fix: preserve omitted capability metadata

src/wf_api/draft_authoring.py

@@ -584,22 +584,19 @@ class WorkflowDraftAuthoringApi:
             changes["input"] = update.input
 
         changed = current.model_copy(update=changes)
-        if projected is None:
-            step_payload = dict(deepcopy(step))
-            for field in ("desc", "retry", "timeout_seconds"):
-                if field not in update.model_fields_set:
-                    continue
-                value = getattr(update, field)
-                if value is None:
-                    step_payload.pop(field, None)
-                else:
-                    step_payload[field] = value
-        else:
-            step_payload = changed.model_dump(
-                mode="json",
-                by_alias=True,
-                exclude_none=True,
-            )
+        # Mutate a raw copy so omitted fields preserve their exact stored
+        # presence, including legacy explicit-null metadata.
+        step_payload = dict(deepcopy(step))
+        for field in ("desc", "retry", "timeout_seconds"):
+            if field not in update.model_fields_set:
+                continue
+            value = getattr(update, field)
+            if value is None:
+                step_payload.pop(field, None)
+            else:
+                step_payload[field] = value
+        if projected is not None:
+            step_payload["input"] = projected.payload
         input_schema = (
             projected.input_schema
             if projected is not None
@@ -631,7 +628,7 @@ class WorkflowDraftAuthoringApi:
         next_steps[step_id] = step_payload
         next_draft["input_schema"] = input_schema
         next_draft["state_schema"] = state_schema
-        return await self.drafts._replace_validated_draft_document(
+        return await self.drafts.replace_validated_draft_document(
             workspace_id=workspace_id,
             revision=revision,
             draft=next_draft,

src/wf_api/drafts.py

@@ -8,6 +8,7 @@ from wf_artifacts import (
     DraftWorkspaceStore,
     compile_workflow_draft,
     patch_workflow_draft,
+    replace_validated_draft_document,
     validate_workflow_draft,
 )
 from wf_artifacts import (
@@ -19,9 +20,6 @@ from wf_artifacts import (
 from wf_artifacts import (
     patch_draft_workspace as patch_draft_workspace_record,
 )
-from wf_artifacts.draft_workspaces.api import (
-    _replace_validated_draft_document as replace_validated_draft_document_record,
-)
 from wf_core.models.schemas import NodeDef
 from wf_core.models.steps import (
     InputBinding,
@@ -292,7 +290,7 @@ class WorkflowDraftApi:
             node_defs_for_draft=self._node_defs_for_draft,
         )
 
-    async def _replace_validated_draft_document(
+    async def replace_validated_draft_document(
         self,
         *,
         workspace_id: str,
@@ -300,7 +298,7 @@ class WorkflowDraftApi:
         draft: dict[str, Any],
     ) -> dict[str, Any]:
         """Persist a focused, structurally validated edit without provider lookup."""
-        return replace_validated_draft_document_record(
+        return replace_validated_draft_document(
             self._draft_store(),
             workspace_id=workspace_id,
             revision=revision,

src/wf_artifacts/__init__.py

@@ -12,6 +12,7 @@ from .draft_workspaces import (
     ensure_workspace_id,
     get_draft_workspace,
     patch_draft_workspace,
+    replace_validated_draft_document,
     summarize_draft_workspace,
 )
 from .drafts import (
@@ -91,6 +92,7 @@ __all__ = [
     "logical_ref_for_concrete_ref",
     "normalize_plan_node_refs",
     "patch_draft_workspace",
+    "replace_validated_draft_document",
     "patch_workflow_draft",
     "summarize_draft_workspace",
     "validate_deployment_dependencies",

src/wf_artifacts/draft_workspaces/__init__.py

@@ -2,6 +2,7 @@ from .api import (
     create_draft_workspace,
     get_draft_workspace,
     patch_draft_workspace,
+    replace_validated_draft_document,
 )
 from .models import (
     WorkflowDraftWorkspace,
@@ -23,5 +24,6 @@ __all__ = [
     "ensure_workspace_id",
     "get_draft_workspace",
     "patch_draft_workspace",
+    "replace_validated_draft_document",
     "summarize_draft_workspace",
 ]

src/wf_artifacts/draft_workspaces/api.py

@@ -102,7 +102,7 @@ def patch_draft_workspace(
     return summarize_draft_workspace(next_workspace)
 
 
-def _replace_validated_draft_document(
+def replace_validated_draft_document(
     store: DraftWorkspaceStore,
     *,
     workspace_id: str,

src/wf_mcp/workflow_surface/tools.py

@@ -605,7 +605,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         title="Update Capability Step",
         description=(
             "Update capability-step metadata and optionally replace its complete "
-            "canonical input bindings. Preserves use, routes, and outputs."
+            "canonical input bindings. Omitted metadata is preserved; explicit "
+            "null clears it. Preserves use, routes, and outputs."
         ),
     )
     async def update_capability_step(

tests/wf_api/test_drafts_service.py

@@ -187,6 +187,50 @@ async def test_update_capability_step_removes_stored_null_metadata(
     assert "retry" not in after["draft"]["steps"]["echo"]
 
 
+@pytest.mark.asyncio
+async def test_update_capability_step_input_preserves_omitted_null_metadata(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_null_metadata"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["steps"]["echo"].update(
+        {
+            "desc": None,
+            "retry": None,
+            "timeout_seconds": None,
+        }
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=draft)
+
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate(
+            input=[
+                InputValueBinding(
+                    target=LocalPath.of("text"),
+                    value="replacement",
+                )
+            ]
+        ),
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+    step = inspected["draft"]["steps"]["echo"]
+
+    assert result["revision"] == 2
+    assert step["desc"] is None
+    assert step["retry"] is None
+    assert step["timeout_seconds"] is None
+    assert step["input"] == [{"value": "replacement", "target": "text"}]
+
+
 @pytest.mark.asyncio
 async def test_update_capability_step_metadata_does_not_resolve_capability(
     tmp_path: Path,