"""T13 schedule administration API tests (wf_api layer only). Uses real Schedule models plus real FileScheduleStore/FileRunStore/ FileWorkflowArtifactStore in tmp_path, constructing WorkflowScheduleApi directly over a minimal WorkflowOperationContext (same pattern as tests/wf_api/test_run_lifecycle.py). """ from __future__ import annotations import asyncio import threading from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any import pytest from tests.scheduling.controlled import ( DictDeployments, ScriptedDispatcher, ) from wf_api import WorkflowApi from wf_api.durable_context import durable_workflow_api from wf_api.models import TraceRange from wf_api.operation_context import WorkflowOperationContext from wf_api.run_lifecycle import ( create_pinned_environment, materialize_admitted_view, persist_admission, persist_stopped_run, ) from wf_api.runs import WorkflowRunApi from wf_api.saved_subgraphs import SavedSubgraphTree from wf_api.schedules import WorkflowScheduleApi from wf_artifacts import ( FileRunStore, FileWorkflowArtifactStore, WorkflowArtifact, WorkflowDeployment, ) from wf_authoring import NodeSpec from wf_core import InterruptRequest, RunState, RunStatus from wf_platform import CapabilitySource from wf_scheduling.calendar import OneShotSource from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry from wf_scheduling.models import PendingCandidate from wf_scheduling.ownership import SchedulerOwnership from wf_scheduling.poll import Scheduler from wf_scheduling.prepare import SchedulePreparer from wf_scheduling.recovery import recover from wf_scheduling.store import ( FileScheduleStore, ScheduleExistsError, StaleScheduleRevisionError, ) 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 _cron() -> dict[str, Any]: return {"kind": "cron", "expression": "* * * * *", "timezone": "UTC"} def _oneshot(at: datetime) -> dict[str, Any]: return {"kind": "oneshot", "at": at.isoformat()} def _literal_binding(target: str, value: Any) -> dict[str, Any]: return {"target": target, "expression": {"kind": "literal", "value": value}} def _occurrence_binding(target: str, field: str) -> dict[str, Any]: return {"target": target, "expression": {"kind": "occurrence", "field": field}} class DummyEvents: def record_event(self, event: object) -> None: pass def record_workflow_event( self, event_type: str, *, capability_id: str, payload: dict[str, Any], ) -> None: pass class EmptySpecProvider: @property def capability_sources(self) -> dict[str, CapabilitySource]: return {} def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]: raise KeyError(f"unknown capability {qualified_name!r}") class NeverRuntime: """Context runtime double: schedule admin never executes workflows.""" async def run_workflow_from_plan(self, *args: Any, **kwargs: Any) -> Any: raise AssertionError("schedule admin must not execute workflows") async def resume_workflow_from_plan(self, *args: Any, **kwargs: Any) -> Any: raise AssertionError("schedule admin must not resume workflows") class CompletingRuntime: """Resume-only fake completing one interrupted run.""" async def run_workflow_from_plan(self, *args: Any, **kwargs: Any) -> Any: raise AssertionError("test must not start new workflow runs") async def resume_workflow_from_plan( self, plan: Any, run: RunState, *, resume_payload: dict[str, Any], resume_outcome: str, deployment: Any = None, artifact: Any = None, saved_subgraph_tree: Any = None, ) -> RunState: return RunState( workflow_name=plan.name, status=RunStatus.COMPLETED, workflow_input=dict(run.workflow_input), state={}, outcome=resume_outcome, output=dict(resume_payload), ) def _artifact( artifact_id: str = "sched-art", input_schema: dict[str, Any] | None = None, ) -> WorkflowArtifact: schema = ( input_schema if input_schema is not None else {"type": "object", "properties": {}} ) return WorkflowArtifact( id=artifact_id, version=1, title="Sched", input_schema=schema, output_schema={"type": "object", "properties": {}}, outcomes=("ok", "submitted"), plan={ "name": artifact_id, "input_schema": schema, "state_schema": {"type": "object", "properties": {}}, "output_schema": {"type": "object", "properties": {}}, "outcomes": ["ok", "submitted"], "start": "end_submitted", "nodes": [{"id": "end_submitted", "type": "end", "outcome": "submitted"}], "edges": [], }, ) def _deployment( deployment_id: str = "dep.personal", artifact_id: str = "sched-art" ) -> WorkflowDeployment: return WorkflowDeployment( id=deployment_id, artifact_id=artifact_id, artifact_version=1, bindings=[], ) def _context( artifact_store: FileWorkflowArtifactStore, run_store: FileRunStore, sched_store: FileScheduleStore | None, runtime: Any | None = None, ) -> WorkflowOperationContext: return WorkflowOperationContext( artifact_store=artifact_store, draft_workspace_store=None, run_store=run_store, events=DummyEvents(), specs=EmptySpecProvider(), runtime=runtime or NeverRuntime(), live_sources=None, schedule_store=sched_store, ) def _harness( root: Path, ) -> tuple[ WorkflowScheduleApi, FileScheduleStore, FileRunStore, FileWorkflowArtifactStore, WorkflowOperationContext, ]: artifact_store = FileWorkflowArtifactStore(root) artifact_store.save_artifact(_artifact()) artifact_store.save_deployment(_deployment()) run_store = FileRunStore(root) sched_store = FileScheduleStore(root) context = _context(artifact_store, run_store, sched_store) return WorkflowScheduleApi(context), sched_store, run_store, artifact_store, context def _env(artifact_store: FileWorkflowArtifactStore) -> Any: deployment = artifact_store.get_deployment("dep.personal") artifact = artifact_store.get_artifact( deployment.artifact_id, deployment.artifact_version ) return create_pinned_environment( deployment=deployment, artifact=artifact, tree=SavedSubgraphTree(artifacts_by_ref={}, diagnostics=[]), ) def _admit_owned( run_store: FileRunStore, artifact_store: FileWorkflowArtifactStore, run_id: str, intended: datetime, schedule_id: str = "s", resolved_input: dict[str, Any] | None = None, ) -> Any: admission = persist_admission( store=run_store, run_id=run_id, environment=_env(artifact_store), resolved_input=dict(resolved_input or {}), max_steps=None, scheduled_at=intended, schedule_id=schedule_id, schedule_revision=1, ) materialize_admitted_view(store=run_store, admission=admission) return admission def _interrupted_state() -> RunState: return RunState( workflow_name="sched", status=RunStatus.INTERRUPTED, workflow_input={}, state={}, interrupt=InterruptRequest( id="interrupt:approval", frame_id="root", node_id="approval", kind="approval", payload={"question": "continue?"}, ), ) def _history_rows( sched_store: FileScheduleStore, schedule_id: str ) -> list[dict[str, Any]]: page = sched_store.list_occurrences(schedule_id, limit=100) rows = page["occurrences"] assert isinstance(rows, list) return rows class FailConsumedOnce(FileScheduleStore): """Real schedule store failing exactly one watermark write.""" def __init__(self, root: Path) -> None: super().__init__(root) self.armed = False def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None: if self.armed: self.armed = False raise OSError("injected consumed failure") super().save_consumed(schedule_id, consumed_through) def _fault_harness( root: Path, ) -> tuple[WorkflowScheduleApi, FailConsumedOnce, FileRunStore]: artifact_store = FileWorkflowArtifactStore(root) artifact_store.save_artifact(_artifact()) artifact_store.save_deployment(_deployment()) run_store = FileRunStore(root) sched_store = FailConsumedOnce(root) context = _context(artifact_store, run_store, sched_store) return WorkflowScheduleApi(context), sched_store, run_store async def test_create_get_list_round_trip_with_defaults(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "round_trip") created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron(), ) assert created["id"] == "s" assert created["deployment_id"] == "dep.personal" assert created["trigger"] == _cron() assert created["input_bindings"] == [] assert created["overlap"] == "skip" assert created["misfire"] == "skip" assert created["max_active_runs"] == 1 assert created["lateness_allowance_s"] == 60.0 assert created["max_steps"] is None assert created["revision"] == 1 assert created["enabled"] is True assert created["paused"] is False assert created["deleted"] is False assert created["exhausted"] is False assert created["blocked_reason"] is None assert created["created_at"] == created["updated_at"] fetched = await api.get_schedule(schedule_id="s") assert fetched["id"] == "s" assert fetched["revision"] == 1 listed = await api.list_schedules() assert [row["id"] for row in listed["schedules"]] == ["s"] async def test_create_duplicate_rejected_including_deleted(tmp_path: Path) -> None: api, _, _, _, _ = _harness(tmp_path / "dup") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) with pytest.raises(ScheduleExistsError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) await api.delete_schedule(schedule_id="s") # Deleted ids are never reusable: old history stays annexed to the id. with pytest.raises(ScheduleExistsError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) async def test_create_unknown_deployment_keyerror(tmp_path: Path) -> None: api, _, _, _, _ = _harness(tmp_path / "unknown_dep") with pytest.raises(KeyError): await api.create_schedule( schedule_id="s", deployment_id="missing.dep", trigger=_cron() ) @pytest.mark.parametrize( "trigger", [ {"kind": "cron", "expression": "not-a-cron", "timezone": "UTC"}, {"kind": "cron", "expression": "* * *", "timezone": "UTC"}, {"kind": "cron", "expression": "* * * * *", "timezone": "Mars/Olympus"}, {"kind": "hourly"}, {"kind": "oneshot", "at": "2026-09-08T12:00:00"}, ], ) async def test_create_bad_trigger_valueerror( tmp_path: Path, trigger: dict[str, Any] ) -> None: api, _, _, _, _ = _harness(tmp_path / "bad_trigger") with pytest.raises(ValueError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=trigger ) @pytest.mark.parametrize( "bindings", [ [{"target": "x"}], [{"expression": {"kind": "literal", "value": 1}}], [{"target": "x", "expression": {"kind": "path", "path": "input.x"}}], [{"target": "x", "expression": {"kind": "occurrence", "field": "nope"}}], ], ) async def test_create_bad_bindings_valueerror( tmp_path: Path, bindings: list[dict[str, Any]] ) -> None: api, _, _, _, _ = _harness(tmp_path / "bad_bindings") with pytest.raises(ValueError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron(), input_bindings=bindings, ) @pytest.mark.parametrize("max_steps", [0, -3]) async def test_create_bad_max_steps_valueerror(tmp_path: Path, max_steps: int) -> None: api, _, _, _, _ = _harness(tmp_path / "bad_steps") with pytest.raises(ValueError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron(), max_steps=max_steps, ) async def test_create_sample_schema_failure_valueerror(tmp_path: Path) -> None: api, _, _, artifact_store, _ = _harness(tmp_path / "sample") artifact_store.save_artifact( _artifact( artifact_id="strict-art", input_schema={ "type": "object", "properties": {"msg": {"type": "string"}}, "required": ["msg"], }, ) ) artifact_store.save_deployment(_deployment("strict.personal", "strict-art")) # A literal violating the root input schema fails sample validation. with pytest.raises(ValueError): await api.create_schedule( schedule_id="bad", deployment_id="strict.personal", trigger=_cron(), input_bindings=[_literal_binding("msg", 123)], ) created = await api.create_schedule( schedule_id="good", deployment_id="strict.personal", trigger=_cron(), input_bindings=[_literal_binding("msg", "hi")], ) assert created["id"] == "good" async def test_update_happy_path_only_future_work(tmp_path: Path) -> None: api, sched_store, run_store, artifact_store, _ = _harness(tmp_path / "update") created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) # Admit before the edit: the frozen invocation must survive it. run_id = run_store.allocate_run_id() _admit_owned( run_store, artifact_store, run_id, intended, resolved_input={"note": "before"} ) before = datetime.now(UTC) updated = await api.update_schedule( schedule_id="s", expected_revision=1, trigger=_oneshot(intended + timedelta(hours=1)), input_bindings=[_occurrence_binding("note", "scheduled_at")], overlap="parallel", max_active_runs=3, max_steps=50, ) assert updated["revision"] == 2 assert updated["created_at"] == created["created_at"] assert updated["trigger"]["kind"] == "oneshot" assert updated["overlap"] == "parallel" assert updated["max_active_runs"] == 3 assert updated["max_steps"] == 50 # Edit side effects mirror Scheduler.edit_schedule. assert sched_store.get_candidate("s") is None rows = _history_rows(sched_store, "s") superseded = [row for row in rows if row["kind"] == "superseded"] assert len(superseded) == 1 assert superseded[0]["reason"] == "schedule-edit" assert superseded[0]["revision"] == 2 consumed_after_update = sched_store.get_consumed("s") assert consumed_after_update is not None assert consumed_after_update >= before # The admitted run keeps its pinned invocation: edits only affect # future admissions. admission = run_store.get_admission(run_id) assert admission.resolved_input == {"note": "before"} assert admission.schedule_revision == 1 assert run_store.get_run(run_id).status.value == "admitted" async def test_update_stale_revision_rejected(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "stale") created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) consumed_before = sched_store.get_consumed("s") with pytest.raises(StaleScheduleRevisionError): await api.update_schedule(schedule_id="s", expected_revision=99) # Stale edits write nothing: no revision bump, no candidate or # watermark change, no history row. assert sched_store.get_schedule("s").revision == 1 assert sched_store.get_schedule("s").updated_at == datetime.fromisoformat( created["updated_at"] ) candidate = sched_store.get_candidate("s") assert candidate is not None and candidate.intended_at == intended assert sched_store.get_consumed("s") == consumed_before assert _history_rows(sched_store, "s") == [] async def test_admin_mutation_waits_for_poll_transaction(tmp_path: Path) -> None: """A poll transition and an API edit cannot interleave store writes.""" api, sched_store, run_store, artifact_store, _ = _harness(tmp_path / "admin_poll") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) sched_store.save_consumed("s", intended - timedelta(hours=1)) entered = threading.Event() release = threading.Event() block_once = True original_save_consumed = sched_store.save_consumed def block_poll_watermark(schedule_id: str, consumed_through: datetime) -> None: nonlocal block_once if block_once: block_once = False entered.set() if not release.wait(timeout=5): raise TimeoutError("poll transaction did not get released") original_save_consumed(schedule_id, consumed_through) sched_store.save_consumed = block_poll_watermark # type: ignore[method-assign] ownership = SchedulerOwnership(tmp_path / "admin_poll", owner="poll").acquire() scheduler = Scheduler( schedule_store=sched_store, run_store=run_store, sources={"s": OneShotSource(intended)}, capacity=1, preparer=SchedulePreparer( DictDeployments({"dep.personal": {"rev": 1, "required": []}}), lambda sched: create_pinned_environment( deployment=artifact_store.get_deployment(sched.deployment_id), artifact=artifact_store.get_artifact("sched-art", 1), tree=SavedSubgraphTree(artifacts_by_ref={}, diagnostics=[]), ), ), dispatcher=ScriptedDispatcher({"*": "complete"}), ownership=ownership, ) release_timer = threading.Timer(0.2, release.set) try: poll_task = asyncio.create_task(asyncio.to_thread(scheduler.poll, intended)) assert await asyncio.to_thread(entered.wait, 5) edit_task = asyncio.create_task( api.update_schedule( schedule_id="s", expected_revision=1, overlap="parallel", max_active_runs=2, ) ) release_timer.start() poll_result, edited = await asyncio.gather(poll_task, edit_task) finally: release.set() release_timer.cancel() ownership.release() assert poll_result["s"].startswith("admit:") assert edited["revision"] == 2 assert run_store.list_admissions()[0].schedule_revision == 1 assert sched_store.get_schedule("s").revision == 2 async def test_create_initializes_consumed_no_backfill(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "create_consumed") created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron(), misfire="latest", ) created_at = datetime.fromisoformat(created["created_at"]) consumed = sched_store.get_consumed("s") assert consumed is not None and consumed >= created_at async def test_create_watermark_failure_does_not_publish_schedule( tmp_path: Path, ) -> None: """A failed creation watermark cannot expose a backfillable schedule.""" api, sched_store, _ = _fault_harness(tmp_path / "torn_create") sched_store.armed = True with pytest.raises(OSError, match="injected consumed failure"): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron(), misfire="latest", ) with pytest.raises(KeyError): sched_store.get_schedule("s") assert sched_store.get_consumed("s") is None async def test_update_watermark_failure_leaves_revision(tmp_path: Path) -> None: api, sched_store, _ = _fault_harness(tmp_path / "torn_update") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) sched_store.armed = True with pytest.raises(OSError, match="injected consumed failure"): await api.update_schedule( schedule_id="s", expected_revision=1, overlap="parallel", max_active_runs=2, ) # Crash-safe ordering: the watermark advance precedes the revision # bump, so a torn edit leaves the old revision (retryable) and can # never backfill pre-edit instants under the new revision. assert sched_store.get_schedule("s").revision == 1 assert sched_store.get_schedule("s").overlap == "skip" async def test_pause_watermark_failure_leaves_flag(tmp_path: Path) -> None: api, sched_store, _ = _fault_harness(tmp_path / "torn_pause") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) sched_store.armed = True with pytest.raises(OSError, match="injected consumed failure"): await api.pause_schedule(schedule_id="s") assert sched_store.get_schedule("s").paused is False async def test_unknown_schedule_keyerror(tmp_path: Path) -> None: api, _, _, _, _ = _harness(tmp_path / "unknown_sched") with pytest.raises(KeyError): await api.get_schedule(schedule_id="missing") with pytest.raises(KeyError): await api.update_schedule(schedule_id="missing", expected_revision=1) with pytest.raises(KeyError): await api.pause_schedule(schedule_id="missing") with pytest.raises(KeyError): await api.resume_schedule(schedule_id="missing") with pytest.raises(KeyError): await api.delete_schedule(schedule_id="missing") with pytest.raises(KeyError): await api.list_schedule_occurrences(schedule_id="missing") async def test_pause_resume_delete_transitions(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "transitions") created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) # Creation never backfills time before the revision: the consumed # watermark starts at creation. created_consumed = sched_store.get_consumed("s") assert created_consumed is not None assert created_consumed >= datetime.fromisoformat(created["created_at"]) intended = ts(2026, 9, 8, 12, 0) sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) before_pause = datetime.now(UTC) paused = await api.pause_schedule(schedule_id="s") assert paused["paused"] is True assert paused["revision"] == 1 # Pause clears the candidate and advances the watermark past the pause # (no history write: the span is never backfilled). assert sched_store.get_candidate("s") is None consumed = sched_store.get_consumed("s") assert consumed is not None and consumed >= before_pause assert _history_rows(sched_store, "s") == [] sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) before_resume = datetime.now(UTC) resumed = await api.resume_schedule(schedule_id="s") assert resumed["paused"] is False assert sched_store.get_candidate("s") is None resumed_consumed = sched_store.get_consumed("s") assert resumed_consumed is not None assert resumed_consumed >= before_resume assert resumed_consumed >= consumed assert _history_rows(sched_store, "s") == [] sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) deleted = await api.delete_schedule(schedule_id="s") assert deleted["deleted"] is True assert sched_store.get_candidate("s") is None # Delete advances nothing: the watermark is exactly the resume value. assert sched_store.get_consumed("s") == resumed_consumed assert (await api.list_schedules())["schedules"] == [] listed = await api.list_schedules(include_deleted=True) assert [row["id"] for row in listed["schedules"]] == ["s"] async def test_delete_preserves_runs_and_history(tmp_path: Path) -> None: api, sched_store, run_store, artifact_store, _ = _harness(tmp_path / "del_keep") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) run_id = run_store.allocate_run_id() _admit_owned(run_store, artifact_store, run_id, intended) FileScheduleHistoryRecorder(sched_store).record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=intended, run_id=run_id, revision=1, reason="rev=1", created_at=intended, ) ) await api.delete_schedule(schedule_id="s") assert run_store.get_run(run_id).id == run_id page = await api.list_schedule_occurrences(schedule_id="s") assert [row["run_id"] for row in page["occurrences"]] == [run_id] with pytest.raises(ScheduleExistsError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) async def test_occurrences_pagination_and_validation(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "pages") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) base = ts(2026, 9, 8, 12, 0) recorder = FileScheduleHistoryRecorder(sched_store) for index in range(5): instant = base + timedelta(minutes=index) recorder.record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=instant, run_id=f"run-{index}", revision=1, reason="rev=1", created_at=instant, ) ) first = await api.list_schedule_occurrences(schedule_id="s", limit=2) assert first["total"] == 5 assert first["cursor"] is None assert first["limit"] == 2 assert len(first["occurrences"]) == 2 assert first["next_cursor"] is not None second = await api.list_schedule_occurrences( schedule_id="s", cursor=first["next_cursor"], limit=2 ) assert second["total"] == 5 assert second["cursor"] == first["next_cursor"] assert len(second["occurrences"]) == 2 assert {row["run_id"] for row in first["occurrences"]}.isdisjoint( {row["run_id"] for row in second["occurrences"]} ) with pytest.raises(ValueError): await api.list_schedule_occurrences(schedule_id="s", limit=0) with pytest.raises(ValueError): await api.list_schedule_occurrences(schedule_id="s", limit=101) with pytest.raises(ValueError): await api.list_schedule_occurrences(schedule_id="s", cursor="bogus") async def test_occurrences_pending_synthesis_first_page_only(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "pending") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) base = ts(2026, 9, 8, 12, 0) recorder = FileScheduleHistoryRecorder(sched_store) for index in range(3): instant = base + timedelta(minutes=index) recorder.record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=instant, run_id=f"run-{index}", revision=1, reason="rev=1", created_at=instant, ) ) intended = base + timedelta(hours=1) sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) first = await api.list_schedule_occurrences(schedule_id="s", limit=2) assert first["total"] == 4 pending = first["occurrences"][0] assert pending["kind"] == "pending" assert pending["schedule_id"] == "s" assert pending["occurrence_id"] == f"s|{intended.isoformat()}" assert datetime.fromisoformat(pending["resolved_at"]) == intended assert pending["revision"] == 1 assert pending["reason"] == "" assert pending["run_id"] is None assert first["occurrences"][1]["kind"] == "admitted" assert first["next_cursor"] is not None later = await api.list_schedule_occurrences( schedule_id="s", cursor=first["next_cursor"], limit=2 ) assert later["total"] == 3 assert all(row["kind"] != "pending" for row in later["occurrences"]) sched_store.save_candidate(None, schedule_id="s") plain = await api.list_schedule_occurrences(schedule_id="s", limit=10) assert plain["total"] == 3 assert all(row["kind"] != "pending" for row in plain["occurrences"]) async def test_admitted_occurrence_replaces_retained_candidate_after_recovery( tmp_path: Path, ) -> None: """A torn candidate clear cannot expose an admitted occurrence as pending.""" root = tmp_path / "admitted_candidate" api, sched_store, run_store, artifact_store, _ = _harness(root) await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 13, 0) admission = persist_admission( store=run_store, run_id=run_store.allocate_run_id(), environment=_env(artifact_store), resolved_input={}, max_steps=None, scheduled_at=intended, schedule_id="s", schedule_revision=1, ) FileScheduleHistoryRecorder(sched_store).record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=intended, run_id=admission.id, revision=1, reason="rev=1", created_at=intended, ) ) # Fault injection: admission/history are durable, but candidate clearing # was lost at the following persistence boundary. sched_store.save_candidate( PendingCandidate(schedule_id="s", intended_at=intended, revision=1), schedule_id="s", ) before_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1) assert before_recovery["total"] == 1 assert before_recovery["occurrences"][0]["kind"] == "admitted" ownership = SchedulerOwnership(root, owner="test").acquire() try: recover( schedule_store=sched_store, run_store=run_store, now=intended, ownership=ownership, ) finally: ownership.release() assert sched_store.get_candidate("s") is None after_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1) assert after_recovery["total"] == 1 assert after_recovery["occurrences"][0]["kind"] == "admitted" assert after_recovery["occurrences"][0]["run_id"] == admission.id async def test_occurrences_pending_limit_one_traverses_every_row( tmp_path: Path, ) -> None: """A synthesized pending row must not make the first stored row unreachable.""" api, sched_store, _, _, _ = _harness(tmp_path / "pending_limit_one") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) base = ts(2026, 9, 8, 12, 0) recorder = FileScheduleHistoryRecorder(sched_store) for index in range(3): instant = base + timedelta(minutes=index) recorder.record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=instant, run_id=f"run-{index}", revision=1, reason="rev=1", created_at=instant, ) ) sched_store.save_candidate( PendingCandidate( schedule_id="s", intended_at=base + timedelta(hours=1), revision=1, ), schedule_id="s", ) rows: list[dict[str, Any]] = [] cursor: str | None = None for _ in range(5): page = await api.list_schedule_occurrences( schedule_id="s", cursor=cursor, limit=1 ) assert len(page["occurrences"]) <= 1 rows.extend(page["occurrences"]) cursor = page["next_cursor"] if cursor is None: break else: pytest.fail("occurrence pagination did not terminate") assert [row["kind"] for row in rows] == [ "pending", "admitted", "admitted", "admitted", ] assert [row["run_id"] for row in rows[1:]] == [ "run-0", "run-1", "run-2", ] async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) -> None: """One occurrence's admission + stops each appear exactly once (B4). Small-page traversal through the public API must not lose the tied stopped entries sharing the occurrence's ``(resolved_at, occurrence_id)``. """ api, sched_store, _, _, _ = _harness(tmp_path / "tied") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) base = ts(2026, 9, 8, 12, 0) recorder = FileScheduleHistoryRecorder(sched_store) recorder.record( HistoryEntry( schedule_id="s", kind="admitted", resolved_at=base, run_id="run-1", revision=1, reason="rev=1", created_at=base, ) ) recorder.record( HistoryEntry( schedule_id="s", kind="interrupted", resolved_at=base, run_id="run-1", revision=1, reason="fresh-result", checkpoint_id="run-1.000001", created_at=base + timedelta(minutes=5), ) ) recorder.record( HistoryEntry( schedule_id="s", kind="completed", resolved_at=base, run_id="run-1", revision=1, reason="reconciled-on-recovery", checkpoint_id="run-1.000002", created_at=base + timedelta(minutes=9), ) ) seen: list[tuple[str, str | None]] = [] cursor: str | None = None for _ in range(10): page = await api.list_schedule_occurrences( schedule_id="s", cursor=cursor, limit=1 ) assert page["total"] == 3 for row in page["occurrences"]: seen.append((row["kind"], row["checkpoint_id"])) next_cursor = page["next_cursor"] if next_cursor is None: break assert next_cursor != cursor cursor = next_cursor else: pytest.fail("occurrence pagination did not terminate") assert seen == [ ("admitted", None), ("interrupted", "run-1.000001"), ("completed", "run-1.000002"), ] async def test_inspect_admitted_run_without_fabrication(tmp_path: Path) -> None: api, sched_store, run_store, artifact_store, context = _harness( tmp_path / "admitted" ) await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) run_id = run_store.allocate_run_id() _admit_owned(run_store, artifact_store, run_id, intended) summary = await WorkflowRunApi(context).inspect_run(run_id=run_id) assert summary["status"] == "admitted" assert summary["run_id"] == run_id assert summary["trace_count"] == 0 assert summary["output"] is None assert summary["interrupt"] is None assert "trace" not in summary with pytest.raises(ValueError): await WorkflowRunApi(context).read_run_trace( run_id=run_id, trace_range=TraceRange(start=0, limit=1) ) async def test_list_runs_includes_admitted(tmp_path: Path) -> None: _, _, run_store, artifact_store, context = _harness(tmp_path / "list_adm") run_id = run_store.allocate_run_id() _admit_owned(run_store, artifact_store, run_id, ts(2026, 9, 8, 12, 0)) runs = WorkflowRunApi(context) filtered = await runs.list_runs(status="admitted") assert filtered["total"] == 1 assert filtered["runs"][0]["run_id"] == run_id assert filtered["runs"][0]["status"] == "admitted" unfiltered = await runs.list_runs() assert unfiltered["total"] == 1 async def test_schedule_owned_resume_then_recovery_reconciles(tmp_path: Path) -> None: root = tmp_path / "reconcile" api, sched_store, run_store, artifact_store, context = _harness(root) await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) intended = ts(2026, 9, 8, 12, 0) run_id = run_store.allocate_run_id() admission = _admit_owned(run_store, artifact_store, run_id, intended) persist_stopped_run( store=run_store, environment=admission.environment, run=_interrupted_state(), run_id=run_id, ) resume_context = _context( artifact_store, run_store, sched_store, runtime=CompletingRuntime() ) resumed = await WorkflowRunApi(resume_context).resume_run( run_id=run_id, resume_payload={"approved": True}, resume_outcome="submitted", ) assert resumed["status"] == "completed" from wf_scheduling import recovery as sched_recovery from wf_scheduling.ownership import SchedulerOwnership ownership = SchedulerOwnership(root, owner="test").acquire() try: diags = sched_recovery.recover( schedule_store=sched_store, run_store=run_store, now=intended + timedelta(minutes=5), ownership=ownership, ) finally: ownership.release() assert any("terminal-reconciled" in message for message in diags) page = await api.list_schedule_occurrences(schedule_id="s") completed = [row for row in page["occurrences"] if row["kind"] == "completed"] assert len(completed) == 1 assert completed[0]["run_id"] == run_id assert completed[0]["checkpoint_id"] == f"{run_id}.000002" async def test_schedule_api_without_store_raises_keyerror(tmp_path: Path) -> None: root = tmp_path / "no_store" artifact_store = FileWorkflowArtifactStore(root) artifact_store.save_artifact(_artifact()) artifact_store.save_deployment(_deployment()) run_store = FileRunStore(root) context = _context(artifact_store, run_store, None) api = WorkflowScheduleApi(context) with pytest.raises(KeyError): await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) with pytest.raises(KeyError): await api.get_schedule(schedule_id="s") with pytest.raises(KeyError): await api.list_schedules() with pytest.raises(KeyError): await api.update_schedule(schedule_id="s", expected_revision=1) with pytest.raises(KeyError): await api.pause_schedule(schedule_id="s") with pytest.raises(KeyError): await api.resume_schedule(schedule_id="s") with pytest.raises(KeyError): await api.delete_schedule(schedule_id="s") with pytest.raises(KeyError): await api.list_schedule_occurrences(schedule_id="s") facade = WorkflowApi(context) with pytest.raises(KeyError): await facade.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) async def test_facade_and_durable_context_delegate(tmp_path: Path) -> None: _, sched_store, run_store, artifact_store, _ = _harness(tmp_path / "facade") bare = _context(artifact_store, run_store, None) explicit = WorkflowApi(bare, schedule_store=sched_store) created = await explicit.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) assert created["id"] == "s" assert (await explicit.get_schedule(schedule_id="s"))["revision"] == 1 assert [row["id"] for row in (await explicit.list_schedules())["schedules"]] == [ "s" ] assert (await explicit.pause_schedule(schedule_id="s"))["paused"] is True assert (await explicit.resume_schedule(schedule_id="s"))["paused"] is False assert (await explicit.delete_schedule(schedule_id="s"))["deleted"] is True stored_context = _context(artifact_store, run_store, sched_store) via_durable = durable_workflow_api(stored_context) assert (await via_durable.get_schedule(schedule_id="s"))["deleted"] is True async def test_local_server_schedules_flag(tmp_path: Path) -> None: from wf_server import build_local_static_workflow_server plain = build_local_static_workflow_server(tmp_path / "plain") assert not (tmp_path / "plain" / "schedules").exists() with pytest.raises(KeyError): await plain.api.get_schedule(schedule_id="s") server = build_local_static_workflow_server(tmp_path / "srv", schedules=True) assert (tmp_path / "srv" / "schedules").is_dir() server.stores.artifact_store.save_artifact(_artifact()) server.stores.artifact_store.save_deployment(_deployment()) created = await server.api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) assert created["id"] == "s" assert (await server.api.get_schedule(schedule_id="s"))["revision"] == 1