1140 lines
39 KiB
Python
1140 lines
39 KiB
Python
"""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
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
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.history import FileScheduleHistoryRecorder, HistoryEntry
|
|
from wf_scheduling.models import PendingCandidate
|
|
from wf_scheduling.ownership import SchedulerOwnership
|
|
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_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
|
|
while True:
|
|
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"]))
|
|
cursor = page["next_cursor"]
|
|
if cursor is None:
|
|
break
|
|
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
|