Heartbeat per event class, not per source

audit-core's completeness contract landed, and our single per-source beat
is the shape it rules inadequate: it is discharged by whichever class is
busy, so a revocation stream that has gone silent looks identical to a
quiet one — and revocation is the only silence here that matters.

Emit one nothing-to-report assertion per declared class, all four in one
transaction so a partial emission cannot report some classes healthy and
others stalled. Carry type audit-core.heartbeat with class and assertion
on data. Pin that the first beat goes out at startup rather than an
interval later, since a declared-but-never-sent class is their
no_heartbeat_since_registration finding and not a skip.

Declare heartbeat_classes and the reconciliation surface in the source
registration, including the residual neither control covers: a
compromised emitter suppresses the event and its own heartbeat together.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HyybaE7DUXrWYrhbnESCTe

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1275879@bnt-lap001
Assistant-Session: eb464208-f821-41b2-bc5a-a6c33d92a8ad

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):