diff --git a/docs/CadenceObserverAdmission.md b/docs/CadenceObserverAdmission.md index 9000a66..7a98fd9 100644 --- a/docs/CadenceObserverAdmission.md +++ b/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. diff --git a/src/kings_guard/contracts.py b/src/kings_guard/contracts.py index c7d0c01..5d6ff9c 100644 --- a/src/kings_guard/contracts.py +++ b/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) diff --git a/src/kings_guard/live.py b/src/kings_guard/live.py index c3f8d99..6253b19 100644 --- a/src/kings_guard/live.py +++ b/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 ) diff --git a/src/kings_guard/stream.py b/src/kings_guard/stream.py index a8542a7..e59dddf 100644 --- a/src/kings_guard/stream.py +++ b/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)) diff --git a/tests/test_completeness.py b/tests/test_completeness.py index ffe4f0e..30fa9f3 100644 --- a/tests/test_completeness.py +++ b/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( diff --git a/tests/test_heartbeat_class_scope.py b/tests/test_heartbeat_class_scope.py new file mode 100644 index 0000000..ed81c23 --- /dev/null +++ b/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 diff --git a/tests/test_stream.py b/tests/test_stream.py index 3d44a4b..252ec69 100644 --- a/tests/test_stream.py +++ b/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( diff --git a/tests/test_stream_admission.py b/tests/test_stream_admission.py index 2ed00bb..bdbfa67 100644 --- a/tests/test_stream_admission.py +++ b/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 diff --git a/workplans/ADHOC-2026-10-15.md b/workplans/ADHOC-2026-10-15.md new file mode 100644 index 0000000..2e56986 --- /dev/null +++ b/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.