sched: add typed occurrence expression kind for schedule inputs (T04)

src/wf_core/models/input_bindings.py

@@ -103,6 +103,46 @@ type InputExpression = Annotated[
 ]
 
 
+class OccurrenceExpression(BaseModel):
+    """Reference one typed schedule-occurrence field.
+
+    Schedule-only leaf: graph expressions must not accept this kind, and
+    ``GraphSourcePath`` roots stay closed to input/state/context.
+    """
+
+    model_config = ConfigDict(extra="forbid", strict=True)
+
+    kind: Literal["occurrence"]
+    field: Literal["schedule_id", "occurrence_id", "scheduled_at"]
+
+
+class ScheduleArrayExpression(BaseModel):
+    """Resolve ordered child schedule expressions into one JSON array."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["array"]
+    items: list[ScheduleExpression]
+
+
+class ScheduleObjectExpression(BaseModel):
+    """Resolve named child schedule expressions into one JSON object."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["object"]
+    fields: dict[str, ScheduleExpression]
+
+
+type ScheduleExpression = Annotated[
+    LiteralExpression
+    | OccurrenceExpression
+    | ScheduleArrayExpression
+    | ScheduleObjectExpression,
+    Field(discriminator="kind"),
+]
+
+
 def _raise_limit(limit: str, location: str) -> None:
     raise ValueError(f"input expression {limit} limit exceeded at {location}")
 
@@ -206,5 +246,35 @@ StepInputBinding = TypeAliasType(
 )
 
 
-for _model in (ArrayExpression, ObjectExpression, InputExpressionBinding):
+class ScheduleInputBinding(BaseModel):
+    """Assign one schedule-side expression to a workflow-input target.
+
+    Schedule expressions reuse literal/object/array composition and the same
+    budget validator, but leaves are typed occurrence references only. Graph
+    paths are invalid here.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    target: LocalPath
+    expression: ScheduleExpression
+
+    @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
+
+
+for _model in (
+    ArrayExpression,
+    ObjectExpression,
+    InputExpressionBinding,
+    ScheduleArrayExpression,
+    ScheduleObjectExpression,
+    ScheduleInputBinding,
+):
     _model.model_rebuild()

src/wf_core/runtime/input_sources.py

@@ -10,16 +10,22 @@ values through a faked graph ``context`` mapping.
 
 from __future__ import annotations
 
-from collections.abc import Iterator, Mapping
+from collections.abc import Iterator, Mapping, Sequence
 from typing import Any, Protocol
 
 from wf_core.errors import WorkflowExecutionError
+from wf_core.local_paths import LocalPathError, has_overlapping_paths, set_local_value
 from wf_core.models.input_bindings import (
     ArrayExpression,
     InputExpression,
     LiteralExpression,
     ObjectExpression,
+    OccurrenceExpression,
     PathExpression,
+    ScheduleArrayExpression,
+    ScheduleExpression,
+    ScheduleInputBinding,
+    ScheduleObjectExpression,
 )
 from wf_core.models.json_values import JsonValue, validate_strict_json_value
 from wf_core.paths import GraphSourcePath, resolve_graph_path
@@ -114,6 +120,81 @@ def resolve_composed_expression(
     raise WorkflowExecutionError(f"unsupported input expression for {label} {location}")
 
 
+def resolve_schedule_expression(
+    expression: ScheduleExpression,
+    *,
+    occurrence: Mapping[str, Any],
+    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``.
+    """
+    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}"
+    )
+
+
+def resolve_schedule_input_bindings(
+    bindings: Sequence[ScheduleInputBinding],
+    *,
+    occurrence: Mapping[str, Any],
+    label: str,
+) -> dict[str, Any]:
+    """Build one workflow-input object from schedule-side bindings."""
+    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(
+            binding.expression,
+            occurrence=occurrence,
+            label=label,
+            location=location,
+        )
+        try:
+            set_local_value(payload, binding.target, value)
+        except LocalPathError as exc:
+            raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc
+    return payload
+
+
 def walk_expression_paths(
     expression: InputExpression,
 ) -> Iterator[tuple[GraphSourcePath, str]]:

tests/scheduling/test_schedule_expressions.py

@@ -0,0 +1,170 @@
+"""Typed occurrence expressions: schedule-only leaves, shared composition (T04)."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from wf_core.errors import WorkflowExecutionError
+from wf_core.models.input_bindings import (
+    InputExpressionBinding,
+    ScheduleInputBinding,
+    validate_input_expression_limits,
+)
+from wf_core.paths import GraphSourcePath
+from wf_core.runtime.input_sources import (
+    resolve_schedule_expression,
+    resolve_schedule_input_bindings,
+)
+
+OCC = {
+    "schedule_id": "sched-1",
+    "occurrence_id": "sched-1|2026-09-08T13:00:00+00:00",
+    "scheduled_at": "2026-09-08T13:00:00+00:00",
+}
+
+
+def test_graph_union_still_rejects_occurrence_kind() -> None:
+    with pytest.raises(ValidationError):
+        InputExpressionBinding.model_validate(
+            {
+                "target": "x",
+                "expression": {"kind": "occurrence", "field": "scheduled_at"},
+            }
+        )
+
+
+def test_graph_roots_stay_closed() -> None:
+    with pytest.raises(ValueError):
+        GraphSourcePath.parse("occurrence.scheduled_at")
+
+
+def test_schedule_binding_json_round_trip() -> None:
+    binding = ScheduleInputBinding.model_validate(
+        {
+            "target": "report_time",
+            "expression": {"kind": "occurrence", "field": "scheduled_at"},
+        }
+    )
+    assert binding.model_dump(mode="json") == {
+        "target": "report_time",
+        "expression": {"kind": "occurrence", "field": "scheduled_at"},
+    }
+    literal = ScheduleInputBinding.model_validate(
+        {"target": "team", "expression": {"kind": "literal", "value": "eng"}}
+    )
+    assert literal.expression.model_dump(mode="json")["value"] == "eng"
+
+
+def test_literal_and_nested_occurrence_composition() -> None:
+    binding = ScheduleInputBinding.model_validate(
+        {
+            "target": "payload",
+            "expression": {
+                "kind": "object",
+                "fields": {
+                    "team": {"kind": "literal", "value": "eng"},
+                    "when": {"kind": "occurrence", "field": "scheduled_at"},
+                    "tags": {
+                        "kind": "array",
+                        "items": [
+                            {"kind": "occurrence", "field": "schedule_id"},
+                            {"kind": "literal", "value": "x"},
+                        ],
+                    },
+                },
+            },
+        }
+    )
+    resolved = resolve_schedule_input_bindings(
+        [binding], occurrence=OCC, label="schedule sched-1"
+    )
+    assert resolved == {
+        "payload": {
+            "team": "eng",
+            "when": "2026-09-08T13:00:00+00:00",
+            "tags": ["sched-1", "x"],
+        }
+    }
+
+
+def test_missing_occurrence_field_fails_before_dispatch() -> None:
+    binding = ScheduleInputBinding.model_validate(
+        {
+            "target": "x",
+            "expression": {"kind": "occurrence", "field": "scheduled_at"},
+        }
+    )
+    with pytest.raises(WorkflowExecutionError, match="unknown occurrence field"):
+        resolve_schedule_input_bindings(
+            [binding], occurrence={"schedule_id": "s"}, label="schedule s"
+        )
+
+
+def test_graph_only_path_is_invalid_in_schedule_bindings() -> None:
+    with pytest.raises(ValidationError):
+        ScheduleInputBinding.model_validate(
+            {
+                "target": "x",
+                "expression": {"kind": "path", "path": "input.a"},
+            }
+        )
+
+
+def test_invalid_field_rejected() -> None:
+    with pytest.raises(ValidationError):
+        ScheduleInputBinding.model_validate(
+            {
+                "target": "x",
+                "expression": {"kind": "occurrence", "field": "nope"},
+            }
+        )
+
+
+def test_over_budget_schedule_expression_rejected() -> 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)
+    with pytest.raises(ValueError, match="limit exceeded"):
+        ScheduleInputBinding.model_validate({"target": "x", "expression": deep})
+
+
+def test_conflicting_targets_rejected() -> None:
+    first = ScheduleInputBinding.model_validate(
+        {"target": "a.b", "expression": {"kind": "literal", "value": 1}}
+    )
+    second = ScheduleInputBinding.model_validate(
+        {"target": "a.b.c", "expression": {"kind": "literal", "value": 2}}
+    )
+    with pytest.raises(WorkflowExecutionError, match="overlap"):
+        resolve_schedule_input_bindings(
+            [first, second], occurrence=OCC, label="schedule s"
+        )
+
+
+def test_invalid_resolved_input_fails_schema_check() -> None:
+    from wf_core.runtime.ops.schemas import validate_payload_against_schema
+
+    binding = ScheduleInputBinding.model_validate(
+        {"target": "count", "expression": {"kind": "literal", "value": "not-an-int"}}
+    )
+    resolved = resolve_schedule_input_bindings(
+        [binding], occurrence=OCC, label="schedule s"
+    )
+    with pytest.raises(Exception, match="count"):
+        validate_payload_against_schema(
+            {"type": "object", "properties": {"count": {"type": "integer"}}},
+            resolved,
+            "schedule s input",
+        )
+
+
+def test_direct_schedule_expression_resolution_rejects_graph_paths() -> 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
+    )