From 2c84aba96be06c624f681b963965a688d2049bf7 Mon Sep 17 00:00:00 2001 From: lda Date: Wed, 9 Sep 2026 11:49:35 +0700 Subject: [PATCH] sched: schedule RPC methods, transport/client decoding, round trips (T13) --- src/wf_client/__init__.py | 2 + src/wf_client/_http_port.py | 115 +++++ src/wf_client/app.py | 153 ++++++ src/wf_client/codec.py | 30 ++ src/wf_client/protocols.py | 72 +++ src/wf_client/schedules.py | 152 ++++++ src/wf_transport_rpc_http/__init__.py | 16 + src/wf_transport_rpc_http/app.py | 2 + src/wf_transport_rpc_http/client/__init__.py | 2 + src/wf_transport_rpc_http/client/schedules.py | 147 ++++++ .../methods/schedules.py | 138 +++++ src/wf_transport_rpc_http/models.py | 60 +++ tests/wf_client/conftest.py | 24 + tests/wf_client/test_schedules.py | 262 ++++++++++ .../test_openrpc_contract.py | 65 +++ .../test_schedules_rpc.py | 477 ++++++++++++++++++ 16 files changed, 1717 insertions(+) create mode 100644 src/wf_client/schedules.py create mode 100644 src/wf_transport_rpc_http/client/schedules.py create mode 100644 src/wf_transport_rpc_http/methods/schedules.py create mode 100644 tests/wf_client/test_schedules.py create mode 100644 tests/wf_transport_rpc_http/test_schedules_rpc.py diff --git a/src/wf_client/__init__.py b/src/wf_client/__init__.py index 04466a15..9f957063 100644 --- a/src/wf_client/__init__.py +++ b/src/wf_client/__init__.py @@ -21,6 +21,7 @@ from .errors import ( WorkflowClientError, ) from .runs import Run, TracePage +from .schedules import Schedule from .workflows import ( ArtifactRef, Diagnostic, @@ -52,6 +53,7 @@ __all__ = [ "Run", "RunSummary", "RevisionConflict", + "Schedule", "TransportError", "ValidationFailed", "WorkflowClientError", diff --git a/src/wf_client/_http_port.py b/src/wf_client/_http_port.py index 6c12e38c..73f80bfa 100644 --- a/src/wf_client/_http_port.py +++ b/src/wf_client/_http_port.py @@ -16,10 +16,13 @@ from wf_api.models import ( ListCapabilitiesResult, ListDeploymentsResult, ListRunsResult, + ListSchedulesResult, + OccurrencePage, RunResult, RunTraceResult, SaveArtifactResult, SaveDeploymentResult, + ScheduleResult, ValidateArtifactPlanResult, ValidateDeploymentResult, WorkflowArtifactPayload, @@ -318,3 +321,115 @@ class PublicErrorWorkflowClientPort: run_id=run_id, trace_range=trace_range, ) + + 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._invoke( + "workflow.schedules.create", + self._rpc.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._invoke( + "workflow.schedules.get", + self._rpc.get_schedule, + schedule_id=schedule_id, + ) + + async def list_schedules( + self, *, include_deleted: bool = False + ) -> ListSchedulesResult: + return await self._invoke( + "workflow.schedules.list", + self._rpc.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._invoke( + "workflow.schedules.update", + self._rpc.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._invoke( + "workflow.schedules.pause", + self._rpc.pause_schedule, + schedule_id=schedule_id, + ) + + async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult: + return await self._invoke( + "workflow.schedules.resume", + self._rpc.resume_schedule, + schedule_id=schedule_id, + ) + + async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult: + return await self._invoke( + "workflow.schedules.delete", + self._rpc.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._invoke( + "workflow.schedules.occurrences.list", + self._rpc.list_schedule_occurrences, + schedule_id=schedule_id, + cursor=cursor, + limit=limit, + ) diff --git a/src/wf_client/app.py b/src/wf_client/app.py index c34d8593..a0eacdb5 100644 --- a/src/wf_client/app.py +++ b/src/wf_client/app.py @@ -18,7 +18,9 @@ from .codec import ( decode_capabilities_page, decode_capability_inspect, decode_deployments, + decode_occurrence_page, decode_runs_page, + decode_schedule_list, decode_workflow_artifact, ) from .discovery import ArtifactSummary, DeploymentSummary, RunSummary @@ -27,8 +29,11 @@ from .protocols import WorkflowClientPort from .workflows import WorkflowArtifact if TYPE_CHECKING: + from wf_api.models import OccurrencePage + from .deployments import Deployment from .runs import Run + from .schedules import Schedule def _capability_ref(qualified_name: str, source_id: str) -> CapabilityRef: @@ -298,3 +303,151 @@ class App: expected_run_id=run_id, operation="workflow.runs.inspect", ) + + 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, + ) -> Schedule: + """Create one schedule and return its immutable snapshot.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.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, + ), + expected_schedule_id=schedule_id, + operation="workflow.schedules.create", + ) + + async def schedule(self, schedule_id: str) -> Schedule: + """Inspect and reconstruct one immutable schedule snapshot.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.get_schedule(schedule_id=schedule_id), + expected_schedule_id=schedule_id, + operation="workflow.schedules.get", + ) + + async def schedules(self, *, include_deleted: bool = False) -> tuple[Schedule, ...]: + """List schedule snapshots ordered by id (deleted excluded by default).""" + from .schedules import Schedule + + wire = decode_schedule_list( + await self._port.list_schedules(include_deleted=include_deleted) + ) + return tuple( + Schedule.from_payload( + self._port, + row, + operation="workflow.schedules.list", + ) + for row in wire["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, + ) -> Schedule: + """Apply a revision-checked schedule edit; ``None`` leaves a field unpatched.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.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, + ), + expected_schedule_id=schedule_id, + operation="workflow.schedules.update", + ) + + async def pause_schedule(self, schedule_id: str) -> Schedule: + """Pause one schedule and return its new snapshot.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.pause_schedule(schedule_id=schedule_id), + expected_schedule_id=schedule_id, + operation="workflow.schedules.pause", + ) + + async def resume_schedule(self, schedule_id: str) -> Schedule: + """Resume one schedule and return its new snapshot.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.resume_schedule(schedule_id=schedule_id), + expected_schedule_id=schedule_id, + operation="workflow.schedules.resume", + ) + + async def delete_schedule(self, schedule_id: str) -> Schedule: + """Soft-delete one schedule and return its new snapshot.""" + from .schedules import Schedule + + return Schedule.from_payload( + self._port, + await self._port.delete_schedule(schedule_id=schedule_id), + expected_schedule_id=schedule_id, + operation="workflow.schedules.delete", + ) + + async def schedule_occurrences( + self, + schedule_id: str, + *, + cursor: str | None = None, + limit: int = 50, + ) -> OccurrencePage: + """Return one validated occurrence-history page for a schedule.""" + return decode_occurrence_page( + await self._port.list_schedule_occurrences( + schedule_id=schedule_id, + cursor=cursor, + limit=limit, + ) + ) diff --git a/src/wf_client/codec.py b/src/wf_client/codec.py index 9e52eafd..3441c8f5 100644 --- a/src/wf_client/codec.py +++ b/src/wf_client/codec.py @@ -16,11 +16,14 @@ from wf_api.models import ( ListCapabilitiesResult, ListDeploymentsResult, ListRunsResult, + ListSchedulesResult, + OccurrencePage, RawWorkflowPlan, RunResult, RunTraceResult, SaveArtifactResult, SaveDeploymentResult, + ScheduleResult, ValidateArtifactPlanResult, ValidateDeploymentResult, WorkflowArtifactPayload, @@ -304,3 +307,30 @@ def decode_trace_result(payload: object) -> DecodedTracePage: return DecodedTracePage( *(getattr(fields, field.name) for field in dataclass_fields(_DecodedRunFields)) ) + + +def decode_schedule_result( + payload: object, + *, + operation: str = "workflow.schedules.get", +) -> ScheduleResult: + """Validate one schedule definition response at the client boundary.""" + return _validate(payload, ScheduleResult, operation) + + +def decode_schedule_list( + payload: object, + *, + operation: str = "workflow.schedules.list", +) -> ListSchedulesResult: + """Validate one schedule listing response at the client boundary.""" + return _validate(payload, ListSchedulesResult, operation) + + +def decode_occurrence_page( + payload: object, + *, + operation: str = "workflow.schedules.occurrences.list", +) -> OccurrencePage: + """Validate one occurrence-history page at the client boundary.""" + return _validate(payload, OccurrencePage, operation) diff --git a/src/wf_client/protocols.py b/src/wf_client/protocols.py index f90f39e9..2dd97c18 100644 --- a/src/wf_client/protocols.py +++ b/src/wf_client/protocols.py @@ -12,10 +12,13 @@ from wf_api.models import ( ListCapabilitiesResult, ListDeploymentsResult, ListRunsResult, + ListSchedulesResult, + OccurrencePage, RunResult, RunTraceResult, SaveArtifactResult, SaveDeploymentResult, + ScheduleResult, ValidateArtifactPlanResult, ValidateDeploymentResult, WorkflowArtifactPayload, @@ -149,3 +152,72 @@ class WorkflowClientPort(Protocol): run_id: str, trace_range: TraceRangeLike, ) -> RunTraceResult: ... + + 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: ... diff --git a/src/wf_client/schedules.py b/src/wf_client/schedules.py new file mode 100644 index 00000000..47bb2c1c --- /dev/null +++ b/src/wf_client/schedules.py @@ -0,0 +1,152 @@ +"""Immutable snapshots for schedule administration definitions.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any + +from ._identity import require_response_identity +from ._repr import html_repr, short_repr +from .codec import decode_schedule_result +from .protocols import WorkflowClientPort + + +@dataclass(frozen=True, slots=True, init=False) +class Schedule: + """Immutable client snapshot of one schedule definition. + + The snapshot keeps the validated wire values (trigger and JSON data + bindings are plain data, never transport DTO instances) and reloads + through ``refresh``. State transitions (update/pause/resume/delete) + stay on the ``App`` facade so this snapshot remains a minimal, + read-plus-refresh mirror of the ``Run`` surface. + """ + + _port: WorkflowClientPort = field(repr=False, compare=False) + schedule_id: str + deployment_id: str + _trigger: dict[str, Any] = field(repr=False) + _input_bindings: list[dict[str, Any]] = field(repr=False) + revision: int + enabled: bool + paused: bool + deleted: bool + overlap: str + misfire: str + max_active_runs: int + lateness_allowance_s: float + max_steps: int | None + + def __init__( + self, + *, + _port: WorkflowClientPort, + schedule_id: str, + deployment_id: str, + trigger: dict[str, Any], + input_bindings: list[dict[str, Any]], + revision: int, + enabled: bool, + paused: bool, + deleted: bool, + overlap: str, + misfire: str, + max_active_runs: int, + lateness_allowance_s: float, + max_steps: int | None, + ) -> None: + object.__setattr__(self, "_port", _port) + object.__setattr__(self, "schedule_id", schedule_id) + object.__setattr__(self, "deployment_id", deployment_id) + object.__setattr__(self, "_trigger", deepcopy(trigger)) + object.__setattr__(self, "_input_bindings", deepcopy(input_bindings)) + object.__setattr__(self, "revision", revision) + object.__setattr__(self, "enabled", enabled) + object.__setattr__(self, "paused", paused) + object.__setattr__(self, "deleted", deleted) + object.__setattr__(self, "overlap", overlap) + object.__setattr__(self, "misfire", misfire) + object.__setattr__(self, "max_active_runs", max_active_runs) + object.__setattr__(self, "lateness_allowance_s", lateness_allowance_s) + object.__setattr__(self, "max_steps", max_steps) + + @property + def id(self) -> str: + """Return the schedule id (alias matching the wire ``id`` field).""" + return self.schedule_id + + @property + def trigger(self) -> dict[str, Any]: + """Return a defensive copy of the schedule trigger definition.""" + return deepcopy(self._trigger) + + @property + def input_bindings(self) -> list[dict[str, Any]]: + """Return a defensive copy of the schedule JSON data bindings.""" + return deepcopy(self._input_bindings) + + def __repr__(self) -> str: + return short_repr( + type(self).__name__, + schedule_id=self.schedule_id, + deployment_id=self.deployment_id, + revision=self.revision, + enabled=self.enabled, + paused=self.paused, + deleted=self.deleted, + ) + + def _repr_html_(self) -> str: + return html_repr( + type(self).__name__, + schedule_id=self.schedule_id, + deployment_id=self.deployment_id, + revision=self.revision, + enabled=self.enabled, + paused=self.paused, + deleted=self.deleted, + ) + + @classmethod + def from_payload( + cls, + port: WorkflowClientPort, + payload: object, + *, + expected_schedule_id: str | None = None, + operation: str = "workflow.schedules.get", + ) -> Schedule: + """Validate one schedule response and reconstruct its snapshot.""" + wire = decode_schedule_result(payload, operation=operation) + if expected_schedule_id is not None: + require_response_identity( + operation=operation, + actual={"schedule_id": wire["id"]}, + expected={"schedule_id": expected_schedule_id}, + ) + return cls( + _port=port, + schedule_id=wire["id"], + deployment_id=wire["deployment_id"], + trigger=dict(wire["trigger"]), + input_bindings=[dict(binding) for binding in wire["input_bindings"]], + revision=wire["revision"], + enabled=wire["enabled"], + paused=wire["paused"], + deleted=wire["deleted"], + overlap=wire["overlap"], + misfire=wire["misfire"], + max_active_runs=wire["max_active_runs"], + lateness_allowance_s=wire["lateness_allowance_s"], + max_steps=wire["max_steps"], + ) + + async def refresh(self) -> Schedule: + """Read the current server snapshot without mutating this schedule.""" + return self.from_payload( + self._port, + await self._port.get_schedule(schedule_id=self.schedule_id), + expected_schedule_id=self.schedule_id, + operation="workflow.schedules.get", + ) diff --git a/src/wf_transport_rpc_http/__init__.py b/src/wf_transport_rpc_http/__init__.py index 486b5f7a..d892dc71 100644 --- a/src/wf_transport_rpc_http/__init__.py +++ b/src/wf_transport_rpc_http/__init__.py @@ -15,9 +15,12 @@ from .models import ( CreateArtifactFromWorkspaceParams, CreateDraftFromCapabilityParams, CreateEmptyDraftWorkspaceParams, + CreateScheduleParams, CreateWrapperFromWorkspaceParams, DeleteDeploymentParams, + DeleteScheduleParams, GetDraftWorkspaceParams, + GetScheduleParams, HandleDraftBranch, HandleDraftParams, HealthParams, @@ -30,15 +33,19 @@ from .models import ( ListCapabilitiesParams, ListDeploymentsParams, ListDraftWorkspacesParams, + ListOccurrencesParams, + ListSchedulesParams, ListSourcesParams, PatchDraftParams, PatchDraftWorkspaceParams, + PauseScheduleParams, ReadRunTraceParams, RemoveDraftBindingParams, RemoveDraftRouteParams, RemoveDraftStepParams, ReplaceDraftWorkspaceDocumentParams, ResumeRunParams, + ResumeScheduleParams, RouteSourceParams, SaveArtifactParams, SaveDeploymentParams, @@ -55,6 +62,7 @@ from .models import ( StartRunParams, TraceRangeParams, UpdateCapabilityStepParams, + UpdateScheduleParams, ValidateDeploymentParams, ValidateDraftParams, ValidateDraftWorkspaceParams, @@ -72,9 +80,12 @@ __all__ = [ "CreateArtifactFromWorkspaceParams", "CreateDraftFromCapabilityParams", "CreateEmptyDraftWorkspaceParams", + "CreateScheduleParams", "CreateWrapperFromWorkspaceParams", "DeleteDeploymentParams", + "DeleteScheduleParams", "GetDraftWorkspaceParams", + "GetScheduleParams", "HandleDraftBranch", "HandleDraftParams", "HealthParams", @@ -87,15 +98,19 @@ __all__ = [ "ListCapabilitiesParams", "ListDeploymentsParams", "ListDraftWorkspacesParams", + "ListOccurrencesParams", + "ListSchedulesParams", "ListSourcesParams", "PatchDraftParams", "PatchDraftWorkspaceParams", + "PauseScheduleParams", "ReadRunTraceParams", "RemoveDraftBindingParams", "RemoveDraftRouteParams", "RemoveDraftStepParams", "ReplaceDraftWorkspaceDocumentParams", "ResumeRunParams", + "ResumeScheduleParams", "RouteSourceParams", "RpcWorkflowApiClient", "SaveArtifactParams", @@ -113,6 +128,7 @@ __all__ = [ "StartRunParams", "TraceRangeParams", "UpdateCapabilityStepParams", + "UpdateScheduleParams", "ValidateDeploymentParams", "ValidateDraftParams", "ValidateDraftWorkspaceParams", diff --git a/src/wf_transport_rpc_http/app.py b/src/wf_transport_rpc_http/app.py index 26335011..6324f17a 100644 --- a/src/wf_transport_rpc_http/app.py +++ b/src/wf_transport_rpc_http/app.py @@ -20,6 +20,7 @@ from .methods.capabilities import ( from .methods.deployments import register_methods as register_deployment_methods from .methods.drafts import register_methods as register_draft_methods from .methods.runs import register_methods as register_run_methods +from .methods.schedules import register_methods as register_schedule_methods from .methods.source_registry import ( register_methods as register_source_registry_methods, ) @@ -65,6 +66,7 @@ def create_rpc_app( register_artifact_methods(entrypoint, server) register_deployment_methods(entrypoint, server) register_run_methods(entrypoint, server) + register_schedule_methods(entrypoint, server) register_source_methods(entrypoint, server) register_source_registry_methods(entrypoint, server) register_admin_methods(entrypoint, server) diff --git a/src/wf_transport_rpc_http/client/__init__.py b/src/wf_transport_rpc_http/client/__init__.py index d1e3fb4a..61ffac37 100644 --- a/src/wf_transport_rpc_http/client/__init__.py +++ b/src/wf_transport_rpc_http/client/__init__.py @@ -9,6 +9,7 @@ from .capabilities import RpcCapabilityClientMixin from .deployments import RpcDeploymentClientMixin from .drafts import RpcDraftClientMixin from .runs import RpcRunClientMixin +from .schedules import RpcScheduleClientMixin from .source_registry import RpcSourceRegistryClientMixin from .sources import RpcSourceAdminClientMixin @@ -21,6 +22,7 @@ class RpcWorkflowApiClient( RpcArtifactClientMixin, RpcDeploymentClientMixin, RpcRunClientMixin, + RpcScheduleClientMixin, RpcSourceAdminClientMixin, RpcSourceRegistryClientMixin, RpcAdminClientMixin, diff --git a/src/wf_transport_rpc_http/client/schedules.py b/src/wf_transport_rpc_http/client/schedules.py new file mode 100644 index 00000000..7703f6b3 --- /dev/null +++ b/src/wf_transport_rpc_http/client/schedules.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +from typing import Any, cast + +from wf_api.models import ListSchedulesResult, OccurrencePage, ScheduleResult + +from .base import RpcCaller + + +class RpcScheduleClientMixin: + """JSON-RPC implementation of workflow schedule administration methods.""" + + async def create_schedule( + self: RpcCaller, + *, + 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 cast( + ScheduleResult, + await self._call( + "workflow.schedules.create", + { + "schedule_id": schedule_id, + "deployment_id": deployment_id, + "trigger": trigger, + "input_bindings": ( + list(input_bindings) if input_bindings is not None else [] + ), + "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: RpcCaller, *, schedule_id: str) -> ScheduleResult: + return cast( + ScheduleResult, + await self._call( + "workflow.schedules.get", + {"schedule_id": schedule_id}, + ), + ) + + async def list_schedules( + self: RpcCaller, *, include_deleted: bool = False + ) -> ListSchedulesResult: + return cast( + ListSchedulesResult, + await self._call( + "workflow.schedules.list", + {"include_deleted": include_deleted}, + ), + ) + + async def update_schedule( + self: RpcCaller, + *, + 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 cast( + ScheduleResult, + await self._call( + "workflow.schedules.update", + { + "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: RpcCaller, *, schedule_id: str) -> ScheduleResult: + return cast( + ScheduleResult, + await self._call( + "workflow.schedules.pause", + {"schedule_id": schedule_id}, + ), + ) + + async def resume_schedule(self: RpcCaller, *, schedule_id: str) -> ScheduleResult: + return cast( + ScheduleResult, + await self._call( + "workflow.schedules.resume", + {"schedule_id": schedule_id}, + ), + ) + + async def delete_schedule(self: RpcCaller, *, schedule_id: str) -> ScheduleResult: + return cast( + ScheduleResult, + await self._call( + "workflow.schedules.delete", + {"schedule_id": schedule_id}, + ), + ) + + async def list_schedule_occurrences( + self: RpcCaller, + *, + schedule_id: str, + cursor: str | None = None, + limit: int = 50, + ) -> OccurrencePage: + return cast( + OccurrencePage, + await self._call( + "workflow.schedules.occurrences.list", + { + "schedule_id": schedule_id, + "cursor": cursor, + "limit": limit, + }, + ), + ) diff --git a/src/wf_transport_rpc_http/methods/schedules.py b/src/wf_transport_rpc_http/methods/schedules.py new file mode 100644 index 00000000..95050976 --- /dev/null +++ b/src/wf_transport_rpc_http/methods/schedules.py @@ -0,0 +1,138 @@ +"""Schedule JSON-RPC method registration. + +Return annotations stay eagerly evaluated because fastapi-jsonrpc captures them +while registering nested handlers for response validation and OpenRPC output. +""" + +import fastapi_jsonrpc as jsonrpc + +from wf_api.models import ListSchedulesResult, OccurrencePage, ScheduleResult +from wf_server import WorkflowServer + +from ..errors import WorkflowRpcError, raise_workflow_rpc_error +from ..models import ( + CreateScheduleParams, + DeleteScheduleParams, + GetScheduleParams, + ListOccurrencesParams, + ListSchedulesParams, + PauseScheduleParams, + ResumeScheduleParams, + UpdateScheduleParams, +) +from ..params import RpcParams + + +def register_methods( + entrypoint: jsonrpc.Entrypoint, + server: WorkflowServer, +) -> None: + """Register schedule administration JSON-RPC methods. + + No composition-time gate is needed here (unlike drafts): a store-less + server raises ``KeyError`` at call time, which maps to a workflow RPC + error exactly like any other unknown-schedule failure. + """ + + @entrypoint.method(name="workflow.schedules.create", errors=[WorkflowRpcError]) + async def workflow_schedules_create( + params: CreateScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.create_schedule( + schedule_id=params.schedule_id, + deployment_id=params.deployment_id, + trigger=params.trigger, + input_bindings=params.input_bindings, + overlap=params.overlap, + misfire=params.misfire, + max_active_runs=params.max_active_runs, + lateness_allowance_s=params.lateness_allowance_s, + max_steps=params.max_steps, + enabled=params.enabled, + ) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.get", errors=[WorkflowRpcError]) + async def workflow_schedules_get( + params: GetScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.get_schedule(schedule_id=params.schedule_id) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.list", errors=[WorkflowRpcError]) + async def workflow_schedules_list( + params: ListSchedulesParams = RpcParams(), + ) -> ListSchedulesResult: + try: + return await server.api.list_schedules( + include_deleted=params.include_deleted, + ) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.update", errors=[WorkflowRpcError]) + async def workflow_schedules_update( + params: UpdateScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.update_schedule( + schedule_id=params.schedule_id, + expected_revision=params.expected_revision, + deployment_id=params.deployment_id, + trigger=params.trigger, + input_bindings=params.input_bindings, + overlap=params.overlap, + misfire=params.misfire, + max_active_runs=params.max_active_runs, + lateness_allowance_s=params.lateness_allowance_s, + max_steps=params.max_steps, + enabled=params.enabled, + ) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.pause", errors=[WorkflowRpcError]) + async def workflow_schedules_pause( + params: PauseScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.pause_schedule(schedule_id=params.schedule_id) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.resume", errors=[WorkflowRpcError]) + async def workflow_schedules_resume( + params: ResumeScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.resume_schedule(schedule_id=params.schedule_id) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method(name="workflow.schedules.delete", errors=[WorkflowRpcError]) + async def workflow_schedules_delete( + params: DeleteScheduleParams = RpcParams(), + ) -> ScheduleResult: + try: + return await server.api.delete_schedule(schedule_id=params.schedule_id) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) + + @entrypoint.method( + name="workflow.schedules.occurrences.list", errors=[WorkflowRpcError] + ) + async def workflow_schedules_occurrences_list( + params: ListOccurrencesParams = RpcParams(), + ) -> OccurrencePage: + try: + return await server.api.list_schedule_occurrences( + schedule_id=params.schedule_id, + cursor=params.cursor, + limit=params.limit, + ) + except (ValueError, KeyError, LookupError, FileNotFoundError) as exc: + raise_workflow_rpc_error(exc) diff --git a/src/wf_transport_rpc_http/models.py b/src/wf_transport_rpc_http/models.py index fb623e84..8c1fe45a 100644 --- a/src/wf_transport_rpc_http/models.py +++ b/src/wf_transport_rpc_http/models.py @@ -462,6 +462,66 @@ class ResumeRunParams(RpcParamsModel): trace_range: TraceRangeParams | None = None +class CreateScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + deployment_id: str = Field(min_length=1) + trigger: dict[str, Any] + input_bindings: list[dict[str, Any]] = Field(default_factory=list) + overlap: str = "skip" + misfire: str = "skip" + max_active_runs: int = Field(default=1, ge=1) + lateness_allowance_s: float = Field(default=60.0, ge=0) + max_steps: int | None = Field( + default=None, + ge=1, + strict=True, + description=( + "Optional schedule step budget. The server default applies when omitted." + ), + ) + enabled: bool = True + + +class GetScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + + +class ListSchedulesParams(RpcParamsModel): + include_deleted: bool = False + + +class UpdateScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + 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 + + +class PauseScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + + +class ResumeScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + + +class DeleteScheduleParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + + +class ListOccurrencesParams(RpcParamsModel): + schedule_id: str = Field(min_length=1) + cursor: str | None = None + limit: int = Field(default=50, ge=1, le=100) + + class ListRegistryEntriesParams(RpcParamsModel): cursor: str | None = Field(default=None) limit: int = Field(default=50, ge=1, le=100) diff --git a/tests/wf_client/conftest.py b/tests/wf_client/conftest.py index ee1de6a4..a8bb3234 100644 --- a/tests/wf_client/conftest.py +++ b/tests/wf_client/conftest.py @@ -70,6 +70,30 @@ class FakeWorkflowClient: async def read_run_trace(self, **params: Any) -> object: return self._response("workflow.runs.trace", params) + async def create_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.create", params) + + async def get_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.get", params) + + async def list_schedules(self, **params: Any) -> object: + return self._response("workflow.schedules.list", params) + + async def update_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.update", params) + + async def pause_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.pause", params) + + async def resume_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.resume", params) + + async def delete_schedule(self, **params: Any) -> object: + return self._response("workflow.schedules.delete", params) + + async def list_schedule_occurrences(self, **params: Any) -> object: + return self._response("workflow.schedules.occurrences.list", params) + async def _call(self, method: str, params: dict[str, Any]) -> object: return self._response(method, params) diff --git a/tests/wf_client/test_schedules.py b/tests/wf_client/test_schedules.py new file mode 100644 index 00000000..aba43406 --- /dev/null +++ b/tests/wf_client/test_schedules.py @@ -0,0 +1,262 @@ +"""T13 schedule client tests (codec + snapshot + facade accessors).""" + +from __future__ import annotations + +from typing import Any, cast + +import pytest + +from wf_client import App, Schedule +from wf_client.codec import ( + decode_occurrence_page, + decode_schedule_list, + decode_schedule_result, +) +from wf_client.errors import InvalidResponse +from wf_client.protocols import WorkflowClientPort + +from .conftest import FakeWorkflowClient + + +def _schedule_payload(**overrides: Any) -> dict[str, Any]: + payload: dict[str, Any] = { + "id": "s", + "deployment_id": "dep.personal", + "trigger": {"kind": "cron", "expression": "* * * * *", "timezone": "UTC"}, + "input_bindings": [{"target": "msg", "value": "hi"}], + "max_steps": None, + "overlap": "skip", + "misfire": "skip", + "max_active_runs": 1, + "lateness_allowance_s": 60.0, + "revision": 1, + "enabled": True, + "paused": False, + "deleted": False, + "exhausted": False, + "blocked_reason": None, + "created_at": "2026-09-08T12:00:00+00:00", + "updated_at": "2026-09-08T12:00:00+00:00", + } + payload.update(overrides) + return payload + + +def _occurrence_payload(**overrides: Any) -> dict[str, Any]: + payload: dict[str, Any] = { + "occurrences": [ + { + "schedule_id": "s", + "occurrence_id": "s|2026-09-08T12:00:00+00:00", + "kind": "admitted", + "resolved_at": "2026-09-08T12:00:00+00:00", + "run_id": "run-0", + "revision": 1, + "reason": "rev=1", + "admitted_at": "2026-09-08T12:00:00+00:00", + "started_at": None, + "checkpoint_id": None, + "interval_start": None, + "interval_end": None, + "interval_count": 0, + "created_at": "2026-09-08T12:00:00+00:00", + } + ], + "total": 1, + "cursor": None, + "next_cursor": None, + "limit": 50, + } + payload.update(overrides) + return payload + + +class _Port: + def __init__(self, payload: dict[str, Any] | None = None) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + self.payload = payload or _schedule_payload(revision=2) + + async def get_schedule(self, **params: Any) -> object: + self.calls.append(("get_schedule", params)) + return self.payload + + +def test_decode_schedule_result_returns_wire_projection() -> None: + result = decode_schedule_result(_schedule_payload()) + + assert result["id"] == "s" + assert result["deployment_id"] == "dep.personal" + assert result["revision"] == 1 + assert result["enabled"] is True + + +def test_decode_schedule_result_rejects_malformed_payload() -> None: + with pytest.raises(InvalidResponse, match="workflow.schedules.get"): + decode_schedule_result({"id": "s"}) + + +def test_decode_schedule_list_and_occurrence_page() -> None: + listed = decode_schedule_list({"schedules": [_schedule_payload()]}) + page = decode_occurrence_page(_occurrence_payload()) + + assert listed["schedules"][0]["id"] == "s" + assert page["total"] == 1 + assert page["occurrences"][0]["run_id"] == "run-0" + + with pytest.raises(InvalidResponse, match="workflow.schedules.list"): + decode_schedule_list({"schedules": [{"id": "broken"}]}) + with pytest.raises(InvalidResponse, match="workflow.schedules.occurrences.list"): + decode_occurrence_page({"total": "many"}) + + +@pytest.mark.asyncio +async def test_schedule_snapshot_exposes_minimal_surface() -> None: + schedule = Schedule.from_payload( + cast(WorkflowClientPort, _Port()), _schedule_payload() + ) + + assert schedule.schedule_id == "s" + assert schedule.id == "s" + assert schedule.deployment_id == "dep.personal" + assert schedule.trigger == { + "kind": "cron", + "expression": "* * * * *", + "timezone": "UTC", + } + assert schedule.input_bindings == [{"target": "msg", "value": "hi"}] + assert schedule.revision == 1 + assert schedule.enabled is True + assert schedule.paused is False + assert schedule.deleted is False + assert schedule.overlap == "skip" + assert schedule.misfire == "skip" + assert schedule.max_active_runs == 1 + assert schedule.lateness_allowance_s == 60.0 + assert schedule.max_steps is None + + +@pytest.mark.asyncio +async def test_schedule_snapshot_defensively_copies_json_data() -> None: + schedule = Schedule.from_payload( + cast(WorkflowClientPort, _Port()), _schedule_payload() + ) + + exposed_trigger = schedule.trigger + exposed_bindings = schedule.input_bindings + assert isinstance(exposed_trigger, dict) + exposed_trigger["expression"] = "mutated" + exposed_bindings[0]["value"] = "mutated" + exposed_bindings.append({"target": "extra", "value": 1}) + + assert schedule.trigger == { + "kind": "cron", + "expression": "* * * * *", + "timezone": "UTC", + } + assert schedule.input_bindings == [{"target": "msg", "value": "hi"}] + + +@pytest.mark.asyncio +async def test_schedule_rejects_mismatched_identity() -> None: + with pytest.raises(InvalidResponse, match="workflow.schedules.get"): + Schedule.from_payload( + cast(WorkflowClientPort, _Port()), + _schedule_payload(id="other"), + expected_schedule_id="s", + ) + + +@pytest.mark.asyncio +async def test_refresh_returns_a_new_snapshot() -> None: + port = _Port() + original = Schedule.from_payload( + cast(WorkflowClientPort, port), _schedule_payload() + ) + refreshed = await original.refresh() + + assert refreshed is not original + assert refreshed.revision == 2 + assert original.revision == 1 + assert port.calls == [("get_schedule", {"schedule_id": "s"})] + + +@pytest.mark.asyncio +async def test_app_schedule_accessors_round_trip() -> None: + payload = _schedule_payload() + fake = FakeWorkflowClient( + **{ + "workflow.schedules.create": payload, + "workflow.schedules.get": payload, + "workflow.schedules.list": {"schedules": [payload]}, + "workflow.schedules.update": _schedule_payload(revision=2), + "workflow.schedules.pause": _schedule_payload(paused=True), + "workflow.schedules.resume": _schedule_payload(paused=False), + "workflow.schedules.delete": _schedule_payload(deleted=True), + "workflow.schedules.occurrences.list": _occurrence_payload(), + } + ) + app = App._from_port(cast(WorkflowClientPort, fake)) + + created = await app.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger={ + "kind": "cron", + "expression": "* * * * *", + "timezone": "UTC", + }, + ) + fetched = await app.schedule("s") + listed = await app.schedules() + updated = await app.update_schedule( + schedule_id="s", expected_revision=1, max_active_runs=3 + ) + paused = await app.pause_schedule("s") + resumed = await app.resume_schedule("s") + deleted = await app.delete_schedule("s") + page = await app.schedule_occurrences("s", limit=10) + + assert created.schedule_id == "s" + assert created.revision == 1 + assert fetched.revision == 1 + assert [item.schedule_id for item in listed] == ["s"] + assert updated.revision == 2 + assert paused.paused is True + assert resumed.paused is False + assert deleted.deleted is True + assert page["total"] == 1 + assert page["occurrences"][0]["run_id"] == "run-0" + assert fake.calls[0] == ( + "workflow.schedules.create", + { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": { + "kind": "cron", + "expression": "* * * * *", + "timezone": "UTC", + }, + "input_bindings": None, + "overlap": "skip", + "misfire": "skip", + "max_active_runs": 1, + "lateness_allowance_s": 60.0, + "max_steps": None, + "enabled": True, + }, + ) + assert ( + "workflow.schedules.occurrences.list", + {"schedule_id": "s", "cursor": None, "limit": 10}, + ) in fake.calls + + +@pytest.mark.asyncio +async def test_app_schedule_rejects_mismatched_identity() -> None: + fake = FakeWorkflowClient( + **{"workflow.schedules.get": _schedule_payload(id="other")} + ) + app = App._from_port(cast(WorkflowClientPort, fake)) + + with pytest.raises(InvalidResponse, match="workflow.schedules.get"): + await app.schedule("s") diff --git a/tests/wf_transport_rpc_http/test_openrpc_contract.py b/tests/wf_transport_rpc_http/test_openrpc_contract.py index 7124b910..a2d80e8b 100644 --- a/tests/wf_transport_rpc_http/test_openrpc_contract.py +++ b/tests/wf_transport_rpc_http/test_openrpc_contract.py @@ -860,3 +860,68 @@ def test_openrpc_exposes_typed_run_results( component_name=component_name, properties=properties, ) + + +@pytest.mark.parametrize( + ("method_name", "component_name", "properties"), + [ + ( + "workflow.schedules.create", + "ScheduleResult", + { + "id", + "deployment_id", + "trigger", + "revision", + "enabled", + }, + ), + ( + "workflow.schedules.get", + "ScheduleResult", + {"id", "revision", "paused", "deleted"}, + ), + ( + "workflow.schedules.list", + "ListSchedulesResult", + {"schedules"}, + ), + ( + "workflow.schedules.update", + "ScheduleResult", + {"id", "revision", "updated_at"}, + ), + ( + "workflow.schedules.pause", + "ScheduleResult", + {"id", "paused"}, + ), + ( + "workflow.schedules.resume", + "ScheduleResult", + {"id", "paused"}, + ), + ( + "workflow.schedules.delete", + "ScheduleResult", + {"id", "deleted"}, + ), + ( + "workflow.schedules.occurrences.list", + "OccurrencePage", + {"occurrences", "total", "cursor", "next_cursor", "limit"}, + ), + ], +) +def test_openrpc_exposes_typed_schedule_results( + openrpc_document: dict[str, Any], + method_name: str, + component_name: str, + properties: set[str], +) -> None: + _assert_result_component( + openrpc_document, + method_name=method_name, + component_name=component_name, + properties=properties, + ) diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py new file mode 100644 index 00000000..cacae534 --- /dev/null +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -0,0 +1,477 @@ +"""T13 schedule JSON-RPC surface tests (transport + client mixin). + +Uses the green ``httpx2`` ASGI pattern from ``test_client.py`` (the +``httpx``-based ``test_app.py`` cannot even be collected in this env). +History and held-candidate fixtures go through the public file schedule +store at the same root, never through server privates. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +import httpx2 +import pytest +from pydantic import ValidationError + +from wf_api.models import RawWorkflowPlan +from wf_api.surface import WorkflowScheduleSurface +from wf_core import END +from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry +from wf_scheduling.models import PendingCandidate +from wf_scheduling.store import FileScheduleStore +from wf_server import build_local_static_workflow_server +from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app +from wf_transport_rpc_http.client.base import RpcProtocolError +from wf_transport_rpc_http.client.schedules import RpcScheduleClientMixin +from wf_transport_rpc_http.models import ( + CreateScheduleParams, + ListOccurrencesParams, + UpdateScheduleParams, +) + + +def _cron() -> dict[str, Any]: + return {"kind": "cron", "expression": "* * * * *", "timezone": "UTC"} + + +def _constant_plan() -> RawWorkflowPlan: + return RawWorkflowPlan.model_validate( + { + "name": "sched_constant", + "input_schema": {"type": "object", "properties": {}}, + "state_schema": { + "type": "object", + "properties": { + "result": {"type": "string", "reducer": "wf.std.replace"} + }, + }, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + "outcomes": ["ok"], + "start": "constant", + "nodes": [ + { + "id": "constant", + "type": "node", + "node": "wf.std.constant", + "input": [ + { + "value": "hello from schedule", + "target": {"root": "local", "parts": ["value"]}, + } + ], + "output": [ + { + "source": {"root": "local", "parts": ["value"]}, + "target": {"root": "state", "parts": ["result"]}, + } + ], + } + ], + "edges": [{"from": "constant", "outcome": "ok", "to": END}], + "output": [ + { + "path": {"root": "state", "parts": ["result"]}, + "target": {"root": "local", "parts": ["result"]}, + } + ], + } + ) + + +async def _seed_server(tmp_path: Any, name: str = "sched-art") -> Any: + """Build a schedule-enabled server with one artifact + deployment.""" + _ = name + server = build_local_static_workflow_server(tmp_path / "store", schedules=True) + await server.api.create_artifact_from_plan( + artifact_id="sched-art", + version=1, + title="Sched Art", + plan=_constant_plan(), + outcomes=["ok"], + source_bindings={}, + ) + await server.api.save_deployment( + { + "id": "dep.personal", + "artifact_id": "sched-art", + "artifact_version": 1, + "bindings": {}, + } + ) + return server + + +def _client_for(server: Any) -> tuple[RpcWorkflowApiClient, httpx2.AsyncClient]: + """Return an RPC client bound to the server over ASGI transport.""" + app = create_rpc_app(server) + transport = httpx2.ASGITransport(app=app) + http_client = httpx2.AsyncClient(transport=transport, base_url="http://test") + return ( + RpcWorkflowApiClient( + url="http://test/rpc", + timeout_seconds=5, + http_client=http_client, + ), + http_client, + ) + + +async def test_rpc_schedule_crud_lifecycle(tmp_path) -> None: + server = await _seed_server(tmp_path) + client, http_client = _client_for(server) + async with http_client: + created = await client.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 + + fetched = await client.get_schedule(schedule_id="s") + assert fetched["id"] == "s" + assert fetched["revision"] == 1 + + listed = await client.list_schedules() + assert [row["id"] for row in listed["schedules"]] == ["s"] + + updated = await client.update_schedule( + schedule_id="s", + expected_revision=1, + max_active_runs=3, + ) + assert updated["revision"] == 2 + assert updated["max_active_runs"] == 3 + + paused = await client.pause_schedule(schedule_id="s") + assert paused["paused"] is True + + resumed = await client.resume_schedule(schedule_id="s") + assert resumed["paused"] is False + + deleted = await client.delete_schedule(schedule_id="s") + assert deleted["deleted"] is True + assert (await client.list_schedules())["schedules"] == [] + listed_deleted = await client.list_schedules(include_deleted=True) + assert [row["id"] for row in listed_deleted["schedules"]] == ["s"] + + +async def test_rpc_schedule_update_rejects_stale_revision(tmp_path) -> None: + server = await _seed_server(tmp_path) + client, http_client = _client_for(server) + async with http_client: + await client.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger=_cron(), + ) + updated = await client.update_schedule( + schedule_id="s", + expected_revision=1, + overlap="parallel", + ) + assert updated["revision"] == 2 + + with pytest.raises(RpcProtocolError) as raised: + await client.update_schedule( + schedule_id="s", + expected_revision=1, + overlap="skip", + ) + + assert raised.value.code == 5000 + assert isinstance(raised.value.data, dict) + assert raised.value.data["code"] == "StaleScheduleRevisionError" + + +async def test_rpc_schedule_occurrences_pagination_and_pending(tmp_path) -> None: + server = await _seed_server(tmp_path) + client, http_client = _client_for(server) + store = FileScheduleStore(tmp_path / "store") + base = datetime(2026, 9, 8, 12, 0, tzinfo=UTC) + async with http_client: + await client.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger=_cron(), + ) + recorder = FileScheduleHistoryRecorder(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) + store.save_candidate( + PendingCandidate(schedule_id="s", intended_at=intended, revision=1), + schedule_id="s", + ) + + first = await client.list_schedule_occurrences(schedule_id="s", limit=2) + assert first["total"] == 4 + assert first["cursor"] is None + assert first["limit"] == 2 + # The pending synthesis prepends one row to the stored page. + assert len(first["occurrences"]) == 3 + pending = first["occurrences"][0] + assert pending["kind"] == "pending" + assert pending["schedule_id"] == "s" + assert pending["revision"] == 1 + assert pending["run_id"] is None + assert first["occurrences"][1]["kind"] == "admitted" + assert first["next_cursor"] is not None + + # The legacy first-page offset also carries the pending synthesis. + legacy_first = await client.list_schedule_occurrences( + schedule_id="s", cursor="0", limit=2 + ) + assert legacy_first["total"] == 4 + assert legacy_first["occurrences"][0]["kind"] == "pending" + + later = await client.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"]) + + store.save_candidate(None, schedule_id="s") + plain = await client.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_rpc_schedule_error_mapping(tmp_path) -> None: + server = await _seed_server(tmp_path) + client, http_client = _client_for(server) + async with http_client: + with pytest.raises(RpcProtocolError) as unknown: + await client.get_schedule(schedule_id="missing") + + # Trigger shapes fail Schedule model validation inside the service, + # which maps to InvalidParams rather than a workflow error. + with pytest.raises(RpcProtocolError) as bad_trigger: + await client.create_schedule( + schedule_id="bad", + deployment_id="dep.personal", + trigger={"kind": "hourly"}, + ) + + with pytest.raises(RpcProtocolError) as bad_deployment: + await client.create_schedule( + schedule_id="bad-dep", + deployment_id="missing.dep", + trigger=_cron(), + ) + + assert unknown.value.code == 5000 + assert isinstance(unknown.value.data, dict) + assert unknown.value.data["code"] == "ScheduleNotFoundError" + assert bad_trigger.value.code == -32602 + assert bad_deployment.value.code == 5000 + assert isinstance(bad_deployment.value.data, dict) + assert bad_deployment.value.data["code"] == "KeyError" + + +async def test_rpc_schedule_without_store_maps_keyerror(tmp_path) -> None: + # A server built without schedules=True has no schedule store; the call + # must surface as a workflow RPC error, not a new gate or a crash. + server = build_local_static_workflow_server(tmp_path / "store") + client, http_client = _client_for(server) + async with http_client: + with pytest.raises(RpcProtocolError) as raised: + await client.get_schedule(schedule_id="s") + + assert raised.value.code == 5000 + assert isinstance(raised.value.data, dict) + assert raised.value.data["code"] == "KeyError" + + +async def test_rpc_schedule_raw_envelope_reports_workflow_error(tmp_path) -> None: + server = await _seed_server(tmp_path) + app = create_rpc_app(server) + transport = httpx2.ASGITransport(app=app) + async with httpx2.AsyncClient( + transport=transport, base_url="http://test" + ) as http_client: + response = await http_client.post( + "http://test/rpc", + json={ + "jsonrpc": "2.0", + "id": "sched-1", + "method": "workflow.schedules.get", + "params": {"schedule_id": "missing"}, + }, + ) + payload = response.json() + + assert response.status_code == 200 + assert payload["error"]["code"] == 5000 + assert payload["error"]["data"]["code"] == "ScheduleNotFoundError" + + +def test_schedule_params_reject_invalid_envelopes() -> None: + with pytest.raises(ValidationError): + CreateScheduleParams.model_validate( + {"schedule_id": "", "deployment_id": "dep.personal", "trigger": _cron()} + ) + with pytest.raises(ValidationError): + CreateScheduleParams.model_validate( + { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": _cron(), + "max_active_runs": 0, + } + ) + with pytest.raises(ValidationError): + CreateScheduleParams.model_validate( + { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": _cron(), + "max_steps": 0, + } + ) + # Strict budgets reject stringly numbers instead of coercing them. + with pytest.raises(ValidationError): + CreateScheduleParams.model_validate( + { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": _cron(), + "max_steps": "5", + } + ) + with pytest.raises(ValidationError): + ListOccurrencesParams.model_validate({"schedule_id": "s", "limit": 0}) + with pytest.raises(ValidationError): + ListOccurrencesParams.model_validate({"schedule_id": "s", "limit": 101}) + # Misspelled params are rejected early (extra=forbid). + with pytest.raises(ValidationError): + CreateScheduleParams.model_validate( + { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": _cron(), + "schedul_id": "typo", + } + ) + + params = CreateScheduleParams.model_validate( + {"schedule_id": "s", "deployment_id": "dep.personal", "trigger": _cron()} + ) + assert params.input_bindings == [] + assert params.overlap == "skip" + assert params.misfire == "skip" + assert params.max_active_runs == 1 + assert params.lateness_allowance_s == 60.0 + assert params.max_steps is None + assert params.enabled is True + + update = UpdateScheduleParams.model_validate( + {"schedule_id": "s", "expected_revision": 1} + ) + assert update.deployment_id is None + assert update.trigger is None + assert update.input_bindings is None + assert update.max_steps is None + assert update.enabled is None + + occurrences = ListOccurrencesParams.model_validate({"schedule_id": "s"}) + assert occurrences.cursor is None + assert occurrences.limit == 50 + + +async def test_rpc_schedule_client_sends_exact_payloads() -> None: + calls: list[dict[str, Any]] = [] + + class Client(RpcScheduleClientMixin): + async def _call(self, method: str, params: dict[str, object]): + calls.append({"method": method, "params": params}) + return {"id": "s", "revision": 1} + + client = Client() + await client.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger=_cron(), + ) + await client.get_schedule(schedule_id="s") + await client.list_schedules() + await client.update_schedule( + schedule_id="s", expected_revision=1, max_active_runs=3 + ) + await client.pause_schedule(schedule_id="s") + await client.resume_schedule(schedule_id="s") + await client.delete_schedule(schedule_id="s") + await client.list_schedule_occurrences(schedule_id="s", limit=2) + + assert [call["method"] for call in calls] == [ + "workflow.schedules.create", + "workflow.schedules.get", + "workflow.schedules.list", + "workflow.schedules.update", + "workflow.schedules.pause", + "workflow.schedules.resume", + "workflow.schedules.delete", + "workflow.schedules.occurrences.list", + ] + assert calls[0]["params"] == { + "schedule_id": "s", + "deployment_id": "dep.personal", + "trigger": _cron(), + "input_bindings": [], + "overlap": "skip", + "misfire": "skip", + "max_active_runs": 1, + "lateness_allowance_s": 60.0, + "max_steps": None, + "enabled": True, + } + assert calls[3]["params"] == { + "schedule_id": "s", + "expected_revision": 1, + "deployment_id": None, + "trigger": None, + "input_bindings": None, + "overlap": None, + "misfire": None, + "max_active_runs": 3, + "lateness_allowance_s": None, + "max_steps": None, + "enabled": None, + } + assert calls[7]["params"] == { + "schedule_id": "s", + "cursor": None, + "limit": 2, + } + + +def test_rpc_client_satisfies_schedule_surface_static_shape() -> None: + _: type[WorkflowScheduleSurface] = RpcWorkflowApiClient