From 24a5ecd56bee7bbe8ee31c7ae47bd9e1127751d8 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 11:34:42 +0200 Subject: [PATCH] Add bounded source reconciliation counts and clarify receiver dependencies Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1046e-fdbf-7b93-a267-5f72b9b24526 --- approval_engine/api.py | 13 ++- approval_engine/store.py | 45 ++++++++++ cadence.yaml | 20 ++--- docs/emission-cadence.md | 34 +++++++- docs/outbox-contract.md | 4 +- docs/production-dependencies.md | 18 +++- tests/test_reconciliation.py | 86 +++++++++++++++++++ ...duction-readiness-and-consumer-adoption.md | 28 +++++- 8 files changed, 231 insertions(+), 17 deletions(-) create mode 100644 tests/test_reconciliation.py diff --git a/approval_engine/api.py b/approval_engine/api.py index 78bbb2f..607ff19 100644 --- a/approval_engine/api.py +++ b/approval_engine/api.py @@ -4,9 +4,10 @@ from __future__ import annotations import json from typing import Any, Callable +from urllib.parse import parse_qs from .auth import Authenticator, DenyAllAuthenticator, Identity -from .errors import ApprovalError, Forbidden +from .errors import ApprovalError, Forbidden, Unprocessable from .store import Engine FORBIDDEN_DECISION_KEYS = frozenset({"effect", "decision", "allow", "deny"}) @@ -105,6 +106,13 @@ class App: return 200, self.engine.storage_status(integrity=True) if path == "/v1/cadence" and method == "GET": self.identity(environ, "approval:observe") + query = parse_qs(environ.get("QUERY_STRING", ""), keep_blank_values=True) + if query: + if set(query) != {"since", "until"} or any( + len(values) != 1 or not values[0] for values in query.values() + ): + raise Unprocessable("supply exactly one since and until") + return 200, self.engine.reconciliation_counts(query["since"][0], query["until"][0]) return 200, self.engine.transition_counts() | {"form": "heartbeat-or-reconciliation"} if path == "/v1/outbox/stats" and method == "GET": self.identity(environ, "approval:observe") @@ -181,6 +189,7 @@ def call( body: dict[str, Any] | None = None, *, authorization: str | None = "Bearer test-token", + query_string: str = "", ) -> tuple[int, dict[str, Any]]: """In-process WSGI helper for tests.""" raw = json.dumps(body or {}).encode("utf-8") if body is not None else b"" @@ -189,7 +198,7 @@ def call( "PATH_INFO": path, "wsgi.input": _Bytes(raw), "CONTENT_LENGTH": str(len(raw)) if body is not None else "0", - "QUERY_STRING": "", + "QUERY_STRING": query_string, } if authorization is not None: environ["HTTP_AUTHORIZATION"] = authorization diff --git a/approval_engine/store.py b/approval_engine/store.py index c546d68..c30decc 100644 --- a/approval_engine/store.py +++ b/approval_engine/store.py @@ -1055,6 +1055,51 @@ class Engine: counts[row["class"]] = row["n"] return counts + def reconciliation_counts(self, since: str, until: str) -> dict[str, Any]: + """One source snapshot; occurrence and local acknowledgment are distinct. + + Audit Core counts receiver acceptance time. Neither local timestamp + proves that time, especially after delayed delivery or a duplicate ack. + """ + bounds = [] + for value in (since, until): + try: + bound = parse_iso(value) + except (TypeError, ValueError) as exc: + raise Unprocessable("since/until must be timezone-aware timestamps") from exc + if bound.tzinfo is None: + raise Unprocessable("since/until must be timezone-aware timestamps") + bounds.append(bound.astimezone(timezone.utc)) + start, end = bounds + if start >= end: + raise Unprocessable("since must precede until") + try: + # A single statement holds a consistent snapshot while the worker + # acknowledges rows. No payload or identity data is selected. + rows = self._conn().execute( + "SELECT class, created_at, drained_at FROM outbox" + ).fetchall() + except sqlite3.Error as exc: + raise StoreUnavailable(str(exc)) from exc + counts = {c: {"committed": 0, "drained": 0, "pending": 0, + "acknowledged_in_window": 0} for c in EVENT_CLASSES} + for row in rows: + entry = counts[row["class"]] + if start <= parse_iso(row["created_at"]) < end: + entry["committed"] += 1 + entry["pending" if row["drained_at"] is None else "drained"] += 1 + if row["drained_at"] is not None and start <= parse_iso(row["drained_at"]) < end: + entry["acknowledged_in_window"] += 1 + return { + "source": "approval-engine", "tenant": self.tenant, + "since": start.isoformat(), "until": end.isoformat(), + "bounds": "since-inclusive-until-exclusive", + "window_basis": "created_at", "counts": counts, + "means": "Source occurrence counts and local acknowledgments; receiver " + "acceptance-time comparison requires drain-lag and boundary " + "accounting. Agreement does not prove completeness.", + } + def outbox_stats(self) -> dict[str, Any]: conn = self._conn() now = self.now() diff --git a/cadence.yaml b/cadence.yaml index 80d6669..2de79a5 100644 --- a/cadence.yaml +++ b/cadence.yaml @@ -9,8 +9,8 @@ "event_class": "issuance", "form": "heartbeat-or-reconciliation", "reconciliation": { - "compare_local": "committed outbox issuance counts for the same tenant and bounded window; drained counts exclude pending lag", - "compare_observed": "audit-core accepted action=approval.issuance counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds", + "compare_local": "committed outbox issuance occurrence-time counts in the chosen since/until window; drained and pending are separated, local acknowledgment-window counts are distinct", + "compare_observed": "audit-core accepted action=approval.issuance counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds; receiver windows use acceptance time, requiring explicit drain-lag, duplicate-acknowledgment and boundary accounting before comparison", "divergence": "finding", "undrained_local": "lag-not-divergence" }, @@ -43,8 +43,8 @@ "event_class": "use", "form": "heartbeat-or-reconciliation", "reconciliation": { - "compare_local": "committed outbox use counts for the same tenant and bounded window; drained counts exclude pending lag", - "compare_observed": "audit-core accepted action=approval.use counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds", + "compare_local": "committed outbox use occurrence-time counts in the chosen since/until window; drained and pending are separated, local acknowledgment-window counts are distinct", + "compare_observed": "audit-core accepted action=approval.use counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds; receiver windows use acceptance time, requiring explicit drain-lag, duplicate-acknowledgment and boundary accounting before comparison", "divergence": "finding", "undrained_local": "lag-not-divergence" }, @@ -77,8 +77,8 @@ "event_class": "supersession", "form": "heartbeat-or-reconciliation", "reconciliation": { - "compare_local": "committed outbox supersession counts for the same tenant and bounded window; drained counts exclude pending lag", - "compare_observed": "audit-core accepted action=approval.supersession counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds", + "compare_local": "committed outbox supersession occurrence-time counts in the chosen since/until window; drained and pending are separated, local acknowledgment-window counts are distinct", + "compare_observed": "audit-core accepted action=approval.supersession counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds; receiver windows use acceptance time, requiring explicit drain-lag, duplicate-acknowledgment and boundary accounting before comparison", "divergence": "finding", "undrained_local": "lag-not-divergence" }, @@ -111,8 +111,8 @@ "event_class": "revocation", "form": "heartbeat-or-reconciliation", "reconciliation": { - "compare_local": "committed outbox revocation counts for the same tenant and bounded window; drained counts exclude pending lag", - "compare_observed": "audit-core accepted action=approval.revocation counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds", + "compare_local": "committed outbox revocation occurrence-time counts in the chosen since/until window; drained and pending are separated, local acknowledgment-window counts are distinct", + "compare_observed": "audit-core accepted action=approval.revocation counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds; receiver windows use acceptance time, requiring explicit drain-lag, duplicate-acknowledgment and boundary accounting before comparison", "divergence": "finding", "undrained_local": "lag-not-divergence" }, @@ -145,8 +145,8 @@ "event_class": "heartbeat", "form": "heartbeat-or-reconciliation", "reconciliation": { - "compare_local": "committed outbox heartbeat counts for the same tenant and bounded window; drained counts exclude pending lag", - "compare_observed": "audit-core accepted action=audit-core.heartbeat counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds", + "compare_local": "committed outbox heartbeat occurrence-time counts in the chosen since/until window; drained and pending are separated, local acknowledgment-window counts are distinct", + "compare_observed": "audit-core accepted action=audit-core.heartbeat counts for source=approval-engine, tenant=tenant:platform and identical since/until bounds; receiver windows use acceptance time, requiring explicit drain-lag, duplicate-acknowledgment and boundary accounting before comparison", "divergence": "finding", "undrained_local": "lag-not-divergence" }, diff --git a/docs/emission-cadence.md b/docs/emission-cadence.md index 49bb901..879cc3f 100644 --- a/docs/emission-cadence.md +++ b/docs/emission-cadence.md @@ -40,7 +40,9 @@ accepted event counts for `source=approval-engine` and the corresponding | `revocation` | `approval.revocation` | | `heartbeat` | `audit-core.heartbeat` | -Undrained local rows are this engine's lag, not yet a divergence. A row +Account for occurrence versus receiver acceptance time as specified below; +identical window bounds alone do not align those populations. Undrained local +rows are this engine's lag, not yet a divergence. A row with `drained_at` set that `audit-core` does not hold is the omission case §9.6 names. @@ -111,3 +113,33 @@ uv run tools/emission-cadence-profile/emission_cadence_profile.py \ This supplies an owner-authored instance to NK-WP-0035; it does not accept the proposed NetKingdom profile or claim receiver admission/observed cadence. + + +## Bounded source snapshot — 2026-10-04 + +`GET /v1/cadence?since=&until=` requires approval:observe +and the exact store tenant. Supply exactly one timezone-aware since and until, +with since earlier than until; encode plus signs in timezone offsets. Bounds +are inclusive since, exclusive until. The existing query-free lifetime response +is preserved; it cannot substitute for a bounded reconciliation. + +The response identifies source, tenant, normalized UTC bounds and window_basis +created_at. Each class has committed, drained and pending counts for rows created +in that window, plus acknowledged_in_window for local drain acknowledgments in +that window regardless of creation time. All counts come from one database +statement snapshot. Zero classes are explicit; no payloads or identities are +selected or returned, and the source performs no receiver calls. + +Audit Core's published `/v1/reconciliation` uses **receiver acceptance time**, +not occurrence time. Local drained_at is acknowledgment time and may follow +acceptance, especially if an accepted event's response was lost and a duplicate +was acknowledged later. Equal textual since/until bounds alone do not align +those populations. The observer must account for delayed delivery, acknowledgment +boundaries and duplicate acceptance, or use independently correlated event ids +and an agreed settled window before treating a count mismatch as omission. +Pending occurrence rows are lag; neither count agreement nor this source +snapshot proves completeness or resistance to source compromise. + +This implements T04's missing bounded source computation. Receiver registration, +observed lag/findings and native reconciliation remain blocked owner acceptance; +no existing live image is promoted by this change. diff --git a/docs/outbox-contract.md b/docs/outbox-contract.md index eb0f78c..a15bdc8 100644 --- a/docs/outbox-contract.md +++ b/docs/outbox-contract.md @@ -29,7 +29,7 @@ object mutation. | `use` | public CAS accepts first `request_digest` and object becomes `consumed` | `approval.use` | | `supersession` | object becomes `superseded` | `approval.supersession` | | `revocation` | object becomes `revoked` | `approval.revocation` | -| `heartbeat` | signed *nothing to report* plus counts | `approval.heartbeat` | +| `heartbeat` | authenticated per-class *nothing to report* plus counts | `audit-core.heartbeat` | Expiry is a clock crossing, persisted on observation, and is **not** an emitted class. The validity window is already on the object. @@ -116,7 +116,7 @@ So `approval.issuance` and `approval.use` both carry a `threshold` object in | `required_count` | the threshold in force on the object at that moment | | `distinct_approver_count` | distinct subjects who had recorded an entry | | `threshold_met` | whether the evaluation passed | -| `approvers` | `subject_id`, `approved_at`, and `assurance` / `evidence_ref` when recorded | +| `approvers` | `subject_id`, `approved_at`, `principal_type`, nullable verified `tenant_source`, and `assurance` / `evidence_ref` when recorded | An auditor holding only the `use` row can recompute the evaluation without reading live rows — which matters because those rows may since have been diff --git a/docs/production-dependencies.md b/docs/production-dependencies.md index ab1ec6e..16f065e 100644 --- a/docs/production-dependencies.md +++ b/docs/production-dependencies.md @@ -6,7 +6,7 @@ the workplan remains `blocked` after completing reachable source work. | Task | Owner return required | Evidence that clears the wait | | --- | --- | --- | | T01 — authenticated approver evidence | key-cape and informed-decision; founder for release rollout | Exact deployed issuer revision and client/profile receipt for requester and human paths; human sub/type, tenant/source, resource audience, scopes and assurance captured as non-secret facts and correlated with a stored human-control entry. Promote scanned schema-v6 image under per-change approval, back up/migrate once, then verify native entry and issuance/use provenance. No tokens or secrets in receipts. | -| T04 — delivery and reconciliation | audit-core, with railiance-platform for deployed sender/custody acceptance | Accept current cadence.yaml inventory for source approval-engine and tenant tenant:platform, explicit redact policy and four transition heartbeat classes at the requested 172800s bound. Return actual lifecycle event ids/drain acknowledgment and independent accepted counts for the same bounded time window, with pending rows accounted separately. Prove startup beats, missing-class and lag findings (depth 100 / age 3600s), and scoped sender bearer revocation. | +| T04 — delivery and reconciliation | audit-core, with railiance-platform for deployed sender/custody acceptance | Accept current cadence.yaml inventory for source approval-engine and tenant tenant:platform, explicit redact policy and four transition heartbeat classes at the requested 172800s bound. Return actual lifecycle event ids/drain acknowledgment and independent accepted counts for an agreed bounded comparison, accounting for occurrence versus receiver acceptance time, delayed delivery, duplicate acknowledgments and boundary crossings; source pending rows remain lag. Prove startup beats, missing-class and lag findings (depth 100 / age 3600s), and scoped sender bearer revocation. | | T05 — live PEP consumption | secrets-engine, with informed-decision for the act-specific human review and access-engine for decision binding | Name the served PEP route/workload, immutable component revision, engine image/endpoint, authenticated caller identities and exact binding. Prove claim-before-decision and CAS consume after ALLOW before protected side effect; same-digest retry, different-digest conflict, spent-on-failure, revoked/expired/superseded/wrong-binding/unavailable denial with no protected callback. Preserve the native approval and decision digests, declared human_control where required, and consumer-owned side-effect/refusal evidence. | For live whitehat T-06 only, T05's named surface must additionally be covered by @@ -60,3 +60,19 @@ The [deployment runbook](../deploy/README.md) and APPROVAL-IN-0001 also remains open for the unassigned shared Taxonomy schema owner. It does not block this workplan's local claim contract and is not a substitute for the three owner returns above. + + +## Reconciliation follow-up review — 2026-10-04 + +T04 source preparation now includes authenticated bounded cadence snapshots, +with committed/drained/pending occurrence counts and separate local +acknowledgment-window counts. The older lifetime-only surface did not satisfy +the source declaration's bounded-window requirement. The local implementation +is complete; source schema remains v6 and awaits the existing release gate. + +Audit Core docs/stream-completeness.md explicitly defines receiver +acceptance-time windows. The required owner return must state the temporal +population and reconcile lag/duplicate/boundary cases; matching since/until +strings against occurrence counts is insufficient. Existing owner task bindings +remain key-cape (T01), audit-core (T04) and secrets-engine (T05). No new owner +receipt or named live T-06 surface arrived since the preceding review. diff --git a/tests/test_reconciliation.py b/tests/test_reconciliation.py new file mode 100644 index 0000000..a2fc46d --- /dev/null +++ b/tests/test_reconciliation.py @@ -0,0 +1,86 @@ +"""Bounded source counts do not equate occurrence, acknowledgment and acceptance.""" +from dataclasses import replace +from datetime import timedelta, timezone +from urllib.parse import urlencode + +import pytest + +from approval_engine.api import App, call +from approval_engine.auth import StaticTokenAuthenticator +from approval_engine.errors import StoreUnavailable +from approval_engine.store import EVENT_CLASSES +from tests.conftest import FROZEN, approve + + +def query(since=FROZEN, until=FROZEN + timedelta(seconds=1)): + return urlencode({"since": since.isoformat(), "until": until.isoformat()}) + + +def test_occurrence_window_separates_pending_and_delayed_acknowledgments(engine, app): + engine.clock = lambda: FROZEN - timedelta(seconds=1) + approve(engine) + engine.clock = lambda: FROZEN + current = approve(engine) + engine.clock = lambda: FROZEN + timedelta(seconds=1) + engine.consume(current.id, current.binding_digest) + status, before = call(app, "GET", "/v1/cadence", query_string=query()) + assert status == 200 + assert before["counts"]["issuance"] == { + "committed": 1, "drained": 0, "pending": 1, "acknowledged_in_window": 0, + } + assert before["counts"]["use"]["committed"] == 0 # until is exclusive + engine.clock = lambda: FROZEN + timedelta(seconds=2) + engine.drain(lambda _: None) + _, after = call(app, "GET", "/v1/cadence", query_string=query()) + assert after["counts"]["issuance"] == { + "committed": 1, "drained": 1, "pending": 0, "acknowledged_in_window": 0, + } + _, ack = call(app, "GET", "/v1/cadence", query_string=query( + FROZEN + timedelta(seconds=2), FROZEN + timedelta(seconds=3))) + assert ack["counts"]["issuance"]["acknowledged_in_window"] == 2 + assert ack["counts"]["use"]["acknowledged_in_window"] == 1 + assert all(c["committed"] == 0 for c in ack["counts"].values()) + assert set(after["counts"]) == set(EVENT_CLASSES) + assert after["source"] == "approval-engine" and after["tenant"] == engine.tenant + assert after["window_basis"] == "created_at" + assert "acceptance-time" in after["means"] + assert "approvers" not in str(after) and current.id not in str(after) + + +def test_timezone_and_subsecond_bounds_are_preserved(app, engine): + approve(engine) + offset = FROZEN.astimezone(timezone(timedelta(hours=2))) + _, result = call(app, "GET", "/v1/cadence", query_string=query(offset, offset + timedelta(seconds=1))) + assert result["counts"]["issuance"]["committed"] == 1 + _, result = call(app, "GET", "/v1/cadence", query_string=query( + FROZEN + timedelta(microseconds=1), FROZEN + timedelta(seconds=1))) + assert result["counts"]["issuance"]["committed"] == 0 + + +@pytest.mark.parametrize("qs", [ + "since=bad&until=bad", "since=2026-10-04&until=2026-10-05", + query(FROZEN, FROZEN), query(FROZEN + timedelta(seconds=1), FROZEN), + "since=x", "until=x", "since=&until=x", "since=x&since=y&until=z", + query() + "&tenant=other", "window=all", +]) +def test_bad_or_unbounded_queries_are_refused(app, qs): + assert call(app, "GET", "/v1/cadence", query_string=qs)[0] == 422 + + +def test_window_counts_preserve_lifetime_endpoint_and_authentication(app, engine): + approve(engine) + _, counts = call(app, "GET", "/v1/cadence") + assert counts["issuance"] == 1 and counts["form"] == "heartbeat-or-reconciliation" + assert call(app, "GET", "/v1/cadence", authorization=None, query_string=query())[0] == 401 + identity = app.authenticator.identities["test-token"] + for altered in (replace(identity, scopes=frozenset({"approval:read"})), + replace(identity, tenant="tenant:other")): + other = App(engine, StaticTokenAuthenticator({"test-token": altered})) + assert call(other, "GET", "/v1/cadence", query_string=query())[0] == 403 + + +def test_window_counts_fail_closed_on_store_failure(app, engine, monkeypatch): + def unavailable(): + raise StoreUnavailable("unavailable") + monkeypatch.setattr(engine, "_conn", unavailable) + assert call(app, "GET", "/v1/cadence", query_string=query())[0] == 503 diff --git a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md index 1758bb3..e3ff5ca 100644 --- a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md +++ b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md @@ -551,7 +551,7 @@ status: wait flavor: residual priority: high blocked_on: message-from:audit-core -blocking_reason: "Sender custody, native synthetic delivery/readback and service deployment are delivered. Await audit-core acceptance of current cadence/class inventory, actual lifecycle drain and same-window reconciliation, class-silence/lag findings and scoped bearer-revocation receipt. Existing AUDIT-WP-0009 implementation is complete and reused." +blocking_reason: "Sender custody, native synthetic delivery/readback and service deployment are delivered. Await audit-core acceptance of current cadence/class inventory, actual lifecycle drain and reconciliation accounting for occurrence/acceptance-time lag, duplicates and boundaries, class-silence/lag findings and scoped bearer-revocation receipt. Existing AUDIT-WP-0009 implementation is complete and reused." state_hub_task_id: "777a5cf5-c1db-5e07-8de6-ab109db5ccdc" ``` @@ -845,3 +845,29 @@ Validation: 181 tests pass, including 12 new signed-HTTP, malformed-claim, persistent migration/restart and transactional rollback cases. Repository implementation is complete for these open tasks; native owner acceptance and the per-change release/rollout gate remain. Workplan stays blocked. + + +## Bounded reconciliation follow-up — 2026-10-04 + +Repeated the blocked-task review after c9ad5c2. No new owner admission receipt +arrived. T04 had a reachable source gap: cadence.yaml promised bounded local +counts, but `/v1/cadence` returned only lifetime totals and outbox stats lacked +a time window. Implemented optional explicit since/until on the authenticated +source route, retaining lifetime behavior. A single read snapshot returns +per-class committed/drained/pending occurrence counts and separate local +acknowledgment-window counts, without payloads, identity data or receiver calls. + +Receiver contract review exposed a temporal dependency: audit-core counts +acceptance time. Updated the declaration and production dependency requirements +to require lag, duplicate and boundary accounting before comparison; matching +window text does not align occurrence, acknowledgment and acceptance. Corrected +the stale approval.heartbeat alias in docs/outbox-contract.md. + +T01/T04/T05 remain wait with their named owner bindings; the workplan remains +blocked after this source work. Release/rollout, receiver acceptance and a served +PEP remain external gates. No production change, packet or other-repo edit. + +Validation: full suite 195 passed (14 new bounded-count/API cases); imported +owner cadence schema and NetKingdom profile returned contract_valid, +profile_assessed and conformant true with zero findings. git diff --check +passed. No remaining local implementation was identified in this review.