From 5dc09439adbd953633988e8f32fccc873f9ca27d Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 19:30:26 -0500
Subject: [PATCH 1/6] feat(cron): keep a run cut off by a gateway restart as
 pending replay (RIG-8084)

A run the gateway cut off stays pending. Its occurrence counts when the
replay finishes, not at the first shutdown. A replay cut off again counts
at that shutdown and is not replayed again. A retry rung already counted
its occurrence and keeps its ladder across a shutdown.

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 cron/jobs.py                       |  96 +++++++-
 cron/occurrences.py                |  25 +-
 cron/scheduler.py                  |   1 +
 cron/scheduler_gateway_gate.py     |   5 +-
 cron/unreachable_retry.py          |  10 +-
 tests/cron/test_shutdown_replay.py | 361 +++++++++++++++++++++++++++++
 6 files changed, 482 insertions(+), 16 deletions(-)
 create mode 100644 tests/cron/test_shutdown_replay.py

diff --git a/cron/jobs.py b/cron/jobs.py
index 86c83d966c00..c2c8330c57d5 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2436,6 +2436,7 @@ def mark_job_run(
     ladder_rung: bool = False,
     quota_hold_seconds: Optional[float] = None,
     recover_consumed_fire: bool = False,
+    shutdown_interrupted: bool = False,
 ) -> bool:
     """Mark a job as run: update last_run_at/last_status, bump completed, recompute next_run_at,
     and retire the record as a terminal completion when the repeat limit is reached.
@@ -2457,6 +2458,10 @@ def mark_job_run(
     it on every tick. ``recover_consumed_fire`` lets a scheduled sparse cron recover its
     consumed fire when the provider reopens; manual runs retain the natural schedule
     (cron/quota_hold.py, #89376).
+
+    ``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 (Q2 ruling, RIG-8084).
     """
     def apply(jobs, _i, job):
         if execution_id is not None and execution_id in job.get("canonical_completions", []):
@@ -2469,18 +2474,37 @@ def apply(jobs, _i, job):
                     job_id)
                 return False
         now = _hermes_now().isoformat()
+        claim = job.get("fire_claim")
+        owned = claim.get("scheduled_at") if isinstance(claim, dict) else None
+        replay = _claim_replay(job)
+        from cron import quota_hold
+        from cron.occurrences import pending_slot_stamp
+        from cron.unreachable_retry import STATE_KEY, clear_state, is_retry_run, plan_retry
+
+        rung = is_retry_run({"_scheduled_instant": owned, STATE_KEY: job.get(STATE_KEY)})
+
+        # A cut-off first run stays pending, uncounted: the replay counts it. A rung already
+        # counted its occurrence; a replay itself counts once when it finishes or is cut off.
+        keep_pending = shutdown_interrupted and owned is not None and replay is None
+        if owned is not None and not rung:
+            clear_state(job)  # a parked rung for another occurrence is stale (design case 14)
         _record_run_outcome(job, success, error, delivery_error, status, now)
         if execution_id is not None:  # retire ids recovery can no longer replay; latest stays last
             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=ladder_rung)
-        from cron import quota_hold
-        from cron.unreachable_retry import clear_state, plan_retry
+        _advance_after_run(job, now, ladder_rung=rung or keep_pending)
+        if keep_pending and job.get("schedule", {}).get("kind") in {"cron", "interval"} \
+                and not is_terminal_job(job):
+            # Started-at is the claim's first-claim time; heartbeats never move it.
+            started = claim.get("claimed_at") or claim.get("at")
+            job["pending_slot"] = pending_slot_stamp(
+                owned, _hermes_now(), replay={"started_at": started, "interrupted_at": now})
 
         if not success and model_unreachable and not is_terminal_job(job):
-            plan_retry(job)
-        else:
-            # Any run that reached the model (either outcome) resets the re-run ladder.
+            plan_retry(job, replay=replay)
+        elif not (shutdown_interrupted and rung):
+            # Any run that reached the model (either outcome) resets the re-run ladder. A rung
+            # cut off by shutdown keeps its ladder: its replay runs as the same rung.
             clear_state(job)
         if not success and quota_hold_seconds and not is_terminal_job(job):
             quota_hold.plan_hold(job, quota_hold_seconds, recover_consumed_fire=recover_consumed_fire)
@@ -2719,6 +2743,37 @@ def _machine_id() -> str:
     return f"{host}:{os.getpid()}"
 
 
+def _claim_replay(job: dict[str, Any]) -> Optional[dict[str, Any]]:
+    from cron.occurrences import claim_replay
+    return claim_replay(job)
+
+
+def _carry_replay(job: dict[str, Any], owned: Optional[str]) -> None:
+    """Copy the replay marker of the occurrence being claimed onto its fire claim (RIG-8084).
+
+    The marker comes from the pending slot when it holds this occurrence, else from a parked
+    retry rung when that rung is this occurrence. Other occurrences carry none."""
+    from cron.occurrences import scheduled_instant, slot_replay
+    from cron.unreachable_retry import STATE_KEY, ladder_replay
+
+    claim = job.get("fire_claim")
+    if not isinstance(claim, dict) or owned is None:
+        return
+    pending = job.get("pending_slot")
+    restored = job.get("restored_replay")
+    replay = None
+    if isinstance(pending, dict) and scheduled_instant(pending.get("scheduled_at")) == owned:
+        replay = slot_replay(job)
+    if replay is None and isinstance(restored, dict) \
+            and scheduled_instant(restored.get("scheduled_at")) == owned:
+        replay = restored.get("replay") if isinstance(restored.get("replay"), dict) else None
+    if replay is None and scheduled_instant((job.get(STATE_KEY) or {}).get("at")) == owned:
+        replay = ladder_replay(job)
+    job.pop("restored_replay", None)
+    if replay is not None:
+        claim["replay"] = dict(replay)
+
+
 def claim_job_for_fire(
     job_id: str, *, claim_ttl_seconds: int = FIRE_CLAIM_TTL_SECONDS, force: bool = False,
     manual: bool = False, return_job: bool = False, scheduled_at: Optional[str] = None,
@@ -2793,11 +2848,12 @@ def apply(jobs, _i, job):
             _activate_job_record(job)
         # Per-acquisition token: a process may legitimately reclaim its own stale lease, and the
         # previous runner must not heartbeat the new claim merely because hostname + PID match.
-        job["fire_claim"] = {"at": now.isoformat(), "by": token}
+        job["fire_claim"] = {"at": now.isoformat(), "claimed_at": now.isoformat(), "by": token}
         # The occurrence this claim owns: the one the caller dispatches, else the due instant.
         owned = None if manual_fire else (scheduled_instant(scheduled_at) if scheduled_at else instant)
         if owned:
             job["fire_claim"]["scheduled_at"] = owned
+        _carry_replay(job, owned)
         if execution_id:
             bind_row(owned)
         # Claimed: the occurrence is now owned by a run (its ledger row + fire claim carry it).
@@ -3144,6 +3200,12 @@ def _fast_forward_missed_recurring(d: _DueJob, grace: int) -> bool:
             "Job '%s' missed its scheduled time (%s, grace=%ds). "
             "Skipping missed occurrence because cron.catch_up_missed is false; next run: %s",
             d.label, d.next_run, grace, new_next)
+        # A dropped slot takes its replay marker with it (design case 12).
+        d.job.pop("restored_replay", None)
+        rj = d.scan.find(d.job["id"])
+        if rj is not None:
+            rj.pop("restored_replay", None)
+            d.scan.needs_save = True
         d.scan.on_saved.append(lambda job=d.job: record_cron_missed(job))
         return True
     logger.info(
@@ -3225,11 +3287,12 @@ def _restore_unclaimed_slot(job: dict[str, Any], scan: _DueScan) -> Optional[str
     (#107485); returns the restored ``next_run_at`` or None. Restored ONCE: the stamp is dropped
     here, so the slot then meets the ordinary late / fast-forward / ``cron.catch_up_missed``
     policy like any other overdue instant — never a replay of every missed slot."""
-    from cron.occurrences import unclaimed_pending_slot
+    from cron.occurrences import slot_replay, unclaimed_pending_slot
 
     slot = unclaimed_pending_slot(job, scan.now)
     if slot is None:
         return None
+    replay = slot_replay(job)
     logger.warning(
         "Job '%s' (%s): occurrence %s was taken off the schedule but never claimed "
         "(scheduler stopped before dispatch); restoring it as the due instant (was %s).",
@@ -3239,6 +3302,13 @@ def _restore_unclaimed_slot(job: dict[str, Any], scan: _DueScan) -> Optional[str
     rj = scan.find(job["id"])
     if rj is not None:
         rj.pop("pending_slot", None)
+    if replay is not None:
+        # The restored instant is still a replay: keep the marker for the next claim of this slot.
+        job["restored_replay"] = {"scheduled_at": slot, "replay": replay}
+        if rj is not None:
+            rj["restored_replay"] = job["restored_replay"]
+        scan.persist(job["id"], next_run_at=slot, restored_replay=job["restored_replay"])
+        return slot
     scan.persist(job["id"], next_run_at=slot)
     return slot
 
@@ -3282,7 +3352,7 @@ def apply(jobs, _i, job):
             now = _hermes_now()
             # Store in the scheduler's zone: a UTC string would trip the timezone-shift repair.
             slot = datetime.fromisoformat(instant).astimezone(now.tzinfo).isoformat()
-            job["pending_slot"] = pending_slot_stamp(slot, now)
+            job["pending_slot"] = pending_slot_stamp(slot, now, replay=_claim_replay(job))
             drop_claimed_occurrence(job)
         else:
             return False
@@ -3369,11 +3439,15 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float)
         job["last_dispatch"] = dispatch_stamp
         # The tick advances next_run_at past this occurrence before any fire claim exists; the
         # stamp survives a process death in that window so the slot is restored, not lost.
-        from cron.occurrences import pending_slot_stamp
+        from cron.occurrences import pending_slot_stamp, slot_replay
 
+        replay = slot_replay(job)
+        restored = job.get("restored_replay")
+        if replay is None and isinstance(restored, dict) and restored.get("scheduled_at") == next_run:
+            replay = restored.get("replay")  # a restored replay keeps its marker through this stamp
         scan.persist(
             job["id"], last_dispatch=dispatch_stamp,
-            pending_slot=pending_slot_stamp(next_run, now))
+            pending_slot=pending_slot_stamp(next_run, now, replay=replay))
     return True
 
 
diff --git a/cron/occurrences.py b/cron/occurrences.py
index 23c7eed0f682..a310bb1ccfb3 100644
--- a/cron/occurrences.py
+++ b/cron/occurrences.py
@@ -68,11 +68,30 @@ def completed_occurrence(job, instant):
 # expired) was never claimed, so it is restored ONCE as the due instant and then flows through
 # the ordinary late / fast-forward / ``cron.catch_up_missed`` policy — never N replays.
 
-def pending_slot_stamp(next_run, now):
-    """Store value for a recurring occurrence about to be handed to the dispatcher."""
+def pending_slot_stamp(next_run, now, replay=None):
+    """Store value for a recurring occurrence about to be handed to the dispatcher.
+
+    ``replay`` marks an interrupted run kept as pending work (RIG-8084); it is stored only when set."""
     from cron.jobs import _machine_id
 
-    return {"scheduled_at": next_run, "at": now.isoformat(), "by": _machine_id()}
+    stamp = {"scheduled_at": next_run, "at": now.isoformat(), "by": _machine_id()}
+    if replay is not None:
+        stamp["replay"] = replay
+    return stamp
+
+
+def slot_replay(job):
+    """The replay marker of a pending slot, or None."""
+    pending = job.get("pending_slot")
+    replay = pending.get("replay") if isinstance(pending, dict) else None
+    return replay if isinstance(replay, dict) else None
+
+
+def claim_replay(job):
+    """The replay marker of a live fire claim, or None."""
+    claim = job.get("fire_claim")
+    replay = claim.get("replay") if isinstance(claim, dict) else None
+    return replay if isinstance(replay, dict) else None
 
 
 def unclaimed_pending_slot(job, now):
diff --git a/cron/scheduler.py b/cron/scheduler.py
index 421060e1445d..2fdbd1458f88 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -2939,6 +2939,7 @@ def _record_fire_ownership_lost(
     if fire_owner is not None and heartbeat_fire_claim(job_id, expected_owner=fire_owner):
         mark_job_run(
             job_id, False, _OWNERSHIP_LOST_INTERRUPTED, expected_fire_owner=fire_owner,
+            shutdown_interrupted=True,
             **({"ladder_rung": True} if is_retry_run(job) else {}),
         )
         finish_execution(execution_id, success=False, error=_OWNERSHIP_LOST_INTERRUPTED)
diff --git a/cron/scheduler_gateway_gate.py b/cron/scheduler_gateway_gate.py
index 8433472d1c47..15cc6d9e7818 100644
--- a/cron/scheduler_gateway_gate.py
+++ b/cron/scheduler_gateway_gate.py
@@ -57,7 +57,10 @@ def _note_fire_refused(job_ids, detail: str) -> None:
             if job["id"] not in ids:
                 continue
             job["last_fire_error"] = {"at": _hermes_now().isoformat(), "detail": detail[:500]}
-            job.pop("pending_slot", None)
+            # A replay's slot was never handed over: keep it so the next tick still replays it.
+            pending = job.get("pending_slot")
+            if not (isinstance(pending, dict) and isinstance(pending.get("replay"), dict)):
+                job.pop("pending_slot", None)
             if job.get("run_claim") is not None:
                 job["run_claim"] = None
         save_jobs(jobs)
diff --git a/cron/unreachable_retry.py b/cron/unreachable_retry.py
index a576ba5fe7f7..c894ad57534e 100644
--- a/cron/unreachable_retry.py
+++ b/cron/unreachable_retry.py
@@ -132,7 +132,13 @@ def is_retry_run(job: dict[str, Any]) -> bool:
     return instant is not None and scheduled_instant((job.get(STATE_KEY) or {}).get("at")) == instant
 
 
-def plan_retry(job: dict[str, Any]) -> bool:
+def ladder_replay(job: dict[str, Any]) -> dict[str, Any] | None:
+    """The replay marker a parked rung carries, or None."""
+    replay = (job.get(STATE_KEY) or {}).get("replay")
+    return replay if isinstance(replay, dict) else None
+
+
+def plan_retry(job: dict[str, Any], replay: dict[str, Any] | None = None) -> bool:
     """Called under the jobs lock AFTER ``_advance_after_run`` computed the schedule's
     natural ``next_run_at`` for a failed, flagged run. Pulls ``next_run_at`` earlier to
     the ladder instant when that is sooner than the natural occurrence; exhausted or
@@ -155,6 +161,8 @@ def plan_retry(job: dict[str, Any]) -> bool:
     delay = RETRY_DELAYS_SECONDS[attempt]
     retry_at = retry_dt.isoformat()
     job[STATE_KEY] = {"attempt": attempt + 1, "at": retry_at, "expr": job["schedule"].get("expr")}
+    if replay is not None:
+        job[STATE_KEY]["replay"] = replay
     job["next_run_at"] = retry_at
     if job.get("state") != "paused":
         job["state"] = "scheduled"
diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
new file mode 100644
index 000000000000..7d2fbc08c45e
--- /dev/null
+++ b/tests/cron/test_shutdown_replay.py
@@ -0,0 +1,361 @@
+"""Interrupted recurring occurrences remain pending and spend repeat budget once."""
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from cron import jobs, unreachable_retry
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    now = [datetime.now(UTC)]
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now[0])
+    monkeypatch.setattr(unreachable_retry, "_hermes_now", lambda: now[0])
+    return now
+
+
+def claim_due(job_id, clock):
+    clock[0] += timedelta(seconds=1)
+    due = jobs.get_due_jobs()
+    assert any(job["id"] == job_id for job in due)
+    return jobs.claim_job_for_fire(job_id, return_job=True)
+
+
+def test_final_repeat_is_counted_when_replay_finishes(clock):
+    job = jobs.create_job("report", "every 24h", repeat=1)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    owner = first["fire_claim"]["by"]
+    started = first["fire_claim"]["claimed_at"]
+    assert jobs.heartbeat_fire_claim(job["id"], expected_owner=owner)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=owner,
+        shutdown_interrupted=True,
+    )
+    pending = jobs.get_job(job["id"])
+    assert pending["repeat"]["completed"] == 0
+    assert pending["pending_slot"]["replay"]["started_at"] == started
+    replay = claim_due(job["id"], clock)
+    assert replay["fire_claim"]["replay"]["started_at"] == started
+    assert jobs.mark_job_run(
+        job["id"], True, expected_fire_owner=replay["fire_claim"]["by"]
+    )
+    assert jobs.get_job(job["id"])["repeat"]["completed"] == 1
+
+
+def test_second_shutdown_does_not_replay_again(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"],
+        shutdown_interrupted=True,
+    )
+    replay = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=replay["fire_claim"]["by"],
+        shutdown_interrupted=True,
+    )
+    result = jobs.get_job(job["id"])
+    assert result["repeat"]["completed"] == 1
+    assert "pending_slot" not in result
+
+
+def test_interrupted_retry_rung_does_not_count_again(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    assert jobs.mark_job_run(job["id"], False, "offline", model_unreachable=True)
+    retry = jobs.get_job(job["id"])["unreachable_retry"]
+    clock[0] = datetime.fromisoformat(retry["at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"],
+        shutdown_interrupted=True,
+    )
+    assert jobs.get_job(job["id"])["repeat"]["completed"] == 1
+    replay = claim_due(job["id"], clock)
+    assert replay["fire_claim"]["scheduled_at"] == first["fire_claim"]["scheduled_at"]
+    assert "replay" in replay["fire_claim"]
+    assert jobs.mark_job_run(
+        job["id"], True, expected_fire_owner=replay["fire_claim"]["by"]
+    )
+    assert jobs.get_job(job["id"])["repeat"]["completed"] == 1
+
+
+def test_unreachable_replay_carries_marker_to_retry(clock):
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"],
+        shutdown_interrupted=True,
+    )
+    replay = claim_due(job["id"], clock)
+    marker = replay["fire_claim"]["replay"]
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "offline",
+        expected_fire_owner=replay["fire_claim"]["by"],
+        model_unreachable=True,
+    )
+    stored = jobs.get_job(job["id"])
+    assert stored["unreachable_retry"]["replay"] == marker
+    clock[0] = datetime.fromisoformat(stored["unreachable_retry"]["at"])
+    rung = claim_due(job["id"], clock)
+    assert rung["fire_claim"]["replay"] == marker
+    assert jobs.mark_job_run(
+        job["id"], True, expected_fire_owner=rung["fire_claim"]["by"]
+    )
+    assert "unreachable_retry" not in jobs.get_job(job["id"])
+
+
+@pytest.mark.parametrize("claim", [None, "malformed"])
+def test_no_claim_counts_without_replay(clock, claim):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    records = jobs.load_jobs()
+    records[0]["fire_claim"] = claim
+    jobs.save_jobs(records)
+    assert jobs.mark_job_run(job["id"], False, "shutdown", shutdown_interrupted=True)
+    stored = jobs.get_job(job["id"])
+    assert stored["repeat"]["completed"] == 1
+    assert "pending_slot" not in stored
+
+
+def test_restore_unstarted_preserves_marker(clock):
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"],
+        False,
+        "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"],
+        shutdown_interrupted=True,
+    )
+    replay = claim_due(job["id"], clock)
+    claim = replay["fire_claim"]
+    assert jobs.restore_unstarted_occurrence(
+        job["id"], claim["scheduled_at"], claim["by"]
+    )
+    assert jobs.get_job(job["id"])["pending_slot"]["replay"] == claim["replay"]
+
+
+def test_refused_dispatch_keeps_replay_stamp(clock, monkeypatch):
+    from cron import scheduler_gateway_gate
+
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    clock[0] += timedelta(seconds=1)
+    assert any(j["id"] == job["id"] for j in jobs.get_due_jobs())
+    scheduler_gateway_gate._note_fire_refused([job["id"]], "no gateway")
+    stored = jobs.get_job(job["id"])
+    assert stored["pending_slot"]["replay"] is not None
+    assert stored["last_fire_error"]["detail"] == "no gateway"
+
+
+def test_second_shutdown_does_not_stamp_again(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    replay = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=replay["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    stored = jobs.get_job(job["id"])
+    assert stored["repeat"]["completed"] == 1
+    assert "pending_slot" not in stored or "replay" not in stored["pending_slot"]
+
+
+def test_one_shot_interruption_stamps_no_replay(clock):
+    job = jobs.create_job("report", "in 30 minutes", repeat=1)
+    clock[0] = datetime.fromisoformat(jobs.get_job(job["id"])["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    assert "replay" not in (jobs.get_job(job["id"]).get("pending_slot") or {})
+
+
+def test_no_claim_stamps_nothing(clock):
+    job = jobs.create_job("report", "every 24h")
+    assert jobs.mark_job_run(job["id"], False, "shutdown", shutdown_interrupted=True)
+    assert "replay" not in (jobs.get_job(job["id"]).get("pending_slot") or {})
+
+
+def _interrupted_replay_slot(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    return job["id"]
+
+
+def _lock_fd(tmp_path):
+    def acquire(_sched):
+        return open(tmp_path / "tick.lock", "a+")
+
+    return acquire
+
+
+def test_estop_tick_leaves_replay_pending(clock, monkeypatch, tmp_path):
+    from agent import estop
+    from cron import scheduler_tick
+
+    job_id = _interrupted_replay_slot(clock)
+    before = jobs.get_job(job_id)["pending_slot"]
+    monkeypatch.setattr(estop, "check_paused", lambda *_a, **_k: True)
+    monkeypatch.setattr(scheduler_tick, "_acquire_tick_lock_or_degrade", _lock_fd(tmp_path))
+    assert scheduler_tick._tick_admitted(False, None, None, True) == 0
+    assert jobs.get_job(job_id)["pending_slot"] == before
+
+
+def test_draining_tick_leaves_replay_pending(clock, monkeypatch, tmp_path):
+    from cron import scheduler_tick
+
+    job_id = _interrupted_replay_slot(clock)
+    before = jobs.get_job(job_id)["pending_slot"]
+    monkeypatch.setattr(scheduler_tick, "_acquire_tick_lock_or_degrade", _lock_fd(tmp_path))
+    assert scheduler_tick._tick_admitted(False, None, None, True, can_dispatch=lambda: False) == 0
+    assert jobs.get_job(job_id)["pending_slot"] == before
+
+
+def test_crash_between_scan_and_claim_keeps_replay(clock):
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    marker = jobs.get_job(job["id"])["pending_slot"]["replay"]
+    # The scan moves the slot (as a crash would leave it) and no claim is taken.
+    clock[0] += timedelta(seconds=1)
+    assert any(j["id"] == job["id"] for j in jobs.get_due_jobs())
+    assert jobs.get_job(job["id"]).get("fire_claim") is None
+    replay = jobs.claim_job_for_fire(job["id"], return_job=True)
+    assert replay["fire_claim"]["replay"] == marker
+
+
+def test_stale_ladder_plans_first_attempt_for_next_occurrence(clock):
+    from cron import unreachable_retry
+
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    replay = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "offline",
+        expected_fire_owner=replay["fire_claim"]["by"], model_unreachable=True,
+    )
+    # Drop the replay (a schedule edit / pause clears the marker), leaving an exhausted ladder.
+    stored = jobs.load_jobs()
+    stored[0][unreachable_retry.STATE_KEY]["attempt"] = 3
+    stored[0].pop("pending_slot", None)
+    stored[0].pop("restored_replay", None)
+    stored[0]["next_run_at"] = (
+        datetime.fromisoformat(stored[0]["next_run_at"]) + timedelta(hours=1)
+    ).isoformat()
+    jobs.save_jobs(stored)
+    clock[0] = datetime.fromisoformat(jobs.get_job(job["id"])["next_run_at"])
+    nxt = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "offline",
+        expected_fire_owner=nxt["fire_claim"]["by"], model_unreachable=True,
+    )
+    assert jobs.get_job(job["id"])[unreachable_retry.STATE_KEY]["attempt"] == 1
+
+
+def test_catch_up_disabled_drops_replay_slot(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    (tmp_path / "config.yaml").write_text("cron:\n  catch_up_missed: false\n", encoding="utf-8")
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 1h")
+        stored = jobs.load_jobs()
+        stale = (jobs._hermes_now() - timedelta(hours=4)).isoformat()
+        stored[0]["pending_slot"] = {
+            "scheduled_at": stale,
+            "replay": {"scheduled_at": stale, "started_at": stale},
+        }
+        stored[0]["next_run_at"] = stale
+        jobs.save_jobs(stored)
+        assert job["id"] not in [j["id"] for j in jobs.get_due_jobs()]
+        after = jobs.get_job(job["id"])
+        assert "replay" not in (after.get("pending_slot") or {})
+        assert "restored_replay" not in after
+
+
+def test_multiplex_profiles_keep_replay_in_their_own_store(tmp_path, clock):
+    """Case 4: each profile's replay lives in its own store; a gated-out profile's replay stays
+    pending and does not leak into another profile's claim."""
+    stores = {}
+    for name in ("alpha", "beta"):
+        with jobs.use_cron_store(tmp_path / name):
+            job = jobs.create_job(name, "every 24h")
+            stores[name] = job["id"]
+            clock[0] = datetime.fromisoformat(job["next_run_at"])
+            first = claim_due(job["id"], clock)
+            assert jobs.mark_job_run(
+                job["id"], False, "shutdown",
+                expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
+            )
+    # Only beta is admitted this tick; alpha's gate refuses it, so alpha is never claimed.
+    with jobs.use_cron_store(tmp_path / "beta"):
+        beta = claim_due(stores["beta"], clock)
+        assert beta["fire_claim"]["replay"] is not None
+    with jobs.use_cron_store(tmp_path / "alpha"):
+        stored = jobs.get_job(stores["alpha"])
+        assert stored["pending_slot"]["replay"] is not None
+        assert stored.get("fire_claim") is None
+
+
+def test_heartbeat_does_not_move_replay_started_at(clock):
+    """Case 10: the heartbeat refreshes the claim's ``at`` but never ``claimed_at``; the replay's
+    ``started_at`` stays at the original claim, before any message the run posted."""
+    job = jobs.create_job("report", "every 24h")
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    claimed_at = first["fire_claim"]["claimed_at"]
+    owner = first["fire_claim"]["by"]
+    clock[0] += timedelta(minutes=10)
+    assert jobs.heartbeat_fire_claim(job["id"], expected_owner=owner)
+    assert jobs.get_job(job["id"])["fire_claim"]["at"] != claimed_at
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=owner, shutdown_interrupted=True,
+    )
+    replay = jobs.get_job(job["id"])["pending_slot"]["replay"]
+    assert replay["started_at"] == claimed_at

From 443c3bcb48466acd0412b281040889a30e266bf3 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 20:07:46 -0500
Subject: [PATCH 2/6] test(cron): gate case-4 replay through the multiplex
 ticker; document interrupted replay (RIG-8084)

---
 tests/cron/test_shutdown_replay.py            | 69 +++++++++++++++----
 .../docs/developer-guide/cron-internals.md    | 15 ++++
 2 files changed, 69 insertions(+), 15 deletions(-)

diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
index 7d2fbc08c45e..a2a054635ea0 100644
--- a/tests/cron/test_shutdown_replay.py
+++ b/tests/cron/test_shutdown_replay.py
@@ -318,28 +318,67 @@ def test_catch_up_disabled_drops_replay_slot(tmp_path, monkeypatch):
         assert "restored_replay" not in after
 
 
-def test_multiplex_profiles_keep_replay_in_their_own_store(tmp_path, clock):
-    """Case 4: each profile's replay lives in its own store; a gated-out profile's replay stays
-    pending and does not leak into another profile's claim."""
-    stores = {}
-    for name in ("alpha", "beta"):
-        with jobs.use_cron_store(tmp_path / name):
+def test_multiplex_gate_runs_only_admitted_profile_replay(tmp_path, clock):
+    """Case 4: a gated-out profile's replay never runs, while the admitted profile's replay runs
+    from its own store. Drives the real multiplex gate through the ticker loop."""
+    import threading
+    from unittest.mock import patch
+    from cron.scheduler_provider import InProcessCronScheduler
+    from hermes_constants import get_hermes_home
+
+    homes = {}
+    for name in ("own-gateway", "orphan"):
+        home = tmp_path / name
+        (home / "cron").mkdir(parents=True)
+        with jobs.use_cron_store(home / "cron"):
             job = jobs.create_job(name, "every 24h")
-            stores[name] = job["id"]
             clock[0] = datetime.fromisoformat(job["next_run_at"])
             first = claim_due(job["id"], clock)
             assert jobs.mark_job_run(
                 job["id"], False, "shutdown",
                 expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
             )
-    # Only beta is admitted this tick; alpha's gate refuses it, so alpha is never claimed.
-    with jobs.use_cron_store(tmp_path / "beta"):
-        beta = claim_due(stores["beta"], clock)
-        assert beta["fire_claim"]["replay"] is not None
-    with jobs.use_cron_store(tmp_path / "alpha"):
-        stored = jobs.get_job(stores["alpha"])
-        assert stored["pending_slot"]["replay"] is not None
-        assert stored.get("fire_claim") is None
+        homes[name] = home
+
+    stop = threading.Event()
+    ticked: list[str] = []
+
+    def _tick(*args, **kwargs):
+        ticked.append(str(get_hermes_home()))
+        if len(ticked) >= 3:
+            stop.set()
+        return 0
+
+    provider = InProcessCronScheduler()
+    with patch("cron.scheduler.tick", side_effect=_tick):
+        thread = threading.Thread(
+            target=provider.start,
+            args=(stop,),
+            kwargs={
+                "interval": 0,
+                "profile_homes": list(homes.items()),
+                "profile_gate": lambda name, home: name != "own-gateway",
+            },
+            daemon=True,
+        )
+        thread.start()
+        thread.join(timeout=5)
+        stop.set()
+        thread.join(timeout=5)
+
+    # The gate admits only the orphan profile to the tick loop.
+    assert set(ticked) == {str(homes["orphan"])}
+    # Gated-out profile keeps its replay pending; it is not claimed.
+    with jobs.use_cron_store(homes["own-gateway"] / "cron"):
+        gated = jobs.load_jobs()[0]
+        assert gated["pending_slot"]["replay"] is not None
+        assert gated.get("fire_claim") is None
+    # Admitted profile's replay is claimed and dispatched from its own store.
+    with jobs.use_cron_store(homes["orphan"] / "cron"):
+        admitted = jobs.load_jobs()[0]
+        clock[0] = datetime.fromisoformat(admitted["pending_slot"]["scheduled_at"])
+        claimed = claim_due(admitted["id"], clock)
+        assert claimed["fire_claim"]["replay"] is not None
 
 
 def test_heartbeat_does_not_move_replay_started_at(clock):
diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index 11841532c961..53962f91985c 100644
--- a/website/docs/developer-guide/cron-internals.md
+++ b/website/docs/developer-guide/cron-internals.md
@@ -203,6 +203,21 @@ so two profiles may carry identically named jobs without colliding. A profile
 that runs its own gateway is skipped per tick, so the two processes never race
 its store and its deliveries always leave through its own live adapters.
 
+**Interrupted occurrences and replay.** A run cut off by a gateway shutdown
+before its terminal row is not counted when it is interrupted. The executions
+ledger keeps the interrupted attempt as a `failed` row, and the record stores a
+`pending_slot.replay` marker (`started_at`, `interrupted_at`). `started_at` is the
+claim's first `claimed_at`, so heartbeats do not move it. The next claim of that
+same occurrence is the replay: it appears as one further execution row for the
+same `scheduled_instant`, and its rungs (retries) follow it.
+
+The occurrence is counted when the replay completes, not at the shutdown. If the
+replay is interrupted again, that interruption counts at that shutdown and the
+occurrence is not replayed a second time. A retry rung that already counted its
+occurrence does not count it again. A replay that is past grace under
+`cron.catch_up_missed: false` is dropped with its marker, so it cannot replay
+later.
+
 **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 e5fa4b1cce56b9f8ba51704513fac7c2b0581746 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 20:46:49 -0500
Subject: [PATCH 3/6] fix(cron): address replay review (RIG-8084)

Preserve interrupted occurrences through shutdown and local-time due scans; cover replay accounting and admission.
---
 cron/jobs.py                                  |  77 ++++++---
 cron/occurrences.py                           |  14 +-
 cron/scheduler.py                             |   3 +-
 tests/cron/test_shutdown_replay.py            | 149 +++++++++++++++---
 .../docs/developer-guide/cron-internals.md    |  10 +-
 5 files changed, 196 insertions(+), 57 deletions(-)

diff --git a/cron/jobs.py b/cron/jobs.py
index c2c8330c57d5..4125834b85f2 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2116,6 +2116,7 @@ def apply(jobs, i, job):
             # An explicit schedule/lifecycle rewrite supersedes any occurrence the dispatcher
             # left unclaimed — pause/resume/edit must not resurrect a slot from before the edit.
             updated.pop("pending_slot", None)
+            updated.pop("restored_replay", None)
             drop_claimed_occurrence(updated)
         _fill_missing_next_run(updated)
         _reject_terminal_activation(job, updated, job_id)
@@ -2376,8 +2377,8 @@ def _advance_after_run(job: dict[str, Any], now: str, *, ladder_rung: bool = Fal
     """Bump ``repeat.completed`` and recompute ``next_run_at``; retire the record as a terminal
     completion when the repeat limit is reached or a one-shot has no further run.
 
-    ``ladder_rung``: this run re-ran an occurrence that already counted (an unreachable-model
-    re-run, ``cron.unreachable_retry.is_retry_run``), so ``repeat.completed`` is left as is."""
+    ``ladder_rung`` skips counting an already-counted retry or an interrupted occurrence
+    whose count is deferred until replay."""
     # If no next run, decide whether this is terminal completion (one-shot) or a transient failure
     # (recurring schedule couldn't compute — e.g. 'croniter' missing from the runtime env). Recurring jobs
     # must NEVER be silently disabled: that turns a missing runtime dep into "job completed" and the user's
@@ -2461,7 +2462,7 @@ 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 (Q2 ruling, RIG-8084).
+    replay cut off again is counted now and not replayed again.
     """
     def apply(jobs, _i, job):
         if execution_id is not None and execution_id in job.get("canonical_completions", []):
@@ -2476,29 +2477,34 @@ def apply(jobs, _i, job):
         now = _hermes_now().isoformat()
         claim = job.get("fire_claim")
         owned = claim.get("scheduled_at") if isinstance(claim, dict) else None
-        replay = _claim_replay(job)
         from cron import quota_hold
-        from cron.occurrences import pending_slot_stamp
+        from cron.occurrences import claim_replay, pending_slot_stamp
         from cron.unreachable_retry import STATE_KEY, clear_state, is_retry_run, plan_retry
 
-        rung = is_retry_run({"_scheduled_instant": owned, STATE_KEY: job.get(STATE_KEY)})
+        replay = claim_replay(job)
+
+        rung = ladder_rung or is_retry_run({"_scheduled_instant": owned, STATE_KEY: job.get(STATE_KEY)})
 
         # A cut-off first run stays pending, uncounted: the replay counts it. A rung already
         # counted its occurrence; a replay itself counts once when it finishes or is cut off.
         keep_pending = shutdown_interrupted and owned is not None and replay is None
         if owned is not None and not rung:
-            clear_state(job)  # a parked rung for another occurrence is stale (design case 14)
+            clear_state(job)  # A parked rung for another occurrence is stale.
+        pending = job.get("pending_slot") if owned is None else None
         _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
             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)
         if keep_pending and job.get("schedule", {}).get("kind") in {"cron", "interval"} \
                 and not is_terminal_job(job):
-            # Started-at is the claim's first-claim time; heartbeats never move it.
+            # Legacy claims lack claimed_at and fall back to the last heartbeat time.
             started = claim.get("claimed_at") or claim.get("at")
+            slot = datetime.fromisoformat(owned).astimezone(_hermes_now().tzinfo).isoformat()
             job["pending_slot"] = pending_slot_stamp(
-                owned, _hermes_now(), replay={"started_at": started, "interrupted_at": now})
+                slot, _hermes_now(), replay={"started_at": started, "interrupted_at": now})
 
         if not success and model_unreachable and not is_terminal_job(job):
             plan_retry(job, replay=replay)
@@ -2743,13 +2749,17 @@ def _machine_id() -> str:
     return f"{host}:{os.getpid()}"
 
 
-def _claim_replay(job: dict[str, Any]) -> Optional[dict[str, Any]]:
-    from cron.occurrences import claim_replay
-    return claim_replay(job)
+
+def _replay_matches(record: Any, instant: Optional[str]) -> bool:
+    from cron.occurrences import scheduled_instant
+
+    target = scheduled_instant(instant)
+    return (target is not None and isinstance(record, dict) and "replay" in record
+            and scheduled_instant(record.get("scheduled_at")) == target)
 
 
 def _carry_replay(job: dict[str, Any], owned: Optional[str]) -> None:
-    """Copy the replay marker of the occurrence being claimed onto its fire claim (RIG-8084).
+    """Copy the replay marker of the occurrence being claimed onto its fire claim.
 
     The marker comes from the pending slot when it holds this occurrence, else from a parked
     retry rung when that rung is this occurrence. Other occurrences carry none."""
@@ -2762,11 +2772,12 @@ def _carry_replay(job: dict[str, Any], owned: Optional[str]) -> None:
     pending = job.get("pending_slot")
     restored = job.get("restored_replay")
     replay = None
-    if isinstance(pending, dict) and scheduled_instant(pending.get("scheduled_at")) == owned:
+    if _replay_matches(pending, owned):
         replay = slot_replay(job)
-    if replay is None and isinstance(restored, dict) \
-            and scheduled_instant(restored.get("scheduled_at")) == owned:
-        replay = restored.get("replay") if isinstance(restored.get("replay"), dict) else None
+    if replay is None and _replay_matches(restored, owned):
+        replay = restored.get("replay")
+        if not isinstance(replay, dict):
+            replay = {}
     if replay is None and scheduled_instant((job.get(STATE_KEY) or {}).get("at")) == owned:
         replay = ladder_replay(job)
     job.pop("restored_replay", None)
@@ -2857,7 +2868,8 @@ def apply(jobs, _i, job):
         if execution_id:
             bind_row(owned)
         # Claimed: the occurrence is now owned by a run (its ledger row + fire claim carry it).
-        job.pop("pending_slot", None)
+        if owned is not None:
+            job.pop("pending_slot", None)
         if job.get("schedule", {}).get("kind") in {"cron", "interval"}:
             nxt = compute_next_run(job["schedule"], now.isoformat())
             if nxt:
@@ -3078,6 +3090,14 @@ def recompute_next(self) -> Optional[str]:
         return compute_next_run(self.schedule, self.scan.now.isoformat())
 
 
+def _drop_restored_replay(job: dict[str, Any], scan: _DueScan) -> None:
+    job.pop("restored_replay", None)
+    stored = scan.find(job["id"])
+    if stored is not None and "restored_replay" in stored:
+        stored.pop("restored_replay")
+        scan.needs_save = True
+
+
 def _repair_timezone_shifted_cron(d: _DueJob) -> bool:
     """Repair a cron job whose stored offset no longer matches now's (TZ migration).
 
@@ -3100,6 +3120,7 @@ def _repair_timezone_shifted_cron(d: _DueJob) -> bool:
         "Recomputing cron run to preserve local wall-clock intent: %s",
         d.label, d.raw_next_run_dt.utcoffset(), now.utcoffset(), new_next)
     d.scan.persist(d.job["id"], next_run_at=new_next)
+    _drop_restored_replay(d.job, d.scan)
     return True
 
 
@@ -3167,6 +3188,7 @@ def _reanchor_stale_cron(d: _DueJob) -> bool:
             d.label, d.next_run, d.schedule.get("expr"), new_next)
         if new_next:
             d.scan.persist(d.job["id"], next_run_at=new_next)
+            _drop_restored_replay(d.job, d.scan)
         return True
     if stale_class == STALE_CRON_TIMEZONE_MIGRATION:
         logger.warning(
@@ -3200,7 +3222,7 @@ def _fast_forward_missed_recurring(d: _DueJob, grace: int) -> bool:
             "Job '%s' missed its scheduled time (%s, grace=%ds). "
             "Skipping missed occurrence because cron.catch_up_missed is false; next run: %s",
             d.label, d.next_run, grace, new_next)
-        # A dropped slot takes its replay marker with it (design case 12).
+        # A dropped slot takes its replay marker with it.
         d.job.pop("restored_replay", None)
         rj = d.scan.find(d.job["id"])
         if rj is not None:
@@ -3286,13 +3308,14 @@ def _restore_unclaimed_slot(job: dict[str, Any], scan: _DueScan) -> Optional[str
     """Put an occurrence the dispatcher advanced past but never claimed back on the schedule
     (#107485); returns the restored ``next_run_at`` or None. Restored ONCE: the stamp is dropped
     here, so the slot then meets the ordinary late / fast-forward / ``cron.catch_up_missed``
-    policy like any other overdue instant — never a replay of every missed slot."""
+    policy like any other overdue instant. ``restored_replay`` holds the slot's
+    ``scheduled_at`` and ``replay`` marker until it is claimed or discarded."""
     from cron.occurrences import slot_replay, unclaimed_pending_slot
 
     slot = unclaimed_pending_slot(job, scan.now)
     if slot is None:
         return None
-    replay = slot_replay(job)
+    replay = slot_replay(job) if _replay_matches(job.get("pending_slot"), slot) else None
     logger.warning(
         "Job '%s' (%s): occurrence %s was taken off the schedule but never claimed "
         "(scheduler stopped before dispatch); restoring it as the due instant (was %s).",
@@ -3352,7 +3375,8 @@ def apply(jobs, _i, job):
             now = _hermes_now()
             # Store in the scheduler's zone: a UTC string would trip the timezone-shift repair.
             slot = datetime.fromisoformat(instant).astimezone(now.tzinfo).isoformat()
-            job["pending_slot"] = pending_slot_stamp(slot, now, replay=_claim_replay(job))
+            from cron.occurrences import claim_replay
+            job["pending_slot"] = pending_slot_stamp(slot, now, replay=claim_replay(job))
             drop_claimed_occurrence(job)
         else:
             return False
@@ -3393,6 +3417,7 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float)
     from cron.occurrences import completed_occurrence, scheduled_instant
 
     if not manual_run and completed_occurrence(job, next_run):
+        _drop_restored_replay(job, scan)
         new_next = d.recompute_next() if recurring else None
         if new_next:
             scan.persist(job["id"], next_run_at=new_next)
@@ -3441,10 +3466,12 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float)
         # stamp survives a process death in that window so the slot is restored, not lost.
         from cron.occurrences import pending_slot_stamp, slot_replay
 
-        replay = slot_replay(job)
+        replay = slot_replay(job) if _replay_matches(job.get("pending_slot"), next_run) else None
         restored = job.get("restored_replay")
-        if replay is None and isinstance(restored, dict) and restored.get("scheduled_at") == next_run:
-            replay = restored.get("replay")  # a restored replay keeps its marker through this stamp
+        if replay is None and _replay_matches(restored, next_run):
+            replay = restored.get("replay")
+            if not isinstance(replay, dict):
+                replay = {}
         scan.persist(
             job["id"], last_dispatch=dispatch_stamp,
             pending_slot=pending_slot_stamp(next_run, now, replay=replay))
diff --git a/cron/occurrences.py b/cron/occurrences.py
index a310bb1ccfb3..3911e44bc05d 100644
--- a/cron/occurrences.py
+++ b/cron/occurrences.py
@@ -71,7 +71,7 @@ def completed_occurrence(job, instant):
 def pending_slot_stamp(next_run, now, replay=None):
     """Store value for a recurring occurrence about to be handed to the dispatcher.
 
-    ``replay`` marks an interrupted run kept as pending work (RIG-8084); it is stored only when set."""
+    ``replay`` marks an interrupted run kept as pending work; it is stored only when set."""
     from cron.jobs import _machine_id
 
     stamp = {"scheduled_at": next_run, "at": now.isoformat(), "by": _machine_id()}
@@ -83,15 +83,19 @@ def pending_slot_stamp(next_run, now, replay=None):
 def slot_replay(job):
     """The replay marker of a pending slot, or None."""
     pending = job.get("pending_slot")
-    replay = pending.get("replay") if isinstance(pending, dict) else None
-    return replay if isinstance(replay, dict) else None
+    if not isinstance(pending, dict) or "replay" not in pending:
+        return None
+    replay = pending["replay"]
+    return replay if isinstance(replay, dict) else {}
 
 
 def claim_replay(job):
     """The replay marker of a live fire claim, or None."""
     claim = job.get("fire_claim")
-    replay = claim.get("replay") if isinstance(claim, dict) else None
-    return replay if isinstance(replay, dict) else None
+    if not isinstance(claim, dict) or "replay" not in claim:
+        return None
+    replay = claim["replay"]
+    return replay if isinstance(replay, dict) else {}
 
 
 def unclaimed_pending_slot(job, now):
diff --git a/cron/scheduler.py b/cron/scheduler.py
index 2fdbd1458f88..bc7d6e34a7f9 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -1161,7 +1161,8 @@ def mark_running_jobs_interrupted(
         try:
             with use_cron_store(profile_home):
                 if mark_job_run(
-                    job_id, False, reason, expected_fire_owner=fire_owner):
+                    job_id, False, reason, expected_fire_owner=fire_owner,
+                    shutdown_interrupted=True):
                     marked.append(job_id)
         except Exception as e:
             logger.warning("Failed to mark job %s interrupted: %s", job_id, e)
diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
index a2a054635ea0..b898d7691130 100644
--- a/tests/cron/test_shutdown_replay.py
+++ b/tests/cron/test_shutdown_replay.py
@@ -22,6 +22,98 @@ def claim_due(job_id, clock):
     return jobs.claim_job_for_fire(job_id, return_job=True)
 
 
+def test_shutdown_sweep_preserves_interrupted_occurrence(clock, monkeypatch, tmp_path):
+    from cron import scheduler
+
+    with jobs.use_cron_store(tmp_path / "cron"):
+        job = jobs.create_job("report", "every 24h", repeat=1)
+        clock[0] = datetime.fromisoformat(job["next_run_at"])
+        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, "_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"])
+        assert stored["pending_slot"]["replay"]
+        assert stored["repeat"]["completed"] == 0
+
+
+@pytest.mark.parametrize("retry", [False, True])
+def test_non_utc_replay_survives_due_scan(clock, retry):
+    from zoneinfo import ZoneInfo
+
+    clock[0] = datetime(2026, 1, 1, 8, 0, tzinfo=ZoneInfo("America/New_York"))
+    job = jobs.create_job("report", "0 9 * * *", repeat=3)
+    if retry:
+        assert jobs.mark_job_run(job["id"], False, "offline", model_unreachable=True)
+        clock[0] = datetime.fromisoformat(jobs.get_job(job["id"])["next_run_at"])
+    else:
+        clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(job["id"], False, "shutdown", expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True)
+    pending = jobs.get_job(job["id"])["pending_slot"]
+    assert datetime.fromisoformat(pending["scheduled_at"]).utcoffset() == clock[0].utcoffset()
+    replay = claim_due(job["id"], clock)
+    assert replay["fire_claim"]["replay"] == pending["replay"]
+
+
+def test_midrun_schedule_edit_keeps_retry_count(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    assert jobs.mark_job_run(job["id"], False, "offline", model_unreachable=True)
+    clock[0] = datetime.fromisoformat(jobs.get_job(job["id"])["next_run_at"])
+    rung = claim_due(job["id"], clock)
+    jobs.update_job(job["id"], {"next_run_at": (clock[0] + timedelta(days=1)).isoformat()})
+    assert jobs.mark_job_run(job["id"], True, expected_fire_owner=rung["fire_claim"]["by"], ladder_rung=True)
+    assert jobs.get_job(job["id"])["repeat"]["completed"] == 1
+
+
+@pytest.mark.parametrize("marker", ["invalid", [], None])
+def test_malformed_replay_counts_without_another_replay(clock, marker):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    records = jobs.load_jobs()
+    records[0]["fire_claim"]["replay"] = marker
+    jobs.save_jobs(records)
+    assert jobs.mark_job_run(job["id"], False, "shutdown", expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True)
+    result = jobs.get_job(job["id"])
+    assert result["repeat"]["completed"] == 1
+    assert "pending_slot" not in result
+
+
+def test_manual_run_preserves_pending_replay(clock):
+    job_id = _interrupted_replay_slot(clock)
+    pending = jobs.get_job(job_id)["pending_slot"]
+    manual = jobs.claim_job_for_fire(job_id, manual=True, return_job=True)
+    assert jobs.mark_job_run(job_id, True, expected_fire_owner=manual["fire_claim"]["by"])
+    assert jobs.get_job(job_id)["pending_slot"] == pending
+    assert claim_due(job_id, clock)["fire_claim"]["replay"] == pending["replay"]
+
+
+def test_normal_completion_counts_once_without_replay(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(job["id"], True, expected_fire_owner=first["fire_claim"]["by"])
+    result = jobs.get_job(job["id"])
+    assert result["repeat"]["completed"] == 1
+    assert "pending_slot" not in result
+
+
+def test_ownership_loss_records_interruption(clock, monkeypatch):
+    from cron import scheduler
+
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    monkeypatch.setattr(scheduler, "finish_execution", lambda *args, **kwargs: None)
+    scheduler._record_fire_ownership_lost(first, first["fire_claim"]["by"], "execution")
+    stored = jobs.get_job(job["id"])
+    assert stored["repeat"]["completed"] == 0
+    assert stored["pending_slot"]["replay"]
+
+
 def test_final_repeat_is_counted_when_replay_finishes(clock):
     job = jobs.create_job("report", "every 24h", repeat=1)
     clock[0] = datetime.fromisoformat(job["next_run_at"])
@@ -189,7 +281,7 @@ def test_second_shutdown_does_not_stamp_again(clock):
     )
     stored = jobs.get_job(job["id"])
     assert stored["repeat"]["completed"] == 1
-    assert "pending_slot" not in stored or "replay" not in stored["pending_slot"]
+    assert "replay" not in stored.get("pending_slot", {})
 
 
 def test_one_shot_interruption_stamps_no_replay(clock):
@@ -203,10 +295,25 @@ def test_one_shot_interruption_stamps_no_replay(clock):
     assert "replay" not in (jobs.get_job(job["id"]).get("pending_slot") or {})
 
 
-def test_no_claim_stamps_nothing(clock):
-    job = jobs.create_job("report", "every 24h")
-    assert jobs.mark_job_run(job["id"], False, "shutdown", shutdown_interrupted=True)
-    assert "replay" not in (jobs.get_job(job["id"]).get("pending_slot") or {})
+def test_restore_unstarted_one_shot_releases_claim(clock):
+    job = jobs.create_job("report", "in 30 minutes", repeat=1)
+    clock[0] = datetime.fromisoformat(job["next_run_at"])
+    first = claim_due(job["id"], clock)
+    claim = first["fire_claim"]
+    assert jobs.restore_unstarted_occurrence(job["id"], claim["scheduled_at"], claim["by"])
+    stored = jobs.get_job(job["id"])
+    assert stored["fire_claim"] is None
+    assert stored["run_claim"]["outage"] is True
+    assert "pending_slot" not in stored
+    assert any(item["id"] == job["id"] for item in jobs.get_due_jobs())
+
+
+def test_schedule_edit_clears_restored_replay(clock):
+    job_id = _interrupted_replay_slot(clock)
+    assert any(item["id"] == job_id for item in jobs.get_due_jobs())
+    assert "restored_replay" in jobs.get_job(job_id)
+    jobs.update_job(job_id, {"next_run_at": (clock[0] + timedelta(days=1)).isoformat()})
+    assert "restored_replay" not in jobs.get_job(job_id)
 
 
 def _interrupted_replay_slot(clock):
@@ -220,36 +327,37 @@ def _interrupted_replay_slot(clock):
     return job["id"]
 
 
-def _lock_fd(tmp_path):
-    def acquire(_sched):
-        return open(tmp_path / "tick.lock", "a+")
+@pytest.fixture
+def tick_lock(monkeypatch, tmp_path):
+    from cron import scheduler_tick
 
-    return acquire
+    monkeypatch.setattr(
+        scheduler_tick, "_acquire_tick_lock_or_degrade",
+        lambda _sched: open(tmp_path / "tick.lock", "a+"),
+    )
 
 
-def test_estop_tick_leaves_replay_pending(clock, monkeypatch, tmp_path):
+def test_estop_tick_leaves_replay_pending(clock, monkeypatch, tick_lock):
     from agent import estop
     from cron import scheduler_tick
 
     job_id = _interrupted_replay_slot(clock)
     before = jobs.get_job(job_id)["pending_slot"]
     monkeypatch.setattr(estop, "check_paused", lambda *_a, **_k: True)
-    monkeypatch.setattr(scheduler_tick, "_acquire_tick_lock_or_degrade", _lock_fd(tmp_path))
     assert scheduler_tick._tick_admitted(False, None, None, True) == 0
     assert jobs.get_job(job_id)["pending_slot"] == before
 
 
-def test_draining_tick_leaves_replay_pending(clock, monkeypatch, tmp_path):
+def test_draining_tick_leaves_replay_pending(clock, tick_lock):
     from cron import scheduler_tick
 
     job_id = _interrupted_replay_slot(clock)
     before = jobs.get_job(job_id)["pending_slot"]
-    monkeypatch.setattr(scheduler_tick, "_acquire_tick_lock_or_degrade", _lock_fd(tmp_path))
     assert scheduler_tick._tick_admitted(False, None, None, True, can_dispatch=lambda: False) == 0
     assert jobs.get_job(job_id)["pending_slot"] == before
 
 
-def test_crash_between_scan_and_claim_keeps_replay(clock):
+def test_scan_before_claim_keeps_replay(clock):
     job = jobs.create_job("report", "every 24h")
     clock[0] = datetime.fromisoformat(job["next_run_at"])
     first = claim_due(job["id"], clock)
@@ -258,7 +366,7 @@ def test_crash_between_scan_and_claim_keeps_replay(clock):
         expected_fire_owner=first["fire_claim"]["by"], shutdown_interrupted=True,
     )
     marker = jobs.get_job(job["id"])["pending_slot"]["replay"]
-    # The scan moves the slot (as a crash would leave it) and no claim is taken.
+    # A scan can stamp the slot before its fire claim is taken.
     clock[0] += timedelta(seconds=1)
     assert any(j["id"] == job["id"] for j in jobs.get_due_jobs())
     assert jobs.get_job(job["id"]).get("fire_claim") is None
@@ -290,6 +398,7 @@ def test_stale_ladder_plans_first_attempt_for_next_occurrence(clock):
         datetime.fromisoformat(stored[0]["next_run_at"]) + timedelta(hours=1)
     ).isoformat()
     jobs.save_jobs(stored)
+    assert stored[0][unreachable_retry.STATE_KEY]["attempt"] == 3
     clock[0] = datetime.fromisoformat(jobs.get_job(job["id"])["next_run_at"])
     nxt = claim_due(job["id"], clock)
     assert jobs.mark_job_run(
@@ -318,9 +427,8 @@ def test_catch_up_disabled_drops_replay_slot(tmp_path, monkeypatch):
         assert "restored_replay" not in after
 
 
-def test_multiplex_gate_runs_only_admitted_profile_replay(tmp_path, clock):
-    """Case 4: a gated-out profile's replay never runs, while the admitted profile's replay runs
-    from its own store. Drives the real multiplex gate through the ticker loop."""
+def test_multiplex_gate_ticks_only_admitted_profile_with_replay(tmp_path, clock):
+    """The profile gate excludes a profile with a pending replay from ticker calls."""
     import threading
     from unittest.mock import patch
     from cron.scheduler_provider import InProcessCronScheduler
@@ -373,7 +481,7 @@ def _tick(*args, **kwargs):
         gated = jobs.load_jobs()[0]
         assert gated["pending_slot"]["replay"] is not None
         assert gated.get("fire_claim") is None
-    # Admitted profile's replay is claimed and dispatched from its own store.
+    # The admitted profile's replay remains claimable from its own store.
     with jobs.use_cron_store(homes["orphan"] / "cron"):
         admitted = jobs.load_jobs()[0]
         clock[0] = datetime.fromisoformat(admitted["pending_slot"]["scheduled_at"])
@@ -382,8 +490,7 @@ def _tick(*args, **kwargs):
 
 
 def test_heartbeat_does_not_move_replay_started_at(clock):
-    """Case 10: the heartbeat refreshes the claim's ``at`` but never ``claimed_at``; the replay's
-    ``started_at`` stays at the original claim, before any message the run posted."""
+    """Heartbeats do not move a modern claim's original replay start time."""
     job = jobs.create_job("report", "every 24h")
     clock[0] = datetime.fromisoformat(job["next_run_at"])
     first = claim_due(job["id"], clock)
diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index 53962f91985c..bd3caeaf7576 100644
--- a/website/docs/developer-guide/cron-internals.md
+++ b/website/docs/developer-guide/cron-internals.md
@@ -208,15 +208,15 @@ before its terminal row is not counted when it is interrupted. The executions
 ledger keeps the interrupted attempt as a `failed` row, and the record stores a
 `pending_slot.replay` marker (`started_at`, `interrupted_at`). `started_at` is the
 claim's first `claimed_at`, so heartbeats do not move it. The next claim of that
-same occurrence is the replay: it appears as one further execution row for the
-same `scheduled_instant`, and its rungs (retries) follow it.
+same occurrence is the replay. It appears as one further execution row for the
+same `scheduled_instant`. An unreachable-model retry retains the replay marker.
 
 The occurrence is counted when the replay completes, not at the shutdown. If the
 replay is interrupted again, that interruption counts at that shutdown and the
 occurrence is not replayed a second time. A retry rung that already counted its
-occurrence does not count it again. A replay that is past grace under
-`cron.catch_up_missed: false` is dropped with its marker, so it cannot replay
-later.
+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.
 
 **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

From 6216c79ba1b93d2c6edb023e63d8d4f61405040e Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 20:48:53 -0500
Subject: [PATCH 4/6] fix(cron): preserve ordinary claim behavior (RIG-8084)

Only preserve pending replay for explicit manual claims.
---
 cron/jobs.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/cron/jobs.py b/cron/jobs.py
index 4125834b85f2..c43a0c9915e2 100644
--- a/cron/jobs.py
+++ b/cron/jobs.py
@@ -2491,6 +2491,8 @@ def apply(jobs, _i, job):
         if owned is not None and not rung:
             clear_state(job)  # A parked rung for another occurrence is stale.
         pending = job.get("pending_slot") if owned is None else None
+        if not isinstance(pending, dict) or "replay" not in pending:
+            pending = None
         _record_run_outcome(job, success, error, delivery_error, status, now)
         if pending is not None:
             job["pending_slot"] = pending
@@ -2868,7 +2870,7 @@ def apply(jobs, _i, job):
         if execution_id:
             bind_row(owned)
         # Claimed: the occurrence is now owned by a run (its ledger row + fire claim carry it).
-        if owned is not None:
+        if owned is not None or not manual_fire:
             job.pop("pending_slot", None)
         if job.get("schedule", {}).get("kind") in {"cron", "interval"}:
             nxt = compute_next_run(job["schedule"], now.isoformat())

From 414232bc3583924f8078dff12b72ce0023e7fb84 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 20:49:38 -0500
Subject: [PATCH 5/6] test(cron): assert shutdown sweep interruption semantics
 (RIG-8084)

Require each registered owner to receive interruption bookkeeping.
---
 tests/cron/test_shutdown_interrupt.py | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/tests/cron/test_shutdown_interrupt.py b/tests/cron/test_shutdown_interrupt.py
index 87ca1959bbbd..77c1013dbf0b 100644
--- a/tests/cron/test_shutdown_interrupt.py
+++ b/tests/cron/test_shutdown_interrupt.py
@@ -187,15 +187,15 @@ def test_shutdown_sees_all_concurrent_direct_fire_owners(self, monkeypatch):
 
         entered = threading.Barrier(3)
         release = threading.Event()
-        marked_owners: list[str] = []
+        marked_owners: list[tuple[str, bool]] = []
 
         def hold_run(_job, _run):
             entered.wait(timeout=2)
             release.wait(timeout=2)
             return True
 
-        def mark(_job_id, _success, _reason, *, expected_fire_owner):
-            marked_owners.append(expected_fire_owner)
+        def mark(_job_id, _success, _reason, *, expected_fire_owner, shutdown_interrupted):
+            marked_owners.append((expected_fire_owner, shutdown_interrupted))
             return True
 
         monkeypatch.setattr(sched, "_run_with_fire_claim_heartbeat", hold_run)
@@ -212,7 +212,7 @@ def mark(_job_id, _success, _reason, *, expected_fire_owner):
 
         assert sched.get_running_job_ids() == frozenset({"same-job"})
         assert sched.mark_running_jobs_interrupted("shutdown") == ["same-job", "same-job"]
-        assert set(marked_owners) == {"old-owner", "replacement-owner"}
+        assert set(marked_owners) == {("old-owner", True), ("replacement-owner", True)}
 
         release.set()
         for thread in threads:
@@ -232,12 +232,13 @@ def test_shutdown_marks_each_owner_in_its_profile_store(self, monkeypatch, tmp_p
             object(): ("owner-b", profile_b),
         }
 
-        def mark(job_id, success, reason, *, expected_fire_owner):
+        def mark(job_id, success, reason, *, expected_fire_owner, shutdown_interrupted):
             observed.append(
                 (
                     job_id,
                     success,
                     expected_fire_owner,
+                    shutdown_interrupted,
                     cron_jobs._current_cron_store().jobs_file,
                 )
             )
@@ -247,8 +248,8 @@ def mark(job_id, success, reason, *, expected_fire_owner):
 
         assert sched.mark_running_jobs_interrupted("shutdown") == ["same-job", "same-job"]
         assert set(observed) == {
-            ("same-job", False, "owner-a", profile_a / "cron" / "jobs.json"),
-            ("same-job", False, "owner-b", profile_b / "cron" / "jobs.json"),
+            ("same-job", False, "owner-a", True, profile_a / "cron" / "jobs.json"),
+            ("same-job", False, "owner-b", True, profile_b / "cron" / "jobs.json"),
         }
 
 

From 9b77dc2aa3a1f6c03de29ba2736845a54cafc976 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sun, 8 Nov 2026 21:00:25 -0500
Subject: [PATCH 6/6] fix(cron): fail closed on a malformed rung replay marker
 (RIG-8084)

---
 cron/unreachable_retry.py          |  7 +++++--
 tests/cron/test_shutdown_replay.py | 17 +++++++++++++++++
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/cron/unreachable_retry.py b/cron/unreachable_retry.py
index c894ad57534e..da1e73e773d7 100644
--- a/cron/unreachable_retry.py
+++ b/cron/unreachable_retry.py
@@ -134,8 +134,11 @@ def is_retry_run(job: dict[str, Any]) -> bool:
 
 def ladder_replay(job: dict[str, Any]) -> dict[str, Any] | None:
     """The replay marker a parked rung carries, or None."""
-    replay = (job.get(STATE_KEY) or {}).get("replay")
-    return replay if isinstance(replay, dict) else None
+    state = job.get(STATE_KEY)
+    if not isinstance(state, dict) or "replay" not in state:
+        return None
+    replay = state["replay"]
+    return replay if isinstance(replay, dict) else {}
 
 
 def plan_retry(job: dict[str, Any], replay: dict[str, Any] | None = None) -> bool:
diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
index b898d7691130..4236cb1f15dd 100644
--- a/tests/cron/test_shutdown_replay.py
+++ b/tests/cron/test_shutdown_replay.py
@@ -186,6 +186,23 @@ def test_interrupted_retry_rung_does_not_count_again(clock):
     assert jobs.get_job(job["id"])["repeat"]["completed"] == 1
 
 
+def test_malformed_rung_replay_does_not_replay_again(clock):
+    job = jobs.create_job("report", "every 24h", repeat=3)
+    assert jobs.mark_job_run(job["id"], False, "offline", model_unreachable=True)
+    records = jobs.load_jobs()
+    records[0][unreachable_retry.STATE_KEY]["replay"] = "invalid"
+    jobs.save_jobs(records)
+    clock[0] = datetime.fromisoformat(records[0][unreachable_retry.STATE_KEY]["at"])
+    rung = claim_due(job["id"], clock)
+    assert jobs.mark_job_run(
+        job["id"], False, "shutdown",
+        expected_fire_owner=rung["fire_claim"]["by"], shutdown_interrupted=True,
+    )
+    stored = jobs.get_job(job["id"])
+    assert stored["repeat"]["completed"] == 1
+    assert "pending_slot" not in stored
+
+
 def test_unreachable_replay_carries_marker_to_retry(clock):
     job = jobs.create_job("report", "every 24h")
     clock[0] = datetime.fromisoformat(job["next_run_at"])