diff --git a/src/wf_scheduling/history.py b/src/wf_scheduling/history.py index 9b374662..1209bdf8 100644 --- a/src/wf_scheduling/history.py +++ b/src/wf_scheduling/history.py @@ -16,11 +16,15 @@ from datetime import datetime, timezone from typing import Any, Protocol from wf_scheduling.models import OccurrenceKind, OccurrenceRecord -from wf_scheduling.occurrences import occurrence_id +from wf_scheduling.occurrences import _require_aware, occurrence_id UTC = timezone.utc -TERMINAL_KINDS: tuple[str, str, str] = ("completed", "interrupted", "failed") +TERMINAL_KINDS: tuple[OccurrenceKind, ...] = ( + "completed", + "interrupted", + "failed", +) @dataclass(frozen=True) @@ -53,7 +57,7 @@ class HistoryRecorder(Protocol): self, schedule_id: str, run_id: str, - kind: str, + kind: OccurrenceKind, checkpoint_id: str | None = None, ) -> bool: """Whether this exact stopped result was already reconciled.""" @@ -62,15 +66,32 @@ class HistoryRecorder(Protocol): def entry_occurrence_id(entry: HistoryEntry, created: datetime) -> str: """Derive the history occurrence id with the scheduler's standing rules.""" + interval = _normalized_interval(entry) 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}" + if interval is not None: + start, end = interval + return f"{entry.schedule_id}|summary|{start.isoformat()}|{end.isoformat()}" return f"{entry.schedule_id}|summary|{created.isoformat()}" +def _normalized_interval( + entry: HistoryEntry, +) -> tuple[datetime, datetime] | None: + """Validate and normalize an optional summary interval to UTC.""" + start, end = entry.interval_start, entry.interval_end + if (start is None) != (end is None): + raise ValueError("interval_start and interval_end must be provided together") + if start is None: + assert end is None + return None + assert end is not None + return ( + _require_aware(start, op="interval_start").astimezone(UTC), + _require_aware(end, op="interval_end").astimezone(UTC), + ) + + class FileScheduleHistoryRecorder: """History recorder backed by a file schedule store.""" @@ -81,6 +102,7 @@ class FileScheduleHistoryRecorder: created = ( entry.created_at if entry.created_at is not None else datetime.now(UTC) ) + interval = _normalized_interval(entry) self.schedule_store.append_history( OccurrenceRecord( schedule_id=entry.schedule_id, @@ -93,8 +115,8 @@ class FileScheduleHistoryRecorder: 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_start=interval[0] if interval is not None else None, + interval_end=interval[1] if interval is not None else None, interval_count=entry.interval_count, created_at=created, ) @@ -104,7 +126,7 @@ class FileScheduleHistoryRecorder: self, schedule_id: str, run_id: str, - kind: str, + kind: OccurrenceKind, checkpoint_id: str | None = None, ) -> bool: return self.schedule_store.has_history_entry( diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index bdb5d900..4b62974b 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -26,7 +26,13 @@ from pathlib import Path from threading import RLock from typing import Any, cast -from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id +from .models import ( + OccurrenceKind, + OccurrenceRecord, + PendingCandidate, + Schedule, + ensure_schedule_id, +) UTC = timezone.utc @@ -48,16 +54,18 @@ def schedule_store_transaction(store: object) -> Iterator[None]: def _history_identity(record: OccurrenceRecord) -> tuple[object, ...]: - """Return fields that identify one durable occurrence transition.""" + """Return the canonical fields that identify one history transition. + + ``occurrence_id`` already contains the UTC-normalized resolved instant or + summary interval. Do not compare the original datetime fields as well: + equivalent offsets must remain one idempotent history entry. + """ return ( record.schedule_id, record.occurrence_id, record.kind, - record.resolved_at, record.run_id, record.checkpoint_id, - record.interval_start, - record.interval_end, ) @@ -237,7 +245,7 @@ class FileScheduleStore: schedule_id: str, *, run_id: str, - kind: str, + kind: OccurrenceKind, checkpoint_id: str | None = None, ) -> bool: """Whether this exact stopped result already has a history entry.