"""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 # --- AUDIT-WP-0018-T03: an opening event that was never closed --------------- UNCLOSED_OPENING = "opening_without_closing" # How far back an opening event is still examined. The operational custody # window (`recoverable_days=30`): an older start cannot be judged against a # store that does not promise to still hold its close. CLOSING_LOOKBACK = timedelta(days=30) @dataclass(frozen=True) class ClosingFinding: """A start in custody with no matching end past its bound. For the bounded executor: a run that consumed its approval and emitted ``execution.run.started`` but whose ``execution.attribution.receipt`` is not in custody. The run may have partly executed. Audit-core does not infer what it did. """ sender: str source: str opening_class: str closing_class: str event_id: str correlation_id: str | None accepted_at: str bound: str deadline: str overdue_seconds: int def as_dict(self) -> dict[str, Any]: return { "kind": UNCLOSED_OPENING, "sender": self.sender, "source": self.source, "class": self.opening_class, "closed_by": self.closing_class, "event_id": self.event_id, "correlation_id": self.correlation_id, "accepted_at": self.accepted_at, "bound": self.bound, "deadline": self.deadline, "overdue_seconds": self.overdue_seconds, "means": ( "an opening event is in custody and no closing event with the " "same correlation_id has arrived by its deadline. The work may " "have partly executed, failed before emitting, or its close may " "have been lost or suppressed; audit-core does not infer which, " "and does not infer what the work did." ), } def evaluate_closing( identities: Iterable[Any], unclosed: Any, *, now: datetime | None = None, grace_factor: float = 1.5, ) -> list[ClosingFinding]: """Findings for declared opening events with no close past their deadline. ``unclosed(source, opening, closing, since)`` returns the backend's unclosed openings. The clock starts at the receiver's own ``accepted_at``, never a source-supplied time. The deadline is the earlier of the declared ``within_seconds`` ceiling and, when the opening event carries one, its ``data.window.not_after`` — a per-run bound can tighten the declaration but never extend it. ``grace_factor`` adds the same allowance heartbeats get (``within_seconds * (grace_factor - 1)``) so a close accepted moments late does not flap a finding; it widens the deadline and never removes it. """ now = now or datetime.now(timezone.utc) since = (now - CLOSING_LOOKBACK).isoformat() findings: list[ClosingFinding] = [] for identity in identities: for opening, closing, within in getattr(identity, "closing_classes", ()): for source in sorted(identity.sources): if source == "*": continue # no determinate stream to hold to a close for row in unclosed(source, opening, closing, since): started = _parse(row["accepted_at"]) if started is None: continue deadline = started + timedelta(seconds=within) bound = "declared" per_run = _parse(row.get("window_not_after")) if per_run is not None and per_run < deadline: deadline, bound = per_run, "window.not_after" allowance = timedelta(seconds=within * (grace_factor - 1)) overdue = now - (deadline + allowance) if overdue.total_seconds() <= 0: continue findings.append(ClosingFinding( sender=identity.name, source=source, opening_class=opening, closing_class=closing, event_id=row["event_id"], correlation_id=row.get("correlation_id"), accepted_at=started.isoformat(), bound=bound, deadline=deadline.isoformat(), overdue_seconds=int(overdue.total_seconds()), )) return findings