diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index b815c9eb..06e35b94 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -111,9 +111,9 @@ class WorkflowRunApi: # allocate/freeze -> persist admission -> materialize view -> dispatch # captured -> persist stopped. A failed durable admission never # dispatches, and dispatch never re-resolves the deployment. - # TODO(T11): hold the single-owner admission lock around this sequence - # once scheduler ownership lands; manual recheck here is only - # deployment validation (no schedule/capacity/overlap yet). + # Manual runs intentionally do not consume scheduler capacity or + # participate in schedule overlap; scheduler-owned dispatch uses its + # separate ownership and admission protocol. store = self._run_store() run_id = store.allocate_run_id() environment = create_pinned_environment( @@ -129,9 +129,9 @@ class WorkflowRunApi: max_steps=limits.max_steps, ) materialize_admitted_view(store=store, admission=admission) - # TODO(T10): record a dispatch mark between materialize and execute so - # crash-after-dispatch (abandoned, failed without replay) is - # distinguishable from pending-dispatch (safe to dispatch later). + # Manual execution has no scheduler pending-dispatch marker. The + # scheduler path records that marker before handing a run to this API, + # while manual runs retain their existing synchronous lifecycle. plan = raw_plan_from_artifact(admission.environment.root_artifact) captured_tree = saved_subgraph_tree_from_snapshots( admission.environment.child_artifacts diff --git a/src/wf_scheduling/lifecycle.py b/src/wf_scheduling/lifecycle.py index f7266f1e..370bf9ee 100644 --- a/src/wf_scheduling/lifecycle.py +++ b/src/wf_scheduling/lifecycle.py @@ -494,17 +494,20 @@ class SchedulerService: except Exception: schedules = [] fresh: dict[str, Any] = {} + definitions: dict[str, Any] = {} for sched in schedules: if getattr(sched, "deleted", False): continue try: fresh[sched.id] = source_for_trigger(sched.trigger) + definitions[sched.id] = sched.trigger except Exception: # No source entry: the poll raises a loud per-schedule # definition error instead of ticking a stale calendar. continue self._sources.clear() self._sources.update(fresh) + self._scheduler.set_source_definitions(definitions) def _tick(self, now: datetime) -> dict[str, str]: with self._lock: diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index d65e1bbd..67caa8c6 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -141,6 +141,18 @@ class Scheduler: else FileScheduleHistoryRecorder(schedule_store) ) self._poll_cursor = 0 + self._source_definitions: dict[str, Any] = {} + + def set_source_definitions(self, definitions: dict[str, Any]) -> None: + """Record the trigger definitions used to build managed sources. + + The service refreshes this map at tick start. A schedule edit can + commit between that refresh and a schedule's fresh read; the poller + then rebuilds only that schedule's source before resolving an instant. + Direct Scheduler tests without this service-owned map keep their + injected source collaborators unchanged. + """ + self._source_definitions = dict(definitions) def _require_ownership(self) -> None: """Reject schedule mutation/dispatch without proven live ownership. @@ -741,12 +753,23 @@ class Scheduler: consumed = self.schedule_store.get_consumed(sched.id) or EPOCH self.schedule_store.save_consumed(sched.id, max(consumed, now)) return "paused" - try: - src = self.sources[sched.id] - except KeyError as exc: - raise InvalidScheduleDefinitionError( - f"no occurrence source for schedule {sched.id!r}" - ) from exc + source_definition = self._source_definitions.get(sched.id) + if source_definition is not None and source_definition != sched.trigger: + try: + src = source_for_trigger(sched.trigger) + except Exception as exc: + raise InvalidScheduleDefinitionError( + f"invalid trigger for schedule {sched.id!r}: {exc}" + ) from exc + self.sources[sched.id] = src + self._source_definitions[sched.id] = sched.trigger + else: + try: + src = self.sources[sched.id] + except KeyError as exc: + raise InvalidScheduleDefinitionError( + f"no occurrence source for schedule {sched.id!r}" + ) from exc consumed = self.schedule_store.get_consumed(sched.id) or EPOCH if consumed > now: return "clock-rollback-held" diff --git a/src/wf_scheduling/prepare.py b/src/wf_scheduling/prepare.py index 7a45a86a..a64ed019 100644 --- a/src/wf_scheduling/prepare.py +++ b/src/wf_scheduling/prepare.py @@ -100,6 +100,12 @@ class SchedulePreparer: ) except KeyError: return PreparationRejected(reason="deployment-deleted") + if revision != environment.deployment.revision: + # The environment is the pinned artifact/deployment snapshot. A + # directory revision change between its construction and this + # recheck makes the invocation internally contradictory; reject + # it rather than recording newer metadata against older artifacts. + return PreparationRejected(reason="deployment-changed") occurrence = { "schedule_id": sched.id, "occurrence_id": occurrence_id(sched.id, intended), diff --git a/src/wf_server/cli.py b/src/wf_server/cli.py index 067d3da7..adbd3be0 100644 --- a/src/wf_server/cli.py +++ b/src/wf_server/cli.py @@ -120,12 +120,9 @@ def serve( sched_config = server_scheduler_config(workflow_config, enable_scheduler) if sched_config is not None: - config_mcp_sources = ( - workflow_config is not None - and any( - getattr(source, "kind", None) == "mcp" - for source in workflow_config.server.sources - ) + config_mcp_sources = workflow_config is not None and any( + getattr(source, "kind", None) == "mcp" + for source in workflow_config.server.sources ) if mcp_backed or config_mcp_sources: # The scheduler is verified over local/static servers only: @@ -139,7 +136,10 @@ def serve( server, rpc_path=resolved_rpc_path, drafts=True, - lifespan=scheduler_lifespan(server, sched_config) + # fastapi-jsonrpc calls its lifespan with the ASGI app. Keep the + # scheduler context manager lazy so service startup happens inside + # the server lifespan, not while the CLI is assembling the app. + lifespan=(lambda _app: scheduler_lifespan(server, sched_config)) if sched_config is not None else None, ) diff --git a/src/wf_transport_rpc_http/app.py b/src/wf_transport_rpc_http/app.py index 6324f17a..64522b12 100644 --- a/src/wf_transport_rpc_http/app.py +++ b/src/wf_transport_rpc_http/app.py @@ -38,8 +38,9 @@ def create_rpc_app( Transport code owns only JSON-RPC envelope handling. Workflow semantics stay behind server.api, so this package remains swappable with WebSocket/MCP - transports later. ``lifespan`` (e.g. the opt-in scheduler lifespan) is - passed through to the ASGI app; ``None`` preserves existing behavior. + transports later. ``lifespan`` must be the ASGI lifespan factory expected + by fastapi-jsonrpc (for example, an opt-in scheduler factory); ``None`` + preserves existing behavior. """ if not rpc_path.startswith("/"): raise ValueError("rpc_path must start with '/'") diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index 6870b13b..ec3dc3dd 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -13,8 +13,8 @@ from tests.scheduling.controlled import ( ) from wf_artifacts.runs.models import StoredRunStatus from wf_artifacts.runs.store import FileRunStore -from wf_scheduling.calendar import OneShotSource -from wf_scheduling.models import Schedule +from wf_scheduling.calendar import CronSource, OneShotSource +from wf_scheduling.models import CronTrigger, Schedule from wf_scheduling.ownership import SchedulerOwnership from wf_scheduling.poll import SCAN_CAP, Scheduler from wf_scheduling.prepare import SchedulePreparer @@ -249,6 +249,34 @@ def test_schedule_edit_between_poll_snapshot_and_admission_cannot_overwrite_term sched.ownership.release() +def test_trigger_edit_refreshes_managed_calendar_before_polling(tmp_path: Path) -> None: + """A trigger edit cannot be paired with the previous tick's source.""" + sched, store, runs, sources = _harness(tmp_path, script={"*": "complete"}) + start = ts(2026, 9, 8, 11, 0) + now = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "a", + CronSource("0 * * * *", "UTC"), + start, + ) + old_trigger = store.get_schedule("a").trigger + sched.set_source_definitions({"a": old_trigger}) + edited = store.get_schedule("a") + edited.revision = 2 + edited.trigger = CronTrigger( + expression="0 13 * * *", + timezone="UTC", + ) + store.update_schedule(edited, expected_revision=1) + + assert sched.poll(now) == {"a": "idle"} + assert runs.list_admissions() == [] + sched.ownership.release() + + def test_parallel_limits_and_interrupted_slots() -> None: import tempfile diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py index b5a712ad..9bb80e2c 100644 --- a/tests/scheduling/test_preparation.py +++ b/tests/scheduling/test_preparation.py @@ -56,9 +56,15 @@ def _scheduler( ) -> tuple[Scheduler, FileScheduleStore, FileRunStore]: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") + + def revisioned_fixture_environment(sched: Any) -> Any: + environment = fixture_environment(sched) + deployment = environment.deployment.model_copy(update={"revision": 3}) + return environment.model_copy(update={"deployment": deployment}) + preparer = SchedulePreparer( DictDeployments({"dep-1": {"rev": 3, "required": []}}), - fixture_environment, + revisioned_fixture_environment, ) sched = Scheduler( schedule_store=sched_store, @@ -159,6 +165,23 @@ def test_unknown_deployment_rejects_without_a_run(tmp_path: Path) -> None: sched.ownership.release() +def test_deployment_revision_mismatch_rejects_pinned_environment() -> None: + """Preparation fails when directory metadata no longer matches the pin.""" + preparer = SchedulePreparer( + DictDeployments({"dep-1": {"rev": 2, "required": []}}), + fixture_environment, + ) + + result = preparer.prepare( + sched=_sched_model("s"), + intended=ts(2026, 9, 8, 12, 0), + now=ts(2026, 9, 8, 12, 0), + ) + + assert isinstance(result, PreparationRejected) + assert result.reason == "deployment-changed" + + def test_missing_required_input_rejects_without_a_run(tmp_path: Path) -> None: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") diff --git a/tests/scheduling/test_preparation_schema.py b/tests/scheduling/test_preparation_schema.py index 2dee03c5..e016488e 100644 --- a/tests/scheduling/test_preparation_schema.py +++ b/tests/scheduling/test_preparation_schema.py @@ -248,7 +248,7 @@ def test_changed_contract_rejects_before_admission(tmp_path: Path) -> None: ) sched.preparer = SchedulePreparer( - DictDeployments({"dep-1": {"rev": 2, "required": []}}), changed_env + DictDeployments({"dep-1": {"rev": 1, "required": []}}), changed_env ) assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"} _rejected_entry(store) diff --git a/tests/wf_server/test_cli.py b/tests/wf_server/test_cli.py index 8d3a8243..531bd2f4 100644 --- a/tests/wf_server/test_cli.py +++ b/tests/wf_server/test_cli.py @@ -494,7 +494,7 @@ def test_rpc_server_cli_enable_scheduler_with_store_root_builds_app( assert result.exit_code == 0, result.output assert captured["server"] is not None assert captured["app"] is not None - assert captured["lifespan"] is not None + assert callable(captured["lifespan"]) def test_rpc_server_cli_config_scheduler_section_enables_without_flag( @@ -528,7 +528,7 @@ def test_rpc_server_cli_config_scheduler_section_enables_without_flag( result = CliRunner().invoke(app, ["--config", str(config_path)]) assert result.exit_code == 0, result.output - assert captured["lifespan"] is not None + assert callable(captured["lifespan"]) def test_rpc_server_cli_flag_overrides_disabled_config_scheduler( diff --git a/tests/wf_server/test_scheduler_composition.py b/tests/wf_server/test_scheduler_composition.py index cba157d7..d85d9ad1 100644 --- a/tests/wf_server/test_scheduler_composition.py +++ b/tests/wf_server/test_scheduler_composition.py @@ -18,6 +18,7 @@ from wf_server.scheduling import ( scheduler_lifespan, server_scheduler_config, ) +from wf_transport_rpc_http import create_rpc_app def test_server_scheduler_config_disabled_by_default() -> None: @@ -147,3 +148,25 @@ async def test_scheduler_lifespan_releases_lock_on_exit(tmp_path: Path) -> None: assert probe.held is True finally: probe.release() + + +async def test_rpc_scheduler_lifespan_factory_starts_and_stops_service( + tmp_path: Path, +) -> None: + """The RPC app receives a callable that lazily owns scheduler startup.""" + server = build_local_static_workflow_server(tmp_path) + config = SchedulerServiceConfig(auto_tick=False) + app = create_rpc_app( + server, + lifespan=lambda _app: scheduler_lifespan(server, config), + ) + + async with app.router.lifespan_context(app): + assert server.api.schedules._schedule_store().root == tmp_path + + probe = SchedulerOwnership(tmp_path, owner="probe") + probe.acquire() + try: + assert probe.held is True + finally: + probe.release()