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..385983c 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,645 @@ 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, 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
+
+    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
+    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, later):
+        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] == shown[2]
+    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 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)
+    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]
+    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"])
+    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)
+
+    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
+    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
+    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_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.
+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, capsys):
+    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, capsys, released) == ("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")
+    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, capsys
+):
+    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
+    error, lines = _render(done, frozen, capsys, released)
+    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
+):
+    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, capsys, released) == (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}
+    kept = _released_state(frozen)
+
+    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_released_kept(frozen, kept)
+    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(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([
+        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
+    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"}
+        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 env["AGENT_EXTRA"] == "agent"
+        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
+
+    toolgraph_calls = []
+
+    def no_toolgraph(*args, **kwargs):
+        # 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")
+    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 == []
+    assert toolgraph_calls == []