tegwick
83849b75d4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
531 lines
30 KiB
Python
531 lines
30 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 = 2
|
|
_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, 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;""")
|
|
|
|
@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")
|
|
_text(media_type, "media type")
|
|
digest = "sha256:" + hashlib.sha256(content).hexdigest()
|
|
with self._transaction() as db:
|
|
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))
|
|
return digest
|
|
|
|
def save_memo(self, memo: Memo):
|
|
_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")
|
|
with self._transaction() as db:
|
|
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))
|
|
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 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 queue_heartbeats(self, now=None):
|
|
now = time.time() if now is None else now
|
|
count = 0
|
|
with self._transaction() as db:
|
|
for event_class, interval in HEARTBEAT_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)
|