sched: address R3 blockers on pagination, intervals, and fail-closed polling

src/wf_scheduling/models.py

@@ -58,6 +58,32 @@ class CronTrigger(BaseModel):
     expression: str
     timezone: str = "UTC"
 
+    @field_validator("expression")
+    @classmethod
+    def _require_five_field_cron(cls, value: str) -> str:
+        # Thin validation: five fields and croniter-constructible. Full DST
+        # semantics belong to the calendar adapter (croniter owns them).
+        if len(value.split()) != 5:
+            raise ValueError(f"cron expression must have exactly 5 fields: {value!r}")
+        try:
+            from wf_scheduling.calendar import CronSource
+
+            CronSource(value, "UTC")
+        except Exception as exc:
+            raise ValueError(f"invalid cron expression {value!r}: {exc}") from exc
+        return value
+
+    @field_validator("timezone")
+    @classmethod
+    def _require_valid_zone(cls, value: str) -> str:
+        try:
+            from zoneinfo import ZoneInfo
+
+            ZoneInfo(value)
+        except Exception as exc:
+            raise ValueError(f"invalid time zone {value!r}") from exc
+        return value
+
 
 class OneShotTrigger(BaseModel):
     """Single offset-aware instant trigger."""

src/wf_scheduling/poll.py

@@ -22,6 +22,14 @@ SCAN_CAP = 100
 EPOCH = datetime(2020, 1, 1, tzinfo=UTC)
 
 
+class BlockedSchedule(ValueError):
+    """A schedule is blocked by a corrupt record and fails closed."""
+
+
+class InvalidScheduleDefinitionError(ValueError):
+    """A schedule definition or its trigger source is invalid."""
+
+
 class OccurrenceSource(Protocol):
     """Due-instant source over aware datetimes, UTC at the boundary."""
 
@@ -53,7 +61,12 @@ def source_for_trigger(trigger: Any) -> OccurrenceSource:
 
 
 def _is_oneshot(source: OccurrenceSource) -> bool:
-    return hasattr(source, "at") and not hasattr(source, "period")
+    # Calendar and test-double one-shots share the class name and an ``at``
+    # instant; periodic sources carry a ``period`` instead. isinstance covers
+    # the calendar type; the name check covers duck-typed test doubles.
+    if isinstance(source, OneShotSource):
+        return True
+    return type(source).__name__ == "OneShotSource" and hasattr(source, "at")
 
 
 class Scheduler:
@@ -129,12 +142,17 @@ class Scheduler:
         interval: tuple[datetime, datetime] | None = None,
         count: int = 0,
         revision: int | None = None,
+        now: datetime | None = None,
+        admitted_at: datetime | None = None,
+        started_at: datetime | None = None,
     ) -> None:
-        oid = (
-            occurrence_id(sched_id, intended)
-            if intended is not None
-            else f"{sched_id}|no-instant"
-        )
+        created = now if now is not None else datetime.now(UTC)
+        if intended is not None:
+            oid = occurrence_id(sched_id, intended)
+        elif interval is not None:
+            oid = f"{sched_id}|summary|{interval[0].isoformat()}|{interval[1].isoformat()}"
+        else:
+            oid = f"{sched_id}|summary|{created.isoformat()}"
         self.schedule_store.append_history(
             OccurrenceRecord(
                 schedule_id=sched_id,
@@ -144,18 +162,30 @@ class Scheduler:
                 run_id=run_id,
                 revision=revision,
                 reason=reason,
+                admitted_at=admitted_at,
+                started_at=started_at,
                 interval_start=interval[0] if interval else None,
                 interval_end=interval[1] if interval else None,
                 interval_count=count,
-                created_at=datetime.now(UTC),
+                created_at=created,
             )
         )
 
     def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None:
         if getattr(sched, "blocked_reason", None):
-            raise ValueError(getattr(sched, "blocked_reason"))
+            raise BlockedSchedule(getattr(sched, "blocked_reason"))
         if not sched.enabled or sched.deleted or sched.paused:
             return None
+        # Fail closed before any new admission when a corrupt active view
+        # exists for this schedule.
+        for run in self._active(sched.id):
+            try:
+                self.run_store.get_admission(run.id)
+            except KeyError:
+                reason = f"corrupt run view without admission: {run.id}"
+                sched.blocked_reason = reason
+                self.schedule_store.save_schedule(sched)
+                raise BlockedSchedule(reason) from None
         dep = self.deployments.get(sched.deployment_id)
         if dep is None:
             self._record(
@@ -164,7 +194,11 @@ class Scheduler:
                 intended=intended,
                 reason="deployment-deleted",
                 revision=sched.revision,
+                now=now,
             )
+            cand = self.schedule_store.get_candidate(sched.id)
+            if cand is not None and cand.intended_at == intended:
+                self.schedule_store.save_candidate(None, schedule_id=sched.id)
             self.schedule_store.save_consumed(sched.id, intended)
             return None
         frozen = {
@@ -259,6 +293,8 @@ class Scheduler:
             run_id=run_id,
             reason=f"rev={sched.revision}",
             revision=sched.revision,
+            now=now,
+            admitted_at=now,
         )
         if _is_oneshot(self.sources.get(sched.id)):  # type: ignore[arg-type]
             sched.exhausted = True
@@ -277,7 +313,15 @@ class Scheduler:
         from wf_artifacts.runs.models import StoredRunStatus
 
         outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete"))
+        if outcome not in ("complete", "hang", "interrupt"):
+            raise ValueError(f"unknown controlled outcome {outcome!r} for {run_id!r}")
+        try:
             record = self.run_store.get_run(run_id)
+        except KeyError as exc:
+            raise BlockedSchedule(f"dispatch missing run view: {run_id!r}") from exc
+        sched_id = self._run_sched(record)
+        if sched_id is None:
+            raise BlockedSchedule(f"dispatch missing admission: {run_id!r}")
         if outcome == "hang":
             return
         if outcome == "interrupt":
@@ -287,21 +331,24 @@ class Scheduler:
             self.run_store.save_run(updated)
             self._record(
                 kind="interrupted",
-                sched_id=self._run_sched(record) or "",
+                sched_id=sched_id,
                 intended=_admission_intended(self.run_store, run_id),
                 run_id=run_id,
+                now=now,
+                started_at=now,
             )
             return
-        assert outcome == "complete"
         updated = record.model_copy(
             update={"status": StoredRunStatus.COMPLETED, "updated_at": now}
         )
         self.run_store.save_run(updated)
         self._record(
             kind="completed",
-            sched_id=self._run_sched(record) or "",
+            sched_id=sched_id,
             intended=_admission_intended(self.run_store, run_id),
             run_id=run_id,
+            now=now,
+            started_at=now,
         )
 
     # -- polling --------------------------------------------------------
@@ -311,15 +358,16 @@ class Scheduler:
         ids = sorted(item.id for item in schedules)
         if not ids:
             return {}
-        start = self._poll_cursor % len(ids)
+        cursor = self.schedule_store.get_poll_cursor()
+        start = cursor % len(ids)
         order = ids[start:] + ids[:start]
-        self._poll_cursor += 1
+        self.schedule_store.save_poll_cursor(cursor + 1)
         results: dict[str, str] = {}
         by_id = {item.id: item for item in schedules}
         for sid in order:
             sched = by_id[sid]
             if getattr(sched, "blocked_reason", None):
-                raise ValueError(getattr(sched, "blocked_reason"))
+                raise BlockedSchedule(getattr(sched, "blocked_reason"))
             results[sid] = self._poll_one(sched, now)
         return results
 
@@ -328,10 +376,10 @@ class Scheduler:
 
         Recovery NEVER executes (T10): it only completes missing views flagged
         pending for this sweep. Pending runs of blocked schedules stay
-        pending. TODO(T10): introduce the pending-dispatch marker (F11) once
-        dispatch marks distinguish crash-after-dispatch from never-dispatched;
-        until then only runs explicitly flagged ``needs_dispatch`` dispatch
-        here so hanging admitted runs are never re-executed.
+        pending. F11 is descoped to T10 here: the pending-dispatch marker does
+        not exist yet, so only runs explicitly flagged ``needs_dispatch``
+        dispatch in this sweep and hanging admitted runs are never
+        re-executed.
         """
         for run in sorted(self.run_store.list_runs(), key=lambda r: r.id):
             if not getattr(run, "needs_dispatch", False):
@@ -371,7 +419,23 @@ class Scheduler:
             consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
             self.schedule_store.save_consumed(sched.id, max(consumed, now))
             return "paused"
+        # Fail closed on corrupt active views before calendar work (F5).
+        for run in self.run_store.list_runs():
+            if self._status_value(run) not in ("admitted", "interrupted"):
+                continue
+            try:
+                self.run_store.get_admission(run.id)
+            except KeyError:
+                reason = f"corrupt run view without admission: {run.id}"
+                sched.blocked_reason = reason
+                self.schedule_store.save_schedule(sched)
+                raise BlockedSchedule(reason) from None
+        try:
             src = self.sources[sched.id]
+        except KeyError as exc:
+            raise InvalidScheduleDefinitionError(
+                f"no occurrence source for schedule {sched.id!r}"
+            ) from exc
         consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
         if consumed > now:
             return "clock-rollback-held"
@@ -411,8 +475,13 @@ class Scheduler:
                 break
         if jumped:
             if sched.misfire == "latest":
+                from wf_scheduling.calendar import ScheduleExhaustedError
+
                 latest = src.prev_before(now)
-                assert latest is not None
+                if latest is None:
+                    raise ScheduleExhaustedError(
+                        f"latest-missed lookup exhausted for {sched.id!r}"
+                    )
                 old = self.schedule_store.get_candidate(sched.id)
                 if old is not None and old.intended_at != latest:
                     self._record(
@@ -421,6 +490,7 @@ class Scheduler:
                         intended=old.intended_at,
                         reason=f"coalesced-into:{latest.isoformat()}",
                         revision=sched.revision,
+                        now=now,
                     )
                 self.schedule_store.save_candidate(
                     PendingCandidate(
@@ -435,14 +505,25 @@ class Scheduler:
                     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)
-                return self._admit_held_candidate(sched, now) or "candidate-held"
+                held = self._admit_held_candidate(sched, now)
+                if held is None:
+                    # Terminal skip inside admission (e.g. overlap) clears the
+                    # candidate; do not misreport it as held.
+                    return "skipped-overlap"
+                return held
             self._record(
                 kind="interval-summary",
                 sched_id=sched.id,
                 reason="skipped-missed-span",
                 revision=sched.revision,
+                interval=(consumed, now),
+                count=-1,
+                now=now,
             )
             self.schedule_store.save_consumed(sched.id, now)
             return "span-skipped"
@@ -484,7 +565,13 @@ class Scheduler:
                     schedule_id=sched.id,
                 )
                 self.schedule_store.save_consumed(sched.id, instant)
-                last_result = self._admit_held_candidate(sched, now) or "candidate-held"
+                held = self._admit_held_candidate(sched, now)
+                if held is None:
+                    last_result = "skipped-overlap"
+                elif held == "held-undecided":
+                    last_result = "candidate-held"
+                else:
+                    last_result = held
             else:
                 if _is_oneshot(src) and not sched.exhausted:
                     self._record(
@@ -511,7 +598,13 @@ class Scheduler:
         if last_result in ("idle",):
             cand = self.schedule_store.get_candidate(sched.id)
             if cand is not None:
-                last_result = self._admit_held_candidate(sched, now) or "candidate-held"
+                held = self._admit_held_candidate(sched, now)
+                if held is None:
+                    last_result = "skipped-overlap"
+                elif held == "held-undecided":
+                    last_result = "candidate-held"
+                else:
+                    last_result = held
         return last_result
 
     def _admit_held_candidate(self, sched: Any, now: datetime) -> str | None:
@@ -524,11 +617,11 @@ class Scheduler:
                     intended=cand.intended_at,
                     reason="schedule-edit",
                     revision=sched.revision,
+                    now=now,
                 )
                 self.schedule_store.save_candidate(None, schedule_id=sched.id)
             return None
-        result = self._admit(sched, cand.intended_at, now)
-        return result if result != "held-undecided" else None
+        return self._admit(sched, cand.intended_at, now)
 
     # -- administration ---------------------------------------------------
     def resume_schedule(self, sid: str, now: datetime) -> None:

src/wf_scheduling/store.py

@@ -5,10 +5,11 @@ Layout per schedule ``<root>/schedules/<id>/``:
 - ``schedule.json`` — the schedule definition (deleted schedules stay with
   ``deleted=True``; ids are never reused to annex old history).
 - ``candidate.json`` — at most one pending latest candidate.
-- ``consumed.json`` — ``{"consumed_through": <utc iso>}`` watermark plus the
-  schedule revision it belongs to.
+- ``consumed.json`` — ``{"consumed_through": <utc iso>}`` watermark.
 - ``history.json`` — occurrence-history list for cursor pagination over
-  ``(resolved_at, occurrence_id)``.
+  ``(resolved_at, occurrence_id)`` (keyset; integer offsets accepted as a
+  legacy fallback).
+- ``_poll_cursor.json`` — round-robin fairness cursor (persisted).
 
 Per-file writes are atomic (tmp + rename); the lock is process-local only.
 Scheduler ownership across processes arrives in T11.
@@ -131,12 +132,33 @@ class FileScheduleStore:
         return datetime.fromisoformat(value)
 
     def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None:
+        if consumed_through.tzinfo is None or consumed_through.utcoffset() is None:
+            raise ValueError("consumed_through requires an aware datetime")
         with self._lock:
             self._write_json(
                 self._consumed_path(schedule_id),
                 {"consumed_through": consumed_through.astimezone(UTC).isoformat()},
             )
 
+    def get_poll_cursor(self) -> int:
+        """Return the persisted round-robin fairness cursor."""
+        path = self.schedules_dir / "_poll_cursor.json"
+        if not path.exists():
+            return 0
+        try:
+            raw = json.loads(path.read_text(encoding="utf-8"))
+            value = raw.get("cursor", 0) if isinstance(raw, dict) else 0
+            return value if isinstance(value, int) and value >= 0 else 0
+        except ValueError, AttributeError:
+            return 0
+
+    def save_poll_cursor(self, cursor: int) -> None:
+        """Persist the round-robin fairness cursor."""
+        with self._lock:
+            self._write_json(
+                self.schedules_dir / "_poll_cursor.json", {"cursor": cursor}
+            )
+
     # -- history with cursor pagination -----------------------------------
     def append_history(self, record: OccurrenceRecord) -> None:
         """Append one occurrence-history entry for a schedule."""
@@ -146,6 +168,16 @@ class FileScheduleStore:
             entries.append(record.model_dump(mode="json"))
             self._write_json(path, entries)
 
+    @staticmethod
+    def _sort_key(
+        item: OccurrenceRecord,
+    ) -> tuple[bool, datetime, str]:
+        return (
+            item.resolved_at is None,
+            item.resolved_at or datetime.max.replace(tzinfo=UTC),
+            item.occurrence_id,
+        )
+
     def list_occurrences(
         self,
         schedule_id: str,
@@ -153,38 +185,62 @@ class FileScheduleStore:
         cursor: str | None = None,
         limit: int = 50,
     ) -> dict[str, object]:
-        """Return one history page ordered by ``(resolved_at, occurrence_id)``."""
+        """Return one history page ordered by ``(resolved_at, occurrence_id)``.
+
+        Keyset cursors are ``"<resolved_at_iso>|<occurrence_id>"`` (empty iso
+        for rows without a resolved instant); plain integer offsets remain
+        accepted as a legacy fallback.
+        """
         if limit < 1 or limit > 100:
             raise ValueError("limit must be between 1 and 100")
         records = [
             OccurrenceRecord.model_validate(item)
             for item in self._read_history_locked(schedule_id)
         ]
-        records.sort(
-            key=lambda item: (
-                item.resolved_at is None,
-                item.resolved_at or datetime.max.replace(tzinfo=UTC),
-                item.occurrence_id,
-            )
-        )
-        start = 0
+        records.sort(key=self._sort_key)
+        total = len(records)
+        start_index = 0
         if cursor is not None:
+            if "|" in cursor:
+                iso_part, _, oid_part = cursor.partition("|")
+                key = (
+                    True,
+                    datetime.max.replace(tzinfo=UTC),
+                    oid_part,
+                )
+                if iso_part:
                     try:
-                start = int(cursor)
+                        key = (False, datetime.fromisoformat(iso_part), oid_part)
+                    except ValueError as exc:
+                        raise ValueError("invalid keyset cursor") from exc
+                start_index = 0
+                for index, item in enumerate(records):
+                    if self._sort_key(item) > key:
+                        start_index = index
+                        break
+                else:
+                    start_index = total
+            else:
+                try:
+                    start_index = int(cursor)
                 except ValueError as exc:
                     raise ValueError(
-                    "cursor must be a non-negative integer offset"
+                        "cursor must be a keyset '<iso>|<id>' or integer offset"
                     ) from exc
-            if start < 0:
+                if start_index < 0:
                     raise ValueError("cursor must be a non-negative integer offset")
-        total = len(records)
-        page = records[start : start + limit]
-        end = start + limit
+        page = records[start_index : start_index + limit]
+        if start_index + limit < total and page:
+            last = page[-1]
+            iso = "" if last.resolved_at is None else last.resolved_at.isoformat()
+            next_cursor: str | None = f"{iso}|{last.occurrence_id}"
+        else:
+            next_cursor = None
         return {
             "occurrences": [item.model_dump(mode="json") for item in page],
             "total": total,
             "cursor": cursor,
-            "next_cursor": str(end) if end < total else None,
+            "next_cursor": next_cursor,
             "limit": limit,
         }
 

tests/scheduling/test_schedule_store.py

@@ -164,11 +164,17 @@ def test_history_pagination_over_resolved_utc(tmp_path: Path) -> None:
         )
     page = store.list_occurrences("a", limit=2)
     assert page["total"] == 3
-    assert page["next_cursor"] == "2"
+    assert page["next_cursor"] is not None
+    assert "|" in str(page["next_cursor"])
     assert len(page["occurrences"]) == 2  # type: ignore[arg-type]
-    second = store.list_occurrences("a", cursor="2", limit=2)
+    second = store.list_occurrences(
+        "a", cursor=page["next_cursor"], limit=2  # type: ignore[arg-type]
+    )
     assert second["next_cursor"] is None
     assert len(second["occurrences"]) == 1  # type: ignore[arg-type]
+    # Legacy integer offsets remain accepted.
+    legacy = store.list_occurrences("a", cursor="2", limit=2)
+    assert len(legacy["occurrences"]) == 1  # type: ignore[arg-type]
 
 
 def test_inspection_payload_carries_contract_fields(tmp_path: Path) -> None: