fix(api): sanitize batch gather escapes, admin-route authz walk, production docs gate (audit G-5, G-4, G-2)

CONTRIBUTING.md

@@ -73,6 +73,23 @@ python scripts/check_performance.py --baseline docs/benchmark-baseline.json --cu
 python scripts/generate_contracts.py --check
 ```
 
+## Dependabot pip PRs and `uv.lock`
+
+Dependabot bumps grouped pip dependencies in `pyproject.toml` **without**
+regenerating `uv.lock`, so the `lock-check` gate (`uv lock --check`) fails on
+every such PR. Bring the lockfile along by hand:
+
+```bash
+gh pr checkout <pr-number>
+uv lock
+git add uv.lock && git commit -m "chore(deps): regenerate uv.lock for the group bump"
+git push
+gh pr update-branch <pr-number>   # branch protection requires up-to-date branches
+```
+
+Then let CI finish and merge as usual. GitHub Actions and npm bumps do not
+need this — only the pip ecosystem is locked with uv.
+
 ## Architecture decisions
 
 Significant design changes should include an ADR in `docs/decisions/`.

src/serving/api/main.py

@@ -506,6 +506,22 @@ async def demo_mode_guard(request: Request, call_next: RequestResponseEndpoint)
     return await call_next(request)
 
 
+# Interactive docs and the schema are auth-exempt (`_is_exempt_path`), which
+# on a production deployment hands any caller the full route map without a
+# key (audit G-2). Same policy as the CORS wildcard: demo/dev convenience,
+# refused on the production profile.
+_DOCS_PATH_PREFIXES = ("/docs", "/redoc", "/openapi")
+
+
+@app.middleware("http")
+async def production_docs_guard(request: Request, call_next: RequestResponseEndpoint) -> Response:
+    if request.url.path.startswith(_DOCS_PATH_PREFIXES) and (
+        getattr(request.app.state, "profile", "dev") == "production"
+    ):
+        return JSONResponse(status_code=404, content={"detail": "Not found."})
+    return await call_next(request)
+
+
 # Registered last so it wraps every other HTTP middleware and observes the
 # final response status code; route template is populated by the router after
 # call_next() returns. Backs agentflow-api-health.json + api-5xx-spike.md.

src/serving/api/routers/batch.py

@@ -41,6 +41,21 @@ def _safe_item_error(exc: Exception, req: Request) -> str:
     return str(exc)
 
 
+def _unexpected_outcome_error(outcome: object, req: Request) -> str:
+    """Client-safe text for a gather outcome that is not a ``BatchResult``.
+
+    ``_execute_item`` converts every ``Exception`` into a ``BatchResult``
+    itself, so anything landing here escaped that controlled path (an
+    unexpected ``BaseException`` or a bug). Its text has not been through
+    ``_safe_item_error`` and may carry raw engine detail — never echo it to
+    the client (S-2 class, audit G-5); log it under the correlation id.
+    """
+    correlation_id = getattr(req.state, "correlation_id", None)
+    logger.error("batch_unexpected_outcome", detail=repr(outcome), correlation_id=correlation_id)
+    ref = f" (ref {correlation_id})" if correlation_id else ""
+    return f"batch item failed{ref}"
+
+
 class BatchItem(BaseModel):
     id: str
     type: Literal["entity", "metric", "query"]
@@ -214,7 +229,7 @@ async def batch_query(request: BatchRequest, req: Request) -> BatchResponse:
     results = [
         outcome
         if isinstance(outcome, BatchResult)
-        else BatchResult(id=item.id, status="error", error=str(outcome))
+        else BatchResult(id=item.id, status="error", error=_unexpected_outcome_error(outcome, req))
         for item, outcome in zip(request.requests, outcomes, strict=False)
     ]
     return BatchResponse(

tests/unit/test_admin_route_authz_matrix.py

@@ -10,11 +10,11 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 
 from fastapi.params import Depends
 
-from src.serving.api.auth.middleware import _is_exempt_path, require_admin_key
+from src.serving.api.auth.middleware import _is_admin_path, _is_exempt_path, require_admin_key
 from src.serving.api.routers.admin import router as admin_router
 from src.serving.api.routers.admin_ui import router as admin_ui_router
 from src.serving.node import ingest as ingest_module
@@ -55,3 +55,92 @@ def test_node_events_is_auth_middleware_exempt_not_open() -> None:
     source = inspect.getsource(ingest_module.ingest_node_events)
     assert "compare_digest" in source
     assert "_extract_bearer" in source
+
+
+# ── G-4: walk the ASSEMBLED app, not just the two known router objects ──────
+#
+# The router pins above can't see a THIRD admin router registered without the
+# dependency — that route would ship fully open (middleware skips admin paths
+# by design). The walk below closes that gap; the probe test below it proves
+# the checker is able to fail (a detector that can't fail proves nothing).
+
+
+def _iter_leaf_routes(
+    routes: list, prefix: str, inherited: tuple
+) -> Iterator[tuple[str, object, tuple]]:
+    """Yield ``(full_path, route, inherited_dependency_calls)`` leaves.
+
+    FastAPI ≥0.139 no longer flattens ``include_router`` into ``app.routes``:
+    the app holds lazy ``_IncludedRouter`` nodes (no ``path`` attribute) whose
+    children live in ``.original_router.routes`` with the router's own prefix
+    already applied; the include-time prefix/dependencies sit on
+    ``.include_context``. Older FastAPI keeps flattened ``APIRoute`` lists —
+    this walk handles both.
+    """
+    for route in routes:
+        context = getattr(route, "include_context", None)
+        original = getattr(route, "original_router", None)
+        if context is not None and original is not None:
+            include_deps = tuple(
+                _dependency_calls(list(getattr(context, "dependencies", []) or []))
+            )
+            yield from _iter_leaf_routes(
+                original.routes,
+                prefix + (getattr(context, "prefix", "") or ""),
+                inherited + include_deps,
+            )
+            continue
+        yield prefix + getattr(route, "path", ""), route, inherited
+
+
+def _admin_routes_missing_admin_key(app: object) -> list[str]:
+    from fastapi.routing import APIRoute
+
+    missing: list[str] = []
+    for path, route, inherited in _iter_leaf_routes(app.routes, "", ()):  # type: ignore[attr-defined]
+        if not _is_admin_path(path):
+            continue
+        if not isinstance(route, APIRoute):
+            # A mount / raw Starlette route under an admin prefix has no
+            # FastAPI dependency chain at all — flag it.
+            missing.append(path)
+            continue
+        calls = _dependency_calls(list(route.dependant.dependencies)) | set(inherited)
+        if require_admin_key not in calls:
+            missing.append(path)
+    return missing
+
+
+def test_every_assembled_admin_route_requires_admin_key() -> None:
+    from src.serving.api.main import app
+
+    admin_paths = [
+        path
+        for path, _route, _inherited in _iter_leaf_routes(app.routes, "", ())
+        if _is_admin_path(path)
+    ]
+    assert len(admin_paths) >= 2, "expected admin routes on the assembled app"
+    assert _admin_routes_missing_admin_key(app) == []
+
+
+def test_admin_key_checker_flags_an_unprotected_route() -> None:
+    from fastapi import APIRouter, FastAPI
+    from fastapi import Depends as FastDepends
+
+    probe = FastAPI()
+    open_router = APIRouter(prefix="/v1/admin")
+
+    @open_router.get("/leak")
+    def _leak() -> dict:  # pragma: no cover — never called
+        return {}
+
+    guarded_router = APIRouter(prefix="/admin", dependencies=[FastDepends(require_admin_key)])
+
+    @guarded_router.get("/ok")
+    def _ok() -> dict:  # pragma: no cover — never called
+        return {}
+
+    probe.include_router(open_router)
+    probe.include_router(guarded_router)
+
+    assert _admin_routes_missing_admin_key(probe) == ["/v1/admin/leak"]

tests/unit/test_batch_unit.py

@@ -450,3 +450,34 @@ async def test_batch_query_reports_mixed_results_and_duration() -> None:
     assert [r.id for r in response.results] == ["ok-entity", "bad-metric", "ok-query"]
     assert [r.status for r in response.results] == ["ok", "error", "ok"]
     assert response.duration_ms >= 0
+
+
+async def test_batch_query_gather_fallback_never_echoes_exception_text(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A non-BatchResult gather outcome escaped `_execute_item`'s own Exception
+    handling, so its text never went through `_safe_item_error` — the client
+    must see a generic message, not raw engine detail (S-2 class, audit G-5)."""
+    engine = _ModernEngine(entity={"order_id": "ORD-1"})
+    req = _make_req(engine)
+
+    async def _escape(item: BatchItem, _req: Any) -> BatchResult:
+        raise RuntimeError("secret-dsn clickhouse://internal-host:8123/prod")
+
+    monkeypatch.setattr(batch_module, "_execute_item", _escape)
+    request = BatchRequest(
+        requests=[
+            BatchItem(
+                id="i-1", type="entity", params={"entity_type": "order", "entity_id": "ORD-1"}
+            )
+        ]
+    )
+
+    response = await batch_query(request, req)
+
+    (result,) = response.results
+    assert result.status == "error"
+    assert result.error is not None
+    assert "secret-dsn" not in result.error
+    assert "internal-host" not in result.error
+    assert result.error.startswith("batch item failed")

tests/unit/test_docs_profile_gate.py

@@ -0,0 +1,67 @@
+"""Production profile hides interactive docs and the OpenAPI schema (audit G-2).
+
+``/docs``, ``/openapi*`` are auth-exempt (``_is_exempt_path``), so a
+production deployment would hand any unauthenticated caller the full route
+map. ``production_docs_guard`` 404s ``/docs``, ``/redoc`` and ``/openapi*``
+on ``profile=production`` only — demo/dev keep the local DX, mirroring the
+CORS-wildcard policy (P2-3).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.serving.api.auth import AuthManager
+from src.serving.api.main import app
+
+
+@pytest.fixture(autouse=True)
+def _auth_manager() -> Iterator[None]:
+    # TestClient without the lifespan: provide the auth manager the
+    # middleware chain expects (same idiom as the CORS tests).
+    state = app.state._state
+    sentinel = object()
+    prev = state.get("auth_manager", sentinel)
+    manager = AuthManager()
+    manager.load()
+    state["auth_manager"] = manager
+    yield
+    if prev is sentinel:
+        state.pop("auth_manager", None)
+    else:
+        state["auth_manager"] = prev
+
+
+@contextmanager
+def _profile(value: str) -> Iterator[None]:
+    # The guard reads app.state.profile per request (set by lifespan in real
+    # boots); pin it directly and restore so no state leaks across tests.
+    state = app.state._state
+    sentinel = object()
+    prev = state.get("profile", sentinel)
+    state["profile"] = value
+    try:
+        yield
+    finally:
+        if prev is sentinel:
+            state.pop("profile", None)
+        else:
+            state["profile"] = prev
+
+
+def test_docs_and_schema_stay_visible_outside_production() -> None:
+    client = TestClient(app)
+    with _profile("dev"):
+        assert client.get("/docs").status_code == 200
+        assert client.get("/openapi.json").status_code == 200
+
+
+def test_production_profile_hides_docs_and_schema() -> None:
+    client = TestClient(app)
+    with _profile("production"):
+        for path in ("/docs", "/redoc", "/openapi.json"):
+            assert client.get(path).status_code == 404, path