add run workflow deployment

src/wf_mcp/broker/artifact_tools.py

@@ -6,11 +6,14 @@ from mcp.server.fastmcp import FastMCP
 from wf_artifacts import (
     AvailableCapability,
     AvailableSource,
+    DependencyDiagnostic,
+    DiagnosticSeverity,
     WorkflowArtifact,
     WorkflowDeployment,
     validate_deployment_dependencies,
 )
 
+from ..models import RawWorkflowPlan
 from .service import WfMcpService
 
 
@@ -97,6 +100,51 @@ def register_artifact_tools(server: FastMCP, service: WfMcpService) -> None:
             ],
         }
 
+    @server.tool()
+    async def run_workflow_deployment(
+        deployment_id: str,
+        workflow_input: dict[str, Any],
+    ) -> dict[str, Any]:
+        if service.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+
+        deployment = service.artifact_store.get_deployment(deployment_id)
+        artifact = service.artifact_store.get_artifact(
+            deployment.artifact_id,
+            deployment.artifact_version,
+        )
+        diagnostics = validate_deployment_dependencies(
+            artifact=artifact,
+            deployment=deployment,
+            sources=_available_sources(service),
+        )
+        if diagnostics:
+            return _run_payload(
+                deployment=deployment,
+                artifact=artifact,
+                status="unrunnable",
+                diagnostics=diagnostics,
+            )
+
+        unsupported = _unsupported_interrupt_diagnostic(artifact)
+        if unsupported is not None:
+            return _run_payload(
+                deployment=deployment,
+                artifact=artifact,
+                status="unsupported",
+                diagnostics=[unsupported],
+            )
+
+        plan = _raw_plan_from_artifact(artifact)
+        run = await service.run_workflow_from_plan(plan, workflow_input)
+        return _run_payload(
+            deployment=deployment,
+            artifact=artifact,
+            status=run.status.value,
+            output=run.output,
+            trace_count=len(run.trace),
+        )
+
 
 def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     """Convert broker capability sources into artifact validation snapshots."""
@@ -119,3 +167,73 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
             )
         )
     return sources
+
+
+def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
+    """Validate the stored plan shape expected by the broker workflow runner."""
+    return RawWorkflowPlan(
+        name=_plan_field(artifact, "name"),
+        input_schema=_plan_field(artifact, "input_schema"),
+        state_schema=_plan_field(artifact, "state_schema"),
+        output_schema=_plan_field(artifact, "output_schema"),
+        start=_plan_field(artifact, "start"),
+        nodes=_plan_field(artifact, "nodes"),
+        edges=_plan_field(artifact, "edges"),
+    )
+
+
+def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
+    try:
+        return artifact.plan[field_name]
+    except KeyError as exc:
+        raise ValueError(
+            f"workflow artifact {artifact.id}@{artifact.version} "
+            f"is missing plan field {field_name!r}"
+        ) from exc
+
+
+def _unsupported_interrupt_diagnostic(
+    artifact: WorkflowArtifact,
+) -> DependencyDiagnostic | None:
+    if not any(node.get("type") == "interrupt" for node in _plan_nodes(artifact)):
+        return None
+    return DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="interrupting_artifact_unsupported",
+        logical_ref=f"workflow.{artifact.id}.v{artifact.version}",
+        message=(
+            "Running saved workflow artifacts with interrupt nodes is unsupported "
+            "until nested run-state resume is implemented."
+        ),
+        repair_hint=(
+            "Run this workflow as a top-level core workflow or remove interrupt "
+            "nodes before saving it as a runnable deployment."
+        ),
+    )
+
+
+def _plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
+    nodes = artifact.plan.get("nodes", [])
+    return [node for node in nodes if isinstance(node, dict)]
+
+
+def _run_payload(
+    *,
+    deployment: WorkflowDeployment,
+    artifact: WorkflowArtifact,
+    status: str,
+    diagnostics: list[DependencyDiagnostic] | None = None,
+    output: dict[str, Any] | None = None,
+    trace_count: int = 0,
+) -> dict[str, Any]:
+    return {
+        "deployment_id": deployment.id,
+        "artifact_id": artifact.id,
+        "artifact_version": artifact.version,
+        "status": status,
+        "output": output,
+        "diagnostics": [
+            diagnostic.model_dump(mode="json") for diagnostic in diagnostics or []
+        ],
+        "trace_count": trace_count,
+    }

tests/wf_mcp/test_broker_server.py

@@ -22,6 +22,7 @@ from wf_mcp.storage import FileStore
 from .test_support import (
     FailingDiscoveryAdapter,
     FakeAdapter,
+    echo_tool,
     local_temp_root,
 )
 
@@ -316,6 +317,116 @@ def test_broker_saves_and_lists_workflow_deployments() -> None:
     assert list_payload["deployments"][0]["bindings"]["context7"] == "context7.personal"
 
 
+def test_broker_runs_non_interrupting_workflow_deployment() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "broker_run_artifacts")
+    artifact_store.save_artifact(_echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings={"demo": "demo.personal"},
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "broker_run_mcp_store"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    server = create_broker_server(service)
+
+    _content, structured = asyncio.run(
+        server.call_tool(
+            "run_workflow_deployment",
+            {
+                "deployment_id": "echo.personal",
+                "workflow_input": {"text": "hello"},
+            },
+        )
+    )
+    payload = cast(dict[str, Any], cast(object, structured))
+
+    assert payload["deployment_id"] == "echo.personal"
+    assert payload["artifact_id"] == "echo"
+    assert payload["status"] == "completed"
+    assert payload["output"]["echoed"] == "hello"
+    assert payload["diagnostics"] == []
+    assert payload["trace_count"] > 0
+
+
+def test_broker_run_deployment_returns_unrunnable_for_dependency_errors() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "broker_run_unrunnable_artifacts"
+    )
+    artifact_store.save_artifact(_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="summarize_docs.personal",
+            artifact_id="summarize_docs",
+            artifact_version=1,
+            bindings={"context7": "context7.personal"},
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "broker_run_unrunnable_mcp_store"),
+        artifact_store=artifact_store,
+    )
+    server = create_broker_server(service)
+
+    _content, structured = asyncio.run(
+        server.call_tool(
+            "run_workflow_deployment",
+            {
+                "deployment_id": "summarize_docs.personal",
+                "workflow_input": {},
+            },
+        )
+    )
+    payload = cast(dict[str, Any], cast(object, structured))
+
+    assert payload["status"] == "unrunnable"
+    assert payload["output"] is None
+    assert payload["diagnostics"][0]["code"] == "source_missing"
+
+
+def test_broker_run_deployment_rejects_interrupting_artifacts() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "broker_run_interrupt_artifacts"
+    )
+    artifact_store.save_artifact(_interrupt_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="approval.personal",
+            artifact_id="approval",
+            artifact_version=1,
+            bindings={},
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "broker_run_interrupt_mcp_store"),
+        artifact_store=artifact_store,
+    )
+    server = create_broker_server(service)
+
+    _content, structured = asyncio.run(
+        server.call_tool(
+            "run_workflow_deployment",
+            {
+                "deployment_id": "approval.personal",
+                "workflow_input": {"message": "send?"},
+            },
+        )
+    )
+    payload = cast(dict[str, Any], cast(object, structured))
+
+    assert payload["status"] == "unsupported"
+    assert payload["output"] is None
+    assert payload["diagnostics"][0]["code"] == "interrupting_artifact_unsupported"
+
+
 def test_build_service_from_config_uses_store_root_for_artifacts() -> None:
     store_root = local_temp_root() / "broker_config_artifact_store"
     service = build_service_from_config(
@@ -349,3 +460,92 @@ def _artifact() -> WorkflowArtifact:
             )
         },
     )
+
+
+def _echo_artifact() -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id="echo",
+        version=1,
+        title="Echo",
+        description="Echo text through a demo capability.",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        outcomes=("completed",),
+        plan={
+            "name": "echo",
+            "input_schema": {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            },
+            "state_schema": {"fields": {"echoed": {"type": "string"}}},
+            "output_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
+            },
+            "start": "echo",
+            "nodes": [
+                {
+                    "id": "echo",
+                    "type": "node",
+                    "node": "demo.personal.echo_tool",
+                    "in_map": {"input.text": "text"},
+                    "out_map": {"echoed": "state.echoed"},
+                }
+            ],
+            "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
+        },
+        required_capabilities={
+            "demo.echo_tool": RequiredCapability(
+                logical_source="demo",
+                capability_name="echo_tool",
+                kind="node_spec",
+            )
+        },
+    )
+
+
+def _interrupt_artifact() -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id="approval",
+        version=1,
+        title="Approval",
+        input_schema={
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        output_schema={"type": "object", "properties": {}},
+        outcomes=("submitted",),
+        plan={
+            "name": "approval",
+            "input_schema": {
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+                "required": ["message"],
+            },
+            "state_schema": {"fields": {}},
+            "output_schema": {"type": "object", "properties": {}},
+            "start": "approval",
+            "nodes": [
+                {
+                    "id": "approval",
+                    "type": "interrupt",
+                    "kind": "approval",
+                    "request_map": {"input.message": "message"},
+                    "out_map": {},
+                    "outcomes": ["submitted"],
+                }
+            ],
+            "edges": [{"from": "approval", "outcome": "submitted", "to": "__end__"}],
+        },
+    )