From a39ee114cef19eefbbdf6913e66b17b823a10c68 Mon Sep 17 00:00:00 2001 From: memtomem <118508877+memtomem@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:45:42 +0900 Subject: [PATCH 1/3] Run released workflows with the current run, resume, revise and verify (#271) The Workflows family of the persisted-contracts D rows, part 2a. No new fixtures and no src change: the v0.2.0 and v0.3.0 stores frozen in part 1 are now executed by HEAD. Execution goes through WorkflowCoordinator.run, the call the durable job's worker makes; the hermetic suite starts no job. Fake codex and kimi-code node agents record their prompts. - run of released planned, created and revised rows (a revision leaves its parent runnable): each node gets the stored goal and its own stored prompt, the final node its dependency's output; calls_used continues from the released count; a HEAD final artifact, approved. - resume against the stored absolute deadline, with the clock fixed from the row's own deadline_at (three of these rows were stored without a heartbeat). After it: every crash and failure ends "workflow deadline exceeded" before any dispatch, deadline unchanged. Before it: a dead owner's run and a recovered one reuse the released draft bundle and refund the killed node's attempt (3 -> 4 calls); a final artifact published before the crash is reused and approved, and v0.2.0's is not backfilled with final_agent; an exhausted node dispatches nothing. Refused from completed, cancelled and planned, and while the released owner's pid lives, where the refused attempt is appended after the released execution history. Events continue the released sequence. - workflow revise: a v0.2.0 parent gains a lineage (its only changed column), a v0.3.0 lineage continues past its last version. - workflow run/resume up to a stubbed launch: the team peek follows default_team for a v0.2.0 row and not for a v0.3.0 row that recorded none; a final state is refused before any launch. - apply --verify of the released final artifact: v0.2.0 gets the gate's own allowlist only, with the notice; v0.3.0 also the final agent's; both also after a prune. - control plans enabled refuse a released workflow planned without one, before any agent or Toolgraph call. The doc row names these; still not yet: release-written control-plan workflows, a bundle saved just before a crash, and an interrupted workflow-origin apply (part 2b, new fixtures). 13 injected regressions in workflow.py and cli.py each fail a test here. Co-Authored-By: Claude Opus 5.5 --- docs/architecture/persisted-contracts.md | 2 +- tests/test_frozen_workflows.py | 513 +++++++++++++++++++++++ 2 files changed, 514 insertions(+), 1 deletion(-) diff --git a/docs/architecture/persisted-contracts.md b/docs/architecture/persisted-contracts.md index 25b4415..594ea8d 100644 --- a/docs/architecture/persisted-contracts.md +++ b/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`. Not yet: a HEAD `run` or `resume` of a released workflow, `apply --verify` environment narrowing, control-plan (toolgraph) workflows, 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. 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 | | 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 | diff --git a/tests/test_frozen_workflows.py b/tests/test_frozen_workflows.py index 81ce351..10af295 100644 --- a/tests/test_frozen_workflows.py +++ b/tests/test_frozen_workflows.py @@ -13,6 +13,10 @@ alive on this host), and the root and fingerprint of a final run artifact. ``spec_json``, node bundles and ``commit.json`` are never touched: digests cover their bytes. ``_install`` asserts that every other cell is unchanged. + +The later sections run HEAD on the released stores: ``run``, ``resume`` and +``revise`` of released rows, the launch's reads before a job starts, and +``apply --verify`` of a released final artifact. """ from __future__ import annotations @@ -799,3 +803,512 @@ def test_recover_skips_a_released_final_artifact(release, tmp_path, monkeypatch) f"Run {run_id}: skipped \N{EM DASH} run is 'pending', not applying" ] assert (run_dir / "manifest.json").read_bytes() == before + + +# ── HEAD executes a released workflow ───────────────────────────────────── +# +# Through ``WorkflowCoordinator.run``, the call the detached job's worker makes: +# the hermetic suite starts no job. The CLI's reads before the launch +# (``_peek_workflow_team`` and ``_execute_workflow``'s checks) are tested with +# the launch stubbed. + +FIXED = "def add(a, b):\n return a + b\n" +#: The released fake codex's answer, stored in every released ``draft`` bundle. +RELEASED_DRAFT_OUTPUT = "codex (fake) fix." + + +class NodeAgent: + """A workflow node agent: records each prompt; ``fix`` repairs ``calc.py``.""" + + def __init__(self, agent_id: str, output: str, *, fix: bool = False): + self.agent_id = agent_id + self.output = output + self.fix = fix + self.calls: list[str] = [] + + def execute(self, prompt, workdir, timeout): + raise AssertionError("a released plan needs no planner call") + + async def aexecute(self, prompt, workdir, timeout): + from uuid import uuid4 + + from syncmill.models import TaskResult, TaskStatus + + self.calls.append(prompt) + if self.fix: + (Path(workdir) / "calc.py").write_text(FIXED, encoding="utf-8") + return TaskResult( + task_id=uuid4(), agent_id=self.agent_id, status=TaskStatus.COMPLETED, + output=self.output, exit_code=0, duration_seconds=0.01, + ) + + +def _workflow_env(frozen: Frozen, monkeypatch) -> None: + """The configuration every released scenario was written under.""" + monkeypatch.setenv("SYNCMILL_AGENT_PRIORITY", json.dumps(["codex", "kimi-code"])) + monkeypatch.setenv("SYNCMILL_CLAUDE__ENABLED", "false") + if frozen.scenario != "no-changes": + monkeypatch.setenv("SYNCMILL_GATE__COMMANDS", json.dumps([GATE_COMMAND])) + + +def _coordinator(frozen: Frozen, monkeypatch): + from conftest import FakeRegistry + from syncmill.project import load_project_config + from syncmill.workflow import WorkflowCoordinator + + _workflow_env(frozen, monkeypatch) + _, config = load_project_config(frozen.repo) + agents = { + "codex": NodeAgent("codex", "HEAD draft."), + "kimi-code": NodeAgent("kimi-code", "HEAD fix.", fix=True), + } + return WorkflowCoordinator(config, FakeRegistry(list(agents.values()))), agents + + +def _at(monkeypatch, moment) -> None: + """Fix both clocks the workflow module reads: ``datetime.now`` (lease and + deadline bookkeeping) and ``time()`` (the time left before the deadline). + ``monotonic`` stays real; it only measures this run's own waits.""" + _clock(monkeypatch, moment) + monkeypatch.setattr("syncmill.workflow.time", moment.timestamp) + + +def _deadline(manifest: dict): + from datetime import datetime + + return datetime.fromisoformat(manifest["deadline_at"]) + + +def _before_deadline(frozen: Frozen, monkeypatch) -> dict: + """The released manifest, with the clock a minute before its deadline: + resume keeps the stored absolute deadline, and three of these rows were + stored without a heartbeat, so the deadline is the one time they all carry.""" + from datetime import timedelta + + released = json.loads(frozen.row()["manifest_json"]) + _at(monkeypatch, _deadline(released) - timedelta(seconds=60)) + return released + + +def _render(record, frozen: Frozen) -> str: + """What ``workflow run``/``resume`` prints for this invocation.""" + import click + + from syncmill.cli import _render_workflow_completion + + try: + _render_workflow_completion(record, str(frozen.repo)) + except click.ClickException as exc: + return f"Error: {exc.message}" + return "" + + +def _approved_final(coordinator, record, final_agent) -> dict: + from syncmill.artifacts import workflow_approved + from syncmill.workflow import workflow_final_run_id + + assert record.final_run_id == workflow_final_run_id(record.workflow_id) + manifest = coordinator.artifacts.load(record.final_run_id).manifest + assert workflow_approved(manifest) + assert manifest["plan"].get("final_agent") == final_agent + return manifest + + +@pytest.mark.parametrize("target", ["planned", "created", "revised-parent", "revised-child"]) +def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, capsys): + scenario = target.split("-")[0] + if scenario not in scenarios(release): + pytest.skip(f"{release} has no {scenario}") + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + # A revision leaves its parent runnable: either row can be run. + workflow_id = frozen.workflow_ids[0 if target == "revised-parent" else -1] + coordinator, agents = _coordinator(frozen, monkeypatch) + before = {wid: frozen.row(wid) for wid in frozen.workflow_ids} + released = json.loads(before[workflow_id]["manifest_json"]) + + done = coordinator.run(workflow_id) + + assert done.state == "completed" + # Each node gets the goal and its own stored prompt; the final node also + # gets its dependency's output. + [draft_prompt] = agents["codex"].calls + [fix_prompt] = agents["kimi-code"].calls + spec = json.loads(before[workflow_id]["spec_json"]) + prompts = {node["id"]: node["prompt"] for node in spec["nodes"]} + goal = f"Original goal:\n{spec['goal']}\n" + assert goal in draft_prompt and goal in fix_prompt + assert f"task:\n{prompts['draft']}\n" in draft_prompt + assert f"task:\n{prompts['fix']}\n" in fix_prompt and "HEAD draft." in fix_prompt + assert prompts["fix"] not in draft_prompt + # The planner's call (a planner plan) stays counted. + assert released["calls_used"] == (1 if scenario == "planned" else 0) + assert done.manifest["calls_used"] == released["calls_used"] + 2 + assert frozen.row(workflow_id)["spec_json"] == before[workflow_id]["spec_json"] + for other in frozen.workflow_ids: + if other != workflow_id: + assert frozen.row(other) == before[other] + final = _approved_final(coordinator, done, "kimi-code") + assert "+ return a + b" in coordinator.artifacts.patch(final["run_id"]).decode() + assert _render(done, frozen) == "" + assert f"[syncmill] final run: {done.final_run_id}" in capsys.readouterr().out + + +#: Released crashes and failures that HEAD may resume. +RESUMABLE = ["running", "recovery-required", "unapproved-running", "failed", "gate-failed"] + + +@pytest.mark.parametrize("scenario", RESUMABLE) +def test_resume_honours_the_released_deadline(release, scenario, tmp_path, monkeypatch): + from datetime import timedelta + + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + row = frozen.row() + released = json.loads(row["manifest_json"]) + _at(monkeypatch, _deadline(released) + timedelta(seconds=1)) + + done = coordinator.run(frozen.workflow_id, resume=True) + + # The stored absolute deadline holds across the upgrade: it is not reset, + # and it comes before any dispatch or exhaustion decision. + assert (done.state, done.manifest["error"]) == ("failed", "workflow deadline exceeded") + assert done.manifest["deadline_at"] == released["deadline_at"] + assert done.manifest["calls_used"] == released["calls_used"] + assert done.final_run_id == row["final_run_id"] + assert agents["codex"].calls == agents["kimi-code"].calls == [] + assert _render(done, frozen) == "Error: workflow deadline exceeded" + + +@pytest.mark.parametrize("scenario", ["running", "recovery-required"]) +def test_resume_finishes_a_released_crash_on_its_completed_dependency( + release, scenario, tmp_path, monkeypatch, capsys +): + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + released = _before_deadline(frozen, monkeypatch) + draft = released["nodes"]["draft"] + stored = json.loads(frozen.frozen_bytes( + f"workflows/{frozen.workflow_id}/{draft['output_path']}" + )) + assert stored["output"] == RELEASED_DRAFT_OUTPUT + # The dead owner's final node left no bundle: nothing to reclaim. + assert not list((frozen.dir / "workflows" / frozen.workflow_id).rglob("nodes/fix/**/result.json")) + released_events = _store(frozen).events(frozen.workflow_id) + assert released_events + + done = coordinator.run(frozen.workflow_id, resume=True) + + # HEAD's events continue the released sequence; the released ones stay. + events = _store(frozen).events(frozen.workflow_id) + assert events[: len(released_events)] == released_events + assert len(events) > len(released_events) + assert min(e["seq"] for e in events[len(released_events):]) > released_events[-1]["seq"] + + assert done.state == "completed" + # The released draft is reused as written, read from its released bundle. + assert agents["codex"].calls == [] + assert done.manifest["nodes"]["draft"] == draft + [fix_prompt] = agents["kimi-code"].calls + assert RELEASED_DRAFT_OUTPUT in fix_prompt and "apply the fix" in fix_prompt + # The killed attempt had no outcome, so it was refunded before this one. + assert released["nodes"]["fix"]["attempts"] == 1 + assert done.manifest["nodes"]["fix"]["attempts"] == 1 + assert done.manifest["calls_used"] == released["calls_used"] + 1 == 4 + _approved_final(coordinator, done, "kimi-code") + assert _render(done, frozen) == "" + assert f"[syncmill] final run: {done.final_run_id}" in capsys.readouterr().out + + +def test_resume_completes_a_released_published_final_artifact(release, tmp_path, monkeypatch): + frozen = Frozen(release, "unapproved-running", tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + released = _before_deadline(frozen, monkeypatch) + run_id, run_dir = _final(frozen) + patch = (run_dir / "winner.patch").read_bytes() + published = json.loads((run_dir / "manifest.json").read_text()) + assert published["state"] == "pending" and "approved" not in published["plan"] + assert frozen.row()["final_run_id"] is None + + done = coordinator.run(frozen.workflow_id, resume=True) + + assert (done.state, done.final_run_id) == ("completed", run_id) + assert agents["codex"].calls == agents["kimi-code"].calls == [] + assert done.manifest["calls_used"] == released["calls_used"] + # Only the approval stamp is added: v0.2.0's artifact is not backfilled + # with the final agent it never recorded. + manifest = _approved_final( + coordinator, done, None if release == "v0.2.0" else "kimi-code" + ) + approval = manifest["plan"].pop("approved") + assert (approval["workflow_id"], approval["run_id"]) == (frozen.workflow_id, run_id) + assert manifest.pop("updated_at") != published.pop("updated_at") + assert manifest == published + assert (run_dir / "winner.patch").read_bytes() == patch + + +@pytest.mark.parametrize("scenario", ["failed", "gate-failed"]) +def test_resume_of_a_released_exhausted_node_dispatches_nothing( + release, scenario, tmp_path, monkeypatch +): + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + released = _before_deadline(frozen, monkeypatch) + error = "no runnable nodes remain before final_node completed" + + done = coordinator.run(frozen.workflow_id, resume=True) + + assert (done.state, done.manifest["error"]) == ("failed", error) + # A durable failure keeps its attempt: no refund, no new dispatch. + assert done.manifest["nodes"] == released["nodes"] + assert done.manifest["calls_used"] == released["calls_used"] + assert agents["codex"].calls == agents["kimi-code"].calls == [] + assert _render(done, frozen) == f"Error: {error}" + + +@pytest.mark.parametrize( + ("scenario", "state"), + [("completed", "completed"), ("unapproved-completed", "completed"), + ("no-changes", "completed"), ("cancelled", "cancelled"), ("planned", "planned")], +) +def test_resume_refuses_a_released_workflow_in_a_final_or_unstarted_state( + release, scenario, state, tmp_path, monkeypatch +): + from syncmill.workflow import WorkflowError + + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + before = frozen.row() + with pytest.raises(WorkflowError) as refused: + coordinator.run(frozen.workflow_id, resume=True) + assert str(refused.value) == ( + f"cannot resume workflow {frozen.workflow_id} from state {state!r}" + ) + assert frozen.row() == before + assert agents["codex"].calls == agents["kimi-code"].calls == [] + + +def test_resume_never_takes_a_released_lease_whose_owner_lives(release, tmp_path, monkeypatch): + import os + + from syncmill.workflow import WorkflowError + + frozen = Frozen(release, "running", tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + conn = sqlite3.connect(frozen.project_dir / "workflows" / "workflows.db") + try: + conn.execute("UPDATE workflows SET lease_owner = ?", + (f"{os.getpid()}@{socket.gethostname()}",)) + conn.commit() + finally: + conn.close() + # Before the deadline, and long past the lease timeout: only the owner's + # pid keeps the lease. + _before_deadline(frozen, monkeypatch) + before = frozen.row() + with pytest.raises(WorkflowError) as refused: + coordinator.run(frozen.workflow_id, resume=True) + assert str(refused.value) == ( + f"workflow {frozen.workflow_id} is still owned by a live run " + f"({before['lease_owner']}); cancel it first or wait for it to finish" + ) + after = frozen.row() + lease = ("state", "lease_owner", "lease_id", "lease_heartbeat") + assert {k: after[k] for k in lease} == {k: before[k] for k in lease} + assert agents["codex"].calls == agents["kimi-code"].calls == [] + # The refused attempt is recorded after the released execution history, + # which is kept as written; nothing else in the manifest moves. + released, manifest = json.loads(before["manifest_json"]), json.loads(after["manifest_json"]) + *kept, attempt = manifest.pop("executions") + assert kept == released.pop("executions") + assert (attempt["action"], attempt["status"]) == ("resume", "failed") + assert manifest == released + + +# ── the CLI around execution ────────────────────────────────────────────── + + +def _cli_supervisor(frozen: Frozen, monkeypatch) -> None: + """A real Supervisor for workflow commands, without real memtomem.""" + from conftest import FakeMemory + + _workflow_env(frozen, monkeypatch) + monkeypatch.setattr("syncmill.supervisor.SyncmillMemory", lambda _cfg: FakeMemory()) + + +@pytest.mark.parametrize("scenario", ["planned", "revised"]) +def test_revise_a_released_workflow(release, scenario, tmp_path, monkeypatch): + if scenario not in scenarios(release): + pytest.skip(f"{release} has no {scenario}") + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + _cli_supervisor(frozen, monkeypatch) + parent = frozen.workflow_id + document = tmp_path / "revision.json" + spec = json.loads(frozen.row(parent)["spec_json"]) + spec["goal"] = "fix add, revised at HEAD" + document.write_text(json.dumps(spec), encoding="utf-8") + _store(frozen) # a v0.2.0 store gains v0.3.0's columns on open + before = {wid: frozen.row(wid) for wid in frozen.workflow_ids} + + result = frozen.cli("workflow", "revise", parent, "--file", str(document)) + + assert result.exit_code == 0, result.output + [child] = frozen.rows( + "SELECT * FROM workflows WHERE workflow_id NOT IN ({})".format( + ",".join("?" * len(before)) + ), + *before, + ) + after = {wid: frozen.row(wid) for wid in frozen.workflow_ids} + assert (child["parent_workflow_id"], child["plan_origin"], child["state"]) == ( + parent, "revision", "planned" + ) + if release == "v0.2.0": + # A released row has no lineage: the revision gives it one, and that + # one column is all that changes on it. + assert before[parent]["lineage_id"] is None + assert re.fullmatch(r"wl-[0-9a-f]{16}", after[parent]["lineage_id"]) + assert child["lineage_id"] == after[parent]["lineage_id"] + assert child["plan_version"] == 2 + changed = {k for k in before[parent] if before[parent][k] != after[parent][k]} + assert changed == {"lineage_id"} + else: + # The released lineage continues past its last version. + released_versions = {"planned": [1], "revised": [1, 2]}[scenario] + assert child["lineage_id"] == before[parent]["lineage_id"] + assert [before[w]["plan_version"] for w in frozen.workflow_ids] == released_versions + assert child["plan_version"] == released_versions[-1] + 1 + assert after == before + + +class _Launches: + """``workflow run``/``resume`` up to the job launch, which is recorded.""" + + def __init__(self, monkeypatch): + self.launched: list = [] + monkeypatch.setattr("syncmill.cli._workflow_job_store", lambda config, root: self) + + def launch(self, spec, *, resolved_config): + self.launched.append((spec, resolved_config)) + return {"job_id": "0" * 32} + + +@pytest.mark.parametrize(("command", "scenario"), [("run", "planned"), ("resume", "recovery-required")]) +def test_the_launch_follows_the_team_a_released_row_recorded( + release, command, scenario, tmp_path, monkeypatch +): + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + _cli_supervisor(frozen, monkeypatch) + monkeypatch.setenv("SYNCMILL_TEAMS", json.dumps({"pair": {"agents": ["codex", "kimi-code"]}})) + monkeypatch.setenv("SYNCMILL_DEFAULT_TEAM", "pair") + launches = _Launches(monkeypatch) + before = frozen.row() + + result = frozen.cli("workflow", command, frozen.workflow_id, "--detach") + + assert result.exit_code == 0, result.output + [(spec, config)] = launches.launched + assert (spec.workflow_id, spec.resume) == (frozen.workflow_id, command == "resume") + # v0.2.0 never recorded a team, so the default applies; v0.3.0 recorded + # that none applied, and the default must not override that. + assert config.applied_team == ("pair" if release == "v0.2.0" else None) + assert frozen.row()["state"] == before["state"] + + +@pytest.mark.parametrize( + ("command", "scenario", "state"), + [("run", "completed", "completed"), ("resume", "cancelled", "cancelled")], +) +def test_the_cli_refuses_a_released_state_before_any_launch( + release, command, scenario, state, tmp_path, monkeypatch +): + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + _cli_supervisor(frozen, monkeypatch) + launches = _Launches(monkeypatch) + result = frozen.cli("workflow", command, frozen.workflow_id, "--detach") + assert (result.exit_code, result.output.splitlines()[-1]) == ( + 1, f"Error: cannot {command} workflow {frozen.workflow_id} from state {state!r}" + ) + assert launches.launched == [] + + +@pytest.mark.parametrize("prune_first", [False, True]) +def test_apply_verify_narrows_a_released_final_artifact( + release, prune_first, tmp_path, monkeypatch +): + import shlex + import sys + + frozen = Frozen(release, "completed", tmp_path, monkeypatch) + run_id, _ = _final(frozen) + dump = tmp_path / "gate-env.json" + script = tmp_path / "dump_env.py" + script.write_text( + "import json, os, sys\njson.dump(sorted(os.environ), open(sys.argv[1], 'w'))\n", + encoding="utf-8", + ) + monkeypatch.setenv("GATE_EXTRA", "gate") + monkeypatch.setenv("AGENT_EXTRA", "agent") + monkeypatch.setenv("SYNCMILL_GATE__ENV_ALLOWLIST", json.dumps(["GATE_EXTRA"])) + monkeypatch.setenv("SYNCMILL_KIMI_CODE__WORKFLOW_ENV_ALLOWLIST", json.dumps(["AGENT_EXTRA"])) + monkeypatch.setenv("SYNCMILL_GATE__COMMANDS", json.dumps([ + GATE_COMMAND, shlex.join([sys.executable, str(script), str(dump)]), + ])) + if prune_first: + # The narrowing is read from the artifact alone. + _Later.install(monkeypatch) + assert frozen.cli("workflow", "prune", "--keep", "0", "--retention-days", "1").exit_code == 0 + assert frozen.rows("SELECT * FROM workflows") == [] + + result = frozen.cli("apply", run_id, "--verify") + + assert result.exit_code == 0, result.output + assert (frozen.repo / "calc.py").read_text() == FIXED + names = set(json.loads(dump.read_text())) + notices = _lines(result.stderr, "[syncmill] verify: ") + if release == "v0.2.0": + # No final agent recorded: only the gate's own list, and it says so. + assert {"GATE_EXTRA", "AGENT_EXTRA"} & names == {"GATE_EXTRA"} + assert notices == [ + "[syncmill] verify: the workflow's final agent is not recorded on this " + "artifact, so verify runs without that agent's workflow_env_allowlist" + ] + else: + assert {"GATE_EXTRA", "AGENT_EXTRA"} & names == {"GATE_EXTRA", "AGENT_EXTRA"} + assert notices == [] + + +@pytest.mark.parametrize(("scenario", "resume"), [("planned", False), ("recovery-required", True)]) +def test_control_plans_refuse_a_released_workflow_planned_without_one( + release, scenario, resume, tmp_path, monkeypatch +): + from syncmill.workflow import WorkflowError + + def no_toolgraph(*args, **kwargs): + raise AssertionError("refused before any Toolgraph call") + + monkeypatch.setenv("SYNCMILL_INTEGRATION__TOOLGRAPH__ENABLED", "true") + monkeypatch.setenv("SYNCMILL_INTEGRATION__TOOLGRAPH__CONTROL_PLAN_ENABLED", "true") + monkeypatch.setenv("SYNCMILL_INTEGRATION__TOOLGRAPH__CANDIDATE_TOOLS", '["workspace::read"]') + monkeypatch.setenv( + "SYNCMILL_INTEGRATION__TOOLGRAPH__AGENT_MAP", '{"codex": "codex", "kimi-code": "kimi"}' + ) + monkeypatch.setattr("syncmill.preflight._sync_run", no_toolgraph) + frozen = Frozen(release, scenario, tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + assert coordinator.config.integration.toolgraph.control_plan_enabled + # A planned row has no deadline yet; a crashed one is resumed before it. + released = ( + _before_deadline(frozen, monkeypatch) if resume + else json.loads(frozen.row()["manifest_json"]) + ) + assert released["control_plan"] is None + before = frozen.row() + + with pytest.raises(WorkflowError) as refused: + coordinator.run(frozen.workflow_id, resume=resume) + + assert str(refused.value) == ( + "workflow predates control-plan evidence; create a new workflow plan" + ) + assert frozen.row() == before + assert agents["codex"].calls == agents["kimi-code"].calls == [] From 3e40cdac571730c60abc6aeb6e510f7f2815468b Mon Sep 17 00:00:00 2001 From: memtomem <118508877+memtomem@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:08:07 +0900 Subject: [PATCH 2/3] Bind the released-workflow execution tests to what they claim (#271) Review round 1 found assertions that survived the code they protect: - rendering is now judged by the invocation's own execution: it must be the one appended after the released history, and the same record with its state flipped must print the same outcome (invocation() -> None passed before); - the final artifact's patch and bundle digests are recomputed from the final node's stored bytes (dropping both from the plan passed before); - the control-plan refusal records Toolgraph calls instead of raising from the seam, which run_preflight swallows (an advisory call before the refusal passed before); - run and revise keep every released workflow file and event as written (a truncated planner output passed before); - apply --verify: an unlisted variable never reaches the gate, the common ones and the listed values do. 17 injected regressions now each fail a test here. Co-Authored-By: Claude Opus 5.5 --- tests/test_frozen_workflows.py | 120 +++++++++++++++++++++++++++------ 1 file changed, 99 insertions(+), 21 deletions(-) diff --git a/tests/test_frozen_workflows.py b/tests/test_frozen_workflows.py index 10af295..5df45ae 100644 --- a/tests/test_frozen_workflows.py +++ b/tests/test_frozen_workflows.py @@ -890,30 +890,86 @@ def _before_deadline(frozen: Frozen, monkeypatch) -> dict: return released -def _render(record, frozen: Frozen) -> str: - """What ``workflow run``/``resume`` prints for this invocation.""" +def _render(record, frozen: Frozen, capsys, released: dict) -> tuple[str, list[str]]: + """What ``workflow run``/``resume`` prints for this invocation: its error, + or its ``[syncmill]`` lines. + + Judged by the invocation's own execution, appended after the released + history, never by the row's state: rendering the same record with the + state flipped must print the same outcome.""" import click from syncmill.cli import _render_workflow_completion - try: - _render_workflow_completion(record, str(frozen.repo)) - except click.ClickException as exc: - return f"Error: {exc.message}" - return "" + execution = record.invocation() + assert execution is not None and execution == record.manifest["executions"][-1] + assert execution["execution_id"] not in { + e["execution_id"] for e in released.get("executions") or [] + } + assert execution["status"] == record.state + flipped = record.model_copy( + update={"state": "failed" if record.state == "completed" else "completed"} + ) + shown = [] + for candidate in (record, flipped): + try: + _render_workflow_completion(candidate, str(frozen.repo)) + error = "" + except click.ClickException as exc: + error = f"Error: {exc.message}" + lines = _lines(capsys.readouterr().out, "[syncmill] ") + shown.append((error, lines)) + assert shown[0] == shown[1] + return shown[0] def _approved_final(coordinator, record, final_agent) -> dict: + """The approved final artifact, bound to the final node's bundle by + digests recomputed here from the stored bytes.""" from syncmill.artifacts import workflow_approved - from syncmill.workflow import workflow_final_run_id + from syncmill.workflow import bundle_digest, patch_digest, workflow_final_run_id assert record.final_run_id == workflow_final_run_id(record.workflow_id) manifest = coordinator.artifacts.load(record.final_run_id).manifest assert workflow_approved(manifest) - assert manifest["plan"].get("final_agent") == final_agent + plan = manifest["plan"] + assert plan.get("final_agent") == final_agent + node = record.manifest["nodes"][record.spec.final_node] + base = coordinator.store.workflow_dir(record.workflow_id) + result_bytes = (base / node["output_path"]).read_bytes() + patch_bytes = (base / node["patch_path"]).read_bytes() + assert coordinator.artifacts.patch(record.final_run_id) == patch_bytes + assert plan["patch_sha256"] == patch_digest(patch_bytes) + assert plan["bundle_sha256"] == bundle_digest(result_bytes, patch_bytes) return manifest +def _released_state(frozen: Frozen) -> dict: + """Every released workflow file and event, per workflow.""" + state = {} + for workflow_id in frozen.workflow_ids: + root = frozen.dir / "workflows" / workflow_id + files = { + path.relative_to(root).as_posix(): path.read_bytes() + for path in sorted(root.rglob("*")) if path.is_file() + } + assert files + state[workflow_id] = (files, _store(frozen).events(workflow_id)) + return state + + +def _assert_released_kept(frozen: Frozen, state: dict) -> None: + """New files and events may follow; none released is changed or lost.""" + for workflow_id, (files, events) in state.items(): + root = frozen.project_dir / "workflows" / workflow_id + for relative, data in files.items(): + assert (root / relative).read_bytes() == data, (workflow_id, relative) + now = _store(frozen).events(workflow_id) + assert now[: len(events)] == events + if events and len(now) > len(events): + assert now[len(events)]["seq"] > events[-1]["seq"] + + @pytest.mark.parametrize("target", ["planned", "created", "revised-parent", "revised-child"]) def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, capsys): scenario = target.split("-")[0] @@ -925,6 +981,9 @@ def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, coordinator, agents = _coordinator(frozen, monkeypatch) before = {wid: frozen.row(wid) for wid in frozen.workflow_ids} released = json.loads(before[workflow_id]["manifest_json"]) + kept = _released_state(frozen) + # The planner's output is among the files kept as written. + assert any("planner" in name for name in kept[workflow_id][0]) == (scenario == "planned") done = coordinator.run(workflow_id) @@ -949,8 +1008,9 @@ def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, assert frozen.row(other) == before[other] final = _approved_final(coordinator, done, "kimi-code") assert "+ return a + b" in coordinator.artifacts.patch(final["run_id"]).decode() - assert _render(done, frozen) == "" - assert f"[syncmill] final run: {done.final_run_id}" in capsys.readouterr().out + _assert_released_kept(frozen, kept) + error, lines = _render(done, frozen, capsys, released) + assert error == "" and f"[syncmill] final run: {done.final_run_id}" in lines #: Released crashes and failures that HEAD may resume. @@ -958,7 +1018,7 @@ def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, @pytest.mark.parametrize("scenario", RESUMABLE) -def test_resume_honours_the_released_deadline(release, scenario, tmp_path, monkeypatch): +def test_resume_honours_the_released_deadline(release, scenario, tmp_path, monkeypatch, capsys): from datetime import timedelta frozen = Frozen(release, scenario, tmp_path, monkeypatch) @@ -976,7 +1036,7 @@ def test_resume_honours_the_released_deadline(release, scenario, tmp_path, monke assert done.manifest["calls_used"] == released["calls_used"] assert done.final_run_id == row["final_run_id"] assert agents["codex"].calls == agents["kimi-code"].calls == [] - assert _render(done, frozen) == "Error: workflow deadline exceeded" + assert _render(done, frozen, capsys, released) == ("Error: workflow deadline exceeded", []) @pytest.mark.parametrize("scenario", ["running", "recovery-required"]) @@ -1015,11 +1075,13 @@ def test_resume_finishes_a_released_crash_on_its_completed_dependency( assert done.manifest["nodes"]["fix"]["attempts"] == 1 assert done.manifest["calls_used"] == released["calls_used"] + 1 == 4 _approved_final(coordinator, done, "kimi-code") - assert _render(done, frozen) == "" - assert f"[syncmill] final run: {done.final_run_id}" in capsys.readouterr().out + error, lines = _render(done, frozen, capsys, released) + assert error == "" and f"[syncmill] final run: {done.final_run_id}" in lines -def test_resume_completes_a_released_published_final_artifact(release, tmp_path, monkeypatch): +def test_resume_completes_a_released_published_final_artifact( + release, tmp_path, monkeypatch, capsys +): frozen = Frozen(release, "unapproved-running", tmp_path, monkeypatch) coordinator, agents = _coordinator(frozen, monkeypatch) released = _before_deadline(frozen, monkeypatch) @@ -1044,11 +1106,13 @@ def test_resume_completes_a_released_published_final_artifact(release, tmp_path, assert manifest.pop("updated_at") != published.pop("updated_at") assert manifest == published assert (run_dir / "winner.patch").read_bytes() == patch + error, lines = _render(done, frozen, capsys, released) + assert error == "" and f"[syncmill] final run: {run_id}" in lines @pytest.mark.parametrize("scenario", ["failed", "gate-failed"]) def test_resume_of_a_released_exhausted_node_dispatches_nothing( - release, scenario, tmp_path, monkeypatch + release, scenario, tmp_path, monkeypatch, capsys ): frozen = Frozen(release, scenario, tmp_path, monkeypatch) coordinator, agents = _coordinator(frozen, monkeypatch) @@ -1062,7 +1126,7 @@ def test_resume_of_a_released_exhausted_node_dispatches_nothing( assert done.manifest["nodes"] == released["nodes"] assert done.manifest["calls_used"] == released["calls_used"] assert agents["codex"].calls == agents["kimi-code"].calls == [] - assert _render(done, frozen) == f"Error: {error}" + assert _render(done, frozen, capsys, released) == (f"Error: {error}", []) @pytest.mark.parametrize( @@ -1148,6 +1212,7 @@ def test_revise_a_released_workflow(release, scenario, tmp_path, monkeypatch): document.write_text(json.dumps(spec), encoding="utf-8") _store(frozen) # a v0.2.0 store gains v0.3.0's columns on open before = {wid: frozen.row(wid) for wid in frozen.workflow_ids} + kept = _released_state(frozen) result = frozen.cli("workflow", "revise", parent, "--file", str(document)) @@ -1159,6 +1224,7 @@ def test_revise_a_released_workflow(release, scenario, tmp_path, monkeypatch): *before, ) after = {wid: frozen.row(wid) for wid in frozen.workflow_ids} + _assert_released_kept(frozen, kept) assert (child["parent_workflow_id"], child["plan_origin"], child["state"]) == ( parent, "revision", "planned" ) @@ -1243,11 +1309,12 @@ def test_apply_verify_narrows_a_released_final_artifact( dump = tmp_path / "gate-env.json" script = tmp_path / "dump_env.py" script.write_text( - "import json, os, sys\njson.dump(sorted(os.environ), open(sys.argv[1], 'w'))\n", + "import json, os, sys\njson.dump(dict(os.environ), open(sys.argv[1], 'w'))\n", encoding="utf-8", ) monkeypatch.setenv("GATE_EXTRA", "gate") monkeypatch.setenv("AGENT_EXTRA", "agent") + monkeypatch.setenv("UNLISTED_EXTRA", "never") monkeypatch.setenv("SYNCMILL_GATE__ENV_ALLOWLIST", json.dumps(["GATE_EXTRA"])) monkeypatch.setenv("SYNCMILL_KIMI_CODE__WORKFLOW_ENV_ALLOWLIST", json.dumps(["AGENT_EXTRA"])) monkeypatch.setenv("SYNCMILL_GATE__COMMANDS", json.dumps([ @@ -1263,8 +1330,13 @@ def test_apply_verify_narrows_a_released_final_artifact( assert result.exit_code == 0, result.output assert (frozen.repo / "calc.py").read_text() == FIXED - names = set(json.loads(dump.read_text())) + env = json.loads(dump.read_text()) + names = set(env) notices = _lines(result.stderr, "[syncmill] verify: ") + # Closed: a name nothing lists never arrives; the common variables do. + assert "UNLISTED_EXTRA" not in names + assert {"PATH", "HOME"} <= names + assert env["GATE_EXTRA"] == "gate" if release == "v0.2.0": # No final agent recorded: only the gate's own list, and it says so. assert {"GATE_EXTRA", "AGENT_EXTRA"} & names == {"GATE_EXTRA"} @@ -1274,6 +1346,7 @@ def test_apply_verify_narrows_a_released_final_artifact( ] else: assert {"GATE_EXTRA", "AGENT_EXTRA"} & names == {"GATE_EXTRA", "AGENT_EXTRA"} + assert env["AGENT_EXTRA"] == "agent" assert notices == [] @@ -1283,8 +1356,12 @@ def test_control_plans_refuse_a_released_workflow_planned_without_one( ): from syncmill.workflow import WorkflowError + toolgraph_calls = [] + def no_toolgraph(*args, **kwargs): - raise AssertionError("refused before any Toolgraph call") + # Recorded, not raised: run_preflight swallows any exception. + toolgraph_calls.append(args) + raise OSError("no toolgraph here") monkeypatch.setenv("SYNCMILL_INTEGRATION__TOOLGRAPH__ENABLED", "true") monkeypatch.setenv("SYNCMILL_INTEGRATION__TOOLGRAPH__CONTROL_PLAN_ENABLED", "true") @@ -1312,3 +1389,4 @@ def no_toolgraph(*args, **kwargs): ) assert frozen.row() == before assert agents["codex"].calls == agents["kimi-code"].calls == [] + assert toolgraph_calls == [] From 0cf71ad66cde6209f438c00cd0f80b9ff6890819 Mon Sep 17 00:00:00 2001 From: memtomem <118508877+memtomem@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:31:13 +0900 Subject: [PATCH 3/3] Pin which execution and which provenance the released resumes use (#271) Review round 2: - rendering is checked against a later, conflicting execution appended to the same record: the outcome stays this invocation's (returning the latest execution passed before, because the round-1 check equated this invocation with the last entry); - resume of the released published final artifact is refused, without approving, replacing or re-running anything, when its patch bytes, plan digests, source fingerprint, node, workflow or repository disagree (the reuse checks could be removed before); - the final node's prompt never carries the draft node's task. 24 injected regressions now each fail a test here. Co-Authored-By: Claude Opus 5.5 --- tests/test_frozen_workflows.py | 63 +++++++++++++++++++++++++++++++--- 1 file changed, 59 insertions(+), 4 deletions(-) diff --git a/tests/test_frozen_workflows.py b/tests/test_frozen_workflows.py index 5df45ae..385983c 100644 --- a/tests/test_frozen_workflows.py +++ b/tests/test_frozen_workflows.py @@ -907,11 +907,16 @@ def _render(record, frozen: Frozen, capsys, released: dict) -> tuple[str, list[s e["execution_id"] for e in released.get("executions") or [] } assert execution["status"] == record.state - flipped = record.model_copy( - update={"state": "failed" if record.state == "completed" else "completed"} + other = "failed" if record.state == "completed" else "completed" + flipped = record.model_copy(update={"state": other}) + # Another invocation that ended differently and recorded itself later. + later = record.model_copy(deep=True) + later.manifest["executions"].append( + {**execution, "execution_id": "wfx-" + "f" * 24, "status": other, + "error": None if other == "completed" else "a later invocation failed"} ) shown = [] - for candidate in (record, flipped): + for candidate in (record, flipped, later): try: _render_workflow_completion(candidate, str(frozen.repo)) error = "" @@ -919,7 +924,7 @@ def _render(record, frozen: Frozen, capsys, released: dict) -> tuple[str, list[s error = f"Error: {exc.message}" lines = _lines(capsys.readouterr().out, "[syncmill] ") shown.append((error, lines)) - assert shown[0] == shown[1] + assert shown[0] == shown[1] == shown[2] return shown[0] @@ -999,6 +1004,7 @@ def test_run_a_released_planned_workflow(release, target, tmp_path, monkeypatch, assert f"task:\n{prompts['draft']}\n" in draft_prompt assert f"task:\n{prompts['fix']}\n" in fix_prompt and "HEAD draft." in fix_prompt assert prompts["fix"] not in draft_prompt + assert prompts["draft"] not in fix_prompt # The planner's call (a planner plan) stays counted. assert released["calls_used"] == (1 if scenario == "planned" else 0) assert done.manifest["calls_used"] == released["calls_used"] + 2 @@ -1110,6 +1116,55 @@ def test_resume_completes_a_released_published_final_artifact( assert error == "" and f"[syncmill] final run: {run_id}" in lines +def _damage_patch(run_dir: Path, manifest: dict) -> None: + (run_dir / "winner.patch").write_bytes(b"not the published patch\n") + + +def _damage(field: str, value: str): + def damage(run_dir: Path, manifest: dict) -> None: + holder = manifest if field == "repo_root" else manifest["plan"] + assert field in holder + holder[field] = value + (run_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + return damage + + +@pytest.mark.parametrize( + ("damage", "error"), + [ + (_damage_patch, "final artifact patch is missing or corrupt"), + (_damage("patch_sha256", "0" * 64), "final run id collision for workflow {id}"), + (_damage("bundle_sha256", "0" * 64), "final run id collision for workflow {id}"), + (_damage("source_fingerprint", "0" * 64), "final run id collision for workflow {id}"), + (_damage("node_id", "draft"), "final run id collision for workflow {id}"), + (_damage("workflow_id", "wf-" + "0" * 16), "final run id collision for workflow {id}"), + (_damage("repo_root", "/elsewhere"), "final artifact repository provenance mismatch"), + ], + ids=["patch", "patch-digest", "bundle-digest", "source", "node", "workflow", "repo"], +) +def test_resume_refuses_a_released_final_artifact_whose_provenance_disagrees( + release, damage, error, tmp_path, monkeypatch +): + from syncmill.workflow import WorkflowError + + frozen = Frozen(release, "unapproved-running", tmp_path, monkeypatch) + coordinator, agents = _coordinator(frozen, monkeypatch) + _before_deadline(frozen, monkeypatch) + _, run_dir = _final(frozen) + damage(run_dir, json.loads((run_dir / "manifest.json").read_text())) + damaged = {name: (run_dir / name).read_bytes() for name in ("manifest.json", "winner.patch")} + + with pytest.raises(WorkflowError) as refused: + coordinator.run(frozen.workflow_id, resume=True) + + assert str(refused.value) == error.format(id=frozen.workflow_id) + # Never approved, never replaced, nothing re-run: the workflow waits for + # an explicit decision. + assert {name: (run_dir / name).read_bytes() for name in damaged} == damaged + assert (frozen.row()["state"], frozen.row()["final_run_id"]) == ("recovery_required", None) + assert agents["codex"].calls == agents["kimi-code"].calls == [] + + @pytest.mark.parametrize("scenario", ["failed", "gate-failed"]) def test_resume_of_a_released_exhausted_node_dispatches_nothing( release, scenario, tmp_path, monkeypatch, capsys