diff --git a/src/wf_core/models/input_bindings.py b/src/wf_core/models/input_bindings.py index 8fee178c..744b5fbe 100644 --- a/src/wf_core/models/input_bindings.py +++ b/src/wf_core/models/input_bindings.py @@ -205,6 +205,15 @@ def validate_input_expression_limits( visit_expression(value, depth=1, location="expression") +def _check_expression_limits(value: object) -> object: + """Share the budget validator across graph and schedule bindings.""" + raw_value = ( + value.model_dump(mode="python") if isinstance(value, BaseModel) else value + ) + validate_input_expression_limits(raw_value) + return value + + class InputExpressionBinding(BaseModel): """Assign one recursively composed expression to a node-local target.""" @@ -216,11 +225,7 @@ class InputExpressionBinding(BaseModel): @field_validator("expression", mode="before") @classmethod def check_limits(cls, value: object) -> object: - raw_value = ( - value.model_dump(mode="python") if isinstance(value, BaseModel) else value - ) - validate_input_expression_limits(raw_value) - return value + return _check_expression_limits(value) InputBinding = Annotated[ @@ -262,11 +267,7 @@ class ScheduleInputBinding(BaseModel): @field_validator("expression", mode="before") @classmethod def check_limits(cls, value: object) -> object: - raw_value = ( - value.model_dump(mode="python") if isinstance(value, BaseModel) else value - ) - validate_input_expression_limits(raw_value) - return value + return _check_expression_limits(value) for _model in ( diff --git a/src/wf_core/runtime/input_bindings.py b/src/wf_core/runtime/input_bindings.py index 02c46c2d..f55b200b 100644 --- a/src/wf_core/runtime/input_bindings.py +++ b/src/wf_core/runtime/input_bindings.py @@ -3,7 +3,6 @@ from __future__ import annotations from collections.abc import Mapping, Sequence from typing import Any -from wf_core.conditions import safe_resolve_path from wf_core.errors import WorkflowExecutionError from wf_core.local_paths import LocalPathError, set_local_value from wf_core.models.input_bindings import ( @@ -14,7 +13,11 @@ from wf_core.models.input_bindings import ( StepInputBinding, ) from wf_core.models.json_values import JsonValue -from wf_core.runtime.input_sources import SourceResolver +from wf_core.runtime.input_sources import ( + GraphSourceResolver, + SourceResolver, + resolve_composed_expression, +) def resolve_input_expression( @@ -27,11 +30,6 @@ def resolve_input_expression( location: str, ) -> JsonValue: """Resolve one composite expression while preserving its payload location.""" - from wf_core.runtime.input_sources import ( - GraphSourceResolver, - resolve_composed_expression, - ) - resolver = GraphSourceResolver( state=state, workflow_input=workflow_input, context=context ) @@ -48,13 +46,44 @@ def resolve_input_expression_with_resolver( location: str, ) -> JsonValue: """Resolve one expression through an explicit typed source resolver.""" - from wf_core.runtime.input_sources import resolve_composed_expression - return resolve_composed_expression( expression, resolver=resolver, label=label, location=location ) +def resolve_step_input_bindings_with_resolver( + bindings: Sequence[StepInputBinding], + *, + resolver: SourceResolver, + label: str, +) -> dict[str, Any]: + """Build one node-local payload through an explicit graph resolver.""" + payload: dict[str, Any] = {} + for binding in bindings: + location = str(binding.target) + if isinstance(binding, InputValueBinding): + value = binding.value + elif isinstance(binding, InputPathBinding): + try: + value = resolver.resolve_path(binding.path) + except (ValueError, WorkflowExecutionError) as exc: + raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc + elif isinstance(binding, InputExpressionBinding): + value = resolve_input_expression_with_resolver( + binding.expression, + resolver=resolver, + label=label, + location=location, + ) + else: + raise WorkflowExecutionError(f"unsupported input binding for {label}") + try: + set_local_value(payload, binding.target, value) + except LocalPathError as exc: + raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc + return payload + + def resolve_step_input_bindings( bindings: Sequence[StepInputBinding], *, @@ -64,35 +93,9 @@ def resolve_step_input_bindings( label: str, ) -> dict[str, Any]: """Build one node-local payload from simple or composite input bindings.""" - - payload: dict[str, Any] = {} - for binding in bindings: - location = str(binding.target) - if isinstance(binding, InputValueBinding): - value = binding.value - elif isinstance(binding, InputPathBinding): - try: - value = safe_resolve_path( - str(binding.path), - state=state, - workflow_input=workflow_input, - context=context, - ) - except (ValueError, WorkflowExecutionError) as exc: - raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc - elif isinstance(binding, InputExpressionBinding): - value = resolve_input_expression( - binding.expression, - state=state, - workflow_input=workflow_input, - context=context, - label=label, - location=location, - ) - else: - raise WorkflowExecutionError(f"unsupported input binding for {label}") - try: - set_local_value(payload, binding.target, value) - except LocalPathError as exc: - raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc - return payload + resolver = GraphSourceResolver( + state=state, workflow_input=workflow_input, context=context + ) + return resolve_step_input_bindings_with_resolver( + bindings, resolver=resolver, label=label + ) diff --git a/src/wf_core/runtime/input_sources.py b/src/wf_core/runtime/input_sources.py index e2ed1f9b..cdde45cf 100644 --- a/src/wf_core/runtime/input_sources.py +++ b/src/wf_core/runtime/input_sources.py @@ -1,11 +1,17 @@ """Typed source-resolver seam for composed input expressions. -Graph and schedule paths share one composition traversal (literal / array / -object recursion plus budget enforcement) but keep distinct source models: -graph leaves resolve ``input`` / ``state`` / ``context`` paths, while -schedule leaves resolve typed occurrence fields. Do not extend +Graph and schedule paths share one composition traversal (``_resolve_any``: +literal/array/object recursion plus budget enforcement via +``validate_input_expression_limits`` at model validation) but keep distinct +source models: graph leaves resolve ``input`` / ``state`` / ``context`` +paths, while schedule leaves resolve typed occurrence fields. Do not extend ``GraphSourcePath`` with schedule-only roots and do not smuggle occurrence values through a faked graph ``context`` mapping. + +``wf_api.input_expressions.validate_and_project_input_expression`` keeps its +own schema-aware recursion (it carries target fragments/locations, not +runtime values); budget parity there comes from the same model-level +``check_limits`` validator, not from reusing this value traversal. """ from __future__ import annotations @@ -81,37 +87,62 @@ class MappingSourceResolver: return self._values[field] -def resolve_composed_expression( - expression: InputExpression, +def _resolve_any( + expression: InputExpression | ScheduleExpression, *, - resolver: SourceResolver, + graph_resolver: SourceResolver | None, + occurrence_resolver: MappingSourceResolver | None, label: str, location: str, ) -> JsonValue: - """Resolve one composite expression through a typed source resolver.""" + """Resolve one leaf or branch through the shared composition traversal. + + Literal/array/object recursion lives here once for both graph and + schedule expressions (same module, same budget validator, same location + scheme). Graph leaves need ``graph_resolver``; schedule leaves need + ``occurrence_resolver``. Distinct static types keep schedule-only + occurrence references out of graph contracts. + """ if isinstance(expression, LiteralExpression): return expression.value if isinstance(expression, PathExpression): + if graph_resolver is None: + raise WorkflowExecutionError( + f"{label} {location}: graph paths are invalid in schedule bindings" + ) try: - value = resolver.resolve_path(expression.path) + value = graph_resolver.resolve_path(expression.path) return validate_strict_json_value(value) except (ValueError, WorkflowExecutionError) as exc: raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc - if isinstance(expression, ArrayExpression): + if isinstance(expression, OccurrenceExpression): + if occurrence_resolver is None: + raise WorkflowExecutionError( + f"{label} {location}: occurrence references are invalid in graph bindings" + ) + try: + return validate_strict_json_value( + occurrence_resolver.resolve_field(expression.field) + ) + except (ValueError, WorkflowExecutionError) as exc: + raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc + if isinstance(expression, (ArrayExpression, ScheduleArrayExpression)): return [ - resolve_composed_expression( + _resolve_any( item, - resolver=resolver, + graph_resolver=graph_resolver, + occurrence_resolver=occurrence_resolver, label=label, location=f"{location}[{index}]", ) for index, item in enumerate(expression.items) ] - if isinstance(expression, ObjectExpression): + if isinstance(expression, (ObjectExpression, ScheduleObjectExpression)): return { - field: resolve_composed_expression( + field: _resolve_any( value, - resolver=resolver, + graph_resolver=graph_resolver, + occurrence_resolver=occurrence_resolver, label=label, location=f"{location}.{field}", ) @@ -120,71 +151,73 @@ def resolve_composed_expression( raise WorkflowExecutionError(f"unsupported input expression for {label} {location}") +def resolve_composed_expression( + expression: InputExpression, + *, + resolver: SourceResolver, + label: str, + location: str, +) -> JsonValue: + """Resolve one composite expression through a typed source resolver.""" + return _resolve_any( + expression, + graph_resolver=resolver, + occurrence_resolver=None, + label=label, + location=location, + ) + + def resolve_schedule_expression( expression: ScheduleExpression, *, - occurrence: Mapping[str, Any], + occurrence: Mapping[str, Any] | MappingSourceResolver, label: str, location: str, ) -> JsonValue: """Resolve one schedule-side expression against the occurrence environment. - Composition (literal/array/object) is shared with the graph traversal - above; leaves are typed occurrence fields only. Graph paths are invalid - here and never reach ``GraphSourcePath``. + Accepts a raw occurrence mapping or a ``MappingSourceResolver``; a raw + mapping is wrapped so schedule composition reuses the same traversal + without faking a graph ``context``. Graph paths are invalid here and + never reach ``GraphSourcePath``. """ - if isinstance(expression, LiteralExpression): - return expression.value - if isinstance(expression, OccurrenceExpression): - try: - if expression.field not in occurrence: - raise WorkflowExecutionError( - f"unknown occurrence field {expression.field!r}" - ) - return validate_strict_json_value(occurrence[expression.field]) - except (ValueError, WorkflowExecutionError) as exc: - raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc - if isinstance(expression, ScheduleArrayExpression): - return [ - resolve_schedule_expression( - item, - occurrence=occurrence, - label=label, - location=f"{location}[{index}]", - ) - for index, item in enumerate(expression.items) - ] - if isinstance(expression, ScheduleObjectExpression): - return { - field: resolve_schedule_expression( - value, - occurrence=occurrence, - label=label, - location=f"{location}.{field}", - ) - for field, value in expression.fields.items() - } - raise WorkflowExecutionError( - f"unsupported schedule expression for {label} {location}" + resolver = ( + occurrence + if isinstance(occurrence, MappingSourceResolver) + else MappingSourceResolver(occurrence) + ) + return _resolve_any( + expression, + graph_resolver=None, + occurrence_resolver=resolver, + label=label, + location=location, ) def resolve_schedule_input_bindings( bindings: Sequence[ScheduleInputBinding], *, - occurrence: Mapping[str, Any], + occurrence: Mapping[str, Any] | MappingSourceResolver, label: str, ) -> dict[str, Any]: """Build one workflow-input object from schedule-side bindings.""" + resolver = ( + occurrence + if isinstance(occurrence, MappingSourceResolver) + else MappingSourceResolver(occurrence) + ) targets = [str(binding.target) for binding in bindings] if has_overlapping_paths(targets): raise WorkflowExecutionError(f"{label}: schedule targets overlap: {targets!r}") payload: dict[str, Any] = {} for binding in bindings: location = str(binding.target) - value = resolve_schedule_expression( + value = _resolve_any( binding.expression, - occurrence=occurrence, + graph_resolver=None, + occurrence_resolver=resolver, label=label, location=location, ) diff --git a/src/wf_core/runtime/ops/interrupts.py b/src/wf_core/runtime/ops/interrupts.py index 8a9c017a..19d39869 100644 --- a/src/wf_core/runtime/ops/interrupts.py +++ b/src/wf_core/runtime/ops/interrupts.py @@ -13,7 +13,8 @@ from wf_core.run_state import ( RunState, StepExecutionResult, ) -from wf_core.runtime.input_bindings import resolve_step_input_bindings +from wf_core.runtime.input_bindings import resolve_step_input_bindings_with_resolver +from wf_core.runtime.input_sources import GraphSourceResolver from wf_core.runtime.lineage import commit_foreach_aware_patch from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace from wf_core.runtime.ops.index import WorkflowIndex @@ -34,11 +35,11 @@ def build_interrupt_request( public_node_id: str | None = None, route: InterruptRoute | None = None, ) -> InterruptRequest: - payload = resolve_step_input_bindings( + payload = resolve_step_input_bindings_with_resolver( node.request, - state=state, - workflow_input=workflow_input, - context=context, + resolver=GraphSourceResolver( + state=state, workflow_input=workflow_input, context=context + ), label=f"interrupt {node.id!r} request", ) validate_payload_against_schema( diff --git a/src/wf_core/runtime/ops/nodes.py b/src/wf_core/runtime/ops/nodes.py index 9e02d7e4..da5f5601 100644 --- a/src/wf_core/runtime/ops/nodes.py +++ b/src/wf_core/runtime/ops/nodes.py @@ -15,7 +15,8 @@ from wf_core.run_state import ( RuntimeContext, StepExecutionResult, ) -from wf_core.runtime.input_bindings import resolve_step_input_bindings +from wf_core.runtime.input_bindings import resolve_step_input_bindings_with_resolver +from wf_core.runtime.input_sources import GraphSourceResolver from wf_core.runtime.lineage import ( commit_foreach_aware_patch, scope_input_for_frame, @@ -57,11 +58,14 @@ def _resolve_node_execution( context_view = frame_context_view(run, frame) context_values = context_view.graph state_view = state_view_for_frame(run, frame) - resolved_input = resolve_step_input_bindings( - node.input, + resolver = GraphSourceResolver( state=state_view, workflow_input=scope_input_for_frame(run, frame), context=context_values, + ) + resolved_input = resolve_step_input_bindings_with_resolver( + node.input, + resolver=resolver, label=f"node {node.id!r} input", ) validate_payload_against_schema( diff --git a/src/wf_core/runtime/subgraphs.py b/src/wf_core/runtime/subgraphs.py index 42c5bcff..f80f7330 100644 --- a/src/wf_core/runtime/subgraphs.py +++ b/src/wf_core/runtime/subgraphs.py @@ -16,7 +16,8 @@ from wf_core.run_state import ( RuntimeScope, StepExecutionResult, ) -from wf_core.runtime.input_bindings import resolve_step_input_bindings +from wf_core.runtime.input_bindings import resolve_step_input_bindings_with_resolver +from wf_core.runtime.input_sources import GraphSourceResolver from wf_core.runtime.lineage import commit_foreach_aware_patch from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.merges import ReducerDefinition @@ -136,11 +137,13 @@ def _start_subgraph( ) -> None: prepared.workflow.validate_structure().raise_for_errors() parent_scope = run.scopes[frame.scope_id] - child_input = resolve_step_input_bindings( + child_input = resolve_step_input_bindings_with_resolver( step.input, - state=state_view_for_frame(run, frame), - workflow_input=parent_scope.workflow_input, - context=frame_context_view(run, frame).graph, + resolver=GraphSourceResolver( + state=state_view_for_frame(run, frame), + workflow_input=parent_scope.workflow_input, + context=frame_context_view(run, frame).graph, + ), label=f"subgraph {step.id!r}", ) validate_payload_against_schema( diff --git a/src/wf_core/validation/steps.py b/src/wf_core/validation/steps.py index f568ba88..a76003ea 100644 --- a/src/wf_core/validation/steps.py +++ b/src/wf_core/validation/steps.py @@ -33,6 +33,7 @@ from wf_core.paths import ( is_valid_destination_path, is_valid_source_path, ) +from wf_core.runtime.input_sources import walk_expression_paths from wf_core.validation.issues import ValidationIssueCode, ValidationReport @@ -245,11 +246,9 @@ def _validate_expression_sources( ) -> None: """Validate every graph path leaf while keeping one top-level target atomic. - Leaf traversal order is shared with + Leaf traversal order is the canonical order defined by :func:`wf_core.runtime.input_sources.walk_expression_paths`. """ - from wf_core.runtime.input_sources import walk_expression_paths - for path, suffix in walk_expression_paths(expression): _validate_source_path( path, diff --git a/tests/scheduling/test_schedule_expressions.py b/tests/scheduling/test_schedule_expressions.py index ce404f3e..81800298 100644 --- a/tests/scheduling/test_schedule_expressions.py +++ b/tests/scheduling/test_schedule_expressions.py @@ -111,6 +111,53 @@ def test_graph_only_path_is_invalid_in_schedule_bindings() -> None: ) +def test_nested_graph_path_is_invalid_in_schedule_bindings() -> None: + with pytest.raises(ValidationError): + ScheduleInputBinding.model_validate( + { + "target": "x", + "expression": { + "kind": "object", + "fields": { + "nested": {"kind": "path", "path": "input.a"}, + }, + }, + } + ) + with pytest.raises(ValidationError): + ScheduleInputBinding.model_validate( + { + "target": "x", + "expression": { + "kind": "array", + "items": [{"kind": "path", "path": "state.b"}], + }, + } + ) + + +def test_nested_occurrence_is_invalid_in_graph_bindings() -> None: + from wf_core.models.input_bindings import ( + ArrayExpression, + ObjectExpression, + ) + + with pytest.raises(ValidationError): + ArrayExpression.model_validate( + { + "kind": "array", + "items": [{"kind": "occurrence", "field": "scheduled_at"}], + } + ) + with pytest.raises(ValidationError): + ObjectExpression.model_validate( + { + "kind": "object", + "fields": {"when": {"kind": "occurrence", "field": "scheduled_at"}}, + } + ) + + def test_invalid_field_rejected() -> None: with pytest.raises(ValidationError): ScheduleInputBinding.model_validate( @@ -153,7 +200,7 @@ def test_invalid_resolved_input_fails_schema_check() -> None: resolved = resolve_schedule_input_bindings( [binding], occurrence=OCC, label="schedule s" ) - with pytest.raises(Exception, match="count"): + with pytest.raises(WorkflowExecutionError, match="count"): validate_payload_against_schema( {"type": "object", "properties": {"count": {"type": "integer"}}}, resolved, @@ -161,10 +208,27 @@ def test_invalid_resolved_input_fails_schema_check() -> None: ) -def test_direct_schedule_expression_resolution_rejects_graph_paths() -> None: +def test_direct_schedule_expression_resolves_literal() -> None: expr = ScheduleInputBinding.model_validate( {"target": "x", "expression": {"kind": "literal", "value": 1}} ).expression assert ( resolve_schedule_expression(expr, occurrence=OCC, label="s", location="x") == 1 ) + + +def test_graph_path_leaf_rejected_inside_schedule_traversal() -> None: + from wf_core.models.input_bindings import PathExpression + + from wf_core.runtime.input_sources import MappingSourceResolver + + graph_leaf = PathExpression.model_validate( + {"kind": "path", "path": "input.a"} + ) + with pytest.raises(WorkflowExecutionError, match="graph paths are invalid"): + resolve_schedule_expression( + graph_leaf, # type: ignore[arg-type] + occurrence=MappingSourceResolver(OCC), + label="s", + location="x", + )