test: harden scheduling review boundaries

tests/wf_api/test_schedules.py

@@ -8,12 +8,18 @@ tests/wf_api/test_run_lifecycle.py).
 
 from __future__ import annotations
 
+import asyncio
+import threading
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
 import pytest
 
+from tests.scheduling.controlled import (
+    DictDeployments,
+    ScriptedDispatcher,
+)
 from wf_api import WorkflowApi
 from wf_api.durable_context import durable_workflow_api
 from wf_api.models import TraceRange
@@ -36,9 +42,12 @@ from wf_artifacts import (
 from wf_authoring import NodeSpec
 from wf_core import InterruptRequest, RunState, RunStatus
 from wf_platform import CapabilitySource
+from wf_scheduling.calendar import OneShotSource
 from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry
 from wf_scheduling.models import PendingCandidate
 from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import SchedulePreparer
 from wf_scheduling.recovery import recover
 from wf_scheduling.store import (
     FileScheduleStore,
@@ -516,6 +525,74 @@ async def test_update_stale_revision_rejected(tmp_path: Path) -> None:
     assert _history_rows(sched_store, "s") == []
 
 
+async def test_admin_mutation_waits_for_poll_transaction(tmp_path: Path) -> None:
+    """A poll transition and an API edit cannot interleave store writes."""
+    api, sched_store, run_store, artifact_store, _ = _harness(
+        tmp_path / "admin_poll"
+    )
+    await api.create_schedule(
+        schedule_id="s", deployment_id="dep.personal", trigger=_cron()
+    )
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.save_consumed("s", intended - timedelta(hours=1))
+
+    entered = threading.Event()
+    release = threading.Event()
+    block_once = True
+    original_save_consumed = sched_store.save_consumed
+
+    def block_poll_watermark(schedule_id: str, consumed_through: datetime) -> None:
+        nonlocal block_once
+        if block_once:
+            block_once = False
+            entered.set()
+            if not release.wait(timeout=5):
+                raise TimeoutError("poll transaction did not get released")
+        original_save_consumed(schedule_id, consumed_through)
+
+    sched_store.save_consumed = block_poll_watermark  # type: ignore[method-assign]
+    ownership = SchedulerOwnership(tmp_path / "admin_poll", owner="poll").acquire()
+    scheduler = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={"s": OneShotSource(intended)},
+        capacity=1,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep.personal": {"rev": 1, "required": []}}),
+            lambda sched: create_pinned_environment(
+                deployment=artifact_store.get_deployment(sched.deployment_id),
+                artifact=artifact_store.get_artifact("sched-art", 1),
+                tree=SavedSubgraphTree(artifacts_by_ref={}, diagnostics=[]),
+            ),
+        ),
+        dispatcher=ScriptedDispatcher({"*": "complete"}),
+        ownership=ownership,
+    )
+    release_timer = threading.Timer(0.2, release.set)
+    try:
+        poll_task = asyncio.create_task(asyncio.to_thread(scheduler.poll, intended))
+        assert await asyncio.to_thread(entered.wait, 5)
+        edit_task = asyncio.create_task(
+            api.update_schedule(
+                schedule_id="s",
+                expected_revision=1,
+                overlap="parallel",
+                max_active_runs=2,
+            )
+        )
+        release_timer.start()
+        poll_result, edited = await asyncio.gather(poll_task, edit_task)
+    finally:
+        release.set()
+        release_timer.cancel()
+        ownership.release()
+
+    assert poll_result["s"].startswith("admit:")
+    assert edited["revision"] == 2
+    assert run_store.list_admissions()[0].schedule_revision == 1
+    assert sched_store.get_schedule("s").revision == 2
+
+
 async def test_create_initializes_consumed_no_backfill(tmp_path: Path) -> None:
     api, sched_store, _, _, _ = _harness(tmp_path / "create_consumed")
 
@@ -959,16 +1036,20 @@ async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) -
 
     seen: list[tuple[str, str | None]] = []
     cursor: str | None = None
-    while True:
+    for _ in range(10):
         page = await api.list_schedule_occurrences(
             schedule_id="s", cursor=cursor, limit=1
         )
         assert page["total"] == 3
         for row in page["occurrences"]:
             seen.append((row["kind"], row["checkpoint_id"]))
-        cursor = page["next_cursor"]
-        if cursor is None:
+        next_cursor = page["next_cursor"]
+        if next_cursor is None:
             break
+        assert next_cursor != cursor
+        cursor = next_cursor
+    else:
+        pytest.fail("occurrence pagination did not terminate")
     assert seen == [
         ("admitted", None),
         ("interrupted", "run-1.000001"),