Prepare independent deployed cadence acceptance verifier

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2

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 '<owner-approved-exact-tenant>' \
+  --frame-seconds 3600 --now '<actual-observation-time-with-timezone>'
+```
+
+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.

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

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",
+        )

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.