fix: serialize scheduler transitions and coalesce misfires

src/wf_api/schedules.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from datetime import UTC, datetime
+from functools import wraps
 from typing import Any
 
 from wf_core import RunLimits
@@ -21,7 +22,11 @@ from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry
 from wf_scheduling.models import OccurrenceRecord, Schedule
 from wf_scheduling.occurrences import occurrence_id
 from wf_scheduling.prepare import PreparationRejected, SchedulePreparer
-from wf_scheduling.store import ScheduleExistsError, StaleScheduleRevisionError
+from wf_scheduling.store import (
+    ScheduleExistsError,
+    StaleScheduleRevisionError,
+    schedule_store_transaction,
+)
 
 from .models import (
     JsonProjector,
@@ -38,6 +43,18 @@ _PROJECT_LIST_SCHEDULES = JsonProjector(ListSchedulesResult)
 _PROJECT_OCCURRENCE_PAGE = JsonProjector(OccurrencePage)
 
 
+def _serialize_schedule_write(method: Any) -> Any:
+    """Serialize one API schedule mutation with the store's local lock."""
+
+    @wraps(method)
+    async def wrapped(self: Any, *args: Any, **kwargs: Any) -> Any:
+        store = self._schedule_store()
+        with schedule_store_transaction(store):
+            return await method(self, *args, **kwargs)
+
+    return wrapped
+
+
 class _ContextDeploymentDirectory:
     """Deployment contract source over the API artifact store.
 
@@ -177,6 +194,7 @@ class WorkflowScheduleApi:
         self._check_trigger(schedule)
         self._validate_sample(schedule)
 
+    @_serialize_schedule_write
     async def create_schedule(
         self,
         *,
@@ -261,6 +279,7 @@ class WorkflowScheduleApi:
             }
         )
 
+    @_serialize_schedule_write
     async def update_schedule(
         self,
         *,
@@ -351,6 +370,7 @@ class WorkflowScheduleApi:
         stored = store.update_schedule(updated, expected_revision=expected_revision)
         return _PROJECT_SCHEDULE(stored.model_dump(mode="json"))
 
+    @_serialize_schedule_write
     async def pause_schedule(self, *, schedule_id: str) -> ScheduleResult:
         """Pause one schedule (mirror the poll-loop paused branch).
 
@@ -376,6 +396,7 @@ class WorkflowScheduleApi:
         store.save_schedule(schedule)
         return _PROJECT_SCHEDULE(schedule.model_dump(mode="json"))
 
+    @_serialize_schedule_write
     async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult:
         """Resume one schedule (mirror ``Scheduler.resume_schedule``).
 
@@ -398,6 +419,7 @@ class WorkflowScheduleApi:
         store.save_schedule(schedule)
         return _PROJECT_SCHEDULE(schedule.model_dump(mode="json"))
 
+    @_serialize_schedule_write
     async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult:
         """Soft-delete one schedule (mirror the poll-loop deleted branch).
 

src/wf_scheduling/poll.py

@@ -38,6 +38,7 @@ from wf_scheduling.ownership import (
     describe_unsupported_layout,
 )
 from wf_scheduling.prepare import InvocationPreparer, PreparationRejected
+from wf_scheduling.store import schedule_store_transaction
 
 UTC = timezone.utc
 SCAN_CAP = 100
@@ -303,6 +304,23 @@ class Scheduler:
         return None
 
     def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None:
+        """Admit one occurrence under the schedule store's local transaction."""
+        with schedule_store_transaction(self.schedule_store):
+            return self._admit_locked(sched, intended, now)
+
+    def _admit_locked(
+        self, sched: Any, intended: datetime, now: datetime
+    ) -> str | None:
+        # The poller's listing is only a fairness snapshot. Re-read under the
+        # admission transaction so an edit committed before this point wins;
+        # an edit after this point waits and affects a later occurrence.
+        try:
+            current = self.schedule_store.get_schedule(sched.id)
+        except KeyError:
+            return None
+        if current.revision != sched.revision:
+            return "schedule-changed"
+        sched = current
         if getattr(sched, "blocked_reason", None):
             raise BlockedSchedule(getattr(sched, "blocked_reason"))
         if not sched.enabled or sched.deleted or sched.paused:
@@ -820,6 +838,48 @@ class Scheduler:
             )
             self.schedule_store.save_consumed(sched.id, now)
             return "span-skipped"
+        if (
+            sched.misfire == "latest"
+            and due
+            and any(
+                (now - instant).total_seconds() > sched.lateness_allowance_s
+                for instant in due
+            )
+        ):
+            # A bounded catch-up can still contain several missed instants.
+            # Coalesce them before the per-instant loop so ``latest`` never
+            # turns a short downtime into a replay burst.
+            latest = due[-1]
+            old = self.schedule_store.get_candidate(sched.id)
+            if old is not None and old.intended_at != latest:
+                self._record(
+                    kind="superseded",
+                    sched_id=sched.id,
+                    intended=old.intended_at,
+                    reason=f"coalesced-into:{latest.isoformat()}",
+                    revision=sched.revision,
+                    now=now,
+                )
+            self.schedule_store.save_candidate(
+                PendingCandidate(
+                    schedule_id=sched.id,
+                    intended_at=latest,
+                    revision=sched.revision,
+                ),
+                schedule_id=sched.id,
+            )
+            self._record(
+                kind="interval-summary",
+                sched_id=sched.id,
+                reason="coalesced-missed-span",
+                revision=sched.revision,
+                interval=(consumed, now),
+                count=-1,
+                now=now,
+            )
+            self.schedule_store.save_consumed(sched.id, now)
+            held = self._admit_held_candidate(sched, now)
+            return "skipped-overlap" if held is None else held
         last_result = "idle"
         for instant in due:
             stored_consumed = self.schedule_store.get_consumed(sched.id) or EPOCH

src/wf_scheduling/recovery.py

@@ -171,6 +171,23 @@ def recover(
             clear_executing(run_store, run.id)
             clear_pending(run_store, run.id)
             diags.append(f"{run.id}:completion-window-cleared")
+        if status in (
+            StoredRunStatus.ADMITTED.value,
+            StoredRunStatus.INTERRUPTED.value,
+            "admitted",
+            "interrupted",
+        ):
+            try:
+                run_store.get_admission(run.id)
+            except KeyError:
+                # Without an admission there is no trusted schedule owner or
+                # invocation snapshot. Fail the view closed and never let it
+                # remain active merely because recovery cannot attribute it.
+                _fail_run(run_store, run, CORRUPT_VIEW_REASON, now, history)
+                clear_pending(run_store, run.id)
+                clear_executing(run_store, run.id)
+                diags.append(f"{run.id}:failed-closed")
+                continue
         # 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
@@ -283,13 +300,6 @@ def recover(
             if _is_pending(run_store, run.id):
                 diags.append(f"{run.id}:pending-dispatch-kept")
                 continue
-            try:
-                run_store.get_admission(run.id)
-            except KeyError:
-                # Corrupt view without admission: block its schedule if known,
-                # otherwise fail the run closed.
-                diags.append(f"{run.id}:corrupt-blocked")
-                continue
             if active:
                 _fail_run(run_store, run, AMBIGUOUS_REASON, now, history)
             else:

src/wf_scheduling/store.py

@@ -19,15 +19,34 @@ Scheduler ownership across processes arrives in T11.
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
+from contextlib import AbstractContextManager, contextmanager
 from datetime import datetime, timezone
 from pathlib import Path
 from threading import RLock
+from typing import Any, cast
 
 from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id
 
 UTC = timezone.utc
 
 
+@contextmanager
+def schedule_store_transaction(store: object) -> Iterator[None]:
+    """Serialize a multi-file schedule transition when the store supports it.
+
+    The scheduler also accepts test doubles and future transactional stores.
+    Those may not expose this optional local transaction seam, in which case
+    the caller retains its existing behavior.
+    """
+    transaction: Any = getattr(store, "transaction", None)
+    if callable(transaction):
+        with cast(AbstractContextManager[None], transaction()):
+            yield
+    else:
+        yield
+
+
 def _history_identity(record: OccurrenceRecord) -> tuple[object, ...]:
     """Return fields that identify one durable occurrence transition."""
     return (
@@ -66,6 +85,12 @@ class FileScheduleStore:
     def schedules_dir(self) -> Path:
         return self.root / "schedules"
 
+    @contextmanager
+    def transaction(self) -> Iterator[None]:
+        """Hold the process-local store lock across a compound transition."""
+        with self._lock:
+            yield
+
     # -- schedules ------------------------------------------------------
     def create_schedule(self, schedule: Schedule) -> Schedule:
         """Persist a new schedule; deleted ids are never reusable."""

tests/scheduling/test_poll.py

@@ -173,6 +173,82 @@ def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None:
         sched.ownership.release()
 
 
+def test_latest_short_downtime_does_not_burst_admit(tmp_path: Path) -> None:
+    """Latest misfire coalesces several bounded missed starts to one run."""
+    sched, store, runs, sources = _harness(tmp_path, script={"*": "complete"})
+    start = ts(2026, 9, 8, 9, 0)
+    now = ts(2026, 9, 8, 12, 0)
+    _add(
+        sched,
+        store,
+        sources,
+        "m",
+        PeriodicSource(timedelta(hours=1), start),
+        start,
+        misfire="latest",
+    )
+
+    sched.poll(now)
+
+    admitted = [row for row in _history(store, "m") if row["kind"] == "admitted"]
+    assert len(admitted) == 1
+    assert datetime.fromisoformat(admitted[0]["resolved_at"]) == now
+    sched.ownership.release()
+
+
+def test_schedule_edit_between_poll_snapshot_and_admission_cannot_overwrite_terms(
+    tmp_path: Path,
+) -> None:
+    """An edit between the poll snapshot and admission wins authoritatively."""
+
+    class EditOnPollRead(FileScheduleStore):
+        def __init__(self, root: Path) -> None:
+            super().__init__(root)
+            self.armed = False
+
+        def get_schedule(self, schedule_id: str) -> Schedule:
+            current = super().get_schedule(schedule_id)
+            if self.armed and schedule_id == "a":
+                self.armed = False
+                edited = current.model_copy(
+                    update={
+                        "revision": current.revision + 1,
+                        "updated_at": current.updated_at + timedelta(seconds=1),
+                    }
+                )
+                super().update_schedule(edited, expected_revision=current.revision)
+                return current
+            return current
+
+    sched_store = EditOnPollRead(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sources: dict[str, Any] = {}
+    delegate = SchedulePreparer(
+        DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+        fixture_environment,
+    )
+
+    sched = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources=sources,
+        capacity=1,
+        preparer=delegate,
+        dispatcher=ScriptedDispatcher({"*": "complete"}),
+        ownership=SchedulerOwnership(tmp_path, owner="test").acquire(),
+    )
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(sched, sched_store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1))
+    sched_store.armed = True
+    try:
+        result = sched.poll(t0)
+        assert result["a"] == "admit:schedule-changed"
+        assert sched_store.get_schedule("a").revision == 2
+        assert run_store.list_admissions() == []
+    finally:
+        sched.ownership.release()
+
+
 def test_parallel_limits_and_interrupted_slots() -> None:
     import tempfile
 

tests/scheduling/test_recovery.py

@@ -110,6 +110,47 @@ def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> Non
     assert run_store.get_run(admission.id).status.value == "failed"
 
 
+def test_recovery_fails_corrupt_view_without_admission(tmp_path: Path) -> None:
+    """A view with no admission is failed closed instead of left active."""
+    from tests.artifacts.test_run_store import artifact as _artifact
+    from tests.artifacts.test_run_store import deployment as _deployment
+    from wf_artifacts import PinnedRunEnvironment, ResumeReadiness, WorkflowRunRecord
+    from wf_artifacts.runs.models import StoredRunStatus
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    now = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    run_store.save_run(
+        WorkflowRunRecord(
+            id=run_id,
+            status=StoredRunStatus.ADMITTED,
+            resume_readiness=ResumeReadiness.NOT_APPLICABLE,
+            environment=PinnedRunEnvironment(
+                deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[]
+            ),
+            latest_checkpoint_id=None,
+            created_at=now,
+            updated_at=now,
+        )
+    )
+
+    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
+    try:
+        diags = sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=now,
+            ownership=ownership,
+        )
+    finally:
+        ownership.release()
+
+    assert any(f"{run_id}:failed-closed" in item for item in diags)
+    assert run_store.get_run(run_id).status.value == "failed"
+
+
 def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None:
 
     from tests.artifacts.test_run_store import artifact as _artifact