diff --git a/approval_engine/audit.py b/approval_engine/audit.py index 26053a1..31cdb6d 100644 --- a/approval_engine/audit.py +++ b/approval_engine/audit.py @@ -18,6 +18,13 @@ class AuditDeliveryError(RuntimeError): def audit_envelope(payload: dict[str, Any]) -> dict[str, Any]: details = dict(payload.get("details") or {}) + # audit-core's heartbeat contract puts the asserted class and the assertion + # on `data` itself, not inside a producer-shaped details object. + heartbeat = ( + {"class": details.get("class"), "assertion": details.get("assertion")} + if payload["action"] == "audit-core.heartbeat" + else {} + ) return { "id": payload["event_id"], "type": payload["action"], @@ -34,6 +41,7 @@ def audit_envelope(payload: dict[str, Any]) -> dict[str, Any]: "outcome": payload["outcome"], "reason": payload.get("reason"), "details": details, + **heartbeat, }, } diff --git a/approval_engine/store.py b/approval_engine/store.py index 5172bb3..bbad871 100644 --- a/approval_engine/store.py +++ b/approval_engine/store.py @@ -36,6 +36,12 @@ AUDIT_SCHEMA = "audit-core.event.v1alpha1" SOURCE = "approval-engine" SCOPE = "netkingdom-approvals" EVENT_CLASSES = ("issuance", "use", "supersession", "revocation", "heartbeat") + +# Declared per class, not per source (audit-core `docs/stream-completeness.md`, +# AUDIT-WP-0009-T04). A per-source heartbeat is satisfied by the chattiest +# class and says nothing about the quiet one — here that is `revocation`, which +# is the class whose silence actually matters. +HEARTBEAT_CLASSES = ("issuance", "use", "supersession", "revocation") LATEST_SCHEMA_VERSION = 5 SCHEMA = """ @@ -870,7 +876,13 @@ class Engine: "scope": SCOPE, "source": SOURCE, "actor": actor, - "action": f"approval.{event_class}", + # audit-core matches heartbeats on the type string it publishes, + # not on our vocabulary. + "action": ( + "audit-core.heartbeat" + if event_class == "heartbeat" + else f"approval.{event_class}" + ), "resource": resource, "outcome": "success", "reason": None, @@ -943,18 +955,38 @@ class Engine: return obj.status, False, consumed, obj.status def emit_heartbeat(self) -> dict[str, Any]: + """Emit one `nothing-to-report` assertion per declared event class. + + One heartbeat covering the whole source would be discharged by whichever + class happens to be busy, which is the failure heartbeats exist to + catch: a revocation stream that has gone silent looks identical to a + quiet one. Each class asserts for itself. + + All four are written in a single transaction. A partial emission would + report some classes healthy and leave others looking stalled, which is + a worse signal than none. + """ conn = self._conn() counts = self.transition_counts() - counts["heartbeat"] += 1 + counts["heartbeat"] += len(HEARTBEAT_CLASSES) + emitted: dict[str, str] = {} try: conn.execute("BEGIN IMMEDIATE") - event_id = self._outbox_insert( - conn, - "heartbeat", - None, - actor=None, - extra={"assertion": "nothing-to-report", "counts": counts}, - ) + for event_class in HEARTBEAT_CLASSES: + emitted[event_class] = self._outbox_insert( + conn, + "heartbeat", + None, + actor=None, + extra={ + # Overrides the emitting class: audit-core reads this as + # the class being asserted about, not as the class of + # the heartbeat itself. + "class": event_class, + "assertion": "nothing-to-report", + "emitted_count": counts.get(event_class, 0), + }, + ) conn.commit() except sqlite3.Error as exc: conn.rollback() @@ -962,7 +994,11 @@ class Engine: except Exception: conn.rollback() raise - return {"event_id": event_id, "assertion": "nothing-to-report", "counts": counts} + return { + "assertion": "nothing-to-report", + "classes": dict(emitted), + "counts": counts, + } def transition_counts(self) -> dict[str, int]: conn = self._conn() diff --git a/docs/audit-source-registration.md b/docs/audit-source-registration.md index 8b9ef5b..d9c76ab 100644 --- a/docs/audit-source-registration.md +++ b/docs/audit-source-registration.md @@ -27,6 +27,52 @@ no credentials. Redaction findings do not excuse a producer defect. Audit Core's scope overlay does not set this field; the protected sender registry entry must explicitly carry `secret_policy: redact` when provisioned. +## Heartbeat classes + +Per class, not per source (`AUDIT-WP-0009-T04`, audit-core +`docs/stream-completeness.md`). A per-source heartbeat is discharged by +whichever class happens to be busy and says nothing about the quiet one — here +that is `revocation`, whose silence is the only silence that matters. Requested +`heartbeat_classes`, all four at the same bound: + +| Class | Longest gap that is not yet a finding | +| --- | --- | +| `issuance` | 172800s (48h) | +| `use` | 172800s (48h) | +| `supersession` | 172800s (48h) | +| `revocation` | 172800s (48h) | + +Twice the deployed `--heartbeat-seconds` default of 86400, so one missed cycle +is not a finding and two consecutive ones are. The engine emits all four in a +single transaction: a partial emission would report some classes healthy and +others stalled, which is a worse signal than none. + +The first beat is emitted at startup rather than one interval later — +`heartbeat_due` is true when no heartbeat exists — because audit-core raises +`no_heartbeat_since_registration` for a declared-but-never-sent class, which is +the shape a naive compare-against-last-seen would drop silently. + +Heartbeat events carry `type: audit-core.heartbeat` with `class` and +`assertion` on `data`. They are ordinary events on the same append-only chain +and the same durable drain; there is no separate heartbeat path here either. + +## Reconciliation + +`GET /v1/reconciliation?source=&tenant=&since=&until=` returns per-class counts +for this source's own events. `may_read: false` still holds and is not +weakened: a source asking how many of its own events the archive holds learns +nothing it did not itself emit. Both bounds must be supplied — a count whose +window this engine did not choose is not comparable against +`transition_counts()`. A request for another source's counts returns `403`, not +zero; a zero would be a false answer to a completeness question. + +**The bound, restated so it is not lost:** agreement on counts proves neither +completeness nor that any event occurred, and a missing heartbeat is not proof +of suppression. Both controls cover loss, outage, drain failure and accident. +Neither covers this engine lying about itself — a compromised emitter suppresses +the event and its own heartbeat together. That is the residual, and it is not +closed by anything in this document. + The transactional outbox emits issuance, use, supersession and revocation; heartbeat uses the same durable drain. Its atomicity and retry contract remain in [outbox-contract.md](outbox-contract.md). Receiver identity/scope/ingress and diff --git a/tests/test_audit.py b/tests/test_audit.py index 74a1802..4381046 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -100,13 +100,18 @@ def test_worker_emits_due_heartbeat_and_drains(): engine = Engine(":memory:", clock=lambda: now[0]) delivered = [] worker = OutboxWorker(engine, delivered.append, heartbeat_interval_seconds=300) + from approval_engine.store import HEARTBEAT_CLASSES + first = worker.run_once() - assert first["delivered"] == 1 - assert delivered[0]["action"] == "approval.heartbeat" + # The first beat is due immediately, not one interval in: declaring a + # heartbeat and never sending one is audit-core's + # `no_heartbeat_since_registration` finding, not a skip. + assert first["delivered"] == len(HEARTBEAT_CLASSES) + assert {d["action"] for d in delivered} == {"audit-core.heartbeat"} now[0] += timedelta(seconds=301) second = worker.run_once() - assert second["delivered"] == 1 - assert len(delivered) == 2 + assert second["delivered"] == len(HEARTBEAT_CLASSES) + assert len(delivered) == 2 * len(HEARTBEAT_CLASSES) engine.close() @@ -116,3 +121,46 @@ def test_sender_rejects_empty_token(tmp_path): sink = AuditCoreSink("http://audit-core", token) with pytest.raises(AuditDeliveryError, match="credential"): sink({}) + + +def test_heartbeat_envelope_carries_class_and_assertion_on_data(): + """audit-core reads these off `data`, not out of a producer-shaped blob.""" + from approval_engine.audit import audit_envelope + + envelope = audit_envelope( + { + "event_id": "e1", + "action": "audit-core.heartbeat", + "source": "approval-engine", + "resource": "approval-engine:heartbeat", + "tenant": "tenant:platform", + "observed_at": "2026-09-10T00:00:00+00:00", + "schema_version": "audit-core.event.v1alpha1", + "scope": "netkingdom-approvals", + "outcome": "success", + "details": {"class": "revocation", "assertion": "nothing-to-report"}, + } + ) + assert envelope["type"] == "audit-core.heartbeat" + assert envelope["data"]["class"] == "revocation" + assert envelope["data"]["assertion"] == "nothing-to-report" + + +def test_ordinary_event_envelope_gains_no_heartbeat_fields(): + from approval_engine.audit import audit_envelope + + envelope = audit_envelope( + { + "event_id": "e2", + "action": "approval.issuance", + "source": "approval-engine", + "resource": "approval:a1", + "tenant": "tenant:platform", + "observed_at": "2026-09-10T00:00:00+00:00", + "schema_version": "audit-core.event.v1alpha1", + "scope": "netkingdom-approvals", + "outcome": "success", + "details": {"class": "issuance"}, + } + ) + assert "assertion" not in envelope["data"] diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 29f11f0..aa68c5e 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -58,15 +58,46 @@ def test_drain_marks_delivered(engine): assert sink[0]["action"] == "approval.issuance" -def test_heartbeat_is_positive_claim(engine): +def test_heartbeat_is_a_positive_claim_per_declared_class(engine): + """One assertion per class, because silence is class-shaped. + + A single per-source heartbeat is discharged by whichever class is busy. + `revocation` is the class whose silence matters here, and it is the one a + per-source beat would hide behind `issuance`. + """ + from approval_engine.store import HEARTBEAT_CLASSES + approve(engine) beat = engine.emit_heartbeat() assert beat["assertion"] == "nothing-to-report" assert beat["counts"]["issuance"] == 1 - assert beat["counts"]["heartbeat"] == 1 + assert set(beat["classes"]) == set(HEARTBEAT_CLASSES) + pending = [p for p in engine.undrained() if p["class"] == "heartbeat"] - assert len(pending) == 1 - assert pending[0]["payload"]["details"]["assertion"] == "nothing-to-report" + assert len(pending) == len(HEARTBEAT_CLASSES) + asserted = {p["payload"]["details"]["class"] for p in pending} + assert asserted == set(HEARTBEAT_CLASSES) + assert "revocation" in asserted + for p in pending: + assert p["payload"]["details"]["assertion"] == "nothing-to-report" + assert p["payload"]["action"] == "audit-core.heartbeat" + + +def test_heartbeat_classes_are_emitted_atomically(engine): + """Partial emission would report some classes healthy and others stalled.""" + from approval_engine.store import HEARTBEAT_CLASSES + + engine.fail_outbox = True + try: + with pytest.raises(Exception): + engine.emit_heartbeat() + finally: + engine.fail_outbox = False + assert [p for p in engine.undrained() if p["class"] == "heartbeat"] == [] + engine.emit_heartbeat() + assert len( + [p for p in engine.undrained() if p["class"] == "heartbeat"] + ) == len(HEARTBEAT_CLASSES) def test_revocation_event_class(engine): diff --git a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md index ca51615..1c4aee5 100644 --- a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md +++ b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md @@ -513,6 +513,32 @@ the same tenant. This closes those source-input waits, not live registration. Platform still owes linked receiver/sender custody and the protected token; T04 stays wait for admission and live drain/reconciliation evidence. +2026-09-10: audit-core landed the detection half (`AUDIT-WP-0009` T04/T06/T07 +at `b098fb1`, contract in their `docs/stream-completeness.md`), so the sender +side is no longer describing a cadence that does not operate. Implemented the +per-class heartbeat contract: `HEARTBEAT_CLASSES` covers `issuance`, `use`, +`supersession` and `revocation`, each asserting `nothing-to-report` for itself, +all four written in one transaction. The previous single per-source beat was +exactly the shape audit-core rules inadequate — it is discharged by whichever +class is busy, so a silent `revocation` stream hides behind `issuance`, and +`revocation` is the only silence here that matters. Heartbeats now carry +`type: audit-core.heartbeat` with `class` and `assertion` on `data` rather than +inside our details object. The first beat is emitted at startup, not one +interval later, because a declared-but-never-sent class is their +`no_heartbeat_since_registration` finding rather than a skip — `heartbeat_due` +already returned true on an empty outbox and that is now pinned by test. +`docs/audit-source-registration.md` declares `heartbeat_classes` at 172800s for +all four (twice the deployed 86400s default, so one missed cycle is not a +finding and two are), records the reconciliation surface and both bounds, and +restates the residual audit-core stated on every response: counts and +heartbeats cover loss, outage, drain failure and accident, and neither covers +this engine lying about itself, since a compromised emitter suppresses the +event and its own heartbeat together. Tests: per-class emission including +`revocation`, atomicity under outbox failure, the envelope placement, and that +an ordinary event gains no heartbeat fields. 155 pass. T04 stays `wait`: live +drain still needs the sender registration and credential admission, and the +`heartbeat_classes` values above are a request until audit-core accepts them. + ## Prove one live PEP consumption path ```task