feat: expose workflow runs over json rpc

src/wf_transport_rpc_http/app.py

@@ -12,10 +12,14 @@ from .errors import WorkflowRpcError, raise_workflow_rpc_error
 from .models import (
     CreateDraftFromCapabilityParams,
     InspectCapabilityParams,
+    InspectRunParams,
     ListCapabilitiesParams,
     PatchDraftParams,
+    ReadRunTraceParams,
+    ResumeRunParams,
     SaveArtifactParams,
     SaveDeploymentParams,
+    StartRunParams,
     ValidateDeploymentParams,
     ValidateDraftParams,
 )
@@ -138,5 +142,61 @@ def create_rpc_app(server: WorkflowServer) -> jsonrpc.API:
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(name="workflow.runs.start", errors=[WorkflowRpcError])
+    async def workflow_runs_start(
+        params: StartRunParams = Params(...),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.run_deployment(
+                deployment_id=params.deployment_id,
+                workflow_input=params.workflow_input,
+                trace_range=(
+                    params.trace_range.to_api_trace_range()
+                    if params.trace_range is not None
+                    else None
+                ),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.inspect", errors=[WorkflowRpcError])
+    async def workflow_runs_inspect(
+        params: InspectRunParams = Params(...),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.inspect_run(run_id=params.run_id)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.trace", errors=[WorkflowRpcError])
+    async def workflow_runs_trace(
+        params: ReadRunTraceParams = Params(...),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.read_run_trace(
+                run_id=params.run_id,
+                trace_range=params.trace_range.to_api_trace_range(),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.resume", errors=[WorkflowRpcError])
+    async def workflow_runs_resume(
+        params: ResumeRunParams = Params(...),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.resume_run(
+                run_id=params.run_id,
+                resume_payload=params.resume_payload,
+                resume_outcome=params.resume_outcome,
+                trace_range=(
+                    params.trace_range.to_api_trace_range()
+                    if params.trace_range is not None
+                    else None
+                ),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     app.bind_entrypoint(entrypoint)
     return app

tests/wf_transport_rpc_http/test_app.py

@@ -5,6 +5,8 @@ from typing import Any
 
 import httpx
 
+from wf_api.models import RawWorkflowPlan
+from wf_core import END
 from wf_server import build_local_static_workflow_server
 from wf_transport_rpc_http.app import create_rpc_app
 
@@ -175,3 +177,108 @@ def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
         assert validate_deployment["result"]["status"] == "runnable"
 
     asyncio.run(scenario())
+
+
+def _constant_plan() -> RawWorkflowPlan:
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": "rpc_constant",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "result": {"type": "string", "reducer": "wf.std.replace"}
+                },
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"result": {"type": "string"}},
+                "required": ["result"],
+            },
+            "outcomes": ["ok"],
+            "start": "constant",
+            "nodes": [
+                {
+                    "id": "constant",
+                    "type": "node",
+                    "node": "wf.std.constant",
+                    "input": [
+                        {
+                            "value": "hello over rpc",
+                            "target": {"root": "local", "parts": ["value"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["value"]},
+                            "target": {"root": "state", "parts": ["result"]},
+                        }
+                    ],
+                }
+            ],
+            "edges": [{"from": "constant", "outcome": "ok", "to": END}],
+            "output": [
+                {
+                    "path": {"root": "state", "parts": ["result"]},
+                    "target": {"root": "local", "parts": ["result"]},
+                }
+            ],
+        }
+    )
+
+
+def test_rpc_runs_deployment_and_reads_bounded_trace(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        await server.api.create_artifact_from_plan(
+            artifact_id="rpc_constant",
+            version=1,
+            title="RPC Constant",
+            plan=_constant_plan(),
+            outcomes=["ok"],
+            source_bindings={"wf.std": "wf.std"},
+        )
+        await server.api.save_deployment(
+            {
+                "id": "rpc_constant.default",
+                "artifact_id": "rpc_constant",
+                "artifact_version": 1,
+                "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
+            }
+        )
+
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            run = await _rpc(
+                client,
+                "workflow.runs.start",
+                {
+                    "deployment_id": "rpc_constant.default",
+                    "workflow_input": {},
+                    "trace_range": {"start": 0, "limit": 1},
+                },
+            )
+            inspected = await _rpc(
+                client,
+                "workflow.runs.inspect",
+                {"run_id": run["result"]["run_id"]},
+            )
+            trace = await _rpc(
+                client,
+                "workflow.runs.trace",
+                {
+                    "run_id": run["result"]["run_id"],
+                    "trace_range": {"start": 0, "limit": 1},
+                },
+            )
+
+        assert run["result"]["status"] == "completed"
+        assert run["result"]["output"]["result"] == "hello over rpc"
+        assert "trace" not in inspected["result"]
+        assert inspected["result"]["trace_count"] >= 1
+        assert trace["result"]["trace_start"] == 0
+        assert trace["result"]["trace_limit"] == 1
+        assert len(trace["result"]["trace"]) == 1
+
+    asyncio.run(scenario())