fix: honor live scheduling revisions at dispatch

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

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:

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"

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

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

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 '/'")

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
 

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

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)

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(

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