diff --git a/SCOPE.md b/SCOPE.md index d7c4d55..4096c38 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -234,10 +234,9 @@ Stated here because a scope file that only lists capabilities overstates them. (MFA backdated on reuse; a downstream fetch missing split-horizon headers) are fixed and deployed. The original ask — Authelia `prompt=login`/`max_age` support, so a step-up can be requested rather than merely checked after the - fact, plus an actionable freshness error in the browser instead of the flat - "Review unavailable" — remains open. Authelia 4.38 rejected that parameter - once already (`cbea539`); retrying it is gated on the Authelia upgrade this - repository does not own. + fact, is complete in INFD-WP-0006, with live fresh-login/p01 acceptance. + An actionable freshness error in the browser instead of the flat + "Review unavailable" remains open in the existing intake record. - **`INFD-IN-0007`** — see Layer. No deadline. - **A reusable account/session-status component**, requested of `user-engine` as `USER-IN-0002` on 2026-09-27: identity, MFA-freshness, logout, to replace @@ -267,7 +266,11 @@ terminal dispositions atomically queue optional hub resolution; a leased worker retries delivery with stable ids and leaves disabled/unconfigured adapters pending. An uncertain approval create never reposts automatically; retained receipts can resume local loading. Tests cover schema-1/2 upgrade, rollback, replay/restart, -revocation and write-back. This is offline implementation, not a deployed intake -listener. Owner contracts, workload custody and runtime composition are tracked +revocation and write-back. T07 also supplies opt-in scheduled delivery, matching +hub message receipts and stable write keys, and immutable posed audit evidence +with receipts, readiness, heartbeat and reconciliation. Actual Audit Core +component tests exercise lost-reply deduplication and counts. This is local +implementation, not a deployed intake listener. Attribute contracts, forwarding, +workload custody, private listener composition and live acceptance are tracked as blocked tasks in INFD-WP-0005, with acceptance requirements in `docs/intake-dependency-requirements.md`. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 04739b8..a3f2824 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -24,7 +24,7 @@ | task | INFD-WP-0005-T04 | done | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | | task | INFD-WP-0005-T05 | wait | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | | task | INFD-WP-0005-T06 | wait | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | -| task | INFD-WP-0005-T07 | wait | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | +| task | INFD-WP-0005-T07 | done | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | | task | INFD-WP-0005-T08 | wait | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | | task | INFD-WP-0005-T09 | done | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | | task | INFD-WP-0005-T10 | wait | — | workplans/INFD-WP-0005-frictionless-decision-intake.md | diff --git a/docs/intake-dependency-requirements.md b/docs/intake-dependency-requirements.md index b99dc18..a6ec9c9 100644 --- a/docs/intake-dependency-requirements.md +++ b/docs/intake-dependency-requirements.md @@ -10,8 +10,10 @@ a bounded transport. Store schema 3 migrates existing schema 1/2 databases and adds the submission ledger and integration outbox. Packet bytes, pinned memo, ready ledger and posed/notification jobs commit in one transaction. Terminal human dispositions queue hub resolution in their existing evidence transaction. -`IntakeDeliveryWorker` dispatches injected audit, notification and hub transports -with leases, stable ids, bounded retries and a disabled-by-default write-back. +`IntakeDeliveryWorker` and the optional scheduled runtime dispatch audit, +notification and hub transports with leases, stable ids, bounded batches and a +disabled-by-default write-back. Posed events use the ordinary immutable audit +outbox, receipts, per-class reconciliation and readiness checks. No listener is composed into the deployed runtime yet; shipped registries are empty, so no caller is admitted. No live deployment occurred. @@ -75,31 +77,50 @@ condition on offline tests or initial dev rollout. Existing request: `23b670ac-809e-49bf-9233-e0b62cf0a039`. -Register `informed-decision.memo.posed` for source informed-decision and tenant +Completed locally by AUDIT-WP-0011 (`audit_core/ingestion.py` and sender registry). +The registered contract accepts `informed-decision.memo.posed` for source informed-decision and tenant platform; accept the locally retained event envelope containing memo/approval ids, carried binding digest, authenticated forwarding submitter, memo class and -submission digest. No packet, brief, or secret content. Confirm sender custody, -Idempotency-Key receipts and heartbeat/reconciliation requirements for the new -class. Provide positive/duplicate/refusal fixtures. Only then connect posed -jobs to AuditCoreSink and include the class in runtime readiness/reconciliation. +submission digest. No packet, brief, or secret content. T07 now connects posed +jobs to AuditCoreSink, preserving the original event id on recovery and duplicate +receipts. The new class participates in readiness, reconciliation and optional +86400-second heartbeat. Actual receiver tests cover acceptance, lost replies and +duplicate receipts. Sender custody and deployed receiver/version acceptance +remain part of T10; owner-local registration is not deployment evidence. ## state-hub — T07, T08 Existing request: `e30dea31-90e3-4fc0-954a-ed8412b52aca`. Deliver forwarding with the projected intake-audience SA token, bounded envelope, -stable idempotency key, durable retry and receipt retention. Identify the actual -SA to review into the submitter registry. Forwarding proves the hub identity; -drafted_by remains asserted and must be visibly described that way. Add a dev -fixture for pose/replay/conflict and a deduplicated notification contract using -our stable delivery id. Notifications carry only id/question/class/submitter; +stable idempotency key, durable retry and receipt retention. Owner commit +`49eeddb` provides a dev-only forwarding fixture and retained notification +receipts. Its `_bounded_tree` accepts only int/float/null scalar leaves and +rejects governed memo booleans (`requires_new_bind`, `required_ack`, `sealed`) +with `invalid_type`; a cross-repo negative fixture reproduces this using p01. +Accept boolean leaves without removing or coercing shared-schema binding fields, +then provide pose/replay/conflict and browser/PDP acceptance evidence. The +fixture is unscheduled and does not choose/admit an account. Identify the actual +SA and supply runtime forwarding to review into the submitter registry. +Forwarding proves the hub identity; drafted_by remains visibly asserted. +The dedicated endpoint `POST /informed-decision/notifications` accepts only +id/question/class/submitter using our stable delivery id as Idempotency-Key. +Its receipt and founder message commit in one transaction; the receipt is +retained independently of generic write-key TTL. `HubDelivery` checks the +delivery id, message UUID, all echoed metadata, accepted status and +signed-in-overview reference; queued +relay acknowledgements remain pending. Notifications carry only id/question/class/submitter; the signed-in overview is the authoritative review entry point. Confirm resolve semantics: terminal human accept/decline only; return/discuss and resolves: never do not resolve. A 409 must allow reading an existing resolution to correlate it. Delivery failure retains local evidence and retries; -other-memo conflicts block for review. Bind the hub transports and schedule -IntakeDeliveryWorker in the runtime after these contracts are accepted. Keep +other-memo conflicts block for review. T07 binds hub transports and schedules +IntakeDeliveryWorker in the runtime with explicit `intake_delivery` configuration. +Generic resolution idempotency retention defaults to 14 days. A resolved +decision refuses a repeated resolve and can be correlated by read after a 409; +unreadable conflicts remain retryable. Queued or mismatched success responses +are not delivery receipts. Keep INFD_HUB_WRITEBACK unset until deployment approval and end-to-end evidence. ## gate-house and founder — T09, T10 @@ -114,13 +135,33 @@ retirement of exact-record admission after its memos become terminal. T10 resumes after T05–T08 acceptance and founder approval for the concrete rollout. Record dev positive/negative traces first, then one live human answer with audit and write-back receipts. The provisioning sitting's human binding -remains blocked in INFD-WP-ADHOC-2026-10-03-T02. +was completed in INFD-WP-ADHOC-2026-10-03-T02 (p01 accepted with native engine +and audit receipts); it does not approve a new intake rollout. + +## Opt-in delivery configuration + +The private review runtime v1 file may include: + +```json +"intake_delivery": { + "hub_origin": "http://state-hub.state-hub.svc.cluster.local:8000" +} +``` + +Omitting this field preserves the existing runtime: no intake deliveries or new +posed heartbeat are scheduled. Configuration uses the existing audit custody; +it neither enables a listener nor grants submitter access. `INFD_HUB_WRITEBACK` +still defaults off. Terminal resolution failures retain the human's existing +evidence and pending delivery; they never repeat the human act. ## Verification -`uv run --extra dev python -m pytest -q`: 437 passed, 67 skipped (optional -component fixtures are not enabled in this run). After the final receipt-recovery -concurrency guard and nonempty-entry response refusal, the affected intake, -runtime migration, durable store and write-back suites: 63 passed, 1 skipped. -`git diff --check` passed. State Hub consistency reconciled the blocked plan, -completed T04/T09 and waiting T07; no registrar fallback was required. +Full suite with `INFD_AUDIT_CORE_SOURCE=/home/worsch/audit-core` and +`INFD_STATE_HUB_SOURCE=/home/worsch/state-hub`: 452 passed, 65 skipped +(optional Approval Engine/Flex Auth component fixtures are not enabled). +Checks include posed receiver acceptance/lost-reply replay, immutable local +evidence/counts, readiness refusal, notification contract/receipt matching, +resolution retry on unreadable conflicts, explicit runtime configuration and +the State Hub forwarder's governed-boolean refusal. `git diff --check` passed. +No new tasks/workplans or live changes were created. T07 is complete locally; +T05/T06/T08/T10 remain waiting and INFD-WP-0005 remains blocked. diff --git a/docs/specs/FrictionlessDecisionIntake.md b/docs/specs/FrictionlessDecisionIntake.md index c020172..30d8821 100644 --- a/docs/specs/FrictionlessDecisionIntake.md +++ b/docs/specs/FrictionlessDecisionIntake.md @@ -256,9 +256,11 @@ intake fact from a signed intake receipt. Then the PEP carries only ids. (`deploy/review-groups.json`: review group → named recipient KeyCape `sub`), never from memo content. Draft principal `pending-human-session` stays in the unsigned record; the live bind still substitutes the authenticated subject. -- Notification: a State Hub message to the founder's inbox (`to_agent` - configured, default `bernd`) with memo id, question, class, submitter and the - review link. The signed-in overview (INFD-WP-0003) lists it as pending. +- Notification: the retained State Hub notification endpoint creates a message + to the founder's inbox (`bernd`) with memo id, question, class and submitter. + The owner contract prohibits URLs and names the signed-in overview as the + authoritative review entry point, replacing the initially proposed review + link. The overview (INFD-WP-0003) lists admissible memos as pending. No packet content, no highlights, no secrets in the notification. Email/push is a later channel owned elsewhere. - audit-core receives `informed-decision.memo.posed` (submitter, class, @@ -369,7 +371,9 @@ R2/R5 re-arming includes authenticated drafting identity and independent grant review. The unsigned PDP attribution gap remains explicitly declared. The offline service and its durable integration worker are implemented; shipped -registries are empty and transports are injected. Runtime composition and live +registries are empty. Optional review-runtime delivery configuration now binds +State Hub transports and schedules posed audit evidence, notification and +disabled-by-default resolution. Private intake listener composition and live acceptance remain blocked on the owner requirements in `docs/intake-dependency-requirements.md`. These requirements do not arm a live control or grant custody/deployment approval. diff --git a/informed_decision/evidence.py b/informed_decision/evidence.py index 40eccf3..c56d479 100644 --- a/informed_decision/evidence.py +++ b/informed_decision/evidence.py @@ -38,6 +38,7 @@ class EventClass(str, Enum): PRESENTATION = "informed-decision.presentation" DISPOSITION = "informed-decision.disposition" STANCE_APPLICATION = "informed-decision.stance_application" + MEMO_POSED = "informed-decision.memo.posed" HEARTBEAT = "audit-core.heartbeat" diff --git a/informed_decision/hub_writeback.py b/informed_decision/hub_writeback.py index 1a6347b..c7fdf11 100644 --- a/informed_decision/hub_writeback.py +++ b/informed_decision/hub_writeback.py @@ -90,9 +90,11 @@ def write_back(answer: Answer, serves: Mapping | None, *, base_url: str, if status == 409: # Already resolved: ours if the hub rationale names this memo. try: - _, current = transport("GET", f"{base_url.rstrip('/')}/decisions/{decision_id}", {}) + read_status, current = transport("GET", f"{base_url.rstrip('/')}/decisions/{decision_id}", {}) except OSError as exc: return WritebackResult("failed", decision_id, payload, type(exc).__name__) + if read_status != 200 or not isinstance(current, dict) or not isinstance(current.get('rationale'), str): + return WritebackResult("failed", decision_id, payload, "resolution_unreadable") if f"memo {answer.memo_id} " in (current.get("rationale") or ""): return WritebackResult("already_resolved", decision_id, payload) return WritebackResult("conflict", decision_id, payload, json.dumps(body)[:200]) diff --git a/informed_decision/intake_delivery.py b/informed_decision/intake_delivery.py index 14b63dc..7265236 100644 --- a/informed_decision/intake_delivery.py +++ b/informed_decision/intake_delivery.py @@ -9,6 +9,7 @@ import time import uuid from .hub_writeback import Answer, enabled, write_back +from .audit import OutboxWorker class IntakeDeliveryWorker: @@ -21,16 +22,18 @@ class IntakeDeliveryWorker: def run_once(self, limit=10): if type(limit) is not int or not 1 <= limit <= 100: raise ValueError('delivery limit must be 1..100') - kinds = [] + delivered = 0 if self.audit: - kinds.append('memo-posed') + self.store.stage_intake_audit() + OutboxWorker(self.store, self.audit).run_once(limit=limit) + delivered += self.store.settle_intake_audit() + kinds = [] if self.notify: kinds.append('notification') if self.hub_transport and enabled(self.env): kinds.append('hub-resolution') if not kinds: - return 0 - delivered = 0 + return delivered for _ in range(limit): now, lease = self.clock(), str(uuid.uuid4()) with self.store._transaction() as db: @@ -45,10 +48,7 @@ class IntakeDeliveryWorker: state = 'pending' try: payload = json.loads(row['payload']) - if row['kind'] == 'memo-posed': - self.audit.deliver(row['id'], row['payload']) - state = 'delivered' - elif row['kind'] == 'notification': + if row['kind'] == 'notification': # notify(id, metadata) returns only after a deduplicated receiver receipt. if self.notify(row['id'], payload) is True: state = 'delivered' diff --git a/informed_decision/intake_hub.py b/informed_decision/intake_hub.py new file mode 100644 index 0000000..8b8152f --- /dev/null +++ b/informed_decision/intake_hub.py @@ -0,0 +1,57 @@ +"""Bounded delivery to State Hub's retained memo-notification receipts.""" +import hashlib +import json +import re +import uuid + +from .http_transport import JSONTransport, fixed_origin + + +class HubDelivery: + def __init__(self, origin, *, transport=None): + self.origin = fixed_origin(origin, allow_internal_http=True) + self.transport = transport or JSONTransport(allow_internal_http=True) + + def _request(self, method, url, payload, key): + body = json.dumps(payload, sort_keys=True, separators=(',', ':'), allow_nan=False).encode() + return self.transport.request(method, url, body=body if method == 'POST' else None, + headers={'Content-Type': 'application/json', 'Idempotency-Key': key, + 'X-StateHub-Source-Agent': 'informed-decision'}) + + def notify(self, delivery_id, metadata): + uuid.UUID(delivery_id) + if set(metadata) != {'memo_id', 'question', 'memo_class', 'submitter'}: + raise ValueError('invalid notification metadata') + if any(not isinstance(v, str) or not v or len(v) > 4096 for v in metadata.values()): + raise ValueError('invalid notification metadata') + if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{0,255}', metadata['memo_id']): + raise ValueError('invalid memo id') + status, receipt = self._request('POST', self.origin + '/informed-decision/notifications', + metadata, delivery_id) + # A queued relay receipt is pending, not successful central delivery. + if status not in (200, 201) or not isinstance(receipt, dict): + return False + try: + uuid.UUID(receipt['message_id']) + except (ValueError, TypeError, KeyError, AttributeError): + return False + return (receipt.get('delivery_id') == delivery_id and receipt.get('status') == 'accepted' + and receipt.get('authoritative_review') == 'signed-in-overview' + and all(receipt.get(k) == v for k, v in metadata.items())) + + def resolution(self, method, url, payload): + suffix = '/resolve' if method == 'POST' else '' + pattern = r'/decisions/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}' + suffix + if (method not in ('POST', 'GET') or not url.startswith(self.origin + '/decisions/') + or not re.fullmatch(pattern, url[len(self.origin):])): + raise ValueError('invalid hub resolution route') + material = json.dumps([method, url, payload], sort_keys=True, separators=(',', ':')).encode() + status, receipt = self._request(method, url, payload, + 'infd-resolution:' + hashlib.sha256(material).hexdigest()) + if 200 <= status < 300: + ident = url[len(self.origin + '/decisions/'):].split('/')[0] + if (status != 200 or not isinstance(receipt, dict) or receipt.get('id') != ident + or (method == 'POST' and (receipt.get('status') != 'resolved' + or any(receipt.get(k) != payload[k] for k in ('rationale', 'decided_by'))))): + return 503, {} # Queued or mismatched response is not a resolution receipt. + return status, receipt diff --git a/informed_decision/runtime.py b/informed_decision/runtime.py index efec7ef..29db1cf 100644 --- a/informed_decision/runtime.py +++ b/informed_decision/runtime.py @@ -36,8 +36,9 @@ def token_file(path): class AuditPump: """One process, bounded batches, 30s ticks; never drops a failed record.""" - def __init__(self, store, sink, *, clock=time.time): + def __init__(self, store, sink, *, clock=time.time, intake_worker=None): self.store, self.worker, self.clock = store, OutboxWorker(store, sink), clock + self.intake_worker = intake_worker self._stop = threading.Event() self._thread = None self._lock = threading.Lock() @@ -50,8 +51,12 @@ class AuditPump: def tick(self): try: - self.store.queue_heartbeats() + if self.intake_worker: + self.intake_worker.run_once(limit=10) + self.store.queue_heartbeats(include_posed=self.intake_worker is not None) result = self.worker.run_once(limit=10) + if self.intake_worker: + self.store.settle_intake_audit() now = self.clock() if now - self._last_reconciled >= 300: stamp = lambda t: datetime.fromtimestamp(t, timezone.utc).isoformat(timespec="microseconds") @@ -71,6 +76,8 @@ class AuditPump: self._last_reconciled = now healthy = not result["retrying"] and not result["blocked"] and not any( row["state"] != "delivered" for row in self.store.outbox()) + if self.intake_worker and self.store.intake_audit_pending(): + healthy = False with self._lock: self._last_ok = now if healthy else None except Exception: @@ -109,7 +116,7 @@ class Runtime: or info.st_size > 16384): raise ValueError("runtime configuration must be an owned private 0600 file") data = json.loads(path.read_text()) - if ((set(data) - {"ordinary_approval_ids"}) != {"schema", "evidence_db", "policy", "approval_origin", "audit"} + if ((set(data) - {"ordinary_approval_ids", "intake_delivery"}) != {"schema", "evidence_db", "policy", "approval_origin", "audit"} or data["schema"] != "informed-decision.review-runtime.v1" or set(data["policy"]) != {"origin", "package", "version", "package_digest", "caller_token_file"} or set(data["audit"]) != {"origin", "sender_token_file"}): @@ -125,7 +132,19 @@ class Runtime: allow_internal_http=True) sink = AuditCoreSink(data["audit"]["origin"], token_file(data["audit"]["sender_token_file"]), allow_internal_http=True) + delivery = data.get('intake_delivery') + hub = None + if 'intake_delivery' in data: + from .intake_hub import HubDelivery + if not isinstance(delivery, dict) or set(delivery) != {'hub_origin'}: + raise ValueError('invalid intake delivery configuration') + hub = HubDelivery(delivery['hub_origin']) store = Store(data["evidence_db"]) controller = ReviewController(store, client, lambda session: ApprovalHTTPClient( approval_origin, session, allow_internal_http=True, ordinary_approval_ids=tuple(ordinary))) - return cls(controller, AuditPump(store, sink)) + worker = None + if hub: + from .intake_delivery import IntakeDeliveryWorker + worker = IntakeDeliveryWorker(store, audit=sink, notify=hub.notify, + hub_transport=hub.resolution, hub_base_url=hub.origin) + return cls(controller, AuditPump(store, sink, intake_worker=worker)) diff --git a/informed_decision/store.py b/informed_decision/store.py index b62e0fd..a3bac3b 100644 --- a/informed_decision/store.py +++ b/informed_decision/store.py @@ -495,11 +495,56 @@ class Store: 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): + def stage_intake_audit(self): + """Promote retained posed envelopes into the ordinary immutable audit outbox. + + Same ids preserve receiver deduplication, including a previously sent + integration job. Delivery receipts and reconciliation now use one path. + """ + with self._transaction() as db: + for row in db.execute("SELECT i.id,i.payload FROM intake_deliveries i LEFT JOIN evidence e ON e.id=i.id " + "WHERE i.kind='memo-posed' AND (e.id IS NULL OR e.envelope!=i.payload) " + "ORDER BY i.rowid LIMIT 100").fetchall(): + event = json.loads(row['payload']) + fields = {'id', 'type', 'source', 'subject', 'tenant', 'correlation_id', 'occurred_at', 'data'} + data_fields = {'memo_id', 'approval_id', 'approval_binding_digest', 'submitter', 'memo_class', 'submission_digest'} + if (set(event) != fields or not isinstance(event.get('data'), dict) + or set(event['data']) != data_fields + or any(not isinstance(v, str) or not v for v in event['data'].values()) + or event.get('id') != row['id'] or event.get('type') != EventClass.MEMO_POSED.value + or event.get('source') != 'informed-decision' or event.get('tenant') != 'tenant:platform'): + raise Conflict('invalid retained posed event') + occurred = datetime.fromisoformat(event['occurred_at']) + if occurred.tzinfo is None: + raise Conflict('invalid retained posed time') + stamp = occurred.astimezone(timezone.utc).isoformat(timespec='microseconds') + existing = db.execute('SELECT envelope FROM evidence WHERE id=?', (row['id'],)).fetchone() + if existing and existing[0] != row['payload']: + raise Conflict('posed event id collision') + db.execute('INSERT OR IGNORE INTO evidence VALUES (?,?,?,?,?)', + (row['id'], EventClass.MEMO_POSED.value, stamp, row['payload'], row['payload'])) + db.execute('INSERT OR IGNORE INTO outbox(id) VALUES (?)', (row['id'],)) + + def settle_intake_audit(self): + with self._transaction() as db: + result = db.execute("UPDATE intake_deliveries SET state='delivered',lease=NULL,lease_until=NULL " + "WHERE kind='memo-posed' AND state!='delivered' AND id IN " + "(SELECT id FROM outbox WHERE state='delivered')") + return result.rowcount + + def intake_audit_pending(self): + with self._connection() as db: + return db.execute("SELECT 1 FROM intake_deliveries WHERE kind='memo-posed' " + "AND state!='delivered' LIMIT 1").fetchone() is not None + + def queue_heartbeats(self, now=None, *, include_posed=False): now = time.time() if now is None else now count = 0 with self._transaction() as db: - for event_class, interval in HEARTBEAT_CLASSES.items(): + classes = {**HEARTBEAT_CLASSES} + if include_posed: + classes[EventClass.MEMO_POSED.value] = 86400 + for event_class, interval in 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(): diff --git a/tests/test_hub_writeback.py b/tests/test_hub_writeback.py index b23251c..a4aa2de 100644 --- a/tests/test_hub_writeback.py +++ b/tests/test_hub_writeback.py @@ -87,3 +87,10 @@ def test_bad_decision_id_refused(): with pytest.raises(ValueError): write_back(answer(), {"hub_decision_id": "D1", "resolves": "on-answer"}, base_url=BASE, transport=Hub(), env=ON) + + +@pytest.mark.parametrize('read_status,body', [(503, {}), (404, {}), (200, {}), (200, None)]) +def test_unreadable_conflict_remains_retryable(read_status, body): + hub = Hub((409, {}), (read_status, body)) + assert write_back(answer(), {'hub_decision_id': D1, 'resolves': 'on-answer'}, + base_url=BASE, transport=hub, env=ON).status == 'failed' diff --git a/tests/test_intake_delivery_runtime.py b/tests/test_intake_delivery_runtime.py new file mode 100644 index 0000000..1b0d55e --- /dev/null +++ b/tests/test_intake_delivery_runtime.py @@ -0,0 +1,198 @@ +"""Delivery contracts, recovery and opt-in runtime composition.""" +import json +import os +from pathlib import Path +import sqlite3 +import sys +import uuid + +import pytest + +from informed_decision.audit import AuditDeliveryError, OutboxWorker +from informed_decision.evidence import EventClass +from informed_decision.intake_delivery import IntakeDeliveryWorker +from informed_decision.intake_hub import HubDelivery +from informed_decision.runtime import AuditPump, Runtime +from informed_decision.store import Store +from test_durable_component import receiver +from test_intake import envelope +from test_intake_service import setup, request +from test_review_runtime import config + + +class HubTransport: + def __init__(self, status=201): + self.status, self.calls = status, [] + + def request(self, method, url, *, headers, body): + payload = json.loads(body) if body else None + self.calls.append((method, url, headers, body)) + if url.endswith('/notifications'): + receipt = {'message_id': str(uuid.uuid4()), 'delivery_id': headers['Idempotency-Key'], + 'status': 'accepted', 'authoritative_review': 'signed-in-overview', **payload} + else: + receipt = {'id': url.split('/decisions/')[1].split('/')[0], + 'status': 'resolved', **(payload or {})} + return self.status, receipt + + +def test_notification_replay_uses_identical_request_and_checks_receipt(): + transport = HubTransport() + hub = HubDelivery('https://hub.test', transport=transport) + ident = str(uuid.uuid4()) + metadata = {'memo_id': 'review-1', 'question': 'Proceed?', 'memo_class': 'credential-lane', 'submitter': 'state-hub'} + assert hub.notify(ident, metadata) + assert hub.notify(ident, metadata) + assert transport.calls[0] == transport.calls[1] + assert json.loads(transport.calls[0][3]) == metadata + assert transport.calls[0][1].endswith('/informed-decision/notifications') + assert transport.calls[0][2]['Idempotency-Key'] == ident + transport.status = 202 + assert not hub.notify(ident, metadata) # Relay queue acknowledgement is not a receipt. + with pytest.raises(ValueError): + hub.notify(ident, {**metadata, 'packet': 'private'}) + transport.request = lambda *args, **kwargs: (201, {'id': str(uuid.uuid4())}) + assert not hub.notify(ident, metadata) + + +def test_resolution_is_fixed_origin_and_idempotent(): + transport = HubTransport(status=200) + hub = HubDelivery('https://hub.test', transport=transport) + url = 'https://hub.test/decisions/' + str(uuid.uuid4()) + for _ in range(2): + hub.resolution('POST', url + '/resolve', {'rationale': 'memo review-1', 'decided_by': 'human'}) + assert transport.calls[0] == transport.calls[1] + hub.resolution('GET', url, {}) + transport.status = 202 + assert hub.resolution('POST', url + '/resolve', {'rationale': 'memo review-1', 'decided_by': 'human'}) == (503, {}) + for method, target in [('GET', url + '/resolve'), ('POST', url), + ('POST', url.replace('hub.test', 'other.test') + '/resolve'), + ('POST', 'https://hub.test/decisions/' + '-' * 36 + '/resolve')]: + with pytest.raises(ValueError): + hub.resolution(method, target, {}) + + +def test_runtime_delivery_is_opt_in_and_writeback_remains_off(tmp_path, monkeypatch): + monkeypatch.delenv('INFD_HUB_WRITEBACK', raising=False) + path, data = config(tmp_path) + assert Runtime.from_file(path).pump.intake_worker is None + data['intake_delivery'] = {'hub_origin': 'https://hub.test'} + path.write_text(json.dumps(data)) + worker = Runtime.from_file(path).pump.intake_worker + assert worker is not None and worker.audit and worker.notify and worker.hub_transport + from informed_decision.hub_writeback import enabled + assert not enabled(worker.env) + for invalid in [None, {}, {**data['intake_delivery'], 'allow_all': True}, + {'hub_origin': 'http://public.example.com'}]: + data['intake_delivery'] = invalid + path.write_text(json.dumps(data)) + with pytest.raises(ValueError): + Runtime.from_file(path) + + +def test_posed_evidence_survives_restart_and_is_immutable(setup): + service, store, _, _ = setup + assert request(service, envelope())[0] == 201 + store.stage_intake_audit() + before = store.evidence() + assert len(before) == 1 and before[0]['class'] == EventClass.MEMO_POSED.value + Store(store.path).stage_intake_audit() + assert store.evidence() == before and len(store.outbox()) == 1 + with store._connection() as db: + with pytest.raises(sqlite3.IntegrityError): + db.execute('DELETE FROM evidence') + assert store.intake_audit_pending() + + +def test_actual_posed_receiver_lost_reply_deduplicates_and_reconciles(setup, receiver): + service, store, _, _ = setup + assert request(service, envelope())[0] == 201 + sink, transport, _ = receiver + store.stage_intake_audit() + first = store.claim_delivery(now=10) + transport.lose_reply = True + with pytest.raises(AuditDeliveryError): + sink.deliver(first[0], first[2]) + 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) + assert store.settle_intake_audit() == 1 and not store.intake_audit_pending() + report = OutboxWorker(store, sink).reconcile('2020-01-01T00:00:00Z', '2100-01-01T00:00:00Z') + assert report['count_values_match'] + assert report['counts'][EventClass.MEMO_POSED.value] == {'source': 1, 'receiver': 1} + assert store.outbox()[0]['receiver_reference'] == 'audit:' + first[0] + + +def test_pump_includes_posed_evidence_and_retains_failed_notifications(setup, receiver): + service, store, _, _ = setup + assert request(service, envelope())[0] == 201 + sink, _, _ = receiver + worker = IntakeDeliveryWorker(store, audit=sink, notify=lambda *args: False) + pump = AuditPump(store, sink, intake_worker=worker) + pump.tick() + assert pump.ready() and not store.intake_audit_pending() + with store._connection() as db: + assert db.execute("SELECT state FROM intake_deliveries WHERE kind='notification'").fetchone()[0] == 'pending' + snapshot = json.loads((store.path.parent / 'audit-reconciliation.json').read_text()) + assert snapshot['count_values_match'] and EventClass.MEMO_POSED.value in snapshot['counts'] + + +def test_failed_posed_delivery_blocks_readiness_and_no_empty_heartbeat(setup, receiver): + service, store, _, _ = setup + assert request(service, envelope())[0] == 201 + sink, _, _ = receiver + sink.token_provider = lambda: 'invalid-synthetic-token' + pump = AuditPump(store, sink, intake_worker=IntakeDeliveryWorker(store, audit=sink)) + pump.tick() + assert not pump.ready() and store.intake_audit_pending() + heartbeats = [json.loads(row['envelope']) for row in store.evidence() + if row['class'] == EventClass.HEARTBEAT.value] + assert all(row['data']['class'] != EventClass.MEMO_POSED.value for row in heartbeats) + + +@pytest.fixture +def hub_contract(): + source = os.environ.get('INFD_STATE_HUB_SOURCE') + if not source: + pytest.skip('set INFD_STATE_HUB_SOURCE for actual hub contract checks') + assert (Path(source) / 'api/services/memo_forwarding.py').is_file() + sys.path.insert(0, source) + from api.services import informed_decision_contract, memo_forwarding + return informed_decision_contract, memo_forwarding + + +def test_actual_hub_forwarder_refuses_governed_boolean_fields(setup, hub_contract, tmp_path): + service, store, creates, _ = setup + contract, forwarding = hub_contract + body = envelope() + body['idempotency_key'] = contract.stable_idempotency_key(body['serves']['hub_decision_id'], body['memo']['id']) + calls = [] + def transport(method, url, payload, headers): + assert method == 'POST' and url == 'https://intake.test/v1/memos' + assert headers['Authorization'] == 'Bearer projected-token' + calls.append(payload) + return request(service, payload) + fixture = forwarding.DevIntakeFixture(tmp_path / 'hub.sqlite', transport=transport, + intake_base_url='https://intake.test', hub_base_url='https://hub.test', + token_reader=lambda: 'projected-token') + # Owner commit 49eeddb only permits int/float/null leaves, rejecting the + # shared schema's actual bools. Do not strip or change those binding fields. + with pytest.raises(contract.ContractError) as refused: + fixture.submit(body) + assert refused.value.code == 'invalid_type' + assert not store.memos_for('keycape-human') and not creates and not calls + assert b'projected-token' not in Path(fixture.path).read_bytes() + + +def test_notification_body_matches_actual_hub_parser(hub_contract): + contract, _ = hub_contract + class Transport(HubTransport): + def request(self, method, url, *, headers, body): + parsed = contract.parse_notification(json.loads(body)) + assert parsed.as_body() == json.loads(body) + assert contract.parse_delivery_id(headers['Idempotency-Key']) + return super().request(method, url, headers=headers, body=body) + hub = HubDelivery('https://hub.test', transport=Transport()) + assert hub.notify(str(uuid.uuid4()), {'memo_id': 'review-1', 'question': 'Proceed?', + 'memo_class': 'credential-lane', 'submitter': 'state-hub'}) diff --git a/workplans/INFD-WP-0005-frictionless-decision-intake.md b/workplans/INFD-WP-0005-frictionless-decision-intake.md index 1d0d0f5..341a736 100644 --- a/workplans/INFD-WP-0005-frictionless-decision-intake.md +++ b/workplans/INFD-WP-0005-frictionless-decision-intake.md @@ -24,8 +24,8 @@ state_hub_workstream_id: "c9407d2e-fd51-5afe-be4c-24483d1b2dbc" Design: `docs/specs/FrictionlessDecisionIntake.md`. Lifecycle review (2026-10-03): blocked after completing the local intake -service, durable ledger/outbox and doctrine-response incorporation. The remaining -runtime composition, custody, policy, receiver contracts and rollout require +service, durable ledger/outbox, scheduled delivery and doctrine-response incorporation. The remaining +intake listener composition, custody, policy, forwarding and rollout require owner deliveries or founder approval. Requirements and acceptance evidence are in `docs/intake-dependency-requirements.md`; no live change was made. @@ -88,7 +88,7 @@ state_hub_task_id: "b91af9d2-30ef-5c46-93f0-5faeba923b5d" Hub write-back adapter `informed_decision/hub_writeback.py` behind `INFD_HUB_WRITEBACK=1` (default off), with tests in `tests/test_hub_writeback.py`. -Not wired into the disposition path yet (T07). +Wired into the terminal-disposition outbox and opt-in scheduled runtime (T07). ## Authenticated intake service and ledger @@ -144,19 +144,21 @@ Replaces OPERATOR step A4. ```task id: INFD-WP-0005-T07 -status: wait -blocked_on: message-from:audit-core -blocking_reason: "Local queued notification/posed events and disposition write-back implemented; audit-core event registration and State Hub deduplicated notification contracts required for runtime wiring (docs/intake-dependency-requirements.md)" +status: done priority: medium depends_on: [INFD-WP-0005-T04] state_hub_task_id: "22d361cc-4bf2-59d4-9b97-728d42b1aeee" ``` Notification and evidence: hub message to the founder on each posed memo -(id, question, class, submitter, link; no content); emit +(id, question, class, submitter, signed-in-overview reference; no content); emit `informed-decision.memo.posed` to audit-core once registered (msg `23b670ac`); wire `hub_writeback.write_back` into the terminal-disposition path with a local -outbox. Replaces OPERATOR step A5. +outbox. Replaces OPERATOR step A5. Implemented locally using the registered +AUDIT-WP-0011 receiver contract and State Hub's retained notification receipts +(the owner contract prohibits notification URLs). +The optional runtime `intake_delivery` configuration schedules immutable posed +evidence, notification and gated resolution delivery. Live acceptance remains T10. ## State Hub forwarding acceptance @@ -165,12 +167,18 @@ id: INFD-WP-0005-T08 status: wait priority: medium blocked_on: message-from:state-hub -blocking_reason: "State Hub as first registered submitter (forwarding decision memos with its SA token) is state-hub's change (msg e30dea31)" +blocking_reason: "Owner dev forwarder 49eeddb refuses governed bool fields (invalid_type); state-hub must fix its contract, choose the actual SA and supply reviewed grant/runtime forwarding (msg e30dea31), alongside T05/T06" state_hub_task_id: "ce2b36b4-ef82-53b7-aee7-e0063855370b" ``` Register state-hub as a submitter and run an end-to-end pose from a hub decision through to the founder's review page in a dev stack. +The owner's DevIntakeFixture now has a cross-repo negative fixture against the +real p01 envelope: `_bounded_tree` refuses `requires_new_bind`, `required_ack` +and `sealed` booleans. The owner must accept shared-schema boolean leaves; +removing or coercing binding fields here is not a compatible workaround. +Production submitter registration and browser/PDP acceptance also wait on +T05/T06 and the owner's chosen SA/runtime forwarding. ## Doctrine response @@ -219,10 +227,17 @@ revocation, replay/restart, collisions, daily budget, atomic rollback, uncertain create refusal and receipt-based load recovery. No deployed listener is enabled. T05 local create-only transport is implemented; workload token exchange and -custody remain external. T07 local integration jobs are durable and terminal -human dispositions enqueue resolution atomically; the worker preserves stable -ids and leaves unconfigured adapters and disabled write-back pending. Receiver -contracts and scheduled runtime composition remain blocked. +custody remain external. T07 is complete locally: posed jobs enter the ordinary +immutable audit outbox with receipt retention, per-class reconciliation and +optional 24-hour heartbeat; notifications require matching hub receipts and +stable idempotency keys. The scheduled runtime drains these and terminal human +resolutions when explicitly configured; disabled write-back remains pending. +Tests exercise the actual Audit Core receiver, lost-reply replay, counts, +readiness refusal and failed notification retention. Nothing was deployed. + +Loose-end audit: all other current and archived workplans have only done/cancel +tasks. T05, T06, T08 and T10 remain waiting on the named external deliveries and +live acceptance; this plan remains blocked. No new task or workplan was opened. T09: reply 4a290496 incorporated into the spec, with proposed-decision status preserved. No founder ruling or deployment permission is inferred.