feat: replace narrow draft output bind with general bind

src/wf_api/draft_authoring.py

@@ -12,7 +12,7 @@ from wf_core.models.steps import (
     InputBinding,
     OutputBinding,
 )
-from wf_core.paths import GraphSourcePath
+from wf_core.paths import GraphSourcePath, LocalPath
 
 from .constants import (
     DEFAULT_CALL_STEP_ID,
@@ -33,9 +33,26 @@ from .drafts import (
     WorkflowDraftApi,
     _draft_input_maps,
     _draft_output_map,
+    _input_map_from_payload,
 )
 from .operation_context import WorkflowOperationContext
-from .schema_projection import project_output_property_to_state_schema
+from .schema_projection import (
+    project_output_property_to_state_schema,
+    project_property_to_schema_path,
+)
+
+
+def _graph_parts(path: str) -> tuple[str, tuple[str, ...]]:
+    parsed = GraphSourcePath.parse(path)
+    return parsed.root, parsed.parts
+
+
+def _local_field(path: str) -> str:
+    raw = path.removeprefix("local.")
+    parsed = LocalPath.parse(raw)
+    if len(parsed.parts) != 1:
+        raise ValueError("local path must name one capability field")
+    return parsed.parts[0]
 
 
 class WorkflowDraftAuthoringApi:
@@ -149,64 +166,87 @@ class WorkflowDraftAuthoringApi:
             draft=draft,
         )
 
-    async def bind_output_to_state(
+    async def bind_draft(
         self,
         *,
         workspace_id: str,
         revision: int,
         step_id: str,
-        output_field: str,
+        source_path: str,
-        state_path: str,
+        target_path: str,
     ) -> dict[str, Any]:
-        """Declare a state field from a step output and bind that output to it.
+        """Bind a graph path to/from one capability local field with schema projection."""
-
-        This is the common draft-authoring repair for validation errors where a
-        step writes to ``state.x`` before ``state_schema.properties.x`` exists.
-        It deliberately edits only one root state field and one step output map.
-        Route changes remain explicit through ``set_draft_route``.
-        """
         workspace = self.drafts._draft_store().get_workspace(workspace_id)
         step = draft_step(workspace.draft, step_id)
         capability_name = step.get("use")
         if not isinstance(capability_name, str) or not capability_name:
-            raise ValueError(
+            raise ValueError(f"draft step {step_id!r} does not declare a capability use")
-                f"draft step {step_id!r} does not declare a capability use"
+        spec = self.context.specs.get_qualified_spec(capability_name)
+
+        source_root, source_parts = _graph_parts(source_path) if not source_path.startswith("local.") else ("local", LocalPath.parse(source_path).parts)
+        target_root, target_parts = _graph_parts(target_path) if not target_path.startswith("local.") else ("local", LocalPath.parse(target_path).parts)
+
+        if target_root == "local" and source_root in {"input", "state"}:
+            local_field = _local_field(target_path)
+            input_schema = spec.input_schema_contract or spec.input_model.model_json_schema()
+            schema_key = "input_schema" if source_root == "input" else "state_schema"
+            target_schema = workspace.draft.get(schema_key, {})
+            if not isinstance(target_schema, dict):
+                raise ValueError(f"draft {schema_key} must be an object")
+            projected = project_property_to_schema_path(
+                target_schema=target_schema,
+                source_schema=input_schema,
+                source_field=local_field,
+                target_parts=source_parts,
+            )
+            input_map = {
+                **_input_map_from_payload(step.get("input", [])),
+                source_path: local_field,
+            }
+            return await self.drafts.patch_draft_workspace(
+                workspace_id=workspace_id,
+                revision=revision,
+                patch=[
+                    {"op": "replace", "path": f"/{schema_key}", "value": projected},
+                    {
+                        "op": "replace",
+                        "path": f"/steps/{escape_json_pointer(step_id)}/input",
+                        "value": input_bindings_payload(input_map, {}),
+                    },
+                ],
             )
 
-        state_field = state_root_field(state_path)
+        if source_root == "local" and target_root in {"state", "output"}:
-        spec = self.context.specs.get_qualified_spec(capability_name)
+            local_field = _local_field(source_path)
-        output_schema = (
+            output_schema = spec.output_schema_contract or spec.output_model.model_json_schema()
-            spec.output_schema_contract or spec.output_model.model_json_schema()
+            schema_key = "state_schema" if target_root == "state" else "output_schema"
-        )
+            target_schema = workspace.draft.get(schema_key, {})
-        state_schema = workspace.draft.get("state_schema", {})
+            if not isinstance(target_schema, dict):
-        if not isinstance(state_schema, dict):
+                raise ValueError(f"draft {schema_key} must be an object")
-            raise ValueError("draft state_schema must be an object")
+            projected = project_property_to_schema_path(
-        projected = project_output_property_to_state_schema(
+                target_schema=target_schema,
-            state_schema=state_schema,
+                source_schema=output_schema,
-            output_schema=output_schema,
+                source_field=local_field,
-            output_field=output_field,
+                target_parts=target_parts,
-            state_field=state_field,
+            )
-        )
+            output_map = {
-        output_map = {
+                **self.drafts._step_output_map(workspace_id=workspace_id, step_id=step_id),
-            **self.drafts._step_output_map(workspace_id=workspace_id, step_id=step_id),
+                local_field: target_path,
-            output_field: state_path,
+            }
-        }
+            return await self.drafts.patch_draft_workspace(
-        return await self.drafts.patch_draft_workspace(
+                workspace_id=workspace_id,
-            workspace_id=workspace_id,
+                revision=revision,
-            revision=revision,
+                patch=[
-            patch=[
+                    {"op": "replace", "path": f"/{schema_key}", "value": projected},
-                {
+                    {
-                    "op": "replace",
+                        "op": "replace",
-                    "path": "/state_schema",
+                        "path": f"/steps/{escape_json_pointer(step_id)}/output",
-                    "value": projected,
+                        "value": output_bindings_payload(output_map),
-                },
+                    },
-                {
+                ],
-                    "op": "replace",
+            )
-                    "path": f"/steps/{escape_json_pointer(step_id)}/output",
+
-                    "value": output_bindings_payload(output_map),
+        raise ValueError(f"unsupported bind direction: {source_path!r} -> {target_path!r}")
-                },
-            ],
-        )
 
     async def add_step_from_capability(
         self,

src/wf_api/draft_payloads.py

@@ -39,7 +39,7 @@ def input_bindings_payload(
 def output_bindings_payload(output_map: dict[str, str]) -> list[dict[str, Any]]:
     """Serialize draft output maps into canonical string-path binding payloads."""
     return [
-        {"source": _local_path_payload(source), "target": _state_path_payload(target)}
+        {"source": _local_path_payload(source), "target": _graph_source_path_payload(target)}
         for source, target in output_map.items()
     ]
 
@@ -61,5 +61,9 @@ def _graph_path_payload(value: str | GraphSourcePath) -> str:
     return str(path)
 
 
+def _graph_source_path_payload(value: str) -> str:
+    return str(GraphSourcePath.parse(value))
+
+
 def _state_path_payload(value: str) -> str:
     return str(StatePath.parse(value))

src/wf_api/drafts.py

@@ -375,6 +375,21 @@ def _input_maps_from_payload(
     return input_map, input_values
 
 
+def _input_map_from_payload(payload: Any) -> dict[str, str]:
+    """Read stored canonical input bindings back into a source -> local field map."""
+    input_map: dict[str, str] = {}
+    if not isinstance(payload, list):
+        return input_map
+    for item in payload:
+        if not isinstance(item, Mapping):
+            continue
+        if "path" in item and "target" in item:
+            input_map[_path_text(item["path"])] = _path_text(
+                item["target"], expected_root="local"
+            )
+    return input_map
+
+
 def _output_map_from_payload(payload: Any) -> dict[str, str]:
     """Read stored canonical output bindings back into the focused output map."""
     output_map: dict[str, str] = {}
@@ -386,7 +401,6 @@ def _output_map_from_payload(payload: Any) -> dict[str, str]:
         if "source" in item and "target" in item:
             output_map[_path_text(item["source"], expected_root="local")] = _path_text(
                 item["target"],
-                expected_root="state",
             )
     return output_map
 

src/wf_api/service.py

@@ -371,21 +371,21 @@ class WorkflowApi:
             merge=merge,
         )
 
-    async def bind_output_to_state(
+    async def bind_draft(
         self,
         *,
         workspace_id: str,
         revision: int,
         step_id: str,
-        output_field: str,
+        source_path: str,
-        state_path: str,
+        target_path: str,
     ) -> dict[str, Any]:
-        return await self.draft_authoring.bind_output_to_state(
+        return await self.draft_authoring.bind_draft(
             workspace_id=workspace_id,
             revision=revision,
             step_id=step_id,
-            output_field=output_field,
+            source_path=source_path,
-            state_path=state_path,
+            target_path=target_path,
         )
 
     async def add_step_from_capability(

src/wf_api/surface.py

@@ -114,14 +114,14 @@ class WorkflowDraftSurface(Protocol):
         merge: bool = False,
     ) -> dict[str, Any]: ...
 
-    async def bind_output_to_state(
+    async def bind_draft(
         self,
         *,
         workspace_id: str,
         revision: int,
         step_id: str,
-        output_field: str,
+        source_path: str,
-        state_path: str,
+        target_path: str,
     ) -> dict[str, Any]: ...
 
     async def add_step_from_capability(

tests/wf_api/test_drafts_service.py

@@ -530,91 +530,51 @@ async def test_facade_delegates_semantic_authoring_to_authoring_service(
         workspace_id="ws1",
         draft=_echo_draft(),
     )
-    result = await facade.bind_output_to_state(
+    result = await facade.bind_draft(
         workspace_id="ws1",
         revision=1,
         step_id="echo",
-        output_field="echoed",
+        source_path="local.echoed",
-        state_path="state.echoed",
+        target_path="state.echoed",
     )
     assert result["revision"] == 2
 
 
 @pytest.mark.asyncio
-async def test_bind_output_to_state_projects_schema_and_merges_output_map(
+@pytest.mark.asyncio
+async def test_bind_draft_workflow_input_to_step_input_projects_input_schema(
     tmp_path: Path,
 ) -> None:
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_output_state")
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_input")
-    api, service, authoring = _draft_api(artifact_store)
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
-    service.register_connection(
+    draft = {**_echo_draft(), "input_schema": {"type": "object", "properties": {}}}
-        ConnectionConfig(id="demo.personal", server="demo", account="personal")
-    )
-    service.register_specs("demo.personal", _snapshot_tool)
     await api.create_draft_workspace(
-        workspace_id="snapshot_ws",
+        workspace_id="bind_ws",
-        draft={
+        draft=draft,
-            "name": "snapshot",
-            "input_schema": {"type": "object", "properties": {}},
-            "state_schema": {
-                "type": "object",
-                "properties": {"before": {"type": "object"}},
-            },
-            "output_schema": {"type": "object", "properties": {}},
-            "start": "snap",
-            "steps": {
-                "snap": {
-                    "use": "demo.personal.snapshot_tool",
-                    "input": [],
-                    "output": [
-                        {
-                            "source": {"root": "local", "parts": ["before"]},
-                            "target": {"root": "state", "parts": ["before"]},
-                        }
-                    ],
-                }
-            },
-            "routes": {"snap": {"ok": "__end__"}},
-        },
     )
 
-    updated = await authoring.bind_output_to_state(
+    result = await authoring.bind_draft(
-        workspace_id="snapshot_ws",
+        workspace_id="bind_ws",
         revision=1,
-        step_id="snap",
+        step_id="echo",
-        output_field="after",
+        source_path="input.text",
-        state_path="state.after",
+        target_path="local.text",
-    )
-    fetched = await api.get_draft_workspace(
-        workspace_id="snapshot_ws",
-        include_draft=True,
     )
+    workspace = await api.get_draft_workspace(workspace_id="bind_ws", include_draft=True)
 
-    draft = fetched["draft"]
+    assert result["revision"] == 2
-    assert updated["revision"] == 2
+    assert workspace["draft"]["input_schema"]["properties"]["text"]["type"] == "string"
-    assert draft["state_schema"]["properties"]["after"]["$ref"] == "#/$defs/_Snapshot"
+    assert workspace["draft"]["steps"]["echo"]["input"] == [
-    assert draft["state_schema"]["$defs"]["_Snapshot"]["properties"]["clicked"] == {
+        {"target": "text", "path": "input.text"}
-        "title": "Clicked",
-        "type": "boolean",
-    }
-    assert draft["steps"]["snap"]["output"] == [
-        {
-            "source": "before",
-            "target": "state.before",
-        },
-        {
-            "source": "after",
-            "target": "state.after",
-        },
     ]
 
 
 @pytest.mark.asyncio
-async def test_bind_output_to_state_rejects_nested_state_path(tmp_path: Path) -> None:
+async def test_bind_draft_output_to_nested_state_projects_state_schema(
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_nested_state")
+    tmp_path: Path,
-    api, service, authoring = _draft_api(artifact_store)
+) -> None:
-    service.register_connection(
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_output_nested")
-        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
-    )
     service.register_specs("demo.personal", _snapshot_tool)
     await api.create_draft_workspace(
         workspace_id="snapshot_ws",
@@ -624,97 +584,43 @@ async def test_bind_output_to_state_rejects_nested_state_path(tmp_path: Path) ->
             "state_schema": {"type": "object", "properties": {}},
             "output_schema": {"type": "object", "properties": {}},
             "start": "snap",
-            "steps": {
+            "steps": {"snap": {"use": "demo.personal.snapshot_tool", "input": [], "output": []}},
-                "snap": {
-                    "use": "demo.personal.snapshot_tool",
-                    "input": [],
-                    "output": [],
-                }
-            },
             "routes": {"snap": {"ok": "__end__"}},
         },
     )
 
-    with pytest.raises(ValueError, match="state_path must name one root field"):
+    result = await authoring.bind_draft(
-        await authoring.bind_output_to_state(
+        workspace_id="snapshot_ws",
-            workspace_id="snapshot_ws",
+        revision=1,
-            revision=1,
+        step_id="snap",
-            step_id="snap",
+        source_path="local.after",
-            output_field="after",
+        target_path="state.session.after",
-            state_path="state.after.clicked",
+    )
-        )
+    workspace = await api.get_draft_workspace(workspace_id="snapshot_ws", include_draft=True)
+
+    assert result["revision"] == 2
+    assert (
+        workspace["draft"]["state_schema"]["properties"]["session"]["properties"]["after"]["$ref"]
+        == "#/$defs/_Snapshot"
+    )
+    assert workspace["draft"]["steps"]["snap"]["output"] == [
+        {"source": "after", "target": "state.session.after"}
+    ]
 
 
 @pytest.mark.asyncio
-async def test_bind_output_to_state_rejects_missing_output_field(
+async def test_bind_draft_rejects_unsupported_direction(tmp_path: Path) -> None:
-    tmp_path: Path,
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_bad_direction")
-) -> None:
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_missing_output")
+    await api.create_draft_workspace(workspace_id="bind_ws", draft=_echo_draft())
-    api, service, authoring = _draft_api(artifact_store)
-    service.register_connection(
-        ConnectionConfig(id="demo.personal", server="demo", account="personal")
-    )
-    service.register_specs("demo.personal", _snapshot_tool)
-    await api.create_draft_workspace(
-        workspace_id="snapshot_ws",
-        draft={
-            "name": "snapshot",
-            "input_schema": {"type": "object", "properties": {}},
-            "state_schema": {"type": "object", "properties": {}},
-            "output_schema": {"type": "object", "properties": {}},
-            "start": "snap",
-            "steps": {
-                "snap": {
-                    "use": "demo.personal.snapshot_tool",
-                    "input": [],
-                    "output": [],
-                }
-            },
-            "routes": {"snap": {"ok": "__end__"}},
-        },
-    )
 
-    with pytest.raises(ValueError, match="output field 'missing'"):
+    with pytest.raises(ValueError, match="unsupported bind direction"):
-        await authoring.bind_output_to_state(
+        await authoring.bind_draft(
-            workspace_id="snapshot_ws",
+            workspace_id="bind_ws",
             revision=1,
-            step_id="snap",
+            step_id="echo",
-            output_field="missing",
+            source_path="input.message",
-            state_path="state.missing",
+            target_path="state.message",
-        )
-
-
-@pytest.mark.asyncio
-async def test_bind_output_to_state_rejects_step_without_capability_use(
-    tmp_path: Path,
-) -> None:
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_no_use")
-    api, _service, authoring = _draft_api(artifact_store)
-    await api.create_draft_workspace(
-        workspace_id="snapshot_ws",
-        draft={
-            "name": "snapshot",
-            "input_schema": {"type": "object", "properties": {}},
-            "state_schema": {"type": "object", "properties": {}},
-            "output_schema": {"type": "object", "properties": {}},
-            "start": "snap",
-            "steps": {
-                "snap": {
-                    "input": [],
-                    "output": [],
-                }
-            },
-            "routes": {"snap": {"ok": "__end__"}},
-        },
-    )
-
-    with pytest.raises(ValueError, match="does not declare a capability use"):
-        await authoring.bind_output_to_state(
-            workspace_id="snapshot_ws",
-            revision=1,
-            step_id="snap",
-            output_field="after",
-            state_path="state.after",
         )