diff --git a/src/wf_core/runtime/input_bindings.py b/src/wf_core/runtime/input_bindings.py index dde20561..02c46c2d 100644 --- a/src/wf_core/runtime/input_bindings.py +++ b/src/wf_core/runtime/input_bindings.py @@ -7,17 +7,14 @@ 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 ( - ArrayExpression, InputExpression, InputExpressionBinding, InputPathBinding, InputValueBinding, - LiteralExpression, - ObjectExpression, - PathExpression, StepInputBinding, ) -from wf_core.models.json_values import JsonValue, validate_strict_json_value +from wf_core.models.json_values import JsonValue +from wf_core.runtime.input_sources import SourceResolver def resolve_input_expression( @@ -30,45 +27,32 @@ 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, + ) - if isinstance(expression, LiteralExpression): - return expression.value - if isinstance(expression, PathExpression): - try: - value = safe_resolve_path( - str(expression.path), - state=state, - workflow_input=workflow_input, - context=context, - ) - return validate_strict_json_value(value) - except (ValueError, WorkflowExecutionError) as exc: - raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc - if isinstance(expression, ArrayExpression): - return [ - resolve_input_expression( - item, - state=state, - workflow_input=workflow_input, - context=context, - label=label, - location=f"{location}[{index}]", - ) - for index, item in enumerate(expression.items) - ] - if isinstance(expression, ObjectExpression): - return { - field: resolve_input_expression( - value, - state=state, - workflow_input=workflow_input, - context=context, - label=label, - location=f"{location}.{field}", - ) - for field, value in expression.fields.items() - } - raise WorkflowExecutionError(f"unsupported input expression for {label} {location}") + resolver = GraphSourceResolver( + state=state, workflow_input=workflow_input, context=context + ) + return resolve_composed_expression( + expression, resolver=resolver, label=label, location=location + ) + + +def resolve_input_expression_with_resolver( + expression: InputExpression, + *, + resolver: SourceResolver, + label: str, + 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( diff --git a/src/wf_core/runtime/input_sources.py b/src/wf_core/runtime/input_sources.py new file mode 100644 index 00000000..f17f0c24 --- /dev/null +++ b/src/wf_core/runtime/input_sources.py @@ -0,0 +1,135 @@ +"""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 +``GraphSourcePath`` with schedule-only roots and do not smuggle occurrence +values through a faked graph ``context`` mapping. +""" + +from __future__ import annotations + +from collections.abc import Iterator, Mapping +from typing import Any, Protocol + +from wf_core.errors import WorkflowExecutionError +from wf_core.models.input_bindings import ( + ArrayExpression, + InputExpression, + LiteralExpression, + ObjectExpression, + PathExpression, +) +from wf_core.models.json_values import JsonValue, validate_strict_json_value +from wf_core.paths import GraphSourcePath, resolve_graph_path + + +class SourceResolver(Protocol): + """Resolve one leaf source path to a JSON-compatible value.""" + + def resolve_path(self, path: GraphSourcePath) -> Any: + """Return the value for one graph source path.""" + ... + + +class GraphSourceResolver: + """Resolve ``input`` / ``state`` / ``context`` paths from live mappings.""" + + def __init__( + self, + *, + state: Mapping[str, Any], + workflow_input: Mapping[str, Any], + context: Mapping[str, Any], + ) -> None: + self._state = state + self._workflow_input = workflow_input + self._context = context + + def resolve_path(self, path: GraphSourcePath) -> Any: + """Resolve one graph path against the captured mappings.""" + return resolve_graph_path( + path, + state=self._state, + workflow_input=self._workflow_input, + context=self._context, + ) + + +class MappingSourceResolver: + """Resolve occurrence-style fields from a flat mapping (schedule side). + + This resolver exists so schedule composition reuses the same traversal + without faking a graph ``context``. It only understands occurrence field + names; graph paths are invalid here. + """ + + def __init__(self, values: Mapping[str, Any]) -> None: + self._values = dict(values) + + def resolve_field(self, field: str) -> Any: + """Return one occurrence field value.""" + if field not in self._values: + raise WorkflowExecutionError(f"unknown occurrence field {field!r}") + return self._values[field] + + +def resolve_composed_expression( + expression: InputExpression, + *, + resolver: SourceResolver, + label: str, + location: str, +) -> JsonValue: + """Resolve one composite expression through a typed source resolver.""" + if isinstance(expression, LiteralExpression): + return expression.value + if isinstance(expression, PathExpression): + try: + value = 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): + return [ + resolve_composed_expression( + item, + resolver=resolver, + label=label, + location=f"{location}[{index}]", + ) + for index, item in enumerate(expression.items) + ] + if isinstance(expression, ObjectExpression): + return { + field: resolve_composed_expression( + value, + resolver=resolver, + label=label, + location=f"{location}.{field}", + ) + for field, value in expression.fields.items() + } + raise WorkflowExecutionError(f"unsupported input expression for {label} {location}") + + +def walk_expression_paths( + expression: InputExpression, +) -> Iterator[tuple[GraphSourcePath, str]]: + """Yield ``(graph path, location)`` leaves in one shared traversal order. + + Locations use the same ``items[i]`` / ``fields.name`` / ``path`` suffix + shape as :func:`wf_core.validation.steps._validate_expression_sources`, + so validation and runtime traverse leaves in the same order. + """ + if isinstance(expression, PathExpression): + yield expression.path, "path" + elif isinstance(expression, ArrayExpression): + for index, item in enumerate(expression.items): + for path, suffix in walk_expression_paths(item): + yield path, f"items[{index}].{suffix}" + elif isinstance(expression, ObjectExpression): + for field, item in expression.fields.items(): + for path, suffix in walk_expression_paths(item): + yield path, f"fields.{field}.{suffix}" diff --git a/src/wf_core/validation/steps.py b/src/wf_core/validation/steps.py index b47654c6..f568ba88 100644 --- a/src/wf_core/validation/steps.py +++ b/src/wf_core/validation/steps.py @@ -10,13 +10,7 @@ from wf_core.models.conditions import ( PathOperand, VariadicCondition, ) -from wf_core.models.input_bindings import ( - ArrayExpression, - InputExpression, - InputExpressionBinding, - ObjectExpression, - PathExpression, -) +from wf_core.models.input_bindings import InputExpression, InputExpressionBinding from wf_core.models.schemas import NodeDef from wf_core.models.steps import ( ConditionNode, @@ -249,39 +243,23 @@ def _validate_expression_sources( error_code: ValidationIssueCode = ValidationIssueCode.INVALID_SOURCE_PATH, message: str = "source path must start with input., state., or context. and reference a declared root field when applicable", ) -> None: - """Validate every graph path leaf while keeping one top-level target atomic.""" - if isinstance(expression, PathExpression): + """Validate every graph path leaf while keeping one top-level target atomic. + + Leaf traversal order is shared with + :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( - expression.path, - f"{input_path}.path", + path, + f"{input_path}.{suffix}", report, state_root_fields, input_root_fields, error_code=error_code, message=message, ) - elif isinstance(expression, ArrayExpression): - for index, item in enumerate(expression.items): - _validate_expression_sources( - item, - input_path=f"{input_path}.items[{index}]", - report=report, - state_root_fields=state_root_fields, - input_root_fields=input_root_fields, - error_code=error_code, - message=message, - ) - elif isinstance(expression, ObjectExpression): - for field, item in expression.fields.items(): - _validate_expression_sources( - item, - input_path=f"{input_path}.fields.{field}", - report=report, - state_root_fields=state_root_fields, - input_root_fields=input_root_fields, - error_code=error_code, - message=message, - ) def _local_root(path: str | LocalPath) -> str | None: diff --git a/tests/core/test_input_sources.py b/tests/core/test_input_sources.py new file mode 100644 index 00000000..928df4d3 --- /dev/null +++ b/tests/core/test_input_sources.py @@ -0,0 +1,108 @@ +"""Source-resolver seam: one traversal, distinct graph/schedule models (T03).""" + +from __future__ import annotations + +import pytest + +from wf_core.errors import WorkflowExecutionError +from wf_core.models.input_bindings import ( + ObjectExpression, + PathExpression, + validate_input_expression_limits, +) +from wf_core.runtime.input_bindings import ( + resolve_input_expression, + resolve_input_expression_with_resolver, +) +from wf_core.runtime.input_sources import ( + GraphSourceResolver, + MappingSourceResolver, + resolve_composed_expression, + walk_expression_paths, +) + + +def test_graph_resolver_matches_legacy_resolution() -> None: + expr = ObjectExpression.model_validate( + { + "kind": "object", + "fields": { + "team": {"kind": "literal", "value": "eng"}, + "tags": {"kind": "array", "items": [{"kind": "literal", "value": "a"}]}, + "req": {"kind": "path", "path": "input.request_id"}, + }, + } + ) + kwargs = { + "state": {}, + "workflow_input": {"request_id": "r1"}, + "context": {}, + "label": "probe", + "location": "$", + } + assert resolve_input_expression(expr, **kwargs) == { + "team": "eng", + "tags": ["a"], + "req": "r1", + } + resolver = GraphSourceResolver( + state={}, workflow_input={"request_id": "r1"}, context={} + ) + assert resolve_composed_expression( + expr, resolver=resolver, label="probe", location="$" + ) == {"team": "eng", "tags": ["a"], "req": "r1"} + assert resolve_input_expression_with_resolver( + expr, resolver=resolver, label="probe", location="$" + ) == {"team": "eng", "tags": ["a"], "req": "r1"} + + +def test_graph_resolver_does_not_smuggle_occurrence_values() -> None: + # Occurrence values must not be faked through graph context: a path that + # looks like an occurrence root is rejected by GraphSourcePath itself, + # and a resolver carrying occurrence data under context is a contract + # violation, not a supported seam. + from wf_core.paths import GraphSourcePath + + with pytest.raises(ValueError): + GraphSourcePath.parse("occurrence.scheduled_at") + resolver = GraphSourceResolver(state={}, workflow_input={}, context={}) + expr = PathExpression.model_validate({"kind": "path", "path": "input.missing"}) + with pytest.raises(WorkflowExecutionError): + resolve_composed_expression(expr, resolver=resolver, label="t", location="$") + + +def test_mapping_resolver_is_distinct_from_graph_resolver() -> None: + occ = MappingSourceResolver( + {"schedule_id": "s", "scheduled_at": "2026-09-08T13:00:00+00:00"} + ) + assert occ.resolve_field("schedule_id") == "s" + with pytest.raises(WorkflowExecutionError, match="unknown occurrence field"): + occ.resolve_field("nope") + assert not hasattr(occ, "resolve_path") + + +def test_walk_order_matches_validation_suffixes() -> None: + expr = ObjectExpression.model_validate( + { + "kind": "object", + "fields": { + "a": {"kind": "path", "path": "input.x"}, + "b": { + "kind": "array", + "items": [{"kind": "path", "path": "state.y"}], + }, + }, + } + ) + assert [suffix for _, suffix in walk_expression_paths(expr)] == [ + "fields.a.path", + "fields.b.items[0].path", + ] + + +def test_budget_validator_is_shared_entry_point() -> None: + deep: dict = {"kind": "literal", "value": 0} + for _ in range(70): + deep = {"kind": "array", "items": [deep]} + with pytest.raises(ValueError, match="limit exceeded"): + validate_input_expression_limits(deep)