fix(cron): preserve completed jobs after claim loss

cron/scheduler.py

@@ -544,6 +544,14 @@ def set(self) -> None:
         for event in self._events:
             event.set()
 
+    def fire_claim_is_set(self) -> bool:
+        """Whether the scheduler's fire-claim heartbeat, rather than another caller, cancelled."""
+        return bool(self._events) and self._events[0].is_set()
+
+    def other_cancel_is_set(self) -> bool:
+        """Whether a transport-level cancellation source requested an interruption."""
+        return any(event.is_set() for event in self._events[1:])
+
 
 def get_running_job_ids() -> "frozenset[str]":
     """Thread-safe snapshot of executing job IDs (dispatch until ``_process_job`` returns). Read by
@@ -2645,10 +2653,15 @@ def side_effect_fence(self):
         return fire_claim_fence(self.job["id"], expected_owner=self.owner)
 
     def lost(self) -> bool:
-        if self.cancel_event is not None and self.cancel_event.is_set():
+        claim_loss_signaled = self.claim_lost is not None and self.claim_lost.is_set()
+        if isinstance(self.claim_lost, _CombinedCancelEvent):
+            if self.claim_lost.other_cancel_is_set():
+                return True
+            claim_loss_signaled = self.claim_lost.fire_claim_is_set()
+        if self.transport_cancelled():
             return True
         if self.owner is None:
-            return False
+            return claim_loss_signaled
         if self_removal_delivery_allowed(self.job["id"]):
             # The run deleted its own record; there is no claim left to re-resolve.
             return False
@@ -2658,7 +2671,8 @@ def lost(self) -> bool:
         except Exception:
             logger.debug(
                 "Job '%s': fire_claim ownership validation failed", self.job["id"], exc_info=True)
-            return False
+            # An already-signaled loss is fail-closed; a first validation failure remains best-effort.
+            return claim_loss_signaled
         if self.claim_lost is not None:
             self.claim_lost.set()
         return True

tests/cron/test_shutdown_interrupt.py

@@ -12,6 +12,7 @@
 """
 
 import threading
+from contextlib import contextmanager
 from unittest.mock import patch
 
 import pytest
@@ -540,6 +541,104 @@ def test_second_fire_cannot_claim_while_first_ownership_live(self, tmp_path):
             assert refreshed["last_status"] == "ok"
 
 
+class TestFireClaimLossAfterCompletedRun:
+    """A transient heartbeat loss must not overwrite a completed run's terminal state."""
+
+    @staticmethod
+    @contextmanager
+    def _owned_fence(*_args, **_kwargs):
+        yield True
+
+    def _run_body(self, monkeypatch, *, heartbeat_result, completed=True):
+        import cron.scheduler as sched
+
+        claim_lost = threading.Event()
+        job = {
+            "id": "long-running-job",
+            "name": "long running job",
+            "execution_id": "completed-execution",
+            "fire_claim": {"by": "fire-owner"},
+        }
+        marked = []
+        finished = []
+        saved = []
+
+        monkeypatch.setattr(sched, "claim_dispatch", lambda *_args: True)
+        monkeypatch.setattr(sched, "mark_execution_running", lambda *_args: {})
+        monkeypatch.setattr(sched, "heartbeat_fire_claim", lambda *_args, **_kwargs: heartbeat_result)
+        monkeypatch.setattr(sched, "fire_claim_fence", self._owned_fence)
+        monkeypatch.setattr(
+            sched, "run_job",
+            lambda *_args, **_kwargs: (
+                claim_lost.set() or completed,
+                "complete durable report" if completed else "partial output",
+                "completed response" if completed else "",
+                None if completed else "cancelled before terminal completion",
+            ),
+        )
+        monkeypatch.setattr(
+            sched, "save_job_output", lambda *_args: saved.append(True) or "/tmp/report.md"
+        )
+        monkeypatch.setattr(
+            sched, "mark_job_run", lambda *args, **kwargs: marked.append((args, kwargs)) or True
+        )
+        monkeypatch.setattr(
+            sched, "finish_execution", lambda *args, **kwargs: finished.append((args, kwargs))
+        )
+        monkeypatch.setattr(sched, "_deliver_result", lambda *_args, **_kwargs: None)
+        monkeypatch.setattr(sched, "_is_cron_silence_response", lambda *_args: False)
+        monkeypatch.setattr(sched, "_resolve_incidents_for_recovered_job", lambda *_args: None)
+        monkeypatch.setattr("agent.secret_scope.build_profile_secret_scope", lambda *_args: None)
+        monkeypatch.setattr("agent.secret_scope.set_secret_scope", lambda *_args: None)
+        monkeypatch.setattr("agent.secret_scope.reset_secret_scope", lambda *_args: None)
+        monkeypatch.setattr("tools.terminal_scope.install_profile_terminal_scope", lambda *_args: None)
+        monkeypatch.setattr("tools.terminal_scope.reset_terminal_scope", lambda *_args: None)
+
+        result = sched._run_one_job_body(
+            job, claim_lost=claim_lost, execution_token=object()
+        )
+        return result, marked, finished, saved
+
+    def test_completed_response_is_preserved_when_ownership_revalidates(self, monkeypatch):
+        result, marked, finished, saved = self._run_body(monkeypatch, heartbeat_result=True)
+
+        assert result is True
+        assert saved == [True]
+        assert marked == [
+            (("long-running-job", True, None), {"delivery_error": None, "expected_fire_owner": "fire-owner"})
+        ]
+        assert finished[0][1]["success"] is True
+
+    def test_incomplete_run_with_lost_owner_stays_fenced_out(self, monkeypatch):
+        result, marked, finished, saved = self._run_body(
+            monkeypatch, heartbeat_result=False, completed=False
+        )
+
+        assert result is True
+        assert saved == []
+        assert marked == []
+        assert finished == [
+            (("completed-execution",), {
+                "success": False,
+                "error": "Fire claim ownership lost; stale result was discarded.",
+            })
+        ]
+
+    def test_transport_cancellation_remains_interrupted_when_claim_is_healthy(self, monkeypatch):
+        import cron.scheduler as sched
+
+        fire_claim_lost = threading.Event()
+        transport_cancel = threading.Event()
+        transport_cancel.set()
+        combined = sched._CombinedCancelEvent(fire_claim_lost, transport_cancel)
+        fence = sched._FireOwnership(
+            {"id": "transport-cancel", "fire_claim": {"by": "fire-owner"}}, combined
+        )
+        monkeypatch.setattr(sched, "heartbeat_fire_claim", lambda *_args, **_kwargs: True)
+
+        assert fence.lost() is True
+
+
 class TestRunOneJobHonoursInterruptedFlag:
     """run_one_job() must not let a job's own completion overwrite a
     status the shutdown path already wrote for the same run."""