diff --git a/libs/svcforge_core/svcforge_core/adapters/helm.py b/libs/svcforge_core/svcforge_core/adapters/helm.py
index abef23b..9f1f76d 100644
--- a/libs/svcforge_core/svcforge_core/adapters/helm.py
+++ b/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",
@@ -433,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)
diff --git a/libs/svcforge_core/svcforge_core/adapters/k8s.py b/libs/svcforge_core/svcforge_core/adapters/k8s.py
index fdcd752..75ae139 100644
--- a/libs/svcforge_core/svcforge_core/adapters/k8s.py
+++ b/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)
diff --git a/libs/svcforge_core/svcforge_core/adapters/redis.py b/libs/svcforge_core/svcforge_core/adapters/redis.py
index d2a3d90..ebf7035 100644
--- a/libs/svcforge_core/svcforge_core/adapters/redis.py
+++ b/libs/svcforge_core/svcforge_core/adapters/redis.py
@@ -178,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
@@ -187,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))
 
 
@@ -258,6 +262,7 @@ class RateLimiter:
                 limit=self._limit,
                 remaining=self._limit,
                 reset_at=reset_at,
+                checked_at=self._clock.now(),
                 degraded=True,
             )
         return RateLimitResult(
@@ -265,6 +270,7 @@ class RateLimiter:
             limit=self._limit,
             remaining=int(remaining),
             reset_at=reset_at,
+            checked_at=self._clock.now(),
         )
 
 
diff --git a/libs/svcforge_core/svcforge_core/adapters/tempyaml.py b/libs/svcforge_core/svcforge_core/adapters/tempyaml.py
new file mode 100644
index 0000000..6e38bf8
--- /dev/null
+++ b/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)
diff --git a/libs/svcforge_core/svcforge_core/domain/catalog.py b/libs/svcforge_core/svcforge_core/domain/catalog.py
index cd66cb7..95c3178 100644
--- a/libs/svcforge_core/svcforge_core/domain/catalog.py
+++ b/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.
diff --git a/libs/svcforge_core/svcforge_core/domain/states.py b/libs/svcforge_core/svcforge_core/domain/states.py
index 61dea7b..d0c1120 100644
--- a/libs/svcforge_core/svcforge_core/domain/states.py
+++ b/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."""
 
 
diff --git a/libs/svcforge_core/svcforge_core/domain/windows.py b/libs/svcforge_core/svcforge_core/domain/windows.py
index 46ea52f..21dac8a 100644
--- a/libs/svcforge_core/svcforge_core/domain/windows.py
+++ b/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."""
 
 
diff --git a/libs/svcforge_core/svcforge_core/repo/db.py b/libs/svcforge_core/svcforge_core/repo/db.py
index 63df47e..8215337 100644
--- a/libs/svcforge_core/svcforge_core/repo/db.py
+++ b/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]
diff --git a/libs/svcforge_core/svcforge_core/repo/instances.py b/libs/svcforge_core/svcforge_core/repo/instances.py
index 415aa22..edb55d1 100644
--- a/libs/svcforge_core/svcforge_core/repo/instances.py
+++ b/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,
diff --git a/libs/svcforge_core/svcforge_core/repo/reconcile.py b/libs/svcforge_core/svcforge_core/repo/reconcile.py
index 0665aad..04ed9da 100644
--- a/libs/svcforge_core/svcforge_core/repo/reconcile.py
+++ b/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
diff --git a/libs/svcforge_core/svcforge_core/repo/tasks.py b/libs/svcforge_core/svcforge_core/repo/tasks.py
index 4984034..64ee488 100644
--- a/libs/svcforge_core/svcforge_core/repo/tasks.py
+++ b/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,7 +213,7 @@ 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),
                 )
                 # Dead-lettering the task is correct for every kind. Moving the INSTANCE to
                 # `failed` is correct only for provision: a provisioning instance that never
@@ -235,7 +235,7 @@ class TaskRepo:
                     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".
diff --git a/libs/svcforge_core/svcforge_core/settings.py b/libs/svcforge_core/svcforge_core/settings.py
index 6dfbac8..a55faa4 100644
--- a/libs/svcforge_core/svcforge_core/settings.py
+++ b/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."""
diff --git a/services/_runtime.py b/services/_runtime.py
new file mode 100644
index 0000000..dc703b0
--- /dev/null
+++ b/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)
diff --git a/services/api/main.py b/services/api/main.py
index a56767b..3a3340a 100644
--- a/services/api/main.py
+++ b/services/api/main.py
@@ -55,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)
diff --git a/services/reconciler/main.py b/services/reconciler/main.py
index 4018b5c..b60b28f 100644
--- a/services/reconciler/main.py
+++ b/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__":
diff --git a/services/worker/handlers.py b/services/worker/handlers.py
index fce124b..e8f78d3 100644
--- a/services/worker/handlers.py
+++ b/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:
-        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},
-        )
+        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:
diff --git a/services/worker/main.py b/services/worker/main.py
index 6ea063e..f855ab4 100644
--- a/services/worker/main.py
+++ b/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)
diff --git a/tests/fakes.py b/tests/fakes.py
index 48de2c7..73c340c 100644
--- a/tests/fakes.py
+++ b/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(),
         )
 
 
diff --git a/tests/integration/test_redis.py b/tests/integration/test_redis.py
index 4c28772..7cee970 100644
--- a/tests/integration/test_redis.py
+++ b/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()