lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7477a12b5183748e49c3cd3a3cb16a1fb6bee91e
lda-wf/docs/superpowers/plans/2026-05-20-core-path-bindings.md
T

945 lines
31 KiB
Markdown
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:
```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.
Reference in New Issue View Git Blame Copy Permalink