feat: bind nested local draft paths

docs/superpowers/plans/2026-07-22-nested-local-draft-bindings.md

@@ -241,7 +241,7 @@
 - Preserves: `WorkflowDraftAuthoringApi.bind_draft(...)` request and response signatures.
 - Produces: complete nested `LocalPath` values in existing input/output binding payloads.
 
-- [ ] **Step 1: Add a nested capability fixture with Pydantic-style definitions**
+- [x] **Step 1: Add a nested capability fixture with Pydantic-style definitions**
 
   Add this test-only capability beside the existing echo/snapshot fixtures in `tests/wf_api/test_drafts_service.py`:
 
@@ -275,7 +275,7 @@
   service.register_specs("demo.personal", _nested_report)
   ```
 
-- [ ] **Step 2: Add failing input/state-to-local tests**
+- [x] **Step 2: Add failing input/state-to-local tests**
 
   Add tests that call:
 
@@ -312,7 +312,7 @@
 
   For the state case, assert the nested state schema remains valid and the stored target is exactly `report.title`.
 
-- [ ] **Step 3: Add failing nested local-output tests**
+- [x] **Step 3: Add failing nested local-output tests**
 
   Cover both supported directions:
 
@@ -343,13 +343,13 @@
 
   Also assert both `state_schema` and `output_schema` contain the projected nested string schema.
 
-- [ ] **Step 4: Add failure atomicity and revision-precedence tests**
+- [x] **Step 4: Add failure atomicity and revision-precedence tests**
 
   Add one current-revision test using `local.report.missing -> state.report.missing`. Assert the complete path appears in the `ValueError`, then fetch the workspace and assert `revision == 1` and the step output remains unchanged.
 
   Add one stale-revision test with the same invalid nested path after first moving the workspace to revision 2. Assert the returned diagnostic code is `revision_conflict` and no nested-path `ValueError` escapes.
 
-- [ ] **Step 5: Run the new API tests and confirm the one-field restriction fails**
+- [x] **Step 5: Run the new API tests and confirm the one-field restriction fails**
 
   Run:
 
@@ -359,7 +359,7 @@
 
   Expected: nested local endpoints fail with `local path must name one capability field`.
 
-- [ ] **Step 6: Replace root-field handling with complete local parts**
+- [x] **Step 6: Replace root-field handling with complete local parts**
 
   In `WorkflowDraftAuthoringApi.bind_draft`:
 
@@ -391,7 +391,7 @@
   }
   ```
 
-- [ ] **Step 7: Run focused API and canonical-model regressions**
+- [x] **Step 7: Run focused API and canonical-model regressions**
 
   Run:
 
@@ -404,7 +404,7 @@
 
   Expected: nested and existing single-field cases pass, and canonical nested mapping validation remains green.
 
-- [ ] **Step 8: Commit focused nested bind support**
+- [x] **Step 8: Commit focused nested bind support**
 
   ```bash
   git add src/wf_api/draft_authoring.py tests/wf_api/test_drafts_service.py

src/wf_api/draft_authoring.py

@@ -58,6 +58,7 @@ from .operation_context import WorkflowOperationContext
 from .schema_projection import (
     project_output_property_to_state_schema,
     project_property_to_schema_path,
+    project_schema_path_to_schema_path,
     schema_path_exists,
 )
 
@@ -67,13 +68,6 @@ def _graph_parts(path: str) -> tuple[str, tuple[str, ...]]:
     return parsed.root, parsed.parts
 
 
-def _local_field(path: str) -> str:
-    parts = _local_parts(path)
-    if len(parts) != 1:
-        raise ValueError("local path must name one capability field")
-    return parts[0]
-
-
 def _local_parts(path: str) -> tuple[str, ...]:
     """Parse a CLI local-root path as the rootless core LocalPath value."""
     return LocalPath.parse(path.removeprefix("local.")).parts
@@ -318,7 +312,7 @@ class WorkflowDraftAuthoringApi:
         source_path: str,
         target_path: str,
     ) -> dict[str, Any]:
-        """Bind a graph path to/from one capability local field, projecting missing schema when needed."""
+        """Bind a graph path to or from one capability-local path."""
         checked = self._workspace_if_revision_matches(
             workspace_id=workspace_id,
             revision=revision,
@@ -354,7 +348,7 @@ class WorkflowDraftAuthoringApi:
             target_root, target_parts = _graph_parts(target_path)
 
         if target_root == "local" and source_root in {"input", "state"}:
-            local_field = _local_field(target_path)
+            local_path = format_toml_path_segments(target_parts)
             input_schema = (
                 spec.input_schema_contract or spec.input_model.model_json_schema()
             )
@@ -365,15 +359,15 @@ class WorkflowDraftAuthoringApi:
             if schema_path_exists(target_schema, source_parts):
                 projected = target_schema
             else:
-                projected = project_property_to_schema_path(
+                projected = project_schema_path_to_schema_path(
                     target_schema=target_schema,
                     source_schema=input_schema,
-                    source_field=local_field,
+                    source_parts=target_parts,
                     target_parts=source_parts,
                 )
             input_map = {
                 **_input_map_from_payload(step.get("input", [])),
-                source_path: local_field,
+                source_path: local_path,
             }
             return await self.drafts.patch_draft_workspace(
                 workspace_id=workspace_id,
@@ -389,7 +383,7 @@ class WorkflowDraftAuthoringApi:
             )
 
         if source_root == "local" and target_root == "output":
-            local_field = _local_field(source_path)
+            local_path = format_toml_path_segments(source_parts)
             output_schema_source = (
                 spec.output_schema_contract or spec.output_model.model_json_schema()
             )
@@ -399,10 +393,10 @@ class WorkflowDraftAuthoringApi:
             state_schema = workspace.draft.get("state_schema", {})
             if not isinstance(state_schema, dict):
                 raise ValueError("draft state_schema must be an object")
-            projected_state = project_property_to_schema_path(
+            projected_state = project_schema_path_to_schema_path(
                 target_schema=state_schema,
                 source_schema=output_schema_source,
-                source_field=local_field,
+                source_parts=source_parts,
                 target_parts=target_parts,
                 allow_existing_equivalent=True,
             )
@@ -410,10 +404,10 @@ class WorkflowDraftAuthoringApi:
             output_schema = workspace.draft.get("output_schema", {})
             if not isinstance(output_schema, dict):
                 raise ValueError("draft output_schema must be an object")
-            projected_output = project_property_to_schema_path(
+            projected_output = project_schema_path_to_schema_path(
                 target_schema=output_schema,
                 source_schema=output_schema_source,
-                source_field=local_field,
+                source_parts=source_parts,
                 target_parts=target_parts,
                 allow_existing_equivalent=True,
             )
@@ -421,10 +415,10 @@ class WorkflowDraftAuthoringApi:
             current_output_map = self.drafts._step_output_map(
                 workspace_id=workspace_id, step_id=step_id
             )
-            previous_state_path = current_output_map.get(local_field)
+            previous_state_path = current_output_map.get(local_path)
             output_map = {
                 **current_output_map,
-                local_field: state_path_str,
+                local_path: state_path_str,
             }
 
             existing_output = workspace.draft.get("output")
@@ -474,17 +468,17 @@ class WorkflowDraftAuthoringApi:
             )
 
         if source_root == "local" and target_root == "state":
-            local_field = _local_field(source_path)
+            local_path = format_toml_path_segments(source_parts)
             output_schema = (
                 spec.output_schema_contract or spec.output_model.model_json_schema()
             )
             target_schema = workspace.draft.get("state_schema", {})
             if not isinstance(target_schema, dict):
                 raise ValueError("draft state_schema must be an object")
-            projected = project_property_to_schema_path(
+            projected = project_schema_path_to_schema_path(
                 target_schema=target_schema,
                 source_schema=output_schema,
-                source_field=local_field,
+                source_parts=source_parts,
                 target_parts=target_parts,
                 allow_existing_equivalent=True,
             )
@@ -492,7 +486,7 @@ class WorkflowDraftAuthoringApi:
                 **self.drafts._step_output_map(
                     workspace_id=workspace_id, step_id=step_id
                 ),
-                local_field: target_path,
+                local_path: target_path,
             }
             return await self.drafts.patch_draft_workspace(
                 workspace_id=workspace_id,

tests/wf_api/test_drafts_service.py

@@ -73,6 +73,47 @@ def _snapshot_tool(payload: _SnapshotInput) -> _SnapshotOutput:
     return _SnapshotOutput(after=_Snapshot(clicked=True))
 
 
+class _ReportInputValue(BaseModel):
+    title: str
+
+
+class _NestedReportInput(BaseModel):
+    report: _ReportInputValue
+
+
+class _ReportOutputValue(BaseModel):
+    markdown: str
+
+
+class _NestedReportOutput(BaseModel):
+    report: _ReportOutputValue
+
+
+@node(name="nested_report", outcomes=("ok",))
+def _nested_report(payload: _NestedReportInput) -> _NestedReportOutput:
+    return _NestedReportOutput(
+        report=_ReportOutputValue(markdown=f"# {payload.report.title}")
+    )
+
+
+def _nested_report_draft() -> dict[str, Any]:
+    return {
+        "name": "nested_report",
+        "input_schema": {"type": "object", "properties": {}},
+        "state_schema": {"type": "object", "properties": {}},
+        "output_schema": {"type": "object", "properties": {}},
+        "start": "render",
+        "steps": {
+            "render": {
+                "use": "demo.personal.nested_report",
+                "input": [],
+                "output": [],
+            }
+        },
+        "routes": {"render": {"ok": "__end__"}},
+    }
+
+
 def _draft_api(
     artifact_store: FileWorkflowArtifactStore,
     *,
@@ -1085,6 +1126,243 @@ async def test_bind_draft_output_to_nested_state_projects_state_schema(
     ]
 
 
+@pytest.mark.asyncio
+async def test_bind_draft_projects_nested_local_input_schema(tmp_path: Path) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "nested_local_input"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="input.title",
+        target_path="local.report.title",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+    draft = workspace["draft"]
+
+    assert result["revision"] == 2
+    assert draft["input_schema"]["properties"]["title"]["type"] == "string"
+    assert draft["steps"]["render"]["input"] == [
+        {"target": "report.title", "path": "input.title"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_reuses_nested_state_for_nested_local_input(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "nested_local_state_input"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    draft = _nested_report_draft()
+    draft["state_schema"] = {
+        "type": "object",
+        "properties": {
+            "report": {
+                "type": "object",
+                "properties": {"title": {"type": "string"}},
+            }
+        },
+    }
+    await api.create_draft_workspace(workspace_id="nested", draft=draft)
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="state.report.title",
+        target_path="local.report.title",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+
+    assert result["revision"] == 2
+    assert workspace["draft"]["steps"]["render"]["input"] == [
+        {"target": "report.title", "path": "state.report.title"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_projects_nested_local_output_to_state(tmp_path: Path) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "nested_local_output_state"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="local.report.markdown",
+        target_path="state.report.markdown",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+    draft = workspace["draft"]
+
+    assert result["revision"] == 2
+    assert draft["steps"]["render"]["output"] == [
+        {"source": "report.markdown", "target": "state.report.markdown"}
+    ]
+    assert (
+        draft["state_schema"]["properties"]["report"]["properties"]["markdown"]["type"]
+        == "string"
+    )
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_lowers_nested_local_output_to_public_output(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "nested_local_public_output"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="local.report.markdown",
+        target_path="output.report.markdown",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+    draft = workspace["draft"]
+
+    assert result["revision"] == 2
+    assert draft["steps"]["render"]["output"] == [
+        {"source": "report.markdown", "target": "state.report.markdown"}
+    ]
+    assert draft["output"] == [
+        {"path": "state.report.markdown", "target": "report.markdown"}
+    ]
+    assert (
+        draft["state_schema"]["properties"]["report"]["properties"]["markdown"]["type"]
+        == "string"
+    )
+    assert (
+        draft["output_schema"]["properties"]["report"]["properties"]["markdown"]["type"]
+        == "string"
+    )
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_rebinds_nested_local_public_output(tmp_path: Path) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "rebind_nested_local_public_output"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+    first = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="local.report.markdown",
+        target_path="output.report.markdown",
+    )
+
+    second = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=first["revision"],
+        step_id="render",
+        source_path="local.report.markdown",
+        target_path="output.published.markdown",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+
+    assert second["revision"] == 3
+    assert workspace["draft"]["steps"]["render"]["output"] == [
+        {"source": "report.markdown", "target": "state.published.markdown"}
+    ]
+    assert workspace["draft"]["output"] == [
+        {"path": "state.published.markdown", "target": "published.markdown"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_rejects_missing_nested_local_output_without_mutation(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "missing_nested_local_output"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+
+    with pytest.raises(
+        ValueError,
+        match="source schema path 'report.missing' is not declared",
+    ):
+        await authoring.bind_draft(
+            workspace_id="nested",
+            revision=1,
+            step_id="render",
+            source_path="local.report.missing",
+            target_path="state.report.missing",
+        )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+
+    assert workspace["revision"] == 1
+    assert workspace["draft"]["steps"]["render"]["output"] == []
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_stale_revision_precedes_nested_local_path_error(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "stale_nested_local_output"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    await api.create_draft_workspace(
+        workspace_id="nested",
+        draft=_nested_report_draft(),
+    )
+    await api.patch_draft_workspace(
+        workspace_id="nested",
+        revision=1,
+        patch=[{"op": "replace", "path": "/name", "value": "nested_v2"}],
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="local.report.missing",
+        target_path="state.report.missing",
+    )
+
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+
+
 @pytest.mark.asyncio
 async def test_bind_draft_rejects_unsupported_direction(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_bad_direction")