sched: persist executing transition before executor; abandon on crash (F1)

src/wf_artifacts/runs/store.py

@@ -64,6 +64,15 @@ class RunStore:
     def is_pending_dispatch(self, run_id: str) -> bool:
         raise NotImplementedError
 
+    def mark_executing(self, run_id: str) -> None:
+        raise NotImplementedError
+
+    def clear_executing(self, run_id: str) -> None:
+        raise NotImplementedError
+
+    def is_executing(self, run_id: str) -> bool:
+        raise NotImplementedError
+
 
 class FileRunStore(RunStore):
     """JSON file-backed admitted- and stopped-run store for local dev/tests.
@@ -215,6 +224,30 @@ class FileRunStore(RunStore):
         except ValueError:
             return False
 
+    def mark_executing(self, run_id: str) -> None:
+        """Persist the undispatched→executing transition before the executor.
+
+        After this write, a crash must abandon the run (failed, no replay):
+        the executor may already have produced external effects.
+        """
+        with self._lock:
+            self._write_json(self._executing_path(run_id), {"executing": True})
+
+    def clear_executing(self, run_id: str) -> None:
+        with self._lock:
+            try:
+                path = self._executing_path(run_id)
+            except ValueError:
+                return
+            if path.exists():
+                path.unlink()
+
+    def is_executing(self, run_id: str) -> bool:
+        try:
+            return self._executing_path(run_id).exists()
+        except ValueError:
+            return False
+
     def _write_json(self, path: Path, payload: object) -> None:
         path.parent.mkdir(parents=True, exist_ok=True)
         temp_path = path.with_suffix(".json.tmp")
@@ -241,5 +274,8 @@ class FileRunStore(RunStore):
     def _pending_path(self, run_id: str) -> Path:
         return self._run_directory(run_id) / "pending_dispatch.json"
 
+    def _executing_path(self, run_id: str) -> Path:
+        return self._run_directory(run_id) / "executing.json"
+
     def _checkpoint_path(self, run_id: str, sequence: int) -> Path:
         return self._run_directory(run_id) / "checkpoints" / f"{sequence:06d}.json"

src/wf_scheduling/poll.py

@@ -300,38 +300,31 @@ class Scheduler:
         self.schedule_store.save_consumed(sched.id, intended)
         from wf_api.run_lifecycle import materialize_admitted_view
 
-        # Dispatch mark precedes the view so a crash after admission but
-        # before/during first dispatch stays pending (not abandoned). Cleared
-        # after dispatch returns regardless of outcome (hang still dispatched).
         self.run_store.mark_pending_dispatch(run_id)
         materialize_admitted_view(store=self.run_store, admission=admission)
-        try:
+        self._execute_guarded(run_id, now)
-            self._dispatch(run_id, now)
-        finally:
-            self.run_store.clear_pending_dispatch(run_id)
         return run_id
 
-    def _dispatch(self, run_id: str, now: datetime) -> None:
+    def _execute_guarded(self, run_id: str, now: datetime) -> None:
-        """Persist one stopped result for an admitted run via the dispatcher.
+        """Dispatch one provably undispatched run behind a durable transition.
 
-        The dispatcher returns a genuine stopped :class:`wf_core.RunState`
+        The undispatched→executing mark is persisted BEFORE the executor is
-        (or :class:`StillRunning` when the outcome is unknown); persistence
+        invoked, and the pending marker is cleared with it. Clearing is
-        goes through the shared ``wf_api.run_lifecycle`` boundary so
+        explicit after a stopped result is durably persisted — there is
-        scheduler dispatches share the manual-run torn-write protocol.
+        deliberately no ``finally``: a terminated process must leave the
+        executing mark so recovery abandons the run instead of retrying it.
         """
         from wf_api.run_lifecycle import persist_stopped_run
         from wf_artifacts.runs.models import StoredRunStatus
 
-        try:
-            self.run_store.get_run(run_id)
-        except KeyError as exc:
-            raise BlockedSchedule(f"dispatch missing run view: {run_id!r}") from exc
         try:
             admission = self.run_store.get_admission(run_id)
         except KeyError as exc:
             raise BlockedSchedule(f"dispatch missing admission: {run_id!r}") from exc
         if admission.schedule_id is None:
             raise BlockedSchedule(f"dispatch missing schedule owner: {run_id!r}")
+        self.run_store.mark_executing(run_id)
+        self.run_store.clear_pending_dispatch(run_id)
         result = self.dispatcher.dispatch(admission=admission, now=now)
         if isinstance(result, StillRunning):
             return
@@ -341,6 +334,52 @@ class Scheduler:
             run=result.result,
             run_id=run_id,
         )
+        self.run_store.clear_executing(run_id)
+        kind = {
+            StoredRunStatus.COMPLETED: "completed",
+            StoredRunStatus.INTERRUPTED: "interrupted",
+            StoredRunStatus.FAILED: "failed",
+        }[stopped.status]
+        self._record(
+            kind=kind,  # type: ignore[arg-type]
+            sched_id=admission.schedule_id,
+            intended=_admission_intended(self.run_store, run_id),
+            run_id=run_id,
+            now=now,
+            started_at=now,
+        )
+
+    def record_stopped_execution(self, run_id: str, result: Any, now: datetime) -> None:
+        """Settle a hanging execution with its late stopped result.
+
+        Async-completion seam for dispatchers that returned
+        :class:`StillRunning`: the executor later produced a genuine stopped
+        :class:`wf_core.RunState`. Persists through the shared lifecycle
+        boundary, clears the executing mark, and records terminal history.
+        Never re-invokes the dispatcher.
+        """
+        from wf_api.run_lifecycle import persist_stopped_run
+        from wf_artifacts.runs.models import StoredRunStatus
+
+        try:
+            admission = self.run_store.get_admission(run_id)
+        except KeyError as exc:
+            raise BlockedSchedule(f"settle missing admission: {run_id!r}") from exc
+        if admission.schedule_id is None:
+            raise BlockedSchedule(f"settle missing schedule owner: {run_id!r}")
+        try:
+            record = self.run_store.get_run(run_id)
+        except KeyError as exc:
+            raise BlockedSchedule(f"settle missing run view: {run_id!r}") from exc
+        if self._status_value(record) != "admitted":
+            raise BlockedSchedule(f"settle non-admitted run: {run_id!r}")
+        stopped = persist_stopped_run(
+            store=self.run_store,
+            environment=admission.environment,
+            run=result,
+            run_id=run_id,
+        )
+        self.run_store.clear_executing(run_id)
         kind = {
             StoredRunStatus.COMPLETED: "completed",
             StoredRunStatus.INTERRUPTED: "interrupted",
@@ -380,20 +419,56 @@ class Scheduler:
         """Dispatch recovery-materialized runs through capacity checks.
 
         Recovery NEVER executes: it only completes missing views flagged
-        pending for this sweep. Pending runs of blocked schedules stay
+        pending for this sweep. Each pending marker is validated before it
-        pending. Hanging admitted runs without a pending marker are never
+        is trusted: markers without an admission (or without a schedule
-        re-executed here.
+        owner) fail the run closed instead of dispatching; markers on
+        stopped runs are stale and cleared without redispatch; unknown
+        owners and blocked schedules are left untouched. Hanging admitted
+        runs without a pending marker are never re-executed here.
         """
-        from wf_scheduling.recovery import _is_pending, clear_pending
+        from wf_scheduling.recovery import (
+            CORRUPT_PENDING_REASON,
+            _fail_run,
+            clear_executing,
+            clear_pending,
+            is_executing,
+        )
+        from wf_scheduling.recovery import _is_pending as _pending
 
         for run in sorted(self.run_store.list_runs(), key=lambda r: r.id):
-            if not _is_pending(self.run_store, run.id):
+            if not _pending(self.run_store, run.id):
                 continue
             try:
                 admission = self.run_store.get_admission(run.id)
             except KeyError:
+                _fail_run(
+                    self.run_store,
+                    run,
+                    CORRUPT_PENDING_REASON,
+                    now,
+                    None,
+                )
+                clear_pending(self.run_store, run.id)
+                clear_executing(self.run_store, run.id)
                 continue
             if admission.schedule_id is None:
+                _fail_run(
+                    self.run_store,
+                    run,
+                    CORRUPT_PENDING_REASON,
+                    now,
+                    None,
+                )
+                clear_pending(self.run_store, run.id)
+                clear_executing(self.run_store, run.id)
+                continue
+            if self._status_value(run) != "admitted":
+                # Stale marker on a stopped run: clear it, never redispatch.
+                # Terminal-history reconciliation is owned by recovery (which
+                # dedups); the sweep only removes the untrustworthy marker.
+                clear_pending(self.run_store, run.id)
+                if is_executing(self.run_store, run.id):
+                    clear_executing(self.run_store, run.id)
                 continue
             try:
                 sched = self.schedule_store.get_schedule(admission.schedule_id)
@@ -403,8 +478,7 @@ class Scheduler:
                 continue
             if self._task_load() >= self.capacity:
                 continue
-            self._dispatch(run.id, now)
+            self._execute_guarded(run.id, now)
-            clear_pending(self.run_store, run.id)
 
     def _poll_one(self, sched: Any, now: datetime) -> str:
         if sched.deleted:

src/wf_scheduling/recovery.py

@@ -6,6 +6,29 @@ the poll sweep), fails abandoned/ambiguous runs with external-effects
 disclosure (no replay), matches stopped results to the ACTIVE attempt by
 identity, reconciles missing terminal records, fails corrupt views closed
 and blocks the schedule, and preserves stopped interruptions in their slots.
+
+Restart behavior at each marker/admission/view write boundary (admission
+persist → pending mark → view materialize → executing mark → pending
+clear → executor → stopped persist → executing clear → history):
+
+- admission only: view is materialized and flagged pending-dispatch; the
+  occurrence was already consumed, so the later poll dispatches exactly
+  once through capacity checks.
+- admission + pending, no view: same as above (the view write was lost).
+- admission + view + pending, no executing mark: provably undispatched
+  (the executor is unreachable without the executing mark); kept pending.
+- any admitted run with the executing mark: the executor may already have
+  produced external effects; failed (ACTIVE attempt: ambiguous, else
+  abandoned) with external-effects disclosure, never redispatched.
+- admitted + view with neither mark and no ACTIVE attempt: legacy or
+  manually cleared state whose outcome is unprovable; failed closed
+  without replay (current code never produces this shape on crash).
+- stopped summary + executing/pending marks: completion-window leftovers
+  (stopped result persisted, marker clearing lost); markers are cleared,
+  the stopped status stands, nothing is re-executed. Checkpoint-vs-summary
+  reconciliation across the torn ``save_checkpoint``/``save_run`` boundary
+  is owned by the F2 reconcile step, which runs before marker handling.
+- stopped summary, no marks: existing terminal/attempt reconciliation.
 """
 
 from __future__ import annotations
@@ -25,6 +48,10 @@ AMBIGUOUS_REASON = (
     "ambiguous resume attempt: may have executed; external effects may "
     "already have occurred; no retry"
 )
+CORRUPT_PENDING_REASON = (
+    "corrupt pending marker: no admission owns this run; the marker cannot "
+    "be trusted for dispatch"
+)
 
 
 def recover(
@@ -60,6 +87,20 @@ def recover(
         status = getattr(run.status, "value", run.status)
         attempt = run_store.get_resume_attempt(run.id)
         active = attempt is not None and attempt.state == "ACTIVE"
+        if status in (
+            StoredRunStatus.INTERRUPTED.value,
+            StoredRunStatus.COMPLETED.value,
+            StoredRunStatus.FAILED.value,
+            "interrupted",
+            "completed",
+            "failed",
+        ) and (is_executing(run_store, run.id) or _is_pending(run_store, run.id)):
+            # Completion-window leftovers: the stopped result was persisted
+            # but marker clearing was lost. The stopped status stands;
+            # nothing is re-executed.
+            clear_executing(run_store, run.id)
+            clear_pending(run_store, run.id)
+            diags.append(f"{run.id}:completion-window-cleared")
         if status == StoredRunStatus.INTERRUPTED.value or status == "interrupted":
             if active:
                 assert attempt is not None
@@ -87,6 +128,18 @@ def recover(
             else:
                 diags.append(f"{run.id}:waiting-resumable")
         elif status == StoredRunStatus.ADMITTED.value or status == "admitted":
+            if is_executing(run_store, run.id):
+                # The executor may already have produced external effects:
+                # abandon, never retry. Both markers are cleared so a later
+                # recovery does not re-fail the now-terminal run.
+                if active:
+                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+                else:
+                    _fail_run(run_store, run, ABANDONED_REASON, now, record_history)
+                clear_pending(run_store, run.id)
+                clear_executing(run_store, run.id)
+                diags.append(f"{run.id}:failed-closed")
+                continue
             if _is_pending(run_store, run.id):
                 diags.append(f"{run.id}:pending-dispatch-kept")
                 continue
@@ -178,6 +231,15 @@ def _mark_pending(run_store: Any, run_id: str) -> None:
     run_store.mark_pending_dispatch(run_id)
 
 
+def is_executing(run_store: Any, run_id: str) -> bool:
+    return bool(run_store.is_executing(run_id))
+
+
+def clear_executing(run_store: Any, run_id: str) -> None:
+    """Clear the executing mark after a stopped result is durably persisted."""
+    run_store.clear_executing(run_id)
+
+
 def _is_pending(run_store: Any, run_id: str) -> bool:
     return bool(run_store.is_pending_dispatch(run_id))
 

tests/scheduling/test_dispatch_transition.py

@@ -0,0 +1,396 @@
+"""Durable dispatch transition: executing is persisted BEFORE the executor (R4/F1).
+
+Pending means provably undispatched and safe to dispatch. The scheduler
+persists the undispatched→executing transition before invoking the
+executor; a crash afterwards abandons the run (failed, no replay) instead
+of retrying. ``finally`` cannot be trusted across process termination, so
+marker clearing is explicit on the stopped path only.
+"""
+
+from __future__ import annotations
+
+import multiprocessing as mp
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+
+from tests.scheduling.controlled import (
+    DictDeployments,
+    ScriptedDispatcher,
+    fixture_environment,
+    stopped_state,
+)
+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
+from wf_scheduling.dispatch import StillRunning, Stopped
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import SchedulePreparer
+from wf_scheduling.store import FileScheduleStore
+
+
+def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime:
+    return datetime(y, mo, d, h, mi, tzinfo=UTC)
+
+
+def _sched_model(sid: str, **kw: Any) -> Schedule:
+    now = ts(2026, 9, 8, 12, 0)
+    base: dict[str, Any] = {
+        "id": sid,
+        "deployment_id": "dep-1",
+        "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"},
+        "input_bindings": [],
+        "created_at": now.isoformat(),
+        "updated_at": now.isoformat(),
+    }
+    base.update(kw)
+    return Schedule.model_validate(base)
+
+
+def _harness(
+    root: Path,
+    ownership: SchedulerOwnership,
+    *,
+    capacity: int = 4,
+    script: dict | None = None,
+) -> tuple[Scheduler, FileScheduleStore, FileRunStore]:
+    sched_store = FileScheduleStore(root / "sched")
+    run_store = FileRunStore(root / "runs")
+    sched = Scheduler(
+        schedule_store=sched_store,
+        run_store=run_store,
+        sources={},
+        capacity=capacity,
+        preparer=SchedulePreparer(
+            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+            fixture_environment,
+        ),
+        dispatcher=ScriptedDispatcher(script),
+        ownership=ownership,
+    )
+    return sched, sched_store, run_store
+
+
+def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None:
+    store.create_schedule(_sched_model("a"))
+    store.save_consumed("a", intended - timedelta(hours=1))
+    sched.sources["a"] = OneShotSource(intended)
+
+
+def test_executing_is_set_before_executor_runs(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        observed: dict[str, bool] = {}
+
+        def probe(admission: Any, now: datetime) -> Any:
+            observed["executing_at_entry"] = FileRunStore(
+                tmp_path / "runs"
+            ).is_executing(admission.id)
+            observed["pending_at_entry"] = FileRunStore(
+                tmp_path / "runs"
+            ).is_pending_dispatch(admission.id)
+            return StillRunning()
+
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": probe})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        assert observed == {"executing_at_entry": True, "pending_at_entry": False}
+        run_id = runs.list_runs()[0].id
+        assert runs.is_executing(run_id)
+        assert not runs.is_pending_dispatch(run_id)
+    finally:
+        ownership.release()
+
+
+def test_failure_before_transition_never_dispatches(tmp_path: Path) -> None:
+    calls: list[str] = []
+
+    class FailMarkOnce(FileRunStore):
+        def __init__(self, root: Path) -> None:
+            super().__init__(root)
+            self.armed = True
+
+        def mark_executing(self, run_id: str) -> None:
+            if self.armed:
+                self.armed = False
+                raise OSError("injected mark_executing failure")
+            super().mark_executing(run_id)
+
+    def spy(admission: Any, now: datetime) -> Any:
+        calls.append(admission.id)
+        return StillRunning()
+
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailMarkOnce(tmp_path / "runs")
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=run_store,
+            sources={},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+                fixture_environment,
+            ),
+            dispatcher=ScriptedDispatcher({"*": spy}),
+            ownership=ownership,
+        )
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, sched_store, intended)
+        with pytest.raises(OSError, match="injected mark_executing"):
+            sched.poll(intended)
+        # The executor never ran; the run stays provably undispatched.
+        assert calls == []
+        run_id = run_store.list_runs()[0].id
+        assert run_store.is_pending_dispatch(run_id)
+        assert not run_store.is_executing(run_id)
+        # Next poll dispatches exactly once.
+        sched.poll(intended + timedelta(minutes=1))
+        assert calls == [run_id]
+    finally:
+        ownership.release()
+
+
+def test_capacity_shortage_keeps_pending_undispatched(tmp_path: Path) -> None:
+    calls: list[str] = []
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, capacity=0)
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+
+        def spy(admission: Any, now: datetime) -> Any:
+            calls.append(admission.id)
+            return StillRunning()
+
+        cast(SD, sched.dispatcher).script = {"*": spy}
+        assert sched.poll(intended) == {"a": "admit:held-undecided"}
+        assert calls == []
+        # A held candidate is pending-but-unadmitted, not an admitted run.
+        assert runs.list_runs() == []
+        sched.capacity = 4
+        result = sched.poll(intended + timedelta(minutes=1))
+        assert result["a"].startswith("admit:run-")
+        assert len(calls) == 1
+    finally:
+        ownership.release()
+
+
+def test_pending_on_terminal_run_clears_without_redispatch(tmp_path: Path) -> None:
+    calls: list[str] = []
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership)
+
+        def spy(admission: Any, now: datetime) -> Any:
+            calls.append(admission.id)
+            return Stopped(result=stopped_state(admission, "complete"))
+
+        cast(SD, sched.dispatcher).script = {"*": spy}
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        assert calls != []
+        run_id = runs.list_runs()[0].id
+        assert runs.get_run(run_id).status.value == "completed"
+        # A stale pending marker on a terminal run must not redispatch.
+        runs.mark_pending_dispatch(run_id)
+        before = len(calls)
+        sched.poll(intended + timedelta(minutes=5))
+        assert len(calls) == before
+        assert not runs.is_pending_dispatch(run_id)
+        assert runs.get_run(run_id).status.value == "completed"
+        page = store.list_occurrences("a", limit=100)
+        completed = [
+            r
+            for r in cast(list[dict[str, Any]], page["occurrences"])
+            if r["kind"] == "completed"
+        ]
+        assert len(completed) == 1
+    finally:
+        ownership.release()
+
+
+def test_pending_without_admission_fails_closed(tmp_path: Path) -> None:
+    from datetime import datetime as _dt
+
+    from tests.artifacts.test_run_store import artifact as _artifact
+    from tests.artifacts.test_run_store import deployment as _deployment
+    from wf_artifacts import PinnedRunEnvironment
+    from wf_artifacts.runs.models import (
+        ResumeReadiness,
+        StoredRunStatus,
+        WorkflowRunRecord,
+    )
+
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        now = _dt.now(UTC)
+        runs.save_run(
+            WorkflowRunRecord(
+                id="run-999009",
+                status=StoredRunStatus.ADMITTED,
+                resume_readiness=ResumeReadiness.NOT_APPLICABLE,
+                environment=PinnedRunEnvironment(
+                    deployment=_deployment(),
+                    root_artifact=_artifact(),
+                    child_artifacts=[],
+                ),
+                latest_checkpoint_id=None,
+                created_at=now,
+                updated_at=now,
+            )
+        )
+        runs.mark_pending_dispatch("run-999009")
+        sched.poll(intended)
+        assert runs.get_run("run-999009").status.value == "failed"
+        assert not runs.is_pending_dispatch("run-999009")
+    finally:
+        ownership.release()
+
+
+def test_executing_admitted_is_abandoned_by_recovery(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        assert runs.is_executing(run_id)
+        # Fresh objects simulate a restart: executing work is abandoned, never
+        # redispatched.
+        fresh_runs = FileRunStore(tmp_path / "runs")
+        fresh_sched_store = FileScheduleStore(tmp_path / "sched")
+        diags = sched_recovery.recover(
+            schedule_store=fresh_sched_store,
+            run_store=fresh_runs,
+            now=intended + timedelta(minutes=1),
+            ownership=ownership,
+        )
+        assert any("failed-closed" in d for d in diags)
+        assert fresh_runs.get_run(run_id).status.value == "failed"
+        # A second poll after recovery never re-executes the abandoned run.
+        sched2, _, runs2 = _harness(tmp_path, ownership, script={"*": "hang"})
+        sched2.poll(intended + timedelta(minutes=2))
+        assert runs2.get_run(run_id).status.value == "failed"
+    finally:
+        ownership.release()
+
+
+def test_settle_hanging_run_persists_and_clears(tmp_path: Path) -> None:
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
+        intended = ts(2026, 9, 8, 12, 0)
+        _due(sched, store, intended)
+        sched.poll(intended)
+        run_id = runs.list_runs()[0].id
+        assert runs.is_executing(run_id)
+        # The late stopped result settles without re-invoking the dispatcher.
+        dispatcher = cast(SD, sched.dispatcher)
+        admission = runs.get_admission(run_id)
+        sched.record_stopped_execution(
+            run_id, dispatcher.finish(admission, "complete"), intended
+        )
+        assert runs.get_run(run_id).status.value == "completed"
+        assert not runs.is_executing(run_id)
+        assert not runs.is_pending_dispatch(run_id)
+        checkpoint = runs.get_latest_checkpoint(run_id)
+        assert checkpoint.reason.value == "completed"
+    finally:
+        ownership.release()
+
+
+def _child_poll_and_die(
+    root: str, side_effect: str, intended_iso: str, sched_id: str = "a"
+) -> None:
+    """Child entry: poll once; the dispatcher records a side effect then dies."""
+    import os as _os
+    from datetime import datetime as _dt
+
+    from tests.scheduling.controlled import (
+        DictDeployments as _Dict,
+    )
+    from tests.scheduling.controlled import (
+        ScriptedDispatcher as _SD,
+    )
+    from tests.scheduling.controlled import (
+        fixture_environment as _env,
+    )
+    from wf_artifacts.runs.store import FileRunStore as _Runs
+    from wf_scheduling.calendar import OneShotSource as _OneShot
+    from wf_scheduling.ownership import SchedulerOwnership as _Own
+    from wf_scheduling.poll import Scheduler as _Sched
+    from wf_scheduling.prepare import SchedulePreparer as _Prep
+    from wf_scheduling.store import FileScheduleStore as _SchedStore
+
+    base = Path(root)
+    intended = _dt.fromisoformat(intended_iso)
+
+    def killer(admission: Any, now: _dt) -> Any:
+        with open(side_effect, "a", encoding="utf-8") as fh:
+            fh.write(admission.id + "\n")
+        _os._exit(1)
+
+    sched = _Sched(
+        schedule_store=_SchedStore(base / "sched"),
+        run_store=_Runs(base / "runs"),
+        sources={sched_id: _OneShot(intended)},
+        capacity=4,
+        preparer=_Prep(_Dict({"dep-1": {"rev": 1, "required": []}}), _env),
+        dispatcher=_SD({"*": killer}),
+        ownership=_Own(base / "sched", owner="child").acquire(),
+    )
+    sched.poll(intended)
+
+
+def test_subprocess_death_after_side_effect_dispatches_once(tmp_path: Path) -> None:
+    ctx = mp.get_context("spawn")
+    root = tmp_path / "death"
+    root.mkdir()
+    sched_store = FileScheduleStore(root / "sched")
+    intended = ts(2026, 9, 8, 12, 0)
+    sched_store.create_schedule(_sched_model("a"))
+    sched_store.save_consumed("a", intended - timedelta(hours=1))
+    side_effect = root / "effects.log"
+    proc = ctx.Process(
+        target=_child_poll_and_die,
+        args=(str(root), str(side_effect), intended.isoformat()),
+    )
+    proc.start()
+    proc.join(120)
+    assert proc.exitcode not in (None, 0), f"child must die, got {proc.exitcode}"
+    assert side_effect.read_text(encoding="utf-8").strip() != ""
+    first_effects = side_effect.read_text(encoding="utf-8").strip().splitlines()
+
+    # Restart with fresh objects: the side effect count stays one and the run
+    # is abandoned, never redispatched.
+    ownership = SchedulerOwnership(root / "sched", owner="parent").acquire()
+    try:
+        sched, store, runs = _harness(root, ownership, script={"*": "hang"})
+        sched.sources["a"] = OneShotSource(intended)
+        diags = sched_recovery.recover(
+            schedule_store=store,
+            run_store=runs,
+            now=intended + timedelta(minutes=1),
+            ownership=ownership,
+        )
+        assert any("failed-closed" in d for d in diags)
+        sched.poll(intended + timedelta(minutes=1))
+        effects = side_effect.read_text(encoding="utf-8").strip().splitlines()
+        assert effects == first_effects
+        assert len(effects) == 1
+        assert runs.get_run(first_effects[0]).status.value == "failed"
+    finally:
+        ownership.release()