diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 193380eb..39c12e39 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -26,7 +26,6 @@ from wf_core.models.workflow import Edge, Workflow from wf_core.tokens import END type ContextAvailability = Literal["available", "conditional"] -type FrameScope = str | None _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH = 32 @@ -377,6 +376,7 @@ def _foreach_item_schema( _schema_document( workflow, foreach.over.root, + stack=controller_stack, foreach_nodes=foreach_nodes, owner_stack_by_node=owner_stack_by_node, ), @@ -424,7 +424,6 @@ def _schema_document( 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, ) -> Mapping[str, object]: @@ -433,13 +432,7 @@ 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,) + resolved_stack: ForeachOwnerStack = stack or () current: dict[str, object] = { field.name: field.schema for field in STANDARD_CONTEXT_FIELDS } diff --git a/src/wf_core/runtime/ops/flow.py b/src/wf_core/runtime/ops/flow.py index ef6b644e..920e7cfe 100644 --- a/src/wf_core/runtime/ops/flow.py +++ b/src/wf_core/runtime/ops/flow.py @@ -5,6 +5,7 @@ from typing import Any from wf_core.errors import WorkflowExecutionError from wf_core.models.workflow import Workflow from wf_core.run_state import ( + ROOT_FRAME_ID, ExecutionFrame, FrameStatus, RunState, @@ -12,6 +13,7 @@ from wf_core.run_state import ( StepExecutionResult, TraceEntry, ) +from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.schemas import validate_payload_against_schema from wf_core.runtime.ops.state import project_output from wf_core.runtime.scheduler import ( @@ -163,9 +165,6 @@ def finalize_run(workflow: Workflow, run: RunState) -> RunState: run.outcome = "ok" # Root workflow output keeps standard root facts consistent by projecting # against the root frame's derived context rather than an empty mapping. - from wf_core.run_state import ROOT_FRAME_ID - from wf_core.runtime.ops.frames import frame_context_view - root_frame = run.frames.get(ROOT_FRAME_ID) root_context: dict[str, Any] = ( dict(frame_context_view(run, root_frame).graph) diff --git a/src/wf_core/runtime/scheduler.py b/src/wf_core/runtime/scheduler.py index badd1df4..c0fe9787 100644 --- a/src/wf_core/runtime/scheduler.py +++ b/src/wf_core/runtime/scheduler.py @@ -74,7 +74,9 @@ class ForeachIterationMetadata: raise WorkflowExecutionError( f"malformed foreach activation id for frame {frame.id!r}" ) - if not isinstance(loop_index, int): + # `bool` is an `int` subclass; an index of True/False is corrupt + # persisted metadata, not item 1/0. + if not isinstance(loop_index, int) or isinstance(loop_index, bool): raise WorkflowExecutionError( f"malformed foreach loop index for frame {frame.id!r}" ) diff --git a/src/wf_core/validation/context_paths.py b/src/wf_core/validation/context_paths.py index ab37e3d4..e20b127d 100644 --- a/src/wf_core/validation/context_paths.py +++ b/src/wf_core/validation/context_paths.py @@ -41,6 +41,7 @@ def validate_context_paths( *, context_schemas: Mapping[str, ContextSchema], report: ValidationReport, + control_regions: ControlRegionAnalysis | None = None, ) -> None: """Validate every ``context.*`` path against its consuming location schema. @@ -49,10 +50,14 @@ def validate_context_paths( only if every literal segment is a declared object property in the consuming node's generated schema. The whole ``context`` object and the ``context.foreach`` map remain readable; unknown dynamic keys do not. + The shared control-region analysis is threaded through so validation runs + it once; alias ownership never triggers a second traversal. """ nodes_by_index = list(workflow.nodes) node_index_by_id = {node.id: idx for idx, node in enumerate(nodes_by_index)} - _validate_alias_ownership(workflow, node_index_by_id, report) + _validate_alias_ownership( + workflow, node_index_by_id, report, control_regions=control_regions + ) for idx, node in enumerate(nodes_by_index): schema = context_schemas.get(node.id) if isinstance(node, NodeUse): @@ -177,14 +182,50 @@ def _validate_one_context_path( "no context schema for this program location", ) return - if not _path_in_schema(schema, path.parts): + failing, available = _failing_segment(schema, path.parts) + if failing is not None: + listed = f" (available: {available})" if available else "" report.add( ValidationIssueCode.INVALID_CONTEXT_PATH, location, - f"invalid context path {str(path)!r} at {node_id or location!r}", + f"invalid context path {str(path)!r} at {node_id or location!r}: " + f"unknown segment {failing!r}{listed}", ) +def _failing_segment( + schema: Mapping[str, Any], parts: tuple[str, ...] +) -> tuple[str | None, str]: + """Return the first unknown segment plus the keys available there. + + Returns ``(None, "")`` when the path walks declared properties (or + permissive unconstrained schemas). Diagnostics only; validity follows + the same walk as :func:`_path_in_schema`. + """ + if not parts: + return None, "" + current: Any = schema + for part in parts: + if not isinstance(current, Mapping): + return part, "" + while isinstance(current.get("$ref"), str): + return part, "" + properties = current.get("properties") + if not isinstance(properties, Mapping): + if current == {}: + return None, "" + if ( + current.get("type") == "object" + and current.get("additionalProperties", True) is not False + ): + return None, "" + return part, "" + if part not in properties: + return part, ",".join(sorted(str(key) for key in properties)) + current = properties[part] + return None, "" + + def _path_in_schema(schema: Mapping[str, Any], parts: tuple[str, ...]) -> bool: """Return whether literal parts walk declared object properties. @@ -251,6 +292,8 @@ def _validate_alias_ownership( workflow: Workflow, node_index_by_id: dict[str, int], report: ValidationReport, + *, + control_regions: ControlRegionAnalysis | None = None, ) -> None: """Reject reserved or colliding active foreach aliases. @@ -258,11 +301,14 @@ def _validate_alias_ownership( ``loop_item``, and ``loop_index`` (that is, ``RESERVED_CONTEXT_KEYS``). Siblings in separate control regions may reuse an alias because they are never active together; only aliases active in the same owner stack - collide. Failures point at the inner foreach's ``as`` field. + collide. Failures point at the inner foreach's ``as`` field. The shared + control-region analysis is reused; this helper never traverses alone. """ - from wf_core.analysis.control_regions import analyze_control_regions + if control_regions is None: + from wf_core.analysis.control_regions import analyze_control_regions - analysis: ControlRegionAnalysis = analyze_control_regions(workflow) + control_regions = analyze_control_regions(workflow) + analysis = control_regions foreach_by_id = { node.id: node for node in workflow.nodes if isinstance(node, ForeachNode) } diff --git a/src/wf_core/validation/core.py b/src/wf_core/validation/core.py index 9b11f925..382c23e2 100644 --- a/src/wf_core/validation/core.py +++ b/src/wf_core/validation/core.py @@ -53,7 +53,12 @@ def validate_workflow(workflow: Workflow) -> ValidationReport: issue.message, ) context_schemas = context_schemas_by_node(workflow, control_regions=analysis) - validate_context_paths(workflow, context_schemas=context_schemas, report=report) + validate_context_paths( + workflow, + context_schemas=context_schemas, + report=report, + control_regions=analysis, + ) return report diff --git a/src/wf_core/validation/steps.py b/src/wf_core/validation/steps.py index 7e829fae..b47654c6 100644 --- a/src/wf_core/validation/steps.py +++ b/src/wf_core/validation/steps.py @@ -329,16 +329,17 @@ def validate_foreach_node( input_root_fields: set[str], workflow: Workflow, ) -> None: - # Interim permissive gate: context-rooted `over` paths reach runtime, where - # structured ancestry resolution handles them. Task 5 replaces this with - # location-aware validation against the consuming node's context schema. + # Context-rooted `over` paths pass this coarse gate and reach the + # location-aware `validate_context_paths` pass, which checks them against + # the consuming controller's structured context schema. if not is_valid_source_path( node.over, state_root_fields, input_root_fields, allow_context=True ): report.add( ValidationIssueCode.INVALID_FOREACH_SOURCE, f"nodes[{index}].over", - "foreach source path must start with input. or state. and reference a declared root field", + "foreach source path must start with input., state., or context. " + "and reference a declared root field when applicable", ) if node.item_error.action != "collect": return diff --git a/tests/core/test_canonical_node_bindings.py b/tests/core/test_canonical_node_bindings.py index 3202acf6..901c6ed0 100644 --- a/tests/core/test_canonical_node_bindings.py +++ b/tests/core/test_canonical_node_bindings.py @@ -11,7 +11,12 @@ from wf_core.models.steps import ( InterruptNode, NodeUse, ) -from wf_core.paths import GraphSourcePath, LocalPath, StatePath +from wf_core.paths import ( + GraphSourcePath, + LocalPath, + PathResolutionError, + StatePath, +) def test_node_use_accepts_canonical_input_and_output_bindings(): @@ -465,9 +470,5 @@ def test_foreach_ref_item_index_are_literal_structured_paths() -> None: 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: + with pytest.raises(PathResolutionError): GraphSourcePath.parse("output.result") - except Exception: - pass - else: - raise AssertionError("expected output root to be rejected") diff --git a/tests/core/test_structured_context_validation.py b/tests/core/test_structured_context_validation.py index 7e2b9086..2eaebb2c 100644 --- a/tests/core/test_structured_context_validation.py +++ b/tests/core/test_structured_context_validation.py @@ -399,3 +399,137 @@ def test_sibling_foreach_aliases_may_match_when_never_active_together() -> None: for issue in report.errors if issue.code == ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT ] + + +def test_child_workflow_cannot_address_caller_foreach_context() -> None: + """A child scope must receive caller values through declared input. + + The child is validated alone, so a path naming the caller's foreach id + is a missing id in the child scope and fails closed. + """ + from wf_core.validation import validate_workflow + + child = Workflow( + name="child", + input_schema=SchemaRef(type="object", properties={"order": {}}), + state_schema=StateSchema.from_field_map({}), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[_record_def()], + start="work", + nodes=[_node_use("work", path="context.foreach.orders.item")], + edges=[Edge.model_validate({"from": "work", "outcome": "ok", "to": END})], + ) + report = validate_workflow(child) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[0].input[0].path" + ) + assert issue is not None + assert "context.foreach.orders.item" in issue.message + + +def test_object_expression_and_nested_conditions_report_exact_paths() -> None: + from wf_core.models.steps import ConditionNode, InterruptNode + from wf_core.validation import validate_workflow + + bad = "context.foreach.missing.item" + workflow = _base_workflow() + workflow.nodes[2] = _node_use( + "work", + expression={ + "kind": "object", + "fields": {"order": {"kind": "path", "path": bad}}, + }, + ) + report = validate_workflow(workflow) + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].input[0].expression.fields.order.path", + ) + is not None + ) + + workflow = _base_workflow() + workflow.nodes[2] = ConditionNode.model_validate( + { + "id": "work", + "type": "condition", + "check": { + "op": "not", + "arg": { + "op": "and", + "args": [ + {"op": "exists", "path": bad}, + { + "op": "eq", + "left": {"path": bad}, + "right": {"value": 1}, + }, + ], + }, + }, + } + ) + workflow.edges = [ + Edge.model_validate({"from": "customers", "outcome": "loop", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + 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": "customers", "outcome": "done", "to": END}), + ] + report = validate_workflow(workflow) + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].check.arg.args[0].path", + ) + is not None + ) + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].check.arg.args[1].left.path", + ) + is not None + ) + + workflow = _base_workflow() + workflow.nodes[2] = InterruptNode.model_validate( + { + "id": "work", + "type": "interrupt", + "kind": "approval", + "request": [ + { + "target": "order", + "expression": {"kind": "path", "path": bad}, + } + ], + } + ) + workflow.edges = [ + Edge.model_validate({"from": "customers", "outcome": "loop", "to": "orders"}), + 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": "customers", "outcome": "done", "to": END}), + ] + report = validate_workflow(workflow) + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].request[0].expression.path", + ) + is not None + ) diff --git a/tests/core/test_structured_runtime_context.py b/tests/core/test_structured_runtime_context.py index 67efe03e..188395e2 100644 --- a/tests/core/test_structured_runtime_context.py +++ b/tests/core/test_structured_runtime_context.py @@ -683,6 +683,8 @@ def test_concurrent_items_receive_distinct_frame_and_lineage_context() -> None: run = execute_workflow(workflow, {"items": ["a", "b"]}, {"record": record}) assert run.status == RunStatus.COMPLETED assert len(contexts) == 2 + # Items admitted in one foreach visit share that visit's activation but + # own distinct item frames and lineages. assert contexts[0].activation_id == contexts[1].activation_id assert contexts[0].frame_id != contexts[1].frame_id assert contexts[0].lineage_id != contexts[1].lineage_id @@ -832,3 +834,92 @@ def test_interrupt_resume_recreates_structured_context_identities() -> None: 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 + + +def test_single_foreach_exposes_one_structured_entry() -> None: + from wf_core import ( + END, + Edge, + ForeachNode, + NodeDef, + NodeUse, + SchemaRef, + Workflow, + execute_workflow, + ) + from wf_core.models.schemas import StateField, StateSchema + + workflow = Workflow( + name="single_structured", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map({"items": 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="each", + nodes=[ + ForeachNode.model_validate( + { + "id": "each", + "type": "foreach", + "over": "state.items", + "as": "item", + "mode": "serial", + } + ), + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.item"}], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate({"from": "each", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "each"}), + Edge.model_validate({"from": "each", "outcome": "done", "to": END}), + ], + ) + seen: list[RuntimeContext] = [] + + def record(_payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + seen.append(ctx) + return {"outcome": "ok", "output": {}} + + run = execute_workflow(workflow, {"items": ["a"]}, {"record": record}) + assert run.status == RunStatus.COMPLETED + assert len(seen) == 1 + assert tuple(seen[0].foreach) == ("each",) + assert seen[0].foreach["each"].item == "a" + assert seen[0].foreach["each"].index == 0 + + +def test_bool_loop_index_metadata_fails_closed() -> None: + run = _run_with_frames( + [ + ExecutionFrame( + id="bad", + kind="foreach_iteration", + node_id="body", + scope_id="root", + metadata={ + "foreach_node_id": "each", + "activation_id": "act-1", + "loop_index": True, + "loop_item": "a", + "loop_alias": "item", + }, + ) + ] + ) + with pytest.raises(WorkflowExecutionError, match="malformed foreach loop index"): + frame_context_view(run, run.frames["bad"]) diff --git a/tests/core/test_subgraph_step.py b/tests/core/test_subgraph_step.py index dc6d4253..34481e2b 100644 --- a/tests/core/test_subgraph_step.py +++ b/tests/core/test_subgraph_step.py @@ -563,6 +563,7 @@ def test_subgraph_does_not_inherit_caller_foreach_context() -> None: assert isinstance(ctx, RuntimeContext) assert tuple(ctx.foreach) == ("orders",) assert ctx.foreach["orders"].item == "child-item" - assert ctx.foreach["orders"].scope_id != "root" + parent_scope_id = run.scopes["root"].id + assert ctx.foreach["orders"].scope_id != parent_scope_id assert pre_seen["foreach"] == {} assert pre_seen["input_order"] == {"sku": "A-17"}