diff --git a/src/wf_core/models/steps.py b/src/wf_core/models/steps.py index 6aac2a30..0f9b78f7 100644 --- a/src/wf_core/models/steps.py +++ b/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.""" diff --git a/tests/authoring/test_builder.py b/tests/authoring/test_builder.py index fa17164e..95bc33f8 100644 --- a/tests/authoring/test_builder.py +++ b/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" diff --git a/tests/authoring/test_subgraph.py b/tests/authoring/test_subgraph.py index 5487d206..96901517 100644 --- a/tests/authoring/test_subgraph.py +++ b/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" diff --git a/tests/core/test_canonical_node_bindings.py b/tests/core/test_canonical_node_bindings.py index 74b1dd11..3202acf6 100644 --- a/tests/core/test_canonical_node_bindings.py +++ b/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")