sched: typed history recorder with exactly-once terminal reconcile (F4)

src/wf_scheduling/history.py

@@ -0,0 +1,115 @@
+"""Typed occurrence-history interface with idempotent reconciliation (R4/F4).
+
+Both the poll loop and startup recovery record through :class:`HistoryRecorder`,
+so completed/failed/interrupted entries share one shape and one idempotency
+identity: ``(run_id, kind, checkpoint_id)``. A resumed run that interrupts
+again produces a new checkpoint id and therefore a new entry; repeating
+recovery never duplicates an entry. The file-backed recorder derives entry
+identity the same way the scheduler always has (occurrence instants hash to
+``occurrence_id``; interval summaries use their span key).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from wf_scheduling.models import OccurrenceKind, OccurrenceRecord
+from wf_scheduling.occurrences import occurrence_id
+
+UTC = timezone.utc
+
+TERMINAL_KINDS: tuple[str, str, str] = ("completed", "interrupted", "failed")
+
+
+@dataclass(frozen=True)
+class HistoryEntry:
+    """One occurrence-history entry with a stable idempotency identity."""
+
+    schedule_id: str
+    kind: OccurrenceKind
+    resolved_at: datetime | None = None
+    run_id: str | None = None
+    revision: int | None = None
+    reason: str = ""
+    admitted_at: datetime | None = None
+    started_at: datetime | None = None
+    checkpoint_id: str | None = None
+    interval_start: datetime | None = None
+    interval_end: datetime | None = None
+    interval_count: int = 0
+    created_at: datetime | None = None
+
+
+class HistoryRecorder(Protocol):
+    """Occurrence-history sink shared by polling and recovery."""
+
+    def record(self, entry: HistoryEntry) -> None:
+        """Append one entry (callers dedup stopped results first)."""
+        ...
+
+    def has_terminal(
+        self,
+        schedule_id: str,
+        run_id: str,
+        kind: str,
+        checkpoint_id: str | None = None,
+    ) -> bool:
+        """Whether this exact stopped result was already reconciled."""
+        ...
+
+
+def entry_occurrence_id(entry: HistoryEntry, created: datetime) -> str:
+    """Derive the history occurrence id with the scheduler's standing rules."""
+    if entry.resolved_at is not None:
+        return occurrence_id(entry.schedule_id, entry.resolved_at)
+    if entry.interval_start is not None and entry.interval_end is not None:
+        start = entry.interval_start.isoformat()
+        end = entry.interval_end.isoformat()
+        return f"{entry.schedule_id}|summary|{start}|{end}"
+    return f"{entry.schedule_id}|summary|{created.isoformat()}"
+
+
+class FileScheduleHistoryRecorder:
+    """History recorder backed by a file schedule store."""
+
+    def __init__(self, schedule_store: Any) -> None:
+        self.schedule_store = schedule_store
+
+    def record(self, entry: HistoryEntry) -> None:
+        created = (
+            entry.created_at if entry.created_at is not None else datetime.now(UTC)
+        )
+        self.schedule_store.append_history(
+            OccurrenceRecord(
+                schedule_id=entry.schedule_id,
+                occurrence_id=entry_occurrence_id(entry, created),
+                kind=entry.kind,
+                resolved_at=entry.resolved_at,
+                run_id=entry.run_id,
+                revision=entry.revision,
+                reason=entry.reason,
+                admitted_at=entry.admitted_at,
+                started_at=entry.started_at,
+                checkpoint_id=entry.checkpoint_id,
+                interval_start=entry.interval_start,
+                interval_end=entry.interval_end,
+                interval_count=entry.interval_count,
+                created_at=created,
+            )
+        )
+
+    def has_terminal(
+        self,
+        schedule_id: str,
+        run_id: str,
+        kind: str,
+        checkpoint_id: str | None = None,
+    ) -> bool:
+        return self.schedule_store.has_history_entry(
+            schedule_id,
+            run_id=run_id,
+            kind=kind,
+            checkpoint_id=checkpoint_id,
+        )

src/wf_scheduling/models.py

@@ -156,6 +156,14 @@ class OccurrenceRecord(BaseModel):
     reason: str = ""
     admitted_at: datetime | None = None
     started_at: datetime | None = None
+    checkpoint_id: str | None = Field(
+        default=None,
+        description=(
+            "Stopped checkpoint this entry reconciles, for exactly-once "
+            "recovery dedup: a resumed run that stops again carries a new "
+            "checkpoint id and therefore a new entry."
+        ),
+    )
     interval_start: datetime | None = None
     interval_end: datetime | None = None
     interval_count: int = 0

src/wf_scheduling/poll.py

@@ -25,8 +25,12 @@ from wf_scheduling.calendar import (
     OneShotSource,
 )
 from wf_scheduling.dispatch import RunDispatcher, StillRunning
-from wf_scheduling.models import OccurrenceRecord, PendingCandidate
-from wf_scheduling.occurrences import occurrence_id
+from wf_scheduling.history import (
+    FileScheduleHistoryRecorder,
+    HistoryEntry,
+    HistoryRecorder,
+)
+from wf_scheduling.models import PendingCandidate
 from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
 from wf_scheduling.prepare import InvocationPreparer, PreparationRejected
 
@@ -116,6 +120,7 @@ class Scheduler:
         preparer: InvocationPreparer,
         dispatcher: RunDispatcher,
         ownership: SchedulerOwnership,
+        history: HistoryRecorder | None = None,
     ) -> None:
         self.schedule_store = schedule_store
         self.run_store = run_store
@@ -124,6 +129,11 @@ class Scheduler:
         self.preparer = preparer
         self.dispatcher = dispatcher
         self.ownership = ownership
+        self.history: HistoryRecorder = (
+            history
+            if history is not None
+            else FileScheduleHistoryRecorder(schedule_store)
+        )
         self._poll_cursor = 0
 
     def _require_ownership(self) -> None:
@@ -191,18 +201,12 @@ class Scheduler:
         now: datetime | None = None,
         admitted_at: datetime | None = None,
         started_at: datetime | None = None,
+        checkpoint_id: str | None = None,
     ) -> None:
         created = now if now is not None else datetime.now(UTC)
-        if intended is not None:
-            oid = occurrence_id(sched_id, intended)
-        elif interval is not None:
-            oid = f"{sched_id}|summary|{interval[0].isoformat()}|{interval[1].isoformat()}"
-        else:
-            oid = f"{sched_id}|summary|{created.isoformat()}"
-        self.schedule_store.append_history(
-            OccurrenceRecord(
+        self.history.record(
+            HistoryEntry(
                 schedule_id=sched_id,
-                occurrence_id=oid,
                 kind=kind,  # type: ignore[arg-type]
                 resolved_at=intended,
                 run_id=run_id,
@@ -210,6 +214,7 @@ class Scheduler:
                 reason=reason,
                 admitted_at=admitted_at,
                 started_at=started_at,
+                checkpoint_id=checkpoint_id,
                 interval_start=interval[0] if interval else None,
                 interval_end=interval[1] if interval else None,
                 interval_count=count,
@@ -217,6 +222,20 @@ class Scheduler:
             )
         )
 
+    def _existing_occurrence_run(self, sched_id: str, intended: datetime) -> str | None:
+        """Return the run already owning this occurrence, if any.
+
+        Identity is ``(schedule_id, resolved UTC instant)`` from the durable
+        admission record — the admission persist is the decision point, so
+        admissions are scanned rather than views (a crashed admission may
+        not have a view yet). Manual runs carry no scheduled instant and
+        never match.
+        """
+        for admission in self.run_store.list_admissions():
+            if admission.schedule_id == sched_id and admission.scheduled_at == intended:
+                return admission.id
+        return None
+
     def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None:
         if getattr(sched, "blocked_reason", None):
             raise BlockedSchedule(getattr(sched, "blocked_reason"))
@@ -232,6 +251,30 @@ class Scheduler:
                 sched.blocked_reason = reason
                 self.schedule_store.save_schedule(sched)
                 raise BlockedSchedule(reason) from None
+        existing = self._existing_occurrence_run(sched.id, intended)
+        if existing is not None:
+            # The occurrence already owns a run (crash between the admission
+            # persist and the watermark/history writes, or a lost watermark
+            # write): an occurrence is immutable and never replayed. Advance
+            # the watermark, reconcile a missing admitted entry exactly once,
+            # and return the owner without dispatching (the pending sweep
+            # owns dispatch).
+            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)
+            _save_consumed_max(self.schedule_store, sched.id, intended)
+            if not self.history.has_terminal(sched.id, existing, "admitted", None):
+                self._record(
+                    kind="admitted",
+                    sched_id=sched.id,
+                    intended=intended,
+                    run_id=existing,
+                    reason=f"rev={sched.revision}",
+                    revision=sched.revision,
+                    now=now,
+                    admitted_at=now,
+                )
+            return existing
         prepared = self.preparer.prepare(sched=sched, intended=intended, now=now)
         if isinstance(prepared, PreparationRejected):
             self._record(
@@ -377,6 +420,7 @@ class Scheduler:
             run_id=run_id,
             now=now,
             started_at=now,
+            checkpoint_id=stopped.latest_checkpoint_id,
         )
 
     def record_stopped_execution(self, run_id: str, result: Any, now: datetime) -> None:
@@ -422,6 +466,7 @@ class Scheduler:
             run_id=run_id,
             now=now,
             started_at=now,
+            checkpoint_id=stopped.latest_checkpoint_id,
         )
 
     # -- polling --------------------------------------------------------
@@ -476,7 +521,7 @@ class Scheduler:
                     run,
                     CORRUPT_PENDING_REASON,
                     now,
-                    None,
+                    self.history,
                 )
                 clear_pending(self.run_store, run.id)
                 clear_executing(self.run_store, run.id)
@@ -487,7 +532,7 @@ class Scheduler:
                     run,
                     CORRUPT_PENDING_REASON,
                     now,
-                    None,
+                    self.history,
                 )
                 clear_pending(self.run_store, run.id)
                 clear_executing(self.run_store, run.id)

src/wf_scheduling/recovery.py

@@ -36,6 +36,11 @@ from __future__ import annotations
 from datetime import datetime, timezone
 from typing import Any
 
+from wf_scheduling.history import (
+    FileScheduleHistoryRecorder,
+    HistoryEntry,
+    HistoryRecorder,
+)
 from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
 
 UTC = timezone.utc
@@ -60,7 +65,7 @@ def recover(
     run_store: Any,
     now: datetime,
     ownership: SchedulerOwnership,
-    record_history: Any | None = None,
+    history: HistoryRecorder | None = None,
 ) -> list[str]:
     """Reconcile durable state after a restart without executing work."""
     from wf_artifacts.runs.models import StoredRunStatus
@@ -70,6 +75,8 @@ def recover(
             "scheduler ownership is required before recovery: an unowned "
             "recovery could abandon or redispatch another owner's work"
         )
+    if history is None:
+        history = FileScheduleHistoryRecorder(schedule_store)
     diags: list[str] = []
     # Admission record is the recovery authority: admitted but never
     # materialized views are completed here and flagged pending for the
@@ -102,40 +109,51 @@ def recover(
             clear_pending(run_store, run.id)
             diags.append(f"{run.id}:completion-window-cleared")
         if status == StoredRunStatus.INTERRUPTED.value or status == "interrupted":
+            checkpoint_id, result_attempt = _latest_result(run_store, run.id)
+            sched_id, intended, revision = _attribution(run_store, run.id)
             if active:
                 assert attempt is not None
-                try:
-                    latest = run_store.get_latest_checkpoint(run.id)
-                    result_attempt = latest.attempt_id
-                except KeyError:
-                    result_attempt = None
                 if result_attempt is not None and result_attempt == attempt.attempt_id:
-                    from wf_artifacts.runs.models import ResumeAttempt
-
-                    run_store.save_resume_attempt(
-                        ResumeAttempt(
-                            run_id=run.id,
-                            attempt_id=attempt.attempt_id,
-                            state="DONE",
-                            created_at=attempt.created_at,
-                            updated_at=now,
-                        )
-                    )
+                    _complete_attempt(run_store, run.id, attempt, now)
                     diags.append(f"{run.id}:fresh-result-resumable")
+                    if _reconcile_terminal(
+                        history,
+                        sched_id,
+                        run.id,
+                        "interrupted",
+                        checkpoint_id,
+                        intended,
+                        revision,
+                        "fresh-result",
+                        now,
+                    ):
+                        diags.append(f"{run.id}:terminal-reconciled")
                 else:
-                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, history)
                     diags.append(f"{run.id}:failed-closed")
             else:
                 diags.append(f"{run.id}:waiting-resumable")
+                if _reconcile_terminal(
+                    history,
+                    sched_id,
+                    run.id,
+                    "interrupted",
+                    checkpoint_id,
+                    intended,
+                    revision,
+                    "reconciled-on-recovery",
+                    now,
+                ):
+                    diags.append(f"{run.id}:terminal-reconciled")
         elif status == StoredRunStatus.ADMITTED.value or status == "admitted":
             if is_executing(run_store, run.id):
                 # The executor may already have produced external effects:
                 # abandon, never retry. Both markers are cleared so a later
                 # recovery does not re-fail the now-terminal run.
                 if active:
-                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, history)
                 else:
-                    _fail_run(run_store, run, ABANDONED_REASON, now, record_history)
+                    _fail_run(run_store, run, ABANDONED_REASON, now, history)
                 clear_pending(run_store, run.id)
                 clear_executing(run_store, run.id)
                 diags.append(f"{run.id}:failed-closed")
@@ -151,82 +169,179 @@ def recover(
                 diags.append(f"{run.id}:corrupt-blocked")
                 continue
             if active:
-                _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+                _fail_run(run_store, run, AMBIGUOUS_REASON, now, history)
             else:
-                _fail_run(run_store, run, ABANDONED_REASON, now, record_history)
+                _fail_run(run_store, run, ABANDONED_REASON, now, history)
             diags.append(f"{run.id}:failed-closed")
         elif status == StoredRunStatus.COMPLETED.value or status == "completed":
+            checkpoint_id, completed_attempt = _latest_result(run_store, run.id)
+            sched_id, intended, revision = _attribution(run_store, run.id)
             if active:
                 assert attempt is not None
-                try:
-                    latest = run_store.get_latest_checkpoint(run.id)
-                    completed_attempt = latest.attempt_id
-                except KeyError:
-                    completed_attempt = None
                 if (
                     completed_attempt is not None
                     and completed_attempt == attempt.attempt_id
                 ):
+                    _complete_attempt(run_store, run.id, attempt, now)
+                    diags.append(f"{run.id}:attempt-reconciled")
+                else:
+                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, history)
+                    diags.append(f"{run.id}:failed-closed")
+                    continue
+            if _reconcile_terminal(
+                history,
+                sched_id,
+                run.id,
+                "completed",
+                checkpoint_id,
+                intended,
+                revision,
+                "reconciled-on-recovery",
+                now,
+            ):
+                diags.append(f"{run.id}:terminal-reconciled")
+        elif status == StoredRunStatus.FAILED.value or status == "failed":
+            sched_id, intended, revision = _attribution(run_store, run.id)
+            if _reconcile_terminal(
+                history,
+                sched_id,
+                run.id,
+                "failed",
+                run.latest_checkpoint_id,
+                intended,
+                revision,
+                "reconciled-on-recovery",
+                now,
+            ):
+                diags.append(f"{run.id}:terminal-reconciled")
+    return diags
+
+
+def _attribution(
+    run_store: Any, run_id: str
+) -> tuple[str | None, datetime | None, int | None]:
+    """Return ``(schedule_id, scheduled_at, schedule_revision)`` for a run.
+
+    ``(None, None, None)`` when no admission owns the run: schedule history
+    cannot attribute it, so callers persist the reason on the run itself.
+    """
+    try:
+        admission = run_store.get_admission(run_id)
+    except KeyError:
+        return None, None, None
+    return (
+        admission.schedule_id,
+        admission.scheduled_at,
+        admission.schedule_revision,
+    )
+
+
+def _latest_result(run_store: Any, run_id: str) -> tuple[str | None, Any]:
+    """Return ``(checkpoint_id, attempt_id)`` of the latest stopped checkpoint."""
+    try:
+        latest = run_store.get_latest_checkpoint(run_id)
+    except KeyError:
+        return None, None
+    return latest.id, latest.attempt_id
+
+
+def _complete_attempt(run_store: Any, run_id: str, attempt: Any, now: datetime) -> None:
+    """Mark the ACTIVE attempt DONE after its fresh result was recognized."""
     from wf_artifacts.runs.models import ResumeAttempt
 
     run_store.save_resume_attempt(
         ResumeAttempt(
-                            run_id=run.id,
+            run_id=run_id,
             attempt_id=attempt.attempt_id,
             state="DONE",
             created_at=attempt.created_at,
             updated_at=now,
         )
     )
-                    diags.append(f"{run.id}:attempt-reconciled")
-                else:
-                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
-                    diags.append(f"{run.id}:failed-closed")
-                    continue
-            if record_history is not None and not _has_terminal(
-                schedule_store, run.id, "completed"
-            ):
-                record_history(
-                    kind="completed",
-                    run_id=run.id,
-                    reason="reconciled-on-recovery",
+
+
+def _reconcile_terminal(
+    history: HistoryRecorder,
+    sched_id: str | None,
+    run_id: str,
+    kind: str,
+    checkpoint_id: str | None,
+    intended: datetime | None,
+    revision: int | None,
+    reason: str,
+    now: datetime,
+) -> bool:
+    """Record one stopped-result entry unless this exact result is known.
+
+    Returns whether an entry was appended. Idempotency identity is
+    ``(run_id, kind, checkpoint_id)``: repeating recovery never duplicates
+    an entry, while a resumed run that stops again (new checkpoint id)
+    records a new one.
+    """
+    if sched_id is None:
+        return False
+    if history.has_terminal(sched_id, run_id, kind, checkpoint_id):
+        return False
+    history.record(
+        HistoryEntry(
+            schedule_id=sched_id,
+            kind=kind,  # type: ignore[arg-type]
+            resolved_at=intended,
+            run_id=run_id,
+            revision=revision,
+            reason=reason,
+            checkpoint_id=checkpoint_id,
+            created_at=now,
         )
-                diags.append(f"{run.id}:terminal-reconciled")
-    return diags
+    )
+    return True
 
 
 def _fail_run(
-    run_store: Any, run: Any, reason: str, now: datetime, record_history: Any | None
+    run_store: Any,
+    run: Any,
+    reason: str,
+    now: datetime,
+    history: HistoryRecorder,
 ) -> None:
+    """Fail a run closed with its truthful reason durably preserved.
+
+    The reason is always written into the run summary diagnostics; when an
+    admission attributes the run to a schedule, a failed occurrence entry is
+    reconciled exactly once through the history recorder.
+    """
+    from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
     from wf_artifacts.runs.models import StoredRunStatus
 
+    diagnostic = DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="schedule-recovery",
+        logical_ref=run.id,
+        message=reason,
+        repair_hint=None,
+    )
     updated = run.model_copy(
-        update={"status": StoredRunStatus.FAILED, "updated_at": now}
+        update={
+            "status": StoredRunStatus.FAILED,
+            "updated_at": now,
+            "diagnostics": [*run.diagnostics, diagnostic],
+        }
     )
     run_store.save_run(updated)
-    if record_history is not None:
-        try:
-            admission = run_store.get_admission(run.id)
-            sched_id = admission.schedule_id or ""
-            intended = admission.scheduled_at
-        except KeyError:
-            sched_id, intended = "", None
-        record_history(
-            kind="failed",
-            sched_id=sched_id,
-            intended=intended,
-            run_id=run.id,
-            reason=reason,
+    sched_id, intended, revision = _attribution(run_store, run.id)
+    _reconcile_terminal(
+        history,
+        sched_id,
+        run.id,
+        "failed",
+        run.latest_checkpoint_id,
+        intended,
+        revision,
+        reason,
+        now,
     )
 
 
-def _has_terminal(schedule_store: Any, run_id: str, kind: str) -> bool:
-    # Scheduler history lives per schedule; without a schedule index, skip
-    # dedup here (poll reconciliation in T08 already guards re-admission via
-    # consumed watermarks). Kept as a seam for T13 inspection.
-    return False
-
-
 def _mark_pending(run_store: Any, run_id: str) -> None:
     run_store.mark_pending_dispatch(run_id)
 

src/wf_scheduling/store.py

@@ -168,6 +168,29 @@ class FileScheduleStore:
             entries.append(record.model_dump(mode="json"))
             self._write_json(path, entries)
 
+    def has_history_entry(
+        self,
+        schedule_id: str,
+        *,
+        run_id: str,
+        kind: str,
+        checkpoint_id: str | None = None,
+    ) -> bool:
+        """Whether this exact stopped result already has a history entry.
+
+        Idempotency identity is ``(run_id, kind, checkpoint_id)``: repeats
+        of one recovery pass dedup, while a resumed run that stops again
+        (new checkpoint id) records a new entry.
+        """
+        for item in self._read_history_locked(schedule_id):
+            if (
+                item.get("run_id") == run_id
+                and item.get("kind") == kind
+                and item.get("checkpoint_id") == checkpoint_id
+            ):
+                return True
+        return False
+
     @staticmethod
     def _sort_key(
         item: OccurrenceRecord,

tests/scheduling/test_history_reconcile.py

@@ -0,0 +1,490 @@
+"""Occurrence-history reconciliation through one typed interface (R4/F4).
+
+Polling and recovery share :class:`HistoryRecorder`: stopped results carry
+``(run_id, kind, checkpoint_id)`` identity, failure reasons persist on the
+run summary even when no schedule attribution exists, and repeating
+recovery never duplicates an entry while a resumed re-interruption records
+anew.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+from tests.artifacts.test_run_store import artifact as _artifact
+from tests.artifacts.test_run_store import deployment as _deployment
+from tests.scheduling.controlled import (
+    DictDeployments,
+    ScriptedDispatcher,
+    fixture_environment,
+)
+from wf_api.run_lifecycle import (
+    materialize_admitted_view,
+    persist_admission,
+    persist_stopped_run,
+)
+from wf_artifacts import PinnedRunEnvironment
+from wf_artifacts.runs.models import ResumeAttempt
+from wf_artifacts.runs.store import FileRunStore
+from wf_core import RunState, RunStatus
+from wf_scheduling import recovery as sched_recovery
+from wf_scheduling.history import FileScheduleHistoryRecorder
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import SchedulePreparer
+from wf_scheduling.store import FileScheduleStore
+
+
+def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime:
+    return datetime(y, mo, d, h, mi, tzinfo=UTC)
+
+
+def _sched_model(sid: str, **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 _env() -> PinnedRunEnvironment:
+    return PinnedRunEnvironment(
+        deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[]
+    )
+
+
+def _admit_owned(
+    run_store: FileRunStore,
+    run_id: str,
+    intended: datetime,
+    *,
+    revision: int = 1,
+) -> Any:
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id="a",
+        schedule_revision=revision,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    return admission
+
+
+def _stop(
+    run_store: FileRunStore,
+    run_id: str,
+    status: RunStatus,
+    *,
+    attempt_id: int | None = None,
+) -> Any:
+    record = run_store.get_run(run_id)
+    return persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(workflow_name="sched", status=status, workflow_input={}, state={}),
+        run_id=run_id,
+        attempt_id=attempt_id,
+    )
+
+
+def _mark_active(
+    run_store: FileRunStore, run_id: str, attempt_id: int, now: datetime
+) -> None:
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=attempt_id,
+            state="ACTIVE",
+            created_at=now,
+            updated_at=now,
+        )
+    )
+
+
+def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, Any]]:
+    page = store.list_occurrences(sid, limit=100)
+    return [
+        r for r in cast(list[dict[str, Any]], page["occurrences"]) if r["kind"] == kind
+    ]
+
+
+def _recover(
+    sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime
+) -> list[str]:
+    ownership = SchedulerOwnership(sched_store.root, owner="test").acquire()
+    try:
+        return sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=now,
+            ownership=ownership,
+        )
+    finally:
+        ownership.release()
+
+
+def test_failed_run_persists_reason_without_callback(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    _admit_owned(run_store, run_id, intended)
+    run_store.mark_executing(run_id)
+    diags = _recover(sched_store, run_store, intended)
+    assert any("failed-closed" in d for d in diags)
+    record = run_store.get_run(run_id)
+    assert record.status.value == "failed"
+    assert any(sched_recovery.ABANDONED_REASON in d.message for d in record.diagnostics)
+    failed = _entries(sched_store, "a", "failed")
+    assert len(failed) == 1
+    assert failed[0]["run_id"] == run_id
+    assert sched_recovery.ABANDONED_REASON in failed[0]["reason"]
+
+
+def test_waiting_interrupted_reconciles_exactly_once(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    _admit_owned(run_store, run_id, intended)
+    stopped = _stop(run_store, run_id, RunStatus.INTERRUPTED)
+    diags = _recover(sched_store, run_store, intended)
+    assert any("waiting-resumable" in d for d in diags)
+    assert any("terminal-reconciled" in d for d in diags)
+    first = _entries(sched_store, "a", "interrupted")
+    assert len(first) == 1
+    assert first[0]["checkpoint_id"] == stopped.latest_checkpoint_id
+    diags = _recover(sched_store, run_store, intended + timedelta(minutes=1))
+    assert _entries(sched_store, "a", "interrupted") == first
+
+
+def test_fresh_interrupted_completes_attempt_and_reconciles(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    _admit_owned(run_store, run_id, intended)
+    attempt_id = run_store.allocate_resume_attempt_id()
+    _mark_active(run_store, run_id, attempt_id, intended)
+    _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id)
+    diags = _recover(sched_store, run_store, intended)
+    assert any("fresh-result-resumable" in d for d in diags)
+    assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+    assert len(_entries(sched_store, "a", "interrupted")) == 1
+    _recover(sched_store, run_store, intended + timedelta(minutes=1))
+    assert len(_entries(sched_store, "a", "interrupted")) == 1
+
+
+def test_completed_reconciles_exactly_once(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    _admit_owned(run_store, run_id, intended)
+    stopped = _stop(run_store, run_id, RunStatus.COMPLETED)
+    diags = _recover(sched_store, run_store, intended)
+    assert any("terminal-reconciled" in d for d in diags)
+    first = _entries(sched_store, "a", "completed")
+    assert len(first) == 1
+    assert first[0]["checkpoint_id"] == stopped.latest_checkpoint_id
+    assert first[0]["run_id"] == run_id
+    _recover(sched_store, run_store, intended + timedelta(minutes=1))
+    assert _entries(sched_store, "a", "completed") == first
+
+
+def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> None:
+    class FailCompletedHistoryOnce(FileScheduleStore):
+        def __init__(self, root: Path) -> None:
+            super().__init__(root)
+            self.armed = True
+
+        def append_history(self, record: Any) -> None:
+            if self.armed and getattr(record, "kind", None) == "completed":
+                self.armed = False
+                raise OSError("injected history failure")
+            super().append_history(record)
+
+    sched_store = FailCompletedHistoryOnce(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=run_store,
+            sources={},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+                fixture_environment,
+            ),
+            dispatcher=ScriptedDispatcher({"*": "complete"}),
+            ownership=ownership,
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        sched_store.create_schedule(_sched_model("a"))
+        sched_store.save_consumed("a", intended - timedelta(hours=1))
+        from wf_scheduling.calendar import OneShotSource
+
+        sched.sources["a"] = OneShotSource(intended)
+        try:
+            sched.poll(intended)
+            raise AssertionError("history failure must propagate")
+        except OSError as exc:
+            assert "injected history failure" in str(exc)
+        run_id = run_store.list_runs()[0].id
+        assert run_store.get_run(run_id).status.value == "completed"
+        assert _entries(sched_store, "a", "completed") == []
+        diags = sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=intended,
+            ownership=ownership,
+        )
+        assert any("terminal-reconciled" in d for d in diags)
+        assert len(_entries(sched_store, "a", "completed")) == 1
+        sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=intended + timedelta(minutes=1),
+            ownership=ownership,
+        )
+        assert len(_entries(sched_store, "a", "completed")) == 1
+    finally:
+        ownership.release()
+
+
+def test_dispatch_and_recovery_share_entry_identity(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched_store = FileScheduleStore(tmp_path / "sched")
+        run_store = FileRunStore(tmp_path / "runs")
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=run_store,
+            sources={},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+                fixture_environment,
+            ),
+            dispatcher=ScriptedDispatcher({"*": "complete"}),
+            ownership=ownership,
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        sched_store.create_schedule(_sched_model("a"))
+        sched_store.save_consumed("a", intended - timedelta(hours=1))
+        from wf_scheduling.calendar import OneShotSource
+
+        sched.sources["a"] = OneShotSource(intended)
+        sched.poll(intended)
+        run_id = run_store.list_runs()[0].id
+        dispatched = _entries(sched_store, "a", "completed")
+        assert len(dispatched) == 1
+        checkpoint_id = run_store.get_run(run_id).latest_checkpoint_id
+        assert dispatched[0]["checkpoint_id"] == checkpoint_id
+        recorder = FileScheduleHistoryRecorder(sched_store)
+        assert recorder.has_terminal("a", run_id, "completed", checkpoint_id)
+        # Recovery on fresh objects recognizes the entry: no duplicate.
+        diags = sched_recovery.recover(
+            schedule_store=FileScheduleStore(tmp_path / "sched"),
+            run_store=FileRunStore(tmp_path / "runs"),
+            now=intended,
+            ownership=ownership,
+        )
+        assert not any("terminal-reconciled" in d for d in diags)
+        assert len(_entries(sched_store, "a", "completed")) == 1
+    finally:
+        ownership.release()
+
+
+def test_resumed_reinterruption_records_anew(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    _admit_owned(run_store, run_id, intended)
+    first = run_store.allocate_resume_attempt_id()
+    _mark_active(run_store, run_id, first, intended)
+    _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=first)
+    _recover(sched_store, run_store, intended)
+    assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+    # The resumed run interrupts again under a new attempt: a new checkpoint
+    # id means a new history entry, not a dedup hit.
+    second = run_store.allocate_resume_attempt_id()
+    assert second != first
+    _mark_active(run_store, run_id, second, intended)
+    _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=second)
+    _recover(sched_store, run_store, intended + timedelta(minutes=5))
+    entries = _entries(sched_store, "a", "interrupted")
+    assert len(entries) == 2
+    assert {e["checkpoint_id"] for e in entries} == {
+        f"{run_id}.000001",
+        f"{run_id}.000002",
+    }
+
+
+def _owned_scheduler(
+    sched_store: FileScheduleStore,
+    run_store: FileRunStore,
+    ownership: SchedulerOwnership,
+    *,
+    script: dict | None = None,
+) -> Scheduler:
+    return Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={},
+        capacity=4,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+            fixture_environment,
+        ),
+        dispatcher=ScriptedDispatcher(script),
+        ownership=ownership,
+    )
+
+
+def test_admission_tear_returns_existing_run_without_dup(tmp_path: Path) -> None:
+    from wf_scheduling.calendar import OneShotSource
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    # Crash shape: admission + view persisted, history/consumed writes lost.
+    run_id = run_store.allocate_run_id()
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id="a",
+        schedule_revision=1,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    sched_store.save_consumed("a", intended - timedelta(hours=1))
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = _owned_scheduler(
+            sched_store, run_store, ownership, script={"*": "hang"}
+        )
+        sched.sources["a"] = OneShotSource(intended)
+        assert sched.poll(intended) == {"a": f"admit:{run_id}"}
+        assert [r.id for r in run_store.list_runs()] == [run_id]
+        assert [a.id for a in run_store.list_admissions()] == [run_id]
+        assert sched_store.get_consumed("a") == intended
+        admitted = _entries(sched_store, "a", "admitted")
+        assert len(admitted) == 1
+        assert admitted[0]["run_id"] == run_id
+        # Re-polling never duplicates the occurrence or its history entry.
+        sched.poll(intended)
+        assert [r.id for r in run_store.list_runs()] == [run_id]
+        assert len(_entries(sched_store, "a", "admitted")) == 1
+    finally:
+        ownership.release()
+
+
+def test_parallel_tear_never_double_admits(tmp_path: Path) -> None:
+    from wf_scheduling.calendar import OneShotSource
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(
+        _sched_model("a", overlap="parallel", max_active_runs=4)
+    )
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = run_store.allocate_run_id()
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id="a",
+        schedule_revision=1,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    sched_store.save_consumed("a", intended - timedelta(hours=1))
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = _owned_scheduler(
+            sched_store, run_store, ownership, script={"*": "hang"}
+        )
+        sched.sources["a"] = OneShotSource(intended)
+        sched.poll(intended)
+        sched.poll(intended)
+        assert [r.id for r in run_store.list_runs()] == [run_id]
+        assert [a.id for a in run_store.list_admissions()] == [run_id]
+    finally:
+        ownership.release()
+
+
+def test_consumed_write_failure_recovers_without_dup(tmp_path: Path) -> None:
+    from wf_scheduling.calendar import OneShotSource
+
+    class FailConsumedOnce(FileScheduleStore):
+        def __init__(self, root: Path) -> None:
+            super().__init__(root)
+            self.armed = False
+
+        def save_consumed(self, schedule_id: str, consumed_through: Any) -> None:
+            if self.armed:
+                self.armed = False
+                raise OSError("injected consumed failure")
+            super().save_consumed(schedule_id, consumed_through)
+
+    sched_store = FailConsumedOnce(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.save_consumed("a", intended - timedelta(hours=1))
+    sched_store.armed = True  # arm only the poll's watermark write
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = _owned_scheduler(
+            sched_store, run_store, ownership, script={"*": "hang"}
+        )
+        sched.sources["a"] = OneShotSource(intended)
+        with __import__("pytest").raises(OSError, match="injected consumed"):
+            sched.poll(intended)
+        assert [a.id for a in run_store.list_admissions()] != []
+        first = run_store.list_admissions()[0].id
+        # Retry: the one-shot is exhausted so no new admission is possible;
+        # the failed watermark write cannot duplicate the occurrence.
+        assert sched.poll(intended) == {"a": "exhausted"}
+        assert [a.id for a in run_store.list_admissions()] == [first]
+        # Recovery completes the orphaned admission; the sweep dispatches it
+        # exactly once with no duplicate run.
+        diags = sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=intended,
+            ownership=ownership,
+        )
+        assert any("pending-dispatch" in d for d in diags)
+        sched.poll(intended)
+        assert [r.id for r in run_store.list_runs()] == [first]
+        assert run_store.is_executing(first)
+    finally:
+        ownership.release()

tests/scheduling/test_preparation.py

@@ -83,6 +83,7 @@ def test_scheduler_requires_typed_collaborators() -> None:
         "preparer",
         "dispatcher",
         "ownership",
+        "history",
     }