sched: add schedule models, file store, and deployment revisions (T07)

src/wf_artifacts/models.py

@@ -222,6 +222,14 @@ class WorkflowDeployment(BaseModel):
         default_factory=list
     )
     drift_policy: DriftPolicy = DriftPolicy.BLOCK
+    revision: int = Field(
+        default=1,
+        ge=1,
+        description=(
+            "Mutable deployment revision for edit-race rechecks. "
+            "Incremented on every save; admission freezes the revision it saw."
+        ),
+    )
 
     def binding_map(self) -> dict[str, str]:
         """Return bindings keyed by dot-joined logical source ref."""

src/wf_artifacts/store.py

@@ -112,6 +112,15 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore):
 
     def save_deployment(self, deployment: WorkflowDeployment) -> None:
         path = self._deployment_path(deployment.id)
+        if path.exists():
+            existing = WorkflowDeployment.model_validate_json(
+                path.read_text(encoding="utf-8")
+            )
+            deployment = deployment.model_copy(
+                update={"revision": existing.revision + 1}
+            )
+        elif deployment.revision != 1:
+            deployment = deployment.model_copy(update={"revision": 1})
         path.write_text(
             json.dumps(deployment.model_dump(mode="json"), indent=2),
             encoding="utf-8",

src/wf_scheduling/models.py

@@ -0,0 +1,136 @@
+"""Schedule, candidate, and occurrence-history models.
+
+A schedule follows its deployment: admission freezes the deployment revision,
+resolved input, limits, and pinned artifact tree, so later edits cannot change
+that occurrence or its run. Overlap is per schedule; manual runs and other
+schedules never participate in the check.
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from wf_core.models.input_bindings import ScheduleInputBinding
+
+SCHEDULE_ID_PATTERN = r"^[A-Za-z0-9_][A-Za-z0-9_.-]*$"
+
+OverlapPolicy = Literal["skip", "parallel"]
+MisfirePolicy = Literal["skip", "latest"]
+
+OccurrenceKind = Literal[
+    "pending",
+    "coalesced",
+    "superseded",
+    "skipped-overlap",
+    "skipped-misfire",
+    "preflight-rejected",
+    "admitted",
+    "running",
+    "interrupted",
+    "completed",
+    "failed",
+    "exhausted",
+    "interval-summary",
+]
+
+
+def ensure_schedule_id(schedule_id: str) -> str:
+    """Reject ids that cannot safely identify one schedule directory."""
+    if not re.fullmatch(SCHEDULE_ID_PATTERN, schedule_id):
+        raise ValueError(
+            "schedule_id must start with alphanumeric or underscore and contain "
+            "only [A-Za-z0-9_.-]"
+        )
+    return schedule_id
+
+
+class CronTrigger(BaseModel):
+    """Recurring five-field Unix-cron trigger in one explicit IANA zone."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["cron"] = "cron"
+    expression: str
+    timezone: str = "UTC"
+
+
+class OneShotTrigger(BaseModel):
+    """Single offset-aware instant trigger."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["oneshot"] = "oneshot"
+    at: datetime
+
+    @field_validator("at")
+    @classmethod
+    def _require_offset(cls, value: datetime) -> datetime:
+        if value.tzinfo is None or value.utcoffset() is None:
+            raise ValueError("one-shot timestamp must include an offset")
+        return value
+
+
+class Schedule(BaseModel):
+    """One deployment schedule with its revision and lifecycle flags."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str = Field(pattern=SCHEDULE_ID_PATTERN)
+    deployment_id: str
+    trigger: CronTrigger | OneShotTrigger = Field(discriminator="kind")
+    input_bindings: list[ScheduleInputBinding] = Field(default_factory=list)
+    max_steps: int | None = Field(default=None, ge=1)
+    overlap: OverlapPolicy = "skip"
+    misfire: MisfirePolicy = "skip"
+    max_active_runs: int = Field(default=1, ge=1)
+    lateness_allowance_s: float = Field(default=60.0, ge=0.0)
+    revision: int = Field(default=1, ge=1)
+    enabled: bool = True
+    paused: bool = False
+    deleted: bool = False
+    exhausted: bool = False
+    blocked_reason: str | None = None
+    created_at: datetime
+    updated_at: datetime
+
+    @field_validator("lateness_allowance_s")
+    @classmethod
+    def _require_finite_allowance(cls, value: float) -> float:
+        if not math.isfinite(value):
+            raise ValueError("lateness allowance must be a finite duration")
+        return value
+
+
+class PendingCandidate(BaseModel):
+    """At most one unadmitted latest candidate per schedule."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schedule_id: str = Field(pattern=SCHEDULE_ID_PATTERN)
+    intended_at: datetime
+    revision: int = Field(ge=1)
+
+
+class OccurrenceRecord(BaseModel):
+    """One inspectable occurrence-history entry for a schedule."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schedule_id: str = Field(pattern=SCHEDULE_ID_PATTERN)
+    occurrence_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
+    interval_start: datetime | None = None
+    interval_end: datetime | None = None
+    interval_count: int = 0
+    created_at: datetime

src/wf_scheduling/store.py

@@ -0,0 +1,223 @@
+"""File-backed schedule store for local development and tests.
+
+Layout per schedule ``<root>/schedules/<id>/``:
+
+- ``schedule.json`` — the schedule definition (deleted schedules stay with
+  ``deleted=True``; ids are never reused to annex old history).
+- ``candidate.json`` — at most one pending latest candidate.
+- ``consumed.json`` — ``{"consumed_through": <utc iso>}`` watermark plus the
+  schedule revision it belongs to.
+- ``history.json`` — occurrence-history list for cursor pagination over
+  ``(resolved_at, occurrence_id)``.
+
+Per-file writes are atomic (tmp + rename); the lock is process-local only.
+Scheduler ownership across processes arrives in T11.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from threading import RLock
+
+from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id
+
+UTC = timezone.utc
+
+
+class ScheduleExistsError(ValueError):
+    """A schedule id is already taken (deleted ids are never reused)."""
+
+
+class ScheduleNotFoundError(KeyError):
+    """No schedule exists for the requested id."""
+
+
+class StaleScheduleRevisionError(ValueError):
+    """A schedule edit carried an outdated revision."""
+
+
+class FileScheduleStore:
+    """JSON file-backed schedule definitions, candidates, and history."""
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self._lock = RLock()
+        self.schedules_dir.mkdir(parents=True, exist_ok=True)
+
+    @property
+    def schedules_dir(self) -> Path:
+        return self.root / "schedules"
+
+    # -- schedules ------------------------------------------------------
+    def create_schedule(self, schedule: Schedule) -> Schedule:
+        """Persist a new schedule; deleted ids are never reusable."""
+        with self._lock:
+            path = self._schedule_path(schedule.id)
+            if path.exists():
+                raise ScheduleExistsError(
+                    f"schedule id already exists: {schedule.id!r}"
+                )
+            self._write_json(path, schedule.model_dump(mode="json"))
+            return schedule
+
+    def get_schedule(self, schedule_id: str) -> Schedule:
+        path = self._schedule_path(schedule_id)
+        if not path.exists():
+            raise ScheduleNotFoundError(f"unknown schedule {schedule_id!r}")
+        return Schedule.model_validate_json(path.read_text(encoding="utf-8"))
+
+    def list_schedules(self, *, include_deleted: bool = False) -> list[Schedule]:
+        schedules = [
+            Schedule.model_validate_json(path.read_text(encoding="utf-8"))
+            for path in sorted(self.schedules_dir.glob("*/schedule.json"))
+        ]
+        if not include_deleted:
+            schedules = [item for item in schedules if not item.deleted]
+        return sorted(schedules, key=lambda item: item.id)
+
+    def update_schedule(
+        self, schedule: Schedule, *, expected_revision: int
+    ) -> Schedule:
+        """Replace a schedule after a revision check (stale edits rejected)."""
+        with self._lock:
+            current = self.get_schedule(schedule.id)
+            if current.revision != expected_revision:
+                raise StaleScheduleRevisionError(
+                    f"stale schedule revision for {schedule.id!r}: "
+                    f"expected {expected_revision}, found {current.revision}"
+                )
+            self._write_json(
+                self._schedule_path(schedule.id), schedule.model_dump(mode="json")
+            )
+            return schedule
+
+    def save_schedule(self, schedule: Schedule) -> None:
+        """Persist a schedule without a revision check (ownership internals)."""
+        with self._lock:
+            self._write_json(
+                self._schedule_path(schedule.id), schedule.model_dump(mode="json")
+            )
+
+    # -- candidates (at most one) ----------------------------------------
+    def get_candidate(self, schedule_id: str) -> PendingCandidate | None:
+        path = self._candidate_path(schedule_id)
+        if not path.exists():
+            return None
+        return PendingCandidate.model_validate_json(path.read_text(encoding="utf-8"))
+
+    def save_candidate(
+        self, candidate: PendingCandidate | None, *, schedule_id: str
+    ) -> None:
+        """Persist or clear the single pending candidate for a schedule."""
+        with self._lock:
+            path = self._candidate_path(schedule_id)
+            if candidate is None:
+                if path.exists():
+                    path.unlink()
+                return
+            self._write_json(path, candidate.model_dump(mode="json"))
+
+    # -- consumed watermark ----------------------------------------------
+    def get_consumed(self, schedule_id: str) -> datetime | None:
+        path = self._consumed_path(schedule_id)
+        if not path.exists():
+            return None
+        raw = json.loads(path.read_text(encoding="utf-8"))
+        value = raw.get("consumed_through")
+        if value is None:
+            return None
+        return datetime.fromisoformat(value)
+
+    def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None:
+        with self._lock:
+            self._write_json(
+                self._consumed_path(schedule_id),
+                {"consumed_through": consumed_through.astimezone(UTC).isoformat()},
+            )
+
+    # -- history with cursor pagination -----------------------------------
+    def append_history(self, record: OccurrenceRecord) -> None:
+        """Append one occurrence-history entry for a schedule."""
+        with self._lock:
+            path = self._history_path(record.schedule_id)
+            entries = self._read_history_locked(record.schedule_id)
+            entries.append(record.model_dump(mode="json"))
+            self._write_json(path, entries)
+
+    def list_occurrences(
+        self,
+        schedule_id: str,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, object]:
+        """Return one history page ordered by ``(resolved_at, occurrence_id)``."""
+        if limit < 1 or limit > 100:
+            raise ValueError("limit must be between 1 and 100")
+        records = [
+            OccurrenceRecord.model_validate(item)
+            for item in self._read_history_locked(schedule_id)
+        ]
+        records.sort(
+            key=lambda item: (
+                item.resolved_at is None,
+                item.resolved_at or datetime.max.replace(tzinfo=UTC),
+                item.occurrence_id,
+            )
+        )
+        start = 0
+        if cursor is not None:
+            try:
+                start = int(cursor)
+            except ValueError as exc:
+                raise ValueError(
+                    "cursor must be a non-negative integer offset"
+                ) from exc
+            if start < 0:
+                raise ValueError("cursor must be a non-negative integer offset")
+        total = len(records)
+        page = records[start : start + limit]
+        end = start + limit
+        return {
+            "occurrences": [item.model_dump(mode="json") for item in page],
+            "total": total,
+            "cursor": cursor,
+            "next_cursor": str(end) if end < total else None,
+            "limit": limit,
+        }
+
+    def _read_history_locked(self, schedule_id: str) -> list[dict]:
+        path = self._history_path(schedule_id)
+        if not path.exists():
+            return []
+        raw = json.loads(path.read_text(encoding="utf-8"))
+        return raw if isinstance(raw, list) else []
+
+    # -- paths ------------------------------------------------------------
+    def _schedule_dir(self, schedule_id: str) -> Path:
+        safe_id = ensure_schedule_id(schedule_id)
+        root = self.schedules_dir.resolve()
+        path = (self.schedules_dir / safe_id).resolve()
+        if path.parent != root:
+            raise ValueError(f"schedule id escapes schedule store: {schedule_id!r}")
+        return path
+
+    def _schedule_path(self, schedule_id: str) -> Path:
+        return self._schedule_dir(schedule_id) / "schedule.json"
+
+    def _candidate_path(self, schedule_id: str) -> Path:
+        return self._schedule_dir(schedule_id) / "candidate.json"
+
+    def _consumed_path(self, schedule_id: str) -> Path:
+        return self._schedule_dir(schedule_id) / "consumed.json"
+
+    def _history_path(self, schedule_id: str) -> Path:
+        return self._schedule_dir(schedule_id) / "history.json"
+
+    def _write_json(self, path: Path, payload: object) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temp_path = path.with_name(path.name + ".tmp")
+        temp_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        temp_path.replace(path)

tests/scheduling/test_schedule_store.py

@@ -0,0 +1,218 @@
+"""Schedule models and file store: revisions, lifecycle, history (T07)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from wf_scheduling.models import (
+    CronTrigger,
+    OccurrenceRecord,
+    PendingCandidate,
+    Schedule,
+)
+from wf_scheduling.store import (
+    FileScheduleStore,
+    ScheduleExistsError,
+    StaleScheduleRevisionError,
+)
+
+
+def _schedule(sid: str = "sched-1", **over: object) -> Schedule:
+    now = datetime(2026, 9, 8, 12, 0, tzinfo=UTC)
+    base: dict[str, object] = {
+        "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(over)
+    return Schedule.model_validate(base)
+
+
+def test_schedule_defaults_match_settled_policies() -> None:
+    sched = _schedule()
+    assert sched.overlap == "skip"
+    assert sched.misfire == "skip"
+    assert sched.lateness_allowance_s == 60.0
+    assert sched.revision == 1
+    assert not sched.paused and not sched.deleted and not sched.exhausted
+    assert sched.enabled
+
+
+def test_cron_trigger_and_lateness_validation() -> None:
+    assert (
+        CronTrigger.model_validate(
+            {"kind": "cron", "expression": "* * * * *", "timezone": "UTC"}
+        ).timezone
+        == "UTC"
+    )
+    with pytest.raises(ValidationError):
+        _schedule(lateness_allowance_s=-1.0)
+    with pytest.raises(ValidationError):
+        _schedule(lateness_allowance_s=float("inf"))
+    with pytest.raises(ValidationError):
+        Schedule.model_validate(
+            {
+                "id": "s",
+                "deployment_id": "d",
+                "trigger": {"kind": "oneshot", "at": "2026-09-08T12:00:00"},
+                "input_bindings": [],
+                "created_at": "2026-09-08T12:00:00+00:00",
+                "updated_at": "2026-09-08T12:00:00+00:00",
+            }
+        )
+
+
+def test_parallel_requires_positive_max_active() -> None:
+    with pytest.raises(ValidationError):
+        _schedule(overlap="parallel", max_active_runs=0)
+
+
+def test_create_get_list_and_id_never_reused(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    store.create_schedule(_schedule("b"))
+    assert [s.id for s in store.list_schedules()] == ["a", "b"]
+    with pytest.raises(ScheduleExistsError):
+        store.create_schedule(_schedule("a"))
+    # Delete clears future admission but keeps history; id stays taken.
+    doomed = store.get_schedule("a")
+    doomed.deleted = True
+    store.save_schedule(doomed)
+    assert [s.id for s in store.list_schedules()] == ["b"]
+    assert [s.id for s in store.list_schedules(include_deleted=True)] == ["a", "b"]
+    with pytest.raises(ScheduleExistsError):
+        store.create_schedule(_schedule("a"))
+
+
+def test_stale_revision_edits_rejected(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    current = store.get_schedule("a")
+    updated = current.model_copy(update={"revision": 2, "paused": True})
+    store.update_schedule(updated, expected_revision=1)
+    assert store.get_schedule("a").revision == 2
+    with pytest.raises(StaleScheduleRevisionError):
+        store.update_schedule(updated, expected_revision=1)
+
+
+def test_pause_and_disable_exclude_interval_no_backfill(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    now = datetime(2026, 9, 8, 12, 30, tzinfo=UTC)
+    sched = store.get_schedule("a")
+    # Pause clears the pending candidate and advances the watermark; resume
+    # starts from the next future instant (poll semantics tested in T08).
+    store.save_candidate(
+        PendingCandidate(
+            schedule_id="a",
+            intended_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC),
+            revision=sched.revision,
+        ),
+        schedule_id="a",
+    )
+    paused = sched.model_copy(update={"paused": True, "updated_at": now})
+    store.save_schedule(paused)
+    store.save_candidate(None, schedule_id="a")
+    store.save_consumed("a", now)
+    assert store.get_candidate("a") is None
+    assert store.get_consumed("a") == now
+
+
+def test_candidate_is_at_most_one(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    first = PendingCandidate(
+        schedule_id="a",
+        intended_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC),
+        revision=1,
+    )
+    second = PendingCandidate(
+        schedule_id="a",
+        intended_at=datetime(2026, 9, 8, 13, 0, tzinfo=UTC),
+        revision=1,
+    )
+    store.save_candidate(first, schedule_id="a")
+    store.save_candidate(second, schedule_id="a")
+    assert store.get_candidate("a") is not None
+    assert store.get_candidate("a") is not None
+    assert store.get_candidate("a").intended_at == datetime(  # type: ignore[union-attr]
+        2026, 9, 8, 13, 0, tzinfo=UTC
+    )
+
+
+def test_history_pagination_over_resolved_utc(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    for hour in (12, 13, 14):
+        store.append_history(
+            OccurrenceRecord(
+                schedule_id="a",
+                occurrence_id=f"a|2026-09-08T{hour:02d}:00:00+00:00",
+                kind="admitted",
+                resolved_at=datetime(2026, 9, 8, hour, 0, tzinfo=UTC),
+                run_id=f"run-{hour}",
+                revision=1,
+                created_at=datetime(2026, 9, 8, hour, 0, tzinfo=UTC),
+            )
+        )
+    page = store.list_occurrences("a", limit=2)
+    assert page["total"] == 3
+    assert page["next_cursor"] == "2"
+    assert len(page["occurrences"]) == 2  # type: ignore[arg-type]
+    second = store.list_occurrences("a", cursor="2", limit=2)
+    assert second["next_cursor"] is None
+    assert len(second["occurrences"]) == 1  # type: ignore[arg-type]
+
+
+def test_inspection_payload_carries_contract_fields(tmp_path: Path) -> None:
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    store.append_history(
+        OccurrenceRecord(
+            schedule_id="a",
+            occurrence_id="a|2026-09-08T12:00:00+00:00",
+            kind="skipped-overlap",
+            resolved_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC),
+            revision=2,
+            reason="active=['run-1']",
+            admitted_at=None,
+            started_at=None,
+            created_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC),
+        )
+    )
+    page = store.list_occurrences("a", limit=10)
+    row = page["occurrences"][0]  # type: ignore[index]
+    assert row["resolved_at"] is not None
+    assert row["revision"] == 2
+    assert row["reason"] == "active=['run-1']"
+
+
+def test_deployment_revision_increments_on_save(tmp_path: Path) -> None:
+    from wf_artifacts import FileWorkflowArtifactStore, WorkflowDeployment
+
+    artifacts = FileWorkflowArtifactStore(tmp_path)
+    artifacts.save_deployment(
+        WorkflowDeployment(
+            id="dep-1",
+            artifact_id="wf-1",
+            artifact_version=1,
+            bindings=[],
+        )
+    )
+    assert artifacts.get_deployment("dep-1").revision == 1
+    artifacts.save_deployment(
+        WorkflowDeployment(
+            id="dep-1",
+            artifact_id="wf-1",
+            artifact_version=2,
+            bindings=[],
+        )
+    )
+    assert artifacts.get_deployment("dep-1").revision == 2