Files

T

lda 0d34174a84 docs: archive completed superpowers plans

2026-06-04 22:29:32 +07:00

31 KiB

Raw Blame History

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.

Current status as of 2026-05-20: The implementation has moved past the original checklist. Typed path values, canonical node bindings, canonical runtime input/output handling, atomic state patches, canonical validation, JSON-Schema-native state reducers, schema validation, and authoring canonical emission are present in the tree. The remaining checklist item is the full compatibility/regression pass in Task 9. If future code changes touch this area, prefer adding focused tests to the existing tests/core/test_*path*, tests/core/test_*mapping*, and tests/authoring/test_builder.py coverage rather than reimplementing the earlier tasks.

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:

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:

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:

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:

@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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

from jsonschema import SchemaError
from jsonschema.validators import Draft202012Validator, validator_for
from pydantic import model_validator

Add an after validator to SchemaRef:

@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:

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:

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:

uvx ruff check
uv run basedpyright --level error

Expected: ruff passes and basedpyright reports 0 errors.

Step 5: Format touched files

Run:

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.

31 KiB Raw Blame History

Core Path Bindings Implementation Plan

File Structure

Task 1: Add Typed Path Values

Task 2: Add Canonical Node Binding Models

Task 3: Move Runtime Node Input Resolution To Canonical Bindings

Task 4: Move Runtime Output Writes To Canonical Bindings And Atomic Patches

Task 5: Update Validation For Canonical Bindings

Task 6: Add Canonical State Schema Fields

Task 7: Harden SchemaRef With JSON Schema Validation

Task 8: Update Authoring Helpers To Emit Canonical Bindings

Task 9: Full Compatibility And Regression Pass

Self-Review Notes

31 KiB

Raw Blame History