sched: bind ownership to store composition; stable recovery failure; schema-checked prepare; guarded settle (R4 wave 2)

src/wf_scheduling/ownership.py

@@ -9,11 +9,18 @@ startup is rejected instead of running unprotected.
 
 Locking design (Windows-tested, per the store transaction boundary): one
 new module holding ``msvcrt.locking`` (Windows) / ``fcntl.flock`` (POSIX)
-on ``<store root>/scheduler.lock``. No ad-hoc per-run lock files.
+on ``<composition root>/scheduler.lock``. No ad-hoc per-run lock files.
+
+The lock is acquired on the composition root that contains the protected
+schedule and run stores. Entry-point guards do not trust the caller to
+have passed the right lock: they validate the held guard against the
+actual store roots (canonicalized for platform aliases) before any write
+or side effect.
 """
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import BinaryIO
 
@@ -26,8 +33,31 @@ class StartupRejected(Exception):
     """Scheduler startup rejected where locking is unsupported."""
 
 
+def canonical_store_path(path: Path | str) -> str:
+    """Canonicalize a store path for ownership-coverage comparison.
+
+    ``realpath`` resolves symlinks, junctions, ``..`` segments, and
+    relative paths; ``normcase`` folds case and separators on Windows
+    (identity on POSIX, which stays case-sensitive). Both sides of every
+    comparison pass through this function, so platform aliases of one
+    directory compare equal. Exotic aliases (subst drives, UNC-vs-mapped
+    paths, 8.3 short names where the OS preserves them) are out of scope:
+    operators should acquire the lock on the same canonical composition
+    root the stores live under.
+    """
+    return os.path.normcase(os.path.realpath(os.fspath(path)))
+
+
+def _is_within(candidate: str, base: str) -> bool:
+    """Whether canonical ``candidate`` lies beneath canonical ``base``."""
+    if candidate == base:
+        return True
+    prefix = base if base.endswith(os.sep) else base + os.sep
+    return candidate.startswith(prefix)
+
+
 class SchedulerOwnership:
-    """Held exclusive ownership of a schedule store root."""
+    """Held exclusive ownership of a store composition root."""
 
     def __init__(self, root: Path, *, owner: str) -> None:
         self.root = root
@@ -49,6 +79,27 @@ class SchedulerOwnership:
         """
         return self._locked
 
+    def canonical_root(self) -> str:
+        """Return the canonical composition root this lock was acquired on."""
+        return canonical_store_path(self.root)
+
+    def covers(self, *store_roots: Path | str | None) -> bool:
+        """Whether this held lock protects the given store roots.
+
+        Coverage means each root canonicalizes to the lock's composition
+        root or beneath it. An unrelated (even held) lock, a released lock,
+        and rootless stores are never covered: guards reject those before
+        any write or side effect instead of trusting the caller's promise.
+        """
+        if not self._locked or not store_roots:
+            return False
+        base = self.canonical_root()
+        return all(
+            store_root is not None
+            and _is_within(canonical_store_path(store_root), base)
+            for store_root in store_roots
+        )
+
     def acquire(self) -> SchedulerOwnership:
         """Acquire the held lock non-blockingly or raise SecondOwnerError."""
         self.lock_path.parent.mkdir(parents=True, exist_ok=True)

src/wf_scheduling/poll.py

@@ -139,13 +139,20 @@ class Scheduler:
     def _require_ownership(self) -> None:
         """Reject schedule mutation/dispatch without proven live ownership.
 
-        Runs before any store write or dispatcher side effect: without a
-        held lock this process cannot prove exclusive ownership, so polling
+        Runs before any store write or dispatcher side effect: the held
+        lock must cover the actual schedule and run store composition, not
+        merely be held on some unrelated directory. Without covering
+        ownership this process cannot prove exclusive ownership, so polling
         or administering schedules would risk double admission.
         """
-        if self.ownership is None or not self.ownership.held:
+        ownership = self.ownership
+        if ownership is None or not ownership.covers(
+            getattr(self.schedule_store, "root", None),
+            getattr(self.run_store, "root", None),
+        ):
             raise SecondOwnerError(
-                "scheduler ownership is required before polling or mutating schedules"
+                "scheduler ownership must cover the schedule and run stores "
+                "before polling or mutating schedules"
             )
 
     # -- helpers ------------------------------------------------------
@@ -431,6 +438,12 @@ class Scheduler:
         :class:`wf_core.RunState`. Persists through the shared lifecycle
         boundary, clears the executing mark, and records terminal history.
         Never re-invokes the dispatcher.
+
+        An admitted status alone is not enough: pending, never-dispatched
+        work is also admitted. Settlement requires the durable executing
+        transition and refuses contradictory pending/executing state (a
+        crash between the transition writes owns that run now, not this
+        caller); it also refuses runs that already stopped.
         """
         self._require_ownership()
         from wf_api.run_lifecycle import persist_stopped_run
@@ -448,6 +461,12 @@ class Scheduler:
             raise BlockedSchedule(f"settle missing run view: {run_id!r}") from exc
         if self._status_value(record) != "admitted":
             raise BlockedSchedule(f"settle non-admitted run: {run_id!r}")
+        if self.run_store.is_pending_dispatch(run_id):
+            raise BlockedSchedule(
+                f"settle contradictory pending/executing run: {run_id!r}"
+            )
+        if not self.run_store.is_executing(run_id):
+            raise BlockedSchedule(f"settle without executing transition: {run_id!r}")
         stopped = persist_stopped_run(
             store=self.run_store,
             environment=admission.environment,

src/wf_scheduling/prepare.py

@@ -17,7 +17,9 @@ from datetime import datetime
 from typing import Any, Protocol
 
 from wf_artifacts.runs.models import PinnedRunEnvironment
+from wf_core.errors import WorkflowExecutionError
 from wf_core.runtime.input_sources import resolve_schedule_input_bindings
+from wf_core.runtime.ops.schemas import validate_payload_against_schema
 from wf_scheduling.occurrences import occurrence_id
 
 
@@ -63,7 +65,14 @@ class SchedulePreparer:
     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.
+    run identity is allocated. The resolved object is additionally
+    validated against the pinned root artifact's input schema with the same
+    validator the runtime uses for fresh runs, so a schedule cannot admit
+    work the deployment contract would refuse.
+
+    The environment is built once per preparation: the deployment revision,
+    required inputs, and input schema all describe that single captured
+    snapshot, never independently resolved versions.
     """
 
     def __init__(
@@ -77,9 +86,12 @@ class SchedulePreparer:
     def prepare(
         self, *, sched: Any, intended: datetime, now: datetime
     ) -> PreparedInvocation | PreparationRejected:
+        environment = self._build_environment(sched)
         try:
-            revision = self._deployments.deployment_revision(sched.deployment_id)
-            required = list(self._deployments.required_inputs(sched.deployment_id))
+            revision = self._deployments.deployment_revision(environment.deployment.id)
+            required = list(
+                self._deployments.required_inputs(environment.deployment.id)
+            )
         except KeyError:
             return PreparationRejected(reason="deployment-deleted")
         occurrence = {
@@ -98,8 +110,16 @@ class SchedulePreparer:
         missing = [key for key in required if key not in resolved]
         if missing:
             return PreparationRejected(reason=f"missing-input:{missing}")
+        try:
+            validate_payload_against_schema(
+                environment.root_artifact.input_schema,
+                dict(resolved),
+                f"schedule {sched.id} input",
+            )
+        except WorkflowExecutionError as exc:
+            return PreparationRejected(reason=f"invalid-input:schema:{exc}")
         return PreparedInvocation(
-            environment=self._build_environment(sched),
+            environment=environment,
             resolved_input=dict(resolved),
             max_steps=getattr(sched, "max_steps", None),
             deployment_revision=revision,

src/wf_scheduling/recovery.py

@@ -31,6 +31,13 @@ clear → executor → stopped persist → executing clear → history):
   reconciliation across the torn ``save_checkpoint``/``save_run`` boundary
   is owned by the F2 reconcile step, which runs before marker handling.
 - stopped summary, no marks: existing terminal/attempt reconciliation.
+- failed summary: a durable recovery decision is never re-processed. When
+  the checkpoint pointer still matches the decided state, history is
+  ensured without touching status, readiness, or diagnostics (repeat
+  recovery is silent). Only a genuinely newer checkpoint (pointer
+  mismatch) may reopen the run via the reconcile path; an older
+  checkpoint can never supersede the abandonment. Failed runs carry
+  not-applicable resume readiness.
 """
 
 from __future__ import annotations
@@ -72,10 +79,14 @@ def recover(
     """Reconcile durable state after a restart without executing work."""
     from wf_artifacts.runs.models import StoredRunStatus
 
-    if ownership is None or not ownership.held:
+    if ownership is None or not ownership.covers(
+        getattr(schedule_store, "root", None),
+        getattr(run_store, "root", None),
+    ):
         raise SecondOwnerError(
-            "scheduler ownership is required before recovery: an unowned "
-            "recovery could abandon or redispatch another owner's work"
+            "scheduler ownership must cover the schedule and run stores "
+            "before recovery: an unowned recovery could abandon or "
+            "redispatch another owner's work"
         )
     if history is None:
         history = FileScheduleHistoryRecorder(schedule_store)
@@ -96,6 +107,27 @@ def recover(
         status = getattr(run.status, "value", run.status)
         attempt = run_store.get_resume_attempt(run.id)
         active = attempt is not None and attempt.state == "ACTIVE"
+        if status in (
+            StoredRunStatus.INTERRUPTED.value,
+            StoredRunStatus.COMPLETED.value,
+            StoredRunStatus.FAILED.value,
+            "interrupted",
+            "completed",
+            "failed",
+        ) and (is_executing(run_store, run.id) or _is_pending(run_store, run.id)):
+            # Completion-window leftovers: the stopped result was persisted
+            # but marker clearing was lost. The stopped status stands;
+            # nothing is re-executed.
+            clear_executing(run_store, run.id)
+            clear_pending(run_store, run.id)
+            diags.append(f"{run.id}:completion-window-cleared")
+        # Durable-decision stability: a FAILED summary for the current
+        # checkpoint state is never re-processed. Only a genuinely newer
+        # checkpoint (pointer mismatch) may reopen it via the reconcile
+        # path below. History is ensured without mutating the summary.
+        if status == StoredRunStatus.FAILED.value or status == "failed":
+            if _failed_decision_stable(run_store, run, history, now, diags):
+                continue
         # Checkpoint-first reconcile across the torn
         # save_checkpoint/save_run boundary: a durable stopped checkpoint
         # wins over a stale summary (status, checkpoint pointer, and
@@ -105,6 +137,25 @@ def recover(
         try:
             changed, status = _reconcile_summary_from_checkpoint(run_store, run, now)
         except (ValueError, OSError) as exc:
+            if _has_recovery_decision(run):
+                # A decided run stays decided even when checkpoint state
+                # corrupts afterwards: ensure history from the summary
+                # pointer, note the corruption, never re-fail.
+                sched_id, intended, revision = _attribution(run_store, run.id)
+                if _reconcile_terminal(
+                    history,
+                    sched_id,
+                    run.id,
+                    "failed",
+                    run.latest_checkpoint_id,
+                    intended,
+                    revision,
+                    _decision_reason(run) or "reconciled-on-recovery",
+                    now,
+                ):
+                    diags.append(f"{run.id}:terminal-reconciled")
+                diags.append(f"{run.id}:corrupt-checkpoint-noted")
+                continue
             _fail_run(run_store, run, f"corrupt checkpoint: {exc}", now, history)
             diags.append(f"{run.id}:failed-closed")
             continue
@@ -119,9 +170,8 @@ def recover(
             "completed",
             "failed",
         ) and (is_executing(run_store, run.id) or _is_pending(run_store, run.id)):
-            # Completion-window leftovers: the stopped result was persisted
-            # but marker clearing was lost. The stopped status stands;
-            # nothing is re-executed.
+            # A repaired summary may carry completion-window leftovers (the
+            # stopped result was persisted but marker clearing was lost).
             clear_executing(run_store, run.id)
             clear_pending(run_store, run.id)
             diags.append(f"{run.id}:completion-window-cleared")
@@ -277,6 +327,67 @@ def _reconcile_summary_from_checkpoint(
     return True, expected.value
 
 
+def _has_recovery_decision(run: Any) -> bool:
+    """Whether the summary carries a durable recovery-failure decision."""
+    return any(
+        getattr(item, "code", None) == "schedule-recovery" for item in run.diagnostics
+    )
+
+
+def _decision_reason(run: Any) -> str:
+    """Return the first durable recovery-failure reason, if any."""
+    for item in run.diagnostics:
+        if getattr(item, "code", None) == "schedule-recovery":
+            return str(getattr(item, "message", ""))
+    return ""
+
+
+def _failed_decision_stable(
+    run_store: Any,
+    run: Any,
+    history: HistoryRecorder,
+    now: datetime,
+    diags: list[str],
+) -> bool:
+    """Leave a decided FAILED summary untouched, ensuring history only.
+
+    Returns True when the run needs no further processing this pass: the
+    summary already reflects a recovery decision (or a legacy/external
+    failure) for the current checkpoint state, so re-running reconcile or
+    re-failing would only duplicate diagnostics and rewrite readiness. A
+    genuinely newer checkpoint (pointer mismatch) returns False so the
+    reconcile path may repair the summary. Unreadable checkpoint state
+    likewise returns False: the reconcile path fails closed for undecided
+    runs and keeps decided ones via its own branch.
+    """
+    try:
+        latest = run_store.get_latest_checkpoint(run.id)
+    except KeyError:
+        latest = None
+    except ValueError, OSError:
+        return False
+    pointer = run.latest_checkpoint_id
+    latest_id = latest.id if latest is not None else None
+    if latest_id != pointer:
+        # Genuinely newer (or torn-away) checkpoint state: not stable.
+        return False
+    sched_id, intended, revision = _attribution(run_store, run.id)
+    reason = _decision_reason(run) or "reconciled-on-recovery"
+    if _reconcile_terminal(
+        history,
+        sched_id,
+        run.id,
+        "failed",
+        pointer,
+        intended,
+        revision,
+        reason,
+        now,
+    ):
+        diags.append(f"{run.id}:terminal-reconciled")
+    return True
+
+
 def _attribution(
     run_store: Any, run_id: str
 ) -> tuple[str | None, datetime | None, int | None]:
@@ -366,12 +477,13 @@ def _fail_run(
 ) -> None:
     """Fail a run closed with its truthful reason durably preserved.
 
-    The reason is always written into the run summary diagnostics; when an
-    admission attributes the run to a schedule, a failed occurrence entry is
-    reconciled exactly once through the history recorder.
+    The reason is always written into the run summary diagnostics, and a
+    failed run is never resumable. When an admission attributes the run to
+    a schedule, a failed occurrence entry is reconciled exactly once
+    through the history recorder.
     """
     from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
-    from wf_artifacts.runs.models import StoredRunStatus
+    from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus
 
     diagnostic = DependencyDiagnostic(
         severity=DiagnosticSeverity.ERROR,
@@ -383,6 +495,7 @@ def _fail_run(
     updated = run.model_copy(
         update={
             "status": StoredRunStatus.FAILED,
+            "resume_readiness": ResumeReadiness.NOT_APPLICABLE,
             "updated_at": now,
             "diagnostics": [*run.diagnostics, diagnostic],
         }

tests/scheduling/test_dispatch_transition.py

@@ -83,7 +83,7 @@ def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None
 
 
 def test_executing_is_set_before_executor_runs(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         observed: dict[str, bool] = {}
 
@@ -128,7 +128,7 @@ def test_failure_before_transition_never_dispatches(tmp_path: Path) -> None:
 
     sched_store = FileScheduleStore(tmp_path / "sched")
     run_store = FailMarkOnce(tmp_path / "runs")
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = Scheduler(
             schedule_store=sched_store,
@@ -160,7 +160,7 @@ def test_failure_before_transition_never_dispatches(tmp_path: Path) -> None:
 
 def test_capacity_shortage_keeps_pending_undispatched(tmp_path: Path) -> None:
     calls: list[str] = []
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _harness(tmp_path, ownership, capacity=0)
         intended = ts(2026, 9, 8, 12, 0)
@@ -185,7 +185,7 @@ def test_capacity_shortage_keeps_pending_undispatched(tmp_path: Path) -> None:
 
 def test_pending_on_terminal_run_clears_without_redispatch(tmp_path: Path) -> None:
     calls: list[str] = []
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _harness(tmp_path, ownership)
 
@@ -230,7 +230,7 @@ def test_pending_without_admission_fails_closed(tmp_path: Path) -> None:
         WorkflowRunRecord,
     )
 
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
         intended = ts(2026, 9, 8, 12, 0)
@@ -260,7 +260,7 @@ def test_pending_without_admission_fails_closed(tmp_path: Path) -> None:
 
 
 def test_executing_admitted_is_abandoned_by_recovery(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
         intended = ts(2026, 9, 8, 12, 0)
@@ -289,7 +289,7 @@ def test_executing_admitted_is_abandoned_by_recovery(tmp_path: Path) -> None:
 
 
 def test_settle_hanging_run_persists_and_clears(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
         intended = ts(2026, 9, 8, 12, 0)
@@ -350,7 +350,7 @@ def _child_poll_and_die(
         capacity=4,
         preparer=_Prep(_Dict({"dep-1": {"rev": 1, "required": []}}), _env),
         dispatcher=_SD({"*": killer}),
-        ownership=_Own(base / "sched", owner="child").acquire(),
+        ownership=_Own(base, owner="child").acquire(),
     )
     sched.poll(intended)
 
@@ -376,7 +376,7 @@ def test_subprocess_death_after_side_effect_dispatches_once(tmp_path: Path) -> N
 
     # Restart with fresh objects: the side effect count stays one and the run
     # is abandoned, never redispatched.
-    ownership = SchedulerOwnership(root / "sched", owner="parent").acquire()
+    ownership = SchedulerOwnership(root, owner="parent").acquire()
     try:
         sched, store, runs = _harness(root, ownership, script={"*": "hang"})
         sched.sources["a"] = OneShotSource(intended)

tests/scheduling/test_history_reconcile.py

@@ -124,7 +124,7 @@ def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, An
 def _recover(
     sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime
 ) -> list[str]:
-    ownership = SchedulerOwnership(sched_store.root, owner="test").acquire()
+    ownership = SchedulerOwnership(sched_store.root.parent, owner="test").acquire()
     try:
         return sched_recovery.recover(
             schedule_store=sched_store,
@@ -223,7 +223,7 @@ def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> No
 
     sched_store = FailCompletedHistoryOnce(tmp_path / "sched")
     run_store = FileRunStore(tmp_path / "runs")
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = Scheduler(
             schedule_store=sched_store,
@@ -271,7 +271,7 @@ def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> No
 
 
 def test_dispatch_and_recovery_share_entry_identity(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched_store = FileScheduleStore(tmp_path / "sched")
         run_store = FileRunStore(tmp_path / "runs")
@@ -383,7 +383,7 @@ def test_admission_tear_returns_existing_run_without_dup(tmp_path: Path) -> None
     )
     materialize_admitted_view(store=run_store, admission=admission)
     sched_store.save_consumed("a", intended - timedelta(hours=1))
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = _owned_scheduler(
             sched_store, run_store, ownership, script={"*": "hang"}
@@ -426,7 +426,7 @@ def test_parallel_tear_never_double_admits(tmp_path: Path) -> None:
     )
     materialize_admitted_view(store=run_store, admission=admission)
     sched_store.save_consumed("a", intended - timedelta(hours=1))
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = _owned_scheduler(
             sched_store, run_store, ownership, script={"*": "hang"}
@@ -460,7 +460,7 @@ def test_consumed_write_failure_recovers_without_dup(tmp_path: Path) -> None:
     intended = ts(2026, 9, 8, 12, 0)
     sched_store.save_consumed("a", intended - timedelta(hours=1))
     sched_store.armed = True  # arm only the poll's watermark write
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = _owned_scheduler(
             sched_store, run_store, ownership, script={"*": "hang"}

tests/scheduling/test_latest_boundary.py

@@ -98,7 +98,7 @@ def _admitted(store: FileScheduleStore, sid: str) -> list[dict[str, Any]]:
 
 
 def test_exact_boundary_admits_latest_once(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "hang"}
@@ -120,7 +120,7 @@ def test_exact_boundary_admits_latest_once(tmp_path: Path) -> None:
 
 
 def test_second_poll_at_same_instant_admits_nothing(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "hang"}
@@ -140,7 +140,7 @@ def test_second_poll_at_same_instant_admits_nothing(tmp_path: Path) -> None:
 
 
 def test_between_ticks_admits_latest(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "hang"}
@@ -163,7 +163,7 @@ def test_between_ticks_admits_latest(tmp_path: Path) -> None:
 
 def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None:
     now = ts(2026, 9, 8, 12, 0)
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, _, sources = _harness(tmp_path, ownership, script={"*": "hang"})
         store.create_schedule(
@@ -174,7 +174,7 @@ def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None:
         sched.poll(now)
     finally:
         ownership.release()
-    ownership2 = SchedulerOwnership(tmp_path / "sched", owner="test2").acquire()
+    ownership2 = SchedulerOwnership(tmp_path, owner="test2").acquire()
     try:
         sched2, store2, runs2, sources2 = _harness(
             tmp_path, ownership2, script={"*": "hang"}
@@ -188,7 +188,7 @@ def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None:
 
 
 def test_long_catchup_stays_bounded(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "complete"}
@@ -208,7 +208,7 @@ def test_long_catchup_stays_bounded(tmp_path: Path) -> None:
 def test_skip_policy_exact_boundary_spans_without_admission(
     tmp_path: Path,
 ) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "hang"}
@@ -229,7 +229,7 @@ def test_skip_policy_exact_boundary_spans_without_admission(
 
 
 def test_overlap_skip_exact_boundary_skips_once(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs, sources = _harness(
             tmp_path, ownership, script={"*": "hang"}

tests/scheduling/test_ownership_binding.py

@@ -0,0 +1,179 @@
+"""Ownership is bound to the protected store composition (R4 binding).
+
+A held lock on an unrelated directory must not authorize mutation of a
+store composition it does not cover. The guard validates the held guard
+against the actual schedule/run store roots (canonicalized for platform
+aliases) instead of trusting the caller to have passed the right lock.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+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 StillRunning
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import 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(
+    sched_store: FileScheduleStore,
+    run_store: FileRunStore,
+    ownership: SchedulerOwnership,
+    *,
+    script: dict | None = None,
+) -> Scheduler:
+    return Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={},
+        capacity=4,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+            fixture_environment,
+        ),
+        dispatcher=ScriptedDispatcher(script),
+        ownership=ownership,
+    )
+
+
+def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None:
+    store.create_schedule(_sched_model("a"))
+    store.save_consumed("a", intended - timedelta(hours=1))
+    sched.sources["a"] = OneShotSource(intended)
+
+
+def test_unrelated_held_lock_rejected_before_writes(tmp_path: Path) -> None:
+    calls: list[str] = []
+
+    def spy(admission: Any, now: datetime) -> Any:
+        calls.append(admission.id)
+        return StillRunning()
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    ownership = SchedulerOwnership(tmp_path / "other", owner="unrelated").acquire()
+    try:
+        sched = _scheduler(sched_store, run_store, ownership)
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, sched_store, intended)
+        with pytest.raises(SecondOwnerError):
+            sched.poll(intended)
+        assert calls == []
+        assert run_store.list_runs() == []
+        assert run_store.list_admissions() == []
+        assert sched_store.list_occurrences("a", limit=100)["total"] == 0
+    finally:
+        ownership.release()
+
+
+def test_two_lock_dirs_cannot_operate_same_stores(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    intended = ts(2026, 9, 8, 12, 0)
+    covering = SchedulerOwnership(tmp_path, owner="covering").acquire()
+    other = SchedulerOwnership(tmp_path / "other", owner="other").acquire()
+    try:
+        first = _scheduler(sched_store, run_store, covering, script={"*": "hang"})
+        _due(first, sched_store, intended)
+        assert first.poll(intended)["a"].startswith("admit:run-")
+        # The other directory's lock acquires fine (different lock file) but
+        # must not authorize the same stores.
+        second = _scheduler(sched_store, run_store, other, script={"*": "hang"})
+        second.sources["a"] = OneShotSource(intended)
+        with pytest.raises(SecondOwnerError):
+            second.poll(intended + timedelta(minutes=1))
+        assert len(run_store.list_runs()) == 1
+    finally:
+        covering.release()
+        other.release()
+
+
+def test_recover_with_unrelated_lock_rejected(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 wf_api.run_lifecycle import materialize_admitted_view, persist_admission
+    from wf_artifacts import PinnedRunEnvironment
+    from wf_scheduling import recovery as sched_recovery
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_store.allocate_run_id(),
+        environment=PinnedRunEnvironment(
+            deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[]
+        ),
+        resolved_input={},
+        max_steps=None,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    ownership = SchedulerOwnership(tmp_path / "other", owner="unrelated").acquire()
+    try:
+        with pytest.raises(SecondOwnerError):
+            sched_recovery.recover(
+                schedule_store=sched_store,
+                run_store=run_store,
+                now=ts(2026, 9, 8, 12, 0),
+                ownership=ownership,
+            )
+        assert run_store.get_run(admission.id).status.value == "admitted"
+    finally:
+        ownership.release()
+
+
+def test_canonicalizer_handles_aliases(tmp_path: Path) -> None:
+    from wf_scheduling.ownership import canonical_store_path
+
+    base = tmp_path / "sched"
+    dotted = tmp_path / "sub" / ".." / "sched"
+    assert canonical_store_path(dotted) == canonical_store_path(base)
+    if os.name == "nt":
+        assert canonical_store_path("C:\\Temp\\SCHED") == canonical_store_path(
+            "c:/temp/sched"
+        )
+    else:
+        assert canonical_store_path(dotted) != canonical_store_path(tmp_path / "other")
+
+
+def test_covers_rejects_empty_root_set(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        assert ownership.held
+        assert ownership.covers() is False
+        assert ownership.covers(tmp_path) is True
+    finally:
+        ownership.release()

tests/scheduling/test_ownership_guard.py

@@ -83,7 +83,7 @@ def test_poll_without_ownership_rejects_before_writes(tmp_path: Path) -> None:
         calls.append(admission.id)
         raise AssertionError("dispatcher must not run without ownership")
 
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="never-acquired")
+    ownership = SchedulerOwnership(tmp_path, owner="never-acquired")
     sched, store, runs = _scheduler(tmp_path, ownership)
     sched.dispatcher = ScriptedDispatcher({"*": spy})
     intended = ts(2026, 9, 8, 12, 0)
@@ -98,7 +98,7 @@ def test_poll_without_ownership_rejects_before_writes(tmp_path: Path) -> None:
 
 
 def test_poll_with_released_ownership_rejects(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     sched, store, runs = _scheduler(tmp_path, ownership)
     intended = ts(2026, 9, 8, 12, 0)
     _due_setup(sched, store, intended)
@@ -133,7 +133,7 @@ def test_recover_without_ownership_rejects_before_writes(tmp_path: Path) -> None
         schedule_revision=1,
     )
     materialize_admitted_view(store=run_store, admission=admission)
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="never-acquired")
+    ownership = SchedulerOwnership(tmp_path, owner="never-acquired")
     with pytest.raises(SecondOwnerError):
         sched_recovery.recover(
             schedule_store=sched_store,
@@ -146,7 +146,7 @@ def test_recover_without_ownership_rejects_before_writes(tmp_path: Path) -> None
 
 
 def test_held_ownership_allows_poll_and_recovery(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _scheduler(tmp_path, ownership, script={"*": "hang"})
         intended = ts(2026, 9, 8, 12, 0)
@@ -164,9 +164,9 @@ def test_held_ownership_allows_poll_and_recovery(tmp_path: Path) -> None:
 
 
 def test_second_owner_cannot_acquire_for_poll(tmp_path: Path) -> None:
-    first = SchedulerOwnership(tmp_path / "sched", owner="first").acquire()
+    first = SchedulerOwnership(tmp_path, owner="first").acquire()
     try:
-        second = SchedulerOwnership(tmp_path / "sched", owner="second")
+        second = SchedulerOwnership(tmp_path, owner="second")
         with pytest.raises(SecondOwnerError):
             second.acquire()
         assert not second.held
@@ -181,7 +181,7 @@ def test_second_owner_cannot_acquire_for_poll(tmp_path: Path) -> None:
 
 
 def test_admin_mutations_require_ownership(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test")
+    ownership = SchedulerOwnership(tmp_path, owner="test")
     sched, store, _ = _scheduler(tmp_path, ownership)
     store.create_schedule(_sched_model("a", paused=True))
     with pytest.raises(SecondOwnerError):
@@ -193,7 +193,7 @@ def test_admin_mutations_require_ownership(tmp_path: Path) -> None:
 
 
 def test_settle_requires_ownership(tmp_path: Path) -> None:
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched, store, runs = _scheduler(tmp_path, ownership, script={"*": "hang"})
         intended = ts(2026, 9, 8, 12, 0)

tests/scheduling/test_poll.py

@@ -88,7 +88,7 @@ def _harness(
         capacity=capacity,
         preparer=preparer,
         dispatcher=ScriptedDispatcher(script),
-        ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(),
+        ownership=SchedulerOwnership(tmp_path, owner="test").acquire(),
     )
     return sched, sched_store, run_store, sources
 

tests/scheduling/test_preparation.py

@@ -67,7 +67,7 @@ def _scheduler(
         capacity=4,
         preparer=preparer,
         dispatcher=ScriptedDispatcher(script),
-        ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(),
+        ownership=SchedulerOwnership(tmp_path, owner="test").acquire(),
     )
     return sched, sched_store, run_store
 
@@ -173,7 +173,7 @@ def test_missing_required_input_rejects_without_a_run(tmp_path: Path) -> None:
         capacity=4,
         preparer=preparer,
         dispatcher=ScriptedDispatcher({"*": "hang"}),
-        ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(),
+        ownership=SchedulerOwnership(tmp_path, owner="test").acquire(),
     )
     intended = ts(2026, 9, 8, 12, 0)
     sched_store.create_schedule(_sched_model("need"))

tests/scheduling/test_preparation_schema.py

@@ -0,0 +1,298 @@
+"""Resolved input validates against the pinned workflow schema (R4 item 3).
+
+Preparation checks more than required key names: the resolved object must
+satisfy the pinned root artifact's input schema (same snapshot that is
+captured in the admission) before any run identity is allocated. Invalid
+input rejects the occurrence without admission, run, or dispatch.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+from tests.artifacts.test_run_store import artifact as _artifact
+from tests.artifacts.test_run_store import deployment as _deployment
+from tests.scheduling.controlled import DictDeployments, ScriptedDispatcher
+from wf_artifacts.runs.models import PinnedRunEnvironment
+from wf_artifacts.runs.store import FileRunStore
+from wf_scheduling.calendar import OneShotSource
+from wf_scheduling.dispatch import StillRunning
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import 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)
+
+
+STRICT_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "properties": {
+        "count": {"type": "integer"},
+        "nested": {
+            "type": "object",
+            "properties": {"name": {"type": "string"}},
+            "required": ["name"],
+            "additionalProperties": False,
+        },
+    },
+    "required": ["count"],
+    "additionalProperties": False,
+}
+
+
+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 _strict_env(sched: Any) -> PinnedRunEnvironment:
+    artifact = _artifact().model_copy(
+        update={"input_schema": dict(STRICT_SCHEMA)},
+    )
+    deployment = _deployment().model_copy(update={"id": sched.deployment_id})
+    return PinnedRunEnvironment(
+        deployment=deployment, root_artifact=artifact, child_artifacts=[]
+    )
+
+
+def _scheduler(
+    tmp_path: Path,
+    ownership: SchedulerOwnership,
+    bindings: list[dict[str, Any]],
+    *,
+    script: dict | None = None,
+) -> tuple[Scheduler, FileScheduleStore, FileRunStore, list[str]]:
+    calls: list[str] = []
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={},
+        capacity=4,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+            _strict_env,
+        ),
+        dispatcher=ScriptedDispatcher(script),
+        ownership=ownership,
+    )
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.create_schedule(_sched_model("s", input_bindings=bindings))
+    sched_store.save_consumed("s", intended - timedelta(hours=1))
+    sched.sources["s"] = OneShotSource(intended)
+    dispatches = cast(ScriptedDispatcher, sched.dispatcher)
+    outcome = (script or {}).get("*", "complete")
+    assert isinstance(outcome, str)
+
+    def spy(admission: Any, now: datetime) -> Any:
+        calls.append(admission.id)
+        if outcome == "hang":
+            return StillRunning()
+        from wf_scheduling.dispatch import Stopped
+
+        return Stopped(result=dispatches.finish(admission, outcome))
+
+    dispatches.script = {"*": spy}
+    return sched, sched_store, run_store, calls
+
+
+def _rejected_entry(store: FileScheduleStore) -> dict[str, Any]:
+    page = store.list_occurrences("s", limit=100)
+    rows = cast(list[dict[str, Any]], page["occurrences"])
+    assert len(rows) == 1
+    assert rows[0]["kind"] == "preflight-rejected"
+    return rows[0]
+
+
+def test_wrong_scalar_type_rejects_without_admission(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs, calls = _scheduler(
+            tmp_path,
+            ownership,
+            [
+                {
+                    "target": "count",
+                    "expression": {
+                        "kind": "literal",
+                        "value": "definitely not integer",
+                    },
+                }
+            ],
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        assert sched.poll(intended) == {"s": "admit:None"}
+        entry = _rejected_entry(store)
+        assert "invalid-input" in entry["reason"]
+        assert runs.list_runs() == []
+        assert runs.list_admissions() == []
+        assert calls == []
+    finally:
+        ownership.release()
+
+
+def test_nested_constraint_violation_rejects(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs, calls = _scheduler(
+            tmp_path,
+            ownership,
+            [
+                {"target": "count", "expression": {"kind": "literal", "value": 3}},
+                {
+                    "target": "nested",
+                    "expression": {
+                        "kind": "object",
+                        "fields": {"name": {"kind": "literal", "value": 7}},
+                    },
+                },
+            ],
+        )
+        assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"}
+        _rejected_entry(store)
+        assert runs.list_runs() == []
+        assert calls == []
+    finally:
+        ownership.release()
+
+
+def test_disallowed_extra_property_rejects(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs, calls = _scheduler(
+            tmp_path,
+            ownership,
+            [
+                {"target": "count", "expression": {"kind": "literal", "value": 3}},
+                {
+                    "target": "surprise",
+                    "expression": {"kind": "literal", "value": "x"},
+                },
+            ],
+        )
+        assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"}
+        _rejected_entry(store)
+        assert runs.list_runs() == []
+        assert calls == []
+    finally:
+        ownership.release()
+
+
+def test_valid_structured_input_admits(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs, calls = _scheduler(
+            tmp_path,
+            ownership,
+            [
+                {"target": "count", "expression": {"kind": "literal", "value": 3}},
+                {
+                    "target": "nested",
+                    "expression": {
+                        "kind": "object",
+                        "fields": {"name": {"kind": "literal", "value": "ok"}},
+                    },
+                },
+            ],
+            script={"*": "hang"},
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        result = sched.poll(intended)
+        assert result["s"].startswith("admit:run-")
+        run_id = result["s"].split(":", 1)[1]
+        assert runs.get_admission(run_id).resolved_input == {
+            "count": 3,
+            "nested": {"name": "ok"},
+        }
+        assert len(calls) == 1
+    finally:
+        ownership.release()
+
+
+def test_changed_contract_rejects_before_admission(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs, calls = _scheduler(
+            tmp_path,
+            ownership,
+            [
+                {"target": "count", "expression": {"kind": "literal", "value": 3}},
+            ],
+        )
+        # The pinned contract now additionally requires a token the
+        # schedule does not provide.
+        strict = dict(STRICT_SCHEMA)
+        strict["required"] = ["count", "token"]
+        artifact = _artifact().model_copy(update={"input_schema": strict})
+
+        def changed_env(sched: Any) -> PinnedRunEnvironment:
+            deployment = _deployment().model_copy(update={"id": sched.deployment_id})
+            return PinnedRunEnvironment(
+                deployment=deployment, root_artifact=artifact, child_artifacts=[]
+            )
+
+        sched.preparer = SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 2, "required": []}}), changed_env
+        )
+        assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"}
+        _rejected_entry(store)
+        assert runs.list_runs() == []
+        assert runs.list_admissions() == []
+        assert calls == []
+    finally:
+        ownership.release()
+
+
+def test_environment_built_once_per_preparation(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        builds: list[str] = []
+
+        def counting_env(sched: Any) -> PinnedRunEnvironment:
+            builds.append(sched.id)
+            return _strict_env(sched)
+
+        sched_store = FileScheduleStore(tmp_path / "sched")
+        run_store = FileRunStore(tmp_path / "runs")
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=run_store,
+            sources={},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}), counting_env
+            ),
+            dispatcher=ScriptedDispatcher({"*": "hang"}),
+            ownership=ownership,
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        sched_store.create_schedule(
+            _sched_model(
+                "s",
+                input_bindings=[
+                    {"target": "count", "expression": {"kind": "literal", "value": 1}}
+                ],
+            )
+        )
+        sched_store.save_consumed("s", intended - timedelta(hours=1))
+        sched.sources["s"] = OneShotSource(intended)
+        sched.poll(intended)
+        assert builds == ["s"]
+    finally:
+        ownership.release()

tests/scheduling/test_recovery.py

@@ -58,7 +58,7 @@ def test_recovery_materializes_missing_view_as_pending(tmp_path: Path) -> None:
         resolved_input={},
         max_steps=None,
     )
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         diags = sched_recovery.recover(
             schedule_store=sched_store,
@@ -96,7 +96,7 @@ def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> Non
         schedule_revision=1,
     )
     materialize_admitted_view(store=run_store, admission=admission)
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         diags = sched_recovery.recover(
             schedule_store=sched_store,
@@ -133,7 +133,7 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None:
         schedule_id="a",
         schedule_revision=1,
     )
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         diags = sched_recovery.recover(
             schedule_store=sched_store,
@@ -147,7 +147,7 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None:
     # Recovery itself produced no terminal history; the poll sweep dispatches.
     assert sched_store.list_occurrences("a", limit=100)["total"] == 0
     sources = {"a": OneShotSource(ts(2026, 9, 8, 12, 0))}
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = Scheduler(
             schedule_store=sched_store,

tests/scheduling/test_recovery_stability.py

@@ -0,0 +1,307 @@
+"""Durable recovery failure is stable across repeated recovery (R4 item 2).
+
+An older checkpoint cannot supersede a durable abandonment decision: once
+recovery fails a run, repeating recovery must leave status, readiness,
+diagnostics, and history stable. Only a genuinely newer durable stopped
+result may repair a torn summary. Failed recovery carries not-applicable
+resume readiness.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+from tests.artifacts.test_run_store import artifact as _artifact
+from tests.artifacts.test_run_store import deployment as _deployment
+from wf_api.run_lifecycle import (
+    load_stored_run,
+    materialize_admitted_view,
+    persist_admission,
+    persist_stopped_run,
+    restore_interrupted_run,
+)
+from wf_artifacts import PinnedRunEnvironment
+from wf_artifacts.runs.models import ResumeAttempt
+from wf_artifacts.runs.store import FileRunStore
+from wf_core import RunState, RunStatus
+from wf_scheduling import recovery as sched_recovery
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership
+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 _env() -> PinnedRunEnvironment:
+    return PinnedRunEnvironment(
+        deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[]
+    )
+
+
+def _recover(
+    sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime
+) -> list[str]:
+    ownership = SchedulerOwnership(sched_store.root.parent, owner="test").acquire()
+    try:
+        return sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=now,
+            ownership=ownership,
+        )
+    finally:
+        ownership.release()
+
+
+def _stopped(
+    run_store: FileRunStore,
+    run_id: str,
+    status: RunStatus,
+    *,
+    attempt_id: int | None = None,
+) -> Any:
+    record = run_store.get_run(run_id)
+    return persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(workflow_name="sched", status=status, workflow_input={}, state={}),
+        run_id=run_id,
+        attempt_id=attempt_id,
+    )
+
+
+def _admit_interrupted(
+    run_store: FileRunStore, intended: datetime, *, attempt_id: int | None = None
+) -> str:
+    run_id = run_store.allocate_run_id()
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id="a",
+        schedule_revision=1,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id)
+    return run_id
+
+
+def _mark_active(
+    run_store: FileRunStore, run_id: str, attempt_id: int, now: datetime
+) -> None:
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=attempt_id,
+            state="ACTIVE",
+            created_at=now,
+            updated_at=now,
+        )
+    )
+
+
+def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, Any]]:
+    page = store.list_occurrences(sid, limit=100)
+    return [
+        r for r in cast(list[dict[str, Any]], page["occurrences"]) if r["kind"] == kind
+    ]
+
+
+def test_abandonment_decision_is_stable_across_recovery() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched_store = FileScheduleStore(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = _admit_interrupted(run_store, intended)
+        _mark_active(run_store, run_id, 9, intended)
+        first = _recover(sched_store, run_store, intended)
+        assert any("failed-closed" in d for d in first)
+        record = run_store.get_run(run_id)
+        assert record.status.value == "failed"
+        assert record.resume_readiness.value == "not_applicable"
+        assert len(record.diagnostics) == 1
+        # Fresh store objects across the restart boundary: everything stable.
+        sched_store2 = FileScheduleStore(root / "sched")
+        run_store2 = FileRunStore(root / "runs")
+        second = _recover(sched_store2, run_store2, intended + timedelta(minutes=1))
+        assert not any(run_id in d for d in second)
+        again = run_store2.get_run(run_id)
+        assert again.status.value == "failed"
+        assert again.resume_readiness.value == "not_applicable"
+        assert len(again.diagnostics) == 1
+        assert _entries(sched_store2, "a", "failed") == _entries(
+            sched_store, "a", "failed"
+        )
+        assert len(_entries(sched_store2, "a", "failed")) == 1
+
+
+def test_failed_readiness_and_inspection_agree() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched_store = FileScheduleStore(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = _admit_interrupted(run_store, intended)
+        _mark_active(run_store, run_id, 9, intended)
+        _recover(sched_store, run_store, intended)
+        record, _ = load_stored_run(run_store, run_id)
+        assert record.resume_readiness.value == "not_applicable"
+        try:
+            restore_interrupted_run(run_store, run_id)
+            raise AssertionError("failed run must not restore as interrupted")
+        except ValueError:
+            pass
+
+
+def test_genuinely_newer_result_repairs_after_decision() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched_store = FileScheduleStore(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = _admit_interrupted(run_store, intended)
+        _mark_active(run_store, run_id, 9, intended)
+        _recover(sched_store, run_store, intended)
+        assert run_store.get_run(run_id).status.value == "failed"
+        # A genuinely newer stopped result under a new matching attempt: the
+        # newer checkpoint repairs the summary and completes the attempt.
+        _mark_active(run_store, run_id, 10, intended)
+        _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=10)
+        diags = _recover(sched_store, run_store, intended + timedelta(minutes=1))
+        assert any("fresh-result-resumable" in d for d in diags)
+        record = run_store.get_run(run_id)
+        assert record.status.value == "interrupted"
+        assert record.resume_readiness.value == "ready"
+        assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+        interrupted = _entries(sched_store, "a", "interrupted")
+        assert {e["checkpoint_id"] for e in interrupted} == {f"{run_id}.000002"}
+
+
+def test_completed_mismatch_decision_is_stable() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched_store = FileScheduleStore(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = run_store.allocate_run_id()
+        admission = persist_admission(
+            store=run_store,
+            run_id=run_id,
+            environment=_env(),
+            resolved_input={},
+            max_steps=None,
+            scheduled_at=intended,
+            schedule_id="a",
+            schedule_revision=1,
+        )
+        materialize_admitted_view(store=run_store, admission=admission)
+        _stopped(run_store, run_id, RunStatus.COMPLETED, attempt_id=3)
+        _mark_active(run_store, run_id, 5, intended)
+        _recover(sched_store, run_store, intended)
+        assert run_store.get_run(run_id).status.value == "failed"
+        second = _recover(
+            FileScheduleStore(root / "sched"),
+            FileRunStore(root / "runs"),
+            intended + timedelta(minutes=1),
+        )
+        assert not any(run_id in d for d in second)
+        assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1
+
+
+def test_legacy_failed_run_gains_history_without_refail() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched_store = FileScheduleStore(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = _admit_interrupted(run_store, intended)
+        record = run_store.get_run(run_id)
+        from wf_artifacts.runs.models import StoredRunStatus
+
+        run_store.save_run(record.model_copy(update={"status": StoredRunStatus.FAILED}))
+        assert _entries(sched_store, "a", "failed") == []
+        diags = _recover(sched_store, run_store, intended)
+        assert any("terminal-reconciled" in d for d in diags)
+        assert not any("failed-closed" in d for d in diags)
+        assert run_store.get_run(run_id).diagnostics == []
+        assert len(_entries(sched_store, "a", "failed")) == 1
+
+
+def test_crash_between_decision_writes_recovers_history_once() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+
+        class FailFailedHistoryOnce(FileScheduleStore):
+            def __init__(self, root: Path) -> None:
+                super().__init__(root)
+                self.armed = False
+
+            def append_history(self, record: Any) -> None:
+                if self.armed and getattr(record, "kind", None) == "failed":
+                    self.armed = False
+                    raise OSError("injected failed-history failure")
+                super().append_history(record)
+
+        sched_store = FailFailedHistoryOnce(root / "sched")
+        run_store = FileRunStore(root / "runs")
+        sched_store.create_schedule(_sched_model("a"))
+        intended = ts(2026, 9, 8, 12, 0)
+        run_id = _admit_interrupted(run_store, intended)
+        _mark_active(run_store, run_id, 9, intended)
+        sched_store.armed = True
+        import pytest
+
+        with pytest.raises(OSError, match="injected failed-history failure"):
+            _recover(sched_store, run_store, intended)
+        record = run_store.get_run(run_id)
+        assert record.status.value == "failed"
+        assert len(record.diagnostics) == 1
+        assert _entries(sched_store, "a", "failed") == []
+        # The decision (status + reason) survived; only history is missing.
+        second = _recover(
+            FileScheduleStore(root / "sched"),
+            FileRunStore(root / "runs"),
+            intended + timedelta(minutes=1),
+        )
+        assert not any("failed-closed" in d for d in second)
+        assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1
+        assert len(_entries(FileScheduleStore(root / "sched"), "a", "failed")) == 1

tests/scheduling/test_settle_guards.py

@@ -0,0 +1,252 @@
+"""Late settlement requires a real executing transition (R4 item 4).
+
+An admitted status also describes pending, never-dispatched work, so the
+async-completion seam must demand the executing marker and refuse
+contradictory pending/executing state before persisting a late result.
+Settling a stopped run twice is rejected; ownership stays enforced.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+
+from tests.scheduling.controlled import (
+    DictDeployments,
+    ScriptedDispatcher,
+    fixture_environment,
+)
+from tests.scheduling.controlled import ScriptedDispatcher as SD
+from wf_artifacts.runs.store import FileRunStore
+from wf_scheduling.calendar import OneShotSource
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
+from wf_scheduling.poll import BlockedSchedule, Scheduler
+from wf_scheduling.prepare import 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 _harness(
+    root: Path,
+    ownership: SchedulerOwnership,
+    *,
+    script: dict | None = None,
+) -> tuple[Scheduler, FileScheduleStore, FileRunStore]:
+    sched_store = FileScheduleStore(root / "sched")
+    run_store = FileRunStore(root / "runs")
+    sched = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={},
+        capacity=4,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+            fixture_environment,
+        ),
+        dispatcher=ScriptedDispatcher(script),
+        ownership=ownership,
+    )
+    return sched, sched_store, run_store
+
+
+def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None:
+    store.create_schedule(_sched_model("a"))
+    store.save_consumed("a", intended - timedelta(hours=1))
+    sched.sources["a"] = OneShotSource(intended)
+
+
+def test_pending_undispatched_run_rejects_settlement(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 wf_api.run_lifecycle import materialize_admitted_view, persist_admission
+    from wf_artifacts import PinnedRunEnvironment
+    from wf_core import RunState, RunStatus
+
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        admission = persist_admission(
+            store=runs,
+            run_id=runs.allocate_run_id(),
+            environment=PinnedRunEnvironment(
+                deployment=_deployment(),
+                root_artifact=_artifact(),
+                child_artifacts=[],
+            ),
+            resolved_input={},
+            max_steps=None,
+            scheduled_at=ts(2026, 9, 8, 12, 0),
+            schedule_id="a",
+            schedule_revision=1,
+        )
+        materialize_admitted_view(store=runs, admission=admission)
+        runs.mark_pending_dispatch(admission.id)
+        state = RunState(
+            workflow_name="sched",
+            status=RunStatus.COMPLETED,
+            workflow_input={},
+            state={},
+        )
+        with pytest.raises(BlockedSchedule):
+            sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0))
+        assert runs.get_run(admission.id).status.value == "admitted"
+        assert runs.is_pending_dispatch(admission.id)
+        assert not runs.is_executing(admission.id)
+        with pytest.raises(KeyError):
+            runs.get_latest_checkpoint(admission.id)
+    finally:
+        ownership.release()
+
+
+def test_missing_executing_marker_rejects_settlement(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 wf_api.run_lifecycle import materialize_admitted_view, persist_admission
+    from wf_artifacts import PinnedRunEnvironment
+    from wf_core import RunState, RunStatus
+
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        admission = persist_admission(
+            store=runs,
+            run_id=runs.allocate_run_id(),
+            environment=PinnedRunEnvironment(
+                deployment=_deployment(),
+                root_artifact=_artifact(),
+                child_artifacts=[],
+            ),
+            resolved_input={},
+            max_steps=None,
+            scheduled_at=ts(2026, 9, 8, 12, 0),
+            schedule_id="a",
+            schedule_revision=1,
+        )
+        materialize_admitted_view(store=runs, admission=admission)
+        assert not runs.is_pending_dispatch(admission.id)
+        state = RunState(
+            workflow_name="sched",
+            status=RunStatus.COMPLETED,
+            workflow_input={},
+            state={},
+        )
+        with pytest.raises(BlockedSchedule):
+            sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0))
+        assert runs.get_run(admission.id).status.value == "admitted"
+        with pytest.raises(KeyError):
+            runs.get_latest_checkpoint(admission.id)
+    finally:
+        ownership.release()
+
+
+def test_contradictory_pending_and_executing_rejects_settlement(
+    tmp_path: Path,
+) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        assert runs.is_executing(run_id)
+        # Crash shape between mark_executing and pending-clear: both markers.
+        runs.mark_pending_dispatch(run_id)
+        dispatcher = cast(SD, sched.dispatcher)
+        admission = runs.get_admission(run_id)
+        with pytest.raises(BlockedSchedule):
+            sched.record_stopped_execution(
+                run_id, dispatcher.finish(admission, "complete"), intended
+            )
+        assert runs.get_run(run_id).status.value == "admitted"
+        assert runs.is_executing(run_id)
+        assert runs.is_pending_dispatch(run_id)
+        with pytest.raises(KeyError):
+            runs.get_latest_checkpoint(run_id)
+    finally:
+        ownership.release()
+
+
+def test_genuine_execution_settles_successfully(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        dispatcher = cast(SD, sched.dispatcher)
+        admission = runs.get_admission(run_id)
+        sched.record_stopped_execution(
+            run_id, dispatcher.finish(admission, "complete"), intended
+        )
+        assert runs.get_run(run_id).status.value == "completed"
+        assert not runs.is_executing(run_id)
+    finally:
+        ownership.release()
+
+
+def test_duplicate_settlement_cannot_overwrite_stopped_result(
+    tmp_path: Path,
+) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        dispatcher = cast(SD, sched.dispatcher)
+        admission = runs.get_admission(run_id)
+        sched.record_stopped_execution(
+            run_id, dispatcher.finish(admission, "complete"), intended
+        )
+        with pytest.raises(BlockedSchedule):
+            sched.record_stopped_execution(
+                run_id, dispatcher.finish(admission, "interrupt"), intended
+            )
+        assert runs.get_run(run_id).status.value == "completed"
+        assert len(runs.list_checkpoints(run_id)) == 1
+    finally:
+        ownership.release()
+
+
+def test_settle_without_ownership_rejected(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        dispatcher = cast(SD, sched.dispatcher)
+        admission = runs.get_admission(run_id)
+        ownership.release()
+        with pytest.raises(SecondOwnerError):
+            sched.record_stopped_execution(
+                run_id, dispatcher.finish(admission, "complete"), intended
+            )
+        assert runs.get_run(run_id).status.value == "admitted"
+    finally:
+        ownership.release()

tests/scheduling/test_torn_reconcile.py

@@ -92,7 +92,7 @@ class FailSaveRunOnce(FileRunStore):
 def _recover(
     sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime
 ) -> list[str]:
-    ownership = SchedulerOwnership(sched_store.root, owner="test").acquire()
+    ownership = SchedulerOwnership(sched_store.root.parent, owner="test").acquire()
     try:
         return sched_recovery.recover(
             schedule_store=sched_store,
@@ -373,7 +373,7 @@ def test_real_workflow_interrupt_resume_interrupt_with_torn_resume(
     run_store = FailSaveRunOnce(tmp_path / "runs")
     sched_store.create_schedule(_sched_model("a"))
     intended = ts(2026, 9, 8, 12, 0)
-    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
     try:
         sched = Scheduler(
             schedule_store=sched_store,