refactor: centralize schema path projection

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

@@ -36,7 +36,7 @@
 - Preserves: `project_property_to_schema_path(...)` as a root-property compatibility wrapper.
 - Removes: private duplicate `_schema_path_exists` implementations from `wf_api.drafts` and `wf_api.draft_authoring`.
 
-- [ ] **Step 1: Add failing tests for inline and referenced nested source paths**
+- [x] **Step 1: Add failing tests for inline and referenced nested source paths**
 
   Extend `tests/wf_api/test_schema_projection.py` imports with the two new public operations, then add these cases:
 
@@ -109,7 +109,7 @@
       assert "Report" in projected["definitions"]
   ```
 
-- [ ] **Step 2: Add failing tests for existence and precise failures**
+- [x] **Step 2: Add failing tests for existence and precise failures**
 
   Add tests that pin the bounded behavior:
 
@@ -175,7 +175,7 @@
 
   Retain the existing target-conflict and equivalent-target tests; they are regression coverage for the generalized operation.
 
-- [ ] **Step 3: Run the schema tests and confirm the new imports fail**
+- [x] **Step 3: Run the schema tests and confirm the new imports fail**
 
   Run:
 
@@ -185,7 +185,7 @@
 
   Expected: collection fails because `schema_path_exists` and `project_schema_path_to_schema_path` are not exported yet.
 
-- [ ] **Step 4: Implement one bounded path resolver and the generalized projector**
+- [x] **Step 4: Implement one bounded path resolver and the generalized projector**
 
   In `src/wf_api/schema_projection.py`:
 
@@ -198,7 +198,7 @@
 
   The implementation must retain the selected leaf unchanged. For example, when the leaf is `{"$ref": "#/$defs/Markdown"}`, copy that reference and merge the source definition blocks rather than replacing the leaf with the resolved definition.
 
-- [ ] **Step 5: Replace both duplicate existence helpers with the shared function**
+- [x] **Step 5: Replace both duplicate existence helpers with the shared function**
 
   In `src/wf_api/drafts.py` and `src/wf_api/draft_authoring.py`:
 
@@ -208,7 +208,7 @@
 
   Delete each private `_schema_path_exists` definition and replace its callers with `schema_path_exists(...)`. Remove now-unused `Mapping` or `Sequence` imports only when no other symbol in that module needs them.
 
-- [ ] **Step 6: Run focused tests and quality checks**
+- [x] **Step 6: Run focused tests and quality checks**
 
   Run:
 
@@ -221,7 +221,7 @@
 
   Expected: all schema projection and draft service tests pass; all quality checks are clean.
 
-- [ ] **Step 7: Commit the shared schema operation**
+- [x] **Step 7: Commit the shared schema operation**
 
   ```bash
   git add src/wf_api/schema_projection.py src/wf_api/drafts.py src/wf_api/draft_authoring.py tests/wf_api/test_schema_projection.py

src/wf_api/draft_authoring.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any
 
@@ -58,6 +58,7 @@ from .operation_context import WorkflowOperationContext
 from .schema_projection import (
     project_output_property_to_state_schema,
     project_property_to_schema_path,
+    schema_path_exists,
 )
 
 
@@ -78,18 +79,6 @@ def _local_parts(path: str) -> tuple[str, ...]:
     return LocalPath.parse(path.removeprefix("local.")).parts
 
 
-def _schema_path_exists(schema: Mapping[str, Any], parts: Sequence[str]) -> bool:
-    current: Any = schema
-    for part in parts:
-        if not isinstance(current, Mapping):
-            return False
-        properties = current.get("properties")
-        if not isinstance(properties, Mapping) or part not in properties:
-            return False
-        current = properties[part]
-    return True
-
-
 class WorkflowDraftAuthoringApi:
     """Capability-aware semantic edits over revisioned workflow drafts."""
 
@@ -373,7 +362,7 @@ class WorkflowDraftAuthoringApi:
             target_schema = workspace.draft.get(schema_key, {})
             if not isinstance(target_schema, dict):
                 raise ValueError(f"draft {schema_key} must be an object")
-            if _schema_path_exists(target_schema, source_parts):
+            if schema_path_exists(target_schema, source_parts):
                 projected = target_schema
             else:
                 projected = project_property_to_schema_path(
@@ -628,7 +617,7 @@ class WorkflowDraftAuthoringApi:
                 if source_root == "input"
                 else projected_state_schema
             )
-            if _schema_path_exists(target_schema, source_parts):
+            if schema_path_exists(target_schema, source_parts):
                 continue
             projected = project_property_to_schema_path(
                 target_schema=target_schema,

src/wf_api/drafts.py

@@ -45,7 +45,7 @@ from .draft_payloads import (
     output_bindings_payload as _draft_output_bindings_payload,
 )
 from .operation_context import WorkflowOperationContext
-from .schema_projection import project_property_to_schema_path
+from .schema_projection import project_property_to_schema_path, schema_path_exists
 
 
 def _empty_object_schema() -> dict[str, Any]:
@@ -541,7 +541,7 @@ class WorkflowDraftApi:
                 target_parts = parse_toml_path_segments(target)
             except ValueError:
                 continue
-            if _schema_path_exists(projected, target_parts):
+            if schema_path_exists(projected, target_parts):
                 continue
             try:
                 source_path = GraphSourcePath.parse(source)
@@ -599,18 +599,6 @@ def _workflow_source_schema(
     return schema if isinstance(schema, dict) else None
 
 
-def _schema_path_exists(schema: Mapping[str, Any], parts: Sequence[str]) -> bool:
-    current: Any = schema
-    for part in parts:
-        if not isinstance(current, Mapping):
-            return False
-        properties = current.get("properties")
-        if not isinstance(properties, Mapping) or part not in properties:
-            return False
-        current = properties[part]
-    return True
-
-
 def _draft_input_maps(
     *,
     input: Sequence[InputBinding] | None,

src/wf_api/schema_projection.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from typing import Any
 
@@ -8,29 +9,38 @@ from jsonschema import Draft202012Validator, SchemaError
 JsonObject = dict[str, Any]
 
 
-def project_property_to_schema_path(
+def schema_path_exists(
+    schema: Mapping[str, Any],
+    parts: Sequence[str],
+) -> bool:
+    """Return whether an object-property path exists in a JSON Schema document."""
+    try:
+        _schema_at_path(schema, parts, label="schema")
+    except ValueError:
+        return False
+    return True
+
+
+def project_schema_path_to_schema_path(
     *,
     target_schema: JsonObject,
     source_schema: JsonObject,
-    source_field: str,
+    source_parts: tuple[str, ...],
     target_parts: tuple[str, ...],
     allow_existing_equivalent: bool = False,
 ) -> JsonObject:
-    """Copy one source property schema into a target JSON Schema object path.
+    """Copy one nested source subschema into a target object-property path."""
-
+    if not source_parts:
-    ``allow_existing_equivalent`` accepts exact schema equality only. It does
+        raise ValueError("source schema path must not be empty")
-    not attempt semantic JSON Schema compatibility analysis.
-    """
     if not target_parts:
         raise ValueError("target schema path must not be empty")
     _check_schema("target_schema", target_schema)
     _check_schema("source_schema", source_schema)
-    source_properties = source_schema.get("properties")
+    source_value = _schema_at_path(
-    if not isinstance(source_properties, dict) or source_field not in source_properties:
+        source_schema,
-        raise ValueError(f"source field {source_field!r} is not declared")
+        source_parts,
-    source_property = source_properties[source_field]
+        label="source schema",
-    if not isinstance(source_property, dict):
+    )
-        raise ValueError(f"source field {source_field!r} is not a JSON Schema object")
 
     projected = deepcopy(target_schema)
     _ensure_object_schema(projected, "target_schema")
@@ -55,13 +65,13 @@ def project_property_to_schema_path(
     )
     leaf = target_parts[-1]
     if leaf in properties:
-        if allow_existing_equivalent and properties[leaf] == source_property:
+        if allow_existing_equivalent and properties[leaf] == source_value:
             _merge_definition_block(projected, source_schema, "$defs")
             _merge_definition_block(projected, source_schema, "definitions")
             _check_schema("projected target_schema", projected)
             return projected
         raise ValueError(f"schema path {'.'.join(target_parts)!r} already exists")
-    properties[leaf] = deepcopy(source_property)
+    properties[leaf] = deepcopy(source_value)
 
     _merge_definition_block(projected, source_schema, "$defs")
     _merge_definition_block(projected, source_schema, "definitions")
@@ -69,6 +79,40 @@ def project_property_to_schema_path(
     return projected
 
 
+def project_property_to_schema_path(
+    *,
+    target_schema: JsonObject,
+    source_schema: JsonObject,
+    source_field: str,
+    target_parts: tuple[str, ...],
+    allow_existing_equivalent: bool = False,
+) -> JsonObject:
+    """Copy one source property schema into a target JSON Schema object path.
+
+    ``allow_existing_equivalent`` accepts exact schema equality only. It does
+    not attempt semantic JSON Schema compatibility analysis.
+    """
+    try:
+        return project_schema_path_to_schema_path(
+            target_schema=target_schema,
+            source_schema=source_schema,
+            source_parts=(source_field,),
+            target_parts=target_parts,
+            allow_existing_equivalent=allow_existing_equivalent,
+        )
+    except ValueError as exc:
+        message = str(exc)
+        if message == f"source schema path {source_field!r} is not declared":
+            raise ValueError(f"source field {source_field!r} is not declared") from exc
+        if message == (
+            f"source schema path {source_field!r} is not a JSON Schema object"
+        ):
+            raise ValueError(
+                f"source field {source_field!r} is not a JSON Schema object"
+            ) from exc
+        raise
+
+
 def project_output_property_to_state_schema(
     *,
     state_schema: JsonObject,
@@ -119,6 +163,88 @@ def _check_schema(name: str, schema: JsonObject) -> None:
         raise ValueError(f"{name} is not valid JSON Schema: {exc.message}") from exc
 
 
+def _schema_at_path(
+    root_schema: Mapping[str, Any],
+    parts: Sequence[str],
+    *,
+    label: str,
+) -> Mapping[str, Any]:
+    """Select an object-property subschema, following bounded local references."""
+    current = root_schema
+    traversed: tuple[str, ...] = ()
+    for part in parts:
+        current = _resolve_local_reference(
+            root_schema,
+            current,
+            label=".".join(traversed) or label,
+        )
+        schema_type = current.get("type")
+        if schema_type is not None and schema_type != "object":
+            blocking_path = ".".join(traversed) or label
+            raise ValueError(f"{label} path {blocking_path!r} is not an object")
+        properties = current.get("properties")
+        full_path = ".".join((*traversed, part))
+        if not isinstance(properties, Mapping) or part not in properties:
+            raise ValueError(f"{label} path {full_path!r} is not declared")
+        child = properties[part]
+        if not isinstance(child, Mapping):
+            raise ValueError(f"{label} path {full_path!r} is not a JSON Schema object")
+        current = child
+        traversed = (*traversed, part)
+    if parts:
+        # Validate a selected leaf reference without replacing it. Projection must
+        # preserve the reference itself so the copied schema can share merged defs.
+        _resolve_local_reference(
+            root_schema,
+            current,
+            label=".".join(traversed),
+        )
+    return current
+
+
+def _resolve_local_reference(
+    root_schema: Mapping[str, Any],
+    candidate: Mapping[str, Any],
+    *,
+    label: str,
+) -> Mapping[str, Any]:
+    """Resolve repository-generated local refs without becoming a full resolver."""
+    current = candidate
+    seen: set[str] = set()
+    while "$ref" in current:
+        reference = current["$ref"]
+        if not isinstance(reference, str):
+            raise ValueError(f"schema path {label!r} has a non-string reference")
+        if reference in seen:
+            raise ValueError(f"cyclic reference {reference!r} at schema path {label!r}")
+        seen.add(reference)
+
+        if reference.startswith("#/$defs/"):
+            pointer = reference.removeprefix("#/")
+        elif reference.startswith("#/definitions/"):
+            pointer = reference.removeprefix("#/")
+        else:
+            raise ValueError(
+                f"unsupported reference {reference!r} at schema path {label!r}"
+            )
+
+        resolved: Any = root_schema
+        for raw_part in pointer.split("/"):
+            part = raw_part.replace("~1", "/").replace("~0", "~")
+            if not isinstance(resolved, Mapping) or part not in resolved:
+                raise ValueError(
+                    f"unresolved reference {reference!r} at schema path {label!r}"
+                )
+            resolved = resolved[part]
+        if not isinstance(resolved, Mapping):
+            raise ValueError(
+                f"reference {reference!r} at schema path {label!r} "
+                "does not select a JSON Schema object"
+            )
+        current = resolved
+    return current
+
+
 def _ensure_object_schema(schema: JsonObject, label: str) -> None:
     schema_type = schema.get("type")
     if schema_type is not None and schema_type != "object":

tests/wf_api/test_schema_projection.py

@@ -5,6 +5,8 @@ import pytest
 from wf_api.schema_projection import (
     project_output_property_to_state_schema,
     project_property_to_schema_path,
+    project_schema_path_to_schema_path,
+    schema_path_exists,
 )
 
 
@@ -193,3 +195,149 @@ def test_project_schema_property_rejects_non_object_ancestor() -> None:
             source_field="after",
             target_parts=("session", "after"),
         )
+
+
+def test_project_schema_path_copies_inline_nested_source() -> None:
+    projected = project_schema_path_to_schema_path(
+        target_schema={"type": "object", "properties": {}},
+        source_schema={
+            "type": "object",
+            "properties": {
+                "report": {
+                    "type": "object",
+                    "properties": {"title": {"type": "string"}},
+                }
+            },
+        },
+        source_parts=("report", "title"),
+        target_parts=("document", "title"),
+    )
+
+    assert projected["properties"]["document"]["properties"]["title"] == {
+        "type": "string"
+    }
+
+
+def test_project_schema_path_traverses_pydantic_defs_reference() -> None:
+    projected = project_schema_path_to_schema_path(
+        target_schema={"type": "object", "properties": {}},
+        source_schema={
+            "type": "object",
+            "properties": {"report": {"$ref": "#/$defs/Report"}},
+            "$defs": {
+                "Report": {
+                    "type": "object",
+                    "properties": {"markdown": {"$ref": "#/$defs/Markdown"}},
+                },
+                "Markdown": {"type": "string", "minLength": 1},
+            },
+        },
+        source_parts=("report", "markdown"),
+        target_parts=("report", "markdown"),
+    )
+
+    assert projected["properties"]["report"]["properties"]["markdown"] == {
+        "$ref": "#/$defs/Markdown"
+    }
+    assert projected["$defs"]["Markdown"]["minLength"] == 1
+
+
+def test_project_schema_path_traverses_legacy_definitions_reference() -> None:
+    projected = project_schema_path_to_schema_path(
+        target_schema={"type": "object", "properties": {}},
+        source_schema={
+            "type": "object",
+            "properties": {"report": {"$ref": "#/definitions/Report"}},
+            "definitions": {
+                "Report": {
+                    "type": "object",
+                    "properties": {"title": {"type": "string"}},
+                }
+            },
+        },
+        source_parts=("report", "title"),
+        target_parts=("title",),
+    )
+
+    assert projected["properties"]["title"] == {"type": "string"}
+    assert "Report" in projected["definitions"]
+
+
+def test_schema_path_exists_follows_local_defs() -> None:
+    schema = {
+        "type": "object",
+        "properties": {"report": {"$ref": "#/$defs/Report"}},
+        "$defs": {
+            "Report": {
+                "type": "object",
+                "properties": {"title": {"type": "string"}},
+            }
+        },
+    }
+
+    assert schema_path_exists(schema, ("report", "title")) is True
+    assert schema_path_exists(schema, ("report", "missing")) is False
+
+
+def test_project_schema_path_rejects_missing_nested_source() -> None:
+    with pytest.raises(
+        ValueError,
+        match="source schema path 'report.missing' is not declared",
+    ):
+        project_schema_path_to_schema_path(
+            target_schema={"type": "object", "properties": {}},
+            source_schema={
+                "type": "object",
+                "properties": {"report": {"type": "object", "properties": {}}},
+            },
+            source_parts=("report", "missing"),
+            target_parts=("value",),
+        )
+
+
+def test_project_schema_path_rejects_scalar_source_ancestor() -> None:
+    with pytest.raises(
+        ValueError,
+        match="source schema path 'report' is not an object",
+    ):
+        project_schema_path_to_schema_path(
+            target_schema={"type": "object", "properties": {}},
+            source_schema={
+                "type": "object",
+                "properties": {"report": {"type": "string"}},
+            },
+            source_parts=("report", "title"),
+            target_parts=("title",),
+        )
+
+
+def test_project_schema_path_rejects_remote_reference() -> None:
+    with pytest.raises(
+        ValueError,
+        match="unsupported reference 'https://example.com/report.json'",
+    ):
+        project_schema_path_to_schema_path(
+            target_schema={"type": "object", "properties": {}},
+            source_schema={
+                "type": "object",
+                "properties": {"report": {"$ref": "https://example.com/report.json"}},
+            },
+            source_parts=("report", "title"),
+            target_parts=("title",),
+        )
+
+
+def test_project_schema_path_rejects_remote_reference_at_selected_leaf() -> None:
+    with pytest.raises(
+        ValueError,
+        match="unsupported reference 'https://example.com/report.json'",
+    ):
+        project_schema_path_to_schema_path(
+            target_schema={"type": "object", "properties": {}},
+            source_schema={
+                "type": "object",
+                "properties": {"report": {"$ref": "https://example.com/report.json"}},
+            },
+            source_parts=("report",),
+            target_parts=("report",),
+        )