From c389590a9ce8ba8ec01ddcef9eec0f3eaa8dee2e Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 22:40:07 +0700 Subject: [PATCH] docs: publish structured runtime context --- docs/README.md | 4 +- docs/current_roadmap.md | 2 +- .../2026-09-04-structured-runtime-context.md | 0 ...09-04-structured-runtime-context-design.md | 2 +- docs/wf_authoring_control_flow.md | 26 +++ skills/wf-python/SKILL.md | 23 +++ .../wf-python/references/python-lifecycle.md | 23 +++ src/wf_api/authoring_contracts.py | 14 +- src/wf_core/analysis/context_scopes.py | 19 +-- src/wf_core/validation/context_paths.py | 23 +-- src/wf_core/validation/core.py | 8 +- tests/authoring/test_builder.py | 10 +- tests/authoring/test_subgraph.py | 5 +- .../test_structured_context_validation.py | 60 ++++--- tests/core/test_structured_runtime_context.py | 155 ++++++++++++++++-- tests/core/test_subgraph_step.py | 141 ++++++++++++++++ tests/wf_api/test_authoring_contracts.py | 8 +- 17 files changed, 431 insertions(+), 92 deletions(-) rename docs/{ => historical}/superpowers/plans/2026-09-04-structured-runtime-context.md (100%) diff --git a/docs/README.md b/docs/README.md index bbe05873..36a6dd39 100644 --- a/docs/README.md +++ b/docs/README.md @@ -117,8 +117,8 @@ docs as the active references: - [`superpowers/specs/2026-09-04-foreach-back-edge-design.md`](superpowers/specs/2026-09-04-foreach-back-edge-design.md): approved canonical foreach body-return and validation semantics. - [`superpowers/specs/2026-09-04-structured-runtime-context-design.md`](superpowers/specs/2026-09-04-structured-runtime-context-design.md): - proposed same-scope nested foreach context, path, schema, and authoring-ref - semantics. + implemented same-scope nested foreach context, path, schema, and + authoring-ref semantics. - [`superpowers/specs/2026-09-04-run-step-budget-design.md`](superpowers/specs/2026-09-04-run-step-budget-design.md): proposed persisted run-wide protection against unbounded graph execution. - [`superpowers/specs/2026-05-24-native-subgraphs-design.md`](superpowers/specs/2026-05-24-native-subgraphs-design.md): diff --git a/docs/current_roadmap.md b/docs/current_roadmap.md index 86c7e18e..63290863 100644 --- a/docs/current_roadmap.md +++ b/docs/current_roadmap.md @@ -48,7 +48,7 @@ and barrier foundations in this order. The implementation plan is ready and its feature branch is under review: - [`structured runtime context design`](superpowers/specs/2026-09-04-structured-runtime-context-design.md) -- [`structured runtime context implementation plan`](superpowers/plans/2026-09-04-structured-runtime-context.md) +- [`structured runtime context implementation plan`](historical/superpowers/plans/2026-09-04-structured-runtime-context.md) This slice gives runtime code, expressions, validation, and authoring references one model for run data and same-scope foreach activations. Subgraphs continue to diff --git a/docs/superpowers/plans/2026-09-04-structured-runtime-context.md b/docs/historical/superpowers/plans/2026-09-04-structured-runtime-context.md similarity index 100% rename from docs/superpowers/plans/2026-09-04-structured-runtime-context.md rename to docs/historical/superpowers/plans/2026-09-04-structured-runtime-context.md diff --git a/docs/superpowers/specs/2026-09-04-structured-runtime-context-design.md b/docs/superpowers/specs/2026-09-04-structured-runtime-context-design.md index b4e25ce9..6ad72420 100644 --- a/docs/superpowers/specs/2026-09-04-structured-runtime-context-design.md +++ b/docs/superpowers/specs/2026-09-04-structured-runtime-context-design.md @@ -2,7 +2,7 @@ ## Status -Approved for implementation on 2026-09-04. This document specifies structured +Implemented on 2026-09-04. This document specifies structured foreach context inside one runtime scope. It complements the foreach back-edge design without expanding that implementation slice. diff --git a/docs/wf_authoring_control_flow.md b/docs/wf_authoring_control_flow.md index 355dc590..88e4fa90 100644 --- a/docs/wf_authoring_control_flow.md +++ b/docs/wf_authoring_control_flow.md @@ -248,6 +248,32 @@ See `examples/authoring_concurrent_foreach.py` for a runnable example covering: - the replace-conflict case when sibling item writes target a non-mergeable state path. +## Structured `foreach` context + +Nested bodies read every active same-scope iteration through structured paths. +Prefer declared input bindings via the foreach reference: + +```python +orders = graph.foreach( + id="orders", + over=state_path("orders"), + as_="order", +) +charge = graph.use( + charge_order, + input=[input_from(orders.item, "order")], +) +graph.set_route(orders, "loop", charge) +graph.set_route(charge, "ok", orders) +``` + +The compiled binding is ordinary protocol data (`context.foreach.orders.item`). +Normal capabilities receive foreach values through declared inputs. Advanced +handlers may inspect `ctx.foreach["orders"].index` and stable runtime +identities (`activation_id`, `frame_id`, `scope_id`, `lineage_id`). Child +workflows do not inherit caller context and must receive input; +`loop_item`, `loop_index`, and aliases are migration conveniences. + ## Deprecated `route` `route()` is a compatibility shim: diff --git a/skills/wf-python/SKILL.md b/skills/wf-python/SKILL.md index 97db0b6b..a9419841 100644 --- a/skills/wf-python/SKILL.md +++ b/skills/wf-python/SKILL.md @@ -226,5 +226,28 @@ except WorkflowClientError as error: - Use the public objects before inspecting `wf_api`, RPC clients, codecs, or stores. Drop below the client boundary only when implementing the client. +## Structured foreach context + +Prefer declared input bindings via the foreach reference: + +```python +orders = graph.foreach( + id="orders", + over=state_path("orders"), + as_="order", +) +charge = graph.use( + charge_order, + input=[input_from(orders.item, "order")], +) +graph.set_route(orders, "loop", charge) +graph.set_route(charge, "ok", orders) +``` + +Normal capabilities receive foreach values through declared inputs. Advanced +handlers may inspect `ctx.foreach["orders"].index` and stable runtime +identities. Child workflows do not inherit caller context and must receive +input. `loop_item`, `loop_index`, and aliases are migration conveniences. + Read [references/python-lifecycle.md](references/python-lifecycle.md) when a complete typed lifecycle or an editing/debugging recipe is needed. diff --git a/skills/wf-python/references/python-lifecycle.md b/skills/wf-python/references/python-lifecycle.md index f58abb13..d32ee8de 100644 --- a/skills/wf-python/references/python-lifecycle.md +++ b/skills/wf-python/references/python-lifecycle.md @@ -175,3 +175,26 @@ for frame in trace.frames: Catch the specific public errors useful to the application and retain a final `WorkflowClientError` fallback. Unknown server errors remain inspectable `ProtocolError` values with `code`, `message`, and `data`. + +## Structured foreach context + +Prefer declared input bindings via the foreach reference: + +```python +orders = graph.foreach( + id="orders", + over=state_path("orders"), + as_="order", +) +charge = graph.use( + charge_order, + input=[input_from(orders.item, "order")], +) +graph.set_route(orders, "loop", charge) +graph.set_route(charge, "ok", orders) +``` + +Normal capabilities receive foreach values through declared inputs. Advanced +handlers may inspect `ctx.foreach["orders"].index` and stable runtime +identities. Child workflows do not inherit caller context and must receive +input. `loop_item`, `loop_index`, and aliases are migration conveniences. diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py index 0fddf695..4694808b 100644 --- a/src/wf_api/authoring_contracts.py +++ b/src/wf_api/authoring_contracts.py @@ -275,9 +275,7 @@ def _nested_foreach_path_options( # bounded object schema, mirroring input/state inventory behavior. if prop_name == "item": options.extend( - _nested_item_subpaths( - prop_schema, owner_id, availability, depth=0 - ) + _nested_item_subpaths(prop_schema, owner_id, availability, depth=0) ) return options @@ -302,7 +300,10 @@ def _nested_item_subpaths( 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": + 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( @@ -339,7 +340,10 @@ def _nested_item_subpaths( ) options.extend( _nested_item_subpaths( - sub_schema, owner_id, availability, depth=depth + 1, + sub_schema, + owner_id, + availability, + depth=depth + 1, prefix_parts=(*prefix_parts, name), ) ) diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index fe09ffdb..193380eb 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -296,9 +296,7 @@ def _context_schema_for_stack( continue entry_schemas[owner_id] = foreach_entry_schema( owner_id, - _foreach_item_schema( - workflow, foreach, foreach_nodes, owner_stack_by_node - ), + _foreach_item_schema(workflow, foreach, foreach_nodes, owner_stack_by_node), ) properties[FOREACH_CONTEXT_KEY] = { "type": "object", @@ -306,9 +304,9 @@ def _context_schema_for_stack( "required": sorted(entry_schemas), "additionalProperties": False, } - required: list[str] = [ - field.name for field in STANDARD_CONTEXT_FIELDS - ] + [FOREACH_CONTEXT_KEY] + 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: @@ -445,10 +443,7 @@ def _schema_document( current: dict[str, object] = { field.name: field.schema for field in STANDARD_CONTEXT_FIELDS } - if ( - foreach_nodes is not None - and owner_stack_by_node is not None - ): + if foreach_nodes is not None and owner_stack_by_node is not None: entry_schemas: dict[str, object] = {} for owner_id in resolved_stack: foreach = foreach_nodes.get(owner_id) @@ -480,7 +475,9 @@ def _schema_document( 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] + 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} diff --git a/src/wf_core/validation/context_paths.py b/src/wf_core/validation/context_paths.py index 2d7a8892..ab37e3d4 100644 --- a/src/wf_core/validation/context_paths.py +++ b/src/wf_core/validation/context_paths.py @@ -64,12 +64,8 @@ def validate_context_paths( node.input, f"nodes[{idx}].input", node.id, schema, report ) elif isinstance(node, ConditionNode): - for location, path in _condition_paths( - node.check, f"nodes[{idx}].check" - ): - _validate_one_context_path( - path, location, node.id, schema, report - ) + for location, path in _condition_paths(node.check, f"nodes[{idx}].check"): + _validate_one_context_path(path, location, node.id, schema, report) elif isinstance(node, ForeachNode): # Context-rooted `over` paths reach this pass; the old # input/state-only check stays permissive for them. @@ -110,9 +106,7 @@ def _validate_step_input_bindings( binding.expression, f"{binding_location}.expression" ): if path.root == "context": - _validate_one_context_path( - path, location, node_id, schema, report - ) + _validate_one_context_path(path, location, node_id, schema, report) def _expression_paths( @@ -219,9 +213,10 @@ def _path_in_schema(schema: Mapping[str, Any], parts: tuple[str, ...]) -> bool: # allows subpaths; scalar or closed schemas do not. if current == {}: return True - if current.get("type") == "object" and current.get( - "additionalProperties", True - ) is not False: + if ( + current.get("type") == "object" + and current.get("additionalProperties", True) is not False + ): return True return False if part not in properties: @@ -280,9 +275,7 @@ def _validate_alias_ownership( continue alias = foreach.as_ idx = node_index_by_id.get(owner_id) - location = ( - f"nodes[{idx}].as" if idx is not None else f"nodes[{owner_id}]" - ) + location = f"nodes[{idx}].as" if idx is not None else f"nodes[{owner_id}]" if not alias or alias in RESERVED_CONTEXT_KEYS: if owner_id not in reported: reported.add(owner_id) diff --git a/src/wf_core/validation/core.py b/src/wf_core/validation/core.py index 9685bad7..9b11f925 100644 --- a/src/wf_core/validation/core.py +++ b/src/wf_core/validation/core.py @@ -52,12 +52,8 @@ def validate_workflow(workflow: Workflow) -> ValidationReport: issue.path, issue.message, ) - context_schemas = context_schemas_by_node( - workflow, control_regions=analysis - ) - validate_context_paths( - workflow, context_schemas=context_schemas, report=report - ) + context_schemas = context_schemas_by_node(workflow, control_regions=analysis) + validate_context_paths(workflow, context_schemas=context_schemas, report=report) return report diff --git a/tests/authoring/test_builder.py b/tests/authoring/test_builder.py index 95bc33f8..c7772f1f 100644 --- a/tests/authoring/test_builder.py +++ b/tests/authoring/test_builder.py @@ -801,13 +801,9 @@ def test_foreach_reference_treats_dotted_id_as_one_literal_segment() -> None: state_schema={"type": "object"}, output_schema={"type": "object"}, ) - each = builder.foreach( - id="orders.v2", over=state_path("orders"), as_="order" - ) + each = builder.foreach(id="orders.v2", over=state_path("orders"), as_="order") - assert each.item == GraphSourcePath( - "context", ("foreach", "orders.v2", "item") - ) + 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' @@ -840,5 +836,5 @@ def test_foreach_ref_works_in_node_input_binding() -> None: input=[input_from(each.item, "order")], ) binding = work.input[0] - assert isinstance(binding, object) + assert isinstance(binding, InputPathBinding) assert str(binding.path) == "context.foreach.orders.item" diff --git a/tests/authoring/test_subgraph.py b/tests/authoring/test_subgraph.py index 96901517..472233ee 100644 --- a/tests/authoring/test_subgraph.py +++ b/tests/authoring/test_subgraph.py @@ -277,6 +277,7 @@ def test_workflow_builder_subgraph_adds_native_subgraph_node() -> None: def test_foreach_ref_works_in_subgraph_input_binding() -> None: from wf_authoring import state_path + from wf_core.models.steps import InputPathBinding child = build_demo_workflow() parent = WorkflowBuilder( @@ -291,4 +292,6 @@ def test_foreach_ref_works_in_subgraph_input_binding() -> None: id="run_child", input=[input_from(each.item, "order")], ) - assert str(step.input[0].path) == "context.foreach.orders.item" + binding = step.input[0] + assert isinstance(binding, InputPathBinding) + assert str(binding.path) == "context.foreach.orders.item" diff --git a/tests/core/test_structured_context_validation.py b/tests/core/test_structured_context_validation.py index 5769f3af..7e2b9086 100644 --- a/tests/core/test_structured_context_validation.py +++ b/tests/core/test_structured_context_validation.py @@ -86,11 +86,14 @@ def test_active_structured_foreach_item_path_is_valid() -> None: workflow = _base_workflow(work_path="context.foreach.orders.item") report = validate_workflow(workflow) - assert _issue( - report, - ValidationIssueCode.INVALID_CONTEXT_PATH, - "nodes[2].input[0].path", - ) is None + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].input[0].path", + ) + is None + ) def test_nested_body_can_read_outer_and_inner_entries() -> None: @@ -104,18 +107,21 @@ def test_nested_body_can_read_outer_and_inner_entries() -> None: ): workflow = _base_workflow(work_path=path) report = validate_workflow(workflow) - assert _issue( - report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path" - ) is None, path + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].input[0].path", + ) + is None + ), path def test_inactive_foreach_entry_is_rejected() -> None: from wf_core.validation import validate_workflow workflow = _base_workflow() - workflow.nodes[3] = _node_use( - "after_inner", path="context.foreach.orders.item" - ) + workflow.nodes[3] = _node_use("after_inner", path="context.foreach.orders.item") report = validate_workflow(workflow) issue = _issue( report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[3].input[0].path" @@ -172,9 +178,7 @@ def test_workflow_output_cannot_read_completed_foreach_entry() -> None: ) ] report = validate_workflow(workflow) - issue = _issue( - report, ValidationIssueCode.INVALID_CONTEXT_PATH, "output[0].path" - ) + issue = _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "output[0].path") assert issue is not None @@ -197,9 +201,7 @@ def test_workflow_output_cannot_read_completed_foreach_entry() -> None: "work", expression={ "kind": "array", - "items": [ - {"kind": "path", "path": "context.foreach.missing.item"} - ], + "items": [{"kind": "path", "path": "context.foreach.missing.item"}], }, ), "nodes[2].input[0].expression.items[0].path", @@ -235,7 +237,9 @@ def test_all_model_surfaces_reject_missing_foreach_id() -> None: ) report = validate_workflow(workflow) assert ( - _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path") + _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path" + ) is not None ) @@ -250,14 +254,14 @@ def test_all_model_surfaces_reject_missing_foreach_id() -> None: Edge.model_validate({"from": "work", "outcome": "true", "to": "orders"}), Edge.model_validate({"from": "work", "outcome": "false", "to": "orders"}), Edge.model_validate({"from": "orders", "outcome": "done", "to": "after_inner"}), - Edge.model_validate({"from": "after_inner", "outcome": "ok", "to": "customers"}), + Edge.model_validate( + {"from": "after_inner", "outcome": "ok", "to": "customers"} + ), Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), ] report = validate_workflow(workflow) assert ( - _issue( - report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].check.path" - ) + _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].check.path") is not None ) @@ -289,7 +293,9 @@ def test_all_model_surfaces_reject_missing_foreach_id() -> None: Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), Edge.model_validate({"from": "work", "outcome": "submitted", "to": "orders"}), Edge.model_validate({"from": "orders", "outcome": "done", "to": "after_inner"}), - Edge.model_validate({"from": "after_inner", "outcome": "ok", "to": "customers"}), + Edge.model_validate( + {"from": "after_inner", "outcome": "ok", "to": "customers"} + ), Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), ] report = validate_workflow(workflow) @@ -306,9 +312,7 @@ def test_all_model_surfaces_reject_missing_foreach_id() -> None: from wf_core.models.steps import InputPathBinding as _IPB workflow = _base_workflow() - workflow.output = [ - _IPB.model_validate({"target": "result", "path": bad}) - ] + workflow.output = [_IPB.model_validate({"target": "result", "path": bad})] report = validate_workflow(workflow) assert ( _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "output[0].path") @@ -381,7 +385,9 @@ def test_sibling_foreach_aliases_may_match_when_never_active_together() -> None: Edge.model_validate({"from": "left", "outcome": "loop", "to": "left_body"}), Edge.model_validate({"from": "left_body", "outcome": "ok", "to": "left"}), Edge.model_validate({"from": "left", "outcome": "done", "to": "right"}), - Edge.model_validate({"from": "right", "outcome": "loop", "to": "right_body"}), + Edge.model_validate( + {"from": "right", "outcome": "loop", "to": "right_body"} + ), Edge.model_validate({"from": "right_body", "outcome": "ok", "to": "right"}), Edge.model_validate({"from": "right", "outcome": "done", "to": "join"}), Edge.model_validate({"from": "join", "outcome": "ok", "to": END}), diff --git a/tests/core/test_structured_runtime_context.py b/tests/core/test_structured_runtime_context.py index e4c0dd07..67efe03e 100644 --- a/tests/core/test_structured_runtime_context.py +++ b/tests/core/test_structured_runtime_context.py @@ -68,7 +68,9 @@ def test_root_frame_has_empty_structured_foreach_context() -> None: def test_nested_same_scope_frames_expose_outermost_to_innermost_context() -> None: run = _run_with_frames( [ - ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + ExecutionFrame( + id="root", kind="root", node_id="customers", scope_id="root" + ), _item_frame( frame_id="outer-item", parent_id="root", @@ -119,7 +121,9 @@ def test_nested_same_scope_frames_expose_outermost_to_innermost_context() -> Non def test_graph_context_values_keep_all_aliases_and_innermost_loop_keys() -> None: run = _run_with_frames( [ - ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + ExecutionFrame( + id="root", kind="root", node_id="customers", scope_id="root" + ), _item_frame( frame_id="outer-item", parent_id="root", @@ -158,7 +162,9 @@ def test_graph_context_values_keep_all_aliases_and_innermost_loop_keys() -> None def test_context_ancestry_stops_at_runtime_scope_boundary() -> None: run = _run_with_frames( [ - ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + ExecutionFrame( + id="root", kind="root", node_id="customers", scope_id="root" + ), _item_frame( frame_id="outer-item", parent_id="root", @@ -290,7 +296,9 @@ def test_structured_context_rejects_duplicate_active_alias() -> None: def test_context_read_does_not_mutate_run_state() -> None: run = _run_with_frames( [ - ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + ExecutionFrame( + id="root", kind="root", node_id="customers", scope_id="root" + ), _item_frame( frame_id="outer-item", parent_id="root", @@ -369,7 +377,9 @@ def _nested_workflow(*, inner_over: str = "state.orders_list"): ), Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), - Edge.model_validate({"from": "orders", "outcome": "done", "to": "customers"}), + Edge.model_validate( + {"from": "orders", "outcome": "done", "to": "customers"} + ), Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), ], ) @@ -398,9 +408,7 @@ def test_nested_handler_receives_outer_and_inner_typed_entries() -> None: {"record": record}, ) assert run.status == RunStatus.COMPLETED - assert seen == [ - (("customers", "orders"), {"name": "Ada"}, {"sku": "A-17"}, 0) - ] + assert seen == [(("customers", "orders"), {"name": "Ada"}, {"sku": "A-17"}, 0)] def test_nested_graph_bindings_resolve_outer_and_inner_items() -> None: @@ -672,9 +680,7 @@ def test_concurrent_items_receive_distinct_frame_and_lineage_context() -> None: contexts.append(ctx.foreach["each"]) return {"outcome": "ok", "output": {}} - run = execute_workflow( - workflow, {"items": ["a", "b"]}, {"record": record} - ) + run = execute_workflow(workflow, {"items": ["a", "b"]}, {"record": record}) assert run.status == RunStatus.COMPLETED assert len(contexts) == 2 assert contexts[0].activation_id == contexts[1].activation_id @@ -686,9 +692,7 @@ def test_concurrent_items_receive_distinct_frame_and_lineage_context() -> None: def test_nested_foreach_over_resolves_structured_outer_item_path() -> None: from wf_core import execute_workflow - workflow = _nested_workflow( - inner_over="context.foreach.customers.item.orders" - ) + workflow = _nested_workflow(inner_over="context.foreach.customers.item.orders") captured: list[object] = [] def record(payload: dict[str, object], _ctx: RuntimeContext) -> dict[str, object]: @@ -705,3 +709,126 @@ def test_nested_foreach_over_resolves_structured_outer_item_path() -> None: ) assert run.status == RunStatus.COMPLETED assert captured == [{"sku": "A-17"}] + + +def test_interrupt_resume_recreates_structured_context_identities() -> None: + from wf_core import ( + END, + Edge, + ForeachNode, + InterruptNode, + NodeDef, + NodeUse, + SchemaRef, + Workflow, + execute_workflow, + resume_workflow, + ) + from wf_core.models.schemas import StateField, StateSchema + from wf_core.run_codec import dump_run_state, load_run_state + + workflow = Workflow( + name="nested_interrupt_resume", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "customers": StateField(type="array"), + "orders_list": StateField(type="array"), + } + ), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[ + NodeDef( + name="record", + input_schema=SchemaRef(type="object", properties={"value": {}}), + output_schema=SchemaRef(type="object", properties={}), + outcomes=["ok"], + ) + ], + start="outer", + nodes=[ + ForeachNode.model_validate( + { + "id": "outer", + "type": "foreach", + "over": "state.customers", + "as": "customer", + "mode": "serial", + } + ), + ForeachNode.model_validate( + { + "id": "inner", + "type": "foreach", + "over": "state.orders_list", + "as": "order", + "mode": "serial", + } + ), + NodeUse.model_validate( + { + "id": "pre", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.order"}], + "output": [], + } + ), + InterruptNode.model_validate( + {"id": "ask", "type": "interrupt", "kind": "approval", "request": []} + ), + NodeUse.model_validate( + { + "id": "post", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.order"}], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate({"from": "outer", "outcome": "loop", "to": "inner"}), + Edge.model_validate({"from": "inner", "outcome": "loop", "to": "pre"}), + Edge.model_validate({"from": "pre", "outcome": "ok", "to": "ask"}), + Edge.model_validate({"from": "ask", "outcome": "submitted", "to": "post"}), + Edge.model_validate({"from": "post", "outcome": "ok", "to": "inner"}), + Edge.model_validate({"from": "inner", "outcome": "done", "to": "outer"}), + Edge.model_validate({"from": "outer", "outcome": "done", "to": END}), + ], + ) + before: list[dict[str, ForeachContext]] = [] + after: list[dict[str, ForeachContext]] = [] + phase = {"value": "before"} + + def record(_payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + snapshot = dict(ctx.foreach) + if phase["value"] == "before": + before.append(snapshot) + else: + after.append(snapshot) + return {"outcome": "ok", "output": {}} + + run = execute_workflow( + workflow, + {"customers": [{"name": "Ada"}], "orders_list": [{"sku": "A-17"}]}, + {"record": record}, + ) + assert run.status == RunStatus.INTERRUPTED + assert len(before) == 1 + # Genuinely test reconstruction: dump, reload, resume from loaded state. + payload = dump_run_state(run) + reloaded = load_run_state(payload) + phase["value"] = "after" + resumed = resume_workflow(workflow, reloaded, {"record": record}, resume_payload={}) + assert resumed.status == RunStatus.COMPLETED + assert len(after) == 1 + before_outer = before[0]["outer"] + after_outer = after[0]["outer"] + before_inner = before[0]["inner"] + after_inner = after[0]["inner"] + assert after_outer.activation_id == before_outer.activation_id + assert after_inner.activation_id == before_inner.activation_id + assert after_inner.frame_id == before_inner.frame_id + assert after_inner.lineage_id == before_inner.lineage_id + assert after_inner.item == before_inner.item diff --git a/tests/core/test_subgraph_step.py b/tests/core/test_subgraph_step.py index 5d7955d2..dc6d4253 100644 --- a/tests/core/test_subgraph_step.py +++ b/tests/core/test_subgraph_step.py @@ -11,6 +11,7 @@ from wf_core import ( NodeUse, PreparedSubgraph, RunState, + RunStatus, SchemaRef, StateField, StateSchema, @@ -425,3 +426,143 @@ def _interrupting_child_workflow() -> Workflow: def _schema(properties: dict[str, object]) -> SchemaRef: return SchemaRef.model_validate({"type": "object", "properties": properties}) + + +def test_subgraph_does_not_inherit_caller_foreach_context() -> None: + from wf_core import ForeachNode, RuntimeContext + + child_seen: dict[str, object] = {} + pre_seen: dict[str, object] = {} + + def probe(payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + if ctx.current_node_id == "pre": + pre_seen["foreach"] = dict(ctx.foreach) + pre_seen["input_order"] = payload.get("order") + else: + child_seen["input_order"] = payload.get("order") + child_seen["context"] = ctx + return {"outcome": "ok", "output": {}} + + child = Workflow( + name="child.workflow", + input_schema=_schema({"order": {"type": "object"}}), + state_schema=StateSchema.from_field_map( + { + "order": StateField(type="object"), + "child_items": StateField(type="array", default=["child-item"]), + } + ), + output_schema=_schema({}), + node_defs=[ + NodeDef( + name="probe", + input_schema=_schema({"order": {}}), + output_schema=_schema({}), + outcomes=["ok"], + ) + ], + start="pre", + nodes=[ + NodeUse.model_validate( + { + "id": "pre", + "type": "node", + "node": "probe", + "input": [{"target": "order", "path": "input.order"}], + } + ), + ForeachNode.model_validate( + { + "id": "orders", + "type": "foreach", + "over": "state.child_items", + "as": "c_order", + "mode": "serial", + } + ), + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "probe", + "input": [{"target": "order", "path": "input.order"}], + } + ), + ], + edges=[ + Edge.model_validate({"from": "pre", "outcome": "ok", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": END}), + ], + ) + parent = Workflow( + name="parent_orders", + input_schema=_schema({}), + state_schema=StateSchema.from_field_map( + {"parent_orders": StateField(type="array")} + ), + output_schema=_schema({}), + node_defs=[ + NodeDef( + name="noop", + input_schema=_schema({}), + output_schema=_schema({}), + outcomes=["ok"], + ) + ], + start="orders", + nodes=[ + ForeachNode.model_validate( + { + "id": "orders", + "type": "foreach", + "over": "state.parent_orders", + "as": "p_order", + "mode": "serial", + } + ), + SubgraphNode.model_validate( + { + "id": "run_child", + "type": "subgraph", + "workflow": "child.workflow", + "input_schema": _schema({"order": {"type": "object"}}), + "output_schema": _schema({}), + "input": [ + { + "target": "order", + "path": "context.foreach.orders.item", + } + ], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate( + {"from": "orders", "outcome": "loop", "to": "run_child"} + ), + Edge.model_validate({"from": "run_child", "outcome": "ok", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": END}), + ], + ) + run = execute_workflow( + parent, + {"parent_orders": [{"sku": "A-17"}]}, + {}, + subgraphs={ + "child.workflow": PreparedSubgraph( + workflow=child, registry={"probe": probe} + ) + }, + ) + assert run.status == RunStatus.COMPLETED + assert child_seen["input_order"] == {"sku": "A-17"} + ctx = child_seen["context"] + assert isinstance(ctx, RuntimeContext) + assert tuple(ctx.foreach) == ("orders",) + assert ctx.foreach["orders"].item == "child-item" + assert ctx.foreach["orders"].scope_id != "root" + assert pre_seen["foreach"] == {} + assert pre_seen["input_order"] == {"sku": "A-17"} diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py index f5e82024..9ad23bf7 100644 --- a/tests/wf_api/test_authoring_contracts.py +++ b/tests/wf_api/test_authoring_contracts.py @@ -491,8 +491,12 @@ def test_structured_foreach_paths_appear_in_authoring_inventory() -> None: 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": "inner_body", "outcome": "ok", "to": "orders"} + ), + Edge.model_validate( + {"from": "orders", "outcome": "done", "to": "customers"} + ), Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), ], )