plan

docs/superpowers/plans/2026-05-20-core-path-bindings.md

@@ -0,0 +1,925 @@
+# Core Path Bindings Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace loose core path/map strings with typed path objects and canonical list-of-struct node bindings while keeping deprecated shapes parse-compatible.
+
+**Architecture:** Add immutable path value objects in `wf_core.paths`, then introduce canonical binding models in `wf_core.models.steps`. Runtime and validation move to the canonical bindings, while old `in_map`, `input_values`, `out_map`, and dict-shaped state fields are accepted only by model validators.
+
+**Tech Stack:** Python 3.14, Pydantic v2, pytest, jsonschema, basedpyright, ruff.
+
+---
+
+## File Structure
+
+- Modify `src/wf_core/paths.py`: own typed graph/state/local path objects and graph path resolution helpers.
+- Modify `src/wf_core/local_paths.py`: keep compatibility wrappers over `LocalPath` plus local get/set helpers.
+- Modify `src/wf_core/models/steps.py`: add `InputPathBinding`, `InputValueBinding`, `OutputBinding`, and canonical `NodeUse.input` / `NodeUse.output`.
+- Modify `src/wf_core/models/conditions.py`: type condition path operands with `GraphSourcePath`.
+- Modify `src/wf_core/models/schemas.py`: harden `SchemaRef` and add canonical state field declarations.
+- Modify `src/wf_core/runtime/ops/nodes.py`: resolve canonical node input bindings.
+- Modify `src/wf_core/runtime/ops/state.py`: apply canonical output bindings through an atomic state patch.
+- Modify `src/wf_core/runtime/ops/schemas.py`: expose focused JSON Schema validation helpers.
+- Modify `src/wf_core/validation/steps.py`: validate canonical bindings and typed paths.
+- Modify `src/wf_authoring/dsl/paths.py`: emit core path objects while preserving ergonomic helpers.
+- Modify `src/wf_authoring/dsl/conditions.py`: compile authoring expressions to core typed condition models.
+- Add `tests/core/test_path_values.py`: path parsing, serialization, JSON Schema, and error tests.
+- Add `tests/core/test_canonical_node_bindings.py`: canonical model parsing and deprecated compatibility tests.
+- Add `tests/core/test_atomic_state_patches.py`: output binding, reducer, overlap, and atomicity tests.
+- Update existing `tests/core/test_mapping_validation.py`, `tests/core/test_nested_mappings.py`, `tests/core/test_nested_state_paths.py`, and authoring tests as needed.
+
+## Task 1: Add Typed Path Values
+
+**Files:**
+- Modify: `src/wf_core/paths.py`
+- Modify: `src/wf_core/local_paths.py`
+- Create: `tests/core/test_path_values.py`
+
+- [ ] **Step 1: Write path value tests**
+
+Add tests for parsing, string serialization, equality/hashability, invalid segments, root-only graph source reads, and no bare write state:
+
+```python
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from wf_core.paths import GraphSourcePath, LocalPath, PathResolutionError, StatePath
+
+
+def test_graph_source_path_accepts_root_and_nested_paths():
+    assert str(GraphSourcePath.parse("state")) == "state"
+    assert str(GraphSourcePath.parse("input.user")) == "input.user"
+    assert str(GraphSourcePath.context("loop_item")) == "context.loop_item"
+
+
+def test_state_path_rejects_bare_state_write_target():
+    with pytest.raises(PathResolutionError, match="state path"):
+        StatePath.parse("state")
+
+
+def test_local_path_supports_root_marker():
+    assert str(LocalPath.root()) == "."
+    assert str(LocalPath.of("user.name")) == "user.name"
+
+
+@pytest.mark.parametrize("raw", ["", "state.", "state.items.0", "state.user-name"])
+def test_paths_reject_invalid_segments(raw: str):
+    with pytest.raises(PathResolutionError):
+        GraphSourcePath.parse(raw)
+
+
+def test_path_objects_are_hashable():
+    paths = {StatePath.of("person.name"), StatePath.of("person.name")}
+    assert len(paths) == 1
+
+
+def test_pydantic_accepts_path_strings_and_serializes_strings():
+    class Payload(BaseModel):
+        source: GraphSourcePath
+        target: StatePath
+        local: LocalPath
+
+    payload = Payload.model_validate(
+        {"source": "input.user", "target": "state.person", "local": "user"}
+    )
+    assert payload.source == GraphSourcePath.input("user")
+    assert payload.model_dump(mode="json")["target"] == "state.person"
+
+
+def test_pydantic_rejects_bad_path_string():
+    class Payload(BaseModel):
+        source: GraphSourcePath
+
+    with pytest.raises(ValidationError):
+        Payload.model_validate({"source": "output.foo"})
+```
+
+- [ ] **Step 2: Run path tests to verify they fail**
+
+Run: `uv run --with pytest pytest tests/core/test_path_values.py -q`
+
+Expected: failures because `GraphSourcePath`, `StatePath`, and `LocalPath` classes do not exist or do not validate strictly.
+
+- [ ] **Step 3: Implement path value classes**
+
+In `src/wf_core/paths.py`, add frozen dataclasses and shared parsing helpers. Keep existing helper function names as compatibility wrappers where practical.
+
+Implementation shape:
+
+```python
+from dataclasses import dataclass
+import re
+from typing import Any, ClassVar, Literal
+
+SEGMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+
+@dataclass(frozen=True)
+class LocalPath:
+    """Node-local payload path. `.` means the whole local payload."""
+
+    parts: tuple[str, ...]
+
+    @classmethod
+    def root(cls) -> "LocalPath":
+        return cls(())
+
+    @classmethod
+    def of(cls, *fragments: str) -> "LocalPath":
+        return cls(_parse_fragments(*fragments, allow_empty=False))
+
+    @classmethod
+    def parse(cls, raw: str) -> "LocalPath":
+        if raw == ".":
+            return cls.root()
+        return cls.of(raw)
+
+    def __str__(self) -> str:
+        return "." if not self.parts else ".".join(self.parts)
+```
+
+Also add:
+
+```python
+GraphRoot = Literal["input", "state", "context"]
+
+
+@dataclass(frozen=True)
+class GraphSourcePath:
+    """Readable workflow graph path rooted at input, state, or context."""
+
+    root: GraphRoot
+    parts: tuple[str, ...] = ()
+
+    @classmethod
+    def parse(cls, raw: str) -> "GraphSourcePath": ...
+
+    @classmethod
+    def input(cls, *fragments: str) -> "GraphSourcePath": ...
+
+    @classmethod
+    def state(cls, *fragments: str) -> "GraphSourcePath": ...
+
+    @classmethod
+    def context(cls, *fragments: str) -> "GraphSourcePath": ...
+```
+
+And:
+
+```python
+@dataclass(frozen=True)
+class StatePath:
+    """Writable workflow state path. Bare `state` is intentionally invalid."""
+
+    parts: tuple[str, ...]
+
+    @classmethod
+    def parse(cls, raw: str) -> "StatePath":
+        parsed = GraphSourcePath.parse(raw)
+        if parsed.root != "state" or not parsed.parts:
+            raise PathResolutionError("expected state path such as state.foo")
+        return cls(parsed.parts)
+
+    @classmethod
+    def of(cls, *fragments: str) -> "StatePath": ...
+```
+
+Add Pydantic `__get_pydantic_core_schema__` and `__get_pydantic_json_schema__` hooks for each class so strings validate into objects and serialize back to strings.
+
+- [ ] **Step 4: Update local path wrappers**
+
+In `src/wf_core/local_paths.py`, keep public functions but delegate parsing to `LocalPath.parse`:
+
+```python
+def split_local_path(path: str | LocalPath) -> list[str]:
+    """Split one node-local path, accepting the new typed path object."""
+    parsed = path if isinstance(path, LocalPath) else LocalPath.parse(path)
+    return list(parsed.parts)
+```
+
+Update `paths_overlap` and `has_overlapping_paths` to accept `str | LocalPath`.
+
+- [ ] **Step 5: Run path tests**
+
+Run: `uv run --with pytest pytest tests/core/test_path_values.py -q`
+
+Expected: all tests in `test_path_values.py` pass.
+
+## Task 2: Add Canonical Node Binding Models
+
+**Files:**
+- Modify: `src/wf_core/models/steps.py`
+- Test: `tests/core/test_canonical_node_bindings.py`
+
+- [ ] **Step 1: Write canonical binding tests**
+
+Create `tests/core/test_canonical_node_bindings.py`:
+
+```python
+import pytest
+from pydantic import ValidationError
+
+from wf_core.models.steps import NodeUse
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+
+
+def test_node_use_accepts_canonical_input_and_output_bindings():
+    node = NodeUse.model_validate(
+        {
+            "id": "echo",
+            "type": "node",
+            "node": "echo",
+            "input": [
+                {"target": "message", "path": "input.message"},
+                {"target": "mode", "value": None},
+            ],
+            "output": [{"source": "echoed", "target": "state.echoed"}],
+        }
+    )
+
+    assert node.input[0].target == LocalPath.of("message")
+    assert node.input[0].path == GraphSourcePath.input("message")
+    assert node.input[1].value is None
+    assert node.output[0].target == StatePath.of("echoed")
+
+
+def test_node_use_converts_old_maps_to_canonical_bindings():
+    node = NodeUse.model_validate(
+        {
+            "id": "echo",
+            "type": "node",
+            "node": "echo",
+            "in_map": {"input.message": "message"},
+            "input_values": {"mode": "fast"},
+            "out_map": {"echoed": "state.echoed"},
+        }
+    )
+
+    dumped = node.model_dump(mode="json")
+    assert "in_map" not in dumped
+    assert "input_values" not in dumped
+    assert "out_map" not in dumped
+    assert dumped["input"][0]["path"] == "input.message"
+    assert dumped["input"][1]["value"] == "fast"
+    assert dumped["output"][0]["target"] == "state.echoed"
+
+
+def test_node_use_rejects_mixed_old_and_new_binding_styles():
+    with pytest.raises(ValidationError):
+        NodeUse.model_validate(
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "echo",
+                "input": [{"target": "message", "path": "input.message"}],
+                "in_map": {"input.other": "other"},
+            }
+        )
+
+
+def test_input_binding_rejects_path_and_value_together():
+    with pytest.raises(ValidationError):
+        NodeUse.model_validate(
+            {
+                "id": "bad",
+                "type": "node",
+                "node": "bad",
+                "input": [
+                    {"target": "message", "path": "input.message", "value": "x"}
+                ],
+            }
+        )
+```
+
+- [ ] **Step 2: Run binding tests to verify they fail**
+
+Run: `uv run --with pytest pytest tests/core/test_canonical_node_bindings.py -q`
+
+Expected: failures because canonical binding fields do not exist yet.
+
+- [ ] **Step 3: Implement binding models**
+
+In `src/wf_core/models/steps.py`, add:
+
+```python
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+
+
+class InputPathBinding(BaseModel):
+    """Map one graph source path into one node-local input path."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    target: LocalPath
+    path: GraphSourcePath
+
+
+class InputValueBinding(BaseModel):
+    """Map one static JSON-compatible value into one node-local input path."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    target: LocalPath
+    value: object
+
+
+InputBinding = Annotated[
+    InputPathBinding | InputValueBinding,
+    Field(union_mode="left_to_right"),
+]
+
+
+class OutputBinding(BaseModel):
+    """Map one node-local output path into one workflow state path."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    source: LocalPath
+    target: StatePath
+```
+
+Update `NodeUse`:
+
+```python
+class NodeUse(BaseModel):
+    ...
+    input: list[InputBinding] = Field(default_factory=list)
+    output: list[OutputBinding] = Field(default_factory=list)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_deprecated_maps(cls, data: object) -> object:
+        ...
+```
+
+The validator should:
+
+- If `input` or `output` is present, reject any of `in_map`, `input_values`, `out_map`.
+- Convert `input_values` entries to `{"target": key, "value": value}` preserving order.
+- Convert `in_map` entries to `{"target": destination, "path": source}` preserving order.
+- Convert `out_map` entries to `{"source": source, "target": destination}` preserving order.
+- Remove old keys from the normalized data.
+
+- [ ] **Step 4: Run binding tests**
+
+Run: `uv run --with pytest pytest tests/core/test_canonical_node_bindings.py -q`
+
+Expected: all tests in `test_canonical_node_bindings.py` pass.
+
+## Task 3: Move Runtime Node Input Resolution To Canonical Bindings
+
+**Files:**
+- Modify: `src/wf_core/runtime/ops/nodes.py`
+- Test: `tests/core/test_nested_mappings.py`
+- Test: `tests/core/test_canonical_node_bindings.py`
+
+- [ ] **Step 1: Add runtime tests for canonical input binding behavior**
+
+In `tests/core/test_nested_mappings.py`, add a test that builds the existing minimal workflow style but uses `input` / `output` instead of old maps:
+
+```python
+def test_canonical_bindings_resolve_input_values_and_paths():
+    workflow = Workflow.model_validate(
+        {
+            "name": "canonical",
+            "input_schema": {"type": "object", "properties": {"message": {"type": "string"}}},
+            "state_schema": {"fields": {"echoed": {"type": "string"}}},
+            "output_schema": {"type": "object", "properties": {"echoed": {"type": "string"}}},
+            "start": "echo",
+            "node_defs": [
+                {
+                    "name": "echo",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"message": {"type": "string"}, "mode": {"type": "string"}},
+                        "required": ["message", "mode"],
+                    },
+                    "output_schema": {"type": "object", "properties": {"echoed": {"type": "string"}}},
+                    "outcomes": ["ok"],
+                }
+            ],
+            "nodes": [
+                {
+                    "id": "echo",
+                    "type": "node",
+                    "node": "echo",
+                    "input": [
+                        {"target": "message", "path": "input.message"},
+                        {"target": "mode", "value": "fast"},
+                    ],
+                    "output": [{"source": "echoed", "target": "state.echoed"}],
+                }
+            ],
+            "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
+        }
+    )
+
+    result = execute_workflow(
+        workflow,
+        {"message": "hi"},
+        registry={"echo": lambda payload, _ctx: {"echoed": f"{payload['mode']}:{payload['message']}"}},
+    )
+
+    assert result.output["echoed"] == "fast:hi"
+```
+
+- [ ] **Step 2: Run the focused test to verify failure**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_mappings.py::test_canonical_bindings_resolve_input_values_and_paths -q`
+
+Expected: failure because runtime still reads `node.input_values`, `node.in_map`, and `node.out_map`.
+
+- [ ] **Step 3: Update `_resolve_node_execution`**
+
+In `src/wf_core/runtime/ops/nodes.py`, import binding classes and use `node.input`.
+
+Implementation shape:
+
+```python
+from wf_core.models.steps import InputPathBinding, InputValueBinding
+
+
+for binding in node.input:
+    if isinstance(binding, InputValueBinding):
+        value = binding.value
+    else:
+        value = safe_resolve_path(
+            str(binding.path),
+            state=run.state,
+            workflow_input=run.workflow_input,
+            context=context_values,
+        )
+    set_local_value(resolved_input, binding.target, value)
+```
+
+`set_local_value` should accept `LocalPath` after Task 1.
+
+- [ ] **Step 4: Run canonical runtime test**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_mappings.py::test_canonical_bindings_resolve_input_values_and_paths -q`
+
+Expected: pass.
+
+## Task 4: Move Runtime Output Writes To Canonical Bindings And Atomic Patches
+
+**Files:**
+- Modify: `src/wf_core/runtime/ops/state.py`
+- Modify: `src/wf_core/runtime/ops/nodes.py`
+- Test: `tests/core/test_atomic_state_patches.py`
+
+- [ ] **Step 1: Write atomic patch tests**
+
+Create `tests/core/test_atomic_state_patches.py`:
+
+```python
+import pytest
+
+from wf_core.errors import WorkflowExecutionError
+from wf_core.models.workflow import Workflow
+from wf_core.runtime.ops.state import apply_output_bindings
+
+
+def _workflow() -> Workflow:
+    return Workflow.model_validate(
+        {
+            "name": "patch",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {
+                "fields": {
+                    "person": {"type": "object"},
+                    "person.name": {"type": "string"},
+                }
+            },
+            "output_schema": {"type": "object", "properties": {}},
+            "start": "n",
+            "nodes": [],
+            "edges": [],
+        }
+    )
+
+
+def test_output_bindings_commit_patch_atomically():
+    workflow = _workflow()
+    state = {"person": {"name": "old"}}
+
+    with pytest.raises(WorkflowExecutionError):
+        apply_output_bindings(
+            workflow,
+            [
+                {"source": "person.name", "target": "state.person.name"},
+                {"source": "missing", "target": "state.person.extra"},
+            ],
+            {"person": {"name": "new"}},
+            state,
+        )
+
+    assert state["person"]["name"] == "old"
+
+
+def test_output_bindings_reject_overlapping_write_targets():
+    workflow = _workflow()
+    state = {}
+
+    with pytest.raises(WorkflowExecutionError, match="overlapping"):
+        apply_output_bindings(
+            workflow,
+            [
+                {"source": "person", "target": "state.person"},
+                {"source": "person.name", "target": "state.person.name"},
+            ],
+            {"person": {"name": "Ada"}},
+            state,
+        )
+```
+
+- [ ] **Step 2: Run atomic patch tests to verify failure**
+
+Run: `uv run --with pytest pytest tests/core/test_atomic_state_patches.py -q`
+
+Expected: failure because `apply_output_bindings` does not exist.
+
+- [ ] **Step 3: Implement `apply_output_bindings`**
+
+In `src/wf_core/runtime/ops/state.py`, add a canonical function:
+
+```python
+from wf_core.models.steps import OutputBinding
+from wf_core.paths import StatePath
+
+
+def apply_output_bindings(
+    workflow: Workflow,
+    bindings: Sequence[OutputBinding],
+    node_output: dict[str, Any],
+    state: dict[str, Any],
+    reducers: Mapping[str, ReducerDefinition] | None = None,
+) -> dict[str, Any]:
+    """Prepare and commit one atomic state patch from canonical output bindings."""
+```
+
+Function behavior:
+
+- Validate no overlapping `binding.target`.
+- Resolve every `binding.source` from `node_output` first.
+- Build a prepared patch keyed by `StatePath`.
+- Compute reducers into prepared merged values without mutating `state`.
+- Commit all prepared values only after all prior steps succeed.
+- Return JSON-friendly `dict[str, Any]` state changes using `str(path)` keys for now, until trace is separately migrated.
+
+Keep `apply_output_map` as a compatibility wrapper that converts old map entries into `OutputBinding` and calls `apply_output_bindings`.
+
+- [ ] **Step 4: Update node finalization**
+
+In `src/wf_core/runtime/ops/nodes.py`, call `apply_output_bindings(workflow, node.output, result.output, run.state, reducers=reducers)` instead of `apply_output_map(...)`.
+
+- [ ] **Step 5: Run state patch tests**
+
+Run: `uv run --with pytest pytest tests/core/test_atomic_state_patches.py tests/core/test_nested_mappings.py -q`
+
+Expected: pass.
+
+## Task 5: Update Validation For Canonical Bindings
+
+**Files:**
+- Modify: `src/wf_core/validation/steps.py`
+- Test: `tests/core/test_mapping_validation.py`
+- Test: `tests/core/test_canonical_node_bindings.py`
+
+- [ ] **Step 1: Add validation tests for canonical fields**
+
+In `tests/core/test_mapping_validation.py`, add tests for invalid source paths, invalid destination paths, overlapping local input targets, and overlapping state output targets using canonical `input` / `output`.
+
+Example:
+
+```python
+def test_validate_workflow_reports_overlapping_canonical_output_targets():
+    workflow = workflow_with_node(
+        node_use={
+            "id": "n",
+            "type": "node",
+            "node": "n",
+            "output": [
+                {"source": "person", "target": "state.person"},
+                {"source": "person.name", "target": "state.person.name"},
+            ],
+        }
+    )
+
+    report = workflow.validate_structure()
+
+    assert any(issue.code == ValidationIssueCode.INVALID_DESTINATION_PATH for issue in report.issues)
+```
+
+Use the existing helper style in `tests/core/test_mapping_validation.py` rather than inventing a second full workflow factory if one already exists.
+
+- [ ] **Step 2: Run mapping validation tests**
+
+Run: `uv run --with pytest pytest tests/core/test_mapping_validation.py -q`
+
+Expected: new canonical validation tests fail until validation reads `node.input` / `node.output`.
+
+- [ ] **Step 3: Update `validate_node_use`**
+
+In `src/wf_core/validation/steps.py`:
+
+- Iterate `node.input`.
+- For `InputValueBinding`, validate target local root against node input schema.
+- For `InputPathBinding`, validate target and source graph path.
+- Iterate `node.output`.
+- Validate output source local root against node output schema.
+- Validate destination `StatePath`.
+- Use typed overlap helpers instead of raw map values.
+- Keep issue paths readable, e.g. `nodes[0].input[1].target`.
+
+- [ ] **Step 4: Run validation tests**
+
+Run: `uv run --with pytest pytest tests/core/test_mapping_validation.py tests/core/test_canonical_node_bindings.py -q`
+
+Expected: pass.
+
+## Task 6: Add Canonical State Schema Fields
+
+**Files:**
+- Modify: `src/wf_core/models/schemas.py`
+- Modify: `src/wf_core/runtime/ops/state.py`
+- Modify: `src/wf_core/validation/steps.py`
+- Test: `tests/core/test_nested_state_paths.py`
+- Test: `tests/core/test_schema_validation.py`
+
+- [ ] **Step 1: Write state schema canonical shape tests**
+
+In `tests/core/test_nested_state_paths.py`, add:
+
+```python
+from wf_core.models.schemas import StateSchema
+from wf_core.paths import StatePath
+
+
+def test_state_schema_accepts_canonical_field_list():
+    schema = StateSchema.model_validate(
+        {
+            "fields": [
+                {"path": "state.person", "type": "object"},
+                {"path": "state.person.name", "type": "string", "reducer": "wf.std.replace"},
+            ]
+        }
+    )
+
+    assert schema.fields[0].path == StatePath.of("person")
+    assert schema.field_map()["person.name"].type == "string"
+
+
+def test_state_schema_accepts_deprecated_dict_shape():
+    schema = StateSchema.model_validate(
+        {"fields": {"person.name": {"type": "string"}}}
+    )
+
+    assert schema.model_dump(mode="json")["fields"][0]["path"] == "state.person.name"
+```
+
+- [ ] **Step 2: Run state schema tests to verify failure**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_state_paths.py -q`
+
+Expected: failure because `StateSchema.fields` is still a dict.
+
+- [ ] **Step 3: Implement canonical `StateFieldDecl`**
+
+In `src/wf_core/models/schemas.py`:
+
+```python
+class StateFieldDecl(BaseModel):
+    """One declared state path plus validation and reducer metadata."""
+
+    path: StatePath
+    schema: SchemaRef = Field(default_factory=lambda: SchemaRef(type="object"))
+    reducer: ReducerRef = Field(default_factory=lambda: ReducerRef(name="wf.std.replace"))
+    trace: bool = True
+    default: Any = None
+```
+
+Preserve compatibility for old `type` directly on the field:
+
+- For old dict values like `{"type": "string"}`, convert to `{"schema": {"type": "string"}}`.
+- For canonical values, allow either `schema` or simple `type` as input if that keeps existing tests stable.
+
+Update `StateSchema`:
+
+```python
+class StateSchema(BaseModel):
+    fields: list[StateFieldDecl] = Field(default_factory=list)
+
+    def field_map(self) -> dict[str, StateFieldDecl]:
+        return {".".join(field.path.parts): field for field in self.fields}
+```
+
+Add a model validator to accept old dict shape and normalize to list.
+
+- [ ] **Step 4: Update callers of `workflow.state_schema.fields`**
+
+Search: `rg 'state_schema\\.fields|\\.fields\\.get|set\\(workflow\\.state_schema\\.fields\\)' src tests`
+
+Update code to use `workflow.state_schema.field_map()` when it needs lookup by rootless path.
+
+Important updates:
+
+- `src/wf_core/runtime/ops/state.py`
+- `src/wf_core/validation/steps.py`
+- any authoring or artifact code constructing state field maps.
+
+- [ ] **Step 5: Run state schema tests**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_state_paths.py tests/core/test_schema_validation.py -q`
+
+Expected: pass.
+
+## Task 7: Harden SchemaRef With JSON Schema Validation
+
+**Files:**
+- Modify: `src/wf_core/models/schemas.py`
+- Modify: `src/wf_core/runtime/ops/schemas.py`
+- Test: `tests/core/test_schema_validation.py`
+
+- [ ] **Step 1: Add schema validation tests**
+
+In `tests/core/test_schema_validation.py`, add tests:
+
+```python
+import pytest
+from pydantic import ValidationError
+
+from wf_core.models.schemas import SchemaRef
+
+
+def test_schema_ref_accepts_valid_json_schema_with_defs():
+    schema = SchemaRef.model_validate(
+        {
+            "$schema": "https://json-schema.org/draft/2020-12/schema",
+            "type": "object",
+            "$defs": {"Name": {"type": "string"}},
+            "properties": {"name": {"$ref": "#/$defs/Name"}},
+        }
+    )
+
+    assert schema.model_extra["$defs"]["Name"]["type"] == "string"
+
+
+def test_schema_ref_rejects_invalid_json_schema():
+    with pytest.raises(ValidationError):
+        SchemaRef.model_validate({"type": 123})
+```
+
+- [ ] **Step 2: Run schema tests to verify failure**
+
+Run: `uv run --with pytest pytest tests/core/test_schema_validation.py -q`
+
+Expected: invalid schema is currently accepted.
+
+- [ ] **Step 3: Add `jsonschema` validation**
+
+In `src/wf_core/models/schemas.py`, import:
+
+```python
+from jsonschema import SchemaError
+from jsonschema.validators import Draft202012Validator, validator_for
+from pydantic import model_validator
+```
+
+Add an after validator to `SchemaRef`:
+
+```python
+@model_validator(mode="after")
+def _validate_json_schema(self) -> "SchemaRef":
+    raw = self.model_dump(mode="python", exclude_none=True)
+    validator_cls = validator_for(raw, default=Draft202012Validator)
+    try:
+        validator_cls.check_schema(raw)
+    except SchemaError as exc:
+        raise ValueError(f"invalid JSON Schema: {exc.message}") from exc
+    return self
+```
+
+- [ ] **Step 4: Run schema tests**
+
+Run: `uv run --with pytest pytest tests/core/test_schema_validation.py -q`
+
+Expected: pass.
+
+## Task 8: Update Authoring Helpers To Emit Canonical Bindings
+
+**Files:**
+- Modify: `src/wf_authoring/dsl/paths.py`
+- Modify: `src/wf_authoring/dsl/conditions.py`
+- Modify: `src/wf_authoring/builder/core.py`
+- Test: `tests/authoring/test_builder.py`
+- Test: `tests/authoring/test_conditions.py`
+- Test: `tests/authoring/test_control_flow_examples.py`
+
+- [ ] **Step 1: Add authoring tests for canonical dumps**
+
+In `tests/authoring/test_builder.py`, add a test that builds a workflow and asserts the dumped node uses canonical `input` / `output`, not old maps:
+
+```python
+def test_builder_emits_canonical_node_bindings():
+    workflow = (
+        WorkflowBuilder("canonical")
+        .schemas(
+            input_schema={"type": "object", "properties": {"message": {"type": "string"}}},
+            state_schema={"fields": {"echoed": {"type": "string"}}},
+            output_schema={"type": "object", "properties": {"echoed": {"type": "string"}}},
+        )
+        .use(echo_node, id="echo", in_map={"input.message": "message"}, out_map={"echoed": "state.echoed"})
+        .start_at("echo")
+        .end("echo", "ok")
+        .build()
+    )
+
+    dumped_node = workflow.model_dump(mode="json")["nodes"][0]
+    assert "input" in dumped_node
+    assert "output" in dumped_node
+    assert "in_map" not in dumped_node
+    assert "out_map" not in dumped_node
+```
+
+Adapt helper names to the current builder API in the file.
+
+- [ ] **Step 2: Run authoring builder tests**
+
+Run: `uv run --with pytest pytest tests/authoring/test_builder.py tests/authoring/test_conditions.py -q`
+
+Expected: new canonical dump test may fail until builder emits or model normalizes canonical shapes.
+
+- [ ] **Step 3: Update path/condition authoring wrappers**
+
+In `src/wf_authoring/dsl/paths.py`, make ergonomic helpers return wrappers around core path values or values accepted by core models. Preserve existing public behavior where possible:
+
+```python
+def state_path(*parts: str) -> GraphPath:
+    return GraphPath(str(GraphSourcePath.state(*parts)))
+```
+
+In `src/wf_authoring/dsl/conditions.py`, make `PathExpr` compile using `GraphSourcePath.parse` for `PathOperand`.
+
+- [ ] **Step 4: Update builder to rely on canonical model normalization**
+
+In `src/wf_authoring/builder/core.py`, either emit canonical binding dicts directly or keep passing old maps into `NodeUse.model_validate`. Prefer direct canonical emission where the builder already has enough structure.
+
+Do not remove user-facing `in_map` / `out_map` builder parameters in this pass.
+
+- [ ] **Step 5: Run authoring tests**
+
+Run: `uv run --with pytest pytest tests/authoring -q`
+
+Expected: authoring tests pass.
+
+## Task 9: Full Compatibility And Regression Pass
+
+**Files:**
+- Modify docs/examples only if tests show stale serialized shapes.
+- Test: full repo.
+
+- [ ] **Step 1: Run core tests**
+
+Run: `uv run --with pytest pytest tests/core tests/authoring tests/rewrite -q`
+
+Expected: pass.
+
+- [ ] **Step 2: Run artifact and MCP workflow-surface tests**
+
+Run: `uv run --with pytest pytest tests/artifacts tests/wf_mcp/test_workflow_surface.py tests/wf_mcp/test_workflow_wrappers.py tests/wf_mcp/test_mcp_workflow_surface_example.py -q`
+
+Expected: pass.
+
+- [ ] **Step 3: Run full test suite**
+
+Run: `uv run --with pytest pytest -q`
+
+Expected: pass, allowing any existing intentionally skipped environment-dependent tests.
+
+- [ ] **Step 4: Run static checks**
+
+Run:
+
+```bash
+uvx ruff check
+uv run basedpyright --level error
+```
+
+Expected: ruff passes and basedpyright reports 0 errors.
+
+- [ ] **Step 5: Format touched files**
+
+Run:
+
+```bash
+uvx ruff format src/wf_core src/wf_authoring tests/core tests/authoring
+```
+
+Expected: files format cleanly.
+
+## Self-Review Notes
+
+- Spec coverage: typed paths, canonical bindings, parse-only compatibility, null/missing semantics, dynamic traversal deferral, state patch atomicity, reducer behavior, JSON Schema validation, authoring updates, and tracing shape are covered. Full trace migration is intentionally not implemented beyond returning string-keyed `state_changes` for compatibility.
+- Placeholder scan: this plan avoids `TBD` and names concrete files, tests, commands, and behavior.
+- Type consistency: `LocalPath`, `GraphSourcePath`, `StatePath`, `InputPathBinding`, `InputValueBinding`, `OutputBinding`, and `StateFieldDecl` are introduced before later tasks use them.

docs/superpowers/specs/2026-05-20-core-path-bindings-design.md

@@ -0,0 +1,331 @@
+# Core Path Bindings Design
+
+## Purpose
+
+`wf_core` currently uses plain strings for graph paths, local node paths, input
+maps, output maps, and state field keys. That made the early system simple, but
+it also pushes too much meaning into ad hoc string parsing. This design makes
+paths and bindings first-class core concepts while keeping JSON serialization
+simple.
+
+The goal is not to make every dynamic JSON traversal statically provable. The
+goal is to make normal workflow data movement explicit, validated, serializable,
+and easy for authoring/MCP layers to generate.
+
+## Goals
+
+- Replace loose map fields with canonical list-of-struct binding models.
+- Store typed path value objects internally while serializing them as strings.
+- Keep missing values distinct from explicit `null`.
+- Make state writes atomic and reducer-aware.
+- Keep compatibility with old workflow shapes through parse-only deprecated
+  fields.
+- Keep dynamic/open-object traversal out of core and in explicit nodes.
+
+## Non-Goals
+
+- No list/index path syntax such as `items.0.name` or `items[0].name`.
+- No arbitrary JSON-pointer support.
+- No whole-state replacement writes in this pass.
+- No implicit defaults for missing paths.
+- No business logic in reducers or bindings.
+- No custom schema language replacing JSON Schema.
+
+## Path Types
+
+Core introduces distinct path value objects:
+
+```text
+LocalPath(parts)
+GraphSourcePath(root, parts)  # root: input | state | context
+StatePath(parts)              # serializes as state.<parts>
+```
+
+These objects are immutable/hashable and are accepted by Pydantic from either
+strings or existing instances. JSON serialization emits strings.
+
+Examples:
+
+```text
+LocalPath.of("user.name")                -> "user.name"
+LocalPath.root()                         -> "."
+GraphSourcePath.state("person.name")     -> "state.person.name"
+GraphSourcePath.parse("input")           -> "input"
+StatePath.of("person.name")              -> "state.person.name"
+```
+
+Path parsing rules:
+
+- Segments use `[A-Za-z_][A-Za-z0-9_]*`.
+- Dots separate segments.
+- Empty segments are invalid.
+- Numeric/positional list segments are rejected.
+- `LocalPath.root()` / `"."` is the only local root marker.
+- Root-only graph source paths `input`, `state`, and `context` are valid reads.
+
+`StatePath` write targets should not accept bare `state` in this pass. Whole
+state replacement is too broad because it interacts with reducers, validation,
+trace, and accidental deletion.
+
+There is no reducer for bare `state`. Reducers attach only to declared non-root
+`StatePath` fields.
+
+## Canonical Node Bindings
+
+`NodeUse` gets canonical binding fields:
+
+```text
+NodeUse.input: list[InputBinding]
+NodeUse.output: list[OutputBinding]
+```
+
+Input bindings are distinguished by shape, not by an extra `kind` field:
+
+```text
+InputPathBinding:
+  target: LocalPath
+  path: GraphSourcePath
+
+InputValueBinding:
+  target: LocalPath
+  value: JsonValue
+```
+
+Output bindings are:
+
+```text
+OutputBinding:
+  source: LocalPath
+  target: StatePath
+```
+
+The same field name can mean different path kinds by position. For example,
+input binding `target` is node-local, while output binding `target` is workflow
+state. Documentation and model field descriptions should make this explicit.
+
+Root local path `"."` is valid:
+
+- input target `"."` means the whole node input payload.
+- output source `"."` means the whole node output payload.
+- a `"."` input binding must be the only input binding.
+
+Examples:
+
+```json
+{
+  "input": [
+    {"target": "user.email", "path": "state.person.email"},
+    {"target": "mode", "value": "fast"}
+  ],
+  "output": [
+    {"source": "result", "target": "state.result"}
+  ]
+}
+```
+
+Whole payload input:
+
+```json
+{
+  "input": [
+    {"target": ".", "path": "state.rates"}
+  ]
+}
+```
+
+Whole payload literal input:
+
+```json
+{
+  "input": [
+    {"target": ".", "value": {"mode": "fast"}}
+  ]
+}
+```
+
+## Compatibility
+
+Old fields are accepted only as deprecated parse inputs:
+
+```text
+in_map
+input_values
+out_map
+```
+
+After validation, `NodeUse` stores only canonical `input` and `output`
+bindings. Canonical serialization emits only the new fields. Payloads that mix
+canonical fields with deprecated fields should fail instead of merging two
+styles.
+
+This shape makes it easy to remove compatibility later: delete the parser
+adapters without changing runtime internals.
+
+State schema gets the same compatibility shape. The canonical form is:
+
+```text
+StateSchema.fields: list[StateFieldDecl]
+
+StateFieldDecl:
+  path: StatePath
+  schema: SchemaRef
+  reducer: ReducerRef
+```
+
+Old dict-shaped fields can be accepted at parse time and normalized:
+
+```json
+{
+  "fields": {
+    "person.tags": {"type": "array", "reducer": "wf.std.append"}
+  }
+}
+```
+
+Canonical serialization should emit list-of-structs with serialized state paths.
+
+## Runtime Semantics
+
+Node execution flow:
+
+1. Validate workflow input against `workflow.input_schema`.
+2. Resolve node input bindings into the node payload.
+3. Validate node payload against `node_def.input_schema`.
+4. Execute the node.
+5. Coerce the node result.
+6. Validate node output against `node_def.output_schema`.
+7. Prepare an atomic state patch from output bindings.
+8. Validate focused state patch values.
+9. Commit the patch.
+10. At `END`, project final output from top-level state fields using
+    `workflow.output_schema.properties`.
+
+State writes are atomic. Runtime should resolve all output sources, detect
+overlap conflicts, compute reducer results, validate patch values, then commit.
+If any step fails, prior state is unchanged.
+
+Reducers run during patch preparation. They receive current value, incoming
+value, and optional config, then return the merged value. Reducers must not
+mutate workflow state directly.
+
+## Missing, Null, And Dynamic Data
+
+Explicit `null` is a real value. It is not the same as a missing path.
+
+Rules:
+
+- `exists(path)` returns true when the path resolves, even if the value is null.
+- `exists(path)` returns false when the path is missing.
+- Comparisons such as `eq(null)` and `ne(null)` are valid.
+- Comparisons against missing paths fail clearly.
+- Input path bindings fail before node execution when the source path is
+  missing.
+- Missing source paths must not silently bind null.
+- Input value bindings may intentionally bind null.
+
+`allow extra` / open object schemas do not authorize speculative deep traversal.
+If a workflow needs dynamic object traversal, it should use an explicit node
+that receives the relevant value and decides what to extract. For example, use
+an `extract_title` node instead of trying to make core prove
+`state.person.occupations.1.title`.
+
+## Validation Rules
+
+Path value objects validate syntax and path kind only. Workflow validation checks
+whether a path is legal in a particular workflow.
+
+Examples of workflow-specific checks:
+
+- `input.foo` exists in `workflow.input_schema` when statically knowable.
+- `state.foo` has a declared state root.
+- write destinations are `StatePath`.
+- output write targets do not overlap.
+
+Overlap rules:
+
+- read paths may overlap.
+- input targets must not overlap.
+- output targets must not overlap.
+- exact duplicate targets are invalid.
+- parent/child write targets such as `state.person` and `state.person.name`
+  are invalid together.
+
+State write validation:
+
+- Validate against the exact declared state field schema when one exists.
+- Do not validate the whole workflow state on every write.
+- Replacement writes must satisfy the full target schema.
+- Replacement writes with forbidden extra fields fail.
+- Partial object patches require an explicit merge-style reducer such as
+  `merge_object`.
+- Do not silently treat `replace` as merge.
+
+## JSON Schema Boundary
+
+Core should not invent a schema language. `SchemaRef` should represent a JSON
+Schema object boundary and be validated with the standard `jsonschema` library.
+
+Rules:
+
+- Respect `$schema` when present with `jsonschema.validators.validator_for`.
+- Default to Draft 2020-12 when `$schema` is absent.
+- Validate workflow input/output schemas and node input/output schemas at model
+  boundaries.
+- Keep JSON Schema object keys separate from workflow path syntax.
+
+Path fields should expose clear JSON Schema as strings with pattern and
+description metadata, not `{root, parts}` objects.
+
+## Authoring Layer
+
+`wf_authoring` keeps ergonomic helpers:
+
+- `state_path(...)`
+- `input_path(...)`
+- `context_path(...)`
+- expression helpers such as `eq`, `ne`, `lt`, `le`, `gt`, `ge`, and `exists`
+
+Those helpers should compile to core path objects and core `Condition` models.
+Authoring may infer mappings for convenience, but it must emit explicit
+canonical bindings into core models.
+
+## Tracing
+
+Runtime may use flat `dict[StatePath, value]` patches internally. Public trace
+serialization should prefer list-of-structs:
+
+```text
+StateChange:
+  path: StatePath
+  value: JsonValue
+```
+
+This avoids custom JSON object-key serialization and gives MCP/LLM clients
+clearer output.
+
+## Implementation Phases
+
+1. Add core path value objects and parser helpers.
+2. Add canonical binding models with parse-only compatibility for old fields.
+3. Update validation to reason over path objects and binding structs.
+4. Update runtime input resolution and output patching to use canonical
+   bindings.
+5. Add focused state patch preparation with atomic commit semantics.
+6. Add JSON Schema validation hardening for `SchemaRef`.
+7. Update `wf_authoring` builders/helpers to emit canonical bindings.
+8. Update docs/examples and mark old fields as deprecated.
+
+Each phase should keep existing tests green, with compatibility tests proving
+old shapes still parse until support is intentionally removed.
+
+## Open Risks
+
+- Pydantic core-schema hooks for frozen path value objects may need careful
+  implementation to keep JSON Schema clean.
+- Existing examples and MCP-facing draft tools may rely on old dict-shaped maps.
+  Compatibility adapters should isolate that churn.
+- Focused schema validation for nested state writes depends on how much schema
+  information is available for exact declared state paths.
+- Workflow input currently seeds initial state. This design keeps that
+  compatibility for now, but explicit initialization remains the target
+  direction.