tegwick
4f546ae3b1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
315 lines
17 KiB
Python
315 lines
17 KiB
Python
"""Internal lifecycle persistence, independent of browser and HTTP adapters.
|
|
|
|
Like Store, this repository is NOT an authenticated service. Its caller must
|
|
authenticate identity and obtain current access-engine entitlement before any
|
|
read or write. Attribution is imported context, never proof of authentication.
|
|
No production route exposes this repository yet.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import uuid
|
|
|
|
from .disposition import Actor, ActorKind
|
|
from .memo import Memo
|
|
from .records import dumps
|
|
from .store import Conflict, EvidenceUnavailable, Store, _now, _text
|
|
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE lifecycle_records (
|
|
memo_id TEXT PRIMARY KEY, tenant TEXT NOT NULL,
|
|
registered_at TEXT NOT NULL);
|
|
CREATE TABLE lifecycle_events (
|
|
sequence INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL UNIQUE,
|
|
tenant TEXT NOT NULL, memo_id TEXT NOT NULL REFERENCES lifecycle_records(memo_id),
|
|
version INTEGER NOT NULL, kind TEXT NOT NULL, actor TEXT NOT NULL,
|
|
at TEXT NOT NULL, body TEXT NOT NULL,
|
|
FOREIGN KEY(memo_id,version) REFERENCES memos(id,version));
|
|
CREATE INDEX lifecycle_events_record ON lifecycle_events(tenant,memo_id,sequence);
|
|
CREATE TABLE lifecycle_operations (
|
|
tenant TEXT NOT NULL, actor TEXT NOT NULL, operation TEXT NOT NULL,
|
|
key TEXT NOT NULL, digest TEXT NOT NULL,
|
|
event_id TEXT NOT NULL REFERENCES lifecycle_events(id),
|
|
PRIMARY KEY(tenant,actor,operation,key));
|
|
CREATE TABLE lifecycle_pending (
|
|
event_id TEXT PRIMARY KEY REFERENCES lifecycle_events(id));
|
|
"""
|
|
# Pending events are intentionally not sent to audit-core under invented types.
|
|
# A later delivery adapter must retain these until a receiver contract is admitted.
|
|
for _table in ("lifecycle_records", "lifecycle_events", "lifecycle_operations"):
|
|
for _operation in ("UPDATE", "DELETE"):
|
|
SCHEMA += (f"CREATE TRIGGER immutable_{_table}_{_operation} BEFORE {_operation} ON {_table} "
|
|
"BEGIN SELECT RAISE(ABORT,'immutable lifecycle record'); END;\n")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Attribution:
|
|
tenant: str
|
|
actor: Actor
|
|
|
|
def __post_init__(self):
|
|
_text(self.tenant, "tenant")
|
|
_text(self.actor.sub, "actor")
|
|
if not isinstance(self.actor.kind, ActorKind):
|
|
raise ValueError("invalid actor kind")
|
|
|
|
|
|
class LifecycleRepository:
|
|
"""Tenant-scoped storage primitives; authorization belongs to the caller."""
|
|
|
|
def __init__(self, store: Store):
|
|
self.store = store
|
|
|
|
@staticmethod
|
|
def _record(db, tenant, memo_id):
|
|
if not db.execute("SELECT 1 FROM lifecycle_records WHERE tenant=? AND memo_id=?",
|
|
(tenant, memo_id)).fetchone():
|
|
raise EvidenceUnavailable("lifecycle record unavailable")
|
|
|
|
@staticmethod
|
|
def _request(context, operation, key, payload):
|
|
_text(key, "idempotency key")
|
|
actor = dumps(context.actor)
|
|
encoded = dumps(payload).encode()
|
|
if len(encoded) > 256 * 1024:
|
|
raise ValueError("lifecycle command too large")
|
|
return (context.tenant, actor, operation, key), hashlib.sha256(encoded).hexdigest()
|
|
|
|
@staticmethod
|
|
def _replay(db, scope, digest):
|
|
row = db.execute("SELECT digest,event_id FROM lifecycle_operations "
|
|
"WHERE tenant=? AND actor=? AND operation=? AND key=?", scope).fetchone()
|
|
if row:
|
|
if row["digest"] != digest:
|
|
raise Conflict("idempotency key has different command content")
|
|
return row["event_id"]
|
|
return None
|
|
|
|
@staticmethod
|
|
def _append(db, scope, digest, memo, kind, body, policy_id=None):
|
|
event_id = str(uuid.uuid4())
|
|
if policy_id is not None:
|
|
body = {**body, "policy_observation_id": policy_id}
|
|
db.execute("INSERT INTO lifecycle_events "
|
|
"(id,tenant,memo_id,version,kind,actor,at,body) VALUES (?,?,?,?,?,?,?,?)",
|
|
(event_id, scope[0], memo.id, memo.version, kind, scope[1], _now(), dumps(body)))
|
|
db.execute("INSERT INTO lifecycle_pending VALUES (?)", (event_id,))
|
|
db.execute("INSERT INTO lifecycle_operations VALUES (?,?,?,?,?,?)", (*scope, digest, event_id))
|
|
return event_id
|
|
|
|
def register(self, context: Attribution, memo: Memo, *, key: str, guard=None, documents=None) -> str:
|
|
"""Register a complete, unbound version-one memo without creating approval."""
|
|
if memo.version != 1 or memo.sealed or memo.approval_id is not None:
|
|
raise ValueError("registration requires an unbound version-one memo")
|
|
from .lifecycle_commands import decode_documents
|
|
documents = [] if documents is None else documents
|
|
decoded = decode_documents(documents, memo)
|
|
payload = {"memo": memo, "documents": documents} if documents else memo
|
|
scope, digest = self._request(context, "register", key, payload)
|
|
with self.store._transaction() as db:
|
|
policy_id = guard(db) if guard is not None else None
|
|
replay = self._replay(db, scope, digest)
|
|
if replay:
|
|
return replay
|
|
# Never take over an existing intake/imported memo, even with identical content.
|
|
if db.execute("SELECT 1 FROM memos WHERE id=?", (memo.id,)).fetchone():
|
|
raise Conflict("memo identifier already exists")
|
|
if {item.hash for item in memo.packet} - {item[0] for item in decoded}:
|
|
raise EvidenceUnavailable("registration requires packet content in the command")
|
|
for document in decoded:
|
|
self.store._put_document(db, *document)
|
|
self.store._save_memo(db, memo)
|
|
db.execute("INSERT INTO lifecycle_records VALUES (?,?,?)", (memo.id, context.tenant, _now()))
|
|
return self._append(db, scope, digest, memo, "registered", {"memo_version": memo.version}, policy_id)
|
|
|
|
def revise(self, context: Attribution, memo: Memo, *, expected_version: int,
|
|
summary: str, key: str, guard=None, documents=None) -> str:
|
|
"""Compare-and-append; retain existing Store guards on outstanding entries."""
|
|
if type(expected_version) is not int or expected_version < 1:
|
|
raise ValueError("expected version must be a positive integer")
|
|
if not isinstance(summary, str) or not summary.strip() or len(summary) > 4096:
|
|
raise ValueError("revision summary required, at most 4096 characters")
|
|
if memo.version != expected_version + 1 or memo.sealed or memo.approval_id is not None:
|
|
raise ValueError("revision requires the next unbound memo version")
|
|
from .lifecycle_commands import decode_documents
|
|
documents = [] if documents is None else documents
|
|
decoded = decode_documents(documents, memo)
|
|
payload = {"memo": memo, "expected_version": expected_version, "summary": summary}
|
|
if documents:
|
|
payload["documents"] = documents
|
|
scope, digest = self._request(context, "revise", key, payload)
|
|
with self.store._transaction() as db:
|
|
policy_id = guard(db) if guard is not None else None
|
|
self._record(db, context.tenant, memo.id)
|
|
replay = self._replay(db, scope, digest)
|
|
if replay:
|
|
return replay
|
|
current = self.store._memo(db, memo.id)
|
|
if current.version != expected_version:
|
|
raise Conflict("memo changed since it was retrieved")
|
|
if current.sealed or current.approval_id is not None:
|
|
raise Conflict("bound memo requires the review integration path")
|
|
available = {item.hash for item in current.packet} | {item[0] for item in decoded}
|
|
if {item.hash for item in memo.packet} - available:
|
|
raise EvidenceUnavailable("revision requires new packet content in the command")
|
|
for document in decoded:
|
|
self.store._put_document(db, *document)
|
|
self.store._save_memo(db, memo)
|
|
return self._append(db, scope, digest, memo, "revised", {"summary": summary}, policy_id)
|
|
|
|
def annotate(self, context: Attribution, memo_id: str, version: int, *, kind: str,
|
|
body: dict, key: str, guard=None) -> str:
|
|
"""Append a source, advisory evaluation or discussion; never a disposition."""
|
|
self._validate_annotation(kind, body)
|
|
if type(version) is not int or version < 1:
|
|
raise ValueError("version must be a positive integer")
|
|
scope, digest = self._request(context, kind, key,
|
|
{"memo_id": memo_id, "version": version, "body": body})
|
|
with self.store._transaction() as db:
|
|
policy_id = guard(db) if guard is not None else None
|
|
self._record(db, context.tenant, memo_id)
|
|
replay = self._replay(db, scope, digest)
|
|
if replay:
|
|
return replay
|
|
memo = self.store._memo(db, memo_id, version)
|
|
if kind == "evaluation":
|
|
for source in body["source_ids"]:
|
|
if not db.execute("SELECT 1 FROM lifecycle_events WHERE id=? AND tenant=? "
|
|
"AND memo_id=? AND version=? AND kind='source'",
|
|
(source, context.tenant, memo_id, version)).fetchone():
|
|
raise EvidenceUnavailable("evaluation source unavailable for this revision")
|
|
return self._append(db, scope, digest, memo, kind, body, policy_id)
|
|
|
|
def publish(self, context, memo_id, version, *, audience, key, guard=None):
|
|
audience = publication_audience(audience)
|
|
if type(version) is not int or version < 1:
|
|
raise ValueError("positive revision required")
|
|
scope, digest = self._request(context, "publish", key,
|
|
{"memo_id": memo_id, "version": version, "audience": audience})
|
|
with self.store._transaction() as db:
|
|
policy_id = guard(db) if guard is not None else None
|
|
self._record(db, context.tenant, memo_id)
|
|
replay = self._replay(db, scope, digest)
|
|
if replay:
|
|
return replay
|
|
memo = self.store._memo(db, memo_id)
|
|
if memo.version != version:
|
|
raise Conflict("publish requires the current revision")
|
|
return self._append(db, scope, digest, memo, "published", {"audience": audience}, policy_id)
|
|
|
|
def policy_context(self, tenant, memo_id, version=None):
|
|
with self.store._connection() as db:
|
|
return self._policy_context(db, tenant, memo_id, version)
|
|
|
|
@staticmethod
|
|
def _policy_context(db, tenant, memo_id, version=None):
|
|
row = db.execute("SELECT e.actor, (SELECT MAX(version) FROM memos WHERE id=l.memo_id) AS current_version "
|
|
"FROM lifecycle_records l JOIN lifecycle_events e ON e.memo_id=l.memo_id "
|
|
"WHERE l.tenant=? AND l.memo_id=? AND e.kind='registered'", (tenant, memo_id)).fetchone()
|
|
if row is None:
|
|
return {"registered": False}
|
|
published = db.execute("SELECT body FROM lifecycle_events WHERE tenant=? AND memo_id=? "
|
|
"AND version=? AND kind='published' ORDER BY sequence DESC LIMIT 1",
|
|
(tenant, memo_id, version if version is not None else row["current_version"])).fetchone()
|
|
revision = db.execute("SELECT actor FROM lifecycle_events WHERE tenant=? AND memo_id=? "
|
|
"AND version=? AND kind IN ('registered','revised') ORDER BY sequence DESC LIMIT 1",
|
|
(tenant, memo_id, version if version is not None else row["current_version"])).fetchone()
|
|
return {"registered": True, "registered_by": json.loads(row["actor"]),
|
|
"revision_by": json.loads(revision["actor"]) if revision else None,
|
|
"current_version": row["current_version"],
|
|
"published_audience": json.loads(published["body"])["audience"] if published else None}
|
|
|
|
def candidates(self, tenant, *, query="", after="", limit=100):
|
|
"""Internal bounded search candidates; callers must authorize each result."""
|
|
if (not isinstance(query, str) or len(query) > 200 or not isinstance(after, str)
|
|
or len(after) > 256 or type(limit) is not int or not 1 <= limit <= 100):
|
|
raise ValueError("invalid search parameters")
|
|
with self.store._connection() as db:
|
|
return [row[0] for row in db.execute(
|
|
"SELECT l.memo_id FROM lifecycle_records l JOIN memos m ON m.id=l.memo_id "
|
|
"WHERE l.tenant=? AND l.memo_id>? AND m.version=(SELECT MAX(version) FROM memos WHERE id=l.memo_id) "
|
|
"AND (instr(lower(l.memo_id),lower(?))>0 OR instr(lower(json_extract(m.body,'$.question')),lower(?))>0 "
|
|
"OR instr(lower(json_extract(m.body,'$.brief')),lower(?))>0) ORDER BY l.memo_id LIMIT ?",
|
|
(tenant, after, query, query, query, limit))]
|
|
|
|
@staticmethod
|
|
def _validate_annotation(kind, body):
|
|
fields = {
|
|
"comment": ({"text"}, {"text", "locator"}),
|
|
"source": ({"reference", "retrieved_at", "digest", "summary"},
|
|
{"reference", "retrieved_at", "digest", "summary"}),
|
|
"evaluation": ({"assessment", "source_ids", "uncertainty"},
|
|
{"assessment", "source_ids", "uncertainty"}),
|
|
}
|
|
if kind not in fields or not isinstance(body, dict):
|
|
raise ValueError("unsupported annotation")
|
|
required, allowed = fields[kind]
|
|
if not required <= body.keys() or body.keys() - allowed:
|
|
raise ValueError("invalid annotation fields")
|
|
for field, value in body.items():
|
|
if field == "source_ids":
|
|
if not isinstance(value, list) or len(value) > 100:
|
|
raise ValueError("source_ids must be a bounded list")
|
|
for source in value:
|
|
_text(source, "source id")
|
|
elif not isinstance(value, str) or not value.strip() or len(value) > 32768:
|
|
raise ValueError("annotation fields must be bounded non-empty text")
|
|
if kind == "source":
|
|
if not re.fullmatch(r"sha256:[a-f0-9]{64}", body["digest"]):
|
|
raise ValueError("invalid source digest")
|
|
try:
|
|
retrieved_at = datetime.fromisoformat(body["retrieved_at"].replace("Z", "+00:00"))
|
|
except ValueError:
|
|
raise ValueError("invalid source retrieval time") from None
|
|
if retrieved_at.tzinfo is None:
|
|
raise ValueError("source retrieval time requires timezone")
|
|
|
|
def retrieve(self, tenant: str, memo_id: str, version: int | None = None) -> Memo:
|
|
with self.store._connection() as db:
|
|
self._record(db, tenant, memo_id)
|
|
return self.store._memo(db, memo_id, version)
|
|
|
|
def packet(self, tenant, memo_id, version, index):
|
|
if type(version) is not int or version < 1 or type(index) is not int or index < 0:
|
|
raise ValueError("invalid packet reference")
|
|
with self.store._connection() as db:
|
|
self._record(db, tenant, memo_id)
|
|
memo = self.store._memo(db, memo_id, version)
|
|
if index >= len(memo.packet):
|
|
raise EvidenceUnavailable("packet unavailable")
|
|
row = db.execute("SELECT content FROM documents WHERE digest=?", (memo.packet[index].hash,)).fetchone()
|
|
if row is None:
|
|
raise EvidenceUnavailable("packet unavailable")
|
|
return bytes(row[0])
|
|
|
|
def events(self, tenant: str, memo_id: str, *, after: int = 0, limit: int = 100) -> list[dict]:
|
|
if type(after) is not int or after < 0 or type(limit) is not int or not 1 <= limit <= 100:
|
|
raise ValueError("invalid event page")
|
|
with self.store._connection() as db:
|
|
self._record(db, tenant, memo_id)
|
|
rows = db.execute("SELECT * FROM lifecycle_events WHERE tenant=? AND memo_id=? "
|
|
"AND sequence>? ORDER BY sequence LIMIT ?",
|
|
(tenant, memo_id, after, limit)).fetchall()
|
|
return [{**dict(row), "actor": json.loads(row["actor"]), "body": json.loads(row["body"])}
|
|
for row in rows]
|
|
|
|
|
|
def publication_audience(value):
|
|
if not isinstance(value, dict) or set(value) != {"subjects", "groups"}:
|
|
raise ValueError("explicit publication audience required")
|
|
result = {}
|
|
for name, items in value.items():
|
|
if not isinstance(items, list) or len(items) > 32:
|
|
raise ValueError("invalid publication audience")
|
|
for item in items:
|
|
_text(item, "audience member")
|
|
if not item.strip() or item == "*":
|
|
raise ValueError("explicit audience member required")
|
|
result[name] = sorted(set(items))
|
|
if not any(result.values()):
|
|
raise ValueError("publication audience must not be empty")
|
|
return result
|