From 7f00422fd7b070c3f246b93d2006cb40235d54df Mon Sep 17 00:00:00 2001 From: lda Date: Wed, 9 Sep 2026 11:33:12 +0700 Subject: [PATCH] sched: schedule administration API, admitted inspection, occurrence pages (T13) --- src/wf_api/__init__.py | 15 +- src/wf_api/durable_context.py | 11 +- src/wf_api/models/__init__.py | 4 + src/wf_api/models/schedules.py | 51 ++ src/wf_api/operation_context.py | 4 + src/wf_api/runs.py | 31 +- src/wf_api/schedules.py | 421 ++++++++++++++++ src/wf_api/service.py | 115 +++++ src/wf_api/surface.py | 83 ++++ src/wf_server/context.py | 15 +- tests/wf_api/test_schedules.py | 832 ++++++++++++++++++++++++++++++++ 11 files changed, 1572 insertions(+), 10 deletions(-) create mode 100644 src/wf_api/models/schedules.py create mode 100644 src/wf_api/schedules.py create mode 100644 tests/wf_api/test_schedules.py diff --git a/src/wf_api/__init__.py b/src/wf_api/__init__.py index f96a2ab2..5dbec893 100644 --- a/src/wf_api/__init__.py +++ b/src/wf_api/__init__.py @@ -23,7 +23,13 @@ from .drafts import WorkflowDraftApi from .durable_context import durable_workflow_api, require_workflow_stores from .listing import matches_query, paged_list_payload from .local_sources import builtin_sources, get_qualified_spec, qualify_spec -from .models import RawWorkflowPlan, TraceRange +from .models import ( + ListSchedulesResult, + OccurrencePage, + RawWorkflowPlan, + ScheduleResult, + TraceRange, +) from .next_actions import NextActionPatchExample, NextActions, NextActionTool from .operation_context import ( WorkflowEventRecorder, @@ -35,6 +41,7 @@ from .operation_context import ( from .refs import WorkflowSurfaceCapabilityId, parse_workflow_surface_capability_id from .runs import WorkflowRunApi from .runtime_dependencies import RuntimeDependencies, resolve_runtime_dependencies +from .schedules import WorkflowScheduleApi from .service import WorkflowApi from .source_admin import WorkflowSourceAdminApi from .source_refs import SourceResourceRef @@ -53,6 +60,7 @@ from .surface import ( WorkflowDeploymentSurface, WorkflowDraftSurface, WorkflowRunSurface, + WorkflowScheduleSurface, WorkflowSourceAdminSurface, WorkflowSourceRegistrySurface, ) @@ -78,6 +86,7 @@ __all__ = [ "AuthRecord", "AuthStore", "CapabilityStepUpdate", + "ListSchedulesResult", "MissingDecision", "MissingDecisionKind", "NextActionPatchExample", @@ -85,9 +94,11 @@ __all__ = [ "NextActions", "OutcomeCandidate", "OutcomeCandidateKind", + "OccurrencePage", "RawWorkflowPlan", "RouteSource", "RuntimeDependencies", + "ScheduleResult", "SourceResourceRef", "TraceRange", "WorkflowAdminApi", @@ -111,6 +122,8 @@ __all__ = [ "WorkflowRunApi", "WorkflowRunSurface", "WorkflowRuntimeRunner", + "WorkflowScheduleApi", + "WorkflowScheduleSurface", "WorkflowSourceAdminApi", "WorkflowSourceAdminSurface", "WorkflowSourceRegistryApi", diff --git a/src/wf_api/durable_context.py b/src/wf_api/durable_context.py index ef62f5b3..e6d38447 100644 --- a/src/wf_api/durable_context.py +++ b/src/wf_api/durable_context.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Any + from .operation_context import WorkflowOperationContext from .service import WorkflowApi from .stores import WorkflowStores @@ -38,10 +40,15 @@ def durable_workflow_api( context: WorkflowOperationContext, *, drafts: bool = False, + schedule_store: Any | None = None, ) -> WorkflowApi: - """Construct a durable API, optionally omitting the draft product surface.""" + """Construct a durable API, optionally omitting the draft product surface. + + An explicit schedule store enables the T13 administration surface; + without one the schedule sub-API raises KeyError, like runs do. + """ require_workflow_stores(context, drafts=drafts) - return WorkflowApi(context, drafts=drafts) + return WorkflowApi(context, drafts=drafts, schedule_store=schedule_store) __all__ = ["durable_workflow_api", "require_workflow_stores"] diff --git a/src/wf_api/models/__init__.py b/src/wf_api/models/__init__.py index a06b3ae5..3999da79 100644 --- a/src/wf_api/models/__init__.py +++ b/src/wf_api/models/__init__.py @@ -101,6 +101,7 @@ from .runs import ( TraceEntryPayload, WorkflowRefPayload, ) +from .schedules import ListSchedulesResult, OccurrencePage, ScheduleResult from .source_registry import ( ApplyRegistryChangesResult, InspectRegistryEntryResult, @@ -186,11 +187,13 @@ __all__ = [ "ListRunsResult", "ListRegistryEntriesResult", "ListSourcesResult", + "ListSchedulesResult", "NextActionPatchExamplePayload", "NextActionsPayload", "NodeSpecCapabilityDetail", "NodeSpecCapabilitySummary", "NodeSpecInventoryPayload", + "OccurrencePage", "PageMetadataPayload", "PatchDraftResult", "PatchedDraftInvalidResult", @@ -208,6 +211,7 @@ __all__ = [ "RequiredCapabilityPayload", "RemoveRegistryEntryResult", "SaveArtifactResult", + "ScheduleResult", "ValidateArtifactPlanResult", "SavedDraftArtifactResult", "SaveDeploymentResult", diff --git a/src/wf_api/models/schedules.py b/src/wf_api/models/schedules.py new file mode 100644 index 00000000..d5a9e592 --- /dev/null +++ b/src/wf_api/models/schedules.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from typing import Any, TypedDict + + +class ScheduleResult(TypedDict): + """JSON projection of one schedule definition. + + Mirrors ``Schedule.model_dump(mode="json")`` exactly; ``created_at`` and + ``updated_at`` are ISO strings in this projection. + """ + + id: str + deployment_id: str + trigger: dict[str, Any] + input_bindings: list[dict[str, Any]] + max_steps: int | None + overlap: str + misfire: str + max_active_runs: int + lateness_allowance_s: float + revision: int + enabled: bool + paused: bool + deleted: bool + exhausted: bool + blocked_reason: str | None + created_at: str + updated_at: str + + +class ListSchedulesResult(TypedDict): + """Schedule listing payload shared by the admin surface.""" + + schedules: list[ScheduleResult] + + +class OccurrencePage(TypedDict): + """One occurrence-history page, optionally with a live pending row. + + Matches the schedule-store page shape (``occurrences``, ``total``, + ``cursor``, ``next_cursor``, ``limit``); see + ``WorkflowScheduleApi.list_schedule_occurrences`` for the pending-row + synthesis contract. + """ + + occurrences: list[dict[str, Any]] + total: int + cursor: str | None + next_cursor: str | None + limit: int diff --git a/src/wf_api/operation_context.py b/src/wf_api/operation_context.py index 19dff342..c90d6978 100644 --- a/src/wf_api/operation_context.py +++ b/src/wf_api/operation_context.py @@ -110,6 +110,10 @@ class WorkflowOperationContext: specs: WorkflowSpecProvider runtime: WorkflowRuntimeRunner live_sources: WorkflowLiveSourceChecker | None = None + # Optional schedule store for the T13 administration surface. None keeps + # every existing construction site working; the schedule API raises + # KeyError when it is absent, exactly like runs with a missing store. + schedule_store: Any | None = None __all__ = [ diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index 845d2ebc..cfd5030d 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -337,13 +337,34 @@ class WorkflowRunApi: } async def inspect_run(self, *, run_id: str) -> RunResult: - """Return one durable stopped-run summary without debug trace entries. + """Return one durable run summary without debug trace entries. - Admitted runs with no stopped checkpoint fail closed here (no - fabricated trace/output); checkpoint-free inspection arrives with - the scheduling administration surface (T13). + Admitted runs with no stopped checkpoint report their durable + admission truthfully (status admitted, no checkpoint id, no + trace slice, output, or interrupt) instead of failing closed; + inspection never fabricates checkpoint-derived state. + ``read_run_trace`` still rejects checkpoint-less runs. """ - record, run = load_stored_run(self._run_store(), run_id) + store = self._run_store() + record = store.get_run(run_id) + if record.status is StoredRunStatus.ADMITTED: + environment = record.environment + try: + max_steps = store.get_admission(run_id).max_steps + except KeyError: + max_steps = None + return _run_payload( + deployment=environment.deployment, + artifact=environment.root_artifact, + status=record.status.value, + run_id=record.id, + resume_readiness=record.resume_readiness.value, + diagnostics=record.diagnostics, + trace_count=0, + max_steps=max_steps, + steps_executed=0, + ) + record, run = load_stored_run(store, run_id) environment = record.environment return _run_payload( deployment=environment.deployment, diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py new file mode 100644 index 00000000..1c6a8596 --- /dev/null +++ b/src/wf_api/schedules.py @@ -0,0 +1,421 @@ +"""Schedule administration operations (T13). + +Same-process admin surface over the file schedule store: create, inspect, +list, revision-checked update, pause, resume, delete, and paginated +occurrence inspection with a live pending-candidate synthesis. + +There is deliberately no ownership requirement on these admin ops: this is +the same-process administration surface, the service observes fresh reads +from the store on every call, and second-process schedulers remain +unsupported per the store boundary (see ``wf_scheduling.ownership``). +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any + +from wf_core import RunLimits +from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry +from wf_scheduling.models import OccurrenceRecord, Schedule +from wf_scheduling.occurrences import occurrence_id +from wf_scheduling.prepare import PreparationRejected, SchedulePreparer + +from .models import ( + JsonProjector, + ListSchedulesResult, + OccurrencePage, + ScheduleResult, +) +from .operation_context import WorkflowOperationContext +from .run_lifecycle import create_pinned_environment +from .saved_subgraphs import resolve_saved_subgraph_tree + +_PROJECT_SCHEDULE = JsonProjector(ScheduleResult) +_PROJECT_LIST_SCHEDULES = JsonProjector(ListSchedulesResult) +_PROJECT_OCCURRENCE_PAGE = JsonProjector(OccurrencePage) + + +class _ContextDeploymentDirectory: + """Deployment contract source over the API artifact store. + + Mirrors ``StoreDeploymentDirectory`` (``wf_scheduling.lifecycle``) + without importing the server-lifecycle layer: the deployment revision + plus the required keys of the pinned root artifact's input schema. + Unknown ids raise ``KeyError``. + """ + + def __init__(self, artifact_store: Any) -> None: + self._artifact_store = artifact_store + + def deployment_revision(self, deployment_id: str) -> int: + return int(self._artifact_store.get_deployment(deployment_id).revision) + + def required_inputs(self, deployment_id: str) -> list[str]: + deployment = self._artifact_store.get_deployment(deployment_id) + artifact = self._artifact_store.get_artifact( + deployment.artifact_id, deployment.artifact_version + ) + schema = getattr(artifact, "input_schema", None) + if not isinstance(schema, dict): + return [] + required = schema.get("required", []) + if not isinstance(required, list): + return [] + return [key for key in required if isinstance(key, str)] + + +def _build_pinned_environment(artifact_store: Any) -> Callable[[Any], Any]: + """Pin deployment + root artifact + subgraph tree for sample validation.""" + + def build(sched: Any) -> Any: + deployment = artifact_store.get_deployment(sched.deployment_id) + artifact = artifact_store.get_artifact( + deployment.artifact_id, deployment.artifact_version + ) + tree = resolve_saved_subgraph_tree( + root_artifact=artifact, artifact_store=artifact_store + ) + return create_pinned_environment( + deployment=deployment, artifact=artifact, tree=tree + ) + + return build + + +class WorkflowScheduleApi: + """Schedule administration over the configured schedule store. + + The store resolves from the operation context (or from an explicitly + passed store, which wins); every method raises ``KeyError`` when no + schedule store is configured, exactly like runs with a missing store. + """ + + def __init__( + self, + context: WorkflowOperationContext, + *, + schedule_store: Any | None = None, + ) -> None: + self.context = context + self._explicit_schedule_store = schedule_store + + def _schedule_store(self) -> Any: + store = self._explicit_schedule_store + if store is None: + store = getattr(self.context, "schedule_store", None) + if store is None: + raise KeyError("workflow schedule store is not configured") + return store + + def _artifact_store(self) -> Any: + if self.context.artifact_store is None: + raise KeyError("workflow artifact store is not configured") + return self.context.artifact_store + + @staticmethod + def _check_trigger(schedule: Schedule) -> None: + """Reject triggers the poll loop could not build a source for.""" + from wf_scheduling.poll import source_for_trigger + + try: + source_for_trigger(schedule.trigger) + except Exception as exc: + raise ValueError(f"invalid schedule trigger: {exc}") from exc + + @staticmethod + def _check_max_steps(max_steps: int | None) -> None: + """Reuse the manual-run step-budget validator for schedules.""" + if max_steps is None: + return + try: + RunLimits(max_steps=max_steps) + except (TypeError, ValueError) as exc: + raise ValueError(f"invalid max_steps {max_steps!r}: {exc}") from exc + + def _validate_sample(self, schedule: Schedule) -> None: + """Resolve the bindings against a synthetic occurrence and validate. + + A real ``SchedulePreparer`` wired to the context artifact store + resolves the schedule's input bindings for the current instant and + validates the result against the pinned root artifact's input + schema, so a schedule cannot be saved when its very first + occurrence would fail preflight. Failures raise ``ValueError``. + """ + artifact_store = self._artifact_store() + preparer = SchedulePreparer( + _ContextDeploymentDirectory(artifact_store), + _build_pinned_environment(artifact_store), + ) + now = datetime.now(UTC) + result = preparer.prepare(sched=schedule, intended=now, now=now) + if isinstance(result, PreparationRejected): + raise ValueError(f"schedule sample occurrence rejected: {result.reason}") + + def _validate_definition(self, schedule: Schedule) -> None: + """Run the trigger/budget/sample checks shared by create and update.""" + self._check_max_steps(schedule.max_steps) + self._check_trigger(schedule) + self._validate_sample(schedule) + + async def create_schedule( + self, + *, + schedule_id: str, + deployment_id: str, + trigger: dict[str, Any], + input_bindings: list[dict[str, Any]] | None = None, + overlap: str = "skip", + misfire: str = "skip", + max_active_runs: int = 1, + lateness_allowance_s: float = 60.0, + max_steps: int | None = None, + enabled: bool = True, + ) -> ScheduleResult: + """Create one schedule after deployment, trigger, and sample checks. + + Unknown deployments raise ``KeyError``; bad triggers, binding + shapes, budgets, and sample-occurrence rejections raise + ``ValueError``; duplicate ids (deleted ids are never reusable) + raise ``ScheduleExistsError``. + """ + store = self._schedule_store() + now = datetime.now(UTC) + # KeyError first: the deployment must exist before any validation. + self._artifact_store().get_deployment(deployment_id) + schedule = Schedule.model_validate( + { + "id": schedule_id, + "deployment_id": deployment_id, + "trigger": trigger, + "input_bindings": input_bindings if input_bindings is not None else [], + "max_steps": max_steps, + "overlap": overlap, + "misfire": misfire, + "max_active_runs": max_active_runs, + "lateness_allowance_s": lateness_allowance_s, + "revision": 1, + "enabled": enabled, + "paused": False, + "deleted": False, + "exhausted": False, + "blocked_reason": None, + "created_at": now, + "updated_at": now, + } + ) + self._validate_definition(schedule) + stored = store.create_schedule(schedule) + return _PROJECT_SCHEDULE(stored.model_dump(mode="json")) + + async def get_schedule(self, *, schedule_id: str) -> ScheduleResult: + """Return one schedule payload; unknown ids raise ``KeyError``.""" + schedule = self._schedule_store().get_schedule(schedule_id) + return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + + async def list_schedules( + self, *, include_deleted: bool = False + ) -> ListSchedulesResult: + """Return schedule payloads ordered by id (deleted excluded by default).""" + schedules = self._schedule_store().list_schedules( + include_deleted=include_deleted + ) + return _PROJECT_LIST_SCHEDULES( + { + "schedules": [ + _PROJECT_SCHEDULE(item.model_dump(mode="json")) + for item in schedules + ] + } + ) + + async def update_schedule( + self, + *, + schedule_id: str, + expected_revision: int, + deployment_id: str | None = None, + trigger: dict[str, Any] | None = None, + input_bindings: list[dict[str, Any]] | None = None, + overlap: str | None = None, + misfire: str | None = None, + max_active_runs: int | None = None, + lateness_allowance_s: float | None = None, + max_steps: int | None = None, + enabled: bool | None = None, + ) -> ScheduleResult: + """Apply a partial edit under an optimistic revision check. + + Only provided fields change; ``revision`` becomes + ``expected_revision + 1`` and ``updated_at`` becomes now, while + ``created_at`` and the admitted-run snapshots stay untouched (later + edits only affect future admissions). Stale revisions raise + ``StaleScheduleRevisionError``. Side effects mirror + ``Scheduler.edit_schedule``: a held candidate is cleared with a + ``superseded``/``schedule-edit`` history row, and the consumed + watermark advances to at least now (no backfill). + """ + store = self._schedule_store() + now = datetime.now(UTC) + current = store.get_schedule(schedule_id) + data = current.model_dump(mode="python") + if deployment_id is not None: + data["deployment_id"] = deployment_id + if trigger is not None: + data["trigger"] = trigger + if input_bindings is not None: + data["input_bindings"] = input_bindings + if overlap is not None: + data["overlap"] = overlap + if misfire is not None: + data["misfire"] = misfire + if max_active_runs is not None: + data["max_active_runs"] = max_active_runs + if lateness_allowance_s is not None: + data["lateness_allowance_s"] = lateness_allowance_s + if max_steps is not None: + data["max_steps"] = max_steps + if enabled is not None: + data["enabled"] = enabled + data["revision"] = expected_revision + 1 + data["updated_at"] = now + # KeyError first, like create: the (possibly repointed) deployment + # must exist before definition validation. + self._artifact_store().get_deployment(data["deployment_id"]) + updated = Schedule.model_validate(data) + self._validate_definition(updated) + stored = store.update_schedule(updated, expected_revision=expected_revision) + old_candidate = store.get_candidate(schedule_id) + if old_candidate is not None: + FileScheduleHistoryRecorder(store).record( + HistoryEntry( + schedule_id=schedule_id, + kind="superseded", + resolved_at=old_candidate.intended_at, + revision=stored.revision, + reason="schedule-edit", + created_at=now, + ) + ) + store.save_candidate(None, schedule_id=schedule_id) + consumed = store.get_consumed(schedule_id) + store.save_consumed( + schedule_id, max(consumed, now) if consumed is not None else now + ) + return _PROJECT_SCHEDULE(stored.model_dump(mode="json")) + + async def pause_schedule(self, *, schedule_id: str) -> ScheduleResult: + """Pause one schedule (mirror the poll-loop paused branch). + + Sets ``paused``, clears any held candidate, and advances the + consumed watermark to at least now so the paused span is never + backfilled on resume. No history row is written. + """ + store = self._schedule_store() + now = datetime.now(UTC) + schedule = store.get_schedule(schedule_id) + schedule.paused = True + store.save_schedule(schedule) + store.save_candidate(None, schedule_id=schedule_id) + consumed = store.get_consumed(schedule_id) + store.save_consumed( + schedule_id, max(consumed, now) if consumed is not None else now + ) + return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + + async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult: + """Resume one schedule (mirror ``Scheduler.resume_schedule``). + + Clears ``paused`` (resume selects the next future occurrence), + clears any held candidate, and advances the consumed watermark to + at least now. No history row is written. + """ + store = self._schedule_store() + now = datetime.now(UTC) + schedule = store.get_schedule(schedule_id) + schedule.paused = False + store.save_schedule(schedule) + consumed = store.get_consumed(schedule_id) + store.save_consumed( + schedule_id, max(consumed, now) if consumed is not None else now + ) + store.save_candidate(None, schedule_id=schedule_id) + return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + + async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult: + """Soft-delete one schedule (mirror the poll-loop deleted branch). + + Sets ``deleted`` and clears any held candidate. Runs and history + are untouched, the consumed watermark does not advance, and the id + stays reserved (re-creation is rejected by create). + """ + store = self._schedule_store() + schedule = store.get_schedule(schedule_id) + schedule.deleted = True + store.save_schedule(schedule) + store.save_candidate(None, schedule_id=schedule_id) + return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + + async def list_schedule_occurrences( + self, + *, + schedule_id: str, + cursor: str | None = None, + limit: int = 50, + ) -> OccurrencePage: + """Return one occurrence-history page with a live pending synthesis. + + The stored page keeps the schedule-store shape (``occurrences``, + ``total``, ``cursor``, ``next_cursor``, ``limit``). When a + candidate is currently held AND this is the first page (``cursor`` + is ``None`` or the legacy first-page offset ``"0"``), one + synthesized ``pending`` row for that candidate + is prepended: ``occurrence_id`` derives from + ``(schedule_id, intended instant)``, ``resolved_at`` is the + intended instant, ``revision`` is the candidate revision, and + ``reason`` is empty; ``total`` grows by one. Later pages never + carry the row (a repeat fetch after the candidate admits therefore + shows the durable ``admitted`` entry instead of a duplicate + pending projection). + """ + store = self._schedule_store() + # KeyError first: an unknown schedule must not leak pagination. + store.get_schedule(schedule_id) + if limit < 1 or limit > 100: + raise ValueError("limit must be between 1 and 100") + page = store.list_occurrences(schedule_id, cursor=cursor, limit=limit) + candidate = store.get_candidate(schedule_id) + if candidate is not None and cursor in (None, "0"): + now = datetime.now(UTC) + intended = candidate.intended_at + pending = OccurrenceRecord( + schedule_id=schedule_id, + occurrence_id=occurrence_id(schedule_id, intended), + kind="pending", + resolved_at=intended, + run_id=None, + revision=candidate.revision, + reason="", + admitted_at=None, + started_at=None, + checkpoint_id=None, + interval_start=None, + interval_end=None, + interval_count=0, + created_at=now, + ) + page = { + "occurrences": [ + pending.model_dump(mode="json"), + *page["occurrences"], + ], + "total": page["total"] + 1, + "cursor": page["cursor"], + "next_cursor": page["next_cursor"], + "limit": page["limit"], + } + return _PROJECT_OCCURRENCE_PAGE(page) + + +__all__ = ["WorkflowScheduleApi"] diff --git a/src/wf_api/service.py b/src/wf_api/service.py index 30ff67b9..6aaa8afc 100644 --- a/src/wf_api/service.py +++ b/src/wf_api/service.py @@ -39,6 +39,8 @@ from .models import ( ListDeploymentsResult, ListDraftWorkspacesResult, ListRunsResult, + ListSchedulesResult, + OccurrencePage, PatchDraftResult, RawWorkflowPlan, RunResult, @@ -46,6 +48,7 @@ from .models import ( SaveArtifactResult, SavedDraftArtifactResult, SaveDeploymentResult, + ScheduleResult, ValidateArtifactPlanResult, ValidateDeploymentResult, ValidateDraftResult, @@ -54,6 +57,7 @@ from .models import ( ) from .operation_context import WorkflowOperationContext from .runs import TraceRangeLike, WorkflowRunApi +from .schedules import WorkflowScheduleApi def _authoring_schema( @@ -105,6 +109,7 @@ class WorkflowApi: context: WorkflowOperationContext, *, drafts: bool = False, + schedule_store: Any | None = None, ) -> None: self.context = context self.capabilities = WorkflowCapabilityApi(context, drafts=drafts) @@ -120,6 +125,10 @@ class WorkflowApi: self.artifacts = WorkflowArtifactApi(context, drafts=drafts) self.deployments = WorkflowDeploymentApi(context) self.runs = WorkflowRunApi(context) + # Unlike drafts, the schedule sub-API is always constructed: it + # raises KeyError only when the store is absent, exactly like runs + # with a missing store. An explicit store wins over the context. + self.schedules = WorkflowScheduleApi(context, schedule_store=schedule_store) def _require_drafts(self) -> WorkflowDraftApi: """Return the draft service for an explicitly draft-enabled API.""" @@ -1089,3 +1098,109 @@ class WorkflowApi: run_id=run_id, trace_range=trace_range, ) + + # -- schedules -- + + async def create_schedule( + self, + *, + schedule_id: str, + deployment_id: str, + trigger: dict[str, Any], + input_bindings: list[dict[str, Any]] | None = None, + overlap: str = "skip", + misfire: str = "skip", + max_active_runs: int = 1, + lateness_allowance_s: float = 60.0, + max_steps: int | None = None, + enabled: bool = True, + ) -> ScheduleResult: + return await self.schedules.create_schedule( + schedule_id=schedule_id, + deployment_id=deployment_id, + trigger=trigger, + input_bindings=input_bindings, + overlap=overlap, + misfire=misfire, + max_active_runs=max_active_runs, + lateness_allowance_s=lateness_allowance_s, + max_steps=max_steps, + enabled=enabled, + ) + + async def get_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: + return await self.schedules.get_schedule(schedule_id=schedule_id) + + async def list_schedules( + self, + *, + include_deleted: bool = False, + ) -> ListSchedulesResult: + return await self.schedules.list_schedules(include_deleted=include_deleted) + + async def update_schedule( + self, + *, + schedule_id: str, + expected_revision: int, + deployment_id: str | None = None, + trigger: dict[str, Any] | None = None, + input_bindings: list[dict[str, Any]] | None = None, + overlap: str | None = None, + misfire: str | None = None, + max_active_runs: int | None = None, + lateness_allowance_s: float | None = None, + max_steps: int | None = None, + enabled: bool | None = None, + ) -> ScheduleResult: + return await self.schedules.update_schedule( + schedule_id=schedule_id, + expected_revision=expected_revision, + deployment_id=deployment_id, + trigger=trigger, + input_bindings=input_bindings, + overlap=overlap, + misfire=misfire, + max_active_runs=max_active_runs, + lateness_allowance_s=lateness_allowance_s, + max_steps=max_steps, + enabled=enabled, + ) + + async def pause_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: + return await self.schedules.pause_schedule(schedule_id=schedule_id) + + async def resume_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: + return await self.schedules.resume_schedule(schedule_id=schedule_id) + + async def delete_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: + return await self.schedules.delete_schedule(schedule_id=schedule_id) + + async def list_schedule_occurrences( + self, + *, + schedule_id: str, + cursor: str | None = None, + limit: int = 50, + ) -> OccurrencePage: + return await self.schedules.list_schedule_occurrences( + schedule_id=schedule_id, + cursor=cursor, + limit=limit, + ) diff --git a/src/wf_api/surface.py b/src/wf_api/surface.py index d8c3a12f..708783f5 100644 --- a/src/wf_api/surface.py +++ b/src/wf_api/surface.py @@ -36,7 +36,9 @@ from .models import ( ListDraftWorkspacesResult, ListRegistryEntriesResult, ListRunsResult, + ListSchedulesResult, ListSourcesResult, + OccurrencePage, PatchDraftResult, RegistryEntryMutationResult, RemoveRegistryEntryResult, @@ -44,6 +46,7 @@ from .models import ( RunTraceResult, SaveArtifactResult, SaveDeploymentResult, + ScheduleResult, SourceDiagnosisResult, ValidateArtifactPlanResult, ValidateDeploymentResult, @@ -543,12 +546,91 @@ class WorkflowRunSurface(Protocol): ) -> RunTraceResult: ... +class WorkflowScheduleSurface(Protocol): + """Schedule administration methods exposed by workflow frontends. + + Same-process admin surface: no ownership requirement (the service + observes fresh store reads; second-process schedulers remain + unsupported per the store boundary). + """ + + async def create_schedule( + self, + *, + schedule_id: str, + deployment_id: str, + trigger: dict[str, Any], + input_bindings: list[dict[str, Any]] | None = None, + overlap: str = "skip", + misfire: str = "skip", + max_active_runs: int = 1, + lateness_allowance_s: float = 60.0, + max_steps: int | None = None, + enabled: bool = True, + ) -> ScheduleResult: ... + + async def get_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: ... + + async def list_schedules( + self, + *, + include_deleted: bool = False, + ) -> ListSchedulesResult: ... + + async def update_schedule( + self, + *, + schedule_id: str, + expected_revision: int, + deployment_id: str | None = None, + trigger: dict[str, Any] | None = None, + input_bindings: list[dict[str, Any]] | None = None, + overlap: str | None = None, + misfire: str | None = None, + max_active_runs: int | None = None, + lateness_allowance_s: float | None = None, + max_steps: int | None = None, + enabled: bool | None = None, + ) -> ScheduleResult: ... + + async def pause_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: ... + + async def resume_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: ... + + async def delete_schedule( + self, + *, + schedule_id: str, + ) -> ScheduleResult: ... + + async def list_schedule_occurrences( + self, + *, + schedule_id: str, + cursor: str | None = None, + limit: int = 50, + ) -> OccurrencePage: ... + + class WorkflowApiSurface( WorkflowCapabilitySurface, WorkflowDraftSurface, WorkflowArtifactSurface, WorkflowDeploymentSurface, WorkflowRunSurface, + WorkflowScheduleSurface, Protocol, ): """Public workflow operation surface shared by local and remote adapters.""" @@ -660,6 +742,7 @@ __all__ = [ "WorkflowDeploymentSurface", "WorkflowDraftSurface", "WorkflowRunSurface", + "WorkflowScheduleSurface", "WorkflowSourceAdminSurface", "WorkflowSourceRegistrySurface", ] diff --git a/src/wf_server/context.py b/src/wf_server/context.py index 486a0559..4e4cf484 100644 --- a/src/wf_server/context.py +++ b/src/wf_server/context.py @@ -299,8 +299,14 @@ def build_local_static_workflow_server( *, extra_sources: Mapping[str, CapabilitySource] | None = None, drafts: bool = False, + schedules: bool = False, ) -> WorkflowServer: - """Build a durable local/static server, with drafts as an explicit opt-in.""" + """Build a durable local/static server, with drafts as an explicit opt-in. + + ``schedules=True`` additionally constructs a file schedule store (which + creates its directory) and enables the T13 schedule administration + surface; the default ``False`` keeps zero behavior change. + """ config = WorkflowServerConfig(store_root=Path(root)) stores = file_workflow_stores(config.store_root, drafts=drafts) events = InMemoryWorkflowEventRecorder() @@ -324,7 +330,12 @@ def build_local_static_workflow_server( runtime=runtime, live_sources=None, ) - api = durable_workflow_api(context, drafts=drafts) + schedule_store = None + if schedules: + from wf_scheduling.store import FileScheduleStore + + schedule_store = FileScheduleStore(config.store_root) + api = durable_workflow_api(context, drafts=drafts, schedule_store=schedule_store) source_admin = WorkflowSourceAdminApi(context) admin = WorkflowAdminApi( connections=EmptyWorkflowConnectionProvider(), diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py new file mode 100644 index 00000000..32b8212e --- /dev/null +++ b/tests/wf_api/test_schedules.py @@ -0,0 +1,832 @@ +"""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.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 + + +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, _, _, _, _ = _harness(tmp_path / "stale") + + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + with pytest.raises(StaleScheduleRevisionError): + await api.update_schedule(schedule_id="s", expected_revision=99) + + +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") + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + assert sched_store.get_consumed("s") is None + + 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_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