diff --git a/SCOPE.md b/SCOPE.md index 709f4df..12f3857 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -6,11 +6,11 @@ ## Status — 2026-09-10 -**The domain core, browser sign-in shell and Approval Engine HTTP adapter -are implemented. The approval surface is not deployed.** +**The domain core, browser sign-in shell, Approval Engine HTTP adapter, +durable evidence store and Audit Core delivery adapter are implemented. The approval surface is not deployed.** -What exists and is tested (206 tests, including the explicit real-engine -component suite; 100 were present before browser integration): +What exists and is tested (258 tests, including explicit checks against the +actual Approval Engine and Audit Core implementations with synthetic identities): - layer and stance declarations — `layer.yaml`, `pep-stance.yaml`, `informed_decision/stance.py`, with published-equals-shipped asserted; @@ -19,22 +19,31 @@ component suite; 100 were present before browser integration): - the **domain core**: `memo.py` (the Decision Memo, its versions and the binding document), `presentation.py` (the sole writer of `view_hash`), `disposition.py` (the verb vocabulary and guards `G_NOAGENT`, `G_STEP`, - `G_PRES`, `G_ACK`, `G_REASONS`, `G_SEALED`), `provenance.py` (claim routes, + `G_PRES`, `G_ACTOR`, `G_ACK`, `G_REASONS`, `G_SEALED`), `provenance.py` (claim routes, A-16), `evidence.py` (the local outbox and commitment records); - `approval_client.py` — the seam to `approval-engine` plus a fake; - `oidc.py` and `web.py` — public-client PKCE sign-in, verified human/MFA profile, bounded server-side sessions, protected cookies and CSRF sign-out; - `approval_http.py` and `http_transport.py` — get-by-id and human-entry transport, declared-control checks, real entry correlation, no consume route - or automatic mutation retry. This adapter has no public browser mutation route. + or automatic mutation retry. This adapter has no public browser mutation route; +- `store.py` / `records.py` — private durable packet/memo/presentation/ + disposition storage, append-only acknowledgments, transactional outbox and + submission correlation, safe reservation and backup/restore; +- `audit.py` — idempotent Audit Core delivery, bounded retry/blocked states, + explicit per-class count/time-basis comparison. Heartbeats are generated + without hiding undelivered evidence; host scheduling remains pending. -Remaining: durable memo/presentation/disposition storage and transactional -evidence outbox, entitlement-before-render integration, L3 review/acknowledgment/ -binding UI, independent audit delivery and native deployment proof. The existing -`Outbox` is in memory. Browser sessions are ephemeral, with no approval state. +Remaining: admitted policy package/caller and entitlement-before-render +integration, L3 review/acknowledgment/binding UI, policy observation persistence, +visible unresolved-entry recovery, scheduled independent audit delivery and +native deployment/custody proof. The legacy `evidence.Outbox` remains an +in-memory test double; the new `Store` supplies durable atomicity. Browser +sessions are ephemeral, with no approval state. `/readyz` returns 503 until the protected approval path is connected. The origin `decisions.coulomb.social` still serves an nginx placeholder. -See [browser-authentication.md](docs/browser-authentication.md). +See [browser-authentication.md](docs/browser-authentication.md) and +[durable-review-evidence.md](docs/durable-review-evidence.md). `INFD-WP-0001-T08` remains open for the live end-to-end proof, which is gated on `APPROVAL-WP-0002-T01` and a deployed `approval-engine`. @@ -82,7 +91,7 @@ trail. - The unreachable-engine stance map, built to v0.8 obligation 3, with published-equals-shipped asserted by test (`tests/test_layer_conformance.py`). -**Built as domain operations; durable HTTP integration remains:** +**Built as domain operations with durable custody; protected HTTP integration remains:** - The presentation record: what was rendered, to whom, when, in which locale and UI release. diff --git a/docs/browser-authentication.md b/docs/browser-authentication.md index 8ec0976..766fb21 100644 --- a/docs/browser-authentication.md +++ b/docs/browser-authentication.md @@ -97,11 +97,13 @@ The client is an internal seam, not a sufficient binding flow. Before a browser route can call it, T08 must connect: 1. The access-engine entitlement decision before rendering a named memo. -2. A durable presentation/version, actor match and required acknowledgments. -3. Dispositions and a transactional evidence outbox, including return/discuss. +2. Wire the durable presentation/version, actor match and required acknowledgments. +3. Wire persisted dispositions and the transactional outbox, including return/discuss. 4. Independent audit custody/delivery and entry-correlation reconciliation. 5. Native registered KeyCape login and the deployed Approval Engine proof. The browser currently exposes no memo or entry route and no consume capability. -The current `evidence.Outbox` is in memory and cannot satisfy durable evidence -requirements. These remaining items stay in the active T08 record. +The legacy `evidence.Outbox` is an in-memory test double. `Store` now supplies +durable atomic state/evidence, and `audit.py` supplies delivery to the real +receiver contract; see [durable-review-evidence.md](durable-review-evidence.md). +Native custody, policy/controller integration and live admission stay in T08. diff --git a/docs/durable-review-evidence.md b/docs/durable-review-evidence.md new file mode 100644 index 0000000..7242c43 --- /dev/null +++ b/docs/durable-review-evidence.md @@ -0,0 +1,144 @@ +# Durable review evidence and audit delivery + +Implemented under `INFD-WP-0001-T08`. `store.py` replaces the domain tests' +in-memory outbox with a persistent internal store. It does not expose a browser +route, decide entitlement, or authorize an approval entry. `web.py` still has +no memo/entry routes and `/readyz` remains 503. + +## Storage and recovery + +The store uses a private SQLite file (0600), inside an existing directory owned +by the process user (0700). Symlinks, hard-linked databases, unsafe permissions, +unrecognized databases and unsupported schema versions are refused. WAL, +foreign keys and FULL synchronous commits are enabled. Connections are scoped +to operations; concurrent writers serialize through SQLite transactions. +This is a local Unix filesystem design, not shared network storage. + +The existing domain objects are serialized to JSON, without pickle or a second +Decision Memo schema. Memo versions, packet documents, original presentations, +acknowledgments, dispositions and evidence rows are immutable. SQL triggers +reject ordinary updates/deletes; they do not protect against a database +administrator or a compromised process with file access. + +Packet bytes must exist in the store before a memo can reference their SHA-256 +hashes. Presentation content includes the memo and the exact binding/awareness +documents. A presentation is produced only through `presentation.render`, the +existing writer of `view_hash`. Packet contents are checked against their +hashes on custody retrieval. A retrieval failure names Informed Decision as +the custodian that cannot produce the material; it never returns a blank as +though no evidence had been promised. + +Every presentation, explicit acknowledgment, disposition and submission result +commits its evidence and outbox row in the same local transaction. An outbox +write failure rolls the state change back. Acknowledgments append separate +records; they never overwrite the original presentation. A disposition retains +the acknowledgment snapshot that existed at that act, so later acknowledgment +does not retroactively strengthen its evidence. + +Actor and presentation recipient must match. Binding also requires verified +human provenance, the current memo version, all required acknowledgments and +the existing step/seal guards. An accept intent requires the native approval id +and carried digest. Return, discussion and decline remain distinct local acts; +only accept creates an external-submission record. A dependency outage is a +stance application, never an invented human decline. + +`Store.backup(new_path)` uses SQLite's backup API; it refuses an existing +destination. Restoring the resulting private database preserves evidence, +outbox ids and submission state. Tests cover process death before commit, +reopening after commit and restoring pending/confirmed records. An admitted +production backup location, schedule and restoration drill are still owed. +There is no automatic deletion/retention job in this implementation. + +## Approval submission and uncertainty + +These are internal delivery records, not cached approval validity: + +1. Record the human's accept intent and its evidence before an engine POST. +2. The controller must obtain a fresh applicable access-engine decision, verify + the session and compare the memo's carried digest with the live approval. + **That controller/policy integration is still pending.** +3. `begin_submission` atomically reserves one attempt. A stale presentation + cannot start; a memo cannot be revised while its entry attempt is in flight + or unresolved. Network I/O happens outside the SQLite transaction. +4. A successful non-duplicate entry response is correlated to the original + disposition using `(approval_id, subject, approved_at)`, atomically with an + evidence record. This confirms delivery, not permission to execute the act. + +Operation ids make the same click idempotent; reuse with different content is +refused. A second presentation cannot acquire another intent for the same +approval and approver. A known confirmed correlation retrieves the original +presentation and acknowledgment snapshot. + +**An unknown duplicate or lost response is unresolved.** It may refer to an +entry created before this presentation. The engine discards the POST body and +does not store a caller's presentation/operation id, so finding an entry later +does not prove which presentation caused it. The store does not attach it to +the latest view or automatically POST again. A crash after reservation likewise +does not reopen dispatch. The records remain visible for explicit controller/ +operator recovery; no recovery path in this slice fabricates that missing link. +The remaining T08 controller must handle this state visibly before shipping. + +## Audit Core contract + +`audit.py` sends the envelope stored at event creation, with stable event id, +`Idempotency-Key`, source `informed-decision` and tenant `tenant:platform`. +The sender credential is supplied by an owner-provided callback, remains out of +the database and is never sent to a redirected endpoint. No native credential +is provisioned or read by this implementation task. + +The receiver must return exactly `202 accepted` or `200 duplicate` with +`reference=audit:` before local delivery is marked complete. Lost +responses replay the exact bytes; Audit Core deduplicates them. A 30-second +local lease bounds concurrent claims; stale workers cannot overwrite a newer +delivery result. Retryable failures back off to a five-minute maximum. +Malformed events, conflicts and credential refusals remain visibly blocked; +an explicit requeue after repair preserves the original event id and body. +No failure drops the evidence or stores the upstream response body as an error. + +Only commitments and metadata travel: no brief, packet content or discussion +note. Native sender custody, receiver registration/application and independent +operator access still need live proof. Component tests use Audit Core's actual +ingestion/read APIs with a distinct SQLite fixture and synthetic scoped sender +and auditor identities. That is contract proof, not production custody proof. + +Per-class heartbeat generation uses the existing 86,400-second declarations. +It does not emit “nothing to report” for a class with undelivered evidence. +The host must schedule heartbeat generation, bounded draining and monitoring; +no background service or cadence is installed by this source change. + +## Reconciliation has two time bases + +Audit Core's current `/v1/reconciliation` counts rows by **accepted_at** and +returns `[{"class": ..., "count": ...}]`. The source counts events by +**occurred_at**. A delayed delivery can fall into different windows even when +nothing is missing. The client verifies source, tenant, explicit time window and +count shape, and reports both time bases with `count_values_match`. + +It deliberately sets `automatic_loss_finding=false`, `completeness_proven=false` +and `reconstructability_proven=false`. Compare stable catch-up totals and inspect +pending/blocked events and receiver references before interpreting divergence. +A production reconciliation procedure that accounts for delayed acceptance is +still part of T08 admission. Equal counts cannot establish equal membership, +complete emission, truthful presentation or availability of source-held content. +The GH-DEC-2026-014 commitment-only limitation remains. + +## Verification and next integration + +```sh +INFD_APPROVAL_ENGINE_SOURCE=/home/worsch/approval-engine \ +INFD_AUDIT_CORE_SOURCE=/home/worsch/audit-core \ +uv run python -m pytest -q +``` + +258 tests pass: 52 added tests cover transactional rollback, process death, +restore, actor/version/ack guards, concurrent clicks, original correlation, +ambiguous submissions, audit retry/refusal and actual receiver contracts. +The opt-in suites use the actual Approval Engine and Audit Core implementations +with synthetic identity/custody. They make no policy decision and spend nothing. + +Remaining under T08: an admitted PDP package/caller and exact read/bind request, +protected review/ack/accept/return/discuss routes, durable policy observations, +visible unresolved-entry recovery, scheduled audit delivery/reconciliation, +native registered human login and deployed-engine/custody proof. No Informed +Decision policy package or registration was found in the checked Flex Auth +examples, registry and docs at `88b3543`; do not substitute a local allow rule. diff --git a/docs/evidence/2026-09-10-durable-review-evidence.json b/docs/evidence/2026-09-10-durable-review-evidence.json new file mode 100644 index 0000000..4a455bd --- /dev/null +++ b/docs/evidence/2026-09-10-durable-review-evidence.json @@ -0,0 +1,60 @@ +{ + "schema": "informed-decision.durable-review-evidence.v1", + "observed_at": "2026-09-10T21:10:46.040050+00:00", + "base_commit": "0e48355b9fcb28ccd0ba09398ee5ac4311e66011", + "contract_source_commits": { + "approval-engine": "a0a602976eef818f36dde35f76f7f2e589bd051b", + "audit-core": "5c0ad522fb36092aa7ec2e8d72f63a5a91853b5b", + "flex-auth": "88b354377c8e26b162f1234e673072f1c06dcd89", + "key-cape": "139994cfac28ff97163ce4bf263f2bb035bbe0a7" + }, + "source_task": "INFD-WP-0001-T08", + "source_task_status": "progress", + "verification": { + "command": "INFD_APPROVAL_ENGINE_SOURCE=/home/worsch/approval-engine INFD_AUDIT_CORE_SOURCE=/home/worsch/audit-core python -m pytest -q", + "tests_passed": 258, + "tests_failed": 0, + "tests_skipped": 0, + "new_tests": 52, + "actual_api_component_tests_total": 7, + "actual_api_component_tests_new": 4, + "identity": "locally signed synthetic fixture", + "audit_custody": "actual Audit Core SQLite fixture, separate sender/auditor identities", + "native_policy_exercised": false + }, + "implemented": [ + "private persistent SQLite review/evidence store with immutable versions and packet custody", + "atomic presentation, explicit acknowledgment, disposition, submission result and outbox writes", + "actor-recipient, current-version and verified-human provenance checks", + "single-attempt entry reservation and original presentation correlation", + "explicit unresolved state after lost entry response or unknown duplicate; no automatic entry retry", + "Audit Core exact-byte idempotent delivery with bounded retries and retained blocked records", + "explicit backup/restore, heartbeat generation and bounded two-time-base reconciliation" + ], + "limits": [ + "internal store only; no new public approval routes", + "controller must obtain native PDP decision and validate session and live carried binding digest before render/dispatch", + "unknown entry causation remains unresolved; engine stores no caller presentation/operation id", + "heartbeats and outbox delivery require host scheduling and monitoring", + "source counts occurred_at while Audit Core counts accepted_at; equality does not prove completeness or reconstructability", + "no native sender custody or registration/login proof", + "no native PDP policy package found in checked Flex Auth examples, registry and docs; owner admission still required" + ], + "source_sha256": { + "informed_decision/records.py": "82f696151895c51b168144a3078d05764da3499b96917c465629ce0836fa8e33", + "informed_decision/store.py": "a26fffe647b866b061d40aec8ec4f1754d9a8df167973ccdd30ae5dfdb4d7ed4", + "informed_decision/audit.py": "06da36a8f3451b4abc7a46da2df66d5939524c170658b879499b2a47a1b10cd3", + "informed_decision/disposition.py": "5ddd19c2444f8e6c9bb090d04045cf1227a34554e000f07b3f677325bf8089f8", + "informed_decision/evidence.py": "5f84997f66409c7b6cec6ab57e73ad5bba2c1b2d669604adcbf36e72a824a88d", + "tests/test_durable_store.py": "e69b8c6c7dfe30291ed1eab2397f3740aa128a38da7c644894e7d04f3f365a29", + "tests/test_audit_delivery.py": "c5f8a35335393d74739cd21c1a4d5b5306b30dfd16966d223119d087c18d4fe0", + "tests/test_durable_component.py": "7f9e7b5075f7e66104b2b634d5bb14556bf4ff23e7060aae166e3697dc138c29", + "tests/test_skeleton.py": "1b9ab97c166870dca24c34149d5f9214ec186de31787407014cbb2342acb1d92" + }, + "browser_approval_routes_exposed": false, + "deployed": false, + "factory_attempts": 0, + "paid_model_calls": 0, + "remaining_live_task": "INFD-WP-0001-T08", + "factory_gate_task": "HFACT-WP-0001-T03" +} diff --git a/informed_decision/audit.py b/informed_decision/audit.py new file mode 100644 index 0000000..c75afb8 --- /dev/null +++ b/informed_decision/audit.py @@ -0,0 +1,108 @@ +"""Deliver immutable outbox bytes to Audit Core with its idempotency contract.""" + +from urllib.parse import urlencode + +from .http_transport import JSONTransport, TransportError, fixed_origin +from .store import bounded_window + + +class AuditDeliveryError(RuntimeError): + def __init__(self, code, permanent=False): + super().__init__(code) + self.code = code + self.permanent = permanent + + +class AuditCoreSink: + def __init__(self, origin, token_provider, *, transport=None, allow_internal_http=False): + self.origin = fixed_origin(origin, allow_internal_http=allow_internal_http) + self.token_provider = token_provider + self.transport = transport or JSONTransport(allow_internal_http=allow_internal_http) + + def _request(self, method, path, *, body=None, event_id=None): + try: + token = self.token_provider() + except (OSError, ValueError): + raise AuditDeliveryError("unauthorized", permanent=True) from None + if not isinstance(token, str) or not token or not token.isascii() or len(token) > 8192 or any(c.isspace() for c in token): + raise AuditDeliveryError("unauthorized", permanent=True) + headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} + if event_id: + headers["Idempotency-Key"] = event_id + try: + status, data = self.transport.request(method, self.origin + path, headers=headers, body=body) + except TransportError: + raise AuditDeliveryError("unavailable") from None + if status in (401, 403): + raise AuditDeliveryError("unauthorized", permanent=True) + if status == 400: + raise AuditDeliveryError("rejected", permanent=True) + if status == 409: + raise AuditDeliveryError("conflict", permanent=True) + if status >= 500: + raise AuditDeliveryError("unavailable") + return status, data + + def deliver(self, event_id, envelope_json): + status, data = self._request("POST", "/v1/events", body=envelope_json.encode(), event_id=event_id) + if ((status, data.get("status")) not in ((202, "accepted"), (200, "duplicate")) + or data.get("reference") != "audit:" + event_id): + raise AuditDeliveryError("invalid_receipt") + return data["reference"] + + def counts(self, since, until): + since, until = bounded_window(since, until) + query = {"source": "informed-decision", "tenant": "tenant:platform", "since": since, "until": until} + status, data = self._request("GET", "/v1/reconciliation?" + urlencode(query)) + try: + same_window = bounded_window(data.get("since"), data.get("until")) == (since, until) + except (TypeError, ValueError): + same_window = False + rows = data.get("counts") + if (status != 200 or data.get("source") != query["source"] or data.get("tenant") != query["tenant"] + or not same_window or not isinstance(rows, list)): + raise AuditDeliveryError("invalid_receipt") + counts = {} + for row in rows: + if (not isinstance(row, dict) or not isinstance(row.get("class"), str) or not row["class"] + or row["class"] in counts or type(row.get("count")) is not int or row["count"] < 0): + raise AuditDeliveryError("invalid_receipt") + counts[row["class"]] = row["count"] + return counts + + +class OutboxWorker: + def __init__(self, store, sink): + self.store = store + self.sink = sink + + def run_once(self, *, limit=100): + if type(limit) is not int or not 1 <= limit <= 1000: + raise ValueError("delivery batch limit must be 1..1000") + result = {"delivered": 0, "retrying": 0, "blocked": 0} + for _ in range(limit): + delivery = self.store.claim_delivery() + if delivery is None: + break + event_id, lease, body = delivery + try: + reference = self.sink.deliver(event_id, body) + except AuditDeliveryError as error: + self.store.finish_delivery(event_id, lease, error=error.code, permanent=error.permanent) + result["blocked" if error.permanent else "retrying"] += 1 + else: + self.store.finish_delivery(event_id, lease, reference=reference) + result["delivered"] += 1 + return result + + def reconcile(self, since, until): + since, until = bounded_window(since, until) + local = self.store.counts_by_class(since, until) + remote = self.sink.counts(since, until) + classes = sorted(set(local) | set(remote)) + return {"source": "informed-decision", "tenant": "tenant:platform", "since": since, "until": until, + "counts": {k: {"source": local.get(k, 0), "receiver": remote.get(k, 0)} for k in classes}, + "count_values_match": all(local.get(k, 0) == remote.get(k, 0) for k in classes), + "source_time_basis": "occurred_at", "receiver_time_basis": "accepted_at", + "automatic_loss_finding": False, + "completeness_proven": False, "reconstructability_proven": False} diff --git a/informed_decision/disposition.py b/informed_decision/disposition.py index aa8d14f..13d1e91 100644 --- a/informed_decision/disposition.py +++ b/informed_decision/disposition.py @@ -114,9 +114,10 @@ def record( Guards, in the order a defect is most likely to be caught: - - ``G_NOAGENT`` humans bind, agents draft. No upstream backstop exists. + - ``G_NOAGENT`` humans bind, agents draft; declared controls also guard upstream. - ``G_STEP`` the verb must be legal for this step kind. - ``G_PRES`` the presentation must be of this memo AND this version. + - ``G_ACTOR`` the acting person must be the presentation's recipient. - ``G_ACK`` required highlights acked before any binding verb. - ``G_REASONS`` a return carries at least one coded reason. - ``G_SEALED`` binding verbs on a sealed version are illegal. @@ -125,7 +126,7 @@ def record( raise DispositionRefused( "G_NOAGENT", f"{actor.kind.value} principals may draft but never bind " - "(INTENT principle 10; approval-engine provides no upstream backstop)", + "(INTENT principle 10)", ) if verb not in legal_verbs(memo.step_kind): @@ -145,6 +146,9 @@ def record( f"{memo.version} — no silent upgrade", ) + if actor.sub != presentation.principal_sub: + raise DispositionRefused("G_ACTOR", "actor did not receive this presentation") + if verb in BINDING_VERBS: if memo.sealed: raise DispositionRefused("G_SEALED", "binding verbs on a sealed version are illegal") diff --git a/informed_decision/evidence.py b/informed_decision/evidence.py index 5b7336d..40eccf3 100644 --- a/informed_decision/evidence.py +++ b/informed_decision/evidence.py @@ -1,4 +1,7 @@ -"""The local transactional outbox and the commitment records it queues. +"""Commitment records and the in-memory outbox used by domain tests. + +The persistent transactional implementation is ``store.Store``; this module +constructs the commitments used by both implementations. Payload is **commitment-only**, granted for Stage 1 by `GH-DEC-2026-014`: hashes, principal, timestamps, acks, the co-referenced approval id. Never the @@ -190,7 +193,7 @@ def heartbeat(event_class: EventClass) -> Commitment: class Outbox: - """Local, transactional. Written in the same transaction as the state change. + """In-memory domain-test double; use ``store.Store`` for durable atomicity. Emit-after-commit is a defect. The queue is local so an `audit-core` outage never blocks a binding act — the same reasoning that keeps it from blocking diff --git a/informed_decision/records.py b/informed_decision/records.py new file mode 100644 index 0000000..1f28268 --- /dev/null +++ b/informed_decision/records.py @@ -0,0 +1,48 @@ +"""JSON persistence of the existing domain objects; never pickle or a new memo schema.""" + +from dataclasses import asdict +import json + +from .disposition import Actor, ActorKind, Disposition, Verb +from .memo import (BindingLevel, BindingSlice, Highlight, Identifier, Memo, + PacketItem, Principal, Scope, StepKind) +from .presentation import Phase, Presentation +from .provenance import Claim, Route + + +def dumps(value) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, + allow_nan=False, default=lambda x: sorted(x) if isinstance(x, frozenset) else asdict(x)) + + +def memo_from(data: dict) -> Memo: + data = dict(data) + binding = dict(data["binding"]) + principal = dict(binding["principal"]) + principal["identifiers"] = tuple(Identifier(**i) for i in principal.get("identifiers", [])) + binding["principal"] = Principal(**principal) + binding["target"] = Scope(**binding["target"]) + data["binding"] = BindingSlice(**binding) + data["binding_level"] = BindingLevel(data["binding_level"]) + data["step_kind"] = StepKind(data["step_kind"]) + data["packet"] = tuple(PacketItem(**p) for p in data["packet"]) + data["highlights"] = tuple(Highlight(**h) for h in data["highlights"]) + return Memo(**data) + + +def presentation_from(data: dict) -> Presentation: + data = dict(data) + data["phase"] = Phase(data["phase"]) + data["acked_highlight_ids"] = frozenset(data["acked_highlight_ids"]) + for field in ("tenant", "principal_type"): + if data[field] is not None: + data[field] = Claim(data[field]["value"], Route(data[field]["route"])) + return Presentation(**data) + + +def disposition_from(data: dict) -> Disposition: + data = dict(data) + data["verb"] = Verb(data["verb"]) + data["actor"] = Actor(data["actor"]["sub"], ActorKind(data["actor"]["kind"])) + data["reasons"] = tuple(data["reasons"]) + return Disposition(**data) diff --git a/informed_decision/store.py b/informed_decision/store.py new file mode 100644 index 0000000..bfeebca --- /dev/null +++ b/informed_decision/store.py @@ -0,0 +1,459 @@ +"""Durable source-held evidence and transactional outbox for the review domain. + +This is an internal store, not an authorization endpoint. Callers owe an +access-engine decision before exposing its content or dispatching an entry. +Submission state describes local delivery, never approval validity. +""" + +from contextlib import contextmanager +from dataclasses import replace +from datetime import datetime, timezone +import hashlib +import json +import os +from pathlib import Path +import re +import sqlite3 +import stat +import time +import uuid + +from .approval_client import EntryResult +from .disposition import Actor, BINDING_VERBS, DispositionRefused, Verb, record +from .evidence import EventClass, commit_disposition, commit_presentation, commit_stance_application, heartbeat, HEARTBEAT_CLASSES +from .memo import Memo +from .presentation import render +from .provenance import Claim, Route, assert_human_control_dischargeable +from .stance import resolve +from .records import disposition_from, dumps, memo_from, presentation_from + +SCHEMA_VERSION = 1 +_SCHEMA = """ +CREATE TABLE documents (digest TEXT PRIMARY KEY, media_type TEXT NOT NULL, content BLOB NOT NULL); +CREATE TABLE memos (id TEXT NOT NULL, version INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY(id,version)); +CREATE TABLE presentations (id TEXT PRIMARY KEY, memo_id TEXT NOT NULL, version INTEGER NOT NULL, body TEXT NOT NULL, + FOREIGN KEY(memo_id,version) REFERENCES memos(id,version)); +CREATE TABLE acknowledgments (presentation_id TEXT NOT NULL REFERENCES presentations(id), + highlight_id TEXT NOT NULL, at TEXT NOT NULL, PRIMARY KEY(presentation_id,highlight_id)); +CREATE TABLE dispositions (id TEXT PRIMARY KEY, operation_id TEXT UNIQUE NOT NULL, request TEXT NOT NULL, + presentation_id TEXT NOT NULL REFERENCES presentations(id), body TEXT NOT NULL); +CREATE TABLE evidence (id TEXT PRIMARY KEY, class TEXT NOT NULL, at TEXT NOT NULL, + envelope TEXT NOT NULL, content TEXT NOT NULL); +CREATE TABLE outbox (id TEXT PRIMARY KEY REFERENCES evidence(id), state TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, next_attempt REAL NOT NULL DEFAULT 0, + lease TEXT, lease_until REAL, last_error TEXT, receiver_reference TEXT); +CREATE TABLE submissions (disposition_id TEXT PRIMARY KEY REFERENCES dispositions(id), + approval_id TEXT NOT NULL, subject TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'prepared', + attempt TEXT, approved_at TEXT, UNIQUE(approval_id,subject)); +""" + + +class StoreError(RuntimeError): + pass + + +class Conflict(StoreError): + pass + + +class EvidenceUnavailable(StoreError): + """Retrieval failure attributable to informed-decision, not an empty result.""" + + +def _text(value, name): + if not isinstance(value, str) or not value or len(value) > 256: + raise ValueError(f"invalid {name}") + + +def _now(): + return datetime.now(timezone.utc).isoformat(timespec="microseconds") + + +class Store: + def __init__(self, path: str | Path): + self.path = Path(path).absolute() + parent = self.path.parent + info = parent.lstat() + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() + or stat.S_IMODE(info.st_mode) != 0o700): + raise StoreError("evidence directory must be owned by this user with mode 0700") + try: + fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600) + except FileExistsError: + pass + else: + os.close(fd) + info = self.path.lstat() + if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid() + or stat.S_IMODE(info.st_mode) != 0o600): + raise StoreError("evidence database must be a private, owned, non-linked 0600 file") + with self._connection() as db: + db.execute("PRAGMA journal_mode=WAL") + version = db.execute("PRAGMA user_version").fetchone()[0] + if version == 0: + if db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchone(): + raise StoreError("unrecognized evidence database") + triggers = "" + for table in ("documents", "memos", "presentations", "acknowledgments", "dispositions", "evidence"): + for operation in ("UPDATE", "DELETE"): + triggers += (f"CREATE TRIGGER immutable_{table}_{operation} BEFORE {operation} ON {table} " + "BEGIN SELECT RAISE(ABORT,'immutable evidence'); END;\n") + db.executescript("BEGIN IMMEDIATE;" + _SCHEMA + triggers + "PRAGMA user_version=1; COMMIT;") + elif version != SCHEMA_VERSION: + raise StoreError("unsupported evidence schema version") + + @contextmanager + def _connection(self): + db = sqlite3.connect(self.path.as_uri() + "?mode=rw", uri=True, timeout=5, isolation_level=None) + db.row_factory = sqlite3.Row + db.execute("PRAGMA foreign_keys=ON") + db.execute("PRAGMA synchronous=FULL") + try: + yield db + finally: + db.close() + + @contextmanager + def _transaction(self): + with self._connection() as db: + db.execute("BEGIN IMMEDIATE") + try: + yield db + db.commit() + except BaseException: + db.rollback() + raise + + def put_document(self, content: bytes, media_type="text/plain") -> str: + if not isinstance(content, bytes) or len(content) > 16 * 1024 * 1024: + raise ValueError("packet document must be bytes, at most 16 MiB") + _text(media_type, "media type") + digest = "sha256:" + hashlib.sha256(content).hexdigest() + with self._transaction() as db: + existing = db.execute("SELECT media_type,content FROM documents WHERE digest=?", (digest,)).fetchone() + if existing and (existing["media_type"] != media_type or existing["content"] != content): + raise Conflict("document digest already has different content or media type") + if not existing: + db.execute("INSERT INTO documents VALUES (?,?,?)", (digest, media_type, content)) + return digest + + def save_memo(self, memo: Memo): + _text(memo.id, "memo id") + if type(memo.version) is not int or memo.version < 1: + raise ValueError("memo version must be a positive integer") + body = dumps(memo) + if len(body.encode()) > 256 * 1024: + raise ValueError("memo is too large") + with self._transaction() as db: + previous = db.execute("SELECT version,body FROM memos WHERE id=? ORDER BY version DESC LIMIT 1", (memo.id,)).fetchone() + if previous and previous["version"] == memo.version and previous["body"] == body: + return + if memo.version != (previous["version"] + 1 if previous else 1): + raise Conflict("memo version must advance by one; existing versions are immutable") + if db.execute("SELECT 1 FROM submissions s JOIN dispositions d ON d.id=s.disposition_id " + "JOIN presentations p ON p.id=d.presentation_id WHERE p.memo_id=? " + "AND s.state IN ('in_flight','unresolved') LIMIT 1", (memo.id,)).fetchone(): + raise Conflict("resolve the outstanding entry attempt before revising this memo") + for item in memo.packet: + if not db.execute("SELECT 1 FROM documents WHERE digest=?", (item.hash,)).fetchone(): + raise EvidenceUnavailable("informed-decision does not hold the referenced packet content") + db.execute("INSERT INTO memos VALUES (?,?,?)", (memo.id, memo.version, body)) + + def _memo(self, db, memo_id, version=None): + if version is None: + row = db.execute("SELECT body FROM memos WHERE id=? ORDER BY version DESC LIMIT 1", (memo_id,)).fetchone() + else: + row = db.execute("SELECT body FROM memos WHERE id=? AND version=?", (memo_id, version)).fetchone() + if row is None: + raise EvidenceUnavailable("informed-decision cannot produce the named memo version") + return memo_from(json.loads(row["body"])) + + def memo(self, memo_id): + with self._connection() as db: + return self._memo(db, memo_id) + + def _presentation(self, db, presentation_id): + row = db.execute("SELECT body FROM presentations WHERE id=?", (presentation_id,)).fetchone() + if row is None: + raise EvidenceUnavailable("informed-decision cannot produce the named presentation") + original = presentation_from(json.loads(row["body"])) + acks = db.execute("SELECT highlight_id FROM acknowledgments WHERE presentation_id=?", (presentation_id,)) + return replace(original, acked_highlight_ids=frozenset(r[0] for r in acks)) + + def presentation(self, presentation_id): + with self._connection() as db: + return self._presentation(db, presentation_id) + + def _event(self, db, commitment, content, extra=None): + # Construct the stable envelope once, in the same transaction as content + # and state. Retries send these exact bytes and the same idempotency key. + data = {**commitment.data, **(extra or {})} + at = datetime.fromisoformat(commitment.at.replace("Z", "+00:00")).astimezone(timezone.utc).isoformat(timespec="microseconds") + envelope = {"id": commitment.id, "type": commitment.event_class.value, + "source": "informed-decision", "tenant": "tenant:platform", + "subject": data.get("memo_id", "informed-decision"), + "correlation_id": data.get("approval_id") or commitment.id, + "occurred_at": at, "data": data} + encoded = dumps(envelope) + if len(encoded.encode()) > 256 * 1024: + raise StoreError("audit commitment exceeds the receiver limit") + db.execute("INSERT INTO evidence VALUES (?,?,?,?,?)", + (commitment.id, commitment.event_class.value, at, encoded, dumps(content))) + db.execute("INSERT INTO outbox(id) VALUES (?)", (commitment.id,)) + return commitment.id + + def present(self, memo_id, *, principal_sub, tenant, principal_type, awareness=None): + _text(principal_sub, "principal") + if (not isinstance(tenant, Claim) or tenant.value != "tenant:platform" + or tenant.route not in (Route.DIRECTORY, Route.REGISTRATION) + or not isinstance(principal_type, Claim)): + raise ValueError("platform tenant and principal provenance are required") + with self._transaction() as db: + memo = self._memo(db, memo_id) + presentation = render(memo, principal_sub=principal_sub, tenant=tenant, + principal_type=principal_type, awareness=awareness) + content = {"memo": json.loads(dumps(memo)), "presentation": json.loads(dumps(presentation)), + "binding_document": memo.binding_document(), "awareness_document": memo.awareness_document(awareness)} + db.execute("INSERT INTO presentations VALUES (?,?,?,?)", + (presentation.id, memo.id, memo.version, dumps(presentation))) + commitment = commit_presentation(presentation, custody="informed-decision:presentations:" + presentation.id) + self._event(db, commitment, content) + return presentation + + def acknowledge(self, presentation_id, actor: Actor, highlight_ids): + ids = frozenset(highlight_ids) + with self._transaction() as db: + p = self._presentation(db, presentation_id) + memo = self._memo(db, p.memo_id) + if actor.sub != p.principal_sub: + raise DispositionRefused("G_ACTOR", "actor did not receive this presentation") + if memo.version != p.memo_version: + raise DispositionRefused("G_PRES", "presentation is stale") + if ids - {h.id for h in memo.highlights}: + raise DispositionRefused("G_ACK", "unknown highlight") + fresh = ids - p.acked_highlight_ids + if not fresh: + return p + at = _now() + for highlight_id in sorted(fresh): + db.execute("INSERT INTO acknowledgments VALUES (?,?,?)", (p.id, highlight_id, at)) + p = replace(p, acked_highlight_ids=p.acked_highlight_ids | fresh) + c = commit_presentation(p, custody="informed-decision:presentations:" + p.id) + self._event(db, c, {"presentation": json.loads(dumps(p))}, + {"event_kind": "acknowledgment", "acknowledged_at": at}) + return p + + def record_disposition(self, presentation_id, actor, verb, *, operation_id, reasons=(), note=None): + _text(operation_id, "operation id") + if note is not None and (not isinstance(note, str) or len(note) > 8192): + raise ValueError("disposition note is too large or invalid") + if any(not isinstance(r, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", r) for r in reasons): + raise ValueError("reasons must be bounded codes, not free text") + request = dumps({"presentation": presentation_id, "actor": actor, "verb": verb, "reasons": reasons, "note": note}) + with self._transaction() as db: + existing = db.execute("SELECT request,body FROM dispositions WHERE operation_id=?", (operation_id,)).fetchone() + if existing: + if existing["request"] != request: + raise Conflict("operation id already names a different disposition") + return disposition_from(json.loads(existing["body"])) + p = self._presentation(db, presentation_id) + memo = self._memo(db, p.memo_id) + d = record(memo, p, verb, actor, reasons=tuple(reasons), note=note) + if verb in BINDING_VERBS: + if p.principal_type is None: + raise DispositionRefused("G_IDENTITY", "verified human provenance required") + assert_human_control_dischargeable(p.principal_type) + if verb is Verb.ACCEPT: + if not memo.approval_id or not memo.approval_binding_digest: + raise DispositionRefused("G_BINDING", "the native approval id and carried digest are required") + if db.execute("SELECT 1 FROM submissions WHERE approval_id=? AND subject=?", (memo.approval_id, actor.sub)).fetchone(): + raise Conflict("this approver already has an intent for the approval; recover the original") + db.execute("INSERT INTO dispositions VALUES (?,?,?,?,?)", (d.id, operation_id, request, p.id, dumps(d))) + if verb is Verb.ACCEPT: + db.execute("INSERT INTO submissions(disposition_id,approval_id,subject) VALUES (?,?,?)", (d.id, memo.approval_id, actor.sub)) + c = commit_disposition(d, custody="informed-decision:dispositions:" + d.id) + self._event(db, c, {"disposition": json.loads(dumps(d)), "presentation": json.loads(dumps(p))}, + {"submission_state": "prepared"} if verb is Verb.ACCEPT else None) + return d + + def begin_submission(self, disposition_id): + """Reserve one external attempt, AFTER the caller's fresh policy check. + + No network occurs in this store. A crash after this reservation stays + unresolved; restart never silently permits another POST. + """ + with self._transaction() as db: + row = db.execute("SELECT * FROM submissions WHERE disposition_id=?", (disposition_id,)).fetchone() + if row is None or row["state"] != "prepared": + raise Conflict("submission is absent, already attempted or unresolved") + d = disposition_from(json.loads(db.execute("SELECT body FROM dispositions WHERE id=?", (disposition_id,)).fetchone()[0])) + p = self._presentation(db, d.presentation_id) + if self._memo(db, p.memo_id).version != p.memo_version: + raise DispositionRefused("G_PRES", "presentation is stale") + attempt = str(uuid.uuid4()) + db.execute("UPDATE submissions SET state='in_flight',attempt=? WHERE disposition_id=?", (attempt, disposition_id)) + return attempt + + def finish_submission(self, disposition_id, attempt, result: EntryResult | None = None): + with self._transaction() as db: + row = db.execute("SELECT * FROM submissions WHERE disposition_id=?", (disposition_id,)).fetchone() + if row is None or row["state"] != "in_flight" or row["attempt"] != attempt: + raise Conflict("submission attempt does not match") + if result is not None and (result.approval_id != row["approval_id"] or result.subject != row["subject"]): + raise Conflict("approval entry does not match the original intent") + if result is not None: + at = datetime.fromisoformat(result.approved_at.replace("Z", "+00:00")) + if at.tzinfo is None: + raise ValueError("approval timestamp must carry its timezone") + # An unknown duplicate may predate this presentation. Never attach it + # to a new one. A lost response likewise cannot prove causation. + state = "confirmed" if result is not None and not result.duplicate else "unresolved" + db.execute("UPDATE submissions SET state=?,approved_at=? WHERE disposition_id=?", + (state, result.approved_at if state == "confirmed" else None, disposition_id)) + d = disposition_from(json.loads(db.execute("SELECT body FROM dispositions WHERE id=?", (disposition_id,)).fetchone()[0])) + c = commit_disposition(d, custody="informed-decision:dispositions:" + d.id) + extra = {"event_kind": "submission_result", "submission_state": state} + if state == "confirmed": + extra["entry_correlation"] = list(result.correlation) + self._event(db, c, {"disposition": json.loads(dumps(d)), **extra}, extra) + return state + + def submission(self, disposition_id): + with self._connection() as db: + row = db.execute("SELECT * FROM submissions WHERE disposition_id=?", (disposition_id,)).fetchone() + return dict(row) if row else None + + def retrieve_presentation(self, presentation_id): + """Internal custody retrieval, not a browser export/entitlement route.""" + with self._connection() as db: + p = self._presentation(db, presentation_id) + memo = self._memo(db, p.memo_id, p.memo_version) + docs = {} + for item in memo.packet: + row = db.execute("SELECT content FROM documents WHERE digest=?", (item.hash,)).fetchone() + if row is None or "sha256:" + hashlib.sha256(row[0]).hexdigest() != item.hash: + raise EvidenceUnavailable("informed-decision cannot produce intact packet content") + docs[item.item_id] = row[0] + return memo, p, docs + + def evidence(self): + with self._connection() as db: + return [dict(r) for r in db.execute("SELECT * FROM evidence ORDER BY rowid")] + + def record_unreachable(self, memo_id, dependency): + if dependency not in {"access-engine", "approval-engine", "key-cape", "audit-core"}: + raise ValueError("unknown dependency") + with self._transaction() as db: + memo = self._memo(db, memo_id) + stance, state = resolve(memo.binding_level.value) + c = commit_stance_application(memo_id=memo.id, memo_version=memo.version, + binding_level=memo.binding_level.value, binding_level_state=state.value, + stance=stance, dependency=dependency, custody="informed-decision:memos:" + memo.id) + return self._event(db, c, {"memo": json.loads(dumps(memo)), "dependency": dependency}) + + def retrieve_disposition(self, disposition_id): + with self._connection() as db: + row = db.execute("SELECT body FROM dispositions WHERE id=?", (disposition_id,)).fetchone() + snapshot = db.execute("SELECT content FROM evidence WHERE class=? AND " + "json_extract(envelope,'$.data.disposition_id')=? ORDER BY rowid LIMIT 1", + (EventClass.DISPOSITION.value, disposition_id)).fetchone() + if row is None or snapshot is None: + raise EvidenceUnavailable("informed-decision cannot produce the named disposition") + d = disposition_from(json.loads(row[0])) + original_presentation = presentation_from(json.loads(snapshot[0])["presentation"]) + memo, _, documents = self.retrieve_presentation(d.presentation_id) + return d, original_presentation, memo, documents + + def presentation_for_entry(self, approval_id, subject, approved_at): + with self._connection() as db: + row = db.execute("SELECT disposition_id FROM submissions WHERE state='confirmed' " + "AND approval_id=? AND subject=? AND approved_at=?", (approval_id, subject, approved_at)).fetchone() + if row is None: + raise EvidenceUnavailable("informed-decision has no confirmed presentation correlation for this entry") + return self.retrieve_disposition(row[0]) + + def outbox(self): + with self._connection() as db: + return [dict(r) for r in db.execute("SELECT * FROM outbox ORDER BY rowid")] + + def queue_heartbeats(self, now=None): + now = time.time() if now is None else now + count = 0 + with self._transaction() as db: + for event_class, interval in HEARTBEAT_CLASSES.items(): + if db.execute("SELECT 1 FROM outbox o JOIN evidence e ON e.id=o.id WHERE o.state!='delivered' " + "AND (e.class=? OR (e.class=? AND json_extract(e.envelope,'$.data.class')=?)) LIMIT 1", + (event_class, EventClass.HEARTBEAT.value, event_class)).fetchone(): + continue # Never let a nothing-to-report assertion mask undelivered evidence. + recent = db.execute("SELECT MAX(at) FROM evidence WHERE class=? OR " + "(class=? AND json_extract(envelope,'$.data.class')=?)", + (event_class, EventClass.HEARTBEAT.value, event_class)).fetchone()[0] + if recent and now - datetime.fromisoformat(recent.replace("Z", "+00:00")).timestamp() < interval: + continue + c = heartbeat(EventClass(event_class)) + c = replace(c, at=datetime.fromtimestamp(now, timezone.utc).isoformat(timespec="microseconds")) + self._event(db, c, {"class": event_class}) + count += 1 + return count + + def claim_delivery(self, now=None): + now = time.time() if now is None else now + with self._transaction() as db: + row = db.execute("SELECT o.id,e.envelope FROM outbox o JOIN evidence e ON e.id=o.id " + "WHERE o.state='pending' AND o.next_attempt<=? AND (o.lease IS NULL OR o.lease_until<=?) ORDER BY o.rowid LIMIT 1", + (now, now)).fetchone() + if row is None: + return None + lease = str(uuid.uuid4()) + db.execute("UPDATE outbox SET lease=?,lease_until=?,attempts=attempts+1 WHERE id=?", (lease, now + 30, row["id"])) + return row["id"], lease, row["envelope"] + + def finish_delivery(self, event_id, lease, *, reference=None, error=None, permanent=False, now=None): + now = time.time() if now is None else now + with self._transaction() as db: + row = db.execute("SELECT attempts FROM outbox WHERE id=? AND state='pending' AND lease=?", (event_id, lease)).fetchone() + if row is None: + raise Conflict("delivery lease does not match") + if reference: + db.execute("UPDATE outbox SET state='delivered',receiver_reference=?,lease=NULL,lease_until=NULL,last_error=NULL WHERE id=?", (reference, event_id)) + else: + if error not in {"unavailable", "unauthorized", "rejected", "conflict", "invalid_receipt"}: + raise ValueError("a bounded delivery failure code is required") + delay = min(300, 2 ** min(row["attempts"], 8)) + db.execute("UPDATE outbox SET state=?,last_error=?,next_attempt=?,lease=NULL,lease_until=NULL WHERE id=?", + ("blocked" if permanent else "pending", error, now + delay, event_id)) + + def requeue_blocked(self, event_id): + """Explicit operator retry after credential/config repair; same event bytes.""" + with self._transaction() as db: + db.execute("UPDATE outbox SET state='pending',next_attempt=0 WHERE id=? AND state='blocked'", (event_id,)) + + def counts_by_class(self, since=None, until=None): + with self._connection() as db: + if since is None and until is None: + rows = db.execute("SELECT class,COUNT(*) FROM evidence GROUP BY class") + else: + start, end = bounded_window(since, until) + rows = db.execute("SELECT class,COUNT(*) FROM evidence WHERE at>=? AND at= result[1]: + raise ValueError("window must be increasing") + return tuple(result) diff --git a/tests/test_audit_delivery.py b/tests/test_audit_delivery.py new file mode 100644 index 0000000..1e67f07 --- /dev/null +++ b/tests/test_audit_delivery.py @@ -0,0 +1,112 @@ +from datetime import datetime, timezone +import json + +import pytest + +from informed_decision.audit import AuditCoreSink, AuditDeliveryError, OutboxWorker +from informed_decision.http_transport import TransportError +from informed_decision.store import Store +from test_durable_store import storage, present + + +class Replies: + def __init__(self, *replies): + self.replies = list(replies) + self.calls = [] + + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + reply = self.replies.pop(0) + if isinstance(reply, Exception): raise reply + return reply + + +@pytest.mark.parametrize("status,name", [(202, "accepted"), (200, "duplicate")]) +def test_only_confirmed_receiver_receipt_marks_delivered(storage, status, name): + store, memo = storage + present(store, memo) + event = store.evidence()[0] + transport = Replies((status, {"status": name, "reference": "audit:" + event["id"]})) + sink = AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport) + assert OutboxWorker(store, sink).run_once()["delivered"] == 1 + row = Store(store.path).outbox()[0] + assert row["state"] == "delivered" and row["receiver_reference"] == "audit:" + event["id"] + method, url, params = transport.calls[0] + assert (method, url) == ("POST", "https://audit.test/v1/events") + assert params["headers"]["Idempotency-Key"] == event["id"] + assert params["body"] == event["envelope"].encode() + assert "synthetic-token" not in event["envelope"] + + +@pytest.mark.parametrize("reply,code,blocked", [ + ((400, {"error": "secret-sentinel"}), "rejected", True), + ((401, {}), "unauthorized", True), ((403, {}), "unauthorized", True), + ((409, {}), "conflict", True), ((503, {}), "unavailable", False), + ((500, {}), "unavailable", False), (TransportError("secret-sentinel"), "unavailable", False), + ((202, {"status": "accepted", "reference": "audit:other-event"}), "invalid_receipt", False), + ((200, {"status": "ok", "reference": "audit:fixture"}), "invalid_receipt", False), +]) +def test_refusals_preserve_pending_evidence_with_bounded_error(storage, reply, code, blocked): + store, memo = storage + present(store, memo) + before = store.evidence() + transport = Replies(reply) + worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport)) + result = worker.run_once() + assert result["blocked" if blocked else "retrying"] == 1 + row = Store(store.path).outbox()[0] + assert row["state"] == ("blocked" if blocked else "pending") + assert row["last_error"] == code and "secret-sentinel" not in str(row) + assert store.evidence() == before + assert worker.run_once() == {"delivered": 0, "retrying": 0, "blocked": 0} + + +def test_repaired_credential_requeues_original_bytes(storage): + store, memo = storage + present(store, memo) + event = store.evidence()[0] + transport = Replies((401, {}), (202, {"status": "accepted", "reference": "audit:" + event["id"]})) + tokens = iter(["old-synthetic", "new-synthetic"]) + worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: next(tokens), transport=transport)) + assert worker.run_once()["blocked"] == 1 + store.requeue_blocked(event["id"]) + assert worker.run_once()["delivered"] == 1 + assert transport.calls[0][2]["body"] == transport.calls[1][2]["body"] + assert transport.calls[1][2]["headers"]["Authorization"] == "Bearer new-synthetic" + + +def test_unavailable_policy_record_is_never_a_human_decline(storage): + store, memo = storage + store.record_unreachable(memo.id, "access-engine") + envelope = json.loads(store.evidence()[0]["envelope"]) + assert envelope["type"] == "informed-decision.stance_application" + assert envelope["data"]["stance_applied"] == "fail_closed" + assert envelope["data"]["decision_attributable"] is False + assert "verb" not in envelope["data"] + + +def test_reconciliation_reports_gap_without_claiming_completeness(storage): + store, memo = storage + present(store, memo) + since, until = "2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z" + remote = {"source": "informed-decision", "tenant": "tenant:platform", "since": since, "until": until, "counts": []} + transport = Replies((200, remote)) + worker = OutboxWorker(store, AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=transport)) + report = worker.reconcile(since, until) + assert report["count_values_match"] is False + assert report["counts"]["informed-decision.presentation"] == {"source": 1, "receiver": 0} + assert report["completeness_proven"] is report["reconstructability_proven"] is False + assert report["source_time_basis"] == "occurred_at" and report["receiver_time_basis"] == "accepted_at" + assert report["automatic_loss_finding"] is False + with pytest.raises(ValueError): worker.reconcile(None, until) + + +@pytest.mark.parametrize("change", [{"source": "other-source"}, {"tenant": "other-tenant"}, + {"since": "2019-01-01T00:00:00Z"}, {"counts": [{"class": "presentation", "count": True}]}, + {"counts": [{"class": "presentation", "count": -1}]}, + {"counts": [{"class": "presentation", "count": 1}, {"class": "presentation", "count": 1}]}]) +def test_reconciliation_rejects_wrong_scope_window_or_shape(change): + body = {"source": "informed-decision", "tenant": "tenant:platform", "since": "2020-01-01T00:00:00Z", + "until": "2100-01-01T00:00:00Z", "counts": [], **change} + sink = AuditCoreSink("https://audit.test", lambda: "synthetic-token", transport=Replies((200, body))) + with pytest.raises(AuditDeliveryError): sink.counts("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z") diff --git a/tests/test_durable_component.py b/tests/test_durable_component.py new file mode 100644 index 0000000..e8a9b54 --- /dev/null +++ b/tests/test_durable_component.py @@ -0,0 +1,143 @@ +"""Actual Approval Engine and Audit Core APIs, with synthetic identities/custody.""" + +import io +import json +import os +from pathlib import Path +import sys +from urllib.parse import urlsplit + +import pytest + +from informed_decision.audit import AuditCoreSink, AuditDeliveryError, OutboxWorker +from informed_decision.disposition import Actor, ActorKind, Verb +from informed_decision.http_transport import TransportError +from informed_decision.memo import PacketItem +from informed_decision.store import Store, Conflict +from test_approval_component import component, signing_key +from test_skeleton import make_memo +from test_durable_store import storage, present + + +@pytest.fixture +def receiver(tmp_path): + source = os.environ.get("INFD_AUDIT_CORE_SOURCE") + if not source: + pytest.skip("set INFD_AUDIT_CORE_SOURCE for actual Audit Core contract checks") + assert (Path(source) / "audit_core/ingestion.py").is_file() + sys.path.insert(0, source) + from audit_core.ingestion import IngestionApplication + from audit_core.senders import SenderIdentity, SenderRegistry + from audit_core.sqlite_backend import SQLiteAuditBackend + backend = SQLiteAuditBackend(str(tmp_path / "independent-receiver.sqlite")) + registry = SenderRegistry([SenderIdentity(name="informed-decision", tokens=("synthetic-audit-token",), + sources=frozenset({"informed-decision"}), tenants=frozenset({"tenant:platform"}), + evidence_kind="load-bearing", secret_policy="redact", may_read=False), + SenderIdentity(name="independent-fixture-reader", tokens=("synthetic-auditor-token",), + sources=frozenset({"informed-decision"}), tenants=frozenset({"tenant:platform"}), + may_read=True, may_write=False)]) + app = IngestionApplication(backend, registry) + + class Transport: + lose_reply = False + def request(self, method, url, *, headers=None, body=None): + parsed = urlsplit(url) + environ = {"PATH_INFO": parsed.path, "QUERY_STRING": parsed.query, "REQUEST_METHOD": method, + "CONTENT_LENGTH": str(len(body or b"")), "wsgi.input": io.BytesIO(body or b""), + "HTTP_AUTHORIZATION": headers.get("Authorization", ""), + "HTTP_IDEMPOTENCY_KEY": headers.get("Idempotency-Key", "")} + result = {} + out = b"".join(app(environ, lambda status, headers: result.update(status=int(status.split()[0])))) + if self.lose_reply: + self.lose_reply = False + raise TransportError("simulated lost receiver response") + return result["status"], json.loads(out) + + transport = Transport() + sink = AuditCoreSink("https://audit.test", lambda: "synthetic-audit-token", transport=transport) + return sink, transport, backend + + +def durable_setup(tmp_path, component): + client, engine, transport, session, _ = component + private = tmp_path / "holder" + private.mkdir(mode=0o700) + store = Store(private / "review.sqlite") + digest = store.put_document(b"synthetic factory review packet") + approval = client.get_approval("fixture") + memo = make_memo(approval_id="fixture", approval_binding_digest=approval["binding"]["digest"], + packet=(PacketItem("doc-1", "Factory fixture", digest),)) + store.save_memo(memo) + p = store.present(memo.id, principal_sub=session.subject, tenant=session.tenant, principal_type=session.principal_type) + actor = Actor(session.subject, ActorKind.PERSON) + store.acknowledge(p.id, actor, ["h-1"]) + return store, p, actor + + +def test_actual_entry_and_independent_receiver_survive_holder_restart(component, receiver, tmp_path): + client, engine, _, _, _ = component + sink, receiver_transport, backend = receiver + store, p, actor = durable_setup(tmp_path, component) + d = store.record_disposition(p.id, actor, Verb.ACCEPT, operation_id="synthetic-click") + attempt = store.begin_submission(d.id) # Harness only; native policy admission is not exercised. + result = client.add_entry("fixture") + assert store.finish_submission(d.id, attempt, result) == "confirmed" + worker = OutboxWorker(Store(store.path), sink) + assert worker.run_once()["delivered"] == 4 + assert worker.reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")["count_values_match"] + recovered = Store(store.path).presentation_for_entry(*result.correlation) + assert recovered[0].id == d.id and recovered[1].id == p.id + assert recovered[3] == {"doc-1": b"synthetic factory review packet"} + assert len(engine.get("fixture").entries) == 1 + # This sender can count its stream without gaining archive read privileges. + with pytest.raises(AuditDeliveryError, match="unauthorized"): + sink._request("GET", "/v1/events") + event = store.evidence()[0] + status, body = receiver_transport.request("GET", "https://audit.test/v1/events/" + event["id"], + headers={"Authorization": "Bearer synthetic-auditor-token"}) + assert status == 200 and body["details"]["data"]["view_hash"] == p.view_hash + assert "synthetic factory review packet" not in json.dumps(body) + assert not any("synthetic factory review packet" in row["envelope"] for row in store.evidence()) + + +def test_lost_receiver_reply_reuses_exact_event_and_deduplicates(component, receiver, tmp_path): + sink, transport, backend = receiver + store, _, _ = durable_setup(tmp_path, component) + first = store.claim_delivery(now=10) + transport.lose_reply = True + with pytest.raises(AuditDeliveryError, match="unavailable"): sink.deliver(first[0], first[2]) + # Simulate crash before marking local delivery; its lease expires after restart. + replay = Store(store.path).claim_delivery(now=41) + assert replay[0] == first[0] and replay[2] == first[2] + reference = sink.deliver(replay[0], replay[2]) + store.finish_delivery(replay[0], replay[1], reference=reference) + OutboxWorker(store, sink).run_once() + report = OutboxWorker(store, sink).reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z") + assert report["count_values_match"] and report["counts"]["informed-decision.presentation"] == {"source": 2, "receiver": 2} + + +def test_lost_engine_reply_does_not_rebind_or_reassign_presentation(component, tmp_path): + client, engine, transport, _, _ = component + store, p, actor = durable_setup(tmp_path, component) + d = store.record_disposition(p.id, actor, Verb.ACCEPT, operation_id="synthetic-click") + attempt = store.begin_submission(d.id) + client.add_entry("fixture") # Real API commits; the caller's response is deliberately discarded. + assert store.finish_submission(d.id, attempt) == "unresolved" + calls = len(transport.calls) + with pytest.raises(Conflict): Store(store.path).begin_submission(d.id) + assert len(transport.calls) == calls and len(engine.get("fixture").entries) == 1 + assert store.submission(d.id)["approved_at"] is None + + +def test_delayed_acceptance_has_a_different_window_without_being_lost(storage, receiver, monkeypatch): + from informed_decision import evidence, presentation + monkeypatch.setattr(evidence, "_now", lambda: "2020-06-01T12:00:00Z") + monkeypatch.setattr(presentation, "_now", lambda: "2020-06-01T12:00:00Z") + store, memo = storage + present(store, memo) + sink, _, backend = receiver + worker = OutboxWorker(store, sink) + assert worker.run_once()["delivered"] == 1 + historical = worker.reconcile("2020-01-01T00:00:00Z", "2021-01-01T00:00:00Z") + assert not historical["count_values_match"] and not historical["automatic_loss_finding"] + assert worker.reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")["count_values_match"] diff --git a/tests/test_durable_store.py b/tests/test_durable_store.py new file mode 100644 index 0000000..a2d2ca1 --- /dev/null +++ b/tests/test_durable_store.py @@ -0,0 +1,318 @@ +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +import json +import os +import sqlite3 +import subprocess +import sys +import time + +import pytest + +from informed_decision.approval_client import EntryResult +from informed_decision.disposition import Actor, ActorKind, DispositionRefused, Verb, record +from informed_decision.evidence import EventClass +from informed_decision.memo import Highlight, PacketItem +from informed_decision.provenance import Claim, HumanControlNotDischargeable, Route +from informed_decision.store import Conflict, EvidenceUnavailable, Store, StoreError +from test_skeleton import HUMAN, make_memo + + +@pytest.fixture +def storage(tmp_path): + private = tmp_path / "private" + private.mkdir(mode=0o700) + store = Store(private / "evidence.sqlite") + digest = store.put_document(b"fixture packet material") + memo = make_memo(packet=(PacketItem("doc-1", "Fixture document", digest),), + brief="private brief sentinel", approval_binding_digest="sha256:" + "a" * 64) + store.save_memo(memo) + return store, memo + + +def present(store, memo, *, route=Route.AUTHENTICATION): + return store.present(memo.id, principal_sub=HUMAN.sub, + tenant=Claim("tenant:platform", Route.REGISTRATION), principal_type=Claim("human", route)) + + +def prepare(storage, operation_id="click-1"): + store, memo = storage + p = present(store, memo) + store.acknowledge(p.id, HUMAN, ["h-1"]) + d = store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id=operation_id) + return p, d + + +def abort_outbox(store): + with sqlite3.connect(store.path) as db: + db.execute("CREATE TRIGGER injected_failure BEFORE INSERT ON outbox BEGIN SELECT RAISE(ABORT,'injected disk failure'); END") + + +def test_packet_and_presentation_content_survive_reopen(storage): + store, memo = storage + p = present(store, memo) + reopened = Store(store.path) + saved, original, documents = reopened.retrieve_presentation(p.id) + assert saved == memo and original == p + assert documents == {"doc-1": b"fixture packet material"} + row = reopened.evidence()[0] + assert json.loads(row["content"])["memo"]["brief"] == "private brief sentinel" + assert "private brief sentinel" not in row["envelope"] + assert "fixture packet material" not in row["envelope"] + assert json.loads(row["envelope"])["data"]["content_exists"] is True + assert reopened.outbox()[0]["state"] == "pending" + + +def test_missing_packet_content_prevents_false_custody_assertion(storage): + store, memo = storage + altered = replace(memo, id="missing", packet=(PacketItem("doc-1", "Missing", "sha256:" + "b" * 64),)) + with pytest.raises(EvidenceUnavailable): store.save_memo(altered) + with pytest.raises(EvidenceUnavailable): store.memo("missing") + assert store.evidence() == [] + + +def test_versions_are_immutable_and_old_presentation_cannot_bind(storage): + store, memo = storage + p = present(store, memo) + store.save_memo(memo) # Exact import is harmless. + with pytest.raises(Conflict): store.save_memo(replace(memo, brief="silently changed")) + store.save_memo(memo.next_version(brief="revised")) + with pytest.raises(DispositionRefused, match="G_PRES"): + store.acknowledge(p.id, HUMAN, ["h-1"]) + with pytest.raises(DispositionRefused, match="G_PRES"): + store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="stale-click") + saved, _, _ = store.retrieve_presentation(p.id) + assert saved.brief == memo.brief # Retrieval uses the historic version. + + +@pytest.mark.parametrize("verb", [Verb.ACCEPT, Verb.DECLINE, Verb.ACKNOWLEDGE, Verb.RETURN, Verb.DISCUSS]) +def test_actor_cannot_use_someone_elses_presentation(storage, verb): + store, memo = storage + p = present(store, memo) + other = Actor("other-person", ActorKind.PERSON) + with pytest.raises(DispositionRefused, match="G_ACTOR"): + record(memo, p.with_ack("h-1"), verb, other, reasons=("wrong_scope",)) + before = store.evidence() + with pytest.raises(DispositionRefused, match="G_ACTOR"): + store.record_disposition(p.id, other, verb, operation_id="stolen-click", reasons=("wrong_scope",)) + assert store.evidence() == before + + +def test_acknowledgments_are_explicit_append_only_and_idempotent(storage): + store, memo = storage + p = present(store, memo) + with pytest.raises(DispositionRefused, match="G_ACK"): + store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="before-ack") + with pytest.raises(DispositionRefused, match="G_ACK"): + store.acknowledge(p.id, HUMAN, ["does-not-exist"]) + with pytest.raises(DispositionRefused, match="G_ACTOR"): + store.acknowledge(p.id, Actor("other", ActorKind.PERSON), ["h-1"]) + after = store.acknowledge(p.id, HUMAN, ["h-1"]) + assert after.view_hash == p.view_hash and after.acked_highlight_ids == {"h-1"} + assert store.acknowledge(p.id, HUMAN, ["h-1"]) == after + assert len(store.evidence()) == 2 + with sqlite3.connect(store.path) as db: + original = json.loads(db.execute("SELECT body FROM presentations WHERE id=?", (p.id,)).fetchone()[0]) + assert original["acked_highlight_ids"] == [] + + +def test_registration_human_assertion_cannot_bind(storage): + store, memo = storage + p = present(store, memo, route=Route.REGISTRATION) + store.acknowledge(p.id, HUMAN, ["h-1"]) + with pytest.raises(HumanControlNotDischargeable): + store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="unverified") + + +def test_return_discuss_and_decline_remain_distinct_without_engine_jobs(storage): + store, memo = storage + p = present(store, memo) + with pytest.raises(DispositionRefused, match="G_REASONS"): + store.record_disposition(p.id, HUMAN, Verb.RETURN, operation_id="empty-return") + returned = store.record_disposition(p.id, HUMAN, Verb.RETURN, operation_id="return", + reasons=("wrong_scope",), note="private note sentinel") + discussed = store.record_disposition(p.id, HUMAN, Verb.DISCUSS, operation_id="discuss", note="question") + store.acknowledge(p.id, HUMAN, ["h-1"]) + declined = store.record_disposition(p.id, HUMAN, Verb.DECLINE, operation_id="decline") + assert [returned.verb, discussed.verb, declined.verb] == [Verb.RETURN, Verb.DISCUSS, Verb.DECLINE] + assert all(store.submission(d.id) is None for d in [returned, discussed, declined]) + assert store.retrieve_disposition(returned.id)[0].note == "private note sentinel" + assert all("private note sentinel" not in r["envelope"] for r in store.evidence()) + + +def test_click_idempotency_and_parallel_submission_reservation(storage): + store, memo = storage + p = present(store, memo) + store.acknowledge(p.id, HUMAN, ["h-1"]) + def click(_): + return Store(store.path).record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="same-click") + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(click, range(2))) + assert results[0] == results[1] + assert len(store.evidence()) == 3 + with pytest.raises(Conflict): + store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="same-click", note="changed") + def reserve(_): + try: return Store(store.path).begin_submission(results[0].id) + except Conflict: return None + with ThreadPoolExecutor(max_workers=2) as pool: + attempts = list(pool.map(reserve, range(2))) + assert sum(a is not None for a in attempts) == 1 + + +@pytest.mark.parametrize("operation", ["present", "acknowledge", "disposition", "completion"]) +def test_state_and_outbox_rollback_together(storage, operation): + store, memo = storage + p = present(store, memo) + if operation in ("disposition", "completion"): + store.acknowledge(p.id, HUMAN, ["h-1"]) + if operation == "completion": + d = store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="intent") + attempt = store.begin_submission(d.id) + before = store.evidence() + abort_outbox(store) + with pytest.raises(sqlite3.IntegrityError, match="injected"): + if operation == "present": present(store, memo) + elif operation == "acknowledge": store.acknowledge(p.id, HUMAN, ["h-1"]) + elif operation == "disposition": store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="intent") + else: store.finish_submission(d.id, attempt, EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved")) + assert store.evidence() == before + assert len(store.outbox()) == len(before) + if operation == "acknowledge": assert not store.presentation(p.id).acked_highlight_ids + if operation == "completion": assert store.submission(d.id)["state"] == "in_flight" + with sqlite3.connect(store.path) as db: + assert db.execute("SELECT COUNT(*) FROM presentations").fetchone()[0] == 1 + assert db.execute("SELECT COUNT(*) FROM dispositions").fetchone()[0] == (1 if operation == "completion" else 0) + + +def test_process_death_before_commit_loses_neither_half(storage): + store, memo = storage + code = '''import os, sys +from informed_decision.store import Store +from informed_decision.provenance import Claim, Route +s=Store(sys.argv[1]); original=s._event +def die(*args, **kwargs): + original(*args, **kwargs) + os._exit(91) +s._event=die +s.present("memo-1", principal_sub="bernd", tenant=Claim("tenant:platform", Route.REGISTRATION), principal_type=Claim("human", Route.AUTHENTICATION)) +''' + result = subprocess.run([sys.executable, "-c", code, str(store.path)]) + assert result.returncode == 91 + reopened = Store(store.path) + assert reopened.memo(memo.id) == memo and reopened.evidence() == [] and reopened.outbox() == [] + with sqlite3.connect(store.path) as db: + assert db.execute("SELECT COUNT(*) FROM presentations").fetchone()[0] == 0 + + +def test_confirmation_points_to_original_ack_snapshot(storage): + store, memo = storage + extra = Highlight("h-2", "doc-1", "optional") + store.save_memo(memo.next_version(highlights=(*memo.highlights, extra))) + p, d = prepare((store, store.memo(memo.id))) + attempt = store.begin_submission(d.id) + result = EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved") + assert store.finish_submission(d.id, attempt, result) == "confirmed" + store.acknowledge(p.id, HUMAN, ["h-2"]) + recovered, snapshot, _, _ = Store(store.path).presentation_for_entry(*result.correlation) + assert recovered.id == d.id and snapshot.acked_highlight_ids == {"h-1"} + assert store.presentation(p.id).acked_highlight_ids == {"h-1", "h-2"} + p2 = present(store, store.memo(memo.id)) + store.acknowledge(p2.id, HUMAN, ["h-1"]) + with pytest.raises(Conflict): store.record_disposition(p2.id, HUMAN, Verb.ACCEPT, operation_id="new-click") + + +@pytest.mark.parametrize("kind", ["lost-reply", "old-duplicate", "crash-after-reserve"]) +def test_uncertain_entry_never_gets_a_new_presentation_or_automatic_retry(storage, kind): + store, memo = storage + p, d = prepare(storage) + attempt = store.begin_submission(d.id) + if kind != "crash-after-reserve": + result = EntryResult("appr-1", HUMAN.sub, "2020-01-01T00:00:00Z", "approved", True) if kind == "old-duplicate" else None + assert store.finish_submission(d.id, attempt, result) == "unresolved" + reopened = Store(store.path) + with pytest.raises(Conflict): reopened.begin_submission(d.id) + with pytest.raises(EvidenceUnavailable): reopened.presentation_for_entry("appr-1", HUMAN.sub, "2020-01-01T00:00:00Z") + assert reopened.submission(d.id)["approved_at"] is None + assert not any(json.loads(r["envelope"])["data"].get("entry_correlation") for r in reopened.evidence()) + + +def test_updated_memo_stops_already_prepared_submission(storage): + store, memo = storage + _, d = prepare(storage) + store.save_memo(memo.next_version(brief="new question")) + with pytest.raises(DispositionRefused, match="G_PRES"): store.begin_submission(d.id) + assert store.submission(d.id)["state"] == "prepared" + + +def test_revision_cannot_race_an_in_flight_or_uncertain_entry(storage): + store, memo = storage + _, d = prepare(storage) + attempt = store.begin_submission(d.id) + with pytest.raises(Conflict): store.save_memo(memo.next_version(brief="racing edit")) + store.finish_submission(d.id, attempt) + with pytest.raises(Conflict): store.save_memo(memo.next_version(brief="unresolved edit")) + + +def test_wrong_entry_or_attempt_cannot_confirm(storage): + store, _ = storage + _, d = prepare(storage) + attempt = store.begin_submission(d.id) + with pytest.raises(Conflict): store.finish_submission(d.id, "wrong-attempt") + with pytest.raises(Conflict): + store.finish_submission(d.id, attempt, EntryResult("other-approval", HUMAN.sub, "2026-09-10T20:00:00Z", "approved")) + assert store.submission(d.id)["state"] == "in_flight" + + +def test_evidence_is_append_only_and_backup_preserves_pending_and_confirmed(storage): + store, _ = storage + p, d = prepare(storage) + attempt = store.begin_submission(d.id) + result = EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved") + store.finish_submission(d.id, attempt, result) + with sqlite3.connect(store.path) as db: + for statement in ["DELETE FROM evidence", "UPDATE presentations SET body='{}'", "DELETE FROM documents", "DELETE FROM dispositions"]: + with pytest.raises(sqlite3.IntegrityError, match="immutable"): db.execute(statement) + destination = store.path.parent / "backup.sqlite" + store.backup(destination) + backup = Store(destination) + assert backup.outbox() == store.outbox() + assert backup.presentation_for_entry(*result.correlation)[0].id == d.id + with pytest.raises(Conflict): store.backup(destination) + + +def test_store_refuses_unsafe_paths_and_unknown_schema(tmp_path): + private = tmp_path / "private" + private.mkdir(mode=0o755) + with pytest.raises(StoreError): Store(private / "db") + private.chmod(0o700) + db = private / "db" + db.touch(mode=0o644) + with pytest.raises(StoreError): Store(db) + db.chmod(0o600) + with sqlite3.connect(db) as conn: conn.execute("PRAGMA user_version=99") + with pytest.raises(StoreError, match="schema"): Store(db) + link = private / "linked" + link.symlink_to(db) + with pytest.raises(StoreError): Store(link) + + +def test_expired_delivery_lease_replays_same_id_and_bytes(storage): + store, memo = storage + present(store, memo) + first = store.claim_delivery(now=10) + assert store.claim_delivery(now=11) is None + again = Store(store.path).claim_delivery(now=41) + assert first[0] == again[0] and first[2] == again[2] and first[1] != again[1] + with pytest.raises(Conflict): store.finish_delivery(first[0], first[1], reference="old-receipt") + + +def test_heartbeats_do_not_mask_backlog_and_are_per_class(storage): + store, memo = storage + present(store, memo) + future = time.time() + 90000 + assert store.queue_heartbeats(now=future) == 2 + assert store.queue_heartbeats(now=future) == 0 + beats = [json.loads(r["envelope"]) for r in store.evidence() if r["class"] == EventClass.HEARTBEAT.value] + assert {b["data"]["class"] for b in beats} == {EventClass.DISPOSITION.value, EventClass.STANCE_APPLICATION.value} + assert all(b["data"]["assertion"] == "nothing-to-report" for b in beats) diff --git a/tests/test_skeleton.py b/tests/test_skeleton.py index 2d50dd0..6ea7113 100644 --- a/tests/test_skeleton.py +++ b/tests/test_skeleton.py @@ -181,7 +181,7 @@ def test_agent_cannot_perform_a_binding_verb(verb): def test_agent_may_still_comment(): memo = make_memo() - pres = render(memo, principal_sub="bot") + pres = render(memo, principal_sub=AGENT.sub) assert record(memo, pres, Verb.COMMENT, AGENT).verb is Verb.COMMENT @@ -434,7 +434,7 @@ def test_drain_failure_leaves_the_record_pending_never_lost(): def test_drain_delivers_and_counts_reconcile_per_class(): outbox = Outbox() memo = make_memo() - pres = render(memo, principal_sub="b").with_ack("h-1") + pres = render(memo, principal_sub=HUMAN.sub).with_ack("h-1") outbox.append(commit_presentation(pres, custody=CUSTODY)) outbox.append(commit_disposition(record(memo, pres, Verb.ACCEPT, HUMAN), custody=CUSTODY)) sent: list[dict] = [] diff --git a/workplans/INFD-WP-0001-founding-specs-and-approver-ui-ownership.md b/workplans/INFD-WP-0001-founding-specs-and-approver-ui-ownership.md index 9c95774..785b3f6 100644 --- a/workplans/INFD-WP-0001-founding-specs-and-approver-ui-ownership.md +++ b/workplans/INFD-WP-0001-founding-specs-and-approver-ui-ownership.md @@ -526,6 +526,48 @@ before a UI retry. `/readyz` deliberately remains 503 and browser entry routes are absent until this protected path is wired. The task remains `progress`. See `docs/browser-authentication.md`. +2026-09-10 — **durable evidence and receiver integration implemented.** +`store.py` / `records.py` persist packet bytes, immutable memo versions and +presentations, explicit append-only acknowledgments, dispositions, submission +correlation and a same-transaction evidence outbox in private SQLite storage. +The new G_ACTOR guard refuses use of another person's presentation. The two +older tests with mismatched placeholder actors were corrected; a dedicated +negative regression now pins the real guard. Return/discuss remain local acts. + +Operation ids and atomic reservation prevent double submission. Lost engine +responses and unknown duplicates remain unresolved and cannot be attached to a +new presentation or silently retried. A confirmed correlation retrieves the +original acknowledgment snapshot; later acks do not strengthen earlier evidence. +Revisions cannot race in-flight/unresolved submissions. Backup/restore preserves +the evidence and pending delivery state. + +`audit.py` uses the current Audit Core ingestion contract, stable event ids and +bytes, scoped metadata-only envelopes, receiver references, retry/backoff and +visible blocked records. Writer-only reconciliation and a separate auditor read +were exercised against the actual receiver. Heartbeats are per class and do not +mask undelivered evidence. The receiver counts by accepted_at, while the source +counts by occurred_at: the report now carries both, and delayed acceptance is +not automatically called a loss. Exact window shape came from the real API, +not from a local fixture assumption. + +258 tests pass (52 added), including process death before commit, injected +outbox failures, concurrent clicks/reservations, restore, lost engine response +and receiver deduplication. Actual Approval Engine and Audit Core APIs are +exercised with synthetic identities and development custody. Evidence: +`docs/evidence/2026-09-10-durable-review-evidence.json`. No native policy, human +login, production audit custody, deployed binding UI or factory run is claimed. + +**Next within T08:** define/admit the exact PDP read/bind request and caller, +then connect protected review/ack/accept/return/discuss routes to this store and +its original-entry recovery rules. Persist the obtained policy observation and +retain decision_attributable=false while its upstream gap remains. No Informed +Decision package/registration was found in Flex Auth's checked examples, +registry or docs at `88b3543`; a local allow rule is not a replacement. +Schedule and admit audit draining/heartbeats, account for acceptance-time delay +in reconciliation, supply native custody and registered human/deployed-engine +proof. These remain live work in this task; `/readyz` stays 503 and browser bind +routes remain absent. Details: `docs/durable-review-evidence.md`. + ## Known risks - **T02 is a hard gate.** Writing the blueprint before the layer ruling risks