From 15bd472bdc0acfd19c6f7131e8287843ef651866 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 11:19:48 +0700 Subject: [PATCH] sched: extract typed preparer/dispatcher seams; test impl out of production (F6) --- src/wf_scheduling/dispatch.py | 41 +++++ src/wf_scheduling/poll.py | 194 +++++++++--------------- src/wf_scheduling/prepare.py | 106 +++++++++++++ src/wf_scheduling/recovery.py | 11 +- tests/scheduling/controlled.py | 90 +++++++++++ tests/scheduling/test_poll.py | 54 ++++--- tests/scheduling/test_preparation.py | 216 +++++++++++++++++++++++++++ tests/scheduling/test_recovery.py | 14 +- 8 files changed, 564 insertions(+), 162 deletions(-) create mode 100644 src/wf_scheduling/dispatch.py create mode 100644 src/wf_scheduling/prepare.py create mode 100644 tests/scheduling/controlled.py create mode 100644 tests/scheduling/test_preparation.py diff --git a/src/wf_scheduling/dispatch.py b/src/wf_scheduling/dispatch.py new file mode 100644 index 00000000..69f488f2 --- /dev/null +++ b/src/wf_scheduling/dispatch.py @@ -0,0 +1,41 @@ +"""Typed dispatch collaborator seam (R4/F6). + +The scheduler owns durable markers, stopped-result persistence, and +history; the dispatcher owns execution only. It receives the admitted +invocation and returns a genuine :class:`wf_core.RunState` (or +:class:`StillRunning` when the outcome is not yet known). Stopped +results are persisted by the scheduler through the shared +``wf_api.run_lifecycle`` boundary, so scheduler dispatches and manual +runs share one torn-write protocol. Controlled scripted dispatchers live +in ``tests/scheduling/controlled.py``; production wiring arrives in T12. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol, Union + +from wf_artifacts.runs.models import RunAdmission +from wf_core import RunState + + +@dataclass(frozen=True) +class Stopped: + """Execution finished with a genuine stopped run state.""" + + result: RunState + + +@dataclass(frozen=True) +class StillRunning: + """Dispatched with an unknown outcome; the run stays admitted.""" + + +DispatchResult = Union[Stopped, StillRunning] + + +class RunDispatcher(Protocol): + """Execution collaborator behind the durable dispatch transition.""" + + def dispatch(self, *, admission: RunAdmission, now: datetime) -> DispatchResult: ... diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 24bd2707..87462829 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -2,20 +2,32 @@ Mirrors the reference state model (probes/deployment_scheduling_verify/ test_schedule_state_model.py) against real file stores. Calendar iteration -uses ``OccurrenceSource`` (``next_after``/``prev_before`` only, never -enumeration); latest-missed catch-up is one bounded ``prev_before`` query -(F1). Overlap decisions precede capacity checks; terminal skips never -reappear; ``latest`` retains at most one candidate. +uses the canonical :class:`wf_scheduling.calendar.OccurrenceSource` +(``next_after``/``prev_before`` only, never enumeration); latest-missed +catch-up is one bounded ``prev_before`` query (F1). Overlap decisions +precede capacity checks; terminal skips never reappear; ``latest`` retains +at most one candidate. + +Admission preparation and execution arrive as typed collaborators +(:mod:`wf_scheduling.prepare`, :mod:`wf_scheduling.dispatch`): this module +contains no fixture input, fixture environments, or canned results. """ from __future__ import annotations from datetime import datetime, timezone -from typing import Any, Protocol +from typing import Any -from wf_scheduling.calendar import CronSource, OneShotSource +from wf_scheduling.calendar import ( + CronSource, + InvalidScheduleDefinitionError, + OccurrenceSource, + OneShotSource, +) +from wf_scheduling.dispatch import RunDispatcher, StillRunning from wf_scheduling.models import OccurrenceRecord, PendingCandidate from wf_scheduling.occurrences import occurrence_id +from wf_scheduling.prepare import InvocationPreparer, PreparationRejected UTC = timezone.utc SCAN_CAP = 100 @@ -26,17 +38,6 @@ class BlockedSchedule(ValueError): """A schedule is blocked by a corrupt record and fails closed.""" -class InvalidScheduleDefinitionError(ValueError): - """A schedule definition or its trigger source is invalid.""" - - -class OccurrenceSource(Protocol): - """Due-instant source over aware datetimes, UTC at the boundary.""" - - def next_after(self, instant: datetime) -> datetime | None: ... - def prev_before(self, instant: datetime) -> datetime | None: ... - - def source_for_trigger(trigger: Any) -> OccurrenceSource: """Build a calendar source from a schedule trigger model.""" kind = trigger.kind if hasattr(trigger, "kind") else trigger.get("kind") @@ -61,22 +62,17 @@ def source_for_trigger(trigger: Any) -> OccurrenceSource: def _is_oneshot(source: OccurrenceSource) -> bool: - # Calendar and test-double one-shots share the class name and an ``at`` - # instant; periodic sources carry a ``period`` instead. isinstance covers - # the calendar type; the name check covers duck-typed test doubles. - if isinstance(source, OneShotSource): - return True - return type(source).__name__ == "OneShotSource" and hasattr(source, "at") + return isinstance(source, OneShotSource) class Scheduler: - """File-store scheduler core with injected clock and controlled dispatch. + """File-store scheduler core with injected clock and collaborators. - ``outcomes`` maps run ids (or ``"*"``) to ``complete`` | ``hang`` | - ``interrupt`` for deterministic tests. Production dispatch (real runtime) - is wired in T12; here ``hang`` keeps an admitted run occupying its - schedule slot, ``interrupt`` marks it interrupted (schedule slot only), - and ``complete`` marks it completed (releases overlap). + Invocation preparation (:class:`wf_scheduling.prepare.InvocationPreparer`) + and execution (:class:`wf_scheduling.dispatch.RunDispatcher`) are typed + collaborators: production code here never fabricates input, environments, + or outcomes. Stopped results are persisted through the shared + ``wf_api.run_lifecycle`` boundary. """ def __init__( @@ -86,15 +82,15 @@ class Scheduler: run_store: Any, sources: dict[str, OccurrenceSource], capacity: int, - outcomes: dict[str, str] | None = None, - deployments: dict[str, dict] | None = None, + preparer: InvocationPreparer, + dispatcher: RunDispatcher, ) -> None: self.schedule_store = schedule_store self.run_store = run_store self.sources = sources self.capacity = capacity - self.outcomes = outcomes or {} - self.deployments = deployments or {"dep-1": {"rev": 1, "required": []}} + self.preparer = preparer + self.dispatcher = dispatcher self._poll_cursor = 0 # -- helpers ------------------------------------------------------ @@ -191,13 +187,13 @@ class Scheduler: sched.blocked_reason = reason self.schedule_store.save_schedule(sched) raise BlockedSchedule(reason) from None - dep = self.deployments.get(sched.deployment_id) - if dep is None: + prepared = self.preparer.prepare(sched=sched, intended=intended, now=now) + if isinstance(prepared, PreparationRejected): self._record( kind="preflight-rejected", sched_id=sched.id, intended=intended, - reason="deployment-deleted", + reason=prepared.reason, revision=sched.revision, now=now, ) @@ -206,26 +202,6 @@ class Scheduler: self.schedule_store.save_candidate(None, schedule_id=sched.id) self.schedule_store.save_consumed(sched.id, intended) return None - frozen = { - "team": "eng", - "report_time": intended.isoformat(), - "sched": sched.id, - "dep_rev": dep["rev"], - } - missing = [k for k in dep.get("required", []) if k not in frozen] - if missing: - self._record( - kind="preflight-rejected", - sched_id=sched.id, - intended=intended, - reason=f"missing-input:{missing}", - revision=sched.revision, - ) - cand = self.schedule_store.get_candidate(sched.id) - if cand is not None and cand.intended_at == intended: - self.schedule_store.save_candidate(None, schedule_id=sched.id) - self.schedule_store.save_consumed(sched.id, intended) - return None active = self._active(sched.id) if sched.overlap == "skip" and active: self._record( @@ -277,20 +253,19 @@ class Scheduler: return None return "held-undecided" run_id = self.run_store.allocate_run_id() - from wf_artifacts.runs.models import RunAdmission + from wf_api.run_lifecycle import persist_admission - admission = RunAdmission( - id=run_id, - environment=_test_environment(sched), - resolved_input=dict(frozen), - max_steps=getattr(sched, "max_steps", None), + admission = persist_admission( + store=self.run_store, + run_id=run_id, + environment=prepared.environment, + resolved_input=prepared.resolved_input, + max_steps=prepared.max_steps, scheduled_at=intended, schedule_id=sched.id, schedule_revision=sched.revision, - deployment_revision=dep["rev"], - created_at=now, + deployment_revision=prepared.deployment_revision, ) - self.run_store.save_admission(admission) self._record( kind="admitted", sched_id=sched.id, @@ -313,56 +288,52 @@ class Scheduler: # Dispatch mark precedes the view so a crash after admission but # before/during first dispatch stays pending (not abandoned). Cleared # after dispatch returns regardless of outcome (hang still dispatched). - try: - self.run_store.mark_pending_dispatch(run_id) - except AttributeError: - pass + self.run_store.mark_pending_dispatch(run_id) materialize_admitted_view(store=self.run_store, admission=admission) try: self._dispatch(run_id, now) finally: - try: - self.run_store.clear_pending_dispatch(run_id) - except AttributeError: - pass + self.run_store.clear_pending_dispatch(run_id) return run_id def _dispatch(self, run_id: str, now: datetime) -> None: + """Persist one stopped result for an admitted run via the dispatcher. + + The dispatcher returns a genuine stopped :class:`wf_core.RunState` + (or :class:`StillRunning` when the outcome is unknown); persistence + goes through the shared ``wf_api.run_lifecycle`` boundary so + scheduler dispatches share the manual-run torn-write protocol. + """ + from wf_api.run_lifecycle import persist_stopped_run from wf_artifacts.runs.models import StoredRunStatus - outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete")) - if outcome not in ("complete", "hang", "interrupt"): - raise ValueError(f"unknown controlled outcome {outcome!r} for {run_id!r}") try: - record = self.run_store.get_run(run_id) + self.run_store.get_run(run_id) except KeyError as exc: raise BlockedSchedule(f"dispatch missing run view: {run_id!r}") from exc - sched_id = self._run_sched(record) - if sched_id is None: - raise BlockedSchedule(f"dispatch missing admission: {run_id!r}") - if outcome == "hang": + try: + admission = self.run_store.get_admission(run_id) + except KeyError as exc: + raise BlockedSchedule(f"dispatch missing admission: {run_id!r}") from exc + if admission.schedule_id is None: + raise BlockedSchedule(f"dispatch missing schedule owner: {run_id!r}") + result = self.dispatcher.dispatch(admission=admission, now=now) + if isinstance(result, StillRunning): return - if outcome == "interrupt": - updated = record.model_copy( - update={"status": StoredRunStatus.INTERRUPTED, "updated_at": now} - ) - self.run_store.save_run(updated) - self._record( - kind="interrupted", - sched_id=sched_id, - intended=_admission_intended(self.run_store, run_id), - run_id=run_id, - now=now, - started_at=now, - ) - return - updated = record.model_copy( - update={"status": StoredRunStatus.COMPLETED, "updated_at": now} + stopped = persist_stopped_run( + store=self.run_store, + environment=admission.environment, + run=result.result, + run_id=run_id, ) - self.run_store.save_run(updated) + kind = { + StoredRunStatus.COMPLETED: "completed", + StoredRunStatus.INTERRUPTED: "interrupted", + StoredRunStatus.FAILED: "failed", + }[stopped.status] self._record( - kind="completed", - sched_id=sched_id, + kind=kind, # type: ignore[arg-type] + sched_id=admission.schedule_id, intended=_admission_intended(self.run_store, run_id), run_id=run_id, now=now, @@ -672,29 +643,6 @@ class Scheduler: self.schedule_store.save_consumed(sid, max(consumed, now)) -def _test_environment(sched: Any) -> Any: - from wf_artifacts import PinnedRunEnvironment, WorkflowArtifact, WorkflowDeployment - - deployment = WorkflowDeployment( - id=getattr(sched, "deployment_id", "dep-1"), - artifact_id="wf-1", - artifact_version=1, - bindings=[], - ) - artifact = WorkflowArtifact( - id="wf-1", - version=1, - title="Wf-1", - input_schema={"type": "object", "properties": {}}, - output_schema={"type": "object", "properties": {}}, - outcomes=("ok",), - plan={"name": "wf-1", "nodes": [], "edges": []}, - ) - return PinnedRunEnvironment( - deployment=deployment, root_artifact=artifact, child_artifacts=[] - ) - - def _admission_intended(run_store: Any, run_id: str) -> datetime | None: try: return run_store.get_admission(run_id).scheduled_at diff --git a/src/wf_scheduling/prepare.py b/src/wf_scheduling/prepare.py new file mode 100644 index 00000000..03fb8595 --- /dev/null +++ b/src/wf_scheduling/prepare.py @@ -0,0 +1,106 @@ +"""Typed invocation preparation for schedule admission (R4/F6). + +Production schedule-side admission decisions live here: resolving the +schedule's real input bindings against the occurrence environment and +rechecking the deployment contract. Only artifact-tree construction is +injected (``build_environment``) until the administration surface (T13) +supplies store-backed environment pinning; controlled fixture +environments and deployment directories live in +``tests/scheduling/controlled.py``, never here. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Protocol + +from wf_artifacts.runs.models import PinnedRunEnvironment +from wf_core.runtime.input_sources import resolve_schedule_input_bindings +from wf_scheduling.occurrences import occurrence_id + + +class DeploymentDirectory(Protocol): + """Deployment contract source for admission rechecks. + + Implementations raise ``KeyError`` for unknown deployment ids, which + preparation reports as a ``deployment-deleted`` preflight rejection. + """ + + def deployment_revision(self, deployment_id: str) -> int: ... + def required_inputs(self, deployment_id: str) -> Sequence[str]: ... + + +@dataclass(frozen=True) +class PreparedInvocation: + """Frozen invocation captured at admission for exactly one occurrence.""" + + environment: PinnedRunEnvironment + resolved_input: dict[str, Any] + max_steps: int | None + deployment_revision: int + + +@dataclass(frozen=True) +class PreparationRejected: + """Preflight rejection: no run is invented for this occurrence.""" + + reason: str + + +class InvocationPreparer(Protocol): + """Schedule-side admission preparation (production or controlled).""" + + def prepare( + self, *, sched: Any, intended: datetime, now: datetime + ) -> PreparedInvocation | PreparationRejected: ... + + +class SchedulePreparer: + """Production preparation over the schedule's real input bindings. + + Occurrence references resolve through the shared schedule-expression + contract (never graph paths); target conflicts, over-budget trees, and + invalid resolved input fail closed as preflight rejections before any + run identity is allocated. + """ + + def __init__( + self, + deployments: DeploymentDirectory, + build_environment: Callable[[Any], PinnedRunEnvironment], + ) -> None: + self._deployments = deployments + self._build_environment = build_environment + + def prepare( + self, *, sched: Any, intended: datetime, now: datetime + ) -> PreparedInvocation | PreparationRejected: + try: + revision = self._deployments.deployment_revision(sched.deployment_id) + required = list(self._deployments.required_inputs(sched.deployment_id)) + except KeyError: + return PreparationRejected(reason="deployment-deleted") + occurrence = { + "schedule_id": sched.id, + "occurrence_id": occurrence_id(sched.id, intended), + "scheduled_at": intended.isoformat(), + } + try: + resolved = resolve_schedule_input_bindings( + sched.input_bindings, + occurrence=occurrence, + label=f"schedule {sched.id}", + ) + except Exception as exc: + return PreparationRejected(reason=f"invalid-input:{exc}") + missing = [key for key in required if key not in resolved] + if missing: + return PreparationRejected(reason=f"missing-input:{missing}") + return PreparedInvocation( + environment=self._build_environment(sched), + resolved_input=dict(resolved), + max_steps=getattr(sched, "max_steps", None), + deployment_revision=revision, + ) diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index c44bc094..fef29198 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -171,16 +171,9 @@ def _mark_pending(run_store: Any, run_id: str) -> None: def _is_pending(run_store: Any, run_id: str) -> bool: - try: - return bool(run_store.is_pending_dispatch(run_id)) - except AttributeError: - # Legacy stores without the pending protocol: treat as not pending. - return False + return bool(run_store.is_pending_dispatch(run_id)) def clear_pending(run_store: Any, run_id: str) -> None: """Clear the pending-dispatch marker after the poll sweep dispatches.""" - try: - run_store.clear_pending_dispatch(run_id) - except AttributeError: - return + run_store.clear_pending_dispatch(run_id) diff --git a/tests/scheduling/controlled.py b/tests/scheduling/controlled.py new file mode 100644 index 00000000..eee9b4d7 --- /dev/null +++ b/tests/scheduling/controlled.py @@ -0,0 +1,90 @@ +"""Controlled scheduler collaborators for tests (fixture side of F6 seams). + +``DictDeployments``, ``fixture_environment``, and ``ScriptedDispatcher`` +are the test doubles behind the production +:mod:`wf_scheduling.prepare` / :mod:`wf_scheduling.dispatch` protocols. +They carry the fixture behavior the old production ``Scheduler`` used to +hardcode (fixture input, fixture environments, scripted outcomes); the +production package no longer contains any of it. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from datetime import datetime +from typing import Any, Union + +from tests.artifacts.test_run_store import artifact as _artifact +from tests.artifacts.test_run_store import deployment as _deployment +from wf_artifacts.runs.models import PinnedRunEnvironment, RunAdmission +from wf_core import RunState, RunStatus +from wf_scheduling.dispatch import DispatchResult, StillRunning, Stopped + +OutcomeSpec = Union[str, Callable[[RunAdmission, datetime], DispatchResult]] + + +class DictDeployments: + """Deployment directory over a plain ``{id: {"rev": n, "required": [...]}}`` map.""" + + def __init__(self, deployments: dict[str, dict[str, Any]]) -> None: + self._deployments = deployments + + def deployment_revision(self, deployment_id: str) -> int: + return int(self._deployments[deployment_id]["rev"]) + + def required_inputs(self, deployment_id: str) -> Sequence[str]: + return list(self._deployments[deployment_id].get("required", [])) + + +def fixture_environment(sched: Any) -> PinnedRunEnvironment: + """Build the historic fixture environment for schedule tests.""" + deployment = _deployment().model_copy( + update={"id": getattr(sched, "deployment_id", "dep-1")} + ) + return PinnedRunEnvironment( + deployment=deployment, root_artifact=_artifact(), child_artifacts=[] + ) + + +def stopped_state(admission: RunAdmission, outcome: str) -> RunState: + """Build a genuine stopped RunState echoing the admitted input.""" + status = { + "complete": RunStatus.COMPLETED, + "interrupt": RunStatus.INTERRUPTED, + "fail": RunStatus.FAILED, + }[outcome] + return RunState( + workflow_name="sched", + status=status, + workflow_input=dict(admission.resolved_input), + state={}, + ) + + +class ScriptedDispatcher: + """Scripted execution double: ``{run_id | "*": outcome | callable}``. + + Outcomes are ``complete`` | ``interrupt`` | ``fail`` | ``hang``. + Callables receive ``(admission, now)`` and may raise or terminate the + process (crash tests). ``finish`` builds a stopped state for a hanging + run so tests can settle it through the scheduler's async-completion + seam without re-dispatching. + """ + + def __init__(self, script: dict[str, OutcomeSpec] | None = None) -> None: + self.script: dict[str, OutcomeSpec] = dict(script or {}) + + def dispatch(self, *, admission: RunAdmission, now: datetime) -> DispatchResult: + spec = self.script.get(admission.id, self.script.get("*", "complete")) + if callable(spec): + return spec(admission, now) + assert isinstance(spec, str), f"unknown scripted outcome {spec!r}" + if spec == "hang": + return StillRunning() + if spec in ("complete", "interrupt", "fail"): + return Stopped(result=stopped_state(admission, spec)) + raise ValueError(f"unknown scripted outcome {spec!r} for {admission.id!r}") + + def finish(self, admission: RunAdmission, outcome: str) -> RunState: + """Build the stopped state that settles a hanging run.""" + return stopped_state(admission, outcome) diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index aa49b7f5..6ddb68cc 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -4,12 +4,19 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import Any +from typing import Any, cast +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) from wf_artifacts.runs.models import StoredRunStatus from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import OneShotSource from wf_scheduling.models import Schedule from wf_scheduling.poll import SCAN_CAP, Scheduler +from wf_scheduling.prepare import SchedulePreparer from wf_scheduling.store import FileScheduleStore UTC_TZ = UTC @@ -41,21 +48,6 @@ class PeriodicSource: return self.start + n * self.period -class OneShotSource: - def __init__(self, at: datetime) -> None: - self.at = at - self.next_calls = 0 - self.prev_calls = 0 - - def next_after(self, instant: datetime) -> datetime | None: - self.next_calls += 1 - return self.at if instant < self.at else None - - def prev_before(self, instant: datetime) -> datetime | None: - self.prev_calls += 1 - return self.at if instant > self.at else None - - def _sched_model(sid: str, start_hint: str = "cron", **kw: Any) -> Schedule: now = ts(2026, 9, 8, 12, 0) base: dict[str, Any] = { @@ -74,21 +66,27 @@ def _harness( tmp_path: Path, *, capacity: int = 4, - outcomes: dict[str, str] | None = None, + script: dict | None = None, deployments: dict[str, dict] | None = None, ) -> tuple[Scheduler, FileScheduleStore, FileRunStore, dict]: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") sources: dict = {} + preparer = SchedulePreparer( + DictDeployments( + deployments + if deployments is not None + else {"dep-1": {"rev": 1, "required": []}} + ), + fixture_environment, + ) sched = Scheduler( schedule_store=sched_store, run_store=run_store, sources=sources, capacity=capacity, - outcomes=outcomes, - deployments=deployments - if deployments is not None - else {"dep-1": {"rev": 1, "required": []}}, + preparer=preparer, + dispatcher=ScriptedDispatcher(script), ) return sched, sched_store, run_store, sources @@ -119,7 +117,7 @@ def test_overlap_skip_blocks_and_late_drops() -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) - sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"}) + sched, store, runs, sources = _harness(root, capacity=4, script={"*": "hang"}) t0 = ts(2026, 9, 8, 12, 0) _add( sched, @@ -161,7 +159,7 @@ def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None: # Capacity returns at 13:00 while 13:00 is also due: exactly one # admission for 13:00, 12:00 superseded, never both. sched.capacity = 4 - sched.outcomes = {"*": "complete"} + cast(ScriptedDispatcher, sched.dispatcher).script = {"*": "complete"} sched.poll(ts(2026, 9, 8, 13, 0)) admitted = [r for r in _history(store, "h") if r["kind"] == "admitted"] assert len(admitted) == 1 @@ -176,7 +174,7 @@ def test_parallel_limits_and_interrupted_slots() -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) - sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"}) + sched, store, runs, sources = _harness(root, capacity=4, script={"*": "hang"}) t0 = ts(2026, 9, 8, 12, 0) _add( sched, @@ -216,7 +214,7 @@ def test_pause_is_not_downtime() -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) sched, store, runs, sources = _harness( - root, capacity=4, outcomes={"*": "complete"} + root, capacity=4, script={"*": "complete"} ) _add( sched, @@ -244,7 +242,7 @@ def test_pause_is_not_downtime() -> None: def test_long_downtime_is_bounded(tmp_path: Path) -> None: sched, store, runs, sources = _harness( - tmp_path, capacity=4, outcomes={"*": "complete"} + tmp_path, capacity=4, script={"*": "complete"} ) src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) _add(sched, store, sources, "m", src, ts(2023, 9, 8, 12, 0), misfire="latest") @@ -259,7 +257,7 @@ def test_long_downtime_is_bounded(tmp_path: Path) -> None: def test_fairness_slow_schedule_not_starved(tmp_path: Path) -> None: - sched, store, runs, sources = _harness(tmp_path, capacity=1, outcomes={"*": "hang"}) + sched, store, runs, sources = _harness(tmp_path, capacity=1, script={"*": "hang"}) t0 = ts(2026, 9, 8, 12, 0) _add( sched, @@ -288,6 +286,6 @@ def test_capacity_wait_then_expire_for_skip(tmp_path: Path) -> None: assert sched.poll(t0) == {"a": "admit:held-undecided"} assert store.get_candidate("a") is None sched.capacity = 1 - sched.outcomes = {"*": "complete"} + cast(ScriptedDispatcher, sched.dispatcher).script = {"*": "complete"} sched.poll(t0 + timedelta(seconds=30)) assert len([r for r in _history(store, "a") if r["kind"] == "admitted"]) == 1 diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py new file mode 100644 index 00000000..6edbdc58 --- /dev/null +++ b/tests/scheduling/test_preparation.py @@ -0,0 +1,216 @@ +"""Typed invocation preparation over real schedule bindings (R4/F6). + +The scheduler no longer hardcodes fixture input: ``SchedulePreparer`` +resolves the schedule's own ``input_bindings`` against the occurrence +environment through the shared schedule-expression contract, rechecks +the deployment directory, and reports preflight rejections without +inventing a run. +""" + +from __future__ import annotations + +import inspect +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import OneShotSource +from wf_scheduling.dispatch import RunDispatcher +from wf_scheduling.models import Schedule +from wf_scheduling.poll import Scheduler +from wf_scheduling.prepare import ( + InvocationPreparer, + PreparationRejected, + SchedulePreparer, +) +from wf_scheduling.store import FileScheduleStore + + +def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime: + return datetime(y, mo, d, h, mi, tzinfo=UTC) + + +def _sched_model(sid: str, **kw: Any) -> Schedule: + now = ts(2026, 9, 8, 12, 0) + base: dict[str, Any] = { + "id": sid, + "deployment_id": "dep-1", + "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"}, + "input_bindings": [], + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + } + base.update(kw) + return Schedule.model_validate(base) + + +def _scheduler( + tmp_path: Path, script: dict | None = None +) -> tuple[Scheduler, FileScheduleStore, FileRunStore]: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + preparer = SchedulePreparer( + DictDeployments({"dep-1": {"rev": 3, "required": []}}), + fixture_environment, + ) + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={}, + capacity=4, + preparer=preparer, + dispatcher=ScriptedDispatcher(script), + ) + return sched, sched_store, run_store + + +def test_scheduler_requires_typed_collaborators() -> None: + params = inspect.signature(Scheduler.__init__).parameters + assert set(params) == { + "self", + "schedule_store", + "run_store", + "sources", + "capacity", + "preparer", + "dispatcher", + } + + +def test_production_scheduler_has_no_fixture_seams() -> None: + import wf_scheduling.poll as poll_module + + source = inspect.getsource(poll_module) + assert "_test_environment" not in source + assert "self.outcomes" not in source + assert "self.deployments" not in source + assert "FixturePreparer" not in source + assert '"eng"' not in source and "'eng'" not in source + # Collaborators are protocols consumed here, implemented elsewhere. + assert "InvocationPreparer" in source + assert "RunDispatcher" in source + assert isinstance( + poll_module.Scheduler.__init__.__annotations__["preparer"], object + ) + + +def test_preparer_contract_is_a_protocol() -> None: + assert issubclass(InvocationPreparer, object) + assert issubclass(RunDispatcher, object) + + +def test_occurrence_bindings_resolve_into_admission(tmp_path: Path) -> None: + sched, store, runs = _scheduler(tmp_path, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + model = _sched_model( + "b", + input_bindings=[ + { + "target": "team", + "expression": {"kind": "literal", "value": "engineering"}, + }, + { + "target": "report_time", + "expression": {"kind": "occurrence", "field": "scheduled_at"}, + }, + { + "target": "which", + "expression": {"kind": "occurrence", "field": "schedule_id"}, + }, + ], + ) + store.create_schedule(model) + store.save_consumed("b", intended - timedelta(hours=1)) + sched.sources["b"] = OneShotSource(intended) + result = sched.poll(intended) + assert result["b"].startswith("admit:run-") + run_id = result["b"].split(":", 1)[1] + admission = runs.get_admission(run_id) + assert admission.resolved_input["team"] == "engineering" + assert admission.resolved_input["report_time"] == intended.isoformat() + assert admission.resolved_input["which"] == "b" + assert admission.deployment_revision == 3 + assert admission.schedule_revision == 1 + + +def test_unknown_deployment_rejects_without_a_run(tmp_path: Path) -> None: + sched, store, runs = _scheduler(tmp_path) + intended = ts(2026, 9, 8, 12, 0) + store.create_schedule(_sched_model("gone", deployment_id="dep-missing")) + store.save_consumed("gone", intended - timedelta(hours=1)) + sched.sources["gone"] = OneShotSource(intended) + assert sched.poll(intended) == {"gone": "admit:None"} + assert runs.list_runs() == [] + assert runs.list_admissions() == [] + page = store.list_occurrences("gone", limit=100) + kinds = [r["kind"] for r in cast(list[dict[str, Any]], page["occurrences"])] + assert kinds == ["preflight-rejected"] + + +def test_missing_required_input_rejects_without_a_run(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + preparer = SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": ["token"]}}), + fixture_environment, + ) + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={}, + capacity=4, + preparer=preparer, + dispatcher=ScriptedDispatcher({"*": "hang"}), + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule(_sched_model("need")) + sched_store.save_consumed("need", intended - timedelta(hours=1)) + sched.sources["need"] = OneShotSource(intended) + assert sched.poll(intended) == {"need": "admit:None"} + assert run_store.list_runs() == [] + page = sched_store.list_occurrences("need", limit=100) + entry = cast(list[dict[str, Any]], page["occurrences"])[0] + assert entry["kind"] == "preflight-rejected" + assert "missing-input" in entry["reason"] + + +def test_conflicting_schedule_targets_reject_without_a_run(tmp_path: Path) -> None: + sched, store, runs = _scheduler(tmp_path) + intended = ts(2026, 9, 8, 12, 0) + store.create_schedule( + _sched_model( + "conflict", + input_bindings=[ + { + "target": "a", + "expression": {"kind": "literal", "value": 1}, + }, + { + "target": "a.b", + "expression": {"kind": "literal", "value": 2}, + }, + ], + ) + ) + store.save_consumed("conflict", intended - timedelta(hours=1)) + sched.sources["conflict"] = OneShotSource(intended) + assert sched.poll(intended) == {"conflict": "admit:None"} + assert runs.list_runs() == [] + page = store.list_occurrences("conflict", limit=100) + entry = cast(list[dict[str, Any]], page["occurrences"])[0] + assert entry["kind"] == "preflight-rejected" + assert entry["reason"].startswith("invalid-input:") + + +def test_preparer_rejection_type_shape() -> None: + import dataclasses + + rejected = PreparationRejected(reason="deployment-deleted") + assert rejected.reason == "deployment-deleted" + assert dataclasses.is_dataclass(PreparationRejected) diff --git a/tests/scheduling/test_recovery.py b/tests/scheduling/test_recovery.py index c1c293cd..c54b7e21 100644 --- a/tests/scheduling/test_recovery.py +++ b/tests/scheduling/test_recovery.py @@ -6,10 +6,17 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) from wf_artifacts.runs.store import FileRunStore from wf_scheduling import recovery as sched_recovery +from wf_scheduling.calendar import OneShotSource from wf_scheduling.models import Schedule from wf_scheduling.poll import Scheduler +from wf_scheduling.prepare import SchedulePreparer from wf_scheduling.store import FileScheduleStore @@ -92,7 +99,6 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: from tests.artifacts.test_run_store import artifact as _artifact from tests.artifacts.test_run_store import deployment as _deployment - from tests.scheduling.test_poll import OneShotSource from wf_api.run_lifecycle import persist_admission from wf_artifacts import PinnedRunEnvironment @@ -124,7 +130,11 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: run_store=run_store, sources=sources, # type: ignore[arg-type] capacity=4, - outcomes={"*": "complete"}, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=ScriptedDispatcher({"*": "complete"}), ) sched_store.save_consumed("a", ts(2026, 9, 8, 12, 0)) sched.poll(ts(2026, 9, 8, 12, 1))