fix(mcp ux): even more error handling

who asked blud to conventional commits

docs/durable_run_operations.md

@@ -52,6 +52,7 @@ The compact response includes:
 - `run_id`: durable handle for this execution attempt
 - `status`: runtime status such as `completed`, `failed`, or `interrupted`
 - `outcome`: terminal workflow outcome when available
+- `error`: failed-run error text when execution failed before a terminal outcome
 - `output`: projected workflow output when available
 - `diagnostics`: dependency/runtime diagnostics
 - `trace_count`: total trace entry count
@@ -70,8 +71,8 @@ Reads one stopped run by `run_id` without returning trace entries:
 ```
 
 Use this when a client already has a `run_id` and needs the current durable
-summary: status, outcome/output if available, diagnostics, and checkpoint
+summary: status, outcome/output if available, failed-run error text,
-metadata.
+diagnostics, and checkpoint metadata.
 
 ## `read_run_trace`
 
@@ -129,6 +130,8 @@ the core rule that external callers resume by `run_id`.
 - Prefer `inspect_run` before reading trace detail.
 - Use `read_run_trace` with explicit small ranges.
 - Treat `trace_count` as metadata, not an instruction to fetch the entire trace.
+- If a run failed with `trace_count: 0`, read the top-level `error` first; the
+  failure may have happened before any trace entry could be emitted.
 - If `resume_run` is blocked, repair the reported dependency issue and retry
   with the same `run_id`.
 - If the run failed because a live source errored during execution, do not retry
@@ -141,4 +144,3 @@ the core rule that external callers resume by `run_id`.
 - No protocol-native MCP Tasks integration yet.
 - No dynamic saved-workflow-as-tool projection requirement.
 - No automatic pause on disconnected sources; source failures are failed runs.
-

docs/wf_mcp_operator_manual.md

@@ -282,7 +282,8 @@ top-level input/output field names, while full JSON schemas stay behind
 `wf.workflow.call_capability` is the REPL-style test step. Its result is
 self-describing: `kind` is either `node_spec` or `wrapper_artifact`,
 `source_id` identifies the owner when applicable, and `diagnostics` is empty for
-successful calls.
+successful calls. Failed test calls return a structured diagnostic with
+`outcome="runtime_error"` instead of leaking raw transport exceptions.
 
 ### 4. Manage Saved Workflows
 
@@ -354,9 +355,10 @@ brand-new MCP tools. Many LLM harnesses do not reliably refresh callable tool
 schemas mid-session.
 
 The default `run_deployment` response is intentionally compact. It includes run
-status, output, diagnostics, and `trace_count`, where `trace_count` is the total
+status, terminal outcome when available, failed-run `error` text when available,
-number of trace entries in the original run. If the caller needs node-level
+output, diagnostics, and `trace_count`, where `trace_count` is the total number
-debug detail, pass an explicit `trace_range` object such as
+of trace entries in the original run. If the caller needs node-level debug
+detail, pass an explicit `trace_range` object such as
 `{"start": 0, "limit": 10}`; otherwise trace entries stay out of the normal
 response. Trace entries may include resolved node inputs, node outputs, and
 state changes, so treat them as debug payloads rather than ordinary list/summary

docs/workflow_drafts.md

@@ -600,11 +600,12 @@ Workspaces are mutable and revisioned. Artifacts are immutable and versioned.
 Patch calls must include the current `revision`; stale revisions return
 `revision_conflict` and do not mutate the workspace.
 
-`create_minimal_draft_workspace` is intentionally only a bootstrapper. It wires
+`create_minimal_draft_workspace` is intentionally only a bootstrapper. For
-an `error` outcome for naive MCP wrappers only when `error_message_source` is
+naive MCP wrappers with an `error` outcome, it wires `wf.std.runtime_error` with
-provided or a state path can be derived from canonical `output` bindings or the
+a static default message unless `error_message_source` is explicitly provided.
-compatibility `output_map`. Provider-specific
+It does not guess that a normal output state path is also an error message.
-error envelopes still belong in saved wrapper artifacts or follow-up patches.
+Provider-specific error envelopes still belong in saved wrapper artifacts or
+follow-up patches.
 
 In MCP Inspector, workspace mutation tools accept a single `request` object.
 This is deliberate: the request object carries descriptions and validation for

src/wf_artifacts/factory.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping
 
 from wf_core import ReducerRef, Workflow
-from wf_platform import NodeSpecInventory
+from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema
 
 from .models import ArtifactKind, JsonObject, RequiredCapability, WorkflowArtifact
 from .references import normalize_plan_node_refs
@@ -32,6 +32,7 @@ def create_workflow_artifact_from_plan(
     _validate_workflow_plan(normalized_plan)
     required = {
         **_required_reducers_from_plan(normalized_plan),
+        **_required_node_specs_from_plan(normalized_plan, observed_node_specs),
         **node_requirements,
         **dict(required_capabilities or {}),
     }
@@ -106,6 +107,60 @@ def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapabili
     return requirements
 
 
+def _required_node_specs_from_plan(
+    plan: JsonObject,
+    observed_node_specs: Mapping[str, NodeSpecInventory] | None,
+) -> dict[str, RequiredCapability]:
+    """Infer direct node-spec dependencies that were not rewritten by bindings.
+
+    Bound artifact creation already records concrete-to-logical rewrites in
+    `normalize_plan_node_refs`. This fallback covers drafts saved with concrete
+    refs and no source bindings, so validation still knows the workflow depends
+    on that live source capability.
+    """
+    requirements: dict[str, RequiredCapability] = {}
+    nodes = plan.get("nodes")
+    if not isinstance(nodes, list):
+        return requirements
+
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        raw_ref = node.get("node")
+        if not isinstance(raw_ref, str):
+            continue
+        try:
+            parsed = CapabilityRef.parse(raw_ref)
+        except ValueError:
+            continue
+        observed = (
+            observed_node_specs.get(raw_ref)
+            if observed_node_specs is not None
+            else None
+        )
+        requirements[raw_ref] = RequiredCapability(
+            ref=parsed,
+            kind="node_spec",
+            input_schema_hash=(
+                hash_json_schema(observed.input_schema)
+                if observed is not None
+                else None
+            ),
+            input_schema_snapshot=(
+                observed.input_schema if observed is not None else None
+            ),
+            output_schema_hash=(
+                hash_json_schema(observed.output_schema)
+                if observed is not None
+                else None
+            ),
+            output_schema_snapshot=(
+                observed.output_schema if observed is not None else None
+            ),
+        )
+    return requirements
+
+
 def _iter_state_schema_reducer_payloads(state_schema: JsonObject) -> list[object]:
     """Read reducer refs from canonical JSON Schema and legacy field metadata."""
     reducer_payloads: list[object] = []

src/wf_mcp/workflow_surface/handlers.py

@@ -9,6 +9,7 @@ from wf_artifacts import (
     AvailableCapability,
     AvailableSource,
     DependencyDiagnostic,
+    DiagnosticSeverity,
     DraftWorkspaceStore,
     RequiredCapability,
     RunStore,
@@ -201,13 +202,40 @@ class WorkflowSurfaceHandlers:
 
         spec = self.service._get_qualified_spec(qualified_name)
         handler = build_async_registry(spec)[spec.name]
-        result = await handler(payload, RuntimeContext(current_node_id=spec.name))
+        source_id = _source_id_for_capability(
-        return {
-            "qualified_name": spec.name,
-            "source_id": _source_id_for_capability(
             self.service.capability_sources,
             spec.name,
+        )
+        try:
+            result = await handler(payload, RuntimeContext(current_node_id=spec.name))
+        except Exception as exc:
+            return {
+                "qualified_name": spec.name,
+                "source_id": source_id,
+                "kind": "node_spec",
+                "deployment_id": None,
+                "outcome": "runtime_error",
+                "output": None,
+                "diagnostics": [
+                    DependencyDiagnostic(
+                        severity=DiagnosticSeverity.ERROR,
+                        code="capability_call_failed",
+                        logical_ref=spec.name,
+                        bound_source=source_id,
+                        message=(
+                            f"Capability {spec.name!r} failed during test call: {exc}"
                         ),
+                        repair_hint=(
+                            "Check the source runtime, then retry the capability "
+                            "or inspect the deployment run if this happened inside "
+                            "a workflow."
+                        ),
+                    ).model_dump(mode="json")
+                ],
+            }
+        return {
+            "qualified_name": spec.name,
+            "source_id": source_id,
             "kind": "node_spec",
             "deployment_id": None,
             "outcome": result["outcome"],
@@ -710,19 +738,22 @@ class WorkflowSurfaceHandlers:
         routes: dict[str, dict[str, str]] = {
             DEFAULT_CALL_STEP_ID: {DEFAULT_OK_OUTCOME: "__end__"}
         }
-        error_source = error_message_source or _first_state_path(draft_output)
+        if DEFAULT_ERROR_OUTCOME in outcomes:
-        if DEFAULT_ERROR_OUTCOME in outcomes and error_source is not None:
             # The bootstrapper cannot infer provider-specific error envelopes.
-            # It only wires an error route when the caller gave, or output_map
+            # Use a static default unless the caller explicitly supplies the
-            # exposes, a concrete state path that can become a runtime message.
+            # state path containing a better provider error message.
+            error_input: dict[str, Any] = {
+                "target": {"root": "local", "parts": ["message"]},
+                "value": "Capability call failed",
+            }
+            if error_message_source is not None:
+                error_input = {
+                    "target": {"root": "local", "parts": ["message"]},
+                    "path": _graph_path_payload(error_message_source),
+                }
             steps[DEFAULT_ERROR_STEP_ID] = {
                 "use": RUNTIME_ERROR_CAPABILITY,
-                "input": [
+                "input": [error_input],
-                    {
-                        "target": {"root": "local", "parts": ["message"]},
-                        "path": _graph_path_payload(error_source),
-                    }
-                ],
                 "output": [],
             }
             routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = DEFAULT_ERROR_STEP_ID
@@ -954,6 +985,7 @@ class WorkflowSurfaceHandlers:
             resume_readiness=record.resume_readiness.value,
             interrupt=_interrupt_payload(run),
             outcome=run.outcome,
+            error=run.error,
             output=run.output,
             trace_count=len(run.trace),
             trace=(
@@ -1003,6 +1035,7 @@ class WorkflowSurfaceHandlers:
                 resume_readiness=blocked.resume_readiness.value,
                 interrupt=_interrupt_payload(stopped_run),
                 outcome=stopped_run.outcome,
+                error=stopped_run.error,
                 output=stopped_run.output,
                 diagnostics=diagnostics,
                 trace_count=len(stopped_run.trace),
@@ -1032,6 +1065,7 @@ class WorkflowSurfaceHandlers:
             resume_readiness=next_record.resume_readiness.value,
             interrupt=_interrupt_payload(run),
             outcome=run.outcome,
+            error=run.error,
             output=run.output,
             trace_count=len(run.trace),
             trace=(
@@ -1064,6 +1098,7 @@ class WorkflowSurfaceHandlers:
             resume_readiness=record.resume_readiness.value,
             interrupt=_interrupt_payload(run),
             outcome=run.outcome,
+            error=run.error,
             output=run.output,
             diagnostics=record.diagnostics,
             trace_count=len(run.trace),
@@ -1241,14 +1276,6 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
-def _first_state_path(output_map: dict[str, str]) -> str | None:
-    """Return the first mapped state path for minimal error-route bootstraps."""
-    for target in output_map.values():
-        if target.startswith("state."):
-            return target
-    return None
-
-
 def _draft_input_maps(
     *,
     input: Sequence[InputBinding] | None,
@@ -1407,6 +1434,7 @@ def _run_payload(
     resume_readiness: str | None = None,
     interrupt: dict[str, Any] | None = None,
     outcome: str | None = None,
+    error: str | None = None,
     diagnostics: list[DependencyDiagnostic] | None = None,
     output: dict[str, Any] | None = None,
     trace_count: int = 0,
@@ -1424,6 +1452,7 @@ def _run_payload(
         "resume_readiness": resume_readiness,
         "interrupt": interrupt,
         "outcome": outcome,
+        "error": error,
         "output": output,
         "diagnostics": [
             diagnostic.model_dump(mode="json") for diagnostic in diagnostics or []

src/wf_mcp/workflow_surface/models.py

@@ -280,7 +280,8 @@ class CreateMinimalDraftWorkspaceRequest(BaseModel):
         default=None,
         description=(
             "Optional state path used as runtime_error.message when the capability "
-            "has an error outcome, for example state.error_message."
+            "has an error outcome, for example state.error_message. If omitted, "
+            "the generated error route uses a static default message."
         ),
     )
     title: str | None = Field(default=None, description="Optional workspace title.")
@@ -335,7 +336,8 @@ class CreateDraftWorkspaceFromCapabilityRequest(BaseModel):
         default=None,
         description=(
             "Optional state path used as runtime_error.message when the capability "
-            "has an error outcome, for example state.error_message."
+            "has an error outcome, for example state.error_message. If omitted, "
+            "the generated error route uses a static default message."
         ),
     )
 

tests/wf_mcp/test_workflow_surface.py

@@ -60,6 +60,11 @@ def mcp_echo_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
     return ChangedEchoOutput(echoed=payload.message)
 
 
+@node(name="failing_tool")
+def failing_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
+    raise RuntimeError("upstream exploded")
+
+
 @reducer(name="custom.default.multiply")
 def multiply(current: int | None, incoming: int) -> int:
     return (current or 1) * incoming
@@ -162,6 +167,37 @@ def test_workflow_surface_filters_stdlib_capabilities_by_source() -> None:
     assert payload["capabilities"][0]["source_id"] == "wf.std"
 
 
+def test_workflow_surface_call_capability_returns_structured_error() -> None:
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_call_capability_error_mcp"),
+        artifact_store=FileWorkflowArtifactStore(
+            local_temp_root() / "surface_call_capability_error"
+        ),
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", failing_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    payload = asyncio.run(
+        handlers.call_capability(
+            qualified_name="demo.personal.failing_tool",
+            payload={"message": "hello"},
+        )
+    )
+
+    assert payload["qualified_name"] == "demo.personal.failing_tool"
+    assert payload["source_id"] == "demo.personal"
+    assert payload["kind"] == "node_spec"
+    assert payload["outcome"] == "runtime_error"
+    assert payload["output"] is None
+    assert payload["diagnostics"][0]["code"] == "capability_call_failed"
+    assert payload["diagnostics"][0]["severity"] == "error"
+    assert "demo.personal.failing_tool" in payload["diagnostics"][0]["message"]
+    assert "upstream exploded" in payload["diagnostics"][0]["message"]
+
+
 def test_workflow_surface_lists_saved_wrapper_capabilities() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_wrapper_caps"
@@ -738,7 +774,7 @@ def test_workflow_surface_creates_minimal_draft_workspace_with_error_route() ->
 
     result = asyncio.run(
         handlers.create_minimal_draft_workspace(
-            workspace_id="echo_draft_canonical_error",
+            workspace_id="echo_draft_static_error",
             name="echo",
             capability_name="demo.personal.mcp_echo_tool",
             input_schema={
@@ -757,18 +793,53 @@ def test_workflow_surface_creates_minimal_draft_workspace_with_error_route() ->
         )
     )
     assert service.draft_workspace_store is not None
-    workspace = service.draft_workspace_store.get_workspace(
+    workspace = service.draft_workspace_store.get_workspace("echo_draft_static_error")
-        "echo_draft_canonical_error"
-    )
 
-    assert result["workspace_id"] == "echo_draft_canonical_error"
+    assert result["workspace_id"] == "echo_draft_static_error"
     assert workspace.draft["routes"]["call"]["ok"] == "__end__"
     assert workspace.draft["routes"]["call"]["error"] == "tool_error"
     assert workspace.draft["steps"]["tool_error"]["use"] == "wf.std.runtime_error"
     assert workspace.draft["steps"]["tool_error"]["input"] == [
         {
             "target": {"root": "local", "parts": ["message"]},
-            "path": {"root": "state", "parts": ["echoed"]},
+            "value": "Capability call failed",
+        }
+    ]
+
+
+def test_workflow_surface_minimal_draft_honors_explicit_error_message_source() -> None:
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_minimal_explicit_error_mcp"),
+        artifact_store=FileWorkflowArtifactStore(
+            local_temp_root() / "surface_minimal_explicit_error"
+        ),
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", mcp_echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    asyncio.run(
+        handlers.create_minimal_draft_workspace(
+            workspace_id="echo_draft_explicit_error",
+            name="echo",
+            capability_name="demo.personal.mcp_echo_tool",
+            input_schema={"type": "object"},
+            state_schema={"fields": {"error_message": {"type": "string"}}},
+            output_schema={"type": "object"},
+            input_map={"input.text": "text"},
+            output_map={"echoed": "state.echoed"},
+            error_message_source="state.error_message",
+        )
+    )
+    assert service.draft_workspace_store is not None
+    workspace = service.draft_workspace_store.get_workspace("echo_draft_explicit_error")
+
+    assert workspace.draft["steps"]["tool_error"]["input"] == [
+        {
+            "target": {"root": "local", "parts": ["message"]},
+            "path": {"root": "state", "parts": ["error_message"]},
         }
     ]
 
@@ -900,6 +971,49 @@ def test_workflow_surface_creates_artifact_from_workspace() -> None:
     artifact = artifact_store.get_artifact("workspace_echo", 1)
     assert result["saved"] is True
     assert artifact.id == "workspace_echo"
+    assert artifact.plan["nodes"][0]["node"] == "demo.echo_tool"
+    required = artifact.required_capability_map()["demo.echo_tool"]
+    assert required.kind == "node_spec"
+    assert str(required.observed_concrete_source) == "demo.personal"
+    assert required.input_schema_snapshot is not None
+    assert required.output_schema_snapshot is not None
+
+
+def test_workflow_surface_workspace_artifact_infers_raw_concrete_dependency() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_workspace_artifact_raw_dependency"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_workspace_artifact_raw_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+    asyncio.run(
+        handlers.create_draft_workspace(
+            workspace_id="echo_draft",
+            draft=_echo_draft(),
+        )
+    )
+
+    asyncio.run(
+        handlers.create_artifact_from_workspace(
+            workspace_id="echo_draft",
+            artifact_id="workspace_echo_raw_dependency",
+            version=1,
+            title="Workspace Echo Raw Dependency",
+            outcomes=("completed",),
+        )
+    )
+
+    artifact = artifact_store.get_artifact("workspace_echo_raw_dependency", 1)
+    required = artifact.required_capability_map()["demo.personal.echo_tool"]
+    assert required.kind == "node_spec"
+    assert required.input_schema_snapshot is not None
+    assert required.output_schema_snapshot is not None
 
 
 def test_workflow_surface_creates_wrapper_from_workspace() -> None:
@@ -1000,6 +1114,44 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     assert traced["trace_truncated"] is False
 
 
+def test_workflow_surface_failed_deployment_exposes_error_on_run_and_inspect() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_failed_run_error"
+    )
+    artifact_store.save_artifact(_failing_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="fail.personal",
+            artifact_id="fail",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_failed_run_error_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", failing_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    payload = asyncio.run(
+        handlers.run_deployment(
+            deployment_id="fail.personal",
+            workflow_input={"message": "hello"},
+        )
+    )
+    inspected = asyncio.run(handlers.inspect_run(run_id=payload["run_id"]))
+
+    assert payload["status"] == "failed"
+    assert "upstream exploded" in payload["error"]
+    assert payload["trace_count"] == 0
+    assert inspected["status"] == "failed"
+    assert inspected["error"] == payload["error"]
+
+
 def test_workflow_surface_run_deployment_can_include_trace_detail() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_run_trace_detail"
@@ -1507,6 +1659,48 @@ def _logical_echo_artifact() -> WorkflowArtifact:
     )
 
 
+def _failing_artifact() -> WorkflowArtifact:
+    plan: dict[str, Any] = {
+        "name": "fail",
+        "input_schema": {
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+        },
+        "start": "fail",
+        "nodes": [
+            {
+                "id": "fail",
+                "type": "node",
+                "node": "demo.personal.failing_tool",
+                "input": [input_binding("input.message", "message")],
+                "output": [output_binding("echoed", "state.echoed")],
+            }
+        ],
+        "edges": [{"from": "fail", "outcome": "ok", "to": "__end__"}],
+    }
+    return WorkflowArtifact(
+        id="fail",
+        version=1,
+        title="Fail",
+        input_schema=plan["input_schema"],
+        output_schema=plan["output_schema"],
+        outcomes=("completed",),
+        plan=plan,
+        required_capabilities={
+            "demo.failing_tool": RequiredCapability(
+                ref="demo.failing_tool",
+                kind="node_spec",
+            )
+        },
+    )
+
+
 def _custom_reducer_artifact() -> WorkflowArtifact:
     plan: dict[str, Any] = {
         "name": "multiply",