use new input/output field

docs/authoring_sketch.md

@@ -51,26 +51,26 @@ Responsibilities:
 
 Typical entry points:
 
-- `use(node_spec, id=..., in_map=..., out_map=...)`
+- `use(node_spec, id=..., input=[...], output=[...])`
 - `condition(...)`
 - `foreach(...)`
 - `interrupt(...)`
 - `connect(...)`
 - `compile()`
 
-`in_map` and `out_map` remain authoring convenience parameters. The compiled
+`input` and `output` use the same canonical binding-list shape as core
-core `NodeUse` shape is canonical `input` and `output` binding lists:
+`NodeUse`:
 
 ```json
 {
-  "input": [{"target": "text", "path": "input.text"}],
+  "input": [{"target": {"root": "local", "parts": ["text"]}, "path": {"root": "input", "parts": ["text"]}}],
-  "output": [{"source": "echoed", "target": "state.echoed"}]
+  "output": [{"source": {"root": "local", "parts": ["echoed"]}, "target": {"root": "state", "parts": ["echoed"]}}]
 }
 ```
 
-Authoring should not teach client LLMs that `in_map`, `input_values`, or
+`in_map`, `input_values`, and `out_map` are deprecated compatibility sugar for
-`out_map` are the preferred core model. Those names are compatibility inputs
+Python authors. Client LLMs and MCP/JSON callers should use canonical binding
-and Python-builder sugar.
+lists so structural paths live inside structs, not as unhashable map keys.
 
 ### `NodeCatalog`
 

docs/core_state_mapping_and_merge.md

@@ -75,9 +75,9 @@ Whole-payload mapping uses the local root path `"."`:
 
 Deprecated compatibility inputs are still accepted at model-parse boundaries:
 `in_map`, `input_values`, and `out_map`. Validated `NodeUse` models store and
-dump only canonical `input` and `output` bindings. `wf_authoring` may still
+dump only canonical `input` and `output` bindings. `wf_authoring` exposes the
-accept `in_map` and `out_map` as builder convenience parameters, but it compiles
+same canonical binding lists and keeps `in_map`, `input_values`, and `out_map`
-them into canonical bindings.
+only as deprecated Python-builder sugar that compiles into canonical bindings.
 
 ## Explicitness Rules
 

docs/structural_refs.md

@@ -107,20 +107,34 @@ state("person.name", "email") # state -> "person.name" -> email
 state(("person.name",))       # state -> "person.name"
 ```
 
-Builder maps use the path kind implied by position:
+Builder canonical bindings use the path kind implied by position:
 
 ```python
 g.use(
     node,
-    in_map={input_path('"email.address"'): ("payload.email",)},
+    input=[
-    out_map={("result.score",): state_path("score")},
+        {
+            "target": {"root": "local", "parts": ["payload.email"]},
+            "path": {"root": "input", "parts": ["email.address"]},
+        }
+    ],
+    output=[
+        {
+            "source": {"root": "local", "parts": ["result.score"]},
+            "target": {"root": "state", "parts": ["score"]},
+        }
+    ],
 )
 ```
 
-In an input map, the key is a graph source path and the value is a node-local
+In an input binding, `path` is a graph source path and `target` is a
-input path. In an output map, the key is a node-local output path and the value
+node-local input path. In an output binding, `source` is a node-local output
-is a workflow state destination path. This lets authors keep concise helpers
+path and `target` is a workflow state destination path.
-without forcing saved workflow JSON back through dotted display strings.
+
+`in_map`, `input_values`, and `out_map` remain deprecated Python sugar for
+concise authoring. Structural path dicts are not valid map keys because Python
+dict keys must be hashable. Use canonical binding lists when working from
+JSON/MCP or when path segments contain display punctuation.
 
 ```json
 {

src/wf_authoring/builder/core.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any, Literal, cast
 import warnings
@@ -40,13 +40,17 @@ from ..schemas import SchemaLike, StateSchemaLike, schema_ref_from, state_schema
 from ..nodes import NodeSpec
 from .ids import next_step_id, slug_id
 from .mapping import (
+    InputBindingArg,
     MapArg,
+    OutputBindingArg,
     auto_input_map,
     auto_output_map,
     coerce_path,
+    normalize_input_bindings,
     normalize_input_mapping,
     normalize_input_values,
     normalize_mapping,
+    normalize_output_bindings,
     normalize_output_mapping,
 )
 from .refs import (
@@ -99,6 +103,49 @@ def _canonical_output_bindings(
     ]
 
 
+def _reject_mixed_binding_styles(
+    *,
+    input: object | None,
+    output: object | None,
+    in_map: object | None,
+    input_values: object | None,
+    out_map: object | None,
+) -> None:
+    """Keep canonical binding lists and deprecated map sugar from mixing."""
+    if input is not None and (in_map is not None or input_values is not None):
+        raise TypeError(
+            "cannot mix canonical input with deprecated in_map/input_values"
+        )
+    if output is not None and out_map is not None:
+        raise TypeError("cannot mix canonical output with deprecated out_map")
+
+
+def _warn_deprecated_binding_sugar(
+    *,
+    in_map: object | None,
+    input_values: object | None,
+    out_map: object | None,
+) -> None:
+    """Warn when callers explicitly use map sugar instead of canonical bindings."""
+    used = [
+        name
+        for name, value in (
+            ("in_map", in_map),
+            ("input_values", input_values),
+            ("out_map", out_map),
+        )
+        if value is not None
+    ]
+    if not used:
+        return
+    warnings.warn(
+        f"{', '.join(used)} are deprecated WorkflowBuilder sugar; use canonical "
+        "input/output binding lists instead",
+        DeprecationWarning,
+        stacklevel=3,
+    )
+
+
 @dataclass(slots=True)
 class WorkflowBuilder:
     name: str
@@ -122,14 +169,31 @@ class WorkflowBuilder:
         spec: NodeSpec[Any, Any],
         *,
         id: str | None = None,
+        input: Sequence[InputBindingArg] | None = None,
+        output: Sequence[OutputBindingArg] | None = None,
         in_map: MapArg | None = None,
         input_values: Mapping[Any, Any] | None = None,
         out_map: MapArg | None = None,
         desc: str | None = None,
     ) -> NodeUse:
+        _reject_mixed_binding_styles(
+            input=input,
+            output=output,
+            in_map=in_map,
+            input_values=input_values,
+            out_map=out_map,
+        )
+        _warn_deprecated_binding_sugar(
+            in_map=in_map,
+            input_values=input_values,
+            out_map=out_map,
+        )
         self.node_specs[spec.name] = spec
         normalized_input_schema = cast(SchemaRef, self.input_schema)
         normalized_state_schema = cast(StateSchema, self.state_schema)
+        if input is not None:
+            node_input = normalize_input_bindings(input)
+        else:
             raw_in_map = (
                 auto_input_map(
                     spec,
@@ -139,24 +203,28 @@ class WorkflowBuilder:
                 if in_map is None
                 else in_map
             )
-        normalized_in_map = normalize_input_mapping(raw_in_map)
+            node_input = _canonical_input_bindings(
-        normalized_input_values = normalize_input_values(input_values)
+                normalize_input_mapping(raw_in_map),
+                normalize_input_values(input_values),
+            )
+        if output is not None:
+            node_output = normalize_output_bindings(output)
+        else:
             raw_out_map = (
                 auto_output_map(spec, state_schema=normalized_state_schema)
                 if out_map is None
                 else out_map
             )
-        normalized_out_map = normalize_output_mapping(raw_out_map)
+            node_output = _canonical_output_bindings(
+                normalize_output_mapping(raw_out_map)
+            )
         node = NodeUse(
             id=id or self._next_step_id(slug_id(spec.name)),
             type="node",
             node=spec.name,
             desc=desc or spec.description,
-            input=_canonical_input_bindings(
+            input=node_input,
-                normalized_in_map,
+            output=node_output,
-                normalized_input_values,
-            ),
-            output=_canonical_output_bindings(normalized_out_map),
         )
         self.nodes.append(node)
         return node
@@ -166,6 +234,8 @@ class WorkflowBuilder:
         name: str,
         *,
         id: str | None = None,
+        input: Sequence[InputBindingArg] | None = None,
+        output: Sequence[OutputBindingArg] | None = None,
         in_map: MapArg | None = None,
         input_values: Mapping[Any, Any] | None = None,
         out_map: MapArg | None = None,
@@ -178,19 +248,38 @@ class WorkflowBuilder:
         hatch for MCP/saved-workflow capability refs that are resolved later by
         the environment runner into node definitions and registry handlers.
         """
-        normalized_in_map = normalize_input_mapping(in_map)
+        _reject_mixed_binding_styles(
-        normalized_input_values = normalize_input_values(input_values)
+            input=input,
-        normalized_out_map = normalize_output_mapping(out_map)
+            output=output,
+            in_map=in_map,
+            input_values=input_values,
+            out_map=out_map,
+        )
+        _warn_deprecated_binding_sugar(
+            in_map=in_map,
+            input_values=input_values,
+            out_map=out_map,
+        )
+        node_input = (
+            normalize_input_bindings(input)
+            if input is not None
+            else _canonical_input_bindings(
+                normalize_input_mapping(in_map),
+                normalize_input_values(input_values),
+            )
+        )
+        node_output = (
+            normalize_output_bindings(output)
+            if output is not None
+            else _canonical_output_bindings(normalize_output_mapping(out_map))
+        )
         node = NodeUse(
             id=id or self._next_step_id(slug_id(name)),
             type="node",
             node=name,
             desc=desc,
-            input=_canonical_input_bindings(
+            input=node_input,
-                normalized_in_map,
+            output=node_output,
-                normalized_input_values,
-            ),
-            output=_canonical_output_bindings(normalized_out_map),
         )
         self.nodes.append(node)
         return node
@@ -263,14 +352,16 @@ class WorkflowBuilder:
     ) -> ForeachNode:
         # Core foreach still stores `over` as a string. Keep this compatibility
         # path isolated until ForeachNode grows a typed GraphSourcePath field.
-        node = ForeachNode.model_validate({
+        node = ForeachNode.model_validate(
+            {
                 "id": id or self._next_step_id(f"foreach_{slug_id(as_)}"),
                 "type": "foreach",
                 "over": coerce_path(over),
                 "as": as_,
                 "mode": mode,
                 "on_item_error": on_item_error,
-        })
+            }
+        )
         self.nodes.append(node)
         return node
 
@@ -304,11 +395,13 @@ class WorkflowBuilder:
         source = self._resolve_branch_ref(from_)
         target = self._resolve_branch_ref(to)
         self.edges.append(
-            Edge.model_validate({
+            Edge.model_validate(
+                {
                     "from": step_id(source),
                     "outcome": outcome,
                     "to": step_id(target),
-            })
+                }
+            )
         )
         return source, target
 

src/wf_authoring/builder/mapping.py

@@ -1,9 +1,15 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from typing import Any, TypeAlias
 
 from wf_core import SchemaRef, StateSchema
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+)
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 from ..dsl import GraphPath
@@ -18,6 +24,8 @@ MapArg: TypeAlias = Mapping[Any, Any]
 InputMap: TypeAlias = dict[GraphSourcePath, LocalPath]
 OutputMap: TypeAlias = dict[LocalPath, StatePath]
 InputValues: TypeAlias = dict[LocalPath, Any]
+InputBindingArg: TypeAlias = InputBinding | Mapping[str, object]
+OutputBindingArg: TypeAlias = OutputBinding | Mapping[str, object]
 
 
 def coerce_path(value: object) -> str:
@@ -41,16 +49,26 @@ def normalize_mapping(mapping: MapArg | None) -> dict[str, str]:
     }
 
 
+def _reject_mapping_path_key(value: object, *, field_name: str) -> None:
+    """Reject structural dict keys; use canonical binding lists for JSON shapes."""
+    if isinstance(value, Mapping):
+        raise TypeError(
+            f"structural path dicts cannot be map keys in {field_name}; "
+            "use canonical input/output binding lists instead"
+        )
+
+
 def normalize_input_mapping(mapping: MapArg | None) -> InputMap:
     """Normalize `in_map`: graph source path -> node-local input path."""
     if mapping is None:
         return {}
-    return {
+    normalized: InputMap = {}
-        coerce_graph_path(source.path if isinstance(source, GraphPath) else source): (
+    for source, destination in mapping.items():
-            coerce_local_path(destination)
+        _reject_mapping_path_key(source, field_name="in_map")
-        )
+        normalized[
-        for source, destination in mapping.items()
+            coerce_graph_path(source.path if isinstance(source, GraphPath) else source)
-    }
+        ] = coerce_local_path(destination)
+    return normalized
 
 
 def normalize_input_values(mapping: Mapping[Any, Any] | None) -> InputValues:
@@ -60,17 +78,57 @@ def normalize_input_values(mapping: Mapping[Any, Any] | None) -> InputValues:
     return {coerce_local_path(target): value for target, value in mapping.items()}
 
 
+def normalize_input_bindings(
+    bindings: Sequence[InputBindingArg] | None,
+) -> list[InputBinding]:
+    """Validate canonical input binding structs for WorkflowBuilder.use()."""
+    if bindings is None:
+        return []
+    normalized: list[InputBinding] = []
+    for binding in bindings:
+        if isinstance(binding, InputPathBinding | InputValueBinding):
+            normalized.append(binding)
+            continue
+        if not isinstance(binding, Mapping):
+            raise TypeError(f"unsupported input binding {binding!r}")
+        if "path" in binding:
+            normalized.append(InputPathBinding.model_validate(binding))
+        elif "value" in binding:
+            normalized.append(InputValueBinding.model_validate(binding))
+        else:
+            raise TypeError("input binding must contain either 'path' or 'value'")
+    return normalized
+
+
 def normalize_output_mapping(mapping: MapArg | None) -> OutputMap:
     """Normalize `out_map`: node-local output path -> workflow state path."""
     if mapping is None:
         return {}
-    return {
+    normalized: OutputMap = {}
-        coerce_local_path(source): coerce_state_path(
+    for source, target in mapping.items():
+        _reject_mapping_path_key(source, field_name="out_map")
+        normalized[coerce_local_path(source)] = coerce_state_path(
             target.path if isinstance(target, GraphPath) else target,
             allow_legacy_root=True,
         )
-        for source, target in mapping.items()
+    return normalized
-    }
+
+
+def normalize_output_bindings(
+    bindings: Sequence[OutputBindingArg] | None,
+) -> list[OutputBinding]:
+    """Validate canonical output binding structs for WorkflowBuilder.use()."""
+    if bindings is None:
+        return []
+    normalized: list[OutputBinding] = []
+    for binding in bindings:
+        if isinstance(binding, OutputBinding):
+            normalized.append(binding)
+            continue
+        if not isinstance(binding, Mapping):
+            raise TypeError(f"unsupported output binding {binding!r}")
+        normalized.append(OutputBinding.model_validate(binding))
+    return normalized
 
 
 def auto_input_map(

src/wf_authoring/dsl/mapping.py

@@ -22,9 +22,9 @@ def normalize_path(path: PathArg) -> str:
 
 def bind_fields(**mapping: PathArg) -> dict[str, str]:
     return {
-        str(coerce_graph_path(source.path if isinstance(source, GraphPath) else source)): (
+        str(
-            destination
+            coerce_graph_path(source.path if isinstance(source, GraphPath) else source)
-        )
+        ): (destination)
         for destination, source in mapping.items()
     }
 

src/wf_authoring/dsl/path_inputs.py

@@ -7,12 +7,7 @@ from typing import TypeAlias, cast
 from wf_core.paths import GraphRoot, GraphSourcePath, LocalPath, StatePath
 
 PathInput: TypeAlias = (
-    str
+    str | Iterable[str] | Mapping[str, object] | GraphSourcePath | StatePath | LocalPath
-    | Iterable[str]
-    | Mapping[str, object]
-    | GraphSourcePath
-    | StatePath
-    | LocalPath
 )
 
 

tests/authoring/test_builder.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import warnings
+from collections.abc import Iterator, Mapping
+
 import pytest
 
 from wf_authoring import WorkflowBuilder, input_path, state, state_path
 from wf_core import END, RunStatus, WorkflowExecutionError
-from wf_core.models.steps import InputPathBinding
+from wf_core.models.steps import InputPathBinding, InputValueBinding
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 from tests.authoring.helpers import (
@@ -13,6 +16,7 @@ from tests.authoring.helpers import (
     AutoBindState,
     auto_bind_node,
 )
+from wf_authoring.builder.mapping import normalize_input_mapping
 
 
 def test_builder_auto_binds_matching_node_inputs_and_outputs_to_state() -> None:
@@ -87,6 +91,44 @@ def test_builder_use_accepts_typed_paths_and_literal_iterable_paths() -> None:
     assert step.output[0].target == StatePath(("state field",))
 
 
+def test_builder_use_accepts_canonical_binding_dicts_with_structural_paths() -> None:
+    builder = WorkflowBuilder(
+        name="canonical_binding_dicts",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    step = builder.use(
+        auto_bind_node,
+        input=[
+            {
+                "target": {"root": "local", "parts": ["payload.text"]},
+                "path": {"root": "input", "parts": ["text.with.dot"]},
+            },
+            {
+                "target": {"root": "local", "parts": ["static.limit"]},
+                "value": 3,
+            },
+        ],
+        output=[
+            {
+                "source": {"root": "local", "parts": ["payload.text"]},
+                "target": {"root": "state", "parts": ["text.with.dot"]},
+            }
+        ],
+    )
+
+    assert isinstance(step.input[0], InputPathBinding)
+    assert step.input[0].path == GraphSourcePath("input", ("text.with.dot",))
+    assert step.input[0].target == LocalPath(("payload.text",))
+    assert isinstance(step.input[1], InputValueBinding)
+    assert step.input[1].target == LocalPath(("static.limit",))
+    assert step.input[1].value == 3
+    assert step.output[0].source == LocalPath(("payload.text",))
+    assert step.output[0].target == StatePath(("text.with.dot",))
+
+
 def test_builder_preserves_explicit_root_node_local_maps() -> None:
     builder = WorkflowBuilder(
         name="root_local_maps",
@@ -272,3 +314,119 @@ def test_builder_use_ref_creates_external_node_use_without_node_def() -> None:
     assert step.output[0].source == LocalPath.of("echoed")
     assert step.output[0].target == StatePath.of("echoed")
     assert workflow.node_defs == []
+
+
+def test_builder_use_ref_accepts_canonical_binding_dicts() -> None:
+    builder = WorkflowBuilder(
+        name="external_ref_canonical_bindings",
+        input_schema={},
+        state_schema={"fields": {}},
+        output_schema={},
+    )
+
+    step = builder.use_ref(
+        "demo.echo",
+        id="echo",
+        input=[
+            {
+                "target": {"root": "local", "parts": ["text"]},
+                "path": {"root": "input", "parts": ["text"]},
+            }
+        ],
+        output=[
+            {
+                "source": {"root": "local", "parts": ["echoed"]},
+                "target": {"root": "state", "parts": ["echoed"]},
+            }
+        ],
+    )
+
+    assert step.node == "demo.echo"
+    assert isinstance(step.input[0], InputPathBinding)
+    assert step.input[0].path == GraphSourcePath.input("text")
+    assert step.output[0].target == StatePath.of("echoed")
+
+
+def test_builder_warns_when_explicit_deprecated_maps_are_used() -> None:
+    builder = WorkflowBuilder(
+        name="deprecated_maps",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    with pytest.warns(DeprecationWarning, match="canonical input/output"):
+        builder.use(
+            auto_bind_node,
+            in_map={"input.text": "text"},
+            out_map={"text": "state.text"},
+        )
+
+
+def test_builder_auto_mapping_does_not_warn() -> None:
+    builder = WorkflowBuilder(
+        name="auto_map_no_warning",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        builder.use(auto_bind_node)
+
+
+def test_builder_rejects_mixed_canonical_and_deprecated_input_styles() -> None:
+    builder = WorkflowBuilder(
+        name="mixed_input_styles",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    with pytest.raises(TypeError, match="cannot mix canonical input"):
+        builder.use(
+            auto_bind_node,
+            input=[{"target": "text", "path": "input.text"}],
+            in_map={"input.text": "text"},
+        )
+
+
+def test_builder_rejects_mixed_canonical_and_deprecated_output_styles() -> None:
+    builder = WorkflowBuilder(
+        name="mixed_output_styles",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    with pytest.raises(TypeError, match="cannot mix canonical output"):
+        builder.use(
+            auto_bind_node,
+            output=[{"source": "text", "target": "state.text"}],
+            out_map={"text": "state.text"},
+        )
+
+
+class _StructuralKeyMap(Mapping[object, object]):
+    def __getitem__(self, key: object) -> object:
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(())
+
+    def __len__(self) -> int:
+        return 1
+
+    def items(self) -> list[tuple[dict[str, object], str]]:
+        return [
+            (
+                {"root": "input", "parts": ["email.address"]},
+                "payload.email",
+            )
+        ]
+
+
+def test_input_map_rejects_structural_dict_keys_with_clear_message() -> None:
+    with pytest.raises(TypeError, match="structural path dicts cannot be map keys"):
+        normalize_input_mapping(_StructuralKeyMap())