AUDIT-WP-0018-T03: report bounded-executor runs that started without a receipt

A sender may declare closing_classes; nk-bounded-executor declares
execution.run.started closed by execution.attribution.receipt within
7200s. GET /v1/stream-findings reports opening_without_closing past the
bound, counted from the receiver's accepted_at plus the heartbeat grace
allowance; a per-run data.window.not_after tightens but never extends.
Identifiers and timing only. The overlay may add or shorten, never
remove, rebind or lengthen. Both backends: 403 passed (1 unrelated
live-catalog env failure).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 482098@bnt-lap001
Assistant-Session: 9643c95a-a398-4f4a-b56a-6119be4e845e

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 |

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,
         })

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(

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 ``{"<opening>": {"closed_by": "<closing>", "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

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(

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

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
+      }
+    }
   }
 ]

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
+          }
+        }
       }
     ]

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.

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.

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,
+            }
+        },
     }
 
 

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]

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`.