feat: expose workflow output bindings over rpc

src/wf_api/service.py

@@ -471,6 +471,19 @@ class WorkflowApi:
             merge=merge,
         )
 
+    async def set_workflow_output_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]:
+        return await self.draft_authoring.set_workflow_output_bindings(
+            workspace_id=workspace_id,
+            revision=revision,
+            bindings=bindings,
+        )
+
     async def bind_draft(
         self,
         *,

src/wf_api/surface.py

@@ -175,6 +175,14 @@ class WorkflowDraftSurface(Protocol):
         merge: bool = False,
     ) -> dict[str, Any]: ...
 
+    async def set_workflow_output_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]: ...
+
     async def bind_draft(
         self,
         *,

src/wf_transport_rpc_http/__init__.py

@@ -47,6 +47,7 @@ from .models import (
     SetStepInputMapParams,
     SetStepOutputBindingsParams,
     SetStepOutputMapParams,
+    SetWorkflowOutputBindingsParams,
     SetWorkflowOutputMapParams,
     StartRunParams,
     TraceRangeParams,
@@ -99,6 +100,7 @@ __all__ = [
     "SetStepInputMapParams",
     "SetStepOutputBindingsParams",
     "SetStepOutputMapParams",
+    "SetWorkflowOutputBindingsParams",
     "SetWorkflowOutputMapParams",
     "StartRunParams",
     "TraceRangeParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -257,6 +257,22 @@ class RpcDraftClientMixin:
             },
         )
 
+    async def set_workflow_output_bindings(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.set_workflow_output_bindings",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "bindings": [binding.model_dump(mode="json") for binding in bindings],
+            },
+        )
+
     async def bind_draft(
         self: RpcCaller,
         *,

src/wf_transport_rpc_http/methods/drafts.py

@@ -35,6 +35,7 @@ from ..models import (
     SetStepInputMapParams,
     SetStepOutputBindingsParams,
     SetStepOutputMapParams,
+    SetWorkflowOutputBindingsParams,
     SetWorkflowOutputMapParams,
     ValidateDraftParams,
     ValidateDraftWorkspaceParams,
@@ -297,6 +298,22 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.set_workflow_output_bindings",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_draft_workspaces_set_workflow_output_bindings(
+        params: SetWorkflowOutputBindingsParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.set_workflow_output_bindings(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                bindings=params.bindings,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.bind",
         errors=[WorkflowRpcError],

src/wf_transport_rpc_http/models.py

@@ -238,6 +238,12 @@ class SetWorkflowOutputMapParams(RpcParamsModel):
     merge: bool = False
 
 
+class SetWorkflowOutputBindingsParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    bindings: list[InputBinding]
+
+
 class BindDraftParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)

tests/wf_transport_rpc_http/test_app.py

@@ -1071,6 +1071,92 @@ async def test_rpc_draft_workspace_set_workflow_output_map(tmp_path) -> None:
     ]
 
 
+async def test_rpc_set_workflow_output_bindings_preserves_union_and_order(
+    tmp_path,
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "output_bindings_ws",
+                "capability_name": "wf.std.constant",
+                "name": "output_bindings_test",
+            },
+        )
+        contracted = await _rpc(
+            client,
+            "workflow.draft_workspaces.set_contract",
+            {
+                "workspace_id": "output_bindings_ws",
+                "revision": created["result"]["revision"],
+                "output_schema": {
+                    "type": "object",
+                    "properties": {
+                        "format": {"type": "string"},
+                    },
+                },
+            },
+        )
+        result = await _rpc(
+            client,
+            "workflow.draft_workspaces.set_workflow_output_bindings",
+            {
+                "workspace_id": "output_bindings_ws",
+                "revision": contracted["result"]["revision"],
+                "bindings": [
+                    {"path": "state.value", "target": "value"},
+                    {"value": "markdown", "target": "format"},
+                ],
+            },
+        )
+        fetched = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "output_bindings_ws", "include_draft": True},
+        )
+
+    assert "result" in result, result
+    assert result["result"]["revision"] == 3
+    assert fetched["result"]["draft"]["output"] == [
+        {"path": "state.value", "target": "value"},
+        {"value": "markdown", "target": "format"},
+    ]
+
+
+@pytest.mark.parametrize(
+    "binding",
+    [
+        {"path": "state.value", "value": "duplicate", "target": "value"},
+        {"path": "state.value"},
+        {"path": "state.value", "target": "value", "extra": True},
+        {"path": "local.value", "target": "value"},
+    ],
+)
+async def test_rpc_set_workflow_output_bindings_rejects_malformed_binding(
+    tmp_path,
+    binding: dict[str, Any],
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        rejected = await _rpc(
+            client,
+            "workflow.draft_workspaces.set_workflow_output_bindings",
+            {
+                "workspace_id": "missing_ws",
+                "revision": 1,
+                "bindings": [binding],
+            },
+        )
+
+    assert rejected["error"]["code"] == -32602
+
+
 async def test_rpc_draft_workspace_remove_route(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)

tests/wf_transport_rpc_http/test_client.py

@@ -15,7 +15,12 @@ from wf_artifacts.drafts.models import (
     DraftStep,
 )
 from wf_core import END
-from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+)
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_server import build_local_static_workflow_server
 from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
@@ -731,6 +736,41 @@ async def test_rpc_client_serializes_canonical_step_input_bindings() -> None:
     )
 
 
+async def test_rpc_client_set_workflow_output_bindings_preserves_union_order() -> None:
+    calls: list[dict[str, Any]] = []
+
+    class Client(RpcDraftClientMixin):
+        async def _call(self, method: str, params: dict[str, object]):
+            calls.append({"method": method, "params": params})
+            return {"revision": 4}
+
+    client = Client()
+    bindings: list[InputBinding] = [
+        InputPathBinding(
+            path=GraphSourcePath.state("value"),
+            target=LocalPath.of("value"),
+        ),
+        InputValueBinding(
+            target=LocalPath.of("format"),
+            value="markdown",
+        ),
+    ]
+
+    await client.set_workflow_output_bindings(
+        workspace_id="ws",
+        revision=3,
+        bindings=bindings,
+    )
+
+    assert calls[-1]["method"] == (
+        "workflow.draft_workspaces.set_workflow_output_bindings"
+    )
+    assert calls[-1]["params"]["bindings"] == [
+        {"target": "value", "path": "state.value"},
+        {"target": "format", "value": "markdown"},
+    ]
+
+
 async def test_rpc_client_serializes_step_output_bindings() -> None:
     calls: list[dict[str, Any]] = []