diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index f0d2fc72..9fb22cad 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -432,6 +432,7 @@ class Scheduler: boundary, clears the executing mark, and records terminal history. Never re-invokes the dispatcher. """ + self._require_ownership() from wf_api.run_lifecycle import persist_stopped_run from wf_artifacts.runs.models import StoredRunStatus diff --git a/tests/scheduling/test_ownership_guard.py b/tests/scheduling/test_ownership_guard.py index fd93e7c5..05076272 100644 --- a/tests/scheduling/test_ownership_guard.py +++ b/tests/scheduling/test_ownership_guard.py @@ -9,7 +9,7 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import Any +from typing import Any, cast import pytest @@ -18,6 +18,7 @@ from tests.scheduling.controlled import ( ScriptedDispatcher, fixture_environment, ) +from tests.scheduling.controlled import ScriptedDispatcher as SD from wf_artifacts.runs.store import FileRunStore from wf_scheduling import recovery as sched_recovery from wf_scheduling.calendar import OneShotSource @@ -189,3 +190,25 @@ def test_admin_mutations_require_ownership(tmp_path: Path) -> None: sched.edit_schedule("a", ts(2026, 9, 8, 12, 0)) assert store.get_schedule("a").paused is True assert store.get_schedule("a").revision == 1 + + +def test_settle_requires_ownership(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs = _scheduler(tmp_path, ownership, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + _due_setup(sched, store, intended) + sched.poll(intended) + run_id = runs.list_runs()[0].id + assert runs.is_executing(run_id) + ownership.release() + admission = runs.get_admission(run_id) + state = cast(SD, sched.dispatcher).finish(admission, "complete") + with pytest.raises(SecondOwnerError): + sched.record_stopped_execution( + run_id, state, intended + timedelta(minutes=1) + ) + assert runs.get_run(run_id).status.value == "admitted" + assert runs.is_executing(run_id) + finally: + ownership.release()