From 0ded9b0d401d69a2fe5c6522e8c27e61172c5d9a Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:09:31 +0700 Subject: [PATCH] sched: add typed occurrence expression kind for schedule inputs (T04) --- src/wf_core/models/input_bindings.py | 72 +++++++- src/wf_core/runtime/input_sources.py | 83 ++++++++- tests/scheduling/test_schedule_expressions.py | 170 ++++++++++++++++++ 3 files changed, 323 insertions(+), 2 deletions(-) create mode 100644 tests/scheduling/test_schedule_expressions.py diff --git a/src/wf_core/models/input_bindings.py b/src/wf_core/models/input_bindings.py index 765c7c26..8fee178c 100644 --- a/src/wf_core/models/input_bindings.py +++ b/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() diff --git a/src/wf_core/runtime/input_sources.py b/src/wf_core/runtime/input_sources.py index f17f0c24..e2ed1f9b 100644 --- a/src/wf_core/runtime/input_sources.py +++ b/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]]: diff --git a/tests/scheduling/test_schedule_expressions.py b/tests/scheduling/test_schedule_expressions.py new file mode 100644 index 00000000..ce404f3e --- /dev/null +++ b/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 + )