Freeze released bundle, control-plan and apply-crash workflows (#271)

docs/architecture/persisted-contracts.md

@@ -53,7 +53,7 @@ A golden over the current writer does not protect these. A reader can stop accep
 |---|---|---|---|
 | Artifacts | run `manifest.json`, `result.json`, `winner.patch`, apply receipts (`applications/*.json`), in every state a standalone `run` leaves them (room-, session- and workflow-origin artifacts are frozen with those families) | `runs list` / `show` / `prune`, `conversations list` / `show`, `diff`, `apply`, `recover` (`_bound_receipt`), `continue`, the TUI artifact view | `test_frozen_artifacts.py` (v0.1.2, v0.3.0) |
 | Rooms | the room database (`rooms.sqlite3`, with the WAL a killed process leaves): rooms, rounds, candidates, members, workspaces, events, `sessions_json`, `selection_json` v1, `provenance_json` v1/v2, `conversation_json` v1; candidate `manifest.json`, patch and output; room-bound run artifacts. No release wrote `conversation_json` v2 or judgment rows | `room list` / `rounds` / `show` / `candidates` / `diff` / `compare` / `export`, `select`, `apply`, `judge`, `recover`, `close`, `prune` `RoomStore.member_sessions` (and through these `RoomCandidateStore.load`, `_parse_sessions_json`, `selection_record_status`, `room_outcome_status`, `round_provenance_status`, `round_conversation_status`); the standalone `apply` / `recover` refusal and skip for room-bound artifacts; a HEAD `room round` on a released room (`branch-from` resuming the released sessions on the released base patch, `fresh`, a first round on a draft, and the `continue-own` refusal when the released workspaces are gone); `room attach` over the released member log paths (tmux faked); the `apply --verify` settings notices; raw export (`--include` preview and confirm); the TUI Rooms tab's readers and pure renders (`build_room_evidence`, `render_room`, `render_round_table`, and the Diff, Output and Details views), not the Textual app | `test_frozen_rooms.py` (v0.1.2, v0.2.0, v0.3.0) |
-| Workflows | the workflow database (`workflows.db`): rows with `spec_json`, manifest `schema_version` 3 with `calls_used`, events, and from v0.3.0 the lineage columns and `file`/`revision` origins; per workflow `spec.json`, planner output, node bundles (`result.json`, `winner.patch`, `commit.json` v3) and stream logs; workflow-origin run artifacts, approved or not, with `plan.final_agent` from v0.3.0. No release wrote manifest v1 or v2 (both normalize to v3 before the insert); those readers stay covered by `test_workflow_durable.py::test_marked_v1_migration_preserves_exact_count_and_control_fields` and `test_workflow.py::test_a_workflow_whose_call_accounting_is_ambiguous_is_refused` | `workflow list` / `show` (`--prompts`) / `status` / `logs` / `lineage` / `export` / `cancel` / `prune`, a dead release-written lease recovered by `show` and `status`, `WorkflowStore.events` and `read_node_bundle`; for the final artifact `diff`, `runs show`, `apply` (approved, after a prune, approval healed from the database, refused for a workflow that never completed, `no_changes`), the standalone `recover` skip. A final artifact naming another run, and an unreadable store: `test_cli.py::test_apply_refuses_an_artifact_the_named_workflow_did_not_finalize` and `test_apply_guard_fails_closed_when_the_workflow_store_cannot_be_read`. HEAD execution (`WorkflowCoordinator.run`, what the job worker calls): `run` of released planned, created and revised rows (accounting, stored prompts, dependency output); `resume` of released crashes and failures against the stored deadline (expired: fails before dispatch; before it: the completed dependency's bundle is reused and the unfinished node's attempt refunded, a published final artifact is reused and approved without backfilling `final_agent`, an exhausted node dispatches nothing), refused from a final or unstarted state and while the owner lives, with the released execution history and events kept; `workflow revise` (a v0.2.0 parent gains a lineage); the launch's team peek and state check; `apply --verify` narrowing (v0.2.0: the gate's list only, with a notice; also after a prune); control plans enabled refuse a released workflow planned without one. Not yet: control-plan (toolgraph) workflows written by a release, a node bundle saved just before a crash (reclaimed on resume), recovery of an interrupted workflow-origin apply | `test_frozen_workflows.py` (v0.2.0, v0.3.0), partly |
+| Workflows | the workflow database (`workflows.db`): rows with `spec_json`, manifest `schema_version` 3 with `calls_used`, events, and from v0.3.0 the lineage columns and `file`/`revision` origins; per workflow `spec.json`, planner output, node bundles (`result.json`, `winner.patch`, `commit.json` v3) and stream logs; workflow-origin run artifacts, approved or not, with `plan.final_agent` from v0.3.0. No release wrote manifest v1 or v2 (both normalize to v3 before the insert); those readers stay covered by `test_workflow_durable.py::test_marked_v1_migration_preserves_exact_count_and_control_fields` and `test_workflow.py::test_a_workflow_whose_call_accounting_is_ambiguous_is_refused` | `workflow list` / `show` (`--prompts`) / `status` / `logs` / `lineage` / `export` / `cancel` / `prune`, a dead release-written lease recovered by `show` and `status`, `WorkflowStore.events` and `read_node_bundle`; for the final artifact `diff`, `runs show`, `apply` (approved, after a prune, approval healed from the database, refused for a workflow that never completed, `no_changes`), the standalone `recover` skip. A final artifact naming another run, and an unreadable store: `test_cli.py::test_apply_refuses_an_artifact_the_named_workflow_did_not_finalize` and `test_apply_guard_fails_closed_when_the_workflow_store_cannot_be_read`. HEAD execution (`WorkflowCoordinator.run`, what the job worker calls): `run` of released planned, created and revised rows (accounting, stored prompts, dependency output); `resume` of released crashes and failures against the stored deadline (expired: fails before dispatch; before it: the completed dependency's bundle is reused and the unfinished node's attempt refunded, a published final artifact is reused and approved without backfilling `final_agent`, an exhausted node dispatches nothing), refused from a final or unstarted state and while the owner lives, with the released execution history and events kept; `workflow revise` (a v0.2.0 parent gains a lineage); the launch's team peek and state check; `apply --verify` narrowing (v0.2.0: the gate's list only, with a notice; also after a prune); control plans enabled refuse a released workflow planned without one. Written later by the same releases (`freeze.py --part 2b`, `bindings-2b.json`): a node bundle saved just before the crash, reclaimed on resume through `read_node_result` / `read_node_bundle` with its usage added once (`_record_usage`), and re-run when its patch, its commit record or its digest disagrees; control-plan workflows, whose released `control-plan.json` is the control preflight's request, whose binding and validation topology refuse a changed configuration, and whose stored limits and node timeouts bound a run with control plans on, off, or Toolgraph off (`_execution_limits`), plus `show`, `read_control_plan`, `conversations show` and `apply` of a completed one; an interrupted apply of a final artifact, classified by `recover` (completed, not applied, manual recovery; `_bound_receipt`) without touching its approval or the workflow row. Not covered: the TUI's `_read_artifact_result`; no reader revalidates a released execution's `control-preflight.json` | `test_frozen_workflows.py` (v0.2.0, v0.3.0) |
 | Sessions and project | isolated-session documents and their checkpoint run artifacts, project marker `.syncmill/project.json` | `IsolatedSessionStore.load` / `list`, `_marker_project_id`; the artifact readers for checkpoint runs | not yet |
 | Board | item rows and their evidence and inputs, decisions | `_row_to_item`, `_load_evidence`, `_row_to_decision` | not yet |
 | Memory backends | Mem0 and LangGraph stored payloads | the backends' read paths | not yet |

tests/fixtures/persisted/workflows/freeze.py

@@ -32,6 +32,23 @@
 copied byte for byte once every writer has stopped, including any WAL a
 killed process leaves. Workspaces and TUI job records are not copied.
 `bindings.json` records the values bound to the generating machine.
+
+`--part 2b` adds scenarios to an existing release directory without writing
+anything already in it: each new scenario directory and `bindings-2b.json`
+must not exist. Its scenarios, from a fresh box of their own:
+- `bundle-saved`: the worker exits right after the final node's bundle is
+  saved, before the manifest records the node's outcome; then
+  `bundle-saved-recovered`, after the release's own `status`. The fake kimi
+  reports token usage in these, so the bundle carries some.
+- `control-planned`, `control-completed`: with Toolgraph control plans on and
+  a fake `toolgraph` that answers both of its protocols from the request.
+- `apply-unreceipted`, `apply-receipted`: a completed workflow's final
+  artifact applied in the release interpreter, dying right after the
+  `applying` record or right before the `applied` one (the artifacts family's
+  crash script).
+
+    python tests/fixtures/persisted/workflows/freeze.py --part 2b \\
+        --release-root /tmp/wt-v0.3.0 --out tests/fixtures/persisted/workflows/v0.3.0
 """
 
 from __future__ import annotations
@@ -102,6 +119,9 @@
         {"role": "meta", "type": "session.resume_hint", "session_id": "fake-kimi-session",
          "content": "To resume this session: kimi -r fake-kimi-session"},
     ]
+    if control.get("kimi_usage"):
+        events.append({"role": "meta", "type": "usage",
+                       "usage": {"input_tokens": 7, "output_tokens": 3}})
 for event in events:
     print(json.dumps(event))
 '''
@@ -141,10 +161,98 @@ def save(self, *args, **kwargs):
                 os._exit(0)
             return saved
         ArtifactStore.save = save
+    elif point == "bundle":
+        from syncmill.workflow import WorkflowStore
+
+        original_save = WorkflowStore.save_node_result
+
+        def save_node_result(self, workflow_id, node_id, *args, **kwargs):
+            saved = original_save(self, workflow_id, node_id, *args, **kwargs)
+            if node_id == "fix":
+                os._exit(0)  # bundle on disk, the node's outcome never recorded
+            return saved
+        WorkflowStore.save_node_result = save_node_result
 
 
 _install()
 '''
+# Both Toolgraph protocols (`preflight.py` is identical in v0.2.0, v0.3.0 and
+# HEAD), each answered from the request it gets, never a fixed golden.
+FAKE_TOOLGRAPH = '''\
+import hashlib, json, sys
+from pathlib import Path
+args = sys.argv[1:]
+out = Path(args[args.index("--out") + 1])
+profile = args[args.index("--profile") + 1]
+if args[0] == "control-preflight":
+    raw = Path(args[1]).read_bytes()
+    plan = json.loads(raw)
+    doc = {
+        "schema_version": 1, "kind": "toolgraph.control-preflight",
+        "run_id": plan["run_id"], "created_at": "2026-11-10T00:00:00+00:00",
+        "mode": "advisory", "profile": profile,
+        "plan_digest": "sha256:" + hashlib.sha256(raw).hexdigest(),
+        "graph_state": {"instance_id": "frozen-graph", "generation": 7},
+        "decision": "advisory_allow",
+        "evaluations": [
+            {"principal": row["id"], "agent_found": True,
+             "eligible": list(row["candidate_tools"]), "rejected": []}
+            for row in plan["principals"]
+        ],
+        "findings": [],
+    }
+elif args[0] == "preflight":
+    agent, tools = args[1], args[2:args.index("--profile")]
+    doc = {
+        "schema_version": 1, "kind": "toolgraph.preflight",
+        "run_id": args[args.index("--run-id") + 1],
+        "created_at": "2026-11-10T00:00:00+00:00", "graph_generation": 7,
+        "agent": agent, "agent_found": True, "profile": profile, "mode": "advisory",
+        "eligible": list(tools), "rejected": [], "features": [],
+        "decision": "advisory_allow",
+    }
+else:
+    sys.exit("fake toolgraph: unknown command " + repr(args[:1]))
+out.write_text(json.dumps(doc), encoding="utf-8")
+'''
+TOOLGRAPH_ENV = {
+    "SYNCMILL_INTEGRATION__TOOLGRAPH__ENABLED": "true",
+    "SYNCMILL_INTEGRATION__TOOLGRAPH__CONTROL_PLAN_ENABLED": "true",
+    "SYNCMILL_INTEGRATION__TOOLGRAPH__CANDIDATE_TOOLS": json.dumps(
+        ["workspace::read", "workspace::write"]
+    ),
+    "SYNCMILL_INTEGRATION__TOOLGRAPH__AGENT_MAP": json.dumps(
+        {"codex": "codex", "kimi-code": "kimi"}
+    ),
+}
+
+# The artifacts family's crash script: runs inside the released interpreter,
+# applies, then dies at one manifest write.
+CRASH_APPLY = '''\
+import os, sys
+from pathlib import Path
+from syncmill.artifacts import ArtifactStore
+from syncmill.worktree import repository_fingerprint
+runs_root, project, run_id, repo, die_at = sys.argv[1:6]
+original = ArtifactStore._compare_and_swap_manifest
+def die(self, run_dir, expected, replacement):
+    state = replacement.get("state")
+    if die_at == "before-applied" and state == "applied":
+        os._exit(0)  # receipt written, patch in the source, flip never happens
+    original(self, run_dir, expected, replacement)
+    if die_at == "after-applying" and state == "applying":
+        os._exit(0)  # applying recorded, source not yet touched, no receipt
+ArtifactStore._compare_and_swap_manifest = die
+store = ArtifactStore(Path(runs_root), project)
+store.apply(run_id, Path(repo), repository_fingerprint(Path(repo)))
+sys.exit("apply returned; the crash hook never fired")
+'''
+
+PART_2B = (
+    "bundle-saved", "bundle-saved-recovered", "control-planned", "control-completed",
+    "apply-unreceipted", "apply-receipted",
+)
+
 HOOK_MODULE = "_syncmill_freeze_hook"
 FROZEN_HOST = "frozen-host"
 
@@ -191,17 +299,27 @@ def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--release-root", type=Path, required=True)
     parser.add_argument("--out", type=Path, required=True)
+    parser.add_argument("--part", choices=("1", "2b"), default="1")
     args = parser.parse_args()
     release = args.release_root.resolve()
     out = args.out.resolve()
-    if out.exists():
-        sys.exit(f"{out} exists; frozen fixtures are never regenerated")
     python = release / ".venv" / "bin" / "python"
     syncmill = release / ".venv" / "bin" / "syncmill"
     version = subprocess.run(
         [str(python), "-c", "import importlib.metadata as m; print(m.version('syncmill'))"],
         check=True, capture_output=True, text=True,
     ).stdout.strip()
+    if args.part == "1":
+        if out.exists():
+            sys.exit(f"{out} exists; frozen fixtures are never regenerated")
+    else:
+        # Additive: the release directory exists, and nothing this part
+        # writes may be there already.
+        if json.loads((out / "bindings.json").read_text())["release"] != version:
+            sys.exit(f"{out} holds another release's fixtures")
+        taken = [name for name in (*PART_2B, "bindings-2b.json") if (out / name).exists()]
+        if taken:
+            sys.exit(f"{out} already has {taken}; frozen fixtures are never regenerated")
     site = _site_packages(python)
     hook_files = [site / f"{HOOK_MODULE}.py", site / f"zz{HOOK_MODULE}.pth"]
     if any(path.exists() for path in hook_files):
@@ -214,14 +332,16 @@ def main() -> None:
         hook_files[0].write_text(CRASH_HOOK, encoding="utf-8")
         hook_files[1].write_text(f"import {HOOK_MODULE}\n", encoding="utf-8")
         with tempfile.TemporaryDirectory() as tmp:
-            _generate(Path(tmp).resolve(), runtime, syncmill, version, out)
+            _generate(Path(tmp).resolve(), runtime, python, syncmill, version, out, args.part)
     finally:
         for path in hook_files:
             path.unlink(missing_ok=True)
         shutil.rmtree(runtime, ignore_errors=True)
 
 
-def _generate(box: Path, runtime: Path, syncmill: Path, version: str, out: Path) -> None:
+def _generate(
+    box: Path, runtime: Path, python: Path, syncmill: Path, version: str, out: Path, part: str
+) -> None:
     repo, home, state, fakebin = (box / n for n in ("repo", "home", "state", "fakebin"))
     for folder in (repo, home, state, fakebin):
         folder.mkdir()
@@ -230,6 +350,11 @@ def _generate(box: Path, runtime: Path, syncmill: Path, version: str, out: Path)
         (fakebin / binary).write_text(f"#!{sys.executable}\n" + FAKE_AGENT, encoding="utf-8")
         (fakebin / binary).chmod(0o755)
     (fakebin / "python3").symlink_to(sys.executable)
+    if part == "2b":
+        (fakebin / "toolgraph").write_text(
+            f"#!{sys.executable}\n" + FAKE_TOOLGRAPH, encoding="utf-8"
+        )
+        (fakebin / "toolgraph").chmod(0o755)
     base_env = {
         "HOME": str(home),
         "PATH": os.pathsep.join([str(fakebin), "/usr/bin", "/bin"]),
@@ -258,7 +383,7 @@ def git(*argv: str) -> None:
     git("add", "-A")
     git("commit", "-q", "-m", "frozen tree")
 
-    out.mkdir(parents=True)
+    out.mkdir(parents=True, exist_ok=part == "2b")
     bindings: dict = {
         "release": version,
         "box_root": str(box),
@@ -269,10 +394,11 @@ def git(*argv: str) -> None:
     }
 
     class Scenario:
-        def __init__(self, name: str, *, gate: bool = True, **control: str):
+        def __init__(self, name: str, *, gate: bool = True, env: dict | None = None,
+                     **control):
             self.name = name
             self.project = f"frozen-workflows-{name}"
-            self.env = {**base_env, "SYNCMILL_PROJECT__ID": self.project}
+            self.env = {**base_env, **(env or {}), "SYNCMILL_PROJECT__ID": self.project}
             if gate:
                 self.env["SYNCMILL_GATE__COMMANDS"] = json.dumps(GATE)
             self.control = {"spec": SPEC, **control}
@@ -392,6 +518,11 @@ def run_until(sc: Scenario, workflow_id: str, marker: str) -> None:
                 sys.exit(f"{sc.name}: the worker or its agent survived SIGKILL")
             time.sleep(0.1)
 
+    if part == "2b":
+        _part_2b(Scenario, freeze, git, home, state, repo, python)
+        (out / "bindings-2b.json").write_text(json.dumps(bindings, indent=2) + "\n")
+        return
+
     # ── the lifecycle ────────────────────────────────────────────────────
     sc = Scenario("planned")
     sc.plan()
@@ -469,5 +600,44 @@ def run_until(sc: Scenario, workflow_id: str, marker: str) -> None:
     (out / "bindings.json").write_text(json.dumps(bindings, indent=2) + "\n")
 
 
+def _part_2b(Scenario, freeze, git, home: Path, state: Path, repo: Path, python: Path) -> None:
+    def crashed_run(sc, workflow_id: str, point: str) -> None:
+        sc.prepare(crash_at=point)
+        sc.workflow("run", workflow_id, ok=False)
+        (home / "crash-at").unlink()
+
+    # The worker exits right after the final node's bundle is saved; then
+    # the release's own `status` recovers the dead lease.
+    sc = Scenario("bundle-saved", kimi_usage=True)
+    workflow_id = sc.plan()
+    crashed_run(sc, workflow_id, "bundle")
+    freeze(sc)
+    sc.name = "bundle-saved-recovered"
+    sc.workflow("status", workflow_id)
+    freeze(sc)
+
+    sc = Scenario("control-planned", env=TOOLGRAPH_ENV)
+    sc.plan()
+    freeze(sc)
+
+    sc = Scenario("control-completed", env=TOOLGRAPH_ENV)
+    workflow_id = sc.plan()
+    sc.workflow("run", workflow_id)
+    freeze(sc)
+
+    for name, die_at in (("apply-unreceipted", "after-applying"),
+                         ("apply-receipted", "before-applied")):
+        sc = Scenario(name)
+        workflow_id = sc.plan()
+        sc.workflow("run", workflow_id)
+        subprocess.run(
+            [str(python), "-c", CRASH_APPLY, str(state / "runs"), sc.project,
+             f"{workflow_id}-final", str(repo), die_at],
+            cwd=repo, env=sc.env, check=True,
+        )
+        freeze(sc)
+        git("checkout", "--", ".")
+
+
 if __name__ == "__main__":
     main()

tests/fixtures/persisted/workflows/v0.2.0/apply-receipted/runs/wf-15778afc04544c85-final/applications/20261109T213723644855Z.json

@@ -0,0 +1,7 @@
+{
+  "applied_at": "2026-11-09T21:37:23.644881+00:00",
+  "repo_root": "/private/var/folders/4n/mt3km5rs1bn_b0w_d6hvlmhc0000gn/T/tmpnhjq8lqv/repo",
+  "patch_sha256": "35dc711418e60edc7974186bd853c7ab08db843f16e43066417b07a53d8566ba",
+  "apply_id": "94ef2849-5ce2-40cd-b87c-284c34be0fc5",
+  "run_id": "wf-15778afc04544c85-final"
+}