sched: extract shared input traversal behind SourceResolver seam (T03)

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,
+    resolver = GraphSourceResolver(
+        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}]",
+    return resolve_composed_expression(
+        expression, resolver=resolver, label=label, location=location
     )
-            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}",
+
+
+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
     )
-            for field, value in expression.fields.items()
-        }
-    raise WorkflowExecutionError(f"unsupported input expression for {label} {location}")
 
 
 def resolve_step_input_bindings(

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}"

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:

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)