tegwick
ee472acc87
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
647 lines
38 KiB
Python
647 lines
38 KiB
Python
"""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 = 4
|
|
_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 not in (1, 2, 3, SCHEMA_VERSION):
|
|
raise StoreError("unsupported evidence schema version")
|
|
if version in (0, 1):
|
|
db.executescript("""BEGIN IMMEDIATE;
|
|
CREATE TABLE policy_observations (id TEXT PRIMARY KEY, at TEXT NOT NULL,
|
|
request TEXT NOT NULL, decision TEXT, outcome TEXT NOT NULL,
|
|
usable_until REAL NOT NULL, response_sha256 TEXT,
|
|
decision_attributable INTEGER NOT NULL CHECK(decision_attributable=0));
|
|
CREATE TRIGGER immutable_policy_observations_UPDATE BEFORE UPDATE ON policy_observations
|
|
BEGIN SELECT RAISE(ABORT,'immutable evidence'); END;
|
|
CREATE TRIGGER immutable_policy_observations_DELETE BEFORE DELETE ON policy_observations
|
|
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;""")
|
|
|
|
if version in (0, 1, 2, 3):
|
|
from .lifecycle import SCHEMA as lifecycle_schema
|
|
db.executescript("BEGIN IMMEDIATE;" + lifecycle_schema
|
|
+ "PRAGMA user_version=4; COMMIT;")
|
|
|
|
@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")
|
|
digest = "sha256:" + hashlib.sha256(content).hexdigest()
|
|
with self._transaction() as db:
|
|
self._put_document(db, digest, media_type, content)
|
|
return digest
|
|
|
|
def _put_document(self, db, digest, media_type, content):
|
|
_text(media_type, "media type")
|
|
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))
|
|
|
|
def save_memo(self, memo: Memo):
|
|
with self._transaction() as db:
|
|
if db.execute("SELECT 1 FROM lifecycle_records WHERE memo_id=?", (memo.id,)).fetchone():
|
|
raise Conflict("registered lifecycle memos require lifecycle revision commands")
|
|
self._save_memo(db, memo)
|
|
|
|
def _save_memo(self, db, memo: Memo):
|
|
"""Shared transaction boundary for imports and lifecycle writes."""
|
|
_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")
|
|
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, version=None):
|
|
with self._connection() as db:
|
|
return self._memo(db, memo_id, version)
|
|
|
|
def observe_policy(self, observation):
|
|
request = json.loads(observation.request_json)
|
|
_text(request["id"], "policy observation id")
|
|
with self._transaction() as db:
|
|
db.execute("INSERT INTO policy_observations VALUES (?,?,?,?,?,?,?,0)",
|
|
(request["id"], _now(), observation.request_json, observation.decision_json,
|
|
observation.outcome, observation.usable_until, observation.response_sha256))
|
|
return request["id"]
|
|
|
|
def policy_observations(self):
|
|
with self._connection() as db:
|
|
return [dict(r) for r in db.execute("SELECT * FROM policy_observations ORDER BY rowid")]
|
|
|
|
def _guard_policy(self, db, policy_id, memo, subject, action):
|
|
if policy_id is None:
|
|
return # Existing internal domain API; the browser always supplies one.
|
|
row = db.execute("SELECT * FROM policy_observations WHERE id=?", (policy_id,)).fetchone()
|
|
if row is None or row["outcome"] != "allow" or row["usable_until"] <= time.time():
|
|
raise Conflict("a current recorded policy observation is required")
|
|
request = json.loads(row["request"])
|
|
if (request["action"] != action or request["subject"]["id"] != subject
|
|
or request["resource"]["id"] != "memo:" + memo.id
|
|
or request["context"] != {"memo_version": memo.version, "approval_id": memo.approval_id,
|
|
"approval_binding_digest": memo.approval_binding_digest}):
|
|
raise Conflict("policy observation does not name this act")
|
|
|
|
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,
|
|
expected_version=None, policy_id=None, approval_binding=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)
|
|
if expected_version is not None and memo.version != expected_version:
|
|
raise Conflict("memo changed after the entitlement check")
|
|
self._guard_policy(db, policy_id, memo, principal_sub, "read")
|
|
if approval_binding is not None and approval_binding.get("digest") != memo.approval_binding_digest:
|
|
raise Conflict("observed act binding does not match this memo")
|
|
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)}
|
|
if policy_id is not None:
|
|
content["policy_observation_id"] = policy_id
|
|
if approval_binding is not None:
|
|
content["approval_binding"] = approval_binding # Observed act, never approval validity/state.
|
|
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, *, policy_id=None):
|
|
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")
|
|
self._guard_policy(db, policy_id, memo, actor.sub, "acknowledge")
|
|
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)), "policy_observation_id": policy_id},
|
|
{"event_kind": "acknowledgment", "acknowledged_at": at})
|
|
return p
|
|
|
|
def record_disposition(self, presentation_id, actor, verb, *, operation_id, reasons=(), note=None, policy_id=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)
|
|
self._guard_policy(db, policy_id, memo, actor.sub, verb.value)
|
|
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))
|
|
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},
|
|
{"submission_state": "prepared"} if verb is Verb.ACCEPT else None)
|
|
return d
|
|
|
|
def begin_submission(self, disposition_id, *, policy_id=None):
|
|
"""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")
|
|
self._guard_policy(db, policy_id, self._memo(db, p.memo_id), d.actor.sub, "accept")
|
|
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 intent_for(self, approval_id, subject):
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT s.*,d.presentation_id,d.operation_id FROM submissions s "
|
|
"JOIN dispositions d ON d.id=s.disposition_id WHERE s.approval_id=? AND s.subject=?",
|
|
(approval_id, subject)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def dispositions_for(self, presentation_id):
|
|
with self._connection() as db:
|
|
return [disposition_from(json.loads(r[0])) for r in db.execute(
|
|
"SELECT body FROM dispositions WHERE presentation_id=? ORDER BY rowid", (presentation_id,))]
|
|
|
|
def memos_for(self, subject):
|
|
"""Legacy review candidates; unpublished lifecycle records stay private."""
|
|
with self._connection() as db:
|
|
rows = db.execute("SELECT m.body FROM memos m WHERE m.version="
|
|
"(SELECT MAX(version) FROM memos WHERE id=m.id) "
|
|
"AND NOT EXISTS (SELECT 1 FROM lifecycle_records l WHERE l.memo_id=m.id) "
|
|
"ORDER BY m.id").fetchall()
|
|
memos = (memo_from(json.loads(r[0])) for r in rows)
|
|
return [m for m in memos if m.binding.principal.id == subject and m.binding.principal.kind == "person"]
|
|
|
|
def lifecycle_counts(self):
|
|
"""Content-free operational counts, including undelivered lifecycle evidence."""
|
|
with self._connection() as db:
|
|
return {name: db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
|
|
for name, table in (("records", "lifecycle_records"),
|
|
("events", "lifecycle_events"),
|
|
("pending", "lifecycle_pending"))}
|
|
|
|
def dispositions_by(self, subject, memo_id):
|
|
"""``subject``'s dispositions on every version of a memo, oldest first, with any submission."""
|
|
with self._connection() as db:
|
|
rows = db.execute("SELECT d.body,s.state,s.approved_at FROM dispositions d "
|
|
"JOIN presentations p ON p.id=d.presentation_id "
|
|
"LEFT JOIN submissions s ON s.disposition_id=d.id "
|
|
"WHERE p.memo_id=? ORDER BY d.rowid", (memo_id,)).fetchall()
|
|
result = []
|
|
for row in rows:
|
|
d = disposition_from(json.loads(row["body"]))
|
|
if d.actor.sub == subject:
|
|
result.append((d, {"state": row["state"], "approved_at": row["approved_at"]} if row["state"] else None))
|
|
return result
|
|
|
|
def presentation_content(self, presentation_id):
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT content FROM evidence WHERE class=? AND "
|
|
"json_extract(envelope,'$.data.presentation_id')=? ORDER BY rowid LIMIT 1",
|
|
(EventClass.PRESENTATION.value, presentation_id)).fetchone()
|
|
if row is None:
|
|
raise EvidenceUnavailable("informed-decision cannot produce the presentation content")
|
|
return json.loads(row[0])
|
|
|
|
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 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:
|
|
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():
|
|
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<? GROUP BY class", (start, end))
|
|
return {row[0]: row[1] for row in rows}
|
|
|
|
def backup(self, destination: str | Path):
|
|
if Path(destination).exists() or Path(destination).is_symlink():
|
|
raise Conflict("backup destination must not already exist")
|
|
target = Store(destination)
|
|
with self._connection() as source, target._connection() as dest:
|
|
source.backup(dest)
|
|
|
|
|
|
def bounded_window(since, until):
|
|
result = []
|
|
for value in (since, until):
|
|
if not isinstance(value, str):
|
|
raise ValueError("an explicit bounded time window is required")
|
|
at = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if at.tzinfo is None:
|
|
raise ValueError("window must carry its timezone")
|
|
result.append(at.astimezone(timezone.utc).isoformat(timespec="microseconds"))
|
|
if result[0] >= result[1]:
|
|
raise ValueError("window must be increasing")
|
|
return tuple(result)
|