feat: expose foreach context paths

src/wf_core/models/steps.py

@@ -274,6 +274,24 @@ class ForeachNode(BaseModel):
             raise ValueError("concurrent policy is only valid when mode='concurrent'")
         return self
 
+    @property
+    def item(self) -> GraphSourcePath:
+        """Return the structured item path for this foreach controller.
+
+        The constructor uses literal tuple segments so a foreach id
+        containing dots stays one TOML-quoted segment instead of being
+        reparsed as nested path separators.
+        """
+        return GraphSourcePath("context", ("foreach", self.id, "item"))
+
+    @property
+    def index(self) -> GraphSourcePath:
+        """Return the structured index path for this foreach controller.
+
+        Like :attr:`item`, the foreach id is one literal segment.
+        """
+        return GraphSourcePath("context", ("foreach", self.id, "index"))
+
 
 class JoinNode(BaseModel):
     """Control-flow step that marks a branch or frame as joined."""

tests/authoring/test_builder.py

@@ -773,3 +773,72 @@ class _StructuralKeyMap:
 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(cast(Mapping[object, object], _StructuralKeyMap()))
+
+
+def test_foreach_reference_exposes_item_and_index_paths() -> None:
+    from wf_core.paths import GraphSourcePath
+
+    builder = WorkflowBuilder(
+        name="foreach_ref",
+        input_schema={"type": "object"},
+        state_schema={"type": "object"},
+        output_schema={"type": "object"},
+    )
+    each = builder.foreach(id="orders", over=state_path("orders"), as_="order")
+
+    assert each.item == GraphSourcePath("context", ("foreach", "orders", "item"))
+    assert each.index == GraphSourcePath("context", ("foreach", "orders", "index"))
+    assert str(each.item) == "context.foreach.orders.item"
+    assert str(each.index) == "context.foreach.orders.index"
+
+
+def test_foreach_reference_treats_dotted_id_as_one_literal_segment() -> None:
+    from wf_core.paths import GraphSourcePath
+
+    builder = WorkflowBuilder(
+        name="foreach_dotted",
+        input_schema={"type": "object"},
+        state_schema={"type": "object"},
+        output_schema={"type": "object"},
+    )
+    each = builder.foreach(
+        id="orders.v2", over=state_path("orders"), as_="order"
+    )
+
+    assert each.item == GraphSourcePath(
+        "context", ("foreach", "orders.v2", "item")
+    )
+    assert str(each.item) == 'context.foreach."orders.v2".item'
+    assert str(each.index) == 'context.foreach."orders.v2".index'
+
+
+def test_foreach_computed_paths_are_not_serialized_fields() -> None:
+    builder = WorkflowBuilder(
+        name="foreach_serialized",
+        input_schema={"type": "object"},
+        state_schema={"type": "object"},
+        output_schema={"type": "object"},
+    )
+    each = builder.foreach(id="orders", over=state_path("orders"), as_="order")
+
+    dumped = each.model_dump(mode="json")
+    assert "item" not in dumped
+    assert "index" not in dumped
+
+
+def test_foreach_ref_works_in_node_input_binding() -> None:
+    builder = WorkflowBuilder(
+        name="foreach_node_binding",
+        input_schema={"type": "object"},
+        state_schema={"type": "object"},
+        output_schema={"type": "object"},
+    )
+    each = builder.foreach(id="orders", over=state_path("orders"), as_="order")
+
+    work = builder.use(
+        auto_bind_node,
+        input=[input_from(each.item, "order")],
+    )
+    binding = work.input[0]
+    assert isinstance(binding, object)
+    assert str(binding.path) == "context.foreach.orders.item"

tests/authoring/test_subgraph.py

@@ -273,3 +273,22 @@ def test_workflow_builder_subgraph_adds_native_subgraph_node() -> None:
     assert workflow.nodes[0].id == "run_child"
     assert workflow.nodes[0].outcomes == child.outcomes
     assert workflow.validate_structure().errors == []
+
+
+def test_foreach_ref_works_in_subgraph_input_binding() -> None:
+    from wf_authoring import state_path
+
+    child = build_demo_workflow()
+    parent = WorkflowBuilder(
+        name="foreach_subgraph_parent",
+        input_schema={"type": "object"},
+        state_schema={"type": "object"},
+        output_schema={"type": "object"},
+    )
+    each = parent.foreach(id="orders", over=state_path("orders"), as_="order")
+    step = parent.subgraph(
+        workflow=child,
+        id="run_child",
+        input=[input_from(each.item, "order")],
+    )
+    assert str(step.input[0].path) == "context.foreach.orders.item"

tests/core/test_canonical_node_bindings.py

@@ -448,3 +448,26 @@ def test_foreach_node_serializes_over_path_as_canonical_string():
 
     assert node.over == GraphSourcePath.state("items")
     assert node.model_dump(mode="json")["over"] == "state.items"
+
+
+def test_foreach_ref_item_index_are_literal_structured_paths() -> None:
+    node = ForeachNode.model_validate(
+        {
+            "id": "orders.v2",
+            "type": "foreach",
+            "over": "state.orders",
+            "as": "order",
+        }
+    )
+    assert node.item == GraphSourcePath("context", ("foreach", "orders.v2", "item"))
+    assert node.index == GraphSourcePath("context", ("foreach", "orders.v2", "index"))
+    assert str(node.item) == 'context.foreach."orders.v2".item'
+    assert "item" not in node.model_dump(mode="json")
+    assert "index" not in node.model_dump(mode="json")
+    # GraphSourcePath still rejects an output root.
+    try:
+        GraphSourcePath.parse("output.result")
+    except Exception:
+        pass
+    else:
+        raise AssertionError("expected output root to be rejected")