sched: one canonical lock identity per composition; checkpoint coherence and ordering authority (R4 wave 3)

src/wf_scheduling/ownership.py

@@ -11,11 +11,22 @@ Locking design (Windows-tested, per the store transaction boundary): one
 new module holding ``msvcrt.locking`` (Windows) / ``fcntl.flock`` (POSIX)
 on ``<composition root>/scheduler.lock``. No ad-hoc per-run lock files.
 
+One store composition has exactly one lock identity: the canonical
+nearest common ancestor of its store roots (see
+:func:`canonical_lock_root`). Any ancestor is NOT good enough — different
+ancestors yield different lock files and therefore independent OS locks,
+so ancestor containment can never prove exclusivity. Guards require the
+held lock's frozen identity to equal the composition identity; overlapping
+compositions that share no common ancestor are rejected instead of
+claimed safe.
+
 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.
+or side effect. The acquired identity is frozen at acquisition: later
+mutation of the public ``root`` attribute cannot redirect a held handle's
+authority.
 """
 
 from __future__ import annotations
@@ -48,12 +59,28 @@ def canonical_store_path(path: Path | str) -> str:
     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)
+def canonical_lock_root(*store_roots: Path | str | None) -> str | None:
+    """Return the one canonical lock identity for a store composition.
+
+    The identity is the canonical nearest common ancestor of the store
+    roots; the lock file lives at ``<identity>/scheduler.lock``. Both the
+    held guard's frozen root and the guarded stores map through this
+    function, so exactly one lock file can authorize one composition.
+    Returns None when a root is missing or the roots share no common
+    ancestor (e.g. different drives): such compositions are unsupported
+    and guards reject them instead of claiming safety.
+    """
+    canonical: list[str] = []
+    for store_root in store_roots:
+        if store_root is None:
+            return None
+        canonical.append(canonical_store_path(store_root))
+    if not canonical:
+        return None
+    try:
+        return os.path.commonpath(canonical)
+    except ValueError:
+        return None
 
 
 class SchedulerOwnership:
@@ -64,6 +91,7 @@ class SchedulerOwnership:
         self.owner = owner
         self._handle: BinaryIO | None = None
         self._locked = False
+        self._acquired_root: str | None = None
 
     @property
     def lock_path(self) -> Path:
@@ -80,25 +108,35 @@ class SchedulerOwnership:
         return self._locked
 
     def canonical_root(self) -> str:
-        """Return the canonical composition root this lock was acquired on."""
+        """Return the canonical composition root this lock was acquired on.
+
+        The identity is frozen at acquisition: mutating the public
+        ``root`` afterwards cannot redirect a held handle's authority.
+        Before acquisition (or after release) it reports the live root.
+        """
+        if self._acquired_root is not None:
+            return self._acquired_root
         return canonical_store_path(self.root)
 
     def covers(self, *store_roots: Path | str | None) -> bool:
-        """Whether this held lock protects the given store roots.
+        """Whether this held lock is the one lock for the given stores.
 
-        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.
+        Coverage means the frozen acquisition identity equals the
+        composition's canonical lock identity
+        (:func:`canonical_lock_root`), not merely that the stores sit
+        beneath some held ancestor: a lock on a parent directory is an
+        independent OS lock from the composition's own lock and proves
+        nothing. An unrelated (even held) lock, a released lock, a
+        mutated handle, and rootless or ancestor-less 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
-        )
+        if self._acquired_root is None:
+            return False
+        expected = canonical_lock_root(*store_roots)
+        return expected is not None and self._acquired_root == expected
 
     def acquire(self) -> SchedulerOwnership:
         """Acquire the held lock non-blockingly or raise SecondOwnerError."""
@@ -120,6 +158,10 @@ class SchedulerOwnership:
             raise StartupRejected(f"unsupported scheduler locking: {exc}") from exc
         self._handle = handle
         self._locked = True
+        # Freeze the authority identity now: the public ``root`` may be
+        # reassigned later, but a held handle must keep proving exactly the
+        # composition it locked, never the new value.
+        self._acquired_root = canonical_store_path(self.root)
         return self
 
     def release(self) -> None:
@@ -134,6 +176,7 @@ class SchedulerOwnership:
             finally:
                 self._handle = None
                 self._locked = False
+                self._acquired_root = None
 
     def __enter__(self) -> SchedulerOwnership:
         return self.acquire()

src/wf_scheduling/recovery.py

@@ -31,13 +31,25 @@ 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.
+- failed summary: a durable recovery decision is never re-processed. A
+  matching checkpoint pointer only ensures history; it never revalidates
+  the decision. A mismatched pointer reopens the run only for a
+  genuinely newer result: the referenced checkpoint must still exist, the
+  latest checkpoint must have a strictly greater sequence with attempt
+  provenance (an attributed result never yields to an older attempt, and
+  an unattributed result never supersedes an attributed decision), and
+  the newer checkpoint must be internally coherent (see below).
+  Anything else — a missing referenced checkpoint, an older stray, a
+  stale attempt, or an incoherent newer checkpoint — keeps the decision
+  with history ensured; corruption is additionally noted once as a
+  durable diagnostic. Failed runs carry not-applicable resume readiness.
+- checkpoint authority: before any summary rewrite, the latest checkpoint
+  must prove coherence — same run identity, checkpoint id matching its
+  sequence, runtime state decodable through core validation, and the
+  outer reason agreeing with the decoded stopped state. The outer reason
+  enum alone is never trusted. Contradictory or corrupt checkpoints fail
+  closed with durable diagnostics; a successful result is never
+  fabricated from disagreement.
 """
 
 from __future__ import annotations
@@ -121,19 +133,24 @@ def recover(
             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.
+        # Durable-decision stability: a FAILED summary for the decided
+        # checkpoint state is never re-processed — history is ensured
+        # without mutating the summary. A mismatched pointer reopens the
+        # run only for a validated genuinely-newer result (referenced
+        # checkpoint present, strictly greater sequence, attempt
+        # provenance, coherent content); anything else keeps the decision.
         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
-        # readiness are rewritten together from the checkpoint's own
-        # persisted fields; nothing is fabricated). Attempt matching below
-        # still distinguishes fresh results from stale ones.
+        # save_checkpoint/save_run boundary: a validated durable stopped
+        # checkpoint wins over a stale summary (status, checkpoint pointer,
+        # and readiness are rewritten together from the checkpoint's own
+        # persisted fields; nothing is fabricated). Checkpoints that fail
+        # authority validation (wrong run, id/sequence disagreement,
+        # undecodable state, reason disagreeing with the decoded stopped
+        # state) raise below and fail closed. Attempt matching below still
+        # distinguishes fresh results from stale ones.
         try:
             changed, status = _reconcile_summary_from_checkpoint(run_store, run, now)
         except (ValueError, OSError) as exc:
@@ -284,17 +301,57 @@ def recover(
     return diags
 
 
+def _check_checkpoint_coherence(run: Any, checkpoint: Any) -> None:
+    """Prove a checkpoint may speak for a run summary, or raise.
+
+    Raises KeyError only from the caller's own loads; raises ValueError
+    (or OSError for unreadable state) when the checkpoint disagrees with
+    the run: a foreign run identity, a checkpoint id that does not match
+    its sequence, runtime state that core validation rejects, or an outer
+    reason that contradicts the decoded stopped state. The outer reason
+    enum alone is never authority: a COMPLETED label over an interrupted
+    runtime snapshot must fail closed, never release overlap.
+    """
+    from wf_core import load_run_state_with_upgrade
+
+    if checkpoint.run_id != run.id:
+        raise ValueError(
+            f"checkpoint {checkpoint.id!r} names run {checkpoint.run_id!r}, "
+            f"not {run.id!r}"
+        )
+    if checkpoint.id != f"{run.id}.{checkpoint.sequence:06d}":
+        raise ValueError(
+            f"checkpoint {checkpoint.id!r} disagrees with its sequence "
+            f"{checkpoint.sequence}"
+        )
+    try:
+        decoded, _ = load_run_state_with_upgrade(
+            checkpoint.state.model_dump(mode="json")
+        )
+    except ValueError as exc:
+        raise ValueError(
+            f"checkpoint {checkpoint.id!r} has undecodable runtime state"
+        ) from exc
+    if decoded.status.value != checkpoint.reason.value:
+        raise ValueError(
+            f"checkpoint {checkpoint.id!r} reason {checkpoint.reason.value!r} "
+            f"contradicts its runtime state {decoded.status.value!r}"
+        )
+
+
 def _reconcile_summary_from_checkpoint(
     run_store: Any, run: Any, now: datetime
 ) -> tuple[bool, str]:
-    """Rewrite a stale summary from its durable stopped checkpoint.
+    """Rewrite a stale summary from its validated durable stopped checkpoint.
 
     Returns ``(changed, status)``. Only the checkpoint's own persisted
     fields (status, checkpoint id, readiness derived exactly as
     ``persist_stopped_run`` derives it) are copied; creation time,
     environment, and diagnostics are preserved. A run without checkpoints
     is untouched (an admitted summary with no checkpoint is genuinely
-    undispatched-or-unknown, handled by the marker rules).
+    undispatched-or-unknown, handled by the marker rules). Authority
+    validation runs first: incoherent checkpoints raise and fail closed
+    in the caller instead of rewriting the summary.
     """
     from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus
 
@@ -302,6 +359,7 @@ def _reconcile_summary_from_checkpoint(
         latest = run_store.get_latest_checkpoint(run.id)
     except KeyError:
         return False, getattr(run.status, "value", run.status)
+    _check_checkpoint_coherence(run, latest)
     expected = StoredRunStatus(latest.reason.value)
     readiness = (
         ResumeReadiness.READY
@@ -342,6 +400,87 @@ def _decision_reason(run: Any) -> str:
     return ""
 
 
+KEPT_DECISION_NOTE_CODE = "schedule-recovery-note"
+
+
+def _ensure_failed_history(
+    run_store: Any,
+    run: Any,
+    history: HistoryRecorder,
+    now: datetime,
+    diags: list[str],
+) -> None:
+    """Ensure the failed terminal entry without touching the summary."""
+    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",
+        run.latest_checkpoint_id,
+        intended,
+        revision,
+        reason,
+        now,
+    ):
+        diags.append(f"{run.id}:terminal-reconciled")
+
+
+def _note_decision_kept(
+    run_store: Any, run: Any, message: str, now: datetime, diags: list[str]
+) -> None:
+    """Durably note once that a decision survived unresolvable state.
+
+    Status and readiness are untouched; the note is appended only when no
+    identical note exists, so repeated recovery stays silent.
+    """
+    from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
+
+    if any(
+        getattr(item, "code", None) == KEPT_DECISION_NOTE_CODE
+        and str(getattr(item, "message", "")) == message
+        for item in run.diagnostics
+    ):
+        return
+    run_store.save_run(
+        run.model_copy(
+            update={
+                "updated_at": now,
+                "diagnostics": [
+                    *run.diagnostics,
+                    DependencyDiagnostic(
+                        severity=DiagnosticSeverity.ERROR,
+                        code=KEPT_DECISION_NOTE_CODE,
+                        logical_ref=run.id,
+                        message=message,
+                        repair_hint=None,
+                    ),
+                ],
+            }
+        )
+    )
+    diags.append(f"{run.id}:decision-kept-noted")
+
+
+def _supersedes_decision(referenced: Any, latest: Any) -> bool:
+    """Whether a newer checkpoint result may reopen a durable decision.
+
+    Ordering alone is not provenance: the latest checkpoint must carry a
+    strictly greater sequence AND attempt provenance — an attributed
+    result never yields to an older attempt, and an unattributed result
+    never supersedes an attributed decision (unattributed results predate
+    the resume protocol that produced the decision). Sequence ties and
+    regressions never supersede: a different checkpoint is not
+    necessarily a newer result.
+    """
+    if latest.sequence <= referenced.sequence:
+        return False
+    if latest.attempt_id is None or referenced.attempt_id is None:
+        return referenced.attempt_id is None
+    return latest.attempt_id >= referenced.attempt_id
+
+
 def _failed_decision_stable(
     run_store: Any,
     run: Any,
@@ -353,38 +492,65 @@ def _failed_decision_stable(
 
     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.
+    failure) and no validated newer result supersedes it, so re-running
+    reconcile or re-failing would only duplicate diagnostics and rewrite
+    readiness. A pointer match ensures history and stays silent. A
+    mismatch reopens the run only for a genuinely newer result —
+    referenced checkpoint present, strictly greater sequence, attempt
+    provenance, coherent content — via the reconcile path. A missing
+    referenced checkpoint, an older or unattributable newer checkpoint,
+    or an incoherent newer checkpoint keeps the decision: missing state
+    never rolls a run backward and never undoes a durable abandonment.
+    Corruption is noted once as a durable diagnostic. Unreadable
+    checkpoint state 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
+        checkpoints = run_store.list_checkpoints(run.id)
     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.
+    referenced = (
+        next((c for c in checkpoints if c.id == pointer), None)
+        if pointer is not None
+        else None
+    )
+    latest = max(checkpoints, key=lambda c: c.sequence, default=None)
+    if pointer is not None and referenced is None:
+        _ensure_failed_history(run_store, run, history, now, diags)
+        _note_decision_kept(
+            run_store,
+            run,
+            f"durable recovery decision kept: referenced checkpoint "
+            f"{pointer!r} is missing",
+            now,
+            diags,
+        )
+        return True
+    if latest is None or latest.id == pointer:
+        _ensure_failed_history(run_store, run, history, now, diags)
+        return True
+    if referenced is None:
+        # A decision without a checkpoint pointer cannot order stray
+        # checkpoints: keep the decision with history ensured.
+        _ensure_failed_history(run_store, run, history, now, diags)
+        return True
+    if _supersedes_decision(referenced, latest):
+        try:
+            _check_checkpoint_coherence(run, latest)
+        except (ValueError, OSError) as exc:
+            _ensure_failed_history(run_store, run, history, now, diags)
+            _note_decision_kept(
+                run_store,
+                run,
+                f"durable recovery decision kept: newer checkpoint "
+                f"{latest.id!r} is incoherent: {exc}",
+                now,
+                diags,
+            )
+            return True
         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")
+    _ensure_failed_history(run_store, run, history, now, diags)
     return True
 
 

tests/scheduling/test_checkpoint_authority.py

@@ -0,0 +1,551 @@
+"""Checkpoint authority: coherence and ordering before recovery trusts state.
+
+A checkpoint becomes recovery authority only when it proves coherence —
+same run identity, checkpoint id matching its sequence, runtime state
+decodable through core validation, and the outer reason agreeing with the
+decoded stopped state. The outer reason enum alone is never trusted.
+
+A durable recovery decision is superseded only by a validated genuinely
+newer result (referenced checkpoint present, strictly greater sequence,
+attempt provenance, coherent content). Missing referenced state never
+rolls a run backward.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+
+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,
+    fixture_environment,
+)
+from wf_api.run_lifecycle import (
+    load_stored_run,
+    materialize_admitted_view,
+    persist_admission,
+    persist_stopped_run,
+    restore_interrupted_run,
+)
+from wf_artifacts import CheckpointReason, PinnedRunEnvironment
+from wf_artifacts.runs.models import ResumeAttempt, RunCheckpoint
+from wf_artifacts.runs.store import FileRunStore
+from wf_core import PersistedRunState, RunState, RunStatus, dump_run_state
+from wf_scheduling import recovery as sched_recovery
+from wf_scheduling.calendar import OneShotSource
+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) -> Schedule:
+    now = ts(2026, 9, 8, 12, 0)
+    return Schedule.model_validate(
+        {
+            "id": sid,
+            "deployment_id": "dep-1",
+            "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"},
+            "input_bindings": [],
+            "created_at": now.isoformat(),
+            "updated_at": now.isoformat(),
+        }
+    )
+
+
+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 _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)
+    record = run_store.get_run(run_id)
+    persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(
+            workflow_name="sched",
+            status=RunStatus.INTERRUPTED,
+            workflow_input={},
+            state={},
+        ),
+        run_id=run_id,
+        attempt_id=attempt_id,
+    )
+    return run_id
+
+
+def _mark_attempt(
+    run_store: FileRunStore, run_id: str, attempt_id: int, state: str, now: datetime
+) -> None:
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=attempt_id,
+            state=cast(Any, state),
+            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 _craft_checkpoint(
+    run_id: str,
+    sequence: int,
+    status: RunStatus,
+    *,
+    checkpoint_run_id: str | None = None,
+    checkpoint_id: str | None = None,
+    attempt_id: int | None = None,
+    now: datetime | None = None,
+) -> RunCheckpoint:
+    """Build a storable checkpoint with caller-controlled identity fields."""
+    return RunCheckpoint(
+        id=checkpoint_id or f"{run_id}.{sequence:06d}",
+        run_id=checkpoint_run_id or run_id,
+        sequence=sequence,
+        reason=CheckpointReason(status.value),
+        state=PersistedRunState.model_validate(
+            dump_run_state(
+                RunState(
+                    workflow_name="sched",
+                    status=status,
+                    workflow_input={},
+                    state={},
+                )
+            )
+        ),
+        attempt_id=attempt_id,
+        created_at=now or datetime.now(UTC),
+    )
+
+
+def _checkpoint_path(root: Path, run_id: str, sequence: int) -> Path:
+    return root / "runs" / "runs" / run_id / "checkpoints" / f"{sequence:06d}.json"
+
+
+def _tamper_reason(root: Path, run_id: str, sequence: int, reason: str) -> None:
+    path = _checkpoint_path(root, run_id, sequence)
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    payload["reason"] = reason
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+
+def _decided_at_000002(
+    tmp_path: Path, intended: datetime
+) -> tuple[FileScheduleStore, FileRunStore, str]:
+    """Failed decision pointing at .000002 with only .000001 older state.
+
+    Genuine multi-restart sequence: ambiguous fail at .000001, a newer
+    stopped result under attempt 10, then stale under attempt 11. The
+    second durable abandonment points at .000002; attempt 11 is DONE so
+    no attempt matching can mask the ordering decision.
+    """
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    run_id = _admit_interrupted(run_store, intended)
+    _mark_attempt(run_store, run_id, 9, "ACTIVE", intended)
+    first = _recover(sched_store, run_store, intended)
+    assert any("failed-closed" in d for d in first)
+    _mark_attempt(run_store, run_id, 10, "ACTIVE", intended)
+    record = run_store.get_run(run_id)
+    persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(
+            workflow_name="sched",
+            status=RunStatus.INTERRUPTED,
+            workflow_input={},
+            state={},
+        ),
+        run_id=run_id,
+        attempt_id=10,
+    )
+    _mark_attempt(run_store, run_id, 11, "ACTIVE", intended)
+    second = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("failed-closed" in d for d in second)
+    decided = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert decided.status.value == "failed"
+    assert decided.latest_checkpoint_id == f"{run_id}.000002"
+    assert decided.resume_readiness.value == "not_applicable"
+    _mark_attempt(FileRunStore(tmp_path / "runs"), run_id, 11, "DONE", intended)
+    return (
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        run_id,
+    )
+
+
+def test_reason_state_disagreement_fails_closed(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_interrupted(run_store, intended)
+    _tamper_reason(tmp_path, run_id, 1, "completed")
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("failed-closed" in d for d in diags)
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert record.status.value == "failed"
+    assert record.resume_readiness.value == "not_applicable"
+    assert any("contradicts its runtime state" in d.message for d in record.diagnostics)
+    # Nothing successful fabricated: no completed/interrupted terminal entries.
+    fresh_sched = FileScheduleStore(tmp_path / "sched")
+    assert _entries(fresh_sched, "a", "completed") == []
+    assert _entries(fresh_sched, "a", "interrupted") == []
+    assert len(_entries(fresh_sched, "a", "failed")) == 1
+    # Repeat recovery is stable: the decision stands, nothing new appended.
+    rerun = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=2),
+    )
+    assert not any(run_id in d for d in rerun)
+    again = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert again.status.value == "failed"
+    assert len(again.diagnostics) == len(record.diagnostics)
+
+
+def test_run_identity_disagreement_fails_closed(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_interrupted(run_store, intended)
+    foreign = _craft_checkpoint(
+        run_id, 2, RunStatus.INTERRUPTED, checkpoint_run_id="run-999999"
+    )
+    _checkpoint_path(tmp_path, run_id, 2).write_text(
+        json.dumps(foreign.model_dump(mode="json")), encoding="utf-8"
+    )
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("failed-closed" in d for d in diags)
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert record.status.value == "failed"
+    assert any("names run" in d.message for d in record.diagnostics)
+
+
+def test_sequence_pointer_incoherence_fails_closed(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_interrupted(run_store, intended)
+    mismatched = _craft_checkpoint(
+        run_id, 2, RunStatus.INTERRUPTED, checkpoint_id=f"{run_id}.000001"
+    )
+    _checkpoint_path(tmp_path, run_id, 2).write_text(
+        json.dumps(mismatched.model_dump(mode="json")), encoding="utf-8"
+    )
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("failed-closed" in d for d in diags)
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert record.status.value == "failed"
+    assert any("disagrees with its sequence" in d.message for d in record.diagnostics)
+
+
+def test_undecodable_runtime_state_fails_closed(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_interrupted(run_store, intended)
+    bogus = _craft_checkpoint(run_id, 2, RunStatus.INTERRUPTED)
+    payload = bogus.model_dump(mode="json")
+    payload["state"] = {"version": 2, "state": {"bogus": True}}
+    _checkpoint_path(tmp_path, run_id, 2).write_text(
+        json.dumps(payload), encoding="utf-8"
+    )
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("failed-closed" in d for d in diags)
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert record.status.value == "failed"
+    assert record.resume_readiness.value == "not_applicable"
+    assert any("undecodable runtime state" in d.message for d in record.diagnostics)
+
+
+def test_genuine_newer_completed_repairs_torn_summary(tmp_path: Path) -> None:
+    """Coherent newer checkpoints still repair torn summaries (no over-block)."""
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_interrupted(run_store, intended)
+    # Torn boundary: the completed checkpoint persisted, the summary did not.
+    run_store.save_checkpoint(_craft_checkpoint(run_id, 2, RunStatus.COMPLETED))
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert any("summary-reconciled" in d for d in diags)
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert record.status.value == "completed"
+    assert record.latest_checkpoint_id == f"{run_id}.000002"
+    fresh_sched = FileScheduleStore(tmp_path / "sched")
+    assert {e["checkpoint_id"] for e in _entries(fresh_sched, "a", "completed")} == {
+        f"{run_id}.000002"
+    }
+
+
+def test_missing_referenced_checkpoint_keeps_decision(tmp_path: Path) -> None:
+    sched_store, run_store, run_id = _decided_at_000002(tmp_path, ts(2026, 9, 8, 12, 0))
+    before = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert len(before.diagnostics) == 1
+    # Referenced durable state goes missing; only older .000001 remains.
+    _checkpoint_path(tmp_path, run_id, 2).unlink()
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        ts(2026, 9, 8, 12, 2),
+    )
+    assert any("decision-kept-noted" in d for d in diags)
+    assert not any("summary-reconciled" in d for d in diags)
+    kept = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert kept.status.value == "failed"
+    assert kept.resume_readiness.value == "not_applicable"
+    assert len(kept.diagnostics) == 2
+    assert any("is missing" in d.message for d in kept.diagnostics)
+    # Both genuine decisions keep their terminal entries; the keep adds none.
+    assert len(_entries(FileScheduleStore(tmp_path / "sched"), "a", "failed")) == 2
+    # Repeat recovery is fully silent and history stays idempotent.
+    rerun = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        ts(2026, 9, 8, 12, 3),
+    )
+    assert not any(run_id in d for d in rerun)
+    again = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert again.status.value == "failed"
+    assert len(again.diagnostics) == 2
+    assert _entries(FileScheduleStore(tmp_path / "sched"), "a", "failed") == _entries(
+        sched_store, "a", "failed"
+    )
+
+
+@pytest.mark.parametrize(
+    ("newer_attempt", "tamper", "expect_repair"),
+    [
+        (11, False, True),  # newer sequence + newer attempt: genuine repair
+        (9, False, False),  # newer sequence + older attempt: stale, keep decision
+        (None, False, False),  # newer sequence + unattributed: keep decision
+        (11, True, False),  # newer + provenance but incoherent: keep + note
+    ],
+)
+def test_attempt_provenance_combinations(
+    tmp_path: Path, newer_attempt: int | None, tamper: bool, expect_repair: bool
+) -> None:
+    intended = ts(2026, 9, 8, 12, 0)
+    _, _, run_id = _decided_at_000002(tmp_path, intended)
+    run_store = FileRunStore(tmp_path / "runs")
+    _mark_attempt(run_store, run_id, 11, "ACTIVE", intended)
+    run_store.save_checkpoint(
+        _craft_checkpoint(run_id, 3, RunStatus.INTERRUPTED, attempt_id=newer_attempt)
+    )
+    if tamper:
+        _tamper_reason(tmp_path, run_id, 3, "completed")
+    diags = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=5),
+    )
+    record = FileRunStore(tmp_path / "runs").get_run(run_id)
+    if expect_repair:
+        assert any("fresh-result-resumable" in d for d in diags)
+        assert record.status.value == "interrupted"
+        assert record.resume_readiness.value == "ready"
+        assert run_store.get_resume_attempt(run_id) is not None
+        assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+    else:
+        assert record.status.value == "failed"
+        assert record.resume_readiness.value == "not_applicable"
+        assert not any("summary-reconciled" in d for d in diags)
+        if tamper:
+            assert any("decision-kept-noted" in d for d in diags)
+            assert any("is incoherent" in d.message for d in record.diagnostics)
+        else:
+            # Stale provenance is expected debris, not corruption: silent keep.
+            assert not any("decision-kept-noted" in d for d in diags)
+        rerun = _recover(
+            FileScheduleStore(tmp_path / "sched"),
+            FileRunStore(tmp_path / "runs"),
+            intended + timedelta(minutes=6),
+        )
+        assert not any(run_id in d for d in rerun)
+
+
+def test_pointerless_decision_ignores_stray_checkpoints(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "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)
+    run_store.mark_executing(run_id)
+    diags = _recover(sched_store, run_store, intended)
+    assert any("failed-closed" in d for d in diags)
+    decided = run_store.get_run(run_id)
+    assert decided.status.value == "failed"
+    assert decided.latest_checkpoint_id is None
+    # A stray checkpoint with no ordering basis cannot reopen the decision.
+    run_store.save_checkpoint(_craft_checkpoint(run_id, 1, RunStatus.INTERRUPTED))
+    rerun = _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=1),
+    )
+    assert not any("summary-reconciled" in d for d in rerun)
+    kept = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert kept.status.value == "failed"
+    assert kept.resume_readiness.value == "not_applicable"
+
+
+def test_inspection_resume_and_poll_agree_after_reconciliation(
+    tmp_path: Path,
+) -> None:
+    intended = ts(2026, 9, 8, 12, 0)
+    _, _, run_id = _decided_at_000002(tmp_path, intended)
+    _checkpoint_path(tmp_path, run_id, 2).unlink()
+    _recover(
+        FileScheduleStore(tmp_path / "sched"),
+        FileRunStore(tmp_path / "runs"),
+        intended + timedelta(minutes=2),
+    )
+    # Inspection agrees: failed, not applicable.
+    record, _ = load_stored_run(FileRunStore(tmp_path / "runs"), run_id)
+    assert record.status.value == "failed"
+    assert record.resume_readiness.value == "not_applicable"
+    # Resume refuses.
+    try:
+        restore_interrupted_run(FileRunStore(tmp_path / "runs"), run_id)
+        raise AssertionError("failed run must not restore as interrupted")
+    except ValueError:
+        pass
+    # A poll sweep with nothing due leaves the decided run untouched and
+    # dispatches nothing from its slot.
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    sched_store.save_consumed("a", intended)
+    ownership = SchedulerOwnership(tmp_path, owner="poll").acquire()
+    try:
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=FileRunStore(tmp_path / "runs"),
+            sources={"a": OneShotSource(intended)},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+                fixture_environment,
+            ),
+            dispatcher=ScriptedDispatcher({"*": "hang"}),
+            ownership=ownership,
+        )
+        sched.poll(intended)
+    finally:
+        ownership.release()
+    after = FileRunStore(tmp_path / "runs").get_run(run_id)
+    assert after.status.value == "failed"
+    assert [d.message for d in after.diagnostics] == [
+        d.message for d in record.diagnostics
+    ]
+    assert [r.id for r in FileRunStore(tmp_path / "runs").list_runs()] == [run_id]
+
+
+def test_wrong_root_rejected_before_writes_or_dispatch(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store = FileScheduleStore(comp / "sched")
+    run_store = FileRunStore(comp / "runs")
+    outsider = SchedulerOwnership(tmp_path, owner="outsider").acquire()
+    try:
+        assert outsider.held
+        assert outsider.covers(sched_store.root, run_store.root) is False
+        with pytest.raises(SecondOwnerError):
+            sched_recovery.recover(
+                schedule_store=sched_store,
+                run_store=run_store,
+                now=datetime.now(UTC),
+                ownership=outsider,
+            )
+        assert run_store.list_runs() == []
+        assert run_store.list_admissions() == []
+    finally:
+        outsider.release()

tests/scheduling/test_lock_identity.py

@@ -0,0 +1,207 @@
+"""One store composition has one lock identity (R4 wave 3, item A).
+
+Ancestor containment is not authority: a lock on a parent directory is an
+independent OS lock from the composition's own lock, so both can be held
+simultaneously. Guards require the held lock's frozen identity to equal
+the composition's canonical lock identity
+(:func:`canonical_lock_root`), and the acquired identity is frozen so a
+later root mutation cannot redirect a held handle.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+
+from wf_artifacts.runs.store import FileRunStore
+from wf_scheduling import recovery as sched_recovery
+from wf_scheduling.ownership import (
+    SchedulerOwnership,
+    SecondOwnerError,
+    canonical_lock_root,
+    canonical_store_path,
+)
+from wf_scheduling.store import FileScheduleStore
+
+
+def _stores(root: Path) -> tuple[FileScheduleStore, FileRunStore]:
+    return FileScheduleStore(root / "sched"), FileRunStore(root / "runs")
+
+
+def test_canonical_lock_identity_is_deterministic(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store, run_store = _stores(comp)
+    assert canonical_lock_root(
+        sched_store.root, run_store.root
+    ) == canonical_store_path(comp)
+    # Aliased spellings of the same composition share the identity.
+    dotted = tmp_path / "sub" / ".." / "composition"
+    assert canonical_lock_root(
+        dotted / "sched", dotted / "runs"
+    ) == canonical_store_path(comp)
+    # A different composition has a different identity; missing roots have none.
+    assert canonical_lock_root(tmp_path / "other") != canonical_store_path(comp)
+    assert canonical_lock_root(None, run_store.root) is None  # type: ignore[arg-type]
+    assert canonical_lock_root() is None
+
+
+def test_parent_and_child_locks_cannot_both_authorize(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store, run_store = _stores(comp)
+    parent_lock = SchedulerOwnership(tmp_path, owner="parent").acquire()
+    try:
+        # A different lock file: simultaneous hold proves independence.
+        child_lock = SchedulerOwnership(comp, owner="child").acquire()
+        try:
+            assert parent_lock.held and child_lock.held
+            assert child_lock.covers(sched_store.root, run_store.root) is True
+            assert parent_lock.covers(sched_store.root, run_store.root) is False
+            now = datetime.now(UTC)
+            sched_recovery.recover(
+                schedule_store=sched_store,
+                run_store=run_store,
+                now=now,
+                ownership=child_lock,
+            )
+            before_runs = run_store.list_runs()
+            before_admissions = run_store.list_admissions()
+            with pytest.raises(SecondOwnerError):
+                sched_recovery.recover(
+                    schedule_store=FileScheduleStore(comp / "sched"),
+                    run_store=FileRunStore(comp / "runs"),
+                    now=now,
+                    ownership=parent_lock,
+                )
+            # Wrong-root rejection happens before any write.
+            assert FileRunStore(comp / "runs").list_runs() == before_runs
+            assert FileRunStore(comp / "runs").list_admissions() == before_admissions
+        finally:
+            child_lock.release()
+    finally:
+        parent_lock.release()
+
+
+def test_strict_ancestor_lock_rejected_for_nested_composition(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store, run_store = _stores(comp)
+    grandparent = SchedulerOwnership(tmp_path.parent, owner="grandparent").acquire()
+    try:
+        assert grandparent.covers(sched_store.root, run_store.root) is False
+    finally:
+        grandparent.release()
+
+
+def test_competing_processes_share_the_canonical_lock(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    comp.mkdir(parents=True)
+    sched_store, run_store = _stores(comp)
+    identity = Path(canonical_lock_root(sched_store.root, run_store.root) or "")
+    holder = SchedulerOwnership(identity, owner="parent").acquire()
+    try:
+        script_path = tmp_path / "compete_canonical.py"
+        script_path.write_text(
+            "import sys\n"
+            "sys.path.insert(0, '.')\n"
+            "from pathlib import Path\n"
+            "from wf_scheduling.ownership import canonical_lock_root, SchedulerOwnership, SecondOwnerError\n"
+            f"sched_root = Path({str(sched_store.root)!r})\n"
+            f"runs_root = Path({str(run_store.root)!r})\n"
+            "identity = Path(canonical_lock_root(sched_root, runs_root))\n"
+            "try:\n"
+            "    SchedulerOwnership(identity, owner='child').acquire()\n"
+            "except SecondOwnerError:\n"
+            "    print('second-owner-rejected')\n"
+            "    raise SystemExit(0)\n"
+            "print('unexpectedly-acquired')\n"
+            "raise SystemExit(1)\n",
+            encoding="utf-8",
+        )
+        proc = subprocess.run(
+            [sys.executable, str(script_path)],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        assert proc.returncode == 0, proc.stderr
+        assert "second-owner-rejected" in proc.stdout
+        # A lock on the parent is an independent file: it acquires, but it
+        # cannot authorize the composition while the canonical lock is held.
+        outsider = SchedulerOwnership(tmp_path, owner="outsider").acquire()
+        try:
+            assert outsider.held
+            assert outsider.covers(sched_store.root, run_store.root) is False
+            with pytest.raises(SecondOwnerError):
+                sched_recovery.recover(
+                    schedule_store=FileScheduleStore(comp / "sched"),
+                    run_store=FileRunStore(comp / "runs"),
+                    now=datetime.now(UTC),
+                    ownership=outsider,
+                )
+        finally:
+            outsider.release()
+    finally:
+        holder.release()
+
+
+def test_released_or_missing_ownership_rejected(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store, run_store = _stores(comp)
+    never = SchedulerOwnership(comp, owner="never-acquired")
+    assert never.covers(sched_store.root, run_store.root) is False
+    with pytest.raises(SecondOwnerError):
+        sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=datetime.now(UTC),
+            ownership=never,
+        )
+    held = SchedulerOwnership(comp, owner="test").acquire()
+    assert held.covers(sched_store.root, run_store.root) is True
+    held.release()
+    assert held.covers(sched_store.root, run_store.root) is False
+    with pytest.raises(SecondOwnerError):
+        sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=datetime.now(UTC),
+            ownership=held,
+        )
+
+
+def test_root_mutation_cannot_redirect_held_authority(tmp_path: Path) -> None:
+    comp = tmp_path / "composition"
+    sched_store, run_store = _stores(comp)
+    other_sched, other_runs = _stores(tmp_path / "elsewhere")
+    ownership = SchedulerOwnership(comp, owner="test").acquire()
+    try:
+        assert ownership.covers(sched_store.root, run_store.root) is True
+        ownership.root = tmp_path / "elsewhere"
+        # Frozen identity: still proves the locked composition, never the
+        # mutated one.
+        assert ownership.canonical_root() == canonical_store_path(comp)
+        assert ownership.covers(sched_store.root, run_store.root) is True
+        assert ownership.covers(other_sched.root, other_runs.root) is False
+        with pytest.raises(SecondOwnerError):
+            sched_recovery.recover(
+                schedule_store=other_sched,
+                run_store=other_runs,
+                now=datetime.now(UTC),
+                ownership=ownership,
+            )
+    finally:
+        ownership.release()
+
+
+def test_canonical_windows_path_handling_retained(tmp_path: Path) -> None:
+    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"
+        )