diff --git a/cron/executions.py b/cron/executions.py
index 001d0d2c53d22..d0ab961b2f959 100644
--- a/cron/executions.py
+++ b/cron/executions.py
@@ -712,6 +712,16 @@ def give_up_execution(execution_id: str, *, reason: str) -> bool:
     return True
 
 
+def mark_unknown_skipped(execution_id: str, detail: str) -> bool:
+    """Record that an unknown shutdown execution's scheduled slot was skipped."""
+    with _transaction() as conn:
+        cur = conn.execute(
+            "UPDATE executions SET error=? WHERE id=? AND status='unknown'",
+            (str(detail), str(execution_id)),
+        )
+        return cur.rowcount == 1
+
+
 def list_executions(
     *, job_id: Optional[str] = None, limit: int = 50, before_claimed_at: Optional[str] = None,
 ) -> list[dict[str, Any]]:
diff --git a/cron/jobs.py b/cron/jobs.py
index c43a0c9915e2d..736d5e458dcea 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2462,10 +2462,11 @@ def mark_job_run(
 
     ``shutdown_interrupted``: the gateway stopped this run. An occurrence not yet counted stays
     pending as a replay (``pending_slot.replay``) and is counted when the replay finishes. A
-    replay cut off again is counted now and not replayed again.
+    recurring unknown execution is stamped by ID so receipt recovery can skip that occurrence.
     """
     def apply(jobs, _i, job):
-        if execution_id is not None and execution_id in job.get("canonical_completions", []):
+        if (execution_id is not None
+                and execution_id in (job.get("canonical_completions") or [])):
             return True
         if expected_fire_owner is not None:
             claim = job.get("fire_claim")
@@ -2496,7 +2497,9 @@ def apply(jobs, _i, job):
         _record_run_outcome(job, success, error, delivery_error, status, now)
         if pending is not None:
             job["pending_slot"] = pending
-        if execution_id is not None:  # retire ids recovery can no longer replay; latest stays last
+        # Defer the first interrupted occurrence's tombstone until replay finishes; otherwise
+        # completed_occurrence would reject the required replay as already completed.
+        if execution_id is not None and not keep_pending:
             from cron.scheduler_authority import replayable_completions as live
             job["canonical_completions"] = [*live(job_id, job.get("canonical_completions")), execution_id]
         _advance_after_run(job, now, ladder_rung=rung or keep_pending)
@@ -2507,6 +2510,11 @@ def apply(jobs, _i, job):
             slot = datetime.fromisoformat(owned).astimezone(_hermes_now().tzinfo).isoformat()
             job["pending_slot"] = pending_slot_stamp(
                 slot, _hermes_now(), replay={"started_at": started, "interrupted_at": now})
+            if execution_id is not None:
+                job["shutdown_interrupted_execution"] = execution_id
+        elif (execution_id is not None
+              and (execution_id == job.get("shutdown_interrupted_execution") or replay is not None)):
+            job.pop("shutdown_interrupted_execution", None)
 
         if not success and model_unreachable and not is_terminal_job(job):
             plan_retry(job, replay=replay)
@@ -2530,6 +2538,57 @@ def locked():
 
     return _under_fire_fence(job_id, locked)
 
+def skip_unknown_occurrence(
+    job_id: str, execution_id: str, error: str, *, pause_recorded=None,
+) -> bool:
+    """Book a shutdown-killed unknown occurrence as skipped, without replaying its side effects."""
+    def apply(jobs, _i, job):
+        completed = job.get("canonical_completions")
+        if isinstance(completed, list) and execution_id in completed:
+            if isinstance(pause_recorded, dict) and job.get("state") == "paused" \
+                    and (job.get("paused_at"), job.get("paused_reason")) \
+                    == (pause_recorded.get("paused_at"), pause_recorded.get("reason")):
+                job.update(enabled=True, state="scheduled", paused_at=None, paused_reason=None)
+                save_jobs(jobs)
+            return True
+        now = _hermes_now().isoformat()
+        detail = f"Skipped unknown shutdown occurrence: {error}"
+        job["last_run_at"] = now
+        job["last_status"] = "skipped"
+        job["last_error"] = detail
+        job["last_failure"] = {"at": now, "detail": detail}
+        from cron.scheduler_authority import replayable_completions
+        job["canonical_completions"] = [
+            *replayable_completions(job_id, completed), execution_id,
+        ]
+        job.pop("pending_slot", None)
+        job.pop("restored_replay", None)
+        job["fire_claim"] = None
+        if job.get("run_claim") is not None:
+            job["run_claim"] = None
+        if (isinstance(pause_recorded, dict) and job.get("state") == "paused"
+                and (job.get("paused_at"), job.get("paused_reason"))
+                == (pause_recorded.get("paused_at"), pause_recorded.get("reason"))):
+            job.update(enabled=True, state="scheduled", paused_at=None, paused_reason=None)
+        job["next_run_at"] = compute_next_run(job["schedule"], now)
+        job.pop("shutdown_interrupted_execution", None)
+        if job["next_run_at"] is not None:
+            if job.get("state") != "paused":
+                job["state"] = "scheduled"
+        else:
+            job["state"] = "error"
+            logger.error(
+                "Job '%s' (%s) could not compute next_run_at; "
+                "leaving enabled and marking state=error so the job is not silently disabled.",
+                job.get("name", job.get("id", "?")), job.get("schedule", {}).get("kind"))
+        save_jobs(jobs)
+        record_cron_missed(job)
+        logger.warning("CRON_UNKNOWN_OUTCOME_SKIPPED: job %s execution %s: %s",
+                       job_id, execution_id, error)
+        return True
+
+    return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
+
 
 def _write_oneshot_diagnostic(job: dict[str, Any], text: str, what: str) -> bool:
     """Best-effort operator-visible trace in the job's output dir; never breaks the caller."""
diff --git a/cron/scheduler.py b/cron/scheduler.py
index bc7d6e34a7f99..4f6fb8f15c010 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -597,7 +597,7 @@ def _inflight_home_path(home_key: str) -> Path:
 # In-flight state below is keyed by ``_inflight_key``; the public accessors still report bare job
 # ids so host-wide consumers (shutdown drain, idle-exit, metrics) see the union across profiles.
 _running_job_ids: set = set()
-_running_fire_owners: dict[tuple, dict[object, tuple[Optional[str], Path]]] = {}
+_running_fire_owners: dict[tuple, dict[object, tuple[Optional[str], Path, Optional[str]]]] = {}
 # Parent gateway threads synchronously waiting on restart-safe scope workers.
 # Shutdown must not misclassify these as ownerless in-process runs: the tool
 # process sweep cannot reach the worker's transient scope.
@@ -1123,30 +1123,30 @@ def mark_running_jobs_interrupted(
     with _running_lock:
         restart_safe_waiters = set(_restart_safe_waiter_job_ids)
         active_fires = [
-            (token, key, owner, profile_home)
+            (token, key, owner, profile_home, execution_id)
             for key, executions in _running_fire_owners.items()
             if key not in restart_safe_waiters
-            for token, (owner, profile_home) in executions.items()
+            for token, (owner, profile_home, execution_id) in executions.items()
         ]
         if only_owners is not None:
             active_fires = [fire for fire in active_fires if (fire[1][1], fire[2]) in only_owners]
-        registered_keys = {key for _t, key, _o, _p in active_fires}
+        registered_keys = {key for _t, key, _o, _p, _e in active_fires}
         if only_owners is None:
             # The key's home half IS the profile home this claim belongs to — the only record of
             # it for a claim that never reached ``_running_fire_owners``. Read the real Path back
             # from the key, never ``Path(key[0])``: the key is normcased.
             active_fires.extend(
-                (None, key, None, _inflight_home_path(key[0]))
+                (None, key, None, _inflight_home_path(key[0]), None)
                 for key in (
                     _running_job_ids - registered_keys - restart_safe_waiters
                 )
             )
         _interrupted_job_ids.update(
             token if token is not None else key
-            for token, key, _owner, _profile_home in active_fires
+            for token, key, _owner, _profile_home, _execution_id in active_fires
         )
     marked = []
-    for _token, key, fire_owner, profile_home in active_fires:
+    for _token, key, fire_owner, profile_home, execution_id in active_fires:
         job_id = key[1]
         if not fire_owner:
             logger.warning(
@@ -1162,6 +1162,7 @@ def mark_running_jobs_interrupted(
             with use_cron_store(profile_home):
                 if mark_job_run(
                     job_id, False, reason, expected_fire_owner=fire_owner,
+                    **({"execution_id": execution_id} if execution_id is not None else {}),
                     shutdown_interrupted=True):
                     marked.append(job_id)
         except Exception as e:
@@ -2901,7 +2902,7 @@ def run_one_job(
     _fire_key = _inflight_key(job["id"])
     with _running_lock:
         _running_fire_owners.setdefault(_fire_key, {})[execution_token] = (
-            fire_owner or None, profile_home)
+            fire_owner or None, profile_home, execution_id)
     try:
         with self_removal_delivery_scope(job["id"]):
             return _run_with_fire_claim_heartbeat(
diff --git a/cron/scheduler_authority.py b/cron/scheduler_authority.py
index eea00d6598a1f..ba7c54d868f07 100644
--- a/cron/scheduler_authority.py
+++ b/cron/scheduler_authority.py
@@ -156,9 +156,9 @@ def reconcile_pending(*, allow_connect=True):
     from hermes_constants import get_hermes_home
     from hermes_cli.gateway_client import connect_gateway
     from cron.executions import execution_owner_live, get_execution
-    from cron.jobs import get_job, pause_job
+    from cron.jobs import get_job, pause_job, skip_unknown_occurrence
     from cron.scheduler import (_RunDelivery, _FireOwnership, _apply_agent_failure_marker,
-                                _save_compose_deliver)
+                                _deliver_crash_failure, _fire_secret_scope, _save_compose_deliver)
     from cron.scheduler_bookkeeping import fail_empty_response, finish_completed_run
     from utils import atomic_json_write
     import logging
@@ -206,11 +206,53 @@ async def observe():
                 state = {'status': 'terminal', 'job': current,
                          'result': [False, f'# Cron Job: {params["job_id"]} (FAILED)\n\n{error}\n', '', error]}
             if state['status'] != 'terminal':
-                if state['status'] == 'unknown' and not record.get('pause_recorded'):
-                    reason = 'Canonical cron unknown; no automatic re-execution'
-                    record['pause_recorded'] = _pause_identity(pause_job(params['job_id'], reason=reason))
-                    atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
-                continue
+                if state['status'] == 'unknown':
+                    row = get_execution(params['request_id'])
+                    error = str((row or {}).get('error') or 'Canonical cron outcome is unknown')
+                    current = get_job(params['job_id'])
+                    if current is None:
+                        journal.unlink(missing_ok=True)
+                        continue
+                    recurring = current.get('schedule', {}).get('kind') in {'cron', 'interval'}
+                    if params['request_id'] in (current.get('canonical_completions') or []):
+                        _lift_pause(params['job_id'], record.get('pause_recorded'))
+                        journal.unlink(missing_ok=True)
+                        continue
+                    if recurring:
+                        from cron.occurrences import completed_occurrence
+                        scheduled_instant = (row or {}).get('scheduled_instant')
+                        if scheduled_instant and completed_occurrence(current, scheduled_instant):
+                            _lift_pause(params['job_id'], record.get('pause_recorded'))
+                            journal.unlink(missing_ok=True)
+                            continue
+                        shutdown_error = current.get('last_error') or error
+                        if current.get('shutdown_interrupted_execution') == params['request_id']:
+                            detail = f'Skipped unknown shutdown occurrence: {shutdown_error}'
+                            from cron.executions import mark_unknown_skipped
+                            # False means the unknown row was already settled or pruned.
+                            mark_unknown_skipped(params['request_id'], detail)
+                            if not skip_unknown_occurrence(
+                                    params['job_id'], params['request_id'], shutdown_error,
+                                    pause_recorded=record.get('pause_recorded')):
+                                raise RuntimeError('Could not book skipped unknown cron occurrence')
+                            journal.unlink(missing_ok=True)
+                            continue
+                    if not record.get('pause_recorded'):
+                        reason = 'Canonical cron unknown; no automatic re-execution'
+                        paused = pause_job(params['job_id'], reason=reason)
+                        record['pause_recorded'] = _pause_identity(paused)
+                        atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
+                    if record.get('pause_recorded') and not record.get('pause_notified'):
+                        record['pause_notified'] = True
+                        atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
+                        with _fire_secret_scope():
+                            _deliver_crash_failure(
+                                dict(current, execution_id=params['request_id']), error,
+                                adapters=None, loop=None)
+                        logging.getLogger(__name__).warning(
+                            'CRON_UNKNOWN_OUTCOME_PAUSED: job %s execution %s: %s',
+                            params['job_id'], params['request_id'], error)
+                    continue
             success, output, answer, error = state['result']
             job = dict(state['job'], **(state.get('job_flags') or {}))
             job['execution_id'] = params['request_id']
diff --git a/tests/cron/test_canonical_completion_retention.py b/tests/cron/test_canonical_completion_retention.py
index c22bb9ef9bfcd..0222b8fe43036 100644
--- a/tests/cron/test_canonical_completion_retention.py
+++ b/tests/cron/test_canonical_completion_retention.py
@@ -24,3 +24,20 @@ def test_completion_ids_retire_with_their_journal(tmp_path, monkeypatch):
         # The still-journaled completion replayed by recovery is a no-op, not a second run.
         assert jobs.mark_job_run(job_id, True, execution_id='pending') is True
         assert jobs.get_job(job_id)['repeat']['completed'] == completed
+
+
+def test_duplicate_completion_succeeds_after_fire_owner_changes(tmp_path, monkeypatch):
+    monkeypatch.setenv('HERMES_HOME', str(tmp_path / 'home'))
+    with jobs.use_cron_store(tmp_path / 'cron'):
+        job_id = jobs.create_job(prompt='report', schedule='every 1m', deliver='local')['id']
+        journal = scheduler_authority.journal_path(job_id, 'already-booked')
+        journal.parent.mkdir(parents=True, exist_ok=True)
+        journal.write_text('{}', encoding='utf-8')
+        assert jobs.mark_job_run(job_id, True, execution_id='already-booked')
+        records = jobs.load_jobs()
+        records[0]['fire_claim'] = {'by': 'replacement-owner'}
+        jobs.save_jobs(records)
+        assert jobs.mark_job_run(
+            job_id, True, expected_fire_owner='original-owner',
+            execution_id='already-booked')
+        assert jobs.get_job(job_id)['repeat']['completed'] == 1
diff --git a/tests/cron/test_shutdown_interrupt.py b/tests/cron/test_shutdown_interrupt.py
index 77c1013dbf0bb..76e627d347042 100644
--- a/tests/cron/test_shutdown_interrupt.py
+++ b/tests/cron/test_shutdown_interrupt.py
@@ -75,8 +75,8 @@ def test_marks_every_in_flight_job(self):
         profile_home = sched._get_hermes_home().resolve()
         sched._running_fire_owners.update(
             {
-                sched._inflight_key("job-1"): {object(): ("owner-1", profile_home)},
-                sched._inflight_key("job-2"): {object(): ("owner-2", profile_home)},
+                sched._inflight_key("job-1"): {object(): ("owner-1", profile_home, None)},
+                sched._inflight_key("job-2"): {object(): ("owner-2", profile_home, None)},
             }
         )
 
@@ -92,6 +92,7 @@ def test_marks_every_in_flight_job(self):
             assert c.args[1] is False
             assert "gateway shutdown" in c.args[2]
             assert c.kwargs["expected_fire_owner"] in {"owner-1", "owner-2"}
+            assert c.kwargs.get("execution_id") is None
 
     def test_sets_interrupted_flag_for_consumption_by_run_one_job(self):
         import cron.scheduler as sched
@@ -114,8 +115,8 @@ def test_one_job_marking_failure_does_not_block_the_others(self):
         profile_home = sched._get_hermes_home().resolve()
         sched._running_fire_owners.update(
             {
-                sched._inflight_key("job-1"): {object(): ("owner-1", profile_home)},
-                sched._inflight_key("job-2"): {object(): ("owner-2", profile_home)},
+                sched._inflight_key("job-1"): {object(): ("owner-1", profile_home, None)},
+                sched._inflight_key("job-2"): {object(): ("owner-2", profile_home, None)},
             }
         )
 
@@ -150,7 +151,7 @@ def test_stale_shutdown_cannot_clear_replacement_owner(self, tmp_path):
 
             sched._running_job_ids.add(sched._inflight_key(created["id"]))
             sched._running_fire_owners[sched._inflight_key(created["id"])] = {
-                object(): (stale_owner, profile_home)
+                object(): (stale_owner, profile_home, None)
             }
             marked = sched.mark_running_jobs_interrupted("shutdown")
             refreshed = jobs.get_job(created["id"])
@@ -167,12 +168,13 @@ def test_run_one_job_registers_owner_only_while_active(self):
 
         job = {
             "id": "owned-job",
+            "execution_id": "execution-1",
             "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner-1"},
         }
 
         def _observe_registry(current_job, run):
             assert list(sched._running_fire_owners[sched._inflight_key(current_job["id"])].values()) == [
-                ("owner-1", sched._get_hermes_home().resolve())
+                ("owner-1", sched._get_hermes_home().resolve(), "execution-1")
             ]
             return True
 
@@ -194,7 +196,8 @@ def hold_run(_job, _run):
             release.wait(timeout=2)
             return True
 
-        def mark(_job_id, _success, _reason, *, expected_fire_owner, shutdown_interrupted):
+        def mark(_job_id, _success, _reason, *, expected_fire_owner, shutdown_interrupted,
+                 execution_id=None):
             marked_owners.append((expected_fire_owner, shutdown_interrupted))
             return True
 
@@ -228,11 +231,12 @@ def test_shutdown_marks_each_owner_in_its_profile_store(self, monkeypatch, tmp_p
         profile_b = tmp_path / "b"
         observed = []
         sched._running_fire_owners[sched._inflight_key("same-job")] = {
-            object(): ("owner-a", profile_a),
-            object(): ("owner-b", profile_b),
+            object(): ("owner-a", profile_a, None),
+            object(): ("owner-b", profile_b, None),
         }
 
-        def mark(job_id, success, reason, *, expected_fire_owner, shutdown_interrupted):
+        def mark(job_id, success, reason, *, expected_fire_owner, shutdown_interrupted,
+                 execution_id=None):
             observed.append(
                 (
                     job_id,
@@ -300,7 +304,7 @@ def test_interruption_targets_only_the_interrupted_execution(self):
         profile_home = sched._get_hermes_home().resolve()
         old_token = object()
         sched._running_fire_owners[sched._inflight_key("job-1")] = {
-            old_token: ("owner-1", profile_home),
+            old_token: ("owner-1", profile_home, None),
         }
 
         with patch("cron.scheduler.mark_job_run", return_value=True):
@@ -319,8 +323,10 @@ def test_only_owners_marks_only_targeted_executions(self):
 
         profile_home = sched._get_hermes_home().resolve()
         token_a, token_b = object(), object()
-        sched._running_fire_owners[sched._inflight_key("job-a")] = {token_a: ("owner-a", profile_home)}
-        sched._running_fire_owners[sched._inflight_key("job-b")] = {token_b: ("owner-b", profile_home)}
+        sched._running_fire_owners[sched._inflight_key("job-a")] = {
+            token_a: ("owner-a", profile_home, None)}
+        sched._running_fire_owners[sched._inflight_key("job-b")] = {
+            token_b: ("owner-b", profile_home, None)}
 
         with patch("cron.scheduler.mark_job_run", return_value=True) as mock_mark:
             marked = sched.mark_running_jobs_interrupted(
@@ -342,7 +348,7 @@ def test_replacement_execution_of_same_job_is_not_poisoned(self):
         profile_home = sched._get_hermes_home().resolve()
         stale_token = object()
         sched._running_fire_owners[sched._inflight_key("job-1")] = {
-            stale_token: ("stale-owner", profile_home),
+            stale_token: ("stale-owner", profile_home, None),
         }
         with patch("cron.scheduler.mark_job_run", return_value=True):
             sched.mark_running_jobs_interrupted("shutdown")
diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
index 4236cb1f15ddf..d0bb2305ab6bc 100644
--- a/tests/cron/test_shutdown_replay.py
+++ b/tests/cron/test_shutdown_replay.py
@@ -31,7 +31,7 @@ def test_shutdown_sweep_preserves_interrupted_occurrence(clock, monkeypatch, tmp
         first = claim_due(job["id"], clock)
         owner = first["fire_claim"]["by"]
         key = scheduler._inflight_key(job["id"])
-        monkeypatch.setattr(scheduler, "_running_fire_owners", {key: {object(): (owner, tmp_path / "cron")}})
+        monkeypatch.setattr(scheduler, "_running_fire_owners", {key: {object(): (owner, tmp_path / "cron", None)}})
         monkeypatch.setattr(scheduler, "_interrupted_job_ids", set())
         assert job["id"] in scheduler.mark_running_jobs_interrupted("shutdown", only_owners={(job["id"], owner)})
         stored = jobs.get_job(job["id"])
diff --git a/tests/cron/test_unknown_outcome_pause.py b/tests/cron/test_unknown_outcome_pause.py
new file mode 100644
index 0000000000000..6dbb47e763905
--- /dev/null
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -0,0 +1,523 @@
+"""Unknown cron outcomes skip shutdown-killed recurring slots and alert on retained pauses."""
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime, timedelta
+import json
+import pytest
+
+from cron import executions, jobs, scheduler, scheduler_authority
+from hermes_cli import gateway_client
+
+
+_SHUTDOWN_ERROR = (
+    "Gateway shutdown ({phase}) killed the job's tool subprocess before the run finished."
+)
+
+
+def _peer(monkeypatch):
+    class Peer:
+        async def rpc(self, method, **params):
+            assert method == "cron.recover"
+            return {"status": "unknown", "result": None}
+
+    @asynccontextmanager
+    async def connected():
+        yield Peer()
+
+    monkeypatch.setattr(gateway_client, "connect_gateway", connected)
+
+
+def _journal(job, execution_id):
+    params = {"job_id": job["id"], "request_id": execution_id, "extra_prompt": None}
+    journal = scheduler_authority.journal_path(job["id"], execution_id)
+    journal.parent.mkdir(parents=True, exist_ok=True)
+    journal.write_text(json.dumps({"params": params, "receipt": None}), encoding="utf-8")
+    return journal
+
+
+def _make_unknown(execution_id, error="unknown_execution: owner lost"):
+    with executions._transaction() as conn:
+        conn.execute(
+            "UPDATE executions SET status='unknown', error=? WHERE id=?",
+            (error, execution_id),
+        )
+
+
+@pytest.mark.parametrize("shutdown_phase", ["post-interrupt", "final-cleanup"])
+def test_shutdown_unknown_recurring_slot_is_skipped_without_replay(
+        tmp_path, monkeypatch, caplog, shutdown_phase):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        killed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        shutdown_error = _SHUTDOWN_ERROR.format(phase=shutdown_phase)
+        key = scheduler._inflight_key(job["id"], tmp_path / "cron")
+        token = object()
+        scheduler._running_fire_owners[key] = {
+            token: (claim["fire_claim"]["by"], tmp_path / "cron", execution["id"]),
+        }
+        try:
+            assert job["id"] in scheduler.mark_running_jobs_interrupted(
+                shutdown_error, only_owners={(job["id"], claim["fire_claim"]["by"])})
+        finally:
+            scheduler._running_fire_owners.pop(key, None)
+            scheduler._interrupted_job_ids.discard(token)
+        assert jobs.get_job(job["id"])["pending_slot"]["replay"]
+        assert jobs.get_job(job["id"])["shutdown_interrupted_execution"] == execution["id"]
+        assert executions.give_up_execution(execution["id"], reason=shutdown_error)
+        journal = _journal(job, execution["id"])
+        scheduler_authority.pause_for_unknown(
+            job["id"], execution["id"], "unknown_execution: cron admission was not replayed")
+        now[0] += timedelta(hours=2, minutes=15)
+        expected_next = jobs.compute_next_run(job["schedule"], now[0].isoformat())
+
+        scheduler_authority.reconcile_pending()
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved["paused_reason"] is None
+        assert datetime.fromisoformat(saved["next_run_at"]) == datetime.fromisoformat(expected_next)
+        assert datetime.fromisoformat(saved["next_run_at"]) > now[0]
+        assert saved["last_status"] == "skipped"
+        assert "skipped" in saved["last_error"].lower()
+        assert shutdown_error in saved["last_error"]
+        assert saved.get("pending_slot") is None
+        assert jobs.get_due_jobs() == []
+        assert executions.get_execution(execution["id"])["status"] == "unknown"
+        assert "CRON_UNKNOWN_OUTCOME_SKIPPED" in caplog.text
+        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
+        assert not journal.exists()
+
+
+def test_replay_completion_before_unknown_receipt_recovery_is_not_skipped(
+        tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        killed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        first = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        shutdown_error = _SHUTDOWN_ERROR.format(phase="post-interrupt")
+        key = scheduler._inflight_key(job["id"], tmp_path / "cron")
+        token = object()
+        scheduler._running_fire_owners[key] = {
+            token: (first["fire_claim"]["by"], tmp_path / "cron", execution["id"]),
+        }
+        try:
+            assert job["id"] in scheduler.mark_running_jobs_interrupted(
+                shutdown_error, only_owners={(job["id"], first["fire_claim"]["by"])})
+        finally:
+            scheduler._running_fire_owners.pop(key, None)
+            scheduler._interrupted_job_ids.discard(token)
+        assert jobs.get_job(job["id"])["shutdown_interrupted_execution"] == execution["id"]
+        _journal(job, execution["id"])
+        assert executions.give_up_execution(execution["id"], reason=shutdown_error)
+
+        # This ordering is legal when the job replay finishes before the unknown receipt is
+        # reconciled. The replay carries the same scheduled instant in its own execution row.
+        now[0] += timedelta(minutes=1)
+        due = jobs.get_due_jobs()
+        assert any(candidate["id"] == job["id"] for candidate in due)
+        replay_execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        replay = jobs.claim_job_for_fire(
+            job["id"], return_job=True, execution_id=replay_execution["id"])
+        assert replay["fire_claim"]["replay"]
+        assert executions.mark_execution_running(replay_execution["id"])
+        assert jobs.mark_job_run(
+            job["id"], True, expected_fire_owner=replay["fire_claim"]["by"],
+            execution_id=replay_execution["id"])
+        assert executions.finish_execution(replay_execution["id"], success=True)
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["repeat"]["completed"] == 1
+        assert saved["last_status"] == "ok"
+        assert saved["state"] == "scheduled"
+        assert saved.get("pending_slot") is None
+        assert saved.get("shutdown_interrupted_execution") is None
+        assert executions.get_execution(replay_execution["id"])["status"] == "completed"
+        assert not scheduler_authority.journal_path(job["id"], execution["id"]).exists()
+
+
+def test_second_shutdown_during_replay_is_retired_without_pausing(
+        tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+    notices = []
+    monkeypatch.setattr(
+        scheduler, "_deliver_crash_failure",
+        lambda job, error, *, adapters, loop: notices.append((job["id"], error)))
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        first_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(first_slot)
+        first = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=first_slot)
+        first_claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=first_slot, execution_id=first["id"])
+        assert executions.mark_execution_running(first["id"])
+        first_key = scheduler._inflight_key(job["id"], tmp_path / "cron")
+        first_token = object()
+        scheduler._running_fire_owners[first_key] = {
+            first_token: (first_claim["fire_claim"]["by"], tmp_path / "cron", first["id"]),
+        }
+        try:
+            assert job["id"] in scheduler.mark_running_jobs_interrupted(
+                "gateway shutdown", only_owners={(job["id"], first_claim["fire_claim"]["by"])})
+        finally:
+            scheduler._running_fire_owners.pop(first_key, None)
+            scheduler._interrupted_job_ids.discard(first_token)
+        assert executions.finish_execution(first["id"], success=False, error="gateway shutdown")
+
+        now[0] += timedelta(minutes=1)
+        assert any(candidate["id"] == job["id"] for candidate in jobs.get_due_jobs())
+        replay = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=first_slot)
+        journal = _journal(job, replay["id"])
+        replay_claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, execution_id=replay["id"])
+        assert replay_claim["fire_claim"]["replay"]
+        assert executions.mark_execution_running(replay["id"])
+        replay_error = "Gateway shutdown (final-cleanup) killed the replay subprocess."
+        replay_key = scheduler._inflight_key(job["id"], tmp_path / "cron")
+        replay_token = object()
+        scheduler._running_fire_owners[replay_key] = {
+            replay_token: (replay_claim["fire_claim"]["by"], tmp_path / "cron", replay["id"]),
+        }
+        try:
+            assert job["id"] in scheduler.mark_running_jobs_interrupted(
+                replay_error,
+                only_owners={(job["id"], replay_claim["fire_claim"]["by"])})
+        finally:
+            scheduler._running_fire_owners.pop(replay_key, None)
+            scheduler._interrupted_job_ids.discard(replay_token)
+        expected_next = jobs.get_job(job["id"])["next_run_at"]
+        assert jobs.get_job(job["id"])["last_status"] == "error"
+        _make_unknown(replay["id"], replay_error)
+        scheduler_authority.reconcile_pending()
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved.get("paused_reason") is None
+        assert saved["last_status"] == "error"
+        assert saved["next_run_at"] == expected_next
+        assert saved["repeat"]["completed"] == 1
+        assert notices == []
+        assert not journal.exists()
+
+
+def test_completed_recurring_slot_retires_non_shutdown_unknown_receipt(
+        tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+    notices = []
+    monkeypatch.setattr(
+        scheduler, "_deliver_crash_failure",
+        lambda job, error, *, adapters, loop: notices.append((job["id"], error)))
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        completed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(completed_slot)
+        completed = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=completed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=completed_slot,
+            execution_id=completed["id"])
+        assert executions.mark_execution_running(completed["id"])
+        assert jobs.mark_job_run(
+            job["id"], True, expected_fire_owner=claim["fire_claim"]["by"],
+            execution_id=completed["id"])
+        assert executions.finish_execution(completed["id"], success=True)
+        expected_next = jobs.get_job(job["id"])["next_run_at"]
+
+        unknown = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=completed_slot)
+        assert executions.mark_execution_running(unknown["id"])
+        _make_unknown(unknown["id"], "unknown_execution: non-shutdown outcome")
+        journal = _journal(job, unknown["id"])
+        scheduler_authority.pause_for_unknown(
+            job["id"], unknown["id"], "unknown_execution: non-shutdown outcome")
+
+        scheduler_authority.reconcile_pending()
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved.get("paused_reason") is None
+        assert saved["next_run_at"] == expected_next
+        assert saved["repeat"]["completed"] == 1
+        assert notices == []
+        assert not journal.exists()
+
+
+def test_shutdown_skip_retries_after_ledger_settlement_crash(tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        killed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        shutdown_error = _SHUTDOWN_ERROR.format(phase="post-interrupt")
+        assert jobs.mark_job_run(
+            job["id"], False, shutdown_error,
+            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
+            shutdown_interrupted=True,
+        )
+        _make_unknown(execution["id"], shutdown_error)
+        journal = _journal(job, execution["id"])
+        scheduler_authority.pause_for_unknown(job["id"], execution["id"], "unknown outcome")
+        real_skip = jobs.skip_unknown_occurrence
+        calls = 0
+
+        def crash_once(*args, **kwargs):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                assert "skipped unknown shutdown occurrence" in (
+                    executions.get_execution(execution["id"])["error"].lower())
+                raise RuntimeError("simulated crash after ledger settlement")
+            return real_skip(*args, **kwargs)
+
+        monkeypatch.setattr(jobs, "skip_unknown_occurrence", crash_once)
+        scheduler_authority.reconcile_pending()
+        assert journal.exists()
+        assert jobs.get_job(job["id"])["shutdown_interrupted_execution"] == execution["id"]
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert calls == 2
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved["paused_reason"] is None
+        assert saved["last_status"] == "skipped"
+        assert execution["id"] in saved["canonical_completions"]
+        assert executions.get_execution(execution["id"])["status"] == "unknown"
+        assert not journal.exists()
+
+
+def test_shutdown_skip_retires_journal_after_job_write_crash(tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        killed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        shutdown_error = _SHUTDOWN_ERROR.format(phase="final-cleanup")
+        assert jobs.mark_job_run(
+            job["id"], False, shutdown_error,
+            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
+            shutdown_interrupted=True,
+        )
+        _make_unknown(execution["id"], shutdown_error)
+        journal = _journal(job, execution["id"])
+        scheduler_authority.pause_for_unknown(job["id"], execution["id"], "unknown outcome")
+        real_skip = jobs.skip_unknown_occurrence
+        calls = 0
+
+        def crash_after_skip(*args, **kwargs):
+            nonlocal calls
+            calls += 1
+            assert real_skip(*args, **kwargs)
+            raise RuntimeError("simulated crash before journal unlink")
+
+        monkeypatch.setattr(jobs, "skip_unknown_occurrence", crash_after_skip)
+        scheduler_authority.reconcile_pending()
+        after_skip = jobs.get_job(job["id"])
+        next_run_at = after_skip["next_run_at"]
+        assert journal.exists()
+        assert after_skip["state"] == "scheduled" and after_skip["enabled"] is True
+        assert after_skip["paused_reason"] is None
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert calls == 1
+        assert saved["last_status"] == "skipped"
+        assert saved["next_run_at"] == next_run_at
+        assert saved["repeat"]["completed"] == 0
+        assert not journal.exists()
+
+
+def test_shutdown_skip_is_booked_when_unknown_ledger_row_is_missing(tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        killed_slot = job["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        shutdown_error = _SHUTDOWN_ERROR.format(phase="post-interrupt")
+        assert jobs.mark_job_run(
+            job["id"], False, shutdown_error,
+            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
+            shutdown_interrupted=True,
+        )
+        _make_unknown(execution["id"], shutdown_error)
+        journal = _journal(job, execution["id"])
+        scheduler_authority.pause_for_unknown(job["id"], execution["id"], "unknown outcome")
+        with executions._transaction() as conn:
+            conn.execute("DELETE FROM executions WHERE id=?", (execution["id"],))
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved["paused_reason"] is None
+        assert saved["last_status"] == "skipped"
+        assert execution["id"] in saved["canonical_completions"]
+        assert not journal.exists()
+
+
+def test_shutdown_unknown_final_repeat_stays_active_without_spending_repeat(
+        tmp_path, monkeypatch):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", repeat=2, deliver="local")
+        now[0] = datetime.fromisoformat(job["next_run_at"])
+        assert jobs.mark_job_run(job["id"], True)
+        after_first_run = jobs.get_job(job["id"])
+        assert after_first_run["repeat"]["completed"] == 1
+        killed_slot = after_first_run["next_run_at"]
+        now[0] = datetime.fromisoformat(killed_slot)
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=killed_slot)
+        claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, scheduled_at=killed_slot, execution_id=execution["id"])
+        assert executions.mark_execution_running(execution["id"])
+        assert jobs.mark_job_run(
+            job["id"], False, _SHUTDOWN_ERROR.format(phase="post-interrupt"),
+            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
+            shutdown_interrupted=True,
+        )
+        _make_unknown(execution["id"], _SHUTDOWN_ERROR.format(phase="post-interrupt"))
+        _journal(job, execution["id"])
+        scheduler_authority.pause_for_unknown(
+            job["id"], execution["id"], "unknown_execution: cron admission was not replayed")
+        now[0] += timedelta(hours=2, minutes=15)
+        expected_next = jobs.compute_next_run(job["schedule"], now[0].isoformat())
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved["repeat"] == {"times": 2, "completed": 1}
+        assert datetime.fromisoformat(saved["next_run_at"]) == datetime.fromisoformat(expected_next)
+        assert datetime.fromisoformat(saved["next_run_at"]) > now[0]
+        assert saved["last_status"] == "skipped"
+        assert executions.get_execution(execution["id"])["status"] == "unknown"
+        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
+
+
+def test_recurring_unknown_without_shutdown_marker_pauses_and_notifies_once(
+        tmp_path, monkeypatch, caplog):
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: datetime(2026, 11, 9, 8, 0, tzinfo=UTC))
+    _peer(monkeypatch)
+    notices = []
+
+    def deliver(job, error, *, adapters, loop):
+        notices.append((job["id"], error))
+        return None, "delivered"
+
+    monkeypatch.setattr(scheduler, "_deliver_crash_failure", deliver)
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h", deliver="local")
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=job["next_run_at"])
+        assert executions.mark_execution_running(execution["id"])
+        _make_unknown(execution["id"], _SHUTDOWN_ERROR.format(phase="final-cleanup"))
+        _journal(job, execution["id"])
+
+        scheduler_authority.reconcile_pending()
+        scheduler_authority.reconcile_pending()
+        assert len(notices) == 1
+        assert notices[0][0] == job["id"]
+        assert notices[0][1] == _SHUTDOWN_ERROR.format(phase="final-cleanup")
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "paused" and saved["enabled"] is False
+        assert saved["paused_reason"] == "Canonical cron unknown; no automatic re-execution"
+        assert "CRON_UNKNOWN_OUTCOME_PAUSED" in caplog.text
+
+
+
+def test_one_shot_unknown_pause_notifies_once_across_recovery_ticks(tmp_path, monkeypatch, caplog):
+    now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(executions, "_hermes_now", lambda: now[0])
+    _peer(monkeypatch)
+    notices = []
+
+    def deliver(job, error, *, adapters, loop):
+        notices.append((job["id"], error, adapters, loop))
+        return None, "delivered"
+
+    monkeypatch.setattr(scheduler, "_deliver_crash_failure", deliver)
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "in 2h", deliver="local")
+        execution = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=job["next_run_at"])
+        assert executions.mark_execution_running(execution["id"])
+        _make_unknown(execution["id"])
+        _journal(job, execution["id"])
+
+        scheduler_authority.reconcile_pending()
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "paused" and saved["enabled"] is False
+        assert saved["paused_reason"] == "Canonical cron unknown; no automatic re-execution"
+        assert len(notices) == 1
+        assert notices[0][0] == job["id"]
+        assert "unknown" in notices[0][1].lower()
+        assert "CRON_UNKNOWN_OUTCOME_PAUSED" in caplog.text
diff --git a/tests/gateway/test_cron_active_work_drain.py b/tests/gateway/test_cron_active_work_drain.py
index 30f27b6cc56cd..eafedb397b257 100644
--- a/tests/gateway/test_cron_active_work_drain.py
+++ b/tests/gateway/test_cron_active_work_drain.py
@@ -89,7 +89,7 @@ async def test_in_flight_cron_job_marked_interrupted_on_forced_kill(self, monkey
 
         sched._running_job_ids.add(sched._inflight_key("job-1"))
         sched._running_fire_owners[sched._inflight_key("job-1")] = {
-            object(): ("owner-1", sched._get_hermes_home().resolve())
+            object(): ("owner-1", sched._get_hermes_home().resolve(), None)
         }
 
         monkeypatch.setattr(_pr.process_registry, "kill_all", lambda task_id=None: 1)
diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index bd3caeaf7576a..1e06a9290065f 100644
--- a/website/docs/developer-guide/cron-internals.md
+++ b/website/docs/developer-guide/cron-internals.md
@@ -218,6 +218,14 @@ occurrence does not count it again. If a replay is past grace and
 `cron.catch_up_missed` is false, the scheduler drops the occurrence and its
 marker. It cannot replay later.
 
+If an attempt is unknown after gateway shutdown killed its subprocess, the in-flight
+execution ID must match the execution stamped by the shutdown sweep. A recurring job
+then records the occurrence as skipped and advances to the next schedule slot. The
+unknown attempt remains in run history with the skipped detail. This does not consume
+`repeat.completed`: a finite schedule retains its remaining repeat so the next slot
+can still run. One-shot jobs and other unknown outcomes stay paused; recovery delivers
+their failure notice through the job's configured failure target.
+
 **Fire-claim lease during a run.** A firing run holds `fire_claim = {at, by}` and
 a heartbeat thread refreshes `at` every 60 s (the lease is 300 s). A heartbeat
 sample that reads the claim as someone else's is re-sampled once before it