diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 89be23f..c684867 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -114,7 +114,7 @@ | task | AUDIT-WP-0017-T07 | wait | — | workplans/AUDIT-WP-0017-production-audit-sources-and-correlation.md | | task | AUDIT-WP-0018-T01 | done | — | workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md | | task | AUDIT-WP-0018-T02 | done | — | workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md | -| task | AUDIT-WP-0018-T03 | todo | — | workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md | +| task | AUDIT-WP-0018-T03 | done | — | workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md | | intake | AUDIT-IN-0001 | closed | — | intakes/intakes.md | | intake | AUDIT-IN-0002 | closed | — | intakes/intakes.md | | intake | AUDIT-IN-0003 | closed | — | intakes/intakes.md | diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index a4e1bac..532a973 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -436,7 +436,8 @@ class IngestionApplication: }) def _stream_findings(self, start_response): - """Missing-heartbeat and below-cadence findings (AUDIT-WP-0009-T04, T05, T07).""" + """Missing-heartbeat, below-cadence and unclosed-opening findings + (AUDIT-WP-0009-T04, T05, T07; AUDIT-WP-0018-T03).""" from audit_core.emission_cadence import evaluate as evaluate_cadence from audit_core.stream_findings import evaluate @@ -465,6 +466,15 @@ class IngestionApplication: lambda source, since, until: counter(source, since, until, None), ) ] + unclosed = getattr(self.backend, "unclosed_openings", None) + if callable(unclosed): + # AUDIT-WP-0018-T03: a declared opening event with no close. + from audit_core.stream_findings import evaluate_closing + + findings += [ + f.as_dict() + for f in evaluate_closing(self.senders.identities, unclosed) + ] return self._json(start_response, HTTPStatus.OK, { "stream_findings": findings, }) diff --git a/audit_core/postgres_backend.py b/audit_core/postgres_backend.py index 964ad72..7a33463 100644 --- a/audit_core/postgres_backend.py +++ b/audit_core/postgres_backend.py @@ -482,6 +482,29 @@ class PostgresAuditBackend: sql += " GROUP BY action ORDER BY action" return [{"class": action, "count": count} for action, count in self._query(sql, params)] + def unclosed_openings( + self, source: str, opening: str, closing: str, since: str, limit: int = 500 + ) -> list[dict]: + """Opening-class events with no closing-class event on the same + ``correlation_id`` (AUDIT-WP-0018-T03). Identifiers and timing only.""" + rows = self._query( + "SELECT o.event_id, o.correlation_id, o.accepted_at, " + "CASE WHEN jsonb_typeof(o.record->'details'->'data'->'window'->'not_after') = 'string' " + "THEN o.record->'details'->'data'->'window'->>'not_after' END " + f"FROM {self._events} o " + "WHERE o.source = %s AND o.action = %s AND o.accepted_at >= %s " + f"AND NOT EXISTS (SELECT 1 FROM {self._events} c " + "WHERE c.correlation_id = o.correlation_id " + "AND c.source = %s AND c.action = %s) " + "ORDER BY o.accepted_at, o.event_id LIMIT %s", + (source, opening, since, source, closing, int(limit)), + ) + return [ + {"event_id": eid, "correlation_id": cid, "accepted_at": _iso(at), + "window_not_after": na} + for eid, cid, at, na in rows + ] + def last_heartbeats(self, source: str) -> dict[str, str]: """Most recent heartbeat per class for one source.""" rows = self._query( diff --git a/audit_core/senders.py b/audit_core/senders.py index aecae07..9f4b3d8 100644 --- a/audit_core/senders.py +++ b/audit_core/senders.py @@ -87,6 +87,14 @@ class SenderIdentity: # parsed from info-tech-canon wire schema 0.1 — the §17 shape the heartbeat # field above deliberately did not invent. ``None`` means undeclared. emission_cadence: Any = None + # AUDIT-WP-0018-T03. Pairs ``(opening class, closing class, within + # seconds)``: every event of the opening class must be followed by an + # event of the closing class with the same ``correlation_id``. The bounded + # executor's ``execution.run.started`` is closed by its + # ``execution.attribution.receipt``. A start with no close past the bound + # is a stream finding — the run may have partly executed and left only + # its start in custody. + closing_classes: tuple[tuple[str, str, int], ...] = () def __post_init__(self) -> None: if not self.name: @@ -132,6 +140,21 @@ class SenderIdentity: f"be a positive number of seconds, got {interval!r}" ) + for opening, closing, within in self.closing_classes: + if not str(opening).strip() or not str(closing).strip(): + raise ValueError( + f"sender {self.name!r}: a closing class pair needs both classes" + ) + if opening == closing: + raise ValueError( + f"sender {self.name!r}: {opening!r} cannot close itself" + ) + if not isinstance(within, int) or isinstance(within, bool) or within <= 0: + raise ValueError( + f"sender {self.name!r}: closing bound for {opening!r} must be " + f"a positive number of seconds, got {within!r}" + ) + @property def is_load_bearing(self) -> bool: return self.evidence_kind == EVIDENCE_LOAD_BEARING @@ -152,6 +175,10 @@ class SenderIdentity: "emission_cadence": ( self.emission_cadence.summary() if self.emission_cadence else None ), + "closing_classes": { + opening: {"closed_by": closing, "within_seconds": within} + for opening, closing, within in self.closing_classes + }, "detection_surface": None, } @@ -324,6 +351,7 @@ def _apply_scope_overlay( ), heartbeat_classes=_overlay_heartbeats(identity, extra), emission_cadence=_overlay_cadence(identity, extra), + closing_classes=_overlay_closing(identity, extra), ) ) return merged @@ -405,6 +433,55 @@ def _overlay_heartbeats( return tuple(sorted(proposed.items())) +def _overlay_closing( + identity: SenderIdentity, extra: dict[str, Any] +) -> tuple[tuple[str, str, int], ...]: + """The overlay may add a closing pair or shorten its bound; it may never + remove a pair, rebind its closing class, or lengthen the bound. + + Same asymmetry as heartbeats: a ConfigMap refresh must not quietly widen + the window in which a run with no receipt goes unnoticed. + """ + if "closing_classes" not in extra: + return identity.closing_classes + declared = {o: (c, w) for o, c, w in identity.closing_classes} + proposed = {o: (c, w) for o, c, w in _closing_classes(extra["closing_classes"])} + for opening, (closing, within) in declared.items(): + if opening not in proposed: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not remove " + f"closing class pair {opening!r}" + ) + if proposed[opening][0] != closing: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not rebind " + f"{opening!r} from {closing!r} to {proposed[opening][0]!r}" + ) + if proposed[opening][1] > within: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not lengthen " + f"the closing bound for {opening!r} " + f"({within}s -> {proposed[opening][1]}s)" + ) + return tuple(sorted((o, c, w) for o, (c, w) in proposed.items())) + + +def _closing_classes(value: Any) -> tuple[tuple[str, str, int], ...]: + """Parse ``{"": {"closed_by": "", "within_seconds": N}}``.""" + if not value: + return () + if not isinstance(value, dict): + raise ValueError("closing_classes must be an object keyed by opening class") + pairs = [] + for opening, spec in value.items(): + if not isinstance(spec, dict) or set(spec) != {"closed_by", "within_seconds"}: + raise ValueError( + f"closing_classes[{opening!r}] needs exactly closed_by and within_seconds" + ) + pairs.append((str(opening), str(spec["closed_by"]), spec["within_seconds"])) + return tuple(sorted(pairs)) + + def _overlay_cadence(identity: SenderIdentity, extra: dict[str, Any]) -> Any: """The overlay may not touch an emission cadence declaration. @@ -447,6 +524,7 @@ def _identity_from(entry: Any) -> SenderIdentity: 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")), + closing_classes=_closing_classes(entry.get("closing_classes")), emission_cadence=( parse_declaration(entry["emission_cadence"]) if entry.get("emission_cadence") is not None diff --git a/audit_core/sqlite_backend.py b/audit_core/sqlite_backend.py index e4ed522..187a848 100644 --- a/audit_core/sqlite_backend.py +++ b/audit_core/sqlite_backend.py @@ -246,6 +246,34 @@ class SQLiteAuditBackend: sql += " GROUP BY action ORDER BY action" return [{"class": action, "count": count} for action, count in self._query(sql, params)] + def unclosed_openings( + self, source: str, opening: str, closing: str, since: str, limit: int = 500 + ) -> list[dict]: + """Opening-class events with no closing-class event on the same + ``correlation_id`` (AUDIT-WP-0018-T03). + + Returns identifiers and timing only, never payloads. An opening event + without a correlation id can never be closed and is returned as such. + """ + rows = self._query( + "SELECT o.event_id, o.correlation_id, o.accepted_at, " + "json_extract(o.record, '$.details.data.window.not_after') " + "FROM events o " + "WHERE json_extract(o.record, '$.source') = ? " + "AND json_extract(o.record, '$.action') = ? AND o.accepted_at >= ? " + "AND NOT EXISTS (SELECT 1 FROM events c " + "WHERE c.correlation_id = o.correlation_id " + "AND json_extract(c.record, '$.source') = ? " + "AND json_extract(c.record, '$.action') = ?) " + "ORDER BY o.accepted_at, o.event_id LIMIT ?", + (source, opening, since, source, closing, int(limit)), + ) + return [ + {"event_id": eid, "correlation_id": cid, "accepted_at": at, + "window_not_after": na if isinstance(na, str) else None} + for eid, cid, at, na in rows + ] + def last_heartbeats(self, source: str) -> dict[str, str]: """Most recent heartbeat per class for one source.""" rows = self._query( diff --git a/audit_core/stream_findings.py b/audit_core/stream_findings.py index 1e16fc0..83b7202 100644 --- a/audit_core/stream_findings.py +++ b/audit_core/stream_findings.py @@ -129,3 +129,108 @@ def evaluate( 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 diff --git a/deploy/senders-scope.json b/deploy/senders-scope.json index 65e5fd0..93c21a5 100644 --- a/deploy/senders-scope.json +++ b/deploy/senders-scope.json @@ -61,6 +61,12 @@ "may_write": true, "may_read": false, "secret_policy": "reject", - "evidence_kind": "load-bearing" + "evidence_kind": "load-bearing", + "closing_classes": { + "execution.run.started": { + "closed_by": "execution.attribution.receipt", + "within_seconds": 7200 + } + } } ] diff --git a/deploy/senders-scope.yaml b/deploy/senders-scope.yaml index 236318f..1e402b1 100644 --- a/deploy/senders-scope.yaml +++ b/deploy/senders-scope.yaml @@ -11,6 +11,10 @@ # (AUDIT-WP-0018-T02): nk-bounded-executor receipts are kept by hash, so a # secret-shaped field must refuse the event rather than be silently masked. # +# closing_classes may be added or their bound shortened here, never removed, +# rebound or lengthened (AUDIT-WP-0018-T03): a started run with no receipt past +# the bound is a stream finding. +# # evidence_kind may be raised here (attributive -> load-bearing) but never # lowered: a ConfigMap refresh must not be able to drop a source's §9.6 # atomicity and detection obligations without anyone deciding to. @@ -87,6 +91,12 @@ data: "may_write": true, "may_read": false, "secret_policy": "reject", - "evidence_kind": "load-bearing" + "evidence_kind": "load-bearing", + "closing_classes": { + "execution.run.started": { + "closed_by": "execution.attribution.receipt", + "within_seconds": 7200 + } + } } ] diff --git a/docs/bounded-executor-receipts.md b/docs/bounded-executor-receipts.md index f55d002..236babb 100644 --- a/docs/bounded-executor-receipts.md +++ b/docs/bounded-executor-receipts.md @@ -206,3 +206,24 @@ Remaining work: and the 202 fix. - **railiance-platform:** custody of the sender bearer (ESO projection to `sso`). + +## Status 2026-11-03 — T03 detection implemented + +A `started` event that has no receipt is reported on `GET /v1/stream-findings` +as `opening_without_closing` (see `docs/stream-completeness.md`, "Opening +events that never close"). The pair is declared on `nk-bounded-executor` in +`deploy/senders-scope.{json,yaml}` with a 7200 s bound, which matches the pilot +windows. The finding appears 3 h after acceptance at the latest, because the +heartbeat grace allowance is added. + +The executor's `started` body (net-kingdom `bounded-executor/executor.py`) +does not yet carry the admission window. If NetKingdom adds +`data.window.not_after`, each run is held to its own window wherever that is +earlier than the declared bound. Without it, admissions longer than 2 h rely +on the declared ceiling. + +The executor's emitter now accepts 202 (accepted) and 200 (duplicate) +(`clients.AuditCore.emit`), which closes the defect recorded on 2026-10-31. + +The new detection runs once the ConfigMap is applied with a receiver image +built from this commit or later. Image `sha256:304cfb83` predates it. diff --git a/docs/stream-completeness.md b/docs/stream-completeness.md index 03696f6..ba35797 100644 --- a/docs/stream-completeness.md +++ b/docs/stream-completeness.md @@ -180,3 +180,48 @@ schema is identical). The result, stated at no more than it is: onto audit-core's evaluation by itself: a source needs `heartbeat_classes` on its registration too, and the two can drift apart. That is fixable on either side and has not been fixed on either. + +## Opening events that never close (AUDIT-WP-0018-T03) + +Some streams come in pairs. The bounded executor emits +`execution.run.started` once it has consumed its single approval, and +`execution.attribution.receipt` when the run ends. A crash in between leaves +only the start in custody, and neither heartbeats nor rate counts can see that. + +A sender's registration may therefore declare `closing_classes`: + +```json +"closing_classes": { + "execution.run.started": { + "closed_by": "execution.attribution.receipt", + "within_seconds": 7200 + } +} +``` + +`GET /v1/stream-findings` then carries an `opening_without_closing` finding +for every opening event that has no closing event with the same +`correlation_id` from the same source by its deadline. + +- **Clock.** The deadline counts from the receiver's own `accepted_at`, never + from a source-supplied time. +- **Bound.** The deadline is the declared `within_seconds`. If the opening event + carries `data.window.not_after`, the earlier of the two applies (`bound` names + which one). A per-run window can tighten the declaration but never extend it. +- **Grace.** The same 1.5× allowance as heartbeats is added + (`within_seconds × 0.5`), so a receipt accepted moments late does not flap + the finding. For the executor that means a finding at 3 h at the latest. +- **Lookback.** Opening events accepted in the last 30 days, the operational + custody window, are examined. +- **No correlation id.** An opening event without a `correlation_id` can never + be closed and is always reported. +- **Disclosure.** A finding carries identifiers and timing only: event id, + correlation id, accepted time, deadline. No payload is disclosed. +- **Overlay.** The scope overlay may add a pair or shorten its bound. It may + never remove a pair, rebind its closing class, or lengthen the bound. + +The finding's `means` says what it does not prove. The run may have partly +executed, failed before emitting, or had its receipt lost or suppressed. +Audit-core does not infer which, and does not infer what the run did. It is a +finding for the verifier and the founder. As with heartbeats, a compromised +emitter that suppresses its start and its receipt together is not covered. diff --git a/tests/test_bounded_executor_sender.py b/tests/test_bounded_executor_sender.py index 2621062..01553f2 100644 --- a/tests/test_bounded_executor_sender.py +++ b/tests/test_bounded_executor_sender.py @@ -147,6 +147,12 @@ def test_scope_entry_is_exact_write_only_reject_load_bearing(): "may_read": False, "secret_policy": "reject", "evidence_kind": "load-bearing", + "closing_classes": { + "execution.run.started": { + "closed_by": "execution.attribution.receipt", + "within_seconds": 7200, + } + }, } diff --git a/tests/test_run_receipt_findings.py b/tests/test_run_receipt_findings.py new file mode 100644 index 0000000..202be21 --- /dev/null +++ b/tests/test_run_receipt_findings.py @@ -0,0 +1,294 @@ +"""AUDIT-WP-0018-T03. A started run with no receipt is a stream finding. + +The bounded executor emits ``execution.run.started`` once it has consumed its +approval, and ``execution.attribution.receipt`` when the run ends. A crash in +between leaves only the start in custody. The scope overlay declares the pair +(``closing_classes``) and its bound; ``GET /v1/stream-findings`` reports every +start whose receipt has not arrived by then. Audit-core does not infer what +the run did. +""" + +import copy +import hashlib +import json +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +from audit_core import stream_findings +from audit_core.interface import AuditEvent +from audit_core.senders import SenderIdentity, SenderRegistry +from audit_core.sqlite_backend import SQLiteAuditBackend +from audit_core.stream_findings import UNCLOSED_OPENING, evaluate_closing + +from test_bounded_executor_sender import ( # noqa: E402 (tests/ is on sys.path) + RUN_ID, + app, # noqa: F401 (fixture) + post, + receipt_event, + started_event, +) +from test_stream_findings import invoke # noqa: E402 + +SOURCE = "net-kingdom.bounded-executor" +STARTED = "execution.run.started" +RECEIPT = "execution.attribution.receipt" +PG_URL = os.environ.get("AUDIT_CORE_TEST_DATABASE_URL") + + +def _postgres_available() -> bool: + if not PG_URL: + return False + try: + import psycopg + + with psycopg.connect(PG_URL, connect_timeout=3): + return True + except Exception: + return False + + +HAVE_PG = _postgres_available() + + +@pytest.fixture(params=["sqlite", "postgres"]) +def backend(request, tmp_path): + if request.param == "sqlite": + yield SQLiteAuditBackend(str(tmp_path / "runs.db")) + return + if not HAVE_PG: + pytest.skip("no PostgreSQL reachable (set AUDIT_CORE_TEST_DATABASE_URL)") + from audit_core.postgres_backend import PostgresAuditBackend + + schema = f"runs_{uuid.uuid4().hex[:12]}" + instance = PostgresAuditBackend(PG_URL, schema=schema) + try: + yield instance + finally: + with instance.pool.connection() as conn: + conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + instance.close() + + +def store(backend, event_id, action, correlation_id, *, source=SOURCE, data=None): + details = {"data": data or {"k": "v"}} + if correlation_id is not None: + details["correlation_id"] = correlation_id + event = AuditEvent( + event_id=event_id, source=source, action=action, resource="admission:a", + outcome="recorded", tenant="platform", scope="tenant", details=details, + observed_at="2026-11-03T10:00:00+00:00", + ) + digest = hashlib.sha256(json.dumps(event.as_record(), sort_keys=True).encode()).hexdigest() + backend.accept(event, digest) + + +EPOCH = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() + + +# --- backend query, both backends -------------------------------------------- + + +def test_a_start_without_a_receipt_is_unclosed(backend): + store(backend, "run:r1:started", STARTED, "r1") + rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH) + assert [(r["event_id"], r["correlation_id"]) for r in rows] == [("run:r1:started", "r1")] + assert rows[0]["accepted_at"] + assert rows[0]["window_not_after"] is None + + +def test_a_receipt_on_the_same_correlation_closes_the_start(backend): + store(backend, "run:r1:started", STARTED, "r1") + store(backend, "run:r1:receipt", RECEIPT, "r1") + store(backend, "run:r2:started", STARTED, "r2") + rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH) + assert [r["correlation_id"] for r in rows] == ["r2"] + + +def test_a_receipt_from_another_source_does_not_close(backend): + store(backend, "run:r1:started", STARTED, "r1") + store(backend, "other:r1:receipt", RECEIPT, "r1", source="someone-else") + rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH) + assert [r["correlation_id"] for r in rows] == ["r1"] + + +def test_a_start_without_a_correlation_id_can_never_close(backend): + store(backend, "run:x:started", STARTED, None) + store(backend, "run:x:receipt", RECEIPT, None) + rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH) + assert [r["event_id"] for r in rows] == ["run:x:started"] + + +def test_the_per_run_window_is_surfaced_and_a_malformed_one_is_not(backend): + store(backend, "run:r1:started", STARTED, "r1", + data={"window": {"not_after": "2026-11-03T12:00:00Z"}}) + store(backend, "run:r2:started", STARTED, "r2", data={"window": {"not_after": 5}}) + rows = {r["correlation_id"]: r for r in + backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)} + assert rows["r1"]["window_not_after"] == "2026-11-03T12:00:00Z" + assert rows["r2"]["window_not_after"] is None + + +def test_only_starts_accepted_since_the_lookback_are_examined(backend): + store(backend, "run:r1:started", STARTED, "r1") + future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + assert backend.unclosed_openings(SOURCE, STARTED, RECEIPT, future) == [] + + +# --- evaluator --------------------------------------------------------------- + +EXECUTOR = SenderIdentity( + name="nk-bounded-executor", tokens=("t",), sources=frozenset({SOURCE}), + tenants=frozenset({"platform"}), evidence_kind="load-bearing", + closing_classes=((STARTED, RECEIPT, 7200),), +) +T0 = datetime(2026, 11, 3, 10, 0, tzinfo=timezone.utc) + + +def rows(**extra): + row = {"event_id": "run:r1:started", "correlation_id": "r1", + "accepted_at": T0.isoformat(), "window_not_after": None} + row.update(extra) + return lambda source, opening, closing, since: [row] + + +def test_no_finding_inside_the_bound_and_its_grace(): + # 2 h bound plus the 1.5x grace allowance (1 h): due at 13:00. + assert evaluate_closing([EXECUTOR], rows(), now=T0 + timedelta(hours=2, minutes=59)) == [] + + +def test_a_finding_past_the_bound_and_grace(): + findings = evaluate_closing([EXECUTOR], rows(), now=T0 + timedelta(hours=3, minutes=1)) + assert len(findings) == 1 + body = findings[0].as_dict() + assert body["kind"] == UNCLOSED_OPENING + assert body["class"] == STARTED and body["closed_by"] == RECEIPT + assert body["correlation_id"] == "r1" + assert body["bound"] == "declared" + assert body["overdue_seconds"] == 60 + assert "does not infer what the work did" in body["means"] + + +def test_a_per_run_window_tightens_the_deadline(): + early = rows(window_not_after=(T0 + timedelta(minutes=30)).isoformat()) + findings = evaluate_closing([EXECUTOR], early, now=T0 + timedelta(hours=1, minutes=31)) + assert [f.bound for f in findings] == ["window.not_after"] + + +def test_a_per_run_window_never_extends_the_declared_bound(): + late = rows(window_not_after=(T0 + timedelta(days=1)).isoformat()) + findings = evaluate_closing([EXECUTOR], late, now=T0 + timedelta(hours=3, minutes=1)) + assert [f.bound for f in findings] == ["declared"] + + +def test_the_lookback_is_the_custody_window(): + seen = {} + + def spy(source, opening, closing, since): + seen["since"] = since + return [] + + evaluate_closing([EXECUTOR], spy, now=T0) + assert seen["since"] == (T0 - timedelta(days=30)).isoformat() + + +def test_a_wildcard_source_is_not_held_to_a_close(): + wild = SenderIdentity( + name="w", tokens=("w",), sources=frozenset({"*"}), + closing_classes=((STARTED, RECEIPT, 60),), + ) + assert evaluate_closing([wild], rows(), now=T0 + timedelta(days=2)) == [] + + +# --- declaration and overlay ------------------------------------------------- + + +def _overlay(tmp_path, closing): + scope = tmp_path / "scope.json" + scope.write_text(json.dumps([{"name": "x", "closing_classes": closing}])) + secret = json.dumps([{ + "name": "x", "tokens": ["x"], "sources": [SOURCE], + "closing_classes": {STARTED: {"closed_by": RECEIPT, "within_seconds": 7200}}, + }]) + return SenderRegistry.from_env({ + "AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), + }) + + +def test_the_overlay_may_shorten_the_bound(tmp_path): + reg = _overlay(tmp_path, {STARTED: {"closed_by": RECEIPT, "within_seconds": 3600}}) + assert reg.authenticate("Bearer x").closing_classes == ((STARTED, RECEIPT, 3600),) + + +@pytest.mark.parametrize("closing", [ + {STARTED: {"closed_by": RECEIPT, "within_seconds": 7201}}, + {STARTED: {"closed_by": "something.else", "within_seconds": 7200}}, + {"other.start": {"closed_by": RECEIPT, "within_seconds": 60}}, +]) +def test_the_overlay_may_not_lengthen_rebind_or_remove(tmp_path, closing): + with pytest.raises(ValueError): + _overlay(tmp_path, closing) + + +@pytest.mark.parametrize("closing", [ + {STARTED: {"closed_by": RECEIPT}}, + {STARTED: {"closed_by": RECEIPT, "within_seconds": 0}}, + {STARTED: {"closed_by": STARTED, "within_seconds": 60}}, + [STARTED, RECEIPT, 60], +]) +def test_a_malformed_declaration_is_refused(closing): + secret = json.dumps([{"name": "x", "tokens": ["x"], "sources": [SOURCE], + "closing_classes": closing}]) + with pytest.raises(ValueError): + SenderRegistry.from_env({"AUDIT_CORE_SENDERS": secret}) + + +# --- end to end on the executor's own envelopes and the Git overlay ---------- + + +@pytest.fixture +def three_hours_later(monkeypatch): + real = stream_findings.evaluate_closing + later = datetime.now(timezone.utc) + timedelta(hours=3, minutes=1) + monkeypatch.setattr(stream_findings, "evaluate_closing", + lambda ids, unclosed: real(ids, unclosed, now=later)) + + +def _closing_findings(app): # noqa: F811 + status, body = invoke(app, "/v1/stream-findings", token="operator") + assert status.startswith("200") + return [f for f in body["stream_findings"] if f["kind"] == UNCLOSED_OPENING] + + +def test_a_fresh_start_is_not_yet_a_finding(app): # noqa: F811 + assert post(app, started_event())[0] == 202 + assert _closing_findings(app) == [] + + +def test_a_start_with_no_receipt_past_the_bound_is_reported(app, three_hours_later): # noqa: F811 + assert post(app, started_event())[0] == 202 + findings = _closing_findings(app) + assert [(f["sender"], f["correlation_id"]) for f in findings] == [ + ("nk-bounded-executor", RUN_ID) + ] + assert findings[0]["event_id"] == f"run:{RUN_ID}:started" + # Identifiers and timing only: nothing from the run's data is disclosed. + assert "data" not in findings[0] and "details" not in findings[0] + + +def test_the_receipt_clears_the_finding(app, three_hours_later): # noqa: F811 + assert post(app, started_event())[0] == 202 + assert post(app, receipt_event())[0] == 202 + assert _closing_findings(app) == [] + + +def test_another_run_receipt_does_not_clear_this_one(app, three_hours_later): # noqa: F811 + assert post(app, started_event())[0] == 202 + other = copy.deepcopy(receipt_event()) + other["id"] = "run:other:receipt" + other["correlation_id"] = "other" + other["data"]["run_id"] = "other" + assert post(app, other)[0] == 202 + assert [f["correlation_id"] for f in _closing_findings(app)] == [RUN_ID] diff --git a/workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md b/workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md index a353851..af4f651 100644 --- a/workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md +++ b/workplans/AUDIT-WP-0018-bounded-executor-receipt-custody.md @@ -9,7 +9,7 @@ flavor: implementation owner: audit-core topic_slug: audit-evidence-dependencies created: "2026-10-31" -updated: "2026-10-31" +updated: "2026-11-03" related: - NK-WP-0056 - NK-WP-0040 @@ -87,7 +87,7 @@ See `docs/bounded-executor-receipts.md`, "Status 2026-10-31". ```task id: AUDIT-WP-0018-T03 -status: todo +status: done priority: medium state_hub_task_id: "ec2cd9b7-be65-5792-b546-ffd0b3937856" ``` @@ -103,6 +103,28 @@ Unblocked 2026-10-31. NetKingdom adopted the started event and set the bound: the admission's `window.not_after`. Pilot windows are 2 h. The receipt also carries `time_interval.end`. +Done 2026-11-03. A sender may declare `closing_classes`. For +`nk-bounded-executor`, `execution.run.started` must be closed by +`execution.attribution.receipt` with the same `correlation_id` within 7200 s. +`GET /v1/stream-findings` reports `opening_without_closing` past that bound. + +- **Deadline:** counted from the receiver's own `accepted_at`, with the + heartbeat grace allowance added, so a finding appears after 3 h at the + latest. A per-run `data.window.not_after` tightens the deadline but never + extends it. +- **Lookback:** 30 days. +- **Disclosure:** identifiers and timing only. +- **Overlay:** may add a pair or shorten its bound; may never remove, rebind or + lengthen. + +Code: `audit_core/stream_findings.py` (`evaluate_closing`) and +`unclosed_openings` on both backends. Tests: +`tests/test_run_receipt_findings.py` (30 tests, both backends). Docs: +`docs/stream-completeness.md`. + +- **Asked of NetKingdom:** put `window.not_after` in the started body. +- **Founder-run:** release an image from this commit and apply the ConfigMap. + ## Founder decision 2026-10-31 Approved by the founder (`proposed` -> `ready`) after the NK-WP-0056-T02 owner answers (hub decision 19afd15e). Founder adopted the `execution.run.started` event plus one receipt per run with an admission-level target/action, `steps[]` detail and labelled `decision_refs`.