# DISPOSABLE SCHEDULING STATE-MODEL PROBE — NOT PRODUCTION CODE.
# See README.md in this directory. Pure stdlib; runs in the repo env:
#   uv run pytest -q probes/deployment_scheduling_verify/test_schedule_state_model.py
"""Executable reference model of the deployment-scheduling state machine.

Covers the spec's state gates with injected clocks and controlled
execution (no sleeps, no threads):
overlap=skip|parallel x misfire=skip|latest, latest-means-ONE-candidate,
supersession, terminal skips, slot accounting, pause-vs-downtime,
edit/delete/restart, capacity/fairness, fault boundaries, crash-during-
resume, exclusive ownership.

The calendar is an abstract due-instant source here (next_after /
prev_before only — deliberately NO iter_between, to forbid unbounded
enumeration). Calendar math itself is probed in test_calendar_probe.py.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

import pytest

UTC = timezone.utc


def ts(y, mo, d, h=0, mi=0, s=0) -> datetime:
    return datetime(y, mo, d, h, mi, s, tzinfo=UTC)


# --------------------------------------------------------------------------
# Errors
# --------------------------------------------------------------------------


class InjectedFault(Exception):
    pass


class SecondOwnerError(Exception):
    pass


class StartupRejected(Exception):
    pass


class BlockedSchedule(Exception):
    pass


class ExecutorCrashed(Exception):
    pass


# --------------------------------------------------------------------------
# Records
# --------------------------------------------------------------------------

RUNNING = "running"
INTERRUPTED = "interrupted"  # durably waiting, resumable
COMPLETED = "completed"
FAILED = "failed"
ACTIVE_STATES = (RUNNING, INTERRUPTED)


@dataclass
class Schedule:
    id: str
    rev: int = 1
    enabled: bool = True
    paused: bool = False
    deleted: bool = False
    overlap: str = "skip"  # skip | parallel
    misfire: str = "skip"  # skip | latest
    max_active: int = 1
    allowance_s: float = 60.0
    deployment_id: str = "dep-1"
    created_dep_rev: int = 1
    exhausted: bool = False
    blocked_reason: str | None = None


@dataclass
class Candidate:
    sched_id: str
    intended: datetime
    rev: int


@dataclass
class Run:
    id: str
    sched_id: str
    intended: datetime
    rev: int
    frozen_input: dict
    state: str = RUNNING
    dispatched_unknown: bool = False  # dispatched, no stopped result yet
    needs_dispatch: bool = False  # admitted + view exists, never dispatched
    attempt_id: int = 0  # resume attempt the run currently belongs to
    result_attempt: int | None = None  # attempt that produced the stopped result
    fail_reason: str = ""


@dataclass
class Record:
    kind: str  # admitted|completed|failed|skipped-overlap|skipped-misfire|
    # superseded|preflight-rejected|exhausted|interval-summary|interrupted
    sched_id: str
    intended: datetime | None = None
    run_id: str | None = None
    reason: str = ""
    interval: tuple[datetime, datetime] | None = None
    count: int = 0


# --------------------------------------------------------------------------
# Occurrence sources: next_after / prev_before ONLY (no enumeration seam)
# --------------------------------------------------------------------------


class CountingMixin:
    def __init__(self) -> None:
        self.next_calls = 0
        self.prev_calls = 0


class PeriodicSource(CountingMixin):
    def __init__(self, period: timedelta, start: datetime) -> None:
        super().__init__()
        self.period = period
        self.start = start

    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(CountingMixin):
    def __init__(self, at: datetime) -> None:
        super().__init__()
        self.at = at

    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


# --------------------------------------------------------------------------
# Store with fault injection
# --------------------------------------------------------------------------


@dataclass
class MemStore:
    lockable: bool = True
    lock_holder: str | None = None
    id_seq: int = 0  # store-backed run counter: survives scheduler restart
    attempt_seq: int = 0  # store-backed resume-attempt counter: same reason
    schedules: dict[str, Schedule] = field(default_factory=dict)
    candidates: dict[str, Candidate | None] = field(default_factory=dict)
    consumed: dict[str, datetime] = field(default_factory=dict)
    admissions: dict[str, dict] = field(default_factory=dict)  # run_id -> record
    runs: dict[str, Run] = field(default_factory=dict)
    resume_attempts: dict[str, str] = field(default_factory=dict)  # run -> ACTIVE|DONE
    history: list[Record] = field(default_factory=list)
    deployments: dict[str, dict] = field(default_factory=dict)  # id -> {rev, required}
    faults: dict[str, str] = field(default_factory=dict)  # op -> before|after
    op_log: list[str] = field(default_factory=list)
    poll_cursor: int = 0

    def check_fault(self, op: str) -> None:
        mode = self.faults.pop(op, None)
        self.op_log.append(op)
        if mode == "before":
            raise InjectedFault(f"{op}:before")
        if mode == "after":
            self.op_log.append(f"{op}:after-pending")

    def after_ok(self, op: str) -> None:
        # A test driver calls this after the op's effect to confirm the
        # injected 'after' fault (crash between effect and next step).
        if f"{op}:after-pending" in self.op_log:
            self.op_log.remove(f"{op}:after-pending")
            raise InjectedFault(f"{op}:after")


EPOCH = ts(2020, 1, 1)


# --------------------------------------------------------------------------
# Scheduler reference model
# --------------------------------------------------------------------------

SCAN_CAP = 100  # max next_after calls per schedule per poll before jumping


class Scheduler:
    def __init__(
        self,
        store: MemStore,
        owner: str,
        capacity: int,
        sources: dict[str, CountingMixin],
        outcomes: dict[str, str] | None = None,
    ) -> None:
        if not store.lockable:
            raise StartupRejected("unsupported locking rejects scheduler startup")
        if store.lock_holder is not None and store.lock_holder != owner:
            raise SecondOwnerError(f"store owned by {store.lock_holder}")
        store.lock_holder = owner
        self.store = store
        self.owner = owner
        self.capacity = capacity
        self.sources = sources
        self.outcomes = outcomes or {}

    def close(self) -> None:
        if self.store.lock_holder == self.owner:
            self.store.lock_holder = None

    def __enter__(self) -> Scheduler:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

    # -- helpers ---------------------------------------------------------
    def _active(self, sched_id: str) -> list[Run]:
        return [
            r
            for r in self.store.runs.values()
            if r.sched_id == sched_id and r.state in ACTIVE_STATES
        ]

    def _task_load(self) -> int:
        # Admitted-but-never-dispatched runs hold a schedule slot but no
        # executing-task slot: nothing is running on their behalf.
        return sum(
            1
            for r in self.store.runs.values()
            if r.state == RUNNING and not r.needs_dispatch
        )

    def _record(self, **kw: object) -> None:
        self.store.history.append(Record(**kw))  # type: ignore[arg-type]

    def _admit(self, sched: Schedule, intended: datetime, now: datetime) -> str | None:
        """Ordered admission. Returns run_id, 'held', or None (terminal)."""
        st = self.store
        if sched.blocked_reason:
            raise BlockedSchedule(sched.blocked_reason)
        if not sched.enabled or sched.deleted or sched.paused:
            return None
        # 0. recheck + preflight against CURRENT deployment contract
        dep = st.deployments.get(sched.deployment_id)
        if dep is None:
            self._record(
                kind="preflight-rejected",
                sched_id=sched.id,
                intended=intended,
                reason="deployment-deleted",
            )
            st.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["required"] if k not in frozen]
        if missing:
            self._record(
                kind="preflight-rejected",
                sched_id=sched.id,
                intended=intended,
                reason=f"missing-input:{missing}",
            )
            if (
                st.candidates.get(sched.id) is not None
                and st.candidates[sched.id].intended == intended
            ):  # type: ignore[union-attr]
                st.candidates[sched.id] = None
            st.consumed[sched.id] = intended
            return None
        # 1. overlap first (takes precedence over capacity)
        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={[r.id for r in active]}",
            )
            if (
                st.candidates.get(sched.id) is not None
                and st.candidates[sched.id].intended == intended
            ):  # type: ignore[union-attr]
                st.candidates[sched.id] = None
            st.consumed[sched.id] = intended
            return None
        if sched.overlap == "parallel" and len(active) >= sched.max_active:
            self._record(
                kind="skipped-overlap",
                sched_id=sched.id,
                intended=intended,
                reason=f"max_active={sched.max_active}",
            )
            if (
                st.candidates.get(sched.id) is not None
                and st.candidates[sched.id].intended == intended
            ):  # type: ignore[union-attr]
                st.candidates[sched.id] = None
            st.consumed[sched.id] = intended
            return None
        # 2. capacity: skip expires at deadline, latest holds one candidate
        if self._task_load() >= self.capacity:
            if sched.misfire == "latest":
                st.candidates[sched.id] = Candidate(sched.id, intended, sched.rev)
                st.consumed[sched.id] = intended
                return "held"
            if (now - intended).total_seconds() > sched.allowance_s:
                self._record(
                    kind="skipped-misfire",
                    sched_id=sched.id,
                    intended=intended,
                    reason="capacity-deadline",
                )
                st.consumed[sched.id] = intended
                return None
            return "held-undecided"  # retry next poll, consumed NOT advanced
        # 3. allocate + freeze (store-backed counter: no reuse on restart)
        st.id_seq += 1
        run_id = f"run-{sched.id}-{st.id_seq}"
        # 4. persist admission record BEFORE dispatch (fault boundary).
        # The occurrence is decided here: history entry, candidate
        # clearing, consumed progress, and one-shot exhaustion all belong
        # to this persist, so a later crash can never re-decide it.
        st.check_fault("admission")
        st.admissions[run_id] = {
            "sched": sched.id,
            "intended": intended,
            "rev": sched.rev,
            "input": dict(frozen),
        }
        # The occurrence-status entry is part of the admission persist itself.
        self._record(
            kind="admitted",
            sched_id=sched.id,
            intended=intended,
            run_id=run_id,
            reason=f"rev={sched.rev}",
        )
        # One-shot admission exhausts the schedule (admit exactly once).
        if isinstance(self.sources.get(sched.id), OneShotSource):
            sched.exhausted = True
        if (
            st.candidates.get(sched.id) is not None
            and st.candidates[sched.id].intended == intended
        ):  # type: ignore[union-attr]
            st.candidates[sched.id] = None
        st.consumed[sched.id] = intended
        st.after_ok("admission")
        # 5. materialize run view with same identity (fault boundary)
        st.check_fault("materialize")
        st.runs[run_id] = Run(run_id, sched.id, intended, sched.rev, dict(frozen))
        st.after_ok("materialize")
        # 6. dispatch the captured invocation without re-resolving (faults)
        st.check_fault("dispatch")
        try:
            self._dispatch(run_id, now)
        finally:
            st.after_ok("dispatch")
        return run_id

    def _dispatch(self, run_id: str, now: datetime) -> None:
        st = self.store
        run = st.runs[run_id]
        outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete"))
        if outcome == "hang":
            run.dispatched_unknown = True  # task alive in-process; restart abandons it
            return  # stays RUNNING, occupies schedule + task slots
        if outcome == "crash":
            run.dispatched_unknown = True
            raise ExecutorCrashed(run_id)
        if outcome == "interrupt":
            st.check_fault("interrupt-persist")
            run.state = INTERRUPTED
            st.after_ok("interrupt-persist")
            self._record(
                kind="interrupted",
                sched_id=run.sched_id,
                intended=run.intended,
                run_id=run_id,
            )
            return
        assert outcome == "complete"
        st.check_fault("complete")
        run.state = COMPLETED
        st.after_ok("complete")
        self._record(
            kind="completed",
            sched_id=run.sched_id,
            intended=run.intended,
            run_id=run_id,
        )

    # -- polling ----------------------------------------------------------
    def poll(self, now: datetime) -> dict[str, str]:
        st = self.store
        if st.lock_holder != self.owner:
            raise SecondOwnerError("lost ownership")
        self._dispatch_pending(now)
        ids = sorted(s.id for s in st.schedules.values())
        if not ids:
            return {}
        start = st.poll_cursor % len(ids)
        order = ids[start:] + ids[:start]
        st.poll_cursor += 1
        results: dict[str, str] = {}
        for sid in order:
            results[sid] = self._poll_one(st.schedules[sid], now)
        return results

    def _dispatch_pending(self, now: datetime) -> None:
        """Dispatch runs that recovery materialized but never executed.

        Recovery NEVER executes: it only completes missing views and flags
        them pending. Execution happens here, through the same capacity
        checks as normal admission — never inside recovery. Pending runs
        of blocked schedules stay pending (fail closed).
        """
        st = self.store
        for run in sorted(st.runs.values(), key=lambda r: r.id):
            if not run.needs_dispatch:
                continue
            sched = st.schedules.get(run.sched_id)
            if sched is None or sched.blocked_reason:
                continue
            if self._task_load() >= self.capacity:
                continue  # stays pending until a slot frees
            run.needs_dispatch = False
            try:
                self._dispatch(run.id, now)
            except ExecutorCrashed:
                # Dispatched with unknown outcome: next recovery fails it
                # closed as abandoned (no replay), like any dispatch crash.
                run.dispatched_unknown = True

    def _poll_one(self, sched: Schedule, now: datetime) -> str:
        st = self.store
        if sched.blocked_reason:
            raise BlockedSchedule(sched.blocked_reason)
        if sched.deleted:
            # Deletion clears pending candidates; admitted runs/history stay.
            if st.candidates.get(sched.id) is not None:
                st.candidates[sched.id] = None
            return "deleted"
        if not sched.enabled:
            # Resolved edge (no spec disable concept): administrative disable
            # behaves like pause for catch-up — excluded, never backfilled.
            if st.candidates.get(sched.id) is not None:
                st.candidates[sched.id] = None
            st.consumed[sched.id] = max(st.consumed.get(sched.id, EPOCH), now)
            return "disabled"
        if sched.exhausted:
            return "exhausted"
        if sched.paused:
            # Explicit pause: clear unadmitted candidates, exclude interval.
            if st.candidates.get(sched.id) is not None:
                st.candidates[sched.id] = None
            st.consumed[sched.id] = max(st.consumed.get(sched.id, EPOCH), now)
            return "paused"
        src = self.sources[sched.id]
        consumed = st.consumed.get(sched.id, EPOCH)
        if consumed > now:
            return "clock-rollback-held"  # never re-admit consumed instants
        # Bounded scan: at most SCAN_CAP next_after calls, then jump.
        due: list[datetime] = []
        cursor = consumed
        jumped = False
        while True:
            nxt = src.next_after(cursor)  # type: ignore[attr-defined]
            if nxt is None or nxt > now:
                if (
                    not due
                    and isinstance(src, OneShotSource)
                    and nxt is None
                    and not sched.exhausted
                    and sched.misfire == "skip"
                    and self._is_oneshot_expired(src, now)
                ):
                    self._record(
                        kind="exhausted",
                        sched_id=sched.id,
                        intended=src.at,
                        reason="oneshot-expired-skip",
                    )
                    sched.exhausted = True
                    st.consumed[sched.id] = now
                    st.candidates[sched.id] = None
                    return "exhausted"
                break
            due.append(nxt)
            cursor = nxt
            if len(due) >= SCAN_CAP:
                jumped = True
                break
        if jumped:
            # Never enumerate further: one bounded query + interval summary.
            if sched.misfire == "latest":
                latest = src.prev_before(now)  # type: ignore[attr-defined]
                assert latest is not None
                old = st.candidates.get(sched.id)
                if old is not None and old.intended != latest:
                    self._record(
                        kind="superseded",
                        sched_id=sched.id,
                        intended=old.intended,
                        reason=f"coalesced-into:{latest.isoformat()}",
                    )
                st.candidates[sched.id] = Candidate(sched.id, latest, sched.rev)
                self._record(
                    kind="interval-summary",
                    sched_id=sched.id,
                    interval=(consumed, now),
                    count=-1,
                    reason="coalesced-missed-span",
                )
                st.consumed[sched.id] = now
                return self._admit_held_candidate(sched, now) or "candidate-held"
            self._record(
                kind="interval-summary",
                sched_id=sched.id,
                interval=(consumed, now),
                count=-1,
                reason="skipped-missed-span",
            )
            st.consumed[sched.id] = now
            return "span-skipped"
        # Normal path: decide each due instant in order.
        last_result = "idle"
        for instant in due:
            if instant <= st.consumed.get(sched.id, EPOCH):
                continue
            age = (now - instant).total_seconds()
            if age <= sched.allowance_s:
                # A timely admission consumes any older held candidate: the
                # newer due instant supersedes it (recorded, never admitted).
                old = st.candidates.get(sched.id)
                if old is not None and old.intended != instant:
                    self._record(
                        kind="superseded",
                        sched_id=sched.id,
                        intended=old.intended,
                        reason=f"admitted-newer:{instant.isoformat()}",
                    )
                    st.candidates[sched.id] = None
                r = self._admit(sched, instant, now)
                last_result = f"admit:{r}"
            elif sched.misfire == "latest":
                old = st.candidates.get(sched.id)
                if old is not None and old.intended != instant:
                    self._record(
                        kind="superseded",
                        sched_id=sched.id,
                        intended=old.intended,
                        reason=f"coalesced-into:{instant.isoformat()}",
                    )
                # A newer due time supersedes the unadmitted candidate; an
                # admitted run is never touched (candidates only).
                st.candidates[sched.id] = Candidate(sched.id, instant, sched.rev)
                st.consumed[sched.id] = instant
                last_result = self._admit_held_candidate(sched, now) or "candidate-held"
            else:
                if isinstance(src, OneShotSource) and not sched.exhausted:
                    self._record(
                        kind="exhausted",
                        sched_id=sched.id,
                        intended=instant,
                        reason="oneshot-expired-skip",
                    )
                    sched.exhausted = True
                    st.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",
                )
                st.consumed[sched.id] = instant
                last_result = "skipped-misfire"
        # Also try a held candidate whose slot may have freed.
        if last_result in ("idle",) and st.candidates.get(sched.id) is not None:
            last_result = self._admit_held_candidate(sched, now) or "candidate-held"
        return last_result

    def _is_oneshot_expired(self, src: OneShotSource, now: datetime) -> bool:
        return src.at <= now

    def _admit_held_candidate(self, sched: Schedule, now: datetime) -> str | None:
        st = self.store
        cand = st.candidates.get(sched.id)
        if cand is None or cand.rev != sched.rev:
            if cand is not None and cand.rev != sched.rev:
                self._record(
                    kind="superseded",
                    sched_id=sched.id,
                    intended=cand.intended,
                    reason="schedule-edit",
                )
                st.candidates[sched.id] = None
            return None
        r = self._admit(sched, cand.intended, now)
        return r if r != "held-undecided" else None

    # -- administration ----------------------------------------------------
    def resume_schedule(self, sid: str, now: datetime) -> None:
        """Unpause: resume selects the next future occurrence; the paused
        interval is excluded from catch-up under both policies."""
        sched = self.store.schedules[sid]
        sched.paused = False
        self.store.consumed[sid] = max(self.store.consumed.get(sid, EPOCH), now)
        self.store.candidates[sid] = None

    def edit_schedule(self, sid: str, now: datetime) -> None:
        """Definition edit: new revision, discard old candidates, begin at
        the edit time (creation/edits never backfill)."""
        st = self.store
        sched = st.schedules[sid]
        sched.rev += 1
        old = st.candidates.get(sid)
        if old is not None:
            self._record(
                kind="superseded",
                sched_id=sid,
                intended=old.intended,
                reason="schedule-edit",
            )
            st.candidates[sid] = None
        st.consumed[sid] = max(st.consumed.get(sid, EPOCH), now)

    # -- resume ------------------------------------------------------------
    def resume_run(self, run_id: str, now: datetime) -> str:
        st = self.store
        run = st.runs[run_id]
        assert run.state == INTERRUPTED, "only waiting interruptions resume"
        if st.resume_attempts.get(run_id) == "ACTIVE":
            raise BlockedSchedule("ambiguous attempt already active")
        if self._task_load() >= self.capacity:
            return "blocked-capacity"  # stays waiting; slot needed
        # Durably mark the active attempt BEFORE executing again. The mark
        # carries a fresh attempt identity that later stopped results echo
        # back, so recovery can tell a new result from the old checkpoint.
        st.check_fault("resume-mark")
        st.attempt_seq += 1
        run.attempt_id = st.attempt_seq
        st.resume_attempts[run_id] = "ACTIVE"
        st.after_ok("resume-mark")
        run.state = RUNNING
        run.dispatched_unknown = True
        st.check_fault("resume-execute")
        try:
            outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete"))
            if outcome == "crash":
                raise ExecutorCrashed(run_id)
            if outcome == "interrupt":
                return self._finish_resume_interrupted(run)
            return self._finish_resume_completed(run)
        finally:
            st.after_ok("resume-execute")

    def _finish_resume_completed(self, run: Run) -> str:
        # Completion, attempt-clearing, and history recording are separate
        # persists with a fault boundary between each pair; a resumed run
        # may also interrupt again instead of completing.
        st = self.store
        st.check_fault("resume-complete")
        run.state = COMPLETED
        run.dispatched_unknown = False
        run.result_attempt = run.attempt_id
        st.after_ok("resume-complete")
        st.check_fault("resume-attempt-clear")
        st.resume_attempts[run.id] = "DONE"
        st.after_ok("resume-attempt-clear")
        self._record(
            kind="completed",
            sched_id=run.sched_id,
            intended=run.intended,
            run_id=run.id,
            reason="resumed-complete",
        )
        return "resumed-complete"

    def _finish_resume_interrupted(self, run: Run) -> str:
        st = self.store
        st.check_fault("resume-interrupt")
        run.state = INTERRUPTED
        run.dispatched_unknown = False
        run.result_attempt = run.attempt_id
        st.after_ok("resume-interrupt")
        st.check_fault("resume-attempt-clear")
        st.resume_attempts[run.id] = "DONE"
        st.after_ok("resume-attempt-clear")
        self._record(
            kind="interrupted",
            sched_id=run.sched_id,
            intended=run.intended,
            run_id=run.id,
            reason="resumed-reinterrupted",
        )
        return "resumed-interrupted"

    # -- recovery ------------------------------------------------------------
    def recover(self, now: datetime) -> list[str]:
        """Startup recovery under exclusive ownership. Returns diagnostics."""
        st = self.store
        if st.lock_holder != self.owner:
            raise SecondOwnerError("lost ownership")
        diags: list[str] = []
        # Admission record is the recovery authority. Recovery NEVER
        # executes: it only materializes missing views (flagged pending for
        # the capacity-checked dispatch sweep in poll()), fails abandoned /
        # ambiguous runs closed, reconciles terminal records, and blocks
        # corrupt schedules. Any RUNNING run lost its in-memory task with
        # the old process: with an admission record it is abandoned
        # (failed, no replay); without one it is corrupt (fail closed).
        # Interrupted runs carry the identity of the attempt that produced
        # them: a result matching the ACTIVE attempt is fresh (resumable);
        # anything else under an ACTIVE attempt is stale (ambiguous, failed).
        for run_id, rec in st.admissions.items():
            if run_id not in st.runs:
                # Admitted but never materialized/dispatched: complete the
                # missing view and flag it pending. The next poll dispatches
                # the captured invocation through capacity checks — exactly
                # once, because the occurrence is already consumed.
                st.runs[run_id] = Run(
                    run_id,
                    rec["sched"],
                    rec["intended"],
                    rec["rev"],
                    dict(rec["input"]),
                )
                st.runs[run_id].needs_dispatch = True
                diags.append(f"{run_id}:view-completed-pending-dispatch")
        for run in st.runs.values():
            if run.state == INTERRUPTED:
                if st.resume_attempts.get(run.id) == "ACTIVE":
                    if (
                        run.result_attempt is not None
                        and run.result_attempt == run.attempt_id
                    ):
                        # The stopped result belongs to the active attempt:
                        # the re-interruption landed before the crash. Fresh,
                        # resumable; the attempt is done executing.
                        st.resume_attempts[run.id] = "DONE"
                        diags.append(f"{run.id}:fresh-result-resumable")
                    else:
                        run.state = FAILED
                        run.fail_reason = (
                            "ambiguous resume attempt: may have executed; "
                            "external effects may already have occurred; no retry"
                        )
                        self._record(
                            kind="failed",
                            sched_id=run.sched_id,
                            intended=run.intended,
                            run_id=run.id,
                            reason=run.fail_reason,
                        )
                        diags.append(f"{run.id}:failed-closed")
                else:
                    diags.append(f"{run.id}:waiting-resumable")
            elif run.state == RUNNING:
                if run.needs_dispatch:
                    # Admitted and materialized but provably never
                    # dispatched: stays pending for the poll sweep, never
                    # failed as abandoned.
                    diags.append(f"{run.id}:pending-dispatch-kept")
                    continue
                if run.id not in st.admissions:
                    sched = st.schedules[run.sched_id]
                    sched.blocked_reason = (
                        f"corrupt run view without admission: {run.id}"
                    )
                    diags.append(f"{run.id}:corrupt-blocked")
                    continue
                if st.resume_attempts.get(run.id) == "ACTIVE":
                    run.state = FAILED
                    run.fail_reason = (
                        "ambiguous resume attempt: may have executed; "
                        "external effects may already have occurred; no retry"
                    )
                else:
                    run.state = FAILED
                    run.fail_reason = (
                        "abandoned execution: outcome unknown; external "
                        "effects may already have occurred; no replay"
                    )
                self._record(
                    kind="failed",
                    sched_id=run.sched_id,
                    intended=run.intended,
                    run_id=run.id,
                    reason=run.fail_reason,
                )
                diags.append(f"{run.id}:failed-closed")
        # Reconcile terminal runs whose history entry was lost to a crash
        # between the state persist and the record append (idempotent: only
        # appends when no terminal record exists for the run). A COMPLETED
        # run with a still-ACTIVE attempt crashed between completion and
        # attempt-clearing: mark DONE, reconcile the record, never re-run.
        for run in st.runs.values():
            if run.state == COMPLETED and st.resume_attempts.get(run.id) == "ACTIVE":
                st.resume_attempts[run.id] = "DONE"
                diags.append(f"{run.id}:attempt-reconciled")
            if run.state == COMPLETED and not any(
                r.kind == "completed" and r.run_id == run.id for r in st.history
            ):
                self._record(
                    kind="completed",
                    sched_id=run.sched_id,
                    intended=run.intended,
                    run_id=run.id,
                    reason="reconciled-on-recovery",
                )
                diags.append(f"{run.id}:terminal-reconciled")
            elif (
                run.state == INTERRUPTED
                and st.resume_attempts.get(run.id) != "ACTIVE"
                and not any(
                    r.kind == "interrupted" and r.run_id == run.id for r in st.history
                )
            ):
                self._record(
                    kind="interrupted",
                    sched_id=run.sched_id,
                    intended=run.intended,
                    run_id=run.id,
                    reason="reconciled-on-recovery",
                )
                diags.append(f"{run.id}:terminal-reconciled")
        return diags


def make_store() -> MemStore:
    st = MemStore()
    st.deployments["dep-1"] = {"rev": 1, "required": ["team", "report_time"]}
    return st


def add_sched(
    st: MemStore,
    sid: str,
    src: CountingMixin,
    sources: dict,
    start: datetime,
    **kw: object,
) -> Schedule:
    if sid in st.schedules:
        raise ValueError(f"schedule id reused: {sid}")
    sched = Schedule(id=sid, **kw)  # type: ignore[arg-type]
    st.schedules[sid] = sched
    st.candidates[sid] = None
    st.consumed[sid] = start
    sources[sid] = src
    return sched


# --------------------------------------------------------------------------
# Tests
# --------------------------------------------------------------------------


def test_defaults_skip_skip():
    s = Schedule(id="s")
    assert s.overlap == "skip" and s.misfire == "skip"


def test_overlap_skip_x_misfire_skip_running_blocks_and_late_drops():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources,
        t0 - timedelta(minutes=10),
    )
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch:
        sch.poll(t0)  # admits 12:00, hangs (RUNNING)
        (run_id,) = list(st.runs)
        sch.poll(t0 + timedelta(minutes=10))  # 12:10 due while active
        kinds = [(r.kind, r.intended) for r in st.history if r.sched_id == "a"]
        assert ("skipped-overlap", ts(2026, 9, 8, 12, 10)) in kinds
        # Late instant beyond allowance with skip: skipped-misfire.
        sch.poll(t0 + timedelta(minutes=30))  # 12:20,12:30 missed (>60s)
        kinds = [(r.kind, r.intended) for r in st.history if r.sched_id == "a"]
        assert ("skipped-misfire", ts(2026, 9, 8, 12, 20)) in kinds
        assert st.runs[run_id].state == RUNNING


def test_overlap_skip_x_misfire_latest_single_catchup_and_supersession():
    st = make_store()
    sources: dict = {}
    add_sched(
        st,
        "h",
        PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
        sources,
        ts(2026, 9, 8, 9, 0),
        misfire="latest",
    )
    with Scheduler(st, "p1", 0, sources) as sch:  # no capacity: hold candidates
        sch.poll(ts(2026, 9, 8, 12, 20))  # missed 10:00,11:00,12:00
        cand = st.candidates["h"]
        assert cand is not None and cand.intended == ts(2026, 9, 8, 12, 0)
        runs_for_h = [r for r in st.history if r.kind == "admitted"]
        assert runs_for_h == [], "ONE pending candidate, not replay-all"
        sup = [r for r in st.history if r.kind == "superseded"]
        assert {r.intended for r in sup} == {
            ts(2026, 9, 8, 10, 0),
            ts(2026, 9, 8, 11, 0),
        }
    # Capacity returns at 13:00 while 13:00 is also due: exactly one admission,
    # for the 13:00 instant (newer supersedes the unadmitted 12:00).
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch2:
        sch2.poll(ts(2026, 9, 8, 13, 0))
        admitted = [r for r in st.history if r.kind == "admitted"]
        assert len(admitted) == 1
        assert admitted[0].intended == ts(2026, 9, 8, 13, 0)
        assert st.candidates["h"] is None, "no stale candidate survives admission"
        assert any(
            r.kind == "superseded" and r.intended == ts(2026, 9, 8, 12, 0)
            for r in st.history
        )
        assert admitted[0].run_id is not None
        assert (
            st.runs[admitted[0].run_id].frozen_input["report_time"]
            == "2026-09-08T13:00:00+00:00"
        )


def test_newer_due_never_supersedes_admitted_run():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources,
        t0 - timedelta(minutes=10),
        misfire="latest",
    )
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch:
        first = sch.poll(t0)["a"]
        run_id = first.split(":", 1)[1]
        assert st.runs[run_id].intended == t0
        sch.poll(t0 + timedelta(minutes=10))
        # 12:10 is skipped-overlap (terminal); the admitted 12:00 run is
        # untouched and no candidate resurrects 12:10.
        assert st.runs[run_id].state == RUNNING
        assert st.candidates["a"] is None
        assert any(
            r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=10)
            for r in st.history
        )


def test_terminal_overlap_skip_never_reappears_through_catchup():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources,
        t0 - timedelta(minutes=10),
        misfire="latest",
    )
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch:
        sch.poll(t0)
        sch.poll(t0 + timedelta(minutes=10))  # terminal skipped-overlap @12:10
    # Restart abandons the hanging in-flight task (failed, no replay), but
    # the terminal 12:10 skip is never reconstructed as a candidate.
    with Scheduler(st, "p1", 0, sources) as sch2:
        diags = sch2.recover(t0 + timedelta(minutes=11))
        assert any("failed-closed" in d for d in diags)
        sch2.poll(t0 + timedelta(minutes=25))  # 12:20 missed -> held candidate
        cand = st.candidates["a"]
        assert cand is not None and cand.intended == t0 + timedelta(minutes=20)
        intents = [
            r.intended
            for r in st.history
            if r.kind == "admitted" and r.intended == t0 + timedelta(minutes=10)
        ]
        assert intents == [], "terminally skipped 12:10 must never be admitted"


def test_parallel_limits_count_interrupted_and_waiting_frees_task_slot():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "p",
        PeriodicSource(timedelta(minutes=5), t0),
        sources,
        t0 - timedelta(minutes=5),
        overlap="parallel",
        max_active=2,
    )
    with Scheduler(st, "owner", 1, sources, {"*": "hang"}) as sch:
        sch.poll(t0)  # run1 RUNNING (task slot taken)
        assert sch._task_load() == 1
        st.runs["run-p-1"].state = INTERRUPTED  # scripted durable interrupt
        st.history.append(
            Record(kind="interrupted", sched_id="p", intended=t0, run_id="run-p-1")
        )
        assert sch._task_load() == 0, "waiting interruptions hold no task slot"
        sch.poll(t0 + timedelta(minutes=5))  # run2 admitted (1 task slot free)
        assert st.runs["run-p-2"].state == RUNNING
        # max_active=2 reached (1 waiting + 1 running): 12:10 skipped-overlap.
        sch.poll(t0 + timedelta(minutes=10))
        assert any(
            r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=10)
            for r in st.history
        )
        # Lowering the limit never cancels; blocks new admission instead.
        st.schedules["p"].max_active = 1
        st.runs["run-p-2"].state = INTERRUPTED
        sch.poll(t0 + timedelta(minutes=15))
        assert any(
            r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=15)
            for r in st.history
        )
        assert st.runs["run-p-1"].state == INTERRUPTED  # not cancelled
        # Resume needs a task slot: none free while... free one by completing.
        st.runs["run-p-2"].state = COMPLETED
        assert (
            sch.resume_run("run-p-1", t0 + timedelta(minutes=16)) == "resumed-complete"
        )


def test_explicit_pause_is_not_downtime():
    st = make_store()
    sources: dict = {}
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
        sources,
        ts(2026, 9, 8, 9, 0),
        misfire="latest",
    )
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        sch.poll(ts(2026, 9, 8, 9, 0))
        st.schedules["a"].paused = True
        sch.poll(ts(2026, 9, 8, 10, 30))  # paused polls exclude the interval
        assert st.candidates["a"] is None
        sch.poll(ts(2026, 9, 8, 11, 30))
        sch.resume_schedule("a", ts(2026, 9, 8, 12, 30))  # next future only
        sch.poll(ts(2026, 9, 8, 12, 30))
        assert st.candidates["a"] is None, "paused times are not replayed"
        admitted_intents = [r.intended for r in st.history if r.kind == "admitted"]
        assert ts(2026, 9, 8, 10, 0) not in admitted_intents
        assert ts(2026, 9, 8, 11, 0) not in admitted_intents
        assert ts(2026, 9, 8, 12, 0) not in admitted_intents
        # Contrast: enabled downtime DOES catch up under latest.
        st2 = make_store()
        sources2: dict = {}
        add_sched(
            st2,
            "b",
            PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
            sources2,
            ts(2026, 9, 8, 9, 0),
            misfire="latest",
        )
        sch2 = Scheduler(st2, "p1", 0, sources2)
        sch2.poll(ts(2026, 9, 8, 12, 20))  # 3 missed while "down", enabled
        assert st2.candidates["b"] is not None
        assert st2.candidates["b"].intended == ts(2026, 9, 8, 12, 0)  # type: ignore[union-attr]
        sch2.close()


def test_edit_discards_candidates_no_backfill_delete_keeps_history():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)),
        sources,
        ts(2026, 9, 8, 9, 0),
        misfire="latest",
    )
    with Scheduler(st, "p1", 0, sources) as sch:
        sch.poll(ts(2026, 9, 8, 12, 20))
        assert st.candidates["a"] is not None
        sch.edit_schedule("a", ts(2026, 9, 8, 12, 21))  # definition edit
        sch.poll(ts(2026, 9, 8, 12, 21))
        assert st.candidates["a"] is None, "edits discard old-revision candidates"
        assert any(
            r.kind == "superseded" and r.reason == "schedule-edit" for r in st.history
        )
        # No backfill before the edit: consumed advanced to edit time.
        assert st.consumed["a"] >= ts(2026, 9, 8, 12, 21)
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch:
        sch.poll(ts(2026, 9, 8, 13, 0))
        (run_id,) = [r for r in st.runs if st.runs[r].state == RUNNING]
        st.schedules["a"].deleted = True  # delete clears pending, keeps rest
        sch.poll(ts(2026, 9, 8, 14, 0))
        assert st.candidates["a"] is None
        assert st.runs[run_id].state == RUNNING, "delete must not cancel active runs"
        n_history = len(st.history)
        assert n_history > 0, "history remains intact"
        # In-flight work can still finish after delete; history is appended.
        st.runs[run_id].state = COMPLETED
        st.history.append(
            Record(
                kind="completed",
                sched_id="a",
                intended=st.runs[run_id].intended,
                run_id=run_id,
            )
        )
        assert len(st.history) == n_history + 1
        with pytest.raises(ValueError, match="schedule id reused"):
            add_sched(st, "a", PeriodicSource(timedelta(hours=1), t0), sources, t0)


def test_restart_survives_candidate_and_rollback_never_readmits():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources,
        t0 - timedelta(minutes=10),
        misfire="latest",
    )
    with Scheduler(st, "p1", 0, sources) as sch:
        sch.poll(t0 + timedelta(minutes=25))  # candidate 12:20
        assert st.candidates["a"] is not None
    with Scheduler(st, "p1", 0, sources) as sch2:  # restart, still no capacity
        sch2.recover(t0 + timedelta(minutes=26))
        sch2.poll(t0 + timedelta(minutes=26))
        assert st.candidates["a"] is not None
        assert st.candidates["a"].intended == t0 + timedelta(minutes=20)  # type: ignore[union-attr]
        # Clock rollback: consumed instants are never re-admitted.
        sch2.poll(t0 - timedelta(hours=1))
        admitted = [r for r in st.history if r.kind == "admitted"]
        assert admitted == []


def test_long_downtime_bounded_and_summarized():
    st = make_store()
    sources: dict = {}
    src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0))
    add_sched(st, "m", src, sources, ts(2023, 9, 8, 12, 0), misfire="latest")
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        sch.poll(ts(2026, 9, 8, 12, 0, 0))  # 3 years of per-minute misses
        total_calls = src.next_calls + src.prev_calls
        assert total_calls <= SCAN_CAP + 2, f"must not enumerate: {total_calls} calls"
        # Capacity available: latest means ONE prompt admission for the latest
        # missed instant (11:59), using the candidate's intended time.
        admitted = [r for r in st.history if r.kind == "admitted"]
        assert len(admitted) == 1
        assert admitted[0].intended == ts(2026, 9, 8, 11, 59)
        assert st.candidates["m"] is None  # consumed by the prompt admission
        summaries = [r for r in st.history if r.kind == "interval-summary"]
        assert len(summaries) == 1, "one coalesced summary, not per-minute rows"
    # Same gap with no capacity: exactly ONE held candidate, zero admissions.
    st1b = make_store()
    sources1b: dict = {}
    src1b = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0))
    add_sched(st1b, "m", src1b, sources1b, ts(2023, 9, 8, 12, 0), misfire="latest")
    with Scheduler(st1b, "p1", 0, sources1b, {"*": "complete"}) as sch:
        sch.poll(ts(2026, 9, 8, 12, 0, 0))
        assert src1b.next_calls + src1b.prev_calls <= SCAN_CAP + 2
        assert st1b.candidates["m"] is not None
        assert st1b.candidates["m"].intended == ts(2026, 9, 8, 11, 59)  # type: ignore[union-attr]
        assert [r for r in st1b.history if r.kind == "admitted"] == []
    # skip policy: same boundedness, zero admissions, next future selected.
    st2 = make_store()
    sources2: dict = {}
    src2 = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0))
    add_sched(st2, "m", src2, sources2, ts(2023, 9, 8, 12, 0), misfire="skip")
    with Scheduler(st2, "p1", 4, sources2, {"*": "complete"}) as sch:
        sch.poll(ts(2026, 9, 8, 12, 0, 30))
        assert src2.next_calls + src2.prev_calls <= SCAN_CAP + 2
        assert [r for r in st2.history if r.kind == "admitted"] == []
        assert st2.candidates["m"] is None


def test_fairness_frequent_schedule_cannot_monopolize():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "fast",
        PeriodicSource(timedelta(minutes=1), t0),
        sources,
        t0 - timedelta(minutes=1),
    )
    add_sched(st, "slow", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    with Scheduler(st, "p1", 1, sources, {"*": "hang"}) as sch:
        sch.poll(t0)  # rotation starts at fast (sorted first): fast admitted
        assert any(r.sched_id == "fast" and r.kind == "admitted" for r in st.history)
        # Complete fast's run externally, next poll must serve slow first.
        for r in st.runs.values():
            r.state = COMPLETED
        res = sch.poll(t0 + timedelta(seconds=30))
        slow_admitted = [
            r for r in st.history if r.sched_id == "slow" and r.kind == "admitted"
        ]
        assert slow_admitted, f"slow schedule starved: {res}"


def test_fault_before_admission_never_dispatches():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["admission"] = "before"
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False, "fault must propagate"
        except InjectedFault:
            pass
        assert st.runs == {}, "no dispatch before durable admission"
        assert st.admissions == {}
        sch.poll(t0)  # retry after the fault is clean
        assert len(st.runs) == 1


def test_fault_between_admission_and_view_recovers_without_redispatch():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["materialize"] = "before"
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except InjectedFault:
            pass
        assert len(st.admissions) == 1 and len(st.runs) == 0
        # Recovery NEVER executes: it completes the view and flags it
        # pending. No outcome exists yet.
        diags = sch.recover(t0)
        assert any("pending-dispatch" in d for d in diags)
        assert len(st.runs) == 1
        run = st.runs["run-a-1"]
        assert run.state == RUNNING and run.needs_dispatch
        assert not run.dispatched_unknown
        assert [r for r in st.history if r.kind == "completed"] == []
        # The next poll dispatches the captured invocation through the
        # capacity checks — exactly once, no second admission.
        sch.poll(t0)
        assert run.state == COMPLETED and not run.needs_dispatch
        n_admitted = len([r for r in st.history if r.kind == "admitted"])
        assert n_admitted == 1, "reconciliation must be idempotent"
        assert len([r for r in st.history if r.kind == "completed"]) == 1


def test_recovery_pending_dispatch_waits_for_capacity():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["materialize"] = "before"
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except InjectedFault:
            pass
        sch.recover(t0)
        sch.capacity = 0  # slots full: pending dispatch must wait
        sch.poll(t0 + timedelta(seconds=1))
        assert st.runs["run-a-1"].needs_dispatch
        assert [r for r in st.history if r.kind == "completed"] == []
        sch.capacity = 4
        sch.poll(t0 + timedelta(seconds=2))
        assert not st.runs["run-a-1"].needs_dispatch
        assert st.runs["run-a-1"].state == COMPLETED


def test_run_ids_survive_restart_without_reuse():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources,
        t0 - timedelta(minutes=10),
    )
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch:
        sch.poll(t0)
        assert st.runs["run-a-1"].intended == t0
    # Restart: the counter lives in the store, so the next admission gets
    # a fresh identity instead of overwriting run-a-1.
    with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch2:
        sch2.recover(t0 + timedelta(seconds=1))  # hanging run abandoned
        assert st.runs["run-a-1"].state == FAILED
        sch2.poll(t0 + timedelta(minutes=10))
        assert st.runs["run-a-1"].intended == t0, "old run untouched"
        assert st.runs["run-a-2"].intended == t0 + timedelta(minutes=10)
        assert len(st.runs) == 2


def test_crash_after_dispatch_marks_abandoned_failed_without_replay():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(hours=1), t0),
        sources,
        t0 - timedelta(hours=1),
    )
    with Scheduler(st, "p1", 4, sources, {"run-a-1": "crash", "*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except ExecutorCrashed:
            pass
        run = st.runs["run-a-1"]
        assert run.state == RUNNING and run.dispatched_unknown
        diags = sch.recover(t0 + timedelta(seconds=5))
        assert run.state == FAILED
        assert "may already have occurred" in run.fail_reason
        assert any("failed-closed" in d for d in diags)
        # Future occurrences proceed; the failed one is never replayed.
        sch.poll(t0 + timedelta(hours=1))
        intents = [r.intended for r in st.history if r.kind == "admitted"]
        assert t0 not in intents[1:] or intents.count(t0) == 1


def test_crash_during_resume_must_not_look_safe_to_retry():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch:
        sch.poll(t0)
        (run_id,) = list(st.runs)
        assert st.runs[run_id].state == INTERRUPTED
    # Restart: a merely-waiting interruption is safe to resume later...
    with Scheduler(st, "p1", 4, sources, {"*": "crash"}) as sch2:
        sch2.recover(t0)
        assert st.runs[run_id].state == INTERRUPTED
        assert run_id not in st.resume_attempts
        try:
            sch2.resume_run(run_id, t0)
            assert False
        except ExecutorCrashed:
            pass
        # ...but the crash left a durably marked ACTIVE attempt: recovery
        # must fail it closed instead of presenting the old checkpoint again.
        assert st.resume_attempts[run_id] == "ACTIVE"
        diags = sch2.recover(t0)
        assert st.runs[run_id].state == FAILED
        assert "ambiguous" in st.runs[run_id].fail_reason
        assert any("failed-closed" in d for d in diags)


def test_preflight_rejection_invents_no_run_and_freezes_input():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(hours=1), t0),
        sources,
        t0 - timedelta(hours=1),
        misfire="latest",
    )
    with Scheduler(st, "p1", 0, sources) as sch:
        sch.poll(t0 + timedelta(minutes=5))  # held candidate @12:00
        assert st.candidates["a"] is not None
        # Deployment edit changes the expected input before admission.
        st.deployments["dep-1"] = {
            "rev": 2,
            "required": ["team", "report_time", "region"],
        }
        sch2 = Scheduler(st, "p1", 4, sources, {"*": "complete"})
        sch2.poll(t0 + timedelta(minutes=6))
        assert st.runs == {}, "preflight rejection must invent no run"
        assert any(r.kind == "preflight-rejected" for r in st.history)
        sch2.close()
    # Admitted runs freeze their invocation: later edits change nothing.
    st3 = make_store()
    sources3: dict = {}
    add_sched(st3, "a", OneShotSource(t0), sources3, t0 - timedelta(hours=1))
    with Scheduler(st3, "p1", 4, sources3, {"*": "hang"}) as sch:
        sch.poll(t0)
        (run_id,) = list(st3.runs)
        before = dict(st3.runs[run_id].frozen_input)
        st3.deployments["dep-1"] = {"rev": 9, "required": ["team"]}
        st3.runs[run_id].state = COMPLETED
        assert st3.runs[run_id].frozen_input == before


def test_exclusive_ownership_and_unsupported_locking():
    st = make_store()
    sources: dict = {}
    sch1 = Scheduler(st, "proc-A", 1, sources)
    try:
        Scheduler(st, "proc-B", 1, sources)
        assert False, "second owner must be rejected"
    except SecondOwnerError:
        pass
    # A held lock never expires while the owner lives (no lease timeout).
    assert not hasattr(st, "lease_expiry")
    sch1.close()  # process death releases
    sch2 = Scheduler(st, "proc-B", 1, sources)  # new owner starts, recovers
    sch2.close()
    bad = MemStore(lockable=False)
    try:
        Scheduler(bad, "proc-C", 1, {})
        assert False, "unsupported locking must reject scheduler startup"
    except StartupRejected:
        pass


def test_corrupt_view_without_admission_fails_closed():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "a",
        PeriodicSource(timedelta(hours=1), t0),
        sources,
        t0 - timedelta(hours=1),
    )
    st.runs["ghost-1"] = Run("ghost-1", "a", t0, 1, {"team": "eng"})
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        diags = sch.recover(t0)
        assert any("corrupt-blocked" in d for d in diags)
        assert st.schedules["a"].blocked_reason is not None
        try:
            sch.poll(t0 + timedelta(hours=1))
            assert False, "corrupt records fail closed with diagnostics"
        except BlockedSchedule:
            pass


def test_overlap_parallel_x_misfire_latest():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(
        st,
        "p",
        PeriodicSource(timedelta(minutes=5), t0),
        sources,
        t0 - timedelta(minutes=5),
        overlap="parallel",
        max_active=2,
        misfire="latest",
    )
    with Scheduler(st, "p1", 0, sources) as sch:  # no task slots: hold
        sch.poll(t0 + timedelta(minutes=12))  # 12:00,05,10 missed
        cand = st.candidates["p"]
        assert cand is not None and cand.intended == t0 + timedelta(minutes=10)
        assert [r for r in st.history if r.kind == "admitted"] == []
    with Scheduler(st, "p1", 1, sources, {"*": "hang"}) as sch:
        sch.poll(t0 + timedelta(minutes=13))  # admits held 12:10, hangs
        assert st.runs["run-p-1"].intended == t0 + timedelta(minutes=10)
        # Task slot taken: 12:15 is held as the one latest candidate.
        sch.poll(t0 + timedelta(minutes=15))
        assert list(st.runs) == ["run-p-1"]
        assert st.candidates["p"] is not None
        assert st.candidates["p"].intended == t0 + timedelta(minutes=15)  # type: ignore[union-attr]
        # Slot frees: the held 12:15 candidate is admitted (hangs).
        st.runs["run-p-1"].state = COMPLETED
        sch.poll(t0 + timedelta(minutes=16))
        assert st.runs["run-p-2"].intended == t0 + timedelta(minutes=15)
        # run-p-2 waits durably: schedule slot held, task slot free. Admit a
        # second hanging run to reach the cap, then the next due instant is
        # terminally skipped-overlap.
        st.runs["run-p-2"].state = INTERRUPTED
        st.history.append(
            Record(
                kind="interrupted",
                sched_id="p",
                intended=t0 + timedelta(minutes=15),
                run_id="run-p-2",
            )
        )
        sch.poll(t0 + timedelta(minutes=20))  # admits 12:20, hangs
        assert st.runs["run-p-3"].intended == t0 + timedelta(minutes=20)
        sch.poll(t0 + timedelta(minutes=25))  # 2 active >= max: terminal skip
        assert any(
            r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=25)
            for r in st.history
        )
        # And a later catch-up admits the earliest missed instant ASAP
        # (12:30), holds only the newest (12:45), and never resurrects the
        # skipped 12:25.
        st.runs["run-p-2"].state = COMPLETED
        st.runs["run-p-3"].state = COMPLETED
        sch.poll(t0 + timedelta(minutes=45))  # 12:30..45 missed
        just = [
            r
            for r in st.history
            if r.kind == "admitted" and r.intended == t0 + timedelta(minutes=30)
        ]
        assert len(just) == 1
        cand = st.candidates["p"]
        assert cand is not None and cand.intended == t0 + timedelta(minutes=45)
        assert all(
            r.intended != t0 + timedelta(minutes=25)
            for r in st.history
            if r.kind == "admitted"
        )


def test_manual_and_other_schedule_runs_are_overlap_independent():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    add_sched(st, "b", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    # Manual runs and other schedules' runs never join this schedule's check.
    st.runs["manual-1"] = Run("manual-1", "manual", t0, 1, {"team": "eng"})
    st.runs["run-b-0"] = Run("run-b-0", "b", t0, 1, {"team": "eng"})
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        sch.poll(t0)
        admitted_a = [
            r for r in st.history if r.kind == "admitted" and r.sched_id == "a"
        ]
        assert len(admitted_a) == 1, "overlap=skip ignores manual/other runs"
        # ...while b is blocked by its OWN running run (per-schedule scope
        # cuts both ways: b's check sees run-b-0, a's check does not).
        skipped_b = [
            r for r in st.history if r.kind == "skipped-overlap" and r.sched_id == "b"
        ]
        assert len(skipped_b) == 1 and skipped_b[0].intended == t0


def test_capacity_wait_then_admit_or_expire_for_skip():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    with Scheduler(st, "p1", 0, sources) as sch:  # full: within allowance
        assert sch.poll(t0) == {"a": "admit:held-undecided"}
        assert st.consumed["a"] < t0, "undecided instants stay unconsumed"
        assert st.candidates["a"] is None, "skip holds no candidate"
    with Scheduler(st, "p1", 1, sources, {"*": "complete"}) as sch:
        sch.poll(t0 + timedelta(seconds=30))  # still within allowance
        admitted = [r for r in st.history if r.kind == "admitted"]
        assert len(admitted) == 1 and admitted[0].intended == t0
    # Past the deadline instead: capacity-delayed skip expires.
    st2 = make_store()
    sources2: dict = {}
    add_sched(
        st2,
        "a",
        PeriodicSource(timedelta(minutes=10), t0),
        sources2,
        t0 - timedelta(hours=1),
    )
    with Scheduler(st2, "p1", 0, sources2) as sch:
        sch.poll(t0)
        assert sch.poll(t0 + timedelta(seconds=61))["a"] == "skipped-misfire"
        assert any(
            r.kind == "skipped-misfire" and r.intended == t0 for r in st2.history
        )


def test_fault_after_admission_recovers_exactly_once():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["admission"] = "after"  # record persisted, crash before view
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except InjectedFault:
            pass
        assert len(st.admissions) == 1 and len(st.runs) == 0
        # Recovery materializes the view but never executes: pending.
        diags = sch.recover(t0)
        assert any("pending-dispatch" in d for d in diags)
        run = st.runs["run-a-1"]
        assert run.state == RUNNING and run.needs_dispatch
        assert [r for r in st.history if r.kind == "completed"] == []
        sch.poll(t0)  # sweep dispatches exactly once
        assert run.state == COMPLETED and not run.needs_dispatch
        assert len([r for r in st.history if r.kind == "admitted"]) == 1
        assert len([r for r in st.history if r.kind == "completed"]) == 1


def test_fault_after_complete_reconciles_terminal_record():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["complete"] = "after"  # COMPLETED persisted, record lost
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs["run-a-1"].state == COMPLETED
        assert [r for r in st.history if r.kind == "completed"] == []
        diags = sch.recover(t0)
        assert any("terminal-reconciled" in d for d in diags)
        assert len([r for r in st.history if r.kind == "completed"]) == 1
        sch.poll(t0 + timedelta(hours=1))  # consumed advanced: no re-admit
        assert len([r for r in st.history if r.kind == "admitted"]) == 1


def test_fault_after_resume_mark_fails_closed_not_retried():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch:
        sch.poll(t0)
        (run_id,) = list(st.runs)
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        st.faults["resume-mark"] = "after"  # ACTIVE persisted, never executed
        try:
            sch.resume_run(run_id, t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs[run_id].state == INTERRUPTED  # never re-ran
        assert st.resume_attempts[run_id] == "ACTIVE"
        diags = sch.recover(t0)  # conservative: ambiguous, never retried
        assert st.runs[run_id].state == FAILED
        assert "ambiguous" in st.runs[run_id].fail_reason
        assert any("failed-closed" in d for d in diags)


def test_fault_after_interrupt_reconciles_waiting_state():
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    st.faults["interrupt-persist"] = "after"  # INTERRUPTED kept, record lost
    with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch:
        try:
            sch.poll(t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs["run-a-1"].state == INTERRUPTED
        diags = sch.recover(t0)
        assert st.runs["run-a-1"].state == INTERRUPTED  # still resumable
        assert any("terminal-reconciled" in d for d in diags)
        # A resumed run may interrupt AGAIN: re-interruption is a durable
        # terminal persist of its own, and the run stays resumable after it.
        assert sch.resume_run("run-a-1", t0) == "resumed-interrupted"
        assert st.runs["run-a-1"].state == INTERRUPTED
        assert st.resume_attempts["run-a-1"] == "DONE"
        assert any(
            r.kind == "interrupted" and r.reason == "resumed-reinterrupted"
            for r in st.history
        )
        sch.outcomes["run-a-1"] = "complete"
        assert sch.resume_run("run-a-1", t0) == "resumed-complete"


def _interrupted_run() -> tuple[MemStore, dict, str]:
    st = make_store()
    sources: dict = {}
    t0 = ts(2026, 9, 8, 12, 0)
    add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1))
    with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch:
        sch.poll(t0)
        (run_id,) = list(st.runs)
    assert st.runs[run_id].state == INTERRUPTED
    return st, sources, run_id


def test_fault_after_resume_complete_reconciles_without_rerun():
    st, sources, run_id = _interrupted_run()
    t0 = ts(2026, 9, 8, 12, 0)
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        st.faults["resume-complete"] = "after"  # COMPLETED kept, rest lost
        try:
            sch.resume_run(run_id, t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs[run_id].state == COMPLETED
        assert st.resume_attempts[run_id] == "ACTIVE"
        assert [r for r in st.history if r.kind == "completed"] == []
        diags = sch.recover(t0)
        # Reconciled, never re-executed: exactly one completion, DONE marker.
        assert st.resume_attempts[run_id] == "DONE"
        assert any("attempt-reconciled" in d for d in diags)
        assert len([r for r in st.history if r.kind == "completed"]) == 1
        sch.poll(t0 + timedelta(hours=1))
        assert len([r for r in st.history if r.kind == "completed"]) == 1


def test_fault_after_resume_attempt_clear_reconciles_record():
    st, sources, run_id = _interrupted_run()
    t0 = ts(2026, 9, 8, 12, 0)
    with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch:
        st.faults["resume-attempt-clear"] = "after"  # DONE kept, record lost
        try:
            sch.resume_run(run_id, t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs[run_id].state == COMPLETED
        assert st.resume_attempts[run_id] == "DONE"
        assert [r for r in st.history if r.kind == "completed"] == []
        diags = sch.recover(t0)
        assert any("terminal-reconciled" in d for d in diags)
        assert len([r for r in st.history if r.kind == "completed"]) == 1


def test_fault_before_resume_interrupt_fails_closed():
    st, sources, run_id = _interrupted_run()
    t0 = ts(2026, 9, 8, 12, 0)
    with Scheduler(st, "p1", 4, sources, {"run-a-1": "interrupt"}) as sch:
        st.faults["resume-interrupt"] = "before"  # re-interrupt never persisted
        try:
            sch.resume_run(run_id, t0)
            assert False
        except InjectedFault:
            pass
        # The ACTIVE marker superseded the old waiting checkpoint, but the
        # re-interruption never landed: fail closed, never retry the old one.
        assert st.runs[run_id].state == RUNNING
        assert st.resume_attempts[run_id] == "ACTIVE"
        assert st.runs[run_id].result_attempt != st.runs[run_id].attempt_id
        diags = sch.recover(t0)
        assert st.runs[run_id].state == FAILED
        assert "ambiguous" in st.runs[run_id].fail_reason
        assert any("failed-closed" in d for d in diags)


def test_fault_after_resume_interrupt_stays_resumable():
    # Exact repro shape: the re-interruption persisted WITH the attempt
    # identity, then the crash hit before attempt-clearing. Recovery must
    # match result to attempt and keep the run resumable — not fail it.
    st, sources, run_id = _interrupted_run()
    t0 = ts(2026, 9, 8, 12, 0)
    with Scheduler(st, "p1", 4, sources, {"run-a-1": "interrupt"}) as sch:
        st.faults["resume-interrupt"] = "after"
        try:
            sch.resume_run(run_id, t0)
            assert False
        except InjectedFault:
            pass
        assert st.runs[run_id].state == INTERRUPTED
        assert st.resume_attempts[run_id] == "ACTIVE"
        assert st.runs[run_id].result_attempt == st.runs[run_id].attempt_id
        assert st.runs[run_id].attempt_id > 0
        diags = sch.recover(t0)
        assert st.runs[run_id].state == INTERRUPTED, "fresh result: resumable"
        assert st.resume_attempts[run_id] == "DONE"
        assert any("fresh-result-resumable" in d for d in diags)
        assert len([r for r in st.history if r.kind == "interrupted"]) == 1
        sch.outcomes[run_id] = "complete"
        assert sch.resume_run(run_id, t0) == "resumed-complete"