refactor: converge the patterns multiple authors left divergent

The codebase was written by several agents and had the same concept done more
than one way. This makes it read as one voice, with no behaviour change.

Dedup, each to a single canonical form:
  - INSTANCE_COLUMNS: the 13-column instances SELECT list existed as _COLUMNS in
    instances.py and reconcile.py (byte-identical) and inlined a third time in
    the worker. One exported constant now.
  - Settings.runtime_dsn: the three entrypoints each chose between str(pg_dsn)
    and pg_dsn.unicode_string(). One property.
  - yaml_tempfile: helm._ValuesFile and k8s._ManifestFile were the same
    write-yaml-to-a-temp-dir context manager. One helper in adapters/tempyaml.py.
  - services/_runtime.py: sleep_or_stop and install_stop_signals were copied
    between the worker and reconciler loops. One module, so shutdown behaviour
    cannot drift between them.
  - k8s.ensure_namespace used MANAGED_BY_LABEL/VALUE from helm.py instead of a
    hardcoded literal, so the managed-by label has one definition.
  - SvcforgeError is now the root of every svcforge exception (CatalogError,
    IllegalTransition, BadWindow, HandlerError), keeping each stdlib base in the
    MRO, so `except SvcforgeError` means what errors.py says it does.
  - ERROR_MAX_CHARS replaces the repeated `[-2000:]` truncation feeding the same
    error columns.
  - the reconciler reads settings.metrics_port like the worker, dropping its
    duplicate DEFAULT_METRICS_PORT and redundant --metrics-port option; the
    SVCFORGE_METRICS_PORT env override still applies through pydantic.

Two smaller correctness/consistency fixes:
  - RateLimitResult.retry_after_s computed its delta against datetime.now(UTC)
    while the limiter runs on an injectable clock, so it was meaningless under a
    FakeClock and drifted by request latency in production. It now carries a
    checked_at from the same clock as reset_at.
  - handle_provision's notifier.send is wrapped like the reconciler's: a flaky
    webhook after the READY CAS would fail the task, and the retry would hit the
    READY early-return and drop the notification, turning a good provision into a
    failed one.

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)

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

@@ -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(),
         )
 
 

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,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".

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

@@ -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)

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_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()