audit-core/audit_core/stream_findings.py

"""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