diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py index ae0af01b..0fddf695 100644 --- a/src/wf_api/authoring_contracts.py +++ b/src/wf_api/authoring_contracts.py @@ -10,6 +10,7 @@ from wf_core.analysis.context_scopes import ( context_fields_by_node, ) from wf_core.models.workflow import Workflow +from wf_core.paths import GraphSourcePath from .models.authoring_contracts import ( AuthoringContractInventoryPayload, @@ -173,6 +174,8 @@ def context_path_options( ) -> list[AuthoringPathOptionPayload]: """Project analyzed runtime context fields into Task 1 path payloads.""" options: list[AuthoringPathOptionPayload] = [] + foreach_schema: Mapping[str, Any] | None = None + foreach_availability: str = "available" for field in fields: if isinstance(field, ContextFieldAvailability): name = field.name @@ -212,6 +215,13 @@ def context_path_options( if reason is not None: option["reason"] = reason options.append(option) + if name == "foreach" and isinstance(schema, Mapping): + foreach_schema = schema + foreach_availability = availability + if foreach_schema is not None: + options.extend( + _nested_foreach_path_options(foreach_schema, foreach_availability) + ) return options @@ -223,6 +233,119 @@ def context_path_options_for_node( return context_path_options(context_fields_by_node(workflow).get(node_id, ())) +def _nested_foreach_path_options( + foreach_schema: Mapping[str, Any], + availability: str, +) -> list[AuthoringPathOptionPayload]: + """Emit literal structured paths beneath the ``foreach`` map. + + Foreach ids are literal TOML segments formatted through + ``GraphSourcePath`` so dotted ids stay quoted as one segment. The walk is + bounded like other authoring traversals; arrays contribute no invented + child names. + """ + from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload + + options: list[_Payload] = [] + properties = foreach_schema.get("properties") + if not isinstance(properties, Mapping): + return options + for owner_id, entry_schema in properties.items(): + if not isinstance(owner_id, str) or not isinstance(entry_schema, Mapping): + continue + entry_properties = entry_schema.get("properties") + if not isinstance(entry_properties, Mapping): + continue + for prop_name, prop_schema in entry_properties.items(): + if not isinstance(prop_name, str) or not isinstance(prop_schema, Mapping): + continue + # Construct dotted ids as literal segments, never by string concat. + path = str(GraphSourcePath("context", ("foreach", owner_id, prop_name))) + option: _Payload = { + "path": path, + "label": prop_name.replace("_", " ").replace("-", " ").title(), + "origin": "runtime_context", + "schema": deepcopy(dict(prop_schema)), + "required": False, + "availability": availability, # type: ignore[typeddict-item] + "uses": ["step_input"], + } + options.append(option) + # Emit nested object properties beneath `.item` when the item is a + # bounded object schema, mirroring input/state inventory behavior. + if prop_name == "item": + options.extend( + _nested_item_subpaths( + prop_schema, owner_id, availability, depth=0 + ) + ) + return options + + +def _nested_item_subpaths( + item_schema: Mapping[str, Any], + owner_id: str, + availability: str, + *, + depth: int, + prefix_parts: tuple[str, ...] = (), +) -> list[AuthoringPathOptionPayload]: + """Emit bounded object children beneath one foreach ``item`` schema.""" + from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload + + if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: + return [] + properties = item_schema.get("properties") + if not isinstance(properties, Mapping): + return [] + options: list[_Payload] = [] + for name, sub_schema in properties.items(): + if not isinstance(name, str) or not isinstance(sub_schema, Mapping): + continue + if isinstance(sub_schema.get("type"), str) and sub_schema.get("type") == "array": + # Arrays are whole values; item indexes need real runtime indexes. + path = str( + GraphSourcePath( + "context", ("foreach", owner_id, "item", *prefix_parts, name) + ) + ) + options.append( + { + "path": path, + "label": name.replace("_", " ").replace("-", " ").title(), + "origin": "runtime_context", + "schema": deepcopy(dict(sub_schema)), + "required": False, + "availability": availability, # type: ignore[typeddict-item] + "uses": ["step_input"], + } + ) + continue + path = str( + GraphSourcePath( + "context", ("foreach", owner_id, "item", *prefix_parts, name) + ) + ) + options.append( + { + "path": path, + "label": name.replace("_", " ").replace("-", " ").title(), + "origin": "runtime_context", + "schema": deepcopy(dict(sub_schema)), + "required": False, + "availability": availability, # type: ignore[typeddict-item] + "uses": ["step_input"], + } + ) + options.extend( + _nested_item_subpaths( + sub_schema, owner_id, availability, depth=depth + 1, + prefix_parts=(*prefix_parts, name), + ) + ) + return options + + def _context_entries_for_inventory( entries: Sequence[AuthoringPathOptionPayload], ) -> list[AuthoringPathOptionPayload]: diff --git a/src/wf_core/analysis/__init__.py b/src/wf_core/analysis/__init__.py index 63815931..03c82910 100644 --- a/src/wf_core/analysis/__init__.py +++ b/src/wf_core/analysis/__init__.py @@ -4,6 +4,9 @@ from .context_scopes import ( ContextFieldAvailability, context_analysis_warnings, context_fields_by_node, + context_schema_for_node, + context_schemas_by_node, + root_context_schema, ) from .control_regions import ( ControlRegionAnalysis, @@ -22,4 +25,7 @@ __all__ = [ "analyze_control_regions", "context_analysis_warnings", "context_fields_by_node", + "context_schema_for_node", + "context_schemas_by_node", + "root_context_schema", ] diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 86f3d020..fe09ffdb 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -6,14 +6,20 @@ from dataclasses import dataclass from typing import Literal from wf_core.analysis.control_regions import ( + ControlRegionAnalysis, ForeachOwnerStack, analyze_control_regions, ) from wf_core.context_contracts import ( + FOREACH_CONTEXT_KEY, + LOOP_INDEX_CONTEXT_KEY, + LOOP_ITEM_CONTEXT_KEY, STANDARD_CONTEXT_FIELDS, ContextFieldContract, ContextSchema, foreach_context_fields, + foreach_entry_schema, + structured_foreach_contract, ) from wf_core.models.steps import ForeachNode from wf_core.models.workflow import Edge, Workflow @@ -65,17 +71,19 @@ class _Warnings: def context_fields_by_node( workflow: Workflow, + *, + control_regions: ControlRegionAnalysis | None = None, ) -> dict[str, tuple[ContextFieldAvailability, ...]]: """Return runtime context contracts for every reachable graph node. This is an abstract execution-frame analysis keyed by static control region: each node use belongs to exactly one foreach-owner stack, and - that stack decides which foreach aliases the node exposes. A node + that stack decides which foreach entries the node exposes. A node reachable under two stacks is a region conflict and receives no foreach fields. The traversal still memoizes node id and owner stack so cyclic graphs terminate. """ - return _analyze(workflow).fields_by_node + return _analyze(workflow, control_regions=control_regions).fields_by_node def context_analysis_warnings(workflow: Workflow) -> tuple[str, ...]: @@ -83,11 +91,69 @@ def context_analysis_warnings(workflow: Workflow) -> tuple[str, ...]: return _analyze(workflow).warnings -def _analyze(workflow: Workflow) -> _ContextAnalysis: +def context_schemas_by_node( + workflow: Workflow, + *, + control_regions: ControlRegionAnalysis | None = None, +) -> dict[str, ContextSchema]: + """Return schemas for all unambiguous, reachable program locations. + + A conflicted or unreachable node has no generated per-node schema; + callers treat that absence as invalid, not as root context. Schemas are + composed from the same owner-stack analysis as field contracts, not from + a second graph traversal. + """ + analysis = control_regions or analyze_control_regions(workflow) + foreach_nodes = { + node.id: node for node in workflow.nodes if isinstance(node, ForeachNode) + } + schemas: dict[str, ContextSchema] = {} + for node_id, stack in analysis.owner_stack_by_node.items(): + # Conflicted nodes are already dropped from owner_stack_by_node by the + # control-region analysis, so every entry here is unambiguous. + schemas[node_id] = _context_schema_for_stack( + workflow, foreach_nodes, analysis.owner_stack_by_node, stack + ) + return schemas + + +def context_schema_for_node(workflow: Workflow, node_id: str) -> ContextSchema: + """Return the complete graph-visible context object schema at one node.""" + schemas = context_schemas_by_node(workflow) + try: + return schemas[node_id] + except KeyError as exc: + raise KeyError(f"no context schema for node {node_id!r}") from exc + + +def root_context_schema() -> ContextSchema: + """Return standard fields plus an empty structured foreach map.""" + properties: dict[str, ContextSchema] = { + field.name: deepcopy(field.schema) for field in STANDARD_CONTEXT_FIELDS + } + properties[FOREACH_CONTEXT_KEY] = { + "type": "object", + "properties": {}, + "required": [], + "additionalProperties": False, + } + return { + "type": "object", + "properties": properties, + "required": sorted(properties), + "additionalProperties": False, + } + + +def _analyze( + workflow: Workflow, + *, + control_regions: ControlRegionAnalysis | None = None, +) -> _ContextAnalysis: """Derive context contracts from static foreach control regions. - Each unambiguous node use has exactly one owner stack; its active foreach - is the final stack item. A canonical return edge pops the stack, so the + Each unambiguous node use has exactly one owner stack; all active entries + in that stack are exposed. A canonical return edge pops the stack, so the controller itself stays in the outer context. Conflicted nodes receive no foreach fields. """ @@ -118,7 +184,7 @@ def _analyze(workflow: Workflow) -> _ContextAnalysis: warnings.add(f"workflow start targets missing node {workflow.start!r}") return _ContextAnalysis({}, tuple(warnings.values)) - analysis = analyze_control_regions(workflow) + analysis = control_regions or analyze_control_regions(workflow) for issue in analysis.issues: warnings.add( f"control region {issue.kind.value} at {issue.path}: {issue.message}" @@ -126,12 +192,11 @@ def _analyze(workflow: Workflow) -> _ContextAnalysis: fields_by_node: dict[str, tuple[ContextFieldAvailability, ...]] = {} for node_id, stack in analysis.owner_stack_by_node.items(): - active_foreach_id = stack[-1] if stack else None fields_by_node[node_id] = _available_fields( workflow, foreach_nodes, analysis.owner_stack_by_node, - active_foreach_id, + stack, ) return _ContextAnalysis(fields_by_node, tuple(warnings.values)) @@ -140,30 +205,67 @@ def _available_fields( workflow: Workflow, foreach_nodes: Mapping[str, ForeachNode], owner_stack_by_node: Mapping[str, ForeachOwnerStack], - active_scope: FrameScope, + stack: ForeachOwnerStack, ) -> tuple[ContextFieldAvailability, ...]: """Return contracts for one static owner stack; all are guaranteed. A single node use has one control region, so foreach fields are either - present (inside a body) or absent (outside). Conditional availability is - not used to represent multiple owner stacks: a node reached under two - stacks is a region conflict and receives no foreach fields at all. + present (inside bodies) or absent (outside). Every owner in the stack + contributes one required ``foreach.`` entry, every active configured + alias, and ``loop_item``/``loop_index`` from the final owner only. + Conditional availability is not used to represent multiple owner stacks: + a node reached under two stacks is a region conflict and receives no + foreach fields at all. """ contracts = list(STANDARD_CONTEXT_FIELDS) - if active_scope is not None: - foreach = foreach_nodes.get(active_scope) - if foreach is not None: - contracts.extend( - foreach_context_fields( - foreach.as_, - _foreach_item_schema( - workflow, - foreach, - foreach_nodes, - owner_stack_by_node, - ), - ) + if not stack: + return tuple( + ContextFieldAvailability( + contract=ContextFieldContract( + contract.name, + deepcopy(contract.schema), + contract.description, + ), + availability="available", ) + for contract in contracts + ) + entry_schemas: dict[str, ContextSchema] = {} + for owner_id in stack: + foreach = foreach_nodes.get(owner_id) + if foreach is None: + continue + entry_schemas[owner_id] = foreach_entry_schema( + owner_id, + _foreach_item_schema( + workflow, + foreach, + foreach_nodes, + owner_stack_by_node, + ), + ) + if entry_schemas: + contracts.append(structured_foreach_contract(entry_schemas)) + # Aliases for every active owner, plus innermost loop keys. + for owner_id in stack: + foreach = foreach_nodes.get(owner_id) + if foreach is None: + continue + item_schema = _foreach_item_schema( + workflow, foreach, foreach_nodes, owner_stack_by_node + ) + # Only the innermost owner contributes loop_item/loop_index; every + # owner contributes its configured alias when unambiguous. Alias + # collision handling is a validation concern; here we expose what the + # stack declares. + if owner_id == stack[-1]: + contracts.extend(foreach_context_fields(foreach.as_, item_schema)) + elif foreach.as_: + # Outer aliases are exposed without re-emitting loop keys. + outer_fields = foreach_context_fields(foreach.as_, item_schema) + for field in outer_fields: + if field.name not in (LOOP_ITEM_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY): + contracts.append(field) return tuple( ContextFieldAvailability( contract=ContextFieldContract( @@ -177,6 +279,70 @@ def _available_fields( ) +def _context_schema_for_stack( + workflow: Workflow, + foreach_nodes: Mapping[str, ForeachNode], + owner_stack_by_node: Mapping[str, ForeachOwnerStack], + stack: ForeachOwnerStack, +) -> ContextSchema: + """Build the complete graph-visible context object schema for one stack.""" + properties: dict[str, ContextSchema] = { + field.name: deepcopy(field.schema) for field in STANDARD_CONTEXT_FIELDS + } + entry_schemas: dict[str, ContextSchema] = {} + for owner_id in stack: + foreach = foreach_nodes.get(owner_id) + if foreach is None: + continue + entry_schemas[owner_id] = foreach_entry_schema( + owner_id, + _foreach_item_schema( + workflow, foreach, foreach_nodes, owner_stack_by_node + ), + ) + properties[FOREACH_CONTEXT_KEY] = { + "type": "object", + "properties": entry_schemas, + "required": sorted(entry_schemas), + "additionalProperties": False, + } + required: list[str] = [ + field.name for field in STANDARD_CONTEXT_FIELDS + ] + [FOREACH_CONTEXT_KEY] + if stack: + innermost = foreach_nodes.get(stack[-1]) + if innermost is not None: + inner_item = _foreach_item_schema( + workflow, innermost, foreach_nodes, owner_stack_by_node + ) + properties[LOOP_ITEM_CONTEXT_KEY] = deepcopy(inner_item) + properties[LOOP_INDEX_CONTEXT_KEY] = {"type": "integer"} + required.extend([LOOP_ITEM_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY]) + for owner_id in stack: + foreach = foreach_nodes.get(owner_id) + if foreach is not None and foreach.as_: + properties[foreach.as_] = deepcopy( + _foreach_item_schema( + workflow, foreach, foreach_nodes, owner_stack_by_node + ) + ) + required.append(foreach.as_) + # Deduplicate while preserving order; aliases could theoretically repeat + # a standard name, but validation rejects those collisions. + seen: set[str] = set() + deduped_required: list[str] = [] + for name in required: + if name not in seen: + seen.add(name) + deduped_required.append(name) + return { + "type": "object", + "properties": properties, + "required": sorted(deduped_required), + "additionalProperties": False, + } + + def _foreach_item_schema( workflow: Workflow, foreach: ForeachNode, @@ -185,18 +351,17 @@ def _foreach_item_schema( ) -> ContextSchema: """Resolve one controller's item schema in its own static context. - An inner foreach may declare ``over="context.outer_item"``; the lookup - uses the controller's own owner stack, not the inner body stack. + An inner foreach may declare ``over="context.foreach.outer.item.children"``; + the lookup uses the controller's own owner stack, not the inner body stack. """ controller_stack = owner_stack_by_node.get(foreach.id) if controller_stack is None: return {} - controller_scope: FrameScope = controller_stack[-1] if controller_stack else None source_schema = _schema_at_path( workflow, foreach.over.root, foreach.over.parts, - controller_scope, + controller_stack, foreach_nodes, owner_stack_by_node, ) @@ -228,7 +393,7 @@ def _schema_at_path( workflow: Workflow, root: str, parts: tuple[str, ...], - active_scope: FrameScope, + stack: ForeachOwnerStack, foreach_nodes: Mapping[str, ForeachNode], owner_stack_by_node: Mapping[str, ForeachOwnerStack], ) -> Mapping[str, object] | None: @@ -236,7 +401,7 @@ def _schema_at_path( schema_document = _schema_document( workflow, root, - active_scope=active_scope, + stack=stack, foreach_nodes=foreach_nodes, owner_stack_by_node=owner_stack_by_node, ) @@ -260,6 +425,7 @@ def _schema_document( workflow: Workflow, root: str, *, + stack: ForeachOwnerStack | None = None, active_scope: FrameScope = None, foreach_nodes: Mapping[str, ForeachNode] | None = None, owner_stack_by_node: Mapping[str, ForeachOwnerStack] | None = None, @@ -269,30 +435,54 @@ def _schema_document( if root == "state": return workflow.state_schema.model_dump(mode="json", exclude_none=True) if root == "context": + # Prefer the full owner stack when available; fall back to the legacy + # single active scope for callers that have not migrated yet. + resolved_stack: ForeachOwnerStack = () + if stack is not None: + resolved_stack = stack + elif active_scope is not None: + resolved_stack = (active_scope,) current: dict[str, object] = { field.name: field.schema for field in STANDARD_CONTEXT_FIELDS } if ( - active_scope is not None - and foreach_nodes is not None + foreach_nodes is not None and owner_stack_by_node is not None ): - foreach = foreach_nodes.get(active_scope) - if foreach is not None: - current.update( - { - field.name: field.schema - for field in foreach_context_fields( - foreach.as_, - _foreach_item_schema( - workflow, - foreach, - foreach_nodes, - owner_stack_by_node, - ), - ) - } - ) + entry_schemas: dict[str, object] = {} + for owner_id in resolved_stack: + foreach = foreach_nodes.get(owner_id) + if foreach is not None: + entry_schemas[owner_id] = foreach_entry_schema( + owner_id, + _foreach_item_schema( + workflow, + foreach, + foreach_nodes, + owner_stack_by_node, + ), + ) + current[FOREACH_CONTEXT_KEY] = { + "type": "object", + "properties": entry_schemas, + "required": sorted(entry_schemas), + "additionalProperties": False, + } + for owner_id in resolved_stack: + foreach = foreach_nodes.get(owner_id) + if foreach is not None: + item_schema = _foreach_item_schema( + workflow, + foreach, + foreach_nodes, + owner_stack_by_node, + ) + for field in foreach_context_fields(foreach.as_, item_schema): + # Innermost loop keys win; outer aliases accumulate. + # Validation owns collision diagnostics. + is_innermost = bool(resolved_stack) and owner_id == resolved_stack[-1] + if field.name not in current or is_innermost: + current[field.name] = field.schema return {"type": "object", "properties": current} return {} diff --git a/src/wf_core/context_contracts.py b/src/wf_core/context_contracts.py index 11063000..fc6f0530 100644 --- a/src/wf_core/context_contracts.py +++ b/src/wf_core/context_contracts.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Mapping from copy import deepcopy from dataclasses import dataclass from typing import Any @@ -87,3 +88,50 @@ def foreach_context_fields( ) ) return tuple(fields) + + +def foreach_entry_schema(node_id: str, item_schema: ContextSchema) -> ContextSchema: + """Return the JSON schema for one structured ``foreach.`` entry. + + Identity fields disclose the dynamic activation/frame/scope/lineage when + advanced runtime code needs them; ``item`` carries the inferred + collection item schema. + """ + return { + "type": "object", + "properties": { + "node_id": {"const": node_id}, + "activation_id": {"type": "string"}, + "frame_id": {"type": "string"}, + "scope_id": {"type": "string"}, + "lineage_id": {"type": "string"}, + "index": {"type": "integer"}, + "item": deepcopy(item_schema), + }, + "required": [ + "node_id", + "activation_id", + "frame_id", + "scope_id", + "lineage_id", + "index", + "item", + ], + "additionalProperties": False, + } + + +def structured_foreach_contract( + entry_schemas: Mapping[str, ContextSchema], +) -> ContextFieldContract: + """Return the ``foreach`` map contract for one static owner stack.""" + return ContextFieldContract( + FOREACH_CONTEXT_KEY, + { + "type": "object", + "properties": dict(entry_schemas), + "required": sorted(entry_schemas), + "additionalProperties": False, + }, + "Structured foreach context by static node id", + ) diff --git a/tests/core/test_context_scopes.py b/tests/core/test_context_scopes.py index 5eaa9c28..9df17090 100644 --- a/tests/core/test_context_scopes.py +++ b/tests/core/test_context_scopes.py @@ -260,12 +260,54 @@ def test_nested_foreach_replaces_inner_scope_and_restores_outer_scope() -> None: inner = _field_map(workflow, "inner_body") after_inner = _field_map(workflow, "after_inner") - assert "outer_item" not in inner - assert inner["inner_item"].availability == "available" + outer_item_schema = {"type": "string"} + inner_item_schema = {"type": "integer"} + + assert inner["outer_item"].schema == outer_item_schema + assert inner["inner_item"].schema == inner_item_schema + assert inner["loop_item"].schema == inner_item_schema + foreach_schema = inner["foreach"].schema + assert set(foreach_schema["properties"]) == {"outer", "inner"} + assert foreach_schema["properties"]["outer"]["properties"]["item"] == ( + outer_item_schema + ) + assert foreach_schema["properties"]["inner"]["properties"]["index"] == { + "type": "integer" + } assert after_inner["outer_item"].availability == "available" assert "inner_item" not in after_inner +def test_inner_completion_schema_restores_outer_structured_entry() -> None: + workflow = _workflow( + start="outer", + nodes=[ + _foreach("outer", alias="outer_item"), + _foreach("inner", alias="inner_item", over="state.inner_items"), + _node("inner_body"), + _node("after_inner"), + ], + edges=[ + {"from": "outer", "outcome": "loop", "to": "inner"}, + {"from": "inner", "outcome": "loop", "to": "inner_body"}, + {"from": "inner", "outcome": "done", "to": "after_inner"}, + {"from": "inner_body", "outcome": "ok", "to": "inner"}, + {"from": "after_inner", "outcome": "ok", "to": "outer"}, + {"from": "outer", "outcome": "done", "to": END}, + ], + state_schema={ + "type": "object", + "properties": { + "items": {"type": "array", "items": {"type": "string"}}, + "inner_items": {"type": "array", "items": {"type": "integer"}}, + }, + }, + ) + after_inner = _field_map(workflow, "after_inner") + foreach_schema = after_inner["foreach"].schema + assert set(foreach_schema["properties"]) == {"outer"} + + def test_nested_foreach_preserves_context_backed_item_schema() -> None: workflow = _workflow( start="outer", diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py index e26487cd..f5e82024 100644 --- a/tests/wf_api/test_authoring_contracts.py +++ b/tests/wf_api/test_authoring_contracts.py @@ -441,3 +441,68 @@ def test_project_inventory_does_not_offer_context_for_workflow_output() -> None: assert inventory["readable_sources"][0]["path"] == "context.item" assert inventory["readable_sources"][0]["uses"] == ["step_input"] + + +def test_structured_foreach_paths_appear_in_authoring_inventory() -> None: + from wf_api.authoring_contracts import context_path_options_for_node + from wf_core import END, Edge, ForeachNode, NodeUse, SchemaRef, Workflow + from wf_core.models.schemas import StateSchema + + def _foreach(node_id: str, *, over: str, alias: str) -> ForeachNode: + return ForeachNode.model_validate( + {"id": node_id, "type": "foreach", "over": over, "as": alias} + ) + + workflow = Workflow( + name="inventory_structured", + input_schema=SchemaRef(type="object"), + state_schema=StateSchema.model_validate( + { + "type": "object", + "properties": { + "customers": { + "type": "array", + "items": { + "type": "object", + "properties": {"name": {"type": "string"}}, + }, + }, + "orders": { + "type": "array", + "items": { + "type": "object", + "properties": {"sku": {"type": "string"}}, + }, + }, + }, + } + ), + output_schema=SchemaRef(type="object"), + start="customers", + nodes=[ + _foreach("customers", over="state.customers", alias="customer"), + _foreach("orders", over="state.orders", alias="order"), + NodeUse(id="inner_body", type="node", node="noop"), + ], + edges=[ + Edge.model_validate( + {"from": "customers", "outcome": "loop", "to": "orders"} + ), + Edge.model_validate( + {"from": "orders", "outcome": "loop", "to": "inner_body"} + ), + Edge.model_validate({"from": "inner_body", "outcome": "ok", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": "customers"}), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ], + ) + options = context_path_options_for_node(workflow, "inner_body") + paths = {option["path"] for option in options} + assert "context.foreach.customers.item" in paths + assert "context.foreach.customers.index" in paths + assert "context.foreach.orders.item" in paths + assert "context.foreach.orders.index" in paths + for option in options: + if option["path"].startswith("context.foreach."): + assert option["origin"] == "runtime_context" + assert option["uses"] == ["step_input"]