sched: crash-safe admin ordering, create watermark, poll freshness (T13 fix)

src/wf_api/schedules.py

@@ -21,6 +21,7 @@ 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 StaleScheduleRevisionError
 
 from .models import (
     JsonProjector,
@@ -207,6 +208,10 @@ class WorkflowScheduleApi:
         )
         self._validate_definition(schedule)
         stored = store.create_schedule(schedule)
+        # Creation never backfills time before the revision: the consumed
+        # watermark starts at creation so catch-up only ever covers
+        # downtime after this revision, never pre-creation instants.
+        store.save_consumed(schedule_id, now)
         return _PROJECT_SCHEDULE(stored.model_dump(mode="json"))
 
     async def get_schedule(self, *, schedule_id: str) -> ScheduleResult:
@@ -251,14 +256,24 @@ class WorkflowScheduleApi:
         ``expected_revision + 1`` and ``updated_at`` becomes now, while
         ``created_at`` and the admitted-run snapshots stay untouched (later
         edits only affect future admissions). Stale revisions raise
-        ``StaleScheduleRevisionError``. Side effects mirror
-        ``Scheduler.edit_schedule``: a held candidate is cleared with a
-        ``superseded``/``schedule-edit`` history row, and the consumed
-        watermark advances to at least now (no backfill).
+        ``StaleScheduleRevisionError`` before any write. Crash-safe
+        ordering: the candidate is cleared (with a ``superseded`` /
+        ``schedule-edit`` history row) and the consumed watermark advances
+        BEFORE the revision bump is persisted, so a crash can only leave
+        the edit unapplied (safe over-skip under the old revision, freely
+        retryable) and never a bumped revision that backfills pre-edit
+        instants on restart. A concurrent edit that commits first still
+        wins via the store's authoritative revision check; our already
+        applied watermark advance is a safe over-skip in that case too.
         """
         store = self._schedule_store()
         now = datetime.now(UTC)
         current = store.get_schedule(schedule_id)
+        if current.revision != expected_revision:
+            raise StaleScheduleRevisionError(
+                f"stale schedule revision for {schedule_id!r}: "
+                f"expected {expected_revision}, found {current.revision}"
+            )
         data = current.model_dump(mode="python")
         if deployment_id is not None:
             data["deployment_id"] = deployment_id
@@ -285,7 +300,11 @@ class WorkflowScheduleApi:
         self._artifact_store().get_deployment(data["deployment_id"])
         updated = Schedule.model_validate(data)
         self._validate_definition(updated)
-        stored = store.update_schedule(updated, expected_revision=expected_revision)
+        # Crash-safe ordering: discard the old revision's unadmitted work
+        # and advance the watermark BEFORE the revision bump below. A
+        # crash here leaves the edit unapplied under the old revision
+        # (retryable); the bumped revision can never observe pre-edit
+        # instants on restart.
         old_candidate = store.get_candidate(schedule_id)
         if old_candidate is not None:
             FileScheduleHistoryRecorder(store).record(
@@ -293,7 +312,7 @@ class WorkflowScheduleApi:
                     schedule_id=schedule_id,
                     kind="superseded",
                     resolved_at=old_candidate.intended_at,
-                    revision=stored.revision,
+                    revision=expected_revision + 1,
                     reason="schedule-edit",
                     created_at=now,
                 )
@@ -303,6 +322,7 @@ class WorkflowScheduleApi:
         store.save_consumed(
             schedule_id, max(consumed, now) if consumed is not None else now
         )
+        stored = store.update_schedule(updated, expected_revision=expected_revision)
         return _PROJECT_SCHEDULE(stored.model_dump(mode="json"))
 
     async def pause_schedule(self, *, schedule_id: str) -> ScheduleResult:
@@ -310,18 +330,24 @@ class WorkflowScheduleApi:
 
         Sets ``paused``, clears any held candidate, and advances the
         consumed watermark to at least now so the paused span is never
-        backfilled on resume. No history row is written.
+        backfilled on resume. No history row is written. Crash-safe
+        ordering: the candidate is cleared and the watermark advances
+        BEFORE the flag flip is persisted, so a crash can only leave the
+        schedule unpaused (retryable) and never a paused flag whose span
+        backfills on resume.
         """
         store = self._schedule_store()
         now = datetime.now(UTC)
-        schedule = store.get_schedule(schedule_id)
-        schedule.paused = True
-        store.save_schedule(schedule)
+        # Existence first: unknown ids raise KeyError before any write.
+        store.get_schedule(schedule_id)
         store.save_candidate(None, schedule_id=schedule_id)
         consumed = store.get_consumed(schedule_id)
         store.save_consumed(
             schedule_id, max(consumed, now) if consumed is not None else now
         )
+        schedule = store.get_schedule(schedule_id)
+        schedule.paused = True
+        store.save_schedule(schedule)
         return _PROJECT_SCHEDULE(schedule.model_dump(mode="json"))
 
     async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult:
@@ -329,28 +355,37 @@ class WorkflowScheduleApi:
 
         Clears ``paused`` (resume selects the next future occurrence),
         clears any held candidate, and advances the consumed watermark to
-        at least now. No history row is written.
+        at least now. No history row is written. Crash-safe ordering like
+        pause: candidate and watermark first, flag flip last.
         """
         store = self._schedule_store()
         now = datetime.now(UTC)
-        schedule = store.get_schedule(schedule_id)
-        schedule.paused = False
-        store.save_schedule(schedule)
+        # Existence first: unknown ids raise KeyError before any write.
+        store.get_schedule(schedule_id)
+        store.save_candidate(None, schedule_id=schedule_id)
         consumed = store.get_consumed(schedule_id)
         store.save_consumed(
             schedule_id, max(consumed, now) if consumed is not None else now
         )
-        store.save_candidate(None, schedule_id=schedule_id)
+        schedule = store.get_schedule(schedule_id)
+        schedule.paused = False
+        store.save_schedule(schedule)
         return _PROJECT_SCHEDULE(schedule.model_dump(mode="json"))
 
     async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult:
         """Soft-delete one schedule (mirror the poll-loop deleted branch).
 
-        Sets ``deleted`` and clears any held candidate. Runs and history
+        Clears any held candidate, then sets ``deleted``. Runs and history
         are untouched, the consumed watermark does not advance, and the id
-        stays reserved (re-creation is rejected by create).
+        stays reserved (re-creation is rejected by create). Either crash
+        half heals: an unclearable candidate on a deleted schedule is
+        dropped by the poll loop, and a cleared candidate on a live
+        schedule is rebuilt from the untouched watermark.
         """
         store = self._schedule_store()
+        # Existence first: unknown ids raise KeyError before any write.
+        store.get_schedule(schedule_id)
+        store.save_candidate(None, schedule_id=schedule_id)
         schedule = store.get_schedule(schedule_id)
         schedule.deleted = True
         store.save_schedule(schedule)

src/wf_client/schedules.py

@@ -32,6 +32,8 @@ class Schedule:
     enabled: bool
     paused: bool
     deleted: bool
+    exhausted: bool
+    blocked_reason: str | None
     overlap: str
     misfire: str
     max_active_runs: int
@@ -50,6 +52,8 @@ class Schedule:
         enabled: bool,
         paused: bool,
         deleted: bool,
+        exhausted: bool,
+        blocked_reason: str | None,
         overlap: str,
         misfire: str,
         max_active_runs: int,
@@ -65,6 +69,8 @@ class Schedule:
         object.__setattr__(self, "enabled", enabled)
         object.__setattr__(self, "paused", paused)
         object.__setattr__(self, "deleted", deleted)
+        object.__setattr__(self, "exhausted", exhausted)
+        object.__setattr__(self, "blocked_reason", blocked_reason)
         object.__setattr__(self, "overlap", overlap)
         object.__setattr__(self, "misfire", misfire)
         object.__setattr__(self, "max_active_runs", max_active_runs)
@@ -135,6 +141,8 @@ class Schedule:
             enabled=wire["enabled"],
             paused=wire["paused"],
             deleted=wire["deleted"],
+            exhausted=wire["exhausted"],
+            blocked_reason=wire["blocked_reason"],
             overlap=wire["overlap"],
             misfire=wire["misfire"],
             max_active_runs=wire["max_active_runs"],

src/wf_scheduling/poll.py

@@ -582,6 +582,22 @@ class Scheduler:
             self._execute_guarded(run.id, now)
 
     def _poll_one(self, sched: Any, now: datetime) -> str:
+        # Fresh read per schedule: the tick lists schedules up front while
+        # same-process administration may commit an edit, pause, resume, or
+        # delete mid-tick. All admission decisions below (policies,
+        # revision, flags) use this fresh copy, never the listing
+        # snapshot. Two narrow residuals remain: calendar iteration uses
+        # the tick-start occurrence source (a trigger edit takes effect
+        # on the next tick), and an admission decision already in flight
+        # cannot observe a concurrent edit — admission stays idempotent
+        # per (schedule, instant), so the worst case is one run admitted
+        # under just-superseded terms, never a duplicate or replay.
+        try:
+            sched = self.schedule_store.get_schedule(sched.id)
+        except KeyError:
+            # Soft deletes never remove the file; a vanishing schedule is
+            # unexpectedly gone — treat it as deleted work, never admit.
+            return "deleted"
         if sched.deleted:
             if self.schedule_store.get_candidate(sched.id) is not None:
                 self.schedule_store.save_candidate(None, schedule_id=sched.id)
@@ -806,22 +822,33 @@ class Scheduler:
 
     # -- administration ---------------------------------------------------
     def resume_schedule(self, sid: str, now: datetime) -> None:
-        """Unpause: resume selects the next future occurrence."""
+        """Unpause: resume selects the next future occurrence.
+
+        Crash-safe ordering shared with the API surface: the candidate
+        is cleared and the watermark advances BEFORE the flag flip is
+        persisted, so a crash can only leave the schedule paused
+        (retryable) and never an unpaused flag whose span backfills.
+        """
         self._require_ownership()
+        self.schedule_store.get_schedule(sid)
+        self.schedule_store.save_candidate(None, schedule_id=sid)
+        consumed = self.schedule_store.get_consumed(sid) or EPOCH
+        self.schedule_store.save_consumed(sid, max(consumed, now))
         sched = self.schedule_store.get_schedule(sid)
         sched.paused = False
         self.schedule_store.save_schedule(sched)
-        consumed = self.schedule_store.get_consumed(sid) or EPOCH
-        self.schedule_store.save_consumed(sid, max(consumed, now))
-        self.schedule_store.save_candidate(None, schedule_id=sid)
 
     def edit_schedule(self, sid: str, now: datetime) -> None:
-        """Definition edit: new revision, discard old candidates, no backfill."""
+        """Definition edit: new revision, discard old candidates, no backfill.
+
+        Crash-safe ordering shared with the API surface: old candidates
+        are discarded (with a superseded row) and the watermark advances
+        BEFORE the revision bump is persisted, so a crash can only leave
+        the edit unapplied under the old revision (retryable) and never
+        a bumped revision that backfills pre-edit instants on restart.
+        """
         self._require_ownership()
         sched = self.schedule_store.get_schedule(sid)
-        sched.revision += 1
-        sched.updated_at = now
-        self.schedule_store.save_schedule(sched)
         old = self.schedule_store.get_candidate(sid)
         if old is not None:
             self._record(
@@ -829,11 +856,15 @@ class Scheduler:
                 sched_id=sid,
                 intended=old.intended_at,
                 reason="schedule-edit",
-                revision=sched.revision,
+                revision=sched.revision + 1,
             )
             self.schedule_store.save_candidate(None, schedule_id=sid)
         consumed = self.schedule_store.get_consumed(sid) or EPOCH
         self.schedule_store.save_consumed(sid, max(consumed, now))
+        sched = self.schedule_store.get_schedule(sid)
+        sched.revision += 1
+        sched.updated_at = now
+        self.schedule_store.save_schedule(sched)
 
 
 def _admission_intended(run_store: Any, run_id: str) -> datetime | None:

tests/scheduling/test_poll.py

@@ -303,3 +303,53 @@ def test_capacity_wait_then_expire_for_skip(tmp_path: Path) -> None:
     sched.poll(t0 + timedelta(seconds=30))
     assert len([r for r in _history(store, "a") if r["kind"] == "admitted"]) == 1
     sched.ownership.release()
+
+
+def test_poll_one_rereads_pause_before_deciding(tmp_path: Path) -> None:
+    sched, store, runs, sources = _harness(tmp_path, script={"*": "hang"})
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1))
+    stale = store.get_schedule("a")
+    # Same-process administration lands after the tick listed schedules:
+    # the decision must observe the pause, not the stale snapshot.
+    live = store.get_schedule("a")
+    live.paused = True
+    store.save_schedule(live)
+    assert sched._poll_one(stale, t0 + timedelta(seconds=1)) == "paused"
+    assert runs.list_runs() == []
+    assert runs.list_admissions() == []
+    sched.ownership.release()
+
+
+def test_poll_one_uses_fresh_definition_after_edit(tmp_path: Path) -> None:
+    sched, store, runs, sources = _harness(tmp_path, script={"*": "hang"})
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(
+        sched,
+        store,
+        sources,
+        "a",
+        OneShotSource(t0),
+        t0 - timedelta(hours=1),
+        input_bindings=[
+            {"target": "team", "expression": {"kind": "literal", "value": "old"}}
+        ],
+    )
+    stale = store.get_schedule("a")
+    live = store.get_schedule("a")
+    live.revision = 2
+    from wf_core.models.input_bindings import ScheduleInputBinding
+
+    live.input_bindings = [
+        ScheduleInputBinding.model_validate(
+            {"target": "team", "expression": {"kind": "literal", "value": "new"}}
+        )
+    ]
+    store.save_schedule(live)
+    result = sched._poll_one(stale, t0 + timedelta(seconds=1))
+    assert result.startswith("admit:run-")
+    run_id = result.split(":", 1)[1]
+    admission = runs.get_admission(run_id)
+    assert admission.resolved_input["team"] == "new"
+    assert admission.schedule_revision == 2
+    sched.ownership.release()

tests/wf_api/test_schedules.py

@@ -260,6 +260,32 @@ def _history_rows(
     return rows
 
 
+class FailConsumedOnce(FileScheduleStore):
+    """Real schedule store failing exactly one watermark write."""
+
+    def __init__(self, root: Path) -> None:
+        super().__init__(root)
+        self.armed = False
+
+    def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None:
+        if self.armed:
+            self.armed = False
+            raise OSError("injected consumed failure")
+        super().save_consumed(schedule_id, consumed_through)
+
+
+def _fault_harness(
+    root: Path,
+) -> tuple[WorkflowScheduleApi, FailConsumedOnce, FileRunStore]:
+    artifact_store = FileWorkflowArtifactStore(root)
+    artifact_store.save_artifact(_artifact())
+    artifact_store.save_deployment(_deployment())
+    run_store = FileRunStore(root)
+    sched_store = FailConsumedOnce(root)
+    context = _context(artifact_store, run_store, sched_store)
+    return WorkflowScheduleApi(context), sched_store, run_store
+
+
 async def test_create_get_list_round_trip_with_defaults(tmp_path: Path) -> None:
     api, sched_store, _, _, _ = _harness(tmp_path / "round_trip")
 
@@ -463,13 +489,81 @@ async def test_update_happy_path_only_future_work(tmp_path: Path) -> None:
 
 
 async def test_update_stale_revision_rejected(tmp_path: Path) -> None:
-    api, _, _, _, _ = _harness(tmp_path / "stale")
+    api, sched_store, _, _, _ = _harness(tmp_path / "stale")
+
+    created = await api.create_schedule(
+        schedule_id="s", deployment_id="dep.personal", trigger=_cron()
+    )
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.save_candidate(
+        PendingCandidate(schedule_id="s", intended_at=intended, revision=1),
+        schedule_id="s",
+    )
+    consumed_before = sched_store.get_consumed("s")
+    with pytest.raises(StaleScheduleRevisionError):
+        await api.update_schedule(schedule_id="s", expected_revision=99)
+    # Stale edits write nothing: no revision bump, no candidate or
+    # watermark change, no history row.
+    assert sched_store.get_schedule("s").revision == 1
+    assert sched_store.get_schedule("s").updated_at == datetime.fromisoformat(
+        created["updated_at"]
+    )
+    candidate = sched_store.get_candidate("s")
+    assert candidate is not None and candidate.intended_at == intended
+    assert sched_store.get_consumed("s") == consumed_before
+    assert _history_rows(sched_store, "s") == []
+
+
+async def test_create_initializes_consumed_no_backfill(tmp_path: Path) -> None:
+    api, sched_store, _, _, _ = _harness(tmp_path / "create_consumed")
+
+    created = await api.create_schedule(
+        schedule_id="s",
+        deployment_id="dep.personal",
+        trigger=_cron(),
+        misfire="latest",
+    )
+    created_at = datetime.fromisoformat(created["created_at"])
+    consumed = sched_store.get_consumed("s")
+    assert consumed is not None and consumed >= created_at
+
+
+async def test_update_watermark_failure_leaves_revision(tmp_path: Path) -> None:
+    api, sched_store, _ = _fault_harness(tmp_path / "torn_update")
 
     await api.create_schedule(
         schedule_id="s", deployment_id="dep.personal", trigger=_cron()
     )
-    with pytest.raises(StaleScheduleRevisionError):
-        await api.update_schedule(schedule_id="s", expected_revision=99)
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.save_candidate(
+        PendingCandidate(schedule_id="s", intended_at=intended, revision=1),
+        schedule_id="s",
+    )
+    sched_store.armed = True
+    with pytest.raises(OSError, match="injected consumed failure"):
+        await api.update_schedule(
+            schedule_id="s",
+            expected_revision=1,
+            overlap="parallel",
+            max_active_runs=2,
+        )
+    # Crash-safe ordering: the watermark advance precedes the revision
+    # bump, so a torn edit leaves the old revision (retryable) and can
+    # never backfill pre-edit instants under the new revision.
+    assert sched_store.get_schedule("s").revision == 1
+    assert sched_store.get_schedule("s").overlap == "skip"
+
+
+async def test_pause_watermark_failure_leaves_flag(tmp_path: Path) -> None:
+    api, sched_store, _ = _fault_harness(tmp_path / "torn_pause")
+
+    await api.create_schedule(
+        schedule_id="s", deployment_id="dep.personal", trigger=_cron()
+    )
+    sched_store.armed = True
+    with pytest.raises(OSError, match="injected consumed failure"):
+        await api.pause_schedule(schedule_id="s")
+    assert sched_store.get_schedule("s").paused is False
 
 
 async def test_unknown_schedule_keyerror(tmp_path: Path) -> None:
@@ -491,10 +585,14 @@ async def test_unknown_schedule_keyerror(tmp_path: Path) -> None:
 
 async def test_pause_resume_delete_transitions(tmp_path: Path) -> None:
     api, sched_store, _, _, _ = _harness(tmp_path / "transitions")
-    await api.create_schedule(
+    created = await api.create_schedule(
         schedule_id="s", deployment_id="dep.personal", trigger=_cron()
     )
-    assert sched_store.get_consumed("s") is None
+    # Creation never backfills time before the revision: the consumed
+    # watermark starts at creation.
+    created_consumed = sched_store.get_consumed("s")
+    assert created_consumed is not None
+    assert created_consumed >= datetime.fromisoformat(created["created_at"])
 
     intended = ts(2026, 9, 8, 12, 0)
     sched_store.save_candidate(

tests/wf_client/test_schedules.py

@@ -128,6 +128,8 @@ async def test_schedule_snapshot_exposes_minimal_surface() -> None:
     assert schedule.enabled is True
     assert schedule.paused is False
     assert schedule.deleted is False
+    assert schedule.exhausted is False
+    assert schedule.blocked_reason is None
     assert schedule.overlap == "skip"
     assert schedule.misfire == "skip"
     assert schedule.max_active_runs == 1