From 12430dec86a82809800a9136dd0d86e2353c7998 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 1 Nov 2026 22:03:13 +0100 Subject: [PATCH] feat: add ordered sheet execution for admitted case steps Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a198e4-7903-75b3-aaf3-c89ddac9437d --- src/warden/case_sheet.py | 255 ++++++++++++ tests/test_case_sheet.py | 390 ++++++++++++++++++ ...EN-WP-0044-security-case-follow-through.md | 20 +- 3 files changed, 663 insertions(+), 2 deletions(-) create mode 100644 src/warden/case_sheet.py create mode 100644 tests/test_case_sheet.py diff --git a/src/warden/case_sheet.py b/src/warden/case_sheet.py new file mode 100644 index 0000000..71ae020 --- /dev/null +++ b/src/warden/case_sheet.py @@ -0,0 +1,255 @@ +"""Ordered sheet over steps already admitted by the execution kernel. + +This is not ``warden case run`` and it does not make a case live. Entry-point +prose is never executed, no adapter is installed, and a closed sheet is only +local reducer closure. Native acceptance stays on owner descriptors, structural +materialization, and a real first run. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass + +from .case_receipt_state import Outcome, Refused +from .case_run import IDENTIFIER + +_CLOSED = frozenset({Outcome.SUCCESS, Outcome.NO_CHANGE}) +_BLOCKING = frozenset( + { + Outcome.UNCERTAIN, + Outcome.FAILED_AFTER, + Outcome.RUNNING, + Outcome.ROLLBACK_RUNNING, + Outcome.ROLLBACK_UNCERTAIN, + } +) +_CHECK = frozenset({Outcome.SUCCESS, Outcome.FAILED_BEFORE, Outcome.UNCERTAIN}) + + +@dataclass(frozen=True, slots=True) +class SheetReceipt: + """Metadata only. ``closed`` is not a case status and not native acceptance.""" + + case_pin: str + steps: tuple[tuple[str, str], ...] + verifications: tuple[tuple[str, str], ...] + closed: bool + + +def run_sheet( + executor, + steps: Sequence[tuple], + *, + session: Callable[[str, Callable[[], None]], None] | None = None, + verifications: Sequence[tuple[str, Callable[[], Outcome]]] = (), +) -> SheetReceipt: + """Run admitted steps in order. Stop unless the outcome is closed. + + Consecutive attended steps on one lane share a single ``session(lane, + dispatch)`` call. ``dispatch`` is the only path that may run them. A + missing session, or a session that never calls ``dispatch``, refuses + before any later step. + """ + loaded = _prepare(executor, steps, verifications) + if any(_mode(executor, triple) == "attended" for triple in loaded): + if not callable(session): + raise Refused("attended sheet requires a contained session") + states: list = [] + for key, group in _groups(executor, loaded): + if key[0] == "attended": + produced: list = [] + + def dispatch(group=group, produced=produced) -> None: + for triple in group: + state = executor.run(*triple) + produced.append(state) + if state.last is None or state.last.outcome not in _CLOSED: + return + + _in_session(session, key[1], dispatch) + if not produced: + raise Refused("attended steps left the contained session") + states.extend(produced) + else: + for triple in group: + states.append(executor.run(*triple)) + if states[-1].last is None or states[-1].last.outcome not in _CLOSED: + break + return _receipt(loaded, states, _verify(loaded, states, verifications)) + + +def rollback_sheet( + executor, + steps: Sequence[tuple], + *, + session: Callable[[str, Callable[[], None]], None] | None = None, +) -> tuple: + """Undo verified creations in reverse order. Uncertain steps block the sheet.""" + from .case_receipt_state import recover + + loaded = _prepare(executor, steps, ()) + current = [] + for triple in loaded: + binding = f"{triple[1].step}:{triple[1].target}" + current.append(recover(executor.journal.load(binding, triple[2]))) + if any( + state.last is not None and state.last.outcome in _BLOCKING for state in current + ): + raise Refused("sheet rollback needs owner reconciliation first") + actions = [] + for triple, state in zip(reversed(loaded), reversed(current)): + last = state.last + if last is None or last.outcome in ( + Outcome.NO_CHANGE, + Outcome.FAILED_BEFORE, + Outcome.ROLLED_BACK, + ): + continue + if last.outcome is not Outcome.SUCCESS or not last.created: + raise Refused("sheet rollback needs a verified successful creation receipt") + actions.append(triple) + if ( + any(_mode(executor, triple) == "attended" for triple in actions) + and session is None + ): + raise Refused("attended sheet requires a contained session") + results = [] + index = 0 + while index < len(actions): + if _mode(executor, actions[index]) != "attended": + state = executor.rollback(*actions[index]) + results.append(state) + if state.last is None or state.last.outcome is not Outcome.ROLLED_BACK: + break + index += 1 + continue + lane = actions[index][1].lane + batch = [] + while ( + index < len(actions) + and _mode(executor, actions[index]) == "attended" + and actions[index][1].lane == lane + ): + batch.append(actions[index]) + index += 1 + produced: list = [] + + def dispatch(batch=batch, produced=produced) -> None: + for triple in batch: + state = executor.rollback(*triple) + produced.append(state) + if state.last is None or state.last.outcome is not Outcome.ROLLED_BACK: + return + + _in_session(session, lane, dispatch) + if not produced: + raise Refused("attended steps left the contained session") + results.extend(produced) + if ( + produced[-1].last is None + or produced[-1].last.outcome is not Outcome.ROLLED_BACK + ): + break + return tuple(results) + + +def _prepare(executor, steps, verifications) -> tuple: + if isinstance(steps, (str, bytes)) or not isinstance(steps, Sequence) or not steps: + raise Refused("sheet has no admitted steps") + if isinstance(verifications, (str, bytes)) or not isinstance( + verifications, Sequence + ): + raise Refused("verification inventory is not a closed list") + loaded = tuple(steps) + seen = [] + for triple in loaded: + if not isinstance(triple, tuple) or len(triple) != 3: + raise Refused("sheet step is not an admitted triple") + descriptor, invocation, initial = triple + stored = executor._admitted.get(invocation, ()) + if stored[:2] != (descriptor, initial): + raise Refused("sheet step is not admitted on this executor") + seen.append(invocation.step) + if len(seen) != len(set(seen)): + raise Refused("duplicate sheet step") + pins = {triple[1].pins for triple in loaded} + if ( + len({(pin.case, pin.case_type, pin.catalog, pin.acceptance) for pin in pins}) + != 1 + ): + raise Refused("sheet steps do not share case type catalog and acceptance pins") + for item in verifications: + if ( + not isinstance(item, tuple) + or len(item) != 2 + or type(item[0]) is not str + or not IDENTIFIER.fullmatch(item[0]) + or not callable(item[1]) + ): + raise Refused("verification id or check is not closed") + return loaded + + +def _in_session(session, lane: str, dispatch) -> None: + if not callable(session): + raise Refused("attended sheet requires a contained session") + session(lane, dispatch) + + +def _mode(executor, triple) -> str: + return executor._admitted[triple[1]][3] + + +def _groups(executor, steps: tuple) -> list[tuple[tuple[str, str], tuple]]: + groups: list[tuple[tuple[str, str], list]] = [] + for triple in steps: + mode = _mode(executor, triple) + key = ( + ("attended", triple[1].lane) + if mode == "attended" + else (mode, triple[1].step) + ) + if groups and groups[-1][0] == key and key[0] == "attended": + groups[-1][1].append(triple) + else: + groups.append((key, [triple])) + return [(key, tuple(group)) for key, group in groups] + + +def _verify(steps, states, verifications) -> list[tuple[str, str]]: + if len(states) != len(steps) or any( + state.last is None or state.last.outcome not in _CLOSED for state in states + ): + return [] + rows = [] + for check_id, check in verifications: + try: + outcome = check() + except (KeyboardInterrupt, SystemExit, Refused): + raise + except Exception: + outcome = Outcome.UNCERTAIN + if type(outcome) is not Outcome or outcome not in _CHECK: + outcome = Outcome.UNCERTAIN + rows.append((check_id, outcome.value)) + if outcome is not Outcome.SUCCESS: + break + return rows + + +def _receipt(steps, states, verifications) -> SheetReceipt: + return SheetReceipt( + steps[0][1].pins.case, + tuple( + (state.step_id, state.last.outcome.value) + for state in states + if state.last is not None + ), + tuple(verifications), + len(states) == len(steps) + and all( + state.last is not None and state.last.outcome in _CLOSED for state in states + ) + and all(outcome == Outcome.SUCCESS.value for _, outcome in verifications), + ) diff --git a/tests/test_case_sheet.py b/tests/test_case_sheet.py new file mode 100644 index 0000000..4b63909 --- /dev/null +++ b/tests/test_case_sheet.py @@ -0,0 +1,390 @@ +"""Ordered sheet over admitted steps. Fixture adapters only; nothing native.""" + +from dataclasses import asdict, replace +import json + +import pytest +import yaml + +from warden.case_receipt_state import Outcome, Refused +from warden.case_run import Adapter, Descriptor, Executor, Journal, OwnerResult, digest +from warden.case_sheet import rollback_sheet, run_sheet +from tests.test_case_run import lane_entry + + +SOURCE = "sha256:" + "1" * 64 + + +def _result(inv, outcome, created=()): + return OwnerResult( + outcome, + "owner-receipt", + created, + True, + session_revocation_evidence="session-revoked", + grant_revocation_evidence="grant-revoked", + attempt_id=inv.attempt_id, + operation=inv.operation, + ) + + +def _spec(step, mode, outcome=Outcome.SUCCESS, created=True, lane=None): + return { + "step": step, + "mode": mode, + "outcome": outcome, + "created": created, + "lane": lane, + } + + +def admit_sheet(tmp_path, specs, *, case_id="case-sheet", executor=None): + """Admit fixture steps on one executor. No step is dispatched.""" + modes = { + "warden": ( + "issuer-lane", + True, + False, + "mint", + ("parent-session", "broker-grant"), + None, + ), + "attended": ("login-lane", False, True, "change", ("parent-session",), "login"), + "owner": ("issuer-lane", False, False, "change", ("parent-session",), None), + } + descriptors = {} + follow = [] + entries = {} + adapters = {} + calls = [] + undos = [] + for spec in specs: + lane, executes, attended, effect, scopes, kind = modes[spec["mode"]] + lane = spec["lane"] or lane + if spec["mode"] == "attended" and spec["lane"]: + kind = "login" + object_id = "obj-" + spec["step"] + desc = Descriptor( + "warden.case.execution.v1", + spec["step"], + "owner", + "owner/" + spec["step"] + "@v1", + SOURCE, + "human", + lane, + "client_id", + effect, + (object_id,), + False, + scopes, + ) + encoded = json.dumps(asdict(desc)).encode() + descriptors[spec["step"]] = (desc, encoded) + follow.append( + { + "id": spec["step"], + "owner": "owner", + "lane": lane, + "warden_executes": executes, + "attended": attended, + } + ) + entry = lane_entry(lane) + if kind: + entry["lane"] = kind + entries[lane] = entry + + def invoke(inv, spec=spec, object_id=object_id): + calls.append(spec["step"]) + created = ( + (object_id,) + if spec["created"] and spec["outcome"] is Outcome.SUCCESS + else () + ) + if spec["outcome"] == "raise": + raise RuntimeError("synthetic lost reply") + return _result(inv, spec["outcome"], created) + + def undo(inv, objects, spec=spec): + undos.append(spec["step"]) + if spec.get("undo") == "raise": + raise RuntimeError("synthetic lost undo") + return _result(inv, Outcome.ROLLED_BACK, ()) + + adapters[desc.adapter] = Adapter( + desc.adapter, SOURCE, invoke, lambda *_: True, undo + ) + + typ = yaml.safe_dump( + { + "spec": { + "id": "test", + "version": "1", + "status": "acknowledged", + "parameters": [ + { + "name": "client_id", + "sensitivity": "operational", + "case_settable": True, + } + ], + "execution_descriptors": { + step: digest(encoded) for step, (_, encoded) in descriptors.items() + }, + "follow_through": follow, + } + } + ).encode() + case = yaml.safe_dump( + { + "spec": { + "id": case_id, + "type": "test@1", + "status": "materialized", + "acceptance": "acceptance", + "parameters": {"client_id": "fixture-client"}, + } + } + ).encode() + catalog = yaml.safe_dump({"entries": list(entries.values())}).encode() + if executor is None: + executor = Executor(Journal(tmp_path / "private"), adapters, lambda *_: True) + else: + executor.adapters.update(adapters) + admitted = [] + for spec in specs: + desc, encoded = descriptors[spec["step"]] + admitted.append( + executor.admit( + case, + typ, + encoded, + catalog, + principal="operator", + actor="human", + lane=desc.lane, + acceptance_evidence="acceptance", + materialization_evidence="materialization", + descriptor_evidence="descriptor", + session_evidence="session", + ) + ) + return executor, tuple(admitted), calls, undos + + +def _session(events): + def session(lane, dispatch): + events.append(("session", lane)) + dispatch() + events.append(("session-end", lane)) + + return session + + +def test_stops_before_later_steps_when_a_step_does_not_close(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, calls, _ = admit_sheet( + tmp_path, + [ + _spec("mint-secret", "warden", Outcome.FAILED_BEFORE, created=False), + _spec("bind-env", "warden"), + ], + ) + seen = [] + receipt = run_sheet( + executor, + steps, + verifications=(("consumer-enforces-groups", lambda: seen.append(1)),), + ) + assert calls == ["mint-secret"] + assert seen == [] + assert receipt.closed is False + assert receipt.steps == (("mint-secret", "failed-before-change"),) + assert receipt.verifications == () + + +def test_attended_steps_on_one_lane_share_one_session(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, calls, _ = admit_sheet( + tmp_path, + [ + _spec("ensure-groups", "attended"), + _spec("bind-env", "attended"), + _spec("mint-secret", "warden"), + ], + ) + events = [] + + def session(lane, dispatch): + events.append(("session", lane, tuple(calls))) + dispatch() + events.append(("session-end", lane, tuple(calls))) + + receipt = run_sheet(executor, steps, session=session) + assert events == [ + ("session", "login-lane", ()), + ("session-end", "login-lane", ("ensure-groups", "bind-env")), + ] + assert calls == ["ensure-groups", "bind-env", "mint-secret"] + assert receipt.closed is True + assert receipt.verifications == () + + +def test_different_lanes_and_a_break_are_separate_sessions(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, calls, _ = admit_sheet( + tmp_path, + [ + _spec("ensure-groups", "attended"), + _spec("mint-secret", "warden"), + _spec("register-client", "attended", lane="other-login"), + ], + ) + events = [] + run_sheet(executor, steps, session=_session(events)) + assert [lane for kind, lane in events if kind == "session"] == [ + "login-lane", + "other-login", + ] + assert calls == ["ensure-groups", "mint-secret", "register-client"] + + +def test_attended_dispatch_outside_the_session_is_refused(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, calls, _ = admit_sheet( + tmp_path, [_spec("ensure-groups", "attended")] + ) + with pytest.raises(Refused, match="contained session"): + run_sheet(executor, steps) + assert calls == [] + + def silent(lane, dispatch): + return None + + with pytest.raises(Refused, match="left the contained session"): + run_sheet(executor, steps, session=silent) + assert calls == [] + + +def test_verification_runs_only_after_every_step_closes(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, _, _ = admit_sheet( + tmp_path, [_spec("mint-secret", "warden"), _spec("bind-env", "owner")] + ) + checks = [] + + def check(outcome): + def run(): + checks.append(outcome) + return outcome + + return run + + receipt = run_sheet( + executor, + steps, + verifications=( + ("custody-allowed", check(Outcome.SUCCESS)), + ("consumer-enforces-groups", check(Outcome.FAILED_BEFORE)), + ("not-reached", check(Outcome.SUCCESS)), + ), + ) + assert [item.value for item in checks] == [ + "verified-success", + "failed-before-change", + ] + assert receipt.closed is False + assert receipt.verifications[-1][0] == "consumer-enforces-groups" + + +def test_unknown_verification_result_is_uncertain_and_not_stored_as_text( + tmp_path, monkeypatch +): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, _, _ = admit_sheet(tmp_path, [_spec("mint-secret", "warden")]) + + def leak(): + raise RuntimeError("synthetic-secret-value") + + receipt = run_sheet(executor, steps, verifications=(("readback", leak),)) + assert receipt.closed is False + assert receipt.verifications == (("readback", "uncertain"),) + assert "synthetic-secret-value" not in repr(receipt) + + def wrong(): + return "verified-success" + + again, again_steps, _, _ = admit_sheet( + tmp_path / "again", [_spec("mint-secret", "warden")] + ) + receipt = run_sheet(again, again_steps, verifications=(("readback", wrong),)) + assert receipt.verifications == (("readback", "uncertain"),) + + +def test_refuses_unadmitted_duplicate_and_foreign_pins(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, calls, _ = admit_sheet(tmp_path, [_spec("mint-secret", "warden")]) + moved = (steps[0][0], replace(steps[0][1], target="other-target"), steps[0][2]) + with pytest.raises(Refused, match="not admitted"): + run_sheet(executor, (moved,)) + with pytest.raises(Refused, match="duplicate"): + run_sheet(executor, (steps[0], steps[0])) + _, others, _, _ = admit_sheet( + tmp_path, + [_spec("bind-env", "owner")], + case_id="other-case", + executor=executor, + ) + with pytest.raises(Refused, match="do not share"): + run_sheet(executor, (steps[0], others[0])) + assert calls == [] + + +def test_rollback_skips_no_change_and_stops_on_uncertain_undo(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, _, undos = admit_sheet( + tmp_path, + [ + _spec("mint-secret", "warden"), + _spec("bind-env", "owner", Outcome.NO_CHANGE, created=False), + dict(_spec("register-client", "owner"), undo="raise"), + ], + ) + assert run_sheet(executor, steps).closed is True + done = rollback_sheet(executor, steps) + assert undos == ["register-client"] + assert done[-1].last.outcome is Outcome.ROLLBACK_UNCERTAIN + assert "mint-secret" not in undos + + +def test_uncertain_step_blocks_sheet_rollback(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, _, undos = admit_sheet( + tmp_path, + [ + _spec("mint-secret", "warden"), + _spec("bind-env", "owner", outcome="raise", created=False), + ], + ) + receipt = run_sheet(executor, steps) + assert receipt.steps[-1][1] == "uncertain" + with pytest.raises(Refused, match="owner reconciliation"): + rollback_sheet(executor, steps) + assert undos == [] + + +def test_attended_rollback_stays_inside_the_session(tmp_path, monkeypatch): + monkeypatch.delenv("WARDEN_AGENT_ID", raising=False) + executor, steps, _, undos = admit_sheet( + tmp_path, + [_spec("ensure-groups", "attended"), _spec("bind-env", "attended")], + ) + run_sheet(executor, steps, session=_session([])) + with pytest.raises(Refused, match="contained session"): + rollback_sheet(executor, steps) + assert undos == [] + events = [] + done = rollback_sheet(executor, steps, session=_session(events)) + assert [lane for kind, lane in events if kind == "session"] == ["login-lane"] + assert undos == ["bind-env", "ensure-groups"] + assert all(state.last.outcome is Outcome.ROLLED_BACK for state in done) diff --git a/workplans/WARDEN-WP-0044-security-case-follow-through.md b/workplans/WARDEN-WP-0044-security-case-follow-through.md index 6c09ae9..7011e52 100644 --- a/workplans/WARDEN-WP-0044-security-case-follow-through.md +++ b/workplans/WARDEN-WP-0044-security-case-follow-through.md @@ -9,7 +9,7 @@ flavor: implementation owner: claude-code topic_slug: netkingdom created: "2026-10-08" -updated: "2026-10-11" +updated: "2026-10-22" depends_on: [] related_repos: - net-kingdom @@ -144,7 +144,7 @@ reaches the plan unrendered. id: WARDEN-WP-0044-T03 status: progress depends_on: [WARDEN-WP-0044-T02, MASON-WP-0010-T03] -blocking_reason: "Local durable execution kernel and injected failure harness implemented. Native acceptance still requires authenticated owner machine descriptors/adapters, actual Forgejo materialization receipt, contained session integration, ordered verification and first native run." +blocking_reason: "Ordered admitted-step sheet exists. Native acceptance still requires authenticated owner machine descriptors/adapters, actual Forgejo materialization receipt, a real contained-login session, promoted lanes, the 1.3 pin after founder acknowledgment, and the first native run." priority: high state_hub_task_id: "bcbf6c91-2c04-5c16-ae8c-5d6c06fd6b89" ``` @@ -408,3 +408,19 @@ warden-safe-unit-integration-r1. This completes the known trusted-unit-entry accident-prevention follow-through; direct pytest bypass, native/non-open/preopened file-descriptor limits remain explicit. No OS sandbox, isolated agent assignment, production grant or native T03 acceptance follows. T03 stays progress. + +2026-10-22: `src/warden/case_sheet.py` runs steps already admitted on one +Executor, in order, and stops unless the outcome is verified-success or +verified-no-change. Consecutive attended steps on the same lane share one +injected contained-session call; a session that does not perform the dispatch +refuses, and no later step runs. Verification callbacks run only after every +step has closed. A raised check becomes uncertain and its text is not stored. +`SheetReceipt.closed` means that local sheet finished. It does not set a case +to `live` and it is not native acceptance. Rollback walks the same admitted +order in reverse, skips no-change and failed-before-change, and refuses while +any step is uncertain or failed-after-change. There is no CLI, no shell, and +no use of entry-point prose. Tests: `tests/test_case_sheet.py` (10 passed +under `make test`). Owner machine descriptors, Forgejo materialization, +installed adapters, promoted lanes, the 1.3 parameter pin after founder +acknowledgment, the real contained login, and the first native run remain +open. T03 stays progress.