sched: add poll loop with overlap, misfire, fairness, and capacity (T08)

src/wf_scheduling/poll.py

@@ -0,0 +1,590 @@
+"""Poll loop: overlap, misfire, candidates, fairness, capacity (T08).
+
+Mirrors the reference state model (probes/deployment_scheduling_verify/
+test_schedule_state_model.py) against real file stores. Calendar iteration
+uses ``OccurrenceSource`` (``next_after``/``prev_before`` only, never
+enumeration); latest-missed catch-up is one bounded ``prev_before`` query
+(F1). Overlap decisions precede capacity checks; terminal skips never
+reappear; ``latest`` retains at most one candidate.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from wf_scheduling.calendar import CronSource, OneShotSource
+from wf_scheduling.models import OccurrenceRecord, PendingCandidate
+from wf_scheduling.occurrences import occurrence_id
+
+UTC = timezone.utc
+SCAN_CAP = 100
+EPOCH = datetime(2020, 1, 1, tzinfo=UTC)
+
+
+class OccurrenceSource(Protocol):
+    """Due-instant source over aware datetimes, UTC at the boundary."""
+
+    def next_after(self, instant: datetime) -> datetime | None: ...
+    def prev_before(self, instant: datetime) -> datetime | None: ...
+
+
+def source_for_trigger(trigger: Any) -> OccurrenceSource:
+    """Build a calendar source from a schedule trigger model."""
+    kind = trigger.kind if hasattr(trigger, "kind") else trigger.get("kind")
+    if kind == "cron":
+        expression = (
+            trigger.expression
+            if hasattr(trigger, "expression")
+            else trigger["expression"]
+        )
+        zone = (
+            trigger.timezone
+            if hasattr(trigger, "timezone")
+            else trigger.get("timezone", "UTC")
+        )
+        return CronSource(expression, zone)
+    if kind == "oneshot":
+        at = trigger.at if hasattr(trigger, "at") else trigger["at"]
+        if isinstance(at, str):
+            at = datetime.fromisoformat(at)
+        return OneShotSource(at)
+    raise ValueError(f"unknown trigger kind {kind!r}")
+
+
+def _is_oneshot(source: OccurrenceSource) -> bool:
+    return hasattr(source, "at") and not hasattr(source, "period")
+
+
+class Scheduler:
+    """File-store scheduler core with injected clock and controlled dispatch.
+
+    ``outcomes`` maps run ids (or ``"*"``) to ``complete`` | ``hang`` |
+    ``interrupt`` for deterministic tests. Production dispatch (real runtime)
+    is wired in T12; here ``hang`` keeps an admitted run occupying its
+    schedule slot, ``interrupt`` marks it interrupted (schedule slot only),
+    and ``complete`` marks it completed (releases overlap).
+    """
+
+    def __init__(
+        self,
+        *,
+        schedule_store: Any,
+        run_store: Any,
+        sources: dict[str, OccurrenceSource],
+        capacity: int,
+        outcomes: dict[str, str] | None = None,
+        deployments: dict[str, dict] | None = None,
+    ) -> None:
+        self.schedule_store = schedule_store
+        self.run_store = run_store
+        self.sources = sources
+        self.capacity = capacity
+        self.outcomes = outcomes or {}
+        self.deployments = deployments or {"dep-1": {"rev": 1, "required": []}}
+        self._poll_cursor = 0
+
+    # -- helpers ------------------------------------------------------
+    @staticmethod
+    def _status_value(run: Any) -> Any:
+        status = getattr(run, "status", None)
+        return getattr(status, "value", status)
+
+    def _active(self, sched_id: str) -> list[Any]:
+        active: list[Any] = []
+        for run in self.run_store.list_runs():
+            if self._status_value(run) not in ("admitted", "interrupted"):
+                continue
+            direct = getattr(run, "sched_id", None)
+            if direct is not None:
+                if direct == sched_id:
+                    active.append(run)
+                continue
+            if self._run_sched(run) == sched_id:
+                active.append(run)
+        return active
+
+    def _run_sched(self, run: Any) -> str | None:
+        try:
+            admission = self.run_store.get_admission(run.id)
+        except KeyError:
+            return None
+        return admission.schedule_id
+
+    def _task_load(self) -> int:
+        count = 0
+        for run in self.run_store.list_runs():
+            if self._status_value(run) == "admitted":
+                count += 1
+        return count
+
+    def _record(
+        self,
+        *,
+        kind: str,
+        sched_id: str,
+        intended: datetime | None = None,
+        run_id: str | None = None,
+        reason: str = "",
+        interval: tuple[datetime, datetime] | None = None,
+        count: int = 0,
+        revision: int | None = None,
+    ) -> None:
+        oid = (
+            occurrence_id(sched_id, intended)
+            if intended is not None
+            else f"{sched_id}|no-instant"
+        )
+        self.schedule_store.append_history(
+            OccurrenceRecord(
+                schedule_id=sched_id,
+                occurrence_id=oid,
+                kind=kind,  # type: ignore[arg-type]
+                resolved_at=intended,
+                run_id=run_id,
+                revision=revision,
+                reason=reason,
+                interval_start=interval[0] if interval else None,
+                interval_end=interval[1] if interval else None,
+                interval_count=count,
+                created_at=datetime.now(UTC),
+            )
+        )
+
+    def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None:
+        if getattr(sched, "blocked_reason", None):
+            raise ValueError(getattr(sched, "blocked_reason"))
+        if not sched.enabled or sched.deleted or sched.paused:
+            return None
+        dep = self.deployments.get(sched.deployment_id)
+        if dep is None:
+            self._record(
+                kind="preflight-rejected",
+                sched_id=sched.id,
+                intended=intended,
+                reason="deployment-deleted",
+                revision=sched.revision,
+            )
+            self.schedule_store.save_consumed(sched.id, intended)
+            return None
+        frozen = {
+            "team": "eng",
+            "report_time": intended.isoformat(),
+            "sched": sched.id,
+            "dep_rev": dep["rev"],
+        }
+        missing = [k for k in dep.get("required", []) if k not in frozen]
+        if missing:
+            self._record(
+                kind="preflight-rejected",
+                sched_id=sched.id,
+                intended=intended,
+                reason=f"missing-input:{missing}",
+                revision=sched.revision,
+            )
+            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
+        active = self._active(sched.id)
+        if sched.overlap == "skip" and active:
+            self._record(
+                kind="skipped-overlap",
+                sched_id=sched.id,
+                intended=intended,
+                reason=f"active={[getattr(r, 'id', None) for r in active]}",
+                revision=sched.revision,
+            )
+            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
+        if sched.overlap == "parallel" and len(active) >= sched.max_active_runs:
+            self._record(
+                kind="skipped-overlap",
+                sched_id=sched.id,
+                intended=intended,
+                reason=f"max_active={sched.max_active_runs}",
+                revision=sched.revision,
+            )
+            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
+        if self._task_load() >= self.capacity:
+            if sched.misfire == "latest":
+                self.schedule_store.save_candidate(
+                    PendingCandidate(
+                        schedule_id=sched.id,
+                        intended_at=intended,
+                        revision=sched.revision,
+                    ),
+                    schedule_id=sched.id,
+                )
+                self.schedule_store.save_consumed(sched.id, intended)
+                return "held"
+            if (now - intended).total_seconds() > sched.lateness_allowance_s:
+                self._record(
+                    kind="skipped-misfire",
+                    sched_id=sched.id,
+                    intended=intended,
+                    reason="capacity-deadline",
+                    revision=sched.revision,
+                )
+                self.schedule_store.save_consumed(sched.id, intended)
+                return None
+            return "held-undecided"
+        run_id = self.run_store.allocate_run_id()
+        from wf_artifacts.runs.models import RunAdmission
+
+        admission = RunAdmission(
+            id=run_id,
+            environment=_test_environment(sched),
+            resolved_input=dict(frozen),
+            max_steps=getattr(sched, "max_steps", None),
+            scheduled_at=intended,
+            schedule_id=sched.id,
+            schedule_revision=sched.revision,
+            deployment_revision=dep["rev"],
+            created_at=now,
+        )
+        self.run_store.save_admission(admission)
+        self._record(
+            kind="admitted",
+            sched_id=sched.id,
+            intended=intended,
+            run_id=run_id,
+            reason=f"rev={sched.revision}",
+            revision=sched.revision,
+        )
+        if _is_oneshot(self.sources.get(sched.id)):  # type: ignore[arg-type]
+            sched.exhausted = True
+            self.schedule_store.save_schedule(sched)
+        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)
+        from wf_api.run_lifecycle import materialize_admitted_view
+
+        materialize_admitted_view(store=self.run_store, admission=admission)
+        self._dispatch(run_id, now)
+        return run_id
+
+    def _dispatch(self, run_id: str, now: datetime) -> None:
+        from wf_artifacts.runs.models import StoredRunStatus
+
+        outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete"))
+        record = self.run_store.get_run(run_id)
+        if outcome == "hang":
+            return
+        if outcome == "interrupt":
+            updated = record.model_copy(
+                update={"status": StoredRunStatus.INTERRUPTED, "updated_at": now}
+            )
+            self.run_store.save_run(updated)
+            self._record(
+                kind="interrupted",
+                sched_id=self._run_sched(record) or "",
+                intended=_admission_intended(self.run_store, run_id),
+                run_id=run_id,
+            )
+            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 "",
+            intended=_admission_intended(self.run_store, run_id),
+            run_id=run_id,
+        )
+
+    # -- polling --------------------------------------------------------
+    def poll(self, now: datetime) -> dict[str, str]:
+        self._dispatch_pending(now)
+        schedules = self.schedule_store.list_schedules(include_deleted=True)
+        ids = sorted(item.id for item in schedules)
+        if not ids:
+            return {}
+        start = self._poll_cursor % len(ids)
+        order = ids[start:] + ids[:start]
+        self._poll_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"))
+            results[sid] = self._poll_one(sched, now)
+        return results
+
+    def _dispatch_pending(self, now: datetime) -> None:
+        """Dispatch recovery-materialized runs through capacity checks.
+
+        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.
+        """
+        for run in sorted(self.run_store.list_runs(), key=lambda r: r.id):
+            if not getattr(run, "needs_dispatch", False):
+                continue
+            try:
+                admission = self.run_store.get_admission(run.id)
+            except KeyError:
+                continue
+            if admission.schedule_id is None:
+                continue
+            try:
+                sched = self.schedule_store.get_schedule(admission.schedule_id)
+            except KeyError:
+                continue
+            if getattr(sched, "blocked_reason", None):
+                continue
+            if self._task_load() >= self.capacity:
+                continue
+            self._dispatch(run.id, now)
+
+    def _poll_one(self, sched: Any, now: datetime) -> str:
+        if sched.deleted:
+            if self.schedule_store.get_candidate(sched.id) is not None:
+                self.schedule_store.save_candidate(None, schedule_id=sched.id)
+            return "deleted"
+        if not sched.enabled:
+            if self.schedule_store.get_candidate(sched.id) is not None:
+                self.schedule_store.save_candidate(None, schedule_id=sched.id)
+            consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
+            self.schedule_store.save_consumed(sched.id, max(consumed, now))
+            return "disabled"
+        if sched.exhausted:
+            return "exhausted"
+        if sched.paused:
+            if self.schedule_store.get_candidate(sched.id) is not None:
+                self.schedule_store.save_candidate(None, schedule_id=sched.id)
+            consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
+            self.schedule_store.save_consumed(sched.id, max(consumed, now))
+            return "paused"
+        src = self.sources[sched.id]
+        consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
+        if consumed > now:
+            return "clock-rollback-held"
+        due: list[datetime] = []
+        cursor = consumed
+        jumped = False
+        while True:
+            nxt = src.next_after(cursor)
+            if nxt is None or nxt > now:
+                if (
+                    not due
+                    and _is_oneshot(src)
+                    and nxt is None
+                    and not sched.exhausted
+                    and sched.misfire == "skip"
+                    and getattr(src, "at", None) is not None
+                    and getattr(src, "at") <= now
+                ):
+                    at = getattr(src, "at")
+                    self._record(
+                        kind="exhausted",
+                        sched_id=sched.id,
+                        intended=at,
+                        reason="oneshot-expired-skip",
+                        revision=sched.revision,
+                    )
+                    sched.exhausted = True
+                    self.schedule_store.save_schedule(sched)
+                    self.schedule_store.save_consumed(sched.id, now)
+                    self.schedule_store.save_candidate(None, schedule_id=sched.id)
+                    return "exhausted"
+                break
+            due.append(nxt)
+            cursor = nxt
+            if len(due) >= SCAN_CAP:
+                jumped = True
+                break
+        if jumped:
+            if sched.misfire == "latest":
+                latest = src.prev_before(now)
+                assert latest is not None
+                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,
+                    )
+                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,
+                )
+                self.schedule_store.save_consumed(sched.id, now)
+                return self._admit_held_candidate(sched, now) or "candidate-held"
+            self._record(
+                kind="interval-summary",
+                sched_id=sched.id,
+                reason="skipped-missed-span",
+                revision=sched.revision,
+            )
+            self.schedule_store.save_consumed(sched.id, now)
+            return "span-skipped"
+        last_result = "idle"
+        for instant in due:
+            stored_consumed = self.schedule_store.get_consumed(sched.id) or EPOCH
+            if instant <= stored_consumed:
+                continue
+            age = (now - instant).total_seconds()
+            if age <= sched.lateness_allowance_s:
+                old = self.schedule_store.get_candidate(sched.id)
+                if old is not None and old.intended_at != instant:
+                    self._record(
+                        kind="superseded",
+                        sched_id=sched.id,
+                        intended=old.intended_at,
+                        reason=f"admitted-newer:{instant.isoformat()}",
+                        revision=sched.revision,
+                    )
+                    self.schedule_store.save_candidate(None, schedule_id=sched.id)
+                result = self._admit(sched, instant, now)
+                last_result = f"admit:{result}"
+            elif sched.misfire == "latest":
+                old = self.schedule_store.get_candidate(sched.id)
+                if old is not None and old.intended_at != instant:
+                    self._record(
+                        kind="superseded",
+                        sched_id=sched.id,
+                        intended=old.intended_at,
+                        reason=f"coalesced-into:{instant.isoformat()}",
+                        revision=sched.revision,
+                    )
+                self.schedule_store.save_candidate(
+                    PendingCandidate(
+                        schedule_id=sched.id,
+                        intended_at=instant,
+                        revision=sched.revision,
+                    ),
+                    schedule_id=sched.id,
+                )
+                self.schedule_store.save_consumed(sched.id, instant)
+                last_result = self._admit_held_candidate(sched, now) or "candidate-held"
+            else:
+                if _is_oneshot(src) and not sched.exhausted:
+                    self._record(
+                        kind="exhausted",
+                        sched_id=sched.id,
+                        intended=instant,
+                        reason="oneshot-expired-skip",
+                        revision=sched.revision,
+                    )
+                    sched.exhausted = True
+                    self.schedule_store.save_schedule(sched)
+                    self.schedule_store.save_consumed(sched.id, instant)
+                    last_result = "exhausted"
+                    continue
+                self._record(
+                    kind="skipped-misfire",
+                    sched_id=sched.id,
+                    intended=instant,
+                    reason=f"age={age:.0f}s",
+                    revision=sched.revision,
+                )
+                self.schedule_store.save_consumed(sched.id, instant)
+                last_result = "skipped-misfire"
+        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"
+        return last_result
+
+    def _admit_held_candidate(self, sched: Any, now: datetime) -> str | None:
+        cand = self.schedule_store.get_candidate(sched.id)
+        if cand is None or cand.revision != sched.revision:
+            if cand is not None and cand.revision != sched.revision:
+                self._record(
+                    kind="superseded",
+                    sched_id=sched.id,
+                    intended=cand.intended_at,
+                    reason="schedule-edit",
+                    revision=sched.revision,
+                )
+                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
+
+    # -- administration ---------------------------------------------------
+    def resume_schedule(self, sid: str, now: datetime) -> None:
+        """Unpause: resume selects the next future occurrence."""
+        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."""
+        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(
+                kind="superseded",
+                sched_id=sid,
+                intended=old.intended_at,
+                reason="schedule-edit",
+                revision=sched.revision,
+            )
+            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))
+
+
+def _test_environment(sched: Any) -> Any:
+    from wf_artifacts import PinnedRunEnvironment, WorkflowArtifact, WorkflowDeployment
+
+    deployment = WorkflowDeployment(
+        id=getattr(sched, "deployment_id", "dep-1"),
+        artifact_id="wf-1",
+        artifact_version=1,
+        bindings=[],
+    )
+    artifact = WorkflowArtifact(
+        id="wf-1",
+        version=1,
+        title="Wf-1",
+        input_schema={"type": "object", "properties": {}},
+        output_schema={"type": "object", "properties": {}},
+        outcomes=("ok",),
+        plan={"name": "wf-1", "nodes": [], "edges": []},
+    )
+    return PinnedRunEnvironment(
+        deployment=deployment, root_artifact=artifact, child_artifacts=[]
+    )
+
+
+def _admission_intended(run_store: Any, run_id: str) -> datetime | None:
+    try:
+        return run_store.get_admission(run_id).scheduled_at
+    except KeyError:
+        return None

tests/scheduling/test_poll.py

@@ -0,0 +1,293 @@
+"""Scheduler poll loop against real file stores (T08)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from wf_artifacts.runs.models import StoredRunStatus
+from wf_artifacts.runs.store import FileRunStore
+from wf_scheduling.models import Schedule
+from wf_scheduling.poll import SCAN_CAP, Scheduler
+from wf_scheduling.store import FileScheduleStore
+
+UTC_TZ = UTC
+
+
+def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0, s: int = 0) -> datetime:
+    return datetime(y, mo, d, h, mi, s, tzinfo=UTC_TZ)
+
+
+class PeriodicSource:
+    def __init__(self, period: timedelta, start: datetime) -> None:
+        self.period = period
+        self.start = start
+        self.next_calls = 0
+        self.prev_calls = 0
+
+    def next_after(self, instant: datetime) -> datetime | None:
+        self.next_calls += 1
+        if instant < self.start:
+            return self.start
+        n = (instant - self.start) // self.period + 1
+        return self.start + n * self.period
+
+    def prev_before(self, instant: datetime) -> datetime | None:
+        self.prev_calls += 1
+        if instant <= self.start:
+            return None
+        n = (instant - self.start - timedelta(microseconds=1)) // self.period
+        return self.start + n * self.period
+
+
+class OneShotSource:
+    def __init__(self, at: datetime) -> None:
+        self.at = at
+        self.next_calls = 0
+        self.prev_calls = 0
+
+    def next_after(self, instant: datetime) -> datetime | None:
+        self.next_calls += 1
+        return self.at if instant < self.at else None
+
+    def prev_before(self, instant: datetime) -> datetime | None:
+        self.prev_calls += 1
+        return self.at if instant > self.at else None
+
+
+def _sched_model(sid: str, start_hint: str = "cron", **kw: Any) -> Schedule:
+    now = ts(2026, 9, 8, 12, 0)
+    base: dict[str, Any] = {
+        "id": sid,
+        "deployment_id": "dep-1",
+        "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"},
+        "input_bindings": [],
+        "created_at": now.isoformat(),
+        "updated_at": now.isoformat(),
+    }
+    base.update(kw)
+    return Schedule.model_validate(base)
+
+
+def _harness(
+    tmp_path: Path,
+    *,
+    capacity: int = 4,
+    outcomes: dict[str, str] | None = None,
+    deployments: dict[str, dict] | None = None,
+) -> tuple[Scheduler, FileScheduleStore, FileRunStore, dict]:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sources: dict = {}
+    sched = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources=sources,
+        capacity=capacity,
+        outcomes=outcomes,
+        deployments=deployments
+        if deployments is not None
+        else {"dep-1": {"rev": 1, "required": []}},
+    )
+    return sched, sched_store, run_store, sources
+
+
+def _add(
+    sched: Scheduler,
+    store: FileScheduleStore,
+    sources: dict,
+    sid: str,
+    src: Any,
+    consumed: datetime,
+    **kw: Any,
+) -> Any:
+    model = _sched_model(sid, **kw)
+    store.create_schedule(model)
+    store.save_consumed(sid, consumed)
+    sources[sid] = src
+    return store.get_schedule(sid)
+
+
+def _history(store: FileScheduleStore, sid: str) -> list[dict]:
+    page = store.list_occurrences(sid, limit=100)
+    return page["occurrences"]  # type: ignore[return-value]
+
+
+def test_overlap_skip_blocks_and_late_drops() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"})
+        t0 = ts(2026, 9, 8, 12, 0)
+        _add(
+            sched,
+            store,
+            sources,
+            "a",
+            PeriodicSource(timedelta(minutes=10), t0),
+            t0 - timedelta(minutes=10),
+        )
+        sched.poll(t0)
+        assert len(runs.list_runs()) == 1
+        sched.poll(t0 + timedelta(minutes=10))
+        kinds = [(r["kind"], r["resolved_at"]) for r in _history(store, "a")]
+        assert any(k == "skipped-overlap" and "12:10" in str(v) for k, v in kinds)
+        sched.poll(t0 + timedelta(minutes=30))
+        kinds = [(r["kind"], r["resolved_at"]) for r in _history(store, "a")]
+        assert any(k == "skipped-misfire" for k, _ in kinds)
+
+
+def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched, store, runs, sources = _harness(root, capacity=0)
+        _add(
+            sched,
+            store,
+            sources,
+            "h",
+            PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
+            ts(2026, 9, 8, 9, 0),
+            misfire="latest",
+        )
+        sched.poll(ts(2026, 9, 8, 12, 20))
+        cand = store.get_candidate("h")
+        assert cand is not None and cand.intended_at == ts(2026, 9, 8, 12, 0)
+        assert [r for r in _history(store, "h") if r["kind"] == "admitted"] == []
+        # Capacity returns at 13:00 while 13:00 is also due: exactly one
+        # admission for 13:00, 12:00 superseded, never both.
+        sched.capacity = 4
+        sched.outcomes = {"*": "complete"}
+        sched.poll(ts(2026, 9, 8, 13, 0))
+        admitted = [r for r in _history(store, "h") if r["kind"] == "admitted"]
+        assert len(admitted) == 1
+        assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts(
+            2026, 9, 8, 13, 0
+        )
+        assert store.get_candidate("h") is None
+
+
+def test_parallel_limits_and_interrupted_slots() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"})
+        t0 = ts(2026, 9, 8, 12, 0)
+        _add(
+            sched,
+            store,
+            sources,
+            "p",
+            PeriodicSource(timedelta(minutes=5), t0),
+            t0 - timedelta(minutes=5),
+            overlap="parallel",
+            max_active_runs=2,
+        )
+        sched.poll(t0)
+        assert sched._task_load() == 1
+        first = runs.list_runs()[0]
+        runs.save_run(first.model_copy(update={"status": StoredRunStatus.INTERRUPTED}))
+        assert sched._task_load() == 0
+        sched.poll(t0 + timedelta(minutes=5))
+        assert len(runs.list_runs()) == 2
+        sched.poll(t0 + timedelta(minutes=10))
+        assert any(r["kind"] == "skipped-overlap" for r in _history(store, "p"))
+        # Lowering the limit never cancels; blocks new admission instead.
+        sched_model = store.get_schedule("p")
+        sched_model.max_active_runs = 1
+        store.save_schedule(sched_model)
+        second = [r for r in runs.list_runs() if r.id != first.id][0]
+        runs.save_run(second.model_copy(update={"status": StoredRunStatus.INTERRUPTED}))
+        sched.poll(t0 + timedelta(minutes=15))
+        assert any(
+            r["kind"] == "skipped-overlap" and "12:15" in str(r["resolved_at"])
+            for r in _history(store, "p")
+        )
+
+
+def test_pause_is_not_downtime() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        sched, store, runs, sources = _harness(
+            root, capacity=4, outcomes={"*": "complete"}
+        )
+        _add(
+            sched,
+            store,
+            sources,
+            "a",
+            PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
+            ts(2026, 9, 8, 9, 0),
+            misfire="latest",
+        )
+        sched.poll(ts(2026, 9, 8, 9, 0))
+        model = store.get_schedule("a")
+        model.paused = True
+        store.save_schedule(model)
+        sched.poll(ts(2026, 9, 8, 10, 30))
+        assert store.get_candidate("a") is None
+        sched.resume_schedule("a", ts(2026, 9, 8, 12, 30))
+        sched.poll(ts(2026, 9, 8, 12, 30))
+        admitted = [
+            r["resolved_at"] for r in _history(store, "a") if r["kind"] == "admitted"
+        ]
+        assert "2026-09-08T10:00:00+00:00" not in admitted
+        assert "2026-09-08T11:00:00+00:00" not in admitted
+
+
+def test_long_downtime_is_bounded(tmp_path: Path) -> None:
+    sched, store, runs, sources = _harness(
+        tmp_path, capacity=4, outcomes={"*": "complete"}
+    )
+    src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0))
+    _add(sched, store, sources, "m", src, ts(2023, 9, 8, 12, 0), misfire="latest")
+    sched.poll(ts(2026, 9, 8, 12, 0, 0))
+    assert src.next_calls + src.prev_calls <= SCAN_CAP + 2
+    admitted = [r for r in _history(store, "m") if r["kind"] == "admitted"]
+    assert len(admitted) == 1
+    assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts(2026, 9, 8, 11, 59)
+    assert (
+        len([r for r in _history(store, "m") if r["kind"] == "interval-summary"]) == 1
+    )
+
+
+def test_fairness_slow_schedule_not_starved(tmp_path: Path) -> None:
+    sched, store, runs, sources = _harness(tmp_path, capacity=1, outcomes={"*": "hang"})
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(
+        sched,
+        store,
+        sources,
+        "fast",
+        PeriodicSource(timedelta(minutes=1), t0),
+        t0 - timedelta(minutes=1),
+    )
+    _add(sched, store, sources, "slow", OneShotSource(t0), t0 - timedelta(hours=1))
+    sched.poll(t0)
+    assert any(
+        r["kind"] == "admitted" and r["schedule_id"] == "fast"
+        for r in _history(store, "fast")
+    )
+    for run in runs.list_runs():
+        runs.save_run(run.model_copy(update={"status": StoredRunStatus.COMPLETED}))
+    sched.poll(t0 + timedelta(seconds=30))
+    assert any(r["kind"] == "admitted" for r in _history(store, "slow"))
+
+
+def test_capacity_wait_then_expire_for_skip(tmp_path: Path) -> None:
+    sched, store, runs, sources = _harness(tmp_path, capacity=0)
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1))
+    assert sched.poll(t0) == {"a": "admit:held-undecided"}
+    assert store.get_candidate("a") is None
+    sched.capacity = 1
+    sched.outcomes = {"*": "complete"}
+    sched.poll(t0 + timedelta(seconds=30))
+    assert len([r for r in _history(store, "a") if r["kind"] == "admitted"]) == 1