coulomb/informed-decision
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Complete intake delivery wiring and narrow remaining owner blockers

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1029c-c003-7283-acc9-930b434e6cae
This commit is contained in:
tegwick 2026-10-03 21:34:50 +02:00
parent 49f8eb4c4b
commit ff59199c3b
13 changed files with 452 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

1
informed_decision/evidence.py
View file

@ -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"

4
informed_decision/hub_writeback.py
View file

@ -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])

16
informed_decision/intake_delivery.py
View file

@ -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'

57
informed_decision/intake_hub.py Normal file
View file

@ -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

27
informed_decision/runtime.py
View file

@ -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))

49
informed_decision/store.py
View file

@ -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():