Phase 0 milestone 0.9 (code): close plan §6/§7 control gaps

Before flipping control-mapping rows, wire the Phase 0 promises that had
no code behind them yet:

- SI-11 / C.5.7.43: app-wide exception handler returns
  {"error":"internal_error","request_id":...} and the X-Request-ID header.
  Logs exception type + stack frames, not the message — driver errors
  echo bound parameters and key values. Engine set to hide_parameters.
- AU-11 / SI-12 (plan §7 risk): create_app refuses to start if any
  mapped table lacks a retention policy.
- CM-3: migrations/IMMUTABLE.sha256 pins every merged revision; a test
  fails on any edit or on an unpinned new revision.
- migrations/env.py: fileConfig(disable_existing_loggers=False) so
  in-process Alembic (the drift test) doesn't silence app loggers.

163 tests passing; ruff, mypy, bandit, gitleaks clean.

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

migrations/IMMUTABLE.sha256

@@ -0,0 +1,8 @@
+# Merged Alembic revisions and their SHA-256. NIST CM-3; CONTRIBUTING rule 1.
+# tests/test_migration_immutability.py fails if a listed file changes or a
+# revision file is missing from this list. Add a line when you add a
+# migration; never change an existing line.
+995b5def7a41c0f8bb319087675dab8dfefb2bd170d89986d07d1fe8c523d538  0001_baseline.py
+70ce3ffc67466544397843391a445790d26049d632e5fda2061a1e393a9c6612  0002_framework_tables.py
+5092d31c78a6c119b992441bd25dafc9b47bf60cf987d6cf20511ff204608d27  0003_registrants.py
+0f9551c9d94921b1b1f1e86beb9594abd2220ad1bbb1215f157cf933c2a70ee6  0004_stub_entity_tables.py

migrations/env.py

@@ -19,7 +19,9 @@ import ypms.models  # noqa: F401 — every model module; see src/ypms/models.py
 
 config = context.config
 if config.config_file_name is not None:
-    fileConfig(config.config_file_name)
+    # Don't disable loggers configured before Alembic runs (app, or pytest when
+    # the drift test invokes migrations in-process).
+    fileConfig(config.config_file_name, disable_existing_loggers=False)
 
 config.set_main_option("sqlalchemy.url", get_settings().database_url)
 target_metadata = Base.metadata

src/ypms/db.py

@@ -19,7 +19,10 @@ _SessionLocal: sessionmaker[Session] | None = None
 def engine() -> Engine:
     global _engine
     if _engine is None:
-        _engine = create_engine(get_settings().database_url, pool_pre_ping=True, future=True)
+        # hide_parameters: keep bound values (PII) out of exception text and logs.
+        _engine = create_engine(
+            get_settings().database_url, pool_pre_ping=True, future=True, hide_parameters=True
+        )
     return _engine
 
 

src/ypms/framework/retention.py

@@ -132,6 +132,12 @@ def mapped_models() -> list[type[Any]]:
     return sorted((m.class_ for m in Base.registry.mappers), key=lambda c: c.__tablename__)
 
 
+def missing_policies() -> list[str]:
+    """Tables with no resolvable policy. Empty in a correct build; `create_app`
+    refuses to start otherwise (PHASE0_PLAN §7: the scaffold can't be forgotten)."""
+    return [m.__tablename__ for m in mapped_models() if resolve_policy(m) is None]
+
+
 def _count_past(db: Session, model: type[Any], policy: RetentionPolicy, now: datetime) -> int:
     if policy.period_days is None:
         raise ValueError(f"policy {policy.data_class!r} has no period")

src/ypms/main.py

@@ -8,12 +8,16 @@ subsequent milestones.
 
 from __future__ import annotations
 
-from fastapi import FastAPI, status
+import traceback
+
+from fastapi import FastAPI, Request, status
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 from sqlalchemy import text
 
 from ypms.auth.router import router as auth_router
 from ypms.db import engine
+from ypms.framework.retention import missing_policies
 from ypms.logging import configure_logging, get_logger
 from ypms.middleware import RequestIdMiddleware
 from ypms.settings import get_settings
@@ -26,6 +30,11 @@ def create_app() -> FastAPI:
     log = get_logger(__name__)
     settings = get_settings()
 
+    # NIST AU-11 / SI-12: every table carries a retention policy, or no boot.
+    missing = missing_policies()
+    if missing:
+        raise RuntimeError(f"tables without a retention policy: {missing}")
+
     app = FastAPI(
         title="YPMS — Youth Program Management System",
         version="0.1.0",
@@ -43,6 +52,26 @@ def create_app() -> FastAPI:
         expose_headers=["X-Request-ID"],
     )
 
+    @app.exception_handler(Exception)
+    async def unhandled(request: Request, exc: Exception) -> JSONResponse:
+        """NIST SI-11 / source RFP C.5.7.43: no stack traces, DB messages, or
+        versions to the caller. The log gets the exception type and stack frames
+        under the request id, but not the message: driver errors echo bound
+        parameters and key values (names, DOB, SSN ciphertext)."""
+        req_id = getattr(request.state, "req_id", None)
+        frames = [f"{f.filename}:{f.lineno} {f.name}" for f in traceback.extract_tb(exc.__traceback__)]
+        log.error(
+            "request.unhandled_exception",
+            req_id=req_id, path=request.url.path, error_type=type(exc).__name__,
+            frames=frames,
+        )
+        headers = {RequestIdMiddleware.HEADER: req_id} if req_id else None
+        return JSONResponse(
+            {"error": "internal_error", "request_id": req_id},
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            headers=headers,
+        )
+
     app.include_router(auth_router)
     app.include_router(registrant_router)
     for stub in stub_routers():

tests/test_app_guards.py

@@ -0,0 +1,49 @@
+"""Startup and error-surface guards in `create_app` (NIST SI-11, AU-11)."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+import ypms.main
+from ypms.main import create_app
+
+
+def test_unhandled_exception_is_opaque_and_carries_request_id() -> None:
+    app = create_app()
+
+    @app.get("/_boom")
+    def boom() -> None:
+        raise RuntimeError("DETAIL: Key (email)=(kid@example.org) already exists")
+
+    with TestClient(app, raise_server_exceptions=False) as c:
+        r = c.get("/_boom", headers={"X-Request-ID": "req-abc123"})
+
+    assert r.status_code == 500
+    assert r.json() == {"error": "internal_error", "request_id": "req-abc123"}
+    assert r.headers["X-Request-ID"] == "req-abc123"
+    assert "kid@example.org" not in r.text
+    assert "Traceback" not in r.text
+
+
+def test_unhandled_exception_log_omits_message(caplog: pytest.LogCaptureFixture) -> None:
+    app = create_app()
+
+    @app.get("/_boom")
+    def boom() -> None:
+        raise RuntimeError("parameters: ('123456789', 'kid@example.org')")
+
+    with TestClient(app, raise_server_exceptions=False) as c:
+        c.get("/_boom")
+    logged = "\n".join(r.getMessage() for r in caplog.records)
+    assert "request.unhandled_exception" in logged
+    assert "RuntimeError" in logged
+    assert "kid@example.org" not in logged and "123456789" not in logged
+
+
+def test_app_refuses_to_start_without_retention_policies(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(ypms.main, "missing_policies", lambda: ["orphans"])
+    with pytest.raises(RuntimeError, match="orphans"):
+        create_app()

tests/test_migration_immutability.py

@@ -0,0 +1,34 @@
+"""NIST CM-3: merged migrations never change (CONTRIBUTING rule 1).
+
+`migrations/IMMUTABLE.sha256` pins every revision file. Editing a merged
+migration means editing that manifest too — visible in review, and the reason
+belongs in the commit message.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+VERSIONS = Path(__file__).resolve().parents[1] / "migrations" / "versions"
+MANIFEST = VERSIONS.parent / "IMMUTABLE.sha256"
+
+
+def _manifest() -> dict[str, str]:
+    out: dict[str, str] = {}
+    for line in MANIFEST.read_text().splitlines():
+        if line.strip() and not line.startswith("#"):
+            digest, name = line.split()
+            out[name] = digest
+    return out
+
+
+def test_every_revision_is_pinned() -> None:
+    files = {p.name for p in VERSIONS.glob("*.py")}
+    assert files - set(_manifest()) == set(), "add new migrations to IMMUTABLE.sha256"
+
+
+def test_pinned_revisions_unchanged() -> None:
+    for name, digest in _manifest().items():
+        actual = hashlib.sha256((VERSIONS / name).read_bytes()).hexdigest()
+        assert actual == digest, f"{name} changed after merge — add a new revision instead"