sched: extract typed preparer/dispatcher seams; test impl out of production (F6)

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: ...

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
+uses the canonical :class:`wf_scheduling.calendar.OccurrenceSource`
-enumeration); latest-missed catch-up is one bounded ``prev_before`` query
+(``next_after``/``prev_before`` only, never enumeration); latest-missed
-(F1). Overlap decisions precede capacity checks; terminal skips never
+catch-up is one bounded ``prev_before`` query (F1). Overlap decisions
-reappear; ``latest`` retains at most one candidate.
+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``
+    return isinstance(source, OneShotSource)
-    # 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")
 
 
 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`` |
+    Invocation preparation (:class:`wf_scheduling.prepare.InvocationPreparer`)
-    ``interrupt`` for deterministic tests. Production dispatch (real runtime)
+    and execution (:class:`wf_scheduling.dispatch.RunDispatcher`) are typed
-    is wired in T12; here ``hang`` keeps an admitted run occupying its
+    collaborators: production code here never fabricates input, environments,
-    schedule slot, ``interrupt`` marks it interrupted (schedule slot only),
+    or outcomes. Stopped results are persisted through the shared
-    and ``complete`` marks it completed (releases overlap).
+    ``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,
+        preparer: InvocationPreparer,
-        deployments: dict[str, dict] | None = None,
+        dispatcher: RunDispatcher,
     ) -> None:
         self.schedule_store = schedule_store
         self.run_store = run_store
         self.sources = sources
         self.capacity = capacity
-        self.outcomes = outcomes or {}
+        self.preparer = preparer
-        self.deployments = deployments or {"dep-1": {"rev": 1, "required": []}}
+        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)
+        prepared = self.preparer.prepare(sched=sched, intended=intended, now=now)
-        if dep is None:
+        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(
+        admission = persist_admission(
-            id=run_id,
+            store=self.run_store,
-            environment=_test_environment(sched),
+            run_id=run_id,
-            resolved_input=dict(frozen),
+            environment=prepared.environment,
-            max_steps=getattr(sched, "max_steps", None),
+            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"],
+            deployment_revision=prepared.deployment_revision,
-            created_at=now,
         )
-        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
         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
         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)
+        try:
-        if sched_id is None:
+            admission = self.run_store.get_admission(run_id)
-            raise BlockedSchedule(f"dispatch missing admission: {run_id!r}")
+        except KeyError as exc:
-        if outcome == "hang":
+            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":
+        stopped = persist_stopped_run(
-            updated = record.model_copy(
+            store=self.run_store,
-                update={"status": StoredRunStatus.INTERRUPTED, "updated_at": now}
+            environment=admission.environment,
-            )
+            run=result.result,
-            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
+        kind = {
-        updated = record.model_copy(
+            StoredRunStatus.COMPLETED: "completed",
-            update={"status": StoredRunStatus.COMPLETED, "updated_at": now}
+            StoredRunStatus.INTERRUPTED: "interrupted",
-        )
+            StoredRunStatus.FAILED: "failed",
-        self.run_store.save_run(updated)
+        }[stopped.status]
         self._record(
-            kind="completed",
+            kind=kind,  # type: ignore[arg-type]
-            sched_id=sched_id,
+            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

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,
+        )

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

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)

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,
+        preparer=preparer,
-        deployments=deployments
+        dispatcher=ScriptedDispatcher(script),
-        if deployments is not None
-        else {"dep-1": {"rev": 1, "required": []}},
     )
     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

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)

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))