durable stopped workflow runs and resume

docs/current_roadmap.md

@@ -62,9 +62,8 @@ implementation state.
   `WorkflowBuilder.prepare_subgraph()` and `WorkflowBuilder.resume()` make the
   local runnable/resumable path available without core-runtime plumbing.
   Saved interrupting artifacts can now pause and resume through
-  `run_deployment`/`resume_run` for the duration of the MCP server process
+  `run_deployment`/`resume_run` across handler/server recreation by restoring
-  (in-memory only). Persisted resume across process restarts remains future
+  stopped checkpoints and pinned root/child artifact definitions.
-  work.
 - **Concurrent foreach**: implemented in core with explicit scheduling,
   reducer/merge semantics, item error policy, async handler batching, and
   quiescent interrupt behavior. Remaining work is polish and future reuse of
@@ -77,14 +76,14 @@ implementation state.
   metadata and compatibility patches. Scope-root commits now apply to both the
   root workflow and prepared native child scopes through the explicit
   scope/lineage commit helper.
-- **Durable run history and resume**: design is recorded in
+- **Durable run history and resume**: the design is recorded in
   [2026-05-26 durable workflow runs](./superpowers/specs/2026-05-26-durable-workflow-runs-and-resume-design.md).
-  Add a validated `RunState` storage codec and dedicated run/checkpoint store.
+  A validated `RunState` codec and dedicated run/checkpoint store now persist
-  Persist stopped snapshots for interrupted, completed, and failed executions;
+  interrupted, completed, and failed stopped snapshots. Stable `run_id` values
-  replace process-local resume handles with stable `run_id` values; add compact
+  support compact `inspect_run` and bounded `read_run_trace` reads. Resume
-  `inspect_run` and bounded `read_run_trace` APIs. Resuming an interrupted run
+  revalidates its pinned dependency environment and reports `blocked` without
-  must revalidate its pinned dependency environment; ordinary live tool/source
+  consuming input when a required source is unavailable. Ordinary live
-  failures remain failed runs, not implicit pauses.
+  tool/source failures remain failed runs, not implicit pauses.
 - **Protocol-native long-running runs**: investigate MCP tasks/progress
   notifications for long-running workflow execution. Avoid inventing a custom
   "start" convention unless protocol-native behavior is insufficient.
@@ -111,8 +110,9 @@ Frame stress points remaining for native subgraphs and future fork/gather:
 ## Why This Order
 
 The MCP workflow authoring path is now usable enough for real testing. The next
-bottleneck is runtime/platform correctness: durable resume/run history,
+bottleneck is runtime/platform correctness: optional per-use-site child
-optional per-use-site child deployment overrides, and protocol-native progress
+deployment overrides, protocol-native progress reporting, and stronger durable
-reporting. Concurrent foreach, native saved child execution, and process-local
+run operations beyond stopped checkpoints. Concurrent foreach, native saved
-interrupt resume now supply scheduler/lineage precedent. Those remaining pieces
+child execution, and durable interrupt resume now supply scheduler/lineage
-should come before adding more high-level authoring sugar.
+precedent. Those remaining pieces should come before adding more high-level
+authoring sugar.

docs/superpowers/specs/2026-05-24-native-subgraphs-design.md

@@ -1,7 +1,7 @@
 # Native Subgraphs Design
 
-Status: prepared-child execution, saved-artifact resolution, and process-local
+Status: prepared-child execution, saved-artifact resolution, and durable
-interrupt resume implemented; durable resume specified separately
+stopped-run interrupt resume implemented
 
 Native subgraphs should make a workflow usable as a workflow step without
 collapsing the child run into one opaque Python node call. The current
@@ -409,10 +409,11 @@ child artifact twice against different accounts or capability bindings.
 ### Saved Child Interrupt Status
 
 Native prepared children can interrupt and resume in core. The workflow surface
-now exposes process-local `run_deployment` / `resume_run` support for saved
+now exposes durable `run_deployment` / `resume_run` support for saved
 interrupting artifacts, including interrupts raised by saved descendants.
-That support does not survive process restart yet. Durable run persistence is
+Stopped checkpoints pin the deployment and exact root/child artifact
-specified separately in
+definitions; dependency drift can block resume without mutating the stopped
+execution. Durable run persistence is specified in
 [`2026-05-26-durable-workflow-runs-and-resume-design.md`](2026-05-26-durable-workflow-runs-and-resume-design.md).
 
 ## Implementation Slices
@@ -462,8 +463,8 @@ specified separately in
 - Dependencies, missing artifacts, and saved-child cycles validate before
   execution.
 - Tests cover saved child execution through deployment bindings, nested saved
-  child dependencies, missing/cyclic diagnostics, and process-local
+  child dependencies, missing/cyclic diagnostics, and durable interrupt/resume
-  interrupt/resume for saved children.
+  for saved children.
 
 ### Slice 4: Optional Policy Expansion
 
@@ -491,12 +492,8 @@ specified separately in
 
 ## Recommendation
 
-The typed boundary scaffold, prepared-child runtime, and routed interrupt
+The typed boundary scaffold, prepared-child runtime, saved-child resolution,
-resume are complete. Implement Slice 3 in the platform layer: prepare saved
+and durable routed interrupt resume are complete. Do not delete wrapper-node
-non-interrupting child artifacts recursively under one deployment environment
+helpers yet; they remain compatibility APIs while saved native execution
-and pass those prepared dependencies into core execution.
+matures. Next policy work is optional per-use-site child deployment overrides
-
+and richer protocol-native long-running progress/reporting.
-The process-local saved-interrupt execution path is complete. Next add durable
-run/checkpoint persistence so interrupted saved children survive process
-restart. Do not delete wrapper-node helpers yet; they remain compatibility APIs
-while saved native execution matures.

docs/superpowers/specs/2026-05-26-durable-workflow-runs-and-resume-design.md

@@ -1,6 +1,6 @@
 # Durable Workflow Runs and Resume Design
 
-Status: design approved for implementation planning
+Status: v1 implemented; future protocol-native progress and broader recovery remain
 
 Durable workflow runs turn the current process-local `run_deployment` /
 `resume_run` behavior into platform state. The runtime already exposes the
@@ -376,16 +376,16 @@ Once stopped-run persistence is stable:
 5. Add general cross-run memory separately if nodes need it; it is not a
    replacement for checkpoints.
 
-## Recommendation
+## Implemented V1
 
-Implement durable stopped-run snapshots first:
+Durable stopped-run snapshots now provide:
 
-1. Add a validated persisted `RunState` codec.
+1. A validated persisted `RunState` codec.
-2. Add `WorkflowRun`, `RunCheckpoint`, and `RunStore`.
+2. `WorkflowRunRecord`, `RunCheckpoint`, and `RunStore`.
-3. Save interrupted, completed, and failed runs after start/resume returns.
+3. Checkpoints for interrupted, completed, and failed public executions.
-4. Replace process-local `_active_runs` lookup with durable run retrieval.
+4. Durable resume retrieval instead of process-local `_active_runs`.
-5. Add compact `inspect_run` and bounded `read_run_trace`.
+5. Compact `inspect_run` and bounded `read_run_trace` tools.
-6. Revalidate pinned dependencies before resuming interrupted runs.
+6. Pinned dependency revalidation before interrupted runs resume.
 
 This delivers durable human-in-the-loop execution and run inspection while
 preserving the correctness boundary that live external-call failures are not

docs/wf_core_architecture.md

@@ -143,8 +143,8 @@ limits and intended adapter seam.
   the parent routes by the child's terminal workflow outcome. Saved/deployed
   workflow resolution remains outside core; the workflow platform can now
   supply saved child artifacts as prepared dependencies using one inherited
-  deployment binding environment, including process-local pause/resume for
+  deployment binding environment, including durable stopped-run pause/resume
-  child interrupts. For local authoring,
+  for child interrupts. For local authoring,
   `WorkflowBuilder.prepare_subgraph()`
   registers a child builder and `WorkflowBuilder.resume()` continues a paused
   prepared-child interrupt without requiring direct core-runtime calls.

docs/wf_mcp_operator_manual.md

@@ -182,6 +182,11 @@ Primary:
 - `wf.workflow.run_deployment`: execute a saved deployment with input. The
   default response is compact and returns `trace_count`; pass `trace_range`
   only when debugging a failed or surprising run.
+- `wf.workflow.inspect_run`: inspect a durable stopped run without trace detail.
+- `wf.workflow.read_run_trace`: retrieve only an explicit bounded debug trace
+  slice for a durable run.
+- `wf.workflow.resume_run`: resume an interrupted durable run when its pinned
+  dependencies remain available.
 
 Advanced:
 
@@ -357,6 +362,13 @@ response. Trace entries may include resolved node inputs, node outputs, and
 state changes, so treat them as debug payloads rather than ordinary list/summary
 data.
 
+Every started deployment receives a durable `run_id`, including completed and
+failed runs. Use `inspect_run` for the compact stored result and
+`read_run_trace` only for an explicit bounded debug range. Interrupted runs can
+be resumed after server/handler recreation; if a pinned source is missing or
+disabled, `resume_run` returns `resume_readiness="blocked"` without advancing
+the execution checkpoint.
+
 ## Which Tool Do I Use?
 
 | I want to... | Use |
@@ -383,6 +395,9 @@ data.
 | Bind a saved workflow to concrete sources | `wf.workflow.save_deployment` |
 | Check whether a deployment can run | `wf.workflow.validate_deployment` |
 | Execute a saved workflow | `wf.workflow.run_deployment` |
+| Inspect a stopped workflow run | `wf.workflow.inspect_run` |
+| Read bounded debug trace entries | `wf.workflow.read_run_trace` |
+| Resume an interrupted workflow run | `wf.workflow.resume_run` |
 
 ## Common Confusions
 

docs/wf_mcp_troubleshooting.md

@@ -305,7 +305,8 @@ The deployment paused at an interrupt node. The response should include:
 ```text
 status: interrupted
 outcome: null
-run_id: <process-local id>
+run_id: <durable id>
+resume_readiness: ready
 interrupt: <request payload and metadata>
 ```
 
@@ -315,11 +316,17 @@ Send the requested resume payload to:
 wf.workflow.resume_run
 ```
 
-The `run_id` is intentionally process-local. It is valid only while the current
+The `run_id` identifies a stored stopped-state checkpoint and survives handler
-MCP server process keeps the paused run in memory. If the server restarts,
+or server recreation. Before applying the resume payload, the platform
-there is no durable run store yet; rerun the deployment from the beginning.
+revalidates the pinned deployment/source environment. If it returns
-After resume completes, `status` is `completed` and `outcome` reports the
+`resume_readiness: blocked`, inspect the diagnostics, restore the missing or
-workflow terminal outcome such as `ok` or `error`.
+disabled source, and call `resume_run` again; the blocked attempt has not
+advanced workflow state. After resume completes, `status` is `completed` and
+`outcome` reports the workflow terminal outcome such as `ok` or `error`.
+
+For debugging a completed, failed, or interrupted run, call
+`wf.workflow.inspect_run` first. Only call `wf.workflow.read_run_trace` with a
+small explicit range when node-level detail is necessary.
 
 ## A Raw MCP Tool Works But The Workflow Version Is Awkward
 

docs/workflow_artifacts.md

@@ -70,12 +70,10 @@ Dynamic projection of saved workflows as individual MCP tools can exist later,
 but it should be optional. The stable run tool is the reliable base layer.
 
 Current `run_deployment` calls are synchronous request/response executions until
-the workflow pauses or completes. They return compact execution status,
+the workflow pauses or completes. They return a durable `run_id`, compact
-terminal workflow outcome, output, diagnostics, and `trace_count`; optional
+execution status, terminal workflow outcome when available, output, diagnostics,
-ranged trace detail is for debugging
+and `trace_count`; optional ranged trace detail is for debugging only. If a run
-only. If a run pauses at an interrupt, the response includes a process-local
+pauses at an interrupt, use `wf.workflow.resume_run` with that `run_id`.
-`run_id` and interrupt payload. Use `wf.workflow.resume_run` with that `run_id`
-to continue while the same MCP server process is alive.
 
 Non-interrupting saved workflow children can now execute natively through this
 deployment surface. A parent deployment resolves its saved descendants by exact
@@ -86,21 +84,21 @@ the subgraph use site rather than only the child artifact id.
 
 Interrupting saved artifacts can pause and resume through the current
 deployment surface, including interrupts raised inside saved child workflows.
-This support is in-memory only: server restart, reload that replaces process
+Stopped snapshots pin the deployment, root artifact, and saved child artifact
-state, or another frontend process invalidates the `run_id`.
+definitions so handler/server recreation does not invalidate the `run_id`.
 
 Durable run history is specified in
 [`2026-05-26-durable-workflow-runs-and-resume-design.md`](superpowers/specs/2026-05-26-durable-workflow-runs-and-resume-design.md).
-It should replace process-local `run_id` values with durable run records. The
+The implemented surface is:
-planned surface is:
 
 - `run_deployment` starts or completes a run and returns `run_id`
 - `inspect_run(run_id)` returns status, output, diagnostics, and trace metadata
 - `read_run_trace(run_id, range)` returns bounded trace slices
 
-Until that exists, clients should treat the current response as the complete
+Before applying a resume payload, the pinned dependency environment is
-ephemeral run result for this request, or as a process-local resume handle when
+revalidated. If it is unavailable or incompatible, the run stays
-`status` is `interrupted`.
+`interrupted`, returns `resume_readiness="blocked"` and diagnostics, and does
+not append a new execution checkpoint.
 
 For long-running workflow execution, prefer MCP-native execution mechanisms
 where available:
@@ -686,12 +684,10 @@ native child pause and builder-driven resume. In the wrapper example the
 parent trace sees one node call; in the native examples child trace entries
 remain in the parent run state.
 
-Persisted resume is still not implemented. In-memory resume works because the
+Durable resume persists stopped snapshots for interrupted, completed, and
-server keeps the paused `RunState`; the durable run design persists stopped
+failed runs and pins root workflow, prepared child dependencies, deployment
-snapshots for interrupted, completed, and failed runs and pins root workflow,
+bindings, and trace metadata. Only declared interrupts become resumable
-prepared child dependencies, deployment bindings, and trace metadata before a
+pauses; live tool/source failures remain failures.
-paused run can survive restart or move across processes. Only declared
-interrupts become resumable pauses; live tool/source failures remain failures.
 
 Blocking dependency failures happen before workflow execution and are not normal
 workflow outcomes. A missing source, disabled source, unresolved binding, or

docs/workflow_drafts.md

@@ -446,11 +446,11 @@ provided on resume back into workflow state. Older map-shaped `request` and
 `resume` values are accepted only as parse compatibility and dump back to the
 canonical list shape.
 
-Saved interrupting artifacts can pause and resume through deployment runs while
+Saved interrupting artifacts can pause and resume through deployment runs. The
-the MCP server process stays alive. The run response includes a process-local
+run response includes a durable `run_id`; pass that to
-`run_id`; pass that to `wf.workflow.resume_run` with the resume payload.
+`wf.workflow.resume_run` with the resume payload. Before advancing a resumed
-Persisted run storage is still future work, so a server restart invalidates
+run, the platform revalidates its pinned dependency environment and can return
-that in-memory run id.
+`resume_readiness="blocked"` without consuming input.
 
 ### `end`
 

src/wf_artifacts/__init__.py

@@ -39,6 +39,17 @@ from .refs import (
     workflow_ref_from_capability,
 )
 from .store import FileWorkflowArtifactStore, WorkflowArtifactStore
+from .runs import (
+    CheckpointReason,
+    FileRunStore,
+    PinnedRunEnvironment,
+    ResumeReadiness,
+    RunCheckpoint,
+    RunStore,
+    StoredRunStatus,
+    WorkflowRunRecord,
+    ensure_run_id,
+)
 from .validation import validate_deployment_dependencies
 from .references import logical_ref_for_concrete_ref, normalize_plan_node_refs
 
@@ -53,6 +64,7 @@ __all__ = [
     "DraftWorkspaceStore",
     "FileDraftWorkspaceStore",
     "FileWorkflowArtifactStore",
+    "FileRunStore",
     "RequiredCapability",
     "SourceBinding",
     "WorkflowArtifact",
@@ -60,7 +72,15 @@ __all__ = [
     "WorkflowCapabilityRef",
     "WorkflowDraftWorkspace",
     "WorkflowArtifactStore",
+    "WorkflowRunRecord",
     "WorkflowDeployment",
+    "RunStore",
+    "RunCheckpoint",
+    "CheckpointReason",
+    "PinnedRunEnvironment",
+    "ResumeReadiness",
+    "StoredRunStatus",
+    "ensure_run_id",
     "artifact_catalog_entry",
     "artifact_node_name",
     "create_draft_workspace",

src/wf_artifacts/runs/__init__.py

@@ -0,0 +1,22 @@
+from .models import (
+    CheckpointReason,
+    PinnedRunEnvironment,
+    ResumeReadiness,
+    RunCheckpoint,
+    StoredRunStatus,
+    WorkflowRunRecord,
+    ensure_run_id,
+)
+from .store import FileRunStore, RunStore
+
+__all__ = [
+    "CheckpointReason",
+    "FileRunStore",
+    "PinnedRunEnvironment",
+    "ResumeReadiness",
+    "RunCheckpoint",
+    "RunStore",
+    "StoredRunStatus",
+    "WorkflowRunRecord",
+    "ensure_run_id",
+]

src/wf_artifacts/runs/models.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from wf_core import PersistedRunState
+
+from ..models import DependencyDiagnostic, WorkflowArtifact, WorkflowDeployment
+
+RUN_ID_PATTERN = r"^[A-Za-z0-9_.-]+$"
+
+
+def ensure_run_id(run_id: str) -> str:
+    """Reject ids that cannot safely identify one local run directory."""
+    if not re.fullmatch(RUN_ID_PATTERN, run_id):
+        raise ValueError(
+            "run_id must match [A-Za-z0-9_.-]+; path separators are not allowed"
+        )
+    return run_id
+
+
+class StoredRunStatus(StrEnum):
+    """Stopped runtime statuses supported by durable run persistence."""
+
+    INTERRUPTED = "interrupted"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class ResumeReadiness(StrEnum):
+    """Whether an interrupted stored run may currently continue."""
+
+    READY = "ready"
+    BLOCKED = "blocked"
+    NOT_APPLICABLE = "not_applicable"
+
+
+class CheckpointReason(StrEnum):
+    """Why a stopped-state checkpoint was written."""
+
+    INTERRUPTED = "interrupted"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class PinnedRunEnvironment(BaseModel):
+    """Exact execution definitions captured when a run starts."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    deployment: WorkflowDeployment
+    root_artifact: WorkflowArtifact
+    child_artifacts: list[WorkflowArtifact] = Field(default_factory=list)
+
+
+class WorkflowRunRecord(BaseModel):
+    """Durable summary and pinned environment for one started workflow run."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str = Field(pattern=RUN_ID_PATTERN)
+    status: StoredRunStatus
+    resume_readiness: ResumeReadiness
+    environment: PinnedRunEnvironment
+    latest_checkpoint_id: str
+    diagnostics: list[DependencyDiagnostic] = Field(default_factory=list)
+    created_at: datetime
+    updated_at: datetime
+
+
+class RunCheckpoint(BaseModel):
+    """One stopped-state snapshot persisted at an external run boundary."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str = Field(pattern=RUN_ID_PATTERN)
+    run_id: str = Field(pattern=RUN_ID_PATTERN)
+    sequence: int = Field(ge=1)
+    reason: CheckpointReason
+    state: PersistedRunState
+    created_at: datetime

src/wf_artifacts/runs/store.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from threading import RLock
+
+from .models import RunCheckpoint, WorkflowRunRecord, ensure_run_id
+
+
+class RunStore:
+    """Persistence boundary for stopped run summaries and checkpoints."""
+
+    def save_run(self, run: WorkflowRunRecord) -> None:
+        raise NotImplementedError
+
+    def get_run(self, run_id: str) -> WorkflowRunRecord:
+        raise NotImplementedError
+
+    def list_runs(self) -> list[WorkflowRunRecord]:
+        raise NotImplementedError
+
+    def save_checkpoint(self, checkpoint: RunCheckpoint) -> None:
+        raise NotImplementedError
+
+    def get_latest_checkpoint(self, run_id: str) -> RunCheckpoint:
+        raise NotImplementedError
+
+    def list_checkpoints(self, run_id: str) -> list[RunCheckpoint]:
+        raise NotImplementedError
+
+
+class FileRunStore(RunStore):
+    """JSON file-backed stopped-run store for local development and tests."""
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self._lock = RLock()
+        self.runs_dir.mkdir(parents=True, exist_ok=True)
+
+    @property
+    def runs_dir(self) -> Path:
+        return self.root / "runs"
+
+    def save_run(self, run: WorkflowRunRecord) -> None:
+        with self._lock:
+            self._write_json(
+                self._run_path(run.id),
+                run.model_dump(mode="json"),
+            )
+
+    def get_run(self, run_id: str) -> WorkflowRunRecord:
+        path = self._run_path(run_id)
+        if not path.exists():
+            raise KeyError(f"unknown workflow run {run_id!r}")
+        return WorkflowRunRecord.model_validate_json(path.read_text(encoding="utf-8"))
+
+    def list_runs(self) -> list[WorkflowRunRecord]:
+        return [
+            WorkflowRunRecord.model_validate_json(path.read_text(encoding="utf-8"))
+            for path in sorted(self.runs_dir.glob("*/run.json"))
+        ]
+
+    def save_checkpoint(self, checkpoint: RunCheckpoint) -> None:
+        with self._lock:
+            self._write_json(
+                self._checkpoint_path(checkpoint.run_id, checkpoint.sequence),
+                checkpoint.model_dump(mode="json"),
+            )
+
+    def get_latest_checkpoint(self, run_id: str) -> RunCheckpoint:
+        checkpoints = self.list_checkpoints(run_id)
+        if not checkpoints:
+            raise KeyError(f"workflow run {run_id!r} has no checkpoints")
+        return checkpoints[-1]
+
+    def list_checkpoints(self, run_id: str) -> list[RunCheckpoint]:
+        directory = self._run_directory(run_id) / "checkpoints"
+        if not directory.exists():
+            return []
+        return [
+            RunCheckpoint.model_validate_json(path.read_text(encoding="utf-8"))
+            for path in sorted(directory.glob("*.json"))
+        ]
+
+    def _write_json(self, path: Path, payload: object) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temp_path = path.with_suffix(".json.tmp")
+        temp_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        temp_path.replace(path)
+
+    def _run_directory(self, run_id: str) -> Path:
+        safe_id = ensure_run_id(run_id)
+        root = self.runs_dir.resolve()
+        path = (self.runs_dir / safe_id).resolve()
+        if path.parent != root:
+            raise ValueError(f"run id escapes run store: {run_id!r}")
+        return path
+
+    def _run_path(self, run_id: str) -> Path:
+        return self._run_directory(run_id) / "run.json"
+
+    def _checkpoint_path(self, run_id: str, sequence: int) -> Path:
+        return self._run_directory(run_id) / "checkpoints" / f"{sequence:06d}.json"

src/wf_core/__init__.py

@@ -28,8 +28,10 @@ from .runtime import (
     WorkflowExecutionError,
     coerce_node_result,
     execute_workflow_async,
+    execute_workflow_result_async,
     execute_workflow,
     resume_workflow_async,
+    resume_workflow_result_async,
     resume_workflow,
     step_workflow_async,
     step_workflow,
@@ -45,6 +47,7 @@ from .run_state import (
     StepExecutionResult,
     TraceEntry,
 )
+from .run_codec import PersistedRunState, dump_run_state, load_run_state
 from .tokens import END, START
 from .validation import (
     ValidationIssue,
@@ -84,6 +87,9 @@ __all__ = [
     "TraceEntry",
     "InterruptRoute",
     "InterruptRequest",
+    "PersistedRunState",
+    "dump_run_state",
+    "load_run_state",
     "START",
     "END",
     "ValidationIssue",
@@ -94,8 +100,10 @@ __all__ = [
     "WorkflowExecutionError",
     "coerce_node_result",
     "execute_workflow_async",
+    "execute_workflow_result_async",
     "execute_workflow",
     "resume_workflow_async",
+    "resume_workflow_result_async",
     "resume_workflow",
     "step_workflow_async",
     "step_workflow",

src/wf_core/run_codec.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
+
+from .run_state import ROOT_SCOPE_ID, RunState
+
+
+class PersistedRunState(BaseModel):
+    """Versioned JSON storage envelope for one stopped runtime snapshot."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    version: Literal[1] = 1
+    state: dict[str, Any]
+
+
+_RUN_STATE_ADAPTER = TypeAdapter(RunState)
+
+
+def dump_run_state(run: RunState) -> dict[str, object]:
+    """Serialize one stopped `RunState` into the durable v1 envelope."""
+    return PersistedRunState(
+        state=_RUN_STATE_ADAPTER.dump_python(run, mode="json")
+    ).model_dump(mode="json")
+
+
+def load_run_state(payload: object) -> RunState:
+    """Validate and restore one durable v1 runtime snapshot.
+
+    The root scope intentionally shares the compatibility ``RunState.state``
+    dict during runtime. Serialization loses object identity, so restored
+    snapshots must recreate this alias before resumed writes occur.
+    """
+    envelope = PersistedRunState.model_validate(payload)
+    try:
+        run = _RUN_STATE_ADAPTER.validate_python(envelope.state)
+    except ValidationError as exc:
+        raise ValueError("invalid persisted workflow run state") from exc
+    root_scope = run.scopes.get(ROOT_SCOPE_ID)
+    if root_scope is not None:
+        run.state = root_scope.committed_state
+    return run

src/wf_core/runtime/__init__.py

@@ -10,8 +10,10 @@ from wf_core.runtime.ops.nodes import (
 from .engine import (
     execute_workflow,
     execute_workflow_async,
+    execute_workflow_result_async,
     resume_workflow,
     resume_workflow_async,
+    resume_workflow_result_async,
 )
 from .subgraphs import PreparedSubgraph
 from .step import complete_step, step_workflow, step_workflow_async
@@ -26,8 +28,10 @@ __all__ = [
     "execute_node_use_async",
     "execute_workflow",
     "execute_workflow_async",
+    "execute_workflow_result_async",
     "resume_workflow",
     "resume_workflow_async",
+    "resume_workflow_result_async",
     "step_workflow",
     "step_workflow_async",
     "PreparedSubgraph",

src/wf_core/runtime/engine.py

@@ -70,6 +70,32 @@ async def execute_workflow_async(
         raise
 
 
+async def execute_workflow_result_async(
+    workflow: Workflow,
+    workflow_input: dict[str, Any],
+    registry: Mapping[str, AsyncNodeHandler],
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
+) -> RunState:
+    """Execute asynchronously and return failed state instead of raising failures."""
+    run = create_run_state(workflow, workflow_input)
+
+    try:
+        prepare_new_run(workflow, workflow_input, run)
+        return await resume_workflow_async(
+            workflow,
+            run,
+            registry,
+            reducers=reducers,
+            subgraphs=subgraphs,
+        )
+    except Exception as exc:
+        run.status = RunStatus.FAILED
+        run.error = str(exc)
+        return run
+
+
 def resume_workflow(
     workflow: Workflow,
     run: RunState,
@@ -174,6 +200,33 @@ async def resume_workflow_async(
     return finalize_run(workflow, run)
 
 
+async def resume_workflow_result_async(
+    workflow: Workflow,
+    run: RunState,
+    registry: Mapping[str, AsyncNodeHandler],
+    *,
+    resume_payload: dict[str, Any] | None = None,
+    resume_outcome: str = "submitted",
+    reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
+) -> RunState:
+    """Resume asynchronously and return failed state instead of raising failures."""
+    try:
+        return await resume_workflow_async(
+            workflow,
+            run,
+            registry,
+            resume_payload=resume_payload,
+            resume_outcome=resume_outcome,
+            reducers=reducers,
+            subgraphs=subgraphs,
+        )
+    except Exception as exc:
+        run.status = RunStatus.FAILED
+        run.error = str(exc)
+        return run
+
+
 def _interrupt_resume_target(
     root_workflow: Workflow,
     root_reducers: Mapping[str, ReducerDefinition] | None,

src/wf_mcp/broker/artifact_tools.py

@@ -86,7 +86,7 @@ def register_artifact_tools(server: FastMCP, service: WfMcpService) -> None:
         resume_payload: dict[str, Any],
         resume_outcome: str = "submitted",
     ) -> dict[str, Any]:
-        """Resume a process-local interrupted deployment run."""
+        """Resume a durable interrupted deployment run."""
         return await handlers.resume_run(
             run_id=run_id,
             resume_payload=resume_payload,

src/wf_mcp/broker/config.py

@@ -3,7 +3,11 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
-from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
+from wf_artifacts import (
+    FileDraftWorkspaceStore,
+    FileRunStore,
+    FileWorkflowArtifactStore,
+)
 
 from ..control import BrokerConfigFile
 from ..models import BrokerConfig
@@ -27,6 +31,7 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
         store=FileStore(config.store_root),
         artifact_store=FileWorkflowArtifactStore(config.store_root),
         draft_workspace_store=FileDraftWorkspaceStore(config.store_root),
+        run_store=FileRunStore(config.store_root),
         # Discovery can use short-lived SDK sessions. Workflow execution needs
         # a persistent runtime so stateful MCP servers keep session/page state
         # across sequential workflow nodes.

src/wf_mcp/broker/service/core.py

@@ -10,7 +10,9 @@ from pydantic import BaseModel
 from wf_artifacts import (
     DraftWorkspaceStore,
     FileDraftWorkspaceStore,
+    FileRunStore,
     FileWorkflowArtifactStore,
+    RunStore,
     WorkflowArtifact,
     WorkflowArtifactCatalogEntry,
     WorkflowArtifactStore,
@@ -22,8 +24,8 @@ from wf_core import (
     NodeUse,
     RunState,
     Workflow,
-    execute_workflow_async,
+    execute_workflow_result_async,
-    resume_workflow_async,
+    resume_workflow_result_async,
 )
 
 from wf_platform import (
@@ -55,6 +57,7 @@ from ...storage import Store
 from ...workflow.wrappers import _model_from_schema
 from ...workflow_surface.runtime_dependencies import resolve_runtime_dependencies
 from ...workflow_surface.saved_subgraphs import (
+    SavedSubgraphTree,
     prepare_saved_subgraphs,
     resolve_saved_subgraph_tree,
 )
@@ -83,6 +86,7 @@ class WfMcpService:
     include_builtin_specs: bool = True
     artifact_store: WorkflowArtifactStore | None = None
     draft_workspace_store: DraftWorkspaceStore | None = None
+    run_store: RunStore | None = None
     tool_executor: ToolExecutor | None = None
 
     def __post_init__(self) -> None:
@@ -93,6 +97,8 @@ class WfMcpService:
             self.draft_workspace_store = FileDraftWorkspaceStore(
                 _store_root(self.store)
             )
+        if self.run_store is None:
+            self.run_store = FileRunStore(_store_root(self.store))
         if self.include_builtin_specs:
             for source in builtin_sources().values():
                 self.register_capability_source(source)
@@ -684,6 +690,7 @@ class WfMcpService:
         *,
         deployment: WorkflowDeployment | None,
         artifact: WorkflowArtifact | None,
+        saved_subgraph_tree: SavedSubgraphTree | None = None,
     ) -> tuple[Workflow, dict[str, Any], dict[str, Any], dict[str, Any]]:
         """Resolve bindings once into the executable pieces core expects.
 
@@ -710,7 +717,15 @@ class WfMcpService:
             plan_node_names=plan_node_names,
         )
         prepared_subgraphs = {}
-        if artifact is not None and self.artifact_store is not None:
+        if saved_subgraph_tree is not None:
+            tree = saved_subgraph_tree
+            prepared_subgraphs = prepare_saved_subgraphs(
+                tree=tree,
+                deployment=deployment,
+                sources=self.capability_sources,
+                compile_plan=self.compile_plan,
+            )
+        elif artifact is not None and self.artifact_store is not None:
             tree = resolve_saved_subgraph_tree(
                 root_artifact=artifact,
                 artifact_store=self.artifact_store,
@@ -735,6 +750,7 @@ class WfMcpService:
         workflow_input: dict[str, Any],
         deployment: WorkflowDeployment | None = None,
         artifact: WorkflowArtifact | None = None,
+        saved_subgraph_tree: SavedSubgraphTree | None = None,
     ):
         self._record_event(
             make_event(
@@ -748,9 +764,10 @@ class WfMcpService:
                 plan,
                 deployment=deployment,
                 artifact=artifact,
+                saved_subgraph_tree=saved_subgraph_tree,
             )
         )
-        run = await execute_workflow_async(
+        run = await execute_workflow_result_async(
             workflow,
             workflow_input,
             registry,
@@ -775,16 +792,18 @@ class WfMcpService:
         resume_outcome: str = "submitted",
         deployment: WorkflowDeployment | None = None,
         artifact: WorkflowArtifact | None = None,
+        saved_subgraph_tree: SavedSubgraphTree | None = None,
     ) -> RunState:
-        """Resume one in-memory run using freshly resolved runtime dependencies."""
+        """Resume one stopped run using its prepared runtime dependency boundary."""
         workflow, registry, reducers, prepared_subgraphs = (
             self._prepare_workflow_runtime(
                 plan,
                 deployment=deployment,
                 artifact=artifact,
+                saved_subgraph_tree=saved_subgraph_tree,
             )
         )
-        resumed = await resume_workflow_async(
+        resumed = await resume_workflow_result_async(
             workflow,
             run,
             registry,

src/wf_mcp/workflow_surface/handlers.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
-from dataclasses import asdict, dataclass
+from dataclasses import asdict
 from typing import TYPE_CHECKING, Any
-from uuid import uuid4
 
 from wf_artifacts import (
     ArtifactKind,
@@ -12,6 +11,7 @@ from wf_artifacts import (
     DependencyDiagnostic,
     DraftWorkspaceStore,
     RequiredCapability,
+    RunStore,
     WorkflowArtifact,
     WorkflowCapabilityRef,
     WorkflowDeployment,
@@ -53,10 +53,21 @@ from .constants import (
 from .models import TraceRange
 from .refs import parse_workflow_surface_capability_id
 from .saved_subgraphs import (
+    SavedSubgraphTree,
     direct_wrapper_interrupt_diagnostic,
     resolve_saved_subgraph_tree,
+    saved_subgraph_tree_from_snapshots,
     validate_saved_subgraph_tree,
 )
+from .run_lifecycle import (
+    create_pinned_environment,
+    has_blocking_diagnostics,
+    mark_resume_blocked,
+    persist_stopped_run,
+    load_stored_run,
+    restore_interrupted_run,
+    validate_pinned_resume_environment,
+)
 from .wrapper_hints import wrapper_hints_for_capability
 
 if TYPE_CHECKING:
@@ -65,27 +76,11 @@ if TYPE_CHECKING:
     from ..broker.service import WfMcpService
 
 
-@dataclass(slots=True)
-class ActiveWorkflowRun:
-    """In-memory paused deployment run.
-
-    This is intentionally not durable. It only makes interrupt/resume usable
-    while the current MCP server process is alive; persisted run storage remains
-    a separate platform concern.
-    """
-
-    deployment: WorkflowDeployment
-    artifact: WorkflowArtifact
-    plan: RawWorkflowPlan
-    run: "RunState"
-
-
 class WorkflowSurfaceHandlers:
     """Reusable implementation behind MCP workflow artifact tools."""
 
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
-        self._active_runs: dict[str, ActiveWorkflowRun] = {}
 
     async def list_artifacts(
         self,
@@ -903,7 +898,9 @@ class WorkflowSurfaceHandlers:
         }
 
     async def validate_deployment(self, *, deployment_id: str) -> dict[str, Any]:
-        deployment, artifact, diagnostics = self._deployment_validation(deployment_id)
+        deployment, artifact, diagnostics, _tree = self._deployment_validation(
+            deployment_id
+        )
         return {
             "deployment_id": deployment.id,
             "artifact_id": artifact.id,
@@ -921,7 +918,9 @@ class WorkflowSurfaceHandlers:
         workflow_input: dict[str, Any],
         trace_range: TraceRange | None = None,
     ) -> dict[str, Any]:
-        deployment, artifact, diagnostics = self._deployment_validation(deployment_id)
+        deployment, artifact, diagnostics, tree = self._deployment_validation(
+            deployment_id
+        )
         if diagnostics:
             return _run_payload(
                 deployment=deployment,
@@ -936,18 +935,23 @@ class WorkflowSurfaceHandlers:
             workflow_input,
             deployment=deployment,
             artifact=artifact,
+            saved_subgraph_tree=tree,
         )
-        run_id = self._save_active_run(
+        record = persist_stopped_run(
+            store=self._run_store(),
+            environment=create_pinned_environment(
                 deployment=deployment,
                 artifact=artifact,
-            plan=plan,
+                tree=tree,
+            ),
             run=run,
         )
         return _run_payload(
             deployment=deployment,
             artifact=artifact,
             status=run.status.value,
-            run_id=run_id,
+            run_id=record.id,
+            resume_readiness=record.resume_readiness.value,
             interrupt=_interrupt_payload(run),
             outcome=run.outcome,
             output=run.output,
@@ -978,29 +982,54 @@ class WorkflowSurfaceHandlers:
         resume_outcome: str = "submitted",
         trace_range: TraceRange | None = None,
     ) -> dict[str, Any]:
-        """Resume one interrupted in-memory deployment run."""
+        """Resume one durable interrupted deployment run."""
-        active = self._active_runs[run_id]
+        record, stopped_run = restore_interrupted_run(self._run_store(), run_id)
+        environment = record.environment
+        diagnostics = validate_pinned_resume_environment(
+            record=record,
+            sources=_available_sources(self.service),
+        )
+        if has_blocking_diagnostics(diagnostics):
+            blocked = mark_resume_blocked(
+                store=self._run_store(),
+                record=record,
+                diagnostics=diagnostics,
+            )
+            return _run_payload(
+                deployment=environment.deployment,
+                artifact=environment.root_artifact,
+                status=stopped_run.status.value,
+                run_id=blocked.id,
+                resume_readiness=blocked.resume_readiness.value,
+                interrupt=_interrupt_payload(stopped_run),
+                outcome=stopped_run.outcome,
+                output=stopped_run.output,
+                diagnostics=diagnostics,
+                trace_count=len(stopped_run.trace),
+            )
+        plan = _raw_plan_from_artifact(environment.root_artifact)
+        tree = saved_subgraph_tree_from_snapshots(environment.child_artifacts)
         run = await self.service.resume_workflow_from_plan(
-            active.plan,
+            plan,
-            active.run,
+            stopped_run,
             resume_payload=resume_payload,
             resume_outcome=resume_outcome,
-            deployment=active.deployment,
+            deployment=environment.deployment,
-            artifact=active.artifact,
+            artifact=environment.root_artifact,
+            saved_subgraph_tree=tree,
         )
-        active.run = run
+        next_record = persist_stopped_run(
-        next_run_id = self._save_active_run(
+            store=self._run_store(),
-            deployment=active.deployment,
+            environment=environment,
-            artifact=active.artifact,
-            plan=active.plan,
             run=run,
             run_id=run_id,
         )
         return _run_payload(
-            deployment=active.deployment,
+            deployment=environment.deployment,
-            artifact=active.artifact,
+            artifact=environment.root_artifact,
             status=run.status.value,
-            run_id=next_run_id,
+            run_id=next_record.id,
+            resume_readiness=next_record.resume_readiness.value,
             interrupt=_interrupt_payload(run),
             outcome=run.outcome,
             output=run.output,
@@ -1023,33 +1052,62 @@ class WorkflowSurfaceHandlers:
             ),
         )
 
-    def _save_active_run(
+    async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
+        """Return one durable stopped-run summary without debug trace entries."""
+        record, run = load_stored_run(self._run_store(), run_id)
+        environment = record.environment
+        return _run_payload(
+            deployment=environment.deployment,
+            artifact=environment.root_artifact,
+            status=record.status.value,
+            run_id=record.id,
+            resume_readiness=record.resume_readiness.value,
+            interrupt=_interrupt_payload(run),
+            outcome=run.outcome,
+            output=run.output,
+            diagnostics=record.diagnostics,
+            trace_count=len(run.trace),
+        )
+
+    async def read_run_trace(
         self,
         *,
-        deployment: WorkflowDeployment,
+        run_id: str,
-        artifact: WorkflowArtifact,
+        trace_range: TraceRange,
-        plan: RawWorkflowPlan,
+    ) -> dict[str, Any]:
-        run: RunState,
+        """Return only a caller-bounded debug trace slice from a stopped run."""
-        run_id: str | None = None,
+        record, run = load_stored_run(self._run_store(), run_id)
-    ) -> str | None:
+        environment = record.environment
-        """Store only interrupted runs; terminal runs leave no resume handle."""
+        end = trace_range.start + trace_range.limit
-        if run.status.value != "interrupted":
+        return _run_payload(
-            if run_id is not None:
+            deployment=environment.deployment,
-                self._active_runs.pop(run_id, None)
+            artifact=environment.root_artifact,
-            return None
+            status=record.status.value,
-        key = run_id or f"run_{uuid4().hex}"
+            run_id=record.id,
-        self._active_runs[key] = ActiveWorkflowRun(
+            resume_readiness=record.resume_readiness.value,
-            deployment=deployment,
+            diagnostics=record.diagnostics,
-            artifact=artifact,
+            trace_count=len(run.trace),
-            plan=plan,
+            trace=[asdict(entry) for entry in run.trace[trace_range.start : end]],
-            run=run,
+            trace_start=trace_range.start,
+            trace_limit=trace_range.limit,
+            trace_truncated=len(run.trace) > end,
         )
-        return key
+
+    def _run_store(self) -> RunStore:
+        """Return the configured durable run store required by workflow runs."""
+        if self.service.run_store is None:
+            raise KeyError("workflow run store is not configured")
+        return self.service.run_store
 
     def _deployment_validation(
         self,
         deployment_id: str,
-    ) -> tuple[WorkflowDeployment, WorkflowArtifact, list[DependencyDiagnostic]]:
+    ) -> tuple[
+        WorkflowDeployment,
+        WorkflowArtifact,
+        list[DependencyDiagnostic],
+        SavedSubgraphTree,
+    ]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
         deployment = self.service.artifact_store.get_deployment(deployment_id)
@@ -1074,7 +1132,7 @@ class WorkflowSurfaceHandlers:
                 sources=available_sources,
             )
         )
-        return deployment, artifact, diagnostics
+        return deployment, artifact, diagnostics, tree
 
 
 def _available_sources(service: WfMcpService) -> list[AvailableSource]:
@@ -1346,6 +1404,7 @@ def _run_payload(
     artifact: WorkflowArtifact,
     status: str,
     run_id: str | None = None,
+    resume_readiness: str | None = None,
     interrupt: dict[str, Any] | None = None,
     outcome: str | None = None,
     diagnostics: list[DependencyDiagnostic] | None = None,
@@ -1362,6 +1421,7 @@ def _run_payload(
         "artifact_version": artifact.version,
         "status": status,
         "run_id": run_id,
+        "resume_readiness": resume_readiness,
         "interrupt": interrupt,
         "outcome": outcome,
         "output": output,

src/wf_mcp/workflow_surface/run_lifecycle.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from uuid import uuid4
+
+from wf_artifacts import (
+    AvailableSource,
+    CheckpointReason,
+    DependencyDiagnostic,
+    DiagnosticSeverity,
+    PinnedRunEnvironment,
+    ResumeReadiness,
+    RunCheckpoint,
+    RunStore,
+    StoredRunStatus,
+    WorkflowArtifact,
+    WorkflowDeployment,
+    WorkflowRunRecord,
+    validate_deployment_dependencies,
+)
+from wf_core import (
+    PersistedRunState,
+    RunState,
+    RunStatus,
+    dump_run_state,
+    load_run_state,
+)
+
+from .saved_subgraphs import SavedSubgraphTree
+
+
+def create_pinned_environment(
+    *,
+    deployment: WorkflowDeployment,
+    artifact: WorkflowArtifact,
+    tree: SavedSubgraphTree,
+) -> PinnedRunEnvironment:
+    """Capture exact root, deployment, and child definitions for one run."""
+    return PinnedRunEnvironment(
+        deployment=deployment,
+        root_artifact=artifact,
+        child_artifacts=list(tree.artifacts_by_ref.values()),
+    )
+
+
+def persist_stopped_run(
+    *,
+    store: RunStore,
+    environment: PinnedRunEnvironment,
+    run: RunState,
+    run_id: str | None = None,
+) -> WorkflowRunRecord:
+    """Persist one externally visible stopped state and its typed checkpoint."""
+    if run.status not in {
+        RunStatus.INTERRUPTED,
+        RunStatus.COMPLETED,
+        RunStatus.FAILED,
+    }:
+        raise ValueError(
+            f"cannot persist active workflow run with status {run.status!s}"
+        )
+
+    key = run_id or f"run_{uuid4().hex}"
+    now = datetime.now(UTC)
+    sequence = 1
+    created_at = now
+    if run_id is not None:
+        existing = store.get_run(run_id)
+        created_at = existing.created_at
+        sequence = store.get_latest_checkpoint(run_id).sequence + 1
+
+    status = StoredRunStatus(run.status.value)
+    readiness = (
+        ResumeReadiness.READY
+        if status is StoredRunStatus.INTERRUPTED
+        else ResumeReadiness.NOT_APPLICABLE
+    )
+    checkpoint_id = f"{key}.{sequence:06d}"
+    checkpoint = RunCheckpoint(
+        id=checkpoint_id,
+        run_id=key,
+        sequence=sequence,
+        reason=CheckpointReason(status.value),
+        state=PersistedRunState.model_validate(dump_run_state(run)),
+        created_at=now,
+    )
+    record = WorkflowRunRecord(
+        id=key,
+        status=status,
+        resume_readiness=readiness,
+        environment=environment,
+        latest_checkpoint_id=checkpoint_id,
+        created_at=created_at,
+        updated_at=now,
+    )
+    store.save_checkpoint(checkpoint)
+    store.save_run(record)
+    return record
+
+
+def restore_interrupted_run(
+    store: RunStore, run_id: str
+) -> tuple[WorkflowRunRecord, RunState]:
+    """Load a persisted interrupted run and its latest typed runtime state."""
+    record, run = load_stored_run(store, run_id)
+    if record.status is not StoredRunStatus.INTERRUPTED:
+        raise ValueError(f"workflow run {run_id!r} is not interrupted")
+    return record, run
+
+
+def load_stored_run(store: RunStore, run_id: str) -> tuple[WorkflowRunRecord, RunState]:
+    """Load any stopped run record together with its latest typed checkpoint."""
+    record = store.get_run(run_id)
+    checkpoint = store.get_latest_checkpoint(run_id)
+    return record, load_run_state(checkpoint.state.model_dump(mode="json"))
+
+
+def validate_pinned_resume_environment(
+    *,
+    record: WorkflowRunRecord,
+    sources: list[AvailableSource],
+) -> list[DependencyDiagnostic]:
+    """Revalidate exact stored graph definitions before a resume mutates state."""
+    environment = record.environment
+    diagnostics = validate_deployment_dependencies(
+        artifact=environment.root_artifact,
+        deployment=environment.deployment,
+        sources=sources,
+    )
+    for child in environment.child_artifacts:
+        diagnostics.extend(
+            validate_deployment_dependencies(
+                artifact=child,
+                deployment=environment.deployment,
+                sources=sources,
+            )
+        )
+    return diagnostics
+
+
+def has_blocking_diagnostics(diagnostics: list[DependencyDiagnostic]) -> bool:
+    """Return whether dependency diagnostics prohibit executing a resume."""
+    return any(item.severity is DiagnosticSeverity.ERROR for item in diagnostics)
+
+
+def mark_resume_blocked(
+    *,
+    store: RunStore,
+    record: WorkflowRunRecord,
+    diagnostics: list[DependencyDiagnostic],
+) -> WorkflowRunRecord:
+    """Record blocked readiness without writing a new execution checkpoint."""
+    blocked = record.model_copy(
+        update={
+            "resume_readiness": ResumeReadiness.BLOCKED,
+            "diagnostics": diagnostics,
+            "updated_at": datetime.now(UTC),
+        }
+    )
+    store.save_run(blocked)
+    return blocked

src/wf_mcp/workflow_surface/saved_subgraphs.py

@@ -40,6 +40,19 @@ class SavedSubgraphTree:
     diagnostics: list[DependencyDiagnostic]
 
 
+def saved_subgraph_tree_from_snapshots(
+    child_artifacts: list[WorkflowArtifact],
+) -> SavedSubgraphTree:
+    """Restore the exact saved-child definitions pinned by a durable run."""
+    return SavedSubgraphTree(
+        artifacts_by_ref={
+            f"workflow.{artifact.id}.v{artifact.version}": artifact
+            for artifact in child_artifacts
+        },
+        diagnostics=[],
+    )
+
+
 def resolve_saved_subgraph_tree(
     *,
     root_artifact: WorkflowArtifact,
@@ -127,8 +140,8 @@ def direct_wrapper_interrupt_diagnostic(
 ) -> DependencyDiagnostic | None:
     """Reject direct wrapper calls that cannot return a resumable run handle.
 
-    Deployment execution supports interrupt/resume through an in-memory
+    Deployment execution supports interrupt/resume through a durable `run_id`;
-    `run_id`; `call_capability` remains a single-call authoring probe.
+    `call_capability` remains a single-call authoring probe.
     """
     if not any(isinstance(node, InterruptNode) for node in _artifact_steps(artifact)):
         return None

src/wf_mcp/workflow_surface/tools.py

@@ -633,8 +633,9 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         name="wf.workflow.resume_run",
         title="Resume Workflow Run",
         description=(
-            "Resume an interrupted in-memory deployment run returned by "
+            "Resume an interrupted durable deployment run returned by "
-            "run_deployment. Run IDs are process-local and are not durable."
+            "run_deployment. Resume can remain blocked when a pinned source "
+            "dependency is unavailable."
         ),
     )
     async def resume_run(
@@ -657,3 +658,36 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             resume_outcome=resume_outcome,
             trace_range=trace_range,
         )
+
+    @server.tool(
+        name="wf.workflow.inspect_run",
+        title="Inspect Workflow Run",
+        description=(
+            "Return a durable stopped-run summary and result without debug trace "
+            "entries. Use read_run_trace only when trace detail is required."
+        ),
+    )
+    async def inspect_run(run_id: str) -> dict[str, Any]:
+        return await handlers.inspect_run(run_id=run_id)
+
+    @server.tool(
+        name="wf.workflow.read_run_trace",
+        title="Read Workflow Run Trace",
+        description="Read an explicit bounded debug trace slice for a durable run.",
+    )
+    async def read_run_trace(
+        run_id: str,
+        trace_range: Annotated[
+            TraceRange,
+            Field(
+                description=(
+                    "Debug traces range to return. Keep the range small because "
+                    "entries can include resolved inputs, outputs, and state changes."
+                )
+            ),
+        ],
+    ) -> dict[str, Any]:
+        return await handlers.read_run_trace(
+            run_id=run_id,
+            trace_range=trace_range,
+        )

tests/artifacts/test_run_store.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+
+from wf_artifacts import (
+    CheckpointReason,
+    FileRunStore,
+    PinnedRunEnvironment,
+    ResumeReadiness,
+    RunCheckpoint,
+    StoredRunStatus,
+    WorkflowArtifact,
+    WorkflowDeployment,
+    WorkflowRunRecord,
+)
+from wf_core import PersistedRunState, RunState, RunStatus, dump_run_state
+
+
+def artifact(artifact_id: str = "parent") -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id=artifact_id,
+        version=1,
+        title=artifact_id.title(),
+        input_schema={"type": "object", "properties": {}},
+        output_schema={"type": "object", "properties": {}},
+        outcomes=("ok",),
+        plan={"name": artifact_id, "nodes": [], "edges": []},
+    )
+
+
+def deployment() -> WorkflowDeployment:
+    return WorkflowDeployment(
+        id="parent.personal",
+        artifact_id="parent",
+        artifact_version=1,
+        bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+    )
+
+
+def run_record(run_id: str, checkpoint_id: str) -> WorkflowRunRecord:
+    now = datetime.now(UTC)
+    return WorkflowRunRecord(
+        id=run_id,
+        status=StoredRunStatus.INTERRUPTED,
+        resume_readiness=ResumeReadiness.READY,
+        environment=PinnedRunEnvironment(
+            deployment=deployment(),
+            root_artifact=artifact(),
+            child_artifacts=[artifact("child")],
+        ),
+        latest_checkpoint_id=checkpoint_id,
+        created_at=now,
+        updated_at=now,
+    )
+
+
+def checkpoint(run_id: str, sequence: int) -> RunCheckpoint:
+    return RunCheckpoint(
+        id=f"{run_id}.{sequence:06d}",
+        run_id=run_id,
+        sequence=sequence,
+        reason=CheckpointReason.INTERRUPTED,
+        state=PersistedRunState.model_validate(
+            dump_run_state(
+                RunState(
+                    workflow_name="parent",
+                    status=RunStatus.INTERRUPTED,
+                    workflow_input={},
+                    state={},
+                )
+            )
+        ),
+        created_at=datetime.now(UTC),
+    )
+
+
+def test_file_run_store_round_trips_pinned_environment_and_checkpoint(tmp_path) -> None:
+    store = FileRunStore(tmp_path)
+    run = run_record("run_123", "run_123.000001")
+    stored_checkpoint = checkpoint("run_123", 1)
+
+    store.save_run(run)
+    store.save_checkpoint(stored_checkpoint)
+
+    restored_run = store.get_run("run_123")
+    restored_checkpoint = store.get_latest_checkpoint("run_123")
+
+    assert restored_run.environment.root_artifact.id == "parent"
+    assert restored_run.environment.child_artifacts[0].id == "child"
+    assert restored_checkpoint.sequence == 1
+
+
+def test_file_run_store_lists_runs_and_checkpoints_in_order(tmp_path) -> None:
+    store = FileRunStore(tmp_path)
+    store.save_run(run_record("run_b", "run_b.000001"))
+    store.save_run(run_record("run_a", "run_a.000002"))
+    store.save_checkpoint(checkpoint("run_a", 2))
+    store.save_checkpoint(checkpoint("run_a", 1))
+
+    assert [record.id for record in store.list_runs()] == ["run_a", "run_b"]
+    assert [item.sequence for item in store.list_checkpoints("run_a")] == [1, 2]
+
+
+def test_file_run_store_rejects_unsafe_run_id(tmp_path) -> None:
+    store = FileRunStore(tmp_path)
+
+    with pytest.raises(ValueError, match="run_id must match"):
+        store.get_run("../outside")

tests/core/test_execution_results.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from wf_core import (
+    END,
+    Edge,
+    InterruptNode,
+    NodeDef,
+    NodeUse,
+    RunStatus,
+    RuntimeContext,
+    SchemaRef,
+    StateSchema,
+    Workflow,
+    execute_workflow_async,
+    execute_workflow_result_async,
+    resume_workflow_async,
+    resume_workflow_result_async,
+)
+
+
+async def explode(_payload: dict[str, Any], _context: RuntimeContext) -> dict[str, Any]:
+    raise ValueError("boom")
+
+
+def test_execute_result_api_returns_failed_state_without_changing_strict_execute() -> (
+    None
+):
+    workflow = _failing_workflow()
+
+    failed = asyncio.run(
+        execute_workflow_result_async(workflow, {}, {"explode": explode})
+    )
+
+    assert failed.status is RunStatus.FAILED
+    assert failed.error == "boom"
+
+    with pytest.raises(ValueError, match="boom"):
+        asyncio.run(execute_workflow_async(workflow, {}, {"explode": explode}))
+
+
+def test_resume_result_api_returns_failed_state_without_changing_strict_resume() -> (
+    None
+):
+    workflow = _interrupt_then_fail_workflow()
+    interrupted = asyncio.run(
+        execute_workflow_async(workflow, {}, {"explode": explode})
+    )
+
+    failed = asyncio.run(
+        resume_workflow_result_async(
+            workflow,
+            interrupted,
+            {"explode": explode},
+            resume_payload={},
+        )
+    )
+
+    assert failed.status is RunStatus.FAILED
+    assert failed.error == "boom"
+
+    interrupted = asyncio.run(
+        execute_workflow_async(workflow, {}, {"explode": explode})
+    )
+    with pytest.raises(ValueError, match="boom"):
+        asyncio.run(
+            resume_workflow_async(
+                workflow,
+                interrupted,
+                {"explode": explode},
+                resume_payload={},
+            )
+        )
+
+
+def _failing_workflow() -> Workflow:
+    return Workflow(
+        name="failing",
+        input_schema=_schema(),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=_schema(),
+        outcomes=["ok"],
+        start="explode",
+        node_defs=[
+            NodeDef(
+                name="explode",
+                input_schema=_schema(),
+                output_schema=_schema(),
+                outcomes=["ok"],
+            )
+        ],
+        nodes=[NodeUse(id="explode", type="node", node="explode")],
+        edges=[Edge.model_validate({"from": "explode", "outcome": "ok", "to": END})],
+    )
+
+
+def _interrupt_then_fail_workflow() -> Workflow:
+    return Workflow(
+        name="interrupt_then_fail",
+        input_schema=_schema(),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=_schema(),
+        outcomes=["ok"],
+        start="ask",
+        node_defs=[
+            NodeDef(
+                name="explode",
+                input_schema=_schema(),
+                output_schema=_schema(),
+                outcomes=["ok"],
+            )
+        ],
+        nodes=[
+            InterruptNode(id="ask", type="interrupt", kind="approval"),
+            NodeUse(id="explode", type="node", node="explode"),
+        ],
+        edges=[
+            Edge.model_validate(
+                {"from": "ask", "outcome": "submitted", "to": "explode"}
+            ),
+            Edge.model_validate({"from": "explode", "outcome": "ok", "to": END}),
+        ],
+    )
+
+
+def _schema() -> SchemaRef:
+    return SchemaRef(type="object", properties={})

tests/core/test_run_codec.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from wf_core import RunState, RunStatus, dump_run_state, load_run_state
+from wf_core.models.reducers import ReducerRef
+from wf_core.models.workflow_refs import WorkflowRef
+from wf_core.paths import StatePath
+from wf_core.run_state import (
+    ROOT_SCOPE_ID,
+    InterruptRequest,
+    InterruptRoute,
+    LineageState,
+    RuntimeScope,
+    StateWrite,
+)
+
+
+def test_run_state_codec_round_trips_completed_output() -> None:
+    run = RunState(
+        workflow_name="echo",
+        status=RunStatus.COMPLETED,
+        workflow_input={"text": "hi"},
+        state={"echoed": "hi"},
+        outcome="ok",
+        output={"echoed": "hi"},
+    )
+
+    stored = dump_run_state(run)
+    restored = load_run_state(stored)
+
+    assert stored["version"] == 1
+    assert restored.status is RunStatus.COMPLETED
+    assert restored.output["echoed"] == "hi"
+
+
+def test_run_state_codec_round_trips_child_interrupt_lineage_types() -> None:
+    run = RunState(
+        workflow_name="parent",
+        status=RunStatus.INTERRUPTED,
+        workflow_input={},
+        state={},
+    )
+    run.scopes["child"] = RuntimeScope(
+        id="child",
+        workflow_name="child",
+        workflow_ref=WorkflowRef(name="child"),
+    )
+    run.lineages["child-lineage"] = LineageState(
+        id="child-lineage",
+        scope_id="child",
+        writes=[
+            StateWrite(
+                path=StatePath(("count",)),
+                incoming_value=1,
+                visible_value=2,
+                reducer=ReducerRef.model_validate("wf.std.add"),
+            )
+        ],
+    )
+    run.interrupt = InterruptRequest(
+        id="interrupt:child",
+        frame_id="parent-step",
+        node_id="child_step",
+        kind="approval",
+        route=InterruptRoute(
+            frame_id="child-frame",
+            node_id="ask",
+            scope_id="child",
+            lineage_id="child-lineage",
+            parent_frame_id="parent-step",
+            workflow_ref=WorkflowRef(name="child"),
+        ),
+    )
+
+    restored = load_run_state(dump_run_state(run))
+
+    write = restored.lineages["child-lineage"].writes[0]
+    assert isinstance(write.path, StatePath)
+    assert str(write.reducer.ref) == "wf.std.add"
+    assert restored.interrupt is not None
+    assert restored.interrupt.route is not None
+    assert isinstance(restored.interrupt.route.workflow_ref, WorkflowRef)
+
+
+def test_run_state_codec_restores_root_state_alias_for_resume_writes() -> None:
+    """Root-scope commits after restore must be visible to final output."""
+    state = {"text": "before"}
+    run = RunState(
+        workflow_name="root",
+        status=RunStatus.INTERRUPTED,
+        workflow_input={"text": "before"},
+        state=state,
+        scopes={
+            ROOT_SCOPE_ID: RuntimeScope(
+                id=ROOT_SCOPE_ID,
+                workflow_name="root",
+                committed_state=state,
+            )
+        },
+    )
+
+    restored = load_run_state(dump_run_state(run))
+    restored.scopes[ROOT_SCOPE_ID].committed_state["after_resume"] = "visible"
+
+    assert restored.state["after_resume"] == "visible"

tests/wf_mcp/test_broker_server.py

@@ -463,7 +463,8 @@ def test_broker_run_deployment_pauses_and_resumes_interrupting_artifacts() -> No
 
     assert resumed["status"] == "completed"
     assert resumed["outcome"] == "submitted"
-    assert resumed["run_id"] is None
+    assert resumed["run_id"] == payload["run_id"]
+    assert resumed["resume_readiness"] == "not_applicable"
 
 
 def test_build_service_from_config_uses_store_root_for_artifacts() -> None:

tests/wf_mcp/test_saved_subgraphs.py

@@ -6,6 +6,7 @@ from typing import Any
 from wf_artifacts import (
     FileWorkflowArtifactStore,
     RequiredCapability,
+    ResumeReadiness,
     WorkflowArtifact,
     WorkflowDeployment,
 )
@@ -122,6 +123,8 @@ def test_interrupting_saved_child_pauses_and_resumes_through_deployment_surface(
     assert paused["interrupt"]["node_id"] == "child_step"
     assert paused["interrupt"]["payload"]["question"] == "hello"
 
+    # Durable resume must not rely on the process-local handler instance.
+    handlers = _handlers(store)
     resumed = asyncio.run(
         handlers.resume_run(
             run_id=paused["run_id"],
@@ -134,6 +137,52 @@ def test_interrupting_saved_child_pauses_and_resumes_through_deployment_surface(
     assert resumed["output"]["echoed"] == "world"
 
 
+def test_interrupted_saved_child_blocks_resume_until_pinned_source_returns() -> None:
+    store = FileWorkflowArtifactStore(local_temp_root() / "saved_subgraph_blocked")
+    store.save_artifact(_parent_artifact())
+    store.save_artifact(_interrupting_child_artifact(requires_demo=True))
+    store.save_deployment(_deployment())
+    handlers = _handlers(store)
+
+    paused = asyncio.run(
+        handlers.run_deployment(
+            deployment_id="parent.personal",
+            workflow_input={"text": "hello"},
+        )
+    )
+    run_store = handlers.service.run_store
+    assert run_store is not None
+
+    handlers.service.capability_sources["demo.personal"].enabled = False
+    blocked = asyncio.run(
+        handlers.resume_run(
+            run_id=paused["run_id"],
+            resume_payload={"answer": "world"},
+        )
+    )
+
+    assert blocked["status"] == "interrupted"
+    assert blocked["resume_readiness"] == "blocked"
+    assert blocked["diagnostics"][0]["code"] == "source_disabled"
+    assert (
+        run_store.get_run(paused["run_id"]).resume_readiness is ResumeReadiness.BLOCKED
+    )
+    assert run_store.get_latest_checkpoint(paused["run_id"]).sequence == 1
+
+    handlers.service.capability_sources["demo.personal"].enabled = True
+    resumed = asyncio.run(
+        handlers.resume_run(
+            run_id=paused["run_id"],
+            resume_payload={"answer": "world"},
+        )
+    )
+
+    assert resumed["status"] == "completed"
+    assert resumed["resume_readiness"] == "not_applicable"
+    assert resumed["output"]["echoed"] == "world"
+    assert run_store.get_latest_checkpoint(paused["run_id"]).sequence == 2
+
+
 def test_missing_saved_child_is_unrunnable_on_deployment_surface() -> None:
     store = FileWorkflowArtifactStore(local_temp_root() / "saved_subgraph_missing_run")
     store.save_artifact(_parent_artifact())
@@ -288,7 +337,7 @@ def _io_schema(field: str) -> dict[str, Any]:
     }
 
 
-def _interrupting_child_artifact() -> WorkflowArtifact:
+def _interrupting_child_artifact(*, requires_demo: bool = False) -> WorkflowArtifact:
     plan: dict[str, Any] = {
         "name": "child",
         "input_schema": _io_schema("text"),
@@ -314,6 +363,11 @@ def _interrupting_child_artifact() -> WorkflowArtifact:
         output_schema=plan["output_schema"],
         outcomes=("completed",),
         plan=plan,
+        required_capabilities=(
+            [RequiredCapability(ref="demo.echo_tool", kind="node_spec")]
+            if requires_demo
+            else []
+        ),
     )
 
 

tests/wf_mcp/test_server.py

@@ -443,6 +443,8 @@ def test_workflow_tools_have_human_metadata() -> None:
             by_name = {tool.name: tool for tool in tools}
             list_artifacts = by_name["wf.workflow.list_artifacts"]
             run_deployment = by_name["wf.workflow.run_deployment"]
+            inspect_run = by_name["wf.workflow.inspect_run"]
+            read_run_trace = by_name["wf.workflow.read_run_trace"]
 
             assert list_artifacts.title == "List Workflow Artifacts"
             assert "saved workflow artifacts" in (list_artifacts.description or "")
@@ -458,6 +460,10 @@ def test_workflow_tools_have_human_metadata() -> None:
             assert "null" in [
                 option.get("type") for option in trace_range_schema["anyOf"]
             ]
+            assert inspect_run.title == "Inspect Workflow Run"
+            assert "trace" in (inspect_run.description or "").lower()
+            read_trace_schema = read_run_trace.inputSchema["properties"]["trace_range"]
+            assert "Debug traces" in read_trace_schema.get("description", "")
 
     asyncio.run(run_proxy())
 

tests/wf_mcp/test_workflow_surface.py

@@ -975,11 +975,30 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     )
 
     assert payload["status"] == "completed"
+    assert isinstance(payload["run_id"], str)
     assert payload["output"]["echoed"] == "hello"
     assert payload["diagnostics"] == []
     assert payload["trace_count"] == 1
     assert "trace" not in payload
 
+    inspected = asyncio.run(handlers.inspect_run(run_id=payload["run_id"]))
+    traced = asyncio.run(
+        handlers.read_run_trace(
+            run_id=payload["run_id"],
+            trace_range=TraceRange(start=0, limit=1),
+        )
+    )
+
+    assert inspected["run_id"] == payload["run_id"]
+    assert inspected["status"] == "completed"
+    assert inspected["trace_count"] == 1
+    assert "trace" not in inspected
+    assert traced["trace_count"] == 1
+    assert traced["trace_start"] == 0
+    assert traced["trace_limit"] == 1
+    assert traced["trace"][0]["node_id"] == "echo"
+    assert traced["trace_truncated"] is False
+
 
 def test_workflow_surface_run_deployment_can_include_trace_detail() -> None:
     artifact_store = FileWorkflowArtifactStore(