feat: fresh-authorize and audit Human Gate decisions

src/main.py

@@ -561,6 +561,7 @@ class QuietHoursPayload(BaseModel):
 
 
 StepUpAction = Literal[
+    "decide_human_gate",
     "merge_pull",
     "delete_source_branch",
     "prepare_release_rollback",
@@ -1884,11 +1885,35 @@ async def decide_human_gate(
     payload: HumanGateDecision,
     request: Request,
     idempotency_key: str = Header(alias="Idempotency-Key", min_length=1, max_length=128),
+    step_up_grant: str | None = Header(
+        default=None, alias="X-Step-Up-Grant", max_length=128
+    ),
 ):
+    await _require_step_up(
+        request,
+        step_up_grant,
+        action="decide_human_gate",
+        target=gate_id,
+    )
     login = await _human_gate_login(request)
+    journal = _security_event_store()
+    operation_id = None
     try:
         store = _human_gate_store()
         existed = await asyncio.to_thread(store.has_receipt_key, login, idempotency_key)
+        if not existed:
+            try:
+                operation_id = await asyncio.to_thread(
+                    journal.reserve,
+                    "human_gate_decision",
+                    method=payload.decision,
+                    target=gate_id,
+                )
+            except SecurityEventStoreError as error:
+                raise HTTPException(
+                    status_code=503,
+                    detail="Security activity is temporarily unavailable",
+                ) from error
         receipt = await asyncio.to_thread(
             store.decide,
             login,
@@ -1901,7 +1926,14 @@ async def decide_human_gate(
             idempotency_key=idempotency_key,
         )
     except (GateValidationError, GateConflict, GateNotFound, sqlite3.Error) as error:
+        if operation_id is not None:
+            try:
+                await asyncio.to_thread(journal.discard, operation_id)
+            except SecurityEventStoreError:
+                pass
         raise _gate_error(error) from error
+    if operation_id is not None:
+        await asyncio.to_thread(journal.finalize, operation_id)
     return JSONResponse(
         receipt, status_code=200 if existed else 201, headers={"Cache-Control": "no-store"}
     )

tests/test_human_gate_api.py

@@ -5,6 +5,7 @@ import pytest
 
 from src import main
 from src.human_gate_store import HumanGateStore
+from src.security_event_store import SecurityEventStore, SecurityEventStoreError
 
 
 CANDIDATE = {
@@ -68,6 +69,156 @@ async def test_decision_requires_revision_and_returns_durable_receipt(gate_api):
     assert all(response.headers["cache-control"] == "no-store" for response in (decided, repeated, receipt, stale))
 
 
+@pytest.mark.anyio
+async def test_decision_requires_fresh_authorization_bound_to_the_exact_gate(monkeypatch, gate_api):
+    gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-authorized")
+    authorization_calls = []
+
+    async def require_step_up(request, grant, *, action, target):
+        authorization_calls.append({"grant": grant, "action": action, "target": target})
+        if grant != "one-time-grant":
+            raise main.HTTPException(
+                status_code=428,
+                detail={
+                    "detail": "Fresh authorization required",
+                    "code": "step_up_required",
+                    "action": action,
+                    "target": target,
+                },
+            )
+
+    monkeypatch.setattr(main, "_require_step_up", require_step_up)
+    payload = {
+        "expected_revision": gate["revision"],
+        "decision": "hold",
+        "reason": "Needs another browser run",
+        "override_reason": "",
+        "checklist": CHECKLIST,
+    }
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        rejected = await client.post(
+            f"/api/v1/human-gates/{gate['id']}/decision",
+            json=payload,
+            headers={"Idempotency-Key": "decision-authorized"},
+        )
+        pending = await client.get(f"/api/v1/human-gates/{gate['id']}")
+        accepted = await client.post(
+            f"/api/v1/human-gates/{gate['id']}/decision",
+            json=payload,
+            headers={
+                "Idempotency-Key": "decision-authorized",
+                "X-Step-Up-Grant": "one-time-grant",
+            },
+        )
+
+    assert rejected.status_code == 428
+    assert pending.json()["state"] == "pending"
+    assert accepted.status_code == 201
+    assert authorization_calls == [
+        {"grant": None, "action": "decide_human_gate", "target": gate["id"]},
+        {"grant": "one-time-grant", "action": "decide_human_gate", "target": gate["id"]},
+    ]
+
+
+@pytest.mark.anyio
+async def test_successful_decision_records_one_completed_privacy_safe_security_event(
+    monkeypatch, gate_api, tmp_path
+):
+    gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-audited")
+    journal_path = tmp_path / "security.sqlite3"
+    journal = SecurityEventStore(journal_path, clock=lambda: 200)
+    monkeypatch.setattr(main, "_security_event_store", lambda: journal)
+    private_reason = "Private launch context must not enter the security journal"
+    payload = {
+        "expected_revision": gate["revision"],
+        "decision": "hold",
+        "reason": private_reason,
+        "override_reason": "",
+        "checklist": CHECKLIST,
+    }
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        decided = await client.post(
+            f"/api/v1/human-gates/{gate['id']}/decision",
+            json=payload,
+            headers={"Idempotency-Key": "decision-audited"},
+        )
+
+    events = journal.list().events
+    assert decided.status_code == 201
+    assert [
+        {"kind": event.kind, "method": event.method, "target": event.target, "status": event.status}
+        for event in events
+    ] == [
+        {
+            "kind": "human_gate_decision",
+            "method": "hold",
+            "target": gate["id"],
+            "status": "completed",
+        }
+    ]
+    assert private_reason.encode() not in journal_path.read_bytes()
+
+
+@pytest.mark.anyio
+async def test_decision_fails_closed_when_security_event_cannot_be_reserved(
+    monkeypatch, gate_api
+):
+    gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-no-journal")
+
+    class UnavailableJournal:
+        def reserve(self, *args, **kwargs):
+            raise SecurityEventStoreError("private journal path")
+
+    monkeypatch.setattr(main, "_security_event_store", UnavailableJournal)
+    payload = {
+        "expected_revision": gate["revision"],
+        "decision": "release",
+        "reason": "",
+        "override_reason": "",
+        "checklist": CHECKLIST,
+    }
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        rejected = await client.post(
+            f"/api/v1/human-gates/{gate['id']}/decision",
+            json=payload,
+            headers={"Idempotency-Key": "decision-no-journal"},
+        )
+        pending = await client.get(f"/api/v1/human-gates/{gate['id']}")
+
+    assert rejected.status_code == 503
+    assert rejected.json() == {"detail": "Security activity is temporarily unavailable"}
+    assert pending.json()["state"] == "pending"
+
+
+@pytest.mark.anyio
+async def test_rejected_decision_discards_its_pending_security_event(
+    monkeypatch, gate_api, tmp_path
+):
+    gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-conflict")
+    journal = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 200)
+    monkeypatch.setattr(main, "_security_event_store", lambda: journal)
+    payload = {
+        "expected_revision": gate["revision"] + 1,
+        "decision": "hold",
+        "reason": "Outdated review",
+        "override_reason": "",
+        "checklist": CHECKLIST,
+    }
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        rejected = await client.post(
+            f"/api/v1/human-gates/{gate['id']}/decision",
+            json=payload,
+            headers={"Idempotency-Key": "decision-conflict"},
+        )
+
+    assert rejected.status_code == 409
+    assert journal.list().events == []
+
+
 @pytest.mark.anyio
 async def test_recycled_login_cannot_read_another_principal_gates(monkeypatch, tmp_path):
     store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: 100)