From e58c6d939af3bed0846dca5060ee36b64ec3c0a1 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Mon, 9 Nov 2026 02:19:22 -0500
Subject: [PATCH 1/5] fix(cron): skip a shutdown-killed recurring slot instead
 of pausing (RIG-8801)

A recurring job whose run was killed by a gateway shutdown and whose outcome is unknown now skips only that slot and stays scheduled. Any pause that remains notifies through the cron failure delivery path once.

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/executions.py                            |  10 +
 cron/jobs.py                                  |  59 +++++-
 cron/scheduler.py                             |  17 +-
 cron/scheduler_authority.py                   |  57 ++++-
 tests/cron/test_shutdown_interrupt.py         |  34 +--
 tests/cron/test_shutdown_replay.py            |   2 +-
 tests/cron/test_unknown_outcome_pause.py      | 200 ++++++++++++++++++
 tests/gateway/test_cron_active_work_drain.py  |   2 +-
 .../docs/developer-guide/cron-internals.md    |   8 +
 9 files changed, 354 insertions(+), 35 deletions(-)
 create mode 100644 tests/cron/test_unknown_outcome_pause.py

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..0700c38a3e9f3 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2462,11 +2462,9 @@ 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", []):
-            return True
         if expected_fire_owner is not None:
             claim = job.get("fire_claim")
             if not isinstance(claim, dict) or claim.get("by") != expected_fire_owner:
@@ -2474,6 +2472,9 @@ def apply(jobs, _i, job):
                     "mark_job_run: job_id %s fire claim owner changed; discarding stale completion",
                     job_id)
                 return False
+        if (execution_id is not None
+                and execution_id in (job.get("canonical_completions") or [])):
+            return True
         now = _hermes_now().isoformat()
         claim = job.get("fire_claim")
         owned = claim.get("scheduled_at") if isinstance(claim, dict) else None
@@ -2496,7 +2497,7 @@ 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
+        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 +2508,10 @@ 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"):
+            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 +2535,52 @@ 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:
+            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..baeb351ce4026 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
@@ -170,6 +170,9 @@ def reconcile_pending(*, allow_connect=True):
             params = record['params']
             if journal != journal_path(params['job_id'], params['request_id']):
                 raise ValueError('cron journal identity conflict')
+            if record.get('unknown_skipped'):
+                journal.unlink(missing_ok=True)
+                continue
             if execution_owner_live(params['request_id'], params['job_id']):
                 # Its firer (this process included) still owns the row and its tail: re-saving the
                 # output or re-marking the job on every tick would duplicate its side effects.
@@ -206,11 +209,51 @@ 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
+                    skipped_execution = current.get('canonical_completions') or []
+                    if (current.get('schedule', {}).get('kind') in {'cron', 'interval'}
+                            and (current.get('shutdown_interrupted_execution') == params['request_id']
+                                 or (current.get('last_status') == 'skipped'
+                                     and isinstance(skipped_execution, list)
+                                     and params['request_id'] in skipped_execution))):
+                        shutdown_detail = str(current.get('last_error') or error)
+                        if current.get('last_status') != 'skipped':
+                            if not skip_unknown_occurrence(
+                                    params['job_id'], params['request_id'], shutdown_detail,
+                                    pause_recorded=record.get('pause_recorded')):
+                                raise RuntimeError('Could not book skipped unknown cron occurrence')
+                            skipped_detail = f'Skipped unknown shutdown occurrence: {shutdown_detail}'
+                        else:
+                            skipped_detail = shutdown_detail
+                        from cron.executions import mark_unknown_skipped
+                        if not mark_unknown_skipped(params['request_id'], skipped_detail):
+                            raise RuntimeError('Could not record skipped unknown cron execution')
+                        record['pause_recorded'] = None
+                        record['unknown_skipped'] = True
+                        atomic_json_write(journal, record, mode=0o600, fsync_dir=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_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..ad6d8ab72c657
--- /dev/null
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -0,0 +1,200 @@
+"""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)
+        assert jobs.mark_job_run(
+            job["id"], False, shutdown_error,
+            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
+            shutdown_interrupted=True,
+        )
+        assert jobs.get_job(job["id"])["pending_slot"]["replay"]
+        assert jobs.get_job(job["id"])["shutdown_interrupted_execution"] == execution["id"]
+        _make_unknown(execution["id"], 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_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

From e6e793f281914bca8f5e084a27f1cd1a0a499f68 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Mon, 9 Nov 2026 05:09:54 -0500
Subject: [PATCH 2/5] fix(cron): keep replay-before-recovery and crash retries
 single-run (RIG-8801)

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/jobs.py                                  |  12 +-
 cron/scheduler_authority.py                   |  18 ++-
 .../test_canonical_completion_retention.py    |  17 +++
 tests/cron/test_unknown_outcome_pause.py      | 138 +++++++++++++++++-
 4 files changed, 170 insertions(+), 15 deletions(-)

diff --git a/cron/jobs.py b/cron/jobs.py
index 0700c38a3e9f3..5b6ca585383fd 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2465,6 +2465,9 @@ def mark_job_run(
     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") or [])):
+            return True
         if expected_fire_owner is not None:
             claim = job.get("fire_claim")
             if not isinstance(claim, dict) or claim.get("by") != expected_fire_owner:
@@ -2472,9 +2475,6 @@ def apply(jobs, _i, job):
                     "mark_job_run: job_id %s fire claim owner changed; discarding stale completion",
                     job_id)
                 return False
-        if (execution_id is not None
-                and execution_id in (job.get("canonical_completions") or [])):
-            return True
         now = _hermes_now().isoformat()
         claim = job.get("fire_claim")
         owned = claim.get("scheduled_at") if isinstance(claim, dict) else None
@@ -2497,6 +2497,8 @@ def apply(jobs, _i, job):
         _record_run_outcome(job, success, error, delivery_error, status, now)
         if pending is not None:
             job["pending_slot"] = pending
+        # 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]
@@ -2510,7 +2512,8 @@ def apply(jobs, _i, job):
                 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"):
+        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):
@@ -2538,7 +2541,6 @@ def 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:
diff --git a/cron/scheduler_authority.py b/cron/scheduler_authority.py
index baeb351ce4026..fc2ffa6ded9f2 100644
--- a/cron/scheduler_authority.py
+++ b/cron/scheduler_authority.py
@@ -216,12 +216,22 @@ async def observe():
                     if current is None:
                         journal.unlink(missing_ok=True)
                         continue
+                    if current.get('schedule', {}).get('kind') in {'cron', 'interval'}:
+                        from cron.occurrences import completed_occurrence
+                        scheduled_instant = (row or {}).get('scheduled_instant')
+                        if scheduled_instant and completed_occurrence(current, scheduled_instant):
+                            # A shutdown replay already booked this occurrence. Its success owns
+                            # the schedule advancement; this older unknown receipt is now stale.
+                            journal.unlink(missing_ok=True)
+                            continue
                     skipped_execution = current.get('canonical_completions') or []
+                    is_shutdown_skip = (
+                        current.get('shutdown_interrupted_execution') == params['request_id']
+                        or (current.get('last_status') == 'skipped'
+                            and isinstance(skipped_execution, list)
+                            and params['request_id'] in skipped_execution))
                     if (current.get('schedule', {}).get('kind') in {'cron', 'interval'}
-                            and (current.get('shutdown_interrupted_execution') == params['request_id']
-                                 or (current.get('last_status') == 'skipped'
-                                     and isinstance(skipped_execution, list)
-                                     and params['request_id'] in skipped_execution))):
+                            and is_shutdown_skip):
                         shutdown_detail = str(current.get('last_error') or error)
                         if current.get('last_status') != 'skipped':
                             if not skip_unknown_occurrence(
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_unknown_outcome_pause.py b/tests/cron/test_unknown_outcome_pause.py
index ad6d8ab72c657..4a25dbdbc8c11 100644
--- a/tests/cron/test_unknown_outcome_pause.py
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -59,14 +59,20 @@ def test_shutdown_unknown_recurring_slot_is_skipped_without_replay(
             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)
-        assert jobs.mark_job_run(
-            job["id"], False, shutdown_error,
-            expected_fire_owner=claim["fire_claim"]["by"], execution_id=execution["id"],
-            shutdown_interrupted=True,
-        )
+        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"]
-        _make_unknown(execution["id"], shutdown_error)
+        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")
@@ -92,6 +98,126 @@ def test_shutdown_unknown_recurring_slot_is_skipped_without_replay(
         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_skip_reconciliation_finishes_after_crash_following_job_store_write(
+        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_execution: cron admission was not replayed")
+        now[0] += timedelta(hours=2, minutes=15)
+        expected_next = jobs.compute_next_run(job["schedule"], now[0].isoformat())
+        real_mark_unknown_skipped = executions.mark_unknown_skipped
+
+        def crash_after_jobs_write(execution_id, detail):
+            assert jobs.get_job(job["id"])["last_status"] == "skipped"
+            raise RuntimeError("simulated crash after jobs.json skip write")
+
+        monkeypatch.setattr(executions, "mark_unknown_skipped", crash_after_jobs_write)
+        scheduler_authority.reconcile_pending()
+        partial = jobs.get_job(job["id"])
+        assert partial["last_status"] == "skipped"
+        assert partial["state"] == "scheduled"
+        assert "shutdown_interrupted_execution" not in partial
+        assert execution["id"] in partial["canonical_completions"]
+        assert journal.exists()
+        assert not json.loads(journal.read_text(encoding="utf-8")).get("unknown_skipped")
+
+        monkeypatch.setattr(executions, "mark_unknown_skipped", real_mark_unknown_skipped)
+        scheduler_authority.reconcile_pending()
+        assert journal.exists()
+        assert json.loads(journal.read_text(encoding="utf-8")).get("unknown_skipped")
+        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
+
+        scheduler_authority.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled"
+        assert saved["repeat"]["completed"] == 0
+        assert saved["last_status"] == "skipped"
+        assert saved["next_run_at"] == expected_next
+        assert executions.get_execution(execution["id"])["status"] == "unknown"
+        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
+        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)]

From 1d61a6fccb837b3eefaf7a2241a2e47dbb1a0c3c Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Mon, 9 Nov 2026 05:43:15 -0500
Subject: [PATCH 3/5] fix(cron): retire receipts of already-booked replays
 without pausing (RIG-8801)

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/jobs.py                             |   1 +
 cron/scheduler_authority.py              |  15 ++-
 tests/cron/test_unknown_outcome_pause.py | 119 ++++++++++++++++++++++-
 3 files changed, 131 insertions(+), 4 deletions(-)

diff --git a/cron/jobs.py b/cron/jobs.py
index 5b6ca585383fd..db7f031a6ab16 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2541,6 +2541,7 @@ def 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:
diff --git a/cron/scheduler_authority.py b/cron/scheduler_authority.py
index fc2ffa6ded9f2..1b04771db26cc 100644
--- a/cron/scheduler_authority.py
+++ b/cron/scheduler_authority.py
@@ -220,11 +220,22 @@ async def observe():
                         from cron.occurrences import completed_occurrence
                         scheduled_instant = (row or {}).get('scheduled_instant')
                         if scheduled_instant and completed_occurrence(current, scheduled_instant):
-                            # A shutdown replay already booked this occurrence. Its success owns
-                            # the schedule advancement; this older unknown receipt is now stale.
+                            # A completed execution owns the schedule advancement; this unknown
+                            # receipt is stale and must not leave its recovery pause behind.
+                            _lift_pause(params['job_id'], record.get('pause_recorded'))
                             journal.unlink(missing_ok=True)
                             continue
                     skipped_execution = current.get('canonical_completions') or []
+                    already_booked = (
+                        isinstance(skipped_execution, list)
+                        and params['request_id'] in skipped_execution
+                        and current.get('last_status') != 'skipped')
+                    if already_booked:
+                        # A shutdown replay was interrupted a second time. Its occurrence was
+                        # already counted by mark_job_run; do not pause or skip it again.
+                        _lift_pause(params['job_id'], record.get('pause_recorded'))
+                        journal.unlink(missing_ok=True)
+                        continue
                     is_shutdown_skip = (
                         current.get('shutdown_interrupted_execution') == params['request_id']
                         or (current.get('last_status') == 'skipped'
diff --git a/tests/cron/test_unknown_outcome_pause.py b/tests/cron/test_unknown_outcome_pause.py
index 4a25dbdbc8c11..fa65d64b4a613 100644
--- a/tests/cron/test_unknown_outcome_pause.py
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -41,6 +41,7 @@ def _make_unknown(execution_id, error="unknown_execution: owner lost"):
             (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):
@@ -157,6 +158,119 @@ def test_replay_completion_before_unknown_receipt_recovery_is_not_skipped(
         assert executions.get_execution(replay_execution["id"])["status"] == "completed"
         assert not scheduler_authority.journal_path(job["id"], execution["id"]).exists()
 
+
+def test_unknown_after_replay_shutdown_is_already_booked_and_not_paused(
+        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.reconcile_pending()
+
+        saved = jobs.get_job(job["id"])
+        assert saved["state"] == "scheduled" and saved["enabled"] is True
+        assert saved["next_run_at"] == expected_next
+        assert saved["repeat"]["completed"] == 1
+        assert notices == []
+        assert not journal.exists()
+
+
 def test_skip_reconciliation_finishes_after_crash_following_job_store_write(
         tmp_path, monkeypatch):
     now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]
@@ -206,8 +320,8 @@ def crash_after_jobs_write(execution_id, detail):
         assert journal.exists()
         assert json.loads(journal.read_text(encoding="utf-8")).get("unknown_skipped")
         assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
-
         scheduler_authority.reconcile_pending()
+        assert not journal.exists()
 
         saved = jobs.get_job(job["id"])
         assert saved["state"] == "scheduled"
@@ -216,7 +330,7 @@ def crash_after_jobs_write(execution_id, detail):
         assert saved["next_run_at"] == expected_next
         assert executions.get_execution(execution["id"])["status"] == "unknown"
         assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
-        assert not journal.exists()
+
 
 def test_shutdown_unknown_final_repeat_stays_active_without_spending_repeat(
         tmp_path, monkeypatch):
@@ -261,6 +375,7 @@ def test_shutdown_unknown_final_repeat_stays_active_without_spending_repeat(
         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))

From 513dfd916ffaf26b13d1f5dd000f4ad4fbe6ae01 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Mon, 9 Nov 2026 06:09:28 -0500
Subject: [PATCH 4/5] fix(cron): settle a half-recorded skip before retiring
 its receipt (RIG-8801)

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/scheduler_authority.py              | 43 +++++++++++-----
 tests/cron/test_unknown_outcome_pause.py | 62 +++++++++++++++++++++++-
 2 files changed, 91 insertions(+), 14 deletions(-)

diff --git a/cron/scheduler_authority.py b/cron/scheduler_authority.py
index 1b04771db26cc..1c23750067e15 100644
--- a/cron/scheduler_authority.py
+++ b/cron/scheduler_authority.py
@@ -216,23 +216,23 @@ async def observe():
                     if current is None:
                         journal.unlink(missing_ok=True)
                         continue
-                    if current.get('schedule', {}).get('kind') in {'cron', 'interval'}:
-                        from cron.occurrences import completed_occurrence
-                        scheduled_instant = (row or {}).get('scheduled_instant')
-                        if scheduled_instant and completed_occurrence(current, scheduled_instant):
-                            # A completed execution owns the schedule advancement; this unknown
-                            # receipt is stale and must not leave its recovery pause behind.
-                            _lift_pause(params['job_id'], record.get('pause_recorded'))
-                            journal.unlink(missing_ok=True)
-                            continue
                     skipped_execution = current.get('canonical_completions') or []
+                    skip_detail = record.get('skip_detail')
                     already_booked = (
                         isinstance(skipped_execution, list)
                         and params['request_id'] in skipped_execution
                         and current.get('last_status') != 'skipped')
                     if already_booked:
                         # A shutdown replay was interrupted a second time. Its occurrence was
-                        # already counted by mark_job_run; do not pause or skip it again.
+                        # already counted by mark_job_run; settle a pending skip before cleanup.
+                        if skip_detail:
+                            if row is None or row.get('status') != 'unknown':
+                                raise RuntimeError(
+                                    'Cannot retire skipped cron receipt before its unknown ledger row')
+                            from cron.executions import mark_unknown_skipped
+                            if not mark_unknown_skipped(params['request_id'], skip_detail):
+                                raise RuntimeError(
+                                    'Could not record already-booked skipped cron execution')
                         _lift_pause(params['job_id'], record.get('pause_recorded'))
                         journal.unlink(missing_ok=True)
                         continue
@@ -240,18 +240,26 @@ async def observe():
                         current.get('shutdown_interrupted_execution') == params['request_id']
                         or (current.get('last_status') == 'skipped'
                             and isinstance(skipped_execution, list)
-                            and params['request_id'] in skipped_execution))
+                            and params['request_id'] in skipped_execution)
+                        or bool(skip_detail and isinstance(skipped_execution, list)
+                                and params['request_id'] in skipped_execution))
                     if (current.get('schedule', {}).get('kind') in {'cron', 'interval'}
                             and is_shutdown_skip):
                         shutdown_detail = str(current.get('last_error') or error)
-                        if current.get('last_status') != 'skipped':
+                        if skip_detail:
+                            skipped_detail = skip_detail
+                        elif current.get('last_status') != 'skipped':
+                            skipped_detail = f'Skipped unknown shutdown occurrence: {shutdown_detail}'
+                            record['skip_detail'] = skipped_detail
+                            atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
                             if not skip_unknown_occurrence(
                                     params['job_id'], params['request_id'], shutdown_detail,
                                     pause_recorded=record.get('pause_recorded')):
                                 raise RuntimeError('Could not book skipped unknown cron occurrence')
-                            skipped_detail = f'Skipped unknown shutdown occurrence: {shutdown_detail}'
                         else:
                             skipped_detail = shutdown_detail
+                            record['skip_detail'] = skipped_detail
+                            atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
                         from cron.executions import mark_unknown_skipped
                         if not mark_unknown_skipped(params['request_id'], skipped_detail):
                             raise RuntimeError('Could not record skipped unknown cron execution')
@@ -259,6 +267,15 @@ async def observe():
                         record['unknown_skipped'] = True
                         atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
                         continue
+                    if current.get('schedule', {}).get('kind') in {'cron', 'interval'}:
+                        from cron.occurrences import completed_occurrence
+                        scheduled_instant = (row or {}).get('scheduled_instant')
+                        if scheduled_instant and completed_occurrence(current, scheduled_instant):
+                            # A completed execution owns the schedule advancement; this unknown
+                            # receipt is stale and must not leave its recovery pause behind.
+                            _lift_pause(params['job_id'], record.get('pause_recorded'))
+                            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)
diff --git a/tests/cron/test_unknown_outcome_pause.py b/tests/cron/test_unknown_outcome_pause.py
index fa65d64b4a613..0c8ffe1861d4b 100644
--- a/tests/cron/test_unknown_outcome_pause.py
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -260,17 +260,77 @@ def test_completed_recurring_slot_retires_non_shutdown_unknown_receipt(
         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_skip_detail_survives_later_success_before_receipt_recovery(
+        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_execution: admission was not replayed")
+        real_mark_unknown_skipped = executions.mark_unknown_skipped
+
+        def crash_before_ledger_write(execution_id, detail):
+            assert jobs.get_job(job["id"])["last_status"] == "skipped"
+            assert json.loads(journal.read_text(encoding="utf-8"))["skip_detail"] == detail
+            raise RuntimeError("simulated crash before skipped ledger update")
+
+        monkeypatch.setattr(executions, "mark_unknown_skipped", crash_before_ledger_write)
+        scheduler_authority.reconcile_pending()
+        assert executions.get_execution(execution["id"])["error"] == shutdown_error
+
+        monkeypatch.setattr(executions, "mark_unknown_skipped", real_mark_unknown_skipped)
+        now[0] += timedelta(minutes=1)
+        later = executions.create_execution(
+            job["id"], source="builtin", scheduled_instant=jobs.get_job(job["id"])["next_run_at"])
+        later_claim = jobs.claim_job_for_fire(
+            job["id"], return_job=True, execution_id=later["id"])
+        assert executions.mark_execution_running(later["id"])
+        assert jobs.mark_job_run(
+            job["id"], True, expected_fire_owner=later_claim["fire_claim"]["by"],
+            execution_id=later["id"])
+        assert executions.finish_execution(later["id"], success=True)
+        assert jobs.get_job(job["id"])["last_status"] == "ok"
+
+        scheduler_authority.reconcile_pending()
+
+        saved_execution = executions.get_execution(execution["id"])
+        assert saved_execution["status"] == "unknown"
+        assert "skipped unknown shutdown occurrence" in saved_execution["error"].lower()
+        assert not journal.exists()
+
+
 def test_skip_reconciliation_finishes_after_crash_following_job_store_write(
         tmp_path, monkeypatch):
     now = [datetime(2026, 11, 9, 8, 0, tzinfo=UTC)]

From 7a5ccafee5e22ef1404677cd3576b6d688d7fd9a Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Mon, 9 Nov 2026 06:28:07 -0500
Subject: [PATCH 5/5] refactor(cron): one ordered, retry-safe path for unknown
 receipt recovery (RIG-8801)

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/jobs.py                             |   5 +
 cron/scheduler_authority.py              |  69 +++---------
 tests/cron/test_unknown_outcome_pause.py | 138 +++++++++++++----------
 3 files changed, 100 insertions(+), 112 deletions(-)

diff --git a/cron/jobs.py b/cron/jobs.py
index db7f031a6ab16..736d5e458dcea 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2545,6 +2545,11 @@ def skip_unknown_occurrence(
     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}"
diff --git a/cron/scheduler_authority.py b/cron/scheduler_authority.py
index 1c23750067e15..ba7c54d868f07 100644
--- a/cron/scheduler_authority.py
+++ b/cron/scheduler_authority.py
@@ -170,9 +170,6 @@ def reconcile_pending(*, allow_connect=True):
             params = record['params']
             if journal != journal_path(params['job_id'], params['request_id']):
                 raise ValueError('cron journal identity conflict')
-            if record.get('unknown_skipped'):
-                journal.unlink(missing_ok=True)
-                continue
             if execution_owner_live(params['request_id'], params['job_id']):
                 # Its firer (this process included) still owns the row and its tail: re-saving the
                 # output or re-marking the job on every tick would duplicate its side effects.
@@ -216,66 +213,30 @@ async def observe():
                     if current is None:
                         journal.unlink(missing_ok=True)
                         continue
-                    skipped_execution = current.get('canonical_completions') or []
-                    skip_detail = record.get('skip_detail')
-                    already_booked = (
-                        isinstance(skipped_execution, list)
-                        and params['request_id'] in skipped_execution
-                        and current.get('last_status') != 'skipped')
-                    if already_booked:
-                        # A shutdown replay was interrupted a second time. Its occurrence was
-                        # already counted by mark_job_run; settle a pending skip before cleanup.
-                        if skip_detail:
-                            if row is None or row.get('status') != 'unknown':
-                                raise RuntimeError(
-                                    'Cannot retire skipped cron receipt before its unknown ledger row')
-                            from cron.executions import mark_unknown_skipped
-                            if not mark_unknown_skipped(params['request_id'], skip_detail):
-                                raise RuntimeError(
-                                    'Could not record already-booked skipped cron execution')
+                    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
-                    is_shutdown_skip = (
-                        current.get('shutdown_interrupted_execution') == params['request_id']
-                        or (current.get('last_status') == 'skipped'
-                            and isinstance(skipped_execution, list)
-                            and params['request_id'] in skipped_execution)
-                        or bool(skip_detail and isinstance(skipped_execution, list)
-                                and params['request_id'] in skipped_execution))
-                    if (current.get('schedule', {}).get('kind') in {'cron', 'interval'}
-                            and is_shutdown_skip):
-                        shutdown_detail = str(current.get('last_error') or error)
-                        if skip_detail:
-                            skipped_detail = skip_detail
-                        elif current.get('last_status') != 'skipped':
-                            skipped_detail = f'Skipped unknown shutdown occurrence: {shutdown_detail}'
-                            record['skip_detail'] = skipped_detail
-                            atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
-                            if not skip_unknown_occurrence(
-                                    params['job_id'], params['request_id'], shutdown_detail,
-                                    pause_recorded=record.get('pause_recorded')):
-                                raise RuntimeError('Could not book skipped unknown cron occurrence')
-                        else:
-                            skipped_detail = shutdown_detail
-                            record['skip_detail'] = skipped_detail
-                            atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
-                        from cron.executions import mark_unknown_skipped
-                        if not mark_unknown_skipped(params['request_id'], skipped_detail):
-                            raise RuntimeError('Could not record skipped unknown cron execution')
-                        record['pause_recorded'] = None
-                        record['unknown_skipped'] = True
-                        atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
-                        continue
-                    if current.get('schedule', {}).get('kind') in {'cron', 'interval'}:
+                    if recurring:
                         from cron.occurrences import completed_occurrence
                         scheduled_instant = (row or {}).get('scheduled_instant')
                         if scheduled_instant and completed_occurrence(current, scheduled_instant):
-                            # A completed execution owns the schedule advancement; this unknown
-                            # receipt is stale and must not leave its recovery pause behind.
                             _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)
diff --git a/tests/cron/test_unknown_outcome_pause.py b/tests/cron/test_unknown_outcome_pause.py
index 0c8ffe1861d4b..6dbb47e763905 100644
--- a/tests/cron/test_unknown_outcome_pause.py
+++ b/tests/cron/test_unknown_outcome_pause.py
@@ -159,7 +159,7 @@ def test_replay_completion_before_unknown_receipt_recovery_is_not_skipped(
         assert not scheduler_authority.journal_path(job["id"], execution["id"]).exists()
 
 
-def test_unknown_after_replay_shutdown_is_already_booked_and_not_paused(
+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])
@@ -273,8 +273,7 @@ def test_completed_recurring_slot_retires_non_shutdown_unknown_receipt(
         assert not journal.exists()
 
 
-def test_skip_detail_survives_later_success_before_receipt_recovery(
-        tmp_path, monkeypatch):
+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])
@@ -297,42 +296,37 @@ def test_skip_detail_survives_later_success_before_receipt_recovery(
         )
         _make_unknown(execution["id"], shutdown_error)
         journal = _journal(job, execution["id"])
-        scheduler_authority.pause_for_unknown(
-            job["id"], execution["id"], "unknown_execution: admission was not replayed")
-        real_mark_unknown_skipped = executions.mark_unknown_skipped
-
-        def crash_before_ledger_write(execution_id, detail):
-            assert jobs.get_job(job["id"])["last_status"] == "skipped"
-            assert json.loads(journal.read_text(encoding="utf-8"))["skip_detail"] == detail
-            raise RuntimeError("simulated crash before skipped ledger update")
-
-        monkeypatch.setattr(executions, "mark_unknown_skipped", crash_before_ledger_write)
+        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 executions.get_execution(execution["id"])["error"] == shutdown_error
-
-        monkeypatch.setattr(executions, "mark_unknown_skipped", real_mark_unknown_skipped)
-        now[0] += timedelta(minutes=1)
-        later = executions.create_execution(
-            job["id"], source="builtin", scheduled_instant=jobs.get_job(job["id"])["next_run_at"])
-        later_claim = jobs.claim_job_for_fire(
-            job["id"], return_job=True, execution_id=later["id"])
-        assert executions.mark_execution_running(later["id"])
-        assert jobs.mark_job_run(
-            job["id"], True, expected_fire_owner=later_claim["fire_claim"]["by"],
-            execution_id=later["id"])
-        assert executions.finish_execution(later["id"], success=True)
-        assert jobs.get_job(job["id"])["last_status"] == "ok"
+        assert journal.exists()
+        assert jobs.get_job(job["id"])["shutdown_interrupted_execution"] == execution["id"]
 
         scheduler_authority.reconcile_pending()
 
-        saved_execution = executions.get_execution(execution["id"])
-        assert saved_execution["status"] == "unknown"
-        assert "skipped unknown shutdown occurrence" in saved_execution["error"].lower()
+        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_skip_reconciliation_finishes_after_crash_following_job_store_write(
-        tmp_path, monkeypatch):
+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])
@@ -347,7 +341,7 @@ def test_skip_reconciliation_finishes_after_crash_following_job_store_write(
         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")
+        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"],
@@ -355,41 +349,69 @@ def test_skip_reconciliation_finishes_after_crash_following_job_store_write(
         )
         _make_unknown(execution["id"], 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())
-        real_mark_unknown_skipped = executions.mark_unknown_skipped
+        scheduler_authority.pause_for_unknown(job["id"], execution["id"], "unknown outcome")
+        real_skip = jobs.skip_unknown_occurrence
+        calls = 0
 
-        def crash_after_jobs_write(execution_id, detail):
-            assert jobs.get_job(job["id"])["last_status"] == "skipped"
-            raise RuntimeError("simulated crash after jobs.json skip write")
+        def crash_after_skip(*args, **kwargs):
+            nonlocal calls
+            calls += 1
+            assert real_skip(*args, **kwargs)
+            raise RuntimeError("simulated crash before journal unlink")
 
-        monkeypatch.setattr(executions, "mark_unknown_skipped", crash_after_jobs_write)
+        monkeypatch.setattr(jobs, "skip_unknown_occurrence", crash_after_skip)
         scheduler_authority.reconcile_pending()
-        partial = jobs.get_job(job["id"])
-        assert partial["last_status"] == "skipped"
-        assert partial["state"] == "scheduled"
-        assert "shutdown_interrupted_execution" not in partial
-        assert execution["id"] in partial["canonical_completions"]
+        after_skip = jobs.get_job(job["id"])
+        next_run_at = after_skip["next_run_at"]
         assert journal.exists()
-        assert not json.loads(journal.read_text(encoding="utf-8")).get("unknown_skipped")
+        assert after_skip["state"] == "scheduled" and after_skip["enabled"] is True
+        assert after_skip["paused_reason"] is None
 
-        monkeypatch.setattr(executions, "mark_unknown_skipped", real_mark_unknown_skipped)
         scheduler_authority.reconcile_pending()
-        assert journal.exists()
-        assert json.loads(journal.read_text(encoding="utf-8")).get("unknown_skipped")
-        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
-        scheduler_authority.reconcile_pending()
-        assert not journal.exists()
 
         saved = jobs.get_job(job["id"])
-        assert saved["state"] == "scheduled"
+        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 saved["next_run_at"] == expected_next
-        assert executions.get_execution(execution["id"])["status"] == "unknown"
-        assert "skipped" in executions.get_execution(execution["id"])["error"].lower()
+        assert execution["id"] in saved["canonical_completions"]
+        assert not journal.exists()
 
 
 def test_shutdown_unknown_final_repeat_stays_active_without_spending_repeat(