AUDIT-WP-0009 T04/T06/T07 — heartbeats, reconciliation, and a home for findings

The detection half audit-core argued up to a MUST and then could not support.
Two registered sources were waiting on it.

T04, heartbeats. A heartbeat is an ordinary event — same envelope, same
append-only custody, same chain, no special table. Deliberate: a heartbeat
stored outside the chain would be the one record here that could be back-dated.
Declared per class rather than per source, because a per-source heartbeat from
a mixed-volume emitter is satisfied by its chattiest class and says nothing
about the quiet, security-relevant one, which is the only reason heartbeats
exist. Not the §17 cadence schema T05 waits on: cadence describes expected
rate, this says how often a source promises to say "nothing to report" for a
class that may legitimately be silent.

no_heartbeat_since_registration is its own finding kind rather than a skip —
it is the case most likely to be a broken integration and the one a naive
"compare against last seen" implementation silently drops. Grace widens the
window so one late run does not flap; it never removes a finding.

T06, reconciliation. Counts, never payloads. The awkward part is that every
registered sender holds may_read: false, which taken literally makes the §9.6
reconciliation obligation undischargeable by every source actually registered.
Resolved by observing that a source asking how many of its own events we hold
is not reading the archive — it learns nothing it did not itself emit. So the
surface is scoped to the caller's own sources and tenants and returns no
payloads; anything wider stays behind may_read and full tenant scope. Another
source's counts return 403 rather than an empty count, because a zero would
read as "we hold none of yours" — a false answer to a question about
completeness. No default window, since a count whose bounds the caller did not
choose is not comparable to anything the caller computed.

T07, the findings surface. /v1/stream-findings, following the dead-letter and
secret-finding conventions: may_read plus full tenant scope, since findings
span every sender and carry no tenant key to filter on.

The bound is on every response rather than in a document nobody opens beside
it. A missing heartbeat is not proof of suppression, and agreement on counts
proves neither completeness nor that any event occurred. Both controls cover
loss, outage, drain failure and accident; neither covers a source lying about
itself, and where the emitter is compromised both agree with it. Closing that
needs an observer independent of the emitter, which §16 put outside our scope.

The scope overlay may shorten a heartbeat interval or add a class, never
lengthen or remove one — same asymmetry as evidence_kind, and for the same
reason: a ConfigMap refresh must not widen the window in which a suppressed
class goes unnoticed without anyone deciding to.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2069992@bnt-lap001
Assistant-Session: 167dd7f8-2a25-4be1-aa46-3b6f1a5f94c6

audit_core/ingestion.py

@@ -46,7 +46,7 @@ from audit_core.redaction import (
     apply_policy,
     finding_from_path,
 )
-from audit_core.senders import SenderIdentity, SenderRegistry, development_registry
+from audit_core.senders import WILDCARD, SenderIdentity, SenderRegistry, development_registry
 from audit_core.sqlite_backend import SQLiteAuditBackend
 
 MAX_BODY_BYTES = 256 * 1024
@@ -164,10 +164,21 @@ class IngestionApplication:
                 "/v1/secret-findings",
                 "/v1/stats",
                 "/v1/integrity",
+                "/v1/stream-findings",
             )
         ):
             return self._read(start_response, environ, path, identity)
 
+        # Reconciliation is deliberately routed BEFORE the may_read gate.
+        # A source asking how many of its own events audit-core holds is not
+        # reading the archive — it learns nothing it did not itself emit — and
+        # §9.6 makes that comparison the source's own detection obligation. A
+        # writer with may_read: false must therefore be able to ask, or the
+        # obligation audit-core argued for is undischargeable by every sender
+        # actually registered. Cross-source counts stay behind may_read.
+        if method == "GET" and path == "/v1/reconciliation":
+            return self._reconciliation(start_response, environ, identity)
+
         if path != "/v1/events" or method != "POST":
             return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
 
@@ -264,6 +275,8 @@ class IngestionApplication:
                 )
             if path == "/v1/stats":
                 return self._json(start_response, HTTPStatus.OK, self.counters.snapshot())
+            if path == "/v1/stream-findings":
+                return self._stream_findings(start_response)
             if path == "/v1/secret-findings":
                 return self._json(
                     start_response, HTTPStatus.OK,
@@ -309,6 +322,97 @@ class IngestionApplication:
                 start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error": "backend_unavailable"}
             )
 
+    def _reconciliation(self, start_response, environ, identity):
+        """Per-class counts of one source's own events (AUDIT-WP-0009-T06).
+
+        Counts, never payloads, and never another source's. The comparison a
+        source makes against its own state transitions is the §9.6 detection
+        obligation; audit-core supplies the numerator and says nothing about
+        what the answer means.
+        """
+        query = parse_qs(environ.get("QUERY_STRING", ""))
+        source = (query.get("source") or [""])[0]
+        if not source:
+            return self._json(
+                start_response, HTTPStatus.BAD_REQUEST, {"error": "source_required"}
+            )
+        if not identity.permits_source(source):
+            # 403 rather than an empty count. A zero here would read as "we
+            # hold none of yours", which is a materially different and false
+            # answer to a question about completeness.
+            return self._json(
+                start_response, HTTPStatus.FORBIDDEN, {"error": "source_not_allowed"}
+            )
+        window = _window(query)
+        if window is None:
+            return self._json(
+                start_response, HTTPStatus.BAD_REQUEST,
+                {"error": "since_and_until_required"},
+            )
+        since, until = window
+        tenant = (query.get("tenant") or [""])[0] or None
+        if tenant is not None and not identity.permits_tenant(tenant):
+            return self._json(
+                start_response, HTTPStatus.FORBIDDEN, {"error": "tenant_not_allowed"}
+            )
+        if tenant is None and not identity.has_full_tenant_scope():
+            # A scoped credential must name the tenant it is counting, so the
+            # answer cannot silently aggregate across a boundary it may not see.
+            return self._json(
+                start_response, HTTPStatus.BAD_REQUEST, {"error": "tenant_required"}
+            )
+        counter = getattr(self.backend, "event_counts", None)
+        if not callable(counter):
+            return self._json(
+                start_response, HTTPStatus.NOT_FOUND,
+                {"error": "reconciliation_not_supported"},
+            )
+        try:
+            counts = counter(source, since, until, tenant)
+        except BackendUnavailableError as exc:
+            log.error("reconciliation failed: %s", exc)
+            return self._json(
+                start_response, HTTPStatus.SERVICE_UNAVAILABLE,
+                {"error": "backend_unavailable"},
+            )
+        return self._json(start_response, HTTPStatus.OK, {
+            "source": source,
+            "tenant": tenant,
+            "since": since,
+            "until": until,
+            "counts": counts,
+            # Said in the response because this is the number most likely to
+            # be quoted out of context in someone else's conformance argument.
+            "means": (
+                "the count of events audit-core accepted and stored in this "
+                "window. Divergence from the source's own count is a finding "
+                "for the source; agreement proves neither completeness nor "
+                "that any event occurred."
+            ),
+        })
+
+    def _stream_findings(self, start_response):
+        """Missing-heartbeat findings (AUDIT-WP-0009-T04, T07)."""
+        from audit_core.stream_findings import evaluate
+
+        reader = getattr(self.backend, "last_heartbeats", None)
+        if not callable(reader):
+            return self._json(
+                start_response, HTTPStatus.NOT_FOUND,
+                {"error": "stream_findings_not_supported"},
+            )
+        last: dict[tuple[str, str], str | None] = {}
+        for identity in self.senders.identities:
+            for source in identity.sources:
+                if source == WILDCARD:
+                    continue
+                for event_class, at in reader(source).items():
+                    last[(source, event_class)] = at
+        findings = evaluate(self.senders.identities, last)
+        return self._json(start_response, HTTPStatus.OK, {
+            "stream_findings": [finding.as_dict() for finding in findings],
+        })
+
     def _count_secrets(self, payload: Any, identity, outcome: str, findings) -> None:
         """Count secret-shaped fields by path, sender, source and action.
 
@@ -472,9 +576,32 @@ _UNSCOPED_READ_PATHS = frozenset({
     "/v1/stats",
     "/v1/secret-findings",
     "/v1/integrity",
+    # Findings span every registered sender and carry no tenant key, so there
+    # is nothing to filter on. A scoped reader is refused rather than served
+    # instance-wide facts — the same rule as the other three.
+    "/v1/stream-findings",
 })
 
 
+def _window(query: dict[str, list[str]]) -> tuple[str, str] | None:
+    """Require an explicit bounded window for a reconciliation query.
+
+    No default window. A count whose bounds the caller did not choose is not
+    comparable against anything the caller computed, and would be quoted as
+    though it were.
+    """
+    since = (query.get("since") or [""])[0]
+    until = (query.get("until") or [""])[0]
+    if not since or not until:
+        return None
+    try:
+        start = _normalize_timestamp(since)
+        end = _normalize_timestamp(until)
+    except ValueError:
+        return None
+    return (start, end) if start < end else None
+
+
 def _readable_by(identity, record: dict) -> bool:
     """Whether ``identity`` may read ``record``.
 

audit_core/postgres_backend.py

@@ -24,6 +24,7 @@ import time
 from datetime import datetime, timezone
 from typing import Any
 
+from audit_core.stream_findings import HEARTBEAT_ACTION
 from audit_core.interface import (
     AcceptResult,
     AuditEvent,
@@ -408,6 +409,37 @@ class PostgresAuditBackend:
         )
         return [{"accepted_at": _iso(at), **rec} for rec, at in rows]
 
+    def event_counts(
+        self, source: str, since: str, until: str, tenant: str | None = None
+    ) -> list[dict]:
+        """Per-class counts for one source over a bounded window.
+
+        AUDIT-WP-0009-T06. Counts only — never payloads. A reconciliation
+        answer that carried records would turn a completeness check into a read
+        surface, and a source does not gain the right to read the archive by
+        emitting into it.
+        """
+        sql = (
+            f"SELECT action, count(*) FROM {self._events} "
+            "WHERE source = %s AND accepted_at >= %s AND accepted_at < %s"
+        )
+        params: tuple = (source, since, until)
+        if tenant is not None:
+            sql += " AND tenant = %s"
+            params += (tenant,)
+        sql += " GROUP BY action ORDER BY action"
+        return [{"class": action, "count": count} for action, count in self._query(sql, params)]
+
+    def last_heartbeats(self, source: str) -> dict[str, str]:
+        """Most recent heartbeat per class for one source."""
+        rows = self._query(
+            f"SELECT record->'details'->'data'->>'class' AS cls, max(accepted_at) "
+            f"FROM {self._events} WHERE source = %s AND action = %s "
+            "GROUP BY cls",
+            (source, HEARTBEAT_ACTION),
+        )
+        return {cls: _iso(at) for cls, at in rows if cls}
+
     def replay(self, event_id: str) -> AcceptResult:
         """Re-submit a stored event through :meth:`accept`.
 

audit_core/senders.py

@@ -68,6 +68,20 @@ class SenderIdentity:
     # §9.6 requires the trade be declared where the trail is documented, so it
     # travels with the identity rather than living only in prose.
     completeness_trade: str | None = None
+    # §9.6 heartbeat declaration (AUDIT-WP-0009-T04): event class -> the
+    # longest gap, in seconds, that is not yet a finding.
+    #
+    # This is deliberately NOT the §17 emission-cadence schema that T05 waits
+    # on. Cadence describes a stream's expected *rate* and belongs to Taxonomy;
+    # this is a registration property saying how often a source promises to say
+    # "nothing to report" for a class that may legitimately be silent. The two
+    # are complementary and audit-core is not inventing a competing rate shape.
+    #
+    # Keyed per class, not per source, on purpose. A per-source heartbeat from
+    # a mixed-volume emitter is satisfied by its chattiest class and says
+    # nothing about the quiet, security-relevant one — which is the only case
+    # heartbeats exist for.
+    heartbeat_classes: tuple[tuple[str, int], ...] = ()
 
     def __post_init__(self) -> None:
         if not self.name:
@@ -102,6 +116,17 @@ class SenderIdentity:
                 "completeness_trade — §9.6 requires emission atomicity of it"
             )
 
+        for action, interval in self.heartbeat_classes:
+            if not action or not str(action).strip():
+                raise ValueError(
+                    f"sender {self.name!r}: a heartbeat class needs an event class"
+                )
+            if not isinstance(interval, int) or isinstance(interval, bool) or interval <= 0:
+                raise ValueError(
+                    f"sender {self.name!r}: heartbeat interval for {action!r} must "
+                    f"be a positive number of seconds, got {interval!r}"
+                )
+
     @property
     def is_load_bearing(self) -> bool:
         return self.evidence_kind == EVIDENCE_LOAD_BEARING
@@ -118,6 +143,7 @@ class SenderIdentity:
             "evidence_kind": self.evidence_kind,
             "completeness_claimed": False,
             "completeness_trade": self.completeness_trade,
+            "heartbeat_classes": dict(self.heartbeat_classes),
             "detection_surface": None,
         }
 
@@ -288,6 +314,7 @@ def _apply_scope_overlay(
                     if "completeness_trade" in extra
                     else identity.completeness_trade
                 ),
+                heartbeat_classes=_overlay_heartbeats(identity, extra),
             )
         )
     return merged
@@ -313,6 +340,36 @@ def _overlay_evidence_kind(identity: SenderIdentity, extra: dict[str, Any]) -> s
     return declared
 
 
+def _overlay_heartbeats(
+    identity: SenderIdentity, extra: dict[str, Any]
+) -> tuple[tuple[str, int], ...]:
+    """The overlay may add a heartbeat class or shorten an interval, never
+    lengthen or remove one.
+
+    Same asymmetry as ``evidence_kind`` and for the same reason: a ConfigMap
+    refresh must not be able to widen the window in which a suppressed class
+    goes unnoticed, or drop the obligation entirely, without anyone deciding
+    to. Tightening detection is safe in the direction a mistake would take it.
+    """
+    if "heartbeat_classes" not in extra:
+        return identity.heartbeat_classes
+    declared = dict(identity.heartbeat_classes)
+    proposed = dict(_heartbeat_classes(extra["heartbeat_classes"]))
+    for action, interval in declared.items():
+        if action not in proposed:
+            raise ValueError(
+                f"sender {identity.name!r}: the scope overlay may not remove "
+                f"heartbeat class {action!r}"
+            )
+        if proposed[action] > interval:
+            raise ValueError(
+                f"sender {identity.name!r}: the scope overlay may not lengthen "
+                f"the heartbeat interval for {action!r} "
+                f"({interval}s -> {proposed[action]}s)"
+            )
+    return tuple(sorted(proposed.items()))
+
+
 def _parse_identities(raw: str) -> list[SenderIdentity]:
     try:
         entries = json.loads(raw)
@@ -338,9 +395,18 @@ def _identity_from(entry: Any) -> SenderIdentity:
         expires_at=_parse_expiry(entry.get("expires_at")),
         evidence_kind=str(entry.get("evidence_kind", EVIDENCE_ATTRIBUTIVE)),
         completeness_trade=_clean_trade(entry.get("completeness_trade")),
+        heartbeat_classes=_heartbeat_classes(entry.get("heartbeat_classes")),
     )
 
 
+def _heartbeat_classes(value: Any) -> tuple[tuple[str, int], ...]:
+    if not value:
+        return ()
+    if not isinstance(value, dict):
+        raise ValueError("heartbeat_classes must be an object of class -> seconds")
+    return tuple(sorted((str(k), v) for k, v in value.items()))
+
+
 def _clean_trade(value: Any) -> str | None:
     if value in (None, ""):
         return None

audit_core/sqlite_backend.py

@@ -13,6 +13,7 @@ import sqlite3
 import threading
 from datetime import datetime, timezone
 
+from audit_core.stream_findings import HEARTBEAT_ACTION
 from audit_core.interface import (
     AcceptResult,
     AuditEvent,
@@ -218,6 +219,40 @@ class SQLiteAuditBackend:
         )
         return [{"accepted_at": at, **json.loads(rec)} for rec, at in rows]
 
+    def event_counts(
+        self, source: str, since: str, until: str, tenant: str | None = None
+    ) -> list[dict]:
+        """Per-class counts for one source over a bounded window.
+
+        AUDIT-WP-0009-T06. Counts only — never payloads. A reconciliation
+        answer that carried records would turn a completeness check into a read
+        surface, and a source does not gain the right to read the archive by
+        emitting into it.
+        """
+        sql = (
+            "SELECT json_extract(record, '$.action') AS action, count(*) "
+            "FROM events WHERE json_extract(record, '$.source') = ? "
+            "AND accepted_at >= ? AND accepted_at < ?"
+        )
+        params: tuple = (source, since, until)
+        if tenant is not None:
+            sql += " AND tenant = ?"
+            params += (tenant,)
+        sql += " GROUP BY action ORDER BY action"
+        return [{"class": action, "count": count} for action, count in self._query(sql, params)]
+
+    def last_heartbeats(self, source: str) -> dict[str, str]:
+        """Most recent heartbeat per class for one source."""
+        rows = self._query(
+            "SELECT json_extract(record, '$.details.data.class') AS cls, "
+            "max(accepted_at) FROM events "
+            "WHERE json_extract(record, '$.source') = ? "
+            "AND json_extract(record, '$.action') = ? "
+            "GROUP BY cls",
+            (source, HEARTBEAT_ACTION),
+        )
+        return {cls: at for cls, at in rows if cls}
+
     def record_rejection(
         self,
         *,

audit_core/stream_findings.py

@@ -0,0 +1,131 @@
+"""Stream-completeness findings: when the trail itself is the defect.
+
+`AUDIT-WP-0009-T04` and `T07`. `INTENT.md` principle 10 already says a degraded
+audit stream is itself an audit and operations event. The principle was in
+place; the mechanism was not.
+
+**What this can and cannot see.** The hash chain proves records held were not
+altered or truncated. It says nothing about a record that never arrived, and
+§9.6 is explicit that omission is the acute risk for exactly the rare negative
+classes — revocation, denial, containment — where suppression is most valuable
+and least visible. Rate monitoring cannot help there either: a class that is
+legitimately silent for a month is indistinguishable from one being suppressed.
+
+A heartbeat fixes that by inverting the burden. Instead of inferring health
+from events that may never come, the source makes a positive claim —
+*nothing to report for this class* — on a declared interval. The claim can
+itself go missing, and a missing claim is a finding. That is the whole idea,
+and it is the only control here that covers adversarial omission.
+
+**The bound, stated because a finding surface invites over-reading.** A
+compromised source emits a truthful-looking heartbeat while suppressing the
+event it is supposed to be vouching for. Heartbeats cover loss, outage, drain
+failure and accident — most of what actually goes wrong — and do not cover a
+source lying about itself. Nothing an archive holds can close that; it needs an
+observer independent of the emitter.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Iterable
+
+# The event class a source uses to say "nothing to report". A heartbeat is an
+# ordinary event: same envelope, same append-only custody, same chain. It gets
+# no special table, because a heartbeat that lived outside the chain would be
+# the one record in this store that could be back-dated.
+HEARTBEAT_ACTION = "audit-core.heartbeat"
+
+MISSING_HEARTBEAT = "missing_heartbeat"
+NEVER_HEARTBEAT = "no_heartbeat_since_registration"
+
+
+@dataclass(frozen=True)
+class StreamFinding:
+    """One reason to believe a stream is not telling the whole truth."""
+
+    kind: str
+    sender: str
+    source: str
+    event_class: str
+    expected_within_seconds: int
+    last_seen: str | None
+    overdue_seconds: int
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "kind": self.kind,
+            "sender": self.sender,
+            "source": self.source,
+            "class": self.event_class,
+            "expected_within_seconds": self.expected_within_seconds,
+            "last_seen": self.last_seen,
+            "overdue_seconds": self.overdue_seconds,
+            # Said on every finding rather than in a document nobody opens
+            # alongside it: this is the absence of a positive claim, not proof
+            # that an event was suppressed.
+            "means": (
+                "a declared heartbeat did not arrive; the class may be healthy "
+                "and the emitter silent. Absence is not proof of suppression."
+            ),
+        }
+
+
+def _parse(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def evaluate(
+    identities: Iterable[Any],
+    last_heartbeats: dict[tuple[str, str], str | None],
+    *,
+    now: datetime | None = None,
+    grace_factor: float = 1.5,
+) -> list[StreamFinding]:
+    """Findings for every declared heartbeat class that is overdue.
+
+    ``last_heartbeats`` is keyed ``(source, class)``. A class with no heartbeat
+    ever recorded produces a finding of its own kind rather than being skipped:
+    a source that declared a heartbeat and never sent one is the case most
+    likely to be a broken integration, and it is precisely the one a
+    "compare against last seen" implementation silently drops.
+
+    ``grace_factor`` exists so one late run does not flap a finding on and off,
+    matching the reasoning behind the attestation freshness window. It widens
+    the window; it never removes the finding.
+    """
+    now = now or datetime.now(timezone.utc)
+    findings: list[StreamFinding] = []
+    for identity in identities:
+        for event_class, interval in getattr(identity, "heartbeat_classes", ()):
+            for source in sorted(identity.sources):
+                if source == "*":
+                    # A wildcard source cannot be held to a heartbeat: there is
+                    # no determinate set of streams to expect one from. Refused
+                    # rather than guessed at.
+                    continue
+                deadline = timedelta(seconds=interval * grace_factor)
+                last = _parse(last_heartbeats.get((source, event_class)))
+                if last is None:
+                    findings.append(StreamFinding(
+                        kind=NEVER_HEARTBEAT, sender=identity.name, source=source,
+                        event_class=event_class, expected_within_seconds=interval,
+                        last_seen=None, overdue_seconds=-1,
+                    ))
+                    continue
+                overdue = (now - last) - deadline
+                if overdue.total_seconds() > 0:
+                    findings.append(StreamFinding(
+                        kind=MISSING_HEARTBEAT, sender=identity.name, source=source,
+                        event_class=event_class, expected_within_seconds=interval,
+                        last_seen=last.isoformat(),
+                        overdue_seconds=int(overdue.total_seconds()),
+                    ))
+    return findings

docs/stream-completeness.md

@@ -0,0 +1,116 @@
+# Stream completeness: heartbeats, reconciliation, and findings
+
+`AUDIT-WP-0009` T04, T06, T07. Statute §9.6.
+
+The chain proves records held were not altered or truncated. It says nothing
+about a record that never arrived — and §9.6 is explicit that **omission** is
+the acute risk for exactly the rare negative classes (revocation, denial,
+containment) where suppression is most valuable and least visible.
+
+audit-core argued that obligation up to a MUST. A source cannot declare a
+cadence to a system with nowhere to put it, so this is the surface owed.
+
+## Three surfaces, and what each actually covers
+
+| Surface | Covers | Does not cover |
+| --- | --- | --- |
+| Heartbeat (`POST /v1/events`, class `audit-core.heartbeat`) | A class going silent — including one that is *legitimately* silent, which rate monitoring can never distinguish | A compromised source emitting a truthful-looking heartbeat while suppressing the event |
+| Reconciliation (`GET /v1/reconciliation`) | Loss, outage, drain failure — divergence between what a source emitted and what arrived | A compromised source suppressing the event and its own count together |
+| Findings (`GET /v1/stream-findings`) | Surfacing the above where an operator sees them | Anything the two above do not detect |
+
+**The bound is the same in both rows and it is not a footnote.** Where the
+emitter itself is compromised, both controls agree with it. They cover loss,
+outage, drain failure and accident — most of what actually goes wrong — and not
+a source lying about itself. Closing that needs an observer independent of the
+emitter, which §16 placed outside audit-core's scope. No conformance claim may
+read these surfaces as covering adversarial omission by the source.
+
+## Heartbeats
+
+A heartbeat is an **ordinary event**: same envelope, same append-only custody,
+same chain, no special table. That is deliberate — a heartbeat stored outside
+the chain would be the one record in this store that could be back-dated.
+
+```json
+{
+  "id": "...", "type": "audit-core.heartbeat",
+  "source": "approval-engine", "subject": "approval-engine",
+  "tenant": "tenant:platform", "correlation_id": "...",
+  "occurred_at": "2026-09-10T03:00:00+00:00",
+  "data": {"class": "approval.revocation", "assertion": "nothing-to-report"}
+}
+```
+
+### Declared per class, never per source
+
+`heartbeat_classes` on the sender registration maps event class → the longest
+gap in seconds that is not yet a finding.
+
+Per class is the whole point. A per-source heartbeat from a mixed-volume
+emitter is satisfied by its chattiest class and says nothing about the quiet,
+security-relevant one — and the quiet class is the only reason heartbeats
+exist. `informed-decision` raised this shape first for presentations versus
+dispositions; it generalises.
+
+### Not the §17 cadence schema
+
+`AUDIT-WP-0009-T05` waits on the emission-cadence declaration `kings-guard` is
+drafting for Taxonomy, and this does not pre-empt it. Cadence describes a
+stream's expected **rate**; a heartbeat is a registration property saying how
+often a source promises to say *nothing to report* for a class that may
+legitimately be silent. Complementary, not alternatives, and audit-core is not
+inventing a competing rate shape while the real one is being written.
+
+### Findings
+
+`GET /v1/stream-findings` requires `may_read` **and** full tenant scope:
+findings span every registered sender and carry no tenant key, so there is
+nothing to filter on and a scoped reader is refused rather than served
+instance-wide facts. Same rule as dead letters, secret findings and integrity.
+
+Two kinds:
+
+- `no_heartbeat_since_registration` — declared a heartbeat, never sent one.
+  Its own kind rather than skipped, because it is the case most likely to be a
+  broken integration and the one a naive "compare against last seen"
+  implementation silently drops.
+- `missing_heartbeat` — one arrived once and is now overdue, with how late.
+
+A grace factor of 1.5 widens the window so a single late run does not flap a
+finding on and off — the same reasoning as the attestation freshness window. It
+widens; it never removes.
+
+Every finding carries a `means` field saying that absence of a heartbeat is not
+proof of suppression. That travels on the finding rather than in a document
+nobody opens alongside it.
+
+## Reconciliation
+
+`GET /v1/reconciliation?source=&tenant=&since=&until=` returns per-class counts
+of a source's own events. **Counts, never payloads.**
+
+### Why a writer with `may_read: false` may call it
+
+Every registered sender holds `may_read: false` — a source does not gain a read
+surface by emitting. Taken literally that would make the §9.6 reconciliation
+obligation undischargeable by every source actually registered, which is a
+strange place for a rule audit-core argued for to end up.
+
+The resolution is that **a source asking how many of its own events audit-core
+holds is not reading the archive**. It learns nothing it did not itself emit.
+So the surface is scoped to the caller's own permitted sources and tenants, and
+returns no payloads. Anything wider — another source's counts, or an unscoped
+aggregate — stays behind `may_read` and full tenant scope.
+
+Two refusals worth knowing:
+
+- Another source's counts return **403, not an empty count**. A zero would read
+  as "we hold none of yours", which is a materially different and false answer
+  to a question about completeness.
+- There is **no default window**. A count whose bounds the caller did not choose
+  is not comparable against anything the caller computed, and would be quoted as
+  though it were.
+
+The response carries a `means` field stating that agreement proves neither
+completeness nor that any event occurred — because this is the number most
+likely to be quoted out of context in someone else's conformance argument.

tests/test_stream_findings.py

@@ -0,0 +1,272 @@
+"""AUDIT-WP-0009 T04/T06/T07 — heartbeats, reconciliation, findings."""
+
+import io
+import json
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from audit_core.ingestion import IngestionApplication
+from audit_core.senders import SenderIdentity, SenderRegistry
+from audit_core.sqlite_backend import SQLiteAuditBackend
+from audit_core.stream_findings import (
+    HEARTBEAT_ACTION,
+    MISSING_HEARTBEAT,
+    NEVER_HEARTBEAT,
+    evaluate,
+)
+
+
+def invoke(app, path, *, token="approval", method="GET", payload=None, key=None):
+    raw = json.dumps(payload).encode() if payload is not None else b""
+    environ = {
+        "PATH_INFO": path.split("?")[0],
+        "QUERY_STRING": path.split("?")[1] if "?" in path else "",
+        "REQUEST_METHOD": method,
+        "CONTENT_LENGTH": str(len(raw)),
+        "wsgi.input": io.BytesIO(raw),
+        "HTTP_AUTHORIZATION": f"Bearer {token}",
+    }
+    if key:
+        environ["HTTP_IDEMPOTENCY_KEY"] = key
+    result = {}
+    out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
+    return result["status"], (json.loads(out) if out else {})
+
+
+SENDER = SenderIdentity(
+    name="approval-engine",
+    tokens=("approval",),
+    sources=frozenset({"approval-engine"}),
+    tenants=frozenset({"tenant:platform"}),
+    may_write=True,
+    may_read=False,
+    evidence_kind="load-bearing",
+    heartbeat_classes=(("approval.revocation", 86400),),
+)
+OPERATOR = SenderIdentity(
+    name="operator", tokens=("operator",), sources=frozenset({"approval-engine"}),
+    tenants=frozenset({"*"}), may_write=False, may_read=True,
+)
+
+
+@pytest.fixture
+def app(tmp_path):
+    backend = SQLiteAuditBackend(str(tmp_path / "s.db"))
+    return IngestionApplication(backend, SenderRegistry([SENDER, OPERATOR]))
+
+
+def emit(app, event_id, action, *, data=None, tenant="tenant:platform"):
+    payload = {
+        "id": event_id, "type": action, "source": "approval-engine",
+        "subject": "approval:1", "tenant": tenant, "correlation_id": "c-1",
+        "occurred_at": "2026-09-10T00:00:00+00:00", "data": data or {"k": "v"},
+    }
+    return invoke(app, "/v1/events", method="POST", payload=payload, key=event_id)
+
+
+# --- T04: heartbeats -------------------------------------------------------
+
+
+def test_a_heartbeat_is_an_ordinary_event_and_is_chained(app):
+    """No special table. A heartbeat outside the chain could be back-dated."""
+    status, _ = emit(
+        app, "hb-1", HEARTBEAT_ACTION,
+        data={"class": "approval.revocation", "assertion": "nothing-to-report"},
+    )
+    assert status.startswith("202")
+    status, body = invoke(app, "/v1/integrity", token="operator")
+    assert status.startswith("200")
+    assert body["intact"] is True
+    assert body["events"] == 1
+
+
+def test_a_declared_class_with_no_heartbeat_ever_is_its_own_finding():
+    """The case a naive 'compare against last seen' silently drops."""
+    findings = evaluate([SENDER], {})
+    assert [f.kind for f in findings] == [NEVER_HEARTBEAT]
+    assert findings[0].event_class == "approval.revocation"
+    assert findings[0].source == "approval-engine"
+
+
+def test_a_fresh_heartbeat_produces_no_finding():
+    now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc)
+    last = (now - timedelta(hours=2)).isoformat()
+    assert evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now) == []
+
+
+def test_an_overdue_heartbeat_is_a_finding_with_its_lateness():
+    now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc)
+    # 86400s interval, 1.5 grace -> overdue past 36h.
+    last = (now - timedelta(hours=40)).isoformat()
+    findings = evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now)
+    assert [f.kind for f in findings] == [MISSING_HEARTBEAT]
+    assert findings[0].overdue_seconds == int(timedelta(hours=4).total_seconds())
+
+
+def test_grace_widens_the_window_but_never_removes_the_finding():
+    now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc)
+    last = (now - timedelta(days=30)).isoformat()
+    findings = evaluate(
+        [SENDER], {("approval-engine", "approval.revocation"): last},
+        now=now, grace_factor=100.0,
+    )
+    # 86400 * 100 is ~100 days, so this one is inside the window...
+    assert findings == []
+    # ...and no grace factor makes a year-old heartbeat acceptable.
+    older = (now - timedelta(days=400)).isoformat()
+    assert evaluate(
+        [SENDER], {("approval-engine", "approval.revocation"): older},
+        now=now, grace_factor=100.0,
+    )
+
+
+def test_a_wildcard_source_is_not_held_to_a_heartbeat():
+    """No determinate set of streams to expect one from. Refused, not guessed."""
+    wild = SenderIdentity(
+        name="w", tokens=("t",), sources=frozenset({"*"}),
+        heartbeat_classes=(("anything", 60),),
+    )
+    assert evaluate([wild], {}) == []
+
+
+def test_every_finding_says_absence_is_not_proof_of_suppression():
+    """The surface invites over-reading; the bound travels on each finding."""
+    finding = evaluate([SENDER], {})[0].as_dict()
+    assert "not proof of suppression" in finding["means"].lower()
+
+
+# --- T07: the findings surface ---------------------------------------------
+
+
+def test_stream_findings_needs_read_and_full_tenant_scope(app):
+    assert invoke(app, "/v1/stream-findings")[0].startswith("403")
+    status, body = invoke(app, "/v1/stream-findings", token="operator")
+    assert status.startswith("200")
+    assert body["stream_findings"][0]["kind"] == NEVER_HEARTBEAT
+
+
+def test_a_delivered_heartbeat_clears_the_finding(app):
+    emit(
+        app, "hb-2", HEARTBEAT_ACTION,
+        data={"class": "approval.revocation", "assertion": "nothing-to-report"},
+    )
+    _, body = invoke(app, "/v1/stream-findings", token="operator")
+    # Stored today against a 24h interval: no longer a finding.
+    assert body["stream_findings"] == []
+
+
+# --- T06: reconciliation ---------------------------------------------------
+
+
+def test_a_writer_may_count_its_own_events_without_may_read(app):
+    """The §9.6 obligation would be undischargeable otherwise."""
+    assert SENDER.may_read is False
+    emit(app, "e-1", "approval.issuance")
+    emit(app, "e-2", "approval.issuance")
+    emit(app, "e-3", "approval.revocation")
+    status, body = invoke(
+        app,
+        "/v1/reconciliation?source=approval-engine&tenant=tenant:platform"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    assert status.startswith("200")
+    assert body["counts"] == [
+        {"class": "approval.issuance", "count": 2},
+        {"class": "approval.revocation", "count": 1},
+    ]
+
+
+def test_reconciliation_returns_counts_and_never_payloads(app):
+    emit(app, "e-9", "approval.issuance", data={"secret_shaped": "x", "binding": "b"})
+    _, body = invoke(
+        app,
+        "/v1/reconciliation?source=approval-engine&tenant=tenant:platform"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    assert set(body) == {"source", "tenant", "since", "until", "counts", "means"}
+    assert "binding" not in json.dumps(body)
+
+
+def test_a_source_may_not_count_another_source(app):
+    status, body = invoke(
+        app,
+        "/v1/reconciliation?source=user-engine&tenant=tenant:platform"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    # 403, not an empty count: a zero would read as "we hold none of yours",
+    # which is a false answer to a question about completeness.
+    assert status.startswith("403")
+    assert body["error"] == "source_not_allowed"
+
+
+def test_reconciliation_requires_an_explicit_bounded_window(app):
+    status, body = invoke(app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform")
+    assert status.startswith("400")
+    assert body["error"] == "since_and_until_required"
+
+
+def test_a_scoped_credential_must_name_its_tenant(app):
+    status, body = invoke(
+        app,
+        "/v1/reconciliation?source=approval-engine"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    assert status.startswith("400")
+    assert body["error"] == "tenant_required"
+
+
+def test_a_scoped_credential_may_not_count_another_tenant(app):
+    status, body = invoke(
+        app,
+        "/v1/reconciliation?source=approval-engine&tenant=tenant:coulomb"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    assert status.startswith("403")
+    assert body["error"] == "tenant_not_allowed"
+
+
+def test_the_count_response_states_what_it_does_not_prove(app):
+    _, body = invoke(
+        app,
+        "/v1/reconciliation?source=approval-engine&tenant=tenant:platform"
+        "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00",
+    )
+    means = body["means"].lower()
+    assert "proves neither completeness" in means
+
+
+# --- the overlay asymmetry -------------------------------------------------
+
+
+def test_the_overlay_may_shorten_a_heartbeat_interval_but_not_lengthen_it(tmp_path):
+    scope = tmp_path / "scope.json"
+    base = [{
+        "name": "approval-engine", "tokens": ["t"], "sources": ["approval-engine"],
+        "heartbeat_classes": {"approval.revocation": 3600},
+    }]
+
+    scope.write_text(json.dumps(
+        [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 600}}]
+    ))
+    registry = SenderRegistry.from_env({
+        "AUDIT_CORE_SENDERS": json.dumps(base),
+        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope),
+    })
+    assert dict(registry.identities[0].heartbeat_classes)["approval.revocation"] == 600
+
+    scope.write_text(json.dumps(
+        [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 99999}}]
+    ))
+    with pytest.raises(ValueError, match="lengthen"):
+        SenderRegistry.from_env({
+            "AUDIT_CORE_SENDERS": json.dumps(base),
+            "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope),
+        })
+
+    scope.write_text(json.dumps([{"name": "approval-engine", "heartbeat_classes": {}}]))
+    with pytest.raises(ValueError, match="remove"):
+        SenderRegistry.from_env({
+            "AUDIT_CORE_SENDERS": json.dumps(base),
+            "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope),
+        })