feat(workflows): add Phase 2 approvals, immutable artifacts and configuration evidence (#6920)

* feat(workflows): add versioned approvals and run configuration evidence

* docs: trim repeated setup guidance to respect inherited size limit

* docs: leave room in inherited agent guidance

* test(channels): align Buzz capability expectation with upstream registry

* fix(workflows): fence file consumers and preserve review access

* fix(runtime): reuse assembly evidence key at run admission

AGENTS.md

@@ -1,6 +1,6 @@
 # AGENTS.md
 
-Repository guidance source of truth; `CLAUDE.md` imports `@AGENTS.md`.
+Guidance source; `CLAUDE.md` imports `@AGENTS.md`.
 Module guides:
 
 - **[backend/AGENTS.md](backend/AGENTS.md)** — backend depth: harness/app split, agent &
@@ -11,10 +11,8 @@ Module guides:
 
 ## What is DeerFlow
 
-DeerFlow pairs a LangGraph super agent with a Next.js chat UI. The backend provides
-sandboxed execution, persistent memory, subagents, and extensible built-in, MCP,
-and community tools, isolated per thread. Feishu, Slack, Telegram, Discord, and
-DingTalk connect through the Gateway.
+DeerFlow combines LangGraph and Next.js with thread-isolated sandbox execution,
+memory, subagents and extensible tools. IM channels connect through the Gateway.
 
 ## Service Topology
 
@@ -84,13 +82,12 @@ compaction, or authentication. If existing hooks cannot express the feature, add
 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}`.
-[Script workflows guide](examples/deerflow-extension-workflows/AGENTS.md): public
-AgentRuns; POSIX, one process/store; no resume/replay or hard currency budget.
+[Workflow guide](examples/deerflow-extension-workflows/AGENTS.md): public
+AgentRuns; POSIX, one process/store; live resume/approval; no restart replay or hard currency budget.
 
-Runtime config lives at the **repo root**: copy `config.example.yaml` → `config.yaml`
-(main app config) and `extensions_config.example.json` → `extensions_config.json` (MCP
-servers + skills). Both real files are gitignored and may be edited at runtime via the
-Gateway API. Config schema and resolution order are documented in
+Root runtime configs: copy `config.example.yaml` → `config.yaml` and
+`extensions_config.example.json` → `extensions_config.json` (MCP/skills).
+Both are gitignored and Gateway-editable; schema/resolution:
 [backend/AGENTS.md](backend/AGENTS.md).
 
 Skill quality review note:
@@ -164,8 +161,6 @@ cd frontend && pnpm check     # Lint + type check (run before committing)
 cd frontend && pnpm test      # Unit tests
 ```
 
-Root `make` drives the stack; backend `make` and frontend `pnpm` drive their modules.
-
 Host pnpm calls use `scripts/pnpm.py`: native Windows tries `pnpm.cmd` before
 `pnpm`; POSIX reverses the order. Its Corepack fallback applies the same ordering
 to `corepack.cmd` and `corepack`. The runner operates from `frontend/` so
@@ -173,16 +168,9 @@ Corepack honors its pinned package-manager version.
 
 ### Prerequisites before `make dev`
 
-First-time setup (`make dev` does not generate config):
-
-```bash
-make config      # copy config.example.yaml -> config.yaml and extensions_config.example.json -> extensions_config.json (both gitignored)
-make install     # install frontend + backend deps and pre-commit hooks
-make dev         # then start everything
-```
-
-Services require `config.yaml`. Both runtime configs are API-editable and gitignored;
-never commit them.
+Run `make config`, then `make install`, then `make dev` on first setup.
+`make dev` requires existing config; both runtime configs are API-editable and
+gitignored. Never commit them.
 
 ### Run a single test
 

README.md

@@ -3415,8 +3415,12 @@ The [script workflows extension](examples/deerflow-extension-workflows/README.md
 adds inspectable JavaScript orchestration through full AgentRuns, background launch,
 bounded parallel execution, progress, retained results and cancellation. It is
 disabled by default and requires a POSIX Gateway and operator-authorized Agent
-references. Phase 1 does not support resume, human output approval, restart replay
-or a hard monetary budget. Workflow deadlines and child-observation timeouts are
+references. Versioned definitions, retained step outputs, live-process pause/resume
+and authenticated human checkpoints support staged inspection reports. Confirmation
+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
 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

@@ -3,8 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
+import hmac
 import json
 import math
+import secrets
 from collections.abc import Awaitable, Callable, Mapping
 from contextlib import asynccontextmanager
 from typing import Any
@@ -45,6 +48,7 @@ class GatewayAgentRunsHost:
         self.app = app
         self.load_user = load_user
         self.active = True
+        self.revision_key = secrets.token_bytes(32)
 
     def close(self) -> None:
         self.active = False
@@ -121,6 +125,58 @@ class _BoundAgentRuns:
         finally:
             reset_current_user(token)
 
+    async def configuration_evidence(self, *, thread_id, run_id):
+        import copy
+
+        from app.gateway.deps import get_run_manager
+        from deerflow.runtime.assembly_evidence import EVIDENCE_KEY
+
+        async with self._request("runs:read", thread_id) as request:
+            record = await get_run_manager(request).get(run_id, user_id=self._user_id)
+            if record is None or record.thread_id != thread_id:
+                raise AgentRunError(404, "Run unavailable")
+            evidence = (record.metadata or {}).get(EVIDENCE_KEY)
+            if not isinstance(evidence, dict) or evidence.get("run_id") != run_id or evidence.get("thread_id") != thread_id:
+                raise AgentRunError(409, "Host assembly evidence unavailable")
+            return copy.deepcopy(evidence)
+
+    async def read_artifact(self, *, thread_id, path):
+        from app.gateway.extension_artifacts import read_output_snapshot
+        from app.gateway.routers.artifacts import reserve_artifact_write
+
+        async with self._request("threads:read", thread_id) as request:
+            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:
+        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
+
+        if not isinstance(assistant_ids, list) or len(assistant_ids) > 32 or any(not isinstance(name, str) or not name or len(name) > 80 for name in assistant_ids):
+            raise AgentRunError(422, "Invalid Agent references")
+        names = sorted(set(assistant_ids))
+        async with self._request("runs:create"):
+
+            def snapshot():
+                agents = {}
+                for name in names:
+                    if name == "lead_agent":
+                        continue
+                    try:
+                        config = load_agent_config(name, user_id=self._user_id)
+                    except FileNotFoundError:
+                        config = None
+                    if config is None:
+                        agents[name] = None
+                        continue
+                    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()
+
+            return await asyncio.to_thread(snapshot)
+
     async def create_thread(self, *, assistant_id="lead_agent", thread_id=None, metadata=None) -> str:
         from app.gateway.routers.threads import ThreadCreateRequest, create_thread
 

backend/app/gateway/extension_artifacts.py

@@ -0,0 +1,41 @@
+"""Bounded output capture without symlink traversal or mutable-path rereads."""
+
+import os
+import stat
+
+from deerflow_extension_api import AgentRunError
+
+from app.gateway.path_utils import normalize_outputs_virtual_path
+from deerflow.config.paths import get_paths
+
+MAX_BYTES = 2 * 1024 * 1024
+
+
+def read_output_snapshot(user_id, thread_id, path):
+    if os.name != "posix" or not isinstance(path, str):
+        raise AgentRunError(422, "Output capture requires POSIX and an output path")
+    path = normalize_outputs_virtual_path(path)
+    paths = get_paths()
+    root = paths.base_dir.resolve()
+    relative = paths.sandbox_outputs_dir(thread_id, user_id=user_id).relative_to(paths.base_dir)
+    parts = (*relative.parts, *path.removeprefix("/mnt/user-data/outputs/").split("/"))
+    directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        for part in parts[:-1]:
+            next_dir = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
+            os.close(directory)
+            directory = next_dir
+        fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
+        with os.fdopen(fd, "rb") as source:
+            before = os.fstat(source.fileno())
+            if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_BYTES:
+                raise AgentRunError(422, "Artifact must be a regular file of at most 2 MiB")
+            content = source.read(MAX_BYTES + 1)
+            after = os.fstat(source.fileno())
+            if len(content) > MAX_BYTES or (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns):
+                raise AgentRunError(409, "Artifact changed during capture")
+            return content
+    except OSError as exc:
+        raise AgentRunError(422, "Artifact unavailable or unsafe") from exc
+    finally:
+        os.close(directory)

backend/app/gateway/routers/AGENTS.md

@@ -38,3 +38,7 @@ critical section to preserve history ordering. Keep the sidecar distinct from
 the nested projection lock, retain it across requests, and acquire/release both
 off-loop inside the drained worker. Unrelated users retain independent locks;
 external filesystem writers do not participate in this history guarantee.
+
+Plugin actions stamp `ActionContext.interaction=session` only from authenticated
+session provenance. Never accept it from payloads or propagate it to model tools;
+extensions use it to distinguish owner review from model/tool output.

backend/app/gateway/routers/plugins.py

@@ -138,7 +138,15 @@ async def invoke_plugin_action(request: Request, namespace: str, action_name: st
     try:
         async with asyncio.timeout(30):
             runs = resolve_agent_runs(request)
-            return await action.handler(MappingProxyType(payload), ActionContext(principal, MappingProxyType(settings), agent_runs=runs.for_plugin(namespace) if runs is not None else None))
+            return await action.handler(
+                MappingProxyType(payload),
+                ActionContext(
+                    principal,
+                    MappingProxyType(settings),
+                    agent_runs=runs.for_plugin(namespace) if runs is not None else None,
+                    interaction="session" if getattr(request.state, "auth_source", None) == "session" and not principal.is_internal else "other",
+                ),
+            )
     except AgentRunError as exc:
         raise HTTPException(exc.status_code, str(exc)) from exc
     except TimeoutError as exc:

backend/docs/extension-agent-runs.md

@@ -38,6 +38,21 @@ The Gateway worker carries the capability only in its reserved runtime context;
 untrusted request input cannot replace it. Direct embedded or subagent execution
 does not acquire it automatically.
 
+## Configuration invalidation (API 0.2.7)
+
+`configuration_revision(assistant_ids=[...])` returns an opaque process-local revision
+for up to 32 Agent references, after rechecking the caller's `runs:create` grant.
+It covers declarative app/extensions config and owner-scoped Custom Agent config/SOUL;
+Agent order does not affect the revision. Missing definitions are represented as
+missing and their later creation changes the revision. The host uses a per-process
+HMAC key, so secret configuration is never returned or exposed as an unsalted digest.
+
+This is a conservative invalidation guard for retained results. It does not pin the
+subsequent admission atomically or attest effective run assembly, mutable memory,
+referenced Skill/tool files or remote provider state. Unrelated declarative changes
+can invalidate it too. Compare only within one Gateway lifetime; do not serialize
+capabilities or interpret the revision as cross-process recovery authority.
+
 ## Operations
 
 - `create_thread(assistant_id="lead_agent", thread_id=None, metadata=None)` creates
@@ -106,3 +121,20 @@ IDs and workflow progress, then reacquire a capability from a newly authenticate
 request to continue. Existing run/checkpoint durability and orphan recovery remain
 the host's responsibility. The original route-permission ceiling cannot widen; Agent/model/tool policy
 still evaluates the current user. Do not expose a retained handle to other users.
+
+### Immutable file review and run configuration evidence
+
+`AgentRuns.configuration_evidence(thread_id, run_id)` returns host-owned proof
+recorded from actual Agent assembly before graph execution and bound to the run.
+It includes actual model, safe numeric parameters and keyed versions of instructions,
+model configuration, tools, Skills and policies. No raw secrets are exposed.
+Missing evidence fails closed; the declarative revision remains a separate preflight
+guard. Evidence describes assembly-time configuration, not external service state.
+
+`AgentRuns.read_artifact(thread_id, path)` authorizes a bounded (2 MiB) regular
+output-file snapshot, rejects symlink traversal, and reserves the thread against
+concurrent runs/edits while reading off-loop. The workflow extension stores those
+bytes append-only with SHA-256, approves exact versions, and serves preview and
+approved consumer reads from the same snapshot (16 KiB chunks). File versions are
+owner/workflow scoped; consumers cannot use an unapproved version. See the workflow
+extension README for limits and `captureArtifact` / `readArtifact` examples.

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

@@ -44,6 +44,32 @@ class AgentRuns(Protocol):
         """
         ...
 
+    async def configuration_evidence(self, *, thread_id: str, run_id: str) -> dict[str, Any]:
+        """Read immutable host evidence captured from this run's actual assembly.
+
+        Fails closed if the factory/host produced no evidence. No config re-read,
+        credentials, raw instructions or mutable thread-state fallback.
+        """
+        ...
+
+    async def read_artifact(self, *, thread_id: str, path: str) -> bytes:
+        """Read a bounded regular output file through owner authorization.
+
+        Returns at most 2 MiB, rejecting symlinks and concurrent mutations.
+        The caller must retain these bytes to create an immutable version.
+        """
+        ...
+
+    async def configuration_revision(self, *, assistant_ids: list[str]) -> str:
+        """Opaque process-local 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.
+        """
+        ...
+
     async def create_thread(self, *, assistant_id: str = "lead_agent", thread_id: str | None = None, metadata: Mapping[str, Any] | None = None) -> str:
         """Create an owned thread, or return the caller's existing thread ID."""
         ...

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

@@ -21,6 +21,8 @@ class ActionContext:
     principal: ExtensionPrincipal
     settings: Mapping[str, SettingValue]
     agent_runs: AgentRuns | None = field(default=None, kw_only=True)
+    # Host-stamped browser session; model tools and other callers default closed.
+    interaction: Literal["session", "other"] = field(default="other", kw_only=True)
 
 
 @dataclass(frozen=True)

backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -937,7 +937,9 @@ def _complete_assembly(
     from deerflow.extensions import get_agent_build_extensions
 
     resolved_extensions = get_agent_build_extensions()
-    if not resolved_extensions.has_agent_assembly_observers:
+    from deerflow.runtime.assembly_evidence import capture_requested
+
+    if not resolved_extensions.has_agent_assembly_observers and not capture_requested():
         return LeadAgentAssembly(graph=graph, descriptor=None, effective_model=effective_model)
 
     from deerflow.agents.assembly_descriptor import build_assembly_descriptor

backend/packages/harness/deerflow/extensions/AGENTS.md

@@ -449,6 +449,13 @@ and revoked before host shutdown; PAT/internal grants remain unsupported.
 Unstamped internal launches receive no handle and must still start normally.
 Action/tool dispatch scopes handles to each registered plugin namespace for
 idempotency isolation; request-resolved handles use `for_plugin` explicitly.
+`configuration_revision` (API 0.2.7) rechecks grants and returns a process-local
+HMAC of declarative app/extensions and Agent config/SOUL, not effective assembly,
+memory, referenced files or provider state. Never expose raw secret configuration.
+`configuration_evidence` reads persisted, run-bound actual assembly versions; the
+worker captures descriptors even without observers and persists before execution.
+`read_artifact` authorizes bounded output bytes off-loop; extensions retain immutable
+versions themselves. Workflow preview/approval/consumption must share those bytes.
 
 **Recorded extension-first exception.** The project-shelf summary pipeline
 (`projects/summaries.py`) ships in core: its consumer-visible surface (the

backend/packages/harness/deerflow/runtime/assembly_evidence.py

@@ -0,0 +1,63 @@
+"""Host-owned, bounded evidence from the actual assembly, before graph execution."""
+
+import hashlib
+import hmac
+import json
+import secrets
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import asdict
+
+from deerflow_extension_api import AgentAssemblyDescriptor
+
+EVIDENCE_KEY = "deerflow_assembly_evidence"
+_capture = ContextVar("capture_run_assembly", default=False)
+_key = secrets.token_bytes(32)
+
+
+def capture_requested():
+    return _capture.get()
+
+
+@contextmanager
+def capture_assembly():
+    token = _capture.set(True)
+    try:
+        yield
+    finally:
+        _capture.reset(token)
+
+
+def _digest(value):
+    raw = json.dumps(value, sort_keys=True, ensure_ascii=True, allow_nan=False).encode()
+    return hmac.new(_key, raw, hashlib.sha256).hexdigest()
+
+
+def project_assembly_evidence(descriptor, *, thread_id, run_id):
+    if not isinstance(descriptor, AgentAssemblyDescriptor):
+        return None
+    # Do not expose arbitrary provider kwargs, middleware parameters, prompts,
+    # URLs, headers or credentials. Keyed versions cover them without raw data.
+    parameters = {
+        key: value
+        for key, value in descriptor.model_parameters.items()
+        if key in {"temperature", "top_p", "top_k", "max_tokens", "max_completion_tokens", "seed", "frequency_penalty", "presence_penalty"} and type(value) in (bool, int, float)
+    }
+    proof = {
+        "schema_version": 1,
+        "source": "host_assembly",
+        "thread_id": thread_id,
+        "run_id": run_id,
+        "agent": descriptor.agent_name,
+        "model": descriptor.effective_model,
+        "parameters": parameters,
+        "model_version": _digest([descriptor.model_parameters, descriptor.thinking_enabled, descriptor.reasoning_effort]),
+        "instructions_version": _digest(descriptor.base_prompt_hash),
+        "tools_version": _digest([asdict(t) for t in descriptor.tools] + [list(descriptor.deferred_tool_names)]),
+        "skills_version": _digest([list(descriptor.enabled_skills), descriptor.effective_policies.get("skill_catalog_hash")]),
+        "policy_version": _digest([asdict(m) for m in descriptor.middlewares] + [descriptor.effective_policies]),
+    }
+    proof["version"] = _digest({k: v for k, v in proof.items() if k not in ("thread_id", "run_id")})
+    if len(json.dumps(proof).encode()) > 8000:
+        raise ValueError("Assembly evidence limit")
+    return proof

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -676,6 +676,8 @@ class RunManager:
         raw ``IntegrityError`` instead of a ``ConflictError``. Production
         callers should use :meth:`create_or_reject`.
         """
+        from deerflow.runtime.assembly_evidence import EVIDENCE_KEY
+
         run_id = str(uuid.uuid4())
         now = _now_iso()
         user_id = _resolve_record_user_id(user_id)
@@ -687,7 +689,7 @@ class RunManager:
             status=RunStatus.pending,
             on_disconnect=on_disconnect,
             multitask_strategy=multitask_strategy,
-            metadata=metadata or {},
+            metadata={key: value for key, value in (metadata or {}).items() if key != EVIDENCE_KEY},
             kwargs=kwargs or {},
             user_id=user_id,
             created_at=now,
@@ -1282,6 +1284,23 @@ class RunManager:
         except Exception:
             logger.warning("Failed to persist model_name update for run %s", run_id, exc_info=True)
 
+    async def record_assembly_evidence(self, run_id: str, evidence: dict) -> None:
+        """Publish only host assembly evidence, once, before execution."""
+        from deerflow.runtime.assembly_evidence import EVIDENCE_KEY
+
+        async with self._lock:
+            record = self._runs.get(run_id)
+            if record is None or evidence.get("run_id") != run_id or evidence.get("thread_id") != record.thread_id:
+                raise ValueError("Assembly/run binding mismatch")
+            if EVIDENCE_KEY in record.metadata:
+                if record.metadata[EVIDENCE_KEY] != evidence:
+                    raise ValueError("Run assembly evidence is immutable")
+                return
+            record.metadata = {**record.metadata, EVIDENCE_KEY: copy.deepcopy(evidence)}
+            if not await self._persist_to_store(record):
+                record.metadata.pop(EVIDENCE_KEY, None)
+                raise RuntimeError("Unable to persist assembly evidence")
+
     async def update_model_name(self, run_id: str, model_name: str | None) -> None:
         """Update the model name for a run."""
         async with self._lock:
@@ -1685,6 +1704,8 @@ class RunManager:
         partial unique index on ``(thread_id) WHERE status IN
         ('pending','running')``.
         """
+        from deerflow.runtime.assembly_evidence import EVIDENCE_KEY
+
         run_id = str(uuid.uuid4())
         now = _now_iso()
         # Resolve before the idempotency checks below compare it with stored rows.
@@ -1706,7 +1727,7 @@ class RunManager:
             on_disconnect=on_disconnect,
             operation_kind=operation_kind,
             multitask_strategy=multitask_strategy,
-            metadata=metadata or {},
+            metadata={key: value for key, value in (metadata or {}).items() if key != EVIDENCE_KEY},
             kwargs=kwargs or {},
             user_id=user_id,
             created_at=now,

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -1270,14 +1270,21 @@ async def run_agent(
         if ctx.app_config is not None and _agent_factory_supports_app_config(agent_factory):
             agent_factory_kwargs["app_config"] = ctx.app_config
         from deerflow.extensions import bind_agent_build_extensions
+        from deerflow.runtime.assembly_evidence import capture_assembly, project_assembly_evidence
 
-        with bind_agent_build_extensions(extensions):
+        with bind_agent_build_extensions(extensions), capture_assembly():
             # Assemble off-loop: agent construction re-enters
             # get_available_tools(), which may block on MCP cache
             # initialization — it must not stall the calling event loop
             # (issue #5172).
             agent_result = await run_assembly(agent_factory, **agent_factory_kwargs)
             agent = _agent_graph(agent_result)
+        from deerflow.agents.lead_agent.agent import LeadAgentAssembly
+
+        if isinstance(agent_result, LeadAgentAssembly):
+            evidence = project_assembly_evidence(agent_result.descriptor, thread_id=thread_id, run_id=run_id)
+            if evidence is not None:
+                await run_manager.record_assembly_evidence(run_id, evidence)
 
         # Assembly resolves request, agent, and authorization fallbacks. Trace the
         # model that will run, rather than the model name originally requested.

backend/tests/blocking_io/test_script_workflow_io.py

@@ -19,3 +19,22 @@ async def test_workflow_lifetime_and_storage_do_not_block_loop(plugin):
     assert (await plugin[1]["list"]({}, context(plugin[3])))["runs"][0]["run_id"] == result["run_id"]
     await plugin[1]["delete"]({"run_id": result["run_id"]}, context(plugin[3]))
     assert not plugin[2].processes
+
+
+async def test_definition_and_live_checkpoint_storage_do_not_block_loop(plugin):
+    from test_workflow_plugin import waiting
+
+    actions, runs = plugin[1], plugin[3]
+    definition = await actions["save_definition"]({"name": "pause", "script": "await pause({stepId:'p'});return 42;"}, context(runs))
+    assert (await actions["definitions"]({}, context(runs)))["definitions"] == [definition]
+    launched = await actions["launch"]({"request_id": "saved", "definition": definition}, context(runs))
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id, "paused")
+    await actions["resume"]({"run_id": run_id, "step_id": "p", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
+    assert (await done(plugin, run_id))["result"] == 42
+
+
+async def test_file_capture_review_consumption_do_not_block_loop(plugin):
+    from test_workflow_plugin import test_immutable_file_review_and_downstream_consume_same_version
+
+    await test_immutable_file_review_and_downstream_consume_same_version(plugin)

backend/tests/test_buzz_channel.py

@@ -46,7 +46,7 @@ def _channel(**overrides) -> BuzzChannel:
 def test_registered_in_framework_registries():
     assert _CHANNEL_REGISTRY["buzz"] == "app.channels.buzz:BuzzChannel"
     assert _CHANNEL_CREDENTIAL_KEYS["buzz"] == ["private_key"]
-    assert CHANNEL_CAPABILITIES["buzz"] == {"supports_streaming": True, "proactive_notifications": False}
+    assert CHANNEL_CAPABILITIES["buzz"] == {"supports_streaming": True, "proactive_notifications": False, "proactive_group_notifications": False}
     policy = CHANNEL_RUN_POLICY["buzz"]
     assert policy.serialize_thread_runs is True and policy.requires_bound_identity is False
 

backend/tests/test_extension_agent_runs.py

@@ -155,7 +155,26 @@ async def runtime(host, monkeypatch, tmp_path):
         builder.add_node("answer", answer)
         builder.set_entry_point("answer")
         builder.set_finish_point("answer")
-        return builder.compile()
+        from deerflow_extension_api import AgentAssemblyDescriptor
+
+        from deerflow.agents.lead_agent.agent import LeadAgentAssembly
+
+        descriptor = AgentAssemblyDescriptor(
+            namespace="fixture",
+            agent_name="lead-agent",
+            requested_model=None,
+            effective_model="fixture-model",
+            model_parameters={"temperature": 0},
+            thinking_enabled=False,
+            reasoning_effort=None,
+            base_prompt_hash="fixture-prompt",
+            tools=(),
+            middlewares=(),
+            deferred_tool_names=(),
+            enabled_skills=(),
+            effective_policies={"skill_catalog_hash": "fixture-skills"},
+        )
+        return LeadAgentAssembly(graph=builder.compile(), descriptor=descriptor, effective_model="fixture-model")
 
     monkeypatch.setattr("app.gateway.services.resolve_agent_factory", lambda _: factory)
     try:
@@ -364,3 +383,65 @@ def test_action_and_tool_context_keep_positional_compatibility():
     principal = ExtensionPrincipal("owner")
     assert ActionContext(principal, {}).agent_runs is None
     assert ToolContext(principal, {}, "thread").agent_runs is None
+
+
+@pytest.mark.asyncio
+async def test_configuration_revision_is_private_scoped_and_detects_changes(host, monkeypatch):
+    from types import SimpleNamespace
+    from unittest.mock import Mock
+
+    control, request, load_user = host
+    app_config = {"models": [{"api_key": "secret", "name": "a"}]}
+    agent_config = {"name": "reviewer", "model": "a"}
+    soul = Mock(return_value="review carefully")
+    monkeypatch.setattr("deerflow.config.app_config.get_app_config", lambda: SimpleNamespace(model_dump=lambda **kw: app_config))
+    monkeypatch.setattr("deerflow.config.extensions_config.get_extensions_config", lambda: SimpleNamespace(model_dump=lambda **kw: {}))
+    monkeypatch.setattr("deerflow.config.agents_config.load_agent_config", lambda *a, **kw: SimpleNamespace(model_dump=lambda **kw: agent_config))
+    monkeypatch.setattr("deerflow.config.agents_config.load_agent_soul", soul)
+    runs = control.bind(request)
+    before = await runs.configuration_revision(assistant_ids=["reviewer", "lead_agent"])
+    assert len(before) == 64 and "secret" not in before
+    assert before == await runs.configuration_revision(assistant_ids=["lead_agent", "reviewer"])
+    soul.return_value = "changed instructions"
+    assert before != await runs.configuration_revision(assistant_ids=["reviewer", "lead_agent"])
+    soul.return_value = "review carefully"
+    agent_config["model"] = "b"
+    assert before != await runs.configuration_revision(assistant_ids=["reviewer", "lead_agent"])
+    load_user.return_value = None
+    with pytest.raises(AgentRunError, match="revoked"):
+        await runs.configuration_revision(assistant_ids=["lead_agent"])
+
+
+@pytest.mark.asyncio
+async def test_actual_assembly_evidence_is_retained_and_bound(runtime):
+    runs, state, request, _, _ = runtime
+    thread = await runs.create_thread(thread_id="proof-thread")
+    run = await runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "proof"}]})
+    assert (await runs.wait(thread_id=thread, run_id=run.run_id, timeout=5)).status == "success"
+    evidence = await runs.configuration_evidence(thread_id=thread, run_id=run.run_id)
+    assert evidence["model"] == "fixture-model"
+    assert evidence["run_id"] == run.run_id
+    evidence["model"] = "tampered"
+    retained = await runs.configuration_evidence(thread_id=thread, run_id=run.run_id)
+    assert retained["model"] == "fixture-model"
+    with pytest.raises(ValueError, match="immutable"):
+        await state.run_manager.record_assembly_evidence(run.run_id, evidence)
+    other = await runs.create_thread(thread_id="other-proof-thread")
+    with pytest.raises(AgentRunError):
+        await runs.configuration_evidence(thread_id=other, run_id=run.run_id)
+
+
+@pytest.mark.asyncio
+async def test_evidence_persistence_failure_prevents_execution(runtime, monkeypatch):
+    runs, state, _, _, handles = runtime
+
+    async def fail(*args, **kwargs):
+        raise RuntimeError("evidence persistence unavailable")
+
+    monkeypatch.setattr(state.run_manager, "record_assembly_evidence", fail)
+    thread = await runs.create_thread(thread_id="proof-failure")
+    run = await runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "must not run"}]})
+    assert (await runs.wait(thread_id=thread, run_id=run.run_id, timeout=5)).status == "error"
+    assert not handles
+    with pytest.raises(AgentRunError, match="unavailable"):
+        await runs.configuration_evidence(thread_id=thread, run_id=run.run_id)

backend/tests/test_plugin_contributions.py

@@ -39,7 +39,7 @@ def plugin_client():
     @app.middleware("http")
     async def identity(request, call_next):
         request.state.user = SimpleNamespace(id="user-1", system_role=role["value"])
-        request.state.auth_source = "session"
+        request.state.auth_source = role.get("source", "session")
         return await call_next(request)
 
     setattr(app.state, EXTENSION_PRINCIPAL_RESOLVER_KEY, lambda request: ExtensionPrincipal("user-1", is_admin=role["value"] == "admin"))
@@ -70,6 +70,7 @@ def test_one_card_and_deployment_owned_switch(plugin_client):
     role["value"] = "member"
     assert http.post(url + "/actions/check", json={"text": "hello"}).json() == {"length": 5, "limit": 20}
     assert calls[0][1].principal.user_id == "user-1"
+    assert calls[0][1].interaction == "session"
     assert http.post(url + "/actions/check", json={"text": "old-account"}, headers={"x-deerflow-plugin-viewer": "other-account"}).status_code == 409
     with pytest.raises(TypeError):
         calls[0][1].settings["enabled"] = False
@@ -150,3 +151,16 @@ def test_conflicting_manifest_cannot_half_register(plugin_client):
             with pytest.raises(ValueError):
                 registry.plugin(invalid)
             assert registry.mark() == mark
+
+
+@pytest.mark.parametrize("source", ["internal", "auth_disabled", "pat", None])
+def test_action_interaction_cannot_be_forged_in_payload(plugin_client, source):
+    http, role, calls, plugin = plugin_client
+    registry = ExtensionRegistry()
+    with registry.attributed_to("test:install"):
+        registry.plugin(replace(plugin, enabled=True))
+    http.app.state.extensions = registry.build()
+    role["source"] = source
+    response = http.post("/api/plugins/community.check/actions/check", json={"text": "x", "interaction": "session"})
+    assert response.status_code == 200
+    assert calls[-1][1].interaction == "other"

backend/tests/test_worker_langfuse_metadata.py

@@ -40,6 +40,10 @@ class _FakeAgent:
 
 
 class _FakeRunManager:
+    async def record_assembly_evidence(self, run_id: str, evidence: dict) -> None:
+        assert evidence["run_id"] == run_id
+        self.assembly_evidence = evidence
+
     async def try_start(self, _run_id: str) -> RunStartOutcome:
         return RunStartOutcome.started
 

backend/tests/test_workflow_boundaries.py

@@ -0,0 +1,94 @@
+"""Immutable file versions and run-bound configuration evidence."""
+
+import pytest
+
+
+def test_evidence_projects_actual_assembly_without_credentials():
+    from deerflow_extension_api import AgentAssemblyDescriptor
+
+    from deerflow.runtime.assembly_evidence import project_assembly_evidence
+
+    descriptor = AgentAssemblyDescriptor(
+        namespace="deerflow",
+        agent_name="reviewer",
+        requested_model="old",
+        effective_model="actual",
+        model_parameters={"temperature": 0.2, "api_key": "secret", "extra": {"Authorization": "Bearer nested-secret"}},
+        thinking_enabled=False,
+        reasoning_effort=None,
+        base_prompt_hash="prompt-sha",
+        tools=(),
+        middlewares=(),
+        deferred_tool_names=(),
+        enabled_skills=("review",),
+        effective_policies={"skill_catalog_hash": "skill-sha"},
+    )
+    proof = project_assembly_evidence(descriptor, thread_id="thread", run_id="run")
+    assert proof["run_id"] == "run" and proof["model"] == "actual"
+    assert proof["parameters"]["temperature"] == 0.2
+    assert "secret" not in str(proof)
+    assert proof == project_assembly_evidence(descriptor, thread_id="thread", run_id="run")
+    from dataclasses import replace
+
+    changed = project_assembly_evidence(replace(descriptor, base_prompt_hash="new-prompt"), thread_id="thread", run_id="run")
+    assert proof["version"] != changed["version"]
+
+
+def test_artifact_versions_are_append_only_and_owner_scoped(tmp_path, monkeypatch):
+    from pathlib import Path
+
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-workflows"))
+    from deerflow_extension_workflows.store import Store
+
+    store = Store(tmp_path / "workflows.sqlite")
+    one = store.capture_artifact("alice", "workflow", "step", "report.pdf", b"original")
+    two = store.capture_artifact("alice", "workflow", "step2", "report.pdf", b"edited")
+    assert one["version"] != two["version"]
+    assert store.artifact("alice", "workflow", one["version"])[1] == b"original"
+    with pytest.raises(ValueError):
+        store.artifact("bob", "workflow", one["version"])
+    with pytest.raises(ValueError):
+        store.capture_artifact("alice", "workflow", "step", "report.pdf", b"overwrite")
+
+
+def test_host_output_capture_rejects_unsafe_files(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+
+    from deerflow_extension_api import AgentRunError
+    from fastapi import HTTPException
+
+    from app.gateway.extension_artifacts import MAX_BYTES, read_output_snapshot
+
+    output = tmp_path / "outputs"
+    output.mkdir()
+    monkeypatch.setattr("app.gateway.extension_artifacts.get_paths", lambda: SimpleNamespace(base_dir=tmp_path, sandbox_outputs_dir=lambda *a, **k: output))
+    report = output / "report.txt"
+    report.write_bytes(b"original")
+    captured = read_output_snapshot("alice", "thread", "/mnt/user-data/outputs/report.txt")
+    report.write_bytes(b"edited")
+    assert captured == b"original"
+    (output / "link").symlink_to(report)
+    for path in ("link", "../report.txt"):
+        with pytest.raises((AgentRunError, HTTPException)):
+            read_output_snapshot("alice", "thread", "/mnt/user-data/outputs/" + path)
+    report.write_bytes(b"x" * (MAX_BYTES + 1))
+    with pytest.raises(AgentRunError):
+        read_output_snapshot("alice", "thread", "/mnt/user-data/outputs/report.txt")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("admit", ["create", "create_or_reject"])
+@pytest.mark.parametrize("evidence_key", [None, "renamed_host_evidence"])
+async def test_client_cannot_supply_host_evidence(admit, evidence_key, monkeypatch):
+    from deerflow.runtime import RunManager, assembly_evidence
+
+    if evidence_key is not None:
+        monkeypatch.setattr(assembly_evidence, "EVIDENCE_KEY", evidence_key)
+    key = assembly_evidence.EVIDENCE_KEY
+    manager = RunManager()
+    try:
+        record = await getattr(manager, admit)("thread", metadata={key: {"source": "forged"}, "safe": "value"})
+        assert key not in record.metadata
+        assert record.metadata["safe"] == "value"
+    finally:
+        await manager.shutdown()

backend/tests/test_workflow_plugin.py

@@ -34,6 +34,21 @@ class Runs:
         self.jobs = []
         self.denied = False
 
+    revision = "initial"
+
+    async def configuration_evidence(self, *, thread_id, run_id):
+        return {"run_id": run_id, "thread_id": thread_id, "source": "host_assembly", "version": "synthetic-assembly", "model": "test-model"}
+
+    async def read_artifact(self, *, thread_id, path):
+        return self.file_content
+
+    file_content = b"original file"
+
+    async def configuration_revision(self, *, assistant_ids):
+        if self.denied:
+            raise AgentRunError(403, "revoked")
+        return self.revision
+
     async def create_thread(self, *, assistant_id, thread_id, metadata):
         if self.denied:
             raise AgentRunError(403, "revoked")
@@ -632,3 +647,369 @@ async def test_cancelled_launch_caller_keeps_committed_sqlite_launch_owned(plugi
         release.set()
         if stopping:
             await stopping
+
+
+async def waiting(plugin, run_id, status="waiting_approval"):
+    async with asyncio.timeout(5):
+        while True:
+            result = await plugin[1]["get"]({"run_id": run_id}, context(plugin[3]))
+            if result["status"] == status:
+                return result
+            await asyncio.sleep(0.01)
+
+
+@pytest.mark.asyncio
+async def test_phase2_definition_versions_are_immutable_and_owner_scoped(plugin):
+    _, actions, _, runs = plugin
+    first = await actions["save_definition"]({"name": "inspection", "script": "return args;"}, context(runs))
+    assert await actions["save_definition"]({"name": "inspection", "script": "return args;"}, context(runs)) == first
+    second = await actions["save_definition"]({"name": "inspection", "script": "return args.product;"}, context(runs))
+    assert first["version"] != second["version"]
+    launched = await actions["launch"]({"request_id": "versioned", "definition": first, "args": {"product": "A"}}, context(runs))
+    assert (await done(plugin, launched["run_id"]))["result"] == {"product": "A"}
+    with pytest.raises(ValueError):
+        await actions["launch"]({"request_id": "stolen", "definition": first}, context(runs, "bob"))
+
+
+@pytest.mark.asyncio
+async def test_phase2_checkpoint_requires_exact_authenticated_decision_and_dispatches_once(plugin):
+    from dataclasses import replace
+
+    from deerflow_extension_api.plugins import ToolContext
+
+    _, actions, _, runs = plugin
+    launched = await launch(
+        plugin,
+        """
+const evidence = await agent({stepId:'evidence',agentRef:'lead_agent',task:'evidence'});
+await humanCheckpoint({stepId:'confirm',sourceStepId:'evidence',artifactVersion:evidence.artifactVersion});
+return await agent({stepId:'report',agentRef:'lead_agent',task:evidence.output});
+""",
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    checkpoint = paused["checkpoints"][0]
+    assert len(runs.starts) == 1
+    decision = {"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"}
+    human = replace(context(runs), interaction="session")
+    for invalid in (context(runs), replace(human, principal=ExtensionPrincipal("bob")), ToolContext(human.principal, human.settings, "chat", agent_runs=runs, interaction="session")):
+        with pytest.raises(ValueError):
+            await actions["decide"](decision, invalid)
+    with pytest.raises(ValueError):
+        await actions["decide"]({**decision, "artifact_version": "stale"}, human)
+    await asyncio.gather(*(actions["decide"](decision, human) for _ in range(3)))
+    result = await done(plugin, run_id)
+    assert result["status"] == "completed"
+    assert len(runs.starts) == 2
+    assert result["checkpoints"][0]["decided_by"] == "alice"
+    with pytest.raises(ValueError):
+        await actions["decide"]({**decision, "decision": "reject"}, human)
+
+
+@pytest.mark.asyncio
+async def test_phase2_pause_resume_preserves_results_and_invalidates_configuration(plugin):
+    _, actions, _, runs = plugin
+    launched = await launch(
+        plugin,
+        """
+const a = await agent({stepId:'a',agentRef:'lead_agent',task:'original'});
+await pause({stepId:'break'});
+return a.output;
+""",
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id, "paused")
+    runs.threads[runs.starts[0][0]]["values"]["messages"].append({"type": "ai", "content": "changed later"})
+    with pytest.raises(ValueError):
+        await actions["resume"]({"run_id": run_id, "step_id": "break", "attempt": 1, "fingerprint": "changed"}, context(runs))
+    await actions["resume"]({"run_id": run_id, "step_id": "break", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
+    assert (await done(plugin, run_id))["result"] == {"echo": "original"}
+    assert len(runs.starts) == 1
+    launched = await launch(plugin, "await pause({stepId:'break'}); return 1;", request_id="changed-config")
+    paused = await waiting(plugin, launched["run_id"], "paused")
+    runs.revision = "modified"
+    with pytest.raises(ValueError, match="configuration"):
+        await actions["resume"]({"run_id": launched["run_id"], "step_id": "break", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
+    await actions["cancel"]({"run_id": launched["run_id"]}, context(runs))
+    assert (await done(plugin, launched["run_id"]))["status"] == "cancelled"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["reject", "cancel", "expiry"])
+async def test_phase2_checkpoint_stop_never_dispatches_downstream(plugin, kind):
+    from dataclasses import replace
+
+    _, actions, service, runs = plugin
+    if kind == "expiry":
+        service.config["pause_timeout_seconds"] = 1
+    launched = await launch(
+        plugin,
+        "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'data'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return await agent({stepId:'b',agentRef:'lead_agent',task:'never'});",
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    if kind == "reject":
+        await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["checkpoints"][0]["artifact_version"], "decision": "reject"}, replace(context(runs), interaction="session"))
+    elif kind == "cancel":
+        await actions["cancel"]({"run_id": run_id}, context(runs))
+    result = await done(plugin, run_id)
+    assert result["status"] == ("cancelled" if kind == "cancel" else "failed")
+    assert len(runs.starts) == 1
+    assert not service.processes
+
+
+@pytest.mark.asyncio
+async def test_phase2_pause_does_not_consume_execution_budget_and_duplicate_resume_is_fenced(plugin):
+    _, actions, service, runs = plugin
+    service.config["timeout_seconds"] = 1
+    launched = await launch(plugin, "await pause({stepId:'first'});await pause({stepId:'second'});return 42;")
+    run_id = launched["run_id"]
+    first = await waiting(plugin, run_id, "paused")
+    await asyncio.sleep(1.1)
+    payload = {"run_id": run_id, "step_id": "first", "attempt": 1, "fingerprint": first["fingerprint"]}
+    await actions["resume"](payload, context(runs))
+    async with asyncio.timeout(5):
+        while True:
+            second = await actions["get"]({"run_id": run_id}, context(runs))
+            if len(second["checkpoints"]) == 2:
+                break
+            await asyncio.sleep(0.01)
+    await actions["resume"](payload, context(runs))
+    assert (await actions["get"]({"run_id": run_id}, context(runs)))["checkpoints"][1]["status"] == "pending"
+    await actions["resume"]({**payload, "step_id": "second"}, context(runs))
+    assert (await done(plugin, run_id))["result"] == 42
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "script",
+    [
+        "await humanCheckpoint({stepId:'c',sourceStepId:'fake',artifactVersion:'approved'});",
+        "pause({stepId:'p'}); await agent({stepId:'a',agentRef:'lead_agent',task:'forbidden'});",
+        "await pause({stepId:'p'});",
+    ],
+)
+async def test_phase2_fail_closed_checkpoint_and_restart(plugin, script):
+    _, actions, service, runs = plugin
+    launched = await launch(plugin, script)
+    run_id = launched["run_id"]
+    if script == "await pause({stepId:'p'});":
+        paused = await waiting(plugin, run_id, "paused")
+        with pytest.raises(ValueError):
+            await actions["delete"]({"run_id": run_id}, context(runs))
+        await service.stop()
+        # Simulate a crash-retained record and a new Gateway instance.
+        record = await asyncio.to_thread(service.store.get, "alice", run_id)
+        record["status"] = "paused"
+        await asyncio.to_thread(service.store.put, "alice", record)
+        from deerflow_extension_workflows.service import Workflows
+
+        replacement = Workflows(service.config)
+        await replacement.start(None)
+        try:
+            assert (await replacement.get({"run_id": run_id}, context(runs)))["status"] == "interrupted"
+            with pytest.raises(ValueError):
+                await replacement.resume({"run_id": run_id, "step_id": "p", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
+        finally:
+            await replacement.stop()
+    else:
+        assert (await done(plugin, run_id))["status"] == "failed"
+    assert not runs.starts
+
+
+@pytest.mark.asyncio
+async def test_phase2_real_gateway_output_checkpoint(plugin, gateway_runtime):
+    from dataclasses import replace
+
+    runs, *_ = gateway_runtime
+    _, actions, service, _ = plugin
+    launched = await actions["launch"](
+        {"request_id": "real-approval", "script": "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'first'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return a.output;"},
+        context(runs),
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    retained = await actions["get_step"]({"run_id": run_id, "step_id": "a"}, context(runs))
+    assert retained["artifact_version"] == paused["checkpoints"][0]["artifact_version"]
+    await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": retained["artifact_version"], "decision": "approve"}, replace(context(runs), interaction="session"))
+    await service.wait_idle(run_id)
+    assert (await actions["get"]({"run_id": run_id}, context(runs)))["result"] == retained["output"]
+
+
+@pytest.mark.asyncio
+async def test_phase2_disconnected_decision_still_publishes_exactly_once(plugin, monkeypatch):
+    from dataclasses import replace
+
+    _, actions, service, runs = plugin
+    launched = await launch(plugin, "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'x'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return 42;")
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    decision = {"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["checkpoints"][0]["artifact_version"], "decision": "approve"}
+    persisted, release = asyncio.Event(), asyncio.Event()
+    original = service.save
+
+    async def hold(owner, record):
+        await original(owner, record)
+        if record["status"] == "waiting_approval" and record["checkpoints"][0]["status"] == "approved":
+            persisted.set()
+            await release.wait()
+
+    monkeypatch.setattr(service, "save", hold)
+    caller = asyncio.create_task(actions["decide"](decision, replace(context(runs), interaction="session")))
+    await asyncio.wait_for(persisted.wait(), 5)
+    caller.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await caller
+    release.set()
+    assert (await done(plugin, run_id))["result"] == 42
+    assert len(runs.starts) == 1
+
+
+@pytest.mark.asyncio
+async def test_phase2_inspection_example_reviews_each_stage_and_final_output(plugin, monkeypatch):
+    from dataclasses import replace
+
+    _, actions, _, runs = plugin
+    original = runs.answer
+
+    async def answer(run, input):
+        await original(run, input)
+        task = json.loads(input["messages"][0]["content"])
+        if isinstance(task, dict) and task.get("instruction", "").startswith("Return JSON with verdict"):
+            runs.threads[run.thread_id]["values"]["messages"][-1]["content"] = json.dumps({"verdict": "ready"})
+
+    monkeypatch.setattr(runs, "answer", answer)
+    script = (ROOT / "examples/deerflow-extension-workflows/examples/inspection.workflow.js").read_text(encoding="utf-8")
+    launched = await launch(
+        plugin, script, args={"complianceAgent": "lead_agent", "ingredientAgent": "lead_agent", "reportAgent": "lead_agent", "reviewerAgent": "lead_agent", "product": "synthetic", "referenceDate": "2026-10-15", "formula": []}
+    )
+    run_id = launched["run_id"]
+    for step_id, calls in [("confirm-evidence", 1), ("confirm-findings", 2), ("confirm-final-report", 4)]:
+        async with asyncio.timeout(5):
+            while True:
+                result = await actions["get"]({"run_id": run_id}, context(runs))
+                checkpoint = next((c for c in result["checkpoints"] if c["step_id"] == step_id), None)
+                if checkpoint and result["status"] == "waiting_approval":
+                    break
+                await asyncio.sleep(0.01)
+        assert len(runs.starts) == calls
+        await actions["decide"]({"run_id": run_id, "step_id": step_id, "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"}, replace(context(runs), interaction="session"))
+    result = await done(plugin, run_id)
+    assert result["status"] == "completed", result
+    assert result["result"]["status"] == "human_confirmed_report"
+    assert result["result"]["artifactVersion"] == result["checkpoints"][-1]["artifact_version"]
+
+
+@pytest.mark.asyncio
+async def test_immutable_file_review_and_downstream_consume_same_version(plugin):
+    import base64
+    from dataclasses import replace
+
+    _, actions, _, runs = plugin
+    launched = await launch(
+        plugin,
+        """
+await agent({stepId:'report',agentRef:'lead_agent',task:'create file'});
+const file=await captureArtifact({stepId:'file',sourceStepId:'report',path:'/mnt/user-data/outputs/report.pdf'});
+await humanCheckpoint({stepId:'confirm-file',sourceStepId:'file',artifactVersion:file.artifactVersion});
+return await readArtifact({artifactVersion:file.artifactVersion});
+""",
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    file = paused["artifacts"][0]
+    assert file["sha256"] and file["source_run_id"]
+    with pytest.raises(ValueError, match="approved"):
+        await actions["read_artifact"]({"run_id": run_id, "version": file["version"]}, context(runs))
+    preview = await actions["get_artifact"]({"run_id": run_id, "version": file["version"]}, context(runs))
+    runs.file_content = b"overwritten source after preview"
+    decision = {"run_id": run_id, "step_id": "confirm-file", "attempt": 1, "artifact_version": file["version"], "decision": "approve"}
+    with pytest.raises(ValueError):
+        await actions["decide"]({**decision, "artifact_version": "different-version"}, replace(context(runs), interaction="session"))
+    await actions["decide"](decision, replace(context(runs), interaction="session"))
+    result = await done(plugin, run_id)
+    assert result["status"] == "completed", result
+    assert result["result"] == preview
+    assert base64.b64decode(result["result"]["content_base64"]) == b"original file"
+    source = await actions["get_step"]({"run_id": run_id, "step_id": "report"}, context(runs))
+    assert source["configuration_evidence"]["run_id"] == source["run_id"]
+
+
+@pytest.mark.asyncio
+async def test_file_reference_requires_approval_before_child_admission(plugin):
+    launched = await launch(
+        plugin,
+        "await agent({stepId:'a',agentRef:'lead_agent',task:'file'});"
+        "const f=await captureArtifact({stepId:'f',sourceStepId:'a',path:'/mnt/user-data/outputs/report.pdf'});"
+        "return await agent({stepId:'b',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]});",
+    )
+    result = await done(plugin, launched["run_id"])
+    assert result["status"] == "failed"
+    assert len(plugin[3].starts) == 1
+
+
+@pytest.mark.asyncio
+async def test_approved_file_reference_is_bound_to_child_input(plugin):
+    from dataclasses import replace
+
+    _, actions, _, runs = plugin
+    launched = await launch(
+        plugin,
+        "await agent({stepId:'a',agentRef:'lead_agent',task:'file'});"
+        "const f=await captureArtifact({stepId:'f',sourceStepId:'a',path:'/mnt/user-data/outputs/report.pdf'});"
+        "await humanCheckpoint({stepId:'ok',sourceStepId:'f',artifactVersion:f.artifactVersion});return await agent({stepId:'b',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]});",
+    )
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    version = paused["artifacts"][0]["version"]
+    await actions["decide"]({"run_id": run_id, "step_id": "ok", "attempt": 1, "artifact_version": version, "decision": "approve"}, replace(context(runs), interaction="session"))
+    result = await done(plugin, run_id)
+    assert result["status"] == "completed", result
+    step = await actions["get_step"]({"run_id": run_id, "step_id": "b"}, context(runs))
+    task = json.loads(runs.starts[-1][1]["messages"][0]["content"])
+    assert step["input_artifacts"] == task["approved_artifacts"]
+    assert task["approved_artifacts"][0]["version"] == version
+    assert task["approved_artifacts"][0]["sha256"] == paused["artifacts"][0]["sha256"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("duplicate", [False, True])
+async def test_parallel_approved_file_consumers_reserve_atomically(plugin, monkeypatch, duplicate):
+    import threading
+    from dataclasses import replace
+
+    _, actions, service, runs = plugin
+    service.config["max_calls"] = 8 if duplicate else 2
+    script = """
+await agent({stepId:'source',agentRef:'lead_agent',task:'file'});
+const f=await captureArtifact({stepId:'file',sourceStepId:'source',path:'/mnt/user-data/outputs/report.pdf'});
+await humanCheckpoint({stepId:'confirm',sourceStepId:'file',artifactVersion:f.artifactVersion});
+return await Promise.all([
+agent({stepId:'consumer',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]}),
+agent({stepId:args.second,agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]})]);
+"""
+    launched = await launch(plugin, script, args={"second": "consumer" if duplicate else "other"})
+    run_id = launched["run_id"]
+    paused = await waiting(plugin, run_id)
+    original = service.store.artifact
+    barrier = threading.Barrier(2, timeout=5)
+    reads = []
+    lock = threading.Lock()
+
+    def synchronized(*args):
+        value = original(*args)
+        with lock:
+            reads.append(args)
+            decision_read = len(reads) == 1
+        if not decision_read:
+            barrier.wait()
+        return value
+
+    monkeypatch.setattr(service.store, "artifact", synchronized)
+    await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["artifacts"][0]["version"], "decision": "approve"}, replace(context(runs), interaction="session"))
+    result = await done(plugin, run_id)
+    assert len(reads) == 3 and not barrier.broken
+    assert result["status"] == "failed", result
+    assert len(result["steps"]) <= 2
+    assert len(runs.starts) <= 2
+    assert len({s["step_id"] for s in result["steps"]}) == len(result["steps"])

docs/full-stack-plugins.md

@@ -3,7 +3,7 @@
 A deployment-installed Python extension can register a `PluginContribution` with
 optional browser code, authenticated backend actions and model tools. This extends
 the existing `install(registry, config)` workflow. MCP and Skills keep their existing
-APIs and lifecycles. Public contracts live in `deerflow_extension_api` (0.2.6).
+APIs and lifecycles. Public contracts live in `deerflow_extension_api` (0.2.7).
 
 Backend actions and model tools may use the optional host-bound
 `context.agent_runs` capability to create, continue, inspect, resume, and cancel
@@ -40,6 +40,11 @@ the existing positional rollback and source attribution.
   loads a versioned resource manifest from an installed package (details below).
 - `BackendAction(name, handler)` declares an async handler receiving a JSON object and
   `ActionContext(principal, settings)`. The principal comes from host authentication.
+  Since API 0.2.7, keyword-only `interaction` defaults to `"other"`; the Gateway
+  stamps `"session"` only for authenticated session actions. It is not taken from
+  payloads and is never stamped on model `ToolContext`. Plugins implementing human
+  output review must also check resource ownership, exact attempt/version, and
+  reject `ToolContext`; session provenance alone is not business authorization.
 - `ModelTool(name, description, input_schema, handler, group="extensions")` declares
   an async handler with a `ToolContext` that additionally carries the thread ID.
   Object schemas must be inline: reference resolution is rejected. Tools enter the
@@ -510,3 +515,20 @@ is user input, not authorization**: plugins must revalidate the referenced objec
 current installation, and viewer permissions before acting. The host does not
 interpret a reference as delegation or start another Agent. Those behaviors
 belong to the plugin's backend handlers or middleware.
+
+### Immutable file review and run configuration evidence
+
+`AgentRuns.configuration_evidence(thread_id, run_id)` returns host-owned proof
+recorded from actual Agent assembly before graph execution and bound to the run.
+It includes actual model, safe numeric parameters and keyed versions of instructions,
+model configuration, tools, Skills and policies. No raw secrets are exposed.
+Missing evidence fails closed; the declarative revision remains a separate preflight
+guard. Evidence describes assembly-time configuration, not external service state.
+
+`AgentRuns.read_artifact(thread_id, path)` authorizes a bounded (2 MiB) regular
+output-file snapshot, rejects symlink traversal, and reserves the thread against
+concurrent runs/edits while reading off-loop. The workflow extension stores those
+bytes append-only with SHA-256, approves exact versions, and serves preview and
+approved consumer reads from the same snapshot (16 KiB chunks). File versions are
+owner/workflow scoped; consumers cannot use an unapproved version. See the workflow
+extension README for limits and `captureArtifact` / `readArtifact` examples.

examples/deerflow-extension-workflows/AGENTS.md

@@ -18,16 +18,21 @@ Keep the workflow deadline handler outside asyncio.timeout and test external can
 separately. Hash the Agent allowlist in canonical sorted order. Owner scope applies
 to every public read and mutation.
 Own persistence through supervisor publication even if the launching caller cancels.
-Serialize launch admission and shutdown. Hold the process lock across service life;
+Serialize launch admission and shutdown. After artifact IO, validate call limits
+and step uniqueness together with await-free reservation. Hold the process lock across service life;
 do not recover live records while another Gateway owns the file.
 
 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. Phase 1 has no resume, artifact approval, effective-configuration
-fingerprint guarantee, cross-process replay or hard monetary budget; keep limitations
-visible in docs, tool descriptions and UI. Do not serialize capability handles.
+human approval. Live checkpoints retain the worker without replay; restart interrupts
+all waits. 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
+append-only captured bytes and SHA-256; preview and approved consumption read
+that exact version. Never claim hard currency limits or freeze external state. Do not serialize capability handles.
 
 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

examples/deerflow-extension-workflows/README.md

@@ -1,4 +1,4 @@
-# Script workflows (Phase 1)
+# Script workflows (Phase 2)
 
 An opt-in packaged extension that runs inspectable JavaScript orchestration in the
 background. Scripts call full DeerFlow Agents through the public `AgentRuns`
@@ -6,15 +6,15 @@ capability. A chat tool or authenticated plugin action returns a workflow run ID
 promptly; **Script workflows** shows the script, argument hashes, execution limits,
 progress, child conversation/run identities, and final result.
 
-Phase 1 has no saved definition library, resume, authenticated output sign-off,
-continuation serialization, automatic restart replay, or hard monetary budget.
-It therefore does not complete the inspection-report scenario requiring human
-approval between stages. Agent execution success, a script verdict and human
-approval are separate facts.
+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.
 
 ## Install and enable
 
-Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.5,
+Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.7,
 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.
@@ -52,6 +52,7 @@ plugins:
       max_workflows_per_owner: 2
       timeout_seconds: 300
       admission_timeout_seconds: 10
+      pause_timeout_seconds: 86400  # Separate bound for each human/pause wait, maximum 7 days.
       cpu_seconds: 5
       memory_bytes: 16777216
 ```
@@ -107,7 +108,7 @@ log('Two checks finished');
 return checks.map(check => check.output);
 ```
 
-`agent()` resolves to `{output, threadId, runId, status: 'success'}`. `output` is the
+`agent()` resolves to `{output, threadId, runId, status: 'success', artifactVersion}`. `output` is the
 last attributable non-tool assistant answer: parsed JSON if valid, otherwise text
 or content blocks. There is no provider-enforced structured output schema in this
 API; validate verdicts in the script before branching. Every Agent call must be
@@ -129,6 +130,95 @@ operation, and uses its normal agent assembly, models, tools, skills, memory and
 checkpoints. The ledger records the requested agent reference and child IDs;
 it does **not** claim to record a per-run effective assembly fingerprint.
 
+## Saved versions, live resume and human review
+
+Use **Workflow name → Save workflow version**, then select an exact saved version.
+The immutable version is the SHA-256 of the script; identical saves are idempotent.
+Definitions are owner-scoped (100 versions per owner). Launching the same saved version
+with new arguments requires a new request ID and records both argument and script hashes.
+Saved scripts never silently follow the latest version. Editing the script detaches
+it from the selected definition. Chat tools `save_workflow` and `list_workflows`
+provide the same library; `launch_workflow` accepts either `script` or
+`definition: {name, version}` with `args` and `request_id`.
+
+```js
+const evidence = await agent({stepId:'evidence',agentRef:'lead_agent',task:args.product});
+await humanCheckpoint({
+  stepId:'confirm-evidence', sourceStepId:'evidence', artifactVersion:evidence.artifactVersion
+});
+await pause({stepId:'before-report'});
+return await agent({stepId:'report',agentRef:'lead_agent',task:{evidence:evidence.output}});
+```
+
+Checkpoints must be **awaited alone after all outstanding calls**. They share the
+unique step-ID namespace; each attempt is 1 and has at most one decision. Up to
+`max_calls` checkpoints are allowed in addition to the Agent call limit.
+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.
+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.
+
+The host records an opaque process-local **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
+continuation. This is deliberately conservative (even unrelated declarative config
+changes invalidate), and **not** a full effective-assembly fingerprint: mutable
+memory, referenced Skill/tool file contents, remote provider state and external
+side effects are not attested. Completed results are historical snapshots, not a
+claim that rerunning against changed external data would produce the same result.
+Each new child still goes through ordinary current authorization and Agent assembly.
+
+**Review exact output** displays the captured source-step output. Only then are
+**Approve this version / Reject this version** offered. The host decision action
+requires an authenticated browser session for the owner, exact checkpoint ID,
+attempt and server-derived output hash. Model tools, internal/PAT callers and
+auth-disabled sessions cannot approve. Auth-disabled deployments can run workflows
+and ordinary pauses, but must enable login to use human checkpoints. Approvals
+cannot be submitted through script messages; model-written “approved” has no effect.
+Matching duplicate decisions return the saved receipt without duplicate dispatch;
+a contradictory decision conflicts. Rejection stops downstream execution.
+
+For files, use `captureArtifact({stepId, sourceStepId, path})` after the source
+Agent completes. This stores independent bytes with SHA-256 and an immutable
+version, rather than approving a mutable path. Pass the capture step/version to
+`humanCheckpoint`. The browser reads that version, verifies its hash, and offers
+safe text/image/PDF preview or an exact-version download before approval.
+Completed checkpoint records retain this review/download entry after approval
+and when reopened; only pending checkpoints offer decision controls.
+A failed saved-definition lookup blocks launch until a version loads or the user
+explicitly selects a new script or edits the editor.
+`agent({..., artifacts:[version]})` admits only approved versions and includes
+version/hash references in the child input. The child reads bytes through
+`read_workflow_artifact`; scripts can use `readArtifact({artifactVersion, offset})`.
+Both return base64 chunks with `next_offset` (null at EOF). Neither reopens the
+source path. Limits: 2 MiB/file, 16 files/workflow, 64 MiB retained bytes/owner,
+and 16 KiB per read. Deleting a retained workflow also deletes its snapshots.
+See [`examples/file-review.workflow.js`](examples/file-review.workflow.js).
+JSON/text output checkpoints remain supported; links inside them require explicit
+file capture. No Agent success or reviewer verdict implicitly approves a version.
+
+Each completed child also retains **host assembly evidence**, bound to its run ID.
+The host records it before graph execution from resolved instructions, actual
+model/parameters, authorized tools, Skills and middleware/policies. Only safe
+numeric parameters and opaque keyed component versions are exposed, never raw
+prompts, provider kwargs or credentials. `get_step` exposes this historical proof;
+missing proof fails the workflow step. This is separate from the declarative
+preflight guard above. Versions compare within one host lifetime; stored proof
+survives restart. It describes assembly-time configuration, not later external
+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.
+
 ## Limits and status
 
 Limits are deployment-controlled, not writable by generated script or browser input.
@@ -162,6 +252,7 @@ reservation/enforcement contract before advertising that guarantee.
 | completed | Script and awaited child calls returned; output is unapproved task data |
 | failed | Script, resource limit (including wall deadline), attribution, child execution or observation failed |
 | cancelled | Stop requested and known active children interrupted or already terminal |
+| 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 |
@@ -181,8 +272,8 @@ run ID, then explicitly interrupts and observes that child. If admission times
 out or authorization is revoked during cleanup, it retains `needs_reconciliation`
 with stable thread/input/idempotency identity. It never automatically retries an
 uncertain admission. Stable request IDs return an existing workflow record; they
-never resume it or reexecute completed work. Changed scripts/args/limits with the
-same request ID conflict. Phase 2 definition/config invalidation is not implemented.
+never themselves resume it or reexecute completed work; use the explicit checkpoint action. Changed scripts/args/limits with the
+same request ID conflict. Definition versions are pinned; editing a definition creates a new version.
 
 Caller cancellation during launch does not abandon a committed record: the service
 owns persistence through supervisor publication, including shutdown coordination.
@@ -194,13 +285,12 @@ Result collection anchors on the visible human input, including the host's canon
 cannot become the anchor; genuinely later human input still conflicts.
 Result collection checks input message/run attribution and rejects an intervening
 human turn, because `get_state()` is the thread's latest state, not immutable run
-output. Results are copied before further script calls. Raw per-step input/results
-are retained privately; public inspection returns a bounded final result and step
-metadata. Child conversations remain accessible through normal host authorization.
+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. Neither graph resume nor tool permission
+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
 approval is interpreted as authenticated approval of an artifact version.
 
 ## Verify

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.5", name="workflows")
+@extension(api="0.2.7", name="workflows")
 def install(registry, config):
     enabled = config.get("enabled", False)
     if type(enabled) is not bool:
@@ -28,6 +28,7 @@ def install(registry, config):
         "max_workflows_per_owner": 2,
         "timeout_seconds": 300,
         "admission_timeout_seconds": 10,
+        "pause_timeout_seconds": 86400,
         "cpu_seconds": 5,
         "memory_bytes": 16 * 1024 * 1024,
     }
@@ -39,6 +40,7 @@ def install(registry, config):
         "max_workflows_per_owner": 4,
         "timeout_seconds": 600,
         "admission_timeout_seconds": 20,
+        "pause_timeout_seconds": 604800,
         "cpu_seconds": 10,
         "memory_bytes": 32 * 1024 * 1024,
     }
@@ -69,9 +71,19 @@ def install(registry, config):
         "properties": {
             "request_id": {"type": "string", "maxLength": 128},
             "script": {"type": "string", "maxLength": 16000},
+            "definition": {
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string"},
+                    "version": {"type": "string"},
+                },
+                "required": ["name", "version"],
+                "additionalProperties": False,
+            },
             "args": {"type": "object"},
         },
-        "required": ["request_id", "script"],
+        "required": ["request_id"],
+        "oneOf": [{"required": ["script"]}, {"required": ["definition"]}],
         "additionalProperties": False,
     }
     if (
@@ -80,23 +92,72 @@ def install(registry, config):
                 namespace=NAMESPACE,
                 title="Script workflows / 脚本工作流",
                 enabled=enabled,
-                description="Bounded background script orchestration of full Agents. Phase 1: no resume, human sign-off or restart replay; no hard monetary budget.",
+                description="Bounded background script orchestration of full Agents. Versioned definitions, live-process resume and authenticated output checkpoints. No restart replay or hard monetary budget.",
                 frontend=BrowserAssets("workflows.v1", Path(__file__).parent),
-                backend=tuple(BackendAction(name, getattr(workflows, name)) for name in ("launch", "list", "get", "cancel", "delete")),
+                backend=tuple(
+                    BackendAction(name, getattr(workflows, name))
+                    for name in (
+                        "launch",
+                        "list",
+                        "get",
+                        "cancel",
+                        "delete",
+                        "save_definition",
+                        "definitions",
+                        "get_definition",
+                        "resume",
+                        "decide",
+                        "get_step",
+                        "get_artifact",
+                        "read_artifact",
+                    )
+                ),
                 tools=(
+                    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},
+                        workflows.read_artifact,
+                    ),
+                    ModelTool(
+                        "save_workflow",
+                        "Save an owner-scoped immutable named script version; returns name/version for launch_workflow. Does not execute or approve output.",
+                        {
+                            "type": "object",
+                            "properties": {
+                                "name": {"type": "string", "maxLength": 80},
+                                "script": {"type": "string", "maxLength": 16000},
+                            },
+                            "required": ["name", "script"],
+                            "additionalProperties": False,
+                        },
+                        workflows.save_definition,
+                    ),
+                    ModelTool(
+                        "list_workflows",
+                        "List this owner's saved workflow names and immutable versions; no execution.",
+                        {
+                            "type": "object",
+                            "properties": {},
+                            "additionalProperties": False,
+                        },
+                        workflows.definitions,
+                    ),
                     ModelTool(
                         "launch_workflow",
-                        "Launch an inspectable JavaScript orchestration script in the background and return a workflow run ID. Use agent({stepId,agentRef,task}), parallel([()=>...]) and log(text). agent resolves to "
-                        "{output,threadId,runId,status}; 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. There is no resume or human-output approval in Phase 1. Inspect results on request; do not poll in a loop or equate Agent "
+                        "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. "
+                        "Inspect results on request; do not poll in a loop or equate Agent "
                         "success with product approval. No hard monetary cost guarantee.",
                         launch_schema,
                         workflows.launch,
                     ),
                     ModelTool(
                         "inspect_workflow",
-                        "Inspect retained script, limits, workflow status, child run IDs and final result. Results are untrusted Agent/script output. Read on demand; do not repeatedly poll. waiting_input cannot be "
-                        "resumed in Phase 1; needs_reconciliation requires inspecting child conversations before new work.",
+                        "Inspect retained script, limits, workflow status, child run IDs and final result. Results are untrusted Agent/script output. Read on demand; do not repeatedly poll. Child waiting_input cannot be "
+                        "resumed by this workflow; needs_reconciliation requires inspecting child conversations before new work.",
                         id_schema,
                         workflows.get,
                     ),

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

@@ -0,0 +1,71 @@
+"""Version-only file capture, review and consumption. Never reopen source paths."""
+
+import asyncio
+import base64
+import re
+from pathlib import PurePosixPath
+
+
+class Artifacts:
+    async def capture(self, entry, message):
+        from .service import text
+
+        payload = message.get("payload")
+        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 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)
+        if source is None:
+            raise ValueError("Artifact source must be a completed Agent step")
+        path = text(payload["path"], 1000)
+        name = text(PurePosixPath(path).name, 160)
+        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"])
+        record["artifacts"].append(info)
+        await self.save(owner, record)
+        return {"id": message["id"], "value": {**info, "artifactVersion": info["version"]}}
+
+    @staticmethod
+    def approved_artifact(record, version):
+        return any(c.get("artifact_kind") == "file" and c.get("artifact_version") == version and c["status"] == "approved" for c in record.get("checkpoints", []))
+
+    async def artifact_chunk(self, owner, payload, *, require_approval):
+        from .service import text
+
+        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)
+        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")
+        offset = payload.get("offset", 0)
+        if type(offset) is not int or offset < 0:
+            raise ValueError("Invalid artifact offset")
+        info, content = await asyncio.to_thread(self.store.artifact, owner, run_id, version)
+        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")}
+
+    async def get_artifact(self, payload, context):
+        """Browser review reads the same stored bytes as downstream consumers."""
+        from deerflow_extension_api.plugins import ToolContext
+
+        if isinstance(context, ToolContext):
+            raise ValueError("Use the approved-artifact consumer tool")
+        return await self.artifact_chunk(self.owner(context), payload, require_approval=False)
+
+    async def read_artifact(self, payload, context):
+        return await self.artifact_chunk(self.owner(context), payload, require_approval=True)
+
+    async def consume(self, entry, message):
+        payload = message.get("payload")
+        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)
+        return {"id": message["id"], "value": result}

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

@@ -0,0 +1,218 @@
+"""Immutable definitions and live-process checkpoints; never replay script effects."""
+
+import asyncio
+import copy
+import re
+import time
+
+from deerflow_extension_api.plugins import ToolContext
+
+
+class Lifecycle:
+    async def save_definition(self, payload, context):
+        from .service import text
+
+        owner = self.owner(context)
+        if set(payload) != {"name", "script"}:
+            raise ValueError("Definition requires name and script")
+        name = text(payload["name"], 80)
+        if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", name):
+            raise ValueError("Invalid definition name")
+        script, version = self.script_version(payload["script"])
+        return await asyncio.to_thread(self.store.save_definition, owner, name, version, script)
+
+    @staticmethod
+    def script_version(value):
+        import hashlib
+
+        from .service import encoded, text
+
+        script = text(value, 16000)
+        encoded(script, 16000)
+        return script, hashlib.sha256(script.encode()).hexdigest()
+
+    async def definitions(self, payload, context):
+        return {"definitions": await asyncio.to_thread(self.store.definitions, self.owner(context))}
+
+    async def get_definition(self, payload, context):
+        from .service import text
+
+        if set(payload) != {"name", "version"}:
+            raise ValueError("Definition requires exact name and version")
+        name, version = text(payload["name"], 80), text(payload["version"], 64)
+        script = await asyncio.to_thread(self.store.definition, self.owner(context), name, version)
+        return {"name": name, "version": version, "script": script}
+
+    async def get_step(self, payload, context):
+        from .service import encoded, text
+
+        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)
+        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"}
+        encoded(result)
+        return result
+
+    async def check_configuration(self, entry):
+        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"]))
+        if revision is None or revision != current:
+            raise ValueError("Agent configuration changed or unavailable; start a new workflow")
+        from .service import identity
+
+        if record["operator_revision"] != identity({**self.config, "agents": sorted(self.config["agents"])}):
+            raise ValueError("Operator configuration changed; start a new workflow")
+
+    async def checkpoint(self, entry, message, deadline):
+        from .service import text
+
+        record = entry["record"]
+        payload = message.get("payload")
+        human = message["type"] == "humanCheckpoint"
+        keys = {"stepId", "sourceStepId", "artifactVersion"} if human else {"stepId"}
+        if not isinstance(payload, dict) or set(payload) != keys:
+            raise ValueError("Invalid checkpoint fields")
+        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", [])):
+            raise ValueError("Step ID already used")
+        if len(record["checkpoints"]) >= self.config["max_calls"]:
+            raise ValueError("Checkpoint limit")
+        await self.check_configuration(entry)
+        row = {
+            "step_id": step_id,
+            "attempt": 1,
+            "kind": message["type"],
+            "status": "pending",
+        }
+        if human:
+            source = next(
+                (s for s in record["steps"] + record.get("artifacts", []) if s["step_id"] == payload["sourceStepId"] and s["status"] == "completed"),
+                None,
+            )
+            if not source or source.get("artifact_version") != payload["artifactVersion"]:
+                raise ValueError("Checkpoint must reference an exact retained step output version")
+            row.update(
+                artifact_kind=source.get("kind", "output"),
+                source_step_id=source["step_id"],
+                source_attempt=source["attempt"],
+                artifact_version=source["artifact_version"],
+            )
+        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"]
+        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"])
+        signals = [
+            asyncio.create_task(entry["gate"].wait()),
+            asyncio.create_task(entry["cancel"].wait()),
+        ]
+        try:
+            await asyncio.wait(signals, return_when=asyncio.FIRST_COMPLETED)
+        finally:
+            for signal in signals:
+                signal.cancel()
+            await asyncio.gather(*signals, return_exceptions=True)
+        if entry["cancel"].is_set() or row["status"] == "rejected":
+            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["status"] = "running"
+        record.pop("pause_expires_at", None)
+        await self.save(entry["owner"], record)
+        return {"id": message["id"], "value": {"status": row["status"]}}
+
+    async def resume(self, payload, context):
+        return await self.control(payload, context, human=False)
+
+    async def decide(self, payload, context):
+        return await self.control(payload, context, human=True)
+
+    async def control(self, payload, context, *, human):
+        # A cancelled HTTP request must not strand a persisted decision before
+        # signalling the live worker. Service shutdown drains these tasks too.
+        task = asyncio.create_task(self.apply_control(copy.deepcopy(dict(payload)), context, human=human))
+        self.admissions.add(task)
+        task.add_done_callback(self.admission_finished)
+        return await asyncio.shield(task)
+
+    async def apply_control(self, payload, context, *, human):
+        from .service import text
+
+        owner = self.owner(context)
+        if context.agent_runs is None:
+            raise ValueError("Session-bound AgentRuns required")
+        if human and (context.interaction != "session" or isinstance(context, ToolContext) or context.principal.is_internal):
+            raise ValueError("Human decisions require an authenticated browser session")
+        expected = {"run_id", "step_id", "attempt", "artifact_version", "decision"} if human else {"run_id", "fingerprint", "step_id", "attempt"}
+        if set(payload) != expected:
+            raise ValueError("Unexpected resume/decision fields")
+        run_id = text(payload["run_id"], 64)
+        async with self.control_lock:
+            retained = await asyncio.to_thread(self.store.get, owner, run_id)
+            entry = self.entries.get(run_id)
+            record = entry["record"] if entry else retained
+            if human:
+                row = next(
+                    (r for r in record.get("checkpoints", []) if r["step_id"] == payload["step_id"]),
+                    None,
+                )
+                if (
+                    not row
+                    or row["kind"] != "humanCheckpoint"
+                    or type(payload["attempt"]) is not int
+                    or row["attempt"] != payload["attempt"]
+                    or row["artifact_version"] != payload["artifact_version"]
+                    or payload["decision"] not in ("approve", "reject")
+                ):
+                    raise ValueError("Decision does not match exact checkpoint attempt and artifact version")
+                status = "approved" if payload["decision"] == "approve" else "rejected"
+                if row["status"] != "pending":
+                    if row["status"] == status:
+                        return {"run_id": run_id, "status": status}
+                    raise ValueError("Checkpoint already decided")
+            else:
+                if payload["fingerprint"] != record["fingerprint"]:
+                    raise ValueError("Script, arguments or configuration do not match")
+                row = next(
+                    (r for r in reversed(record.get("checkpoints", [])) if r["kind"] == "pause" and r["step_id"] == payload["step_id"]),
+                    None,
+                )
+                if not row or type(payload["attempt"]) is not int or payload["attempt"] != row["attempt"]:
+                    raise ValueError("Workflow is not paused at this attempt")
+                if row["status"] == "resumed":
+                    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")
+            # 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 self.check_configuration(entry)
+            if entry["cancel"].is_set() or record["status"] not in (
+                "paused",
+                "waiting_approval",
+            ):
+                raise ValueError("Workflow stopping")
+            if human and row.get("artifact_kind") == "file":
+                await asyncio.to_thread(self.store.artifact, owner, run_id, row["artifact_version"])
+            previous = copy.deepcopy(row)
+            row.update(status=status, decided_by=owner, decided_at=time.time())
+            try:
+                await self.save(owner, record)
+            except BaseException:
+                row.clear()
+                row.update(previous)
+                raise
+            entry["gate"].set()
+            return {"run_id": run_id, "status": status}

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

@@ -15,6 +15,8 @@ from pathlib import Path
 
 from deerflow_extension_api import AgentRunError
 
+from .artifacts import Artifacts
+from .lifecycle import Lifecycle
 from .store import TERMINAL, Store
 
 NAMESPACE = "community.workflows"
@@ -58,19 +60,27 @@ def project(record, detail=True):
         )
     }
     result["run_id"] = record["id"]
+    result.update(
+        fingerprint=record["fingerprint"],
+        definition=record.get("definition"),
+        configuration_revision=record.get("configuration_revision"),
+    )
     if detail:
         result.update(
+            checkpoints=copy.deepcopy(record.get("checkpoints", [])),
+            artifacts=copy.deepcopy(record.get("artifacts", [])),
+            pause_expires_at=record.get("pause_expires_at"),
             script=record["script"],
             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")} 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:
+class Workflows(Lifecycle, Artifacts):
     def __init__(self, config):
         self.config = config
         self.store = None
@@ -80,6 +90,7 @@ class Workflows:
         self.entries = {}
         self.admissions = set()
         self.launch_lock = asyncio.Lock()
+        self.control_lock = asyncio.Lock()
         self.write_lock = asyncio.Lock()
         self.global_slots = asyncio.Semaphore(config["global_concurrency"])
         self.lockfile = None
@@ -137,9 +148,16 @@ class Workflows:
         owner = self.owner(context)
         if context.agent_runs is None:
             raise ValueError("Session-bound AgentRuns required; PAT/internal callers unsupported")
-        if set(payload) - {"request_id", "script", "args"}:
+        if set(payload) - {"request_id", "script", "args", "definition"}:
             raise ValueError("Unexpected launch fields")
-        script = text(payload.get("script"), 16000)
+        definition = payload.get("definition")
+        if definition is not None:
+            if "script" in payload or not isinstance(definition, dict):
+                raise ValueError("Choose a script or an exact definition version")
+            definition = await self.get_definition(definition, context)
+            script = definition.pop("script")
+        else:
+            script = text(payload.get("script"), 16000)
         encoded(script, 16000)
         request_id = text(payload.get("request_id"), 128)
         args = copy.deepcopy(payload.get("args", {}))
@@ -156,6 +174,7 @@ class Workflows:
                 "admission_timeout_seconds",
                 "max_calls",
                 "timeout_seconds",
+                "pause_timeout_seconds",
                 "cpu_seconds",
                 "memory_bytes",
             )
@@ -168,10 +187,16 @@ class Workflows:
                 "args_hash": args_hash,
                 "limits": limits,
                 "agents": sorted(self.config["agents"]),
+                "definition": definition,
             }
         )
         record = {
             "id": uuid.uuid4().hex,
+            "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,
+            "operator_revision": identity({**self.config, "agents": sorted(self.config["agents"])}),
+            "checkpoints": [],
+            "artifacts": [],
             "request_id": request_id,
             "fingerprint": fingerprint,
             "script_hash": script_hash,
@@ -231,7 +256,11 @@ class Workflows:
         self.workers.pop(run_id, None)
         self.entries.pop(run_id, None)
         if not task.cancelled() and task.exception() is not None:
-            logger.error("Workflow %s supervisor failed (%s); inspect retained child records", run_id, type(task.exception()).__name__)
+            logger.error(
+                "Workflow %s supervisor failed (%s); inspect retained child records",
+                run_id,
+                type(task.exception()).__name__,
+            )
 
     async def get(self, payload, context):
         owner = self.owner(context)
@@ -270,22 +299,29 @@ class Workflows:
         response = {"id": message["id"]}
         row = None
         try:
-            if not isinstance(payload, dict) or set(payload) != {
-                "stepId",
-                "agentRef",
-                "task",
-            }:
+            if not isinstance(payload, dict) or not {"stepId", "agentRef", "task"} <= set(payload) or set(payload) - {"stepId", "agentRef", "task", "artifacts"}:
                 raise ValueError("agent requires stepId, agentRef, task")
             step_id = text(payload["stepId"], 80)
             if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", step_id):
                 raise ValueError("stepId must contain ASCII letters, digits, underscores or hyphens")
             if payload["agentRef"] not in self.config["agents"]:
                 raise ValueError("Agent reference not permitted by operator")
-            if any(s["step_id"] == step_id for s in record["steps"]):
-                raise ValueError("Step ID already used")
-            if len(record["steps"]) >= self.config["max_calls"]:
-                entry["fault"] = "Total Agent call limit"
-                raise ValueError(entry["fault"])
+            versions = payload.get("artifacts", [])
+            if not isinstance(versions, list) or len(versions) > 16 or any(not isinstance(v, str) for v in versions):
+                raise ValueError("artifacts must contain at most 16 version IDs")
+            references = []
+            for version in versions:
+                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")}})
+            task = payload["task"]
+            if references:
+                task = {
+                    "task": task,
+                    "approved_artifacts": references,
+                    "artifact_instruction": "Read these exact versions using read_workflow_artifact (run_id, version, offset). Follow next_offset until null. Do not use source file paths.",
+                }
             # Stable identity/input are recorded BEFORE thread creation or run admission.
             thread_id = "workflow-" + identity([record["id"], step_id])[:32]
             message_id = "workflow-" + identity([record["id"], step_id, 1])[:32]
@@ -294,10 +330,16 @@ class Workflows:
                     {
                         "role": "user",
                         "id": message_id,
-                        "content": encoded(payload["task"], 8000),
+                        "content": encoded(task, 16000 if references else 8000),
                     }
                 ]
             }
+            # 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,
@@ -306,6 +348,7 @@ class Workflows:
                 "run_id": None,
                 "status": "queued",
                 "input": input,
+                "input_artifacts": references,
                 "idempotency_key": f"{record['id']}:{message_id}:1",
             }
             record["steps"].append(row)
@@ -314,6 +357,13 @@ class Workflows:
                 if entry["cancel"].is_set() or entry["fault"]:
                     row["status"] = "cancelled"
                     raise ValueError("Workflow stopping")
+                if record.get("configuration_revision") is not None:
+                    try:
+                        await self.check_configuration(entry)
+                    except Exception:
+                        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.
@@ -334,7 +384,11 @@ class Workflows:
                         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)
                 while True:
                     if entry["cancel"].is_set() or entry["fault"]:
@@ -352,7 +406,7 @@ class Workflows:
                 if state.get("next") or state.get("interrupts") or pending_input(messages):
                     row["status"] = "waiting_input"
                     entry["fault"] = "waiting_input"
-                    raise ValueError("Child requires input; Phase 1 cannot resume or approve")
+                    raise ValueError("Child requires input; workflow resume cannot answer child interrupts")
                 if run.status != "success":
                     row["status"] = "failed"
                     raise ValueError("Agent run did not succeed")
@@ -384,9 +438,22 @@ class Workflows:
                 else:
                     output = content
                 encoded(output, 12000)
+                # Evidence comes from the run's completed host assembly, never a
+                # fresh read of mutable configuration after the Agent finishes.
+                async with asyncio.timeout(5):
+                    proof = await runs.configuration_evidence(thread_id=thread_id, run_id=run.run_id)
+                if proof.get("run_id") != run.run_id or proof.get("thread_id") != thread_id or proof.get("source") != "host_assembly":
+                    raise ValueError("Host configuration evidence does not match child run")
+                row["configuration_evidence"] = proof
+                row["configuration_version"] = proof["version"]
                 # Capture before any later turn can alter the thread checkpoint.
-                row.update(status="completed", output=copy.deepcopy(output))
+                row.update(
+                    status="completed",
+                    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,
@@ -466,7 +533,7 @@ class Workflows:
                     "memory_bytes": self.config["memory_bytes"],
                 },
             )
-            async with asyncio.timeout(self.config["timeout_seconds"]):
+            async with asyncio.timeout(self.config["timeout_seconds"]) as deadline:
                 turns = 0
                 while True:
                     if entry["cancel"].is_set() or entry["fault"]:
@@ -483,6 +550,19 @@ class Workflows:
                     messages = frame["messages"]
                     if not isinstance(messages, list) or len(messages) > 64:
                         raise ValueError("Worker message limit")
+                    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]})
+                        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]})
+                        continue
                     for message in messages:
                         if message.get("type") == "log":
                             if len(record["logs"]) >= 20:
@@ -523,7 +603,10 @@ class Workflows:
         except TimeoutError:
             # Keep this outside asyncio.timeout: its own cancellation must first
             # leave the context manager so it can be translated to TimeoutError.
-            record.update(status="failed", error="Execution deadline exceeded")
+            record.update(
+                status="failed",
+                error="Checkpoint wait deadline exceeded" if record["status"] in ("paused", "waiting_approval") else "Execution deadline exceeded",
+            )
         except asyncio.CancelledError:
             entry["cancel"].set()
             record.update(status="interrupted", error="Host interrupted; child cleanup required")

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

@@ -7,6 +7,8 @@ function node(tag, text = '', attrs = {}) {
 function mount(root, context) {
   const t = (en, zh) => context.locale?.startsWith('zh') ? zh : en;
   let disposed = false, selected = null, generation = 0, timer;
+  const blobUrls = new Set();
+  function releaseFiles(){for(const url of blobUrls)URL.revokeObjectURL(url);blobUrls.clear();}
   const sheet = node('link', '', {rel:'stylesheet',href:new URL('./style.css',import.meta.url).href});
   const panel = node('section','',{className:'script-workflows'});
   const notice = node('p','',{role:'status'}), navigation = node('nav'), detail = node('div');
@@ -14,10 +16,39 @@ function mount(root, context) {
   script.value = "const result = await agent({stepId:'check',agentRef:'lead_agent',task:args.task});\nreturn result.output;";
   const args = node('textarea','',{rows:3,ariaLabel:t('JSON arguments','JSON 参数')});
   args.value = '{"task":"Write a short checklist"}';
+  let definition = null, libraryGeneration = 0, definitionBlocked = false;
+  const definitions = node('select','',{ariaLabel:t('Saved workflow version','已保存工作流版本')});
+  const definitionName = node('input','',{ariaLabel:t('Workflow name','工作流名称'),placeholder:'inspection'});
+  const library = node('div');
+  async function loadDefinitions() {
+    const result = await call('definitions',{});
+    definitions.replaceChildren(node('option',t('New script','新脚本'),{value:''}));
+    for(const item of result.definitions) definitions.append(node('option',`${item.name} · ${item.version.slice(0,12)}`,{value:JSON.stringify(item)}));
+  }
+  definitions.onchange=async()=>{
+    const version=++libraryGeneration, chosen=definitions.value?JSON.parse(definitions.value):null;
+    definition=null;definitionBlocked=Boolean(chosen);submit.disabled=true;
+    try {
+      if(chosen){
+        const item=await call('get_definition',chosen);
+        if(version!==libraryGeneration)return;
+        script.value=item.script;definitionName.value=item.name;definition=chosen;definitionBlocked=false;
+      }
+    } catch(e){if(version===libraryGeneration)notice.textContent=e.message;}
+    finally{if(version===libraryGeneration)submit.disabled=definitionBlocked;}
+  };
+  script.oninput=()=>{libraryGeneration++;definition=null;definitions.value='';definitionBlocked=false;submit.disabled=false;};
+  library.append(definitions,definitionName,button(t('Save workflow version','保存工作流版本'),async()=>{
+    const version=libraryGeneration, source=script.value;
+    const saved=await call('save_definition',{name:definitionName.value,script:source});
+    await loadDefinitions();
+    if(version===libraryGeneration && script.value===source){definition=saved;definitionBlocked=false;submit.disabled=false;definitions.value=JSON.stringify(saved);}
+    notice.textContent=t(`Saved ${saved.name} · ${saved.version}`,`已保存 ${saved.name} · ${saved.version}`);
+  }));
   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('Phase 1: bounded background Agent calls. No resume, human sign-off, restart replay or hard monetary budget. Cancellation preserves child checkpoints and does not undo side effects.','第一阶段：有界后台 Agent 调用。不支持恢复、成果审批、重启回放或金额硬上限。取消保留子对话检查点，不撤销副作用。')),notice,form,navigation,detail);
+  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);
   root.append(sheet,panel);
   const alive = version => !disposed && !context.signal?.aborted && version === generation;
   async function call(action,payload) {
@@ -42,18 +73,71 @@ function mount(root, context) {
     if(!id)return;
     const run=await call('get',{run_id:id});
     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(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({args:run.args,limits:run.limits,script_hash:run.script_hash,args_hash:run.args_hash},null,2)));
+    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){
       const row=node('div',`${step.step_id} · ${step.agent_ref} · ${step.status}`);
       if(step.thread_id && context.openConversation) row.append(button(t('Open child conversation','打开子对话'),()=>context.openConversation(step.thread_id)));
+      if(step.status==='completed') row.append(button(t('Inspect retained output','查看保留输出'),async()=>{
+        const value=await call('get_step',{run_id:id,step_id:step.step_id});
+        if(alive(version)&&id===selected)row.append(node('pre',JSON.stringify(value,null,2)));
+      }));
       row.append(node('code',step.run_id||t('Run admission pending','等待 run admission')));detail.append(row);
     }
+    for(const checkpoint of run.checkpoints||[]) {
+      const review=node('section','',{className:'checkpoint'});
+      review.append(node('h4',`${checkpoint.step_id} · ${checkpoint.status}`),node('pre',JSON.stringify(checkpoint,null,2)));
+      detail.append(review);
+      const canDecide=checkpoint.status==='pending' && run.status==='waiting_approval';
+      if(checkpoint.status==='pending' && checkpoint.kind==='pause' && run.status==='paused'){
+        review.append(button(t('Resume this checkpoint','继续此检查点'),async()=>{
+          await call('resume',{run_id:id,step_id:checkpoint.step_id,attempt:checkpoint.attempt,fingerprint:run.fingerprint});await show();
+        }));
+      }
+      if(checkpoint.kind==='humanCheckpoint'){
+        review.append(button(t('Review exact output','审阅确切输出'),async()=>{
+          const output=await call('get_step',{run_id:id,step_id:checkpoint.source_step_id});
+          if(!alive(version)||id!==selected)return;
+          if(output.artifact_version!==checkpoint.artifact_version)throw Error(t('Artifact version changed','成果版本已变化'));
+          if(output.kind==='file') {
+            if(!Number.isInteger(output.size)||output.size<0||output.size>2097152)throw Error('Invalid file size');
+            const bytes=new Uint8Array(output.size);
+            let offset=0;
+            do {
+              const chunk=await call('get_artifact',{run_id:id,version:output.artifact_version,offset});
+              if(!alive(version)||id!==selected||!review.isConnected)return;
+              if(chunk.version!==output.artifact_version||chunk.sha256!==output.sha256||chunk.offset!==offset)throw Error('File version mismatch');
+              const part=Uint8Array.from(atob(chunk.content_base64),c=>c.charCodeAt(0));
+              if(offset+part.length>bytes.length)throw Error('Invalid file chunk');
+              bytes.set(part,offset);offset+=part.length;
+              if(chunk.next_offset===null){if(offset!==bytes.length)throw Error('Incomplete file');break;}
+              if(!part.length||chunk.next_offset!==offset)throw Error('Invalid file offset');
+            }while(true);
+            const hash=Array.from(new Uint8Array(await crypto.subtle.digest('SHA-256',bytes)),b=>b.toString(16).padStart(2,'0')).join('');
+            if(hash!==output.sha256)throw Error('File hash mismatch');
+            if(!alive(version)||id!==selected||!review.isConnected)return;
+            const ext=output.name.split('.').pop().toLowerCase();
+            const mime=({png:'image/png',jpg:'image/jpeg',jpeg:'image/jpeg',webp:'image/webp',pdf:'application/pdf'})[ext]||'application/octet-stream';
+            const url=URL.createObjectURL(new Blob([bytes],{type:mime}));blobUrls.add(url);
+            review.append(node('p',`${output.name} · SHA-256 ${hash}`));
+            if(['txt','md','json','csv','log','yaml','yml'].includes(ext))review.append(node('pre',new TextDecoder().decode(bytes)));
+            else if(mime.startsWith('image/'))review.append(node('img','',{src:url,alt:output.name}));
+            else if(ext==='pdf'){const frame=node('iframe','',{src:url,title:output.name});frame.setAttribute('sandbox','');review.append(frame);}
+            review.append(node('a',t('Download this exact version','下载此确切版本'),{href:url,download:output.name}));
+          } else review.append(node('pre',JSON.stringify(output.output,null,2)));
+          if(canDecide)for(const decision of ['approve','reject'])review.append(button(decision==='approve'?t('Approve this version','确认此版本'):t('Reject this version','拒绝此版本'),async()=>{
+            await call('decide',{run_id:id,step_id:checkpoint.step_id,attempt:checkpoint.attempt,artifact_version:checkpoint.artifact_version,decision});await show();
+          }));
+        }));
+      }
+    }
     detail.append(node('h4',t('Progress','进度')),node('pre',run.logs.join('\n')),node('h4',t('Final result','最终结果')),node('pre',JSON.stringify(run.result,null,2)));
-    if(['queued','running','cancelling'].includes(run.status)){
+    if(['queued','running','cancelling','paused','waiting_approval'].includes(run.status)){
       detail.append(button(t('Cancel workflow','取消工作流'),async()=>{await call('cancel',{run_id:id});await show();}));
-      clearTimeout(timer);timer=setTimeout(()=>show().catch(e=>{if(!disposed)notice.textContent=e.message;}),2000);
+      clearTimeout(timer);if(['queued','running','cancelling'].includes(run.status))timer=setTimeout(()=>show().catch(e=>{if(!disposed)notice.textContent=e.message;}),2000);
     }else{
       detail.append(button(t('Delete retained record','删除保留记录'),async()=>{await call('delete',{run_id:id});selected=null;generation++;detail.replaceChildren();await list();}));
       await list();
@@ -70,19 +154,23 @@ function mount(root, context) {
     }
   }
   form.onsubmit=async event=>{
-    event.preventDefault();submit.disabled=true;notice.textContent='';
+    event.preventDefault();
+    if(definitionBlocked)return;
+    submit.disabled=true;notice.textContent='';
     try{
       const value=JSON.parse(args.value);
       if(!value || Array.isArray(value)||typeof value!=='object')throw Error(t('Arguments must be an object','参数必须是对象'));
-      const content=JSON.stringify({script:script.value,args:value});
+      const source=definition?{definition}:{script:script.value};
+      const content=JSON.stringify({...source,args:value});
       if(pendingRequest?.content!==content)pendingRequest={content,id:Array.from(crypto.getRandomValues(new Uint8Array(16)),b=>b.toString(16).padStart(2,'0')).join('')};
-      const result=await call('launch',{request_id:pendingRequest.id,script:script.value,args:value});
+      const result=await call('launch',{request_id:pendingRequest.id,...source,args:value});
       // Keep the identity for unchanged input, including after accepted-launch
       // refresh failures. Refresh retries must never submit another launch.
       selected=result.run_id;generation++;clearTimeout(timer);await inspectAccepted(result.run_id);
-    }catch(e){if(!disposed)notice.textContent=e.message;}finally{submit.disabled=false;}
+    }catch(e){if(!disposed)notice.textContent=e.message;}finally{submit.disabled=definitionBlocked;}
   };
+  loadDefinitions().catch(e=>{if(!disposed)notice.textContent=e.message;});
   list().catch(e=>{if(!disposed)notice.textContent=e.message;});
-  return {dispose(){disposed=true;generation++;clearTimeout(timer);sheet.remove();panel.remove();}};
+  return {dispose(){releaseFiles();disposed=true;generation++;clearTimeout(timer);sheet.remove();panel.remove();}};
 }
 export default {apiVersion:1,module:'workflows.v1',surfaces:[{id:'runs',slot:'page',title:'Script workflows',navigation:{label:'Script workflows',labelZh:'脚本工作流'},mount}]};

examples/deerflow-extension-workflows/deerflow_extension_workflows/static/style.css

@@ -6,3 +6,6 @@
 .script-workflows pre { padding: 12px; border: 1px solid var(--border); white-space: pre-wrap; overflow-wrap: anywhere; }
 .script-workflows code { display: block; font-size: 12px; overflow-wrap: anywhere; }
 .script-workflows [role=status] { color: var(--destructive); }
+.script-workflows .checkpoint img { display: block; max-width: 100%; max-height: 640px; object-fit: contain; }
+.script-workflows .checkpoint iframe { display: block; width: 100%; height: 640px; border: 1px solid var(--border); }
+.script-workflows .checkpoint a { display: block; margin: 12px 0; text-decoration: underline; }

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

@@ -1,5 +1,6 @@
 """Owner-scoped, retained records; no capability serialization or automatic replay."""
 
+import hashlib
 import json
 import sqlite3
 from contextlib import closing
@@ -24,6 +25,84 @@ class Store:
         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))")
             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, "
+                "info TEXT NOT NULL, content BLOB NOT NULL, PRIMARY KEY(owner,workflow,step_id), UNIQUE(owner,workflow,version))"
+            )
+            db.execute("CREATE TRIGGER IF NOT EXISTS workflow_artifacts_immutable BEFORE UPDATE ON workflow_artifacts BEGIN SELECT RAISE(ABORT, 'Artifact versions cannot be overwritten'); END")
+            db.execute("CREATE TABLE IF NOT EXISTS workflow_definitions (owner TEXT NOT NULL, name TEXT NOT NULL, version TEXT NOT NULL, script TEXT NOT NULL, PRIMARY KEY(owner,name,version))")
+
+    def capture_artifact(self, owner, workflow, step_id, name, content):
+        if not isinstance(content, bytes) or len(content) > 2 * 1024 * 1024:
+            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"}
+        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()
+            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:
+                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:
+                raise ValueError("Owner artifact limit (64 MiB)")
+            db.execute("INSERT INTO workflow_artifacts VALUES (?,?,?,?,?,?)", (owner, workflow, step_id, version, json.dumps(info), content))
+        return info
+
+    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()
+            if not row:
+                raise ValueError("Artifact unavailable")
+            info, content = json.loads(row[0]), row[1]
+            if hashlib.sha256(content).hexdigest() != info["sha256"] or len(content) != info["size"]:
+                raise ValueError("Artifact integrity failure")
+            return info, content
+
+    def save_definition(self, owner, name, version, script):
+        with closing(self.connect()) as db, db:
+            db.execute("BEGIN IMMEDIATE")
+            existing = db.execute(
+                "SELECT 1 FROM workflow_definitions WHERE owner=? AND name=? AND version=?",
+                (owner, name, version),
+            ).fetchone()
+            if not existing:
+                if (
+                    db.execute(
+                        "SELECT count(*) FROM workflow_definitions WHERE owner=?",
+                        (owner,),
+                    ).fetchone()[0]
+                    >= 100
+                ):
+                    raise ValueError("Definition version limit (100)")
+                db.execute(
+                    "INSERT INTO workflow_definitions VALUES (?,?,?,?)",
+                    (owner, name, version, script),
+                )
+        return {"name": name, "version": version}
+
+    def definitions(self, owner):
+        with closing(self.connect()) as db:
+            return [
+                {"name": name, "version": version}
+                for name, version in db.execute(
+                    "SELECT name,version FROM workflow_definitions WHERE owner=? ORDER BY rowid DESC",
+                    (owner,),
+                )
+            ]
+
+    def definition(self, owner, name, version):
+        with closing(self.connect()) as db:
+            row = db.execute(
+                "SELECT script FROM workflow_definitions WHERE owner=? AND name=? AND version=?",
+                (owner, name, version),
+            ).fetchone()
+            if not row:
+                raise ValueError("Definition unavailable")
+            return row[0]
 
     def connect(self):
         return sqlite3.connect(self.path, timeout=5)
@@ -91,11 +170,12 @@ 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 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')").fetchall():
+            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)
                 record.update(
                     status="interrupted",

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

@@ -18,6 +18,12 @@ BOOT = r"""
     if(pending.size>=64){reject(new Error('Pending call limit'));return;}
     const id=++serial; requests.push({type:'agent',id,payload:clone(payload)});pending.set(id,{resolve,reject});
   })),writable:false,configurable:false});
+  for(const name of ['pause','humanCheckpoint','captureArtifact','readArtifact']) {
+    Object.defineProperty(globalThis,name,{value:Object.freeze(payload=>new Promise((resolve,reject)=>{
+      if(pending.size>=64){reject(new Error('Pending call limit'));return;}
+      const id=++serial; requests.push({type:name,id,payload:clone(payload)});pending.set(id,{resolve,reject});
+    })),writable:false,configurable:false});
+  }
   Object.defineProperty(globalThis,'parallel',{value:Object.freeze(tasks=>{
     if(!Array.isArray(tasks)||tasks.length>64) throw new Error('Parallel item limit');
     return Promise.all(tasks.map(task=>task()));

examples/deerflow-extension-workflows/examples/file-review.workflow.js

@@ -0,0 +1,6 @@
+// Requires authenticated browser approval and read_workflow_artifact enabled.
+await agent({stepId:'write',agentRef:'lead_agent',task:'Write a concise report to /mnt/user-data/outputs/report.md and return its path.'});
+const file = await captureArtifact({stepId:'report-file',sourceStepId:'write',path:'/mnt/user-data/outputs/report.md'});
+await humanCheckpoint({stepId:'confirm-report',sourceStepId:'report-file',artifactVersion:file.artifactVersion});
+const review = await agent({stepId:'review',agentRef:'lead_agent',task:'Read the approved report version using read_workflow_artifact, then summarize it.',artifacts:[file.artifactVersion]});
+return {artifactVersion:file.artifactVersion,sha256:file.sha256,review:review.output};

examples/deerflow-extension-workflows/examples/inspection.workflow.js

@@ -0,0 +1,17 @@
+// Return complete report text, not mutable file links, for exact-output review.
+const evidence = await agent({stepId:'evidence',agentRef:args.complianceAgent,task:{product:args.product,referenceDate:args.referenceDate}});
+await humanCheckpoint({stepId:'confirm-evidence',sourceStepId:'evidence',artifactVersion:evidence.artifactVersion});
+const findings = await agent({stepId:'ingredients',agentRef:args.ingredientAgent,task:{formula:args.formula,evidence:evidence.output}});
+await humanCheckpoint({stepId:'confirm-findings',sourceStepId:'ingredients',artifactVersion:findings.artifactVersion});
+let report, feedback;
+for (let attempt=0; attempt<2; attempt++) {
+  report = await agent({stepId:`report-${attempt}`,agentRef:args.reportAgent,task:{instruction:'Return the complete report as Markdown in the output, not a file link.',evidence:evidence.output,findings:findings.output,previous:report?.output??null,feedback:feedback??null}});
+  const review = await agent({stepId:`review-${attempt}`,agentRef:args.reviewerAgent,task:{report:report.output,instruction:'Return JSON with verdict ready or needs_revision and reasons.'}});
+  if(review.output?.verdict==='ready') {
+    await humanCheckpoint({stepId:'confirm-final-report',sourceStepId:`report-${attempt}`,artifactVersion:report.artifactVersion});
+    return {status:'human_confirmed_report',report:report.output,artifactVersion:report.artifactVersion};
+  }
+  if(review.output?.verdict!=='needs_revision')throw Error('Unexpected reviewer verdict');
+  feedback=review.output;
+}
+return {status:'revision_required',report:report.output};

examples/deerflow-extension-workflows/pyproject.toml

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

examples/deerflow-extension-workflows/verify_browser.mjs

@@ -12,19 +12,30 @@ const server=createServer(async(req,res)=>{
   res.setHeader('Content-Type','text/html');
   res.end(`<!doctype html><html><head><title>Workflow test</title></head><body><main></main><script type="module">
     import plugin from '/index.mjs';
-    window.calls=[];window.opened=[];window.state='running';window.hidden=false;window.failList=0;window.failGet=0;window.loseLaunch=0;window.identities=new Set();
+    window.calls=[];window.opened=[];window.state='running';window.hidden=false;window.failList=0;window.failGet=0;window.loseLaunch=0;window.identities=new Set();window.savedDefinitions=[];window.definitionScript='';
     let saved;
     window.view=plugin.surfaces[0].mount(document.querySelector('main'),{locale:'en',signal:new AbortController().signal,
       openConversation:async id=>window.opened.push(id),
       callBackend:async(action,payload)=>{
         calls.push({action,payload});
+        if(action==='definitions')return {definitions:window.savedDefinitions};
+        if(action==='save_definition'){const item={name:payload.name,version:'saved-version'};window.savedDefinitions=[item];window.definitionScript=payload.script;return item;}
+        if(action==='get_definition'&&window.failDefinition)throw Error('Definition lookup failed');
+        if(action==='get_definition')return {...payload,script:window.definitionScript};
+        const bytes=new TextEncoder().encode('Exact <img src=x onerror=alert(1)> file');
+        const hash=Array.from(new Uint8Array(await crypto.subtle.digest('SHA-256',bytes)),b=>b.toString(16).padStart(2,'0')).join('');
+        if(action==='get_artifact')return {version:'artifact-v1',sha256:hash,offset:0,next_offset:null,content_base64:btoa(new TextDecoder().decode(bytes))};
+        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==='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');
         if(action==='get' && window.failGet-- > 0)throw Error('Inspection refresh failed');
         if(action==='list')return {runs:saved&&!window.hidden?[{run_id:'fixture-run',status:window.state}]:[]};
         if(action==='cancel'){window.state='cancelled';return {status:'cancelling'};}
         if(action==='delete'){window.hidden=true;return {deleted:true};}
-        if(action==='get')return {run_id:'fixture-run',status:window.state,script:saved.script,args:saved.args,script_hash:'script-version',args_hash:'args-version',limits:{concurrency:2,max_calls:8},steps:[{step_id:'check',agent_ref:'lead_agent',status:window.state,thread_id:'child-thread',run_id:'child-run'}],logs:['Checking'],result:'<img src=x onerror=alert(1)>'};
+        if(action==='get')return {fingerprint:'fingerprint',checkpoints:(window.state==='waiting_approval'||(window.fileReview&&window.state==='completed'))?[{step_id:'confirm',attempt:1,kind:'humanCheckpoint',status:window.state==='completed'?'approved':'pending',source_step_id:'check',artifact_version:'artifact-v1'}]:window.state==='paused'?[{step_id:'p',attempt:1,kind:'pause',status:'pending'}]:[],run_id:'fixture-run',status:window.state,script:saved.script,args:saved.args,script_hash:'script-version',args_hash:'args-version',limits:{concurrency:2,max_calls:8},steps:[{step_id:'check',agent_ref:'lead_agent',status:window.state,thread_id:'child-thread',run_id:'child-run'}],logs:['Checking'],result:'<img src=x onerror=alert(1)>'};
       }});
   </script></body></html>`);
 });
@@ -32,7 +43,7 @@ await new Promise(resolve=>server.listen(0,'127.0.0.1',resolve));
 const browser=await chromium.launch({headless:true,...(process.env.PLAYWRIGHT_CHANNEL?{channel:process.env.PLAYWRIGHT_CHANNEL}:{})});
 try{
   const page=await browser.newPage({viewport:{width:1280,height:1000}});
-  const errors=[];page.on('pageerror',e=>errors.push(e.message));
+  const errors=[];page.on('pageerror',e=>{errors.push(e.message);console.error(e.message);});
   await page.goto(`http://127.0.0.1:${server.address().port}/`);
   await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
   await page.getByRole('heading',{name:'fixture-run · running',exact:true}).waitFor();
@@ -68,10 +79,64 @@ try{
     await page.waitForFunction(()=>!document.querySelector('button[type=submit]').disabled);
     assert.equal(await page.evaluate(()=>identities.size),2,failure);
   }
+  await page.evaluate(()=>window.state='waiting_approval');
+  await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
+  await page.getByRole('button',{name:'Review exact output',exact:true}).click();
+  await page.getByRole('button',{name:'Approve this version',exact:true}).click();
+  await page.getByRole('heading',{name:'fixture-run · completed',exact:true}).waitFor();
+  const decision=await page.evaluate(()=>calls.find(c=>c.action==='decide').payload);
+  assert.equal(decision.artifact_version,'artifact-v1');assert.equal(decision.attempt,1);
+  await page.evaluate(()=>{window.state='waiting_approval';window.fileReview=true;window.badHash=true;});
+  await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
+  await page.getByRole('button',{name:'Review exact output',exact:true}).click();
+  await page.waitForFunction(()=>document.querySelector('[role=status]').textContent.includes('mismatch'));
+  assert.equal(await page.getByRole('button',{name:'Approve this version',exact:true}).count(),0);
+  await page.evaluate(()=>window.badHash=false);
+  await page.getByRole('button',{name:'Review exact output',exact:true}).click();
+  await page.getByRole('link',{name:'Download this exact version',exact:true}).waitFor();
+  assert.equal(await page.locator('img').count(),0);
+  await page.getByRole('button',{name:'Approve this version',exact:true}).click();
+  await page.getByRole('heading',{name:'fixture-run · completed',exact:true}).waitFor();
+  await page.getByRole('button',{name:'Review exact output',exact:true}).click();
+  const retainedDownload=page.getByRole('link',{name:'Download this exact version',exact:true});
+  await retainedDownload.waitFor();
+  assert.equal(await retainedDownload.evaluate(async a=>await (await fetch(a.href)).text()),'Exact <img src=x onerror=alert(1)> file');
+  assert.equal(await page.getByRole('button',{name:'Approve this version',exact:true}).count(),0);
+  await page.getByRole('button',{name:'fixture- · completed',exact:true}).click();
+  await page.getByRole('button',{name:'Review exact output',exact:true}).click();
+  await retainedDownload.waitFor();
+  assert.equal(await retainedDownload.evaluate(async a=>await (await fetch(a.href)).text()),'Exact <img src=x onerror=alert(1)> file');
+  await page.evaluate(()=>window.state='paused');
+  await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
+  await page.getByRole('button',{name:'Resume this checkpoint',exact:true}).click();
+  await page.getByRole('heading',{name:'fixture-run · completed',exact:true}).waitFor();
+  assert.equal(await page.evaluate(()=>calls.find(c=>c.action==='resume').payload.step_id),'p');
+  await page.getByLabel('Workflow name',{exact:true}).fill('inspection');
+  await page.getByRole('button',{name:'Save workflow version',exact:true}).click();
+  await page.getByRole('option',{name:'inspection · saved-versio',exact:true}).waitFor({state:'attached'});
+  await page.getByLabel('Saved workflow version',{exact:true}).selectOption(JSON.stringify({name:'inspection',version:'saved-version'}));
+  await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
+  await page.waitForFunction(()=>calls.some(c=>c.action==='launch'&&c.payload.definition?.version==='saved-version'));
+  assert.equal(await page.evaluate(()=>calls.filter(c=>c.action==='launch').at(-1).payload.script),undefined);
+  await page.getByLabel('Script',{exact:true}).fill('return "previous editor";');
+  await page.evaluate(()=>window.failDefinition=true);
+  await page.getByLabel('Saved workflow version',{exact:true}).selectOption(JSON.stringify({name:'inspection',version:'saved-version'}));
+  await page.waitForFunction(()=>document.querySelector('[role=status]').textContent.includes('Definition lookup failed'));
+  assert.equal(await page.getByRole('button',{name:'Launch workflow',exact:true}).isDisabled(),true);
+  const beforeFailedSubmit=await page.evaluate(()=>calls.filter(c=>c.action==='launch').length);
+  await page.locator('form').evaluate(form=>form.dispatchEvent(new Event('submit',{bubbles:true,cancelable:true})));
+  assert.equal(await page.evaluate(()=>calls.filter(c=>c.action==='launch').length),beforeFailedSubmit);
+  await page.getByLabel('Saved workflow version',{exact:true}).selectOption('');
+  assert.equal(await page.getByRole('button',{name:'Launch workflow',exact:true}).isDisabled(),false);
+  await page.getByLabel('Saved workflow version',{exact:true}).selectOption(JSON.stringify({name:'inspection',version:'saved-version'}));
+  await page.waitForFunction(()=>document.querySelector('button[type=submit]').disabled);
+  await page.getByLabel('Script',{exact:true}).fill('return "explicit edit";');
+  await page.getByRole('button',{name:'Launch workflow',exact:true}).click();
+  await page.waitForFunction(()=>calls.filter(c=>c.action==='launch').at(-1).payload.script==='return "explicit edit";');
   await page.evaluate(()=>view.dispose());
   assert.equal(await page.locator('.script-workflows').count(),0);
   assert.deepEqual(errors,[]);
-  console.log('Browser checks passed: launch, versioned inspection, child navigation, cancellation, escaped output, deletion and disposal.');
+  console.log('Browser checks passed: launch retries, saved versions, exact-output review, approval, resume, child navigation, cancellation, escaped output, deletion and disposal.');
 }finally{
   await browser.close();await new Promise(resolve=>server.close(resolve));
 }

frontend/src/core/i18n/locales/en-US.ts

@@ -43,7 +43,7 @@ export const enUS: Translations = {
       workflows: {
         title: "Script workflows",
         description:
-          "Run inspectable Agent workflows with bounded parallel steps, progress and cancellation. Phase 1 has no resume, human approval or hard monetary budget.",
+          "Save versioned Agent workflows, resume within the same Gateway process and review exact retained output versions. No restart replay or hard monetary budget.",
       },
       agentTeams: {
         title: "Agent teams",

frontend/src/core/i18n/locales/zh-CN.ts

@@ -43,7 +43,7 @@ export const zhCN: Translations = {
       workflows: {
         title: "脚本工作流",
         description:
-          "通过可检查的脚本编排 Agent，支持受限并发、进度与取消。第一阶段不支持恢复、人工审批或硬金额预算。",
+          "保存版本化 Agent 工作流，在同一 Gateway 进程内恢复，并确认确切的保留输出版本。不支持重启回放或硬金额预算。",
       },
       agentTeams: {
         title: "Agent 团队",

frontend/tests/e2e/bookmark-plugin.spec.ts

@@ -485,8 +485,8 @@ for (const locale of ["en-US", "zh-CN"]) {
     await expect(workflow).toHaveCount(1);
     await expect(workflow).toContainText(
       zh
-        ? "不支持恢复、人工审批或硬金额预算"
-        : "no resume, human approval or hard monetary budget",
+        ? "不支持重启回放或硬金额预算"
+        : "No restart replay or hard monetary budget",
     );
     await workflow
       .getByRole("button", {

frontend/tests/unit/core/extensions/catalog.test.ts

@@ -111,7 +111,7 @@ test("preserves custom deployment extensions and module failures", () => {
   );
 });
 
-test("workflow metadata is localized, explains Phase 1 limits and merges its registration", () => {
+test("workflow metadata is localized, explains Phase 2 limits and merges its registration", () => {
   const workflow: LoadedContribution = {
     ...contribution,
     namespace: "community.workflows",
@@ -122,9 +122,9 @@ test("workflow metadata is localized, explains Phase 1 limits and merges its reg
     [
       enUS.extensions.catalog,
       "Script workflows",
-      "no resume, human approval or hard monetary budget",
+      "No restart replay or hard monetary budget",
     ],
-    [zhCN.extensions.catalog, "脚本工作流", "不支持恢复、人工审批或硬金额预算"],
+    [zhCN.extensions.catalog, "脚本工作流", "不支持重启回放或硬金额预算"],
   ] as const) {
     const catalogOnly = extensionDirectory([], copy).find(
       (entry) => entry.package === "deerflow-extension-workflows",