"""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_preserves_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 d61ba32 accepts the shared schema's bool leaves. Do not strip # or coerce requires_new_bind, required_ack, or sealed. posed = fixture.submit(body) assert posed.status == 'delivered' and len(calls) == 1 and len(creates) == 1 memo = calls[0]['memo'] assert memo['binding']['target']['requires_new_bind'] is False assert memo['highlights'][0]['required_ack'] is True assert memo['sealed'] is False assert store.memos_for('keycape-human') 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'})