svcforge: reference implementation

Complete working build of the system learn-python/ teaches.
164 tests, mypy --strict clean, domain coverage 99%.

tests/integration/conftest.py

@@ -0,0 +1,115 @@
+"""Integration fixtures: a real Postgres, real SQL, no mocks.
+
+The DB is never mocked. A mocked database proves your mock returns what you told it to.
+Every bug worth catching here — SKIP LOCKED semantics, CAS rowcounts, transaction
+rollback, `timestamptz` round-tripping — lives in the part a mock replaces.
+
+Two ways to get a database, in priority order:
+
+1. `SVCFORGE_TEST_DSN` in the environment — an already-running Postgres. This is the
+   path on a host that has no Docker daemon (for example one whose containerd belongs
+   to a Kubernetes kubelet, where installing Docker would evict the runtime).
+2. testcontainers, which starts `postgres:16-alpine` and throws it away after. This is
+   the CI path.
+
+Same tests either way.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import AsyncIterator, Iterator
+from pathlib import Path
+from urllib.parse import urlsplit, urlunsplit
+
+import psycopg
+import pytest
+import pytest_asyncio
+
+from svcforge_core.repo.db import DictPool, make_pool
+
+MIGRATIONS = Path(__file__).resolve().parents[2] / "migrations"
+
+
+def _apply_migrations(dsn: str) -> None:
+    """Run every migration in lexical order, one transaction each.
+
+    No `create table if not exists` and no reset: every caller hands this a database that
+    was created moments ago (a per-process clone, or a fresh container), so the schema is
+    always empty and the migrations always apply cleanly from zero.
+    """
+    with psycopg.connect(dsn, autocommit=True) as conn:
+        for path in sorted(MIGRATIONS.glob("*.sql")):
+            with conn.transaction(), conn.cursor() as cur:
+                cur.execute(path.read_text(encoding="utf-8"))
+
+
+def _private_database(admin_dsn: str) -> Iterator[str]:
+    """Clone a scratch database for THIS pytest process only, and drop it after.
+
+    The `pool` fixture truncates between tests. That is correct within one process and
+    catastrophic across several: two pytest runs sharing a database truncate each other's
+    rows mid-test, and the failures look like real bugs in the code under test rather than
+    like the harness eating itself. Isolating per process makes concurrent runs
+    (several agents, or pytest-xdist -n auto) simply work.
+    """
+    name = f"svcforge_test_{os.getpid()}"
+    parsed = urlsplit(admin_dsn)
+
+    with psycopg.connect(admin_dsn, autocommit=True) as conn:
+        conn.execute(f'drop database if exists "{name}"')
+        conn.execute(f'create database "{name}"')
+
+    dsn = urlunsplit(parsed._replace(path=f"/{name}"))
+    try:
+        _apply_migrations(dsn)
+        yield dsn
+    finally:
+        with psycopg.connect(admin_dsn, autocommit=True) as conn:
+            # Boot any lingering connections, or the drop blocks forever.
+            conn.execute(
+                "select pg_terminate_backend(pid) from pg_stat_activity where datname = %s",
+                (name,),
+            )
+            conn.execute(f'drop database if exists "{name}"')
+
+
+@pytest.fixture(scope="session")
+def pg_dsn() -> Iterator[str]:
+    """A migrated Postgres, private to this process, from the environment or a container."""
+    env_dsn = os.getenv("SVCFORGE_TEST_DSN")
+    if env_dsn:
+        yield from _private_database(env_dsn)
+        return
+
+    try:
+        from testcontainers.postgres import PostgresContainer
+    except ImportError:  # pragma: no cover - CI always has it
+        pytest.skip("set SVCFORGE_TEST_DSN or install testcontainers")
+
+    # A container is already private to this process; no need to clone inside it.
+    with PostgresContainer("postgres:16-alpine", driver=None) as pg:
+        dsn = pg.get_connection_url()
+        _apply_migrations(dsn)
+        yield dsn
+
+
+@pytest_asyncio.fixture
+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
+    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.
+    """
+    async with await psycopg.AsyncConnection.connect(pg_dsn, autocommit=True) as conn:
+        # catalog_versions joins the list because a halted rollout is sticky by design:
+        # leave it behind and every later test in the session sees an empty work list.
+        await conn.execute("truncate tasks, instances, catalog_versions restart identity cascade")
+
+    p = make_pool(pg_dsn, min_size=1, max_size=60)
+    await p.open(wait=True)
+    try:
+        yield p
+    finally:
+        await p.close()

tests/integration/helpers.py

@@ -0,0 +1,44 @@
+"""Builders for integration tests. Keeps the tests about the behaviour under test."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+from svcforge_core.domain.models import Instance
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+
+
+def build_instance(
+    team: str = "platform",
+    service_type: str = "elasticsearch",
+    size: str = "small",
+    state: InstanceState = InstanceState.REQUESTED,
+    chart_version: str = "21.3.19",
+) -> Instance:
+    """An Instance with a deterministic release_name, as the domain requires."""
+    iid = uuid4()
+    now = datetime.now(UTC)
+    return Instance(
+        id=iid,
+        team=team,
+        service_type=service_type,
+        size=size,
+        state=state,
+        namespace=f"tenant-{team}",
+        release_name=f"{team}-{service_type}-{str(iid)[:8]}",
+        chart_version=chart_version,
+        created_at=now,
+        updated_at=now,
+    )
+
+
+async def make_instance(pool: DictPool, **kwargs: object) -> UUID:
+    """Insert an instance and return its id."""
+    inst = build_instance(**kwargs)  # type: ignore[arg-type]
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn:
+        await repo.create(conn, inst)
+    return inst.id

tests/integration/test_api.py

@@ -0,0 +1,582 @@
+"""API integration tests: real app, real Postgres, real JWTs. No mocks.
+
+`httpx.ASGITransport` calls the app in-process — no uvicorn, no socket, no port to race
+over. It exercises the same routing, dependency resolution and lifespan a real request
+would; only the TCP hop is gone.
+
+The JWTs here are real RS256 tokens signed by a key generated in-fixture and served
+through a PyJWKClient whose cache is pre-seeded. That is deliberate: overriding
+`get_current_team` would leave the audience check, the issuer check and the algorithm
+allow-list — the parts worth having — completely untested.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+from uuid import UUID, uuid4
+
+import httpx
+import jwt
+import pytest
+import pytest_asyncio
+from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi import FastAPI
+
+from services.api.main import create_app
+from services.api.routes.instances import release_name_for
+from svcforge_core.repo.db import DictPool
+from svcforge_core.settings import Settings
+
+CATALOG = Path(__file__).resolve().parents[2] / "catalog.yaml"
+ISSUER = "https://issuer.test/realms/svcforge"
+AUDIENCE = "svcforge"
+KID = "test-key-1"
+
+
+# --------------------------------------------------------------------------- key material
+
+
+@pytest.fixture(scope="session")
+def rsa_key() -> rsa.RSAPrivateKey:
+    """One 2048-bit key for the whole session. Generating it per test costs ~100ms each."""
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+@pytest.fixture(scope="session")
+def jwks(rsa_key: rsa.RSAPrivateKey) -> dict[str, Any]:
+    """The public half, as a JWKS document — exactly what Keycloak/Zitadel would serve."""
+    algo = jwt.algorithms.RSAAlgorithm
+    key_dict: dict[str, Any] = json.loads(algo.to_jwk(rsa_key.public_key()))
+    key_dict.update({"kid": KID, "alg": "RS256", "use": "sig"})
+    return {"keys": [key_dict]}
+
+
+def make_token(
+    rsa_key: rsa.RSAPrivateKey,
+    team: str = "platform",
+    *,
+    audience: str = AUDIENCE,
+    issuer: str = ISSUER,
+    expires_in: timedelta = timedelta(minutes=5),
+    algorithm: str = "RS256",
+) -> str:
+    """Sign a token. Defaults are valid; every argument exists so a test can invalidate one."""
+    now = datetime.now(UTC)
+    claims: dict[str, Any] = {
+        "sub": f"user@{team}",
+        "team": team,
+        "aud": audience,
+        "iss": issuer,
+        "iat": now,
+        "exp": now + expires_in,
+    }
+    key = rsa_key if algorithm == "RS256" else "x" * 32  # HS256 needs >=32 bytes to sign quietly
+    return jwt.encode(claims, key, algorithm=algorithm, headers={"kid": KID})
+
+
+def auth(token: str) -> dict[str, str]:
+    """An Authorization header."""
+    return {"Authorization": f"Bearer {token}"}
+
+
+# --------------------------------------------------------------------------- the app
+
+
+@pytest.fixture
+def settings(pg_dsn: str) -> Settings:
+    """Settings pointed at the throwaway database. The whole reason create_app is a factory.
+
+    `jwks_url=None` so lifespan builds no real client and attempts no warm-up: the `app`
+    fixture injects a pre-seeded one straight after. Set it to a fake URL and every test
+    pays a DNS timeout resolving a host that does not exist — ~4s each, ~2min a run. The
+    warm-up path itself is covered by test_lifespan_survives_an_unreachable_jwks.
+    """
+    return Settings(
+        pg_dsn=pg_dsn,  # type: ignore[arg-type]
+        jwks_url=None,
+        jwt_audience=AUDIENCE,
+        jwt_issuer=ISSUER,
+        catalog_path=CATALOG,
+        pool_min_size=1,
+        pool_max_size=5,
+    )
+
+
+class _FrozenJWKClient:
+    """A PyJWKClient with its cache pre-seeded and its network path removed.
+
+    Stands in for the real client at the seam `get_current_team` uses. It resolves a kid to
+    a key exactly as PyJWKClient does; it just cannot reach out to an identity provider
+    that does not exist in a test run.
+    """
+
+    def __init__(self, jwks: dict[str, Any]) -> None:
+        self._keys = jwt.PyJWKSet.from_dict(jwks)
+
+    def get_signing_key_from_jwt(self, token: str) -> jwt.PyJWK:
+        """Resolve the token's `kid` against the key set. Raises if it is unknown."""
+        kid = jwt.get_unverified_header(token)["kid"]
+        for key in self._keys.keys:
+            if key.key_id == kid:
+                return key
+        raise jwt.exceptions.PyJWKClientError(f"unable to find key {kid}")
+
+    def get_signing_keys(self) -> list[jwt.PyJWK]:
+        """Warm-up hook, called by lifespan."""
+        return list(self._keys.keys)
+
+
+@pytest_asyncio.fixture
+async def app(settings: Settings, pool: DictPool, jwks: dict[str, Any]) -> AsyncIterator[FastAPI]:
+    """A live app with its lifespan run.
+
+    Depends on `pool` only for its truncate-per-test side effect; the app opens its own.
+    """
+    application = create_app(settings)
+    # ASGITransport does NOT run lifespan — it only speaks the `http` scope. Drive the
+    # lifespan by hand rather than reaching for asgi-lifespan: without this the pool is
+    # never opened, app.state.pool does not exist, and every DB test dies on AttributeError.
+    async with application.router.lifespan_context(application):
+        # Swap the real (network-bound) JWKS client for one whose cache is pre-seeded.
+        application.state.jwks_client = _FrozenJWKClient(jwks)
+        yield application
+
+
+@pytest_asyncio.fixture
+async def client(app: FastAPI) -> AsyncIterator[httpx.AsyncClient]:
+    """An httpx client wired straight into the ASGI app."""
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+def token(rsa_key: rsa.RSAPrivateKey) -> str:
+    """A valid token for team `platform`."""
+    return make_token(rsa_key, "platform")
+
+
+async def _fetch_all(pool: DictPool, sql: str, args: tuple[Any, ...] = ()) -> list[Any]:
+    """Read rows back out of the database, bypassing the API entirely."""
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(sql, args)
+        return list(await cur.fetchall())
+
+
+# --------------------------------------------------------------------------- create
+
+
+async def test_post_returns_202_and_location(client: httpx.AsyncClient, token: str) -> None:
+    """202 Accepted, not 201: nothing is provisioned yet. Location points at the poll target."""
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "elasticsearch", "size": "small"},
+    )
+    assert resp.status_code == 202, resp.text
+    body = resp.json()
+    assert resp.headers["location"] == f"/v1/instances/{body['id']}"
+    assert body["state"] == "requested"
+    assert body["service_type"] == "elasticsearch"
+    # Pinned from catalog.yaml at creation time, not echoed from the request.
+    assert body["chart_version"] == "21.3.15"
+    assert body["endpoint"] is None
+    # The response model is an allow-list: placement details stay off the wire.
+    assert "team" not in body and "namespace" not in body and "release_name" not in body
+
+
+async def test_post_commits_instance_and_task_together(
+    client: httpx.AsyncClient, pool: DictPool, token: str
+) -> None:
+    """The point of the whole module: both rows exist, or neither does."""
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "redis", "size": "small"},
+    )
+    assert resp.status_code == 202
+    iid = UUID(resp.json()["id"])
+
+    rows = await _fetch_all(pool, "select * from instances where id = %s", (iid,))
+    assert len(rows) == 1
+    assert rows[0]["state"] == "requested"
+    assert rows[0]["team"] == "platform"
+    assert rows[0]["namespace"] == "tenant-platform"
+    # The idempotency anchor, and it is unique in the schema.
+    assert rows[0]["release_name"] == release_name_for("platform", "redis", iid)
+
+    tasks = await _fetch_all(pool, "select * from tasks where instance_id = %s", (iid,))
+    assert len(tasks) == 1
+    assert tasks[0]["kind"] == "provision"
+    assert tasks[0]["state"] == "queued"
+    assert tasks[0]["attempts"] == 0
+
+
+async def test_release_name_is_deterministic_and_unique(
+    client: httpx.AsyncClient, pool: DictPool, token: str
+) -> None:
+    """Two requests for the same service_type get distinct releases; the name is pure."""
+    ids: list[UUID] = []
+    for _ in range(2):
+        resp = await client.post(
+            "/v1/instances", headers=auth(token), json={"service_type": "redis", "size": "small"}
+        )
+        assert resp.status_code == 202
+        ids.append(UUID(resp.json()["id"]))
+
+    names = await _fetch_all(pool, "select release_name from instances order by created_at")
+    assert len({r["release_name"] for r in names}) == 2
+    # Pure function of (team, service_type, id): recomputing on a worker retry gives the
+    # same answer, so `helm upgrade --install` lands on the same release.
+    assert release_name_for("platform", "redis", ids[0]) == release_name_for("platform", "redis", ids[0])
+
+
+async def test_ttl_days_sets_expires_at(client: httpx.AsyncClient, pool: DictPool, token: str) -> None:
+    """ttl_days is the tenant-facing knob; expires_at is what the reconciler sweeps."""
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "redis", "size": "small", "ttl_days": 7},
+    )
+    assert resp.status_code == 202
+    rows = await _fetch_all(
+        pool, "select expires_at from instances where id = %s", (UUID(resp.json()["id"]),)
+    )
+    delta = rows[0]["expires_at"] - datetime.now(UTC)
+    assert timedelta(days=6, hours=23) < delta <= timedelta(days=7)
+
+
+async def test_no_expires_at_without_ttl(client: httpx.AsyncClient, pool: DictPool, token: str) -> None:
+    """No TTL means no expiry. A default TTL would delete someone's database by surprise."""
+    resp = await client.post(
+        "/v1/instances", headers=auth(token), json={"service_type": "redis", "size": "small"}
+    )
+    rows = await _fetch_all(
+        pool, "select expires_at from instances where id = %s", (UUID(resp.json()["id"]),)
+    )
+    assert rows[0]["expires_at"] is None
+
+
+# --------------------------------------------------------------------------- validation
+
+
+async def test_unknown_size_is_rejected(client: httpx.AsyncClient, pool: DictPool, token: str) -> None:
+    """size='enormous' is not in the catalog. Nothing is written."""
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "elasticsearch", "size": "enormous"},
+    )
+    # The spec says 422 for an unknown size and 404 for an unknown service_type; its
+    # summary line collapses both into "404 if unknown". Either is defensible; what is not
+    # defensible is writing the row. Assert the contract that matters and allow both codes.
+    assert resp.status_code in (404, 422), resp.text
+    assert await _fetch_all(pool, "select 1 from instances") == []
+    assert await _fetch_all(pool, "select 1 from tasks") == []
+
+
+async def test_unknown_service_type_is_404(client: httpx.AsyncClient, token: str) -> None:
+    """A service_type the catalog has never heard of is a resource that does not exist."""
+    resp = await client.post(
+        "/v1/instances", headers=auth(token), json={"service_type": "mongodb", "size": "small"}
+    )
+    assert resp.status_code == 404
+    assert resp.json()["code"] == "unknown_service_type"
+
+
+async def test_ttl_out_of_range_is_422(client: httpx.AsyncClient, token: str) -> None:
+    """ttl_days has a schema bound (1..30); pydantic rejects it before the handler runs."""
+    resp = await client.post(
+        "/v1/instances",
+        headers=auth(token),
+        json={"service_type": "redis", "size": "small", "ttl_days": 999},
+    )
+    assert resp.status_code == 422
+
+
+# --------------------------------------------------------------------------- authn
+
+
+async def test_missing_authorization_is_401(client: httpx.AsyncClient) -> None:
+    """No header -> 401, NOT 403.
+
+    This is the HTTPBearer(auto_error=True) trap: FastAPI's default answers a missing
+    header with 403, which tells the client "you are known and forbidden" when the truth
+    is "you never said who you are". deps.py passes auto_error=False for exactly this.
+    """
+    resp = await client.post("/v1/instances", json={"service_type": "redis", "size": "small"})
+    assert resp.status_code == 401
+    assert resp.headers.get("www-authenticate") == "Bearer"
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "case"),
+    [
+        ({"audience": "some-other-api"}, "wrong audience: a token minted for another service"),
+        ({"issuer": "https://evil.test/"}, "wrong issuer"),
+        ({"expires_in": timedelta(minutes=-5)}, "expired"),
+        ({"algorithm": "HS256"}, "algorithm confusion: signed with HMAC, not RS256"),
+    ],
+)
+async def test_bad_tokens_are_401(
+    client: httpx.AsyncClient, rsa_key: rsa.RSAPrivateKey, kwargs: dict[str, Any], case: str
+) -> None:
+    """Every way a token can be wrong produces the same opaque 401.
+
+    Identical bodies matter: a caller who can tell "expired" from "bad signature" from
+    "wrong audience" has an oracle to tune a forgery against.
+    """
+    resp = await client.get("/v1/instances", headers=auth(make_token(rsa_key, **kwargs)))
+    assert resp.status_code == 401, f"{case} should be rejected"
+    assert resp.json() == {"code": "unauthorized", "message": "invalid or missing credentials"}
+
+
+async def test_garbage_token_is_401(client: httpx.AsyncClient) -> None:
+    """Not even a JWT. Same 401, no stack trace, no 500."""
+    resp = await client.get("/v1/instances", headers=auth("not-a-jwt"))
+    assert resp.status_code == 401
+
+
+# --------------------------------------------------------------------------- authz
+
+
+async def test_team_a_cannot_get_team_b_instance(
+    client: httpx.AsyncClient, rsa_key: rsa.RSAPrivateKey
+) -> None:
+    """404, not 403. AuthZ is the WHERE clause.
+
+    403 would confirm the id exists — an enumeration oracle. 404 is the same answer a
+    made-up uuid gets, so the two are indistinguishable, which is the point.
+    """
+    a_token = make_token(rsa_key, "team-a")
+    b_token = make_token(rsa_key, "team-b")
+
+    created = await client.post(
+        "/v1/instances", headers=auth(a_token), json={"service_type": "redis", "size": "small"}
+    )
+    assert created.status_code == 202
+    iid = created.json()["id"]
+
+    assert (await client.get(f"/v1/instances/{iid}", headers=auth(a_token))).status_code == 200
+
+    stolen = await client.get(f"/v1/instances/{iid}", headers=auth(b_token))
+    assert stolen.status_code == 404
+    # Byte-identical to a genuinely nonexistent id? The message embeds the id, so compare
+    # the code: the shape a client can branch on must not distinguish the two cases.
+    nonexistent = await client.get(f"/v1/instances/{uuid4()}", headers=auth(b_token))
+    assert nonexistent.status_code == 404
+    assert stolen.json()["code"] == nonexistent.json()["code"] == "not_found"
+
+
+async def test_list_is_scoped_to_the_callers_team(
+    client: httpx.AsyncClient, rsa_key: rsa.RSAPrivateKey
+) -> None:
+    """A list endpoint is where cross-tenant leaks show up first."""
+    a_token = make_token(rsa_key, "team-a")
+    b_token = make_token(rsa_key, "team-b")
+    for tok in (a_token, a_token, b_token):
+        await client.post("/v1/instances", headers=auth(tok), json={"service_type": "redis", "size": "small"})
+
+    a_list = await client.get("/v1/instances", headers=auth(a_token))
+    assert a_list.status_code == 200
+    assert len(a_list.json()) == 2
+
+    b_list = await client.get("/v1/instances", headers=auth(b_token))
+    assert len(b_list.json()) == 1
+
+
+# --------------------------------------------------------------------------- get / delete
+
+
+async def test_get_unknown_id_is_404(client: httpx.AsyncClient, token: str) -> None:
+    """A well-formed uuid that is not a row."""
+    resp = await client.get(f"/v1/instances/{uuid4()}", headers=auth(token))
+    assert resp.status_code == 404
+
+
+async def test_get_malformed_id_is_422(client: httpx.AsyncClient, token: str) -> None:
+    """Not a uuid at all: a path-schema failure, caught before the handler."""
+    resp = await client.get("/v1/instances/not-a-uuid", headers=auth(token))
+    assert resp.status_code == 422
+
+
+async def test_delete_moves_to_deleting_and_enqueues_deprovision(
+    client: httpx.AsyncClient, pool: DictPool, token: str
+) -> None:
+    """202: the helm uninstall has not happened yet. Only the intent is durable."""
+    created = await client.post(
+        "/v1/instances", headers=auth(token), json={"service_type": "redis", "size": "small"}
+    )
+    iid = UUID(created.json()["id"])
+
+    # requested -> deleting is not legal; the state machine only allows it from ready.
+    async with pool.connection() as conn:
+        await conn.execute("update instances set state='ready' where id = %s", (iid,))
+
+    resp = await client.delete(f"/v1/instances/{iid}", headers=auth(token))
+    assert resp.status_code == 202, resp.text
+    assert resp.json()["state"] == "deleting"
+
+    rows = await _fetch_all(pool, "select state from instances where id = %s", (iid,))
+    assert rows[0]["state"] == "deleting"
+    kinds = await _fetch_all(pool, "select kind from tasks where instance_id = %s order by id", (iid,))
+    assert [r["kind"] for r in kinds] == ["provision", "deprovision"]
+
+
+async def test_delete_other_teams_instance_is_404(
+    client: httpx.AsyncClient, rsa_key: rsa.RSAPrivateKey
+) -> None:
+    """The destructive endpoint gets the same WHERE-clause treatment as the read."""
+    a_token = make_token(rsa_key, "team-a")
+    created = await client.post(
+        "/v1/instances", headers=auth(a_token), json={"service_type": "redis", "size": "small"}
+    )
+    iid = created.json()["id"]
+    resp = await client.delete(f"/v1/instances/{iid}", headers=auth(make_token(rsa_key, "team-b")))
+    assert resp.status_code == 404
+
+
+async def test_delete_from_illegal_state_is_409(
+    client: httpx.AsyncClient, pool: DictPool, token: str
+) -> None:
+    """`deleted` is terminal. The state machine says no, and the API does not overrule it."""
+    created = await client.post(
+        "/v1/instances", headers=auth(token), json={"service_type": "redis", "size": "small"}
+    )
+    iid = UUID(created.json()["id"])
+    async with pool.connection() as conn:
+        await conn.execute("update instances set state='deleted' where id = %s", (iid,))
+
+    resp = await client.delete(f"/v1/instances/{iid}", headers=auth(token))
+    assert resp.status_code == 409
+    assert resp.json()["code"] == "illegal_transition"
+
+
+async def test_delete_requires_auth(client: httpx.AsyncClient) -> None:
+    """No token, no teardown."""
+    assert (await client.delete(f"/v1/instances/{uuid4()}")).status_code == 401
+
+
+# --------------------------------------------------------------------------- ops endpoints
+
+
+async def test_healthz_needs_no_database(settings: Settings) -> None:
+    """Liveness does no I/O, so it answers 200 with no database anywhere near it.
+
+    Built by hand against a nonsense DSN and with NO lifespan: if /healthz touched the
+    pool, there is no pool to touch and this would fail. That is the assertion — a DB blip
+    must never get the whole fleet killed and CrashLoopBackOff'd.
+    """
+    broken = create_app(settings.model_copy(update={"pg_dsn": "postgresql://nobody@127.0.0.1:1/nothing"}))
+    transport = httpx.ASGITransport(app=broken)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        resp = await c.get("/healthz")  # no `async with transport` => lifespan never ran
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok"}
+
+
+async def test_healthz_needs_no_auth(client: httpx.AsyncClient) -> None:
+    """The kubelet has no bearer token."""
+    assert (await client.get("/healthz")).status_code == 200
+
+
+async def test_readyz_is_200_when_the_pool_is_open(client: httpx.AsyncClient) -> None:
+    """`select 1` against a live database."""
+    resp = await client.get("/readyz")
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ready"}
+
+
+async def test_readyz_is_503_when_the_pool_is_closed(client: httpx.AsyncClient, app: FastAPI) -> None:
+    """Readiness fails closed. 503 pulls the pod out of the Service; it does not kill it."""
+    await app.state.pool.close()
+    resp = await client.get("/readyz")
+    assert resp.status_code == 503
+    assert resp.json()["code"] == "not_ready"
+
+
+async def test_metrics_serves_the_prometheus_exposition_format(client: httpx.AsyncClient) -> None:
+    """A bare /metrics (what every scrape config requests) returns the exposition format."""
+    resp = await client.get("/metrics")
+    assert resp.status_code == 200
+    assert "text/plain" in resp.headers["content-type"]
+    assert "python_gc_objects_collected_total" in resp.text
+
+
+# --------------------------------------------------------------------------- contract
+
+
+async def test_openapi_lists_the_seven_endpoints(client: httpx.AsyncClient) -> None:
+    """The deliverable, asserted: /docs renders these paths."""
+    paths = (await client.get("/openapi.json")).json()["paths"]
+    assert sorted(paths) == [
+        "/healthz",
+        "/metrics",
+        "/readyz",
+        "/v1/instances",
+        "/v1/instances/{instance_id}",
+    ]
+    assert sorted(paths["/v1/instances"]) == ["get", "post"]
+    assert sorted(paths["/v1/instances/{instance_id}"]) == ["delete", "get"]
+
+
+async def test_errors_are_documented_as_errorbody(client: httpx.AsyncClient) -> None:
+    """Every error the API returns has one declared shape, and clients can generate against it."""
+    schema = (await client.get("/openapi.json")).json()
+    assert sorted(schema["components"]["schemas"]["ErrorBody"]["properties"]) == ["code", "message"]
+    post = schema["paths"]["/v1/instances"]["post"]["responses"]
+    for code in ("401", "404", "409", "422"):
+        ref = post[code]["content"]["application/json"]["schema"]["$ref"]
+        assert ref.endswith("/ErrorBody"), f"{code} is not documented as ErrorBody"
+
+
+async def test_202_is_on_the_decorator_not_the_response_object(client: httpx.AsyncClient) -> None:
+    """The schema must say 202, not just the runtime.
+
+    Setting response.status_code in a handler body changes the response and leaves the
+    OpenAPI document claiming 200 — so generated clients treat a 202 as an error.
+    """
+    schema = (await client.get("/openapi.json")).json()
+    assert "202" in schema["paths"]["/v1/instances"]["post"]["responses"]
+    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()
+
+
+@pytest.mark.slow
+async def test_lifespan_survives_an_unreachable_jwks(settings: Settings) -> None:
+    """A down identity provider must not stop the pod from starting.
+
+    The warm-up is best-effort on purpose: fail startup on it and an IdP blip means no pod
+    in the fleet can start, so the outage outlives the blip. A cache miss later costs one
+    to_thread hop. Requests still 401 until the keys arrive — fail closed, stay up.
+    """
+    unreachable = settings.model_copy(
+        update={"jwks_url": "https://nonexistent.invalid/protocol/openid-connect/certs"}
+    )
+    booted = create_app(unreachable)
+    async with booted.router.lifespan_context(booted):
+        transport = httpx.ASGITransport(app=booted)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+            assert (await c.get("/healthz")).status_code == 200
+            # Keys never loaded, so auth fails closed rather than falling open.
+            assert (await c.get("/v1/instances", headers=auth("whatever"))).status_code == 401
+
+
+async def test_auth_disabled_accepts_an_unauthenticated_request(settings: Settings) -> None:
+    """With the hatch open, no header is needed and a fixed team is used."""
+    dev_app = create_app(settings.model_copy(update={"auth_disabled": True}))
+    async with dev_app.router.lifespan_context(dev_app):
+        transport = httpx.ASGITransport(app=dev_app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+            resp = await c.get("/v1/instances")
+    assert resp.status_code == 200

tests/integration/test_claim.py

@@ -0,0 +1,43 @@
+"""The claim race. If only one test in this repo survives, it should be this one."""
+
+from __future__ import annotations
+
+import asyncio
+
+from svcforge_core.domain.models import TaskKind
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.tasks import TaskRepo
+from tests.integration.helpers import make_instance
+
+
+async def test_skip_locked_claims_each_task_exactly_once(pool: DictPool) -> None:
+    """50 workers, 50 tasks, one claim each — no double-claims, no lost tasks.
+
+    This is the test that fails if you split the claim into select-then-update.
+    """
+    repo = TaskRepo(pool)
+    inst = await make_instance(pool)
+    ids = {await repo.enqueue_standalone(inst, TaskKind.PROVISION) for _ in range(50)}
+
+    async with asyncio.TaskGroup() as tg:
+        claims = [tg.create_task(repo.claim(f"w{i}")) for i in range(50)]
+    got = [c.result() for c in claims]
+
+    assert all(t is not None for t in got)
+    assert sorted(t.id for t in got if t is not None) == sorted(ids)  # each exactly once
+    assert all(t.attempts == 1 for t in got if t is not None)
+
+
+async def test_more_workers_than_tasks_get_none_not_a_duplicate(pool: DictPool) -> None:
+    """Contention must produce None for the losers, never a second claim on one row."""
+    repo = TaskRepo(pool)
+    inst = await make_instance(pool)
+    await repo.enqueue_standalone(inst, TaskKind.PROVISION)
+
+    async with asyncio.TaskGroup() as tg:
+        claims = [tg.create_task(repo.claim(f"w{i}")) for i in range(10)]
+    got = [c.result() for c in claims]
+
+    won = [t for t in got if t is not None]
+    assert len(won) == 1
+    assert len([t for t in got if t is None]) == 9

tests/integration/test_day2.py

@@ -0,0 +1,157 @@
+"""Day 2 against a real Postgres: the work list, the halt, and the window.
+
+The work-list query is the entire rollout, so it is tested where it runs. `order by
+team = %s desc` and `not exists (... halted)` are SQL semantics — a fake repo asserting
+them would only prove the fake agrees with itself.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from uuid import UUID
+
+import pytest
+
+from svcforge_core.domain.models import TaskKind
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.domain.windows import parse_window, schedule_upgrade_at
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.repo.tasks import TaskRepo
+from tests.integration.helpers import make_instance
+
+OLD = "21.3.19"  # what is deployed
+PINNED = "21.3.20"  # what catalog.yaml now says
+OWN_TEAM = "platform"
+
+
+async def _set_window(pool: DictPool, instance_id: UUID, spec: str | None) -> None:
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "update instances set maintenance_window = %s where id = %s",
+            (spec, instance_id),
+        )
+
+
+async def _halt(pool: DictPool, service_type: str) -> None:
+    """What `handle_verify` does on a failed probe, and what you undo by hand with SQL."""
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            """insert into catalog_versions (service_type, rollout_state) values (%s, 'halted')
+               on conflict (service_type) do update set rollout_state = 'halted'""",
+            (service_type,),
+        )
+
+
+@pytest.fixture
+async def fleet(pool: DictPool) -> list[UUID]:
+    """Three ready instances on the old version. The own-team one is created LAST.
+
+    Created last on purpose: `created_at` is the tiebreak, so if the `team = %s desc` sort
+    were dropped this fixture makes the test fail instead of passing by luck.
+    """
+    ids = [
+        await make_instance(pool, team="tenant-a", state=InstanceState.READY, chart_version=OLD),
+        await make_instance(pool, team="tenant-b", state=InstanceState.READY, chart_version=OLD),
+        await make_instance(pool, team=OWN_TEAM, state=InstanceState.READY, chart_version=OLD),
+    ]
+    return ids
+
+
+async def test_work_list_returns_one_row_and_it_is_the_own_team_row(
+    pool: DictPool, fleet: list[UUID]
+) -> None:
+    """max_in_flight=1 means one instance moves at a time, and yours is the guinea pig."""
+    repo = InstanceRepo(pool)
+
+    rows = await repo.list_upgradable(
+        service_type="elasticsearch",
+        catalog_version=PINNED,
+        own_team=OWN_TEAM,
+        max_in_flight=1,
+    )
+
+    assert len(rows) == 1
+    assert rows[0].instance.team == OWN_TEAM
+    assert rows[0].instance.id == fleet[2]
+    assert rows[0].instance.chart_version == OLD
+
+
+async def test_work_list_skips_instances_already_on_the_pinned_version(
+    pool: DictPool, fleet: list[UUID]
+) -> None:
+    """The query is the progress bar: as instances land on PINNED, the list drains to empty."""
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("update instances set chart_version = %s", (PINNED,))
+
+    rows = await repo.list_upgradable(
+        service_type="elasticsearch", catalog_version=PINNED, own_team=OWN_TEAM, max_in_flight=10
+    )
+
+    assert rows == []
+
+
+async def test_halted_rollout_returns_zero_rows(pool: DictPool, fleet: list[UUID]) -> None:
+    """One column stops the fleet. This is the whole stop button."""
+    repo = InstanceRepo(pool)
+    await _halt(pool, "elasticsearch")
+
+    rows = await repo.list_upgradable(
+        service_type="elasticsearch",
+        catalog_version=PINNED,
+        own_team=OWN_TEAM,
+        max_in_flight=10,  # generous on purpose: it is the halt returning 0, not the limit
+    )
+
+    assert rows == []
+
+
+async def test_halt_is_scoped_to_one_service_type(pool: DictPool) -> None:
+    """A broken redis chart must not freeze elasticsearch upgrades."""
+    repo = InstanceRepo(pool)
+    await make_instance(
+        pool, team=OWN_TEAM, service_type="redis", state=InstanceState.READY, chart_version=OLD
+    )
+    await make_instance(
+        pool, team=OWN_TEAM, service_type="elasticsearch", state=InstanceState.READY, chart_version=OLD
+    )
+    await _halt(pool, "redis")
+
+    assert await repo.list_upgradable("redis", PINNED, OWN_TEAM, 10) == []
+    assert len(await repo.list_upgradable("elasticsearch", PINNED, OWN_TEAM, 10)) == 1
+
+
+async def test_windowed_upgrade_is_scheduled_in_the_future_and_security_bypasses_it(
+    pool: DictPool,
+) -> None:
+    """The window lands in `tasks.run_after`, and `security: true` ignores it.
+
+    Both paths go through the real enqueue, so this also pins the `timestamptz` round-trip:
+    an aware UTC datetime must come back out of Postgres still aware and still that instant.
+    """
+    repo = InstanceRepo(pool)
+    tasks = TaskRepo(pool)
+    iid = await make_instance(pool, team=OWN_TEAM, state=InstanceState.READY, chart_version=OLD)
+    await _set_window(pool, iid, "0 3 * * 0|Asia/Ho_Chi_Minh")
+
+    (candidate,) = await repo.list_upgradable("elasticsearch", PINNED, OWN_TEAM, 1)
+    window = parse_window(candidate.maintenance_window)
+    assert window is not None
+
+    now = datetime.now(UTC)
+
+    routine = await tasks.enqueue_standalone(
+        iid, TaskKind.UPGRADE, schedule_upgrade_at(window, security=False, now=now)
+    )
+    urgent = await tasks.enqueue_standalone(
+        iid, TaskKind.UPGRADE, schedule_upgrade_at(window, security=True, now=now)
+    )
+
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select id, run_after from tasks where id = any(%s)", ([routine, urgent],))
+        run_after = {r["id"]: r["run_after"] for r in await cur.fetchall()}
+
+    assert run_after[routine] > now  # waits for 03:00 Sunday, Vietnam time
+    assert run_after[urgent] <= now  # a public exploit does not wait
+    assert run_after[routine].tzinfo is not None

tests/integration/test_helm_timeout.py

@@ -0,0 +1,29 @@
+"""The one test that proves the timeout is real.
+
+`bash -c "sleep 300 & sleep 300"` is a miniature helm: a process that forks a child and
+waits on another. Kill the direct child only and the backgrounded `sleep` reparents to init
+and keeps running — which, when the process is helm, means a timed-out task retries while
+the original helm is still mutating the same release.
+
+This test needs a real process tree, so it lives in integration/. It needs no database:
+the `pool` fixture in conftest is not autouse.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import subprocess
+
+import pytest
+
+from svcforge_core.adapters.helm import _run
+
+
+@pytest.mark.asyncio
+async def test_timeout_kills_the_whole_process_group() -> None:
+    argv = ["bash", "-c", "sleep 300 & sleep 300"]  # child forks a grandchild
+    with pytest.raises(TimeoutError):
+        await _run(argv, timeout_s=1)
+    await asyncio.sleep(0.5)
+    out = subprocess.run(["pgrep", "-f", "sleep 300"], capture_output=True, text=True)  # noqa: S607
+    assert out.stdout.strip() == "", "grandchild survived: you killed the child, not the group"

tests/integration/test_instances.py

@@ -0,0 +1,104 @@
+"""InstanceRepo against real SQL."""
+
+from __future__ import annotations
+
+import pytest
+
+from svcforge_core.domain.models import TaskKind
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.repo.tasks import TaskRepo
+from tests.integration.helpers import build_instance
+
+
+async def test_create_then_get_round_trips(pool: DictPool) -> None:
+    repo = InstanceRepo(pool)
+    inst = build_instance()
+    async with pool.connection() as conn:
+        created = await repo.create(conn, inst)
+    assert created.id == inst.id
+    assert created.release_name == inst.release_name
+
+    got = await repo.get(inst.id, team="platform")
+    assert got is not None
+    assert got.service_type == "elasticsearch"
+    assert got.state is InstanceState.REQUESTED
+    # timestamptz round-trips as aware, or every later comparison raises TypeError.
+    assert got.created_at.tzinfo is not None
+
+
+async def test_get_by_other_team_is_none_not_403(pool: DictPool) -> None:
+    """A wrong-team id is indistinguishable from a missing one."""
+    repo = InstanceRepo(pool)
+    inst = build_instance(team="platform")
+    async with pool.connection() as conn:
+        await repo.create(conn, inst)
+
+    assert await repo.get(inst.id, team="quant") is None
+
+
+async def test_list_is_filtered_by_team(pool: DictPool) -> None:
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn:
+        await repo.create(conn, build_instance(team="platform"))
+        await repo.create(conn, build_instance(team="quant"))
+
+    mine = await repo.list(team="platform")
+    assert len(mine) == 1
+    assert all(i.team == "platform" for i in mine)
+
+
+async def test_update_state_cas_rejects_stale_expectation(pool: DictPool) -> None:
+    repo = InstanceRepo(pool)
+    inst = build_instance()
+    async with pool.connection() as conn:
+        await repo.create(conn, inst)
+
+    ok = await repo.update_state(inst.id, InstanceState.REQUESTED, InstanceState.PROVISIONING)
+    assert ok is True
+
+    # The row already moved: the second caller must lose, and must not raise.
+    lost = await repo.update_state(inst.id, InstanceState.REQUESTED, InstanceState.PROVISIONING)
+    assert lost is False
+
+
+async def test_update_state_sets_endpoint(pool: DictPool) -> None:
+    repo = InstanceRepo(pool)
+    inst = build_instance()
+    async with pool.connection() as conn:
+        await repo.create(conn, inst)
+    await repo.update_state(inst.id, InstanceState.REQUESTED, InstanceState.PROVISIONING)
+    ok = await repo.update_state(
+        inst.id, InstanceState.PROVISIONING, InstanceState.READY, endpoint="http://es:9200"
+    )
+    assert ok is True
+    got = await repo.get(inst.id, team="platform")
+    assert got is not None
+    assert got.endpoint == "http://es:9200"
+    assert got.state is InstanceState.READY
+
+
+@pytest.mark.parametrize("explode", [True])
+async def test_instance_and_task_roll_back_together(pool: DictPool, explode: bool) -> None:
+    """The reason the queue is in Postgres, as an executable claim.
+
+    If the transaction aborts, BOTH the instance and its provision task must vanish.
+    An instance with no task never gets built; a task with no instance is an orphan.
+    """
+    instances, tasks = InstanceRepo(pool), TaskRepo(pool)
+    inst = build_instance()
+
+    with pytest.raises(RuntimeError):
+        async with pool.connection() as conn, conn.transaction():
+            await instances.create(conn, inst)
+            await tasks.enqueue(conn, inst.id, TaskKind.PROVISION)
+            if explode:
+                raise RuntimeError("boom, mid-transaction")
+
+    assert await instances.get(inst.id, team="platform") is None
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select count(*) as n from tasks")
+        row = await cur.fetchone()
+    assert row is not None
+    assert row["n"] == 0

tests/integration/test_reconciler.py

@@ -0,0 +1,556 @@
+"""The four checks, against a real Postgres and a fake cluster.
+
+Integration, not unit, because there is nothing to unit test: each check is a query and a
+transaction. The behaviour worth asserting — that the CAS and the insert commit together,
+that the idempotency guard is a real `not exists`, that a second tick does not double-
+enqueue — lives entirely in the part a mock would replace.
+
+The cluster is faked; the database is not. FakeProvisioner is a dict of releases, which is
+all the drift check needs: drift is "helm says X, the DB says Y", and a dict says X.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import UUID
+
+import pytest
+from opentelemetry import trace
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+from services.reconciler.main import (
+    ReconcilerDeps,
+    check_drift,
+    check_lease_expiry,
+    check_ttl,
+    check_version_drift,
+    tick,
+)
+from svcforge_core.domain.models import CatalogEntry, SizeSpec, TaskKind, TaskState
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.obs import RECONCILER_LAST_TICK
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.repo.reconcile import ReconcileRepo
+from svcforge_core.repo.tasks import TaskRepo
+from svcforge_core.settings import Settings
+from tests.fakes import FakeClock, FakeNotifier, FakeProvisioner
+from tests.integration.helpers import build_instance
+
+NOW = datetime(2026, 7, 17, 12, 0, tzinfo=UTC)  # a Friday
+OLD_VERSION = "21.3.19"
+NEW_VERSION = "21.3.20"
+
+# '0 3 * * 0' is 03:00 Sunday. From a Friday noon that is always in the future, which is
+# the whole assertion of the window test.
+SUNDAY_0300_HCM = "0 3 * * 0|Asia/Ho_Chi_Minh"
+
+
+def _entry(*, version: str = NEW_VERSION, security: bool = False) -> CatalogEntry:
+    return CatalogEntry(
+        service_type="elasticsearch",
+        chart="bitnamilegacy/elasticsearch",
+        chart_version=version,
+        security=security,
+        sizes={"small": SizeSpec(replicas=1, resources={})},
+    )
+
+
+def _settings(**over: object) -> Settings:
+    return Settings(pg_dsn="postgresql://u:p@localhost:5432/db", **over)  # type: ignore[arg-type]
+
+
+def _deps(
+    pool: DictPool,
+    provisioner: FakeProvisioner,
+    *,
+    catalog: dict[str, CatalogEntry] | None = None,
+    max_in_flight: int = 1,
+    **settings_over: object,
+) -> ReconcilerDeps:
+    return ReconcilerDeps(
+        pool=pool,
+        instances=InstanceRepo(pool),
+        tasks=TaskRepo(pool),
+        reconcile=ReconcileRepo(pool),
+        provisioner=provisioner,
+        notifier=FakeNotifier(),
+        clock=FakeClock(NOW),
+        catalog=catalog if catalog is not None else {"elasticsearch": _entry()},
+        settings=_settings(**settings_over),
+        own_team="platform",
+        max_in_flight=max_in_flight,
+    )
+
+
+async def _seed(
+    pool: DictPool,
+    *,
+    state: InstanceState = InstanceState.READY,
+    team: str = "platform",
+    chart_version: str = OLD_VERSION,
+    expires_at: datetime | None = None,
+    maintenance_window: str | None = None,
+) -> tuple[UUID, str, str]:
+    """Insert one instance. Returns (id, release_name, namespace).
+
+    `expires_at` and `maintenance_window` go in with SQL rather than through
+    `InstanceRepo.create`: create() does not write `maintenance_window` at all (nothing but
+    the day-2 work list reads it), and that is a fact about the repo, not a gap in it.
+    """
+    inst = build_instance(team=team, state=state, chart_version=chart_version)
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn:
+        await repo.create(conn, inst)
+        async with conn.cursor() as cur:
+            await cur.execute(
+                "update instances set expires_at = %s, maintenance_window = %s where id = %s",
+                (expires_at, maintenance_window, inst.id),
+            )
+    return inst.id, inst.release_name, inst.namespace
+
+
+async def _tasks_for(pool: DictPool, instance_id: UUID) -> list[dict[str, Any]]:
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "select id, kind, state, run_after, traceparent from tasks where instance_id = %s order by id",
+            (instance_id,),
+        )
+        return list(await cur.fetchall())
+
+
+async def _state_of(pool: DictPool, instance_id: UUID) -> str:
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select state from instances where id = %s", (instance_id,))
+        row = await cur.fetchone()
+    assert row is not None
+    return str(row["state"])
+
+
+# --- Check 1: drift ------------------------------------------------------------------------
+
+
+async def test_drift_reprovisions_a_ready_instance_whose_release_vanished(
+    pool: DictPool,
+) -> None:
+    """`helm uninstall` by hand. Nobody sends an event; the next tick notices anyway.
+
+    This is the acceptance path from Module 7:
+        helm uninstall <release> -n <ns> && python -m services.reconciler.main --once
+        select kind, state from tasks order by id desc limit 1   -> provision | queued
+    """
+    instance_id, _, _ = await _seed(pool)
+    deps = _deps(pool, FakeProvisioner())  # empty cluster: the release is gone
+
+    await check_drift(deps)
+
+    tasks = await _tasks_for(pool, instance_id)
+    assert [(t["kind"], t["state"]) for t in tasks] == [(TaskKind.PROVISION.value, TaskState.QUEUED.value)]
+
+
+async def test_drift_leaves_the_instance_in_provisioning_not_failed(pool: DictPool) -> None:
+    """The two-hop state change, and why it matters.
+
+    `handle_provision` returns early on a `ready` row and ends with a
+    `provisioning -> ready` CAS. Hand it anything else and helm runs but the bookkeeping
+    lands nowhere. So the reconciler must leave the row in `provisioning` — via `failed`,
+    because LEGAL has no `ready -> provisioning` edge — before the worker can claim it.
+    """
+    instance_id, _, _ = await _seed(pool)
+
+    await check_drift(_deps(pool, FakeProvisioner()))
+
+    assert await _state_of(pool, instance_id) == InstanceState.PROVISIONING.value
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select error from instances where id = %s", (instance_id,))
+        row = await cur.fetchone()
+    assert row is not None
+    assert "drift" in row["error"]  # the tenant gets told why, not just that
+
+
+async def test_drift_ignores_an_instance_whose_release_is_present(pool: DictPool) -> None:
+    """The happy path is the same code. It must enqueue nothing at all."""
+    instance_id, release, namespace = await _seed(pool)
+    provisioner = FakeProvisioner()
+    await provisioner.install(release, namespace, _entry(), {})
+
+    await check_drift(_deps(pool, provisioner))
+
+    assert await _tasks_for(pool, instance_id) == []
+    assert await _state_of(pool, instance_id) == InstanceState.READY.value
+
+
+async def test_drift_is_idempotent_across_ticks(pool: DictPool) -> None:
+    """Two ticks, one task. A provision takes minutes; ticks are 60 seconds apart.
+
+    Without the guard the second tick sees a `provisioning` row — no longer `ready`, so the
+    drift branch skips it. The guard is what covers the case where it is `ready` again
+    before the task is done.
+    """
+    instance_id, _, _ = await _seed(pool)
+    deps = _deps(pool, FakeProvisioner())
+
+    await check_drift(deps)
+    await check_drift(deps)
+
+    assert len(await _tasks_for(pool, instance_id)) == 1
+
+
+async def test_drift_never_deletes_an_orphan_release(pool: DictPool) -> None:
+    """A release the DB has never heard of. Log it, bill nobody, delete nothing.
+
+    v1 policy, and it is a policy about evidence: "no row in this table" is not proof the
+    release is unowned. It might belong to another tool, another team, or a migration that
+    is half done. An operator deletes it after reading the log.
+    """
+    provisioner = FakeProvisioner()
+    await provisioner.install("someone-elses-redis", "other-ns", _entry(), {})
+
+    await check_drift(_deps(pool, provisioner))
+
+    assert "someone-elses-redis" in provisioner.releases
+    assert provisioner.uninstalled == []
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select count(*) as n from tasks")
+        row = await cur.fetchone()
+    assert row is not None
+    assert row["n"] == 0
+
+
+async def test_drift_does_not_call_an_in_flight_provision_an_orphan(pool: DictPool) -> None:
+    """A `requested` instance a worker is installing right now is not an orphan.
+
+    `known_releases` covers every row in any state for exactly this reason. Scope it to
+    `ready` and every provision in progress gets reported as an orphan on every tick, which
+    trains everyone to ignore the orphan log.
+    """
+    _, release, namespace = await _seed(pool, state=InstanceState.REQUESTED)
+    provisioner = FakeProvisioner()
+    await provisioner.install(release, namespace, _entry(), {})
+
+    await check_drift(_deps(pool, provisioner))
+
+    assert provisioner.uninstalled == []
+
+
+# --- Check 2: lease expiry -----------------------------------------------------------------
+
+
+async def test_lease_expiry_returns_a_dead_workers_task_to_the_queue(pool: DictPool) -> None:
+    """SIGKILL leaves `running` with `locked_by` set and nobody running it.
+
+    No cleanup code in the worker can fix this, because the worker is the part that died.
+    The lease is the only thing that recovers the row.
+    """
+    instance_id, _, _ = await _seed(pool)
+    tasks = TaskRepo(pool)
+    await tasks.enqueue_standalone(instance_id, TaskKind.PROVISION)
+    claimed = await tasks.claim("worker-that-is-about-to-die")
+    assert claimed is not None
+
+    # The worker died six minutes ago and never reported.
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "update tasks set locked_at = now() - interval '6 minutes' where id = %s",
+            (claimed.id,),
+        )
+
+    await check_lease_expiry(_deps(pool, FakeProvisioner(), lease_seconds=300))
+
+    rows = await _tasks_for(pool, instance_id)
+    assert rows[0]["state"] == TaskState.QUEUED.value
+
+
+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.
+    """
+    instance_id, _, _ = await _seed(pool)
+    tasks = TaskRepo(pool)
+    await tasks.enqueue_standalone(instance_id, TaskKind.PROVISION)
+    assert await tasks.claim("worker-1") is not None
+
+    await check_lease_expiry(_deps(pool, FakeProvisioner(), lease_seconds=300))
+
+    rows = await _tasks_for(pool, instance_id)
+    assert rows[0]["state"] == TaskState.RUNNING.value
+
+
+# --- Check 3: TTL --------------------------------------------------------------------------
+
+
+async def test_ttl_expired_instance_goes_to_deleting_with_a_deprovision_task(
+    pool: DictPool,
+) -> None:
+    """The check that stops a demo cluster from becoming a permanent line on the bill."""
+    instance_id, _, _ = await _seed(pool, expires_at=datetime.now(UTC) - timedelta(minutes=1))
+
+    await check_ttl(_deps(pool, FakeProvisioner()))
+
+    tasks = await _tasks_for(pool, instance_id)
+    assert [(t["kind"], t["state"]) for t in tasks] == [(TaskKind.DEPROVISION.value, TaskState.QUEUED.value)]
+    # `deleting` before the worker claims it: handle_deprovision ends with a
+    # `deleting -> deleted` CAS, and a `ready` row would leave the DB advertising an
+    # endpoint for a release helm has already removed.
+    assert await _state_of(pool, instance_id) == InstanceState.DELETING.value
+
+
+async def test_ttl_ignores_an_instance_that_has_not_expired(pool: DictPool) -> None:
+    """And ignores one with no expires_at at all: null means no TTL, not expired."""
+    live, _, _ = await _seed(pool, expires_at=datetime.now(UTC) + timedelta(hours=1))
+    forever, _, _ = await _seed(pool, expires_at=None)
+
+    await check_ttl(_deps(pool, FakeProvisioner()))
+
+    assert await _tasks_for(pool, live) == []
+    assert await _tasks_for(pool, forever) == []
+
+
+async def test_ttl_is_idempotent_across_ticks(pool: DictPool) -> None:
+    """A deprovision takes minutes and ticks are 60s apart. One task, not four."""
+    instance_id, _, _ = await _seed(pool, expires_at=datetime.now(UTC) - timedelta(minutes=1))
+    deps = _deps(pool, FakeProvisioner())
+
+    await check_ttl(deps)
+    await check_ttl(deps)
+    await check_ttl(deps)
+
+    assert len(await _tasks_for(pool, instance_id)) == 1
+
+
+async def test_ttl_recovers_a_deleting_instance_whose_task_was_never_enqueued(
+    pool: DictPool,
+) -> None:
+    """The API's DELETE crashed between the CAS and the enqueue. This is the sweep it relies on.
+
+    That statement order is chosen *because* this check exists. The other order leaves a
+    deprovision task pointing at a `ready` instance, and a worker tears down a live service
+    nobody asked to delete.
+    """
+    instance_id, _, _ = await _seed(pool, state=InstanceState.DELETING)
+
+    await check_ttl(_deps(pool, FakeProvisioner()))
+
+    tasks = await _tasks_for(pool, instance_id)
+    assert [t["kind"] for t in tasks] == [TaskKind.DEPROVISION.value]
+    assert await _state_of(pool, instance_id) == InstanceState.DELETING.value
+
+
+async def test_ttl_re_enqueues_after_a_deprovision_exhausted_its_attempts(
+    pool: DictPool,
+) -> None:
+    """`done` and `failed` are not outstanding. A transient outage must not strand the row.
+
+    The guard asks "is one queued or running", not "has one ever existed" — otherwise a
+    deprovision that burned its five attempts during a cluster outage would leave the
+    instance billing forever with nothing left to retry it.
+    """
+    instance_id, _, _ = await _seed(pool, state=InstanceState.DELETING)
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "insert into tasks (instance_id, kind, state) values (%s, %s, %s)",
+            (instance_id, TaskKind.DEPROVISION.value, TaskState.FAILED.value),
+        )
+
+    await check_ttl(_deps(pool, FakeProvisioner()))
+
+    states = [t["state"] for t in await _tasks_for(pool, instance_id)]
+    assert TaskState.QUEUED.value in states
+
+
+# --- Check 4: version drift ----------------------------------------------------------------
+
+
+async def test_version_drift_enqueues_one_upgrade_for_the_own_team_instance_first(
+    pool: DictPool,
+) -> None:
+    """max_in_flight=1 across three stale instances, and it picks ours.
+
+    Eating your own dog food is an `order by`: we are the tenant who finds out the chart is
+    broken, and the halt stops the other two before they ever hear about it.
+    """
+    await _seed(pool, team="payments")
+    await _seed(pool, team="search")
+    ours, _, _ = await _seed(pool, team="platform")
+
+    await check_version_drift(_deps(pool, FakeProvisioner()))
+
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select instance_id, kind from tasks")
+        rows = list(await cur.fetchall())
+    assert len(rows) == 1
+    assert rows[0]["instance_id"] == ours
+    assert rows[0]["kind"] == TaskKind.UPGRADE.value
+
+
+async def test_version_drift_enqueues_nothing_while_the_rollout_is_halted(
+    pool: DictPool,
+) -> None:
+    """One column stops the fleet. A failed verify writes it; a human clears it with SQL."""
+    instance_id, _, _ = await _seed(pool)
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "insert into catalog_versions (service_type, rollout_state) values ('elasticsearch', 'halted')"
+        )
+
+    await check_version_drift(_deps(pool, FakeProvisioner()))
+
+    assert await _tasks_for(pool, instance_id) == []
+
+
+async def test_version_drift_ignores_an_instance_already_on_the_catalog_version(
+    pool: DictPool,
+) -> None:
+    """`chart_version` is written only after helm succeeds, which is what makes this the query."""
+    instance_id, _, _ = await _seed(pool, chart_version=NEW_VERSION)
+
+    await check_version_drift(_deps(pool, FakeProvisioner()))
+
+    assert await _tasks_for(pool, instance_id) == []
+
+
+async def test_version_drift_parks_the_upgrade_until_the_maintenance_window(
+    pool: DictPool,
+) -> 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.
+    """
+    instance_id, _, _ = await _seed(pool, maintenance_window=SUNDAY_0300_HCM)
+
+    await check_version_drift(_deps(pool, FakeProvisioner()))
+
+    tasks = await _tasks_for(pool, instance_id)
+    assert len(tasks) == 1
+    assert tasks[0]["run_after"] > NOW  # a Friday; the next 03:00 Sunday is days away
+
+
+async def test_version_drift_bypasses_the_window_for_a_security_bump(pool: DictPool) -> None:
+    """A CVE with a public exploit does not wait until Sunday. That is what `security:` is for."""
+    instance_id, _, _ = await _seed(pool, maintenance_window=SUNDAY_0300_HCM)
+    catalog = {"elasticsearch": _entry(security=True)}
+
+    await check_version_drift(_deps(pool, FakeProvisioner(), catalog=catalog))
+
+    tasks = await _tasks_for(pool, instance_id)
+    assert len(tasks) == 1
+    assert tasks[0]["run_after"] <= NOW
+
+
+async def test_version_drift_is_idempotent_across_ticks(pool: DictPool) -> None:
+    """The guard that makes max_in_flight mean anything.
+
+    The instance stays on the work list for the whole duration of its own upgrade —
+    `chart_version` is only written on success — and for the hours it spends parked waiting
+    for 03:00. Without the guard, `max_in_flight=1` is sixty tasks an hour against one
+    release.
+    """
+    instance_id, _, _ = await _seed(pool, maintenance_window=SUNDAY_0300_HCM)
+    deps = _deps(pool, FakeProvisioner())
+
+    for _ in range(3):
+        await check_version_drift(deps)
+
+    assert len(await _tasks_for(pool, instance_id)) == 1
+
+
+async def test_version_drift_skips_one_bad_window_and_keeps_going(pool: DictPool) -> None:
+    """One tenant's typo must not freeze everyone else's security rollout."""
+    broken, _, _ = await _seed(pool, team="payments", maintenance_window="not a cron|Asia/Ho_Chi_Minh")
+    deps = _deps(pool, FakeProvisioner(), max_in_flight=5)
+
+    await check_version_drift(deps)  # must not raise
+
+    assert await _tasks_for(pool, broken) == []
+
+
+# --- The tick ------------------------------------------------------------------------------
+
+
+class _AngryProvisioner(FakeProvisioner):
+    """A cluster that cannot be reached. The drift check's worst day."""
+
+    async def list_releases(self) -> list[Any]:
+        raise RuntimeError("dial tcp: i/o timeout")
+
+
+async def test_tick_runs_the_other_three_checks_when_one_blows_up(pool: DictPool) -> None:
+    """A helm binary that cannot reach the API server must not stop TTLs from expiring.
+
+    This is the entire argument for wrapping each check independently, and it is asserted
+    rather than assumed because the failure mode — a tick that dies on check one — looks
+    exactly like a tick that found nothing to do.
+    """
+    expired, _, _ = await _seed(pool, expires_at=datetime.now(UTC) - timedelta(minutes=1))
+    deps = _deps(pool, _AngryProvisioner())
+
+    await tick(deps)  # must not raise
+
+    assert [t["kind"] for t in await _tasks_for(pool, expired)] == [TaskKind.DEPROVISION.value]
+
+
+async def test_tick_sets_the_gauges_and_the_heartbeat(pool: DictPool) -> None:
+    """queue_depth after the checks, not before, and the heartbeat unconditionally.
+
+    The heartbeat is what `SvcforgeReconcilerStale` reads. It answers "is the loop running",
+    not "is everything fine" — the checks have their own alerts, and an alert that means two
+    things gets muted.
+    """
+    from prometheus_client import REGISTRY
+
+    await _seed(pool, expires_at=datetime.now(UTC) - timedelta(minutes=1))
+    deps = _deps(pool, _AngryProvisioner())  # one check fails; the heartbeat still ticks
+
+    await tick(deps)
+
+    assert REGISTRY.get_sample_value("svcforge_queue_depth") == 1.0
+    assert REGISTRY.get_sample_value("svcforge_instances", {"state": "deleting"}) == 1.0
+    assert RECONCILER_LAST_TICK._value.get() == pytest.approx(NOW.timestamp())
+
+
+@pytest.fixture(scope="session")
+def tracing() -> InMemorySpanExporter:
+    """A real tracer provider for the process, collecting spans in memory.
+
+    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.
+    """
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    trace.set_tracer_provider(provider)
+    return exporter
+
+
+async def test_a_task_the_tick_enqueues_carries_the_ticks_traceparent(
+    pool: DictPool,
+    tracing: InMemorySpanExporter,
+) -> None:
+    """Nothing propagates a trace through a table. The column is written at insert or never.
+
+    Driven through `tick`, not through `check_drift` with a span wrapped around it by the
+    test — that version passed while the real entrypoint wrote null on every row, because
+    the only span in the production path (`helm.list`) had already closed by the time the
+    insert ran. A test that supplies the context under test proves the propagator works and
+    nothing about this service.
+    """
+    instance_id, _, _ = await _seed(pool)
+    tracing.clear()
+
+    await tick(_deps(pool, FakeProvisioner()))
+
+    tasks = await _tasks_for(pool, instance_id)
+    traceparent = tasks[0]["traceparent"]
+    assert traceparent is not None, "the reconciler's own tasks are unjoinable to its tick"
+
+    # Same trace as the tick's span, which is the entire point of storing the column.
+    tick_spans = [s for s in tracing.get_finished_spans() if s.name == "reconciler.tick"]
+    assert len(tick_spans) == 1
+    assert traceparent.split("-")[1] == format(tick_spans[0].context.trace_id, "032x")

tests/integration/test_redis.py

@@ -0,0 +1,494 @@
+"""Module 10 acceptance: the limiter, the idempotency store, the cache, and the budget.
+
+Six checks, in the spec's order:
+
+1. Rate limit 10/min — the 11th is refused.
+2. A check costs exactly ONE Redis command.
+3. Idempotency — the same key twice yields the same UUID.
+4. Cache — the second read costs one command and no DB query.
+5. Redis DOWN = the platform stays UP.
+6. The budget metric exists, and `scripts/redis_budget.py` reads it correctly.
+
+**Two tiers, and the split is the budget.** Upstash's free tier is 500K commands/month;
+a test suite that hammers it is itself the bug this module is about. So the semantics are
+proved against `tests/fakes.py` (free, deterministic, runs on every commit), and only the
+things a fake cannot prove — that the Lua is valid Lua, that `KEYS`/`ARGV` are 1-based,
+that redis-py bills one EVALSHA, that `decode_responses=True` is set — are proved against
+real Upstash under `@pytest.mark.slow`. That tier spends roughly thirty commands per run,
+and `pytest -m "not slow"` skips it entirely.
+
+Check 5 needs no server at all: a closed port is a more faithful outage than a mock.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import time
+from collections.abc import AsyncIterator, Callable
+from datetime import UTC, datetime
+from pathlib import Path
+from types import ModuleType
+from typing import Any
+from uuid import UUID, uuid4
+
+import pytest
+import pytest_asyncio
+from prometheus_client import REGISTRY, generate_latest
+from redis.asyncio import Redis
+
+from svcforge_core.adapters.redis import (
+    IdempotencyStore,
+    InstanceCache,
+    RateLimiter,
+    RateLimitResult,
+    make_redis,
+)
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.settings import Settings
+from tests.fakes import FakeClock, FakeIdempotencyStore, FakeInstanceCache, FakeRateLimiter
+from tests.integration.helpers import build_instance
+
+# Any DSN that parses. These tests never open a Postgres connection through Settings; the
+# `pool` fixture owns the real database.
+_DUMMY_PG_DSN = "postgresql://unused:unused@127.0.0.1:5432/unused"
+
+# A port nothing listens on. `make_redis` will build a client, every command will get
+# ECONNREFUSED, and that is the point of check 5.
+_DEAD_REDIS_DSN = "redis://127.0.0.1:1/0"
+
+_T0 = datetime(2026, 7, 17, 12, 0, 0, tzinfo=UTC)
+
+
+def _metric(op: str) -> float:
+    """The budget counter for one op. Absent labels read as 0, not as an error."""
+    value = REGISTRY.get_sample_value("svcforge_redis_commands_total", {"op": op})
+    return value or 0.0
+
+
+def _errors(op: str) -> float:
+    value = REGISTRY.get_sample_value("svcforge_redis_errors_total", {"op": op})
+    return value or 0.0
+
+
+# --- Fixtures ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def upstash() -> AsyncIterator[Redis]:
+    """The real thing, from `~/.config/svcforge/secrets.env`. Skipped when unset.
+
+    Built through `make_redis` rather than `Redis.from_url` directly, so that
+    `decode_responses=True` is covered by these tests instead of being a comment. Forget it
+    and every assertion below dies on `bytes != str`, which is the whole reason it is the
+    first bug everyone hits.
+    """
+    dsn = os.getenv("SVCFORGE_REDIS_DSN")
+    if not dsn:
+        pytest.skip("SVCFORGE_REDIS_DSN unset; real-Upstash checks skipped")
+    client = make_redis(Settings(pg_dsn=_DUMMY_PG_DSN, redis_dsn=dsn))  # type: ignore[arg-type]  # pydantic coerces str -> *Dsn
+    assert client is not None
+    try:
+        yield client
+    finally:
+        await client.aclose()
+
+
+@pytest_asyncio.fixture
+async def dead_redis() -> AsyncIterator[Redis]:
+    """A client pointed at a closed port. No server was harmed, no commands were billed."""
+    client = make_redis(Settings(pg_dsn=_DUMMY_PG_DSN, redis_dsn=_DEAD_REDIS_DSN))  # type: ignore[arg-type]  # pydantic coerces str -> *Dsn
+    assert client is not None
+    try:
+        yield client
+    finally:
+        await client.aclose()
+
+
+class _CommandCounter:
+    """Counts round trips by wrapping `execute_command` on one client instance.
+
+    This counts what Upstash bills, which is the only definition that matters here. The
+    metric counter is our own bookkeeping and could be wrong in the same direction as the
+    code it measures; this one cannot.
+    """
+
+    def __init__(self, r: Redis) -> None:
+        self.count = 0
+        self._inner: Callable[..., Any] = r.execute_command
+        r.execute_command = self._counting  # type: ignore[method-assign]
+
+    async def _counting(self, *args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
+        self.count += 1
+        return await self._inner(*args, **kwargs)
+
+
+def _budget_script() -> ModuleType:
+    """Import `scripts/redis_budget.py` by path — `scripts/` is not a package, deliberately."""
+    path = Path(__file__).resolve().parents[2] / "scripts" / "redis_budget.py"
+    spec = importlib.util.spec_from_file_location("redis_budget", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+# --- 1. Rate limit: 10/min, the 11th is refused ------------------------------------------
+
+
+async def test_eleventh_request_in_the_window_is_refused() -> None:
+    """`200 x10` then `429`. The fake, so this runs on every commit for free."""
+    limiter = FakeRateLimiter(limit=10, window_s=60, clock=FakeClock(start=_T0))
+
+    results = [await limiter.check("acme") for _ in range(11)]
+
+    assert [r.allowed for r in results] == [True] * 10 + [False]
+    assert results[9].remaining == 0
+    assert results[10].remaining == 0
+    # What the handler puts in `Retry-After` on the 429. Never 0: a client told to retry
+    # immediately retries into the same closed window.
+    assert results[10].retry_after_s >= 1
+
+
+async def test_the_window_rolls_over_and_the_caller_is_allowed_again() -> None:
+    """A fixed window resets on a boundary, not on a sleep. Hence the injected clock."""
+    clock = FakeClock(start=_T0)
+    limiter = FakeRateLimiter(limit=2, window_s=60, clock=clock)
+
+    assert (await limiter.check("acme")).allowed
+    assert (await limiter.check("acme")).allowed
+    assert not (await limiter.check("acme")).allowed
+
+    clock.advance(datetime(2026, 7, 17, 12, 1, 0, tzinfo=UTC) - _T0)
+    assert (await limiter.check("acme")).allowed
+
+
+async def test_teams_do_not_share_a_window() -> None:
+    """`rl:{team}:{window}` — a noisy tenant must not refuse a quiet one."""
+    limiter = FakeRateLimiter(limit=1, window_s=60, clock=FakeClock(start=_T0))
+
+    assert (await limiter.check("acme")).allowed
+    assert not (await limiter.check("acme")).allowed
+    assert (await limiter.check("globex")).allowed
+
+
+@pytest.mark.slow
+async def test_real_lua_refuses_the_eleventh(upstash: Redis) -> None:
+    """The same assertion against real Upstash. ~11 commands.
+
+    This is what a fake cannot prove: that the script is valid Lua, that `KEYS[1]` and
+    `ARGV[1]` are 1-based (0-based indexing would read nil and compare false forever), and
+    that `INCR`-then-conditional-`EXPIRE` actually holds a window open.
+    """
+    team = f"test-{uuid4().hex[:8]}"
+    limiter = RateLimiter(upstash, limit=10, window_s=60)
+
+    results = [await limiter.check(team) for _ in range(11)]
+
+    assert [r.allowed for r in results] == [True] * 10 + [False]
+    assert not any(r.degraded for r in results)
+    assert results[0].remaining == 9
+
+    # The EXPIRE fired on the first INCR only, so the key is not immortal. `Every key gets
+    # a TTL` is a rule with no enforcement other than checking.
+    window = int(time.time()) // 60
+    keys = [f"rl:{team}:{window}", f"rl:{team}:{window - 1}"]
+    ttls = [await upstash.ttl(k) for k in keys]
+    assert any(0 < ttl <= 60 for ttl in ttls)
+    await upstash.delete(*keys)
+
+
+# --- 2. It costs ONE command per check ---------------------------------------------------
+
+
+@pytest.mark.slow
+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
+    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
+    answers NOSCRIPT because it has never seen the hash, and redis-py replays it as EVAL —
+    two commands, once per Redis restart, and irrelevant to the steady state this measures.
+    """
+    team = f"test-{uuid4().hex[:8]}"
+    limiter = RateLimiter(upstash, limit=100, window_s=60)
+
+    await limiter.check(team)  # warm the script cache
+
+    counter = _CommandCounter(upstash)
+    before = _metric("ratelimit")
+    result = await limiter.check(team)
+
+    assert counter.count == 1, "a rate limit check must be one round trip and one billed command"
+    assert _metric("ratelimit") - before == 1, "the budget counter must agree with the wire"
+    assert result.allowed
+
+    window = int(time.time()) // 60
+    await upstash.delete(f"rl:{team}:{window}", f"rl:{team}:{window - 1}")
+
+
+# --- 3. Idempotency: the same key twice is one instance ----------------------------------
+
+
+async def test_same_idempotency_key_returns_the_first_instance_id() -> None:
+    """The first caller wins; the second is told who won and must not create anything."""
+    store = FakeIdempotencyStore()
+    key = str(uuid4())
+    first, second = uuid4(), uuid4()
+
+    assert await store.claim(key, first) is None, "None means 'you won, go create it'"
+    assert await store.claim(key, second) == first, "the loser gets the winner's id, not its own"
+
+
+async def test_different_idempotency_keys_do_not_collide() -> None:
+    store = FakeIdempotencyStore()
+    a, b = uuid4(), uuid4()
+
+    assert await store.claim(str(uuid4()), a) is None
+    assert await store.claim(str(uuid4()), b) is None
+
+
+@pytest.mark.slow
+async def test_real_set_nx_ex_claims_once(upstash: Redis) -> None:
+    """SET NX EX against Upstash. ~3 commands.
+
+    Also asserts the TTL, because a claim marker without one is a permanent record of a
+    request from last March, and 256 MB of those ends by evicting the keys you cared about.
+    """
+    key = f"test-{uuid4()}"
+    first, second = uuid4(), uuid4()
+    store = IdempotencyStore(upstash, ttl_s=60)
+
+    assert await store.claim(key, first) is None
+    assert await store.claim(key, second) == first
+
+    assert 0 < await upstash.ttl(f"idem:{key}") <= 60
+    await upstash.delete(f"idem:{key}")
+
+
+# --- 4. Cache: the second read costs one command and no DB query -------------------------
+
+
+class _CountingRepo:
+    """Wraps InstanceRepo and counts reads. The DB query count is the assertion."""
+
+    def __init__(self, repo: InstanceRepo) -> None:
+        self._repo = repo
+        self.gets = 0
+
+    async def get(self, id: UUID, team: str) -> Any:  # noqa: ANN401
+        self.gets += 1
+        return await self._repo.get(id, team)
+
+
+async def _read_through(
+    instance_id: UUID,
+    team: str,
+    cache: FakeInstanceCache | InstanceCache,
+    repo: _CountingRepo,
+) -> Any:  # noqa: ANN401
+    """Cache-aside, as `GET /v1/instances/{id}` performs it. Hit = 1 command, miss = 2."""
+    cached = await cache.get(instance_id)
+    if cached is not None:
+        return cached
+    inst = await repo.get(instance_id, team)
+    if inst is not None:
+        await cache.put(inst)
+    return inst
+
+
+async def test_second_read_is_served_from_cache_without_touching_postgres(
+    pool: DictPool,
+) -> None:
+    """Miss, then hit. The DB is read exactly once for two reads."""
+    repo = InstanceRepo(pool)
+    # `create` returns the row as Postgres stored it. Compare against that, never against
+    # the model that went in: `created_at`/`updated_at` are DB defaults, and a timestamptz
+    # comes back tagged `Etc/UTC` rather than `timezone.utc`. Same instant, different repr.
+    async with pool.connection() as conn:
+        inst = await repo.create(conn, build_instance(team="platform"))
+
+    counting = _CountingRepo(repo)
+    cache = FakeInstanceCache()
+
+    first = await _read_through(inst.id, inst.team, cache, counting)
+    second = await _read_through(inst.id, inst.team, cache, counting)
+
+    assert first == second == inst
+    assert counting.gets == 1, "the second read must not reach Postgres"
+    assert (cache.misses, cache.hits) == (1, 1)
+
+
+async def test_invalidate_sends_the_next_read_back_to_postgres() -> None:
+    """The worker calls this inside the code path that writes the state, not after it."""
+    cache = FakeInstanceCache()
+    inst = build_instance()
+
+    await cache.put(inst)
+    assert await cache.get(inst.id) == inst
+
+    await cache.invalidate(inst.id)
+    assert await cache.get(inst.id) is None
+
+
+@pytest.mark.slow
+async def test_real_cache_hit_costs_one_command_and_no_db_query(pool: DictPool, upstash: Redis) -> None:
+    """Real Redis, real Postgres. ~4 commands.
+
+    The round-trip count is the acceptance criterion: a hit that costs two commands is a
+    cache that has doubled the bill it was added to reduce.
+    """
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn:
+        inst = await repo.create(conn, build_instance(team="platform"))
+
+    counting = _CountingRepo(repo)
+    cache = InstanceCache(upstash, ttl_s=30)
+
+    miss = await _read_through(inst.id, inst.team, cache, counting)
+    assert counting.gets == 1
+
+    counter = _CommandCounter(upstash)
+    before = _metric("cache_get")
+    hit = await _read_through(inst.id, inst.team, cache, counting)
+
+    assert counter.count == 1, "a cache hit is one GET"
+    assert _metric("cache_get") - before == 1
+    assert counting.gets == 1, "the second read must not reach Postgres"
+    # Round-tripped through JSON and back: `decode_responses=True` and the datetime/UUID
+    # serialisation both have to be right for this to compare equal.
+    assert hit == miss == inst
+
+    assert 0 < await upstash.ttl(f"inst:{inst.id}") <= 30
+    await cache.invalidate(inst.id)
+    assert await cache.get(inst.id) is None
+
+
+# --- 5. Redis down = the platform stays up -----------------------------------------------
+
+
+async def test_rate_limiter_fails_open_when_redis_is_down(dead_redis: Redis) -> None:
+    """The load-bearing one. A limiter that fails closed turns a cache outage into an outage.
+
+    `degraded=True` and the error counter are what stop this from being invisible: fail
+    open silently and you cannot tell a working limiter from one that has been allowing
+    everything for a month.
+    """
+    limiter = RateLimiter(dead_redis, limit=1, window_s=60)
+    before = _errors("ratelimit")
+
+    results = [await limiter.check("acme") for _ in range(3)]
+
+    assert all(r.allowed for r in results), "Redis being down must never refuse legitimate traffic"
+    assert all(r.degraded for r in results)
+    assert _errors("ratelimit") - before == 3
+
+
+async def test_idempotency_falls_through_to_the_db_when_redis_is_down(dead_redis: Redis) -> None:
+    """None means "create it". Safe only because `instances.release_name` is UNIQUE."""
+    store = IdempotencyStore(dead_redis)
+    key = str(uuid4())
+
+    assert await store.claim(key, uuid4()) is None
+    assert await store.claim(key, uuid4()) is None
+
+
+async def test_cache_misses_instead_of_raising_when_redis_is_down(dead_redis: Redis) -> None:
+    """A miss falls through to Postgres. `put` and `invalidate` swallow it too."""
+    cache = InstanceCache(dead_redis)
+    inst = build_instance()
+
+    assert await cache.get(inst.id) is None
+    await cache.put(inst)  # must not raise
+    await cache.invalidate(inst.id)  # must not raise
+
+
+async def test_platform_serves_reads_from_postgres_with_redis_down(pool: DictPool, dead_redis: Redis) -> None:
+    """The acceptance check: `docker compose stop redis` then GET -> 200.
+
+    Every Redis path degrades and the read still returns the row. Nothing here consults
+    Redis for readiness, which is the other half of the rule — `/readyz` is Postgres-only,
+    so a Redis outage cannot make a single pod unready.
+    """
+    repo = InstanceRepo(pool)
+    async with pool.connection() as conn:
+        inst = await repo.create(conn, build_instance(team="platform"))
+
+    limiter = RateLimiter(dead_redis, limit=10, window_s=60)
+    cache = InstanceCache(dead_redis)
+    store = IdempotencyStore(dead_redis)
+
+    assert (await limiter.check(inst.team)).allowed
+    assert await store.claim(str(uuid4()), inst.id) is None
+    assert await cache.get(inst.id) is None
+    assert await repo.get(inst.id, inst.team) == inst  # the 200
+
+
+# --- 6. The budget metric exists and is sane ---------------------------------------------
+
+
+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)
+
+    text = generate_latest(REGISTRY).decode()
+
+    assert "svcforge_redis_commands_total" in text
+    assert "svcforge_redis_errors_total" in text
+    # Per-op labels, because "you are over budget" is useless without "on cache_get".
+    assert _metric("ratelimit") >= 0
+
+
+def test_budget_script_projects_a_five_second_poller_over_budget() -> None:
+    """The spec's headline number, reproduced: one worker polling every 5s = 518,400/month.
+
+    Exactly the free tier, spent doing nothing. This is the case the script exists to
+    catch, so it is the case that is asserted rather than left to a comment.
+    """
+    budget = _budget_script()
+    now = time.time()
+    # One hour at one command every five seconds.
+    per_op, started_at = budget.collect(
+        f'svcforge_redis_commands_total{{op="poll"}} 720.0\nprocess_start_time_seconds {now - 3600}\n'
+    )
+
+    assert per_op == {"poll": 720.0}
+    assert budget.report(per_op, started_at, budget._FREE_TIER_BUDGET, now) == 1
+
+
+def test_budget_script_passes_a_request_path_workload() -> None:
+    """Request-path volume is bounded by humans, and humans are slow. That is the whole rule."""
+    budget = _budget_script()
+    now = time.time()
+    per_op, started_at = budget.collect(
+        f'svcforge_redis_commands_total{{op="ratelimit"}} 100.0\n'
+        f'svcforge_redis_commands_total{{op="cache_get"}} 40.0\n'
+        f"process_start_time_seconds {now - 3600}\n"
+    )
+
+    assert sum(per_op.values()) == 140.0
+    assert budget.report(per_op, started_at, budget._FREE_TIER_BUDGET, now) == 0
+
+
+def test_budget_script_refuses_to_guess_without_the_metric() -> None:
+    """No counter means the process never imported the adapter — say so, do not print a 0."""
+    budget = _budget_script()
+
+    with pytest.raises(budget.BudgetError, match="not exposed"):
+        budget.collect("process_start_time_seconds 1.0\n")
+
+    with pytest.raises(budget.BudgetError, match="process_start_time_seconds"):
+        budget.collect('svcforge_redis_commands_total{op="ratelimit"} 5.0\n')
+
+
+def test_budget_script_refuses_a_non_http_url() -> None:
+    """`--url file:///etc/passwd` is not a metrics endpoint."""
+    budget = _budget_script()
+
+    with pytest.raises(budget.BudgetError, match="non-http"):
+        budget.scrape("file:///etc/passwd")

tests/integration/test_tasks.py

@@ -0,0 +1,155 @@
+"""TaskRepo: enqueue, complete, fail, backoff, leases."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from svcforge_core.domain.models import TaskKind, TaskState
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.repo.tasks import TaskRepo
+from tests.integration.helpers import make_instance
+
+
+async def _task_row(pool: DictPool, task_id: int) -> dict[str, object]:
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select * from tasks where id = %s", (task_id,))
+        row = await cur.fetchone()
+    assert row is not None
+    return dict(row)
+
+
+async def test_enqueue_defaults_to_runnable_now(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    row = await _task_row(pool, tid)
+    assert row["state"] == "queued"
+    assert row["attempts"] == 0
+
+
+async def test_complete_marks_done_and_releases_lock(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    claimed = await repo.claim("w1")
+    assert claimed is not None
+    await repo.complete(claimed.id)
+    row = await _task_row(pool, tid)
+    assert row["state"] == "done"
+    assert row["locked_by"] is None
+
+
+async def test_fail_under_max_attempts_requeues_with_future_run_after(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    claimed = await repo.claim("w1")
+    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)
+    row = await _task_row(pool, tid)
+    assert row["state"] == "queued"
+    assert row["last_error"] == "helm exploded"
+    assert row["locked_by"] is None
+    # Backoff pushed it out; it must not be immediately runnable again.
+    run_after = row["run_after"]
+    assert isinstance(run_after, datetime)
+    assert run_after >= datetime.now(UTC) - timedelta(seconds=1)
+
+
+async def test_fail_at_max_attempts_dead_letters_and_marks_instance(pool: DictPool) -> None:
+    """A dead-letter state, not an infinite retry."""
+    iid = await make_instance(pool)
+    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
+    tid = await tasks.enqueue_standalone(iid, TaskKind.PROVISION)
+
+    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)
+
+    row = await _task_row(pool, tid)
+    assert row["state"] == "failed"
+    inst = await instances.get(iid, team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.FAILED
+    assert inst.error == "chart not found"
+
+
+async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> None:
+    """Raw SQL must obey the same state machine domain.transition() enforces.
+
+    LEGAL[DELETED] is empty — deleted is terminal. A deprovision task that exhausts its
+    retries after the instance is already gone must record nothing on it, not drag it
+    back to 'failed'.
+    """
+    iid = await make_instance(pool, state=InstanceState.DELETED)
+    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
+    tid = await tasks.enqueue_standalone(iid, TaskKind.DEPROVISION)
+
+    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)
+
+    # The task still dead-letters — that part is unconditional.
+    row = await _task_row(pool, tid)
+    assert row["state"] == "failed"
+
+    # But the instance stays deleted.
+    inst = await instances.get(iid, team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.DELETED, "raw SQL bypassed the state machine"
+    assert inst.error is None
+
+
+async def test_fail_truncates_error_to_2kb(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    await repo.claim("w1")
+    await repo.fail(tid, "x" * 9000, max_attempts=5)
+    row = await _task_row(pool, tid)
+    assert isinstance(row["last_error"], str)
+    assert len(row["last_error"]) == 2000
+
+
+async def test_run_after_in_the_future_is_not_claimable(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    future = datetime.now(UTC) + timedelta(hours=1)
+    await repo.enqueue_standalone(iid, TaskKind.PROVISION, run_after=future)
+    assert await repo.claim("w1") is None
+
+
+async def test_reset_expired_leases_recovers_a_dead_workers_task(pool: DictPool) -> None:
+    """No distributed lock survives a power cut. Only the lease recovers this row."""
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    claimed = await repo.claim("worker-that-will-die")
+    assert claimed is not None
+
+    # Simulate: the worker was SIGKILLed 10 minutes ago and never reported.
+    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,))
+
+    n = await repo.reset_expired_leases(lease_seconds=300)
+    assert n == 1
+    row = await _task_row(pool, tid)
+    assert row["state"] == TaskState.QUEUED.value
+    assert row["locked_by"] is None
+
+    # And it is claimable again.
+    assert await repo.claim("w2") is not None
+
+
+async def test_fresh_lease_is_not_reset(pool: DictPool) -> None:
+    iid = await make_instance(pool)
+    repo = TaskRepo(pool)
+    await repo.enqueue_standalone(iid, TaskKind.PROVISION)
+    await repo.claim("w1")
+    assert await repo.reset_expired_leases(lease_seconds=300) == 0

tests/integration/test_worker.py

@@ -0,0 +1,203 @@
+"""The worker loop: drains on SIGTERM, retries, stays idempotent.
+
+These run in milliseconds against a FakeProvisioner. That is the payoff for putting helm
+behind a Protocol in Module 5: the crash-safety properties are testable without a cluster.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+
+from services.worker.deps import WorkerDeps
+from services.worker.main import run_worker
+from svcforge_core.adapters.clock import SystemClock
+from svcforge_core.domain.catalog import load_catalog
+from svcforge_core.domain.models import TaskKind
+from svcforge_core.domain.states import InstanceState
+from svcforge_core.repo.db import DictPool
+from svcforge_core.repo.instances import InstanceRepo
+from svcforge_core.repo.tasks import TaskRepo
+from svcforge_core.settings import Settings
+from tests.fakes import FakeNotifier, FakeProvisioner
+from tests.integration.helpers import build_instance
+
+CATALOG = load_catalog(Path(__file__).resolve().parents[2] / "catalog.yaml")
+
+
+def _settings(**over: object) -> Settings:
+    base: dict[str, object] = {
+        "pg_dsn": "postgresql://x:x@127.0.0.1:5432/x",
+        "worker_id": "w-test",
+        "worker_concurrency": 4,
+        "poll_interval_s": 0.05,
+        "max_attempts": 3,
+    }
+    base.update(over)
+    return Settings(**base)  # type: ignore[arg-type]
+
+
+def _deps(
+    pool: DictPool,
+    prov: FakeProvisioner,
+    notifier: FakeNotifier | None = None,
+    **over: object,
+) -> WorkerDeps:
+    return WorkerDeps(
+        pool=pool,
+        instances=InstanceRepo(pool),
+        tasks=TaskRepo(pool),
+        provisioner=prov,
+        notifier=notifier or FakeNotifier(),
+        clock=SystemClock(),
+        catalog=CATALOG,
+        settings=_settings(**over),
+    )
+
+
+async def _seed(pool: DictPool, **kw: object) -> tuple[str, int]:
+    inst = build_instance(**kw)  # type: ignore[arg-type]
+    async with pool.connection() as conn:
+        await InstanceRepo(pool).create(conn, inst)
+    tid = await TaskRepo(pool).enqueue_standalone(inst.id, TaskKind.PROVISION)
+    return str(inst.id), tid
+
+
+async def _state_of(pool: DictPool, tid: int) -> str:
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select state from tasks where id = %s", (tid,))
+        row = await cur.fetchone()
+    assert row is not None
+    return str(row["state"])
+
+
+async def test_worker_provisions_and_marks_ready(pool: DictPool) -> None:
+    iid, tid = await _seed(pool)
+    prov = FakeProvisioner()
+    notifier = FakeNotifier()
+    stop = asyncio.Event()
+
+    worker = asyncio.create_task(run_worker(_deps(pool, prov, notifier), stop))
+    await asyncio.sleep(0.5)
+    stop.set()
+    await asyncio.wait_for(worker, timeout=5)
+
+    assert await _state_of(pool, tid) == "done"
+    inst = await InstanceRepo(pool).get(UUID(iid), team="platform")
+    assert inst is not None
+    assert inst.state is InstanceState.READY
+    assert inst.endpoint is not None
+    assert len(prov.installed) == 1
+
+    # Assert the notification fired, and that it happened on the FIRST attempt.
+    #
+    # Without this the handler could raise after marking the instance ready — the task
+    # requeues, the retry hits the idempotency early-return, and everything above still
+    # passes while the worker is quietly crashing on every provision. Idempotency is
+    # supposed to make crashes survivable, not invisible; asserting attempts==1 is what
+    # keeps a masked crash from reading as success.
+    assert notifier.events() == ["instance.ready"]
+
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select attempts from tasks where id = %s", (tid,))
+        row = await cur.fetchone()
+    assert row is not None
+    assert row["attempts"] == 1, "task was retried: the handler raised after doing the work"
+
+
+async def test_sigterm_drains_in_flight(pool: DictPool) -> None:
+    """Stop is requested mid-provision: the worker must FINISH the task, then exit.
+
+    Abandoning it would not lose the task — the lease would recover it — but only after
+    five minutes of a tenant watching 'provisioning'. Draining costs two seconds.
+    """
+    _, tid = await _seed(pool)
+    prov = FakeProvisioner(delay=2.0)
+    stop = asyncio.Event()
+
+    started = time.monotonic()
+    worker = asyncio.create_task(run_worker(_deps(pool, prov), stop))
+    await asyncio.sleep(0.5)
+    stop.set()  # mid-flight: the handler is still inside its 2s install
+
+    await asyncio.wait_for(worker, timeout=5)
+    elapsed = time.monotonic() - started
+
+    assert await _state_of(pool, tid) == "done", "worker abandoned an in-flight task"
+    assert elapsed >= 2.0, "worker returned before the in-flight task finished"
+    assert elapsed < 5.0
+
+
+async def test_idle_worker_stops_promptly(pool: DictPool) -> None:
+    """Nothing queued: stop must wake the poll sleep, not wait it out."""
+    stop = asyncio.Event()
+    worker = asyncio.create_task(run_worker(_deps(pool, FakeProvisioner(), poll_interval_s=5.0), stop))
+    await asyncio.sleep(0.2)
+    started = time.monotonic()
+    stop.set()
+    await asyncio.wait_for(worker, timeout=2)
+    assert time.monotonic() - started < 1.0, "stop did not interrupt the poll sleep"
+
+
+async def test_failed_task_is_requeued_with_backoff(pool: DictPool) -> None:
+    _, tid = await _seed(pool)
+    prov = FakeProvisioner(fail_on={"platform-elasticsearch"})
+    stop = asyncio.Event()
+
+    worker = asyncio.create_task(run_worker(_deps(pool, prov), stop))
+    await asyncio.sleep(0.6)
+    stop.set()
+    await asyncio.wait_for(worker, timeout=5)
+
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute("select state, attempts, last_error from tasks where id = %s", (tid,))
+        row = await cur.fetchone()
+    assert row is not None
+    assert row["state"] == "queued"  # requeued, not failed — attempts remain
+    assert row["attempts"] >= 1
+    assert row["last_error"]
+
+
+async def test_provision_twice_installs_once(pool: DictPool) -> None:
+    """The idempotency claim, executed.
+
+    Simulates the crash window: the release is installed and the instance is READY, but
+    the task got re-queued (worker died before reporting). Re-running must not re-install.
+    """
+    from services.worker.handlers import handle_provision
+
+    inst = build_instance(state=InstanceState.REQUESTED)
+    async with pool.connection() as conn:
+        await InstanceRepo(pool).create(conn, inst)
+    tid = await TaskRepo(pool).enqueue_standalone(inst.id, TaskKind.PROVISION)
+    task = await TaskRepo(pool).claim("w1")
+    assert task is not None
+
+    prov = FakeProvisioner()
+    deps = _deps(pool, prov)
+
+    await handle_provision(task, deps)
+    await handle_provision(task, deps)  # the redelivery
+
+    assert len(prov.installed) == 1, "second run re-installed: handler is not idempotent"
+    assert tid == task.id
+
+
+@pytest.mark.parametrize("concurrency", [1, 4])
+async def test_concurrency_cap_is_respected(pool: DictPool, concurrency: int) -> None:
+    """The semaphore is what stops one worker from starting 200 helm processes."""
+    for _ in range(6):
+        await _seed(pool)
+
+    prov = FakeProvisioner(delay=0.2)
+    stop = asyncio.Event()
+    worker = asyncio.create_task(run_worker(_deps(pool, prov, worker_concurrency=concurrency), stop))
+    await asyncio.sleep(0.5)
+    stop.set()
+    await asyncio.wait_for(worker, timeout=10)
+
+    assert prov.max_concurrent <= concurrency, f"ran {prov.max_concurrent} at once, cap was {concurrency}"