docs: bring RUNBOOK and ARCHITECTURE up to date

The runner moved to node0 and the drift check reads release secrets via the
Kubernetes API in-cluster; the docs still described node2 and `helm list`.

- RUNBOOK: the durable image-cache fix is now the node0 hostPath store, not a
  node2 pin; /data is NFS RWX, so the Multi-Attach wait is gone. Cross-references
  entries 9 and 10.
- ARCHITECTURE: the reconciler's edge to the cluster is "list releases", not
  "helm list" (helm is the out-of-cluster fallback).
- ARCHITECTURE: the state diagram and its prose described fail() moving every
  dead-lettered instance to `failed`. Corrected to the per-kind behaviour — only
  provision fails the instance; deprovision stays `deleting` for retry, upgrade
  and verify stay `ready` — matching the fix in tasks.py.

ARCHITECTURE.md

@@ -46,7 +46,7 @@ flowchart LR
     WORKER -->|"claim<br/>SKIP LOCKED"| PG
     WORKER -->|"helm upgrade --install"| K8S
     RECON -->|"drift, leases,<br/>TTL, versions"| PG
-    RECON -->|"helm list"| K8S
+    RECON -->|"list releases"| K8S
 
     classDef truth fill:#2d4a22,stroke:#5a8f3d,color:#fff
     classDef derived fill:#4a3222,stroke:#8f6a3d,color:#fff
@@ -191,12 +191,12 @@ stateDiagram-v2
     ready --> deleting: DELETE, or TTL expired
     deleting --> deleted: helm uninstall succeeded
 
-    requested --> failed: attempts exhausted
-    provisioning --> failed: attempts exhausted
+    requested --> failed: provision attempts exhausted
+    provisioning --> failed: provision attempts exhausted
     ready --> failed: drift — the release vanished
     failed --> provisioning: retry
     failed --> deleting: give up, tear it down
-    deleting --> failed: attempts exhausted
+    deleting --> deleting: deprovision retried, never failed
 
     deleted --> [*]: terminal
 ```
@@ -205,19 +205,31 @@ stateDiagram-v2
 chain of `if`s. `deleted` maps to an **empty frozenset** rather than being absent, so
 "terminal" is stated rather than implied by a missing key.
 
-**The state machine is enforced in SQL too.** `TaskRepo.fail` writes `instances.state`
-directly, so it derives its guard from the same `LEGAL` table:
+**Dead-lettering the task does not fail the instance, except for provision.** When a task
+exhausts its retries `TaskRepo.fail` marks the *task* `failed` for every kind. It moves the
+*instance* to `failed` only for `provision`, because that is the only kind where a dead
+letter means the instance is broken. For the others the instance is still healthy and
+something else owns its recovery:
+
+| kind | instance state on dead-letter | why |
+|------|-------------------------------|-----|
+| provision | `failed` | it never came up; a human re-provisions |
+| deprovision | stays `deleting` | so `due_for_deprovision` re-enqueues it; `failed` would strand it and leak the release |
+| upgrade | stays `ready` | `helm --atomic` rolled back; it still serves the old version |
+| verify | stays `ready` | `handle_verify` already halted the rollout |
+
+The provision write is still guarded by the state machine, derived from the same `LEGAL`
+table rather than restated:
 
 ```python
 _CAN_FAIL = tuple(s.value for s, allowed in LEGAL.items() if InstanceState.FAILED in allowed)
 ...
+if kind == 'provision':
     UPDATE instances SET state='failed' WHERE id=%s AND state = ANY(%s)
 ```
 
-Without that, a deprovision exhausting its retries against an already-`deleted` instance
-would resurrect it into `failed` — a transition `transition()` explicitly forbids,
-performed by raw SQL that never asked it. A state machine only one layer respects is
-decoration.
+The `SvcforgeTaskDeadLettered` alert fires for every kind, so leaving the instance alone
+loses no operator visibility.
 
 ---
 
@@ -278,7 +290,7 @@ One replica. Four checks. Every 60 seconds.
 
 ```mermaid
 flowchart LR
-    TICK(("tick<br/>every 60s")) --> D["<b>drift</b><br/>helm list vs DB"]
+    TICK(("tick<br/>every 60s")) --> D["<b>drift</b><br/>live releases vs DB"]
     TICK --> L["<b>lease expiry</b><br/>running + locked_at old"]
     TICK --> T["<b>TTL</b><br/>ready + expires_at passed"]
     TICK --> V["<b>version drift</b><br/>chart_version ≠ catalog"]

RUNBOOK.md

@@ -118,11 +118,13 @@ kubectl -n gitea exec gitea-actions-runner-0 -c dind -- docker pull \
   ghcr.io/catthehacker/ubuntu:act-24.04@sha256:c710431fbad9eb3bcb102d04e5ff74fbd0ce6e383f78afebfb3770a1a817fdf9
 ```
 
-The durable fix is to stop the runner restarting. Its `/data` PVC is ReadWriteOnce, so
-every reschedule hits `Multi-Attach error` and the pod sits in Init until Longhorn detaches
-from the old node. It is pinned to node2 in `oci-k8s/.../addons/tasks/main.yml` for exactly
-that reason. A dedicated PVC for the image cache would survive restarts outright, but on
-this cluster that volume faulted and blocked the runner, so it is deliberately not used.
+The durable fix is a persistent image store, which the runner now has: `/var/lib/docker`
+is a hostPath on node0 (see `oci-k8s/.../addons/tasks/main.yml`), so the act image survives
+a restart and is not re-pulled. The runner is pinned to **node0**, not node2 — node2 is a
+single-core control-plane node whose pod network was measured 21x slower under its own
+load, which starved every clone and pull. Its `/data` PVC is NFS ReadWriteMany, so a
+reschedule attaches immediately with no `Multi-Attach` wait. Entries 9 and 10 cover the
+caches and the node move in full.
 
 ### 6. Stopping a run, and reading a restarted runner correctly
 

libs/svcforge_core/svcforge_core/adapters/helm.py

@@ -21,17 +21,15 @@ from __future__ import annotations
 import asyncio
 import json
 import os
-import shutil
 import signal
-import tempfile
 from collections.abc import Sequence
 from pathlib import Path
 from typing import Any, Protocol
 
 import httpx
-import yaml
 from pydantic import BaseModel, ConfigDict, Field
 
+from svcforge_core.adapters.tempyaml import yaml_tempfile
 from svcforge_core.domain.models import CatalogEntry
 from svcforge_core.errors import SvcforgeError
 
@@ -220,7 +218,7 @@ class HelmProvisioner:
         # `--wait` is why `ready` in the DB means ready — it returns when the pods are up.
         # `--atomic` rolls back a failed upgrade; it doubles the worst case, which is what
         # `_RUN_TIMEOUT_MARGIN_S` and helm's own `--timeout` are sized around.
-        with _values_file(values) as path:
+        with yaml_tempfile(values, prefix="svcforge-values-", name="values.yaml") as path:
             argv = self._base_argv(
                 "upgrade",
                 "--install",
@@ -278,12 +276,12 @@ class HelmProvisioner:
         sweep. They are not orphans. They were never svcforge's to know about.
 
         Cost is why this does not shell out to helm when it does not have to. `--selector` is
-        NOT pushed down as a server-side selector — helm fetches and decompresses every
-        release secret in the cluster regardless, then filters what it already parsed.
-        Measured: unscoped 23 releases in 4392ms, scoped to 0 in 3988ms, a saving of about
-        10%, not the order of magnitude the flag's shape suggests.
+        applied by helm after it has already fetched and decompressed every release secret in
+        the cluster, so it saves almost nothing: measured unscoped 23 releases in 4392ms,
+        scoped to 0 in 3988ms, about 10%. The flag's shape suggests a server-side filter; it
+        is a client-side one.
 
-        That was not academic. With the CPU request mutated to 0 by a cluster policy, the
+        The cost was real. With the CPU request mutated to 0 by a cluster policy, the
         call took over 330s and timed out on every tick, against ~4s given real CPU. A check
         that never completes reports no drift, which looks exactly like no drift existing.
 
@@ -347,19 +345,27 @@ class HelmProvisioner:
           treating it as missing would have the reconciler re-provision on top of it.
         * helm writes one secret per revision, so a release can still appear more than once
           — 20 releases here had up to 10 revisions each. The newest `version` label wins.
-          Skipping this would report a release that exists as several, which for the
-          reconciler's set difference is harmless, and for anything counting releases is not.
+          Skipping this would report one release as several; the reconciler's set difference
+          tolerates that, but any caller that counts releases would over-count.
 
         `chart`, `status`, `revision` and `app_version` on the returned ReleaseInfo are the
-        subset the labels give away free. `chart` in particular is empty rather than wrong,
-        because the chart name lives only in the compressed payload. Narrowing the model
-        instead would have been honest about this method and dishonest about the helm path,
-        which does populate them.
+        subset the labels give away free. `chart` is empty here because the chart name lives
+        only in the compressed payload. The model keeps those fields so the helm fallback
+        path, which does populate them from `helm list`, returns the same shape.
         """
         try:
             token = _SA_TOKEN.read_text(encoding="utf-8").strip()
         except OSError:
             return None
+        # The CA has to be readable too, and it is checked here rather than left to httpx.
+        # httpx loads the CA eagerly when the client is built, and that load raises OSError,
+        # which is not in the (httpx.HTTPError, json.JSONDecodeError) except below. A
+        # half-mounted ServiceAccount — token present, ca.crt absent or late — would then
+        # crash the tick as a bare bug instead of falling back. A complete ServiceAccount is
+        # the real in-cluster signal, so a missing CA means "not in-cluster" like a missing
+        # token does.
+        if not token or not os.access(_SA_CA, os.R_OK):
+            return None
         host, port = (
             os.environ.get("KUBERNETES_SERVICE_HOST"),
             os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443"),
@@ -370,6 +376,12 @@ class HelmProvisioner:
         selector = f"{_HELM_OWNER_LABEL},{MANAGED_BY_LABEL}={MANAGED_BY_VALUE},{_LIVE_STATUSES}"
         try:
             async with httpx.AsyncClient(verify=str(_SA_CA), timeout=_API_TIMEOUT_S) as client:
+                # No `limit` param, and that is load-bearing: the apiserver only returns a
+                # `metadata.continue` token when the client sets `limit`, so with none set it
+                # returns the full matching set in one response and the single read below is
+                # complete. Adding `limit` here without also looping on `continue` would
+                # silently truncate the list, and the reconciler would read the missing
+                # releases as orphans to delete or as vanished releases to re-provision.
                 resp = await client.get(
                     f"https://{host}:{port}/api/v1/secrets",
                     params={"labelSelector": selector},
@@ -419,29 +431,3 @@ class HelmProvisioner:
                 ),
             )
         return [info for _, info in newest.values()]
-
-
-class _ValuesFile:
-    """Context manager yielding a path to a values.yaml written from a dict.
-
-    A file, not `--set`: `--set` has its own escaping grammar (commas, dots, backslashes) and
-    values carry tenant-shaped strings. Serialising YAML sidesteps the grammar entirely.
-    """
-
-    def __init__(self, values: dict[str, Any]) -> None:
-        self._values = values
-        self._dir: str | None = None
-
-    def __enter__(self) -> Path:
-        self._dir = tempfile.mkdtemp(prefix="svcforge-values-")
-        path = Path(self._dir) / "values.yaml"
-        path.write_text(yaml.safe_dump(self._values, default_flow_style=False), encoding="utf-8")
-        return path
-
-    def __exit__(self, *exc: object) -> None:
-        if self._dir is not None:
-            shutil.rmtree(self._dir, ignore_errors=True)
-
-
-def _values_file(values: dict[str, Any]) -> _ValuesFile:
-    return _ValuesFile(values)

libs/svcforge_core/svcforge_core/adapters/k8s.py

@@ -13,14 +13,11 @@ from __future__ import annotations
 import base64
 import binascii
 import json
-import shutil
-import tempfile
 from pathlib import Path
 from typing import Any
 
-import yaml
-
-from svcforge_core.adapters.helm import HelmError, _run
+from svcforge_core.adapters.helm import MANAGED_BY_LABEL, MANAGED_BY_VALUE, HelmError, _run
+from svcforge_core.adapters.tempyaml import yaml_tempfile
 from svcforge_core.errors import SvcforgeError
 
 _KUBECTL_TIMEOUT_S = 60
@@ -71,10 +68,10 @@ class KubectlClient:
             "kind": "Namespace",
             "metadata": {
                 "name": ns,
-                "labels": {"app.kubernetes.io/managed-by": "svcforge", **(labels or {})},
+                "labels": {MANAGED_BY_LABEL: MANAGED_BY_VALUE, **(labels or {})},
             },
         }
-        with _manifest_file(manifest) as path:
+        with yaml_tempfile(manifest, prefix="svcforge-manifest-", name="manifest.yaml") as path:
             await self._kubectl("apply", "--filename", str(path))
 
     async def read_secret(self, ns: str, name: str) -> dict[str, str]:
@@ -122,25 +119,3 @@ class KubectlClient:
             # timeout path does not come through HelmError. Without this clause a wedged
             # kubectl surfaces as TimeoutError past a caller written to `except K8sError`.
             raise K8sError(f"kubectl {args[0] if args else ''} timed out after {self._timeout_s}s") from exc
-
-
-class _ManifestFile:
-    """A temp file holding one YAML manifest, removed on exit."""
-
-    def __init__(self, manifest: dict[str, Any]) -> None:
-        self._manifest = manifest
-        self._dir: str | None = None
-
-    def __enter__(self) -> Path:
-        self._dir = tempfile.mkdtemp(prefix="svcforge-manifest-")
-        path = Path(self._dir) / "manifest.yaml"
-        path.write_text(yaml.safe_dump(self._manifest, default_flow_style=False), encoding="utf-8")
-        return path
-
-    def __exit__(self, *exc: object) -> None:
-        if self._dir is not None:
-            shutil.rmtree(self._dir, ignore_errors=True)
-
-
-def _manifest_file(manifest: dict[str, Any]) -> _ManifestFile:
-    return _ManifestFile(manifest)

libs/svcforge_core/svcforge_core/adapters/redis.py

@@ -39,7 +39,6 @@ you cared about.
 from __future__ import annotations
 
 import asyncio
-import logging
 import math
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -51,6 +50,7 @@ from pydantic import ValidationError
 from redis.asyncio import Redis
 from redis.exceptions import RedisError
 
+from svcforge_core import obs
 from svcforge_core.adapters.clock import Clock, SystemClock
 from svcforge_core.domain.models import Instance
 
@@ -59,7 +59,10 @@ if TYPE_CHECKING:
 
     from svcforge_core.settings import Settings
 
-_log = logging.getLogger(__name__)
+# structlog via obs, not stdlib logging. The stdlib bridge builds the event dict from the
+# record message alone and drops `extra=` fields — the same trap notify.py documents. A bound
+# logger takes fields as kwargs (team=team) and keeps them. Bound per instance in __init__,
+# which runs after obs.setup() has configured structlog, never at import time.
 
 # --- The budget metric ------------------------------------------------------------------
 #
@@ -175,6 +178,10 @@ class RateLimitResult:
     limit: int
     remaining: int
     reset_at: datetime
+    # When the limiter made this decision, from the same injected clock as reset_at. The two
+    # have to share a clock or retry_after_s (their difference) is meaningless under a
+    # FakeClock, and drifts by the request latency even in production.
+    checked_at: datetime
     degraded: bool = False
 
     @property
@@ -184,7 +191,7 @@ class RateLimitResult:
         Rounded up and floored at one: `Retry-After: 0` invites an immediate retry into
         the same closed window, which is a busy loop with extra steps.
         """
-        delta = (self.reset_at - datetime.now(UTC)).total_seconds()
+        delta = (self.reset_at - self.checked_at).total_seconds()
         return max(1, math.ceil(delta))
 
 
@@ -209,6 +216,7 @@ class RateLimiter:
 
     def __init__(self, r: Redis, limit: int, window_s: int, *, clock: Clock | None = None) -> None:
         """`clock` is injectable so the window boundary is testable without sleeping."""
+        self._log = obs.get_logger(__name__)
         if limit < 1:
             raise ValueError("limit must be >= 1")
         if window_s < 1:
@@ -244,16 +252,17 @@ class RateLimiter:
             allowed, remaining = await self._script(keys=[key], args=[self._limit, self._window_s])
         except _REDIS_DOWN:
             REDIS_ERRORS.labels(op="ratelimit").inc()
-            _log.warning(
+            self._log.warning(
                 "rate limiter degraded: redis unavailable, failing OPEN",
                 exc_info=True,
-                extra={"team": team},
+                team=team,
             )
             return RateLimitResult(
                 allowed=True,
                 limit=self._limit,
                 remaining=self._limit,
                 reset_at=reset_at,
+                checked_at=self._clock.now(),
                 degraded=True,
             )
         return RateLimitResult(
@@ -261,6 +270,7 @@ class RateLimiter:
             limit=self._limit,
             remaining=int(remaining),
             reset_at=reset_at,
+            checked_at=self._clock.now(),
         )
 
 
@@ -294,6 +304,7 @@ class IdempotencyStore:
 
     def __init__(self, r: Redis, ttl_s: int = 86400) -> None:
         """A day is the window a client might reasonably retry in; then the key is garbage."""
+        self._log = obs.get_logger(__name__)
         if ttl_s < 1:
             raise ValueError("ttl_s must be >= 1")
         self._r = r
@@ -320,7 +331,7 @@ class IdempotencyStore:
             existing = await self._r.get(redis_key)
         except _REDIS_DOWN:
             REDIS_ERRORS.labels(op="idempotency").inc()
-            _log.warning(
+            self._log.warning(
                 "idempotency degraded: redis unavailable, falling through to the DB constraint",
                 exc_info=True,
             )
@@ -333,7 +344,7 @@ class IdempotencyStore:
         try:
             return UUID(_as_text(existing))
         except ValueError:
-            _log.warning("idempotency key holds a non-UUID value; ignoring it")
+            self._log.warning("idempotency key holds a non-UUID value; ignoring it")
             return None
 
 
@@ -371,6 +382,7 @@ class InstanceCache:
     """
 
     def __init__(self, r: Redis, ttl_s: int = 30) -> None:
+        self._log = obs.get_logger(__name__)
         if ttl_s < 1:
             raise ValueError("ttl_s must be >= 1")
         self._r = r
@@ -391,7 +403,7 @@ class InstanceCache:
             raw = await self._r.get(self._key(instance_id))
         except _REDIS_DOWN:
             REDIS_ERRORS.labels(op="cache_get").inc()
-            _log.warning("cache read degraded: redis unavailable, falling through to Postgres")
+            self._log.warning("cache read degraded: redis unavailable, falling through to Postgres")
             return None
         if raw is None:
             return None
@@ -400,7 +412,7 @@ class InstanceCache:
         except ValidationError:
             # A model change deployed over a warm cache. Treat it as a miss and let the TTL
             # take the old shape out. Not an error: the truth is in Postgres either way.
-            _log.info("cache entry failed validation; treating as a miss")
+            self._log.info("cache entry failed validation; treating as a miss")
             return None
 
     async def put(self, inst: Instance) -> None:
@@ -410,7 +422,7 @@ class InstanceCache:
             await self._r.set(self._key(inst.id), inst.model_dump_json(), ex=self._ttl_s)
         except _REDIS_DOWN:
             REDIS_ERRORS.labels(op="cache_put").inc()
-            _log.warning("cache write degraded: redis unavailable")
+            self._log.warning("cache write degraded: redis unavailable")
 
     async def invalidate(self, instance_id: UUID) -> None:
         """One DEL. Called by the worker inside the code path that writes the state.
@@ -423,4 +435,4 @@ class InstanceCache:
             await self._r.delete(self._key(instance_id))
         except _REDIS_DOWN:
             REDIS_ERRORS.labels(op="cache_del").inc()
-            _log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL")
+            self._log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL")

libs/svcforge_core/svcforge_core/adapters/tempyaml.py

@@ -0,0 +1,34 @@
+"""One temp YAML file, written from a dict and removed on exit.
+
+helm and kubectl both take their input as a file rather than on the command line: `--set`
+and inline manifests each have their own escaping grammar, and tenant-shaped values would
+have to be escaped into it. Serialising YAML to a file sidesteps the grammar entirely. Both
+adapters needed the same throwaway-file dance, so it lives here once.
+"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+@contextmanager
+def yaml_tempfile(payload: dict[str, Any], *, prefix: str, name: str) -> Iterator[Path]:
+    """Yield a path to `name` inside a fresh temp dir, holding `payload` as YAML.
+
+    The whole dir is removed on exit, `ignore_errors` so a cleanup race never masks the real
+    error from the block.
+    """
+    tmpdir = tempfile.mkdtemp(prefix=prefix)
+    try:
+        path = Path(tmpdir) / name
+        path.write_text(yaml.safe_dump(payload, default_flow_style=False), encoding="utf-8")
+        yield path
+    finally:
+        shutil.rmtree(tmpdir, ignore_errors=True)

libs/svcforge_core/svcforge_core/domain/catalog.py

@@ -10,9 +10,10 @@ import yaml
 from pydantic import ValidationError
 
 from svcforge_core.domain.models import CatalogEntry
+from svcforge_core.errors import SvcforgeError
 
 
-class CatalogError(Exception):
+class CatalogError(SvcforgeError):
     """A catalog file could not be parsed or validated.
 
     `key` names the offending service type, or None when the failure is file-level.

libs/svcforge_core/svcforge_core/domain/states.py

@@ -3,6 +3,8 @@
 from enum import StrEnum
 from typing import Final
 
+from svcforge_core.errors import SvcforgeError
+
 
 class InstanceState(StrEnum):
     """Lifecycle of a provisioned service instance."""
@@ -15,7 +17,7 @@ class InstanceState(StrEnum):
     FAILED = "failed"
 
 
-class IllegalTransition(Exception):
+class IllegalTransition(SvcforgeError):
     """Raised by transition() when cur -> nxt is not in LEGAL."""
 
 

libs/svcforge_core/svcforge_core/domain/windows.py

@@ -19,11 +19,13 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from croniter import croniter
 
+from svcforge_core.errors import SvcforgeError
+
 CRON_FIELDS = 5
 SEPARATOR = "|"
 
 
-class BadWindow(ValueError):
+class BadWindow(SvcforgeError, ValueError):
     """A maintenance window spec is not a 5-field cron plus a known IANA zone."""
 
 

libs/svcforge_core/svcforge_core/repo/db.py

@@ -17,6 +17,11 @@ from psycopg_pool import AsyncConnectionPool
 # against a bare `AsyncConnectionPool`, which resolves to tuple rows. The runtime was
 # always right; without these aliases the annotations quietly disagree with it, and the
 # fix people reach for is `# type: ignore`, which throws away the checking entirely.
+# The cap on error text written to the `instances.error` and `tasks.last_error` columns. A
+# helm failure can emit megabytes; these columns are read by humans. Defined once here rather
+# than as a bare 2000 at each write, so the two call paths that feed the same columns agree.
+ERROR_MAX_CHARS = 2000
+
 type DictRow = dict[str, Any]
 type DictConnection = AsyncConnection[DictRow]
 type DictPool = AsyncConnectionPool[DictConnection]

libs/svcforge_core/svcforge_core/repo/instances.py

@@ -20,7 +20,7 @@ from svcforge_core.domain.models import Instance
 from svcforge_core.domain.states import InstanceState
 from svcforge_core.repo.db import DictPool
 
-_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version,
+INSTANCE_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version,
               endpoint, error, expires_at, created_at, updated_at"""
 
 
@@ -57,7 +57,7 @@ class InstanceRepo:
                                            release_name, chart_version, endpoint, error, expires_at)
                     values (%(id)s, %(team)s, %(service_type)s, %(size)s, %(state)s, %(namespace)s,
                             %(release_name)s, %(chart_version)s, %(endpoint)s, %(error)s, %(expires_at)s)
-                    returning {_COLUMNS}""",  # noqa: S608 - _COLUMNS is a module constant, not input
+                    returning {INSTANCE_COLUMNS}""",  # noqa: S608 - INSTANCE_COLUMNS is a module constant, not input
                 {
                     "id": inst.id,
                     "team": inst.team,
@@ -80,7 +80,7 @@ class InstanceRepo:
         """Fetch one instance owned by `team`. None if it does not exist OR is not theirs."""
         async with self._pool.connection() as conn, conn.cursor() as cur:
             await cur.execute(
-                f"select {_COLUMNS} from instances where id = %s and team = %s",  # noqa: S608
+                f"select {INSTANCE_COLUMNS} from instances where id = %s and team = %s",  # noqa: S608
                 (id, team),
             )
             row = await cur.fetchone()
@@ -90,7 +90,7 @@ class InstanceRepo:
         """The team's instances, newest first."""
         async with self._pool.connection() as conn, conn.cursor() as cur:
             await cur.execute(
-                f"""select {_COLUMNS} from instances
+                f"""select {INSTANCE_COLUMNS} from instances
                     where team = %s order by created_at desc limit %s""",  # noqa: S608
                 (team, limit),
             )
@@ -157,7 +157,7 @@ class InstanceRepo:
         # and would otherwise type these rows as tuples.
         async with self._pool.connection() as conn, conn.cursor(row_factory=dict_row) as cur:
             await cur.execute(
-                f"""select {_COLUMNS}, maintenance_window from instances
+                f"""select {INSTANCE_COLUMNS}, maintenance_window from instances
                      where state = 'ready'
                        and service_type = %(service_type)s
                        and chart_version <> %(catalog_version)s
@@ -166,7 +166,7 @@ class InstanceRepo:
                               where cv.service_type = %(service_type)s
                                 and cv.rollout_state = 'halted')
                      order by team = %(own_team)s desc, created_at
-                     limit %(max_in_flight)s""",  # noqa: S608 - _COLUMNS is a module constant, not input
+                     limit %(max_in_flight)s""",  # noqa: S608 - INSTANCE_COLUMNS is a module constant, not input
                 {
                     "service_type": service_type,
                     "catalog_version": catalog_version,

libs/svcforge_core/svcforge_core/repo/reconcile.py

@@ -25,10 +25,8 @@ from psycopg import AsyncCursor
 from svcforge_core.domain.models import Instance, TaskKind, TaskState
 from svcforge_core.domain.states import InstanceState, transition
 from svcforge_core.obs import inject_traceparent
-from svcforge_core.repo.db import DictPool
-
-_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version,
-              endpoint, error, expires_at, created_at, updated_at"""
+from svcforge_core.repo.db import ERROR_MAX_CHARS, DictPool
+from svcforge_core.repo.instances import INSTANCE_COLUMNS
 
 # A task nobody will ever run again. The idempotency guard on every enqueue below asks
 # "is one already outstanding?", and 'done'/'failed' are not outstanding: a failed
@@ -70,7 +68,7 @@ class ReconcileRepo:
         """Every instance the DB believes is running. The drift check's expectation."""
         async with self._pool.connection() as conn, conn.cursor() as cur:
             await cur.execute(
-                f"select {_COLUMNS} from instances where state = %s",  # noqa: S608 - module constant
+                f"select {INSTANCE_COLUMNS} from instances where state = %s",  # noqa: S608 - module constant
                 (InstanceState.READY.value,),
             )
             rows = await cur.fetchall()
@@ -126,7 +124,7 @@ class ReconcileRepo:
 
                 await cur.execute(
                     "update instances set state = %s, error = %s, updated_at = now() where id = %s",
-                    (provisioning.value, reason[-2000:], instance_id),
+                    (provisioning.value, reason[-ERROR_MAX_CHARS:], instance_id),
                 )
                 return await _insert_task(cur, instance_id, TaskKind.PROVISION)
 
@@ -151,7 +149,7 @@ class ReconcileRepo:
         """
         async with self._pool.connection() as conn, conn.cursor() as cur:
             await cur.execute(
-                f"""select {_COLUMNS} from instances i
+                f"""select {INSTANCE_COLUMNS} from instances i
                      where ((i.state = %(ready)s and i.expires_at < now()) or i.state = %(deleting)s)
                        and not exists (
                              select 1 from tasks t

libs/svcforge_core/svcforge_core/repo/tasks.py

@@ -20,7 +20,7 @@ from svcforge_core.domain.backoff import next_attempt_at
 from svcforge_core.domain.models import Task, TaskKind
 from svcforge_core.domain.states import LEGAL, InstanceState
 from svcforge_core.obs import TASKS_DEAD_LETTERED, inject_traceparent
-from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.db import ERROR_MAX_CHARS, DictPool
 
 # Which states may legally become `failed`, derived from the domain's own table rather
 # than restated here. Without this guard the UPDATE below would happily move a `deleted`
@@ -205,7 +205,7 @@ class TaskRepo:
                               set state='queued', locked_by=null, locked_at=null,
                                   last_error=%s, run_after=%s
                             where id = %s""",
-                        (err[-2000:], next_attempt_at(attempts - 1, now=now), task_id),
+                        (err[-ERROR_MAX_CHARS:], next_attempt_at(attempts - 1, now=now), task_id),
                     )
                     return True
 
@@ -213,15 +213,29 @@ class TaskRepo:
                     """update tasks
                           set state='failed', locked_by=null, locked_at=null, last_error=%s
                         where id = %s""",
-                    (err[-2000:], task_id),
+                    (err[-ERROR_MAX_CHARS:], task_id),
                 )
-                # `state = any(%s)` keeps this honest: a deprovision that exhausts its
-                # retries against an already-deleted instance records nothing rather than
-                # resurrecting it into `failed`.
+                # Dead-lettering the task is correct for every kind. Moving the INSTANCE to
+                # `failed` is correct only for provision: a provisioning instance that never
+                # came up is failed, and nothing recovers it but a human. The other kinds
+                # must leave the instance where it is, because for each of them the instance
+                # is still healthy and something else is responsible for recovery:
+                #   deprovision — still `deleting`, which is exactly what lets
+                #                 due_for_deprovision re-enqueue it on the next sweep. `failed`
+                #                 drops it out of that query and leaks the release forever.
+                #   upgrade     — helm --atomic rolled back, so it is still `ready` and
+                #                 serving the previous version. check_version_drift retries on
+                #                 the next window; `failed` would mislabel a working service
+                #                 and drop it off the upgrade work-list.
+                #   verify      — handle_verify already halted the rollout; the instance is
+                #                 `ready`, and drift re-provisions it if its release vanished.
+                # The dead-letter metric and its alert are the operator signal for all four,
+                # so leaving the instance alone loses no visibility.
+                if row["kind"] == TaskKind.PROVISION.value:
                     await cur.execute(
                         """update instances set error=%s, state=%s, updated_at=now()
                             where id=%s and state = any(%s)""",
-                    (err[-2000:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)),
+                        (err[-ERROR_MAX_CHARS:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)),
                     )
                 # Counted here, not in the worker: this is the only place that knows the
                 # difference between "attempt 2 of 5 failed" and "this task is done trying".

libs/svcforge_core/svcforge_core/settings.py

@@ -83,6 +83,15 @@ class Settings(BaseSettings):
     # just for /metrics. 9000 matches the chart's PodMonitor; change both or neither.
     metrics_port: int = Field(default=9000, ge=1, le=65535)
 
+    @property
+    def runtime_dsn(self) -> str:
+        """The transaction-pooler DSN the services open their pool against, as a string.
+
+        A property so the three entrypoints do not each choose between `str(pg_dsn)` and
+        `pg_dsn.unicode_string()` — the two spellings that were drifting across the services.
+        """
+        return str(self.pg_dsn)
+
     @property
     def migration_dsn(self) -> str:
         """Migrations need a session-mode connection; fall back to the runtime DSN locally."""

services/_runtime.py

@@ -0,0 +1,37 @@
+"""Shared asyncio scaffolding for the long-lived services.
+
+The worker and the reconciler are both a loop that runs until SIGTERM. They wake immediately
+on shutdown rather than sleeping through it, and they install the same loop-safe signal
+handlers. Both lived in each service before; keeping one copy means the shutdown behaviour
+cannot drift between them.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import signal
+
+
+async def sleep_or_stop(stop: asyncio.Event, seconds: float) -> None:
+    """Sleep for `seconds`, but return the instant `stop` is set.
+
+    `await asyncio.sleep(seconds)` would make every SIGTERM cost up to `seconds` of
+    Kubernetes waiting on terminationGracePeriod for nothing.
+    """
+    with contextlib.suppress(TimeoutError):
+        await asyncio.wait_for(stop.wait(), timeout=seconds)
+
+
+def install_stop_signals(stop: asyncio.Event) -> None:
+    """Set `stop` on SIGTERM and SIGINT, loop-safely.
+
+    add_signal_handler, not signal.signal. signal.signal runs the handler at an arbitrary
+    bytecode boundary on whatever thread the C-level handler lands on, and the loop does not
+    notice until its next timer fires — up to a full sleep interval away. add_signal_handler
+    schedules the callback as an ordinary loop callback, so the sleep_or_stop above returns
+    at once.
+    """
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGTERM, signal.SIGINT):
+        loop.add_signal_handler(sig, stop.set)

services/api/main.py

@@ -11,9 +11,11 @@ import asyncio
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from jwt import PyJWKClient
+from starlette.exceptions import HTTPException
 
 from services.api.models import ErrorBody
 from services.api.routes import health, instances
@@ -53,7 +55,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         RateLimiter(redis, limit=settings.rate_limit_per_minute, window_s=60) if redis is not None else None
     )
 
-    pool = make_pool(str(settings.pg_dsn), settings.pool_min_size, settings.pool_max_size)
+    pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size)
     # wait=True fails NOW, loudly, if the DSN is wrong — instead of at the first request,
     # as a PoolTimeout, in front of a user.
     await pool.open(wait=True)
@@ -89,6 +91,13 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo
     Handlers raise `detail={"code": ..., "message": ...}`; FastAPI's default would nest
     that under `{"detail": {...}}`. Plain-string details (raised by FastAPI itself, e.g.
     a 405) are wrapped so clients never have to branch on the body's type.
+
+    Registered on starlette's HTTPException, not fastapi's. fastapi.HTTPException is a
+    subclass, and Starlette matches handlers by walking type(exc).__mro__, so a handler
+    keyed on the subclass never fires for a framework-raised 404 or 405 — which are
+    starlette.HTTPException instances. Keying on the parent catches both: app handlers
+    raise the FastAPI subclass with a dict detail, the framework raises the parent with a
+    str detail, and the branch below renders each into ErrorBody.
     """
     assert isinstance(exc, HTTPException)  # noqa: S101 - registered only for HTTPException
     # Widened to object deliberately. Starlette types `detail` as str, but FastAPI passes
@@ -102,6 +111,21 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo
     return JSONResponse(status_code=exc.status_code, content=body.model_dump(), headers=exc.headers)
 
 
+async def _validation_exception_handler(request: Request, exc: Exception) -> JSONResponse:
+    """Render request-validation failures as ErrorBody too.
+
+    A body that fails validation (a forbidden extra field, a bad type, an out-of-range
+    ttl_days) raises RequestValidationError, which the HTTPException handler above never
+    sees. Without this it returns FastAPI's default `{"detail": [...]}` — a second 422 shape
+    alongside the ErrorBody 422s the handlers raise. This gives every 422 one shape.
+    """
+    assert isinstance(exc, RequestValidationError)  # noqa: S101 - registered only for this
+    return JSONResponse(
+        status_code=422,
+        content=ErrorBody(code="validation_error", message=str(exc.errors())).model_dump(),
+    )
+
+
 def create_app(settings: Settings | None = None) -> FastAPI:
     """App factory: lifespan, routers, exception handler, /metrics."""
     settings = settings or load_settings()
@@ -132,6 +156,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.include_router(instances.router)
 
     app.add_exception_handler(HTTPException, _http_exception_handler)
+    app.add_exception_handler(RequestValidationError, _validation_exception_handler)
     return app
 
 

services/reconciler/main.py

@@ -23,19 +23,18 @@ Three rules hold the design together:
   binary that cannot reach the API server must not stop TTLs from expiring.
 * **Enqueue, never act.** The reconciler diagnoses; workers treat. It writes task rows and
   instance states, and never calls `helm install`. The one exception is reading — the drift
-  check runs `helm list`, because seeing reality is the job.
+  check lists the live releases, because seeing reality is the job.
 """
 
 from __future__ import annotations
 
 import asyncio
-import contextlib
-import signal
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 
 import typer
 
+from services._runtime import install_stop_signals, sleep_or_stop
 from svcforge_core.adapters.clock import Clock, SystemClock
 from svcforge_core.adapters.helm import HelmProvisioner, Provisioner
 from svcforge_core.adapters.notify import LogNotifier, Notifier
@@ -59,9 +58,6 @@ from svcforge_core.settings import Settings, load_settings
 
 log = get_logger("svcforge.reconciler")
 
-# The chart's PodMonitor scrapes the port named `metrics` on 9000. Keep them in step.
-DEFAULT_METRICS_PORT = 9000
-
 
 @dataclass(frozen=True)
 class ReconcilerDeps:
@@ -92,7 +88,7 @@ class ReconcilerDeps:
 
 
 async def check_drift(deps: ReconcilerDeps) -> None:
-    """`helm list -A -o json` versus what the database believes.
+    """The live helm releases versus what the database believes.
 
     This is the only check that looks outside Postgres, and the only one that can catch the
     failure nothing else can: someone ran `helm uninstall` by hand, or a node was drained
@@ -312,12 +308,6 @@ async def _run_checks(deps: ReconcilerDeps) -> None:
         log.exception("gauges.failed")
 
 
-async def _sleep_or_stop(stop: asyncio.Event, seconds: float) -> None:
-    """Sleep, but wake immediately on SIGTERM. A 60s nap must not cost 60s of shutdown."""
-    with contextlib.suppress(TimeoutError):
-        await asyncio.wait_for(stop.wait(), timeout=seconds)
-
-
 async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None:
     """Tick, sleep, repeat, until told to stop.
 
@@ -328,7 +318,7 @@ async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None:
     """
     while not stop.is_set():
         await tick(deps)
-        await _sleep_or_stop(stop, deps.settings.reconcile_interval_s)
+        await sleep_or_stop(stop, deps.settings.reconcile_interval_s)
 
 
 def build_deps(
@@ -353,11 +343,11 @@ def build_deps(
     )
 
 
-async def _amain(once: bool, metrics_port: int, own_team: str, max_in_flight: int) -> None:
+async def _amain(once: bool, own_team: str, max_in_flight: int) -> None:
     settings = load_settings()
     setup("svcforge-reconciler", settings)
 
-    pool = make_pool(settings.pg_dsn.unicode_string(), settings.pool_min_size, settings.pool_max_size)
+    pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size)
     await pool.open(wait=True)
     deps = build_deps(pool, settings, own_team, max_in_flight)
 
@@ -368,17 +358,12 @@ async def _amain(once: bool, metrics_port: int, own_team: str, max_in_flight: in
             await tick(deps)
             return
 
-        start_metrics_server(metrics_port)
+        # settings.metrics_port, like the worker. SVCFORGE_METRICS_PORT still overrides it,
+        # through pydantic rather than a second CLI option, so the port has one definition.
+        start_metrics_server(settings.metrics_port)
 
         stop = asyncio.Event()
-        loop = asyncio.get_running_loop()
-        for sig in (signal.SIGTERM, signal.SIGINT):
-            # add_signal_handler, NOT signal.signal. signal.signal fires the handler at an
-            # arbitrary bytecode boundary on the main thread and the loop does not notice
-            # until its next timer — which here is up to a full 60s tick away. This one is
-            # scheduled as an ordinary loop callback, so the `stop.wait()` above returns
-            # immediately.
-            loop.add_signal_handler(sig, stop.set)
+        install_stop_signals(stop)
 
         await run_reconciler(deps, stop)
     finally:
@@ -391,9 +376,6 @@ app = typer.Typer(add_completion=False, help="svcforge reconciler: the control l
 @app.command()
 def main(
     once: bool = typer.Option(False, "--once", help="Run one tick and exit."),
-    metrics_port: int = typer.Option(
-        DEFAULT_METRICS_PORT, envvar="SVCFORGE_METRICS_PORT", help="Port for /metrics."
-    ),
     own_team: str = typer.Option(
         "platform", envvar="SVCFORGE_OWN_TEAM", help="Team whose instances upgrade first."
     ),
@@ -403,7 +385,7 @@ def main(
 ) -> None:
     """Run the reconciler."""
     # One asyncio.run, at the top, never nested. Everything below it is already async.
-    asyncio.run(_amain(once, metrics_port, own_team, max_in_flight))
+    asyncio.run(_amain(once, own_team, max_in_flight))
 
 
 if __name__ == "__main__":

services/worker/handlers.py

@@ -18,18 +18,21 @@ from typing import Any
 from services.worker.deps import WorkerDeps
 from svcforge_core.domain.models import CatalogEntry, Instance, Task, TaskKind
 from svcforge_core.domain.states import InstanceState
+from svcforge_core.errors import SvcforgeError
+from svcforge_core.obs import get_logger
+from svcforge_core.repo.instances import INSTANCE_COLUMNS
+
+log = get_logger("svcforge.worker")
 
 
-class HandlerError(RuntimeError):
+class HandlerError(SvcforgeError, RuntimeError):
     """A task failed in a way worth retrying. The message lands in tasks.last_error."""
 
 
 async def _load_instance(task: Task, deps: WorkerDeps) -> Instance:
     async with deps.pool.connection() as conn, conn.cursor() as cur:
         await cur.execute(
-            """select id, team, service_type, size, state, namespace, release_name,
-                      chart_version, endpoint, error, expires_at, created_at, updated_at
-                 from instances where id = %s""",
+            f"select {INSTANCE_COLUMNS} from instances where id = %s",  # noqa: S608 - module constant
             (task.instance_id,),
         )
         row = await cur.fetchone()
@@ -75,11 +78,19 @@ async def handle_provision(task: Task, deps: WorkerDeps) -> None:
         inst.id, InstanceState.PROVISIONING, InstanceState.READY, endpoint=endpoint
     )
     if ok:
+        try:
             await deps.notifier.send(
                 "instance.ready",
                 f"instance {inst.id} is ready at {endpoint}",
                 {"instance_id": str(inst.id), "team": inst.team, "service_type": inst.service_type},
             )
+        except Exception:
+            # The provision succeeded and the row is already READY; the notification is a
+            # courtesy. Letting a webhook timeout propagate would fail the task, and the
+            # retry would hit the READY early-return and drop the notification anyway — so a
+            # flaky notifier would turn every provision into a "failed" task. Same guard the
+            # reconciler puts around its own notify.
+            log.exception("notify.failed", instance_id=str(inst.id))
 
 
 async def handle_deprovision(task: Task, deps: WorkerDeps) -> None:

services/worker/main.py

@@ -10,13 +10,12 @@ for something better.
 from __future__ import annotations
 
 import asyncio
-import contextlib
-import signal
 import time
 from collections.abc import Awaitable
 
 from opentelemetry import trace
 
+from services._runtime import install_stop_signals, sleep_or_stop
 from services.worker.deps import WorkerDeps
 from services.worker.handlers import HANDLERS
 from svcforge_core import obs
@@ -33,16 +32,6 @@ from svcforge_core.settings import Settings, load_settings
 log = obs.get_logger("svcforge.worker")
 
 
-async def _sleep_or_stop(stop: asyncio.Event, seconds: float) -> None:
-    """Sleep, but wake immediately on shutdown.
-
-    `await asyncio.sleep(5)` would make every SIGTERM cost up to five seconds of
-    Kubernetes waiting on terminationGracePeriod for no reason.
-    """
-    with contextlib.suppress(TimeoutError):
-        await asyncio.wait_for(stop.wait(), timeout=seconds)
-
-
 async def _report(coro: Awaitable[bool], task_id: int, what: str) -> None:
     """Run a terminal report, and never let its failure escape.
 
@@ -154,12 +143,12 @@ async def run_worker(deps: WorkerDeps, stop: asyncio.Event) -> None:
                 # A DB blip must not kill the worker; back off and try again.
                 log.exception("claim failed")
                 sem.release()
-                await _sleep_or_stop(stop, deps.settings.poll_interval_s)
+                await sleep_or_stop(stop, deps.settings.poll_interval_s)
                 continue
 
             if task is None:
                 sem.release()
-                await _sleep_or_stop(stop, deps.settings.poll_interval_s)
+                await sleep_or_stop(stop, deps.settings.poll_interval_s)
                 continue
 
             tg.create_task(_run_one(deps, task, sem))
@@ -176,7 +165,7 @@ async def _amain() -> None:
     settings.check_production()
     obs.start_metrics_server(settings.metrics_port)
 
-    pool = make_pool(settings.pg_dsn.unicode_string(), settings.pool_min_size, settings.pool_max_size)
+    pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size)
     await pool.open(wait=True)
 
     deps = WorkerDeps(
@@ -191,13 +180,7 @@ async def _amain() -> None:
     )
 
     stop = asyncio.Event()
-    loop = asyncio.get_running_loop()
-    for sig in (signal.SIGTERM, signal.SIGINT):
-        # add_signal_handler, NOT signal.signal. signal.signal runs the handler at an
-        # arbitrary bytecode boundary on whatever thread the C-level handler lands on,
-        # and the event loop will not notice until its next timer fires. This one is
-        # loop-safe: the callback runs as a normal loop callback.
-        loop.add_signal_handler(sig, stop.set)
+    install_stop_signals(stop)
 
     try:
         await run_worker(deps, stop)

tests/fakes.py

@@ -145,7 +145,12 @@ class FakeRateLimiter:
             # Fails OPEN, exactly like the real one. A limiter that refused here would make
             # "Redis is down" indistinguishable from "you are over quota".
             return RateLimitResult(
-                allowed=True, limit=self.limit, remaining=self.limit, reset_at=reset_at, degraded=True
+                allowed=True,
+                limit=self.limit,
+                remaining=self.limit,
+                reset_at=reset_at,
+                checked_at=self.clock.now(),
+                degraded=True,
             )
         key = f"rl:{team}:{window}"
         n = self.counts.get(key, 0) + 1
@@ -155,6 +160,7 @@ class FakeRateLimiter:
             limit=self.limit,
             remaining=max(0, self.limit - n),
             reset_at=reset_at,
+            checked_at=self.clock.now(),
         )
 
 

tests/integration/test_api.py

@@ -298,6 +298,46 @@ async def test_ttl_out_of_range_is_422(client: httpx.AsyncClient, token: str) ->
     assert resp.status_code == 422
 
 
+def _is_error_body(payload: object) -> bool:
+    """The uniform error shape: a dict with `code` and `message`, and no default `detail`."""
+    return isinstance(payload, dict) and "code" in payload and "message" in payload
+
+
+async def test_framework_404_uses_the_error_body_shape(client: httpx.AsyncClient) -> None:
+    """A 404 raised by the router, not a handler, must still be ErrorBody.
+
+    Starlette raises its own HTTPException for an unknown route. The exception handler is
+    registered on that parent class precisely so this body is ErrorBody and not FastAPI's
+    default `{"detail": "Not Found"}` — one shape for every error.
+    """
+    resp = await client.get("/v1/no-such-route")
+    assert resp.status_code == 404
+    assert _is_error_body(resp.json()), resp.text
+
+
+async def test_framework_405_uses_the_error_body_shape(client: httpx.AsyncClient) -> None:
+    """A wrong-method 405 comes from the router too, and must be ErrorBody."""
+    resp = await client.delete("/v1/instances")  # collection route has no DELETE
+    assert resp.status_code == 405
+    assert _is_error_body(resp.json()), resp.text
+
+
+async def test_body_validation_422_uses_the_error_body_shape(client: httpx.AsyncClient, token: str) -> None:
+    """A RequestValidationError 422 must match the handler-raised 422 shape.
+
+    A forbidden extra field trips pydantic's `extra="forbid"` and raises
+    RequestValidationError, which the dedicated handler renders as ErrorBody rather than the
+    default `{"detail": [...]}`.
+    """
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "redis", "size": "small", "surprise": "field"},
+    )
+    assert resp.status_code == 422
+    assert _is_error_body(resp.json()), resp.text
+
+
 # --------------------------------------------------------------------------- authn
 
 

tests/integration/test_redis.py

@@ -434,7 +434,7 @@ async def test_the_budget_metric_is_exposed_and_labelled_by_op() -> None:
     """`curl -s localhost:8000/metrics | grep svcforge_redis_commands_total`."""
     limiter = FakeRateLimiter(limit=10, window_s=60, clock=FakeClock(start=_T0))
     await limiter.check("acme")  # the fake does not touch the real counter
-    RateLimitResult(allowed=True, limit=10, remaining=9, reset_at=_T0)
+    RateLimitResult(allowed=True, limit=10, remaining=9, reset_at=_T0, checked_at=_T0)
 
     text = generate_latest(REGISTRY).decode()
 

tests/integration/test_tasks.py

@@ -110,6 +110,86 @@ async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> Non
     assert inst.error is None
 
 
+async def test_fail_of_deprovision_leaves_the_instance_deleting_to_be_retried(
+    pool: DictPool,
+) -> None:
+    """A dead-lettered deprovision must not strand the instance in `failed`.
+
+    The instance is `deleting`, which is one of the states that CAN legally become `failed`,
+    so the naive blanket UPDATE would move it there. due_for_deprovision only re-selects
+    `ready`(expired) and `deleting`, so `failed` would take the instance out of the recovery
+    sweep and leak the helm release forever. reconcile.py documents that a deprovision which
+    exhausts its retries stays re-enqueueable; this pins that guarantee.
+    """
+    iid = await make_instance(pool, state=InstanceState.DELETING)
+    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
+    tid = await tasks.enqueue_standalone(iid, TaskKind.DEPROVISION)
+
+    claimed = await tasks.claim("w1")
+    assert claimed is not None
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update tasks set attempts = 5 where id = %s", (tid,))
+
+    assert await tasks.fail(tid, "cluster unreachable", "w1", max_attempts=5) is True
+
+    assert (await _task_row(pool, tid))["state"] == "failed"
+    inst = await instances.get(iid, team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.DELETING, "a stranded deprovision leaks the release"
+    assert inst.error is None
+
+
+async def test_fail_of_upgrade_leaves_a_working_instance_ready(pool: DictPool) -> None:
+    """A dead-lettered upgrade must not mark a healthy instance `failed`.
+
+    helm --atomic rolls the release back, so after a failed upgrade the instance is still
+    `ready` and serving the previous version. Marking it `failed` mislabels a working
+    service and drops it off the upgrade work-list. check_version_drift retries on the next
+    window; the dead-letter metric is the operator signal.
+    """
+    iid = await make_instance(pool, state=InstanceState.READY)
+    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
+    tid = await tasks.enqueue_standalone(iid, TaskKind.UPGRADE)
+
+    claimed = await tasks.claim("w1")
+    assert claimed is not None
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update tasks set attempts = 5 where id = %s", (tid,))
+
+    assert await tasks.fail(tid, "upgrade to 1.4.0 kept timing out", "w1", max_attempts=5) is True
+
+    assert (await _task_row(pool, tid))["state"] == "failed"
+    inst = await instances.get(iid, team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.READY, "a failed upgrade mislabelled a healthy instance"
+    assert inst.error is None
+
+
+async def test_fail_of_verify_leaves_the_instance_ready(pool: DictPool) -> None:
+    """A dead-lettered verify must not mark the instance `failed`.
+
+    handle_verify halts the rollout for the service type; the instance itself is `ready`,
+    and drift re-provisions it if its release vanished. `failed` would take it out of both
+    recovery paths.
+    """
+    iid = await make_instance(pool, state=InstanceState.READY)
+    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
+    tid = await tasks.enqueue_standalone(iid, TaskKind.VERIFY)
+
+    claimed = await tasks.claim("w1")
+    assert claimed is not None
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update tasks set attempts = 5 where id = %s", (tid,))
+
+    assert await tasks.fail(tid, "release vanished after upgrade", "w1", max_attempts=5) is True
+
+    assert (await _task_row(pool, tid))["state"] == "failed"
+    inst = await instances.get(iid, team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.READY
+    assert inst.error is None
+
+
 async def test_fail_truncates_error_to_2kb(pool: DictPool) -> None:
     iid = await make_instance(pool)
     repo = TaskRepo(pool)

tests/unit/test_helm_argv.py

@@ -21,6 +21,7 @@ from svcforge_core.adapters import helm
 from svcforge_core.adapters.helm import (
     MANAGED_BY_LABEL,
     MANAGED_BY_VALUE,
+    HelmError,
     HelmProvisioner,
 )
 from svcforge_core.domain.models import CatalogEntry, SizeSpec
@@ -131,8 +132,10 @@ def _fake_api(
 
     token = tmp_path / "token"
     token.write_text("tok", encoding="utf-8")
+    ca = tmp_path / "ca.crt"
+    ca.write_text("ca", encoding="utf-8")  # a complete SA has both; the readability check needs it
     monkeypatch.setattr(helm, "_SA_TOKEN", token)
-    monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
+    monkeypatch.setattr(helm, "_SA_CA", ca)
     monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
     monkeypatch.setenv("KUBERNETES_SERVICE_PORT_HTTPS", "443")
 
@@ -267,3 +270,89 @@ async def test_api_read_handles_kubernetes_serialising_empty_as_null(
     monkeypatch.setattr(httpx, "AsyncClient", _NullClient)
 
     assert await HelmProvisioner().list_releases() == []
+
+
+@pytest.mark.asyncio
+async def test_api_read_passes_tls_verify_and_timeout(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """The API client must verify against the SA CA and carry the read timeout.
+
+    A refactor that dropped `verify` to the default (or None) is a TLS regression the happy
+    path would not reveal, so it is pinned here off the captured client kwargs.
+    """
+    cap = _fake_api(monkeypatch, tmp_path, [])
+
+    await HelmProvisioner().list_releases()
+
+    assert cap["client_kwargs"]["verify"] == str(helm._SA_CA)
+    assert cap["client_kwargs"]["timeout"] == helm._API_TIMEOUT_S
+
+
+@pytest.mark.asyncio
+async def test_api_read_sends_no_limit_param(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """No `limit`, so the apiserver returns the full set and the single read is complete.
+
+    Pins the pagination invariant: adding `limit` without consuming `metadata.continue`
+    would silently truncate the release list.
+    """
+    cap = _fake_api(monkeypatch, tmp_path, [])
+
+    await HelmProvisioner().list_releases()
+
+    assert "limit" not in cap["params"]
+
+
+@pytest.mark.asyncio
+async def test_api_error_raises_helmerror_and_does_not_fall_back(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A reachable-but-erroring apiserver raises HelmError; it must not shell out to helm.
+
+    Falling back would swap a visible error for the 330s helm timeout this method exists to
+    remove. The fallback is only for a ServiceAccount that is not present at all.
+    """
+    (tmp_path / "ca.crt").write_text("ca", encoding="utf-8")
+    monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token")
+    (tmp_path / "token").write_text("tok", encoding="utf-8")
+    monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
+    monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
+    seen = _capture(monkeypatch)
+
+    class _ErrClient:
+        def __init__(self, **kw: object) -> None:
+            pass
+
+        async def __aenter__(self) -> _ErrClient:
+            return self
+
+        async def __aexit__(self, *exc: object) -> None:
+            return None
+
+        async def get(self, url: str, **kw: object) -> object:
+            raise httpx.ConnectError("connection reset by peer")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _ErrClient)
+
+    with pytest.raises(HelmError):
+        await HelmProvisioner().list_releases()
+    assert seen == [], "an API error must not fall back to `helm list`"
+
+
+@pytest.mark.asyncio
+async def test_missing_ca_falls_back_to_helm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """A token without a readable CA is a half-mounted SA: fall back rather than crash.
+
+    httpx loads the CA when the client is built, raising OSError that the API except clause
+    does not catch, so the readability check has to happen before the request. A missing CA
+    means "not in-cluster", the same as a missing token.
+    """
+    monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token")
+    (tmp_path / "token").write_text("tok", encoding="utf-8")
+    monkeypatch.setattr(helm, "_SA_CA", tmp_path / "absent-ca.crt")  # never created
+    monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
+    seen = _capture(monkeypatch)
+
+    await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases()
+
+    assert seen and seen[0][:2] == ["helm", "list"]