feat: expose capability step updates over rpc

src/wf_api/service.py

@@ -11,6 +11,7 @@ from .artifacts import WorkflowArtifactApi
 from .capabilities import WorkflowCapabilityApi
 from .deployments import WorkflowDeploymentApi
 from .draft_authoring import RouteSource, WorkflowDraftAuthoringApi
+from .draft_updates import CapabilityStepUpdate
 from .drafts import WorkflowDraftApi
 from .models import RawWorkflowPlan
 from .operation_context import WorkflowOperationContext
@@ -439,6 +440,21 @@ class WorkflowApi:
             bindings=bindings,
         )
 
+    async def update_capability_step(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        update: CapabilityStepUpdate,
+    ) -> dict[str, Any]:
+        return await self.draft_authoring.update_capability_step(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            update=update,
+        )
+
     async def set_step_output_map(
         self,
         *,
@@ -512,7 +528,11 @@ class WorkflowApi:
         route_from_outcome: str = "ok",
         routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
+        input_bindings: Sequence[InputBinding] | None = None,
         bind_outputs: dict[str, str] | None = None,
+        desc: str | None = None,
+        retry: int | None = None,
+        timeout_seconds: int | None = None,
     ) -> dict[str, Any]:
         return await self.draft_authoring.add_step_from_capability(
             workspace_id=workspace_id,
@@ -523,7 +543,11 @@ class WorkflowApi:
             route_from_outcome=route_from_outcome,
             routes=routes,
             input_map=input_map,
+            input_bindings=input_bindings,
             bind_outputs=bind_outputs,
+            desc=desc,
+            retry=retry,
+            timeout_seconds=timeout_seconds,
         )
 
     async def add_step(

src/wf_api/surface.py

@@ -8,6 +8,7 @@ from wf_artifacts.drafts.models import DraftStep
 from wf_core.models.steps import InputBinding, OutputBinding
 
 from .draft_authoring import RouteSource
+from .draft_updates import CapabilityStepUpdate
 from .runs import TraceRangeLike
 
 
@@ -156,6 +157,15 @@ class WorkflowDraftSurface(Protocol):
         bindings: Sequence[OutputBinding],
     ) -> dict[str, Any]: ...
 
+    async def update_capability_step(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        update: CapabilityStepUpdate,
+    ) -> dict[str, Any]: ...
+
     async def set_step_output_map(
         self,
         *,
@@ -204,7 +214,11 @@ class WorkflowDraftSurface(Protocol):
         route_from_outcome: str = "ok",
         routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
+        input_bindings: Sequence[InputBinding] | None = None,
         bind_outputs: dict[str, str] | None = None,
+        desc: str | None = None,
+        retry: int | None = None,
+        timeout_seconds: int | None = None,
     ) -> dict[str, Any]: ...
 
     async def add_step(

src/wf_transport_rpc_http/__init__.py

@@ -51,6 +51,7 @@ from .models import (
     SetWorkflowOutputMapParams,
     StartRunParams,
     TraceRangeParams,
+    UpdateCapabilityStepParams,
     ValidateDeploymentParams,
     ValidateDraftParams,
     ValidateDraftWorkspaceParams,
@@ -104,6 +105,7 @@ __all__ = [
     "SetWorkflowOutputMapParams",
     "StartRunParams",
     "TraceRangeParams",
+    "UpdateCapabilityStepParams",
     "ValidateDeploymentParams",
     "ValidateDraftParams",
     "ValidateDraftWorkspaceParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from collections.abc import Sequence
 from typing import Any, Literal
 
+from wf_api import CapabilityStepUpdate
 from wf_api.surface import RouteSource
 from wf_artifacts.drafts.models import DraftStep
 from wf_core.models.steps import InputBinding, OutputBinding
@@ -219,6 +220,24 @@ class RpcDraftClientMixin:
             },
         )
 
+    async def update_capability_step(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        update: CapabilityStepUpdate,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.update_capability_step",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "step_id": step_id,
+                "update": update.model_dump(mode="json", exclude_unset=True),
+            },
+        )
+
     async def set_step_output_map(
         self: RpcCaller,
         *,
@@ -304,21 +323,39 @@ class RpcDraftClientMixin:
         route_from_outcome: str = "ok",
         routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
+        input_bindings: Sequence[InputBinding] | None = None,
         bind_outputs: dict[str, str] | None = None,
+        desc: str | None = None,
+        retry: int | None = None,
+        timeout_seconds: int | None = None,
     ) -> dict[str, Any]:
+        if input_map is not None and input_bindings is not None:
+            raise ValueError("input_map and input_bindings are mutually exclusive")
+        params: dict[str, object] = {
+            "workspace_id": workspace_id,
+            "revision": revision,
+            "step_id": step_id,
+            "capability_name": capability_name,
+            "route_from_step": route_from_step,
+            "route_from_outcome": route_from_outcome,
+            "routes": routes,
+            "bind_outputs": bind_outputs or {},
+        }
+        if input_map is not None:
+            params["input_map"] = input_map
+        if input_bindings is not None:
+            params["input_bindings"] = [
+                binding.model_dump(mode="json") for binding in input_bindings
+            ]
+        if desc is not None:
+            params["desc"] = desc
+        if retry is not None:
+            params["retry"] = retry
+        if timeout_seconds is not None:
+            params["timeout_seconds"] = timeout_seconds
         return await self._call(
             "workflow.draft_workspaces.add_step_from_capability",
-            {
+            params,
-                "workspace_id": workspace_id,
-                "revision": revision,
-                "step_id": step_id,
-                "capability_name": capability_name,
-                "route_from_step": route_from_step,
-                "route_from_outcome": route_from_outcome,
-                "routes": routes,
-                "input_map": input_map or {},
-                "bind_outputs": bind_outputs or {},
-            },
         )
 
     async def add_step(

src/wf_transport_rpc_http/methods/drafts.py

@@ -37,6 +37,7 @@ from ..models import (
     SetStepOutputMapParams,
     SetWorkflowOutputBindingsParams,
     SetWorkflowOutputMapParams,
+    UpdateCapabilityStepParams,
     ValidateDraftParams,
     ValidateDraftWorkspaceParams,
 )
@@ -246,6 +247,23 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.update_capability_step",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_draft_workspaces_update_capability_step(
+        params: UpdateCapabilityStepParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.update_capability_step(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                step_id=params.step_id,
+                update=params.update,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.set_step_output_bindings",
         errors=[WorkflowRpcError],
@@ -349,7 +367,11 @@ def register_methods(
                 route_from_outcome=params.route_from_outcome,
                 routes=params.routes,
                 input_map=params.input_map,
+                input_bindings=params.input_bindings,
                 bind_outputs=params.bind_outputs,
+                desc=params.desc,
+                retry=params.retry,
+                timeout_seconds=params.timeout_seconds,
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -4,6 +4,7 @@ from typing import Any, Literal, Self
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from wf_api import CapabilityStepUpdate
 from wf_api.models import TraceRange
 from wf_artifacts.drafts.models import DraftStep
 from wf_core.models.steps import InputBinding, OutputBinding
@@ -252,6 +253,13 @@ class BindDraftParams(RpcParamsModel):
     target_path: str = Field(min_length=1)
 
 
+class UpdateCapabilityStepParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    step_id: str = Field(min_length=1)
+    update: CapabilityStepUpdate
+
+
 class AddStepFromCapabilityParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)
@@ -260,8 +268,18 @@ class AddStepFromCapabilityParams(RpcParamsModel):
     route_from_step: str | None = None
     route_from_outcome: str = Field(default="ok", min_length=1)
     routes: dict[str, str] | None = None
-    input_map: dict[str, str] = Field(default_factory=dict)
+    input_map: dict[str, str] | None = None
+    input_bindings: list[InputBinding] | None = None
     bind_outputs: dict[str, str] = Field(default_factory=dict)
+    desc: str | None = Field(default=None, min_length=1)
+    retry: int | None = Field(default=None, ge=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+
+    @model_validator(mode="after")
+    def validate_input_forms(self) -> Self:
+        if {"input_map", "input_bindings"} <= self.model_fields_set:
+            raise ValueError("input_map and input_bindings are mutually exclusive")
+        return self
 
 
 class BranchDraftParams(RpcParamsModel):

tests/wf_transport_rpc_http/test_app.py

@@ -12,7 +12,11 @@ from wf_core import END
 from wf_server import build_local_static_workflow_server
 from wf_server.config import build_workflow_server_from_workflow_config
 from wf_transport_rpc_http.app import create_rpc_app
-from wf_transport_rpc_http.models import AddDraftStepParams
+from wf_transport_rpc_http.models import (
+    AddDraftStepParams,
+    AddStepFromCapabilityParams,
+    UpdateCapabilityStepParams,
+)
 
 
 async def _rpc(
@@ -26,6 +30,87 @@ async def _rpc(
     return response.json()
 
 
+def test_update_capability_step_params_preserve_nested_field_presence() -> None:
+    params = UpdateCapabilityStepParams.model_validate(
+        {
+            "workspace_id": "report",
+            "revision": 4,
+            "step_id": "publish",
+            "update": {"desc": None, "retry": 0},
+        }
+    )
+
+    assert params.update.model_fields_set == {"desc", "retry"}
+    assert params.update.desc is None
+    assert params.update.retry == 0
+
+
+@pytest.mark.parametrize(
+    "update",
+    [
+        {},
+        {"input": None},
+        {"retry": -1},
+        {"timeout_seconds": 0},
+        {"unknown": True},
+    ],
+)
+def test_update_capability_step_params_reject_invalid_update(
+    update: dict[str, object],
+) -> None:
+    with pytest.raises(ValidationError):
+        UpdateCapabilityStepParams.model_validate(
+            {
+                "workspace_id": "report",
+                "revision": 4,
+                "step_id": "publish",
+                "update": update,
+            }
+        )
+
+
+def test_add_step_from_capability_params_preserve_canonical_inputs() -> None:
+    params = AddStepFromCapabilityParams.model_validate(
+        {
+            "workspace_id": "report",
+            "revision": 1,
+            "step_id": "publish",
+            "capability_name": "demo.report",
+            "desc": "Publish report",
+            "retry": 0,
+            "timeout_seconds": 30,
+            "input_bindings": [
+                {"path": "state.report.title", "target": "request.title"},
+                {"value": "markdown", "target": "request.format"},
+            ],
+        }
+    )
+
+    assert params.desc == "Publish report"
+    assert params.retry == 0
+    assert params.timeout_seconds == 30
+    assert [
+        binding.model_dump(mode="json") for binding in params.input_bindings or []
+    ] == [
+        {"path": "state.report.title", "target": "request.title"},
+        {"value": "markdown", "target": "request.format"},
+    ]
+
+
+def test_add_step_from_capability_params_reject_both_input_forms() -> None:
+    with pytest.raises(ValidationError, match="mutually exclusive"):
+        AddStepFromCapabilityParams.model_validate(
+            {
+                "workspace_id": "report",
+                "revision": 1,
+                "step_id": "publish",
+                "capability_name": "demo.report",
+                "input_map": {},
+                "input_bindings": [],
+            }
+        )
+
+
 async def test_rpc_health_and_capability_methods(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)
@@ -1232,6 +1317,99 @@ async def test_rpc_draft_workspace_add_step_from_capability(tmp_path) -> None:
     assert result["status"] == "valid"
 
 
+async def test_rpc_draft_workspace_updates_capability_step_atomically(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:
+        await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "update_step_ws",
+                "capability_name": "wf.std.constant",
+                "name": "update_step",
+            },
+        )
+        before = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "update_step_ws", "include_draft": True},
+        )
+
+        updated = await _rpc(
+            client,
+            "workflow.draft_workspaces.update_capability_step",
+            {
+                "workspace_id": "update_step_ws",
+                "revision": 1,
+                "step_id": "call",
+                "update": {
+                    "desc": "Return the prepared value",
+                    "retry": 0,
+                    "input": [{"value": "prepared", "target": "value"}],
+                },
+            },
+        )
+        inspected = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "update_step_ws", "include_draft": True},
+        )
+
+    assert updated["result"]["revision"] == 2
+    step = inspected["result"]["draft"]["steps"]["call"]
+    assert step["use"] == "wf.std.constant"
+    assert step["desc"] == "Return the prepared value"
+    assert step["retry"] == 0
+    assert step["input"] == [{"value": "prepared", "target": "value"}]
+    assert step["output"] == before["result"]["draft"]["steps"]["call"]["output"]
+    assert inspected["result"]["draft"]["routes"] == before["result"]["draft"]["routes"]
+
+
+async def test_rpc_add_step_from_capability_accepts_canonical_inputs(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:
+        await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "canonical_add_ws",
+                "capability_name": "wf.std.constant",
+                "name": "canonical_add",
+            },
+        )
+        added = await _rpc(
+            client,
+            "workflow.draft_workspaces.add_step_from_capability",
+            {
+                "workspace_id": "canonical_add_ws",
+                "revision": 1,
+                "step_id": "second",
+                "capability_name": "wf.std.constant",
+                "routes": {"ok": "__end__"},
+                "desc": "Return another prepared value",
+                "retry": 0,
+                "timeout_seconds": 30,
+                "input_bindings": [{"value": "second", "target": "value"}],
+            },
+        )
+        inspected = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "canonical_add_ws", "include_draft": True},
+        )
+
+    assert added["result"]["revision"] == 2
+    step = inspected["result"]["draft"]["steps"]["second"]
+    assert step["desc"] == "Return another prepared value"
+    assert step["retry"] == 0
+    assert step["timeout_seconds"] == 30
+    assert step["input"] == [{"value": "second", "target": "value"}]
+
+
 def test_add_draft_step_params_preserve_typed_step_json() -> None:
     foreach = AddDraftStepParams.model_validate(
         {

tests/wf_transport_rpc_http/test_client.py

@@ -6,6 +6,7 @@ import httpx
 import pytest
 from pydantic import TypeAdapter
 
+from wf_api import CapabilityStepUpdate
 from wf_api.models import RawWorkflowPlan, TraceRange
 from wf_api.surface import RouteSource, WorkflowDraftSurface
 from wf_artifacts.drafts.models import (
@@ -875,6 +876,61 @@ async def test_rpc_client_draft_workspace_add_step_from_capability(tmp_path) ->
     assert result["status"] == "valid"
 
 
+async def test_rpc_client_serializes_capability_step_changes() -> 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": 2}
+
+    client = Client()
+    await client.update_capability_step(
+        workspace_id="ws",
+        revision=1,
+        step_id="publish",
+        update=CapabilityStepUpdate.model_validate({"desc": None, "retry": 0}),
+    )
+    await client.add_step_from_capability(
+        workspace_id="ws",
+        revision=2,
+        step_id="publish",
+        capability_name="demo.report",
+        routes={"ok": "__end__"},
+        desc="Publish report",
+        retry=0,
+        timeout_seconds=30,
+        input_bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "title"),
+                target=LocalPath.of("request", "title"),
+            ),
+            InputValueBinding(
+                target=LocalPath.of("request", "format"),
+                value="markdown",
+            ),
+        ],
+    )
+
+    assert calls[0] == {
+        "method": "workflow.draft_workspaces.update_capability_step",
+        "params": {
+            "workspace_id": "ws",
+            "revision": 1,
+            "step_id": "publish",
+            "update": {"desc": None, "retry": 0},
+        },
+    }
+    assert calls[1]["method"] == "workflow.draft_workspaces.add_step_from_capability"
+    assert calls[1]["params"]["input_bindings"] == [
+        {"path": "state.report.title", "target": "request.title"},
+        {"value": "markdown", "target": "request.format"},
+    ]
+    assert calls[1]["params"]["desc"] == "Publish report"
+    assert calls[1]["params"]["retry"] == 0
+    assert calls[1]["params"]["timeout_seconds"] == 30
+
+
 async def test_rpc_client_preserves_nested_local_path_strings() -> None:
     calls: list[dict[str, Any]] = []