"""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, timezone from pathlib import Path from typing import Any, cast import pytest 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, HistoryEntry, entry_occurrence_id, ) 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.parent, 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_interval_identity_normalizes_offsets_and_deduplicates( tmp_path: Path, ) -> None: store = FileScheduleStore(tmp_path / "sched") recorder = FileScheduleHistoryRecorder(store) utc_start = datetime(2026, 9, 8, 12, 0, tzinfo=UTC) utc_end = datetime(2026, 9, 8, 13, 0, tzinfo=UTC) offset = timezone(timedelta(hours=2)) recorder.record( HistoryEntry( schedule_id="a", kind="interval-summary", interval_start=utc_start, interval_end=utc_end, interval_count=2, created_at=utc_start, ) ) recorder.record( HistoryEntry( schedule_id="a", kind="interval-summary", interval_start=utc_start.astimezone(offset), interval_end=utc_end.astimezone(offset), interval_count=2, created_at=utc_end, ) ) rows = cast( list[dict[str, Any]], store.list_occurrences("a", limit=10)["occurrences"] ) assert len(rows) == 1 row = rows[0] assert row["occurrence_id"] == ( "a|summary|2026-09-08T12:00:00+00:00|2026-09-08T13:00:00+00:00" ) assert row["interval_start"] == "2026-09-08T12:00:00Z" assert row["interval_end"] == "2026-09-08T13:00:00Z" @pytest.mark.parametrize( ("interval_start", "interval_end"), [ (datetime(2026, 9, 8, 12, 0), datetime(2026, 9, 8, 13, 0, tzinfo=UTC)), (datetime(2026, 9, 8, 12, 0, tzinfo=UTC), datetime(2026, 9, 8, 13, 0)), (None, datetime(2026, 9, 8, 13, 0, tzinfo=UTC)), (datetime(2026, 9, 8, 12, 0, tzinfo=UTC), None), ], ) def test_interval_identity_rejects_naive_or_half_specified_bounds( interval_start: datetime | None, interval_end: datetime | None, ) -> None: with pytest.raises(ValueError): entry_occurrence_id( HistoryEntry( schedule_id="a", kind="interval-summary", interval_start=interval_start, interval_end=interval_end, ), datetime(2026, 9, 8, 14, 0, tzinfo=UTC), ) def test_history_identity_without_interval_still_uses_created_time() -> None: created = datetime(2026, 9, 8, 14, 0, tzinfo=UTC) assert ( entry_occurrence_id( HistoryEntry(schedule_id="a", kind="interval-summary"), created ) == "a|summary|2026-09-08T14:00:00+00:00" ) 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, 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, 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, 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, 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, owner="test").acquire() try: sched = _owned_scheduler( sched_store, run_store, ownership, script={"*": "hang"} ) sched.sources["a"] = OneShotSource(intended) with 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()