feat(workflows): add durable restart recovery (Phase 3) (#7075)

* feat(workflows): recover durable workflows after Gateway restart

* docs(workflows): keep recovery guidance within inherited budget

* fix(workflows): retain cancellation ownership and stabilize turn-limit test

* fix(workflows): log service-owned action failures

AGENTS.md

@@ -83,7 +83,8 @@ generic hook to the extension contract plus an extension that uses it — do not
 hard-code business logic into core. Working examples:
 `examples/deerflow-extension-{example,bookmarks,agent-teams}`.
 [Workflow guide](examples/deerflow-extension-workflows/AGENTS.md): public
-AgentRuns; POSIX, one process/store; live resume/approval; no restart replay or hard currency budget.
+AgentRuns; POSIX, one process/store; approval and session-authorized recovery; uncertain admissions need reconciliation;
+no multi-Gateway failover or hard currency budget.
 
 Root runtime configs: copy `config.example.yaml` → `config.yaml` and
 `extensions_config.example.json` → `extensions_config.json` (MCP/skills).

README.md

@@ -3559,7 +3559,8 @@ and authenticated human checkpoints support staged inspection reports. Confirmat
 binds the exact retained output or captured file version (SHA-256); preview and
 downstream reads use the same immutable bytes. Each child retains host-generated
 assembly evidence for its resolved instructions, model, tools and Skills.
-Gateway restart ends resumability. There is no restart replay or hard monetary budget. Workflow deadlines and child-observation timeouts are
+Cancelling a recoverable workflow continues child cleanup even if the requesting browser disconnects.
+After a Gateway restart, sign in and select **Recover workflow** to recover a compatible run using retained results and exact-version approvals. Uncertain child admissions require reconciliation; they are never automatically retried. Recovery requires unchanged runtime/configuration and durable Gateway storage. Multi-Gateway failover remains Phase 4; there is no hard monetary budget. Workflow deadlines and child-observation timeouts are
 reported separately from uncertain admission. Accepted launches retain their request
 identity across inspection refresh failures, with a separate inspection retry action. A runnable example, installation steps
 and QuickJS source-build prerequisites for other Python/CPU combinations are included.

backend/app/gateway/extension_agent_runs.py

@@ -148,7 +148,7 @@ class _BoundAgentRuns:
             async with reserve_artifact_write(request, thread_id, user_id=self._user_id):
                 return await asyncio.to_thread(read_output_snapshot, self._user_id, thread_id, path)
 
-    async def configuration_revision(self, *, assistant_ids: list[str]) -> str:
+    async def configuration_revision(self, *, assistant_ids: list[str], durable: bool = False) -> str:
         from deerflow.config.agents_config import load_agent_config, load_agent_soul
         from deerflow.config.app_config import get_app_config
         from deerflow.config.extensions_config import get_extensions_config
@@ -173,7 +173,12 @@ class _BoundAgentRuns:
                     agents[name] = [config.model_dump(mode="json"), load_agent_soul(name, user_id=self._user_id)]
                 value = [get_app_config().model_dump(mode="json"), get_extensions_config().model_dump(mode="json"), names, agents]
                 raw = json.dumps(value, sort_keys=True, ensure_ascii=True, default=str).encode()
-                return hmac.new(self._host.revision_key, raw, hashlib.sha256).hexdigest()
+                key = self._host.revision_key
+                if durable:
+                    from app.gateway.auth.config import get_auth_config
+
+                    key = hmac.new(get_auth_config().jwt_secret.encode(), b"deerflow:configuration-revision:v1", hashlib.sha256).digest()
+                return hmac.new(key, raw, hashlib.sha256).hexdigest()
 
             return await asyncio.to_thread(snapshot)
 

backend/packages/extension-api/deerflow_extension_api/__init__.py

@@ -102,7 +102,7 @@ from deerflow_extension_api.state import ExtensionData
 
 #: Contract version. Before 1.0, minors may break and patches are additive.
 #: From 1.0 on, bump the major for breaking changes.
-API_VERSION = "0.2.7"
+API_VERSION = "0.2.8"
 
 __all__ = [
     "AGENT_RUNS_CONTEXT_KEY",

backend/packages/extension-api/deerflow_extension_api/agent_runs.py

@@ -60,13 +60,15 @@ class AgentRuns(Protocol):
         """
         ...
 
-    async def configuration_revision(self, *, assistant_ids: list[str]) -> str:
-        """Opaque process-local revision of declarative Agent/host configuration.
+    async def configuration_revision(self, *, assistant_ids: list[str], durable: bool = False) -> str:
+        """Opaque revision of declarative Agent/host configuration.
 
         Revalidates the caller. Includes app/extensions config and Custom Agent
         config/SOUL; excludes mutable memory, file contents referenced by tools,
-        provider state and effective per-run assembly. Compare within one host
-        lifetime only; this is an invalidation guard, not an assembly attestation.
+        provider state and effective per-run assembly. By default compare within
+        one host lifetime only. With durable=True, compare across restarts sharing the host
+        signing secret; secret rotation invalidates revisions. This retains no
+        grant and is an invalidation guard, not an assembly attestation.
         """
         ...
 

backend/packages/extension-api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "deerflow-extension-api"
-version = "0.2.7"
+version = "0.2.8"
 description = "Public contracts for DeerFlow extensions"
 requires-python = ">=3.12"
 # Keep the contract package import-light and independent from the host. Public

backend/packages/harness/pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     # the contract version it implements, extensions declare ranges. A range
     # here would let pip resolve a newer contract package than this harness
     # implements, making newer extensions look supported at runtime.
-    "deerflow-extension-api==0.2.7",
+    "deerflow-extension-api==0.2.8",
     "dotenv>=0.9.9",
     "exa-py>=1.0.0",
     "httpx>=0.28.0",

backend/tests/test_extension_agent_runs.py

@@ -445,3 +445,24 @@ async def test_evidence_persistence_failure_prevents_execution(runtime, monkeypa
     assert not handles
     with pytest.raises(AgentRunError, match="unavailable"):
         await runs.configuration_evidence(thread_id=thread, run_id=run.run_id)
+
+
+@pytest.mark.asyncio
+async def test_durable_configuration_revision_survives_host_replacement_and_rotation(host, monkeypatch):
+    control, request, load_user = host
+    monkeypatch.setattr("deerflow.config.app_config.get_app_config", lambda: SimpleNamespace(model_dump=lambda **kw: {"model": "test"}))
+    monkeypatch.setattr("deerflow.config.extensions_config.get_extensions_config", lambda: SimpleNamespace(model_dump=lambda **kw: {}))
+    secret = SimpleNamespace(jwt_secret="stable-test-signing-key")
+    monkeypatch.setattr("app.gateway.auth.config.get_auth_config", lambda: secret)
+    first = control.bind(request)
+    durable = await first.configuration_revision(assistant_ids=["lead_agent"], durable=True)
+    ephemeral = await first.configuration_revision(assistant_ids=["lead_agent"])
+    control.close()
+    second = GatewayAgentRunsHost(request.app, load_user=load_user).bind(request)
+    assert durable == await second.configuration_revision(assistant_ids=["lead_agent"], durable=True)
+    assert ephemeral != await second.configuration_revision(assistant_ids=["lead_agent"])
+    secret.jwt_secret = "rotated"
+    assert durable != await second.configuration_revision(assistant_ids=["lead_agent"], durable=True)
+    load_user.return_value = None
+    with pytest.raises(AgentRunError):
+        await second.configuration_revision(assistant_ids=["lead_agent"], durable=True)

backend/tests/test_extension_api_contracts.py

@@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package():
     """Every additive contract slice bumps both gates together."""
     from importlib.metadata import version
 
-    assert API_VERSION == "0.2.7"
+    assert API_VERSION == "0.2.8"
     assert API_VERSION == version("deerflow-extension-api")
 
 

backend/tests/test_workflow_plugin.py

@@ -46,7 +46,7 @@ class Runs:
 
     file_content = b"original file"
 
-    async def configuration_revision(self, *, assistant_ids):
+    async def configuration_revision(self, *, assistant_ids, durable=False):
         if self.denied:
             raise AgentRunError(403, "revoked")
         return self.revision
@@ -356,7 +356,7 @@ async def test_service_shutdown_drains_children_and_worker(plugin):
     assert all(r.status == "interrupted" for r in plugin[3].runs.values())
     await plugin[2].start(None)
     final = await plugin[1]["get"]({"run_id": result["run_id"]}, context(plugin[3]))
-    assert final["status"] == "cancelled"
+    assert final["status"] == "needs_reconciliation"
 
 
 @pytest.mark.asyncio
@@ -453,6 +453,7 @@ async def test_startup_marks_unfinished_records_interrupted_without_replay(plugi
     await done(plugin, result["run_id"])
     record = await asyncio.to_thread(plugin[2].store.get, "alice", result["run_id"])
     record["status"] = "running"
+    record.pop("recovery_version", None)
     await asyncio.to_thread(plugin[2].store.put, "alice", record)
     await plugin[2].stop()
     await plugin[2].start(None)
@@ -607,7 +608,10 @@ async def test_cancel_while_agent_pending_stops_promptly(plugin):
 )
 async def test_spinning_script_still_hits_turn_limit(plugin, monkeypatch, script):
     service = plugin[2]
-    monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 50)
+    # Exercise the frame cap without racing 50 durable frame commits against
+    # the fixture's short wall deadline on loaded CI runners.
+    monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 3)
+    monkeypatch.setitem(service.config, "timeout_seconds", 30)
     result = await launch(plugin, script)
     final = await done(plugin, result["run_id"])
     assert final["status"] == "failed", final
@@ -744,7 +748,7 @@ async def test_cancelled_launch_caller_keeps_committed_sqlite_launch_owned(plugi
             await asyncio.wait_for(runs.entered.wait(), 5)
             await plugin[1]["cancel"]({"run_id": retry["run_id"]}, context(runs))
         final = await done(plugin, retry["run_id"])
-        assert final["status"] == "cancelled", final
+        assert final["status"] in ({"recoverable", "needs_reconciliation"} if shutdown else {"cancelled"}), final
         assert len(runs.starts) <= 1
         assert not service.entries and not service.workers and not service.processes
         await plugin[1]["delete"]({"run_id": retry["run_id"]}, context(runs))
@@ -912,7 +916,7 @@ async def test_phase2_fail_closed_checkpoint_and_restart(plugin, script):
         replacement = Workflows(service.config)
         await replacement.start(None)
         try:
-            assert (await replacement.get({"run_id": run_id}, context(runs)))["status"] == "interrupted"
+            assert (await replacement.get({"run_id": run_id}, context(runs)))["status"] == "recoverable"
             with pytest.raises(ValueError):
                 await replacement.resume({"run_id": run_id, "step_id": "p", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
         finally:

backend/tests/test_workflow_recovery.py

@@ -0,0 +1,408 @@
+"""Recovery exercises the isolated script worker and durable SQLite boundaries."""
+
+import asyncio
+import copy
+import os
+from dataclasses import replace
+
+import pytest
+import test_workflow_plugin as fixtures
+from test_workflow_plugin import context, done, launch, waiting
+
+plugin = fixtures.plugin
+
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.skipif(os.name != "posix", reason="Workflow execution requires POSIX")]
+
+
+async def restart_snapshot(plugin, run_id):
+    """Retain the last committed crash image, without shutdown's cleanup writes."""
+    service = plugin[2]
+    record = await asyncio.to_thread(service.store.get, "alice", run_id)
+    await service.stop()
+    current = await asyncio.to_thread(service.store.get, "alice", run_id)
+    record["revision"] = current.get("revision", 0)
+    await asyncio.to_thread(service.store.put, "alice", record)
+    await service.start(None)
+    return record
+
+
+async def test_restart_approval_reuses_outputs_and_exact_version(plugin):
+    _, actions, service, runs = plugin
+    result = await launch(
+        plugin,
+        "const a=await agent({stepId:'a',agentRef:'lead_agent',task:1}); await humanCheckpoint({stepId:'review',sourceStepId:'a',artifactVersion:a.artifactVersion}); return await agent({stepId:'b',agentRef:'reviewer',task:a.output});",
+    )
+    run_id = result["run_id"]
+    paused = await waiting(plugin, run_id, "waiting_approval")
+    await restart_snapshot(plugin, run_id)
+    assert (await actions["get"]({"run_id": run_id}, context(runs)))["status"] == "recoverable"
+    session = replace(context(runs), interaction="session")
+    await actions["recover"]({"run_id": run_id}, session)
+    recovered = await waiting(plugin, run_id, "waiting_approval")
+    assert recovered["checkpoints"] == paused["checkpoints"]
+    checkpoint = recovered["checkpoints"][0]
+    payload = dict(run_id=run_id, step_id="review", attempt=1, artifact_version=checkpoint["artifact_version"], decision="approve")
+    await asyncio.gather(actions["decide"](payload, session), actions["decide"](payload, session))
+    final = await done(plugin, run_id)
+    assert final["status"] == "completed", final
+    assert len(runs.starts) == 2
+
+
+async def test_recovery_rejects_changed_config_and_revoked_permission(plugin):
+    from deerflow_extension_api import AgentRunError
+
+    _, actions, service, runs = plugin
+    run_id = (await launch(plugin, "await pause({stepId:'p'}); return 1;"))["run_id"]
+    await waiting(plugin, run_id, "paused")
+    await restart_snapshot(plugin, run_id)
+    session = replace(context(runs), interaction="session")
+    runs.denied = True
+    with pytest.raises(AgentRunError):
+        await actions["recover"]({"run_id": run_id}, session)
+    runs.denied = False
+    runs.revision = "changed"
+    with pytest.raises(ValueError, match="configuration"):
+        await actions["recover"]({"run_id": run_id}, session)
+    assert not runs.starts
+
+
+async def test_store_rejects_stale_revision(plugin):
+    service = plugin[2]
+    run_id = (await launch(plugin, "return 1;"))["run_id"]
+    await done(plugin, run_id)
+    first = await asyncio.to_thread(service.store.get, "alice", run_id)
+    stale = copy.deepcopy(first)
+    await asyncio.to_thread(service.store.put, "alice", first)
+    with pytest.raises(ValueError, match="revision"):
+        await asyncio.to_thread(service.store.put, "alice", stale)
+
+
+async def install_snapshot(plugin, record):
+    service = plugin[2]
+    await service.stop()
+    latest = await asyncio.to_thread(service.store.get, "alice", record["id"])
+    record = copy.deepcopy(record)
+    record["revision"] = latest.get("revision", 0)
+    await asyncio.to_thread(service.store.put, "alice", record)
+    await service.start(None)
+
+
+async def test_approval_committed_before_dispatch_recovers_once(plugin):
+    _, actions, service, runs = plugin
+    run_id = (
+        await launch(
+            plugin, "const a=await agent({stepId:'a',agentRef:'lead_agent',task:1}); await humanCheckpoint({stepId:'c',sourceStepId:'a',artifactVersion:a.artifactVersion}); return await agent({stepId:'b',agentRef:'reviewer',task:2});"
+        )
+    )["run_id"]
+    await waiting(plugin, run_id, "waiting_approval")
+    record = await asyncio.to_thread(service.store.get, "alice", run_id)
+    record["checkpoints"][0].update(status="approved", decided_by="alice", decided_at=1)
+    await install_snapshot(plugin, record)
+    session = replace(context(runs), interaction="session")
+    await asyncio.gather(actions["recover"]({"run_id": run_id}, session), actions["recover"]({"run_id": run_id}, session))
+    result = await done(plugin, run_id)
+    assert result["status"] == "completed", result
+    assert len(runs.starts) == 2
+
+
+@pytest.mark.parametrize("boundary", ["admission", "result"])
+async def test_child_crash_boundaries_never_repeat_dispatch(plugin, monkeypatch, boundary):
+    _, actions, service, runs = plugin
+    snapshots = []
+    original = runs.start if boundary == "admission" else runs.configuration_evidence
+
+    async def capture(**kw):
+        value = await original(**kw)
+        entry = next(iter(service.entries.values()))
+        snapshots.append(await asyncio.to_thread(service.store.get, "alice", entry["record"]["id"]))
+        return value
+
+    monkeypatch.setattr(runs, "start" if boundary == "admission" else "configuration_evidence", capture)
+    run_id = (await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:1});"))["run_id"]
+    await done(plugin, run_id)
+    await install_snapshot(plugin, snapshots[0])
+    session = replace(context(runs), interaction="session")
+    result = await actions["recover"]({"run_id": run_id}, session)
+    if boundary == "admission":
+        assert result["status"] == "needs_reconciliation"
+    else:
+        final = await done(plugin, run_id)
+        assert final["status"] == "completed", final
+        assert final["result"]["output"] == {"echo": 1}
+    assert len(runs.starts) == 1
+
+
+async def test_parallel_results_and_file_approval_survive_restart(plugin):
+    _, actions, service, runs = plugin
+    script = (
+        "const results=await parallel([()=>agent({stepId:'a',agentRef:'lead_agent',task:1}),()=>agent({stepId:'b',agentRef:'reviewer',task:2})]); "
+        "const file=await captureArtifact({stepId:'f',sourceStepId:'a',path:'report.txt'}); "
+        "await humanCheckpoint({stepId:'c',sourceStepId:'f',artifactVersion:file.artifactVersion}); "
+        "return await readArtifact({artifactVersion:file.artifactVersion});"
+    )
+    run_id = (await launch(plugin, script))["run_id"]
+    paused = await waiting(plugin, run_id, "waiting_approval")
+    await restart_snapshot(plugin, run_id)
+    runs.file_content = b"changed source must not be read"
+    session = replace(context(runs), interaction="session")
+    await actions["recover"]({"run_id": run_id}, session)
+    await waiting(plugin, run_id, "waiting_approval")
+    await actions["decide"](dict(run_id=run_id, step_id="c", attempt=1, artifact_version=paused["checkpoints"][0]["artifact_version"], decision="approve"), session)
+    final = await done(plugin, run_id)
+    assert final["status"] == "completed", final
+    import base64
+
+    assert base64.b64decode(final["result"]["content_base64"]) == b"original file"
+    assert len(runs.starts) == 2
+
+
+async def test_real_process_kill_during_review(plugin):
+    import json
+    import os
+    import sys
+    from pathlib import Path
+
+    _, actions, service, runs = plugin
+    await service.stop()
+    root = Path(__file__).resolve().parents[2]
+    program = r"""
+import asyncio,json,sys
+from deerflow_extension_workflows.service import Workflows
+from test_workflow_plugin import Runs,context
+async def main():
+    service=Workflows(json.loads(sys.argv[1]));await service.start(None)
+    runs=Runs()
+    result=await service.launch({'request_id':'kill-test','script':("const a=await agent({stepId:'a',agentRef:'lead_agent',task:1}); "
+        "await humanCheckpoint({stepId:'c',sourceStepId:'a',artifactVersion:a.artifactVersion}); return a.output;"),'args':{}},context(runs))
+    while True:
+        state=await service.get({'run_id':result['run_id']},context(runs))
+        if state['status']=='waiting_approval':break
+        await asyncio.sleep(.01)
+    print(result['run_id'],flush=True)
+    await asyncio.Event().wait()
+asyncio.run(main())
+"""
+    env = {**os.environ, "PYTHONPATH": os.pathsep.join([str(root / "backend/tests"), str(root / "backend"), str(root / "examples/deerflow-extension-workflows")])}
+    child = await asyncio.create_subprocess_exec(sys.executable, "-c", program, json.dumps(service.config), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env)
+    try:
+        async with asyncio.timeout(30):
+            run_id = (await child.stdout.readline()).decode().strip()
+        assert len(run_id) == 32, (await child.stderr.read()).decode() if child.returncode is not None else run_id
+        child.kill()
+        await child.wait()
+        await service.start(None)
+        state = await actions["get"]({"run_id": run_id}, context(runs))
+        assert state["status"] == "recoverable"
+        session = replace(context(runs), interaction="session")
+        await actions["recover"]({"run_id": run_id}, session)
+        state = await waiting(plugin, run_id, "waiting_approval")
+        await actions["decide"](dict(run_id=run_id, step_id="c", attempt=1, artifact_version=state["checkpoints"][0]["artifact_version"], decision="approve"), session)
+        final = await done(plugin, run_id)
+        assert final["status"] == "completed", final
+        assert final["result"] == {"echo": 1}
+        assert not runs.starts
+    finally:
+        if child.returncode is None:
+            child.kill()
+            await child.wait()
+
+
+async def test_graceful_shutdown_pause_and_cancel_recovery(plugin):
+    _, actions, service, runs = plugin
+    run_id = (await launch(plugin, "await pause({stepId:'p'}); return 7;"))["run_id"]
+    await waiting(plugin, run_id, "paused")
+    await service.stop()
+    await service.start(None)
+    session = replace(context(runs), interaction="session")
+    assert (await actions["get"]({"run_id": run_id}, session))["status"] == "recoverable"
+    await actions["cancel"]({"run_id": run_id}, session)
+    with pytest.raises(ValueError, match="not recoverable"):
+        await actions["recover"]({"run_id": run_id}, session)
+    assert not runs.starts
+
+
+@pytest.mark.parametrize("change", ["runtime", "journal", "expiry", "owner", "tool"])
+async def test_recovery_guards_prevent_new_dispatch(plugin, change):
+    from deerflow_extension_api import ExtensionPrincipal
+    from deerflow_extension_api.plugins import ToolContext
+
+    _, actions, service, runs = plugin
+    run_id = (await launch(plugin, "await pause({stepId:'p'}); return await agent({stepId:'a',agentRef:'lead_agent',task:1});"))["run_id"]
+    await waiting(plugin, run_id, "paused")
+    record = await asyncio.to_thread(service.store.get, "alice", run_id)
+    if change == "runtime":
+        record["runtime_version"] = "incompatible"
+    if change == "journal":
+        record["journal"][0]["frame"]["messages"][0]["payload"]["stepId"] = "changed"
+    if change == "expiry":
+        record["pause_expires_at"] = 1
+    await install_snapshot(plugin, record)
+    session = replace(context(runs), interaction="session")
+    if change == "owner":
+        session = replace(session, principal=ExtensionPrincipal("bob"))
+    if change == "tool":
+        session = ToolContext(session.principal, session.settings, "thread", agent_runs=runs)
+    if change in ("runtime", "owner", "tool"):
+        with pytest.raises(ValueError):
+            await actions["recover"]({"run_id": run_id}, session)
+    else:
+        await actions["recover"]({"run_id": run_id}, session)
+        final = await done(plugin, run_id)
+        assert final["status"] == "failed", final
+    assert not runs.starts
+
+
+async def test_file_capture_commit_before_metadata_reuses_original_bytes(plugin, monkeypatch):
+    _, actions, service, runs = plugin
+    snapshots = []
+    capture = service.store.capture_artifact
+
+    def capture_before_metadata(owner, workflow, *args):
+        result = capture(owner, workflow, *args)
+        snapshots.append(service.store.get(owner, workflow))
+        return result
+
+    monkeypatch.setattr(service.store, "capture_artifact", capture_before_metadata)
+    run_id = (await launch(plugin, "await agent({stepId:'a',agentRef:'lead_agent',task:1}); return await captureArtifact({stepId:'f',sourceStepId:'a',path:'report.txt'});"))["run_id"]
+    final = await done(plugin, run_id)
+    assert final["status"] == "completed"
+    version = final["result"]["artifactVersion"]
+    await install_snapshot(plugin, snapshots[0])
+    runs.file_content = b"new mutable source"
+    await actions["recover"]({"run_id": run_id}, replace(context(runs), interaction="session"))
+    recovered = await done(plugin, run_id)
+    assert recovered["status"] == "completed", recovered
+    assert recovered["result"]["artifactVersion"] == version
+    assert (await asyncio.to_thread(service.store.artifact, "alice", run_id, version))[1] == b"original file"
+    assert len(runs.starts) == 1
+
+
+async def test_recovered_interrupted_child_cannot_dispatch_even_when_script_catches(plugin, monkeypatch):
+    from deerflow_extension_api import AgentRun
+
+    _, actions, service, runs = plugin
+    snapshots = []
+    original = runs.configuration_evidence
+
+    async def snapshot(**kwargs):
+        entry = next(iter(service.entries.values()))
+        snapshots.append(await asyncio.to_thread(service.store.get, "alice", entry["record"]["id"]))
+        return await original(**kwargs)
+
+    monkeypatch.setattr(runs, "configuration_evidence", snapshot)
+    run_id = (await launch(plugin, "try {await agent({stepId:'a',agentRef:'lead_agent',task:1});} catch(e) {} await pause({stepId:'p'}); return await agent({stepId:'b',agentRef:'reviewer',task:2});"))["run_id"]
+    await waiting(plugin, run_id, "paused")
+    record = snapshots[0]
+    row = record["steps"][0]
+    runs.runs[row["run_id"]] = AgentRun(row["thread_id"], row["run_id"], "interrupted")
+    await install_snapshot(plugin, record)
+    await actions["recover"]({"run_id": run_id}, replace(context(runs), interaction="session"))
+    final = await done(plugin, run_id)
+    assert final["status"] == "needs_reconciliation", final
+    assert len(runs.starts) == 1
+
+
+async def test_recovered_cancel_survives_caller_cancellation(plugin, monkeypatch):
+    _, actions, service, runs = plugin
+    run_id = (await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:1});"))["run_id"]
+    await done(plugin, run_id)
+    record = await asyncio.to_thread(service.store.get, "alice", run_id)
+    record["status"] = "running"
+    record["steps"][0]["status"] = "running"
+    await install_snapshot(plugin, record)
+    child_id = record["steps"][0]["run_id"]
+    runs.runs[child_id] = replace(runs.runs[child_id], status="running")
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = service.cancel_child
+
+    async def blocked_cancel(capability, row):
+        entered.set()
+        await release.wait()
+        await original(capability, row)
+
+    monkeypatch.setattr(service, "cancel_child", blocked_cancel)
+    caller = asyncio.create_task(actions["cancel"]({"run_id": run_id}, context(runs)))
+    try:
+        await asyncio.wait_for(entered.wait(), 5)
+        caller.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await caller
+        assert service.admissions, "Service must retain ownership of child cleanup"
+    finally:
+        release.set()
+        await asyncio.gather(caller, *list(service.admissions), return_exceptions=True)
+    final = await actions["get"]({"run_id": run_id}, context(runs))
+    assert final["status"] == "cancelled", final
+    assert final["steps"][0]["child_status"] == "interrupted"
+    assert (await actions["cancel"]({"run_id": run_id}, context(runs)))["status"] == "cancelled"
+
+
+async def test_save_drains_repeated_cancellation_before_unlocking(plugin, monkeypatch):
+    import threading
+
+    service = plugin[2]
+    run_id = (await launch(plugin, "return 1;"))["run_id"]
+    await done(plugin, run_id)
+    record = await asyncio.to_thread(service.store.get, "alice", run_id)
+    revision = record["revision"]
+    entered, release = asyncio.Event(), threading.Event()
+    loop = asyncio.get_running_loop()
+    original = service.store.put
+
+    def blocked_put(owner, snapshot):
+        loop.call_soon_threadsafe(entered.set)
+        assert release.wait(10), "Test did not release SQLite write"
+        return original(owner, snapshot)
+
+    monkeypatch.setattr(service.store, "put", blocked_put)
+    caller = asyncio.create_task(service.save("alice", record))
+    try:
+        await asyncio.wait_for(entered.wait(), 5)
+        for _ in range(2):
+            caller.cancel()
+            for _ in range(3):
+                await asyncio.sleep(0)
+        assert service.write_lock.locked(), "Do not unlock while SQLite is still committing"
+        assert not caller.done(), "Cancellation must wait for revision publication"
+    finally:
+        release.set()
+        await asyncio.gather(caller, return_exceptions=True)
+    assert caller.cancelled()
+    assert record["revision"] == revision + 1
+    assert (await asyncio.to_thread(service.store.get, "alice", run_id))["revision"] == record["revision"]
+    await service.save("alice", record)
+    assert record["revision"] == revision + 2
+
+
+async def test_disconnected_cancel_logs_terminal_save_failure(plugin, monkeypatch, caplog):
+    _, actions, service, runs = plugin
+    run_id = (await launch(plugin, "await pause({stepId:'p'}); return 1;"))["run_id"]
+    await waiting(plugin, run_id, "paused")
+    await restart_snapshot(plugin, run_id)
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = service.save
+
+    async def fail_terminal_save(owner, record):
+        if record["status"] == "cancelled":
+            entered.set()
+            await release.wait()
+            raise OSError("synthetic storage failure")
+        await original(owner, record)
+
+    monkeypatch.setattr(service, "save", fail_terminal_save)
+    caller = asyncio.create_task(actions["cancel"]({"run_id": run_id}, context(runs)))
+    try:
+        await asyncio.wait_for(entered.wait(), 5)
+        caller.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await caller
+    finally:
+        release.set()
+        await asyncio.gather(caller, *list(service.admissions), return_exceptions=True)
+    final = await actions["get"]({"run_id": run_id}, context(runs))
+    assert final["status"] == "cancelling"
+    assert not service.admissions
+    errors = [r for r in caplog.records if r.levelname == "ERROR" and r.name == type(service).__module__]
+    assert len(errors) == 1
+    assert "OSError" in errors[0].message

backend/uv.lock

@@ -924,7 +924,7 @@ extensions = []
 
 [[package]]
 name = "deerflow-extension-api"
-version = "0.2.7"
+version = "0.2.8"
 source = { editable = "packages/extension-api" }
 
 [[package]]

examples/deerflow-extension-workflows/AGENTS.md

@@ -29,8 +29,9 @@ Anchor results on the visible human input, including canonical __user rewrites;
 never use hidden system/memory IDs. Retain UI launch identity across refresh failures.
 Thread latest state is not immutable run output. Check message/run attribution,
 interrupts and pending human input before copying output. Agent success is never
-human approval. Live checkpoints retain the worker without replay; restart interrupts
-all waits. Bind human decisions to source output hash, step/attempt and authenticated
+human approval. Live checkpoints retain the worker. On restart, versioned bridge-journal replay
+requires an authenticated recovery action; reuse committed outputs and decisions.
+Legacy records are inspection-only. Never resubmit uncertain admissions. Bind human decisions to source output hash, step/attempt and authenticated
 session ActionContext, never ToolContext. Persist before waking; fence duplicate
 resume by checkpoint ID. Config revision is a declarative preflight guard. Store actual host assembly
 evidence per child run; never expose raw credentials. File checkpoints bind
@@ -40,3 +41,23 @@ that exact version. Never claim hard currency limits or freeze external state. D
 Tests live in backend/tests/test_workflow_plugin.py. Run the backend offline and
 blocking-I/O targets plus verify_browser.mjs. The browser fixture is loopback-only
 and synthetic; it must never become a deployed authentication bypass.
+
+Phase 3: recovery.py owns session reattachment, exact bridge replay and admission
+reconciliation classification. Store.put uses conditional revisions; drain SQLite
+writes through repeated cancellation before releasing the write lock. Service-own
+recovered-run cancellation through child cleanup and the terminal write, and drain
+it during shutdown even if its caller disconnects. Log failed service-owned actions
+at error level even after caller disconnection; omit raw exception payloads as for
+supervisor failures. Persist bridge intent
+before host effects and responses before worker delivery. Include worker/supervisor
+source and engine version in compatibility checks; Date/random are unavailable.
+Preserve original wait expiry and remaining execution budget. Read captured files
+by stored step/version after crashes, never overwrite them from mutable sources.
+Phase 4 leases/fenced dispatch remain unimplemented: retain the exclusive flock.
+Recovery tests: backend/tests/test_workflow_recovery.py, including real process kill.
+
+AgentRuns.configuration_revision(durable=True) uses a purpose-separated HMAC
+under the persisted host signing secret for restart comparison. Default revisions
+remain process-local. Rotation invalidates comparisons; no raw config or secret is
+returned. Recovery callers must bind a fresh authenticated session capability;
+never serialize handles or reconstruct authority from a stored user ID alone.

examples/deerflow-extension-workflows/README.md

@@ -1,4 +1,4 @@
-# Script workflows (Phase 2)
+# Script workflows (Phase 3)
 
 An opt-in packaged extension that runs inspectable JavaScript orchestration in the
 background. Scripts call full DeerFlow Agents through the public `AgentRuns`
@@ -9,12 +9,12 @@ progress, child conversation/run identities, and final result.
 Versioned definitions, retained output snapshots, explicit same-process pause/resume,
 and authenticated human checkpoints support staged inspection reports. Agent success,
 script verdicts, tool permission approval and human output approval remain separate facts.
-No arbitrary JavaScript continuation serialization, restart replay or hard monetary
-budget is provided.
+Compatible workflows can be recovered explicitly after a single Gateway restart.
+No arbitrary JavaScript continuation serialization or hard monetary budget is provided.
 
 ## Install and enable
 
-Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.7,
+Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.8,
 and a **single Gateway process per storage file**. Windows execution is unsupported.
 The package holds an exclusive process lock; another process using that same file
 fails service startup rather than recovering another worker's live workflows.
@@ -156,14 +156,15 @@ unique step-ID namespace; each attempt is 1 and has at most one decision. Up to
 A pause retains the same isolated worker and its live JS variables. **Resume this
 checkpoint** continues there, reusing completed values without reading newer thread
 state or replaying earlier script code. It cannot retry a failed/cancelled workflow,
-answer a child's graph interrupt, change script/arguments, or resume after restart.
+answer a child's graph interrupt or change script/arguments. After restart use
+**Recover workflow** first, then resume or decide the retained checkpoint.
 An old resume request names its exact step/attempt and cannot release a later pause.
 Paused workers consume normal active workflow slots and memory; cancel unused waits.
 The execution wall budget is suspended while waiting; each wait has its own
 `pause_timeout_seconds` deadline (default 24 hours, maximum 7 days). CPU, call and
-log limits remain cumulative. Expiry fails closed and never approves anything.
+log limits remain cumulative; worker CPU limits apply per worker incarnation. Expiry fails closed and never approves anything.
 
-The host records an opaque process-local **declarative configuration revision**:
+The host records an opaque durable **declarative configuration revision**:
 app configuration, extension configuration, and allowed Custom Agent config/SOUL.
 It rechecks this revision before each new child and when continuing a checkpoint;
 changes require a new workflow. Operator limits/allowlist changes also invalidate
@@ -215,9 +216,9 @@ service state, runtime tool results or deterministic replay.
 
 The UI exposes retained outputs one at a time to keep responses bounded. Checkpoint
 receipts retain who decided, when, source attempt and output version. On restart
-these records remain inspectable but execution is interrupted; there is no automatic
-replay, authorization reattachment or continuation serialization. Start a new run
-only after considering prior external effects.
+these records remain inspectable. Compatible runs become `recoverable`; a fresh
+authenticated browser session must explicitly recover them. No credentials or
+capability handles are serialized. Start new work only after considering prior effects.
 
 ## Limits and status
 
@@ -262,7 +263,8 @@ reservation/enforcement contract before advertising that guarantee.
 | paused / waiting_approval | Live worker retained at an explicit script checkpoint; occupies an active workflow slot |
 | waiting_input | Child has a graph interrupt or human-input artifact; downstream stopped |
 | needs_reconciliation | Admission/cancellation outcome uncertain; inspect child identities before new work |
-| interrupted | Supervisor externally cancelled, or unfinished record recovered after Gateway restart; no execution replay |
+| recoverable | Restart/shutdown retained a compatible execution; sign in and select Recover workflow |
+| interrupted | Legacy record, cancelled startup record, or externally interrupted supervisor; inspect child effects |
 
 Wall-clock expiry reports `failed` with `Execution deadline exceeded` and cleans
 up tracked children and the worker. A timeout after child admission reports child
@@ -286,6 +288,11 @@ Caller cancellation during launch does not abandon a committed record: the servi
 owns persistence through supervisor publication, including shutdown coordination.
 A matching retry returns that same supervised record rather than an unowned queued
 launch. This does not add execution replay or resume.
+Cancellation of a recoverable workflow is also service-owned: disconnecting the
+caller does not abandon child cleanup or its final status write. Shutdown drains
+that cleanup before releasing the store's process lock.
+If a background action fails after disconnection, the Gateway logs its exception
+type at error level so operators can inspect retained workflow and child records.
 
 Result collection anchors on the visible human input, including the host's canonical
 `__user` rewrite after dynamic context injection. Hidden system/memory reminders
@@ -295,9 +302,9 @@ human turn, because `get_state()` is the thread's latest state, not immutable ru
 output. Results are copied before further script calls. Per-step inputs remain private. Owner-scoped `get_step` returns one immutable retained
 output with its attempt and artifact version; run inspection returns bounded metadata. Child conversations remain accessible through normal host authorization.
 
-At startup, unfinished records become `interrupted`; retained IDs do not recreate
-a capability or arbitrary JS continuation. Gateway revokes handles during shutdown,
-so cleanup may require reconciliation. Paused/approval records also become interrupted; their old decisions cannot resume a new worker. Neither graph resume nor tool permission
+At startup, compatible unfinished records become `recoverable`; legacy or cancelling
+records become `interrupted`. Retained IDs do not recreate a capability by themselves. Gateway revokes handles during shutdown,
+so cleanup may require reconciliation. Paused/approval records become recoverable; committed decisions retain their exact version and can continue through the recovery contract below. Neither graph resume nor tool permission
 approval is interpreted as authenticated approval of an artifact version.
 
 ## Verify
@@ -330,3 +337,63 @@ races, limits, owner scope, service shutdown and restart records.
 The implementation uses an original small JSON bridge. `deer-workflow` was evaluated
 for API shapes but is not imported or copied: its dynamic Node module loader,
 unbounded fan-out and external CLI Agent adapters are not used here.
+
+
+## Phase 3 recovery contract
+
+Recovery is explicit and owner-scoped. In **Script workflows**, select a
+`recoverable` run and **Recover workflow**. This is a browser action, not a model
+tool. It obtains a fresh session-bound AgentRuns capability and checks current
+permissions, declarative configuration, operator limits, script/arguments, and
+runtime compatibility before starting a replacement worker. Recovery itself does
+not approve pending human output; review the same retained version and approve
+or reject it separately. Duplicate recovery requests share one active worker.
+
+The extension persists a versioned bridge journal: each emitted request frame is
+committed before host operations; each response batch is committed before it is
+returned to JavaScript. Recovery evaluates the unchanged script from its start,
+verifies its request frames against that journal, and feeds back recorded responses
+in their original order. Confirmed Agent results are reused rather than dispatched
+again. It reconstructs script variables without serializing a JS heap. `Date` and
+`Math.random` are unavailable; pass timestamps/random values in `args`. A divergent
+history stops before issuing new effects. The journal is bounded at 4 MiB and the complete record at
+8 MiB per run; the worker has a 10,000-round limit.
+
+Crash boundaries have explicit outcomes:
+
+| Boundary | Recovery behavior |
+| --- | --- |
+| Before child admission intent | A queued step may be admitted with its original identity/input/key |
+| After admission intent, before run ID is stored | `needs_reconciliation`; never resubmit an uncertain Agent call |
+| Known child run, before result is stored | Observe that run without starting it again; validate success, message attribution and host evidence before retaining the result |
+| Child interrupted/failed or external outcome uncertain | Stop; inspect child identities and reconcile effects before starting new work |
+| Approval persisted, before downstream dispatch | Reuse that exact decision and continue through the recorded boundary |
+| File captured, before workflow metadata is stored | Recover the immutable stored bytes by capture step; never overwrite with a newer source |
+
+There is no automatic retry or operator override of uncertain external effects.
+Cancellation requests interruption of known children and cannot undo committed
+business effects. Recovery does not resume a child's interrupted graph. Legacy
+Phase 1/2 records have no replay journal and remain inspection-only after restart.
+Normal shutdown preserves compatible pending waits; interrupted active children
+may require reconciliation.
+
+Execution time remaining is retained. A crash during active execution consumes
+elapsed wall time including downtime; a persisted human/pause wait suspends the
+execution budget, but its original absolute wait expiry still applies. Replay
+also consumes the remaining execution budget. The CPU limit is per worker;
+Agent-call and log quotas are retained across recovery.
+
+The workflow runtime source and QuickJS version must match. Incompatible code,
+script/arguments, configuration or operator-limit changes reject recovery rather
+than silently migrating execution. The durable configuration digest uses a
+purpose-separated HMAC derived from the host's persisted signing secret. Preserve
+that secret and the Gateway's durable run/thread/checkpoint data across restart;
+secret rotation invalidates recovery. Stored assembly evidence remains historical
+and is distinct from this configuration preflight check.
+
+Phase 3 and Phase 4 share the design: stable step/attempt and idempotency identities,
+conditional record revisions, a durable intent/result journal, and fresh authority
+are the foundations for later leases and fenced dispatch. **Phase 3 still requires
+one Gateway process per workflow store.** It does not implement replica routing,
+leases, automatic takeover or cluster-wide quotas; these belong to Phase 4. Keep
+the exclusive process lock until those guarantees exist.

examples/deerflow-extension-workflows/deerflow_extension_workflows/__init__.py

@@ -15,7 +15,7 @@ from deerflow_extension_api import (
 from .service import NAMESPACE, Workflows
 
 
-@extension(api="0.2.7", name="workflows")
+@extension(api="0.2.8", name="workflows")
 def install(registry, config):
     enabled = config.get("enabled", False)
     if type(enabled) is not bool:
@@ -92,7 +92,10 @@ def install(registry, config):
                 namespace=NAMESPACE,
                 title="Script workflows / 脚本工作流",
                 enabled=enabled,
-                description="Bounded background script orchestration of full Agents. Versioned definitions, live-process resume and authenticated output checkpoints. No restart replay or hard monetary budget.",
+                description=(
+                    "Bounded background script orchestration of full Agents. Versioned definitions, authenticated output checkpoints and explicit single-Gateway recovery. "
+                    "Uncertain child effects require reconciliation. No hard monetary budget."
+                ),
                 frontend=BrowserAssets("workflows.v1", Path(__file__).parent),
                 backend=tuple(
                     BackendAction(name, getattr(workflows, name))
@@ -106,6 +109,7 @@ def install(registry, config):
                         "definitions",
                         "get_definition",
                         "resume",
+                        "recover",
                         "decide",
                         "get_step",
                         "get_artifact",
@@ -116,7 +120,16 @@ def install(registry, config):
                     ModelTool(
                         "read_workflow_artifact",
                         "Read exact approved immutable file bytes, base64 in 16 KiB chunks. Pass workflow run_id, version, offset. Verify sha256 after assembling chunks. Never read the mutable original path as a substitute.",
-                        {"type": "object", "properties": {"run_id": {"type": "string"}, "version": {"type": "string"}, "offset": {"type": "integer", "minimum": 0}}, "required": ["run_id", "version"], "additionalProperties": False},
+                        {
+                            "type": "object",
+                            "properties": {
+                                "run_id": {"type": "string"},
+                                "version": {"type": "string"},
+                                "offset": {"type": "integer", "minimum": 0},
+                            },
+                            "required": ["run_id", "version"],
+                            "additionalProperties": False,
+                        },
                         workflows.read_artifact,
                     ),
                     ModelTool(
@@ -148,7 +161,8 @@ def install(registry, config):
                         "Launch an inspectable JavaScript orchestration script in the background and return a workflow run ID. Use agent({stepId,agentRef,task}), parallel([()=>...]), log(text), pause({stepId}) and "
                         "humanCheckpoint({stepId,sourceStepId,artifactVersion}). Await checkpoints alone after all calls. agent resolves to "
                         "{output,threadId,runId,status,artifactVersion}; args contains JSON arguments. Use unique step IDs, explicit bounded loops and await every Agent call. Only operator-allowed agent references are admitted with "
-                        "this caller's permissions. Keep request_id stable on retries. Resume retains the live worker only; restart ends resumability. Human decisions are browser-only and cover retained step output. "
+                        "this caller's permissions. Keep request_id stable on retries. After restart, an authenticated browser session can explicitly recover a compatible workflow. "
+                        "Pass timestamps and randomness in args; Date and Math.random are unavailable. Human decisions are browser-only and cover retained step output. "
                         "Inspect results on request; do not poll in a loop or equate Agent "
                         "success with product approval. No hard monetary cost guarantee.",
                         launch_schema,

examples/deerflow-extension-workflows/deerflow_extension_workflows/artifacts.py

@@ -11,13 +11,43 @@ class Artifacts:
         from .service import text
 
         payload = message.get("payload")
-        if not isinstance(payload, dict) or set(payload) != {"stepId", "sourceStepId", "path"}:
+        if not isinstance(payload, dict) or set(payload) != {
+            "stepId",
+            "sourceStepId",
+            "path",
+        }:
             raise ValueError("captureArtifact requires stepId, sourceStepId and path")
         record, owner = entry["record"], entry["owner"]
         step_id = text(payload["stepId"], 80)
+        if entry.get("recover_frame"):
+            info = next((a for a in record["artifacts"] if a["step_id"] == step_id), None)
+            if info:
+                await asyncio.to_thread(self.store.artifact, owner, record["id"], info["version"])
+                return {
+                    "id": message["id"],
+                    "value": {**info, "artifactVersion": info["version"]},
+                }
+            retained = await asyncio.to_thread(self.store.artifact_by_step, owner, record["id"], step_id)
+            if retained:
+                source = next(s for s in record["steps"] if s["step_id"] == payload["sourceStepId"])
+                info = retained
+                info.update(
+                    source_step_id=source["step_id"],
+                    source_run_id=source["run_id"],
+                    source_thread_id=source["thread_id"],
+                )
+                record["artifacts"].append(info)
+                await self.save(owner, record)
+                return {
+                    "id": message["id"],
+                    "value": {**info, "artifactVersion": info["version"]},
+                }
         if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", step_id) or any(s["step_id"] == step_id for s in record["steps"] + record["checkpoints"] + record["artifacts"]):
             raise ValueError("Invalid or reused artifact step ID")
-        source = next((s for s in record["steps"] if s["step_id"] == payload["sourceStepId"] and s["status"] == "completed"), None)
+        source = next(
+            (s for s in record["steps"] if s["step_id"] == payload["sourceStepId"] and s["status"] == "completed"),
+            None,
+        )
         if source is None:
             raise ValueError("Artifact source must be a completed Agent step")
         path = text(payload["path"], 1000)
@@ -25,10 +55,17 @@ class Artifacts:
         async with asyncio.timeout(10):
             content = await entry["runs"].read_artifact(thread_id=source["thread_id"], path=path)
         info = await asyncio.to_thread(self.store.capture_artifact, owner, record["id"], step_id, name, content)
-        info.update(source_step_id=source["step_id"], source_run_id=source["run_id"], source_thread_id=source["thread_id"])
+        info.update(
+            source_step_id=source["step_id"],
+            source_run_id=source["run_id"],
+            source_thread_id=source["thread_id"],
+        )
         record["artifacts"].append(info)
         await self.save(owner, record)
-        return {"id": message["id"], "value": {**info, "artifactVersion": info["version"]}}
+        return {
+            "id": message["id"],
+            "value": {**info, "artifactVersion": info["version"]},
+        }
 
     @staticmethod
     def approved_artifact(record, version):
@@ -39,7 +76,10 @@ class Artifacts:
 
         if set(payload) - {"run_id", "version", "offset"}:
             raise ValueError("Unexpected artifact read fields")
-        run_id, version = text(payload.get("run_id"), 64), text(payload.get("version"), 64)
+        run_id, version = (
+            text(payload.get("run_id"), 64),
+            text(payload.get("version"), 64),
+        )
         record = await asyncio.to_thread(self.store.get, owner, run_id)
         if require_approval and not self.approved_artifact(record, version):
             raise ValueError("File version has not been approved")
@@ -50,7 +90,12 @@ class Artifacts:
         if offset > len(content):
             raise ValueError("Offset exceeds artifact size")
         end = min(offset + 16384, len(content))
-        return {**info, "offset": offset, "next_offset": end if end < len(content) else None, "content_base64": base64.b64encode(content[offset:end]).decode("ascii")}
+        return {
+            **info,
+            "offset": offset,
+            "next_offset": end if end < len(content) else None,
+            "content_base64": base64.b64encode(content[offset:end]).decode("ascii"),
+        }
 
     async def get_artifact(self, payload, context):
         """Browser review reads the same stored bytes as downstream consumers."""
@@ -65,7 +110,18 @@ class Artifacts:
 
     async def consume(self, entry, message):
         payload = message.get("payload")
-        if not isinstance(payload, dict) or set(payload) - {"artifactVersion", "offset"}:
+        if not isinstance(payload, dict) or set(payload) - {
+            "artifactVersion",
+            "offset",
+        }:
             raise ValueError("readArtifact requires artifactVersion and optional offset")
-        result = await self.artifact_chunk(entry["owner"], {"run_id": entry["record"]["id"], "version": payload.get("artifactVersion"), "offset": payload.get("offset", 0)}, require_approval=True)
+        result = await self.artifact_chunk(
+            entry["owner"],
+            {
+                "run_id": entry["record"]["id"],
+                "version": payload.get("artifactVersion"),
+                "offset": payload.get("offset", 0),
+            },
+            require_approval=True,
+        )
         return {"id": message["id"], "value": result}

examples/deerflow-extension-workflows/deerflow_extension_workflows/lifecycle.py

@@ -1,4 +1,4 @@
-"""Immutable definitions and live-process checkpoints; never replay script effects."""
+"""Immutable definitions and exact-version live/recovered checkpoints."""
 
 import asyncio
 import copy
@@ -49,7 +49,10 @@ class Lifecycle:
         record = await asyncio.to_thread(self.store.get, self.owner(context), text(payload.get("run_id"), 64))
         row = next((s for s in record["steps"] if s["step_id"] == payload.get("step_id")), None)
         if row is None:
-            row = next((s for s in record.get("artifacts", []) if s["step_id"] == payload.get("step_id")), None)
+            row = next(
+                (s for s in record.get("artifacts", []) if s["step_id"] == payload.get("step_id")),
+                None,
+            )
         if not row or row["status"] != "completed":
             raise ValueError("Completed step unavailable")
         result = {k: copy.deepcopy(v) for k, v in row.items() if k != "input"}
@@ -60,7 +63,7 @@ class Lifecycle:
         record = entry["record"]
         revision = record.get("configuration_revision")
         async with asyncio.timeout(5):
-            current = await entry["runs"].configuration_revision(assistant_ids=sorted(self.config["agents"]))
+            current = await entry["runs"].configuration_revision(assistant_ids=sorted(self.config["agents"]), durable=True)
         if revision is None or revision != current:
             raise ValueError("Agent configuration changed or unavailable; start a new workflow")
         from .service import identity
@@ -80,9 +83,10 @@ class Lifecycle:
         step_id = text(payload["stepId"], 80)
         if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", step_id):
             raise ValueError("Invalid checkpoint stepId")
-        if any(s["step_id"] == step_id for s in record["steps"] + record["checkpoints"] + record.get("artifacts", [])):
+        existing = next((r for r in record["checkpoints"] if r["step_id"] == step_id), None) if entry.get("recover_frame") else None
+        if not existing and any(s["step_id"] == step_id for s in record["steps"] + record["checkpoints"] + record.get("artifacts", [])):
             raise ValueError("Step ID already used")
-        if len(record["checkpoints"]) >= self.config["max_calls"]:
+        if not existing and len(record["checkpoints"]) >= self.config["max_calls"]:
             raise ValueError("Checkpoint limit")
         await self.check_configuration(entry)
         row = {
@@ -104,15 +108,27 @@ class Lifecycle:
                 source_attempt=source["attempt"],
                 artifact_version=source["artifact_version"],
             )
+        if existing:
+            if any(existing.get(k) != v for k, v in row.items() if k != "status"):
+                raise ValueError("Recovery checkpoint version changed")
+            row = existing
+            if row["status"] in ("approved", "resumed"):
+                record.pop("pause_expires_at", None)
+                return {"id": message["id"], "value": {"status": row["status"]}}
+            if row["status"] != "pending":
+                raise ValueError("Checkpoint rejected")
+        else:
+            record["checkpoints"].append(row)
+            record["pause_expires_at"] = time.time() + self.config["pause_timeout_seconds"]
         entry["gate"] = asyncio.Event()
-        record["checkpoints"].append(row)
         record["status"] = "waiting_approval" if human else "paused"
-        record["pause_expires_at"] = time.time() + self.config["pause_timeout_seconds"]
+        record["remaining_seconds"] = max(0, deadline.when() - asyncio.get_running_loop().time())
+        record["budget_deadline"] = None
         await self.save(entry["owner"], record)
         # Keep the same isolated worker/JS heap alive. No code before the gate is
         # replayed; suspend only the remaining execution wall budget while waiting.
         remaining = max(0, deadline.when() - asyncio.get_running_loop().time())
-        deadline.reschedule(asyncio.get_running_loop().time() + self.config["pause_timeout_seconds"])
+        deadline.reschedule(asyncio.get_running_loop().time() + max(0, record["pause_expires_at"] - time.time()))
         signals = [
             asyncio.create_task(entry["gate"].wait()),
             asyncio.create_task(entry["cancel"].wait()),
@@ -127,6 +143,7 @@ class Lifecycle:
             raise ValueError("Human rejected output" if row["status"] == "rejected" else "Cancelled")
         await self.check_configuration(entry)
         deadline.reschedule(asyncio.get_running_loop().time() + remaining)
+        record["budget_deadline"] = time.time() + remaining
         record["status"] = "running"
         record.pop("pause_expires_at", None)
         await self.save(entry["owner"], record)
@@ -194,10 +211,10 @@ class Lifecycle:
                     return {"run_id": run_id, "status": "resumed"}
                 status = "resumed"
             if not entry or record["status"] != ("waiting_approval" if human else "paused") or entry["cancel"].is_set():
-                raise ValueError("Live checkpoint unavailable; restart does not permit resume")
+                raise ValueError("Live checkpoint unavailable; recover the workflow first after restart")
             # Revalidate both the retained capability and this caller's live grant.
             async with asyncio.timeout(5):
-                await context.agent_runs.configuration_revision(assistant_ids=sorted(self.config["agents"]))
+                await context.agent_runs.configuration_revision(assistant_ids=sorted(self.config["agents"]), durable=True)
             await self.check_configuration(entry)
             if entry["cancel"].is_set() or record["status"] not in (
                 "paused",

examples/deerflow-extension-workflows/deerflow_extension_workflows/recovery.py

@@ -0,0 +1,109 @@
+"""Versioned bridge journal replay; authority is reacquired from a live session.
+
+Phase 3 retains the exclusive store owner. Conditional revisions and durable
+bridge intents are the foundation for a future leased, fenced cluster owner.
+"""
+
+import asyncio
+import copy
+import json
+
+from deerflow_extension_api.plugins import ToolContext
+
+RECOVERY_VERSION = 1
+
+
+class Recovery:
+    async def recover(self, payload, context):
+        if set(payload) != {"run_id"}:
+            raise ValueError("Recovery requires run_id")
+        if isinstance(context, ToolContext) or context.interaction != "session" or context.principal.is_internal or context.agent_runs is None:
+            raise ValueError("Recovery requires an authenticated browser session")
+        task = asyncio.create_task(self.restore(payload, context))
+        self.admissions.add(task)
+        task.add_done_callback(self.admission_finished)
+        return await asyncio.shield(task)
+
+    async def restore(self, payload, context):
+        from .service import identity, text
+
+        async with self.launch_lock:
+            owner = self.owner(context)
+            run_id = text(payload["run_id"], 64)
+            record = await asyncio.to_thread(self.store.get, owner, run_id)
+            if run_id in self.entries:
+                return {"run_id": run_id, "status": record["status"]}
+            if record["status"] != "recoverable" or record.get("recovery_version") != RECOVERY_VERSION:
+                raise ValueError("Workflow is not recoverable; inspect retained child outcomes")
+            if record.get("runtime_version") != self.runtime_version:
+                raise ValueError("Workflow runtime version changed; recovery requires the original compatible runtime")
+            if record["script_hash"] != self.script_version(record["script"])[1] or record["args_hash"] != identity(record["args"]):
+                raise ValueError("Recovery script or arguments changed")
+            if len(self.entries) >= self.config["max_workflows"] or sum(e["owner"] == owner for e in self.entries.values()) >= self.config["max_workflows_per_owner"]:
+                raise ValueError("Active workflow limit")
+            entry = self.new_entry(owner, record, context.agent_runs)
+            await self.check_configuration(entry)
+            # A committed admission intent with no receipt is never dispatched
+            # again. An operator must reconcile its external effects first.
+            uncertain = [s for s in record["steps"] if s["status"] not in ("queued", "completed", "running") or (s["status"] == "running" and not s.get("run_id"))]
+            if uncertain:
+                record.update(
+                    status="needs_reconciliation",
+                    error="Uncertain child admission or outcome; inspect child identities before retrying",
+                )
+                await self.save(owner, record)
+                return {"run_id": run_id, "status": record["status"]}
+            for artifact in record.get("artifacts", []):
+                await asyncio.to_thread(self.store.artifact, owner, run_id, artifact["version"])
+            entry["replay"] = copy.deepcopy(record["journal"])
+            record.update(status="queued", error=None)
+            await self.save(owner, record)
+            self.publish(entry)
+            return {"run_id": run_id, "status": "queued"}
+
+    async def begin_frame(self, entry, frame):
+        """Persist intent before effects; replay verifies every emitted request."""
+        record = entry["record"]
+        replay = entry.get("replay", [])
+        index = entry.setdefault("frame_index", 0)
+        entry["frame_index"] = index + 1
+        entry["recover_frame"] = False
+        if index < len(replay):
+            expected = replay[index]
+            if expected["frame"] != frame:
+                raise ValueError("Recovery diverged from committed bridge history")
+            pending = entry.setdefault("replay_pending", {})
+            for message in frame["messages"]:
+                if message.get("type") == "agent":
+                    pending[message["id"]] = message
+            if expected["responses"] is not None:
+                for response in expected["responses"]:
+                    pending.pop(response["id"], None)
+                return expected["responses"]
+            entry["recover_frame"] = True
+        else:
+            item = {"frame": copy.deepcopy(frame), "responses": None}
+            size = len(json.dumps(item).encode())
+            if record.get("journal_bytes", 0) + size > 4 * 1024 * 1024:
+                raise ValueError("Recovery journal limit (4 MiB)")
+            record["journal_bytes"] = record.get("journal_bytes", 0) + size
+            record["journal"].append(item)
+            await self.save(entry["owner"], record)
+        # Restore requests outstanding across earlier response batches too.
+        pending = entry.pop("replay_pending", {})
+        current = {m.get("id") for m in frame["messages"]}
+        for message in pending.values():
+            if message["id"] not in current:
+                entry["tasks"].add(asyncio.create_task(self.step(entry, message, replay=True)))
+        return None
+
+    async def deliver_frame(self, entry, process, responses):
+        record = entry["record"]
+        record["remaining_seconds"] = max(0, entry["deadline"].when() - asyncio.get_running_loop().time())
+        size = len(json.dumps(responses).encode())
+        if record.get("journal_bytes", 0) + size > 4 * 1024 * 1024:
+            raise ValueError("Recovery journal limit (4 MiB)")
+        record["journal_bytes"] = record.get("journal_bytes", 0) + size
+        record["journal"][entry["frame_index"] - 1]["responses"] = copy.deepcopy(responses)
+        await self.save(entry["owner"], record)
+        await self.send(process, {"responses": responses})

examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py

@@ -17,6 +17,7 @@ from deerflow_extension_api import AgentRunError
 
 from .artifacts import Artifacts
 from .lifecycle import Lifecycle
+from .recovery import RECOVERY_VERSION, Recovery
 from .store import TERMINAL, Store
 
 NAMESPACE = "community.workflows"
@@ -75,13 +76,26 @@ def project(record, detail=True):
             args=record["args"],
             logs=record["logs"],
             result=record.get("result"),
-            steps=[{k: v for k, v in step.items() if k not in ("input", "output", "configuration_evidence", "input_artifacts")} for step in record["steps"]],
+            steps=[
+                {
+                    k: v
+                    for k, v in step.items()
+                    if k
+                    not in (
+                        "input",
+                        "output",
+                        "configuration_evidence",
+                        "input_artifacts",
+                    )
+                }
+                for step in record["steps"]
+            ],
         )
     encoded(result, 60 * 1024)
     return result
 
 
-class Workflows(Lifecycle, Artifacts):
+class Workflows(Recovery, Lifecycle, Artifacts):
     def __init__(self, config):
         self.config = config
         self.store = None
@@ -107,18 +121,33 @@ class Workflows(Lifecycle, Artifacts):
                 fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                 store = Store(path)
                 store.interrupt_unfinished()
-                return lock, store
+                # Include worker code and engine version in the conservative
+                # replay compatibility guard; upgrades never silently replay.
+                from importlib.metadata import version
+
+                signature = hashlib.sha256()
+                for name in (
+                    "worker.py",
+                    "service.py",
+                    "recovery.py",
+                    "lifecycle.py",
+                    "artifacts.py",
+                ):
+                    signature.update(Path(__file__).with_name(name).read_bytes())
+                signature.update(version("quickjs").encode())
+                return lock, store, signature.hexdigest()
             except BaseException:
                 lock.close()
                 raise
 
-        self.lockfile, self.store = await asyncio.to_thread(initialize)
+        self.lockfile, self.store, self.runtime_version = await asyncio.to_thread(initialize)
         self.available = True
 
     async def stop(self):
         async with self.launch_lock:
             self.available = False
             for entry in self.entries.values():
+                entry["suspending"] = entry["record"]["status"] != "cancelling"
                 entry["cancel"].set()
             workers = list(self.workers.values())
         # Admissions waiting for launch_lock see unavailable; committed admissions
@@ -142,8 +171,24 @@ class Workflows(Lifecycle, Artifacts):
         return context.principal.user_id
 
     async def save(self, owner, record):
+        # SQLite commits survive cancellation of to_thread. Keep the lock and
+        # observe the new revision before propagating cancellation to cleanup.
+        task = asyncio.create_task(self.save_drained(owner, record))
+        try:
+            await asyncio.shield(task)
+        except asyncio.CancelledError:
+            while not task.done():
+                try:
+                    await asyncio.wait({task})
+                except asyncio.CancelledError:
+                    continue
+            if not task.cancelled():
+                task.exception()
+            raise
+
+    async def save_drained(self, owner, record):
         async with self.write_lock:
-            await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record))
+            record["revision"] = await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record))
 
     async def launch(self, payload, context):
         owner = self.owner(context)
@@ -193,8 +238,11 @@ class Workflows(Lifecycle, Artifacts):
         )
         record = {
             "id": uuid.uuid4().hex,
+            "recovery_version": RECOVERY_VERSION,
+            "runtime_version": self.runtime_version,
+            "journal": [],
             "definition": definition,
-            "configuration_revision": await context.agent_runs.configuration_revision(assistant_ids=sorted(self.config["agents"])) if hasattr(context.agent_runs, "configuration_revision") else None,
+            "configuration_revision": await context.agent_runs.configuration_revision(assistant_ids=sorted(self.config["agents"]), durable=True) if hasattr(context.agent_runs, "configuration_revision") else None,
             "operator_revision": identity({**self.config, "agents": sorted(self.config["agents"])}),
             "checkpoints": [],
             "artifacts": [],
@@ -222,7 +270,12 @@ class Workflows(Lifecycle, Artifacts):
     def admission_finished(self, task):
         self.admissions.discard(task)
         if not task.cancelled():
-            task.exception()  # Consume failures even when the launching caller left.
+            error = task.exception()
+            if error is not None:
+                logger.error(
+                    "Workflow background action failed (%s); inspect retained workflow and child records",
+                    type(error).__name__,
+                )
 
     async def admit(self, owner, record, context):
         async with self.launch_lock:
@@ -238,21 +291,27 @@ class Workflows(Lifecycle, Artifacts):
             if len(self.entries) >= self.config["max_workflows"] or len(current) >= self.config["max_workflows_per_owner"]:
                 raise ValueError("Active workflow limit")
             record, created = await asyncio.to_thread(self.store.create, owner, record)
-            entry = {
-                "owner": owner,
-                "record": record,
-                "runs": context.agent_runs,
-                "cancel": asyncio.Event(),
-                "tasks": set(),
-                "slots": asyncio.Semaphore(self.config["concurrency"]),
-                "fault": None,
-            }
-            self.entries[record["id"]] = entry
-            task = asyncio.create_task(self.execute(entry))
-            self.workers[record["id"]] = task
-            task.add_done_callback(lambda t: self.worker_finished(record["id"], t))
+            self.publish(self.new_entry(owner, record, context.agent_runs))
             return {"run_id": record["id"], "status": record["status"]}
 
+    def new_entry(self, owner, record, runs):
+        return {
+            "owner": owner,
+            "record": record,
+            "runs": runs,
+            "cancel": asyncio.Event(),
+            "tasks": set(),
+            "slots": asyncio.Semaphore(self.config["concurrency"]),
+            "fault": None,
+        }
+
+    def publish(self, entry):
+        run_id = entry["record"]["id"]
+        self.entries[run_id] = entry
+        task = asyncio.create_task(self.execute(entry))
+        self.workers[run_id] = task
+        task.add_done_callback(lambda t: self.worker_finished(run_id, t))
+
     def worker_finished(self, run_id, task):
         self.workers.pop(run_id, None)
         self.entries.pop(run_id, None)
@@ -273,18 +332,41 @@ class Workflows(Lifecycle, Artifacts):
         return {"runs": [project(r, False) for r in records], "limit": 20}
 
     async def cancel(self, payload, context):
-        owner = self.owner(context)
-        run_id = text(payload.get("run_id"), 64)
-        record = await asyncio.to_thread(self.store.get, owner, run_id)
-        entry = self.entries.get(run_id)
-        if entry:
-            entry["cancel"].set()
-            if entry["record"]["status"] not in TERMINAL:
-                entry["record"]["status"] = "cancelling"
-            await self.save(owner, entry["record"])
-        # Terminal waiting_input/restart records may still have child work to inspect;
-        # this action never resumes or claims it cancelled an unknown child.
-        return {"run_id": run_id, "status": "cancelling" if entry else record["status"]}
+        task = asyncio.create_task(self.apply_cancel(payload, context))
+        self.admissions.add(task)
+        task.add_done_callback(self.admission_finished)
+        return await asyncio.shield(task)
+
+    async def apply_cancel(self, payload, context):
+        # Serialize against recovery publication. A recovered record cannot be
+        # published after a concurrent cancellation has committed.
+        async with self.launch_lock:
+            owner = self.owner(context)
+            run_id = text(payload.get("run_id"), 64)
+            record = await asyncio.to_thread(self.store.get, owner, run_id)
+            entry = self.entries.get(run_id)
+            if entry:
+                entry["cancel"].set()
+                entry["suspending"] = False
+                if entry["record"]["status"] not in TERMINAL:
+                    entry["record"]["status"] = "cancelling"
+                await self.save(owner, entry["record"])
+                return {"run_id": run_id, "status": "cancelling"}
+            if record["status"] == "recoverable":
+                if context.agent_runs is None:
+                    raise ValueError("Session-bound AgentRuns required")
+                record["status"] = "cancelling"
+                await self.save(owner, record)
+                for row in record["steps"]:
+                    if row.get("run_id") and row["status"] != "completed":
+                        await self.cancel_child(context.agent_runs, row)
+                uncertain = any(s["status"] not in ("completed", "queued", "cancelled") for s in record["steps"])
+                record.update(
+                    status="needs_reconciliation" if uncertain else "cancelled",
+                    error="Inspect uncertain child effects" if uncertain else None,
+                )
+                await self.save(owner, record)
+            return {"run_id": run_id, "status": record["status"]}
 
     async def delete(self, payload, context):
         owner = self.owner(context)
@@ -294,7 +376,7 @@ class Workflows(Lifecycle, Artifacts):
         await asyncio.to_thread(self.store.delete, owner, run_id)
         return {"deleted": True}
 
-    async def step(self, entry, message):
+    async def step(self, entry, message, *, replay=False):
         owner, record, runs = entry["owner"], entry["record"], entry["runs"]
         payload = message.get("payload")
         response = {"id": message["id"]}
@@ -315,7 +397,12 @@ class Workflows(Lifecycle, Artifacts):
                 if not self.approved_artifact(record, version):
                     raise ValueError("File version has not been approved")
                 info, _ = await asyncio.to_thread(self.store.artifact, owner, record["id"], version)
-                references.append({"run_id": record["id"], **{k: info[k] for k in ("version", "sha256", "name", "size")}})
+                references.append(
+                    {
+                        "run_id": record["id"],
+                        **{k: info[k] for k in ("version", "sha256", "name", "size")},
+                    }
+                )
             task = payload["task"]
             if references:
                 task = {
@@ -335,25 +422,36 @@ class Workflows(Lifecycle, Artifacts):
                     }
                 ]
             }
+            existing = next((s for s in record["steps"] if s["step_id"] == step_id), None) if replay else None
+            if existing:
+                row = existing
+                if row["input"] != input or row["agent_ref"] != payload["agentRef"]:
+                    raise ValueError("Recovery step input changed")
+                if row["status"] == "completed":
+                    response["value"] = self.step_result(row)
+                    return response
+                if row["status"] not in ("queued", "running"):
+                    raise ValueError("Child outcome requires reconciliation")
             # Revalidate after file IO; validation and reservation must not yield.
-            if any(s["step_id"] == step_id for s in record["steps"] + record.get("checkpoints", []) + record.get("artifacts", [])):
-                raise ValueError("Step ID already used")
-            if len(record["steps"]) >= self.config["max_calls"]:
-                entry["fault"] = "Total Agent call limit"
-                raise ValueError(entry["fault"])
-            row = {
-                "step_id": step_id,
-                "attempt": 1,
-                "agent_ref": payload["agentRef"],
-                "thread_id": thread_id,
-                "run_id": None,
-                "status": "queued",
-                "input": input,
-                "input_artifacts": references,
-                "idempotency_key": f"{record['id']}:{message_id}:1",
-            }
-            record["steps"].append(row)
-            await self.save(owner, record)
+            if not existing:
+                if any(s["step_id"] == step_id for s in record["steps"] + record.get("checkpoints", []) + record.get("artifacts", [])):
+                    raise ValueError("Step ID already used")
+                if len(record["steps"]) >= self.config["max_calls"]:
+                    entry["fault"] = "Total Agent call limit"
+                    raise ValueError(entry["fault"])
+                row = {
+                    "step_id": step_id,
+                    "attempt": 1,
+                    "agent_ref": payload["agentRef"],
+                    "thread_id": thread_id,
+                    "run_id": None,
+                    "status": "queued",
+                    "input": input,
+                    "input_artifacts": references,
+                    "idempotency_key": f"{record['id']}:{message_id}:1",
+                }
+                record["steps"].append(row)
+                await self.save(owner, record)
             async with entry["slots"], self.global_slots:
                 if entry["cancel"].is_set() or entry["fault"]:
                     row["status"] = "cancelled"
@@ -365,32 +463,35 @@ class Workflows(Lifecycle, Artifacts):
                         entry["fault"] = "Agent configuration changed or unavailable"
                         row["status"] = "failed"
                         raise
-                row["status"] = "creating_thread"
-                await self.save(owner, record)
-                # Never cancel these awaits just because a workflow cancel was requested.
-                # Keep tracking the response so an admitted child cannot disappear.
-                async with asyncio.timeout(self.config["admission_timeout_seconds"]):
-                    await runs.create_thread(
-                        assistant_id=payload["agentRef"],
-                        thread_id=thread_id,
-                        metadata={"title": f"Workflow {record['id'][:8]} / {step_id}"},
-                    )
-                    if entry["cancel"].is_set():
-                        row["status"] = "cancelled"
-                        raise ValueError("Workflow stopping")
-                    row["status"] = "admitting"
+                if row["status"] == "queued":
+                    row["status"] = "creating_thread"
                     await self.save(owner, record)
-                    run = await runs.start(
-                        thread_id=thread_id,
-                        input=input,
-                        idempotency_key=row["idempotency_key"],
+                    # Never cancel these awaits just because a workflow cancel was requested.
+                    # Keep tracking the response so an admitted child cannot disappear.
+                    async with asyncio.timeout(self.config["admission_timeout_seconds"]):
+                        await runs.create_thread(
+                            assistant_id=payload["agentRef"],
+                            thread_id=thread_id,
+                            metadata={"title": f"Workflow {record['id'][:8]} / {step_id}"},
+                        )
+                        if entry["cancel"].is_set():
+                            row["status"] = "cancelled"
+                            raise ValueError("Workflow stopping")
+                        row["status"] = "admitting"
+                        await self.save(owner, record)
+                        run = await runs.start(
+                            thread_id=thread_id,
+                            input=input,
+                            idempotency_key=row["idempotency_key"],
+                        )
+                    row.update(
+                        run_id=run.run_id,
+                        status="running",
+                        effective_assistant_id=run.assistant_id,
                     )
-                row.update(
-                    run_id=run.run_id,
-                    status="running",
-                    effective_assistant_id=run.assistant_id,
-                )
-                await self.save(owner, record)
+                    await self.save(owner, record)
+                else:
+                    run = await runs.get(thread_id=thread_id, run_id=row["run_id"])
                 while True:
                     if entry["cancel"].is_set() or entry["fault"]:
                         await self.cancel_child(runs, row)
@@ -453,13 +554,7 @@ class Workflows(Lifecycle, Artifacts):
                     output=copy.deepcopy(output),
                     artifact_version=identity([record["id"], step_id, row["attempt"], output]),
                 )
-                response["value"] = {
-                    "artifactVersion": row["artifact_version"],
-                    "output": output,
-                    "threadId": thread_id,
-                    "runId": run.run_id,
-                    "status": "success",
-                }
+                response["value"] = self.step_result(row)
         except TimeoutError:
             if row and row["run_id"] is not None:
                 # Admission is known. Final cleanup interrupts/observes the tracked
@@ -484,11 +579,25 @@ class Workflows(Lifecycle, Artifacts):
                 row["status"] = "failed"
             response["error"] = str(exc)[:240] if isinstance(exc, ValueError) else "Agent step failed"
         finally:
+            if replay and "error" in response:
+                entry["fault"] = "Recovered child outcome requires reconciliation"
+                if row:
+                    row["status"] = "needs_reconciliation"
             if row and "error" in response:
                 row["error"] = response["error"]
             await self.save(owner, record)
         return response
 
+    @staticmethod
+    def step_result(row):
+        return {
+            "artifactVersion": row["artifact_version"],
+            "output": copy.deepcopy(row["output"]),
+            "threadId": row["thread_id"],
+            "runId": row["run_id"],
+            "status": "success",
+        }
+
     async def cancel_child(self, runs, row):
         try:
             async with asyncio.timeout(5):
@@ -507,6 +616,9 @@ class Workflows(Lifecycle, Artifacts):
         process = None
         workspace = None
         try:
+            if record.get("budget_deadline") is not None:
+                record["remaining_seconds"] = max(0, record["budget_deadline"] - time.time())
+            record["budget_deadline"] = time.time() + record.get("remaining_seconds", self.config["timeout_seconds"])
             record["status"] = "running"
             await self.save(owner, record)
             workspace = await asyncio.to_thread(tempfile.TemporaryDirectory, prefix="deerflow-workflow-")
@@ -534,7 +646,8 @@ class Workflows(Lifecycle, Artifacts):
                     "memory_bytes": self.config["memory_bytes"],
                 },
             )
-            async with asyncio.timeout(self.config["timeout_seconds"]) as deadline:
+            async with asyncio.timeout(record.get("remaining_seconds", self.config["timeout_seconds"])) as deadline:
+                entry["deadline"] = deadline
                 turns = 0
                 while True:
                     if entry["cancel"].is_set() or entry["fault"]:
@@ -551,31 +664,47 @@ class Workflows(Lifecycle, Artifacts):
                     messages = frame["messages"]
                     if not isinstance(messages, list) or len(messages) > 64:
                         raise ValueError("Worker message limit")
+                    replayed = await self.begin_frame(entry, frame)
+                    if replayed is not None:
+                        await self.send(process, {"responses": replayed})
+                        continue
                     if any(m.get("type") in ("pause", "humanCheckpoint") for m in messages):
                         if len(messages) != 1 or entry["tasks"] or frame["outcome"] is not None:
                             raise ValueError("Checkpoint must be awaited alone, after all Agent calls")
                         response = await self.checkpoint(entry, messages[0], deadline)
-                        await self.send(process, {"responses": [response]})
+                        await self.deliver_frame(entry, process, [response])
                         continue
                     if any(m.get("type") in ("captureArtifact", "readArtifact") for m in messages):
                         if len(messages) != 1 or entry["tasks"] or frame["outcome"] is not None:
                             raise ValueError("Artifact operations must be awaited alone")
                         operation = self.capture if messages[0]["type"] == "captureArtifact" else self.consume
                         response = await operation(entry, messages[0])
-                        await self.send(process, {"responses": [response]})
+                        await self.deliver_frame(entry, process, [response])
                         continue
-                    for message in messages:
+                    for message_index, message in enumerate(messages):
                         if message.get("type") == "log":
+                            log_id = f"{entry['frame_index']}:{message_index}"
+                            if log_id in record.get("log_ids", []):
+                                continue
                             if len(record["logs"]) >= 20:
                                 raise ValueError("Log limit")
                             log = text(message.get("text"), 160)
                             encoded(log, 160)
                             record["logs"].append(log)
+                            record.setdefault("log_ids", []).append(log_id)
                             await self.save(owner, record)
                         elif message.get("type") == "agent":
                             if type(message.get("id")) is not int or len(entry["tasks"]) >= self.config["max_calls"]:
                                 raise ValueError("Pending Agent call limit")
-                            entry["tasks"].add(asyncio.create_task(self.step(entry, message)))
+                            entry["tasks"].add(
+                                asyncio.create_task(
+                                    self.step(
+                                        entry,
+                                        message,
+                                        replay=entry.get("recover_frame", False),
+                                    )
+                                )
+                            )
                         else:
                             raise ValueError("Unsupported worker message")
                     if frame["outcome"] is not None:
@@ -606,7 +735,7 @@ class Workflows(Lifecycle, Artifacts):
                     finished = [t for t in entry["tasks"] if t.done()]
                     responses = [t.result() for t in finished]
                     entry["tasks"].difference_update(finished)
-                    await self.send(process, {"responses": responses})
+                    await self.deliver_frame(entry, process, responses)
         except TimeoutError:
             # Keep this outside asyncio.timeout: its own cancellation must first
             # leave the context manager so it can be translated to TimeoutError.
@@ -645,6 +774,16 @@ class Workflows(Lifecycle, Artifacts):
                     status="needs_reconciliation",
                     error="Some child outcomes are uncertain; inspect their threads/runs before new work",
                 )
+            if entry.get("suspending") and record["status"] not in (
+                "completed",
+                "needs_reconciliation",
+            ):
+                record.update(status="recoverable", error="Gateway stopped; sign in to recover")
+                if any(s["status"] not in ("queued", "completed", "running") for s in record["steps"]):
+                    record.update(
+                        status="needs_reconciliation",
+                        error="Gateway stopped with unfinished child effects; inspect child runs",
+                    )
             await self.save(owner, record)
             self.entries.pop(record["id"], None)
 

examples/deerflow-extension-workflows/deerflow_extension_workflows/static/index.mjs

@@ -48,7 +48,7 @@ function mount(root, context) {
   const form = node('form');
   const submit = node('button',t('Launch workflow','启动工作流'),{type:'submit'});
   form.append(node('label',t('Inspectable JavaScript script','可检查的 JavaScript 脚本')),script,node('label',t('Arguments (JSON object)','参数（JSON 对象）')),args,submit);
-  panel.append(node('h2',t('Script workflows','脚本工作流')),node('p',t('Versioned workflows with live-process resume and authenticated output review. Restart ends resumability; no hard monetary budget. File approval covers a verified immutable snapshot; linked files require explicit capture.','版本化工作流支持同进程恢复和登录用户成果确认。重启后不能恢复；无金额硬上限。文件确认针对经过哈希验证的不可覆盖快照；链接文件需显式捕获。')),notice,library,form,navigation,detail);
+  panel.append(node('h2',t('Script workflows','脚本工作流')),node('p',t('Versioned workflows with authenticated output review. After restart, sign in and recover compatible runs. Single Gateway only; uncertain child effects require reconciliation; no hard monetary budget. File approval covers a verified immutable snapshot; linked files require explicit capture.','版本化工作流支持登录用户成果确认。重启后请登录并恢复兼容的工作流。仅支持单 Gateway；不确定的子任务效果需要核对；无金额硬上限。文件确认针对经过哈希验证的不可覆盖快照；链接文件需显式捕获。')),notice,library,form,navigation,detail);
   root.append(sheet,panel);
   const alive = version => !disposed && !context.signal?.aborted && version === generation;
   async function call(action,payload) {
@@ -75,7 +75,8 @@ function mount(root, context) {
     if(!alive(version)||id!==selected)return;
     releaseFiles();
     detail.replaceChildren(node('h3',`${run.run_id} · ${run.status}`));
-    if(['paused','waiting_approval'].includes(run.status))detail.append(button(t('Refresh checkpoint','刷新检查点'),()=>show()),node('p',t('Gateway restart ends this live checkpoint.','Gateway 重启后此检查点不能继续。')));
+    if(['paused','waiting_approval'].includes(run.status))detail.append(button(t('Refresh checkpoint','刷新检查点'),()=>show()),node('p',t('After restart, recover this workflow before continuing its checkpoint.','重启后请先恢复工作流，再继续此检查点。')));
+    if(run.status==='recoverable')detail.append(button(t('Recover workflow','恢复工作流'),async()=>{await call('recover',{run_id:id});await show();}),button(t('Cancel workflow','取消工作流'),async()=>{await call('cancel',{run_id:id});await show();}));
     if(run.error)detail.append(node('p',run.error));
     detail.append(node('h4',t('Script','脚本')),node('pre',run.script),node('h4',t('Arguments and execution limits','参数和执行上限')),node('pre',JSON.stringify({definition:run.definition,configuration_revision:run.configuration_revision,args:run.args,limits:run.limits,script_hash:run.script_hash,args_hash:run.args_hash},null,2)));
     for(const step of run.steps){

examples/deerflow-extension-workflows/deerflow_extension_workflows/store.py

@@ -1,4 +1,4 @@
-"""Owner-scoped, retained records; no capability serialization or automatic replay."""
+"""Owner-scoped versioned records and immutable snapshots; never store capabilities."""
 
 import hashlib
 import json
@@ -13,6 +13,7 @@ TERMINAL = {
     "waiting_input",
     "interrupted",
     "needs_reconciliation",
+    "recoverable",
 }
 
 
@@ -24,6 +25,9 @@ class Store:
         self.path.parent.mkdir(parents=True, exist_ok=True)
         with closing(self.connect()) as db, db:
             db.execute("CREATE TABLE IF NOT EXISTS workflows (id TEXT PRIMARY KEY, owner TEXT NOT NULL, request_id TEXT NOT NULL, fingerprint TEXT NOT NULL, status TEXT NOT NULL, data TEXT NOT NULL, UNIQUE(owner,request_id))")
+            columns = {row[1] for row in db.execute("PRAGMA table_info(workflows)")}
+            if "revision" not in columns:
+                db.execute("ALTER TABLE workflows ADD COLUMN revision INTEGER NOT NULL DEFAULT 0")
             db.execute("CREATE INDEX IF NOT EXISTS workflows_owner ON workflows(owner)")
             db.execute(
                 "CREATE TABLE IF NOT EXISTS workflow_artifacts (owner TEXT NOT NULL, workflow TEXT NOT NULL, step_id TEXT NOT NULL, version TEXT NOT NULL, "
@@ -37,24 +41,64 @@ class Store:
             raise ValueError("Artifact limit is 2 MiB")
         digest = hashlib.sha256(content).hexdigest()
         version = hashlib.sha256(json.dumps([owner, workflow, step_id, digest]).encode()).hexdigest()
-        info = {"kind": "file", "step_id": step_id, "attempt": 1, "name": name, "size": len(content), "sha256": digest, "version": version, "artifact_version": version, "status": "completed"}
+        info = {
+            "kind": "file",
+            "step_id": step_id,
+            "attempt": 1,
+            "name": name,
+            "size": len(content),
+            "sha256": digest,
+            "version": version,
+            "artifact_version": version,
+            "status": "completed",
+        }
         with closing(self.connect()) as db, db:
             db.execute("BEGIN IMMEDIATE")
-            old = db.execute("SELECT info FROM workflow_artifacts WHERE owner=? AND workflow=? AND step_id=?", (owner, workflow, step_id)).fetchone()
+            old = db.execute(
+                "SELECT info FROM workflow_artifacts WHERE owner=? AND workflow=? AND step_id=?",
+                (owner, workflow, step_id),
+            ).fetchone()
             if old:
                 if json.loads(old[0]) != info:
                     raise ValueError("Artifact capture step cannot be overwritten")
                 return info
-            if db.execute("SELECT count(*) FROM workflow_artifacts WHERE owner=? AND workflow=?", (owner, workflow)).fetchone()[0] >= 16:
+            if (
+                db.execute(
+                    "SELECT count(*) FROM workflow_artifacts WHERE owner=? AND workflow=?",
+                    (owner, workflow),
+                ).fetchone()[0]
+                >= 16
+            ):
                 raise ValueError("Artifact count limit (16)")
-            if db.execute("SELECT coalesce(sum(length(content)),0) FROM workflow_artifacts WHERE owner=?", (owner,)).fetchone()[0] + len(content) > 64 * 1024 * 1024:
+            if (
+                db.execute(
+                    "SELECT coalesce(sum(length(content)),0) FROM workflow_artifacts WHERE owner=?",
+                    (owner,),
+                ).fetchone()[0]
+                + len(content)
+                > 64 * 1024 * 1024
+            ):
                 raise ValueError("Owner artifact limit (64 MiB)")
-            db.execute("INSERT INTO workflow_artifacts VALUES (?,?,?,?,?,?)", (owner, workflow, step_id, version, json.dumps(info), content))
+            db.execute(
+                "INSERT INTO workflow_artifacts VALUES (?,?,?,?,?,?)",
+                (owner, workflow, step_id, version, json.dumps(info), content),
+            )
         return info
 
+    def artifact_by_step(self, owner, workflow, step_id):
+        with closing(self.connect()) as db:
+            row = db.execute(
+                "SELECT version FROM workflow_artifacts WHERE owner=? AND workflow=? AND step_id=?",
+                (owner, workflow, step_id),
+            ).fetchone()
+        return self.artifact(owner, workflow, row[0])[0] if row else None
+
     def artifact(self, owner, workflow, version):
         with closing(self.connect()) as db:
-            row = db.execute("SELECT info,content FROM workflow_artifacts WHERE owner=? AND workflow=? AND version=?", (owner, workflow, version)).fetchone()
+            row = db.execute(
+                "SELECT info,content FROM workflow_artifacts WHERE owner=? AND workflow=? AND version=?",
+                (owner, workflow, version),
+            ).fetchone()
             if not row:
                 raise ValueError("Artifact unavailable")
             info, content = json.loads(row[0]), row[1]
@@ -121,7 +165,7 @@ class Store:
             if db.execute("SELECT count(*) FROM workflows WHERE owner=?", (owner,)).fetchone()[0] >= 100:
                 raise ValueError("Retained workflow limit (100); delete finished records")
             db.execute(
-                "INSERT INTO workflows VALUES (?,?,?,?,?,?)",
+                "INSERT INTO workflows (id,owner,request_id,fingerprint,status,data) VALUES (?,?,?,?,?,?)",
                 (
                     record["id"],
                     owner,
@@ -134,11 +178,20 @@ class Store:
             return record, True
 
     def put(self, owner, record):
+        revision = record.get("revision", 0)
+        updated = {**record, "revision": revision + 1}
+        raw = json.dumps(updated)
+        if len(raw.encode()) > 8 * 1024 * 1024:
+            raise ValueError("Workflow recovery journal limit (8 MiB)")
         with closing(self.connect()) as db, db:
-            db.execute(
-                "UPDATE workflows SET status=?,data=? WHERE owner=? AND id=?",
-                (record["status"], json.dumps(record), owner, record["id"]),
-            )
+            changed = db.execute(
+                "UPDATE workflows SET status=?,data=?,revision=revision+1 WHERE owner=? AND id=? AND revision=?",
+                (record["status"], raw, owner, record["id"], revision),
+            ).rowcount
+            if changed != 1:
+                raise ValueError("Stale workflow revision")
+        record["revision"] = revision + 1
+        return revision + 1
 
     def get(self, owner, run_id):
         with closing(self.connect()) as db:
@@ -170,18 +223,23 @@ class Store:
             row = db.execute("SELECT status FROM workflows WHERE owner=? AND id=?", (owner, run_id)).fetchone()
             if not row or row[0] not in TERMINAL:
                 raise ValueError("Only finished workflow records can be deleted")
-            db.execute("DELETE FROM workflow_artifacts WHERE owner=? AND workflow=?", (owner, run_id))
+            db.execute(
+                "DELETE FROM workflow_artifacts WHERE owner=? AND workflow=?",
+                (owner, run_id),
+            )
             db.execute("DELETE FROM workflows WHERE owner=? AND id=?", (owner, run_id))
 
     def interrupt_unfinished(self):
         with closing(self.connect()) as db, db:
             for run_id, raw in db.execute("SELECT id,data FROM workflows WHERE status IN ('queued','running','cancelling','paused','waiting_approval')").fetchall():
                 record = json.loads(raw)
+                recoverable = record.get("recovery_version") == 1 and record["status"] != "cancelling"
                 record.update(
-                    status="interrupted",
-                    error="Gateway restarted; execution cannot resume. Inspect child runs before starting new work.",
+                    status="recoverable" if recoverable else "interrupted",
+                    error="Gateway restarted; sign in and recover, or inspect uncertain child outcomes." if recoverable else "Legacy or cancelled execution cannot resume; inspect child runs.",
+                    revision=record.get("revision", 0) + 1,
                 )
                 db.execute(
-                    "UPDATE workflows SET status=?,data=? WHERE id=?",
+                    "UPDATE workflows SET status=?,data=?,revision=revision+1 WHERE id=?",
                     (record["status"], json.dumps(record), run_id),
                 )

examples/deerflow-extension-workflows/deerflow_extension_workflows/worker.py

@@ -10,6 +10,10 @@ MAX_FRAME = 48 * 1024
 
 BOOT = r"""
 (() => {
+  // Replay must not observe wall time or host randomness. These globals are
+  // unavailable; callers pass timestamps and random seeds explicitly in args.
+  Object.defineProperty(globalThis,'Date',{value:undefined,writable:false,configurable:false});
+  Object.defineProperty(Math,'random',{value:()=>{throw new Error('Pass randomness in args');},writable:false,configurable:false});
   const encode=JSON.stringify.bind(JSON), decode=JSON.parse.bind(JSON);
   const requests=[], pending=new Map(); let serial=0, final=null;
   const clone=value=>decode(encode(value));

examples/deerflow-extension-workflows/pyproject.toml

@@ -1,10 +1,10 @@
 [project]
 name = "deerflow-extension-workflows"
-version = "0.2.0"
+version = "0.3.0"
 description = "Opt-in constrained script orchestration of public DeerFlow AgentRuns"
 requires-python = ">=3.12"
 license = "MIT"
-dependencies = ["deerflow-extension-api>=0.2.7,<0.3", "quickjs==1.19.4"]
+dependencies = ["deerflow-extension-api>=0.2.8,<0.3", "quickjs==1.19.4"]
 
 [project.entry-points."deerflow.extensions"]
 workflows = "deerflow_extension_workflows:install"

examples/deerflow-extension-workflows/verify_browser.mjs

@@ -28,6 +28,7 @@ const server=createServer(async(req,res)=>{
         if(action==='get_step'&&window.fileReview)return {kind:'file',name:'report.txt',size:bytes.length,sha256:window.badHash?'bad':hash,artifact_version:'artifact-v1'};
         if(action==='get_step')return {output:{report:'Retained <img src=x onerror=alert(1)> report'},artifact_version:'artifact-v1'};
         if(action==='decide'){window.state='completed';return {status:'approved'};}
+        if(action==='recover'){window.state='paused';return {status:'queued'};}
         if(action==='resume'){window.state='completed';return {status:'resumed'};}
         if(action==='launch'){saved=payload;identities.add(payload.request_id);if(window.loseLaunch-- > 0)throw Error('Launch response lost');return {run_id:'fixture-run',status:'queued'};}
         if(action==='list' && window.failList-- > 0)throw Error('List refresh failed');
@@ -48,6 +49,15 @@ try{
   await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
   await page.getByRole('heading',{name:'fixture-run · running',exact:true}).waitFor();
   const launch=await page.evaluate(()=>calls.find(c=>c.action==='launch'));
+  await page.evaluate(()=>window.state='recoverable');
+  // Refresh the selected run through the existing list action.
+  await page.locator('nav button').first().click();
+  await page.getByRole('button',{name:'Recover workflow',exact:true}).click();
+  assert.equal(await page.evaluate(()=>calls.filter(c=>c.action==='recover').length),1);
+  await page.getByRole('button',{name:'Resume this checkpoint',exact:true}).waitFor();
+  await page.evaluate(()=>window.state='running');
+  await page.locator('nav button').first().click();
+
   assert.equal(launch.payload.request_id.length,32);
   assert.match(launch.payload.script,/await agent/);
   await page.getByRole('button',{name:'Open child conversation',exact:true}).click();