From 96623fd46f9fde05c3379c092c6e586ab60dc280 Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 15 Oct 2026 21:18:13 +0200 Subject: [PATCH] Prepare independent deployed cadence acceptance verifier Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2 --- docs/CadenceObserverAdmission.md | 76 ++++++ src/kings_guard/acceptance.py | 242 ++++++++++++++++++ tests/test_acceptance.py | 182 +++++++++++++ ...-WP-0005-qonto-source-cadence-admission.md | 17 ++ 4 files changed, 517 insertions(+) create mode 100644 src/kings_guard/acceptance.py create mode 100644 tests/test_acceptance.py diff --git a/docs/CadenceObserverAdmission.md b/docs/CadenceObserverAdmission.md index a1d76d9..d252c65 100644 --- a/docs/CadenceObserverAdmission.md +++ b/docs/CadenceObserverAdmission.md @@ -120,3 +120,79 @@ artifacts and records the config digest. The historical Taxonomy draft fixture remains a historical worked example. Live admission must supply the actual configured frame/interval and preserve per-class scope; the source fixture cannot prove that a deployed process emits hourly heartbeats. + +## Deployed-capture verifier — 2026-10-16 + +`kings_guard.acceptance` is an offline consumer of runtime/evidence-owner +artifacts. It opens no network, cluster, database, or credential client. Run +from this repository after the runtime owner has delivered all four inputs: + +```bash +PYTHONPATH=src python3 -m kings_guard.acceptance \ + --events /path/to/source-instance.jsonl \ + --snapshot /path/to/source-snapshot.json \ + --receiver /path/to/receiver-reconciliation.json \ + --receipts /path/to/collector-accepted.jsonl \ + --source qonto-assistant --tenant '' \ + --frame-seconds 3600 --now '' +``` + +The output hashes its exact inputs. Exit0 means the supplied bounded capture +passes; exit1 means failure or invalid evidence. The verifier does not attest +input provenance, signatures, deployed image pins, or the operational matrix +below. A fixture passing the command remains fixture evidence. + +Input contract: + +- Raw JSONL contains exactly one source instance, startup sequence1 through + the source snapshot's last sequence, a real deny, and at least one periodic + heartbeat emitted at least f after startup. The source snapshot is the + unmodified `AuditLogger.reconciliation_snapshot()` response from that instance. +- Receiver JSON is the actual `/v1/reconciliation` response with exact source, + tenant, instance, and explicit acceptance-time since/until. Its counts must + equal the capture's allow/deny counts plus heartbeat transport records. +- Collector JSONL contains one durable original accepted receipt per source + sequence with `event_id`, `status_code`, `accepted_at`, `source`, `tenant`, + `stream_instance_id`, `stream_sequence`, `type`, and + `covered_event_classes`. The reviewed mapping is one source heartbeat to + one `audit-core.heartbeat` covering only `audit.deny`; it preserves original + source fields. This mapping does not claim separate receiver admission of + audit.allow or audit.heartbeat obligations. +- `accepted_at` records collector response receipt time, an upper bound on + acceptance. Original202 receipts can bound delivery lag; a duplicate200 + receipt alone cannot establish when the event first entered custody. The + verifier requires original202 for timing. Preserve earliest successful + receipts across retries, and supply a receiver window enclosing them. + Sequence/snapshot bounds are source occurrence coordinates, not receiver + acceptance times. Receipt correlation bridges those coordinates. + +The runtime owner must keep the authorized candidate active during the real +hour-long capture (for example authorized health polling) while retaining its +configured scale-to-zero behavior. Do not silently change minScale to obtain +heartbeats. Capture restart and scale-to-zero separately: a shutdown heartbeat +is a lifecycle claim, not proof the process remains alive; never present its +later absence as an unexpected omission when shutdown was intended. Resume +with a new instance and sequence1, then reconcile each instance separately. + +`assess_class_silence` independently computes class deadlines from owner-supplied +registered waits and explicit per-class receiver last-seen timestamps, then +compares the exact missing classes with the receiver's sanitized findings. +Use wait2f and grace1.5 (deadline3f). An explicit None is a receiver-observed +never-seen class; an omitted last-seen entry is refused. An unrelated fresh +class cannot conceal a stale rare class. Supply actual observations and their +owner/source/window pins; local projected times are algorithm tests only. + +Before T03 can close, collect independent operational receipts for quiet-active, +deny, periodic timing, bounded per-instance reconciliation, class-specific +missing heartbeat, sequence gap, source/accepted divergence, lag distinguished +from persistent divergence, failed cross-source/tenant reads, restart, and +scale-to-zero. Negative injection must use an owner-approved isolated lane; +do not suppress production evidence or change source registry bounds to make +a test finish sooner. The existing full-frame wait remains necessary. + +Validation: 144 tests pass, including16 new acceptance/correlation tests and +existing class-coverage regressions; changed-file Ruff passes. No live capture +or receiver finding was produced by this source delivery. Knowledge package +0.2.6 digest8d1d6a9d4d07fec7b619618de7dfe2be2254c1790f85f4198c7fe19db9751c02 +returned start/evidence withheld, orientation-only; current source cards and +Staff-layer owner boundary were read directly. This is not a readiness grant. diff --git a/src/kings_guard/acceptance.py b/src/kings_guard/acceptance.py new file mode 100644 index 0000000..30c90aa --- /dev/null +++ b/src/kings_guard/acceptance.py @@ -0,0 +1,242 @@ +"""Offline assessment of owner-delivered deployed Qonto evidence (no clients).""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from collections import Counter +from pathlib import Path +from typing import Any + +from kings_guard.cadence import parse_timestamp as _parse_timestamp +from kings_guard.live import _confirm_stream, _read_jsonl + + +def parse_timestamp(value: str): + if not isinstance(value, str) or not value.strip(): + raise ValueError("timestamp must be a timezone-qualified string") + from datetime import datetime + + if datetime.fromisoformat(value.replace("Z", "+00:00")).tzinfo is None: + raise ValueError("timestamp must include a timezone") + return _parse_timestamp(value) + + +def assess_capture( + events: list[dict[str, Any]], + snapshot: dict[str, Any], + receiver: dict[str, Any], + receipts: list[dict[str, Any]], + *, + source: str, + tenant: str, + frame_seconds: int, + now: str, +) -> dict[str, Any]: + """Assess actual supplied evidence; caller provenance remains an owner claim. + + Receiver time bounds are acceptance time, while source sequence bounds run + from startup to snapshot. Requiring one receipt per sequence links them. + """ + if type(frame_seconds) is not int or frame_seconds <= 0: + raise ValueError("frame_seconds must be a positive integer") + if not source or not tenant: + raise ValueError("exact source and tenant are required") + _, _, notes, corrections = _confirm_stream(events, snapshot) + findings = list(corrections) + now_dt = parse_timestamp(now) + instance = snapshot["stream_instance_id"] + times = [parse_timestamp(event["timestamp"]) for event in events] + snapshot_at = parse_timestamp(snapshot["snapshot_at"]) + if times != sorted(times) or times[-1] > snapshot_at or snapshot_at > now_dt: + findings.append("source timestamps are unordered or outside snapshot/evaluation bounds") + beats = [event for event in events if event["event_class"] == "audit.heartbeat"] + if not beats or beats[0].get("reason") != "startup" or events[0] != beats[0]: + findings.append("capture does not start with the source startup heartbeat") + periodic = [event for event in beats if event.get("reason") == "periodic"] + if ( + not periodic + or not beats + or not any( + (parse_timestamp(event["timestamp"]) - times[0]).total_seconds() >= frame_seconds + for event in periodic + ) + ): + findings.append("a real periodic heartbeat after one full frame is missing") + beat_times = [parse_timestamp(event["timestamp"]) for event in beats] + if any( + (b - a).total_seconds() > 2 * frame_seconds + for a, b in zip(beat_times, beat_times[1:], strict=False) + ): + findings.append("source heartbeat gap exceeds registered two-frame wait") + if not any(event["event_class"] in {"audit.allow", "audit.deny"} for event in events): + findings.append("capture has no request transition") + if not any(event["event_class"] == "audit.deny" for event in events): + findings.append("capture has no denial transition") + stale = not beat_times or (now_dt - beat_times[-1]).total_seconds() > 3 * frame_seconds + if stale: + findings.append("process heartbeat is stale beyond effective three-frame deadline") + if any( + receiver.get(key) != value + for key, value in (("source", source), ("tenant", tenant), ("stream_instance_id", instance)) + ): + findings.append("receiver source/tenant/instance binding does not match") + since, until = parse_timestamp(receiver["since"]), parse_timestamp(receiver["until"]) + if since > until or until > now_dt: + findings.append("receiver acceptance window is invalid or in the future") + rows = receiver.get("counts", []) + received: dict[str, int] = {} + for row in rows: + key, value = row["class"], row["count"] + if key in received or type(value) is not int or value < 0: + raise ValueError("receiver counts must be unique nonnegative integers") + received[key] = value + expected = Counter( + "audit-core.heartbeat" if e["event_class"] == "audit.heartbeat" else e["event_class"] + for e in events + ) + if {k: v for k, v in received.items() if v} != dict(expected): + findings.append("receiver accepted counts diverge from captured instance") + by_sequence: dict[int, dict[str, Any]] = {} + ids: set[str] = set() + lags: list[float] = [] + for receipt in receipts: + sequence = receipt["stream_sequence"] + if type(sequence) is not int or sequence in by_sequence or not 1 <= sequence <= len(events): + raise ValueError("receipt sequences must be unique and within the source capture") + by_sequence[sequence] = receipt + event = events[sequence - 1] + expected_type = ( + "audit-core.heartbeat" + if event["event_class"] == "audit.heartbeat" + else event["event_class"] + ) + accepted_at = parse_timestamp(receipt["accepted_at"]) + event_id = receipt.get("event_id") + if not event_id or event_id in ids: + findings.append("receipt event ID is missing or duplicated") + ids.add(event_id) + if ( + receipt.get("stream_instance_id") != instance + or receipt.get("source") != source + or receipt.get("tenant") != tenant + or receipt.get("type") != expected_type + or receipt.get("status_code") not in (200, 202) + or not since <= accepted_at <= until + ): + findings.append("receipt binding/status/acceptance window is invalid") + if event["event_class"] == "audit.heartbeat" and receipt.get("covered_event_classes") != [ + "audit.deny" + ]: + findings.append("heartbeat receipt must retain exact audit.deny coverage") + lag = (accepted_at - times[sequence - 1]).total_seconds() + if receipt.get("status_code") != 202: + findings.append("first-acceptance 202 receipt is required to assess delivery lag") + else: + lags.append(lag) + if lag < 0 or lag > frame_seconds: + findings.append("delivery lag exceeds one frame or clocks are inconsistent") + if set(by_sequence) != set(range(1, len(events) + 1)): + findings.append("durable acceptance receipt missing for source sequence") + findings = list(dict.fromkeys(findings)) + return { + "schema_version": "kings-guard-capture-assessment/v1", + "result": "pass" if not findings else "fail", + "findings": findings, + "source": source, + "tenant": tenant, + "stream_instance_id": instance, + "frame_seconds": frame_seconds, + "effective_deadline_seconds": 3 * frame_seconds, + "evaluated_at": now, + "max_delivery_lag_upper_bound_seconds": max(lags, default=None), + "source_event_count": len(events), + "notes": list(notes), + "scope": ( + "supplied deployed capture only; provenance and missing-heartbeat " + "injection are separate owner receipts" + ), + "means": ( + "Receipt time bounds acceptance from above; count agreement does not " + "prove completeness or omitted event absence." + ), + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + for name in ("events", "snapshot", "receiver", "receipts"): + parser.add_argument(f"--{name}", required=True, type=Path) + parser.add_argument("--source", required=True) + parser.add_argument("--tenant", required=True) + parser.add_argument("--frame-seconds", required=True, type=int) + parser.add_argument("--now", required=True, help="Explicit timezone-qualified observation time") + args = parser.parse_args() + try: + result = assess_capture( + _read_jsonl(args.events), + json.loads(args.snapshot.read_text()), + json.loads(args.receiver.read_text()), + _read_jsonl(args.receipts), + source=args.source, + tenant=args.tenant, + frame_seconds=args.frame_seconds, + now=args.now, + ) + result["input_sha256"] = { + name: hashlib.sha256(getattr(args, name).read_bytes()).hexdigest() + for name in ("events", "snapshot", "receiver", "receipts") + } + except (ValueError, KeyError, TypeError, IndexError, OSError) as exc: + result = {"result": "invalid", "error": str(exc)} + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["result"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +def assess_class_silence( + *, + source: str, + sender: str, + registered_wait_seconds: dict[str, int], + last_seen: dict[str, str | None], + findings: list[dict[str, Any]], + now: str, +) -> dict[str, Any]: + """Compare owner-provided receiver observations with independent class clocks. + + This cannot attest that supplied observations came from production. Each + class needs an explicit last-seen entry (None is an observed never-seen + value). Healthy unrelated classes cannot cover a missing one. + """ + now_dt = parse_timestamp(now) + expected: set[str] = set() + for event_class, wait in registered_wait_seconds.items(): + if type(wait) is not int or wait <= 0 or event_class not in last_seen: + raise ValueError("each registered class needs a positive wait and explicit last_seen") + value = last_seen[event_class] + stamp = parse_timestamp(value) if value is not None else None + if stamp is not None and stamp > now_dt: + raise ValueError("heartbeat last_seen cannot be in the future") + if stamp is None or (now_dt - stamp).total_seconds() > wait * 1.5: + expected.add(event_class) + actual = { + item["class"] + for item in findings + if item.get("source") == source + and item.get("sender") == sender + and item.get("kind") in {"missing_heartbeat", "no_heartbeat_since_registration"} + } + return { + "result": "pass" if actual == expected else "fail", + "source": source, + "sender": sender, + "evaluated_at": now, + "expected_missing_classes": sorted(expected), + "actual_missing_classes": sorted(actual), + "means": "Independent evaluation of supplied class clocks; no suppression inference.", + } diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py new file mode 100644 index 0000000..58a9f86 --- /dev/null +++ b/tests/test_acceptance.py @@ -0,0 +1,182 @@ +"""All inputs here are test fixtures, never live operational receipts.""" + +from copy import deepcopy + +import pytest + +from kings_guard.acceptance import assess_capture + + +def capture(): + instance = "test-instance" + events = [] + for seq, cls, stamp, reason, count in ( + (1, "audit.heartbeat", "2026-10-15T10:00:00Z", "startup", 0), + (2, "audit.deny", "2026-10-15T10:01:00Z", None, 1), + (3, "audit.heartbeat", "2026-10-15T11:00:00Z", "periodic", 1), + ): + event = dict( + stream_id="qonto-assistant.audit", + stream_instance_id=instance, + stream_sequence=seq, + event_class=cls, + timestamp=stamp, + ) + if reason: + event.update( + reason=reason, + assertion="nothing-to-report" if not count else "transitions-reported", + source_transition_counts={"audit.allow": 0, "audit.deny": count}, + ) + events.append(event) + snapshot = dict( + stream_id="qonto-assistant.audit", + stream_instance_id=instance, + last_stream_sequence=3, + snapshot_at="2026-10-15T11:00:02Z", + source_transition_counts={"audit.allow": 0, "audit.deny": 1}, + ) + receiver = dict( + source="qonto-assistant", + tenant="test-tenant", + stream_instance_id=instance, + since="2026-10-15T09:59:00Z", + until="2026-10-15T11:00:03Z", + counts=[{"class": "audit.deny", "count": 1}, {"class": "audit-core.heartbeat", "count": 2}], + ) + receipts = [ + dict( + event_id=f"test-{e['stream_sequence']}", + status_code=202, + accepted_at=e["timestamp"], + stream_instance_id=instance, + stream_sequence=e["stream_sequence"], + type="audit-core.heartbeat" if e["reason"] else e["event_class"], + source="qonto-assistant", + tenant="test-tenant", + covered_event_classes=["audit.deny"] if e["reason"] else [], + ) + for e in [dict(e, reason=e.get("reason")) for e in events] + ] + return events, snapshot, receiver, receipts + + +def assess(parts): + return assess_capture( + *parts, + source="qonto-assistant", + tenant="test-tenant", + frame_seconds=3600, + now="2026-10-15T11:00:04Z", + ) + + +def test_complete_fixture_capture_passes_only_its_bounded_evidence(): + result = assess(capture()) + assert result["result"] == "pass" + assert result["effective_deadline_seconds"] == 10800 + assert "supplied deployed capture only" in result["scope"] + + +@pytest.mark.parametrize( + "kind", + [ + "gap", + "instance", + "divergence", + "missing_receipt", + "wrong_class", + "duplicate_200", + "lag", + "short_periodic", + "source_binding", + "tenant_binding", + ], +) +def test_refuses_broken_correlation(kind): + events, snapshot, receiver, receipts = deepcopy(capture()) + if kind == "gap": + events[1]["stream_sequence"] = 4 + elif kind == "instance": + receiver["stream_instance_id"] = "other" + elif kind == "divergence": + receiver["counts"][0]["count"] = 2 + elif kind == "missing_receipt": + receipts.pop() + elif kind == "wrong_class": + receipts[0]["covered_event_classes"] = ["audit.allow"] + elif kind == "duplicate_200": + receipts[0]["status_code"] = 200 + elif kind == "lag": + receipts[0]["accepted_at"] = "2026-10-15T11:00:01Z" + elif kind == "short_periodic": + events[2]["timestamp"] = "2026-10-15T10:30:00Z" + elif kind == "source_binding": + receipts[0]["source"] = "other" + elif kind == "tenant_binding": + receipts[0]["tenant"] = "other" + assert assess((events, snapshot, receiver, receipts))["result"] == "fail" + + +def test_timezone_missing_is_invalid(): + parts = capture() + parts[0][0]["timestamp"] = "2026-10-15T10:00:00" + with pytest.raises(ValueError, match="timezone"): + assess(parts) + + +def test_duplicate_receipt_sequence_is_invalid(): + parts = capture() + parts[3].append(parts[3][0]) + with pytest.raises(ValueError, match="unique"): + assess(parts) + + +def test_class_silence_does_not_accept_fresh_unrelated_class(): + from kings_guard.acceptance import assess_class_silence + + result = assess_class_silence( + source="approval-engine", + sender="approval-engine", + registered_wait_seconds={"issuance": 7200, "revocation": 7200}, + last_seen={"issuance": "2026-10-15T11:00:00Z", "revocation": "2026-10-15T07:00:00Z"}, + findings=[], + now="2026-10-15T11:00:01Z", + ) + assert result["result"] == "fail" + assert result["expected_missing_classes"] == ["revocation"] + + +def test_class_silence_matches_receiver_findings_per_class(): + from kings_guard.acceptance import assess_class_silence + + result = assess_class_silence( + source="approval-engine", + sender="approval-engine", + registered_wait_seconds={"issuance": 7200, "revocation": 7200}, + last_seen={"issuance": "2026-10-15T11:00:00Z", "revocation": None}, + findings=[ + dict( + source="approval-engine", + sender="approval-engine", + kind="no_heartbeat_since_registration", + **{"class": "revocation"}, + ) + ], + now="2026-10-15T11:00:01Z", + ) + assert result["result"] == "pass" + + +def test_class_silence_refuses_absent_observation_instead_of_inventing_never_seen(): + from kings_guard.acceptance import assess_class_silence + + with pytest.raises(ValueError, match="last_seen"): + assess_class_silence( + source="a", + sender="a", + registered_wait_seconds={"deny": 7200}, + last_seen={}, + findings=[], + now="2026-10-15T11:00:00Z", + ) diff --git a/workplans/KG-WP-0005-qonto-source-cadence-admission.md b/workplans/KG-WP-0005-qonto-source-cadence-admission.md index bf3eafc..d7bb1c1 100644 --- a/workplans/KG-WP-0005-qonto-source-cadence-admission.md +++ b/workplans/KG-WP-0005-qonto-source-cadence-admission.md @@ -139,3 +139,20 @@ original content remains in f5bb0f4. No audit-core record or existing workplan / task ID was changed, no Hub row or UUID was invented, and no existing record was retired or renumbered. This explicitly records the scoped source correction rather than bypassing a duplicate identity by assigning an arbitrary ID. + +### Operational acceptance preparation — 2026-10-16 + +Under NK-WP-0035, added an executable offline deployed-capture verifier and +independent per-class receiver-silence comparison to `kings_guard.acceptance`. +It checks real periodic elapsed time, contiguous source instance/sequence, +source snapshot, exact receiver source/tenant/instance counts, durable receipt +coverage, and first202 delivery-lag upper bounds. Duplicate200 receipts cannot +establish original acceptance time. Input SHA256 pins and bounded conclusions +prevent local fixture success from becoming a live conformance claim. + +Full suite144 passed (16 new regressions), changed-file Ruff passes. Runtime +owner input and actual negative/read/restart/scale-to-zero observation matrix +remain required; T03 remains wait. Detailed command, artifact contract, health +polling during hourly observation, shutdown interpretation and owner boundaries +are in docs/CadenceObserverAdmission.md. No live resource or credential client +was opened and no live acceptance receipt was fabricated.