models + next action on deployment

docs/workflow_capabilities.md

@@ -361,6 +361,12 @@ clients a compact "what should I call next?" pointer, while diagnostics,
 artifact validation, deployment validation, and runtime status remain the
 source of truth.
 
+Deployment validation and run lifecycle responses also expose `next_actions`.
+For runnable deployments this points to `wf.workflow.run_deployment`; for
+unrunnable deployments it points back to validation after the caller repairs
+bindings, sources, or schema drift. Failed runs never suggest reading an
+unbounded trace; trace guidance always uses a bounded `trace_range`.
+
 ## Relationship To Capability Sources
 
 Sources own capability kinds:

src/wf_mcp/workflow_surface/handlers.py

@@ -1001,6 +1001,10 @@ class WorkflowSurfaceHandlers:
             "diagnostics": [
                 diagnostic.model_dump(mode="json") for diagnostic in diagnostics
             ],
+            "next_actions": NextActions.from_deployment_validation(
+                deployment_id=deployment.id,
+                diagnostics=diagnostics,
+            ).model_dump(mode="json"),
         }
 
     async def run_deployment(
@@ -1584,6 +1588,12 @@ def _run_payload(
             diagnostic.model_dump(mode="json") for diagnostic in diagnostics or []
         ],
         "trace_count": trace_count,
+        "next_actions": NextActions.from_run_result(
+            run_id=run_id,
+            status=status,
+            trace_count=trace_count,
+            diagnostics=diagnostics or [],
+        ).model_dump(mode="json"),
     }
     if trace is not None:
         # Trace entries can grow quickly, so the public run tool only includes

src/wf_mcp/workflow_surface/models.py

@@ -377,6 +377,64 @@ class CreateDraftWorkspaceFromCapabilityResult(DraftWorkspaceResult):
     )
 
 
+class ValidateDeploymentResult(BaseModel):
+    """Response contract for validate_deployment."""
+
+    deployment_id: str = Field(description="Deployment that was validated.")
+    artifact_id: str = Field(description="Artifact targeted by the deployment.")
+    artifact_version: int = Field(description="Artifact version.")
+    status: Literal["runnable", "unrunnable"] = Field(
+        description="Whether the deployment can be run."
+    )
+    diagnostics: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Structured diagnostics for unrunnable deployments.",
+    )
+    next_actions: NextActions = Field(description="Advisory next step guidance.")
+
+
+class RunDeploymentResult(BaseModel):
+    """Response contract for run_deployment, inspect_run, resume_run, and read_run_trace."""
+
+    deployment_id: str = Field(description="Deployment that was run.")
+    artifact_id: str = Field(description="Artifact targeted by the deployment.")
+    artifact_version: int = Field(description="Artifact version.")
+    status: str = Field(description="Run status, such as completed, failed, or interrupted.")
+    run_id: str | None = Field(default=None, description="Durable run identifier.")
+    resume_readiness: str | None = Field(
+        default=None, description="Resume readiness state."
+    )
+    interrupt: dict[str, Any] | None = Field(
+        default=None, description="Interrupt payload if paused."
+    )
+    outcome: str | None = Field(default=None, description="Terminal outcome label.")
+    error: str | None = Field(default=None, description="Error message if failed.")
+    output: dict[str, Any] | None = Field(
+        default=None, description="Workflow output payload."
+    )
+    diagnostics: list[dict[str, Any]] = Field(
+        default_factory=list, description="Structured diagnostics."
+    )
+    trace_count: int = Field(default=0, description="Total trace entries.")
+    next_actions: NextActions = Field(description="Advisory next step guidance.")
+    trace_start: int | None = Field(
+        default=None,
+        description="First trace entry offset included when a trace slice was requested.",
+    )
+    trace_limit: int | None = Field(
+        default=None,
+        description="Maximum trace entries requested when a trace slice was included.",
+    )
+    trace: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="Bounded debug trace slice, present only when explicitly requested.",
+    )
+    trace_truncated: bool = Field(
+        default=False,
+        description="Whether more trace entries exist after the returned slice.",
+    )
+
+
 class CreateArtifactFromWorkspaceRequest(BaseModel):
     """Typed MCP request payload for saving a draft workspace as an artifact."""
 

src/wf_mcp/workflow_surface/next_actions.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from enum import StrEnum
 from typing import Any, Self
 
@@ -106,6 +107,122 @@ class NextActions(BaseModel):
             warnings=notes,
         )
 
+    @classmethod
+    def from_deployment_validation(
+        cls,
+        *,
+        deployment_id: str,
+        diagnostics: Sequence[object],
+    ) -> Self:
+        """Create guidance after validate_deployment."""
+        if not diagnostics:
+            return cls(
+                can_continue=True,
+                can_save_now=None,
+                recommended_next_tool=NextActionTool.RUN_DEPLOYMENT,
+                reason=(
+                    f"Deployment {deployment_id!r} is runnable; call "
+                    "wf.workflow.run_deployment with workflow_input."
+                ),
+                patch_examples=[],
+                warnings=[],
+            )
+
+        codes = {_diagnostic_field(diagnostic, "code") for diagnostic in diagnostics}
+        warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
+        if "source_unreachable" in codes:
+            reason = (
+                "One or more live sources are unreachable; fix or reconnect the "
+                "source, then rerun wf.workflow.validate_deployment with live_check=true."
+            )
+        elif "source_missing" in codes or "binding_missing" in codes:
+            reason = (
+                "Deployment bindings or sources are missing; inspect the deployment "
+                "and save corrected bindings before running."
+            )
+        elif "capability_missing" in codes or "schema_changed" in codes:
+            reason = (
+                "A required capability is missing or drifted; inspect capabilities "
+                "or refresh sources, then validate again."
+            )
+        else:
+            reason = (
+                "Deployment is not runnable; inspect diagnostics, repair the "
+                "deployment or sources, then validate again."
+            )
+        return cls(
+            can_continue=True,
+            can_save_now=None,
+            recommended_next_tool=NextActionTool.VALIDATE_DEPLOYMENT,
+            reason=reason,
+            patch_examples=[],
+            warnings=warnings,
+        )
+
+    @classmethod
+    def from_run_result(
+        cls,
+        *,
+        run_id: str | None,
+        status: str,
+        trace_count: int,
+        diagnostics: Sequence[object],
+    ) -> Self:
+        """Create guidance after run_deployment, inspect_run, resume_run, or read_run_trace."""
+        warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
+        if status == "interrupted" and run_id is not None:
+            return cls(
+                can_continue=True,
+                can_save_now=None,
+                recommended_next_tool=NextActionTool.RESUME_RUN,
+                reason=(
+                    "Run is interrupted; call wf.workflow.resume_run with this "
+                    "run_id and the interrupt response payload."
+                ),
+                patch_examples=[],
+                warnings=warnings,
+            )
+        if status in {"failed", "unrunnable"}:
+            examples = (
+                [_bounded_trace_example(run_id=run_id, trace_count=trace_count)]
+                if run_id is not None and trace_count > 0
+                else []
+            )
+            return cls(
+                can_continue=bool(examples),
+                can_save_now=None,
+                recommended_next_tool=(
+                    NextActionTool.READ_RUN_TRACE if examples else None
+                ),
+                reason=(
+                    "Run failed; read a bounded trace slice for debugging."
+                    if examples
+                    else "Run failed before producing trace entries; inspect diagnostics and error."
+                ),
+                patch_examples=examples,
+                warnings=warnings,
+            )
+        if status == "completed":
+            return cls(
+                can_continue=False,
+                can_save_now=None,
+                recommended_next_tool=None,
+                reason=(
+                    "Run completed. No required next workflow tool; use read_run_trace "
+                    "with a bounded trace_range only if debugging."
+                ),
+                patch_examples=[],
+                warnings=warnings,
+            )
+        return cls(
+            can_continue=False,
+            can_save_now=None,
+            recommended_next_tool=None,
+            reason=f"Run status {status!r} has no obvious next workflow tool.",
+            patch_examples=[],
+            warnings=warnings,
+        )
+
 
 def _wrapper_draft_patch_examples(
     *,
@@ -160,3 +277,45 @@ def _wrapper_draft_patch_examples(
             )
         )
     return examples
+
+
+def _diagnostic_field(diagnostic: object, field: str) -> str | None:
+    """Read a diagnostic field from either a Pydantic model or a JSON dict."""
+    if isinstance(diagnostic, dict):
+        value = diagnostic.get(field)
+    else:
+        value = getattr(diagnostic, field, None)
+    return value if isinstance(value, str) else None
+
+
+def _diagnostic_warning(diagnostic: object) -> str:
+    """Format one compact diagnostic warning for next_actions."""
+    code = _diagnostic_field(diagnostic, "code") or "diagnostic"
+    bound_source = _diagnostic_field(diagnostic, "bound_source")
+    logical_ref = _diagnostic_field(diagnostic, "logical_ref")
+    if bound_source:
+        return f"{code}: {bound_source}"
+    if logical_ref:
+        return f"{code}: {logical_ref}"
+    return code
+
+
+def _bounded_trace_example(
+    *,
+    run_id: str,
+    trace_count: int,
+) -> NextActionPatchExample:
+    """Return a safe read_run_trace request; never suggest full trace reads."""
+    return NextActionPatchExample(
+        description=(
+            "Read a bounded debug trace slice. Increase start/limit only when needed."
+        ),
+        tool=NextActionTool.READ_RUN_TRACE,
+        request={
+            "run_id": run_id,
+            "trace_range": {
+                "start": 0,
+                "limit": 25,
+            },
+        },
+    )

src/wf_mcp/workflow_surface/tools.py

@@ -23,11 +23,13 @@ from .models import (
     DraftWorkspaceListResult,
     DraftWorkspaceResult,
     PatchDraftWorkspaceRequest,
+    RunDeploymentResult,
     SetDraftNameRequest,
     SetDraftRouteRequest,
     SetStepInputMapRequest,
     SetStepOutputMapRequest,
     TraceRange,
+    ValidateDeploymentResult,
     ValidateDraftWorkspaceRequest,
 )
 
@@ -627,11 +629,13 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 )
             ),
         ] = False,
-    ) -> dict[str, Any]:
+    ) -> ValidateDeploymentResult:
-        return await handlers.validate_deployment(
+        return ValidateDeploymentResult.model_validate(
+            await handlers.validate_deployment(
                 deployment_id=deployment_id,
                 live_check=live_check,
             )
+        )
 
     @server.tool(
         name="wf.workflow.run_deployment",
@@ -656,12 +660,14 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 )
             ),
         ] = None,
-    ) -> dict[str, Any]:
+    ) -> RunDeploymentResult:
-        return await handlers.run_deployment(
+        return RunDeploymentResult.model_validate(
+            await handlers.run_deployment(
                 deployment_id=deployment_id,
                 workflow_input=workflow_input,
                 trace_range=trace_range,
             )
+        )
 
     @server.tool(
         name="wf.workflow.resume_run",
@@ -685,13 +691,15 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 )
             ),
         ] = None,
-    ) -> dict[str, Any]:
+    ) -> RunDeploymentResult:
-        return await handlers.resume_run(
+        return RunDeploymentResult.model_validate(
+            await handlers.resume_run(
                 run_id=run_id,
                 resume_payload=resume_payload,
                 resume_outcome=resume_outcome,
                 trace_range=trace_range,
             )
+        )
 
     @server.tool(
         name="wf.workflow.inspect_run",
@@ -701,8 +709,10 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             "entries. Use read_run_trace only when trace detail is required."
         ),
     )
-    async def inspect_run(run_id: str) -> dict[str, Any]:
+    async def inspect_run(run_id: str) -> RunDeploymentResult:
-        return await handlers.inspect_run(run_id=run_id)
+        return RunDeploymentResult.model_validate(
+            await handlers.inspect_run(run_id=run_id)
+        )
 
     @server.tool(
         name="wf.workflow.read_run_trace",
@@ -720,8 +730,10 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 )
             ),
         ],
-    ) -> dict[str, Any]:
+    ) -> RunDeploymentResult:
-        return await handlers.read_run_trace(
+        return RunDeploymentResult.model_validate(
+            await handlers.read_run_trace(
                 run_id=run_id,
                 trace_range=trace_range,
             )
+        )

tests/wf_mcp/server/test_config.py

@@ -125,6 +125,24 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
                 "Advisory"
                 in next_actions_schema["properties"]["can_save_now"]["description"]
             )
+            validate_deployment = tools_by_name["wf.workflow.validate_deployment"]
+            run_deployment = tools_by_name["wf.workflow.run_deployment"]
+
+            validate_output = validate_deployment.outputSchema
+            run_output = run_deployment.outputSchema
+
+            assert validate_output is not None
+            assert "next_actions" in validate_output["properties"]
+            assert (
+                "recommended_next_tool"
+                in validate_output["properties"]["next_actions"]["properties"]
+            )
+            assert run_output is not None
+            assert "next_actions" in run_output["properties"]
+            assert (
+                "recommended_next_tool"
+                in run_output["properties"]["next_actions"]["properties"]
+            )
             wrapper_workspace_input = tools_by_name[
                 "wf.workflow.create_wrapper_from_workspace"
             ].inputSchema

tests/wf_mcp/workflow_surface/test_deployments.py

@@ -57,6 +57,11 @@ def test_workflow_surface_validates_deployment_dependencies() -> None:
 
     assert payload["status"] == "unrunnable"
     assert payload["diagnostics"][0]["code"] == "source_missing"
+    assert payload["next_actions"]["can_continue"] is True
+    assert payload["next_actions"]["recommended_next_tool"] == (
+        "wf.workflow.validate_deployment"
+    )
+    assert payload["next_actions"]["warnings"][0] == "source_missing: context7.personal"
 
 
 def test_workflow_surface_validate_deployment_live_check_is_opt_in() -> None:
@@ -85,6 +90,10 @@ def test_workflow_surface_validate_deployment_live_check_is_opt_in() -> None:
     assert payload["status"] == "runnable"
     assert payload["diagnostics"] == []
     assert adapter.calls == 0
+    assert payload["next_actions"]["can_continue"] is True
+    assert payload["next_actions"]["recommended_next_tool"] == (
+        "wf.workflow.run_deployment"
+    )
 
 
 def test_workflow_surface_validate_deployment_live_check_reports_unreachable_source() -> (

tests/wf_mcp/workflow_surface/test_next_actions.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
+
 from wf_mcp.workflow_surface.next_actions import NextActionTool, NextActions
 
 
@@ -61,3 +63,86 @@ def test_next_actions_from_low_confidence_wrapper_hints_can_patch() -> None:
         "/draft/steps/call/output"
     )
     assert dumped["patch_examples"][1]["request"]["patch"] == []
+
+
+def test_next_actions_from_runnable_deployment_recommends_run() -> None:
+    actions = NextActions.from_deployment_validation(
+        deployment_id="echo.personal",
+        diagnostics=[],
+    )
+
+    dumped = actions.model_dump(mode="json")
+    assert dumped["can_continue"] is True
+    assert dumped["recommended_next_tool"] == NextActionTool.RUN_DEPLOYMENT.value
+    assert "run_deployment" in dumped["reason"]
+    assert dumped["warnings"] == []
+
+
+def test_next_actions_from_unrunnable_deployment_recommends_validation_retry() -> None:
+    diagnostic = DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="source_unreachable",
+        logical_ref="demo.echo_tool",
+        bound_source="demo.personal",
+        message="Live check for upstream source 'demo.personal' failed.",
+        repair_hint="Start or reconnect the source.",
+    )
+
+    actions = NextActions.from_deployment_validation(
+        deployment_id="echo.personal",
+        diagnostics=[diagnostic],
+    )
+
+    dumped = actions.model_dump(mode="json")
+    assert dumped["can_continue"] is True
+    assert dumped["recommended_next_tool"] == NextActionTool.VALIDATE_DEPLOYMENT.value
+    assert "fix or reconnect" in dumped["reason"]
+    assert dumped["warnings"][0] == "source_unreachable: demo.personal"
+
+
+def test_next_actions_from_completed_run_has_no_required_next_tool() -> None:
+    actions = NextActions.from_run_result(
+        run_id="run_123",
+        status="completed",
+        trace_count=2,
+        diagnostics=[],
+    )
+
+    dumped = actions.model_dump(mode="json")
+    assert dumped["can_continue"] is False
+    assert dumped["recommended_next_tool"] is None
+    assert "completed" in dumped["reason"]
+    assert dumped["patch_examples"] == []
+
+
+def test_next_actions_from_failed_run_recommends_bounded_trace() -> None:
+    actions = NextActions.from_run_result(
+        run_id="run_123",
+        status="failed",
+        trace_count=12,
+        diagnostics=[],
+    )
+
+    dumped = actions.model_dump(mode="json")
+    assert dumped["can_continue"] is True
+    assert dumped["recommended_next_tool"] == NextActionTool.READ_RUN_TRACE.value
+    assert "bounded trace" in dumped["reason"]
+    assert dumped["patch_examples"][0]["tool"] == NextActionTool.READ_RUN_TRACE.value
+    assert dumped["patch_examples"][0]["request"]["run_id"] == "run_123"
+    assert dumped["patch_examples"][0]["request"]["trace_range"]["start"] == 0
+    assert dumped["patch_examples"][0]["request"]["trace_range"]["limit"] == 25
+
+
+def test_next_actions_from_interrupted_run_recommends_resume() -> None:
+    actions = NextActions.from_run_result(
+        run_id="run_123",
+        status="interrupted",
+        trace_count=3,
+        diagnostics=[],
+    )
+
+    dumped = actions.model_dump(mode="json")
+    assert dumped["can_continue"] is True
+    assert dumped["recommended_next_tool"] == NextActionTool.RESUME_RUN.value
+    assert "resume_run" in dumped["reason"]
+    assert dumped["patch_examples"] == []

tests/wf_mcp/workflow_surface/test_runs.py

@@ -7,6 +7,7 @@ from wf_mcp.broker import WfMcpService
 from wf_mcp.models import ConnectionConfig
 from wf_mcp.storage import FileStore
 from wf_mcp.workflow_surface import TraceRange, WorkflowSurfaceHandlers
+from wf_mcp.workflow_surface.models import RunDeploymentResult
 from wf_platform import (
     CapabilityBuckets,
     CapabilitySource,
@@ -70,6 +71,9 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     assert payload["diagnostics"] == []
     assert payload["trace_count"] == 1
     assert "trace" not in payload
+    assert payload["next_actions"]["can_continue"] is False
+    assert payload["next_actions"]["recommended_next_tool"] is None
+    assert "completed" in payload["next_actions"]["reason"]
 
     inspected = asyncio.run(h.inspect_run(run_id=payload["run_id"]))
     traced = asyncio.run(
@@ -83,6 +87,8 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     assert inspected["status"] == "completed"
     assert inspected["trace_count"] == 1
     assert "trace" not in inspected
+    assert inspected["next_actions"]["can_continue"] is False
+    assert inspected["next_actions"]["recommended_next_tool"] is None
     assert traced["trace_count"] == 1
     assert traced["trace_start"] == 0
     assert traced["trace_limit"] == 1
@@ -124,8 +130,11 @@ def test_workflow_surface_failed_deployment_exposes_error_on_run_and_inspect() -
     assert payload["status"] == "failed"
     assert "upstream exploded" in payload["error"]
     assert payload["trace_count"] == 0
+    assert payload["next_actions"]["recommended_next_tool"] is None
+    assert "before producing trace" in payload["next_actions"]["reason"]
     assert inspected["status"] == "failed"
     assert inspected["error"] == payload["error"]
+    assert inspected["next_actions"]["recommended_next_tool"] is None
 
 
 def test_workflow_surface_run_deployment_can_include_trace_detail() -> None:
@@ -167,6 +176,13 @@ def test_workflow_surface_run_deployment_can_include_trace_detail() -> None:
     assert len(payload["trace"]) == 1
     assert payload["trace"][0]["node_id"] == "echo"
     assert payload["trace"][0]["outcome"] == "ok"
+    assert payload["next_actions"]["can_continue"] is False
+    assert payload["next_actions"]["patch_examples"] == []
+    validated = RunDeploymentResult.model_validate(payload).model_dump(mode="json")
+    assert validated["trace"][0]["node_id"] == "echo"
+    assert validated["trace_start"] == 0
+    assert validated["trace_limit"] == 10
+    assert validated["trace_truncated"] is False
 
 
 def test_workflow_surface_run_deployment_can_read_empty_trace_range() -> None: