feat(deploy): run migrations as part of the release, and report schema state

Central was serving two revisions behind the code it shipped: review_contracts
did not exist there although its migration was inside the running image. There
was no migration mechanism at all — bare uvicorn CMD, nothing chart-declared —
and nothing surfaced the mismatch. The API starts happily against a schema it
was not built for and only fails when a request touches a missing table.

Adds a chart-managed Helm pre-install/pre-upgrade hook running alembic upgrade
head, weighted to complete before the API rolls. A hook rather than an init
container: init containers run per pod, so more than one replica means
concurrent alembic upgrade with no locking. Failed jobs are deliberately
retained — a migration that fails and vanishes is how this drifted in the first
place.

/state/health now reports applied and expected revisions. "unknown" is
deliberately not "ok": an instance that cannot establish agreement must not
claim it, the same principle as instance_role defaulting to unknown.

Refs STATE-WP-0083-T07

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2583210@bnt-lap001
Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006

api/routers/state.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import noload, selectinload
 
 from api.config import settings
 from api.database import get_session
+from api.services.schema_state import schema_state
 from api.flow_defs import assertion_result_to_dict, load_flow
 from api.models.capability_request import CapabilityRequest
 from api.models.contribution import Contribution, ContributionStatus, ContributionType
@@ -1115,6 +1116,9 @@ async def health_check(session: AsyncSession = Depends(get_session)) -> dict:
             # Identity, so a caller can verify it reached the hub it meant to.
             "instance_role": settings.state_hub_instance_role,
             "instance_label": settings.state_hub_instance_label,
+            # Surfaced, not merely logged: a schema behind the code is a fault
+            # the operator must be able to see (STATE-WP-0083-T07).
+            "schema": await schema_state(session),
         }
     except Exception as exc:
         return JSONResponse(

api/services/schema_state.py

@@ -0,0 +1,63 @@
+"""Report whether the database schema matches the code (STATE-WP-0083-T07).
+
+Central ran two revisions behind the code it was serving, and `review_contracts`
+did not exist there although its migration shipped inside the running image.
+Nothing surfaced that: the API starts happily against a schema it was not built
+for, and only fails when a request happens to touch the missing table.
+
+A hub that cannot say which schema it is running has the same problem as a
+projection that cannot name its source commit — it is asserting correctness it
+cannot demonstrate.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+_MIGRATIONS = Path(__file__).resolve().parents[2] / "migrations"
+
+
+@lru_cache(maxsize=1)
+def code_head_revision() -> str | None:
+    """The head revision the shipped migration scripts define.
+
+    Read from the migration files rather than the database: this is what the
+    code expects, and it must be knowable without a working connection.
+    """
+    try:
+        from alembic.config import Config
+        from alembic.script import ScriptDirectory
+
+        cfg = Config()
+        cfg.set_main_option("script_location", str(_MIGRATIONS))
+        heads = ScriptDirectory.from_config(cfg).get_heads()
+        return heads[0] if len(heads) == 1 else ",".join(sorted(heads)) or None
+    except Exception:
+        return None
+
+
+async def db_revision(session: AsyncSession) -> str | None:
+    try:
+        result = await session.execute(text("select version_num from alembic_version"))
+        return result.scalar()
+    except Exception:
+        return None
+
+
+async def schema_state(session: AsyncSession) -> dict[str, Any]:
+    applied = await db_revision(session)
+    expected = code_head_revision()
+    # "unknown" is deliberately not "ok": an instance that cannot determine its
+    # own schema state must not report agreement it has not established.
+    if applied is None or expected is None:
+        status = "unknown"
+    elif applied == expected:
+        status = "ok"
+    else:
+        status = "behind"
+    return {"status": status, "applied": applied, "expected": expected}

deploy/railiance/apps/charts/state-hub/templates/migrate-job.yaml

@@ -0,0 +1,40 @@
+{{- if .Values.migrations.enabled }}
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ include "statehub.fullname" . }}-migrate
+  labels: {{- include "statehub.labels" . | nindent 4 }}
+  annotations:
+    # Run before the API starts serving, and before an upgrade swaps the image.
+    # A deployment that can serve against a schema it was not built for is the
+    # same class of defect as a projection that cannot name its source commit
+    # (STATE-WP-0083-T07).
+    "helm.sh/hook": pre-install,pre-upgrade
+    "helm.sh/hook-weight": "-5"
+    # Keep a failed job for inspection; a silent migration failure is how the
+    # schema drifted two revisions behind the code in the first place.
+    "helm.sh/hook-delete-policy": before-hook-creation,hook-succeeded
+spec:
+  backoffLimit: {{ .Values.migrations.backoffLimit }}
+  template:
+    metadata:
+      labels: {{- include "statehub.labels" . | nindent 8 }}
+    spec:
+      restartPolicy: Never
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets: {{- toYaml . | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: migrate
+          image: {{ include "statehub.image" . | quote }}
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          command: ["/app/.venv/bin/python", "-m", "alembic", "upgrade", "head"]
+          envFrom:
+            {{- if .Values.config.enabled }}
+            - configMapRef:
+                name: {{ .Values.config.name | quote }}
+            {{- end }}
+            - secretRef:
+                name: {{ .Values.secret.name | quote }}
+          resources: {{- toYaml .Values.migrations.resources | nindent 12 }}
+{{- end }}

deploy/railiance/apps/charts/state-hub/values.yaml

@@ -51,6 +51,21 @@ ingress:
     traefik.ingress.kubernetes.io/router.tls: "true"
     cert-manager.io/cluster-issuer: letsencrypt-prod
 
+# Database migrations (STATE-WP-0083-T07). Runs as a Helm pre-install/pre-upgrade
+# hook rather than an init container: an init container runs per pod, so more
+# than one replica means concurrent `alembic upgrade` with no locking. A hook
+# runs once per release and fails the upgrade if the migration fails.
+migrations:
+  enabled: true
+  backoffLimit: 1
+  resources:
+    requests:
+      cpu: 50m
+      memory: 256Mi
+    limits:
+      cpu: 500m
+      memory: 1Gi
+
 # Classification allowed-values (CUST-WP-0067-T09). The API validates repo
 # classification against the-custodian canon; a container has no such checkout,
 # so the file travels with the release as a ConfigMap. Without it every

tests/test_schema_state.py

@@ -0,0 +1,56 @@
+"""Schema/code agreement must be observable (STATE-WP-0083-T07).
+
+Central served two revisions behind the code with no signal at all. The API
+started fine and would only have failed when a request touched a missing table.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from api.services import schema_state as ss
+
+
+class _Session:
+    def __init__(self, revision=None, raises=False):
+        self._revision = revision
+        self._raises = raises
+
+    async def execute(self, *_a, **_k):
+        if self._raises:
+            raise RuntimeError("no connection")
+        rev = self._revision
+
+        class R:
+            def scalar(self_inner):
+                return rev
+
+        return R()
+
+
+def test_code_head_is_readable_from_the_shipped_migrations():
+    """Must not require a database: it is what the code expects, not what ran."""
+    assert ss.code_head_revision()
+
+
+@pytest.mark.asyncio
+async def test_matching_revision_reports_ok(monkeypatch):
+    monkeypatch.setattr(ss, "code_head_revision", lambda: "abc123")
+    assert (await ss.schema_state(_Session("abc123")))["status"] == "ok"
+
+
+@pytest.mark.asyncio
+async def test_behind_revision_is_reported_with_both_values(monkeypatch):
+    monkeypatch.setattr(ss, "code_head_revision", lambda: "newrev")
+    state = await ss.schema_state(_Session("oldrev"))
+    assert state["status"] == "behind"
+    assert state["applied"] == "oldrev" and state["expected"] == "newrev"
+
+
+@pytest.mark.asyncio
+async def test_unknown_is_not_reported_as_ok(monkeypatch):
+    """An instance that cannot establish agreement must not claim it."""
+    monkeypatch.setattr(ss, "code_head_revision", lambda: "newrev")
+    assert (await ss.schema_state(_Session(raises=True)))["status"] == "unknown"
+    monkeypatch.setattr(ss, "code_head_revision", lambda: None)
+    assert (await ss.schema_state(_Session("oldrev")))["status"] == "unknown"