QONTO-WP-0003-T05: pin REST/MCP audit-schema parity with a test

CapabilityService._emit_audit was already the single audit call site for
both transports since T02, but nothing failed if a future change diverged
one transport's shape. Add tests/test_audit_parity.py: same capability
called through protocol="rest" and protocol="mcp" (allow path and deny
path) must produce identical audit events except request_id/timestamp/
latency_ms (expected to vary) and protocol (expected to differ). Also pins
down that no audit event ever contains a secret-shaped field name.

Confirmed via grep: no State Hub coupling anywhere in src/qonto_assistant/
-- the only audit sink is AuditLogger, so there's no per-call hot-path
write to accidentally wire up.

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

tests/test_audit_parity.py

@@ -0,0 +1,101 @@
+"""QONTO-WP-0003-T05: REST and MCP must emit schema-identical audit events.
+
+CapabilityService._emit_audit is the single audit call site for both
+transports (see service.py) -- these tests pin that down with an explicit
+assertion rather than relying on code inspection, so a future change that
+special-cases either transport's audit shape breaks a test, not just a
+review comment.
+"""
+
+from pathlib import Path
+
+from qonto_assistant.audit import AuditLogger
+from qonto_assistant.contracts import ActorClaims
+from qonto_assistant.errors import PolicyDeniedError
+from qonto_assistant.policy import PolicyEngine
+from qonto_assistant.qonto_client import FixtureQontoClient
+from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
+from qonto_assistant.service import CapabilityService
+
+FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"
+POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
+
+# Fields whose values are expected to vary per call (timing/identifiers) or
+# by design (protocol). Everything else must match exactly between REST and
+# MCP for the same logical call.
+NON_COMPARABLE_FIELDS = {"request_id", "timestamp", "latency_ms", "protocol"}
+SECRET_FIELD_NAMES = {"api_key", "authorization", "authorization_header", "openbao_token", "secret", "token"}
+
+
+def _service() -> tuple[CapabilityService, list[dict[str, object]]]:
+    events: list[dict[str, object]] = []
+    policy = PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False)
+    service = CapabilityService(
+        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
+        policy=policy,
+        audit_logger=AuditLogger(sink=events.append),
+        rate_limiter=RateLimiter(limit=20, window_seconds=60),
+        concurrency_limiter=ConcurrencyLimiter(limit=4),
+    )
+    return service, events
+
+
+def _claims() -> ActorClaims:
+    return ActorClaims(actor_id="parity-test", tenant_id="binky", lane="green")
+
+
+def test_allow_path_audit_schema_identical_across_protocols() -> None:
+    service, events = _service()
+
+    service.get_accounts(claims=_claims(), request_id="req-rest", protocol="rest")
+    service.get_accounts(claims=_claims(), request_id="req-mcp", protocol="mcp")
+
+    rest_event, mcp_event = events
+
+    assert set(rest_event) == set(mcp_event)
+    assert rest_event["protocol"] == "rest"
+    assert mcp_event["protocol"] == "mcp"
+    for field in set(rest_event) - NON_COMPARABLE_FIELDS:
+        assert rest_event[field] == mcp_event[field], f"field {field!r} diverged: {rest_event[field]!r} != {mcp_event[field]!r}"
+
+    assert rest_event["decision"] == "allow"
+    assert rest_event["capability"] == "org_summary"
+
+
+def test_deny_path_audit_schema_identical_across_protocols() -> None:
+    service, events = _service()
+
+    for protocol, request_id in (("rest", "req-rest-deny"), ("mcp", "req-mcp-deny")):
+        try:
+            service.list_transactions(
+                claims=_claims(),
+                request_id=request_id,
+                account_slug=None,
+                page=1,
+                page_size=101,
+                window_days=31,
+                status="completed",
+                side=None,
+                protocol=protocol,
+            )
+        except PolicyDeniedError:
+            pass
+
+    rest_event, mcp_event = events
+
+    assert set(rest_event) == set(mcp_event)
+    for field in set(rest_event) - NON_COMPARABLE_FIELDS:
+        assert rest_event[field] == mcp_event[field]
+
+    assert rest_event["decision"] == "deny"
+    assert rest_event["deny_reason"] == "arg_constraint"
+    assert rest_event["qonto_http_status"] is None
+    assert rest_event["result_count"] is None
+
+
+def test_audit_event_never_contains_secret_field_names() -> None:
+    service, events = _service()
+    service.get_accounts(claims=_claims(), request_id="req-secret-check", protocol="mcp")
+
+    for event in events:
+        assert SECRET_FIELD_NAMES.isdisjoint(event.keys())

workplans/QONTO-WP-0003-mcp-surface.md

@@ -218,7 +218,7 @@ its own scope note. `pytest` → `28 passed` (unchanged).
 
 ```task
 id: QONTO-WP-0003-T05
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "b7d70c3e-fe55-47d9-9180-cd4c0bedc31a"
 ```
@@ -232,6 +232,26 @@ hot-path events — do not wire per-invocation State Hub writes.
 Done when: a test asserts REST and MCP audit records for the same capability
 call are schema-identical modulo protocol field.
 
+**Done 2026-07-23:** This was already true structurally since T02
+(`CapabilityService._emit_audit` is the single audit call site for both
+transports), but it wasn't pinned down by a test — a future change could
+have special-cased one transport's shape without anything failing. Added
+`tests/test_audit_parity.py`: calls the same capability through
+`protocol="rest"` and `protocol="mcp"` for both an allow path
+(`org_summary`) and a deny path (`list_transactions` oversized `page_size`
+→ `arg_constraint`), then asserts the two events have identical key sets
+and identical values on every field except `request_id`/`timestamp`/
+`latency_ms` (expected to vary per call) and `protocol` (expected to
+differ by design). A third test asserts no audit event ever contains a
+secret-shaped field name (`api_key`, `authorization`, `token`, etc.) —
+belt-and-suspenders alongside `tests/test_audit.py`'s existing redaction
+test.
+
+Confirmed (`grep`) there is no State Hub coupling anywhere in
+`src/qonto_assistant/`: the only audit sink is `AuditLogger` (structured
+stdout JSON, or a test-injected `sink` callable) — no per-call State Hub
+write exists to accidentally wire up. `pytest` → `31 passed`.
+
 ## Task: MCP smoke path + operator runbook update
 
 ```task