sched: schedule administration API, admitted inspection, occurrence pages (T13)

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",

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"]

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",

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

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__ = [

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,

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"]

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,
+        )

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",
 ]

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(),

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