Bind heartbeat observations to explicitly covered classes

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2

docs/CadenceObserverAdmission.md

@@ -49,7 +49,10 @@ owner must deliver a bounded capture and snapshot; no audit-core sender/feed is
 declared in the reviewed receiver source. If routed through Audit Core, owners
 must explicitly translate the heartbeat envelope, bind exact sender+tenant,
 retain source instance/sequence/window semantics, and admit its declaration.
-An unbounded database accepted count cannot replace a process-local snapshot.
+AUDIT-WP-0013-T04 now supports explicit `stream_instance_id` on receiver
+reconciliation. A reviewed receiver release is still required, and acceptance
+time must be aligned with source/sequence windows. An unbounded database
+accepted count cannot replace a process-local snapshot.
 
 local-identity has no registered sender or evidence feed, declares no heartbeat,
 and remains the owner's explicit activity-scoped incompatibility. Do not accept
@@ -78,3 +81,28 @@ after the fact. There is no automatic containment authority in this delivery.
 Knowledge orientation reloaded 2026-10-03: Emergent Security0.2.5 verified-files,
 package digest `9e186dbd6f2d5eb455bca24db4b03cd866c63c4e94660430dd15f2ca9856215f`,
 generic start plus evidence source card. Orientation is not a readiness grant.
+
+
+## Heartbeat coverage correction — 2026-10-15
+
+`StreamHeartbeat.covered_event_classes` explicitly names the classes vouched
+for by a source or reviewed adapter. Matching requires both the transport
+event class and that coverage. Shared `audit-core.heartbeat` transport names
+alone cannot let a fresh issuance beat hide an expired revocation beat.
+Missing coverage does not satisfy a covered-class obligation; malformed coverage
+is refused. Count keys and observed traffic never infer this binding.
+
+The Qonto source adapter explicitly maps its shared heartbeat to the three
+classes in its source-owned declaration. This is local source-path evidence,
+not admission of a collector or a receiver feed. The same envelope with an
+explicit receiver `data.class` must map only to that named class; do not set
+all declared classes on every Approval Engine heartbeat.
+
+The existing 2026-09-15 sender bearer-revocation receipt in audit-core already
+proves scoped overlap/drop: replacement bearer 202 then 401, incumbent 202,
+for approval-engine and informed-decision. Reuse that exact historical proof;
+it does not prove lifecycle reconciliation or cadence registry admission.
+Receiver applies 1.5 grace to registered heartbeat intervals: the requested
+172800 seconds would produce a 259200-second effective deadline. Record the
+source/receiver owner decision before admitting that bound; do not silently
+claim a 48-hour alert from that registration.

src/kings_guard/contracts.py

@@ -202,13 +202,23 @@ class ImmuneObservation:
 
 @dataclass(frozen=True, slots=True)
 class StreamHeartbeat:
-    """A signed positive claim that can itself go missing (§9.6)."""
+    """A source positive claim with explicit class coverage that can go missing."""
 
     source_system: str
     timestamp: str
     event_class: str
     assertion: str
     counts: Mapping[str, int] = field(default_factory=dict)
+    # Explicit source/adapter coverage, distinct from the transport event type.
+    # Empty coverage is unassessed; counts and traffic never infer it.
+    covered_event_classes: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        coverage = self.covered_event_classes
+        if (not isinstance(coverage, tuple)
+                or any(not isinstance(value, str) or not value.strip() for value in coverage)
+                or len(set(coverage)) != len(coverage)):
+            raise ValueError("heartbeat coverage must be a tuple of unique nonempty class names")
 
 
 @dataclass(frozen=True, slots=True)

src/kings_guard/live.py

@@ -272,6 +272,9 @@ def _confirm_stream(
             event_class="audit.heartbeat",
             assertion=str(event["assertion"]),
             counts=_transition_counts(event),
+            # Qonto's source declaration maps its shared heartbeat to these
+            # classes; this is not inferred from counts or traffic volume.
+            covered_event_classes=("audit.allow", "audit.deny", "audit.heartbeat"),
         )
         for event in heartbeat_events
     )

src/kings_guard/stream.py

@@ -68,7 +68,7 @@ def evaluate_stream(
 
     for spec in cadence.heartbeats:
         due = (now_dt - watch_start) >= spec.interval
-        latest = _latest_heartbeat(heartbeats, spec.event_class)
+        latest = _latest_heartbeat(heartbeats, spec.event_class, spec.covered_event_class)
         missing = False
         if latest is None:
             missing = due
@@ -146,8 +146,13 @@ def _count_observations(observations: Sequence[ImmuneObservation]) -> dict[str,
 def _latest_heartbeat(
     heartbeats: Sequence[StreamHeartbeat],
     event_class: str,
+    covered_event_class: str,
 ) -> StreamHeartbeat | None:
-    matching = [item for item in heartbeats if item.event_class == event_class]
+    matching = [
+        item for item in heartbeats
+        if item.event_class == event_class
+        and covered_event_class in item.covered_event_classes
+    ]
     if not matching:
         return None
     return max(matching, key=lambda item: parse_timestamp(item.timestamp))

tests/test_completeness.py

@@ -102,6 +102,7 @@ def test_complete_stream_keeps_richness_and_states_completeness() -> None:
                 event_class="audit.heartbeat",
                 assertion="nothing-to-report",
                 counts={"audit.deny": 1},
+                covered_event_classes=("audit.deny",),
             ),
         ),
         reconciliation=ReconciliationView(

tests/test_heartbeat_class_scope.py

@@ -0,0 +1,138 @@
+from dataclasses import replace
+from datetime import timedelta
+
+import pytest
+
+from kings_guard.cadence import HeartbeatCadence, load_qonto_assistant_source_cadence
+from kings_guard.contracts import EvidenceClass, StreamHeartbeat
+from kings_guard.stream import evaluate_stream
+
+NOW = "2026-10-15T12:00:00Z"
+
+
+def approval_cadence():
+    return replace(
+        load_qonto_assistant_source_cadence(),
+        source_system="approval-engine",
+        heartbeats=tuple(
+            HeartbeatCadence(
+                "audit-core.heartbeat",
+                name,
+                EvidenceClass.LOAD_BEARING,
+                timedelta(hours=24),
+                "nothing-to-report",
+            )
+            for name in ("issuance", "revocation")
+        ),
+        reconciliations=(),
+    )
+
+
+def assess(*beats):
+    return evaluate_stream(
+        (), approval_cadence(), now=NOW, watching_since="2026-10-12T12:00:00Z", heartbeats=beats
+    )
+
+
+def test_unscoped_shared_transport_heartbeat_cannot_vouch_for_transition_classes():
+    beat = StreamHeartbeat(
+        "approval-engine",
+        NOW,
+        "audit-core.heartbeat",
+        "nothing-to-report",
+        counts={"issuance": 0, "revocation": 0},
+    )
+    result = assess(beat)
+    assert "stream:heartbeat_missing:issuance" in result.findings
+    assert "stream:heartbeat_missing:revocation" in result.findings
+
+
+def test_fresh_issuance_heartbeat_does_not_hide_stale_revocation():
+    fresh = StreamHeartbeat(
+        "approval-engine",
+        NOW,
+        "audit-core.heartbeat",
+        "nothing-to-report",
+        covered_event_classes=("issuance",),
+    )
+    stale = StreamHeartbeat(
+        "approval-engine",
+        "2026-10-12T12:00:00Z",
+        "audit-core.heartbeat",
+        "nothing-to-report",
+        covered_event_classes=("revocation",),
+    )
+    result = assess(stale, fresh)
+    assert "stream:heartbeat_missing:revocation" in result.findings
+    assert "stream:heartbeat_missing:issuance" not in result.findings
+
+
+def test_fresh_heartbeats_for_both_classes_satisfy_their_own_obligations():
+    beats = tuple(
+        StreamHeartbeat(
+            "approval-engine",
+            NOW,
+            "audit-core.heartbeat",
+            "nothing-to-report",
+            covered_event_classes=(name,),
+        )
+        for name in ("issuance", "revocation")
+    )
+    assert not assess(*beats).findings
+
+
+def test_explicit_coverage_cannot_override_wrong_transport_event_class():
+    beat = StreamHeartbeat(
+        "approval-engine",
+        NOW,
+        "other.heartbeat",
+        "nothing-to-report",
+        covered_event_classes=("issuance", "revocation"),
+    )
+    assert len(assess(beat).findings) == 2
+
+
+@pytest.mark.parametrize(
+    "coverage", ["revocation", ["revocation"], ("",), ("revocation", "revocation"), (None,)]
+)
+def test_malformed_coverage_is_refused_instead_of_matching_by_substring(coverage):
+    with pytest.raises(ValueError, match="coverage"):
+        StreamHeartbeat(
+            "approval-engine",
+            NOW,
+            "audit-core.heartbeat",
+            "nothing-to-report",
+            covered_event_classes=coverage,
+        )
+
+
+def test_qonto_source_adapter_publishes_its_shared_heartbeat_coverage():
+    from kings_guard.live import _confirm_stream
+
+    counts = {"audit.allow": 0, "audit.deny": 0}
+    event = {
+        "stream_id": "qonto-assistant.audit",
+        "stream_instance_id": "instance",
+        "stream_sequence": 1,
+        "event_class": "audit.heartbeat",
+        "timestamp": NOW,
+        "assertion": "nothing-to-report",
+        "source_transition_counts": counts,
+    }
+    snapshot = {
+        "stream_id": event["stream_id"],
+        "stream_instance_id": "instance",
+        "last_stream_sequence": 1,
+        "source_transition_counts": counts,
+    }
+    beats, reconciliation, _, corrections = _confirm_stream([event], snapshot)
+    assert not corrections
+    assert beats[0].covered_event_classes == ("audit.allow", "audit.deny", "audit.heartbeat")
+    result = evaluate_stream(
+        (),
+        load_qonto_assistant_source_cadence(),
+        now=NOW,
+        heartbeats=beats,
+        reconciliation=reconciliation,
+    )
+    assert not result.findings

tests/test_stream.py

@@ -34,6 +34,7 @@ def test_unmet_declared_rate_is_a_stream_finding() -> None:
                 event_class="audit.heartbeat",
                 assertion="nothing-to-report",
                 counts={"audit.deny": 0},
+                covered_event_classes=("audit.deny",),
             ),
         ),
         reconciliation=ReconciliationView(
@@ -82,6 +83,7 @@ def test_reconciliation_divergence_is_a_stream_finding() -> None:
                 event_class="audit.heartbeat",
                 assertion="nothing-to-report",
                 counts={"audit.deny": 2},
+                covered_event_classes=("audit.deny",),
             ),
         ),
         reconciliation=ReconciliationView(

tests/test_stream_admission.py

@@ -51,7 +51,9 @@ def test_absent_or_invalid_reconciliation_counts_leave_completeness_unknown(coun
     result = evaluate_stream(
         (), load_qonto_assistant_source_cadence(), now=NOW,
         heartbeats=[StreamHeartbeat("qonto-assistant", NOW, "audit.heartbeat",
-                                    "nothing-to-report")],
+                                    "nothing-to-report",
+                                    covered_event_classes=("audit.allow", "audit.deny",
+                                                           "audit.heartbeat"))],
         reconciliation=ReconciliationView(counts, counts),
     )
     assert result.completeness is StreamCompleteness.UNKNOWN
@@ -64,7 +66,9 @@ def test_unrelated_or_future_heartbeat_cannot_satisfy_cadence(source, timestamp)
     result = evaluate_stream(
         (), load_qonto_assistant_source_cadence(), now=NOW,
         watching_since="2026-09-03T12:00:00Z",
-        heartbeats=[StreamHeartbeat(source, timestamp, "audit.heartbeat", "nothing-to-report")],
+        heartbeats=[StreamHeartbeat(source, timestamp, "audit.heartbeat", "nothing-to-report",
+                                    covered_event_classes=("audit.allow", "audit.deny",
+                                                           "audit.heartbeat"))],
         reconciliation=ReconciliationView(COUNTS, COUNTS),
     )
     assert result.completeness is StreamCompleteness.DEGRADED
@@ -75,7 +79,9 @@ def test_excess_evidence_is_also_reconciliation_divergence():
     result = evaluate_stream(
         (), load_qonto_assistant_source_cadence(), now=NOW,
         heartbeats=[StreamHeartbeat("qonto-assistant", NOW, "audit.heartbeat",
-                                    "nothing-to-report")],
+                                    "nothing-to-report",
+                                    covered_event_classes=("audit.allow", "audit.deny",
+                                                           "audit.heartbeat"))],
         reconciliation=ReconciliationView({"audit.deny": 0}, {"audit.deny": 1}),
     )
     assert "stream:reconciliation_divergence:audit.deny" in result.findings

workplans/ADHOC-2026-10-15.md

@@ -0,0 +1,39 @@
+---
+id: ADHOC-2026-10-15
+type: workplan
+title: "Keep shared heartbeat transports bound to covered event classes"
+domain: infotech
+repo: kings-guard
+status: finished
+owner: codex
+topic_slug: netkingdom
+created: "2026-10-15"
+updated: "2026-10-15"
+---
+
+## Prevent one class heartbeat from hiding another class silence
+
+```task
+id: ADHOC-2026-10-15-T01
+status: done
+priority: high
+```
+
+NK-WP-0035 review reproduced a local observer defect: any fresh event with a
+shared heartbeat transport name satisfied every covered class. A fresh issuance
+beat could mask absent revocation. Added explicit covered_event_classes to the
+runtime observation contract, required exact transport-plus-coverage matching,
+and migrated Qonto's source adapter and existing positive test inputs. Unscoped
+beats no longer vouch for transition classes; malformed coverage is refused.
+No Tooling client, live feed or deployment is introduced.
+
+Regression first failed on the previous implementation with false COMPLETE for
+an unscoped shared heartbeat. Tests cover stale revocation with fresh issuance,
+positive class-isolated beats, wrong transport, malformed coverage and Qonto's
+explicit shared mapping. Operational acceptance remains KG-WP-0005-T03; its
+receiver/capture prerequisites remain independently owned. No new live authority
+or stream-completeness claim follows from these source tests.
+
+Validation: full King’s Guard suite **128 passed**. Changed-file Ruff checks
+pass. This remains a local observation-contract repair, not signed evidence
+or a live source/receiver admission.