review: fix 26 findings from a 4-agent audit

CORRECTNESS
- lost-lease race: complete()/fail() did not check ownership, so a worker whose
  lease expired could mark a task done while another worker was running it, or
  requeue a task someone else owned. Reproduced, fixed with a CAS on
  (state, locked_by), pinned by two regression tests.
- worker died on report failure: _run_one's docstring claimed no exception
  escapes the TaskGroup; fail()/complete() were outside the guarded block, so a
  DB blip cancelled every sibling provision on the pod.
- claim query used an INNER join, which could strand a just-claimed task and
  report 'queue empty'. LEFT join.
- InstanceRepo.set_error bypassed the state machine and had no callers. Deleted.
- handle_deprovision ignored its CAS result, so a wrong-state instance kept a
  dangling endpoint and got re-provisioned by the drift check 60s later.
- handle_verify re-notified on every retry: five pages for one halt.

DEPLOY-BREAKING
- the migration Job could never succeed: no Dockerfile copied migrations/, and
  migrate.py resolved the path relative to the source tree, which only works for
  an editable install. Added COPY + SVCFORGE_MIGRATIONS_DIR.
- ServiceMonitor selector did not match the Service: API metrics never scraped.
- SvcforgeReconcilerStale fired permanently from every pod, because the gauge is
  module-level and every service exports it as 0. Scoped to the reconciler job.
- SvcforgeTaskFailed latched forever on a monotonic counter. Now increase()[15m].
- the digest guard accepted the all-zeros placeholder.
- worker terminationGracePeriodSeconds was 60s against a 600s helm timeout.

DEAD CODE THAT SHOULD NOT HAVE BEEN
- adapters/k8s.py was never called, so tenant namespaces were never created and
  the first provision for a new team would fail. Wired into handle_provision.
- adapters/redis.py was never imported by any service. Rate limiting is now wired
  into the API, failing open.
- Settings.check_production() had no callers. Given an explicit environment and
  called from every entrypoint.

OBSERVABILITY
- the API never called obs.setup(): no JSON logs, no trace correlation, log_json
  silently inert.
- LogNotifier's structured fields were discarded by the stdlib->structlog bridge.
- bind_task_context cleared the 'service' binding for the life of every task.
- split tasks_failed into task_attempts_failed and tasks_dead_lettered.

SECURITY
- trivy correctly blocked the worker/reconciler images: helm 3.16.2 and kubectl
  1.31.2 carry CRITICAL Go stdlib CVEs. Bumped to helm 3.21.3 and kubectl 1.35.3,
  which also closes a four-minor skew against the v1.35.3 cluster.

TESTS THAT COULD NOT FAIL
- the concurrency cap test passed on a fully serial worker.
- the alert/metric cross-check asserted a hardcoded list instead of reading the
  chart, so it could not catch a rename on the chart side.
- fixed OTel tracer-provider pollution between test files.

DOCS
- ARCHITECTURE.md: mermaid diagrams, user stories, and the helm-vs-ArgoCD
  guarantee (verified with --dry-run=server).
- AGENTS.md + CLAUDE.md.
- prose sweep for back-and-forth phrasing across 19 files.

tests/integration/conftest.py

@@ -98,7 +98,7 @@ def pg_dsn() -> Iterator[str]:
 async def pool(pg_dsn: str) -> AsyncIterator[DictPool]:
     """A clean database and an open pool, per test.
 
-    max_size=60 is not a performance choice. test_skip_locked_claims_each_task_exactly_once
+    max_size=60 is a correctness requirement. test_skip_locked_claims_each_task_exactly_once
     races 50 concurrent claims; a pool smaller than that serialises them at the pool
     instead of at the database, and the test passes while proving nothing.
     """

tests/integration/test_api.py

@@ -545,11 +545,25 @@ async def test_202_is_on_the_decorator_not_the_response_object(client: httpx.Asy
     assert "202" in schema["paths"]["/v1/instances/{instance_id}"]["delete"]["responses"]
 
 
-def test_auth_disabled_short_circuits(settings: Settings) -> None:
-    """The dev escape hatch exists, and Settings.check_production() refuses it in prod."""
-    dev = settings.model_copy(update={"auth_disabled": True})
-    with pytest.raises(ValueError, match="refused outside local development"):
-        dev.check_production()
+def test_auth_disabled_is_allowed_locally_and_refused_everywhere_else(settings: Settings) -> None:
+    """The dev escape hatch exists, and check_production() refuses it outside `local`.
+
+    Both halves matter. The permissive half is why every entrypoint can call this
+    unconditionally at startup; the refusing half is the actual safety property. An
+    earlier version raised unconditionally, which meant it could only be called from a
+    branch that already knew it was production — so nobody ever wrote that branch and the
+    check never ran at all.
+    """
+    dev_locally = settings.model_copy(update={"auth_disabled": True, "environment": "local"})
+    dev_locally.check_production()  # must not raise
+
+    for env in ("prod", "staging"):
+        dev_in_prod = settings.model_copy(update={"auth_disabled": True, "environment": env})
+        with pytest.raises(ValueError, match="refused"):
+            dev_in_prod.check_production()
+
+    # And a correctly-configured production app starts fine.
+    settings.model_copy(update={"auth_disabled": False, "environment": "prod"}).check_production()
 
 
 @pytest.mark.slow

tests/integration/test_reconciler.py

@@ -267,8 +267,8 @@ async def test_lease_expiry_returns_a_dead_workers_task_to_the_queue(pool: DictP
 async def test_lease_expiry_leaves_a_live_worker_alone(pool: DictPool) -> None:
     """A task claimed a second ago is not a dead worker. Reclaiming it would double-provision.
 
-    Handlers are idempotent, so a wrongly-freed lease is survivable — but survivable is not
-    free, and this is why lease_seconds sits above helm's own --timeout.
+    Handlers are idempotent, so a wrongly-freed lease is survivable, though it still costs a
+    duplicated helm run — which is why lease_seconds sits above helm's own --timeout.
     """
     instance_id, _, _ = await _seed(pool)
     tasks = TaskRepo(pool)
@@ -419,8 +419,9 @@ async def test_version_drift_parks_the_upgrade_until_the_maintenance_window(
 ) -> None:
     """The queue does the waiting, in `where run_after <= now()`.
 
-    Not a scheduler and not an in-memory timer: a task parked in Postgres until 03:00 Sunday
-    survives a reconciler restart. That is the whole reason `run_after` exists.
+    The waiting is done by the queue rather than by a scheduler or an in-memory timer: a
+    task parked in Postgres until 03:00 Sunday survives a reconciler restart. That is the
+    whole reason `run_after` exists.
     """
     instance_id, _, _ = await _seed(pool, maintenance_window=SUNDAY_0300_HCM)
 
@@ -520,8 +521,19 @@ def tracing() -> InMemorySpanExporter:
 
     Global because OTEL's is: `trace.set_tracer_provider` takes once per process, and
     `obs.tracer()` resolves it at call time. Session-scoped so the second call never
-    happens.
+    happens from HERE.
+
+    The internals reset is load-bearing rather than cosmetic. `set_tracer_provider` is
+    one-shot: a second call logs "Overriding of current TracerProvider is not allowed" at
+    WARNING and is otherwise ignored. Any earlier test that builds a FastAPI app calls
+    `obs.setup()` and burns that one shot, after which this fixture silently installs
+    nothing, `get_finished_spans()` returns `[]`, and the failure reads as "the reconciler
+    stopped writing traceparents" rather than "another test got there first". Clearing the
+    module globals is the only way to take the shot back.
     """
+    trace._TRACER_PROVIDER = None
+    trace._TRACER_PROVIDER_SET_ONCE._done = False
+
     exporter = InMemorySpanExporter()
     provider = TracerProvider()
     provider.add_span_processor(SimpleSpanProcessor(exporter))

tests/integration/test_redis.py

@@ -206,7 +206,7 @@ async def test_real_lua_refuses_the_eleventh(upstash: Redis) -> None:
 async def test_a_check_costs_exactly_one_redis_command(upstash: Redis) -> None:
     """One EVALSHA. Not GET+INCR+EXPIRE, which is three billed commands and a race.
 
-    At 500K/month the difference is not academic: three commands per request caps the
+    At 500K/month the difference is material: three commands per request caps the
     platform at 166K requests/month instead of 500K, for a limiter that is also wrong.
 
     The first call is excluded from the count on purpose. redis-py sends EVALSHA, Upstash

tests/integration/test_tasks.py

@@ -35,7 +35,7 @@ async def test_complete_marks_done_and_releases_lock(pool: DictPool) -> None:
     tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
     claimed = await repo.claim("w1")
     assert claimed is not None
-    await repo.complete(claimed.id)
+    await repo.complete(claimed.id, "w1")
     row = await _task_row(pool, tid)
     assert row["state"] == "done"
     assert row["locked_by"] is None
@@ -49,7 +49,7 @@ async def test_fail_under_max_attempts_requeues_with_future_run_after(pool: Dict
     assert claimed is not None
     assert claimed.attempts == 1  # attempts increments at CLAIM time, not on failure
 
-    await repo.fail(tid, "helm exploded", max_attempts=5)
+    await repo.fail(tid, "helm exploded", "w1", max_attempts=5)
     row = await _task_row(pool, tid)
     assert row["state"] == "queued"
     assert row["last_error"] == "helm exploded"
@@ -66,10 +66,12 @@ async def test_fail_at_max_attempts_dead_letters_and_marks_instance(pool: DictPo
     tasks, instances = TaskRepo(pool), InstanceRepo(pool)
     tid = await tasks.enqueue_standalone(iid, TaskKind.PROVISION)
 
+    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,))
 
-    await tasks.fail(tid, "chart not found", max_attempts=5)
+    assert await tasks.fail(tid, "chart not found", "w1", max_attempts=5) is True
 
     row = await _task_row(pool, tid)
     assert row["state"] == "failed"
@@ -90,10 +92,12 @@ async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> Non
     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,))
 
-    await tasks.fail(tid, "helm uninstall kept failing", max_attempts=5)
+    assert await tasks.fail(tid, "helm uninstall kept failing", "w1", max_attempts=5) is True
 
     # The task still dead-letters — that part is unconditional.
     row = await _task_row(pool, tid)
@@ -111,7 +115,7 @@ async def test_fail_truncates_error_to_2kb(pool: DictPool) -> None:
     repo = TaskRepo(pool)
     tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
     await repo.claim("w1")
-    await repo.fail(tid, "x" * 9000, max_attempts=5)
+    await repo.fail(tid, "x" * 9000, "w1", max_attempts=5)
     row = await _task_row(pool, tid)
     assert isinstance(row["last_error"], str)
     assert len(row["last_error"]) == 2000
@@ -153,3 +157,56 @@ async def test_fresh_lease_is_not_reset(pool: DictPool) -> None:
     await repo.enqueue_standalone(iid, TaskKind.PROVISION)
     await repo.claim("w1")
     assert await repo.reset_expired_leases(lease_seconds=300) == 0
+
+
+async def test_stale_worker_cannot_complete_a_task_another_worker_now_owns(pool: DictPool) -> None:
+    """The lost-lease race, as a regression test.
+
+    Worker A hangs past its lease. The reconciler requeues the task. Worker B claims it and
+    starts working. Worker A finally returns and reports success. Without the ownership
+    check in `complete`, A marks the task done while B is still running it: B's helm
+    install is unaccounted for, and if B then fails, the task is resurrected and a THIRD
+    worker provisions the same instance. That is the double-provision the claim query's
+    whole design exists to prevent, arriving through the back door.
+    """
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+
+    a = await repo.claim("worker-A")
+    assert a is not None
+
+    # A's lease expires and the reconciler hands the task back to the queue.
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update tasks set locked_at = now() - interval '10 minutes' where id=%s", (tid,))
+    assert await repo.reset_expired_leases(lease_seconds=300) == 1
+
+    b = await repo.claim("worker-B")
+    assert b is not None, "worker-B should have been able to claim the requeued task"
+
+    # A reports. It must lose.
+    assert await repo.complete(tid, "worker-A") is False, "stale worker completed a task it no longer owns"
+
+    row = await _task_row(pool, tid)
+    assert row["state"] == "running", "the task must still belong to worker-B"
+    assert row["locked_by"] == "worker-B"
+
+
+async def test_stale_worker_cannot_fail_a_task_another_worker_now_owns(pool: DictPool) -> None:
+    """The mirror case: a stale failure report must not requeue someone else's task."""
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+
+    assert await repo.claim("worker-A") is not None
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update tasks set locked_at = now() - interval '10 minutes' where id=%s", (tid,))
+    await repo.reset_expired_leases(lease_seconds=300)
+    assert await repo.claim("worker-B") is not None
+
+    assert await repo.fail(tid, "stale report", "worker-A", max_attempts=5) is False
+
+    row = await _task_row(pool, tid)
+    assert row["state"] == "running"
+    assert row["locked_by"] == "worker-B"
+    assert row["last_error"] is None, "a stale worker wrote its error onto another worker's task"

tests/integration/test_worker.py

@@ -41,10 +41,21 @@ def _settings(**over: object) -> Settings:
     return Settings(**base)  # type: ignore[arg-type]
 
 
+class FakeNamespaceEnsurer:
+    """Records the namespaces it was asked to create. Satisfies NamespaceEnsurer."""
+
+    def __init__(self) -> None:
+        self.ensured: list[str] = []
+
+    async def ensure_namespace(self, ns: str, labels: dict[str, str] | None = None) -> None:
+        self.ensured.append(ns)
+
+
 def _deps(
     pool: DictPool,
     prov: FakeProvisioner,
     notifier: FakeNotifier | None = None,
+    namespaces: FakeNamespaceEnsurer | None = None,
     **over: object,
 ) -> WorkerDeps:
     return WorkerDeps(
@@ -52,6 +63,7 @@ def _deps(
         instances=InstanceRepo(pool),
         tasks=TaskRepo(pool),
         provisioner=prov,
+        namespaces=namespaces or FakeNamespaceEnsurer(),
         notifier=notifier or FakeNotifier(),
         clock=SystemClock(),
         catalog=CATALOG,
@@ -200,4 +212,12 @@ async def test_concurrency_cap_is_respected(pool: DictPool, concurrency: int) ->
     stop.set()
     await asyncio.wait_for(worker, timeout=10)
 
+    # BOTH bounds. `<= concurrency` alone passes on a worker whose semaphore is broken to 1,
+    # or that lost concurrency entirely — it only proves the cap is not exceeded, never that
+    # concurrency exists. With 6 tasks seeded and a 0.2s handler, a working worker reaches
+    # its cap exactly.
     assert prov.max_concurrent <= concurrency, f"ran {prov.max_concurrent} at once, cap was {concurrency}"
+    assert prov.max_concurrent == concurrency, (
+        f"only reached {prov.max_concurrent} concurrent with a cap of {concurrency}: "
+        "the worker is not actually running tasks in parallel"
+    )