nested path support for in/out maps, merge strat refactor

validation + runtime for write targets

examples/mcp_tool_workflow.py

@@ -34,41 +34,43 @@ async def run_example() -> dict[str, object]:
 
     await service.refresh_connection_catalog("fixture.personal")
 
-    plan = RawWorkflowPlan(
-        name="mcp_echo_workflow",
-        input_schema={
-            "type": "object",
-            "properties": {"text": {"type": "string"}},
-            "required": ["text"],
-        },
-        state_schema={"fields": {"echoed": {"type": "string"}}},
-        output_schema={
-            "type": "object",
-            "properties": {"echoed": {"type": "string"}},
-            "required": ["echoed"],
-        },
-        start="echo",
-        nodes=[
-            {
-                "id": "echo",
-                "type": "node",
-                "node": "fixture.personal.echo_tool",
-                "in_map": {"input.text": "text"},
-                "out_map": {"echoed": "state.echoed"},
+    plan = RawWorkflowPlan.model_validate(
+        {
+            "name": "mcp_echo_workflow",
+            "input_schema": {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
             },
-            {
-                "id": "raise_mcp_error",
-                "type": "node",
-                "node": "wf.std.runtime_error",
-                "in_map": {"input.text": "message"},
-                "out_map": {},
+            "state_schema": {"fields": {"echoed": {"type": "string"}}},
+            "output_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
             },
-        ],
-        edges=[
-            {"from": "echo", "outcome": "ok", "to": END},
-            {"from": "echo", "outcome": "error", "to": "raise_mcp_error"},
-            {"from": "raise_mcp_error", "outcome": "ok", "to": END},
-        ],
+            "start": "echo",
+            "nodes": [
+                {
+                    "id": "echo",
+                    "type": "node",
+                    "node": "fixture.personal.echo_tool",
+                    "in_map": {"input.text": "text"},
+                    "out_map": {"echoed": "state.echoed"},
+                },
+                {
+                    "id": "raise_mcp_error",
+                    "type": "node",
+                    "node": "wf.std.runtime_error",
+                    "in_map": {"input.text": "message"},
+                    "out_map": {},
+                },
+            ],
+            "edges": [
+                {"from": "echo", "outcome": "ok", "to": END},
+                {"from": "echo", "outcome": "error", "to": "raise_mcp_error"},
+                {"from": "raise_mcp_error", "outcome": "ok", "to": END},
+            ],
+        }
     )
 
     run = await service.run_workflow_from_plan(plan, {"text": "hello from MCP"})

src/wf_core/local_paths.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+
+class LocalPathError(ValueError):
+    """Raised when a node-local dotted path cannot be parsed or resolved."""
+
+
+def split_local_path(path: str) -> list[str]:
+    """Split one dotted node-local path, rejecting empty segments."""
+    parts = path.split(".")
+    if not path or any(not part for part in parts):
+        raise LocalPathError(f"invalid local path {path!r}")
+    return parts
+
+
+def get_local_value(payload: Mapping[str, Any], path: str) -> Any:
+    """Resolve one node-local path from a nested mapping payload."""
+    current: Any = payload
+    for part in split_local_path(path):
+        if not isinstance(current, Mapping) or part not in current:
+            raise LocalPathError(f"local path {path!r} could not be resolved")
+        current = current[part]
+    return current
+
+
+def set_local_value(payload: dict[str, Any], path: str, value: Any) -> None:
+    """Write one value into a nested node-local mapping payload."""
+    parts = split_local_path(path)
+    current = payload
+    for part in parts[:-1]:
+        next_value = current.setdefault(part, {})
+        if not isinstance(next_value, dict):
+            raise LocalPathError(f"local path {path!r} overlaps an existing value")
+        current = next_value
+    current[parts[-1]] = value
+
+
+def paths_overlap(left: str, right: str) -> bool:
+    """Return whether two dotted paths overlap by equality or ancestry."""
+    left_parts = split_local_path(left)
+    right_parts = split_local_path(right)
+    shortest = min(len(left_parts), len(right_parts))
+    return left_parts[:shortest] == right_parts[:shortest]
+
+
+def has_overlapping_paths(paths: Iterable[str]) -> bool:
+    """Return whether any pair of dotted paths overlaps."""
+    seen: list[str] = []
+    for path in paths:
+        if any(paths_overlap(path, prior) for prior in seen):
+            return True
+        seen.append(path)
+    return False

src/wf_core/runtime/ops/merges.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from wf_core.errors import WorkflowExecutionError
+
+
+def apply_builtin_merge(
+    *,
+    strategy: str,
+    current_value: Any,
+    incoming_value: Any,
+    destination_path: str,
+) -> Any:
+    """Apply one built-in merge rule.
+
+    This is the future seam for source-owned reducer libraries. The current core
+    still supports only built-in rules and keeps them pure over current and
+    incoming values.
+    """
+    if strategy == "replace":
+        return incoming_value
+
+    if strategy == "append":
+        if current_value is None:
+            return [incoming_value] if not isinstance(incoming_value, list) else incoming_value
+        if not isinstance(current_value, list):
+            raise WorkflowExecutionError(
+                f"cannot append into non-list state path {destination_path!r}"
+            )
+        return [
+            *current_value,
+            *incoming_value,
+        ] if isinstance(incoming_value, list) else [*current_value, incoming_value]
+
+    if strategy == "merge_object":
+        if current_value is None:
+            if not isinstance(incoming_value, dict):
+                raise WorkflowExecutionError(
+                    f"cannot merge non-object value into {destination_path!r}"
+                )
+            return dict(incoming_value)
+        if not isinstance(current_value, dict) or not isinstance(incoming_value, dict):
+            raise WorkflowExecutionError(
+                f"merge_object requires dict values at {destination_path!r}"
+            )
+        return current_value | incoming_value
+
+    raise WorkflowExecutionError(f"unknown merge strategy {strategy!r}")

src/wf_core/runtime/ops/nodes.py

@@ -5,6 +5,7 @@ from typing import Any, cast
 
 from wf_core.conditions import safe_resolve_path
 from wf_core.errors import WorkflowExecutionError
+from wf_core.local_paths import LocalPathError, set_local_value
 from wf_core.models.results import NodeResult
 from wf_core.models.schemas import NodeDef
 from wf_core.models.steps import NodeUse
@@ -30,15 +31,18 @@ def _resolve_node_execution(
 ) -> tuple[dict[str, Any], RuntimeContext]:
     frame = run.current_frame()
     context_values = frame_context_values(frame)
-    resolved_input = {
-        destination_field: safe_resolve_path(
+    resolved_input: dict[str, Any] = {}
+    for source_path, destination_field in node.in_map.items():
+        value = safe_resolve_path(
             source_path,
             state=run.state,
             workflow_input=run.workflow_input,
             context=context_values,
         )
-        for source_path, destination_field in node.in_map.items()
-    }
+        try:
+            set_local_value(resolved_input, destination_field, value)
+        except LocalPathError as exc:
+            raise WorkflowExecutionError(str(exc)) from exc
     validate_payload_against_schema(
         node_def.input_schema, resolved_input, f"node input for {node.id}"
     )

src/wf_core/runtime/ops/state.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
+from wf_core.local_paths import LocalPathError, get_local_value, has_overlapping_paths
 from wf_core.models.steps import NodeUse
 from wf_core.models.workflow import Workflow
 from wf_core.paths import (
@@ -11,6 +12,7 @@ from wf_core.paths import (
     set_nested_value,
     split_graph_path,
 )
+from wf_core.runtime.ops.merges import apply_builtin_merge
 
 
 def apply_output_map(
@@ -36,16 +38,22 @@ def apply_mapped_state(
     *,
     missing_field_message: str,
 ) -> dict[str, Any]:
-    state_changes: dict[str, Any] = {}
+    if has_overlapping_paths(mapping.values()):
+        raise WorkflowExecutionError("mapped state patch has overlapping destination paths")
+
+    patch: dict[str, Any] = {}
     for source_field, destination_path in mapping.items():
-        if source_field not in source_data:
+        try:
+            value = get_local_value(source_data, source_field)
+        except LocalPathError:
             raise WorkflowExecutionError(
                 missing_field_message.format(field=repr(source_field))
-            )
-        value = source_data[source_field]
+            ) from None
+        patch[destination_path] = value
+
+    for destination_path, value in patch.items():
         write_state_value(workflow, state, destination_path, value)
-        state_changes[destination_path] = value
-    return state_changes
+    return dict(patch)
 
 
 def write_state_value(
@@ -65,44 +73,14 @@ def write_state_value(
     declared_field = workflow.state_schema.fields.get(field_name)
     merge_strategy = declared_field.merge_strategy if declared_field else "replace"
     key_path = parts
-
-    if merge_strategy == "replace":
-        safe_set_nested_value(state, key_path, value)
-        return
-
     current_value = get_nested_value(state, key_path)
-    if merge_strategy == "append":
-        if current_value is None:
-            safe_set_nested_value(
-                state, key_path, [value] if not isinstance(value, list) else value
-            )
-            return
-        if not isinstance(current_value, list):
-            raise WorkflowExecutionError(
-                f"cannot append into non-list state path {destination_path!r}"
-            )
-        if isinstance(value, list):
-            current_value.extend(value)
-        else:
-            current_value.append(value)
-        return
-
-    if merge_strategy == "merge_object":
-        if current_value is None:
-            if not isinstance(value, dict):
-                raise WorkflowExecutionError(
-                    f"cannot merge non-object value into {destination_path!r}"
-                )
-            safe_set_nested_value(state, key_path, dict(value))
-            return
-        if not isinstance(current_value, dict) or not isinstance(value, dict):
-            raise WorkflowExecutionError(
-                f"merge_object requires dict values at {destination_path!r}"
-            )
-        current_value.update(value)
-        return
-
-    raise WorkflowExecutionError(f"unknown merge strategy {merge_strategy!r}")
+    merged_value = apply_builtin_merge(
+        strategy=merge_strategy,
+        current_value=current_value,
+        incoming_value=value,
+        destination_path=destination_path,
+    )
+    safe_set_nested_value(state, key_path, merged_value)
 
 
 def project_output(workflow: Workflow, state: dict[str, Any]) -> dict[str, Any]:

src/wf_core/validation/steps.py

@@ -9,6 +9,7 @@ from wf_core.models.conditions import (
     PathOperand,
     VariadicCondition,
 )
+from wf_core.local_paths import LocalPathError, has_overlapping_paths, split_local_path
 from wf_core.models.schemas import NodeDef
 from wf_core.models.steps import ConditionNode, ForeachNode, InterruptNode, NodeUse
 from wf_core.models.workflow import Workflow
@@ -38,7 +39,11 @@ def validate_node_use(
     input_root_fields = set(workflow.input_schema.properties)
 
     for source_path, destination_field in node.in_map.items():
-        if destination_field not in input_fields:
+        try:
+            destination_root = split_local_path(destination_field)[0]
+        except LocalPathError:
+            destination_root = ""
+        if destination_root not in input_fields:
             report.add(
                 ValidationIssueCode.INVALID_NODE_INPUT_FIELD,
                 f"nodes[{index}].in_map[{source_path!r}]",
@@ -53,8 +58,19 @@ def validate_node_use(
                 "source path must start with input., state., or context. and reference a declared root field when applicable",
             )
 
+    if has_overlapping_paths(node.in_map.values()):
+        report.add(
+            ValidationIssueCode.INVALID_NODE_INPUT_FIELD,
+            f"nodes[{index}].in_map",
+            "in_map has overlapping node-local input paths",
+        )
+
     for source_field, destination_path in node.out_map.items():
-        if source_field not in output_fields:
+        try:
+            source_root = split_local_path(source_field)[0]
+        except LocalPathError:
+            source_root = ""
+        if source_root not in output_fields:
             report.add(
                 ValidationIssueCode.INVALID_NODE_OUTPUT_FIELD,
                 f"nodes[{index}].out_map[{source_field!r}]",
@@ -66,6 +82,12 @@ def validate_node_use(
                 f"nodes[{index}].out_map[{source_field!r}]",
                 "destination path must start with state.",
             )
+    if has_overlapping_paths(node.out_map.values()):
+        report.add(
+            ValidationIssueCode.INVALID_DESTINATION_PATH,
+            f"nodes[{index}].out_map",
+            "out_map has overlapping state destination paths",
+        )
 
 
 def validate_condition_node(

tests/authoring/test_builder.py

@@ -41,6 +41,24 @@ def test_builder_auto_binds_matching_node_inputs_and_outputs_to_state() -> None:
     assert run.state["count"] == 2
 
 
+def test_builder_preserves_explicit_nested_node_local_maps() -> None:
+    builder = WorkflowBuilder(
+        name="nested_local_maps",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+
+    step = builder.use(
+        auto_bind_node,
+        in_map={"state.text": "payload.text"},
+        out_map={"payload.text": "state.text"},
+    )
+
+    assert step.in_map == {"state.text": "payload.text"}
+    assert step.out_map == {"payload.text": "state.text"}
+
+
 def test_builder_can_auto_id_node_uses_from_spec_name() -> None:
     builder = WorkflowBuilder(
         name="auto_id_demo",

tests/core/test_mapping_validation.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from wf_core import Edge, NodeDef, NodeUse, SchemaRef, StateField, StateSchema, Workflow
+
+
+def test_validation_allows_nested_node_local_paths() -> None:
+    report = _workflow(
+        in_map={"input.person.name": "user.name"},
+        out_map={"user.age": "state.person.age"},
+    ).validate_structure()
+
+    assert report.errors == []
+
+
+def test_validation_rejects_overlapping_node_input_destinations() -> None:
+    report = _workflow(
+        in_map={
+            "input.person": "user",
+            "input.person.name": "user.name",
+        },
+        out_map={},
+    ).validate_structure()
+
+    assert any("overlapping node-local input paths" in issue.message for issue in report.errors)
+
+
+def test_validation_rejects_overlapping_state_write_destinations() -> None:
+    report = _workflow(
+        in_map={},
+        out_map={
+            "user": "state.person",
+            "user.age": "state.person.age",
+        },
+    ).validate_structure()
+
+    assert any("overlapping state destination paths" in issue.message for issue in report.errors)
+
+
+def _workflow(*, in_map: dict[str, str], out_map: dict[str, str]) -> Workflow:
+    return Workflow(
+        name="mapping_validation",
+        input_schema=SchemaRef.model_validate(
+            {"type": "object", "properties": {"person": {"type": "object"}}}
+        ),
+        state_schema=StateSchema(fields={"person": StateField(type="object")}),
+        output_schema=SchemaRef(type="object", properties={}),
+        node_defs=[
+            NodeDef(
+                name="tool",
+                input_schema=SchemaRef.model_validate(
+                    {"type": "object", "properties": {"user": {"type": "object"}}}
+                ),
+                output_schema=SchemaRef.model_validate(
+                    {"type": "object", "properties": {"user": {"type": "object"}}}
+                ),
+                outcomes=["ok"],
+            )
+        ],
+        start="tool",
+        nodes=[
+            NodeUse(
+                id="tool",
+                type="node",
+                node="tool",
+                in_map=in_map,
+                out_map=out_map,
+            )
+        ],
+        edges=[Edge.model_validate({"from": "tool", "outcome": "ok", "to": "__end__"})],
+    )

tests/core/test_nested_mappings.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import pytest
+
+from wf_core import (
+    END,
+    Edge,
+    NodeDef,
+    NodeUse,
+    SchemaRef,
+    StateField,
+    StateSchema,
+    Workflow,
+    WorkflowExecutionError,
+    execute_workflow,
+)
+
+
+def test_nested_node_local_paths_build_input_and_read_output() -> None:
+    workflow = _nested_mapping_workflow()
+
+    run = execute_workflow(
+        workflow,
+        {"person": {"name": "Ada"}, "digital": {"email": "[email protected]"}},
+        {
+            "big_tool": lambda payload, _ctx: {
+                "outcome": "ok",
+                "output": {
+                    "user": {"age": 36, "gender": "x"},
+                    "job": {"years": 12},
+                },
+            }
+        },
+    )
+
+    assert run.trace[0].resolved_input == {
+        "user": {"name": "Ada", "email": "[email protected]"}
+    }
+    assert run.state["person"]["age"] == 36
+    assert run.state["person"]["gender"] == "x"
+    assert run.state["experience"]["years"] == 12
+
+
+def test_missing_nested_node_output_path_fails() -> None:
+    workflow = _nested_mapping_workflow()
+
+    with pytest.raises(
+        WorkflowExecutionError,
+        match="did not return required mapped field 'user.gender'",
+    ):
+        execute_workflow(
+            workflow,
+            {"person": {"name": "Ada"}, "digital": {"email": "[email protected]"}},
+            {
+                "big_tool": lambda payload, _ctx: {
+                    "outcome": "ok",
+                    "output": {"user": {"age": 36}, "job": {"years": 12}},
+                }
+            },
+        )
+
+
+def _nested_mapping_workflow() -> Workflow:
+    return Workflow(
+        name="nested_mapping",
+        input_schema=SchemaRef.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "person": {"type": "object"},
+                    "digital": {"type": "object"},
+                },
+            }
+        ),
+        state_schema=StateSchema(
+            fields={
+                "person": StateField(type="object"),
+                "experience": StateField(type="object"),
+            }
+        ),
+        output_schema=SchemaRef(type="object", properties={}),
+        node_defs=[
+            NodeDef(
+                name="big_tool",
+                input_schema=SchemaRef.model_validate(
+                    {
+                        "type": "object",
+                        "properties": {"user": {"type": "object"}},
+                    }
+                ),
+                output_schema=SchemaRef.model_validate(
+                    {
+                        "type": "object",
+                        "properties": {
+                            "user": {"type": "object"},
+                            "job": {"type": "object"},
+                        },
+                    }
+                ),
+                outcomes=["ok"],
+            )
+        ],
+        start="big",
+        nodes=[
+            NodeUse(
+                id="big",
+                type="node",
+                node="big_tool",
+                in_map={
+                    "input.person.name": "user.name",
+                    "input.digital.email": "user.email",
+                },
+                out_map={
+                    "user.age": "state.person.age",
+                    "user.gender": "state.person.gender",
+                    "job.years": "state.experience.years",
+                },
+            )
+        ],
+        edges=[Edge.model_validate({"from": "big", "outcome": "ok", "to": END})],
+    )

tests/wf_mcp/test_service.py

@@ -25,7 +25,7 @@ from .test_support import (
 
 
 def _single_echo_plan(plan_name: str, node_name: str) -> RawWorkflowPlan:
-    return RawWorkflowPlan(
+    return _raw_plan(
         name=plan_name,
         input_schema={
             "type": "object",
@@ -54,6 +54,11 @@ def _single_echo_plan(plan_name: str, node_name: str) -> RawWorkflowPlan:
     )
 
 
+def _raw_plan(**payload: object) -> RawWorkflowPlan:
+    """Parse JSON-shaped workflow input through the public typed boundary."""
+    return RawWorkflowPlan.model_validate(payload)
+
+
 def test_service_builds_namespaced_catalog() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "catalog_store"))
     service.register_connection(
@@ -274,7 +279,7 @@ def test_service_compiles_and_runs_raw_plan() -> None:
     )
     service.register_specs("demo.personal", echo_tool, finalize_tool)
 
-    plan = RawWorkflowPlan(
+    plan = _raw_plan(
         name="demo_plan",
         input_schema={
             "type": "object",
@@ -343,7 +348,7 @@ def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
     )
     service.register_specs("demo.personal", dotted_echo_tool)
 
-    plan = RawWorkflowPlan(
+    plan = _raw_plan(
         name="dotted_local_name_plan",
         input_schema={
             "type": "object",
@@ -407,7 +412,7 @@ def test_service_does_not_resolve_specs_hidden_from_planner() -> None:
         )
     )
 
-    plan = RawWorkflowPlan(
+    plan = _raw_plan(
         name="hidden_source_plan",
         input_schema={
             "type": "object",
@@ -697,7 +702,7 @@ def test_service_records_tool_call_events() -> None:
 
     asyncio.run(service.refresh_connection_catalog("demo.personal"))
 
-    plan = RawWorkflowPlan(
+    plan = _raw_plan(
         name="tool_only_plan",
         input_schema={
             "type": "object",
@@ -750,7 +755,7 @@ def test_service_can_call_upstream_tool_through_wf_mcp_system_node() -> None:
     )
     service.register_adapter("demo", FakeAdapter())
 
-    plan = RawWorkflowPlan(
+    plan = _raw_plan(
         name="system_tool_plan",
         input_schema={
             "type": "object",