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

Add protected headless lifecycle application service

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 18:25:21 +02:00
parent f7094f8313
commit b2a4b863a8
7 changed files with 439 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

8
SCOPE.md
View file

@ -15,8 +15,12 @@ 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
local event queue is durable. `LifecycleService` now wraps these operations
with fresh exact-policy observations, expiry checks inside write transactions,
and a verified human-session adapter. HTTP exposure, workload identity composition
and external event delivery remain unimplemented. See
`docs/specs/LifecycleFoundationContract.md` and
`docs/specs/LifecycleServiceContract.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.

7
docs/specs/LifecycleFoundationContract.md
View file

@ -1,7 +1,8 @@
# 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.
`DecisionLifecycleService.md`. The protected application interface built on this
slice is documented in `LifecycleServiceContract.md`; HTTP exposure is pending.
## Implemented boundary
@ -13,7 +14,7 @@ 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
the repository is an internal persistence interface: the 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.
@ -84,7 +85,7 @@ The existing readiness check does not account for it yet.
## Remaining contracts
T02 remains in progress: authenticated service ports, versioned HTTP schemas,
T02 remains in progress: production identity/policy composition, 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;

90
docs/specs/LifecycleServiceContract.md Normal file
View file

@ -0,0 +1,90 @@
# Protected lifecycle application service
2026-10-04; `INFD-WP-0007-T02/T03`. Implemented in
`informed_decision/lifecycle_service.py`, above `LifecycleRepository`.
This is a headless Python application interface. No HTTP route or production
composition enables it yet.
## Caller and authority
The identity adapter supplies an immutable `Caller`: tenant, actor subject/kind,
finite expiry and JSON identity attributes. This is trusted in-process context,
not a client-controlled payload or credential. `Caller.from_human_session`
imports the existing verified KeyCape session, preserving tenant provenance,
authentication-derived human identity, assurance, roles and groups. It carries
neither the browser token nor CSRF material into policy requests or stored events.
Agent adapters must authenticate the workload before constructing Caller. This
increment does not add an issuer, infer a tenant from a request body, treat a
claimed agent name as authentication, or authorize a new workload audience.
The existing intake TokenReview audience is not silently reused for this service.
Non-cluster/workload composition remains T02/T04 work.
The service consumes `PolicyClient` observations. It neither evaluates policy
nor falls back to local permissions. New proposed action names are
`lifecycle.register`, `lifecycle.revise`, `lifecycle.source`,
`lifecycle.evaluation`, `lifecycle.comment`, `lifecycle.read`, and
`lifecycle.events`. These are not added to the existing review-action allowlist;
review and human disposition contracts remain unchanged.
The request carries the authenticated subject, tenant, resource `memo:<id>`,
system `informed-decision`, type `decision-memo`, exact configured policy pin,
and context:
* `contract`: `informed-decision.lifecycle.v1`;
* `memo_version`: submitted/requested revision, or null for latest retrieval and
record-wide event retrieval;
* `command_digest`: SHA-256 of canonical command JSON, including mutation content,
expected revision and idempotency key, or the read/page parameters.
The command digest correlates the requested operation; it is not `view_hash`,
an approval binding digest or an authorization verdict. No presentation evidence
is supplied as an input to the decision it presents for. Advisory content is not
sent to the PDP in clear text. Identity attributes are supplied by trusted
adapters, not copied from memo content.
## Freshness, replay and evidence
Every operation, including an idempotent retry, obtains a new policy observation
and persists it using the existing Store. A deny, expired or mismatched response,
unsupported obligation or unavailable policy service prevents the operation.
The existing PolicyClient validates resource/subject/context, request digests,
policy pins, response lifetime and the supported effect.
Mutation commands snapshot supplied content before the network call. Under the
repository's write transaction, a guard rechecks caller and decision expiry and
the persisted observation before even returning an idempotent replay. New events
carry `body.policy_observation_id`; receipts replay their original event and
observation link while the fresh retry check remains in policy history. The
identity/policy guard does not perform network calls while holding the write lock.
Read authorization precedes repository lookup. Denied requests therefore cannot
distinguish missing and existing records through lookup errors. The service
rechecks identity and decision expiry after retrieving content and before returning
it. Repository tenant partition enforcement also applies. Record-wide event
access requires its own action; it does not inherit a memo-read grant.
The existing `decision_attributable: false` declaration remains in force. A
matched, recorded response is not proof of signed PDP attribution. New lifecycle
events still reside in the durable pending queue until an Audit Core contract
and delivery adapter are admitted.
## Error and integration contract
`LifecycleError` supplies bounded status/code pairs: 401 for expired identity,
403 for a policy deny, and 503 for unavailable or unusable policy evidence.
The repository retains `Conflict`, `EvidenceUnavailable` and validation errors;
the future HTTP adapter must map these to bounded envelopes without echoing
content, credentials or raw exception diagnostics.
No accept/publish endpoint is implied by this interface. Human acceptance stays
on the existing presentation/disposition path; publication needs an explicit
audience and separately authorized transition. The service supplies the shared
operations that subsequent HTTP and UI adapters will invoke.
Before enabling a listener, T02/T04 must provide: reviewed lifecycle PDP package
and action/context fixtures; admitted human/workload identity adapters; closed
HTTP schemas with canonical URLs and request limits; service admission and
publication audience contracts; lifecycle audit receiver admission and readiness
accounting. Test fixtures exercise the real PolicyClient against synthetic policy
transport. They are not evidence of a deployed lifecycle policy grant.

19
informed_decision/lifecycle.py
View file

@ -90,8 +90,10 @@ class LifecycleRepository:
return None
@staticmethod
def _append(db, scope, digest, memo, kind, body):
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)))
@ -99,12 +101,13 @@ class LifecycleRepository:
db.execute("INSERT INTO lifecycle_operations VALUES (?,?,?,?,?,?)", (*scope, digest, event_id))
return event_id
def register(self, context: Attribution, memo: Memo, *, key: str) -> str:
def register(self, context: Attribution, memo: Memo, *, key: str, guard=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")
scope, digest = self._request(context, "register", key, memo)
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
@ -113,10 +116,10 @@ class LifecycleRepository:
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})
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) -> str:
summary: str, key: str, guard=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")
@ -127,6 +130,7 @@ class LifecycleRepository:
scope, digest = self._request(context, "revise", key,
{"memo": memo, "expected_version": expected_version, "summary": summary})
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:
@ -137,10 +141,10 @@ class LifecycleRepository:
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})
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) -> 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:
@ -148,6 +152,7 @@ class LifecycleRepository:
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:
@ -159,7 +164,7 @@ class LifecycleRepository:
"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)
return self._append(db, scope, digest, memo, kind, body, policy_id)
@staticmethod
def _validate_annotation(kind, body):

142
informed_decision/lifecycle_service.py Normal file
View file

@ -0,0 +1,142 @@
"""Headless application service consuming fresh, exact external policy decisions.
Identity adapters supply Caller after authentication; no request body can mint
one. Runtime composition and lifecycle policy admission remain separate work.
"""
from dataclasses import dataclass
import hashlib
import json
import math
import time
import uuid
from .disposition import Actor, ActorKind
from .lifecycle import Attribution, LifecycleRepository
from .oidc import HumanSession
from .policy import PolicyError
from .provenance import Route
from .records import dumps, memo_from
from .store import _text
@dataclass(frozen=True)
class Caller:
attribution: Attribution
expires_at: float
attributes_json: str
def __post_init__(self):
if (type(self.expires_at) not in (int, float) or not math.isfinite(self.expires_at)
or not isinstance(self.attributes_json, str) or len(self.attributes_json) > 16384):
raise ValueError("invalid authenticated caller")
attributes = json.loads(self.attributes_json)
if not isinstance(attributes, dict):
raise ValueError("invalid caller attributes")
@classmethod
def from_human_session(cls, session: HumanSession):
if (not isinstance(session, HumanSession)
or session.principal_type.value != "human"
or session.principal_type.route is not Route.AUTHENTICATION
or session.tenant.route not in (Route.DIRECTORY, Route.REGISTRATION)):
raise ValueError("verified human identity required")
return cls(Attribution(session.tenant.value, Actor(session.subject, ActorKind.PERSON)),
session.expires_at, dumps({
"tenant_source": session.tenant.route.value,
"principal_type_source": session.principal_type.route.value,
"assurance": session.assurance,
"roles": list(session.roles), "groups": list(session.groups)}))
class LifecycleError(RuntimeError):
def __init__(self, status, code):
self.status, self.code = status, code
super().__init__(code)
class LifecycleService:
def __init__(self, repository: LifecycleRepository, policy, *, clock=time.time):
self.repository, self.policy, self.clock = repository, policy, clock
def _caller(self, caller):
if not isinstance(caller, Caller) or caller.expires_at <= self.clock():
raise LifecycleError(401, "identity_expired")
def _authorize(self, caller, memo_id, operation, payload):
self._caller(caller)
_text(memo_id, "memo id")
actor = caller.attribution.actor
tenant = caller.attribution.tenant
command = dumps(payload).encode()
if len(command) > 256 * 1024:
raise ValueError("lifecycle command too large")
request = {"id": str(uuid.uuid4()), "tenant": tenant,
"subject": {"id": actor.sub, "type": "human" if actor.kind is ActorKind.PERSON else actor.kind.value,
"tenant": tenant, "attributes": json.loads(caller.attributes_json)},
"action": "lifecycle." + operation,
"resource": {"id": "memo:" + memo_id, "type": "decision-memo",
"system": "informed-decision", "tenant": tenant},
"context": {"contract": "informed-decision.lifecycle.v1",
"memo_version": payload.get("memo").version if payload.get("memo") is not None else payload.get("version"),
"command_digest": "sha256:" + hashlib.sha256(command).hexdigest()},
"policy_version": self.policy.version}
observation = self.policy.check(request)
# Defend the application boundary even with an incorrectly composed client.
if json.loads(observation.request_json) != request:
raise LifecycleError(503, "policy_request_mismatch")
policy_id = self.repository.store.observe_policy(observation)
def current():
self._caller(caller)
try:
observation.require_current(self.clock())
except PolicyError:
raise LifecycleError(403 if observation.outcome == "policy_denied" else 503,
observation.outcome if observation.outcome != "allow" else "decision_expired") from None
def guard(db):
current()
recorded = db.execute("SELECT request,outcome,usable_until FROM policy_observations WHERE id=?",
(policy_id,)).fetchone()
if (recorded is None or recorded["request"] != observation.request_json
or recorded["outcome"] != "allow" or recorded["usable_until"] <= self.clock()):
raise LifecycleError(503, "policy_observation_unavailable")
return policy_id
current()
return current, guard
def register(self, caller, memo, *, key):
# Snapshot before the PDP call: mutable caller-owned objects cannot change
# the command between authorization and persistence.
memo = memo_from(json.loads(dumps(memo)))
_, guard = self._authorize(caller, memo.id, "register", {"memo": memo, "key": key})
return self.repository.register(caller.attribution, memo, key=key, guard=guard)
def revise(self, caller, memo, *, expected_version, summary, key):
memo = memo_from(json.loads(dumps(memo)))
_, guard = self._authorize(caller, memo.id, "revise", {
"memo": memo, "expected_version": expected_version, "summary": summary, "key": key})
return self.repository.revise(caller.attribution, memo, expected_version=expected_version,
summary=summary, key=key, guard=guard)
def annotate(self, caller, memo_id, version, *, kind, body, key):
body = json.loads(dumps(body))
self.repository._validate_annotation(kind, body)
_, guard = self._authorize(caller, memo_id, kind, {
"memo_id": memo_id, "version": version, "body": body, "key": key})
return self.repository.annotate(caller.attribution, memo_id, version, kind=kind,
body=body, key=key, guard=guard)
def retrieve(self, caller, memo_id, version=None):
current, _ = self._authorize(caller, memo_id, "read", {"version": version})
result = self.repository.retrieve(caller.attribution.tenant, memo_id, version)
current()
return result
def events(self, caller, memo_id, *, after=0, limit=100):
current, _ = self._authorize(caller, memo_id, "events", {"after": after, "limit": limit})
result = self.repository.events(caller.attribution.tenant, memo_id, after=after, limit=limit)
current()
return result

171
tests/test_lifecycle_service.py Normal file
View file

@ -0,0 +1,171 @@
from dataclasses import replace
import json
import time
import pytest
from informed_decision.lifecycle_service import Caller, LifecycleError, LifecycleService
from informed_decision.policy import Observation
from informed_decision.provenance import Claim, Route
from informed_decision.store import EvidenceUnavailable
from test_lifecycle import AGENT, workspace
from test_policy_client import human, policy, PolicyTransport
def caller():
return Caller(AGENT, time.time() + 600, '{"principal_type_source":"authentication-derived"}')
def service(workspace, transport=None, **kwargs):
repo, memo = workspace
return LifecycleService(repo, policy(transport), **kwargs), memo
def test_agent_and_human_use_shared_service_and_replay_requires_new_decision(workspace):
app, memo = service(workspace)
agent = caller()
event = app.register(agent, memo, key="new")
assert app.register(agent, memo, key="new") == event
assert len(app.policy.transport.calls) == 2
person = Caller.from_human_session(replace(human(), tenant=Claim(AGENT.tenant, Route.DIRECTORY)))
app.annotate(person, memo.id, 1, kind="comment", body={"text": "Consider alternative B"}, key="comment")
app.revise(agent, memo.next_version(brief="Alternative B assessed"), expected_version=1,
summary="Addressed review", key="revise")
assert app.retrieve(person, memo.id).version == 2
events = app.events(agent, memo.id)
assert len(events) == 3
assert events[1]["actor"]["kind"] == "person"
observations = app.repository.store.policy_observations()
assert all(row["decision_attributable"] == 0 for row in observations)
identifiers = {row["id"] for row in observations}
assert all(event["body"]["policy_observation_id"] in identifiers for event in events)
request = app.policy.transport.calls[0][3]
assert request["action"] == "lifecycle.register"
assert request["subject"]["type"] == "agent"
assert request["resource"]["tenant"] == AGENT.tenant
assert "view_hash" not in json.dumps(request)
assert "synthetic-browser-token" not in json.dumps(observations)
@pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"])
def test_no_nonallow_effect_can_register(workspace, effect):
transport = PolicyTransport()
transport.change = lambda data: data.update(effect=effect)
app, memo = service(workspace, transport)
with pytest.raises(LifecycleError):
app.register(caller(), memo, key="new")
with pytest.raises(EvidenceUnavailable):
app.repository.retrieve(AGENT.tenant, memo.id)
assert len(app.repository.store.policy_observations()) == 1
@pytest.mark.parametrize("mutate", [
lambda data: data["binding"].update(action="read"),
lambda data: data["binding"]["context"].update(command_digest="sha256:" + "0" * 64),
lambda data: data["provenance"].update(policy_package="unadmitted"),
lambda data: data.update(obligations=[{"type": "unknown"}]),
])
def test_policy_response_must_match_exact_lifecycle_request(workspace, mutate):
transport = PolicyTransport()
transport.change = mutate
app, memo = service(workspace, transport)
with pytest.raises(LifecycleError):
app.register(caller(), memo, key="new")
with pytest.raises(EvidenceUnavailable):
app.repository.store.memo(memo.id)
def test_revoked_permission_denies_replay_and_existing_and_missing_reads_equally(workspace):
app, memo = service(workspace)
agent = caller()
app.register(agent, memo, key="new")
app.policy.transport.change = lambda data: data.update(effect="deny")
operations = [lambda: app.register(agent, memo, key="new"),
lambda: app.retrieve(agent, memo.id), lambda: app.retrieve(agent, "missing"),
lambda: app.events(agent, memo.id),
lambda: app.annotate(agent, memo.id, 1, kind="comment", body={"text": "No"}, key="no")]
for operation in operations:
with pytest.raises(LifecycleError) as error:
operation()
assert error.value.status == 403
assert len(app.repository.events(AGENT.tenant, memo.id)) == 1
def test_identity_expiry_refuses_before_policy_call(workspace):
app, memo = service(workspace)
with pytest.raises(LifecycleError) as error:
app.register(replace(caller(), expires_at=1), memo, key="new")
assert error.value.status == 401
assert app.policy.transport.calls == []
@pytest.mark.parametrize("expiry", ["identity", "policy"])
def test_expiry_while_waiting_for_write_transaction_refuses_mutation(workspace, monkeypatch, expiry):
now = [time.time()]
app, memo = service(workspace, clock=lambda: now[0])
agent = replace(caller(), expires_at=now[0] + (10 if expiry == "identity" else 600))
original = app.repository.register
def delayed(*args, **kwargs):
now[0] += 40
return original(*args, **kwargs)
monkeypatch.setattr(app.repository, "register", delayed)
with pytest.raises(LifecycleError) as error:
app.register(agent, memo, key="new")
assert error.value.code == ("identity_expired" if expiry == "identity" else "decision_expired")
with pytest.raises(EvidenceUnavailable):
app.repository.store.memo(memo.id)
def test_command_is_snapshotted_before_external_policy_call(workspace):
app, memo = service(workspace)
agent = caller()
app.register(agent, memo, key="new")
body = {"text": "authorized content"}
app.policy.transport.change = lambda data: body.update(text="changed during network call")
app.annotate(agent, memo.id, 1, kind="comment", body=body, key="comment")
event = app.repository.events(AGENT.tenant, memo.id)[1]
assert event["body"]["text"] == "authorized content"
def test_read_checks_expiry_after_loading_content(workspace, monkeypatch):
now = [time.time()]
app, memo = service(workspace, clock=lambda: now[0])
agent = caller()
app.register(agent, memo, key="new")
original = app.repository.retrieve
def delayed(*args, **kwargs):
result = original(*args, **kwargs)
now[0] += 40
return result
monkeypatch.setattr(app.repository, "retrieve", delayed)
with pytest.raises(LifecycleError, match="decision_expired"):
app.retrieve(agent, memo.id)
def test_observation_from_another_request_is_refused(workspace, monkeypatch):
app, memo = service(workspace)
monkeypatch.setattr(app.policy, "check", lambda req: Observation('{}', None, "allow", time.time()+60, None))
with pytest.raises(LifecycleError, match="policy_request_mismatch"):
app.register(caller(), memo, key="new")
assert app.repository.store.policy_observations() == []
def test_missing_persisted_authorization_prevents_write(workspace, monkeypatch):
app, memo = service(workspace)
monkeypatch.setattr(app.repository.store, "observe_policy", lambda observed: "not-persisted")
with pytest.raises(LifecycleError, match="policy_observation_unavailable"):
app.register(caller(), memo, key="new")
with pytest.raises(EvidenceUnavailable):
app.repository.store.memo(memo.id)
@pytest.mark.parametrize("expiry", [float("nan"), float("inf"), True])
def test_unbounded_identity_lifetime_refused(expiry):
with pytest.raises(ValueError):
replace(caller(), expires_at=expiry)
def test_human_identity_must_be_authentication_derived():
with pytest.raises(ValueError):
Caller.from_human_session(replace(human(), principal_type=Claim("human", Route.REGISTRATION)))

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

@ -149,3 +149,17 @@ 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.
## Protected application service — 2026-10-04
Added `LifecycleService` above the repository: verified human-session import,
trusted workload-caller interface, fresh PolicyClient observations for every
operation/retry, immutable command snapshots, transaction-time authorization
guards and post-read expiry checks. Mutation events link the retained policy
observation. Existing human disposition and policy action contracts are unchanged.
`docs/specs/LifecycleServiceContract.md` records the proposed lifecycle action
names/context and remaining production identity/policy/HTTP/audit admission work.
T02/T03 remain in progress; no listener or policy grant was deployed.
Validation: `uv run --extra dev python -m pytest -q` — 501 passed, 72 skipped;
`git diff --check` passed.