fix(cron): preserve heartbeated external fire claims

cron/jobs.py

@@ -2546,6 +2546,17 @@ def apply(jobs, _i, job):
     return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
 
 
+def live_fire_claim_owner(job_id: str) -> str | None:
+    """Return the current unexpired fire owner under the durable claim fence."""
+    def apply(_jobs, _i, job):
+        claim = job.get("fire_claim")
+        if not _claim_is_live(claim, _hermes_now(), FIRE_CLAIM_TTL_SECONDS):
+            return None
+        return str(claim.get("by") or "") or None
+
+    return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, None))
+
+
 # Completed one-shots are retained in jobs.json (final status stays inspectable) and pruned by
 # _sweep_completed_oneshots once they age out.
 COMPLETED_ONESHOT_RETENTION_DAYS = 7

cron/scheduler.py

@@ -835,6 +835,13 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
             ):
                 reason = "ledger-terminal"
             elif age >= allowance:
+                # A restart-safe external worker intentionally outlives this
+                # process's Future. Its durable fire claim is heartbeated in
+                # the worker process; do not let the local in-flight sweeper
+                # steal that live claim and discard a legitimate long result.
+                from cron.jobs import live_fire_claim_owner
+                if live_fire_claim_owner(job_id) is not None:
+                    continue
                 reason = "age"
             else:
                 continue

tests/cron/test_recurring_wedge_selfheal.py

@@ -141,6 +141,26 @@ def test_two_consecutive_auto_fires_after_guard(self, cron_env, monkeypatch):
         assert latest2["status"] == "completed"
         assert latest2["id"] != latest1["id"], "two distinct executions"
 
+    def test_live_durable_fire_claim_outlives_local_future(self, cron_env, monkeypatch):
+        """A heartbeated external worker must not be reaped by the local age sweep."""
+        S, _E, env = self._setup(cron_env, monkeypatch)
+        import cron.jobs as J
+
+        job_id = env["job_id"]
+        claimed = J.claim_job_for_fire(job_id, force=True, return_job=True)
+        assert isinstance(claimed, dict)
+        owner = claimed["fire_claim"]["by"]
+
+        S._running_job_ids.clear()
+        S._running_since.clear()
+        S._running_futures.clear()
+        S._running_job_ids.add(job_id)
+        S._running_since[job_id] = time.time() - 6 * 60 * 60
+
+        assert S.sweep_stale_inflight([J.get_job(job_id)]) == []
+        assert job_id in S.get_running_job_ids()
+        assert J.live_fire_claim_owner(job_id) == owner
+
     def test_guard_stats_reported(self, cron_env, monkeypatch):
         """The guard must surface a countable forced-release signal."""
         S, E, env = self._setup(cron_env, monkeypatch)