Upload folder using huggingface_hub

api/atp.py

@@ -0,0 +1,429 @@
+"""
+ATP (Agentic Training Platform) API routes — see docs/ATP.md §5.
+
+Mounted by api/server.py:
+    from api.atp import router as atp_router
+    app.include_router(atp_router)
+
+Endpoints (all bearer-gated by the default auth middleware — api/auth.py):
+    GET  /atp/data              — full ATP_DATA blob (T0 public seed)
+    GET  /atp/agents            — agent roster
+    GET  /atp/agents/{agent_id} — one agent profile (404 if unknown)
+    GET  /atp/agents/{agent_id}/report.pdf — official report-card PDF
+                                  (Phase 6; any authenticated role;
+                                  live awards org-scoped; 404 if unknown)
+    GET  /atp/certs             — certification catalog
+    GET  /atp/certs/{cert_id}   — one cert (404 if unknown)
+    GET  /atp/standard          — the ATP standard (label grammar + principles)
+    GET  /atp/marketplace       — marketplace listings
+    GET  /atp/hitl              — HITL reward log, latest first (org-scoped)
+    POST /atp/hitl              — body {agentId, layer, signal, reason, rater}
+                                  → appended (enriched) event  [admin|sme]
+    GET  /atp/requests          — expert-composition requests (org-scoped)
+    POST /atp/requests          — body {major, specialty, badgeIds, packId}
+                                  → appended request           [admin|sme]
+
+Certification engine (Phase 3 — docs/HARDENING.md):
+    POST /atp/exams/run         — body {certId, candidateSpec?, agentId?,
+                                  dryRun=true} → {jobId} (poll /jobs/{id};
+                                  the job result is the award dict) [admin|sme]
+    GET  /atp/awards            — org-scoped cert awards, latest first
+    GET  /atp/evidence/{id}     — one signed evidence row (org-scoped; 404
+                                  cross-tenant/absent — no existence oracle)
+    GET  /atp/evidence/{id}/verify — {ok, sigValid, chainPosition}
+    GET  /atp/chain/verify      — full evidence-chain verification for the org
+
+Tenancy (Phase 2 — docs/TENANCY.md):
+  * The seed catalog (data/agents/certs/standard/marketplace) is T0 public:
+    read-only, identical for every org — never merged with tenant rows.
+  * atp_hitl / atp_requests are T1 org-internal: reads and writes go through
+    the store with org_id taken ONLY from the verified request state
+    (request.state.org_id, set by the auth middleware from the JWT) — never
+    from a client-supplied body/query value. Fallback 'org-demo' covers
+    BU_AUTH_DISABLED dev mode and legacy single-admin tokens.
+  * The demo seed rows (RL.hitlLog / EXPERT_REQUESTS) belong to 'org-demo'
+    (TENANCY.md leak surface #1), so they are appended only for that org.
+  * Writes require role admin|sme via api.auth.require_role; GET endpoints
+    are open to any authenticated role (viewer included). Under
+    BU_AUTH_DISABLED=1 the middleware runs every request as org-demo/admin,
+    so dev mode passes every role gate (pre-Phase-2 behavior preserved).
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from pydantic import BaseModel
+
+# require_role(*roles) → FastAPI dependency that 403s when the verified
+# session role is not in `roles` (401 when the middleware never ran).
+from api.auth import require_role
+from atp import store, tenant_db
+
+router = APIRouter(prefix="/atp")
+
+
+# ── Tenancy helpers ────────────────────────────────────────────────────────
+
+def _org(request: Request) -> str:
+    """Org for this request — ONLY from verified auth state (TENANCY.md §4
+    layer 1). 'org-demo' when auth is disabled / legacy single-admin token."""
+    return getattr(request.state, "org_id", None) or "org-demo"
+
+
+# ── Read endpoints — T0 public seed ────────────────────────────────────────
+
+@router.get("/data")
+def atp_data():
+    """Full ATP_DATA blob (same shape as window.ATP_DATA, minus JS helpers).
+    T0 public seed — never merge tenant rows into it (TENANCY.md #9)."""
+    return store.get_data()
+
+
+@router.get("/agents")
+def atp_agents():
+    return {"agents": store.get_data().get("AGENTS", [])}
+
+
+@router.get("/agents/{agent_id}")
+def atp_agent(agent_id: str):
+    for a in store.get_data().get("AGENTS", []):
+        if a.get("id") == agent_id:
+            return a
+    raise HTTPException(404, f"agent {agent_id} not found")
+
+
+@router.get("/agents/{agent_id}/report.pdf")
+def atp_agent_report_pdf(agent_id: str, request: Request):
+    """Official report-card PDF (Phase 6, docs/HARDENING.md).
+
+    Any authenticated role — the PDF is built from the T0 public seed plus
+    the CALLER ORG's live cert awards (org_id from verified request state
+    only, TENANCY.md layer 1; atp/reportcard.py reads them through
+    atp/tenant_db.py). Served as an attachment download; 404 for agents
+    not in the seed roster. Deliberately NOT under /videos or /media —
+    those prefixes are public-cached (TENANCY.md leak surface #4).
+    """
+    # Lazy import: reportlab stays an on-demand dependency — the server
+    # boots (and every other route works) even if it is not installed.
+    from atp import reportcard
+
+    try:
+        pdf = reportcard.build_report_card_pdf(agent_id, org_id=_org(request))
+    except reportcard.UnknownAgentError:
+        raise HTTPException(404, f"agent {agent_id} not found")
+    return Response(
+        content=pdf,
+        media_type="application/pdf",
+        headers={
+            "Content-Disposition":
+                f'attachment; filename="agent-{agent_id}-report-card.pdf"',
+        },
+    )
+
+
+@router.get("/certs")
+def atp_certs():
+    return {"certs": store.get_data().get("CERTS", [])}
+
+
+@router.get("/certs/{cert_id}")
+def atp_cert(cert_id: str):
+    for c in store.get_data().get("CERTS", []):
+        if c.get("id") == cert_id:
+            return c
+    raise HTTPException(404, f"cert {cert_id} not found")
+
+
+@router.get("/standard")
+def atp_standard():
+    return store.get_data().get("STANDARD", {})
+
+
+@router.get("/marketplace")
+def atp_marketplace():
+    return store.get_data().get("MARKETPLACE", {"listings": []})
+
+
+# ── HITL reward log — T1 org-internal ──────────────────────────────────────
+
+@router.get("/hitl")
+def atp_hitl_log(request: Request, limit: int = 100):
+    """HITL reward log, latest first: the requesting org's live events
+    (store.read_hitl already returns latest-first) followed by the seed audit
+    history (stored oldest → newest, so reversed here). The seed rows belong
+    to 'org-demo' only — other tenants never see them."""
+    org = _org(request)
+    live = store.read_hitl(limit=limit, org_id=org)
+    seed = store.get_data().get("RL", {}).get("hitlLog", []) if org == "org-demo" else []
+    return {"events": (live + list(reversed(seed)))[:limit]}
+
+
+class HitlBody(BaseModel):
+    agentId: str
+    layer: int
+    signal: int
+    reason: str = ""
+    rater: str = "anon"
+
+
+@router.post("/hitl", dependencies=[Depends(require_role("admin", "sme"))])
+def atp_hitl_submit(body: HitlBody, request: Request):
+    """Append a human reward signal (role admin|sme). store.append_hitl
+    computes weightedDelta per the RL weighting in docs/ATP.md §1 and logs it
+    append-only under the caller's org."""
+    if not 1 <= body.layer <= 7:
+        raise HTTPException(400, "layer must be an integer in 1..7")
+    if body.signal not in (1, -1):
+        raise HTTPException(400, "signal must be 1 or -1")
+    return store.append_hitl({
+        "agentId": body.agentId,
+        "layer":   body.layer,
+        "signal":  body.signal,
+        "reason":  body.reason,
+        "rater":   body.rater,
+    }, org_id=_org(request))
+
+
+# ── Expert composition / train-to-order (docs/ATP.md §7) ───────────────────
+
+@router.get("/requests")
+def atp_requests(request: Request, limit: int = 100):
+    """Expert-composition requests: the requesting org's live commissions
+    (store.read_requests already returns latest-first) followed by the seed
+    board — which belongs to 'org-demo' only."""
+    org = _org(request)
+    live = store.read_requests(limit=limit, org_id=org)
+    seed = store.get_data().get("EXPERT_REQUESTS", []) if org == "org-demo" else []
+    return {"requests": live + seed}
+
+
+class ComposeBody(BaseModel):
+    major: str
+    specialty: str = ""
+    badgeIds: list[str] = []
+    packId: str | None = None
+
+
+@router.post("/requests", dependencies=[Depends(require_role("admin", "sme"))])
+def atp_request_submit(body: ComposeBody, request: Request):
+    """Commission an expert (role admin|sme — viewers are read-only,
+    TENANCY.md §Principals). store.append_request runs the matching rule
+    against seed agents, builds the pipeline stages, and appends the request
+    to the atp_requests table under the caller's org."""
+    return store.append_request({
+        "major":     body.major,
+        "specialty": body.specialty,
+        "badgeIds":  body.badgeIds,
+        "packId":    body.packId,
+    }, org_id=_org(request))
+
+
+# ── Certification engine (Phase 3 — docs/HARDENING.md) ─────────────────────
+#
+# Exam runs execute in the background (agents/jobs — they call the candidate
+# AND the judge model, 30-120s live) and write signed, chained rows into the
+# append-only atp_evidence / atp_cert_awards tables (migration 002; org_id +
+# RLS from 005). atp/exams.py + atp/signing.py own the exam/signing logic;
+# these routes are the org-scoped HTTP surface over them. Reads go through
+# atp/tenant_db.py ONLY (TENANCY.md layer 2) with org_id from verified
+# request state — never from the client payload.
+
+_DB_READY = False
+_DB_LOCK = threading.Lock()
+_EVIDENCE_COLNAMES: set[str] | None = None
+
+# camelCase key ↔ snake_case column, same mapping idiom as atp/store.py.
+_AWARD_COLS = {
+    "id":            "id",
+    "ts":            "ts",
+    "agentId":       "agent_id",
+    "certId":        "cert_id",
+    "score":         "score",
+    "sectionScores": "section_scores",
+    "itemBreakdown": "item_breakdown",
+    "evidenceIds":   "evidence_ids",
+}
+_AWARD_JSON_COLS = {"section_scores", "item_breakdown", "evidence_ids"}
+
+_EVIDENCE_COLS = {
+    "id":       "id",
+    "ts":       "ts",
+    "agentId":  "agent_id",
+    "certId":   "cert_id",
+    "kind":     "kind",
+    "payload":  "payload",
+    "sig":      "sig",
+    "prevHash": "prev_hash",
+}
+
+
+def _ensure_db() -> None:
+    """Engine + migrations up before the first direct tenant_db read here.
+    run_migrations() is idempotent (schema_migrations bookkeeping), so this
+    is a no-op when the server startup / atp.store already ran it."""
+    global _DB_READY
+    if _DB_READY:
+        return
+    with _DB_LOCK:
+        if not _DB_READY:
+            from atp import db
+            db.run_migrations()
+            _DB_READY = True
+
+
+def _loads_maybe(v):
+    """json.loads a JSON-text column, tolerating NULL / non-JSON text."""
+    if v is None or not isinstance(v, str):
+        return v
+    try:
+        return json.loads(v)
+    except (json.JSONDecodeError, ValueError):
+        return v
+
+
+def _row_to_camel(row: dict, cols: dict, json_cols: set = frozenset()) -> dict:
+    return {key: (_loads_maybe(row.get(col)) if col in json_cols else row.get(col))
+            for key, col in cols.items()}
+
+
+class ExamRunBody(BaseModel):
+    certId: str
+    candidateSpec: str | None = None   # agents/backend.get_backend spec string
+    agentId: str | None = None
+    dryRun: bool = True                # deterministic stub model (CI-safe)
+
+
+@router.post("/exams/run", dependencies=[Depends(require_role("admin", "sme"))])
+def atp_exam_run(body: ExamRunBody, request: Request):
+    """Enqueue a certification exam run (role admin|sme).
+
+    Returns {jobId}; poll GET /jobs/{jobId} — when status='done' the job
+    result is the award dict (score, sectionScores, itemBreakdown,
+    evidenceIds, passed…). The run itself enforces the ATP standard promises
+    (docs/ATP.md §2) inside atp/exams.py: judge/candidate model-family
+    separation and recorded seed/temperature/run-count reproducibility.
+    dryRun (default true) uses the deterministic stub model — no live LLM.
+    """
+    from agents import jobs
+
+    org = _org(request)
+    cert_id = body.certId
+    if not any(c.get("id") == cert_id
+               for c in store.get_data().get("CERTS", [])):
+        raise HTTPException(404, f"cert {cert_id} not found")
+
+    candidate_spec = body.candidateSpec
+    agent_id = body.agentId
+    dry_run = body.dryRun
+
+    def _task():
+        from atp import exams
+        return exams.run_exam(
+            cert_id,
+            candidate_spec,
+            org_id=org,
+            dry_run=dry_run,
+            agent_id=agent_id,
+        )
+
+    job_id = jobs.submit("atp_exam", _task, org_id=org)
+    # jobId is the documented key; job_id keeps BU_API.runJob() compatible.
+    return {"jobId": job_id, "job_id": job_id}
+
+
+@router.get("/awards")
+def atp_awards(request: Request, limit: int = 100):
+    """Org-scoped cert awards, latest first (any authenticated role)."""
+    org = _org(request)
+    _ensure_db()
+    sel = ", ".join(_AWARD_COLS.values())
+    rows = tenant_db.scoped_query(
+        org,
+        f"SELECT {sel} FROM atp_cert_awards"
+        f" WHERE org_id = :org ORDER BY id DESC LIMIT :n",
+        {"org": org, "n": int(limit)})
+    return {"awards": [_row_to_camel(r, _AWARD_COLS, _AWARD_JSON_COLS)
+                       for r in rows]}
+
+
+def _evidence_where() -> str:
+    """Row-match clause for a path id. The surrogate pk is an integer
+    (BIGSERIAL / AUTOINCREMENT — migration 002) so match its text form; if
+    the schema also carries a string business id (evidence_id), accept that
+    too so award.evidenceIds resolve regardless of which form they use."""
+    global _EVIDENCE_COLNAMES
+    if _EVIDENCE_COLNAMES is None:
+        from sqlalchemy import inspect as sa_inspect
+
+        from atp import db
+        _EVIDENCE_COLNAMES = {
+            c["name"] for c in sa_inspect(db.get_engine()).get_columns("atp_evidence")}
+    if "evidence_id" in _EVIDENCE_COLNAMES:
+        return "(evidence_id = :eid OR CAST(id AS TEXT) = :eid)"
+    return "CAST(id AS TEXT) = :eid"
+
+
+def _evidence_row(org: str, evidence_id: str) -> dict | None:
+    """One org-scoped evidence row (raw snake_case columns), or None."""
+    _ensure_db()
+    rows = tenant_db.scoped_query(
+        org,
+        f"SELECT * FROM atp_evidence"
+        f" WHERE org_id = :org AND {_evidence_where()} LIMIT 1",
+        {"org": org, "eid": str(evidence_id)})
+    return rows[0] if rows else None
+
+
+@router.get("/evidence/{evidence_id}")
+def atp_evidence(evidence_id: str, request: Request):
+    """One signed evidence row. Cross-tenant ids 404 with the SAME message
+    as unknown ids — no existence oracle (TENANCY.md #6)."""
+    row = _evidence_row(_org(request), evidence_id)
+    if row is None:
+        raise HTTPException(404, f"evidence {evidence_id} not found")
+    return _row_to_camel(row, _EVIDENCE_COLS, {"payload"})
+
+
+@router.get("/evidence/{evidence_id}/verify")
+def atp_evidence_verify(evidence_id: str, request: Request):
+    """Verify one evidence row: {ok, sigValid, chainPosition}.
+
+    sigValid       — HMAC signature over the row's canonical payload checks
+                     out (atp.signing.verify_row).
+    chainPosition  — 1-based position of the row in the org's evidence chain.
+    ok             — sigValid AND the whole org chain verifies (a valid row
+                     inside a tampered ledger is NOT ok).
+    """
+    from atp import signing
+
+    org = _org(request)
+    row = _evidence_row(org, evidence_id)
+    if row is None:
+        raise HTTPException(404, f"evidence {evidence_id} not found")
+
+    sig_valid = bool(signing.verify_row(row))
+    position = tenant_db.scoped_query(
+        org,
+        "SELECT COUNT(*) AS n FROM atp_evidence"
+        " WHERE org_id = :org AND id <= :rid",
+        {"org": org, "rid": row["id"]})[0]["n"]
+    chain = signing.verify_chain(org)
+    chain_ok = bool(chain.get("ok")) if isinstance(chain, dict) else bool(chain)
+    return {"ok": sig_valid and chain_ok,
+            "sigValid": sig_valid,
+            "chainPosition": position}
+
+
+@router.get("/chain/verify")
+def atp_chain_verify(request: Request):
+    """Verify the caller org's whole evidence chain (sig + prev_hash links).
+    Returns atp.signing.verify_chain's result dict (tamper-evidence promise,
+    docs/HARDENING.md Phase 3)."""
+    from atp import signing
+
+    _ensure_db()
+    result = signing.verify_chain(_org(request))
+    return result if isinstance(result, dict) else {"ok": bool(result)}

api/auth.py

@@ -1,22 +1,34 @@
 """
 Bearer-token authentication for the Brain University FastAPI app.
 
-Token format (URL-safe base64):
-    <username>|<expiry_unix>|<hmac_sha256(username|expiry, BU_SECRET)>
+Two token formats are accepted (Phase 2, docs/TENANCY.md):
 
-Stateless, no DB. Verification recomputes the HMAC and checks expiry.
+  * Org session JWT (api/identity.py) — HS256 over BU_AUTH_SECRET, claims
+    {sub, org_id, role, exp}. Tried FIRST by verify_token().
+  * Legacy single-admin HMAC token (URL-safe base64):
+        <username>|<expiry_unix>|<hmac_sha256(username|expiry, BU_SECRET)>
+    Fallback — kept for backward compatibility with the deployed demo; a
+    legacy token maps to the bootstrap org 'org-demo' with role 'admin'.
+
+Stateless, no DB. Verification recomputes the HMAC / JWT signature and
+checks expiry. On success the middleware sets request.state.user /
+.user_id / .org_id / .role; route handlers gate roles with require_role().
 
 Env vars (read at process start):
     BU_AUTH_USER       — single allowed username (default "admin")
-    BU_AUTH_PASSWORD   — required; refusing to start without it in production
-    BU_AUTH_SECRET     — HMAC key; auto-generated random fallback if unset
-                         (tokens invalidate on restart, fine for single-instance)
+    BU_AUTH_PASSWORD   — REQUIRED; process refuses to start without it
+                         (unless BU_AUTH_DISABLED=1)
+    BU_AUTH_SECRET     — HMAC key; REQUIRED; process refuses to start without
+                         it (unless BU_AUTH_DISABLED=1, which uses an ephemeral
+                         random key — tokens invalidate on restart)
     BU_AUTH_TTL_HOURS  — token lifetime, default 24
-    BU_AUTH_DISABLED   — "1" to bypass auth entirely (dev only)
+    BU_AUTH_DISABLED   — "1" to bypass auth entirely (dev only; used by
+                         scripts/demo_start.sh). Requests run as
+                         org-demo/admin so org-scoped routes keep working.
 
 Routes protected by middleware are everything except an allowlist:
-    /health, /login, /docs, /openapi.json, /redoc, /videos/{...}, /media/{...},
-    OPTIONS preflight.
+    /health, /login, /auth/login, /auth/oidc/{login,callback}, /docs,
+    /openapi.json, /redoc, /videos/{...}, /media/{...}, OPTIONS preflight.
 """
 
 from __future__ import annotations
@@ -26,8 +38,9 @@ import hashlib
 import hmac
 import os
 import secrets
+import sys
 import time
-import warnings
+from typing import NamedTuple
 
 from fastapi import HTTPException, Request
 
@@ -40,29 +53,71 @@ AUTH_SECRET   = os.environ.get("BU_AUTH_SECRET", "").encode("utf-8")
 AUTH_TTL_S    = int(os.environ.get("BU_AUTH_TTL_HOURS", "24")) * 3600
 AUTH_DISABLED = os.environ.get("BU_AUTH_DISABLED", "") == "1"
 
-if not AUTH_SECRET:
-    AUTH_SECRET = secrets.token_bytes(32)
-    warnings.warn(
-        "BU_AUTH_SECRET not set; using ephemeral random key. Tokens invalidate "
-        "on restart. Set BU_AUTH_SECRET to a 32+ byte random string in prod.",
-        RuntimeWarning,
-    )
-
-if not AUTH_PASSWORD and not AUTH_DISABLED:
-    # Soft-fail: keep dev usable but loudly warn. Production deploy scripts
-    # MUST set BU_AUTH_PASSWORD.
-    AUTH_PASSWORD = "password"
-    warnings.warn(
-        "BU_AUTH_PASSWORD not set; defaulting to 'admin/password'. "
-        "Set BU_AUTH_PASSWORD before exposing this server publicly.",
-        RuntimeWarning,
+if AUTH_DISABLED:
+    # Explicit dev/demo opt-in (scripts/demo_start.sh, .claude/launch.json).
+    # Keep usable defaults so the login gate (which always POSTs /login)
+    # works in demo mode without any env setup.
+    print(
+        "=" * 72 + "\n"
+        "WARNING: BU_AUTH_DISABLED=1 — AUTHENTICATION IS BYPASSED (dev/demo\n"
+        "only). The login gate accepts the dev default credentials, and the\n"
+        "HMAC key is ephemeral (all tokens invalidate on restart). NEVER set\n"
+        "BU_AUTH_DISABLED in production. Unset it and provide\n"
+        "BU_AUTH_PASSWORD + BU_AUTH_SECRET to re-enable auth.\n"
+        + "=" * 72,
+        file=sys.stderr,
     )
+    if not AUTH_SECRET:
+        AUTH_SECRET = secrets.token_bytes(32)  # ephemeral; fine for dev
+    if not AUTH_PASSWORD:
+        AUTH_PASSWORD = "password"  # dev-only default; enabled path has NO default
+else:
+    _missing = [name for name, val in (
+        ("BU_AUTH_PASSWORD", AUTH_PASSWORD),
+        ("BU_AUTH_SECRET", AUTH_SECRET),
+    ) if not val]
+    if _missing:
+        raise RuntimeError(
+            "Refusing to start: required auth env var(s) unset: "
+            + ", ".join(_missing)
+            + ". Set them in the deployment environment (see render.yaml / "
+            "DEPLOY.md), or export BU_AUTH_DISABLED=1 for local dev only "
+            "(scripts/demo_start.sh does this)."
+        )
+    if len(AUTH_SECRET) < 32:
+        # HS256 signing key — short keys are brute-forceable. Same fail-loud
+        # posture as the missing-var check above.
+        raise RuntimeError(
+            "Refusing to start: BU_AUTH_SECRET is shorter than 32 bytes. "
+            "Generate one with: python3 -c \"import secrets; "
+            "print(secrets.token_urlsafe(48))\""
+        )
 
 
 # ── Token mint / verify ────────────────────────────────────────────────────
 
+# Bootstrap org for legacy single-admin tokens (seeded by migrations/004).
+LEGACY_ORG_ID = "org-demo"
+LEGACY_ROLE = "admin"
+
+
+class Principal(NamedTuple):
+    """Verified caller identity, copied onto request.state by the middleware.
+
+    user    — display identifier (org users: user id / legacy: username)
+    user_id — users.user_id for org sessions; None for legacy tokens
+    org_id  — tenant id (legacy tokens map to 'org-demo')
+    role    — admin | sme | viewer (legacy tokens map to 'admin')
+    """
+
+    user: str
+    user_id: str | None
+    org_id: str
+    role: str
+
+
 def mint_token(username: str, ttl_s: int | None = None) -> str:
-    """Issue a fresh bearer token."""
+    """Issue a fresh LEGACY bearer token (single-admin HMAC format)."""
     ttl = ttl_s or AUTH_TTL_S
     expiry = int(time.time()) + ttl
     payload = f"{username}|{expiry}"
@@ -71,8 +126,37 @@ def mint_token(username: str, ttl_s: int | None = None) -> str:
     return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=")
 
 
-def verify_token(token: str) -> str | None:
-    """Return username if token valid + unexpired; else None."""
+def verify_token(token: str) -> Principal | None:
+    """Verify a bearer token of either format → Principal, or None.
+
+    Order: org session JWT (api.identity.verify_session) first, then the
+    legacy single-admin HMAC token — which maps to org-demo/admin so the
+    pre-Phase-2 deployed demo keeps working unchanged.
+    """
+    if not token:
+        return None
+    try:
+        from api import identity
+    except ImportError:  # identity deps not installed — legacy still works
+        identity = None
+    if identity is not None:
+        claims = identity.verify_session(token)
+        if claims:
+            return Principal(
+                user=claims["sub"],
+                user_id=claims["sub"],
+                org_id=claims["org_id"],
+                role=claims["role"],
+            )
+    username = _verify_legacy_token(token)
+    if username is not None:
+        return Principal(user=username, user_id=None,
+                         org_id=LEGACY_ORG_ID, role=LEGACY_ROLE)
+    return None
+
+
+def _verify_legacy_token(token: str) -> str | None:
+    """Return username if the legacy HMAC token is valid + unexpired."""
     if not token:
         return None
     try:
@@ -109,6 +193,7 @@ def check_password(username: str, password: str) -> bool:
 
 PUBLIC_PATHS = {
     "/", "/health", "/login",
+    "/auth/login", "/auth/oidc/login", "/auth/oidc/callback",
     "/docs", "/openapi.json", "/redoc",
 }
 PUBLIC_PREFIXES = ("/videos/", "/media/")  # static cached media
@@ -125,9 +210,22 @@ def is_public(path: str, method: str) -> bool:
     return False
 
 
+def _set_principal(request: Request, principal: Principal) -> None:
+    request.state.user = principal.user
+    request.state.user_id = principal.user_id
+    request.state.org_id = principal.org_id
+    request.state.role = principal.role
+
+
 async def auth_middleware(request: Request, call_next):
     """FastAPI ASGI middleware — reject unauthenticated requests."""
     if AUTH_DISABLED:
+        # Dev bypass runs as the legacy admin principal so org-scoped
+        # routes (request.state.org_id / require_role) keep working.
+        _set_principal(request, Principal(
+            user=AUTH_USER, user_id=None,
+            org_id=LEGACY_ORG_ID, role=LEGACY_ROLE,
+        ))
         return await call_next(request)
     if is_public(request.url.path, request.method):
         return await call_next(request)
@@ -140,13 +238,44 @@ async def auth_middleware(request: Request, call_next):
             headers={"WWW-Authenticate": "Bearer"},
         )
     token = auth_hdr.split(" ", 1)[1].strip()
-    user = verify_token(token)
-    if not user:
+    principal = verify_token(token)
+    if not principal:
         from fastapi.responses import JSONResponse
         return JSONResponse(
             {"detail": "invalid or expired token"}, status_code=401,
             headers={"WWW-Authenticate": "Bearer"},
         )
-    # Pass username through so downstream routes can read it
-    request.state.user = user
+    # Pass identity through so downstream routes can read it
+    _set_principal(request, principal)
     return await call_next(request)
+
+
+# ── Role gating — FastAPI dependency factory ───────────────────────────────
+
+def require_role(*roles: str):
+    """Dependency factory: require the verified session role to be in `roles`.
+
+    Usage:
+        @router.post("/thing", dependencies=[Depends(require_role("admin"))])
+        # or, to read the role:
+        def handler(role: str = Depends(require_role("admin", "sme"))): ...
+
+    Roles are matched exactly (no implicit hierarchy) — pass every role that
+    may call the route. With no arguments it only requires authentication.
+    Legacy tokens and the BU_AUTH_DISABLED=1 dev bypass carry role 'admin'.
+    """
+
+    def _dependency(request: Request) -> str:
+        role = getattr(request.state, "role", None)
+        if role is None:
+            # Public/unauthenticated path — the middleware never ran a
+            # verification for this request.
+            raise HTTPException(
+                status_code=401, detail="authentication required",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        if roles and role not in roles:
+            raise HTTPException(status_code=403, detail="insufficient role")
+        return role
+
+    return _dependency

api/billing.py

@@ -0,0 +1,512 @@
+"""
+Billing + marketplace licensing API — docs/HARDENING.md Phase 4.
+
+Two routers, both mounted by api/server.py:
+
+    router          — /billing/*  license admin (org-scoped, T1 per
+                      docs/TENANCY.md) + optional Stripe self-serve checkout
+    experts_router  — /experts/{agent_id}/chat  license-key-gated delivery
+                      proxy (metered per call)
+
+Stripe is OPTIONAL by design. Without STRIPE_API_KEY the platform still
+issues 'manual' licenses via POST /billing/licenses (design partners pay by
+invoice) and all gating / metering / revocation behaves identically; the
+Stripe endpoints (/billing/checkout, /billing/webhook) then 404 cleanly.
+With STRIPE_API_KEY (+ STRIPE_WEBHOOK_SECRET for the webhook) Stripe adds
+self-serve subscription checkout on top: checkout.session.completed issues
+the license, customer.subscription.deleted revokes it.
+
+Key handling (atp/licensing.py stores only the key_id half — never the
+secret):
+  * The plaintext licenseKey is returned exactly ONCE — in the POST
+    /billing/licenses response. Save it then or re-issue.
+  * Webhook-issued keys are DISCARDED here after issuance (the webhook
+    response goes to Stripe, not the customer) — delivery happens
+    out-of-band via the billing portal / support until a customer portal
+    ships.
+  * GET /billing/licenses never returns key material (keyId, the public
+    identifier half, is not key material).
+
+Enforcement model for /experts/{agent_id}/chat: the route is on the
+auth-middleware allowlist (bearer optional) but the handler calls
+atp.licensing.check_license on EVERY request — a DB status read per call is
+the enforcement, there is deliberately NO cache of check results, so
+revoking a license cuts this path off on the very next call.
+
+Org scoping: /billing/* admin routes take the org ONLY from verified
+request.state (set by the auth middleware from the JWT — TENANCY.md layer 1).
+The expert gate authenticates by license key instead; the license ROW
+carries the org its usage is metered under.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Request
+from pydantic import BaseModel
+
+from api.auth import require_role
+from atp import licensing, store
+
+router = APIRouter(prefix="/billing")
+experts_router = APIRouter(prefix="/experts")
+
+#: Backend spec for the expert delivery proxy (agents/backend.get_backend).
+#: The dry-run default keeps the delivery path CI-exercisable with no keys.
+EXPERT_SPEC_ENV = "ATP_EXPERT_SPEC"
+EXPERT_SPEC_DEFAULT = "dryrun:candidate"
+
+
+# ── Tenancy helper (same idiom as api/atp.py) ───────────────────────────────
+
+def _org(request: Request) -> str:
+    """Org for this request — ONLY from verified auth state (TENANCY.md §4
+    layer 1). 'org-demo' when auth is disabled / legacy single-admin token."""
+    return getattr(request.state, "org_id", None) or "org-demo"
+
+
+# ── Seed catalog helpers (T0 — atp/store.get_data) ─────────────────────────
+
+def _agent(agent_id: str) -> dict | None:
+    for a in store.get_data().get("AGENTS", []):
+        if a.get("id") == agent_id:
+            return a
+    return None
+
+
+def _top_cert_label(agent: dict) -> str:
+    """Label of the agent's highest-layer cert (Stripe product naming)."""
+    certs = {c.get("id"): c for c in store.get_data().get("CERTS", [])}
+    held = [certs[cid] for cid in (agent.get("certIds") or []) if cid in certs]
+    if not held:
+        return ""
+    top = max(held, key=lambda c: c.get("layer") or 0)
+    return str(top.get("label") or "")
+
+
+def _unit_amount_usd(price: str | None) -> int | None:
+    """Seed licensing.price ('$4,800/mo') → Stripe unit_amount cents."""
+    m = re.search(r"\$?\s*([0-9][0-9,]*(?:\.[0-9]+)?)", price or "")
+    if not m:
+        return None
+    return int(round(float(m.group(1).replace(",", "")) * 100))
+
+
+def _license_for_subscription(org_id: str,
+                              subscription_id: str | None) -> dict | None:
+    """The org's license bound to a Stripe subscription id, if any."""
+    if not subscription_id:
+        return None
+    for row in licensing.list_licenses(org_id):
+        if row.get("stripeSubscription") == subscription_id:
+            return row
+    return None
+
+
+# ── Stripe (optional — endpoints 404 when unconfigured) ────────────────────
+
+def _stripe_key() -> str | None:
+    return os.environ.get("STRIPE_API_KEY") or None
+
+
+def _stripe_configured_or_404():
+    """Stripe endpoints don't exist without config — plain 404, no oracle."""
+    if not _stripe_key():
+        raise HTTPException(404, "Not Found")
+    try:
+        import stripe
+    except ImportError as e:  # pragma: no cover — requirements.txt has it
+        raise HTTPException(
+            503, "stripe library not installed (pip install stripe)") from e
+    stripe.api_key = _stripe_key()
+    return stripe
+
+
+def _cancel_subscription_best_effort(subscription_id: str | None) -> bool:
+    """Cancel the Stripe subscription behind a revoked license. Best-effort
+    by contract: the license is already revoked in OUR ledger (the expert
+    gate is cut off regardless); a Stripe hiccup must not un-revoke it."""
+    if not subscription_id or not _stripe_key():
+        return False
+    try:
+        import stripe
+        stripe.api_key = _stripe_key()
+        stripe.Subscription.cancel(subscription_id)
+        return True
+    except Exception as e:  # noqa: BLE001 — best-effort by contract
+        print(f"WARNING: stripe cancel of {subscription_id} failed: {e}",
+              file=sys.stderr)
+        return False
+
+
+# ── License admin routes ────────────────────────────────────────────────────
+
+class IssueBody(BaseModel):
+    agentId: str
+    kind: str = "production"   # licensing.VALID_KINDS
+    expiresAt: str | None = None   # ISO-8601 UTC; None = perpetual
+
+
+@router.post("/licenses", dependencies=[Depends(require_role("admin"))])
+def billing_issue(body: IssueBody, request: Request):
+    """Issue a 'manual' license for a marketplace agent (role admin).
+
+    The Stripe-less path (design partners pay by invoice). Returns
+    {license, licenseKey}; the plaintext licenseKey appears ONLY here —
+    the engine stores just its key_id half.
+    """
+    try:
+        return licensing.issue_license(_org(request), body.agentId, body.kind,
+                                       expires_at=body.expiresAt)
+    except licensing.UnknownAgentError as e:
+        raise HTTPException(404, str(e)) from e
+    except licensing.AgentNotLicensableError as e:
+        raise HTTPException(400, str(e)) from e
+    except ValueError as e:  # bad kind / expires_at
+        raise HTTPException(400, str(e)) from e
+    except licensing.LicenseSigningKeyError as e:
+        raise HTTPException(503, str(e)) from e
+
+
+@router.post("/licenses/{license_id}/reissue",
+             dependencies=[Depends(require_role("admin"))])
+def billing_reissue(license_id: str, request: Request):
+    """Mint a fresh key for an active license (role admin). The previous key
+    stops working immediately. This is how Stripe self-serve customers get
+    their key: the webhook stores no plaintext, an org admin re-issues here
+    and delivers it. Key shown ONCE."""
+    try:
+        return licensing.reissue_key(_org(request), license_id)
+    except licensing.LicenseNotFoundError as e:
+        raise HTTPException(404, str(e)) from e
+    except ValueError as e:
+        raise HTTPException(400, str(e)) from e
+    except licensing.LicenseSigningKeyError as e:
+        raise HTTPException(503, str(e)) from e
+
+
+@router.get("/licenses",
+            dependencies=[Depends(require_role("admin", "viewer"))])
+def billing_licenses(request: Request):
+    """Org's licenses, latest-first. Never contains key material — keys are
+    shown once, at issue time (keyId is the public identifier half only)."""
+    return {"licenses": licensing.list_licenses(_org(request))}
+
+
+class RevokeBody(BaseModel):
+    reason: str = ""
+
+
+@router.post("/licenses/{license_id}/revoke",
+             dependencies=[Depends(require_role("admin"))])
+def billing_revoke(license_id: str, body: RevokeBody, request: Request):
+    """Revoke a license (role admin, idempotent). Takes effect on the very
+    next /experts call — the gate re-reads DB state every time. Cross-tenant
+    ids 404 identically to unknown ids (TENANCY.md #6). When the license was
+    Stripe-issued AND Stripe is configured, the subscription is cancelled
+    too, best-effort."""
+    try:
+        revoked = licensing.revoke_license(
+            _org(request), license_id, body.reason.strip() or "revoked by admin")
+    except licensing.LicenseNotFoundError as e:
+        raise HTTPException(404, f"license {license_id} not found") from e
+    revoked["stripeCancelled"] = _cancel_subscription_best_effort(
+        revoked.get("stripeSubscription"))
+    return revoked
+
+
+@router.get("/usage", dependencies=[Depends(require_role("admin", "viewer"))])
+def billing_usage(request: Request, licenseId: str | None = None):
+    """Metered usage for the org (optionally one license):
+    {calls, tokensIn, tokensOut, byAgent}."""
+    return licensing.usage_summary(_org(request), license_id=licenseId)
+
+
+# ── Stripe self-serve checkout ──────────────────────────────────────────────
+
+class CheckoutBody(BaseModel):
+    agentId: str
+    kind: str = "production"
+
+
+@router.post("/checkout", dependencies=[Depends(require_role("admin"))])
+def billing_checkout(body: CheckoutBody, request: Request):
+    """Create a Stripe subscription Checkout Session for a seed listing.
+
+    404 when STRIPE_API_KEY is unset (manual licensing still fully works).
+    Price comes from the seed agent's licensing.price; the license itself is
+    issued by the checkout.session.completed webhook, keyed by the metadata
+    written here ({org_id, agent_id, kind} — org from verified state ONLY,
+    so a tampered client cannot buy a license into another org).
+    """
+    stripe = _stripe_configured_or_404()
+    org = _org(request)
+    agent = _agent(body.agentId)
+    if agent is None:
+        raise HTTPException(404, f"agent {body.agentId} not found")
+    if body.kind not in licensing.VALID_KINDS:
+        raise HTTPException(
+            400, f"kind must be one of {sorted(licensing.VALID_KINDS)}")
+    seed_licensing = agent.get("licensing") or {}
+    if not seed_licensing.get("available"):
+        raise HTTPException(400, f"agent {body.agentId} is not licensable")
+    unit_amount = _unit_amount_usd(seed_licensing.get("price"))
+    if not unit_amount:
+        raise HTTPException(
+            400, f"agent {body.agentId} has no parseable price")
+
+    label = _top_cert_label(agent)
+    product_name = agent.get("name") or body.agentId
+    if label:
+        product_name = f"{product_name} — {label}"
+    metadata = {"org_id": org, "agent_id": body.agentId, "kind": body.kind}
+    base = (os.environ.get("BU_PUBLIC_URL")
+            or request.headers.get("origin")
+            or str(request.base_url)).rstrip("/")
+    try:
+        session = stripe.checkout.Session.create(
+            mode="subscription",
+            line_items=[{
+                "quantity": 1,
+                "price_data": {
+                    "currency": "usd",
+                    "unit_amount": unit_amount,
+                    "recurring": {"interval": "month"},
+                    "product_data": {"name": product_name},
+                },
+            }],
+            # Session metadata drives checkout.session.completed issuance;
+            # subscription_data.metadata puts the SAME keys on the
+            # subscription object so customer.subscription.deleted can
+            # resolve the org + license to revoke.
+            metadata=metadata,
+            subscription_data={"metadata": metadata},
+            success_url=(base +
+                         "/#/billing/success?session_id={CHECKOUT_SESSION_ID}"),
+            cancel_url=base + "/#/billing/cancelled",
+        )
+    except Exception as e:  # noqa: BLE001 — stripe SDK error zoo
+        raise HTTPException(502, f"stripe checkout failed: {e}") from e
+    return {"url": session.url, "sessionId": session.id}
+
+
+# ── Stripe webhook (public path — the signature IS the authentication) ─────
+
+@router.post("/webhook")
+async def billing_webhook(request: Request):
+    """Stripe event sink. api/server.py adds this exact path to the auth
+    allowlist: Stripe cannot send a bearer — every delivery is authenticated
+    by its Stripe-Signature header, verified against STRIPE_WEBHOOK_SECRET
+    (invalid/missing signature → 400; Stripe unconfigured → 404).
+
+    Handled events (anything else is acknowledged and ignored):
+      checkout.session.completed    → issue the license from the session
+        metadata ({org_id, agent_id, kind} written by /billing/checkout),
+        storing the Stripe customer/subscription ids. Idempotent across
+        Stripe's retries (one license per subscription). The plaintext key
+        is DISCARDED here — only its key_id persists, so it is not
+        retrievable later (out-of-band delivery via the billing portal).
+      customer.subscription.deleted → revoke the matching license
+        (reason 'subscription cancelled').
+
+    Responds 200 fast — handling is local DB writes only, no network calls.
+    """
+    if not _stripe_key() or not os.environ.get("STRIPE_WEBHOOK_SECRET"):
+        raise HTTPException(404, "Not Found")
+    import stripe
+
+    payload = await request.body()
+    sig_header = request.headers.get("stripe-signature", "")
+    try:
+        stripe.Webhook.construct_event(
+            payload, sig_header, os.environ["STRIPE_WEBHOOK_SECRET"])
+    except Exception as e:  # noqa: BLE001 — bad payload OR bad signature
+        raise HTTPException(400, "invalid webhook signature") from e
+
+    # construct_event verified signature + JSON; work on the verified bytes
+    # as plain dicts (StripeObject's dict-likeness varies across versions).
+    import json
+    event = json.loads(payload)
+    etype = event.get("type", "")
+    obj = (event.get("data") or {}).get("object") or {}
+
+    if etype == "checkout.session.completed":
+        md = obj.get("metadata") or {}
+        org, agent_id = md.get("org_id"), md.get("agent_id")
+        kind = md.get("kind") or "production"
+        if not (org and agent_id):
+            return {"received": True, "handled": etype,
+                    "skipped": "missing metadata"}
+        subscription = obj.get("subscription")
+        # Stripe retries deliveries — never double-issue for one sub.
+        if subscription and _license_for_subscription(org, subscription):
+            return {"received": True, "handled": etype,
+                    "skipped": "already issued"}
+        try:
+            issued = licensing.issue_license(
+                org, agent_id, kind,
+                stripe_customer=obj.get("customer"),
+                stripe_subscription=subscription)
+        except (licensing.LicensingError, ValueError) as e:
+            # Ack (200) so Stripe stops retrying a permanently-bad event,
+            # but say what was wrong for the webhook log.
+            print(f"WARNING: webhook license issue failed: {e}",
+                  file=sys.stderr)
+            return {"received": True, "handled": etype, "skipped": str(e)}
+        # Hash-only persistence: the plaintext key never leaves this scope.
+        return {"received": True, "handled": etype,
+                "licenseId": issued["license"]["id"]}
+
+    if etype == "customer.subscription.deleted":
+        md = obj.get("metadata") or {}
+        org, sub_id = md.get("org_id"), obj.get("id")
+        if not (org and sub_id):
+            return {"received": True, "handled": etype,
+                    "skipped": "missing metadata"}
+        row = _license_for_subscription(org, sub_id)
+        if row is None:
+            return {"received": True, "handled": etype,
+                    "skipped": "no matching license"}
+        licensing.revoke_license(org, row["id"], "subscription cancelled")
+        return {"received": True, "handled": etype, "licenseId": row["id"]}
+
+    return {"received": True, "ignored": etype}
+
+
+# ── Expert delivery gate — /experts/{agent_id}/chat ─────────────────────────
+
+class ChatBody(BaseModel):
+    messages: list[dict]
+
+
+def _persona(agent: dict) -> str:
+    """Agent persona system prompt from T0 seed fields only (no tenant
+    data — TENANCY.md leak surface #5)."""
+    name = agent.get("name") or agent.get("id")
+    level = agent.get("level")
+    domains = ", ".join(agent.get("domains") or []) or "general"
+    skills = ", ".join(
+        s.get("name", "") for s in (agent.get("skills") or [])[:8])
+    summary = (agent.get("reportCard") or {}).get("summary", "")
+    boundaries = " ".join(
+        f.get("boundary", "") for f in (agent.get("failures") or [])[:2])
+    parts = [
+        f"You are {name}, an ATP Level {level} certified expert agent "
+        f"(domains: {domains}).",
+        f"Certified skills: {skills}." if skills else "",
+        summary,
+        f"Known boundaries: {boundaries}" if boundaries else "",
+        "Answer within your certified scope; when a request falls outside "
+        "it, say so plainly and recommend human review.",
+    ]
+    return " ".join(p for p in parts if p)
+
+
+def _flatten_messages(messages: list[dict]) -> str:
+    """[{role, content}, ...] → one transcript string (backends take a
+    single system + user pair, agents/backend.Backend.complete)."""
+    parts = []
+    for m in messages:
+        if not isinstance(m, dict):
+            continue
+        role = str(m.get("role", "user")).strip().lower() or "user"
+        content = str(m.get("content", "")).strip()
+        if content:
+            parts.append(f"{role.capitalize()}: {content}")
+    return "\n\n".join(parts)
+
+
+def _license_org(license_key: str) -> str | None:
+    """Org the checked license row belongs to.
+
+    check_license()'s public license dict intentionally carries no org key
+    (atp/store.py convention), but metering must land under the license
+    ROW's org (T1) — never under a caller-supplied value, and the keyed
+    machine caller has no bearer/org at all. Resolve it through the same
+    key-id lookup check_license used; the in-package coupling to
+    licensing's _parse_key/_load_by_key_id is deliberate — the alternative
+    (raw licenses SQL here) would break the DAL-only convention
+    (TENANCY.md layer 2).
+    """
+    parsed = licensing._parse_key(license_key)
+    if not parsed:
+        return None
+    row = licensing._load_by_key_id(parsed[0])
+    return (row or {}).get("org_id")
+
+
+_DENIED_403 = ("revoked", "expired")
+
+
+@experts_router.post("/{agent_id}/chat")
+def expert_chat(
+    agent_id: str,
+    body: ChatBody,
+    request: Request,
+    license_key: str | None = Header(default=None, alias="X-ATP-License-Key"),
+):
+    """Licensed expert delivery (HARDENING.md Phase 4).
+
+    Authentication is the LICENSE KEY (X-ATP-License-Key header), not a
+    bearer — api/server.py puts '/experts/' on the middleware allowlist and
+    THIS handler enforces instead: atp.licensing.check_license runs on EVERY
+    request (a DB status read — deliberately uncached, so revocation cuts
+    the path off on the very next call). 401 invalid key / 403 revoked or
+    expired (with the reason). Every successful call appends a usage_events
+    row (metering).
+    """
+    if not license_key or not license_key.strip():
+        raise HTTPException(
+            401, "missing license key (X-ATP-License-Key header)")
+    license_key = license_key.strip()
+
+    try:
+        check = licensing.check_license(license_key)
+    except licensing.LicenseSigningKeyError as e:
+        raise HTTPException(503, str(e)) from e
+    if not check.get("ok"):
+        reason = str(check.get("reason") or "invalid")
+        if reason in _DENIED_403:
+            raise HTTPException(403, f"license {reason}")
+        raise HTTPException(401, f"invalid license key ({reason})")
+    lic = check.get("license") or {}
+
+    if lic.get("agentId") != agent_id:
+        raise HTTPException(
+            403, f"license is for agent {lic.get('agentId')}, not {agent_id}")
+
+    agent = _agent(agent_id)
+    if agent is None:  # licensed against a since-removed seed agent
+        raise HTTPException(404, f"agent {agent_id} not found")
+
+    transcript = _flatten_messages(body.messages or [])
+    if not transcript:
+        raise HTTPException(400, "messages must contain at least one "
+                                 "non-empty {role, content} entry")
+
+    from agents.backend import get_backend
+    spec = os.environ.get(EXPERT_SPEC_ENV, EXPERT_SPEC_DEFAULT)
+    system = _persona(agent)
+    try:
+        reply = get_backend(spec).complete(system, transcript)
+    except Exception as e:  # noqa: BLE001 — backend/network failures
+        raise HTTPException(503, f"expert backend unavailable: {e}") from e
+
+    # Metering — approx token counts (chars/4). No usage row, no reply:
+    # usage_events is the append-only billing record (HARDENING.md Phase 4).
+    org = _license_org(license_key) or "org-demo"
+    try:
+        usage = licensing.record_usage(
+            org, lic.get("id"), agent_id,
+            endpoint=f"/experts/{agent_id}/chat",
+            tokens_in=max(1, (len(system) + len(transcript)) // 4),
+            tokens_out=max(1, len(reply or "") // 4),
+            status="ok",
+        )
+    except Exception as e:  # noqa: BLE001
+        raise HTTPException(503, f"usage metering failed: {e}") from e
+
+    return {"reply": reply, "usage": usage}

api/identity.py

@@ -0,0 +1,474 @@
+"""
+Org-scoped identity for Brain University — Phase 2 (docs/TENANCY.md).
+
+Three ways to obtain a session:
+  1. POST /auth/login {email, password}          — local accounts (argon2id)
+  2. GET  /auth/oidc/login → GET /auth/oidc/callback — OIDC authorization-code
+     flow (issuer/client from env; endpoints 404 when OIDC_ISSUER is unset)
+  3. Legacy POST /login (api/server.py)          — single-admin HMAC token;
+     api.auth.verify_token() maps it to org 'org-demo' with role 'admin', so
+     the deployed demo keeps working with zero new env.
+
+Session = JWT (HS256 over BU_AUTH_SECRET), claims {sub, org_id, role, exp}.
+`sub` is users.user_id (the identity spec's `id` — see migrations/004).
+
+Env (all optional; partial sets fail loudly, P0 style):
+  BU_BOOTSTRAP_ADMIN_EMAIL   — with _PASSWORD: provision this local admin in
+  BU_BOOTSTRAP_ADMIN_PASSWORD  org 'org-demo' on first use if absent.
+                               One without the other = RuntimeError.
+  OIDC_ISSUER                — enables /auth/oidc/*; discovery at
+                               {OIDC_ISSUER}/.well-known/openid-configuration
+  OIDC_CLIENT_ID             — REQUIRED once OIDC_ISSUER is set
+  OIDC_CLIENT_SECRET         — REQUIRED once OIDC_ISSUER is set
+  OIDC_REDIRECT_URL          — REQUIRED once OIDC_ISSUER is set; must match
+                               the IdP-registered callback
+                               (https://host/auth/oidc/callback)
+
+OIDC org resolution: user linked by (iss, sub); on first login the verified
+email's domain must match an orgs.domain row, else 403 'no org for domain'.
+OIDC users default to role 'viewer'.
+
+Mounted by api/server.py:
+    from api.identity import router as identity_router
+    app.include_router(identity_router)
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+import threading
+import time
+from urllib.parse import urlencode
+
+import httpx
+import jwt
+from argon2 import PasswordHasher
+from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import RedirectResponse
+from pydantic import BaseModel
+
+from api import auth as _auth  # BU_AUTH_SECRET / TTL; auth does NOT import us at module level
+
+BOOTSTRAP_ORG_ID = "org-demo"   # seeded by migrations/004
+ROLES = ("admin", "sme", "viewer")
+
+SESSION_ALGO = "HS256"
+STATE_TTL_S = 600               # OIDC state/nonce validity: 10 minutes
+_STATE_AUD = "bu-oidc-state"    # aud claim keeps state JWTs unusable as sessions
+_ID_TOKEN_ALGOS = ["RS256", "RS384", "RS512", "ES256", "ES384", "PS256"]
+
+# ── Env (validated at import — P0 fail-loud style) ──────────────────────────
+
+OIDC_ISSUER        = os.environ.get("OIDC_ISSUER", "").strip()
+OIDC_CLIENT_ID     = os.environ.get("OIDC_CLIENT_ID", "").strip()
+OIDC_CLIENT_SECRET = os.environ.get("OIDC_CLIENT_SECRET", "")
+OIDC_REDIRECT_URL  = os.environ.get("OIDC_REDIRECT_URL", "").strip()
+
+if OIDC_ISSUER:
+    _missing = [name for name, val in (
+        ("OIDC_CLIENT_ID", OIDC_CLIENT_ID),
+        ("OIDC_CLIENT_SECRET", OIDC_CLIENT_SECRET),
+        ("OIDC_REDIRECT_URL", OIDC_REDIRECT_URL),
+    ) if not val]
+    if _missing:
+        raise RuntimeError(
+            "Refusing to start: OIDC_ISSUER is set but required OIDC env "
+            "var(s) unset: " + ", ".join(_missing)
+            + ". Set them all, or unset OIDC_ISSUER to disable OIDC "
+            "(local + legacy logins keep working without it)."
+        )
+
+
+def _bootstrap_env() -> tuple[str, str] | None:
+    """(email, password) when both set, None when both unset; partial raises."""
+    email = os.environ.get("BU_BOOTSTRAP_ADMIN_EMAIL", "").strip()
+    password = os.environ.get("BU_BOOTSTRAP_ADMIN_PASSWORD", "")
+    if not email and not password:
+        return None
+    if not (email and password):
+        missing = "BU_BOOTSTRAP_ADMIN_PASSWORD" if email else "BU_BOOTSTRAP_ADMIN_EMAIL"
+        raise RuntimeError(
+            f"Refusing to start: {missing} unset while its counterpart is "
+            "set. Set both BU_BOOTSTRAP_ADMIN_EMAIL and "
+            "BU_BOOTSTRAP_ADMIN_PASSWORD, or neither."
+        )
+    return email.lower(), password
+
+
+_bootstrap_env()  # fail loudly at import on partial env
+
+
+# ── Password hashing (argon2id) ─────────────────────────────────────────────
+
+_hasher = PasswordHasher()  # argon2id, library-default parameters
+
+
+def hash_password(password: str) -> str:
+    return _hasher.hash(password)
+
+
+def verify_password(pw_hash: str, password: str) -> bool:
+    try:
+        return _hasher.verify(pw_hash, password)
+    except (VerifyMismatchError, VerificationError, InvalidHashError):
+        return False
+
+
+# ── Session JWTs — claims {sub, org_id, role, exp} ──────────────────────────
+
+def mint_session(user: dict, ttl_s: int | None = None) -> str:
+    """Issue a session JWT for a users row (dict with id/org_id/role)."""
+    exp = int(time.time()) + (ttl_s or _auth.AUTH_TTL_S)
+    claims = {
+        "sub": user["id"],
+        "org_id": user["org_id"],
+        "role": user["role"],
+        "exp": exp,
+    }
+    return jwt.encode(claims, _auth.AUTH_SECRET, algorithm=SESSION_ALGO)
+
+
+def verify_session(token: str) -> dict | None:
+    """Return the session claims when `token` is a valid, unexpired org JWT.
+
+    None for anything else (legacy HMAC tokens, state tokens, garbage) —
+    api.auth.verify_token() then falls back to the legacy verifier.
+    """
+    if not token or token.count(".") != 2:  # fast path: not a JWT at all
+        return None
+    try:
+        claims = jwt.decode(token, _auth.AUTH_SECRET, algorithms=[SESSION_ALGO])
+    except jwt.InvalidTokenError:
+        return None
+    if not all(claims.get(k) for k in ("sub", "org_id", "role")):
+        return None
+    if claims["role"] not in ROLES:
+        return None
+    return claims
+
+
+# ── DB helpers (users/orgs — migrations/004; users.user_id is the id) ───────
+
+_USER_COLS = ("user_id, display_name, created_at, org_id, email, role, "
+              "oidc_iss, oidc_sub, pw_hash")
+
+
+def _db():
+    from atp import db
+    return db
+
+
+_MIGRATED = False
+
+
+def _ensure_db() -> None:
+    """Idempotent lazy migration (same pattern as atp/store.py)."""
+    global _MIGRATED
+    if not _MIGRATED:
+        _db().run_migrations()
+        _MIGRATED = True
+
+
+def _to_user(row: dict) -> dict:
+    user = dict(row)
+    user["id"] = user.pop("user_id")
+    return user
+
+
+def get_user_by_email(email: str) -> dict | None:
+    rows = _db().query(
+        f"SELECT {_USER_COLS} FROM users WHERE email = :email",
+        {"email": email.strip().lower()},
+    )
+    return _to_user(rows[0]) if rows else None
+
+
+def get_user_by_oidc(iss: str, sub: str) -> dict | None:
+    rows = _db().query(
+        f"SELECT {_USER_COLS} FROM users"
+        " WHERE oidc_iss = :iss AND oidc_sub = :sub",
+        {"iss": iss, "sub": sub},
+    )
+    return _to_user(rows[0]) if rows else None
+
+
+def get_org_by_domain(domain: str) -> dict | None:
+    rows = _db().query(
+        "SELECT id, name, domain, created_at FROM orgs"
+        " WHERE domain IS NOT NULL AND lower(domain) = lower(:domain)",
+        {"domain": domain},
+    )
+    return dict(rows[0]) if rows else None
+
+
+def create_user(*, email: str, org_id: str, role: str,
+                oidc_iss: str | None = None, oidc_sub: str | None = None,
+                pw_hash: str | None = None,
+                display_name: str | None = None) -> dict:
+    """Insert an identity user; returns the created (or racing-winner) row."""
+    if role not in ROLES:
+        raise ValueError(f"role must be one of {ROLES}, got {role!r}")
+    email = email.strip().lower()
+    try:
+        _db().execute(
+            "INSERT INTO users (user_id, display_name, created_at, org_id,"
+            " email, role, oidc_iss, oidc_sub, pw_hash)"
+            " VALUES (:id, :name, :ts, :org, :email, :role, :iss, :sub, :pw)",
+            {
+                "id": "u-" + secrets.token_hex(8),
+                "name": display_name or email,
+                "ts": time.time(),  # users.created_at is REAL (001)
+                "org": org_id,
+                "email": email,
+                "role": role,
+                "iss": oidc_iss,
+                "sub": oidc_sub,
+                "pw": pw_hash,
+            },
+        )
+    except Exception:
+        # Unique-email race: another worker inserted first — use theirs.
+        existing = get_user_by_email(email)
+        if existing is None:
+            raise
+        return existing
+    user = get_user_by_email(email)
+    assert user is not None
+    return user
+
+
+def link_user_oidc(user_id: str, iss: str, sub: str) -> None:
+    _db().execute(
+        "UPDATE users SET oidc_iss = :iss, oidc_sub = :sub"
+        " WHERE user_id = :id",
+        {"iss": iss, "sub": sub, "id": user_id},
+    )
+
+
+# ── Bootstrap admin (local fallback account, org-demo) ──────────────────────
+
+def provision_bootstrap_admin() -> dict | None:
+    """Create the BU_BOOTSTRAP_ADMIN_* local admin in org-demo if absent.
+
+    Returns the user row (existing or created), or None when the env pair is
+    unset. Partial env = RuntimeError (P0 fail-loud style). Idempotent.
+    """
+    pair = _bootstrap_env()
+    if pair is None:
+        return None
+    email, password = pair
+    _ensure_db()
+    existing = get_user_by_email(email)
+    if existing is not None:
+        return existing
+    return create_user(
+        email=email,
+        org_id=BOOTSTRAP_ORG_ID,
+        role="admin",
+        pw_hash=hash_password(password),
+        display_name="Bootstrap admin",
+    )
+
+
+_READY = False
+_READY_LOCK = threading.Lock()
+
+
+def _ensure_ready() -> None:
+    """Migrations + bootstrap provisioning, once per process, thread-safe."""
+    global _READY
+    if _READY:
+        return
+    with _READY_LOCK:
+        if _READY:
+            return
+        _ensure_db()
+        provision_bootstrap_admin()
+        _READY = True
+
+
+# ── OIDC — discovery, JWKS, signed state ────────────────────────────────────
+
+_DISCOVERY_TTL_S = 3600
+_discovery_cache: dict = {}
+_jwks_clients: dict[str, jwt.PyJWKClient] = {}
+
+
+def oidc_enabled() -> bool:
+    return bool(OIDC_ISSUER)
+
+
+def _discovery() -> dict:
+    """Fetch (and cache) the issuer's openid-configuration."""
+    now = time.time()
+    if _discovery_cache.get("cfg") and now - _discovery_cache["fetched"] < _DISCOVERY_TTL_S:
+        return _discovery_cache["cfg"]
+    url = OIDC_ISSUER.rstrip("/") + "/.well-known/openid-configuration"
+    try:
+        resp = httpx.get(url, timeout=10.0)
+        resp.raise_for_status()
+        cfg = resp.json()
+    except httpx.HTTPError as e:
+        raise HTTPException(502, f"OIDC discovery failed: {e.__class__.__name__}") from e
+    for key in ("authorization_endpoint", "token_endpoint", "jwks_uri", "issuer"):
+        if key not in cfg:
+            raise HTTPException(502, f"OIDC discovery document missing {key!r}")
+    _discovery_cache.update(cfg=cfg, fetched=now)
+    return cfg
+
+
+def _jwks_client(jwks_uri: str) -> jwt.PyJWKClient:
+    client = _jwks_clients.get(jwks_uri)
+    if client is None:
+        client = jwt.PyJWKClient(jwks_uri, cache_keys=True)
+        _jwks_clients[jwks_uri] = client
+    return client
+
+
+def _mint_state(nonce: str) -> str:
+    """Signed CSRF state, 10 min TTL. aud keeps it unusable as a session."""
+    claims = {"aud": _STATE_AUD, "nonce": nonce,
+              "exp": int(time.time()) + STATE_TTL_S}
+    return jwt.encode(claims, _auth.AUTH_SECRET, algorithm=SESSION_ALGO)
+
+
+def _verify_state(state: str) -> dict | None:
+    if not state:
+        return None
+    try:
+        return jwt.decode(state, _auth.AUTH_SECRET,
+                          algorithms=[SESSION_ALGO], audience=_STATE_AUD)
+    except jwt.InvalidTokenError:
+        return None
+
+
+# ── Routes ──────────────────────────────────────────────────────────────────
+
+router = APIRouter()
+
+
+def _public_user(user: dict) -> dict:
+    return {"id": user["id"], "email": user["email"],
+            "org_id": user["org_id"], "role": user["role"]}
+
+
+class LoginBody(BaseModel):
+    email: str
+    password: str
+
+
+@router.post("/auth/login")
+def local_login(body: LoginBody):
+    """Local-account login → session JWT with {sub, org_id, role, exp}."""
+    _ensure_ready()
+    user = get_user_by_email(body.email)
+    ok = (user is not None and user.get("pw_hash")
+          and verify_password(user["pw_hash"], body.password))
+    if not ok:
+        time.sleep(0.25)  # match legacy /login: slow credential stuffing
+        raise HTTPException(401, "invalid credentials")
+    return {"token": mint_session(user), "user": _public_user(user)}
+
+
+@router.get("/auth/oidc/login")
+def oidc_login():
+    """Redirect to the IdP's authorization endpoint (code flow)."""
+    if not oidc_enabled():
+        raise HTTPException(404, "Not Found")
+    cfg = _discovery()
+    nonce = secrets.token_urlsafe(16)
+    params = {
+        "response_type": "code",
+        "client_id": OIDC_CLIENT_ID,
+        "redirect_uri": OIDC_REDIRECT_URL,
+        "scope": "openid email profile",
+        "state": _mint_state(nonce),
+        "nonce": nonce,
+    }
+    return RedirectResponse(
+        cfg["authorization_endpoint"] + "?" + urlencode(params),
+        status_code=302,
+    )
+
+
+@router.get("/auth/oidc/callback")
+def oidc_callback(code: str = "", state: str = "", error: str = "",
+                  error_description: str = ""):
+    """Exchange the code, verify the id_token, provision/link, mint session."""
+    if not oidc_enabled():
+        raise HTTPException(404, "Not Found")
+    _ensure_ready()
+    if error:
+        raise HTTPException(400, f"OIDC error: {error} {error_description}".strip())
+    st = _verify_state(state)
+    if st is None:
+        raise HTTPException(400, "invalid or expired state")
+    if not code:
+        raise HTTPException(400, "missing code")
+
+    cfg = _discovery()
+    try:
+        resp = httpx.post(
+            cfg["token_endpoint"],
+            data={
+                "grant_type": "authorization_code",
+                "code": code,
+                "redirect_uri": OIDC_REDIRECT_URL,
+                "client_id": OIDC_CLIENT_ID,
+                "client_secret": OIDC_CLIENT_SECRET,
+            },
+            timeout=10.0,
+        )
+        resp.raise_for_status()
+        id_token = resp.json().get("id_token", "")
+    except httpx.HTTPError as e:
+        raise HTTPException(502, f"OIDC token exchange failed: {e.__class__.__name__}") from e
+    if not id_token:
+        raise HTTPException(502, "OIDC token response had no id_token")
+
+    try:
+        signing_key = _jwks_client(cfg["jwks_uri"]).get_signing_key_from_jwt(id_token)
+        claims = jwt.decode(
+            id_token,
+            signing_key.key,
+            algorithms=_ID_TOKEN_ALGOS,
+            audience=OIDC_CLIENT_ID,
+            issuer=cfg["issuer"],
+            leeway=30,
+        )
+    except jwt.PyJWKClientError as e:
+        raise HTTPException(502, f"OIDC JWKS fetch failed: {e.__class__.__name__}") from e
+    except jwt.InvalidTokenError:
+        raise HTTPException(401, "invalid id_token")
+    if claims.get("nonce") != st["nonce"]:
+        raise HTTPException(401, "nonce mismatch")
+
+    iss, sub = claims["iss"], claims["sub"]
+    user = get_user_by_oidc(iss, sub)
+    if user is None:
+        email = (claims.get("email") or "").strip().lower()
+        if not email:
+            raise HTTPException(403, "id_token has no email claim")
+        if claims.get("email_verified") is False:
+            raise HTTPException(403, "email not verified by identity provider")
+        existing = get_user_by_email(email)
+        if existing is not None:
+            # Same verified email as an existing account → link identities.
+            link_user_oidc(existing["id"], iss, sub)
+            user = get_user_by_email(email)
+        else:
+            domain = email.split("@", 1)[1]
+            org = get_org_by_domain(domain)
+            if org is None:
+                raise HTTPException(403, "no org for domain")
+            user = create_user(
+                email=email,
+                org_id=org["id"],
+                role="viewer",  # OIDC default; org admins promote later
+                oidc_iss=iss,
+                oidc_sub=sub,
+                display_name=claims.get("name") or email,
+            )
+    return {"token": mint_session(user), "user": _public_user(user)}

api/knowledge.py

@@ -0,0 +1,262 @@
+"""
+Company-knowledge API routes — Phase 6 (docs/HARDENING.md first-customer UX).
+
+Mounted by api/server.py:
+    from api.knowledge import router as knowledge_router
+    app.include_router(knowledge_router)
+
+Endpoints (bearer-gated by the default auth middleware — api/auth.py):
+    POST /knowledge/docs                — multipart upload (md/txt/pdf,
+                                          <= 5 MB) → doc dict   [admin|sme]
+    GET  /knowledge/docs                — org's docs, summaries only [any role]
+    GET  /knowledge/docs/{id}           — full decrypted text    [admin|sme]
+    POST /knowledge/drafts              — {docIds, title} → generated L5
+                                          draft (blueprint + decrypted items
+                                          for review)            [admin|sme]
+    GET  /knowledge/drafts              — org's drafts [any role; items (T2)
+                                          included only for admin|sme]
+    POST /knowledge/drafts/{id}/approve — {note} → sign-off dict [sme ONLY]
+    POST /knowledge/drafts/{id}/reject  — {note} → sign-off dict [sme ONLY]
+    POST /knowledge/drafts/{id}/exam    — {candidateSpec?, agentId?,
+                                          dryRun=true} → {jobId}; job-backed
+                                          like /atp/exams/run    [admin|sme]
+
+Tenancy (docs/TENANCY.md — this is T2 "crown jewels" data):
+  * org_id comes ONLY from verified request state (request.state.org_id, set
+    by the auth middleware from the JWT) — never from body/query. All storage
+    is org-scoped + encrypted at rest in atp/knowledge.py.
+  * Uploads are stored as encrypted DB columns ONLY — no file ever lands on
+    disk, and in particular never under the public /videos or /media
+    prefixes (leak surface #4).
+  * Cross-tenant/unknown ids → the same 404 (no existence oracle, #6).
+
+Separation of duties: admins (or SMEs) manage docs and generate drafts, but
+ONLY role 'sme' can approve/reject — an admin must not be able to sign off
+the exam content they themselves assembled. The BU_AUTH_DISABLED=1 dev
+bypass (which runs every request as org-demo with role 'admin') is exempted
+from that one gate so the demo flow keeps working end-to-end; with auth
+enabled there is no exemption.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
+from pydantic import BaseModel
+
+from api.auth import require_role
+from atp import knowledge
+
+router = APIRouter(prefix="/knowledge")
+
+#: Upload size cap (bytes) — Phase 6 contract: 5 MB.
+MAX_UPLOAD_BYTES = 5 * 1024 * 1024
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+def _org(request: Request) -> str:
+    """Org for this request — ONLY from verified auth state (TENANCY.md §4
+    layer 1). 'org-demo' when auth is disabled / legacy single-admin token."""
+    return getattr(request.state, "org_id", None) or "org-demo"
+
+
+def _user(request: Request) -> str:
+    return getattr(request.state, "user", None) or "anon"
+
+
+def require_sme(request: Request) -> str:
+    """Sign-off gate: role 'sme' ONLY (separation of duties — admins manage,
+    SMEs sign). Dev exception: the BU_AUTH_DISABLED=1 bypass carries role
+    'admin' for every request (api/auth.py), so it is allowed through here —
+    otherwise the demo could never exercise approval. With auth enabled the
+    gate is strict."""
+    role = getattr(request.state, "role", None)
+    if role is None:
+        raise HTTPException(
+            status_code=401, detail="authentication required",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    from api import auth as _auth
+    if role != "sme" and not _auth.AUTH_DISABLED:
+        raise HTTPException(
+            403, "draft sign-off requires role 'sme' (separation of duties: "
+                 "admins manage documents and drafts; SMEs sign)")
+    return role
+
+
+def _http_errors(fn, *args, **kwargs):
+    """Run a knowledge-module call, translating its typed errors to HTTP.
+
+    UnsupportedDocumentError → 415, KnowledgeError → 400, DraftStateError →
+    409. The fail-loud missing-DATA_ENCRYPTION_KEY RuntimeError from
+    atp/crypto.py becomes a 503 with an actionable message (T2 content is
+    NEVER written or served as plaintext instead)."""
+    try:
+        return fn(*args, **kwargs)
+    except knowledge.UnsupportedDocumentError as e:
+        raise HTTPException(415, str(e)) from None
+    except knowledge.KnowledgeError as e:
+        raise HTTPException(400, str(e)) from None
+    except knowledge.DraftStateError as e:
+        raise HTTPException(409, str(e)) from None
+    except RuntimeError as e:
+        if "DATA_ENCRYPTION_KEY" in str(e):
+            raise HTTPException(
+                503, "DATA_ENCRYPTION_KEY is not configured on this "
+                     "deployment — company-knowledge content is T2 and is "
+                     "only ever stored/served encrypted (docs/TENANCY.md)",
+            ) from None
+        raise
+
+
+# ── Documents ────────────────────────────────────────────────────────────────
+
+@router.post("/docs", dependencies=[Depends(require_role("admin", "sme"))])
+async def upload_doc(request: Request, file: UploadFile = File(...)):
+    """Upload one company document (md/txt/pdf, <= 5 MB) into the org vault.
+
+    Content is encrypted under the org's subkey before it touches the DB; the
+    response returns metadata + the heuristic summary, not the full text.
+    """
+    if knowledge.doc_kind(file.filename, file.content_type) is None:
+        raise HTTPException(
+            415, f"unsupported document type (filename={file.filename!r}, "
+                 f"content-type={file.content_type!r}) — allowed: .md, .txt, "
+                 f".pdf")
+    data = await file.read(MAX_UPLOAD_BYTES + 1)
+    if len(data) > MAX_UPLOAD_BYTES:
+        raise HTTPException(
+            413, f"file exceeds the {MAX_UPLOAD_BYTES // (1024 * 1024)} MB "
+                 f"upload cap")
+    if not data:
+        raise HTTPException(400, "empty upload")
+    return _http_errors(
+        knowledge.ingest_doc, _org(request), file.filename,
+        file.content_type, data, uploaded_by=_user(request))
+
+
+@router.get("/docs", dependencies=[Depends(require_role())])
+def get_docs(request: Request):
+    """Org's vault docs, latest first — decrypted SUMMARIES only (any
+    authenticated role; full text stays behind the admin|sme route)."""
+    return {"docs": _http_errors(knowledge.list_docs, _org(request))}
+
+
+@router.get("/docs/{doc_id}",
+            dependencies=[Depends(require_role("admin", "sme"))])
+def get_doc(doc_id: str, request: Request):
+    """One doc with the full decrypted text (admin|sme). Cross-tenant ids
+    404 with the SAME message as unknown ids (TENANCY.md #6)."""
+    doc = _http_errors(knowledge.get_doc, _org(request), doc_id)
+    if doc is None:
+        raise HTTPException(404, f"doc {doc_id} not found")
+    return doc
+
+
+# ── L5 drafts ────────────────────────────────────────────────────────────────
+
+class DraftBody(BaseModel):
+    docIds: list[str]
+    title: str = ""
+
+
+@router.post("/drafts", dependencies=[Depends(require_role("admin", "sme"))])
+def create_draft(body: DraftBody, request: Request):
+    """Generate a deterministic L5 exam draft from the given vault docs
+    (blueprint sections = doc titles; >= 8 recall/apply items). The response
+    includes the DECRYPTED items so the reviewer can read them — this route
+    is admin|sme for exactly that reason."""
+    return _http_errors(
+        knowledge.generate_l5_draft, _org(request), body.docIds,
+        created_by=_user(request), title=body.title)
+
+
+@router.get("/drafts")
+def get_drafts(request: Request, role: str = Depends(require_role())):
+    """Org's drafts, latest first (any authenticated role). The decrypted
+    items (T2 — verbatim company passages) are included only for admin|sme;
+    viewers get blueprint + status metadata."""
+    include_items = role in ("admin", "sme")
+    return {"drafts": _http_errors(
+        knowledge.list_drafts, _org(request), include_items=include_items)}
+
+
+class ReviewBody(BaseModel):
+    note: str = ""
+
+
+@router.post("/drafts/{draft_id}/approve",
+             dependencies=[Depends(require_sme)])
+def approve_draft(draft_id: str, body: ReviewBody, request: Request):
+    """SME sign-off (role sme ONLY): flips the draft to 'approved', appends a
+    signed 'sme_signoff' evidence row to the org chain, and returns the
+    evidence id + examCertRef + examRunnable flag."""
+    out = _http_errors(
+        knowledge.approve_draft, _org(request), draft_id,
+        reviewer=_user(request), note=body.note)
+    if out is None:
+        raise HTTPException(404, f"draft {draft_id} not found")
+    return out
+
+
+@router.post("/drafts/{draft_id}/reject",
+             dependencies=[Depends(require_sme)])
+def reject_draft(draft_id: str, body: ReviewBody, request: Request):
+    """SME rejection (role sme ONLY) — same evidence trail as approval,
+    payload outcome 'rejected'."""
+    out = _http_errors(
+        knowledge.reject_draft, _org(request), draft_id,
+        reviewer=_user(request), note=body.note)
+    if out is None:
+        raise HTTPException(404, f"draft {draft_id} not found")
+    return out
+
+
+# ── Exam over the approved draft bank ────────────────────────────────────────
+
+class DraftExamBody(BaseModel):
+    candidateSpec: str | None = None   # agents/backend.get_backend spec string
+    agentId: str | None = None
+    dryRun: bool = True                # deterministic stub model (CI-safe)
+
+
+@router.post("/drafts/{draft_id}/exam",
+             dependencies=[Depends(require_role("admin", "sme"))])
+def run_draft_exam(draft_id: str, body: DraftExamBody, request: Request):
+    """Run the certification exam against the org's APPROVED draft bank.
+
+    Job-backed exactly like POST /atp/exams/run: returns {jobId}; poll
+    GET /jobs/{jobId} — when status='done' the result is the award dict.
+    The bank is resolved through atp/exams.py's Phase 6 loader path
+    (ORG_CERT_PREFIX ref → atp/knowledge.py in-memory bank; judge/candidate
+    separation and evidence signing apply unchanged).
+    """
+    from agents import jobs
+
+    org = _org(request)
+    draft = _http_errors(knowledge.get_draft, org, draft_id)
+    if draft is None:
+        raise HTTPException(404, f"draft {draft_id} not found")
+    if draft["status"] != "approved":
+        raise HTTPException(
+            409, f"draft {draft_id} is '{draft['status']}' — only an "
+                 f"SME-approved draft can be examined")
+
+    cert_ref = knowledge.EXAM_CERT_PREFIX + draft_id
+    candidate_spec = body.candidateSpec
+    agent_id = body.agentId
+    dry_run = body.dryRun
+
+    def _task():
+        from atp import exams
+        return exams.run_exam(
+            cert_ref,
+            candidate_spec,
+            org_id=org,
+            dry_run=dry_run,
+            agent_id=agent_id,
+        )
+
+    job_id = jobs.submit("l5_draft_exam", _task, org_id=org)
+    # jobId is the documented key; job_id keeps BU_API.runJob() compatible.
+    return {"jobId": job_id, "job_id": job_id, "certRef": cert_ref}

api/ops.py

@@ -0,0 +1,496 @@
+"""
+Runtime ops for the Brain University API — Phase 5 (docs/HARDENING.md).
+
+Three concerns, one middleware (`ops_middleware`) plus one renderer
+(`metrics_response` for the /metrics route in api/server.py):
+
+1. RATE LIMITING — in-process token bucket, keyed (client IP, token subject).
+   * Env `BU_RATE_LIMIT` = "<requests>/<window_seconds>", default "120/60"
+     (120 requests per rolling 60s per key). `BU_RATE_LIMIT=0` disables.
+     Malformed values warn to stderr and fall back to the default (an ops
+     typo must never boot an unlimited OR a bricked API).
+   * The middleware is registered so it runs AFTER auth (see api/server.py):
+     request.state.user is already populated, so one NATed office IP full of
+     distinct authenticated users doesn't share a single bucket — but an
+     unauthenticated scanner (no subject) still collapses to its IP.
+   * Over-limit → 429 with a `Retry-After: <seconds>` header (seconds until
+     the bucket next holds a full token).
+   * Exempt: /health (liveness probes must never 429), /metrics (scrapers),
+     and OPTIONS (CORS preflight carries no credentials to key on).
+   * Buckets live in a bounded LRU (`MAX_BUCKET_KEYS` = 10k keys, least-
+     recently-seen evicted) so hostile key churn can't grow memory unbounded.
+   * Client IP is request.client.host — behind a reverse proxy (Render,
+     Netlify redirects) that is the proxy's address unless uvicorn runs with
+     --proxy-headers (or ProxyHeadersMiddleware) so X-Forwarded-For rewrites
+     scope['client']. Without it, all remote clients share one IP and the
+     bucket key degrades to per-token-subject — still functional, coarser.
+   * SCOPE: single-instance only, by design. Buckets are per-process memory —
+     N instances behind a load balancer enforce N×limit, and a restart resets
+     all buckets. Good enough for the current single-box Render deploy; for
+     multi-instance, swap `TokenBucketLimiter` for a Redis-backed bucket
+     (INCR + EXPIRE, or a Lua token bucket) — the middleware contract
+     (`acquire(key) -> (allowed, retry_after_s)`) is the seam.
+   * BU_AUTH_DISABLED=1 demo mode is untouched: the limiter still runs (it
+     needs no auth), keyed on (ip, dev-admin); set BU_RATE_LIMIT=0 to switch
+     it off entirely.
+
+2. /metrics — Prometheus text exposition format 0.0.4, HAND-ROLLED on
+   purpose: no `prometheus-client` dependency (documented decision — the
+   series below are counters/gauges/one histogram with a small fixed label
+   set; a registry library buys nothing and requirements.txt is heavy
+   enough). Series:
+     bu_up 1                                     — process serving requests
+     bu_requests_total{method,path,status}       — counter
+     bu_request_duration_seconds*{path,status}   — histogram (+_sum/_count)
+     bu_rate_limited_total                       — counter (429s issued here)
+     bu_jobs{status}                             — gauge, DB, cached 5s
+     bu_evidence_chain_length                    — gauge, DB, cached 30s
+     bu_licenses{status}                         — gauge, DB, cached 30s
+   Label cardinality is BOUNDED: `path` is always the ROUTE TEMPLATE
+   (`/atp/evidence/{evidence_id}`, never the raw id), and any path that
+   matches no registered route collapses to the single sentinel
+   `/_unmatched` — a scanner spraying random URLs cannot mint new series.
+   DB-backed gauges query through atp/tenant_db (org-demo scope — cheap
+   bootstrap-org aggregates, NOT cross-tenant sums) and every failure is
+   swallowed: a broken DB drops those series from the scrape, it never 500s
+   /metrics.
+   VISIBILITY TRADEOFF: /metrics is added to the auth allowlist by default
+   because Prometheus scrapers don't hold bearers and the payload carries no
+   tenant data (aggregate counts + route templates only). It DOES leak
+   coarse operational shape (traffic volume, error rates, license counts) to
+   anyone who can reach the port — set `BU_METRICS_PUBLIC=0` to keep it
+   behind the bearer gate and give the scraper a token instead.
+
+3. STRUCTURED REQUEST LOGS — one JSON object per line to stdout:
+     {"ts", "method", "path", "status", "ms", "org", "user", "ip"}
+   * `path` is the route template (same bounded collapse as metrics).
+   * Opt-in via `BU_LOG_JSON=1` (default OFF so `uvicorn --reload` dev logs
+     stay human-readable). Read per-request, so tests can flip it.
+   * REDACTION BY CONSTRUCTION: the log record is built from the fixed field
+     list above — request bodies, query strings, and headers (Authorization,
+     X-ATP-License-Key, cookies, ...) are never read by the logger at all.
+     Keep it that way: add fields to `_JSON_LOG_FIELDS`, never dump
+     request.headers / body. Covered by a redaction test.
+
+Wiring (api/server.py): `app.middleware("http")(ops_middleware)` registered
+BETWEEN the access-log middleware and auth_middleware. Starlette runs the
+last-registered http middleware first, so the runtime order is
+auth → ops → tenant-access-log �� CORS → routes: identity is available for
+the bucket key, and a 429 short-circuits BEFORE the tenant access log, so a
+flood can't amplify into per-request DB writes.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import os
+import sys
+import threading
+import time
+from collections import OrderedDict
+from datetime import datetime, timezone
+
+from fastapi import Request
+from fastapi.responses import JSONResponse, PlainTextResponse
+
+# ── Config ─────────────────────────────────────────────────────────────────
+
+DEFAULT_RATE_LIMIT = "120/60"
+MAX_BUCKET_KEYS = 10_000
+
+#: Paths the rate limiter never touches (liveness probes + scrapers).
+RATE_LIMIT_EXEMPT_PATHS = frozenset({"/health", "/metrics"})
+
+#: /metrics on the auth allowlist? Default yes — see module docstring
+#: tradeoff. BU_METRICS_PUBLIC=0 keeps it behind the bearer gate.
+METRICS_PUBLIC = os.environ.get("BU_METRICS_PUBLIC", "1") != "0"
+
+METRICS_CONTENT_TYPE = "text/plain; version=0.0.4; charset=utf-8"
+
+#: The ONLY fields a JSON request-log line may carry (redaction by
+#: construction — see module docstring §3).
+_JSON_LOG_FIELDS = ("ts", "method", "path", "status", "ms", "org", "user", "ip")
+
+
+def _log_json_enabled() -> bool:
+    """BU_LOG_JSON read per-request: costs ~µs, keeps tests/env flips live."""
+    return os.environ.get("BU_LOG_JSON", "") == "1"
+
+
+# ── Token-bucket rate limiter (bounded LRU of buckets) ─────────────────────
+
+class TokenBucketLimiter:
+    """Classic token bucket per key, stored in a size-bounded LRU.
+
+    capacity = `limit` tokens, refilled continuously at limit/window per
+    second — a full burst of `limit` is allowed instantly, sustained rate
+    converges to limit/window. Single-process scope (module docstring §1).
+    """
+
+    def __init__(self, limit: int, window_s: int,
+                 max_keys: int = MAX_BUCKET_KEYS):
+        if limit <= 0 or window_s <= 0:
+            raise ValueError("limit and window must be positive")
+        self.limit = limit
+        self.window_s = window_s
+        self.capacity = float(limit)
+        self.rate = limit / window_s          # tokens per second
+        self.max_keys = max_keys
+        self._buckets: OrderedDict[str, tuple[float, float]] = OrderedDict()
+        self._lock = threading.Lock()         # sync routes run off-loop; be safe
+
+    def acquire(self, key: str, now: float | None = None) -> tuple[bool, int]:
+        """Try to take one token for `key` → (allowed, retry_after_seconds).
+
+        retry_after is 0 when allowed, else the whole seconds until the
+        bucket next holds >= 1 token (what we put in Retry-After).
+        """
+        if now is None:
+            now = time.monotonic()
+        with self._lock:
+            tokens, last = self._buckets.get(key, (self.capacity, now))
+            tokens = min(self.capacity, tokens + (now - last) * self.rate)
+            if tokens >= 1.0:
+                tokens -= 1.0
+                allowed, retry = True, 0
+            else:
+                allowed = False
+                retry = max(1, math.ceil((1.0 - tokens) / self.rate))
+            self._buckets[key] = (tokens, now)
+            self._buckets.move_to_end(key)
+            while len(self._buckets) > self.max_keys:   # bounded memory
+                self._buckets.popitem(last=False)       # evict least-recent
+            return allowed, retry
+
+    def __len__(self) -> int:
+        return len(self._buckets)
+
+
+def _parse_rate_limit(spec: str) -> tuple[int, int] | None:
+    """'120/60' → (120, 60); '0' → None (disabled). Raises on garbage."""
+    spec = (spec or "").strip()
+    if spec == "0":
+        return None
+    n, _, w = spec.partition("/")
+    limit, window = int(n), int(w)
+    if limit <= 0 or window <= 0:
+        raise ValueError(spec)
+    return limit, window
+
+
+_LIMITER: TokenBucketLimiter | None = None
+
+
+def set_rate_limit(spec: str | None = None) -> None:
+    """(Re)configure the limiter from `spec`, or from env when None.
+
+    Called once at import; exported so tests (and an ops REPL) can retune
+    without reloading the module. Malformed spec → warn + default, never
+    crash or silently disable.
+    """
+    global _LIMITER
+    if spec is None:
+        spec = os.environ.get("BU_RATE_LIMIT", DEFAULT_RATE_LIMIT)
+    try:
+        parsed = _parse_rate_limit(spec)
+    except (ValueError, TypeError):
+        print(f"WARNING: BU_RATE_LIMIT={spec!r} is malformed — expected "
+              f"'<requests>/<seconds>' or '0'; using default "
+              f"{DEFAULT_RATE_LIMIT!r}.", file=sys.stderr)
+        parsed = _parse_rate_limit(DEFAULT_RATE_LIMIT)
+    _LIMITER = None if parsed is None else TokenBucketLimiter(*parsed)
+
+
+set_rate_limit()
+
+
+# ── Metrics registry (stdlib, one lock, bounded labels) ──────────────────��─
+
+#: Histogram bucket upper bounds (seconds). +Inf is implicit.
+HIST_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+
+_KNOWN_METHODS = frozenset(
+    {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"})
+
+_METRICS_LOCK = threading.Lock()
+_STARTED_AT = time.time()
+_REQUESTS: dict[tuple[str, str, int], int] = {}          # (method,path,status)
+_HIST: dict[tuple[str, int], list] = {}                   # (path,status) →
+#                                                  [bucket_counts, sum, count]
+_RATE_LIMITED = 0
+
+
+def _observe(method: str, path_tmpl: str, status: int, elapsed_s: float) -> None:
+    method = method if method in _KNOWN_METHODS else "OTHER"
+    rkey = (method, path_tmpl, status)
+    hkey = (path_tmpl, status)
+    with _METRICS_LOCK:
+        _REQUESTS[rkey] = _REQUESTS.get(rkey, 0) + 1
+        hist = _HIST.get(hkey)
+        if hist is None:
+            hist = _HIST[hkey] = [[0] * (len(HIST_BUCKETS) + 1), 0.0, 0]
+        counts, _, _ = hist
+        for i, le in enumerate(HIST_BUCKETS):
+            if elapsed_s <= le:
+                counts[i] += 1
+        counts[-1] += 1                                    # +Inf
+        hist[1] += elapsed_s
+        hist[2] += 1
+
+
+def _count_rate_limited() -> None:
+    global _RATE_LIMITED
+    with _METRICS_LOCK:
+        _RATE_LIMITED += 1
+
+
+# ── Path templating (bounded label set) ────────────────────────────────────
+
+_ROUTE_TABLE: list | None = None
+
+
+def _path_template(request: Request) -> str:
+    """Collapse the raw path to its route template, else '/_unmatched'.
+
+    Fast path: FastAPI puts the matched route in request.scope['route'].
+    Fallback (unrouted: 429 short-circuits, 404s, OPTIONS): match against
+    every registered route's compiled path_regex — one-time table build.
+    Everything that matches nothing shares ONE sentinel label so arbitrary
+    client paths can never mint new series.
+    """
+    route = request.scope.get("route")
+    fmt = getattr(route, "path_format", None)
+    if fmt:
+        return fmt
+    global _ROUTE_TABLE
+    if _ROUTE_TABLE is None:
+        table = []
+        for r in request.app.routes:      # fully registered before 1st request
+            rx = getattr(r, "path_regex", None)
+            f = getattr(r, "path_format", None)
+            if rx is not None and f:
+                table.append((rx, f))
+        _ROUTE_TABLE = table
+    path = request.url.path
+    for rx, f in _ROUTE_TABLE:
+        if rx.match(path):
+            return f
+    return "/_unmatched"
+
+
+# ── DB-backed gauges (cached; failures drop the series, never 500) ─────────
+
+_DB_CACHE: dict[str, tuple[float, object]] = {}
+_DB_CACHE_LOCK = threading.Lock()
+
+
+def _cached(name: str, ttl_s: float, fn):
+    """Value of fn(), refreshed at most every ttl_s. On error: serve the
+    last good value if any (marked fresh again, so a broken DB is re-probed
+    only once per ttl — /metrics never hammers or propagates)."""
+    now = time.time()
+    with _DB_CACHE_LOCK:
+        hit = _DB_CACHE.get(name)
+        if hit is not None and hit[0] > now:
+            return hit[1]
+    try:
+        value = fn()
+    except Exception:  # noqa: BLE001 — metrics must never take the API down
+        value = hit[1] if hit is not None else None
+    with _DB_CACHE_LOCK:
+        _DB_CACHE[name] = (now + ttl_s, value)
+    return value
+
+
+def _jobs_by_status() -> dict[str, int]:
+    from atp import tenant_db
+    org = tenant_db.DEFAULT_ORG
+    rows = tenant_db.scoped_query(
+        org,
+        "SELECT status, COUNT(*) AS n FROM jobs "
+        "WHERE org_id = :org GROUP BY status",
+        {"org": org},
+    )
+    out = {"queued": 0, "running": 0, "done": 0, "error": 0}
+    for r in rows:
+        out[str(r["status"])] = int(r["n"])
+    return out
+
+
+def _evidence_chain_length() -> int:
+    from atp import tenant_db
+    org = tenant_db.DEFAULT_ORG
+    rows = tenant_db.scoped_query(
+        org,
+        "SELECT COUNT(*) AS n FROM atp_evidence WHERE org_id = :org",
+        {"org": org},
+    )
+    return int(rows[0]["n"])
+
+
+def _licenses_by_status() -> dict[str, int]:
+    from atp import tenant_db
+    org = tenant_db.DEFAULT_ORG
+    rows = tenant_db.scoped_query(
+        org,
+        "SELECT status, COUNT(*) AS n FROM licenses "
+        "WHERE org_id = :org GROUP BY status",
+        {"org": org},
+    )
+    out = {"active": 0, "revoked": 0}     # always emit the two headline series
+    for r in rows:
+        out[str(r["status"])] = int(r["n"])
+    return out
+
+
+# ── Exposition (Prometheus text format 0.0.4) ──────────────────────────────
+
+def _esc(v: str) -> str:
+    """Escape a label value per the exposition format."""
+    return v.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
+
+
+def render_metrics() -> str:
+    lines: list[str] = []
+    add = lines.append
+
+    add("# HELP bu_up 1 while the API process is up and serving.")
+    add("# TYPE bu_up gauge")
+    add("bu_up 1")
+    add("# HELP bu_process_start_time_seconds Unix time the process started.")
+    add("# TYPE bu_process_start_time_seconds gauge")
+    add(f"bu_process_start_time_seconds {_STARTED_AT:.3f}")
+
+    with _METRICS_LOCK:
+        requests = dict(_REQUESTS)
+        hist = {k: [list(v[0]), v[1], v[2]] for k, v in _HIST.items()}
+        rate_limited = _RATE_LIMITED
+
+    add("# HELP bu_requests_total HTTP requests by method, route template, "
+        "and status.")
+    add("# TYPE bu_requests_total counter")
+    for (method, path, status), n in sorted(requests.items(),
+                                            key=lambda kv: kv[0]):
+        add(f'bu_requests_total{{method="{_esc(method)}",path="{_esc(path)}",'
+            f'status="{status}"}} {n}')
+
+    add("# HELP bu_request_duration_seconds HTTP request latency by route "
+        "template and status.")
+    add("# TYPE bu_request_duration_seconds histogram")
+    for (path, status), (counts, total, count) in sorted(
+            hist.items(), key=lambda kv: kv[0]):
+        labels = f'path="{_esc(path)}",status="{status}"'
+        for i, le in enumerate(HIST_BUCKETS):
+            add(f'bu_request_duration_seconds_bucket{{{labels},le="{le}"}} '
+                f"{counts[i]}")
+        add(f'bu_request_duration_seconds_bucket{{{labels},le="+Inf"}} '
+            f"{counts[-1]}")
+        add(f"bu_request_duration_seconds_sum{{{labels}}} {total:.6f}")
+        add(f"bu_request_duration_seconds_count{{{labels}}} {count}")
+
+    add("# HELP bu_rate_limited_total Requests rejected with 429 by the "
+        "token-bucket limiter.")
+    add("# TYPE bu_rate_limited_total counter")
+    add(f"bu_rate_limited_total {rate_limited}")
+
+    jobs = _cached("jobs", 5.0, _jobs_by_status)
+    if jobs is not None:
+        add("# HELP bu_jobs Background jobs by status (org-demo scope, "
+            "cached 5s).")
+        add("# TYPE bu_jobs gauge")
+        for status, n in sorted(jobs.items()):
+            add(f'bu_jobs{{status="{_esc(status)}"}} {n}')
+
+    chain = _cached("evidence_chain", 30.0, _evidence_chain_length)
+    if chain is not None:
+        add("# HELP bu_evidence_chain_length Rows in the signed evidence "
+            "chain (org-demo scope, cached 30s).")
+        add("# TYPE bu_evidence_chain_length gauge")
+        add(f"bu_evidence_chain_length {chain}")
+
+    licenses = _cached("licenses", 30.0, _licenses_by_status)
+    if licenses is not None:
+        add("# HELP bu_licenses Marketplace licenses by status (org-demo "
+            "scope, cached 30s).")
+        add("# TYPE bu_licenses gauge")
+        for status, n in sorted(licenses.items()):
+            add(f'bu_licenses{{status="{_esc(status)}"}} {n}')
+
+    return "\n".join(lines) + "\n"
+
+
+def metrics_response() -> PlainTextResponse:
+    """Response for the /metrics route (api/server.py owns the route def)."""
+    return PlainTextResponse(render_metrics(), media_type=METRICS_CONTENT_TYPE)
+
+
+# ── JSON request log ───────────────────────────────────────────────────────
+
+def _emit_json_log(request: Request, path_tmpl: str, status: int,
+                   elapsed_s: float) -> None:
+    """One JSON line to stdout. Fields fixed by _JSON_LOG_FIELDS — never add
+    headers or bodies here (redaction by construction; see docstring §3)."""
+    record = {
+        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
+        "method": request.method,
+        "path": path_tmpl,
+        "status": status,
+        "ms": round(elapsed_s * 1000.0, 2),
+        "org": getattr(request.state, "org_id", None),
+        "user": getattr(request.state, "user", None),
+        "ip": request.client.host if request.client else None,
+    }
+    assert set(record) == set(_JSON_LOG_FIELDS)
+    print(json.dumps(record, separators=(",", ":")), file=sys.stdout,
+          flush=True)
+
+
+# ── The middleware ─────────────────────────────────────────────────────────
+
+async def ops_middleware(request: Request, call_next):
+    """Rate limit (post-auth identity) + metrics + optional JSON log.
+
+    Runs INSIDE auth (api/server.py registration order), so
+    request.state.user/org_id are set for authenticated requests and 401s
+    never reach us. A 429 returns straight from here — inner layers (tenant
+    access log, CORS, routes) never run for rate-limited requests.
+    """
+    start = time.perf_counter()
+    method = request.method
+    raw_path = request.url.path
+
+    limiter = _LIMITER
+    if (limiter is not None and method != "OPTIONS"
+            and raw_path not in RATE_LIMIT_EXEMPT_PATHS):
+        ip = request.client.host if request.client else "unknown"
+        subject = getattr(request.state, "user", None) or "-"
+        allowed, retry_after = limiter.acquire(f"{ip}|{subject}")
+        if not allowed:
+            _count_rate_limited()
+            elapsed = time.perf_counter() - start
+            tmpl = _path_template(request)
+            _observe(method, tmpl, 429, elapsed)
+            if _log_json_enabled():
+                _emit_json_log(request, tmpl, 429, elapsed)
+            return JSONResponse(
+                {"detail": "rate limit exceeded"},
+                status_code=429,
+                headers={"Retry-After": str(retry_after)},
+            )
+
+    try:
+        response = await call_next(request)
+    except Exception:
+        elapsed = time.perf_counter() - start
+        tmpl = _path_template(request)
+        _observe(method, tmpl, 500, elapsed)
+        if _log_json_enabled():
+            _emit_json_log(request, tmpl, 500, elapsed)
+        raise
+    elapsed = time.perf_counter() - start
+    tmpl = _path_template(request)
+    _observe(method, tmpl, response.status_code, elapsed)
+    if _log_json_enabled():
+        _emit_json_log(request, tmpl, response.status_code, elapsed)
+    return response

api/server.py

@@ -35,7 +35,7 @@ from pathlib import Path
 from typing import Any
 
 try:
-    from fastapi import FastAPI, HTTPException, UploadFile, File, Form
+    from fastapi import FastAPI, HTTPException, Request, UploadFile, File, Form
     from fastapi.middleware.cors import CORSMiddleware
     from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
     from pydantic import BaseModel
@@ -64,12 +64,139 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# ── Tenant access audit (Phase 2 — docs/TENANCY.md §4 layer 4) ────────────
+# Starlette runs the LAST-registered http middleware FIRST, so registering
+# this BEFORE auth_middleware (below) makes it the inner layer: on the
+# request path it runs AFTER auth — request.state.{user,org_id,role} are
+# already populated and 401-rejected requests never reach it. Rows are
+# written fire-and-forget on a single-worker pool so the response is never
+# blocked, and every failure (including atp/tenant_db.py not being deployed
+# yet) is swallowed — the audit trail must never take the API down.
+from concurrent.futures import ThreadPoolExecutor as _AccessLogPool
+
+_ACCESS_LOG_POOL = _AccessLogPool(max_workers=1, thread_name_prefix="bu-accesslog")
+
+
+@app.middleware("http")
+async def access_log_middleware(request: Request, call_next):
+    response = await call_next(request)
+    try:
+        import api.auth as _auth
+        if _auth.AUTH_DISABLED:          # dev mode: no audit rows
+            return response
+        if _auth.is_public(request.url.path, request.method):
+            return response              # public surface: not tenant data
+        user = getattr(request.state, "user", None)
+        if user is None:                 # unauthenticated (shouldn't happen
+            return response              # behind the auth middleware)
+        org_id = getattr(request.state, "org_id", None) or "org-demo"
+        role = getattr(request.state, "role", None) or "admin"
+        method, path = request.method, request.url.path
+        status = response.status_code
+
+        def _write() -> None:
+            try:
+                from atp import tenant_db
+                tenant_db.log_access(
+                    org_id=org_id, user=user, role=role,
+                    method=method, path=path, status=status,
+                )
+            except Exception:  # noqa: BLE001 — fire-and-forget by contract
+                pass
+
+        _ACCESS_LOG_POOL.submit(_write)
+    except Exception:  # noqa: BLE001 — auditing must never block a response
+        pass
+    return response
+
+
+# ── Ops middleware (Phase 5 — api/ops.py: rate limit + metrics + logs) ────
+# Registered AFTER access_log_middleware and BEFORE auth_middleware, so the
+# runtime order (last-registered runs first) is:
+#     auth → ops → tenant-access-log → CORS → routes
+# i.e. the token-bucket key can include the verified token subject, 401s
+# never reach the limiter, and a 429 short-circuits BEFORE the tenant access
+# log (a flood can't amplify into per-request DB writes). Env knobs:
+# BU_RATE_LIMIT ('120/60'; '0' disables), BU_LOG_JSON=1, BU_METRICS_PUBLIC=0.
+from api import ops as _ops
+app.middleware("http")(_ops.ops_middleware)
+
 # Bearer-token auth on every route except an allowlist (/health, /login,
 # /docs, /videos, /media, OPTIONS preflight). Configure via env vars
 # BU_AUTH_USER / BU_AUTH_PASSWORD / BU_AUTH_SECRET. See api/auth.py.
 from api.auth import auth_middleware, mint_token, check_password
 app.middleware("http")(auth_middleware)
 
+# /metrics is on the auth allowlist by default: Prometheus scrapers don't
+# hold bearers and the exposition carries no tenant payload (aggregate
+# counts + route templates only — see api/ops.py §2 for the tradeoff).
+# Set BU_METRICS_PUBLIC=0 to keep it behind the bearer gate instead.
+if _ops.METRICS_PUBLIC:
+    import api.auth as _auth_mod
+    _auth_mod.PUBLIC_PATHS.add("/metrics")
+
+# ATP (Agent University) routes — /atp/* (see api/atp.py + docs/ATP.md §5).
+from api.atp import router as atp_router
+app.include_router(atp_router)
+
+# Identity routes (Phase 2 — OIDC code flow, local org accounts, org-scoped
+# JWT sessions; see api/identity.py + docs/TENANCY.md). Guarded import so the
+# legacy single-admin demo keeps booting if this file ships first.
+try:
+    from api.identity import router as identity_router
+except ImportError:  # pragma: no cover — api/identity.py not deployed yet
+    identity_router = None
+    import sys as _sys
+    print("WARNING: api/identity.py not present — identity routes not "
+          "mounted; legacy /login only.", file=_sys.stderr)
+if identity_router is not None:
+    app.include_router(identity_router)
+
+# Billing + licensed expert delivery (Phase 4 — docs/HARDENING.md; see
+# api/billing.py). Guarded import, same idiom as identity above.
+try:
+    from api.billing import experts_router, router as billing_router
+except ImportError:  # pragma: no cover — api/billing.py not deployed yet
+    billing_router = experts_router = None
+    import sys as _sys
+    print("WARNING: api/billing.py not present — billing/expert routes not "
+          "mounted.", file=_sys.stderr)
+if billing_router is not None:
+    app.include_router(billing_router)
+    app.include_router(experts_router)
+    # Auth-allowlist wiring for the two non-bearer surfaces (api/auth.py owns
+    # the sets; these additions live here with the routes they serve):
+    #   * /billing/webhook — Stripe cannot send a bearer; every delivery is
+    #     authenticated by its Stripe-Signature header inside the handler.
+    #   * /experts/*       — authenticated per-request by license key
+    #     (X-ATP-License-Key); the handler hits the DB on EVERY call, so
+    #     revocation is enforced immediately (bearer optional).
+    import api.auth as _auth
+    _auth.PUBLIC_PATHS.add("/billing/webhook")
+    _auth.PUBLIC_PREFIXES = _auth.PUBLIC_PREFIXES + ("/experts/",)
+
+# Company-knowledge ingestion + L5 exam drafting (Phase 6 —
+# docs/HARDENING.md first-customer UX; see api/knowledge.py). Guarded
+# import, same idiom as identity/billing above. Every /knowledge route is
+# bearer-gated (no public-path additions — uploads are T2 crown jewels).
+try:
+    from api.knowledge import router as knowledge_router
+except ImportError:  # pragma: no cover — api/knowledge.py not deployed yet
+    knowledge_router = None
+    import sys as _sys
+    print("WARNING: api/knowledge.py not present — company-knowledge routes "
+          "not mounted.", file=_sys.stderr)
+if knowledge_router is not None:
+    app.include_router(knowledge_router)
+
+
+def _request_org(request: Request) -> str:
+    """Tenant for this request — ONLY ever from verified auth state
+    (request.state.org_id, set by the auth middleware from the JWT; TENANCY.md
+    §4 layer 1), never from client-supplied values. The 'org-demo' fallback
+    covers BU_AUTH_DISABLED dev mode and legacy single-admin tokens."""
+    return getattr(request.state, "org_id", None) or "org-demo"
+
 
 class LoginBody(BaseModel):
     username: str
@@ -95,12 +222,11 @@ def _graph():
     """Build the merged corpus graph + cluster list once and cache."""
     if "G" in _GRAPH_CACHE:
         return _GRAPH_CACHE
-    from graph.corpus_graph import build_corpus_graph
-    from graph.clusters import cluster_graph
-    from graph.foundations import rank_foundations
-    G = build_corpus_graph()
-    clusters = cluster_graph(G)
-    foundations = rank_foundations(G, clusters)
+    from graph.clusters import build_clusters, load_graph
+    from graph.foundations import recommend
+    clusters = build_clusters()          # cached to data/sessions/clusters.json
+    G = load_graph()
+    foundations = recommend(limit=8)
     _GRAPH_CACHE.update(G=G, clusters=clusters, foundations=foundations)
     return _GRAPH_CACHE
 
@@ -268,7 +394,7 @@ def podcast_local_interject(body: LocalInterjectBody):
 # ── Background jobs (async podcast render) ────────────────────────────────
 
 @app.post("/podcast/render/async")
-def podcast_render_async(body: PodcastRenderBody):
+def podcast_render_async(body: PodcastRenderBody, request: Request):
     """Enqueue a cloud podcast render; returns a job_id to poll at /jobs/{id}."""
     from agents import jobs, podcast_video
 
@@ -278,11 +404,12 @@ def podcast_render_async(body: PodcastRenderBody):
             raise RuntimeError("render failed (check moviepy/gTTS install)")
         return {k: str(v.relative_to(PROJECT_ROOT)) for k, v in r.items()}
 
-    return {"job_id": jobs.submit("podcast_render", task)}
+    return {"job_id": jobs.submit("podcast_render", task,
+                                  org_id=_request_org(request))}
 
 
 @app.post("/podcast/local/async")
-def podcast_local_async(body: LocalPodcastBody):
+def podcast_local_async(body: LocalPodcastBody, request: Request):
     """Enqueue a fully-local podcast render; poll at /jobs/{id}."""
     from agents import jobs
     from agents.local_podcast import generate_local
@@ -294,13 +421,17 @@ def podcast_local_async(body: LocalPodcastBody):
             raise RuntimeError("local podcast generation failed")
         return {"mp3": str(p.relative_to(PROJECT_ROOT))}
 
-    return {"job_id": jobs.submit("podcast_local", task)}
+    return {"job_id": jobs.submit("podcast_local", task,
+                                  org_id=_request_org(request))}
 
 
 @app.get("/jobs/{job_id}")
-def job_status(job_id: str):
+def job_status(job_id: str, request: Request):
+    """Job status, scoped to the caller's org (jobs.get only sees the org's
+    own rows — RLS-backed on Postgres). A cross-tenant id 404s with the SAME
+    message as an unknown id — no existence oracle (TENANCY.md #6)."""
     from agents import jobs
-    j = jobs.get(job_id)
+    j = jobs.get(job_id, org_id=_request_org(request))
     if not j:
         raise HTTPException(404, f"no such job {job_id}")
     return j
@@ -311,6 +442,14 @@ def health():
     return {"ok": True}
 
 
+@app.get("/metrics")
+def metrics():
+    """Prometheus text exposition — hand-rolled, bounded label set (Phase 5,
+    api/ops.py). Public by default (BU_METRICS_PUBLIC=0 gates it behind
+    auth); exempt from rate limiting so scrapers never see a 429."""
+    return _ops.metrics_response()
+
+
 @app.get("/graph")
 def graph_endpoint():
     g = _graph()["G"]
@@ -330,7 +469,7 @@ def trailhead(user_id: str = "anon", interest: str = "", n: int = 5):
 
 def _foundation_card(f) -> dict:
     return {
-        "node": getattr(f, "node", str(f)),
+        "node": getattr(f, "raw_name", None) or getattr(f, "name", str(f)),
         "score": getattr(f, "score", None),
         "rationale": getattr(f, "rationale", []),
         "cluster_id": getattr(f, "cluster_id", None),
@@ -465,40 +604,24 @@ def briefing_endpoint(node: str, interest: str = ""):
 @app.post("/socratic/start")
 def socratic_start(body: SocraticStartBody):
     from agents.socratic import start_session
-    from agents.persistence import open_db
+    from agents.persistence import save_socratic_session
     import uuid
     state = start_session(body.node)
     sid = f"soc_{uuid.uuid4().hex[:12]}"
-    db = open_db()
-    db.execute(
-        """INSERT INTO socratic_sessions
-             (session_id, user_id, node, state_json, updated_at)
-           VALUES (?, ?, ?, ?, ?)""",
-        (sid, body.user_id, body.node, json.dumps(state),
-         _now()),
-    )
-    db.commit()
+    save_socratic_session(sid, body.user_id, body.node, state)
     return {"session_id": sid, "state": state}
 
 
 @app.post("/socratic/step")
 def socratic_step(body: SocraticStepBody):
     from agents.socratic import step
-    from agents.persistence import open_db
-    db = open_db()
-    row = db.execute(
-        "SELECT state_json FROM socratic_sessions WHERE session_id = ?",
-        (body.session_id,),
-    ).fetchone()
-    if not row:
+    from agents.persistence import (load_socratic_session,
+                                    update_socratic_session)
+    session = load_socratic_session(body.session_id)
+    if not session:
         raise HTTPException(404, "session not found")
-    state = json.loads(row[0])
-    new_state = step(state, body.answer)
-    db.execute(
-        "UPDATE socratic_sessions SET state_json = ?, updated_at = ? WHERE session_id = ?",
-        (json.dumps(new_state), _now(), body.session_id),
-    )
-    db.commit()
+    new_state = step(session["state"], body.answer)
+    update_socratic_session(body.session_id, new_state)
     return new_state