diff --git a/.github/workflows/ci-deploy.yml b/.github/workflows/ci-deploy.yml
index 65875fb4b..3f2d3e51d 100644
--- a/.github/workflows/ci-deploy.yml
+++ b/.github/workflows/ci-deploy.yml
@@ -17,7 +17,7 @@ jobs:
     needs: build
     uses: ./.github/workflows/_deploy.yml
     with:
-      image_tag: ${{ needs.build.outputs.image_tag }}
+      image_tag: '${{ needs.build.outputs.image_tag }}@${{ needs.build.outputs.image_digest }}'
       deployment_name: vela
       environment: ${{ github.ref_name == 'main' && 'prod' || 'dev' }}
     secrets: inherit
diff --git a/chart/templates/_helpers.tpl b/chart/templates/_helpers.tpl
index 6c07a864e..5e25eca9a 100644
--- a/chart/templates/_helpers.tpl
+++ b/chart/templates/_helpers.tpl
@@ -3,16 +3,37 @@ Reusable snippets shared across Vela Helm templates.
 */}}
 
 {{/*
-Renders a Postgres init container that waits for the target database to accept connections.
+Returns the plaintext password for a given DB credential key, preserving it across upgrades
+via lookup of the existing vela-controller-secret. On fresh installs the password is derived
+deterministically from the release name, namespace, and key so that all templates in the same
+render produce the same value.
+
+Usage: {{ include "vela.dbPassword" (list "controller-db-password" .) }}
+*/}}
+{{- define "vela.dbPassword" -}}
+{{- $key := index . 0 -}}
+{{- $ctx := index . 1 -}}
+{{- $existingSecret := lookup "v1" "Secret" $ctx.Release.Namespace "vela-controller-secret" -}}
+{{- if and $existingSecret (index $existingSecret.data $key) -}}
+{{- index $existingSecret.data $key | b64dec -}}
+{{- else -}}
+{{- printf "%s-%s-%s" $ctx.Release.Name $ctx.Release.Namespace $key | sha256sum | trunc 32 -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Renders a Postgres init container that waits for the server to be ready.
+When `database` is provided it also waits until that specific database accepts connections.
 The helper accepts a dictionary with the following optional keys:
   - name               : init container name (default: wait-for-database)
   - image              : container image (default: postgres:17-alpine)
   - imagePullPolicy    : pull policy (default: IfNotPresent)
   - host               : database hostname (default: database)
   - port               : database port (default: 5432)
-  - secretName         : Kubernetes secret with credentials (default: database)
-  - usernameKey        : Secret key used for DB username (default: superuser-username)
-  - passwordKey        : Secret key used for DB password (default: superuser-password)
+  - database           : if set, block until this database accepts connections
+  - secretName         : secret containing credentials for the psql check (default: database)
+  - usernameKey        : key for the DB username in secretName (default: superuser-username)
+  - passwordKey        : key for the DB password in secretName (default: superuser-password)
   - securityContext    : optional security context applied to the init container
 */}}
 {{- define "vela.waitForPostgresInitContainer" -}}
@@ -21,6 +42,7 @@ The helper accepts a dictionary with the following optional keys:
 {{- $imagePullPolicy := default "IfNotPresent" .imagePullPolicy -}}
 {{- $host := default "database" .host -}}
 {{- $port := default "5432" .port -}}
+{{- $database := .database -}}
 {{- $secretName := default "database" .secretName -}}
 {{- $usernameKey := default "superuser-username" .usernameKey -}}
 {{- $passwordKey := default "superuser-password" .passwordKey -}}
@@ -32,6 +54,7 @@ The helper accepts a dictionary with the following optional keys:
       value: {{ $host | quote }}
     - name: DB_PORT
       value: {{ $port | quote }}
+{{- if $database }}
     - name: DB_USER
       valueFrom:
         secretKeyRef:
@@ -42,20 +65,22 @@ The helper accepts a dictionary with the following optional keys:
         secretKeyRef:
           name: {{ $secretName }}
           key: {{ $passwordKey }}
+{{- end }}
   command: ["/bin/sh", "-c"]
   args:
     - |
       echo "Waiting for database..."
-      until pg_isready -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER"; do
+      until pg_isready -h "$DB_HOST" -p "$DB_PORT"; do
         sleep 2
       done
       echo "Database is ready"
+{{- if $database }}
 
-      # Ensure postgres user can connect
-      until psql -h "$DB_HOST" -U "$DB_USER" -d postgres -c '\q' 2>/dev/null; do
-        echo "Waiting for Postgres superuser connection..."
+      until psql -h "$DB_HOST" -U "$DB_USER" -d {{ $database | quote }} -c '\q' 2>/dev/null; do
+        echo "Waiting for Postgres connection to {{ $database }}..."
         sleep 2
       done
+{{- end }}
 {{- with .securityContext }}
   securityContext:
 {{ toYaml . | nindent 4 }}
diff --git a/chart/templates/controller.yaml b/chart/templates/controller.yaml
deleted file mode 100644
index 0021b8932..000000000
--- a/chart/templates/controller.yaml
+++ /dev/null
@@ -1,176 +0,0 @@
-apiVersion: v1
-kind: Secret
-metadata:
-  name: vela-controller-secret
-type: Opaque
-data:
-  {{- $existingSecret := lookup "v1" "Secret" .Release.Namespace "vela-controller-secret" }}
-  {{- if $existingSecret }}
-    deployment-password-secret: {{ index $existingSecret.data "deployment-password-secret" }}
-  {{- else }}
-    deployment-password-secret: {{ randAlphaNum 32 | b64enc }}
-  {{- end }}
----
-apiVersion: v1
-kind: ServiceAccount
-metadata:
-  name: vela-controller
-  namespace: {{ .Release.Namespace }}
----
-apiVersion: rbac.authorization.k8s.io/v1
-kind: ClusterRoleBinding
-metadata:
-  name: vela-controller-{{ .Release.Namespace }}-admin
-subjects:
-  - kind: ServiceAccount
-    name: vela-controller
-    namespace: {{ .Release.Namespace }}
-roleRef:
-  apiGroup: rbac.authorization.k8s.io
-  kind: ClusterRole
-  name: cluster-admin
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: vela-controller
-  labels:
-    app.kubernetes.io/name: vela-controller
-    app.kubernetes.io/instance: {{ .Release.Name }}
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app.kubernetes.io/name: vela-controller
-  template:
-    metadata:
-      labels:
-        app.kubernetes.io/name: vela-controller
-        app.kubernetes.io/instance: {{ .Release.Name }}
-    spec:
-      serviceAccountName: vela-controller
-      initContainers:
-{{ include "vela.waitForPostgresInitContainer" (dict) | nindent 8 }}
-      containers:
-        - name: vela-controller
-          image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}"
-          imagePullPolicy: Always
-          ports:
-            - name: http
-              containerPort: 8000
-              protocol: TCP
-          envFrom:
-            - configMapRef:
-                name: vela-controller-config
-          env:
-            - name: VELA_GRAFANA_SECURITY_ADMIN_USER
-              valueFrom:
-                secretKeyRef:
-                  name: vela-grafana-secret
-                  key: VELA_GRAFANA_SECURITY_ADMIN_USER
-            - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
-              valueFrom:
-                secretKeyRef:
-                  name: vela-grafana-secret
-                  key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
-            - name: DB_USER
-              valueFrom:
-                secretKeyRef:
-                  name: database
-                  key: superuser-username
-            - name: DB_PASSWORD
-              valueFrom:
-                secretKeyRef:
-                  name: database
-                  key: superuser-password
-            - name: VELA_DEPLOYMENT_PASSWORD_SECRET
-              valueFrom:
-                secretKeyRef:
-                  name: vela-controller-secret
-                  key: deployment-password-secret
-            - name: DB_HOST
-              value: database
-            - name: VELA_POSTGRES_URL
-              value: 'postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres'
-            - name: VELA_GRAFANA_URL
-              value: "https://{{ .Values.domain }}:{{ .Values.port }}/grafana"
-          livenessProbe:
-            httpGet:
-              path: /health
-              port: http
-            periodSeconds: 5
-            timeoutSeconds: 5
-            failureThreshold: 3
-          readinessProbe:
-            httpGet:
-              path: /health
-              port: http
-            initialDelaySeconds: 10
-          {{- with .Values.containerSecurityContext }}
-          securityContext:
-            {{- toYaml . | nindent 12 }}
-          {{- end }}
-      {{- with .Values.podSecurityContext }}
-      securityContext:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
----
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: vela-controller-config
-data:
-  VELA_ROOT_PATH: {{ .Values.controller.env.VELA_ROOT_PATH | quote }}
-  VELA_JWT_SECRET: {{ .Values.controller.env.VELA_JWT_SECRET | quote }}
-  VELA_GATEWAY_NAME: {{ .Values.gateway.name | quote }}
-  VELA_GATEWAY_NAMESPACE: {{ .Values.gateway.namespace | quote }}
-  VELA_PGMETA_CRYPTO_KEY: {{ .Values.controller.env.VELA_PGMETA_CRYPTO_KEY | quote }}
-  VELA_CORS_ORIGINS: {{ .Values.controller.env.VELA_CORS_ORIGINS | quote }}
-  VELA_KEYCLOAK_URL: {{ .Values.controller.env.VELA_KEYCLOAK_URL | quote }}
-  VELA_KEYCLOAK_ADMIN_NAME: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_NAME | quote }}
-  VELA_KEYCLOAK_ADMIN_SECRET: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_SECRET | quote }}
-  VELA_CLOUDFLARE__API_TOKEN: {{ .Values.controller.env.VELA_CLOUDFLARE_API_TOKEN | quote }}
-  VELA_CLOUDFLARE__ZONE_ID: {{ .Values.controller.env.VELA_CLOUDFLARE_ZONE_ID | quote }}
-  VELA_CLOUDFLARE__BRANCH_REF: {{ .Values.controller.env.VELA_BRANCH_REF | quote }}
-  VELA_CLOUDFLARE__BRANCH_DB_REF: {{ .Values.controller.env.VELA_BRANCH_DB_REF | quote }}
-  VELA_CLOUDFLARE__DOMAIN_SUFFIX: {{ .Values.domainSuffix | default .Values.domain | quote }}
-  VELA_DEPLOYMENT_NAMESPACE_PREFIX: {{ .Values.controller.env.VELA_DEPLOYMENT_NAMESPACE_PREFIX | quote }}
-  VELA_DEPLOYMENT_SERVICE_PORT: {{ .Values.port | quote }}
-  VELA_SERVER_ROOT_URL: "https://{{ .Values.domain }}:{{ .Values.port }}"
-  VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: {{ .Values.controller.env.VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER | quote }}
-  VELA_GRAFANA_URL: {{ .Values.controller.env.VELA_GRAFANA_URL | quote }}
-  VELA_SIMPLYBLOCK_CSI_NAMESPACE: {{ .Values.controller.env.VELA_SIMPLYBLOCK_CSI_NAMESPACE | quote }}
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: vela-controller-service
-spec:
-  selector:
-    app.kubernetes.io/name: vela-controller
-  ports:
-    - name: http
-      port: 8000
-      targetPort: 8000
-      protocol: TCP
-
----
-apiVersion: gateway.networking.k8s.io/v1
-kind: HTTPRoute
-metadata:
-  name: vela-controller
-spec:
-  parentRefs:
-    - name: {{ .Values.gateway.name }}
-      namespace: {{ .Values.gateway.namespace }}
-  hostnames:
-    - "{{ .Values.domain }}"
-  rules:
-    - matches:
-        - path:
-            type: PathPrefix
-            value: /vela
-      backendRefs:
-        - name: vela-controller-service
-          namespace: {{ .Release.Namespace }}
-          port: 8000
diff --git a/chart/templates/controller/configmap.yaml b/chart/templates/controller/configmap.yaml
new file mode 100644
index 000000000..466252610
--- /dev/null
+++ b/chart/templates/controller/configmap.yaml
@@ -0,0 +1,25 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vela-controller-config
+data:
+  VELA_ROOT_PATH: {{ .Values.controller.env.VELA_ROOT_PATH | quote }}
+  VELA_JWT_SECRET: {{ .Values.controller.env.VELA_JWT_SECRET | quote }}
+  VELA_GATEWAY_NAME: {{ .Values.gateway.name | quote }}
+  VELA_GATEWAY_NAMESPACE: {{ .Values.gateway.namespace | quote }}
+  VELA_PGMETA_CRYPTO_KEY: {{ .Values.controller.env.VELA_PGMETA_CRYPTO_KEY | quote }}
+  VELA_CORS_ORIGINS: {{ .Values.controller.env.VELA_CORS_ORIGINS | quote }}
+  VELA_KEYCLOAK_URL: {{ .Values.controller.env.VELA_KEYCLOAK_URL | quote }}
+  VELA_KEYCLOAK_ADMIN_NAME: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_NAME | quote }}
+  VELA_KEYCLOAK_ADMIN_SECRET: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_SECRET | quote }}
+  VELA_CLOUDFLARE__API_TOKEN: {{ .Values.controller.env.VELA_CLOUDFLARE_API_TOKEN | quote }}
+  VELA_CLOUDFLARE__ZONE_ID: {{ .Values.controller.env.VELA_CLOUDFLARE_ZONE_ID | quote }}
+  VELA_CLOUDFLARE__BRANCH_REF: {{ .Values.controller.env.VELA_BRANCH_REF | quote }}
+  VELA_CLOUDFLARE__BRANCH_DB_REF: {{ .Values.controller.env.VELA_BRANCH_DB_REF | quote }}
+  VELA_CLOUDFLARE__DOMAIN_SUFFIX: {{ .Values.domainSuffix | default .Values.domain | quote }}
+  VELA_DEPLOYMENT_NAMESPACE_PREFIX: {{ .Values.controller.env.VELA_DEPLOYMENT_NAMESPACE_PREFIX | quote }}
+  VELA_DEPLOYMENT_SERVICE_PORT: {{ .Values.port | quote }}
+  VELA_SERVER_ROOT_URL: "https://{{ .Values.domain }}:{{ .Values.port }}"
+  VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: {{ .Values.controller.env.VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER | quote }}
+  VELA_GRAFANA_URL: {{ .Values.controller.env.VELA_GRAFANA_URL | quote }}
+  VELA_SIMPLYBLOCK_CSI_NAMESPACE: {{ .Values.controller.env.VELA_SIMPLYBLOCK_CSI_NAMESPACE | quote }}
diff --git a/chart/templates/controller/deployment.yaml b/chart/templates/controller/deployment.yaml
new file mode 100644
index 000000000..161e86367
--- /dev/null
+++ b/chart/templates/controller/deployment.yaml
@@ -0,0 +1,93 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: vela-controller
+  labels:
+    app.kubernetes.io/name: vela-controller
+    app.kubernetes.io/instance: {{ .Release.Name }}
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: vela-controller
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: vela-controller
+        app.kubernetes.io/instance: {{ .Release.Name }}
+    spec:
+      serviceAccountName: vela-controller
+      initContainers:
+{{ include "vela.waitForPostgresInitContainer" (dict
+    "secretName" "vela-controller-secret"
+    "usernameKey" "controller-db-username"
+    "passwordKey" "controller-db-password"
+    "database" "controller"
+  ) | nindent 8 }}
+      containers:
+        - name: vela-controller
+          image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}"
+          imagePullPolicy: Always
+          ports:
+            - name: http
+              containerPort: 8000
+              protocol: TCP
+          envFrom:
+            - configMapRef:
+                name: vela-controller-config
+          env:
+            - name: VELA_GRAFANA_SECURITY_ADMIN_USER
+              valueFrom:
+                secretKeyRef:
+                  name: vela-grafana-secret
+                  key: VELA_GRAFANA_SECURITY_ADMIN_USER
+            - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-grafana-secret
+                  key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
+            - name: CONTROLLER_DB_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: controller-db-password
+            - name: CELERY_DB_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: celery-db-password
+            - name: VELA_DEPLOYMENT_PASSWORD_SECRET
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: deployment-password-secret
+            - name: DB_HOST
+              value: database
+            - name: VELA_POSTGRES_URL
+              value: 'postgresql+asyncpg://vela_controller:$(CONTROLLER_DB_PASSWORD)@$(DB_HOST):5432/controller'
+            - name: VELA_BROKER_URL
+              value: 'sqla+postgresql+psycopg://vela_celery:$(CELERY_DB_PASSWORD)@$(DB_HOST):5432/controller'
+            - name: VELA_RESULT_BACKEND
+              value: 'db+postgresql+psycopg://vela_celery:$(CELERY_DB_PASSWORD)@$(DB_HOST):5432/controller'
+            - name: VELA_GRAFANA_URL
+              value: "https://{{ .Values.domain }}:{{ .Values.port }}/grafana"
+          livenessProbe:
+            httpGet:
+              path: /health
+              port: http
+            periodSeconds: 5
+            timeoutSeconds: 5
+            failureThreshold: 3
+          readinessProbe:
+            httpGet:
+              path: /health
+              port: http
+            initialDelaySeconds: 10
+          {{- with .Values.containerSecurityContext }}
+          securityContext:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+      {{- with .Values.podSecurityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
diff --git a/chart/templates/controller/rbac.yaml b/chart/templates/controller/rbac.yaml
new file mode 100644
index 000000000..9bc42f6fd
--- /dev/null
+++ b/chart/templates/controller/rbac.yaml
@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: vela-controller
+  namespace: {{ .Release.Namespace }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: vela-controller-{{ .Release.Namespace }}-admin
+subjects:
+  - kind: ServiceAccount
+    name: vela-controller
+    namespace: {{ .Release.Namespace }}
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: cluster-admin
diff --git a/chart/templates/controller/secret.yaml b/chart/templates/controller/secret.yaml
new file mode 100644
index 000000000..dfc43083e
--- /dev/null
+++ b/chart/templates/controller/secret.yaml
@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: vela-controller-secret
+  namespace: {{ .Release.Namespace }}
+type: Opaque
+data:
+  {{- $existingSecret := lookup "v1" "Secret" .Release.Namespace "vela-controller-secret" }}
+  deployment-password-secret: {{ if $existingSecret }}{{ index $existingSecret.data "deployment-password-secret" }}{{ else }}{{ randAlphaNum 32 | b64enc }}{{ end }}
+  controller-db-password: {{ include "vela.dbPassword" (list "controller-db-password" .) | b64enc }}
+  celery-db-password: {{ include "vela.dbPassword" (list "celery-db-password" .) | b64enc }}
+  controller-db-username: {{ "vela_controller" | b64enc }}
diff --git a/chart/templates/controller/service.yaml b/chart/templates/controller/service.yaml
new file mode 100644
index 000000000..18532eaca
--- /dev/null
+++ b/chart/templates/controller/service.yaml
@@ -0,0 +1,33 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: vela-controller-service
+spec:
+  selector:
+    app.kubernetes.io/name: vela-controller
+  ports:
+    - name: http
+      port: 8000
+      targetPort: 8000
+      protocol: TCP
+
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: vela-controller
+spec:
+  parentRefs:
+    - name: {{ .Values.gateway.name }}
+      namespace: {{ .Values.gateway.namespace }}
+  hostnames:
+    - "{{ .Values.domain }}"
+  rules:
+    - matches:
+        - path:
+            type: PathPrefix
+            value: /vela
+      backendRefs:
+        - name: vela-controller-service
+          namespace: {{ .Release.Namespace }}
+          port: 8000
diff --git a/chart/templates/controller/worker.yaml b/chart/templates/controller/worker.yaml
new file mode 100644
index 000000000..bd1240e19
--- /dev/null
+++ b/chart/templates/controller/worker.yaml
@@ -0,0 +1,76 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: vela-controller-worker
+  labels:
+    app.kubernetes.io/name: vela-controller-worker
+    app.kubernetes.io/instance: {{ .Release.Name }}
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: vela-controller-worker
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: vela-controller-worker
+        app.kubernetes.io/instance: {{ .Release.Name }}
+    spec:
+      serviceAccountName: vela-controller
+      initContainers:
+{{ include "vela.waitForPostgresInitContainer" (dict
+    "secretName" "vela-controller-secret"
+    "usernameKey" "controller-db-username"
+    "passwordKey" "controller-db-password"
+    "database" "controller"
+  ) | nindent 8 }}
+      containers:
+        - name: vela-controller-worker
+          image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}"
+          imagePullPolicy: Always
+          command: ["celery", "-A", "simplyblock.vela.worker", "worker", "--loglevel=info"]
+          envFrom:
+            - configMapRef:
+                name: vela-controller-config
+          env:
+            - name: VELA_GRAFANA_SECURITY_ADMIN_USER
+              valueFrom:
+                secretKeyRef:
+                  name: vela-grafana-secret
+                  key: VELA_GRAFANA_SECURITY_ADMIN_USER
+            - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-grafana-secret
+                  key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD
+            - name: CONTROLLER_DB_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: controller-db-password
+            - name: CELERY_DB_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: celery-db-password
+            - name: VELA_DEPLOYMENT_PASSWORD_SECRET
+              valueFrom:
+                secretKeyRef:
+                  name: vela-controller-secret
+                  key: deployment-password-secret
+            - name: DB_HOST
+              value: database
+            - name: VELA_POSTGRES_URL
+              value: 'postgresql+asyncpg://vela_controller:$(CONTROLLER_DB_PASSWORD)@$(DB_HOST):5432/controller'
+            - name: VELA_BROKER_URL
+              value: 'sqla+postgresql+psycopg://vela_celery:$(CELERY_DB_PASSWORD)@$(DB_HOST):5432/controller'
+            - name: VELA_RESULT_BACKEND
+              value: 'db+postgresql+psycopg://vela_celery:$(CELERY_DB_PASSWORD)@$(DB_HOST):5432/controller'
+          {{- with .Values.containerSecurityContext }}
+          securityContext:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+      {{- with .Values.podSecurityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
diff --git a/chart/templates/database.yaml b/chart/templates/database.yaml
index ebca909b5..6587e3b98 100644
--- a/chart/templates/database.yaml
+++ b/chart/templates/database.yaml
@@ -46,3 +46,6 @@ spec:
   replication:
     mode: async
     role: ha-read
+  managedSql:
+    scripts:
+    - sgScript: controller-db-init
diff --git a/chart/templates/db-init-script-secret.yaml b/chart/templates/db-init-script-secret.yaml
new file mode 100644
index 000000000..a7d0a9e75
--- /dev/null
+++ b/chart/templates/db-init-script-secret.yaml
@@ -0,0 +1,42 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: vela-db-init-sql
+  namespace: {{ .Release.Namespace }}
+stringData:
+  create-database.sql: |
+    CREATE DATABASE controller;
+
+  init.sql: |
+    DO $$
+    BEGIN
+      IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'vela_controller') THEN
+        CREATE USER vela_controller WITH PASSWORD '{{ include "vela.dbPassword" (list "controller-db-password" .) }}';
+      ELSE
+        ALTER USER vela_controller WITH PASSWORD '{{ include "vela.dbPassword" (list "controller-db-password" .) }}';
+      END IF;
+      IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'vela_celery') THEN
+        CREATE USER vela_celery WITH PASSWORD '{{ include "vela.dbPassword" (list "celery-db-password" .) }}';
+      ELSE
+        ALTER USER vela_celery WITH PASSWORD '{{ include "vela.dbPassword" (list "celery-db-password" .) }}';
+      END IF;
+    END $$;
+
+    GRANT ALL PRIVILEGES ON DATABASE controller TO vela_controller;
+    ALTER DATABASE controller OWNER TO vela_controller;
+    GRANT CONNECT ON DATABASE controller TO vela_celery;
+
+  init-controller-db.sql: |
+    CREATE SCHEMA IF NOT EXISTS celery AUTHORIZATION vela_celery;
+    ALTER ROLE vela_celery IN DATABASE controller SET search_path TO celery;
+
+    DO $$
+    DECLARE r RECORD;
+    BEGIN
+      FOR r IN SELECT tablename FROM pg_tables WHERE schemaname = 'public' LOOP
+        EXECUTE format('ALTER TABLE public.%I OWNER TO vela_controller', r.tablename);
+      END LOOP;
+      FOR r IN SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = 'public' LOOP
+        EXECUTE format('ALTER SEQUENCE public.%I OWNER TO vela_controller', r.sequence_name);
+      END LOOP;
+    END $$;
diff --git a/chart/templates/db-init-sgscript.yaml b/chart/templates/db-init-sgscript.yaml
new file mode 100644
index 000000000..b4354f225
--- /dev/null
+++ b/chart/templates/db-init-sgscript.yaml
@@ -0,0 +1,27 @@
+apiVersion: stackgres.io/v1
+kind: SGScript
+metadata:
+  name: controller-db-init
+  namespace: {{ .Release.Namespace }}
+spec:
+  continueOnError: true
+  scripts:
+  - id: 1
+    name: create-database
+    scriptFrom:
+      secretKeyRef:
+        name: vela-db-init-sql
+        key: create-database.sql
+  - id: 2
+    name: init-users-and-grants
+    scriptFrom:
+      secretKeyRef:
+        name: vela-db-init-sql
+        key: init.sql
+  - id: 3
+    name: init-controller-db-schemas
+    database: controller
+    scriptFrom:
+      secretKeyRef:
+        name: vela-db-init-sql
+        key: init-controller-db.sql
diff --git a/chart/templates/db-migration-job.yaml b/chart/templates/db-migration-job.yaml
new file mode 100644
index 000000000..1343f8fef
--- /dev/null
+++ b/chart/templates/db-migration-job.yaml
@@ -0,0 +1,64 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: vela-db-migration
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    helm.sh/hook: pre-upgrade
+    helm.sh/hook-weight: "0"
+    helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded
+spec:
+  template:
+    spec:
+      restartPolicy: Never
+      containers:
+        - name: db-migration
+          image: postgres:17-alpine
+          env:
+            - name: DB_HOST
+              value: database
+            - name: SUPERUSER
+              valueFrom:
+                secretKeyRef:
+                  name: database
+                  key: superuser-username
+            - name: PGPASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: database
+                  key: superuser-password
+          command: ["/bin/sh", "-c"]
+          args:
+            - |
+              until pg_isready -h "$DB_HOST" -U "$SUPERUSER"; do
+                echo "Waiting for database..."
+                sleep 2
+              done
+
+              # Create controller database if it doesn't exist yet.
+              # The SGScript normally handles this, but it runs asynchronously after
+              # the upgrade so the migration job must bootstrap the database itself.
+              psql -h "$DB_HOST" -U "$SUPERUSER" -d postgres \
+                -c "CREATE DATABASE controller" 2>/dev/null || true
+
+              if psql -h "$DB_HOST" -U "$SUPERUSER" -d controller \
+                   -c "SELECT 1 FROM alembic_version LIMIT 1" 2>/dev/null; then
+                echo "Migration already done, skipping."
+                exit 0
+              fi
+
+              if ! psql -h "$DB_HOST" -U "$SUPERUSER" -d postgres \
+                    -c "SELECT 1 FROM alembic_version LIMIT 1" 2>/dev/null; then
+                echo "No source data found, skipping migration."
+                exit 0
+              fi
+
+              echo "Migrating data from postgres to controller database..."
+              pg_dump -h "$DB_HOST" -U "$SUPERUSER" -d postgres \
+                --schema=public --no-owner --no-privileges \
+                --exclude-table=kombu_message \
+                --exclude-table=kombu_queue \
+                --exclude-table=celery_taskmeta \
+                --exclude-table=celery_tasksetmeta \
+              | psql -h "$DB_HOST" -U "$SUPERUSER" -d controller
+              echo "Migration complete."
diff --git a/chart/templates/vector-config.yaml b/chart/templates/vector-config.yaml
index 9d0a294c8..7b288427a 100644
--- a/chart/templates/vector-config.yaml
+++ b/chart/templates/vector-config.yaml
@@ -126,16 +126,23 @@ data:
         inputs:
           - router.controller
         source: |-
-          parsed, err = parse_regex(.event_message, r'^(?P<time>\d+-\d+-\d+T\d+:\d+:\d+([+|-]\d{2}:?\d{2})|Z)\W+(?P<level>\w+)\W+(?P<msg>.*)$', numeric_groups: true)
+          parsed, err = parse_json(.event_message)
           if err == null {
-              .event_message = parsed.msg
-              .timestamp = now()
-              if exists(parsed.time) {
-                  .timestamp = parse_timestamp!(parsed.time, format: "%FT%T%z")
-              }
-              .level = "info"
-              if exists(parsed.level) {
-                  .level = parsed.level
+              .event_message = parsed.message
+              .timestamp = parse_timestamp!(parsed.timestamp, format: "%FT%T%z")
+              .level = downcase!(parsed.level)
+          } else {
+              parsed, err = parse_regex(.event_message, r'^(?P<time>\d+-\d+-\d+T\d+:\d+:\d+([+|-]\d{2}:?\d{2})|Z)\W+(?P<level>\w+)\W+(?P<msg>.*)$', numeric_groups: true)
+              if err == null {
+                  .event_message = parsed.msg
+                  .timestamp = now()
+                  if exists(parsed.time) {
+                      .timestamp = parse_timestamp!(parsed.time, format: "%FT%T%z")
+                  }
+                  .level = "info"
+                  if exists(parsed.level) {
+                      .level = parsed.level
+                  }
               }
           }
 
diff --git a/containers/Containerfile b/containers/Containerfile
index e0a05a2a2..316362ca4 100644
--- a/containers/Containerfile
+++ b/containers/Containerfile
@@ -17,6 +17,8 @@ RUN --mount=type=cache,target=/var/cache/apt \
 # Installating via the community-apt repository failed when executed as a GitHub action
 RUN curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash
 
+RUN useradd -m -u 1000 vela
+
 WORKDIR /app
 
 COPY .. .
@@ -24,4 +26,6 @@ COPY .. .
 RUN --mount=type=cache,target=/root/.cache/pip pip install .
 RUN rm -r /app/*
 
+USER vela
+
 CMD ["uvicorn", "simplyblock.vela.api:app", "--host", "0.0.0.0", "--proxy-headers"]
diff --git a/containers/compose-dev.yml b/containers/compose-dev.yml
index 523082502..e4b226345 100644
--- a/containers/compose-dev.yml
+++ b/containers/compose-dev.yml
@@ -4,13 +4,15 @@ services:
       context: ..
       dockerfile: containers/Containerfile
     volumes:
-      - ~/.kube/config:/root/.kube/config:ro
+      - ~/.kube/config:/home/vela/.kube/config:ro
       - ..:/app
     environment:
-      VELA_POSTGRES_URL: 'postgresql+asyncpg://postgres:vela@database:5432/vela'
+      VELA_POSTGRES_URL: 'postgresql+asyncpg://vela_controller:controller@database:5432/controller'
       VELA_JWT_SECRET: 'secret'
       VELA_CORS_ORIGINS: '["http://localhost:3000"]'
       VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false'
+      VELA_BROKER_URL: 'sqla+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_RESULT_BACKEND: 'db+postgresql+psycopg://vela_celery:celery@database:5432/controller'
     ports:
       - 8000:8000
     command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && uvicorn --host 0.0.0.0 --reload simplyblock.vela.api:app"'
@@ -25,13 +27,35 @@ services:
     restart: unless-stopped
 
 
+  controller-worker:
+    build:
+      context: ..
+      dockerfile: containers/Containerfile
+    volumes:
+      - ~/.kube/config:/home/vela/.kube/config:ro
+      - ..:/app
+    command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && celery -A simplyblock.vela.worker worker --loglevel=info"'
+    environment:
+      VELA_BROKER_URL: 'sqla+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_RESULT_BACKEND: 'db+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_POSTGRES_URL: 'postgresql+asyncpg://vela_controller:controller@database:5432/controller'
+      VELA_JWT_SECRET: 'secret'
+      VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false'
+    depends_on:
+      database:
+        condition: service_healthy
+    restart: unless-stopped
+
+
   database:
     image: docker.io/postgres:17
     environment:
-      POSTGRES_PASSWORD: vela
-      POSTGRES_DB: vela
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: postgres
+    volumes:
+      - ./init.sql:/docker-entrypoint-initdb.d/90-init.sql:ro
     healthcheck:
-      test: ['CMD-SHELL', 'pg_isready -U postgres -d vela']
+      test: ['CMD-SHELL', 'pg_isready -U postgres -d postgres']
       timeout: 5s
       interval: 5s
       retries: 3
diff --git a/containers/compose.yml b/containers/compose.yml
index e05e1816d..5890892c5 100644
--- a/containers/compose.yml
+++ b/containers/compose.yml
@@ -18,13 +18,13 @@ services:
       context: ..
       dockerfile: containers/Containerfile
     volumes:
-      - ~/.kube/config:/root/.kube/config:ro
+      - ~/.kube/config:/home/vela/.kube/config:ro
       - ..:/app
     command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && uvicorn --host 0.0.0.0 --reload simplyblock.vela.api:app"'
     environment:
       VELA_ROOT_PATH: '/vela'
       VELA_JWT_SECRET: 'http://auth:8080/auth/realms/vela/protocol/openid-connect/certs'
-      VELA_POSTGRES_URL: 'postgresql+asyncpg://postgres:vela@database:5432/vela'
+      VELA_POSTGRES_URL: 'postgresql+asyncpg://vela_controller:controller@database:5432/controller'
       VELA_CORS_ORIGINS: '["http://localhost:3000"]'
       VELA_PGMETA_CRYPTO_KEY: 'secret'
       VELA_KEYCLOAK_URL: 'http://auth:8080/auth/'
@@ -42,6 +42,8 @@ services:
       VELA_GRAFANA_SECURITY_ADMIN_PASSWORD: "password"
       VELA_SIMPLYBLOCK_CSI_NAMESPACE: 'simplyblock-csi'
       VELA_DEPLOYMENT_PASSWORD_SECRET: 'secret'
+      VELA_BROKER_URL: 'sqla+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_RESULT_BACKEND: 'db+postgresql+psycopg://vela_celery:celery@database:5432/controller'
 
     depends_on:
       database:
@@ -54,6 +56,43 @@ services:
     restart: unless-stopped
 
 
+  controller-worker:
+    build:
+      context: ..
+      dockerfile: containers/Containerfile
+    volumes:
+      - ~/.kube/config:/home/vela/.kube/config:ro
+      - ..:/app
+    command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && celery -A simplyblock.vela.worker worker --loglevel=info"'
+    environment:
+      VELA_BROKER_URL: 'sqla+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_RESULT_BACKEND: 'db+postgresql+psycopg://vela_celery:celery@database:5432/controller'
+      VELA_ROOT_PATH: '/vela'
+      VELA_JWT_SECRET: 'http://auth:8080/auth/realms/vela/protocol/openid-connect/certs'
+      VELA_POSTGRES_URL: 'postgresql+asyncpg://vela_controller:controller@database:5432/controller'
+      VELA_CORS_ORIGINS: '["http://localhost:3000"]'
+      VELA_PGMETA_CRYPTO_KEY: 'secret'
+      VELA_KEYCLOAK_URL: 'http://auth:8080/auth/'
+      VELA_KEYCLOAK_ADMIN_NAME: 'admin'
+      VELA_KEYCLOAK_ADMIN_SECRET: 'admin'
+      VELA_CLOUDFLARE__API_TOKEN: 'your_cloudflare_api_token'
+      VELA_CLOUDFLARE__ZONE_ID: 'your_cloudflare_zone_id'
+      VELA_CLOUDFLARE__BRANCH_REF: 'branch-ref.staging.vela.run'
+      VELA_CLOUDFLARE__BRANCH_DB_REF: 'branch-db-ref.staging.vela.run'
+      VELA_CLOUDFLARE__DOMAIN_SUFFIX: 'example.com'
+      VELA_DEPLOYMENT_NAMESPACE_PREFIX: 'vela'
+      VELA_GRAFANA_URL: 'http://grafana:3000'
+      VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false'
+      VELA_GRAFANA_SECURITY_ADMIN_USER: "admin"
+      VELA_GRAFANA_SECURITY_ADMIN_PASSWORD: "password"
+      VELA_SIMPLYBLOCK_CSI_NAMESPACE: 'simplyblock-csi'
+      VELA_DEPLOYMENT_PASSWORD_SECRET: 'secret'
+    depends_on:
+      database:
+        condition: service_healthy
+    restart: unless-stopped
+
+
   auth:
     image: quay.io/keycloak/keycloak:26.4
     environment:
@@ -84,12 +123,12 @@ services:
   database:
     image: docker.io/postgres:17
     environment:
-      POSTGRES_PASSWORD: vela
-      POSTGRES_DB: vela
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: postgres
     volumes:
     - ./init.sql:/docker-entrypoint-initdb.d/90-init.sql:ro
     healthcheck:
-      test: ['CMD-SHELL', 'pg_isready -U postgres -d vela']
+      test: ['CMD-SHELL', 'pg_isready -U postgres -d postgres']
       timeout: 5s
       interval: 5s
       retries: 3
@@ -115,7 +154,7 @@ services:
       GF_AUTH_JWT_JWK_SET_URL: "https://caddy:8443/auth/realms/vela/protocol/openid-connect/certs"
       GF_AUTH_JWT_ALLOW_INSECURE_CERTIFICATE: "true"
       GF_AUTH_JWT_TLS_SKIP_VERIFY_INSECURE: "true"
-    user: "472:0" 
+    user: "472:0"
 
   caddy:
     image: caddy:2
diff --git a/containers/init.sql b/containers/init.sql
index a97f7faf3..cbf3fa28a 100644
--- a/containers/init.sql
+++ b/containers/init.sql
@@ -1,3 +1,16 @@
+CREATE USER vela_controller WITH PASSWORD 'controller';
+CREATE DATABASE controller OWNER vela_controller;
+GRANT ALL PRIVILEGES ON DATABASE controller TO vela_controller;
+
+CREATE USER vela_celery WITH PASSWORD 'celery';
+GRANT CONNECT ON DATABASE controller TO vela_celery;
+
+\c controller
+CREATE SCHEMA IF NOT EXISTS celery AUTHORIZATION vela_celery;
+ALTER ROLE vela_celery IN DATABASE controller SET search_path TO celery;
+
+\c postgres
+
 CREATE USER keycloak WITH PASSWORD 'keycloak';
 CREATE DATABASE keycloak OWNER keycloak;
 GRANT ALL PRIVILEGES ON DATABASE keycloak TO keycloak;
diff --git a/docs/manual-deployment.md b/docs/manual-deployment.md
index e2d804e13..3064ff7dc 100644
--- a/docs/manual-deployment.md
+++ b/docs/manual-deployment.md
@@ -47,6 +47,10 @@ spec:
             - name: proxy
               image: kong/kong-gateway:3.9
               env:
+                - name: KONG_NGINX_HTTP_LARGE_CLIENT_HEADER_BUFFERS
+                  value: "8 32k"
+                - name: KONG_NGINX_HTTP_CLIENT_HEADER_BUFFER_SIZE
+                  value: "32k"
                 - name: KONG_NGINX_PROXY_PROXY_BUFFER_SIZE
                   value: "128k"
                 - name: KONG_NGINX_PROXY_PROXY_BUFFERS
diff --git a/docs/neon-autoscaler.md b/docs/neon-autoscaler.md
index 7777bd3de..0a8105598 100644
--- a/docs/neon-autoscaler.md
+++ b/docs/neon-autoscaler.md
@@ -59,33 +59,33 @@ Demo environment uses the images that are manually deployed.
 # deploy scheduler
 kubectl -n kube-system \
   set image deployment/autoscale-scheduler \
-  autoscale-scheduler=docker.io/simplyblock/autoscaling:autoscale-scheduler-main-cef240f
+  autoscale-scheduler=docker.io/simplyblock/autoscaling:autoscale-scheduler-main-707a485
 kubectl -n kube-system rollout status deployment/autoscale-scheduler 
 
 # deploy autoscaler agent
 kubectl -n kube-system \
   set image daemonset/autoscaler-agent \
-  autoscaler-agent=docker.io/simplyblock/autoscaling:autoscaler-agent-main-cef240f
+  autoscaler-agent=docker.io/simplyblock/autoscaling:autoscaler-agent-main-707a485
 kubectl -n kube-system rollout status daemonset/autoscaler-agent
 
 # deploy vxlan controller
 kubectl -n neonvm-system \
   set image daemonset/neonvm-vxlan-controller \
-  vxlan-controller=docker.io/simplyblock/autoscaling:neonvm-vxlan-controller-main-cef240f
+  vxlan-controller=docker.io/simplyblock/autoscaling:neonvm-vxlan-controller-main-707a485
 kubectl -n neonvm-system rollout status daemonset/neonvm-vxlan-controller
 
 kubectl -n neonvm-system \
   set image deployment/neonvm-controller \
-  manager=docker.io/simplyblock/autoscaling:neonvm-controller-main-cef240f
+  manager=docker.io/simplyblock/autoscaling:neonvm-controller-main-707a485
 kubectl -n neonvm-system rollout status deployment/neonvm-controller
 
 kubectl -n neonvm-system \
   set image daemonset/neonvm-runner-image-loader \
-  neonvm-runner-loader=docker.io/simplyblock/autoscaling:neonvm-runner-main-cef240f
+  neonvm-runner-loader=docker.io/simplyblock/autoscaling:neonvm-runner-main-707a485
 kubectl -n neonvm-system rollout status daemonset/neonvm-runner-image-loader
 
 kubectl -n neonvm-system set env deployment/neonvm-controller \
-  VM_RUNNER_IMAGE=docker.io/simplyblock/autoscaling:neonvm-runner-main-cef240f
+  VM_RUNNER_IMAGE=docker.io/simplyblock/autoscaling:neonvm-runner-main-707a485
 kubectl -n neonvm-system rollout status deployment/neonvm-controller
 ```
 
diff --git a/pyproject.toml b/pyproject.toml
index eb60547e5..90aa5cd04 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -28,6 +28,9 @@ dependencies = [
     'cloudflare',
     'alembic',
     'httpx',
+    'python-json-logger>=4.0',
+    'celery[sqlalchemy]',
+    'asgiref',
 ]
 
 [tool.setuptools]
@@ -86,7 +89,7 @@ testpaths = [
     "tests/test_organizations.py",
     "tests/test_projects.py",
     "tests/test_resources.py",
-    "tests/test_branches.py",
+    "tests/branches",
 ]
 
 [tool.ruff]
diff --git a/src/_util/__init__.py b/src/_util/__init__.py
index 4be2db228..e82c80d8c 100644
--- a/src/_util/__init__.py
+++ b/src/_util/__init__.py
@@ -169,7 +169,7 @@ def validate_ulid(v: Any) -> ULID:
     ),
 ]
 
-Quantity = Annotated[Decimal, BeforeValidator(parse_quantity)]
+Quantity = Annotated[Decimal, BeforeValidator(parse_quantity), PlainSerializer(lambda d: str(d))]
 
 
 def bytes_to_kb(value: int) -> int:
@@ -243,16 +243,6 @@ def quantity_to_milli_cpu(value: str | Decimal | None) -> int | None:
     return int(quantity * Decimal(1000))
 
 
-def quantity_to_bytes(value: str | Decimal | None) -> int | None:
-    """Convert a Kubernetes-style quantity string (e.g. '10Gi', '512Mi') to bytes."""
-
-    quantity = _normalize_quantity(value)
-    if quantity is None:
-        return None
-
-    return int(quantity)
-
-
 def permissive_numeric_timedelta(value: Any) -> Any:
     """Parses the given value into timedelta
 
diff --git a/src/api/__init__.py b/src/api/__init__.py
index 1de6e0c1c..fdd640472 100644
--- a/src/api/__init__.py
+++ b/src/api/__init__.py
@@ -13,12 +13,11 @@
 from fastapi.routing import APIRoute
 from pydantic import BaseModel
 
+from ..database import engine
 from ..deployment.monitors.health import vm_monitor
-from ..deployment.monitors.resize import ResizeMonitor
 from ._util.role import create_access_rights_if_emtpy
 from .backup import router as backup_router
 from .backupmonitor import run_backup_monitor
-from .db import engine
 from .organization import api as organization_api
 from .resources import api as resources_api
 from .resources import monitor_resources
@@ -28,17 +27,30 @@
 
 
 def _logging_config() -> dict[str, Any]:
-    log_format = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
     date_format = "%Y-%m-%dT%H:%M:%S%z"
-    log_level = get_settings().log_level
+    settings = get_settings()
+    log_level = settings.log_level
+    if settings.log_json:
+        formatter: dict[str, Any] = {
+            "()": "pythonjsonlogger.json.JsonFormatter",
+            "fmt": "%(asctime)s %(levelname)s %(name)s %(message)s",
+            "rename_fields": {
+                "asctime": "timestamp",
+                "levelname": "level",
+                "name": "logger",
+            },
+            "datefmt": date_format,
+        }
+    else:
+        formatter = {
+            "format": "%(asctime)s - %(levelname)s - %(name)s - %(message)s",
+            "datefmt": date_format,
+        }
     return {
         "version": 1,
         "disable_existing_loggers": False,
         "formatters": {
-            "default": {
-                "format": log_format,
-                "datefmt": date_format,
-            },
+            "default": formatter,
         },
         "handlers": {
             "default": {
@@ -235,22 +247,18 @@ def health():
 _use_route_names_as_operation_ids(app)
 
 
-_resize_monitor = ResizeMonitor()
-
-
 @app.on_event("startup")
 async def on_startup():
     await _populate_db()
     # start async background monitor
     asyncio.create_task(run_backup_monitor())
     asyncio.create_task(monitor_resources())
-    _resize_monitor.start()
     asyncio.create_task(vm_monitor.run())
 
 
 @app.on_event("shutdown")
 async def on_shutdown():
-    await _resize_monitor.stop()
+    pass
 
 
 __all__ = ["app"]
diff --git a/src/api/_util/backups.py b/src/api/_util/backups.py
index 057196bcf..8b8cc23b3 100644
--- a/src/api/_util/backups.py
+++ b/src/api/_util/backups.py
@@ -17,9 +17,9 @@
 
 if TYPE_CHECKING:
     from ..._util import Identifier
+    from ...database import SessionDep
     from ...models.organization import Organization
     from ...models.project import Project
-    from ..db import SessionDep
 
 
 async def copy_branch_backup_schedules(
diff --git a/src/api/_util/resourcelimit.py b/src/api/_util/resourcelimit.py
index d8f650058..b6c82fa94 100644
--- a/src/api/_util/resourcelimit.py
+++ b/src/api/_util/resourcelimit.py
@@ -12,6 +12,7 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..._util import Identifier, single_or_none
+from ...database import SessionDep
 from ...exceptions import VelaResourceLimitError
 from ...models.branch import Branch, BranchServiceStatus
 from ...models.organization import Organization
@@ -20,6 +21,7 @@
     BranchAllocationPublic,
     BranchProvisioning,
     EntityType,
+    OrganizationLimitDefault,
     ProvisioningLog,
     ResourceLimit,
     ResourceLimitsPublic,
@@ -27,7 +29,6 @@
     ResourceUsageMinute,
     UsageCycle,
 )
-from ..db import SessionDep
 
 
 async def delete_branch_provisioning(session: SessionDep, branch_id: Identifier, *, commit: bool = True):
@@ -134,26 +135,6 @@ async def clone_branch_provisioning(session: SessionDep, source: Branch, target:
     await session.refresh(target)
 
 
-async def initialize_organization_resource_limits(session: SessionDep, organization: Organization):
-    result = await session.execute(select(ResourceLimit).where(ResourceLimit.entity_type == EntityType.system))
-    system_limits = result.scalars().all()
-
-    with session.no_autoflush:
-        for system_limit in system_limits:
-            await session.merge(
-                ResourceLimit(
-                    entity_type=EntityType.org,
-                    org_id=organization.id,
-                    project_id=None,
-                    resource=system_limit.resource,
-                    max_total=system_limit.max_total,
-                    max_per_branch=system_limit.max_per_branch,
-                )
-            )
-    await session.commit()
-    await session.refresh(organization)
-
-
 def dict_to_resource_limits(value: Mapping[ResourceType, int | None]) -> ResourceLimitsPublic:
     return ResourceLimitsPublic(
         milli_vcpu=value.get(ResourceType.milli_vcpu),
@@ -320,7 +301,6 @@ async def get_remaining_project_resources(
     *,
     exclude_branch_ids: Sequence[Identifier] | None = None,
 ) -> ResourceLimitsPublic:
-    system_limits = await get_system_resource_limits(session)
     organization_limits = await get_organization_resource_limits(session, organization_id)
     project_limits = await get_project_resource_limits(session, project_id)
 
@@ -337,16 +317,13 @@ async def get_remaining_project_resources(
 
     effective_limits: dict[ResourceType, int] = {}
     for resource_type in ResourceType:
-        system_limit = system_limits.get(resource_type)
         organization_limit = organization_limits.get(resource_type)
         project_limit = project_limits.get(resource_type)
         per_branch_limit = (
             project_limit.max_per_branch
-            if project_limit and project_limit.max_per_branch is not None
+            if project_limit
             else organization_limit.max_per_branch
-            if organization_limit and organization_limit.max_per_branch is not None
-            else system_limit.max_per_branch
-            if system_limit and system_limit.max_per_branch is not None
+            if organization_limit
             else None
         )
 
@@ -374,17 +351,6 @@ async def get_remaining_project_resources(
     return dict_to_resource_limits(effective_limits)
 
 
-async def get_system_resource_limits(session: SessionDep) -> dict[ResourceType, ResourceLimit]:
-    result = await session.execute(
-        select(ResourceLimit).where(
-            ResourceLimit.entity_type == EntityType.system,
-            ResourceLimit.org_id.is_(None),  # type: ignore[union-attr]
-            ResourceLimit.project_id.is_(None),  # type: ignore[union-attr]
-        )
-    )
-    return _map_resource_limits(list(result.scalars().all()))
-
-
 async def get_organization_resource_limits(
     session: SessionDep, organization_id: Identifier
 ) -> dict[ResourceType, ResourceLimit]:
@@ -638,6 +604,18 @@ def from_database(cls, limits: Sequence[ResourceLimit]) -> Self:
             per_branch=Resources.from_database(limits, "max_per_branch"),
         )
 
+    @classmethod
+    def from_defaults(cls, defaults: Sequence["OrganizationLimitDefault"]) -> Self:
+        return cls(
+            total=Resources.from_database(defaults, "max_total"),
+            per_branch=Resources.from_database(defaults, "max_per_branch"),
+        )
+
+    @classmethod
+    async def organization_defaults(cls, session: AsyncSession) -> Self:
+        result = await session.execute(select(OrganizationLimitDefault))
+        return cls.from_defaults(list(result.scalars().all()))
+
     def to_database(self, entity_type: EntityType) -> list[ResourceLimit]:
         return [
             ResourceLimit(
@@ -651,14 +629,6 @@ def to_database(self, entity_type: EntityType) -> list[ResourceLimit]:
         ]
 
 
-async def system_limits(session: AsyncSession) -> Limits:
-    statement = select(ResourceLimit).where(
-        col(ResourceLimit.org_id).is_(None), col(ResourceLimit.project_id).is_(None)
-    )
-    entries = (await session.exec(statement)).all()
-    return Limits.from_database(entries)
-
-
 async def organization_limits(organization: Organization) -> Limits:
     return Limits.from_database(await organization.awaitable_attrs.limits)
 
@@ -683,11 +653,6 @@ def _allocations():
     )
 
 
-async def system_allocations(session: AsyncSession) -> Resources:
-    allocations = (await session.exec(_allocations())).one()
-    return Resources(**(allocations._asdict()))
-
-
 async def organization_allocations(session: AsyncSession, organization: Organization) -> Resources:
     statement = _allocations().join(Project).where(Project.organization_id == organization.id)
     allocations = (await session.exec(statement)).one()
@@ -700,15 +665,8 @@ async def project_allocations(session: AsyncSession, project: Project) -> Resour
     return Resources(**(allocations._asdict()))
 
 
-async def system_available(session: AsyncSession) -> Resources:
-    return (await system_limits(session)).total - await system_allocations(session)
-
-
 async def organization_available(session: AsyncSession, organization: Organization) -> Resources:
-    return Resources.min(
-        (await organization_limits(organization)).total - await organization_allocations(session, organization),
-        await system_available(session),
-    )
+    return (await organization_limits(organization)).total - await organization_allocations(session, organization)
 
 
 async def project_available(session: AsyncSession, project: Project) -> Resources:
@@ -716,3 +674,15 @@ async def project_available(session: AsyncSession, project: Project) -> Resource
         (await project_limits(project)).total - await project_allocations(session, project),
         await organization_available(session, await project.awaitable_attrs.organization),
     )
+
+
+async def project_branch_maxima(project: Project) -> Resources:
+    """Minimum per-branch limit across the hierarchy (project > organization).
+
+    Returns None for any field where no per-branch limit has been configured at any level.
+    """
+    organization = await project.awaitable_attrs.organization
+    return Resources.min(
+        (await project_limits(project)).per_branch,
+        (await organization_limits(organization)).per_branch,
+    )
diff --git a/src/api/_util/role.py b/src/api/_util/role.py
index 5cb4c0206..a7d4b0462 100644
--- a/src/api/_util/role.py
+++ b/src/api/_util/role.py
@@ -5,9 +5,9 @@
 from ulid import ULID
 
 from ..._util import Identifier
+from ...database import SessionDep
 from ...models.branch import Branch
 from ...models.role import AccessRight, AccessRightPublic, Organization, Role, RoleAccessRight, RoleType, RoleUserLink
-from ..db import SessionDep
 
 
 async def clone_user_role_assignment(
diff --git a/src/api/auth.py b/src/api/auth.py
index 9da6875d9..15b6f32e7 100644
--- a/src/api/auth.py
+++ b/src/api/auth.py
@@ -10,8 +10,8 @@
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from ..database import SessionDep
 from ..models.user import JWT, User
-from .db import SessionDep
 from .settings import get_settings
 
 # HTTPBearer returns 403 instead of 401. Avoid this by raising the error manually
diff --git a/src/api/backup.py b/src/api/backup.py
index 87d20bfad..9f30da70a 100644
--- a/src/api/backup.py
+++ b/src/api/backup.py
@@ -12,6 +12,7 @@
 from sqlmodel import asc, delete, select
 from ulid import ULID
 
+from ..database import SessionDep
 from ..models._util import Identifier
 from ..models.backups import (
     BackupCreatePublic,
@@ -38,7 +39,6 @@
     create_branch_snapshot_bundle,
     delete_branch_snapshot,
 )
-from .db import SessionDep
 from .dependencies import OrganizationDep
 
 router = APIRouter(dependencies=[Depends(authenticated_user)], tags=["backup"])
diff --git a/src/api/backup_snapshots.py b/src/api/backup_snapshots.py
index 6e38bca59..04c7c8f26 100644
--- a/src/api/backup_snapshots.py
+++ b/src/api/backup_snapshots.py
@@ -9,8 +9,8 @@
 from uuid import UUID
 
 import httpx
+from kubernetes.utils import parse_quantity
 
-from .._util import Identifier, quantity_to_bytes
 from ..deployment import AUTOSCALER_PVC_SUFFIX, AUTOSCALER_WAL_PVC_SUFFIX, get_autoscaler_vm_identity
 from ..deployment.kubernetes.snapshot import (
     create_snapshot_from_pvc,
@@ -26,6 +26,7 @@
 if TYPE_CHECKING:
     from ulid import ULID
 
+    from .._util import Identifier
     from ..models.backups import BackupEntry
 
 logger = logging.getLogger(__name__)
@@ -154,7 +155,7 @@ async def _create_snapshot_for_pvc(
     status = snapshot.get("status") or {}
     content_name_payload = status.get("boundVolumeSnapshotContentName")
     content_name = content_name_payload if isinstance(content_name_payload, str) else None
-    size_bytes = quantity_to_bytes(status.get("restoreSize"))
+    size_bytes = parse_quantity(status["restoreSize"])
 
     try:
         if content_name is None:
diff --git a/src/api/dependencies.py b/src/api/dependencies.py
index 50c20b165..37758d12e 100644
--- a/src/api/dependencies.py
+++ b/src/api/dependencies.py
@@ -6,6 +6,7 @@
 from sqlmodel import select
 
 from .._util import Identifier
+from ..database import SessionDep
 from ..models.backups import BackupEntry
 from ..models.branch import Branch, BranchApiKey, BranchServiceStatus
 from ..models.organization import Organization
@@ -13,7 +14,6 @@
 from ..models.role import Role
 from ..models.user import User
 from .auth import authenticated_user
-from .db import SessionDep
 
 
 async def organization_lookup(session: SessionDep, organization_id: Identifier) -> Organization:
diff --git a/src/api/organization/__init__.py b/src/api/organization/__init__.py
index de0de2d43..29c4a74c7 100644
--- a/src/api/organization/__init__.py
+++ b/src/api/organization/__init__.py
@@ -13,9 +13,9 @@
 from ...deployment import delete_deployment
 from ...models.audit import OrganizationAuditLog
 from ...models.organization import Organization, OrganizationCreate, OrganizationUpdate
-from ...models.resources import ResourceTypePublic, ResourceUsageMinute
+from ...models.resources import EntityType, ResourceTypePublic, ResourceUsageMinute
 from .._util import Conflict, Forbidden, NotFound, Unauthenticated, url_path_for
-from .._util.resourcelimit import initialize_organization_resource_limits
+from .._util.resourcelimit import Limits
 from .._util.role import create_organization_admin_role
 from ..auth import authenticated_user
 from ..dependencies import AuthUserDep, OrganizationDep, SessionDep
@@ -87,7 +87,12 @@ async def create(
     user: AuthUserDep,
     response: Literal["empty", "full"] = "empty",
 ) -> JSONResponse:
-    entity = Organization(**parameters.model_dump(), users=[user])
+    default_limits = await Limits.organization_defaults(session)
+    entity = Organization(
+        **parameters.model_dump(),
+        users=[user],
+        limits=default_limits.to_database(EntityType.org),
+    )
     session.add(entity)
     try:
         await session.commit()
@@ -108,9 +113,6 @@ async def create(
     await session.commit()
     await session.refresh(entity)
 
-    # Set up initial organization resource limits
-    await initialize_organization_resource_limits(session, entity)
-
     entity_url = url_path_for(request, "organizations:detail", organization_id=entity.id)
     return JSONResponse(
         content=entity.model_dump() if response == "full" else None,
diff --git a/src/api/organization/member.py b/src/api/organization/member.py
index 36cb1d3dc..39030fd28 100644
--- a/src/api/organization/member.py
+++ b/src/api/organization/member.py
@@ -6,10 +6,10 @@
 from fastapi.responses import JSONResponse
 from sqlalchemy.exc import IntegrityError
 
+from ...database import SessionDep
 from ...models.user import UserID, UserPublic
 from .._util import Forbidden, NotFound, Unauthenticated
 from ..auth import authenticated_user, user_by_id
-from ..db import SessionDep
 from ..dependencies import MemberDep, OrganizationDep, UserDep
 from ..user import public_list as public_user_list
 
diff --git a/src/api/organization/project/__init__.py b/src/api/organization/project/__init__.py
index e07bc8911..c9d8222f7 100644
--- a/src/api/organization/project/__init__.py
+++ b/src/api/organization/project/__init__.py
@@ -10,6 +10,7 @@
 from sqlmodel import select
 
 from ...._util import Identifier
+from ....database import SessionDep
 from ....deployment import delete_deployment, get_autoscaler_vm_identity
 from ....deployment.kubernetes.neonvm import set_virtualmachine_power_state
 from ....exceptions import VelaKubernetesError
@@ -28,7 +29,6 @@
     get_project_limit_totals,
 )
 from ...auth import security
-from ...db import SessionDep
 from ...dependencies import OrganizationDep, ProjectDep
 from . import branch as branch_module
 from .resources import api as resources_api
diff --git a/src/api/organization/project/branch/__init__.py b/src/api/organization/project/branch/__init__.py
index a274dede7..737532843 100644
--- a/src/api/organization/project/branch/__init__.py
+++ b/src/api/organization/project/branch/__init__.py
@@ -17,15 +17,14 @@
 from fastapi.security import HTTPAuthorizationCredentials
 from keycloak.exceptions import KeycloakError
 from kubernetes_asyncio.client.exceptions import ApiException
-from pydantic import AfterValidator, ValidationError
+from pydantic import AfterValidator
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import select
 
 from ....._util import DEFAULT_DB_NAME, DEFAULT_DB_USER, Identifier, storage_backend_bytes_to_db_bytes
 from ....._util.crypto import encrypt_with_passphrase, generate_keys
+from .....database import AsyncSessionLocal, SessionDep
 from .....deployment import (
-    AUTOSCALER_PVC_SUFFIX,
-    STORAGE_PVC_SUFFIX,
     DeploymentParameters,
     ResizeParameters,
     branch_api_domain,
@@ -39,7 +38,6 @@
     kube_service,
     resolve_branch_database_volume_size,
     update_branch_database_password,
-    update_branch_volume_iops,
 )
 from .....deployment._util import deployment_namespace
 from .....deployment.health import (
@@ -64,9 +62,6 @@
     BranchPgbouncerConfigStatus,
     BranchPgbouncerConfigUpdate,
     BranchPublic,
-    BranchResizeService,
-    BranchResizeStatus,
-    BranchResizeStatusEntry,
     BranchServiceStatus,
     BranchSourceDeploymentParameters,
     BranchStatus,
@@ -75,7 +70,6 @@
     CapaResizeKey,
     DatabaseInformation,
     PgbouncerConfig,
-    aggregate_resize_statuses,
 )
 from .....models.resources import BranchAllocationPublic, ResourceLimitsPublic, ResourceType
 from ...._util import Conflict, Forbidden, NotFound, Unauthenticated, url_path_for
@@ -96,12 +90,13 @@
     SNAPSHOT_TIMEOUT_SEC as _SNAPSHOT_TIMEOUT_SECONDS,
 )
 from ....backup_snapshots import branch_snapshots_used_size
-from ....db import AsyncSessionLocal, SessionDep
 from ....dependencies import BranchDep, OrganizationDep, ProjectDep, backup_lookup, branch_lookup
 from ....keycloak import realm_admin
 from ....settings import get_settings as get_api_settings
 from .api_keys import api as api_key_api
 from .auth import api as auth_api
+from .resize_tasks import dispatch_resize
+from .tasks import task_api
 
 api = APIRouter(tags=["branch"])
 
@@ -120,11 +115,6 @@
     BranchServiceStatus.RESIZING,
 }
 _PROTECTED_BRANCH_STATUSES: set[BranchServiceStatus] = {BranchServiceStatus.PAUSED}
-_ACTIVE_RESIZE_STATUSES: set[BranchResizeStatus] = {
-    "PENDING",
-    "RESIZING",
-    "FILESYSTEM_RESIZE_PENDING",
-}
 _CREATING_STATUS_ERROR_GRACE_PERIOD = timedelta(minutes=5)
 _STARTING_STATUS_ERROR_GRACE_PERIOD = timedelta(minutes=5)
 
@@ -190,13 +180,19 @@ async def _cleanup_failed_branch_deployment(branch_id: Identifier) -> None:
 def _should_update_branch_status(
     current: BranchServiceStatus,
     derived: BranchServiceStatus,
-    resize_status: BranchResizeStatus,
+    *,
+    resize_in_progress: bool = True,
 ) -> bool:
     if current == derived:
         return False
-    if current == BranchServiceStatus.RESIZING and resize_status in _ACTIVE_RESIZE_STATUSES:
-        # Keep the explicit RESIZING while a resize is still in progress
-        # unless we detect a hard failure.
+    if current == BranchServiceStatus.RESIZING:
+        if not resize_in_progress:
+            return derived in {
+                BranchServiceStatus.ACTIVE_HEALTHY,
+                BranchServiceStatus.ACTIVE_UNHEALTHY,
+                BranchServiceStatus.STOPPED,
+                BranchServiceStatus.ERROR,
+            }
         return derived == BranchServiceStatus.ERROR
     if current == BranchServiceStatus.STARTING and derived == BranchServiceStatus.STOPPED:
         logger.debug("Ignoring STARTING -> STOPPED transition detected by branch status monitor")
@@ -282,38 +278,14 @@ async def _refresh_branch_status(branch: Branch) -> BranchServiceStatus:
         status,
     )
 
-    if _should_update_branch_status(
-        current_status,
-        status,
-        resize_status=branch.resize_status,
-    ):
+    resize_in_progress = branch.resize_task_id is not None
+    if _should_update_branch_status(current_status, status, resize_in_progress=resize_in_progress):
         branch.set_status(status)
         return status
 
     return current_status
 
 
-def _normalize_resize_statuses(branch: Branch) -> dict[str, BranchResizeStatusEntry]:
-    statuses = branch.resize_statuses or {}
-    if not statuses:
-        return {}
-
-    normalized: dict[str, BranchResizeStatusEntry] = {}
-    for service, entry in statuses.items():
-        if isinstance(entry, BranchResizeStatusEntry):
-            normalized[service] = entry
-            continue
-        try:
-            normalized[service] = BranchResizeStatusEntry.model_validate(entry)
-        except ValidationError:
-            logger.warning(
-                "Skipping invalid resize status entry for branch %s service %s",
-                branch.id,
-                service,
-            )
-    return normalized
-
-
 _DEFAULT_SERVICE_STATUS = BranchStatus(
     database=BranchServiceStatus.UNKNOWN,
     storage=BranchServiceStatus.UNKNOWN,
@@ -717,96 +689,6 @@ async def _build_branch_entity(
     return entity
 
 
-_PARAMETER_TO_SERVICE: dict[CapaResizeKey, BranchResizeService] = {
-    "database_size": "database_disk_resize",
-    "storage_size": "storage_api_disk_resize",
-    "milli_vcpu": "database_cpu_resize",
-    "memory_bytes": "database_memory_resize",
-    "iops": "database_iops_resize",
-}
-
-
-def _track_resize_change(
-    *,
-    parameter_key: CapaResizeKey,
-    new_value: int | None,
-    current_value: int | None,
-    statuses: dict[str, dict[str, Any]],
-    effective: dict[CapaResizeKey, int],
-    timestamp: str,
-) -> None:
-    service_key = _PARAMETER_TO_SERVICE[parameter_key]
-    if new_value is None:
-        return
-    if new_value != current_value:
-        effective[parameter_key] = new_value
-        entry: dict[str, Any] = {"status": "PENDING", "timestamp": timestamp, "requested_at": timestamp}
-        statuses[service_key] = entry
-    elif statuses.get(service_key, {}).get("status") == "PENDING":
-        statuses.pop(service_key, None)
-
-
-async def _apply_resize_operations(
-    session: SessionDep,
-    branch: Branch,
-    effective_parameters: dict[CapaResizeKey, int],
-) -> None:
-    namespace, autoscaler_vm_name = get_autoscaler_vm_identity(branch.id)
-    if "database_size" in effective_parameters:
-        new_database_size = effective_parameters["database_size"]
-        pvc_name = f"{autoscaler_vm_name}{AUTOSCALER_PVC_SUFFIX}"
-        storage_size_bytes = str(new_database_size)
-        await kube_service.resize_pvc_storage(namespace, pvc_name, storage_size_bytes)
-
-    if "storage_size" in effective_parameters:
-        new_storage_size = effective_parameters["storage_size"]
-        pvc_name = f"{autoscaler_vm_name}{STORAGE_PVC_SUFFIX}"
-        storage_size_bytes = str(new_storage_size)
-        await kube_service.resize_pvc_storage(namespace, pvc_name, storage_size_bytes)
-
-    if "iops" in effective_parameters:
-        new_iops = effective_parameters["iops"]
-        await update_branch_volume_iops(branch.id, new_iops)
-        branch.iops = new_iops
-        await create_or_update_branch_provisioning(
-            session,
-            branch,
-            ResourceLimitsPublic(iops=new_iops),
-            commit=False,
-        )
-
-    milli_vcpu = effective_parameters.get("milli_vcpu", branch.milli_vcpu)
-    memory = effective_parameters.get("memory_bytes", branch.memory)
-
-    cpu_changed = "milli_vcpu" in effective_parameters
-    memory_changed = "memory_bytes" in effective_parameters
-
-    if cpu_changed or memory_changed:
-        await kube_service.resize_autoscaler_vm(
-            namespace,
-            autoscaler_vm_name,
-            cpu_milli=milli_vcpu,
-            memory_bytes=memory,
-        )
-        if cpu_changed:
-            branch.milli_vcpu = milli_vcpu
-            await create_or_update_branch_provisioning(
-                session,
-                branch,
-                ResourceLimitsPublic(milli_vcpu=milli_vcpu),
-                commit=False,
-            )
-
-        if memory_changed:
-            branch.memory = memory
-            await create_or_update_branch_provisioning(
-                session,
-                branch,
-                ResourceLimitsPublic(ram=memory),
-                commit=False,
-            )
-
-
 async def _deploy_branch_environment_task(
     *,
     organization_id: Identifier,
@@ -1671,13 +1553,8 @@ async def status(
     _project: ProjectDep,
     branch: BranchDep,
 ) -> BranchStatusPublic:
-    normalized_resize_statuses = _normalize_resize_statuses(branch)
     service_status = await collect_branch_service_health(branch.id)
-    return BranchStatusPublic(
-        resize_status=branch.resize_status,
-        resize_statuses=normalized_resize_statuses,
-        service_status=service_status,
-    )
+    return BranchStatusPublic(service_status=service_status)
 
 
 @instance_api.post(
@@ -1927,15 +1804,14 @@ async def update_pgbouncer_config(
 )
 async def resize(
     session: SessionDep,
-    _organization: OrganizationDep,
-    _project: ProjectDep,
+    request: Request,
+    organization: OrganizationDep,
+    project: ProjectDep,
     parameters: ResizeParameters,
     branch: BranchDep,
 ):
-    branch_in_session = await session.merge(branch)
-
     if parameters.database_size is not None:
-        current_database_size = branch_in_session.database_size
+        current_database_size = branch.database_size
         requested_database_size = parameters.database_size
         if current_database_size is not None and requested_database_size < current_database_size:
             raise HTTPException(
@@ -1947,7 +1823,7 @@ async def resize(
             )
 
     if parameters.storage_size is not None:
-        current_storage = branch_in_session.storage_size
+        current_storage = branch.storage_size
         requested_storage = parameters.storage_size
         if current_storage is not None and requested_storage < current_storage:
             raise HTTPException(
@@ -1958,72 +1834,41 @@ async def resize(
                 ),
             )
 
-    updated_statuses = dict(branch_in_session.resize_statuses or {})
-    timestamp = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+    # Guard against concurrent resizes
+    if branch.resize_task_id is not None:
+        raise HTTPException(status_code=400, detail="A resize operation is already in progress")
 
     effective_parameters: dict[CapaResizeKey, int] = {}
+    for param_key, new_val, current_val in [
+        ("database_size", parameters.database_size, branch.database_size),
+        ("storage_size", parameters.storage_size, branch.storage_size),
+        ("milli_vcpu", parameters.milli_vcpu, branch.milli_vcpu),
+        ("memory_bytes", parameters.memory_bytes, branch.memory),
+        ("iops", parameters.iops, branch.iops),
+    ]:
+        if new_val is not None and new_val != current_val:
+            effective_parameters[cast("CapaResizeKey", param_key)] = new_val
 
-    _track_resize_change(
-        parameter_key="database_size",
-        new_value=parameters.database_size,
-        current_value=branch_in_session.database_size,
-        statuses=updated_statuses,
-        effective=effective_parameters,
-        timestamp=timestamp,
-    )
-    _track_resize_change(
-        parameter_key="storage_size",
-        new_value=parameters.storage_size,
-        current_value=branch_in_session.storage_size,
-        statuses=updated_statuses,
-        effective=effective_parameters,
-        timestamp=timestamp,
-    )
-    _track_resize_change(
-        parameter_key="milli_vcpu",
-        new_value=parameters.milli_vcpu,
-        current_value=branch_in_session.milli_vcpu,
-        statuses=updated_statuses,
-        effective=effective_parameters,
-        timestamp=timestamp,
-    )
-    _track_resize_change(
-        parameter_key="memory_bytes",
-        new_value=parameters.memory_bytes,
-        current_value=branch_in_session.memory,
-        statuses=updated_statuses,
-        effective=effective_parameters,
-        timestamp=timestamp,
-    )
-    _track_resize_change(
-        parameter_key="iops",
-        new_value=parameters.iops,
-        current_value=branch_in_session.iops,
-        statuses=updated_statuses,
-        effective=effective_parameters,
-        timestamp=timestamp,
-    )
+    if not effective_parameters:
+        return Response(status_code=202)
+
+    await _ensure_resize_resource_limits(session, branch, effective_parameters)
+
+    task_id = dispatch_resize(str(branch.id), dict(effective_parameters))
 
-    if effective_parameters:
-        await _ensure_resize_resource_limits(session, branch_in_session, effective_parameters)
-        await _apply_resize_operations(session, branch_in_session, effective_parameters)
-        completion_timestamp = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
-        for param_key, service_key in (
-            ("iops", "database_iops_resize"),
-            ("milli_vcpu", "database_cpu_resize"),
-            ("memory_bytes", "database_memory_resize"),
-        ):
-            if param_key in effective_parameters:
-                updated_statuses[service_key] = {
-                    "status": "COMPLETED",
-                    "timestamp": completion_timestamp,
-                }
-
-    branch_in_session.resize_statuses = updated_statuses
-    branch_in_session.resize_status = aggregate_resize_statuses(updated_statuses)
-    branch_in_session.set_status(BranchServiceStatus.RESIZING)
+    branch.set_status(BranchServiceStatus.RESIZING)
+    branch.resize_task_id = task_id
     await session.commit()
-    return Response(status_code=202)
+
+    task_url = url_path_for(
+        request,
+        "organizations:projects:branch:tasks:detail",
+        organization_id=await organization.awaitable_attrs.id,
+        project_id=await project.awaitable_attrs.id,
+        branch_id=await branch.awaitable_attrs.id,
+        task_id=task_id,
+    )
+    return Response(status_code=202, headers={"Location": task_url})
 
 
 _control_responses: dict[int | str, dict[str, Any]] = {
@@ -2139,6 +1984,7 @@ async def control_branch(
 
 instance_api.include_router(auth_api, prefix="/auth")
 instance_api.include_router(api_key_api, prefix="/apikeys")
+instance_api.include_router(task_api, prefix="/tasks")
 
 
 api.include_router(instance_api)
diff --git a/src/api/organization/project/branch/resize_tasks.py b/src/api/organization/project/branch/resize_tasks.py
new file mode 100644
index 000000000..d45d78105
--- /dev/null
+++ b/src/api/organization/project/branch/resize_tasks.py
@@ -0,0 +1,130 @@
+"""Celery chord tasks for proactive branch resize completion.
+
+The chord dispatches one sub-task per resource type (in the deployment layer)
+and calls ``finalize_resize`` when all sub-tasks have settled.  With
+``task_chord_propagates = False`` the callback always fires, even on partial
+failure, so succeeded fields are persisted and ``resize_task_id`` is cleared
+regardless of individual sub-task outcomes.
+"""
+
+import logging
+from uuid import UUID
+
+from asgiref.sync import async_to_sync
+from celery import chord
+from ulid import ULID
+
+from .....database import AsyncSessionLocal
+from .....deployment.health import collect_branch_service_health, derive_branch_status_from_services
+from .....deployment.resize import resize_cpu_memory, resize_database_pvc, resize_iops, resize_storage_pvc
+from .....models.branch import Branch, BranchServiceStatus
+from .....models.resources import ResourceLimitsPublic
+from .....worker import app
+from ...._util.resourcelimit import create_or_update_branch_provisioning
+
+logger = logging.getLogger(__name__)
+
+_FIELD_TO_BRANCH_ATTR: dict[str, str] = {
+    "database_size": "database_size",
+    "storage_size": "storage_size",
+    "milli_vcpu": "milli_vcpu",
+    "memory_bytes": "memory",
+    "iops": "iops",
+}
+
+_FIELD_TO_RESOURCE_LIMITS_KWARG: dict[str, str] = {
+    "database_size": "database_size",
+    "storage_size": "storage_size",
+    "milli_vcpu": "milli_vcpu",
+    "memory_bytes": "ram",
+    "iops": "iops",
+}
+
+
+async def _apply_succeeded_fields(session: object, branch: Branch, succeeded: dict[str, int]) -> None:
+    """Write each succeeded resize field onto the branch and update provisioning records."""
+    for field, value in succeeded.items():
+        branch_attr = _FIELD_TO_BRANCH_ATTR.get(field)
+        if branch_attr is None:
+            continue
+        if getattr(branch, branch_attr) == value:
+            continue
+        setattr(branch, branch_attr, value)
+        limits_kwarg = _FIELD_TO_RESOURCE_LIMITS_KWARG[field]
+        await create_or_update_branch_provisioning(
+            session,  # type: ignore[arg-type]
+            branch,
+            ResourceLimitsPublic(**{limits_kwarg: value}),
+            commit=False,
+        )
+
+
+async def _async_finalize_resize(
+    job_results: list,
+    branch_id: str,
+    effective_parameters: dict,
+    job_fields: list[list[str]],
+) -> dict:
+    succeeded: dict[str, int] = {}
+    errors: list[str] = []
+
+    for result, fields in zip(job_results, job_fields, strict=False):
+        if isinstance(result, Exception):
+            errors.append(str(result))
+        else:
+            for field in fields:
+                if field in effective_parameters:
+                    succeeded[field] = effective_parameters[field]
+
+    async with AsyncSessionLocal() as session:
+        branch = await session.get(Branch, ULID.from_str(branch_id))
+        if branch is None:
+            logger.error("Branch %s not found during finalize_resize", branch_id)
+            return {**succeeded, "errors": errors}
+
+        await _apply_succeeded_fields(session, branch, succeeded)
+
+        service_status = await collect_branch_service_health(branch.id)
+        if branch.status == BranchServiceStatus.RESIZING:
+            branch.set_status(
+                derive_branch_status_from_services(service_status, storage_enabled=branch.enable_file_storage)
+            )
+
+        branch.resize_task_id = None
+        await session.commit()
+
+    return {**succeeded, "errors": errors}
+
+
+@app.task(name="simplyblock.vela.branch.resize.finalize")
+def finalize_resize(
+    job_results: list,
+    branch_id: str,
+    effective_parameters: dict,
+    job_fields: list[list[str]],
+) -> dict:
+    """Chord callback: persist succeeded fields and clear resize_task_id."""
+    return async_to_sync(_async_finalize_resize)(job_results, branch_id, effective_parameters, job_fields)
+
+
+def dispatch_resize(branch_id: str, effective_parameters: dict) -> UUID:
+    """Build and dispatch a chord of resize sub-tasks; return the chord result UUID."""
+    jobs = []
+    job_fields: list[list[str]] = []
+
+    ep = effective_parameters
+    if "milli_vcpu" in ep or "memory_bytes" in ep:
+        jobs.append(resize_cpu_memory.s(branch_id, ep.get("milli_vcpu"), ep.get("memory_bytes")))
+        job_fields.append([f for f in ("milli_vcpu", "memory_bytes") if f in ep])
+    if "iops" in ep:
+        jobs.append(resize_iops.s(branch_id, ep["iops"]))
+        job_fields.append(["iops"])
+    if "database_size" in ep:
+        jobs.append(resize_database_pvc.s(branch_id, ep["database_size"]))
+        job_fields.append(["database_size"])
+    if "storage_size" in ep:
+        jobs.append(resize_storage_pvc.s(branch_id, ep["storage_size"]))
+        job_fields.append(["storage_size"])
+
+    result = chord(jobs)(finalize_resize.s(branch_id=branch_id, effective_parameters=ep, job_fields=job_fields))
+    return UUID(result.id)
diff --git a/src/api/organization/project/branch/tasks.py b/src/api/organization/project/branch/tasks.py
new file mode 100644
index 000000000..38351a52b
--- /dev/null
+++ b/src/api/organization/project/branch/tasks.py
@@ -0,0 +1,86 @@
+"""Branch task list/detail endpoints.
+
+Exposes Celery task state (currently resize only) under:
+  GET .../branches/{branch_id}/tasks
+  GET .../branches/{branch_id}/tasks/{task_id}
+"""
+
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from ...._util import Forbidden, NotFound, Unauthenticated
+from ....dependencies import BranchDep, OrganizationDep, ProjectDep
+from .resize_tasks import finalize_resize
+
+task_api = APIRouter(tags=["branch"])
+
+_CELERY_STATE_TO_STATUS: dict[str, str] = {
+    "PENDING": "PENDING",
+    "STARTED": "STARTED",
+    "SUCCESS": "COMPLETED",
+    "FAILURE": "FAILED",
+    "REVOKED": "FAILED",
+}
+
+
+class BranchTaskPublic(BaseModel):
+    id: UUID
+    task_type: str
+    status: str
+    parameters: dict
+    result: Any | None
+    error: str | None
+    date_done: datetime | None
+
+
+def _build_task_public(task_id: UUID) -> BranchTaskPublic:
+    result = finalize_resize.AsyncResult(str(task_id))
+    state = result.state
+    status = _CELERY_STATE_TO_STATUS.get(state, state)
+    kwargs: dict = result.kwargs or {}
+    return BranchTaskPublic(
+        id=task_id,
+        task_type="resize",
+        status=status,
+        parameters=kwargs.get("effective_parameters", {}),
+        result=result.result if state == "SUCCESS" else None,
+        error=str(result.traceback) if state == "FAILURE" and result.traceback else None,
+        date_done=result.date_done,
+    )
+
+
+@task_api.get(
+    "/",
+    name="organizations:projects:branch:tasks:list",
+    response_model=list[BranchTaskPublic],
+    responses={401: Unauthenticated, 403: Forbidden, 404: NotFound},
+)
+async def list_tasks(
+    _organization: OrganizationDep,
+    _project: ProjectDep,
+    branch: BranchDep,
+) -> list[BranchTaskPublic]:
+    if branch.resize_task_id is None:
+        return []
+    return [_build_task_public(branch.resize_task_id)]
+
+
+@task_api.get(
+    "/{task_id}",
+    name="organizations:projects:branch:tasks:detail",
+    response_model=BranchTaskPublic,
+    responses={401: Unauthenticated, 403: Forbidden, 404: NotFound},
+)
+async def get_task(
+    _organization: OrganizationDep,
+    _project: ProjectDep,
+    branch: BranchDep,
+    task_id: UUID,
+) -> BranchTaskPublic:
+    if branch.resize_task_id != task_id:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return _build_task_public(task_id)
diff --git a/src/api/organization/project/resources.py b/src/api/organization/project/resources.py
index 19d657f6a..6dc201b62 100644
--- a/src/api/organization/project/resources.py
+++ b/src/api/organization/project/resources.py
@@ -2,7 +2,14 @@
 
 from ....models.resources import EntityType
 from ..._util import Forbidden, NotFound, Unauthenticated
-from ..._util.resourcelimit import Limits, Resources, project_allocations, project_available, project_limits
+from ..._util.resourcelimit import (
+    Limits,
+    Resources,
+    project_allocations,
+    project_available,
+    project_branch_maxima,
+    project_limits,
+)
 from ...dependencies import ProjectDep, SessionDep
 
 api = APIRouter()
@@ -49,3 +56,12 @@ async def allocations(session: SessionDep, project: ProjectDep) -> Resources:
 )
 async def available(session: SessionDep, project: ProjectDep) -> Resources:
     return await project_available(session, project)
+
+
+@api.get(
+    "/branch-maxima/",
+    name="organizations:projects:resources:branch-maxima",
+    responses={401: Unauthenticated, 403: Forbidden, 404: NotFound},
+)
+async def branch_maxima(project: ProjectDep) -> Resources:
+    return await project_branch_maxima(project)
diff --git a/src/api/organization/role.py b/src/api/organization/role.py
index 16470d29f..9ccabdb6e 100644
--- a/src/api/organization/role.py
+++ b/src/api/organization/role.py
@@ -7,6 +7,7 @@
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from ...database import SessionDep
 from ...models._util import Identifier
 from ...models.role import (
     AccessRight,
@@ -29,7 +30,6 @@
 from .._util import Forbidden, NotFound, Unauthenticated
 from ..access_right_utils import check_access
 from ..auth import authenticated_user
-from ..db import SessionDep
 from ..dependencies import OrganizationDep, RoleDep, branch_lookup, project_lookup
 
 api = APIRouter(dependencies=[Depends(authenticated_user)], tags=["role"])
diff --git a/src/api/resources.py b/src/api/resources.py
index 12c5b1d4f..5ac88ad5f 100644
--- a/src/api/resources.py
+++ b/src/api/resources.py
@@ -6,12 +6,14 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException
+from kubernetes.utils import parse_quantity
 from kubernetes_asyncio.client.exceptions import ApiException
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlmodel import select
 
-from .._util import quantity_to_bytes, quantity_to_milli_cpu
+from .._util import quantity_to_milli_cpu
 from ..check_branch_status import get_branch_status
+from ..database import SessionDep
 from ..deployment import (
     get_autoscaler_vm_identity,
     resolve_autoscaler_volume_identifiers,
@@ -40,10 +42,7 @@
     ResourceLimitsPublic,
     ResourceUsageMinute,
 )
-from ._util import Unauthenticated
 from ._util.resourcelimit import (
-    Limits,
-    Resources,
     check_resource_limits,
     create_or_update_branch_provisioning,
     dict_to_resource_limits,
@@ -55,12 +54,8 @@
     get_organization_resource_usage,
     get_project_resource_usage,
     make_usage_cycle,
-    system_allocations,
-    system_available,
-    system_limits,
 )
 from .auth import authenticated_user
-from .db import SessionDep
 from .dependencies import BranchDep, OrganizationDep, ProjectDep
 from .organization.project.branch import refresh_branch_status
 from .settings import get_settings
@@ -321,9 +316,9 @@ def _parse_compute_usage(metrics: dict[str, Any]) -> tuple[int, int]:
 
     usage = cast("dict[str, Any]", compute_usage["usage"])
     cpu_usage = quantity_to_milli_cpu(usage["cpu"])
-    memory_usage = quantity_to_bytes(usage["memory"])
+    memory_usage = parse_quantity(usage["memory"])
 
-    if cpu_usage is None or memory_usage is None:
+    if cpu_usage is None:
         raise ValueError("Metrics API returned empty resource usage for compute container")
 
     return cpu_usage, memory_usage
@@ -494,30 +489,3 @@ async def monitor_resources():
             await asyncio.sleep((interval - elapsed).total_seconds())
         else:
             logger.warning("Resource monitor execution exeeded desired interval")
-
-
-@api.get(
-    "/limits/",
-    name="resources:limits",
-    responses={401: Unauthenticated},
-)
-async def limits(session: SessionDep) -> Limits:
-    return await system_limits(session)
-
-
-@api.get(
-    "/allocations/",
-    name="resources:allocations",
-    responses={401: Unauthenticated},
-)
-async def allocated(session: SessionDep) -> Resources:
-    return await system_allocations(session)
-
-
-@api.get(
-    "/available/",
-    name="resources:available",
-    responses={401: Unauthenticated},
-)
-async def available(session: SessionDep) -> Resources:
-    return await system_available(session)
diff --git a/src/api/settings.py b/src/api/settings.py
index 24e85ef9d..b652afa9d 100644
--- a/src/api/settings.py
+++ b/src/api/settings.py
@@ -22,6 +22,7 @@ class Settings(BaseSettings):
     keycloak_admin_name: str
     keycloak_admin_secret: str
     log_level: Literal["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"] = "INFO"
+    log_json: bool = True
     resource_monitor_interval: Annotated[timedelta, BeforeValidator(permissive_numeric_timedelta)] = timedelta(
         seconds=60
     )
diff --git a/src/api/system.py b/src/api/system.py
index 23ce59c27..fa7fd47ee 100644
--- a/src/api/system.py
+++ b/src/api/system.py
@@ -21,11 +21,10 @@
     VCPU_MILLIS_MIN,
     VCPU_MILLIS_STEP,
 )
-from ..models.resources import ResourceLimitDefinitionPublic, ResourceType
+from ..database import SessionDep
+from ..models.resources import OrganizationLimitDefault, ResourceLimitDefinitionPublic, ResourceType
 from ..models.role import AccessRight
-from ._util.resourcelimit import get_system_resource_limits
 from .auth import authenticated_user
-from .db import SessionDep
 
 api = APIRouter(tags=["system"])
 
@@ -68,10 +67,12 @@ async def list_available_permissions(
 async def list_resource_limit_definitions(
     session: SessionDep,
 ) -> list[ResourceLimitDefinitionPublic]:
-    system_limits = await get_system_resource_limits(session)
+    defaults_result = await session.execute(select(OrganizationLimitDefault))
+    defaults = {d.resource: d for d in defaults_result.scalars().all()}
 
     def _get_limit(resource_type: ResourceType, default: int) -> int:
-        return system_limits[resource_type].max_total if system_limits[resource_type] else default
+        d = defaults.get(resource_type)
+        return d.max_total if d else default
 
     max_vcpu_millis = _get_limit(ResourceType.milli_vcpu, VCPU_MILLIS_MAX)
     max_ram_bytes = _get_limit(ResourceType.ram, MEMORY_MAX)
diff --git a/src/api/user.py b/src/api/user.py
index 5d7c05a66..a0326c782 100644
--- a/src/api/user.py
+++ b/src/api/user.py
@@ -8,12 +8,12 @@
 from pydantic import BaseModel, EmailStr, StrictBool
 from sqlmodel import and_, select
 
+from ..database import SessionDep
 from ..models.membership import Membership
 from ..models.role import AccessRight, Role, RoleAccessRight, RoleUserLink, RoleUserLinkPublic, UserPermissionPublic
 from ..models.user import User, UserParameters, UserPublic
 from ._util import NotFound, Unauthenticated
 from .auth import authenticated_user
-from .db import SessionDep
 from .keycloak import realm_admin
 
 api = APIRouter(dependencies=[Depends(authenticated_user)], tags=["user"])
diff --git a/src/api/db.py b/src/database.py
similarity index 58%
rename from src/api/db.py
rename to src/database.py
index d88a3d8e8..132c0c04a 100644
--- a/src/api/db.py
+++ b/src/database.py
@@ -1,3 +1,4 @@
+import os
 from collections.abc import AsyncGenerator
 from typing import Annotated
 
@@ -5,17 +6,23 @@
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from .settings import get_settings
+from .api.settings import get_settings
 
-# Enable `pool_pre_ping` and periodic recycling so ASGI workers notice connections
-# that Postgres closed while idle (seen as "connection is closed" during requests)
-# and transparently reopen them before handing out a session.
-engine = create_async_engine(
-    str(get_settings().postgres_url),
-    pool_pre_ping=True,
-    pool_recycle=3600,
-)
 
+def _is_worker() -> bool:
+    return bool(os.environ.get("VELA_BROKER_URL"))
+
+
+if _is_worker():
+    from sqlalchemy.pool import NullPool
+
+    engine = create_async_engine(str(get_settings().postgres_url), poolclass=NullPool)
+else:
+    engine = create_async_engine(
+        str(get_settings().postgres_url),
+        pool_pre_ping=True,
+        pool_recycle=3600,
+    )
 
 AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
 
diff --git a/src/deployment/__init__.py b/src/deployment/__init__.py
index 9eb44150d..f30114238 100644
--- a/src/deployment/__init__.py
+++ b/src/deployment/__init__.py
@@ -1226,7 +1226,7 @@ async def set_initial_password(branch_id: Identifier, password: str, admin_passw
     while (
         (status := vm_monitor.status(branch_id)) is None
         or status.services is None
-        or not status.services.get("postgres", False)
+        or not status.services.get("pgbouncer", False)  # Use as a proxy to ensure migrations have run
     ):
         if (datetime.now() - start).total_seconds() > timeout:
             raise TimeoutError("Failed to wait for online database")
diff --git a/src/deployment/deployment.py b/src/deployment/deployment.py
index 8d9a1c067..5dbb146e9 100644
--- a/src/deployment/deployment.py
+++ b/src/deployment/deployment.py
@@ -18,7 +18,7 @@
     },
     "18.1-velaos": {
         "image": "docker.io/simplyblock/vela-image",
-        "tag": "sha-c1980c7",
+        "tag": "sha-51e562f",
     },
 }
 
diff --git a/src/deployment/kubernetes/__init__.py b/src/deployment/kubernetes/__init__.py
index f4e90b4a0..ca96b6c23 100644
--- a/src/deployment/kubernetes/__init__.py
+++ b/src/deployment/kubernetes/__init__.py
@@ -1,16 +1,31 @@
+import asyncio
 import logging
 import math
+import re
+import time
 from collections.abc import Mapping
 from copy import deepcopy
+from decimal import Decimal
 from typing import Any
 
 from aiohttp import ClientError
+from kubernetes.utils import parse_quantity
 from kubernetes_asyncio import client
 
 from ...exceptions import VelaKubernetesError
 from ._util import core_v1_client, custom_api_client, discovery_v1_client, storage_v1_client
 from .neonvm import NeonVM, get_neon_vm
 
+POLL_INTERVAL_SECONDS = 5
+POLL_TIMEOUT_SECONDS = 30 * 60  # 30 minutes
+
+_FAILURE_PATTERN = re.compile(
+    r"\b(resize|resizing|resized)\w*\b.*\b(fail|failure|failed|failing|error|err)\w*\b"
+    r"|"
+    r"\b(fail|failure|failed|failing|error|err)\w*\b.*\b(resize|resizing|resized)\w*\b",
+    flags=re.IGNORECASE,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -292,7 +307,7 @@ async def get_persistent_volume_claim(self, namespace: str, name: str) -> Any:
                     raise VelaKubernetesError(f"PersistentVolumeClaim {namespace!r}/{name!r} not found") from exc
                 raise
 
-    async def resize_pvc_storage(self, namespace: str, name: str, storage: str) -> None:
+    async def resize_pvc_storage(self, namespace: str, name: str, storage: str, *, wait: bool = False) -> None:
         async with core_v1_client() as core_v1:
             try:
                 await core_v1.patch_namespaced_persistent_volume_claim(
@@ -308,6 +323,31 @@ async def resize_pvc_storage(self, namespace: str, name: str, storage: str) -> N
 
         logger.info("Resized PVC %s/%s to %s", namespace, name, storage)
 
+        if wait:
+            await self._poll_pvc_until_complete(namespace, name, parse_quantity(storage))
+
+    async def _poll_pvc_until_complete(self, namespace: str, name: str, target_bytes: Decimal) -> int:
+        """Poll PVC status every 5s until resize completes or fails. Returns actual capacity."""
+        start = time.monotonic()
+        async with core_v1_client() as core_v1:
+            while True:
+                pvc = await core_v1.read_namespaced_persistent_volume_claim(namespace=namespace, name=name)
+
+                capacity_str = (pvc.status.capacity or {}).get("storage")
+                if capacity_str and ((actual := parse_quantity(capacity_str)) >= target_bytes):
+                    return int(actual)
+
+                for condition in pvc.status.conditions or []:
+                    msg = condition.message or ""
+                    if _FAILURE_PATTERN.search(msg):
+                        raise RuntimeError(f"PVC {namespace}/{name} resize failed: {msg}")
+
+                elapsed = time.monotonic() - start
+                if elapsed >= POLL_TIMEOUT_SECONDS:
+                    raise TimeoutError(f"PVC {namespace}/{name} resize timed out after {POLL_TIMEOUT_SECONDS}s")
+
+                await asyncio.sleep(POLL_INTERVAL_SECONDS)
+
     async def get_persistent_volume(self, name: str) -> Any:
         async with core_v1_client() as core_v1:
             try:
@@ -322,7 +362,7 @@ async def resize_autoscaler_vm(
         namespace: str,
         name: str,
         *,
-        cpu_milli: int | None,
+        cpu: Decimal | None,
         memory_bytes: int | None,
     ) -> NeonVM:
         """
@@ -333,13 +373,8 @@ async def resize_autoscaler_vm(
 
         vm_manifest = _build_autoscaler_vm_manifest(vm.model_dump(by_alias=True), namespace, name)
         guest_spec = vm_manifest.setdefault("spec", {}).setdefault("guest", {})
-        cpu_block = guest_spec.setdefault("cpus", {})
-        min_milli = guest.cpus.min_milli
-        max_milli = guest.cpus.max_milli
-        limit_milli = cpu_milli if cpu_milli is not None else guest.cpus.use_milli
-        cpu_block["min"] = _milli_to_cores(min_milli)
-        cpu_block["max"] = _milli_to_cores(max_milli)
-        cpu_block["limit"] = _milli_to_cores(limit_milli)
+        if cpu is not None:
+            guest_spec.setdefault("cpus", {})["limit"] = str(cpu)
 
         if memory_bytes is not None:
             slot_size_bytes = guest.slot_size_bytes
@@ -404,11 +439,6 @@ async def apply_autoscaler_vm(self, namespace: str, name: str, vm_manifest: dict
         return NeonVM.model_validate(applied)
 
 
-def _milli_to_cores(value: int) -> int | float:
-    cores = value / 1000
-    return int(cores) if cores.is_integer() else cores
-
-
 def _build_autoscaler_vm_manifest(vm_obj: dict[str, Any], namespace: str, name: str) -> dict[str, Any]:
     """
     Prepare a clean autoscaler VM manifest for patching.
diff --git a/src/deployment/kubernetes/neonvm.py b/src/deployment/kubernetes/neonvm.py
index 1b335a57c..4dfb7a293 100644
--- a/src/deployment/kubernetes/neonvm.py
+++ b/src/deployment/kubernetes/neonvm.py
@@ -2,10 +2,11 @@
 from typing import Annotated, Any, Literal
 
 from aiohttp.client_exceptions import ClientError
+from kubernetes.utils import parse_quantity
 from kubernetes_asyncio.client.exceptions import ApiException
 from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, PlainSerializer, ValidationError
 
-from ..._util import quantity_to_bytes, quantity_to_milli_cpu
+from ..._util import Quantity, quantity_to_milli_cpu
 from ...exceptions import VelaKubernetesError
 from ._util import custom_api_client
 
@@ -19,16 +20,6 @@ def _require_int(value: Any, field: str) -> int:
         raise VelaKubernetesError(f"Autoscaler VM missing required integer field {field}") from None
 
 
-def _require_quantity_bytes(value: Any, field: str) -> int:
-    if isinstance(value, (int, float)):
-        return int(value)
-    if isinstance(value, str):
-        parsed = quantity_to_bytes(value)
-        if parsed is not None:
-            return parsed
-    raise VelaKubernetesError(f"Autoscaler VM missing required quantity field {field}")
-
-
 def _require_cpu_millis(value: Any, field: str) -> int:
     if isinstance(value, (int, float)):
         return int(value) if value >= 1000 else int(value * 1000)
@@ -66,21 +57,9 @@ class NeonVMStatus(CamelModel):
 
 
 class GuestCPUs(CamelModel):
-    min: Any
-    use: Any
-    max: Any
-
-    @property
-    def use_milli(self) -> int:
-        return _require_cpu_millis(self.use, "guest.cpus.use")
-
-    @property
-    def min_milli(self) -> int:
-        return _require_cpu_millis(self.min, "guest.cpus.min")
-
-    @property
-    def max_milli(self) -> int:
-        return _require_cpu_millis(self.max, "guest.cpus.max")
+    min: Quantity
+    use: Quantity
+    max: Quantity
 
 
 class MemorySlots(CamelModel):
@@ -117,13 +96,13 @@ class Port(CamelModel):
 class Guest(CamelModel):
     cpus: GuestCPUs
     memory_slots: MemorySlots
-    memory_slot_size: Any
+    memory_slot_size: str
     ports: list[Port]
     env: AutoscalerEnv = Field(default_factory=dict)
 
     @property
     def slot_size_bytes(self) -> int:
-        return _require_quantity_bytes(self.memory_slot_size, "guest.memorySlotSize")
+        return int(parse_quantity(self.memory_slot_size))
 
 
 class NeonVMSpec(CamelModel):
diff --git a/src/deployment/monitors/health.py b/src/deployment/monitors/health.py
index a5a464e8c..c8cb4051a 100644
--- a/src/deployment/monitors/health.py
+++ b/src/deployment/monitors/health.py
@@ -3,10 +3,10 @@
 import re
 from datetime import datetime, timedelta
 
-from kubernetes_asyncio import client
 from pydantic import BaseModel
 from ulid import ULID
 
+from ..kubernetes._util import custom_api_client
 from ..kubernetes.neonvm import NeonVM, Phase
 from ..settings import get_settings
 
@@ -56,8 +56,7 @@ async def _check_status(self, vm: NeonVM):
 
     async def run(self):
         logger.info("Started VM monitor")
-        async with client.ApiClient() as api:
-            custom_api = client.CustomObjectsApi(api)
+        async with custom_api_client() as custom_api:
             while True:
                 try:
                     start = datetime.now()
diff --git a/src/deployment/monitors/resize/__init__.py b/src/deployment/monitors/resize/__init__.py
deleted file mode 100644
index 521ea306e..000000000
--- a/src/deployment/monitors/resize/__init__.py
+++ /dev/null
@@ -1,222 +0,0 @@
-"""Resize monitor orchestration.
-
-This module stitches together two complementary monitoring loops that ensure
-branch resources reflect resize requests made through Vela:
-
-PVC resize pipeline
--------------------
-1. ``stream_pvc_events`` opens a long-lived Kubernetes watch filtered to PVC
-   events. The watch feeds a bounded queue to decouple collection from
-   processing, maintaining a resourceVersion for informer-like continuity and
-   exponential backoff for resiliency.
-2. A small worker pool (``_event_worker``) pulls events off the queue and calls
-   ``_handle_pvc_event`` which enriches the raw Kubernetes event, fetching PVC
-   capacity when necessary, and normalizes timestamps and statuses.
-3. ``_apply_volume_status`` persists the synthesized status onto the owning
-   ``Branch`` record, updating aggregate resize state and capacity fields so that
-   API consumers observe consistent progress semantics.
-
-Control surface
----------------
-``run_resize_monitor`` drives both loops concurrently, coordinating shutdown via
-``ResizeMonitor`` which offers a simple start/stop interface for long-lived
-services. The design emphasizes:
-* backpressure-aware queuing so surges of events do not starve handlers;
-* idempotent database writes guarded by status transitions;
-* graceful teardown of asyncio tasks to avoid leaking watches or sessions.
-"""
-
-import asyncio
-import logging
-from contextlib import suppress
-from typing import Any, cast
-
-from kubernetes_asyncio.client import CoreV1Api
-from kubernetes_asyncio.client.models import CoreV1Event
-from sqlmodel import select
-from sqlmodel.ext.asyncio.session import AsyncSession
-from ulid import ULID
-
-from ...._util import storage_backend_bytes_to_db_bytes
-from ....api._util.resourcelimit import create_or_update_branch_provisioning
-from ....api.db import engine
-from ....deployment import deployment_branch
-from ....exceptions import VelaDeploymentError, VelaKubernetesError
-from ....models.branch import (
-    RESIZE_STATUS_PRIORITY,
-    Branch,
-    BranchResizeStatus,
-    aggregate_resize_statuses,
-    should_transition_resize_status,
-)
-from ....models.resources import ResourceLimitsPublic
-from ...health import collect_branch_service_health, derive_branch_status_from_services
-from .pvc_resize import (
-    INITIAL_BACKOFF_SECONDS,
-    VOLUME_SERVICE_MAP,
-    derive_status,
-    fetch_pvc_capacity,
-    normalize_iso_timestamp,
-    resource_from_pvc_name,
-    stream_pvc_events,
-)
-
-logger = logging.getLogger(__name__)
-
-
-async def _apply_volume_status(
-    *,
-    branch_id: ULID,
-    status: BranchResizeStatus | None,
-    event_payload: dict[str, str],
-    resource: str | None,
-    capacity: int | None,
-) -> None:
-    """Persist the PVC resize status and size information on the branch record."""
-    from sqlalchemy.exc import NoResultFound
-
-    async with AsyncSession(engine) as session:
-        statement = select(Branch).where(Branch.id == branch_id)
-        result = await session.exec(statement)
-        try:
-            branch = result.one()
-        except NoResultFound:
-            return
-
-        service_key = VOLUME_SERVICE_MAP.get(resource) if resource else None
-        statuses = dict(branch.resize_statuses or {})
-        status_updated = False
-
-        if status is not None and service_key:
-            current_entry = statuses.get(service_key)
-            current_status_value = None
-            if isinstance(current_entry, dict):
-                current_status_value = current_entry.get("status")
-            if isinstance(current_status_value, str) and current_status_value in RESIZE_STATUS_PRIORITY:
-                current_status = cast("BranchResizeStatus", current_status_value)
-            else:
-                current_status = None
-
-            if should_transition_resize_status(current_status, status):
-                statuses[service_key] = {
-                    "status": status,
-                    "timestamp": event_payload["timestamp"],
-                }
-                branch.resize_statuses = statuses
-                branch.resize_status = aggregate_resize_statuses(statuses)
-                await set_branch_status(branch.resize_status, branch)
-                status_updated = True
-        elif status is not None and should_transition_resize_status(branch.resize_status, status):
-            branch.resize_status = status
-            await set_branch_status(branch.resize_status, branch)
-            status_updated = True
-
-        if status_updated and status == "COMPLETED" and capacity is not None:
-            normalized_capacity = storage_backend_bytes_to_db_bytes(capacity)
-            if resource == "storage":
-                await create_or_update_branch_provisioning(
-                    session,
-                    branch,
-                    ResourceLimitsPublic(storage_size=normalized_capacity),
-                    commit=False,
-                )
-                branch.storage_size = normalized_capacity
-            elif resource == "database":
-                await create_or_update_branch_provisioning(
-                    session,
-                    branch,
-                    ResourceLimitsPublic(database_size=normalized_capacity),
-                    commit=False,
-                )
-                branch.database_size = normalized_capacity
-
-        await session.commit()
-
-
-async def set_branch_status(status: BranchResizeStatus, branch: Branch) -> None:
-    if status not in {"FAILED", "COMPLETED"}:
-        return
-
-    service_status = await collect_branch_service_health(branch.id)
-    branch.set_status(
-        derive_branch_status_from_services(
-            service_status,
-            storage_enabled=branch.enable_file_storage,
-        )
-    )
-
-
-async def _handle_pvc_event(core_v1: CoreV1Api, event: CoreV1Event) -> None:
-    """Map a raw Kubernetes event into branch status updates and capacity changes."""
-    ref = event.involved_object
-    namespace = getattr(ref, "namespace", None)
-    name = getattr(ref, "name", None)
-    if not namespace or not name:
-        return
-    try:
-        branch_id = deployment_branch(namespace)
-    except VelaDeploymentError:
-        logger.debug("Ignoring resize event for namespace %s (not a branch namespace)", namespace)
-        return
-    resource = resource_from_pvc_name(name)
-    status = derive_status(event.reason, event.type, event.message)
-    capacity = None
-    if status == "COMPLETED":
-        capacity = await fetch_pvc_capacity(core_v1, namespace, name)
-
-    status = cast("BranchResizeStatus", status)
-    timestamp_source = (
-        event.last_timestamp or event.event_time or (event.metadata.creation_timestamp if event.metadata else None)
-    )
-    payload = {
-        "timestamp": normalize_iso_timestamp(timestamp_source),
-        "type": event.reason or "",
-        "message": event.message or "",
-    }
-    await _apply_volume_status(
-        branch_id=branch_id,
-        status=status,
-        event_payload=payload,
-        resource=resource,
-        capacity=capacity,
-    )
-
-
-async def run_resize_monitor(stop_event: asyncio.Event) -> None:
-    """Drive PVC resize monitors until the caller signals shutdown."""
-    while not stop_event.is_set():
-        try:
-            await stream_pvc_events(stop_event, _handle_pvc_event)
-        except VelaKubernetesError as exc:
-            logger.warning("PVC resize monitor awaiting Kubernetes configuration: %s", exc)
-            await asyncio.sleep(INITIAL_BACKOFF_SECONDS)
-        except Exception:  # pragma: no cover - defensive guard
-            logger.exception("PVC resize monitor unexpected failure; retrying")
-            await asyncio.sleep(INITIAL_BACKOFF_SECONDS)
-
-
-class ResizeMonitor:
-    """Lightweight wrapper for starting and stopping the resize monitor loop."""
-
-    def __init__(self) -> None:
-        self._stop = asyncio.Event()
-        self._task: asyncio.Task[Any] | None = None
-
-    def start(self) -> None:
-        """Ensure the background resize monitor task is running."""
-        if self._task is None:
-            self._task = asyncio.create_task(run_resize_monitor(self._stop))
-
-    async def stop(self) -> None:
-        """Shut down the background task and reset internal state."""
-        if self._task is None:
-            return
-        self._stop.set()
-        self._task.cancel()
-        with suppress(asyncio.CancelledError):
-            await self._task
-        self._task = None
-        self._stop = asyncio.Event()
-
-
-__all__ = ["ResizeMonitor", "run_resize_monitor"]
diff --git a/src/deployment/monitors/resize/pvc_resize.py b/src/deployment/monitors/resize/pvc_resize.py
deleted file mode 100644
index 3e5309204..000000000
--- a/src/deployment/monitors/resize/pvc_resize.py
+++ /dev/null
@@ -1,291 +0,0 @@
-import asyncio
-import logging
-import re
-from collections.abc import Awaitable, Callable
-from contextlib import suppress
-from datetime import UTC, datetime
-
-from aiohttp import ClientError
-from kubernetes_asyncio import watch
-from kubernetes_asyncio.client import CoreV1Api
-from kubernetes_asyncio.client.exceptions import ApiException
-from kubernetes_asyncio.client.models import CoreV1Event
-
-from ...._util import quantity_to_bytes
-from ....deployment import AUTOSCALER_PVC_SUFFIX, DATABASE_PVC_SUFFIX, STORAGE_PVC_SUFFIX
-from ....deployment.kubernetes._util import core_v1_client
-
-logger = logging.getLogger(__name__)
-
-
-FAILURE_PATTERN = re.compile(
-    r"\b(resize|resizing|resized)\w*\b.*\b(fail|failure|failed|failing|error|err)\w*\b"
-    r"|"
-    r"\b(fail|failure|failed|failing|error|err)\w*\b.*\b(resize|resizing|resized)\w*\b",
-    flags=re.IGNORECASE,
-)
-
-VOLUME_SERVICE_MAP = {
-    "database": "database_disk_resize",
-    "storage": "storage_api_disk_resize",
-}
-
-INITIAL_BACKOFF_SECONDS = 5
-MAX_BACKOFF_SECONDS = 60
-PVC_EVENT_QUEUE_MAXSIZE = 2048
-PVC_WORKER_POOL_SIZE = 4
-PVC_WATCH_TIMEOUT_SECONDS = 60
-PVC_QUEUE_PUT_TIMEOUT_SECONDS = 2
-
-
-def normalize_iso_timestamp(value: datetime | None) -> str:
-    """Return a UTC ISO-8601 string for Kubernetes event timestamps."""
-    if value is None:
-        value = datetime.now(UTC)
-    if value.tzinfo is None:
-        value = value.replace(tzinfo=UTC)
-    return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
-
-
-def resource_from_pvc_name(name: str) -> str | None:
-    """Infer which service a PVC belongs to based on the naming convention."""
-    if name.endswith(STORAGE_PVC_SUFFIX):
-        return "storage"
-    if name.endswith(DATABASE_PVC_SUFFIX):
-        return "database"
-    if name.endswith(AUTOSCALER_PVC_SUFFIX):
-        return "database"
-    return None
-
-
-def derive_status(reason: str | None, event_type: str | None, message: str | None) -> str | None:
-    """Translate a Kubernetes Event into a high-level resize status."""
-    normalized_reason = (reason or "").upper()
-    normalized_message = message or ""
-    if normalized_reason in {"RESIZING", "EXTERNALEXPANDING"}:
-        return "RESIZING"
-    if normalized_reason == "FILESYSTEMRESIZEREQUIRED":
-        return "FILESYSTEM_RESIZE_PENDING"
-    if normalized_reason == "RESIZEFINISHED":
-        return "FILESYSTEM_RESIZE_PENDING"
-    if normalized_reason == "FILESYSTEMRESIZESUCCESSFUL":
-        return "COMPLETED"
-    if normalized_reason in {"VOLUMERESIZEFAILED", "FILESYSTEMRESIZEFAILED"}:
-        return "FAILED"
-    if (event_type or "").upper() == "WARNING" and FAILURE_PATTERN.search(normalized_message):
-        return "FAILED"
-    return None
-
-
-async def fetch_pvc_capacity(core_v1: CoreV1Api, namespace: str, name: str) -> int | None:
-    """Fetch the latest PVC capacity in bytes via the Kubernetes API."""
-    try:
-        pvc = await core_v1.read_namespaced_persistent_volume_claim(namespace=namespace, name=name)
-    except (ApiException, ClientError):
-        logger.exception("Failed to fetch PVC %s/%s for resize completion update", namespace, name)
-        return None
-
-    status_capacity = getattr(pvc.status, "capacity", None)
-    if status_capacity:
-        capacity = quantity_to_bytes(status_capacity.get("storage"))
-        if capacity:
-            return capacity
-
-    resources = getattr(pvc.spec, "resources", None) if pvc.spec else None
-    requests = getattr(resources, "requests", None)
-    if requests:
-        capacity = quantity_to_bytes(requests.get("storage"))
-        if capacity:
-            return capacity
-
-    return None
-
-
-EventHandler = Callable[[CoreV1Api, CoreV1Event], Awaitable[None]]
-
-
-async def _enqueue_event(
-    queue: asyncio.Queue[CoreV1Event | None],
-    event: CoreV1Event,
-    stop_event: asyncio.Event,
-) -> bool:
-    """Attempt to enqueue an event while honoring shutdown signals."""
-    warned = False
-    while True:
-        if stop_event.is_set():
-            return False
-        try:
-            await asyncio.wait_for(queue.put(event), timeout=PVC_QUEUE_PUT_TIMEOUT_SECONDS)
-        except TimeoutError:
-            if not warned:
-                warned = True
-                logger.warning(
-                    "PVC resize monitor queue is saturated (%s/%s); delaying watcher",
-                    queue.qsize(),
-                    PVC_EVENT_QUEUE_MAXSIZE,
-                )
-            continue
-        return True
-
-
-async def _event_worker(
-    worker_id: int,
-    *,
-    queue: asyncio.Queue[CoreV1Event | None],
-    core_v1: CoreV1Api,
-    handler: EventHandler,
-    stop_event: asyncio.Event,
-) -> None:
-    """Drain the PVC event queue and invoke the handler in a dedicated task."""
-    worker_name = f"pvc-resize-worker-{worker_id}"
-    logger.debug("%s started", worker_name)
-    try:
-        while True:
-            try:
-                item = await queue.get()
-            except asyncio.CancelledError:
-                raise
-
-            if item is None:
-                queue.task_done()
-                break
-
-            try:
-                await handler(core_v1, item)
-            except asyncio.CancelledError:
-                raise
-            except Exception:
-                logger.exception("%s failed processing PVC event", worker_name)
-            finally:
-                queue.task_done()
-
-            if stop_event.is_set() and queue.empty():
-                logger.debug("%s draining queue and exiting", worker_name)
-                break
-    finally:
-        logger.debug("%s stopped", worker_name)
-
-
-async def _consume_event_stream(
-    *,
-    core_v1: CoreV1Api,
-    field_selector: str,
-    resource_version: str | None,
-    stop_event: asyncio.Event,
-    queue: asyncio.Queue[CoreV1Event | None],
-) -> str | None:
-    """Stream Kubernetes events until a shutdown signal or queue saturation occurs."""
-    kube_watch = watch.Watch()
-    try:
-        async for item in kube_watch.stream(
-            core_v1.list_event_for_all_namespaces,
-            field_selector=field_selector,
-            resource_version=resource_version,
-            timeout_seconds=PVC_WATCH_TIMEOUT_SECONDS,
-        ):
-            if stop_event.is_set():
-                break
-
-            event: CoreV1Event = item["object"]
-            metadata = getattr(event, "metadata", None)
-            if metadata is not None:
-                resource_version = getattr(metadata, "resource_version", resource_version)
-
-            if not await _enqueue_event(queue, event, stop_event):
-                break
-        return resource_version
-    finally:
-        kube_watch.stop()
-
-
-async def _handle_stream_error(
-    exc: ApiException | ClientError,
-    *,
-    resource_version: str | None,
-    backoff_seconds: int,
-) -> tuple[str | None, int]:
-    """Apply error handling and backoff policy for PVC event stream errors."""
-    if isinstance(exc, ApiException) and exc.status == 410:
-        logger.info("PVC resize monitor resource version expired; resyncing from latest")
-        return "0", INITIAL_BACKOFF_SECONDS
-
-    if isinstance(exc, ApiException):
-        logger.warning("PVC resize monitor API error: %s", exc)
-    else:
-        logger.warning("PVC resize monitor stream client error: %s", exc)
-
-    await asyncio.sleep(backoff_seconds)
-    next_backoff = min(backoff_seconds * 2, MAX_BACKOFF_SECONDS)
-    return resource_version, next_backoff
-
-
-async def _watch_pvc_event_stream(
-    *,
-    core_v1: CoreV1Api,
-    queue: asyncio.Queue[CoreV1Event | None],
-    stop_event: asyncio.Event,
-    field_selector: str,
-) -> None:
-    """Watch PVC events until stopped, applying backoff and error handling."""
-    resource_version: str | None = "0"
-    backoff_seconds = INITIAL_BACKOFF_SECONDS
-    while not stop_event.is_set():
-        try:
-            resource_version = await _consume_event_stream(
-                core_v1=core_v1,
-                field_selector=field_selector,
-                resource_version=resource_version,
-                stop_event=stop_event,
-                queue=queue,
-            )
-            backoff_seconds = INITIAL_BACKOFF_SECONDS
-        except TimeoutError:
-            continue
-        except asyncio.CancelledError:
-            raise
-        except (ApiException, ClientError) as exc:
-            resource_version, backoff_seconds = await _handle_stream_error(
-                exc,
-                resource_version=resource_version,
-                backoff_seconds=backoff_seconds,
-            )
-
-
-async def stream_pvc_events(stop_event: asyncio.Event, handler: EventHandler) -> None:
-    """Watch PVC-related events and push them through a worker-backed queue."""
-    field_selector = "involvedObject.kind=PersistentVolumeClaim"
-    queue: asyncio.Queue[CoreV1Event | None] = asyncio.Queue(maxsize=PVC_EVENT_QUEUE_MAXSIZE)
-    async with core_v1_client() as core_v1:
-        workers = [
-            asyncio.create_task(
-                _event_worker(
-                    worker_id,
-                    queue=queue,
-                    core_v1=core_v1,
-                    handler=handler,
-                    stop_event=stop_event,
-                )
-            )
-            for worker_id in range(PVC_WORKER_POOL_SIZE)
-        ]
-        try:
-            await _watch_pvc_event_stream(
-                core_v1=core_v1,
-                queue=queue,
-                stop_event=stop_event,
-                field_selector=field_selector,
-            )
-        finally:
-            # Allow workers to finish in-flight items before shutting down.
-            try:
-                await asyncio.wait_for(queue.join(), timeout=PVC_WATCH_TIMEOUT_SECONDS)
-            except TimeoutError:
-                logger.warning(
-                    "PVC resize monitor queue did not drain within %s seconds; cancelling workers",
-                    PVC_WATCH_TIMEOUT_SECONDS,
-                )
-            for worker in workers:
-                worker.cancel()
-            for worker in workers:
-                with suppress(asyncio.CancelledError):
-                    await worker
diff --git a/src/deployment/resize.py b/src/deployment/resize.py
new file mode 100644
index 000000000..2508f6a9a
--- /dev/null
+++ b/src/deployment/resize.py
@@ -0,0 +1,67 @@
+"""Celery task and dispatch helper for branch resize operations.
+
+The task lives in the deployment layer and has no DB access.  The API layer
+manages branch state; this module only orchestrates infrastructure changes and
+returns the resulting sizes for the caller to persist.
+"""
+
+import logging
+from decimal import Decimal
+
+from asgiref.sync import async_to_sync
+from ulid import ULID
+
+from ..worker import app
+from . import (
+    AUTOSCALER_PVC_SUFFIX,
+    STORAGE_PVC_SUFFIX,
+    get_autoscaler_vm_identity,
+    kube_service,
+    update_branch_volume_iops,
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def _resize_cpu_memory(deployment_id: ULID, milli_vcpu: int | None, memory_bytes: int | None) -> None:
+    namespace, autoscaler_vm_name = get_autoscaler_vm_identity(deployment_id)
+    await kube_service.resize_autoscaler_vm(
+        namespace,
+        autoscaler_vm_name,
+        cpu=Decimal(milli_vcpu) / 1000 if milli_vcpu is not None else None,
+        memory_bytes=memory_bytes,
+    )
+
+
+@app.task(name="simplyblock.vela.deployment.resize.cpu_memory")
+def resize_cpu_memory(deployment_id: str, milli_vcpu: int | None, memory_bytes: int | None) -> None:
+    """Resize CPU and/or memory for a branch deployment."""
+    async_to_sync(_resize_cpu_memory)(ULID.from_str(deployment_id), milli_vcpu, memory_bytes)
+
+
+async def _resize_iops(deployment_id: ULID, iops: int) -> None:
+    await update_branch_volume_iops(deployment_id, iops)
+
+
+@app.task(name="simplyblock.vela.deployment.resize.iops")
+def resize_iops(deployment_id: str, iops: int) -> None:
+    """Resize IOPS for a branch deployment."""
+    async_to_sync(_resize_iops)(ULID.from_str(deployment_id), iops)
+
+
+async def _resize_pvc(deployment_id: ULID, pvc_suffix: str, target_size: int) -> None:
+    namespace, autoscaler_vm_name = get_autoscaler_vm_identity(deployment_id)
+    pvc_name = f"{autoscaler_vm_name}{pvc_suffix}"
+    await kube_service.resize_pvc_storage(namespace, pvc_name, str(target_size), wait=True)
+
+
+@app.task(name="simplyblock.vela.deployment.resize.database_pvc")
+def resize_database_pvc(deployment_id: str, database_size: int) -> None:
+    """Resize the database PVC for a branch deployment."""
+    async_to_sync(_resize_pvc)(ULID.from_str(deployment_id), AUTOSCALER_PVC_SUFFIX, database_size)
+
+
+@app.task(name="simplyblock.vela.deployment.resize.storage_pvc")
+def resize_storage_pvc(deployment_id: str, storage_size: int) -> None:
+    """Resize the storage PVC for a branch deployment."""
+    async_to_sync(_resize_pvc)(ULID.from_str(deployment_id), STORAGE_PVC_SUFFIX, storage_size)
diff --git a/src/models/branch.py b/src/models/branch.py
index b77059a10..d049981d5 100644
--- a/src/models/branch.py
+++ b/src/models/branch.py
@@ -1,9 +1,9 @@
-from collections.abc import Mapping
+import uuid
 from datetime import UTC, datetime
 from enum import StrEnum
 from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, ValidationError, model_validator
+from pydantic import BaseModel, model_validator
 from pydantic import Field as PydanticField
 from sqlalchemy import BigInteger, Boolean, Column, String, Text, UniqueConstraint, text
 from sqlalchemy.dialects.postgresql import JSONB
@@ -102,19 +102,12 @@ class Branch(AsyncAttrs, Model, table=True):
     jwt_secret: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))]
     anon_key: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))]
     service_key: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))]
-    resize_status: "BranchResizeStatus" = Field(
-        default="NONE",
-        sa_column=Column(String(length=48), nullable=False),
-    )
-    resize_statuses: dict[str, dict[str, Any]] = Field(
-        default_factory=dict,
-        sa_column=Column(JSONB, nullable=False, server_default=text("'{}'::jsonb")),
-    )
     resource_usage: dict[str, Any] = Field(
         default_factory=_default_resource_usage_payload,
         sa_column=Column(JSONB, nullable=False, server_default=text("'{}'::jsonb")),
     )
     pitr_enabled: bool = Field(default=False, sa_column=Column(Boolean, nullable=False, server_default=text("false")))
+    resize_task_id: uuid.UUID | None = Field(default=None, nullable=True)
 
     __table_args__ = (UniqueConstraint("project_id", "name", name="unique_branch_name_per_project"),)
 
@@ -472,102 +465,8 @@ class BranchStatus(BaseModel):
     rest: BranchServiceStatus
 
 
-BranchResizeStatus = Literal[
-    "NONE",
-    "PENDING",
-    "RESIZING",
-    "FILESYSTEM_RESIZE_PENDING",
-    "COMPLETED",
-    "FAILED",
-]
-
 CapaResizeKey = Literal["database_size", "storage_size", "milli_vcpu", "memory_bytes", "iops"]
 
-BranchResizeService = Literal[
-    "database_disk_resize",
-    "storage_api_disk_resize",
-    "database_cpu_resize",
-    "database_memory_resize",
-    "database_iops_resize",
-]
-
-RESIZE_STATUS_PRIORITY: dict[BranchResizeStatus, int] = {
-    "NONE": 0,
-    "PENDING": 1,
-    "RESIZING": 2,
-    "FILESYSTEM_RESIZE_PENDING": 3,
-    "COMPLETED": 4,
-    "FAILED": 5,
-}
-
-
-class BranchResizeStatusEntry(BaseModel):
-    """Single service's resize state and the timestamp when it was observed."""
-
-    model_config = ConfigDict(extra="allow")
-
-    status: BranchResizeStatus
-    timestamp: str
-
-    def timestamp_as_datetime(self) -> datetime:
-        value = self.timestamp
-        if not value:
-            return datetime.min.replace(tzinfo=UTC)
-        try:
-            normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
-            parsed = datetime.fromisoformat(normalized)
-        except ValueError:
-            return datetime.min.replace(tzinfo=UTC)
-        if parsed.tzinfo is None:
-            parsed = parsed.replace(tzinfo=UTC)
-        return parsed.astimezone(UTC)
-
-
-def should_transition_resize_status(current: BranchResizeStatus | None, proposed: BranchResizeStatus) -> bool:
-    """Return True when the proposed status should replace the current one."""
-
-    if proposed not in RESIZE_STATUS_PRIORITY:
-        return False
-    if current == proposed:
-        return False
-    if proposed == "FAILED":
-        return True
-    if current is None:
-        return True
-    return RESIZE_STATUS_PRIORITY[proposed] >= RESIZE_STATUS_PRIORITY[current]
-
-
-def aggregate_resize_statuses(
-    statuses: Mapping[str, BranchResizeStatusEntry | Mapping[str, Any]] | None,
-) -> BranchResizeStatus:
-    """
-    Aggregate per-service resize statuses into a single branch-level status.
-    """
-    highest: BranchResizeStatus = "NONE"
-    if not statuses:
-        return highest
-
-    highest_timestamp = datetime.min.replace(tzinfo=UTC)
-    for entry in statuses.values():
-        if isinstance(entry, BranchResizeStatusEntry):
-            snapshot = entry
-        else:
-            try:
-                snapshot = BranchResizeStatusEntry.model_validate(entry)
-            except ValidationError:
-                continue
-
-        status = snapshot.status
-        if status not in RESIZE_STATUS_PRIORITY:
-            continue
-        timestamp = snapshot.timestamp_as_datetime()
-        if timestamp > highest_timestamp or (
-            timestamp == highest_timestamp and should_transition_resize_status(highest, status)
-        ):
-            highest = status
-            highest_timestamp = timestamp
-    return highest
-
 
 class BranchPublic(BaseModel):
     id: Identifier
@@ -589,8 +488,6 @@ class BranchPublic(BaseModel):
 
 
 class BranchStatusPublic(BaseModel):
-    resize_status: BranchResizeStatus
-    resize_statuses: dict[str, BranchResizeStatusEntry]
     service_status: BranchStatus
 
 
diff --git a/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py b/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py
new file mode 100644
index 000000000..56db63c36
--- /dev/null
+++ b/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py
@@ -0,0 +1,39 @@
+"""Adapt branch model to external task
+
+Revision ID: ad471311850e
+Revises: e8f3a2d51c9b
+Create Date: 2026-03-17 21:55:53.413091
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+import sqlmodel.sql
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'ad471311850e'
+down_revision: Union[str, Sequence[str], None] = 'e8f3a2d51c9b'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('branch', sa.Column('resize_task_id', sa.Uuid(), nullable=True))
+    op.drop_column('branch', 'resize_status')
+    op.drop_column('branch', 'resize_statuses')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('branch', 'resize_task_id')
+    op.add_column('branch', sa.Column('resize_statuses', postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), autoincrement=False, nullable=False))
+    op.add_column('branch', sa.Column('resize_status', sa.VARCHAR(length=48), autoincrement=False, nullable=False))
+    # ### end Alembic commands ###
diff --git a/src/models/migrations/versions/e8f3a2d51c9b_fix_system_limits.py b/src/models/migrations/versions/e8f3a2d51c9b_fix_system_limits.py
new file mode 100644
index 000000000..8a579eb8d
--- /dev/null
+++ b/src/models/migrations/versions/e8f3a2d51c9b_fix_system_limits.py
@@ -0,0 +1,69 @@
+"""fix-system-limits
+
+Revision ID: e8f3a2d51c9b
+Revises: 9bebcc605033
+Create Date: 2026-03-20 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+import sqlmodel
+import sqlmodel.sql
+from alembic import op
+from sqlalchemy.dialects import postgresql
+from ulid import ULID
+
+# revision identifiers, used by Alembic.
+revision: str = "e8f3a2d51c9b"
+down_revision: Union[str, Sequence[str], None] = "9bebcc605033"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    resource_type_enum = postgresql.ENUM(name="resourcetype", create_type=False)
+
+    # Create the organizationlimitdefault table
+    op.create_table(
+        "organizationlimitdefault",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("resource", resource_type_enum, nullable=False),
+        sa.Column("max_total", sa.BigInteger(), nullable=False),
+        sa.Column("max_per_branch", sa.BigInteger(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("uq_org_limit_default_resource", "organizationlimitdefault", ["resource"], unique=True)
+
+    # Migrate system limits → OrganizationLimitDefault
+    conn = op.get_bind()
+    system_rows = conn.execute(
+        sa.text("SELECT resource, max_total, max_per_branch FROM resourcelimit WHERE entity_type = 'system'")
+    ).fetchall()
+
+    org_limit_default = sa.table(
+        "organizationlimitdefault",
+        sa.column("id", sa.UUID),
+        sa.column("resource", resource_type_enum),
+        sa.column("max_total", sa.BigInteger),
+        sa.column("max_per_branch", sa.BigInteger),
+    )
+    for resource, max_total, max_per_branch in system_rows:
+        conn.execute(
+            sa.insert(org_limit_default).values(
+                id=ULID().to_uuid(),
+                resource=resource,
+                max_total=max_total,
+                max_per_branch=max_per_branch,
+            )
+        )
+
+    conn.execute(sa.text("DELETE FROM resourcelimit WHERE entity_type = 'system'"))
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # Data migration is not trivially reversible; just drop the table.
+    op.drop_index("uq_org_limit_default_resource", table_name="organizationlimitdefault")
+    op.drop_table("organizationlimitdefault")
diff --git a/src/models/resources.py b/src/models/resources.py
index b2357124e..28815c833 100644
--- a/src/models/resources.py
+++ b/src/models/resources.py
@@ -29,7 +29,6 @@ class ResourceType(PyEnum):
 
 
 class EntityType(PyEnum):
-    system = "system"
     org = "org"
     project = "project"
 
@@ -85,6 +84,13 @@ class ResourceLimit(AsyncAttrs, Model, table=True):
     max_per_branch: Annotated[int, Field(sa_type=BigInteger)]
 
 
+class OrganizationLimitDefault(AsyncAttrs, Model, table=True):
+    __table_args__ = (Index("uq_org_limit_default_resource", "resource", unique=True),)
+    resource: ResourceType
+    max_total: Annotated[int, Field(sa_type=BigInteger)]
+    max_per_branch: Annotated[int, Field(sa_type=BigInteger)]
+
+
 class BranchProvisioning(AsyncAttrs, Model, table=True):
     branch_id: Identifier | None = Model.foreign_key_field("branch", ondelete="CASCADE")
     resource: ResourceType
diff --git a/src/worker/__init__.py b/src/worker/__init__.py
new file mode 100644
index 000000000..fb20d4e15
--- /dev/null
+++ b/src/worker/__init__.py
@@ -0,0 +1,24 @@
+from celery import Celery
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="vela_", case_sensitive=False)
+    broker_url: str
+    result_backend: str
+
+
+_settings = Settings()  # type: ignore[call-arg]
+
+app = Celery("vela", broker=_settings.broker_url, backend=_settings.result_backend)
+
+# Persist task name, args, and kwargs in celery_taskmeta so AsyncResult can
+# reconstruct full task details without a custom model.
+app.conf.result_extended = True
+
+# Chord callback always fires even when individual sub-tasks fail.
+app.conf.task_chord_propagates = False
+
+# Register tasks — must be imported after `app` is defined.
+from ..api.organization.project.branch import resize_tasks as _api_resize_tasks  # noqa: E402, F401
+from ..deployment import resize as _deployment_resize  # noqa: E402, F401
diff --git a/tests/test_branches.py b/tests/branches/test_basic.py
similarity index 74%
rename from tests/test_branches.py
rename to tests/branches/test_basic.py
index a4562210a..33bb9ef81 100644
--- a/tests/test_branches.py
+++ b/tests/branches/test_basic.py
@@ -1,34 +1,21 @@
 import time
 
-import httpx
 import psycopg
 import pytest
-from conftest import BRANCH_TIMEOUT_SEC, _id, wait_for_status
+from conftest import BRANCH_TIMEOUT_SEC, wait_for_status
 
 pytestmark = pytest.mark.branch
 
+_BRANCH_PASSWORD = "SecurePass1!"
 _BRANCH_NAME = "test-branch"
 _BRANCH_RENAMED = "test-branch-renamed"
-_BRANCH_CREATE_PAYLOAD = {
-    "name": _BRANCH_NAME,
-    "deployment": {
-        "database_password": "SecurePass1!",
-        "database_size": 1000000000,
-        "storage_size": 1000000000,
-        "milli_vcpu": 500,
-        "memory_bytes": 1073741824,
-        "iops": 1000,
-        "database_image_tag": "18.1-velaos",
-        "enable_file_storage": True,
-    },
-}
-
-_state: dict = {}
 
 _DB_CONNECT_TIMEOUT = 10
 _DB_CONNECT_MAX_WAIT = 120
 _DB_CONNECT_RETRY_DELAY = 15
 
+_state: dict = {}
+
 
 def _check_postgres_connection(db_info: dict, password: str) -> None:
     """Verify a Postgres connection can be established with the given credentials.
@@ -69,36 +56,26 @@ def project(make_project, org):
 
 
 @pytest.fixture(scope="module")
-def branch_id(client, org, project):
-    r = client.post(
-        f"organizations/{org}/projects/{project}/branches/",
-        json=_BRANCH_CREATE_PAYLOAD,
-        timeout=60,
-    )
-    assert r.status_code == 201
-    bid = _id(r.headers["Location"])
-    branch_data = wait_for_status(
-        client,
-        f"organizations/{org}/projects/{project}/branches/{bid}/",
-        "ACTIVE_HEALTHY",
-        BRANCH_TIMEOUT_SEC,
+def branch_id(client, make_branch, org, project):
+    bid = make_branch(
+        org,
+        project,
+        _BRANCH_NAME,
+        deployment={
+            "database_password": _BRANCH_PASSWORD,
+            "database_size": 1_000_000_000,
+            "storage_size": 1_000_000_000,
+            "milli_vcpu": 500,
+            "memory_bytes": 1_073_741_824,
+            "iops": 1000,
+            "database_image_tag": "18.1-velaos",
+            "enable_file_storage": True,
+        },
     )
-    _check_postgres_connection(
-        branch_data["database"],
-        _BRANCH_CREATE_PAYLOAD["deployment"]["database_password"],
-    )
-    yield bid
-
-    r = client.delete(f"organizations/{org}/projects/{project}/branches/{bid}/")
-    assert r.status_code == 204
-    with pytest.raises(httpx.HTTPStatusError) as exc:
-        wait_for_status(
-            client,
-            f"organizations/{org}/projects/{project}/branches/{bid}/",
-            "DELETED",  # Dummy status for now, after soft-delete is implemented this will become the actual check
-            BRANCH_TIMEOUT_SEC,
-        )
-    assert exc.value.response.status_code == 404
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{bid}/")
+    r.raise_for_status()
+    _check_postgres_connection(r.json()["database"], _BRANCH_PASSWORD)
+    return bid
 
 
 def test_branch_list_empty(client, org, project):
@@ -147,14 +124,21 @@ def test_branch_list_contains(client, org, project, branch_id):
 def test_branch_status_endpoint(client, org, project, branch_id):
     r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/status")
     assert r.status_code == 200
-    data = r.json()
-    assert "resize_status" in data
 
 
 def test_branch_resize(client, org, project, branch_id):
+    database_size = 6 * 1_000_000_000  # 6 GB, must be a multiple of 1 GB
+    storage_size = 2 * 1_000_000_000  # 2 GB, must be a multiple of 1 GB
+    memory_bytes = 2 * 1024 * 1024 * 1024  # 2 GiB, must be a multiple of 256 MiB
     r = client.post(
         f"organizations/{org}/projects/{project}/branches/{branch_id}/resize",
-        json={"iops": 2000},
+        json={
+            "iops": 2000,
+            "milli_vcpu": 1000,
+            "memory_bytes": memory_bytes,
+            "database_size": database_size,
+            "storage_size": storage_size,
+        },
     )
     assert r.status_code == 202
     wait_for_status(
@@ -163,6 +147,14 @@ def test_branch_resize(client, org, project, branch_id):
         "ACTIVE_HEALTHY",
         BRANCH_TIMEOUT_SEC,
     )
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/")
+    assert r.status_code == 200
+    resources = r.json()["max_resources"]
+    assert resources["iops"] == 2000
+    assert resources["milli_vcpu"] == 1000
+    assert resources["ram_bytes"] == memory_bytes
+    assert resources["nvme_bytes"] == database_size
+    assert resources["storage_bytes"] == storage_size
 
 
 def test_branch_password_reset(client, org, project, branch_id):
diff --git a/tests/branches/test_clone_restore.py b/tests/branches/test_clone_restore.py
new file mode 100644
index 000000000..6486febb0
--- /dev/null
+++ b/tests/branches/test_clone_restore.py
@@ -0,0 +1,76 @@
+import pytest
+
+pytestmark = pytest.mark.backup
+
+_BRANCH_PASSWORD = "SecurePass1!"
+
+
+@pytest.fixture(scope="module")
+def org(make_org):
+    return make_org("test-org-backup", max_backups=10)
+
+
+@pytest.fixture(scope="module")
+def project(make_project, org):
+    return make_project(org, "test-project-backup", max_backups=10)
+
+
+@pytest.fixture(scope="module")
+def branch_id(make_branch, org, project):
+    return make_branch(
+        org,
+        project,
+        "test-branch-backup",
+        deployment={
+            "database_password": _BRANCH_PASSWORD,
+            "database_size": 5_000_000_000,
+            "storage_size": 5_000_000_000,
+            "milli_vcpu": 500,
+            "memory_bytes": 1_073_741_824,
+            "iops": 1000,
+            "database_image_tag": "18.1-velaos",
+            "enable_file_storage": False,
+        },
+    )
+
+
+@pytest.fixture(scope="module")
+def backup_id(client, branch_id):
+    """Trigger a manual backup and return the backup_id."""
+    r = client.post(f"backup/branches/{branch_id}/", timeout=120)
+    assert r.status_code == 200
+    data = r.json()
+    assert data["status"] == "manual backup created"
+    assert "backup_id" in data
+    return data["backup_id"]
+
+
+def test_branch_clone(client, org, project, branch_id, make_branch):
+    """Clone a branch and verify the clone reaches ACTIVE_HEALTHY."""
+    clone_id = make_branch(
+        org,
+        project,
+        "test-branch-clone",
+        source={"branch_id": str(branch_id), "data_copy": True},
+    )
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{clone_id}/")
+    assert r.status_code == 200
+    assert r.json()["status"] == "ACTIVE_HEALTHY"
+
+
+def test_manual_backup(backup_id):
+    """Trigger an on-demand backup for the branch and verify a backup_id is returned."""
+    assert backup_id, "backup_id fixture must return a valid id"
+
+
+def test_restore_branch_from_backup(client, org, project, backup_id, make_branch):
+    """Create a new branch from the manual backup and verify it reaches ACTIVE_HEALTHY."""
+    restored_id = make_branch(
+        org,
+        project,
+        "test-branch-restored",
+        restore={"backup_id": backup_id},
+    )
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{restored_id}/")
+    assert r.status_code == 200
+    assert r.json()["status"] == "ACTIVE_HEALTHY"
diff --git a/tests/conftest.py b/tests/conftest.py
index 07dfd80f1..9568db935 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -84,8 +84,9 @@ def client():
 def make_org(client):
     created: list[ULID] = []
 
-    def _factory(name: str) -> ULID:
-        r = client.post("organizations/", json={"name": name, "max_backups": 0, "environments": ""})
+    def _factory(name: str, **overrides: object) -> ULID:
+        payload = {"name": name, "max_backups": 0, "environments": "", **overrides}
+        r = client.post("organizations/", json=payload)
         assert r.status_code == 201
         uid = _id(r.headers["Location"])
         created.append(uid)
@@ -100,10 +101,11 @@ def _factory(name: str) -> ULID:
 def make_project(client):
     created: list[tuple[ULID, ULID]] = []
 
-    def _factory(org_id: ULID, name: str) -> ULID:
+    def _factory(org_id: ULID, name: str, **overrides: object) -> ULID:
+        payload = {"name": name, "max_backups": 0, "project_limits": {}, "per_branch_limits": {}, **overrides}
         r = client.post(
             f"organizations/{org_id}/projects/",
-            json={"name": name, "max_backups": 0, "project_limits": {}, "per_branch_limits": {}},
+            json=payload,
         )
         assert r.status_code == 201
         uid = _id(r.headers["Location"])
@@ -113,3 +115,48 @@ def _factory(org_id: ULID, name: str) -> ULID:
     yield _factory
     for org_id, uid in created:
         client.delete(f"organizations/{org_id}/projects/{uid}/")
+
+
+@pytest.fixture(scope="session")
+def make_branch(client):
+    """Factory that creates a branch, waits for ACTIVE_HEALTHY, and deletes on teardown.
+
+    Provides sensible deployment defaults.  Pass ``deployment={…}`` to override.
+    For clones pass ``source={…}``; for restores pass ``restore={…}`` – deployment
+    defaults are omitted automatically in those cases.
+    """
+    default_deployment: dict = {
+        "database_password": "SecurePass1!",
+        "database_size": 1_000_000_000,
+        "storage_size": 1_000_000_000,
+        "milli_vcpu": 500,
+        "memory_bytes": 1_073_741_824,
+        "iops": 1000,
+        "database_image_tag": "18.1-velaos",
+        "enable_file_storage": False,
+    }
+    created: list[tuple[ULID, ULID, ULID]] = []
+
+    def _factory(org_id: ULID, project_id: ULID, name: str, **overrides: object) -> ULID:
+        payload: dict = {"name": name, **overrides}
+        if "deployment" not in payload and "source" not in payload and "restore" not in payload:
+            payload["deployment"] = {**default_deployment}
+        r = client.post(
+            f"organizations/{org_id}/projects/{project_id}/branches/",
+            json=payload,
+            timeout=60,
+        )
+        assert r.status_code == 201
+        uid = _id(r.headers["Location"])
+        wait_for_status(
+            client,
+            f"organizations/{org_id}/projects/{project_id}/branches/{uid}/",
+            "ACTIVE_HEALTHY",
+            BRANCH_TIMEOUT_SEC,
+        )
+        created.append((org_id, project_id, uid))
+        return uid
+
+    yield _factory
+    for org_id, project_id, uid in reversed(created):
+        client.delete(f"organizations/{org_id}/projects/{project_id}/branches/{uid}/")
diff --git a/tests/test_organizations.py b/tests/test_organizations.py
index c60a45c0a..81bff8d5c 100644
--- a/tests/test_organizations.py
+++ b/tests/test_organizations.py
@@ -1,3 +1,4 @@
+import pytest
 from conftest import _id
 
 _ORG_NAME = "test-org-lifecycle"
@@ -5,6 +6,14 @@
 _state: dict = {}
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _org_cleanup(client):
+    """Safety net: delete the org if test_org_delete did not run (e.g. early failure)."""
+    yield
+    if "org_id" in _state:
+        client.delete(f"organizations/{_state['org_id']}/")  # no-op if already deleted
+
+
 def test_org_list_empty(client):
     r = client.get("organizations/")
     assert r.status_code == 200
diff --git a/tests/test_projects.py b/tests/test_projects.py
index 2da5ab1f1..0d957aec3 100644
--- a/tests/test_projects.py
+++ b/tests/test_projects.py
@@ -65,3 +65,10 @@ def test_project_list_contains(client, org):
 def test_project_not_found(client, org):
     r = client.get(f"organizations/{org}/projects/00000000000000000000000000/")
     assert r.status_code == 404
+
+
+def test_project_delete(client, org):
+    r = client.delete(f"organizations/{org}/projects/{_state['project_id']}/")
+    assert r.status_code == 204
+    r = client.get(f"organizations/{org}/projects/{_state['project_id']}/")
+    assert r.status_code == 404
diff --git a/tests/test_resources.py b/tests/test_resources.py
index 28b741c0f..f7a6ec797 100644
--- a/tests/test_resources.py
+++ b/tests/test_resources.py
@@ -21,7 +21,6 @@
 _FIELDS = {"milli_vcpu", "ram", "iops", "database_size", "storage_size"}
 
 # Stored during the test run to reuse in later assertions.
-_system_available: dict = {}
 _org_available: dict = {}
 
 
@@ -42,42 +41,34 @@ def project(make_project, org):
 
 
 # ---------------------------------------------------------------------------
-# System
+# Org default limits (must run before test_org_set_limits changes them)
 # ---------------------------------------------------------------------------
 
 
-def test_system_limits(client) -> None:
-    r = client.get("resources/limits/")
+def test_org_has_default_limits_on_creation(client, org) -> None:
+    """New org should have limits populated from OrganizationLimitDefault defaults."""
+    r = client.get(f"organizations/{org}/resources/limits/")
     assert r.status_code == 200
     data = r.json()
     assert set(data.keys()) == {"total", "per_branch"}
     for section in ("total", "per_branch"):
         assert set(data[section].keys()) == _FIELDS
         for v in data[section].values():
-            assert v is not None and isinstance(v, int)
-
+            assert v is not None and isinstance(v, int) and v > 0
 
-def test_system_allocations(client) -> None:
-    r = client.get("resources/allocations/")
-    assert r.status_code == 200
-    data = r.json()
-    assert set(data.keys()) == _FIELDS
-    for v in data.values():
-        assert v == 0
 
-
-def test_system_available(client) -> None:
-    r = client.get("resources/available/")
-    assert r.status_code == 200
-    data = r.json()
-    assert set(data.keys()) == _FIELDS
-    for v in data.values():
-        assert v >= 0
-    _system_available.update(data)
+def test_org_default_limits_are_available(client, org) -> None:
+    """Available resources for a new org should equal its default limits (no branches yet)."""
+    limits_r = client.get(f"organizations/{org}/resources/limits/")
+    avail_r = client.get(f"organizations/{org}/resources/available/")
+    assert limits_r.status_code == 200
+    assert avail_r.status_code == 200
+    for f in _FIELDS:
+        assert avail_r.json()[f] == limits_r.json()["total"][f]
 
 
 # ---------------------------------------------------------------------------
-# Org
+# Org (override limits)
 # ---------------------------------------------------------------------------