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

Start durable headless decision lifecycle foundation

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
This commit is contained in:
tegwick 2026-10-04 17:41:14 +02:00
parent 9289c764c3
commit f7094f8313
8 changed files with 553 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

10
SCOPE.md
View file

@ -4,15 +4,19 @@
> direction belongs in `INTENT.md`; the current stage belongs in `GOAL.md`;
> current work and gates belong in `workplans/`.
## Requested expansion — 2026-10-04 (not implemented)
## Lifecycle expansion — 2026-10-04 (first local persistence slice)
The product now targets the full decision lifecycle for agents and humans with
an API foundation and a UI at `decisions.coulomb.social`. See
`docs/specs/DecisionLifecycleService.md` and `INFD-WP-0007`. Existing memo,
presentation, disposition, storage, review and intake modules are reusable
building blocks; they are not a complete headless lifecycle API. Research,
evaluation, collaborative authoring, audience-scoped publication and revision
workflows still need implementation. Frictionless intake has local implementation
evaluation, collaborative authoring and audience-scoped publication still need
service/UI implementation. The first internal `LifecycleRepository` now supports
registration, immutable revisions, attributed research sources/evaluations/
comments and event retrieval using the existing Memo and Store (schema 4). Its
local event queue is durable; external delivery and authenticated API exposure
are not implemented. See `docs/specs/LifecycleFoundationContract.md`. Frictionless intake has local implementation
but its live delivery remains under `INFD-WP-0005`. The dated deployment snapshot
below is historical, not a fresh observation of production.

6
WORK-RECORDS.md
View file

@ -12,7 +12,7 @@
| workplan | INFD-WP-0003 | finished | — | workplans/INFD-WP-0003-my-decisions-overview.md |
| workplan | INFD-WP-0005 | blocked | — | workplans/INFD-WP-0005-frictionless-decision-intake.md |
| workplan | INFD-WP-0006 | finished | — | workplans/INFD-WP-0006-fresh-review-authentication.md |
| workplan | INFD-WP-0007 | proposed | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| workplan | INFD-WP-0007 | active | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-ADHOC-2026-10-03-T01 | done | — | workplans/ADHOC-2026-10-03.md |
| task | INFD-WP-ADHOC-2026-10-03-T02 | done | — | workplans/ADHOC-2026-10-03.md |
| task | INFD-WP-0003-T01 | done | — | workplans/INFD-WP-0003-my-decisions-overview.md |
@ -32,8 +32,8 @@
| task | INFD-WP-0006-T01 | done | — | workplans/INFD-WP-0006-fresh-review-authentication.md |
| task | INFD-WP-0006-T02 | done | — | workplans/INFD-WP-0006-fresh-review-authentication.md |
| task | INFD-WP-0007-T01 | done | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T02 | todo | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T03 | todo | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T02 | progress | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T03 | progress | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T04 | todo | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T05 | todo | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |
| task | INFD-WP-0007-T06 | todo | — | workplans/INFD-WP-0007-decision-lifecycle-service.md |

92
docs/specs/LifecycleFoundationContract.md Normal file
View file

@ -0,0 +1,92 @@
# Lifecycle foundation: first implementation contract
2026-10-04; `INFD-WP-0007-T02/T03`. This is the internal persistence slice of
`DecisionLifecycleService.md`, not a public API or completed application service.
## Implemented boundary
`LifecycleRepository` depends on the existing `Store`, memo and disposition
types. It imports no UI, HTTP runtime or provider client. It registers complete,
unbound version-one memos and creates immutable successor revisions. The existing
Decision Memo is reused without schema changes. Incomplete authoring drafts and
conversion of legacy/approval-bound records remain future work.
Each call carries `Attribution(tenant, Actor(subject, kind))` from its trusted
caller. This is audit attribution, not an authentication assertion. Like `Store`,
the repository is an internal persistence interface: the future application/API
layer must authenticate callers and enforce current access-engine authorization
before invoking it. Supplying a tenant string is not an access grant. No route
or runtime configuration exposes these operations in this increment.
## Commands and reads
| Method | Contract |
| --- | --- |
| `register(context, memo, key=...)` | Version 1, no approval binding or seal; rejects ids already present in the memo store; returns durable event id |
| `revise(context, memo, expected_version=..., summary=..., key=...)` | Exact current revision precondition and next version; nonempty change summary; returns durable event id |
| `annotate(context, memo_id, version, kind=..., body=..., key=...)` | Append source/evaluation/comment to an existing exact revision; returns durable event id |
| `retrieve(tenant, memo_id, version=None)` | Retrieve current or historical memo only within its registered tenant |
| `events(tenant, memo_id, after=0, limit=100)` | Ordered cursor page of attributed immutable events, maximum 100 |
Registration returns the memo's existing caller-supplied id through retrieval;
generated public ids and canonical URLs are deferred to the API adapter. Memo ids
remain globally unique because existing review URLs and storage use that key.
An existing imported memo cannot be taken over by registration, even if its bytes
match. A tenant mismatch produces the same unavailable error as a missing record.
This is partition enforcement, not a replacement for per-record authorization.
Idempotency scope is tenant + actor kind/subject + operation + key. A SHA-256 of
canonical command JSON detects content conflicts, including changed revision
preconditions. Exact retries return the original event id even after subsequent
revisions. Different authors must use concurrency preconditions; matching content
alone never makes a stale edit succeed. The command bound is 256 KiB.
All writes use one SQLite transaction for memo content, immutable lifecycle event,
pending-delivery row and idempotency receipt. An event queue failure rolls back
the content mutation. Events contain actor, tenant, server timestamp, record and
revision; they do not record an authorization verdict or approval validity.
## Companion records
* Comment: `text`, optional `locator`. This is discussion, not a disposition.
* Source: `reference`, timezone-qualified `retrieved_at`, SHA-256 `digest`,
`summary`. These are supplied research provenance, not a claim that the service
fetched or verified the source. This slice performs no network fetching.
* Evaluation: `assessment`, `source_ids`, `uncertainty`. Referenced source event
ids must belong to this record, tenant and revision. Up to 100 references are
accepted. Empty references are allowed for an explicitly uncited assessment.
Fields are closed and text is bounded at 32,768 characters per field. Annotations
cannot silently enter `view_hash`. Promoting evidence into the commitment requires
a new memo revision using the existing packet/binding rules. Research comments
from agents never go through or weaken the human disposition guards.
## Migration, evidence and rollout
Store schema 4 adds `lifecycle_records`, `lifecycle_events`,
`lifecycle_operations` and `lifecycle_pending`; migrations from versions 1–3
preserve existing memo, presentation, disposition and evidence rows. Historic
records are not automatically registered or assigned a tenant. Existing store
backup includes lifecycle content and pending rows. Old code refuses schema 4;
rollback therefore needs restoration of a pre-migration backup and explicit
handling of intervening writes, not simply an old image against the new database.
Memo persistence is extracted into a shared transaction helper without changing
legacy `save_memo` behavior. Existing stale-presentation guards see the same memo
versions. Approval-bound/sealed revision handling requires the later review
integration; this repository refuses those lifecycle revision commands.
Pending lifecycle events are durable local obligations. They are not emitted as
invented Audit Core event types or reported delivered. The existing review audit
queue remains unchanged. Before exposure, T02/T03 must settle receiver admission,
payload/privacy, delivery retries and readiness accounting for the new queue.
The existing readiness check does not account for it yet.
## Remaining contracts
T02 remains in progress: authenticated service ports, versioned HTTP schemas,
publication audience policy, incomplete drafts, non-cluster workload identity,
legacy record enrollment, source fetching/retention, independent evidence delivery
and non-approval binding integration. T03 is started with this persistence slice;
it is not complete. T04/T05 continue to own HTTP and UI exposure. Native runtime
intake and admission remain with `INFD-WP-0005`.

210
informed_decision/lifecycle.py Normal file
View file

@ -0,0 +1,210 @@
"""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):
event_id = str(uuid.uuid4())
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) -> 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")
scope, digest = self._request(context, "register", key, memo)
with self.store._transaction() as db:
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")
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})
def revise(self, context: Attribution, memo: Memo, *, expected_version: int,
summary: str, key: str) -> 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")
scope, digest = self._request(context, "revise", key,
{"memo": memo, "expected_version": expected_version, "summary": summary})
with self.store._transaction() as db:
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")
self.store._save_memo(db, memo)
return self._append(db, scope, digest, memo, "revised", {"summary": summary})
def annotate(self, context: Attribution, memo_id: str, version: int, *, kind: str,
body: dict, key: str) -> 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:
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)
@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 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]

41
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 = 3
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));
@ -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, 2, SCHEMA_VERSION):
elif version not in (1, 2, 3, SCHEMA_VERSION):
raise StoreError("unsupported evidence schema version")
if version in (0, 1):
db.executescript("""BEGIN IMMEDIATE;
@ -126,6 +126,11 @@ class Store:
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)
@ -162,26 +167,30 @@ class Store:
return digest
def save_memo(self, memo: Memo):
with self._transaction() as db:
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")
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))
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:

186
tests/test_lifecycle.py Normal file
View file

@ -0,0 +1,186 @@
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
import sqlite3
import subprocess
import sys
import pytest
from informed_decision.disposition import Actor, ActorKind, DispositionRefused, Verb
from informed_decision.lifecycle import Attribution, LifecycleRepository
from informed_decision.provenance import Claim, Route
from informed_decision.store import Conflict, EvidenceUnavailable, Store
from test_skeleton import make_memo
AGENT = Attribution("tenant:one", Actor("researcher", ActorKind.AGENT))
HUMAN = Attribution("tenant:one", Actor("reviewer", ActorKind.PERSON))
@pytest.fixture
def workspace(tmp_path):
private = tmp_path / "private"
private.mkdir(mode=0o700)
store = Store(private / "store.sqlite")
memo = make_memo(approval_id=None, packet=(), highlights=())
return LifecycleRepository(store), memo
def test_registration_replay_survives_restart_and_backup(workspace):
repo, memo = workspace
receipt = repo.register(AGENT, memo, key="register")
reopened = LifecycleRepository(Store(repo.store.path))
assert reopened.register(AGENT, memo, key="register") == receipt
assert reopened.retrieve(AGENT.tenant, memo.id) == memo
backup = repo.store.path.parent / "backup.sqlite"
repo.store.backup(backup)
assert LifecycleRepository(Store(backup)).events(AGENT.tenant, memo.id) == repo.events(AGENT.tenant, memo.id)
with repo.store._connection() as db:
assert db.execute("SELECT event_id FROM lifecycle_pending").fetchone()[0] == receipt
assert db.execute("SELECT count(*) FROM submissions").fetchone()[0] == 0
def test_replay_collision_and_existing_memo_takeover_refused(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
with pytest.raises(Conflict):
repo.register(AGENT, replace(memo, brief="different"), key="register")
with pytest.raises(Conflict):
repo.register(HUMAN, memo, key="register")
imported = replace(memo, id="imported")
repo.store.save_memo(imported)
with pytest.raises(Conflict):
repo.register(AGENT, imported, key="import")
def test_idempotency_is_scoped_by_tenant_and_actor(workspace):
repo, memo = workspace
other = Attribution("tenant:two", AGENT.actor)
for context, ident in [(AGENT, "one"), (HUMAN, "two"), (other, "three")]:
repo.register(context, replace(memo, id=ident), key="same-key")
with pytest.raises(EvidenceUnavailable):
repo.retrieve(other.tenant, "one")
with pytest.raises(EvidenceUnavailable):
repo.events(other.tenant, "one")
with pytest.raises(EvidenceUnavailable):
repo.annotate(other, "one", 1, kind="comment", body={"text": "leak"}, key="comment")
def test_concurrent_revision_has_one_winner_and_preserves_history(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
def edit(label):
try:
return repo.revise(HUMAN, memo.next_version(brief=label), expected_version=1, summary=label, key=label)
except Conflict:
return None
with ThreadPoolExecutor(max_workers=2) as pool:
outcomes = list(pool.map(edit, ["first", "second"]))
assert sum(outcome is not None for outcome in outcomes) == 1
assert repo.retrieve(AGENT.tenant, memo.id, 1) == memo
assert repo.retrieve(AGENT.tenant, memo.id).version == 2
assert len(repo.events(AGENT.tenant, memo.id)) == 2
def test_revision_replay_after_later_edits_returns_original_receipt(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
v2 = memo.next_version(brief="updated")
receipt = repo.revise(AGENT, v2, expected_version=1, summary="updated", key="revision")
repo.revise(AGENT, v2.next_version(brief="again"), expected_version=2, summary="again", key="again")
assert repo.revise(AGENT, v2, expected_version=1, summary="updated", key="revision") == receipt
@pytest.mark.parametrize("operation", ["register", "revise", "comment"])
def test_pending_event_failure_rolls_back_entire_mutation(workspace, operation):
repo, memo = workspace
if operation != "register":
repo.register(AGENT, memo, key="register")
with repo.store._connection() as db:
db.execute("CREATE TRIGGER fail_pending BEFORE INSERT ON lifecycle_pending "
"BEGIN SELECT RAISE(ABORT,'injected failure'); END")
with pytest.raises(sqlite3.IntegrityError):
if operation == "register":
repo.register(AGENT, memo, key="register")
elif operation == "revise":
repo.revise(AGENT, memo.next_version(), expected_version=1, summary="change", key="revision")
else:
repo.annotate(AGENT, memo.id, 1, kind="comment", body={"text": "hello"}, key="comment")
if operation == "register":
with pytest.raises(EvidenceUnavailable):
repo.store.memo(memo.id)
else:
assert repo.retrieve(AGENT.tenant, memo.id).version == 1
assert len(repo.events(AGENT.tenant, memo.id)) == 1
def test_sources_evaluation_comments_are_revision_anchored_and_paginated(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
source = repo.annotate(AGENT, memo.id, 1, kind="source", key="source", body={
"reference": "urn:report:1", "retrieved_at": "2026-10-04T10:00:00Z",
"digest": "sha256:" + "a" * 64, "summary": "Source supplied by researcher"})
evaluation = {"assessment": "Alternative A is supported", "source_ids": [source],
"uncertainty": "Costs have not been independently checked"}
repo.annotate(AGENT, memo.id, 1, kind="evaluation", body=evaluation, key="evaluation")
comment = repo.annotate(HUMAN, memo.id, 1, kind="comment", body={"text": "Check the costs"}, key="comment")
repo.revise(AGENT, memo.next_version(), expected_version=1, summary="next", key="next")
with pytest.raises(EvidenceUnavailable):
repo.annotate(AGENT, memo.id, 2, kind="evaluation", body=evaluation, key="evaluation-v2")
first = repo.events(AGENT.tenant, memo.id, limit=2)
following = repo.events(AGENT.tenant, memo.id, after=first[-1]["sequence"])
assert [row["kind"] for row in first + following] == ["registered", "source", "evaluation", "comment", "revised"]
assert following[1]["id"] == comment and following[1]["version"] == 1
assert following[0]["actor"]["kind"] == "agent"
assert following[1]["actor"]["kind"] == "person"
def test_agent_comment_does_not_grant_disposition_capability(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
repo.annotate(AGENT, memo.id, 1, kind="comment", body={"text": "Advisory only"}, key="comment")
presentation = repo.store.present(memo.id, principal_sub=AGENT.actor.sub,
tenant=Claim("tenant:platform", Route.DIRECTORY),
principal_type=Claim("agent", Route.AUTHENTICATION))
with pytest.raises(DispositionRefused, match="G_NOAGENT"):
repo.store.record_disposition(presentation.id, AGENT.actor, Verb.ACCEPT, operation_id="agent-accept")
def test_revision_invalidates_existing_presentation(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
presentation = repo.store.present(memo.id, principal_sub=HUMAN.actor.sub,
tenant=Claim("tenant:platform", Route.DIRECTORY),
principal_type=Claim("human", Route.AUTHENTICATION))
repo.revise(HUMAN, memo.next_version(brief="new"), expected_version=1, summary="new", key="new")
with pytest.raises(DispositionRefused, match="G_PRES"):
repo.store.record_disposition(presentation.id, HUMAN.actor, Verb.ACCEPT, operation_id="stale")
@pytest.mark.parametrize("kind,body", [
("accept", {"text": "yes"}), ("comment", {"text": "yes", "principal_type": "human"}),
("evaluation", {"assessment": "yes", "source_ids": "not-a-list", "uncertainty": "none"}),
("source", {"reference": "ref", "retrieved_at": "yesterday", "digest": "bad", "summary": "test"}),
])
def test_malformed_companion_records_rejected(workspace, kind, body):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
with pytest.raises(ValueError):
repo.annotate(AGENT, memo.id, 1, kind=kind, body=body, key="bad")
assert len(repo.events(AGENT.tenant, memo.id)) == 1
def test_lifecycle_history_is_append_only(workspace):
repo, memo = workspace
repo.register(AGENT, memo, key="register")
with repo.store._connection() as db:
for table in ("lifecycle_records", "lifecycle_events", "lifecycle_operations"):
with pytest.raises(sqlite3.IntegrityError):
db.execute(f"DELETE FROM {table}")
def test_foundation_imports_no_ui_or_http_runtime():
result = subprocess.run([sys.executable, "-c", "import sys; import informed_decision.lifecycle; "
"assert not {'informed_decision.ui', 'informed_decision.web', "
"'informed_decision.runtime'} & sys.modules.keys()"], capture_output=True, text=True)
assert result.returncode == 0, result.stderr

9
tests/test_review_runtime.py
View file

@ -12,13 +12,16 @@ from test_durable_store import storage,present
from test_durable_component import receiver
@pytest.mark.parametrize("version", [1, 2])
@pytest.mark.parametrize("version", [1, 2, 3])
def test_schema_upgrade_preserves_existing_evidence(storage, version):
store,memo=storage;p=present(store,memo)
before=store.evidence()
with sqlite3.connect(store.path) as db:
db.execute('DROP TABLE intake_deliveries')
db.execute('DROP TABLE intake_ledger')
for table in ('lifecycle_pending', 'lifecycle_operations', 'lifecycle_events', 'lifecycle_records'):
db.execute(f'DROP TABLE {table}')
if version < 3:
db.execute('DROP TABLE intake_deliveries')
db.execute('DROP TABLE intake_ledger')
if version == 1:
db.execute('DROP TABLE policy_observations')
db.execute(f'PRAGMA user_version={version}')

29
workplans/INFD-WP-0007-decision-lifecycle-service.md
View file

@ -4,7 +4,7 @@ type: workplan
title: "Decision lifecycle service — headless foundation and human UI"
domain: infotech
repo: informed-decision
status: proposed
status: active
owner: codex
topic_slug: netkingdom
created: "2026-10-04"
@ -21,8 +21,9 @@ state_hub_workstream_id: "2b3e04ed-2137-584f-a6c2-f1abb29636fa"
Contract: `docs/specs/DecisionLifecycleService.md`. This extends the existing
memo and intake implementation. `INFD-WP-0005` continues to own intake runtime,
attribute admission, forwarding and live acceptance. Proposed status reflects
the still-open API/identity/publication contracts, not a deployed capability.
attribute admission, forwarding and live acceptance. Implementation started on
2026-10-04 at the founder's request. API/identity/publication contracts remain
open; the first persistence slice is local and not exposed by a service route.
## Record product direction and architecture
@ -41,7 +42,7 @@ boundary, revision/evidence semantics and acceptance without per-decision deploy
```task
id: INFD-WP-0007-T02
status: todo
status: progress
priority: high
state_hub_task_id: "cde0c932-afce-5bec-8cee-a2f377955b9f"
```
@ -57,7 +58,7 @@ and current review URLs. Produce a reviewed migration and compatibility plan.
```task
id: INFD-WP-0007-T03
status: todo
status: progress
priority: high
depends_on: [INFD-WP-0007-T02]
state_hub_task_id: "2a206bc2-cb92-539d-9014-9ba71f70277c"
@ -130,3 +131,21 @@ After explicit approval of the concrete rollout, deploy and obtain native
agent-to-human-to-agent evidence at decisions.coulomb.social. Only the human
performs acceptance. Update SCOPE from observations and hand any residuals into
live work records before marking this plan finished.
## First implementation slice — 2026-10-04
`docs/specs/LifecycleFoundationContract.md` records the implemented T02 storage
contract and remaining service/API contracts. T03 begins against that bounded
contract while the broader T02 work continues. `informed_decision/lifecycle.py`
provides registration, immutable revisions, attributed sources/evaluations/
comments and event retrieval on the existing Memo and Store. Schema 4 retains
mutation events and pending delivery atomically with memo writes and idempotency
receipts. Tests cover concurrent revisions, restart/backup, tenant partitioning,
rollback, replay, provenance references, schema upgrades and existing human/stale
presentation guards. Neither task is complete: application authorization,
publication, HTTP/UI and admitted evidence delivery remain in their live tasks.
No production route or deployment was changed.
Validation: `uv run --extra dev python -m pytest -q` — 480 passed, 72 skipped.
The system interpreter had dependency-related failures; the project-managed
environment passes. `git diff --check` passes.