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
This commit is contained in:
parent
9fb167871b
commit
2b8c9097ad
13 changed files with 654 additions and 6 deletions
|
|
@ -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 |
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
294
tests/test_run_receipt_findings.py
Normal file
294
tests/test_run_receipt_findings.py
Normal file
|
|
@ -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]
|
||||
|
|
@ -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`.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue