diff --git a/SCOPE.md b/SCOPE.md index 56ec998..8df1add 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -4,29 +4,34 @@ > direction belongs in `INTENT.md`; the current stage belongs in `GOAL.md`; > current work and gates belong in `workplans/`. -## Lifecycle expansion — 2026-10-04 (first local persistence slice) +## Lifecycle expansion — 2026-10-04 (source implementation) -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 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. `LifecycleService` now wraps these operations -with fresh exact-policy observations, expiry checks inside write transactions, -and a verified human-session adapter. An unmounted WSGI -`LifecycleAPI` now provides registration, optimistic revisions with atomic packet -uploads, retrieval, history and annotation routes through that service; production HTTP exposure, workload identity composition and external -event delivery remain unimplemented. Unpublished lifecycle records -are excluded from the legacy approver overview, legacy import cannot bypass their -revision history, and pending lifecycle events close runtime readiness. 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. +INFD-WP-0007 adds a headless service and an optional browser workspace. The shared +foundation supports registration with packet custody, immutable revisions, +attributed sources/evaluations/comments, explicit revision-specific publication, +authorized search, retrieval and history. Every operation consumes fresh Access +Engine policy observations; audience facts do not implement a local ACL. + +The workspace reuses verified KeyCape sessions and CSRF/origin checks for draft +creation, research, discussion, publication, refinement, revision comparison and +packet downloads. Runtime configuration requires a separate lifecycle policy +package. The bearer API adapter remains unmounted pending workload identity and +listener admission. Lifecycle records remain outside the legacy approver overview; +the human binding handoff is not implemented. + +Opt-in delivery commits lifecycle evidence through the existing Audit Core +outbox, with receipt-based settlement, retries and reconciliation. Current delivery +is platform-tenant-only. Pending evidence closes runtime readiness. The deployment +renderer supports separately pinned lifecycle policy and exact network proposals; +no live configuration was changed. Cross-repository validation: 623 passed, +5 skipped, including two agent-created decisions and human collaboration with +unchanged runtime configuration and actual Audit Core delivery. Identity and PDP +responses in that scenario are synthetic; this is not live acceptance evidence. + +Contracts: `docs/specs/DecisionLifecycleService.md`, `LifecycleFoundationContract.md`, +`LifecycleServiceContract.md`, `LifecycleHTTPContract.md` and +`LifecycleWorkspaceDelivery.md` under `docs/specs/`. INFD-WP-0005 retains intake +runtime/admission work. The deployment snapshot below is historical. ## Status — 2026-09-27 diff --git a/docs/specs/LifecycleFoundationContract.md b/docs/specs/LifecycleFoundationContract.md index 6429172..1393688 100644 --- a/docs/specs/LifecycleFoundationContract.md +++ b/docs/specs/LifecycleFoundationContract.md @@ -77,19 +77,19 @@ legacy `save_memo` behavior. Existing stale-presentation guards see the same mem 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. -Runtime readiness now refuses while lifecycle events remain pending, and the -admin inspection command reports content-free lifecycle counts. No events are -marked delivered without the future admitted receiver adapter. +Pending lifecycle events are durable local obligations. The opt-in runtime audit +adapter stages immutable `informed-decision.lifecycle` commitments into the +existing transactional outbox and clears pending obligations only after a receiver +receipt. Content remains in local custody; the independent commitment contains +its digest and existence assertion. Readiness refuses while lifecycle delivery +is pending. See `LifecycleWorkspaceDelivery.md` for the platform tenant boundary, +retries, heartbeats and outstanding production admission. ## Remaining contracts 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 +production publication audience policy, incomplete drafts, non-cluster workload identity, +legacy record enrollment, source fetching/retention, live 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`. diff --git a/docs/specs/LifecycleHTTPContract.md b/docs/specs/LifecycleHTTPContract.md index 212b27c..a476910 100644 --- a/docs/specs/LifecycleHTTPContract.md +++ b/docs/specs/LifecycleHTTPContract.md @@ -6,6 +6,8 @@ existing browser app or public ingress. | Method and path | Input | Result | | --- | --- | --- | +| `GET /v1/decisions` | Optional `q` (up to 200 characters), opaque `cursor` | Authorized current revision summaries and `next_cursor` | +| `POST /v1/decisions/{id}/publications` | `version`, `audience` with explicit `subjects` and `groups` lists | Durable publication event receipt | | `POST /v1/decisions` | `memo`, optional `documents` | Register an unbound version-one memo and return event/record/revision/resource receipt | | `POST /v1/decisions/{id}/revisions` | `memo`, `expected_version`, `summary`, optional `documents` | Compare-and-append successor revision; same receipt shape | | `GET /v1/decisions/{id}` | Optional positive integer `version` | Existing Memo JSON for the current or specified revision | @@ -19,7 +21,7 @@ Ids are 1–256 ASCII letters/digits or `.`, `_`, `:`, `-`. POST requires exact nonempty `Idempotency-Key` up to 256 characters. Mutation receipts are HTTP 200 for both first submission and replay, with `event_id`, `memo_id`, `version`. Unknown fields, duplicate JSON/query keys, non-finite JSON values, boolean -revision numbers and truncated bodies are rejected. Search, publication and disposition routes are not yet exposed. +revision numbers and truncated bodies are rejected. Disposition routes are not exposed. Authentication runs before routing or reading a mutation body. The composition must explicitly supply a bearer-token verifier returning an authenticated @@ -40,8 +42,9 @@ identity, 403 policy denial, 404 unavailable record/route, 405 unsupported metho service or unusable policy evidence. Raw exception text and credentials are never returned. Upstream raw diagnostics are not forwarded. -T04 remains in progress: search/publication contracts, admitted authenticators, -canonical browser URLs and runtime wiring. Current +T04 remains in progress: admitted workload authenticators and API listener +composition. The browser workspace uses verified sessions through a separate +adapter (see LifecycleWorkspaceDelivery.md). Current tests use the real service/repository/PolicyClient with synthetic authenticator and PDP transport; they establish adapter behavior, not live identity admission. Runtime must not expose this adapter before lifecycle policy and independent @@ -64,8 +67,11 @@ Registration is `{ "memo": , "documents": [...] }`. Revision adds `expected_version` and a nonempty `summary`; its memo id must match the path and its version must be exactly expected_version + 1. Every mutation requires Idempotency-Key. Register/revise receipts include a relative API resource link -`/v1/decisions/{id}?version={version}`. This does not claim that an unpublished -draft has a human review URL or that the API is publicly mounted. +`/v1/decisions/{id}?version={version}` and a `workspace_url` at +`https://decisions.coulomb.social/decisions/{id}?version={version}`. The latter +is an authorized authoring workspace, not a binding review link. The API remains +unmounted. Search cursors are caller/tenant/query scoped, expire after five +minutes and are invalidated on process restart. Every page rechecks policy. Each document has exactly `digest`, `media_type`, `content_base64`. At most 100 documents are allowed; the total JSON request remains bounded at 256 KiB and diff --git a/docs/specs/LifecycleServiceContract.md b/docs/specs/LifecycleServiceContract.md index 9aae419..d64c67a 100644 --- a/docs/specs/LifecycleServiceContract.md +++ b/docs/specs/LifecycleServiceContract.md @@ -2,8 +2,8 @@ 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. +This is a headless Python application interface. An optional runtime composition +enables the browser workspace; production admission is still outstanding. ## Caller and authority @@ -23,8 +23,8 @@ 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; +`lifecycle.evaluation`, `lifecycle.comment`, `lifecycle.read`, +`lifecycle.events`, `lifecycle.publish`, `lifecycle.search` and `lifecycle.packet`. 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:`, @@ -35,7 +35,17 @@ and context: * `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. + expected revision and idempotency key, or the read/page parameters; +* `record`: tenant-partitioned registration facts, original registering actor, + current revision and latest publication audience for the requested revision; +* `publication_audience`: explicit subjects/groups for a publication command. + +Search uses resource `decisions`, type `decision-collection`, then independently +authorizes each candidate with `lifecycle.read`. A page examines at most 50 +candidates; denied records are omitted, unavailable policy fails the request. +Opaque cursors are scoped to caller/tenant/query and expire after five minutes. +Packet access checks its own action and a revision-specific packet index; a +known content digest grants no retrieval capability. The command digest correlates the requested operation; it is not `view_hash`, an approval binding digest or an authorization verdict. No presentation evidence @@ -53,7 +63,8 @@ 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 +the persisted observation and unchanged record/audience facts 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. @@ -61,26 +72,27 @@ identity/policy guard does not perform network calls while holding the write loc 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 +it, including unchanged record/audience facts. 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. +events remain pending until the opt-in audit adapter obtains a receiver receipt. +See LifecycleWorkspaceDelivery.md for custody, reconciliation and admission. ## 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 +the HTTP adapter maps 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. +Publication appends an immutable event for the current revision with an explicit +nonempty audience (at most 32 subjects and 32 groups, no wildcard). A successor +revision does not inherit publication. Audience facts are inputs to Access Engine, +never local permissions. Human acceptance requires the separate existing +presentation/disposition path; the lifecycle-to-review handoff is still open. Before enabling a listener, T02/T04 must provide: reviewed lifecycle PDP package and action/context fixtures; admitted human/workload identity adapters; closed diff --git a/docs/specs/LifecycleWorkspaceDelivery.md b/docs/specs/LifecycleWorkspaceDelivery.md new file mode 100644 index 0000000..b699a2a --- /dev/null +++ b/docs/specs/LifecycleWorkspaceDelivery.md @@ -0,0 +1,70 @@ +# Lifecycle workspace and delivery + +2026-10-04; INFD-WP-0007-T02 through T06, source implementation. + +## Human workspace + +With `lifecycle_policy` configured, the signed-in home links to `/decisions`. +The workspace uses the same headless service as the API. Existing KeyCape +sessions establish identity; existing origin and CSRF checks protect forms. +Users can search, prepare a draft, record source metadata and evaluations, +comment, publish to explicit members/groups, refine, compare consecutive +revisions and download authorized packet entries. Source content is not fetched. +Packet uploads currently use the API. All forms receive fresh policy checks; +rendering a button grants no authority. + +Publication records an audience for one revision. Access Engine determines access +using authenticated identity and retained record facts. Publication creates no +approval and no human disposition. New revisions require their own publication. +The workspace explicitly states that human acceptance is unavailable for these +drafts. Connecting a lifecycle revision to the existing protected presentation +and Approval Engine path remains T03/T05 work. + +## Runtime and deployment inputs + +The optional runtime `lifecycle_policy` object requires exactly `origin`, +`package`, `version`, `package_digest`, `caller_token_file`. Its package must +differ from the review package. No default policy or identity fallback exists. +The optional boolean `lifecycle_delivery` defaults to false. Configure both +for an admitted workspace with evidence delivery. + +The deployment renderer accepts the same separate policy pins plus `pod_name`, +requires explicit `lifecycle_delivery: true` with that package, and generates +exact egress and an owner-review ingress proposal. It references the existing +mounted workload policy credential path; this does not authorize that workload +for a new package. Existing admission inputs render without lifecycle changes. +The API still requires its own admitted bearer verifier and listener composition. + +## Independent evidence + +On an enabled audit tick, up to 100 pending mutations are staged as immutable +`informed-decision.lifecycle` commitments in the existing outbox. The event UUID +is the delivery identity. Commitment metadata includes memo id, revision, +operation, a content digest and an existence assertion. The digest covers the +retained event and historical memo. Advisory text stays in source custody; +commitment delivery does not provide independent content reconstructability. + +Pending obligations clear only after the outbox retains a receiver reference. +Lost replies retry with the same identity. The existing worker supplies bounded +delivery, reconciliation and a 24-hour lifecycle heartbeat. Undelivered evidence +keeps runtime readiness false. The current sink/reconciliation identity is +platform-scoped; other tenants fail closed at staging. Multi-tenant receiver +admission is not claimed. Kubernetes traffic readiness behavior is unchanged. + +Tests use the actual local Audit Core receiver contract, including deduplication +and reconciliation. The agent-to-human collaboration scenario creates two +records, publishes/comments/refines and drains five events with unchanged runtime +configuration bytes. Its identity and PDP transport are synthetic. It proves +source behavior, not deployed workload authority or all seven acceptance cases. + +## Remaining delivery work + +T02/T04 require owner-admitted lifecycle policy actions/context, workload identity +and API ingress. The policy must cover all actions in LifecycleServiceContract, +including collection search and per-record checks. Existing compact-sitting pins +do not supply these grants. WP-0005 retains private intake composition work. + +T03/T05 require the durable review handoff and human-only acceptance integration. +T06 requires full acceptance evidence, sender/cadence admission, immutable image, +database backup/migration and rollback verification. T07 requires the separately +approved concrete live rollout. No production pins or resources changed here. diff --git a/informed_decision/evidence.py b/informed_decision/evidence.py index c56d479..53d1ea8 100644 --- a/informed_decision/evidence.py +++ b/informed_decision/evidence.py @@ -39,6 +39,7 @@ class EventClass(str, Enum): DISPOSITION = "informed-decision.disposition" STANCE_APPLICATION = "informed-decision.stance_application" MEMO_POSED = "informed-decision.memo.posed" + LIFECYCLE = "informed-decision.lifecycle" HEARTBEAT = "audit-core.heartbeat" diff --git a/informed_decision/lifecycle.py b/informed_decision/lifecycle.py index 661b650..25b0999 100644 --- a/informed_decision/lifecycle.py +++ b/informed_decision/lifecycle.py @@ -184,6 +184,54 @@ class LifecycleRepository: 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() + return {"registered": True, "registered_by": json.loads(row["actor"]), + "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 = { @@ -221,6 +269,19 @@ class LifecycleRepository: 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") @@ -231,3 +292,20 @@ class LifecycleRepository: (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 diff --git a/informed_decision/lifecycle_http.py b/informed_decision/lifecycle_http.py index 6fd44f7..05e61ae 100644 --- a/informed_decision/lifecycle_http.py +++ b/informed_decision/lifecycle_http.py @@ -11,11 +11,12 @@ from urllib.parse import parse_qs from .lifecycle_service import Caller, LifecycleError from .lifecycle_commands import decode_memo +from .oidc import ORIGIN from .records import dumps from .store import Conflict, EvidenceUnavailable MAX_BODY = 256 * 1024 -ROUTE = re.compile(r"/v1/decisions/([A-Za-z0-9._:-]{1,256})(?:/(events|comments|sources|evaluations|revisions))?") +ROUTE = re.compile(r"/v1/decisions/([A-Za-z0-9._:-]{1,256})(?:/(events|comments|sources|evaluations|revisions|publications))?") def _object(pairs): @@ -77,6 +78,11 @@ class LifecycleAPI: params = parse_qs(query, keep_blank_values=True, strict_parsing=True, max_num_fields=4) if any(len(value) != 1 for value in params.values()): raise ValueError() + if method == "GET" and registration: + if set(params) - {"q", "cursor"}: + raise ValueError() + return 200, self.service.search(caller, query=params.get("q", [""])[0], + cursor=params.get("cursor", [None])[0]) if method == "GET" and resource is None: if set(params) - {"version"}: raise ValueError() @@ -90,7 +96,7 @@ class LifecycleAPI: return 200, {"events": self.service.events(caller, memo_id, after=int(params.get("after", ["0"])[0]), limit=int(params.get("limit", ["100"])[0]))} kinds = {"comments": "comment", "sources": "source", "evaluations": "evaluation"} - if method != "POST" or resource not in {*kinds, "register", "revisions"}: + if method != "POST" or resource not in {*kinds, "register", "revisions", "publications"}: raise LifecycleError(405, "method_not_allowed") if params: raise ValueError() @@ -109,6 +115,11 @@ class LifecycleAPI: key = environ.get("HTTP_IDEMPOTENCY_KEY", "") if not isinstance(key, str) or not key.strip() or len(key) > 256: raise ValueError() + if resource == "publications": + if not isinstance(data, dict) or set(data) != {"version", "audience"}: + raise ValueError() + event_id = self.service.publish(caller, memo_id, data["version"], audience=data["audience"], key=key) + return 200, {"event_id": event_id, "memo_id": memo_id, "version": data["version"]} if resource in {"register", "revisions"}: required = {"memo"} if registration else {"memo", "expected_version", "summary"} if (not isinstance(data, dict) or not required <= data.keys() @@ -124,7 +135,8 @@ class LifecycleAPI: event_id = self.service.revise(caller, memo, key=key, expected_version=data["expected_version"], summary=data["summary"], documents=documents) return 200, {"event_id": event_id, "memo_id": memo.id, "version": memo.version, - "resource": f"/v1/decisions/{memo.id}?version={memo.version}"} + "resource": f"/v1/decisions/{memo.id}?version={memo.version}", + "workspace_url": f"{ORIGIN}/decisions/{memo.id}?version={memo.version}"} if not isinstance(data, dict) or set(data) != {"version", "body"}: raise ValueError() event_id = self.service.annotate(caller, memo_id, data["version"], kind=kinds[resource], diff --git a/informed_decision/lifecycle_service.py b/informed_decision/lifecycle_service.py index 1845a8a..454c88f 100644 --- a/informed_decision/lifecycle_service.py +++ b/informed_decision/lifecycle_service.py @@ -8,11 +8,13 @@ from dataclasses import dataclass import hashlib import json import math +import secrets +import threading import time import uuid from .disposition import Actor, ActorKind -from .lifecycle import Attribution, LifecycleRepository +from .lifecycle import Attribution, LifecycleRepository, publication_audience from .oidc import HumanSession from .policy import PolicyError from .provenance import Route @@ -58,6 +60,8 @@ class LifecycleError(RuntimeError): class LifecycleService: def __init__(self, repository: LifecycleRepository, policy, *, clock=time.time): self.repository, self.policy, self.clock = repository, policy, clock + self._cursors = {} + self._cursor_lock = threading.Lock() def _caller(self, caller): if not isinstance(caller, Caller) or caller.expires_at <= self.clock(): @@ -71,14 +75,21 @@ class LifecycleService: command = dumps(payload).encode() if len(command) > 256 * 1024: raise ValueError("lifecycle command too large") + version = payload.get("expected_version", payload.get("memo").version + if payload.get("memo") is not None else payload.get("version")) + state = (self.repository.policy_context(tenant, memo_id, version) + if operation != "search" else {"registered": False}) 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", + "resource": {"id": "decisions" if operation == "search" else "memo:" + memo_id, + "type": "decision-collection" if operation == "search" else "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"), + "record": state, + **({"publication_audience": payload["audience"]} if operation == "publish" else {}), "command_digest": "sha256:" + hashlib.sha256(command).hexdigest()}, "policy_version": self.policy.version} observation = self.policy.check(request) @@ -102,10 +113,17 @@ class LifecycleService: 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") + if self.repository._policy_context(db, tenant, memo_id, version) != state: + raise LifecycleError(409, "record_changed") return policy_id + def read_current(): + current() + if operation != "search" and self.repository.policy_context(tenant, memo_id, version) != state: + raise LifecycleError(409, "record_changed") + current() - return current, guard + return read_current, guard def register(self, caller, memo, *, key, documents=None): # Snapshot before the PDP call: mutable caller-owned objects cannot change @@ -137,6 +155,56 @@ class LifecycleService: current() return result + def publish(self, caller, memo_id, version, *, audience, key): + audience = publication_audience(audience) + _, guard = self._authorize(caller, memo_id, "publish", + {"version": version, "audience": audience, "key": key}) + return self.repository.publish(caller.attribution, memo_id, version, audience=audience, key=key, guard=guard) + + def packet(self, caller, memo_id, version, index): + current, _ = self._authorize(caller, memo_id, "packet", {"version": version, "index": index}) + result = self.repository.packet(caller.attribution.tenant, memo_id, version, index) + current() + return result + + def search(self, caller, *, query="", cursor=None): + self._caller(caller) + if not isinstance(query, str) or len(query) > 200: + raise ValueError("invalid search query") + scope = (caller.attribution.tenant, dumps(caller.attribution.actor), query) + after = "" + with self._cursor_lock: + self._cursors = {key: row for key, row in self._cursors.items() if row[2] > self.clock()} + if cursor is not None: + row = self._cursors.get(cursor) if isinstance(cursor, str) else None + if row is None or row[0] != scope: + raise ValueError("invalid search cursor") + after = row[1] + current, _ = self._authorize(caller, "decisions", "search", {"query": query, "after": after}) + candidates = self.repository.candidates(caller.attribution.tenant, query=query, after=after, limit=50) + results, checks = [], [current] + for memo_id in candidates: + try: + check, _ = self._authorize(caller, memo_id, "read", {"version": None}) + except LifecycleError as error: + if error.status == 403: + continue + raise + memo = self.repository.retrieve(caller.attribution.tenant, memo_id) + check() + checks.append(check) + results.append({"memo_id": memo.id, "version": memo.version, "question": memo.question}) + for check in checks: + check() + next_cursor = None + if len(candidates) == 50: + with self._cursor_lock: + if len(self._cursors) >= 1000: + raise LifecycleError(503, "search_capacity") + next_cursor = secrets.token_urlsafe(24) + self._cursors[next_cursor] = (scope, candidates[-1], self.clock() + 300) + return {"decisions": results, "next_cursor": next_cursor} + 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) diff --git a/informed_decision/lifecycle_ui.py b/informed_decision/lifecycle_ui.py new file mode 100644 index 0000000..05b2e2c --- /dev/null +++ b/informed_decision/lifecycle_ui.py @@ -0,0 +1,183 @@ +"""Human workspace adapter: forms and rendering, with no storage or policy rules.""" + +from dataclasses import replace +import re +from urllib.parse import urlencode +import uuid + +from .lifecycle_commands import decode_memo +from .lifecycle_service import Caller, LifecycleError +from .memo import Memo, BindingLevel, BindingSlice, Principal, Scope, StepKind +from .records import dumps +from .store import EvidenceUnavailable +from .ui import document, hidden, text +import json + + +def _form(action, session, fields, button): + return (f'
' + + hidden("csrf", session.csrf) + hidden("key", str(uuid.uuid4())) + + fields + f'
') + + +def _textarea(name, label, value=""): + return f'' + + +def _lines(value): + return [line.strip() for line in value.splitlines() if line.strip()] + + +class LifecycleWorkspace: + def __init__(self, service): + self.service = service + + def dispatch(self, method, path, params, session, form=None): + caller = Caller.from_human_session(session) + self.service._caller(caller) + if method == "GET" and path == "/decisions": + if set(params) - {"q", "cursor"}: + raise ValueError() + query = params.get("q", "") + page = self.service.search(caller, query=query, cursor=params.get("cursor")) + content = ('

Decision workspace

Prepare a decision, examine evidence, and invite discussion.

' + '
' + '

Prepare a decision

' + if not page["decisions"]: + content += '

No decisions to display on this page.

' + if page["next_cursor"]: + content += f'

Next page

' + return 200, document("Decision workspace", content, session.subject), [] + if path == "/decisions/new/prepare": + if params: + raise ValueError() + if method == "GET": + fields = (_textarea("question", "Question") + _textarea("brief", "Context, alternatives and recommendation") + + '' + '') + content = '

Prepare a decision

' + _form(path, session, fields, "Save draft") + return 200, document("Prepare a decision", content, session.subject), [] + if method == "POST": + self._fields(form, {"question", "brief", "scope_id", "scope_label"}) + memo = Memo(id=str(uuid.uuid4()), version=1, question=form["question"], brief=form["brief"], + requested_act="accept", binding_level=BindingLevel.ORGANIZATIONAL, + step_kind=StepKind.MITZEICHNUNG, + binding=BindingSlice(Principal("pending-human-session", "person", "To be assigned"), + Scope("resource", form["scope_id"], form["scope_label"]))) + # Stable create identity across an HTTP retry of the same rendered form. + memo = replace(memo, id=str(uuid.uuid5(uuid.NAMESPACE_URL, + dumps([caller.attribution.tenant, session.subject, form["key"]])))) + self.service.register(caller, decode_memo(json.loads(dumps(memo))), key=form["key"]) + return 303, "", [("Location", "/decisions/" + memo.id)] + match = re.fullmatch(r"/decisions/([A-Za-z0-9._:-]{1,256})(?:/(revise|comment|publish|source|evaluate|packet/([0-9]{1,3})))?", path) + if not match: + return 404, "Not found.", [] + memo_id, action, index = match.groups() + if method == "GET" and (action is None or index is not None): + if set(params) - {"version", "after"}: + raise ValueError() + version = int(params["version"]) if "version" in params else None + if index is not None: + if version is None or "after" in params: + raise ValueError() + return 200, self.service.packet(caller, memo_id, version, int(index)), [ + ("Content-Disposition", f'attachment; filename="decision-document-{int(index)+1}.bin"')] + memo = self.service.retrieve(caller, memo_id, version) + events = self.service.events(caller, memo_id, after=int(params.get("after", "0"))) + previous = None + if memo.version > 1: + try: + previous = self.service.retrieve(caller, memo_id, memo.version - 1) + except EvidenceUnavailable: + pass + except LifecycleError as error: + if error.status != 403: + raise + return 200, self._page(memo, events, session, previous), [] + if method == "POST" and action in {"revise", "comment", "publish", "source", "evaluate"}: + if params: + raise ValueError() + fields = {"revise": {"version", "question", "brief", "summary"}, + "comment": {"version", "text"}, "publish": {"version", "subjects", "groups"}, + "source": {"version", "reference", "retrieved_at", "digest", "summary"}, + "evaluate": {"version", "assessment", "source_ids", "uncertainty"}}[action] + self._fields(form, fields) + version = int(form["version"]) + if action == "revise": + previous = self.service.retrieve(caller, memo_id, version) + memo = previous.next_version(question=form["question"], brief=form["brief"]) + self.service.revise(caller, decode_memo(json.loads(dumps(memo))), expected_version=version, + summary=form["summary"], key=form["key"]) + elif action == "publish": + self.service.publish(caller, memo_id, version, key=form["key"], + audience={"subjects": _lines(form["subjects"]), "groups": _lines(form["groups"])}) + else: + body = {key: form[key] for key in fields - {"version"}} + if action == "evaluate": + body["source_ids"] = _lines(body["source_ids"]) + self.service.annotate(caller, memo_id, version, key=form["key"], + kind="evaluation" if action == "evaluate" else action, body=body) + return 303, "", [("Location", "/decisions/" + memo_id)] + return 405, "Method not allowed.", [] + + @staticmethod + def _fields(form, required): + if not isinstance(form, dict) or set(form) != required | {"csrf", "key"}: + raise ValueError() + uuid.UUID(form["key"]) + + def _page(self, memo, events, session, previous=None): + base = "/decisions/" + memo.id + revision = hidden("version", memo.version) + content = (f'

Decision workspace

{text(memo.question)}

' + f'

Revision {memo.version} · {text(memo.binding.target.label)}

' + f'

Context and recommendation

{text(memo.brief)}

' + '

Human acceptance is not available for this draft. Publishing invites discussion; it does not record acceptance.

') + if previous is not None: + content += f'
Compare with revision {previous.version}' + for field, label in (("question", "Question"), ("brief", "Context and recommendation")): + before, after = getattr(previous, field), getattr(memo, field) + if before != after: + content += (f'

{label}

Before

{text(before)}

' + f'

After

{text(after)}

') + content += '
' + if memo.packet: + content += '

Supporting documents

    ' + for index, item in enumerate(memo.packet): + content += f'
  • {text(item.label)}
    • ' + content += '
' + content += '

Discussion and history

' + for event in events: + body = {key: value for key, value in event["body"].items() if key != "policy_observation_id"} + content += (f'

{text(event["kind"])} · revision {event["version"]}

' + f'

{text(event["actor"]["sub"])} ({text(event["actor"]["kind"])}) · {text(event["at"])}

') + if event["kind"] == "source": + content += f'

Source reference: {text(event["id"])}

' + for key, value in body.items(): + content += f'

{text(key.replace("_", " "))}: {text(dumps(value) if isinstance(value, (dict, list)) else value)}

' + content += '
' + if len(events) == 100: + content += f'

More history

' + content += '

Add a comment

' + content += _form(base + "/comment", session, revision + _textarea("text", "Comment"), "Add comment") + '
' + content += '
Refine this decision' + content += _form(base + "/revise", session, revision + _textarea("question", "Question", memo.question) + + _textarea("brief", "Context and recommendation", memo.brief) + + _textarea("summary", "What changed"), "Save new revision") + '
' + content += '
Publish for discussion

Enter explicit member or group identifiers, one per line.

' + content += _form(base + "/publish", session, revision + '' + '', "Publish this revision") + '
' + content += '
Add a research source' + content += _form(base + "/source", session, revision + ''.join(_textarea(name, label) for name, label in ( + ("reference", "Source location or reference"), ("retrieved_at", "Retrieved at (date, time and timezone)"), + ("digest", "Source content digest (sha256:...)"), ("summary", "What the source supports"))), "Add source") + '
' + content += '
Evaluate the evidence' + content += _form(base + "/evaluate", session, revision + _textarea("assessment", "Assessment") + + '' + + _textarea("uncertainty", "Uncertainty and open questions"), "Add evaluation") + '
' + return document("Decision workspace", content, session.subject) diff --git a/informed_decision/runtime.py b/informed_decision/runtime.py index 92ba87b..e179287 100644 --- a/informed_decision/runtime.py +++ b/informed_decision/runtime.py @@ -36,9 +36,10 @@ def token_file(path): class AuditPump: """One process, bounded batches, 30s ticks; never drops a failed record.""" - def __init__(self, store, sink, *, clock=time.time, intake_worker=None): + def __init__(self, store, sink, *, clock=time.time, intake_worker=None, lifecycle_delivery=False): self.store, self.worker, self.clock = store, OutboxWorker(store, sink), clock self.intake_worker = intake_worker + self.lifecycle_delivery = lifecycle_delivery self._stop = threading.Event() self._thread = None self._lock = threading.Lock() @@ -57,10 +58,15 @@ class AuditPump: try: if self.intake_worker: self.intake_worker.run_once(limit=10) - self.store.queue_heartbeats(include_posed=self.intake_worker is not None) + if self.lifecycle_delivery: + self.store.stage_lifecycle_audit() + self.store.queue_heartbeats(include_posed=self.intake_worker is not None, + include_lifecycle=self.lifecycle_delivery) result = self.worker.run_once(limit=10) if self.intake_worker: self.store.settle_intake_audit() + if self.lifecycle_delivery: + self.store.settle_lifecycle_audit() now = self.clock() if now - self._last_reconciled >= 300: stamp = lambda t: datetime.fromtimestamp(t, timezone.utc).isoformat(timespec="microseconds") @@ -112,6 +118,7 @@ class AuditPump: class Runtime: controller: ReviewController pump: AuditPump + lifecycle: object | None = None @classmethod def from_file(cls, filename): @@ -122,11 +129,14 @@ class Runtime: or info.st_size > 16384): raise ValueError("runtime configuration must be an owned private 0600 file") data = json.loads(path.read_text()) - if ((set(data) - {"ordinary_approval_ids", "intake_delivery"}) != {"schema", "evidence_db", "policy", "approval_origin", "audit"} + if ((set(data) - {"ordinary_approval_ids", "intake_delivery", "lifecycle_policy", "lifecycle_delivery"}) != {"schema", "evidence_db", "policy", "approval_origin", "audit"} or data["schema"] != "informed-decision.review-runtime.v1" or set(data["policy"]) != {"origin", "package", "version", "package_digest", "caller_token_file"} or set(data["audit"]) != {"origin", "sender_token_file"}): raise ValueError("invalid review runtime configuration") + lifecycle_delivery = data.get("lifecycle_delivery", False) + if type(lifecycle_delivery) is not bool: + raise ValueError("invalid lifecycle delivery setting") ordinary = ordinary_review_ids(data.get("ordinary_approval_ids", [])) if not Path(data["evidence_db"]).is_absolute(): raise ValueError("absolute evidence database path required") @@ -145,6 +155,17 @@ class Runtime: if not isinstance(delivery, dict) or set(delivery) != {'hub_origin'}: raise ValueError('invalid intake delivery configuration') hub = HubDelivery(delivery['hub_origin']) + lifecycle_policy = None + if "lifecycle_policy" in data: + configuration = data["lifecycle_policy"] + if (not isinstance(configuration, dict) + or set(configuration) != {"origin", "package", "version", "package_digest", "caller_token_file"}): + raise ValueError("invalid lifecycle policy configuration") + if configuration["package"] == policy["package"]: + raise ValueError("lifecycle policy must have a separate admission package") + lifecycle_policy = PolicyClient(configuration["origin"], token_file(configuration["caller_token_file"]), + package=configuration["package"], version=configuration["version"], + package_digest=configuration["package_digest"], allow_internal_http=True) store = Store(data["evidence_db"]) controller = ReviewController(store, client, lambda session: ApprovalHTTPClient( approval_origin, session, allow_internal_http=True, ordinary_approval_ids=tuple(ordinary))) @@ -153,4 +174,10 @@ class Runtime: from .intake_delivery import IntakeDeliveryWorker worker = IntakeDeliveryWorker(store, audit=sink, notify=hub.notify, hub_transport=hub.resolution, hub_base_url=hub.origin) - return cls(controller, AuditPump(store, sink, intake_worker=worker)) + lifecycle = None + if lifecycle_policy is not None: + from .lifecycle import LifecycleRepository + from .lifecycle_service import LifecycleService + lifecycle = LifecycleService(LifecycleRepository(store), lifecycle_policy) + return cls(controller, AuditPump(store, sink, intake_worker=worker, + lifecycle_delivery=lifecycle_delivery), lifecycle) diff --git a/informed_decision/store.py b/informed_decision/store.py index 877d9b1..105f9ec 100644 --- a/informed_decision/store.py +++ b/informed_decision/store.py @@ -20,7 +20,7 @@ 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 .evidence import Commitment, 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 @@ -556,18 +556,50 @@ class Store: "(SELECT id FROM outbox WHERE state='delivered')") return result.rowcount + def stage_lifecycle_audit(self): + """Opt-in commitment-only delivery through the existing audit outbox.""" + with self._transaction() as db: + rows = db.execute("SELECT l.* FROM lifecycle_pending p JOIN lifecycle_events l ON l.id=p.event_id " + "ORDER BY l.sequence LIMIT 100").fetchall() + staged = 0 + for row in rows: + if row["tenant"] != "tenant:platform": + raise Conflict("lifecycle audit tenant not admitted by this runtime") + content = {"event": dict(row), "memo": self._memo(db, row["memo_id"], row["version"])} + existing = db.execute("SELECT class,content FROM evidence WHERE id=?", (row["id"],)).fetchone() + if existing: + if existing["class"] != EventClass.LIFECYCLE.value or existing["content"] != dumps(content): + raise Conflict("lifecycle evidence id collision") + continue + commitment = Commitment(row["id"], EventClass.LIFECYCLE, row["at"], { + "memo_id": row["memo_id"], "memo_version": row["version"], "operation": row["kind"], + "content_sha256": "sha256:" + hashlib.sha256(dumps(content).encode()).hexdigest(), + "content_exists": True, "custody": "informed-decision:lifecycle:" + row["id"]}) + self._event(db, commitment, content) + staged += 1 + return staged + + def settle_lifecycle_audit(self): + with self._transaction() as db: + return db.execute("DELETE FROM lifecycle_pending WHERE event_id IN " + "(SELECT o.id FROM outbox o JOIN evidence e ON e.id=o.id " + "WHERE o.state='delivered' AND o.receiver_reference IS NOT NULL AND e.class=?)", + (EventClass.LIFECYCLE.value,)).rowcount + def intake_audit_pending(self): with self._connection() as db: return db.execute("SELECT 1 FROM intake_deliveries WHERE kind='memo-posed' " "AND state!='delivered' LIMIT 1").fetchone() is not None - def queue_heartbeats(self, now=None, *, include_posed=False): + def queue_heartbeats(self, now=None, *, include_posed=False, include_lifecycle=False): now = time.time() if now is None else now count = 0 with self._transaction() as db: classes = {**HEARTBEAT_CLASSES} if include_posed: classes[EventClass.MEMO_POSED.value] = 86400 + if include_lifecycle: + classes[EventClass.LIFECYCLE.value] = 86400 for event_class, interval in classes.items(): if db.execute("SELECT 1 FROM outbox o JOIN evidence e ON e.id=o.id WHERE o.state!='delivered' " "AND (e.class=? OR (e.class=? AND json_extract(e.envelope,'$.data.class')=?)) LIMIT 1", diff --git a/informed_decision/web.py b/informed_decision/web.py index b3ea736..2e9158c 100644 --- a/informed_decision/web.py +++ b/informed_decision/web.py @@ -17,6 +17,7 @@ from .disposition import DispositionRefused, Verb from .policy import PolicyError from .provenance import HumanControlNotDischargeable from .review import ReviewError +from .lifecycle_service import LifecycleError from .store import Conflict, EvidenceUnavailable, StoreError from .ui import STYLES, document, error_page, overview_section, review_page @@ -38,9 +39,10 @@ def _one(query: str, *, limit=8192, fields=16) -> dict[str, str]: class App: - def __init__(self, login: KeyCapeLogin, review=None, *, readiness=lambda: False): + def __init__(self, login: KeyCapeLogin, review=None, *, readiness=lambda: False, lifecycle=None): self.login = login self.review, self.readiness = review, readiness + self.lifecycle = lifecycle def __call__(self, environ, start_response): path = environ.get("PATH_INFO", "/") @@ -84,6 +86,11 @@ class App: } status, body, content_type = exc.status, error_page(messages.get(exc.code, "The review could not complete its required checks. Reload after the service is available.")), "text/html" + except LifecycleError as exc: + status, body, content_type = exc.status, document("Decision unavailable", + '

Decision unavailable

The request could not complete its required checks. ' + 'Sign in again if your session expired, or reload the decision before retrying.

' + '

Return to the workspace

'), "text/html" except DispositionRefused as exc: status, body, content_type = 409, error_page({ "G_ACK": "Record all required highlight acknowledgments before accepting or declining.", @@ -125,6 +132,14 @@ class App: return 303, "", "text/plain", [("Location", "/"), _cookie(FLOW_COOKIE, "", 0), _cookie(SESSION_COOKIE, new_sid, 900)] session = self.login.session(sid) + if self.lifecycle is not None and (path == "/decisions" or path.startswith("/decisions/")): + if session is None: + raise LifecycleError(401, "identity_expired") + from .lifecycle_ui import LifecycleWorkspace + params = _one(environ.get("QUERY_STRING", "")) + form = self._form(environ, session) if method == "POST" else None + status, body, extra = LifecycleWorkspace(self.lifecycle).dispatch(method, path, params, session, form) + return status, body, "application/octet-stream" if isinstance(body, bytes) else "text/html", extra if self.review is not None and (path == "/review" or path.startswith("/presentations/")): if session is None: raise ReviewError(401, "session_expired") @@ -188,6 +203,7 @@ class App: if method == "GET" and path == "/": if session: content = (f"

Signed in as {html.escape(session.subject)}.

" + + ('

Open the decision workspace

' if self.lifecycle is not None else '') + (self._overview(session) if self.review is not None else '') + ('

Open a decision review

Enter the memo identifier supplied with your review request.

' '
' @@ -244,7 +260,8 @@ def main(): runtime = Runtime.from_file(os.environ["INFD_REVIEW_CONFIG"]) runtime.pump.start() app = App(login, runtime.controller if runtime else None, - readiness=runtime.pump.ready if runtime else lambda: False) + readiness=runtime.pump.ready if runtime else lambda: False, + lifecycle=runtime.lifecycle if runtime else None) try: serve(app, host=host, port=8080, threads=4) finally: diff --git a/tests/test_deployment_candidate.py b/tests/test_deployment_candidate.py index ac7742a..af3a68a 100644 --- a/tests/test_deployment_candidate.py +++ b/tests/test_deployment_candidate.py @@ -1,4 +1,5 @@ import copy +import json import importlib.util from pathlib import Path @@ -75,3 +76,24 @@ def test_policy_revision_changes_immutable_configuration_name(): first=inputs();second=copy.deepcopy(first);second['policy']['version']='v2' names=lambda data:[o['metadata']['name'] for o in module.render(data)['items'] if o['kind']=='ConfigMap'] assert names(first)!=names(second) + + +def test_optional_workspace_requires_distinct_pins_and_exact_policy_network_peer(): + data = inputs() + data['lifecycle_policy'] = {**data['policy'], 'package': 'informed-decision.lifecycle', + 'origin': 'http://flex-auth-lifecycle.flex-auth.svc.cluster.local:8080', + 'pod_name': 'flex-auth-lifecycle'} + with pytest.raises(ValueError): + module.render(data) + data['lifecycle_delivery'] = True + objects = module.render(data)['items'] + config = json.loads(next(row for row in objects if row['kind']=='ConfigMap')['data']['runtime.json']) + assert config['lifecycle_policy']['package']=='informed-decision.lifecycle' + assert config['lifecycle_delivery'] is True + review_network = next(row for row in objects if row['kind']=='NetworkPolicy' and row['metadata']['namespace']=='informed-decision') + assert any(peer.get('podSelector', {}).get('matchLabels', {}).get('app.kubernetes.io/name')=='flex-auth-lifecycle' + for rule in review_network['spec']['egress'] for peer in rule.get('to', [])) + assert any(row['metadata']['name']=='informed-decision-lifecycle-ingress' for row in objects) + data['lifecycle_policy']['package'] = data['policy']['package'] + with pytest.raises(ValueError): + module.render(data) diff --git a/tests/test_lifecycle_audit.py b/tests/test_lifecycle_audit.py new file mode 100644 index 0000000..e8f0a8f --- /dev/null +++ b/tests/test_lifecycle_audit.py @@ -0,0 +1,80 @@ +from dataclasses import replace +import hashlib +import json + +import pytest + +from informed_decision.audit import OutboxWorker +from informed_decision.evidence import EventClass +from informed_decision.lifecycle import LifecycleRepository +from informed_decision.runtime import AuditPump +from informed_decision.store import Conflict, Store +from test_durable_component import receiver +from test_lifecycle import AGENT, workspace + + +def register_platform(workspace): + repo, memo = workspace + actor = replace(AGENT, tenant="tenant:platform") + event_id = repo.register(actor, memo, key="register") + return repo.store, memo, event_id + + +def test_lifecycle_commitment_is_content_free_stable_and_verifiable(workspace): + store, memo, event_id = register_platform(workspace) + assert store.stage_lifecycle_audit() == 1 + assert store.stage_lifecycle_audit() == 0 + evidence = store.evidence()[0] + envelope = json.loads(evidence["envelope"]) + assert envelope["id"] == event_id and envelope["type"] == EventClass.LIFECYCLE.value + assert memo.brief not in evidence["envelope"] and memo.question not in evidence["envelope"] + assert envelope["data"]["content_sha256"] == "sha256:" + hashlib.sha256(evidence["content"].encode()).hexdigest() + assert store.settle_lifecycle_audit() == 0 + assert store.lifecycle_counts()["pending"] == 1 + assert Store(store.path).evidence() == store.evidence() + + +def test_lifecycle_delivery_uses_actual_receiver_and_recovers_lost_reply(workspace, receiver): + store, _, event_id = register_platform(workspace) + sink, transport, _ = receiver + store.stage_lifecycle_audit() + transport.lose_reply = True + worker = OutboxWorker(store, sink) + assert worker.run_once()["retrying"] == 1 + assert store.settle_lifecycle_audit() == 0 + with store._connection() as db: + db.execute("UPDATE outbox SET next_attempt=0 WHERE id=?", (event_id,)) + reopened = Store(store.path) + assert OutboxWorker(reopened, sink).run_once()["delivered"] == 1 + assert reopened.settle_lifecycle_audit() == 1 + assert reopened.lifecycle_counts() == {"records": 1, "events": 1, "pending": 0} + assert worker.reconcile("2020-01-01T00:00:00Z", "2100-01-01T00:00:00Z")["counts"][EventClass.LIFECYCLE.value] == {"source": 1, "receiver": 1} + + +def test_runtime_delivery_is_explicit_and_restores_readiness_only_after_receipts(workspace, receiver): + store, _, _ = register_platform(workspace) + sink, _, _ = receiver + disabled = AuditPump(store, sink) + disabled.tick() + assert not disabled.ready() and store.lifecycle_counts()["pending"] == 1 + enabled = AuditPump(store, sink, lifecycle_delivery=True) + enabled.tick() + assert enabled.ready() and store.lifecycle_counts()["pending"] == 0 + assert store.evidence()[0]["class"] != EventClass.LIFECYCLE.value # Legacy heartbeats predate opt-in. + + +def test_unadmitted_tenant_remains_pending_without_audit_relabelling(workspace): + repo, memo = workspace + repo.register(AGENT, memo, key="register") + with pytest.raises(Conflict, match="tenant not admitted"): + repo.store.stage_lifecycle_audit() + assert repo.store.evidence() == [] + assert repo.store.lifecycle_counts()["pending"] == 1 + + +def test_pending_events_do_not_get_a_nothing_to_report_heartbeat(workspace): + store, _, _ = register_platform(workspace) + store.stage_lifecycle_audit() + store.queue_heartbeats(include_lifecycle=True) + assert not any(json.loads(row["envelope"])["data"].get("class") == EventClass.LIFECYCLE.value + for row in store.evidence() if row["class"] == EventClass.HEARTBEAT.value) diff --git a/tests/test_lifecycle_flow.py b/tests/test_lifecycle_flow.py new file mode 100644 index 0000000..4b2a84d --- /dev/null +++ b/tests/test_lifecycle_flow.py @@ -0,0 +1,74 @@ +import json +from urllib.parse import urlencode +import uuid + +from informed_decision.audit import OutboxWorker +from informed_decision.lifecycle_http import LifecycleAPI +from informed_decision.lifecycle_service import Caller +from informed_decision.runtime import Runtime +from informed_decision.web import App, SESSION_COOKIE +from informed_decision.oidc import ORIGIN +from informed_decision.policy import PolicyClient +from informed_decision.records import dumps +from test_browser_auth import login, signing_key, finish, call +from test_deployment_candidate import inputs, module +from test_durable_component import receiver +from test_lifecycle_http import call as api_call +from test_policy_client import PolicyTransport +from test_review_runtime import config +from test_skeleton import make_memo + + +def test_agent_to_human_collaboration_and_audit_without_configuration_changes(tmp_path, login, receiver): + path, configuration = config(tmp_path) + configuration['lifecycle_policy'] = {**configuration['policy'], 'package': 'informed-decision.lifecycle'} + configuration['lifecycle_delivery'] = True + path.write_text(json.dumps(configuration)) + original_configuration = path.read_bytes() + runtime = Runtime.from_file(path) + transport = PolicyTransport() + transport.change = lambda data: data['provenance'].update(policy_package='informed-decision.lifecycle') + lifecycle = runtime.lifecycle + lifecycle.policy = PolicyClient('https://policy.test', lambda: 'synthetic-caller', + package='informed-decision.lifecycle', version='v1', package_digest=configuration['policy']['package_digest'], + transport=transport) + # Match the immutable configured digest in the simulated PDP return. + transport.change = lambda data: data['provenance'].update( + policy_package='informed-decision.lifecycle', policy_package_digest=configuration['policy']['package_digest']) + sid = finish(login) + session = login[0].session(sid) + from informed_decision.lifecycle import Attribution + from informed_decision.disposition import Actor, ActorKind + agent = Caller(Attribution(session.tenant.value, Actor('synthetic-agent', ActorKind.AGENT)), + session.expires_at, '{"principal_type_source":"authentication-derived"}') + def authenticate(token): + if token != 'fixture': + raise ValueError() + return agent + api = LifecycleAPI(lifecycle, authenticate) + app = App(login[0], runtime.controller, readiness=runtime.pump.ready, lifecycle=lifecycle) + cookie = f'{SESSION_COOKIE}={sid}' + memo = make_memo(id='agent-posed', approval_id=None, packet=(), highlights=()) + status, receipt, _ = api_call(api, '/v1/decisions', method='POST', body={'memo': json.loads(dumps(memo))}) + assert status == 200 and receipt['workspace_url'] == ORIGIN + '/decisions/agent-posed?version=1' + assert call(app, '/decisions/agent-posed', cookie=cookie)['status'] == 200 + def post(action, **values): + values.update(csrf=session.csrf, key=str(uuid.uuid4())) + return call(app, '/decisions/agent-posed/' + action, 'POST', cookie=cookie, origin=ORIGIN, + body=urlencode(values).encode()) + assert post('publish', version='1', subjects=session.subject, groups='')['status'] == 303 + assert post('comment', version='1', text='Please add the alternative.')['status'] == 303 + assert post('revise', version='1', question=memo.question, brief='Alternative added.', summary='Responded to feedback')['status'] == 303 + assert api_call(api, '/v1/decisions/agent-posed')[1]['version'] == 2 + from dataclasses import replace + second = replace(memo, id='second-agent-decision') + assert api_call(api, '/v1/decisions', method='POST', body={'memo': json.loads(dumps(second))}, + HTTP_IDEMPOTENCY_KEY='second')[0] == 200 + assert not runtime.pump.ready() + runtime.pump.worker = OutboxWorker(runtime.controller.store, receiver[0]) + runtime.pump.tick() + assert runtime.pump.ready() + assert runtime.controller.store.lifecycle_counts() == {'records': 2, 'events': 5, 'pending': 0} + assert path.read_bytes() == original_configuration + assert runtime.controller.store.memos_for(memo.binding.principal.id) == [] + assert api_call(api, '/v1/decisions/agent-posed/accept', method='POST', body={})[0] == 404 diff --git a/tests/test_lifecycle_publication.py b/tests/test_lifecycle_publication.py new file mode 100644 index 0000000..c4a69ed --- /dev/null +++ b/tests/test_lifecycle_publication.py @@ -0,0 +1,111 @@ +from dataclasses import replace +import json +import time + +import pytest + +from informed_decision.lifecycle_service import LifecycleError, LifecycleService +from informed_decision.store import Conflict, EvidenceUnavailable, Store +from test_lifecycle import AGENT, workspace +from test_lifecycle_service import caller, service +from test_policy_client import policy + + +AUDIENCE = {"subjects": ["reviewer"], "groups": []} + + +def test_publication_is_revision_pinned_durable_and_replayable(workspace): + app, memo = service(workspace) + actor = caller() + app.register(actor, memo, key="register") + receipt = app.publish(actor, memo.id, 1, audience=AUDIENCE, key="publish") + assert app.publish(actor, memo.id, 1, audience=AUDIENCE, key="publish") == receipt + state = app.repository.policy_context(AGENT.tenant, memo.id) + assert state["published_audience"] == AUDIENCE + reopened = type(app.repository)(Store(app.repository.store.path)) + assert reopened.policy_context(AGENT.tenant, memo.id) == state + assert reopened.events(AGENT.tenant, memo.id)[-1]["kind"] == "published" + assert app.policy.transport.calls[1][3]["context"]["publication_audience"] == AUDIENCE + app.revise(actor, memo.next_version(brief="revised"), expected_version=1, summary="Changed", key="revise") + assert app.repository.policy_context(AGENT.tenant, memo.id)["published_audience"] is None + assert app.repository.policy_context(AGENT.tenant, memo.id, 1)["published_audience"] == AUDIENCE + with pytest.raises(Conflict): + app.publish(actor, memo.id, 1, audience=AUDIENCE, key="new-publish") + with app.repository.store._connection() as db: + assert db.execute("SELECT count(*) FROM dispositions").fetchone()[0] == 0 + assert db.execute("SELECT count(*) FROM submissions").fetchone()[0] == 0 + + +@pytest.mark.parametrize("audience", [{}, {"public": True}, {"subjects": [], "groups": []}, + {"subjects": ["*"], "groups": []}, + {"subjects": "reviewer", "groups": []}]) +def test_publication_requires_explicit_bounded_audience(workspace, audience): + app, memo = service(workspace) + app.register(caller(), memo, key="register") + with pytest.raises(ValueError): + app.publish(caller(), memo.id, 1, audience=audience, key="publish") + assert len(app.repository.events(AGENT.tenant, memo.id)) == 1 + + +def test_audience_is_not_local_permission_and_denial_prevents_publication(workspace): + app, memo = service(workspace) + app.register(caller(), memo, key="register") + app.policy.transport.change = lambda data: data.update(effect="deny") + with pytest.raises(LifecycleError) as error: + app.publish(caller(), memo.id, 1, audience=AUDIENCE, key="publish") + assert error.value.status == 403 + assert app.repository.policy_context(AGENT.tenant, memo.id)["published_audience"] is None + + +def test_audience_change_during_authorization_blocks_read_and_write(workspace, monkeypatch): + app, memo = service(workspace) + app.register(caller(), memo, key="register") + original = app.repository.annotate + def changed(*args, **kwargs): + app.repository.publish(AGENT, memo.id, 1, audience=AUDIENCE, key="concurrent-publish") + return original(*args, **kwargs) + monkeypatch.setattr(app.repository, "annotate", changed) + with pytest.raises(LifecycleError, match="record_changed"): + app.annotate(caller(), memo.id, 1, kind="comment", body={"text": "pending"}, key="comment") + original_read = app.repository.retrieve + def changed_read(*args, **kwargs): + result = original_read(*args, **kwargs) + app.repository.publish(AGENT, memo.id, 1, audience={"subjects": ["another"], "groups": []}, key="replace") + return result + monkeypatch.setattr(app.repository, "retrieve", changed_read) + with pytest.raises(LifecycleError, match="record_changed"): + app.retrieve(caller(), memo.id) + + +def test_search_filters_denied_records_and_opaque_cursor_is_caller_scoped(workspace): + app, memo = service(workspace) + for index in range(51): + app.repository.register(AGENT, replace(memo, id=f"record-{index:03}"), key=f"register-{index}") + # Actual PolicyClient still validates these synthetic PDP responses. + def decide(data): + if data["binding"]["resource"]["id"] == "memo:record-001": + data["effect"] = "deny" + app.policy.transport.change = decide + first = app.search(caller()) + assert len(first["decisions"]) == 49 + assert "record-001" not in json.dumps(first) + assert first["next_cursor"] and "record-049" not in first["next_cursor"] + assert app.search(caller(), cursor=first["next_cursor"])["decisions"][0]["memo_id"] == "record-050" + with pytest.raises(ValueError): + app.search(caller(), query="different", cursor=first["next_cursor"]) + foreign = replace(caller(), attribution=replace(AGENT, tenant="tenant:other")) + with pytest.raises(ValueError): + app.search(foreign, cursor=first["next_cursor"]) + with pytest.raises(ValueError): + LifecycleService(app.repository, policy()).search(caller(), cursor=first["next_cursor"]) + app.policy.transport.change = lambda data: data.update(effect="deny") + with pytest.raises(LifecycleError): + app.search(caller(), cursor=first["next_cursor"]) + + +def test_search_does_not_turn_policy_failure_into_empty_success(workspace): + app, memo = service(workspace) + app.repository.register(AGENT, memo, key="register") + app.policy.transport.status = 503 + with pytest.raises(LifecycleError): + app.search(caller()) diff --git a/tests/test_lifecycle_ui.py b/tests/test_lifecycle_ui.py new file mode 100644 index 0000000..f5ceae7 --- /dev/null +++ b/tests/test_lifecycle_ui.py @@ -0,0 +1,92 @@ +import re +import uuid +from urllib.parse import urlencode + +import pytest + +from informed_decision.lifecycle_service import Caller +from informed_decision.oidc import ORIGIN +from informed_decision.web import App, SESSION_COOKIE +from test_browser_auth import login, signing_key, finish, call +from test_lifecycle import workspace +from test_lifecycle_service import service + + +@pytest.fixture +def browser(workspace, login): + lifecycle, _ = service(workspace) + sid = finish(login) + session = login[0].session(sid) + return App(login[0], lifecycle=lifecycle), lifecycle, session, f"{SESSION_COOKIE}={sid}" + + +def post(browser, path, **values): + app, _, session, cookie = browser + values.setdefault("csrf", session.csrf) + values.setdefault("key", str(uuid.uuid4())) + return call(app, path, "POST", cookie=cookie, origin=ORIGIN, body=urlencode(values).encode()) + + +def create(browser): + response = post(browser, "/decisions/new/prepare", question="Which storage option should we adopt?", + brief="Compare cost, recovery and maintenance.", scope_id="storage", scope_label="Storage strategy") + assert response["status"] == 303 + return dict(response["headers"])["Location"] + + +def test_human_workspace_create_refine_publish_discuss_and_retrieve(browser): + app, lifecycle, session, cookie = browser + assert 'Open the decision workspace' in call(app, cookie=cookie)["body"] + path = create(browser) + page = call(app, path, cookie=cookie) + assert page["status"] == 200 and "Storage strategy" in page["body"] + assert "Human acceptance is not available" in page["body"] + assert post(browser, path + "/comment", version="1", text="Please include restore time.")["status"] == 303 + assert post(browser, path + "/publish", version="1", subjects=session.subject, groups="")["status"] == 303 + assert "published" in call(app, path, cookie=cookie)["body"] + assert post(browser, path + "/revise", version="1", question="Which storage option should we adopt?", + brief="Option A restores in one hour.", summary="Added recovery estimate")["status"] == 303 + assert "Revision 2" in call(app, path, cookie=cookie)["body"] + assert "Compare cost" in call(app, path, cookie=cookie, query="version=1")["body"] + listing = call(app, "/decisions", cookie=cookie, query="q=storage") + assert listing["status"] == 200 and path in listing["body"] + memo_id = path.rsplit("/", 1)[1] + assert lifecycle.repository.policy_context(session.tenant.value, memo_id)["published_audience"] is None + assert lifecycle.repository.store.lifecycle_counts()["pending"] == 4 + + +def test_same_create_form_replay_reuses_record(browser): + fields = {"question": "Which option?", "brief": "Consider alternatives", "scope_id": "target", + "scope_label": "Target", "key": str(uuid.uuid4())} + first = post(browser, "/decisions/new/prepare", **fields) + second = post(browser, "/decisions/new/prepare", **fields) + assert dict(first["headers"])["Location"] == dict(second["headers"])["Location"] + assert browser[1].repository.store.lifecycle_counts()["records"] == 1 + + +def test_csrf_and_origin_are_enforced_before_any_mutation(browser): + app, lifecycle, session, cookie = browser + values = {"question": "Which option?", "brief": "Consider alternatives", "scope_id": "target", + "scope_label": "Target", "key": str(uuid.uuid4()), "csrf": session.csrf} + for origin, csrf in [(None, session.csrf), ("https://other.test", session.csrf), (ORIGIN, "wrong")]: + response = call(app, "/decisions/new/prepare", "POST", cookie=cookie, origin=origin, + body=urlencode({**values, "csrf": csrf}).encode()) + assert response["status"] == 403 + assert lifecycle.repository.store.lifecycle_counts()["records"] == 0 + + +def test_workspace_is_opt_in_and_requires_human_session(browser): + app, _, _, cookie = browser + assert call(app, "/decisions")["status"] == 401 + assert call(App(app.login), "/decisions", cookie=cookie)["status"] == 404 + + +def test_workspace_escapes_content_and_policy_failure_returns_no_memo(browser): + app, lifecycle, _, cookie = browser + path = create(browser) + post(browser, path + "/comment", version="1", text='') + page = call(app, path, cookie=cookie)["body"] + assert "