sched: guard record_stopped_execution with proven ownership

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
 

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