diff --git a/cron/jobs.py b/cron/jobs.py
index 86c83d966c00..c43a0c9915e2 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
@@ -2436,6 +2437,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 +2459,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.
     """
     def apply(jobs, _i, job):
         if execution_id is not None and execution_id in job.get("canonical_completions", []):
@@ -2469,18 +2475,44 @@ 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
+        from cron import quota_hold
+        from cron.occurrences import claim_replay, pending_slot_stamp
+        from cron.unreachable_retry import STATE_KEY, clear_state, is_retry_run, plan_retry
+
+        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.
+        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
         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):
+            # 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(
+                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)
-        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 +2751,42 @@ def _machine_id() -> str:
     return f"{host}:{os.getpid()}"
 
 
+
+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.
+
+    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 _replay_matches(pending, owned):
+        replay = slot_replay(job)
+    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)
+    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,15 +2861,17 @@ 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).
-        job.pop("pending_slot", 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())
             if nxt:
@@ -3022,6 +3092,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).
 
@@ -3044,6 +3122,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
 
 
@@ -3111,6 +3190,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(
@@ -3144,6 +3224,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.
+        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(
@@ -3224,12 +3310,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."""
-    from cron.occurrences import unclaimed_pending_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) 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).",
@@ -3239,6 +3327,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 +3377,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)
+            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
@@ -3323,6 +3419,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)
@@ -3369,11 +3466,17 @@ 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) if _replay_matches(job.get("pending_slot"), next_run) else None
+        restored = job.get("restored_replay")
+        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))
+            pending_slot=pending_slot_stamp(next_run, now, replay=replay))
     return True
 
 
diff --git a/cron/occurrences.py b/cron/occurrences.py
index 23c7eed0f682..3911e44bc05d 100644
--- a/cron/occurrences.py
+++ b/cron/occurrences.py
@@ -68,11 +68,34 @@ 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; 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")
+    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")
+    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 421060e1445d..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)
@@ -2939,6 +2940,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..da1e73e773d7 100644
--- a/cron/unreachable_retry.py
+++ b/cron/unreachable_retry.py
@@ -132,7 +132,16 @@ 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."""
+    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:
     """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 +164,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_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"),
         }
 
 
diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py
new file mode 100644
index 000000000000..4236cb1f15dd
--- /dev/null
+++ b/tests/cron/test_shutdown_replay.py
@@ -0,0 +1,524 @@
+"""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_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"])
+    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_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"])
+    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 "replay" not in stored.get("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_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):
+    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"]
+
+
+@pytest.fixture
+def tick_lock(monkeypatch, tmp_path):
+    from cron import scheduler_tick
+
+    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, 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)
+    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, tick_lock):
+    from cron import scheduler_tick
+
+    job_id = _interrupted_replay_slot(clock)
+    before = jobs.get_job(job_id)["pending_slot"]
+    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_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)
+    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"]
+    # 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
+    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)
+    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(
+        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_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
+    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")
+            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,
+            )
+        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
+    # 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"])
+        claimed = claim_due(admitted["id"], clock)
+        assert claimed["fire_claim"]["replay"] is not None
+
+
+def test_heartbeat_does_not_move_replay_started_at(clock):
+    """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)
+    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
diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index 11841532c961..bd3caeaf7576 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`. 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. 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
 sample that reads the claim as someone else's is re-sampled once before it