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

Implement offline workload intake and record blocked owner dependencies

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 18:45:10 +02:00
parent c1a75e85aa
commit c1cf6cf54c
13 changed files with 790 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

34
informed_decision/intake_approval.py Normal file
View file

@ -0,0 +1,34 @@
"""Create-only approval adapter; token custody/exchange belongs to workload setup."""
from datetime import datetime, timedelta, timezone
from .http_transport import JSONTransport, fixed_origin
from .intake import canonical
from .intake_service import IntakeError
class IntakeApprovalClient:
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 __call__(self, request):
if (set(request) != {'binding', 'human_control', 'required_count'}
or request['human_control'] is not True or type(request['required_count']) is not int
or request['required_count'] != 1 or request['binding'].get('actor') != 'informed-decision'):
raise IntakeError(400, 'invalid_approval_intent')
try:
token = self.token_provider()
if (not isinstance(token, str) or not token or len(token) > 16384
or not token.isascii() or any(c.isspace() for c in token)):
raise ValueError()
now = datetime.now(timezone.utc)
body = {**request, 'pdp_path': False, 'validity': {
'not_before': now.isoformat(), 'expires_at': (now + timedelta(hours=24)).isoformat()}}
status, result = self.transport.request('POST', self.origin + '/v1/approvals',
headers={'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}, body=canonical(body))
if status != 201:
raise ValueError()
return result
except Exception:
raise IntakeError(503, 'approval_create_unresolved') from None

68
informed_decision/intake_delivery.py Normal file
View file

@ -0,0 +1,68 @@
"""Leased local integration outbox. Owner composition supplies bounded transports.
Unknown adapters leave rows pending; hub resolutions remain pending while the
write-back flag is off. Dispatch failure never rolls back a human disposition.
Audit/notification receivers must deduplicate using the stable delivery id.
"""
import json
import time
import uuid
from .hub_writeback import Answer, enabled, write_back
class IntakeDeliveryWorker:
def __init__(self, store, *, audit=None, notify=None, hub_transport=None,
hub_base_url='', env=None, clock=time.time):
self.store, self.audit, self.notify = store, audit, notify
self.hub_transport, self.hub_base_url = hub_transport, hub_base_url
self.env, self.clock = env, clock
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 = []
if self.audit:
kinds.append('memo-posed')
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
for _ in range(limit):
now, lease = self.clock(), str(uuid.uuid4())
with self.store._transaction() as db:
placeholders = ','.join('?' for kind in kinds)
row = db.execute(f"SELECT * FROM intake_deliveries WHERE kind IN ({placeholders}) "
"AND state IN ('pending','in_flight') AND next_attempt<=? "
"AND (lease_until IS NULL OR lease_until<=?) ORDER BY rowid LIMIT 1", (*kinds, now, now)).fetchone()
if row is None:
break
db.execute("UPDATE intake_deliveries SET state='in_flight',lease=?,lease_until=?,attempts=attempts+1 WHERE id=?",
(lease, now + 30, row['id']))
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':
# notify(id, metadata) returns only after a deduplicated receiver receipt.
if self.notify(row['id'], payload) is True:
state = 'delivered'
else:
result = write_back(Answer(**payload['answer']), payload['serves'],
base_url=self.hub_base_url, transport=self.hub_transport, env=self.env)
if result.status in ('resolved', 'already_resolved'):
state = 'delivered'
elif result.status == 'conflict':
state = 'blocked'
except Exception:
pass # Never persist upstream bodies, tokens, or exception text.
with self.store._transaction() as db:
db.execute("UPDATE intake_deliveries SET state=?,lease=NULL,lease_until=NULL,next_attempt=? "
"WHERE id=? AND lease=?", (state, now + min(3600, 2 ** min(row['attempts'] + 1, 11)), row['id'], lease))
delivered += state == 'delivered'
return delivered

218
informed_decision/intake_service.py Normal file
View file

@ -0,0 +1,218 @@
"""Separate workload-only WSGI intake. No browser session or binding route.
The creator and TokenReview transports are injected by the workload composition.
An uncertain create is never automatically repeated: its durable reservation
requires receipt recovery until approval-engine supports idempotent creation.
"""
from dataclasses import replace
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import uuid
from .intake import IntakeRefused, SubmitterGrant, canonical, idempotency_outcome, validate_submission
from .memo import Principal
from .records import dumps
AUDIENCE = "informed-decision-intake"
MAX_BODY = 1024 * 1024
class IntakeError(Exception):
def __init__(self, status, code):
self.status, self.code = status, code
super().__init__(code)
class TokenReview:
"""call(body) performs an authenticated Kubernetes TokenReview using pod credentials.
The adapter must verify the API server's CA and suppress tokens and responses
in its logs. TokenReview is performed afresh on every pose, including replay.
"""
def __init__(self, call):
self.call = call
def authenticate(self, token):
try:
result = self.call({"apiVersion": "authentication.k8s.io/v1", "kind": "TokenReview",
"spec": {"token": token, "audiences": [AUDIENCE]}})
status = result["status"]
user = status["user"]
username = user["username"]
if (status.get("authenticated") is not True or status.get("error")
or AUDIENCE not in status.get("audiences", [])
or not user.get("uid") or not isinstance(username, str)
or not re.fullmatch(r"system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+", username)):
raise ValueError()
return username
except Exception:
raise IntakeError(401, "workload_authentication_failed") from None
def configuration(submitters_path, groups_path):
"""Reload reviewed deployment data on every request so removal revokes immediately.
Empty shipped registries deliberately admit no workload or human recipient.
Entries identify actual service accounts; drafted_by is only an assertion.
"""
try:
submitters = json.loads(Path(submitters_path).read_text())
groups = json.loads(Path(groups_path).read_text())
if not isinstance(submitters, dict) or not isinstance(groups, dict):
raise ValueError()
for username, row in submitters.items():
if (not re.fullmatch(r"system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+", username)
or set(row) != {"submitter_id", "memo_classes", "review_groups", "daily_limit"}
or not isinstance(row["submitter_id"], str)
or not row["submitter_id"] or type(row["daily_limit"]) is not int
or row["daily_limit"] < 1):
raise ValueError()
for field in ("memo_classes", "review_groups"):
if not isinstance(row[field], list) or any(not isinstance(x, str) for x in row[field]):
raise ValueError()
if len({r["submitter_id"] for r in submitters.values()}) != len(submitters):
raise ValueError()
for row in groups.values():
if (set(row) != {"subject", "display_name"} or not all(isinstance(x, str) and x for x in row.values())):
raise ValueError()
return submitters, groups
except Exception:
raise IntakeError(503, "intake_configuration_unavailable") from None
class IntakeService:
def __init__(self, store, authenticator, creator, *, submitters_path, groups_path):
self.store, self.authenticator, self.creator = store, authenticator, creator
self.submitters_path, self.groups_path = submitters_path, groups_path
def pose(self, token, envelope):
username = self.authenticator.authenticate(token)
submitters, groups = configuration(self.submitters_path, self.groups_path)
row = submitters.get(username)
if row is None:
raise IntakeError(403, "submitter_unknown")
grant = SubmitterGrant(row["submitter_id"], frozenset(row["memo_classes"]), frozenset(row["review_groups"]))
submission = validate_submission(envelope, submitter_id=grant.submitter_id, grant=grant)
recipient = groups.get(submission.review_group)
if recipient is None:
raise IntakeError(503, "review_group_unconfigured")
metadata = {"memo_class": submission.memo_class, "review_group": submission.review_group,
"drafted_by": submission.drafted_by, "serves": submission.serves}
recovered = None
# Reserve before any network side effect. SQLite serializes concurrent poses.
with self.store._transaction() as db:
previous = db.execute("SELECT * FROM intake_ledger WHERE submitter=? AND key=?",
(submission.submitter_id, submission.idempotency_key)).fetchone()
if previous:
idempotency_outcome(previous["digest"], submission.submission_digest)
if previous["state"] == "ready":
return 200, json.loads(previous["receipt"])
if previous["state"] == "created":
recovered = json.loads(previous["receipt"])
else:
raise IntakeError(409, "approval_create_unresolved")
if not previous and (db.execute("SELECT 1 FROM memos WHERE id=?", (submission.memo.id,)).fetchone()
or db.execute("SELECT 1 FROM intake_ledger WHERE memo_id=?", (submission.memo.id,)).fetchone()):
raise IntakeError(409, "memo_id_taken")
# Budget counts attempts as well as completed poses; retries do not spend twice.
day = datetime.now(timezone.utc).date().isoformat()
count = db.execute("SELECT COUNT(*) FROM intake_ledger WHERE submitter=? AND json_extract(metadata,'$.day')=?",
(submission.submitter_id, day)).fetchone()[0]
if not previous and count >= row["daily_limit"]:
raise IntakeError(429, "submitter_budget_exhausted")
metadata["day"] = day
if not previous:
db.execute("INSERT INTO intake_ledger VALUES (?,?,?,?,?,?,NULL)",
(submission.submitter_id, submission.idempotency_key, submission.submission_digest,
submission.memo.id, "creating", dumps(metadata)))
if recovered is None:
try:
approval = self.creator({"binding": submission.approval_binding,
"human_control": True, "required_count": 1})
binding = approval["binding"]
approval_id, digest = approval["id"], binding["digest"]
if (not re.fullmatch(r"[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}", approval_id)
or not re.fullmatch(r"sha256:[0-9a-f]{64}", digest)
or binding.get("human_control") is not True
or any(binding.get(k) != v for k, v in submission.approval_binding.items())
or approval.get("status") != "requested" or approval.get("entries")):
raise ValueError()
except Exception:
raise IntakeError(503, "approval_create_unresolved") from None
# Preserve the receipt before local load: recovery can locate this engine object.
receipt = {"memo_id": submission.memo.id, "approval_id": approval_id, "approval_binding_digest": digest}
with self.store._transaction() as db:
db.execute("UPDATE intake_ledger SET state='created',receipt=? WHERE submitter=? AND key=?",
(dumps(receipt), submission.submitter_id, submission.idempotency_key))
else:
receipt = recovered
approval_id, digest = receipt["approval_id"], receipt["approval_binding_digest"]
memo = replace(submission.memo, approval_id=approval_id, approval_binding_digest=digest,
binding=replace(submission.memo.binding, principal=Principal(
recipient["subject"], "person", recipient["display_name"])))
with self.store._transaction() as db:
current = db.execute("SELECT state,receipt FROM intake_ledger WHERE submitter=? AND key=?",
(submission.submitter_id, submission.idempotency_key)).fetchone()
if current["state"] == "ready":
return 200, json.loads(current["receipt"])
for item in envelope["packet"]:
data = submission.packet[item["item_id"]]
document_digest = "sha256:" + hashlib.sha256(data).hexdigest()
existing = db.execute("SELECT * FROM documents WHERE digest=?", (document_digest,)).fetchone()
if existing and (existing["media_type"] != item["media_type"] or existing["content"] != data):
raise IntakeError(409, "document_conflict")
db.execute("INSERT OR IGNORE INTO documents VALUES (?,?,?)", (document_digest, item["media_type"], data))
db.execute("INSERT INTO memos VALUES (?,?,?)", (memo.id, memo.version, dumps(memo)))
event_id = str(uuid.uuid4())
event = {"id": event_id, "type": "informed-decision.memo.posed",
"source": "informed-decision", "tenant": "tenant:platform",
"subject": memo.id, "correlation_id": approval_id,
"occurred_at": datetime.now(timezone.utc).isoformat(),
"data": {**receipt, "submitter": submission.submitter_id,
"memo_class": submission.memo_class,
"submission_digest": submission.submission_digest}}
notification = {"memo_id": memo.id, "question": memo.question,
"memo_class": submission.memo_class, "submitter": submission.submitter_id}
for kind, payload in (("memo-posed", event), ("notification", notification)):
db.execute("INSERT INTO intake_deliveries(id,kind,payload) VALUES (?,?,?)",
(event_id if kind == "memo-posed" else str(uuid.uuid4()), kind, dumps(payload)))
db.execute("UPDATE intake_ledger SET state='ready' WHERE submitter=? AND key=?",
(submission.submitter_id, submission.idempotency_key))
return 201, receipt
def __call__(self, environ, start_response):
try:
if environ.get("PATH_INFO") != "/v1/memos":
raise IntakeError(404, "not_found")
if environ.get("REQUEST_METHOD") != "POST":
raise IntakeError(405, "method_not_allowed")
authorization = environ.get("HTTP_AUTHORIZATION", "")
if not authorization.startswith("Bearer ") or not authorization[7:] or len(authorization) > 16384:
raise IntakeError(401, "workload_token_required")
if environ.get("CONTENT_TYPE", "").split(";")[0] != "application/json":
raise IntakeError(415, "json_required")
length = int(environ.get("CONTENT_LENGTH") or 0)
if length < 1 or length > MAX_BODY:
raise IntakeError(413, "submission_too_large")
raw = environ["wsgi.input"].read(length)
if len(raw) != length:
raise IntakeError(400, "invalid_json")
status, body = self.pose(authorization[7:], json.loads(raw))
except IntakeError as exc:
status, body = exc.status, {"error": exc.code}
except IntakeRefused as exc:
status = 409 if exc.reason == "idempotency_conflict" else 400
body = {"error": exc.reason} # No client-controlled field paths in logs/responses.
except (ValueError, TypeError, KeyError, UnicodeError):
status, body = 400, {"error": "invalid_submission"}
except Exception:
status, body = 503, {"error": "intake_unavailable"}
from http import HTTPStatus
raw = canonical(body)
start_response(f"{status} {HTTPStatus(status).phrase}",
[("Content-Type", "application/json"), ("Content-Length", str(len(raw))),
("Cache-Control", "no-store")])
return [raw]

29
informed_decision/store.py
View file

@ -27,7 +27,7 @@ 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 = 2
SCHEMA_VERSION = 3
_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));
@ -99,7 +99,7 @@ class Store:
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 not in (1, SCHEMA_VERSION):
elif version not in (1, 2, SCHEMA_VERSION):
raise StoreError("unsupported evidence schema version")
if version in (0, 1):
db.executescript("""BEGIN IMMEDIATE;
@ -113,6 +113,19 @@ class Store:
BEGIN SELECT RAISE(ABORT,'immutable evidence'); END;
PRAGMA user_version=2; COMMIT;""")
if version in (0, 1, 2):
db.executescript("""BEGIN IMMEDIATE;
CREATE TABLE intake_ledger (
submitter TEXT NOT NULL, key TEXT NOT NULL, digest TEXT NOT NULL,
memo_id TEXT NOT NULL UNIQUE, state TEXT NOT NULL,
metadata TEXT NOT NULL, receipt TEXT,
PRIMARY KEY(submitter,key));
CREATE TABLE intake_deliveries (
id TEXT PRIMARY KEY, kind TEXT NOT NULL, payload TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'pending', attempts INTEGER NOT NULL DEFAULT 0,
next_attempt REAL NOT NULL DEFAULT 0, lease TEXT, lease_until REAL);
PRAGMA user_version=3; COMMIT;""")
@contextmanager
def _connection(self):
db = sqlite3.connect(self.path.as_uri() + "?mode=rw", uri=True, timeout=5, isolation_level=None)
@ -320,6 +333,18 @@ class Store:
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))
intake = db.execute("SELECT metadata FROM intake_ledger WHERE memo_id=? AND state='ready'", (memo.id,)).fetchone()
if intake and verb in (Verb.ACCEPT, Verb.DECLINE):
from .hub_writeback import Answer, build_resolution
serves = json.loads(intake["metadata"]).get("serves")
if serves and serves.get("resolves") == "on-answer":
answer = Answer(memo.id, memo.version, verb.value, d.id, d.at,
memo.binding.principal.display_name, memo.approval_id,
memo.approval_binding_digest, p.view_hash, actor.kind.value)
payload = {"answer": answer.__dict__, "serves": serves,
"resolution": build_resolution(answer)}
db.execute("INSERT INTO intake_deliveries(id,kind,payload) VALUES (?,?,?)",
(d.id, "hub-resolution", dumps(payload)))
c = commit_disposition(d, custody="informed-decision:dispositions:" + d.id)
self._event(db, c, {"disposition": json.loads(dumps(d)), "presentation": json.loads(dumps(p)),
"policy_observation_id": policy_id},