fix(cron): skip a shutdown-killed recurring slot instead of pausing (RIG-8801)

cron/executions.py

@@ -712,6 +712,16 @@ def give_up_execution(execution_id: str, *, reason: str) -> bool:
     return True
 
 
+def mark_unknown_skipped(execution_id: str, detail: str) -> bool:
+    """Record that an unknown shutdown execution's scheduled slot was skipped."""
+    with _transaction() as conn:
+        cur = conn.execute(
+            "UPDATE executions SET error=? WHERE id=? AND status='unknown'",
+            (str(detail), str(execution_id)),
+        )
+        return cur.rowcount == 1
+
+
 def list_executions(
     *, job_id: Optional[str] = None, limit: int = 50, before_claimed_at: Optional[str] = None,
 ) -> list[dict[str, Any]]:

cron/jobs.py

@@ -2462,10 +2462,11 @@ def mark_job_run(
 
     ``shutdown_interrupted``: the gateway stopped this run. An occurrence not yet counted stays
     pending as a replay (``pending_slot.replay``) and is counted when the replay finishes. A
-    replay cut off again is counted now and not replayed again.
+    recurring unknown execution is stamped by ID so receipt recovery can skip that occurrence.
     """
     def apply(jobs, _i, job):
-        if execution_id is not None and execution_id in job.get("canonical_completions", []):
+        if (execution_id is not None
+                and execution_id in (job.get("canonical_completions") or [])):
             return True
         if expected_fire_owner is not None:
             claim = job.get("fire_claim")
@@ -2496,7 +2497,9 @@ def apply(jobs, _i, job):
         _record_run_outcome(job, success, error, delivery_error, status, now)
         if pending is not None:
             job["pending_slot"] = pending
-        if execution_id is not None:  # retire ids recovery can no longer replay; latest stays last
+        # Defer the first interrupted occurrence's tombstone until replay finishes; otherwise
+        # completed_occurrence would reject the required replay as already completed.
+        if execution_id is not None and not keep_pending:
             from cron.scheduler_authority import replayable_completions as live
             job["canonical_completions"] = [*live(job_id, job.get("canonical_completions")), execution_id]
         _advance_after_run(job, now, ladder_rung=rung or keep_pending)
@@ -2507,6 +2510,11 @@ def apply(jobs, _i, job):
             slot = datetime.fromisoformat(owned).astimezone(_hermes_now().tzinfo).isoformat()
             job["pending_slot"] = pending_slot_stamp(
                 slot, _hermes_now(), replay={"started_at": started, "interrupted_at": now})
+            if execution_id is not None:
+                job["shutdown_interrupted_execution"] = execution_id
+        elif (execution_id is not None
+              and (execution_id == job.get("shutdown_interrupted_execution") or replay is not None)):
+            job.pop("shutdown_interrupted_execution", None)
 
         if not success and model_unreachable and not is_terminal_job(job):
             plan_retry(job, replay=replay)
@@ -2530,6 +2538,57 @@ def locked():
 
     return _under_fire_fence(job_id, locked)
 
+def skip_unknown_occurrence(
+    job_id: str, execution_id: str, error: str, *, pause_recorded=None,
+) -> bool:
+    """Book a shutdown-killed unknown occurrence as skipped, without replaying its side effects."""
+    def apply(jobs, _i, job):
+        completed = job.get("canonical_completions")
+        if isinstance(completed, list) and execution_id in completed:
+            if isinstance(pause_recorded, dict) and job.get("state") == "paused" \
+                    and (job.get("paused_at"), job.get("paused_reason")) \
+                    == (pause_recorded.get("paused_at"), pause_recorded.get("reason")):
+                job.update(enabled=True, state="scheduled", paused_at=None, paused_reason=None)
+                save_jobs(jobs)
+            return True
+        now = _hermes_now().isoformat()
+        detail = f"Skipped unknown shutdown occurrence: {error}"
+        job["last_run_at"] = now
+        job["last_status"] = "skipped"
+        job["last_error"] = detail
+        job["last_failure"] = {"at": now, "detail": detail}
+        from cron.scheduler_authority import replayable_completions
+        job["canonical_completions"] = [
+            *replayable_completions(job_id, completed), execution_id,
+        ]
+        job.pop("pending_slot", None)
+        job.pop("restored_replay", None)
+        job["fire_claim"] = None
+        if job.get("run_claim") is not None:
+            job["run_claim"] = None
+        if (isinstance(pause_recorded, dict) and job.get("state") == "paused"
+                and (job.get("paused_at"), job.get("paused_reason"))
+                == (pause_recorded.get("paused_at"), pause_recorded.get("reason"))):
+            job.update(enabled=True, state="scheduled", paused_at=None, paused_reason=None)
+        job["next_run_at"] = compute_next_run(job["schedule"], now)
+        job.pop("shutdown_interrupted_execution", None)
+        if job["next_run_at"] is not None:
+            if job.get("state") != "paused":
+                job["state"] = "scheduled"
+        else:
+            job["state"] = "error"
+            logger.error(
+                "Job '%s' (%s) could not compute next_run_at; "
+                "leaving enabled and marking state=error so the job is not silently disabled.",
+                job.get("name", job.get("id", "?")), job.get("schedule", {}).get("kind"))
+        save_jobs(jobs)
+        record_cron_missed(job)
+        logger.warning("CRON_UNKNOWN_OUTCOME_SKIPPED: job %s execution %s: %s",
+                       job_id, execution_id, error)
+        return True
+
+    return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
+
 
 def _write_oneshot_diagnostic(job: dict[str, Any], text: str, what: str) -> bool:
     """Best-effort operator-visible trace in the job's output dir; never breaks the caller."""

cron/scheduler.py

@@ -597,7 +597,7 @@ def _inflight_home_path(home_key: str) -> Path:
 # In-flight state below is keyed by ``_inflight_key``; the public accessors still report bare job
 # ids so host-wide consumers (shutdown drain, idle-exit, metrics) see the union across profiles.
 _running_job_ids: set = set()
-_running_fire_owners: dict[tuple, dict[object, tuple[Optional[str], Path]]] = {}
+_running_fire_owners: dict[tuple, dict[object, tuple[Optional[str], Path, Optional[str]]]] = {}
 # Parent gateway threads synchronously waiting on restart-safe scope workers.
 # Shutdown must not misclassify these as ownerless in-process runs: the tool
 # process sweep cannot reach the worker's transient scope.
@@ -1123,30 +1123,30 @@ def mark_running_jobs_interrupted(
     with _running_lock:
         restart_safe_waiters = set(_restart_safe_waiter_job_ids)
         active_fires = [
-            (token, key, owner, profile_home)
+            (token, key, owner, profile_home, execution_id)
             for key, executions in _running_fire_owners.items()
             if key not in restart_safe_waiters
-            for token, (owner, profile_home) in executions.items()
+            for token, (owner, profile_home, execution_id) in executions.items()
         ]
         if only_owners is not None:
             active_fires = [fire for fire in active_fires if (fire[1][1], fire[2]) in only_owners]
-        registered_keys = {key for _t, key, _o, _p in active_fires}
+        registered_keys = {key for _t, key, _o, _p, _e in active_fires}
         if only_owners is None:
             # The key's home half IS the profile home this claim belongs to — the only record of
             # it for a claim that never reached ``_running_fire_owners``. Read the real Path back
             # from the key, never ``Path(key[0])``: the key is normcased.
             active_fires.extend(
-                (None, key, None, _inflight_home_path(key[0]))
+                (None, key, None, _inflight_home_path(key[0]), None)
                 for key in (
                     _running_job_ids - registered_keys - restart_safe_waiters
                 )
             )
         _interrupted_job_ids.update(
             token if token is not None else key
-            for token, key, _owner, _profile_home in active_fires
+            for token, key, _owner, _profile_home, _execution_id in active_fires
         )
     marked = []
-    for _token, key, fire_owner, profile_home in active_fires:
+    for _token, key, fire_owner, profile_home, execution_id in active_fires:
         job_id = key[1]
         if not fire_owner:
             logger.warning(
@@ -1162,6 +1162,7 @@ def mark_running_jobs_interrupted(
             with use_cron_store(profile_home):
                 if mark_job_run(
                     job_id, False, reason, expected_fire_owner=fire_owner,
+                    **({"execution_id": execution_id} if execution_id is not None else {}),
                     shutdown_interrupted=True):
                     marked.append(job_id)
         except Exception as e:
@@ -2901,7 +2902,7 @@ def run_one_job(
     _fire_key = _inflight_key(job["id"])
     with _running_lock:
         _running_fire_owners.setdefault(_fire_key, {})[execution_token] = (
-            fire_owner or None, profile_home)
+            fire_owner or None, profile_home, execution_id)
     try:
         with self_removal_delivery_scope(job["id"]):
             return _run_with_fire_claim_heartbeat(

cron/scheduler_authority.py

@@ -156,9 +156,9 @@ def reconcile_pending(*, allow_connect=True):
     from hermes_constants import get_hermes_home
     from hermes_cli.gateway_client import connect_gateway
     from cron.executions import execution_owner_live, get_execution
-    from cron.jobs import get_job, pause_job
+    from cron.jobs import get_job, pause_job, skip_unknown_occurrence
     from cron.scheduler import (_RunDelivery, _FireOwnership, _apply_agent_failure_marker,
-                                _save_compose_deliver)
+                                _deliver_crash_failure, _fire_secret_scope, _save_compose_deliver)
     from cron.scheduler_bookkeeping import fail_empty_response, finish_completed_run
     from utils import atomic_json_write
     import logging
@@ -206,11 +206,53 @@ async def observe():
                 state = {'status': 'terminal', 'job': current,
                          'result': [False, f'# Cron Job: {params["job_id"]} (FAILED)\n\n{error}\n', '', error]}
             if state['status'] != 'terminal':
-                if state['status'] == 'unknown' and not record.get('pause_recorded'):
-                    reason = 'Canonical cron unknown; no automatic re-execution'
-                    record['pause_recorded'] = _pause_identity(pause_job(params['job_id'], reason=reason))
-                    atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
-                continue
+                if state['status'] == 'unknown':
+                    row = get_execution(params['request_id'])
+                    error = str((row or {}).get('error') or 'Canonical cron outcome is unknown')
+                    current = get_job(params['job_id'])
+                    if current is None:
+                        journal.unlink(missing_ok=True)
+                        continue
+                    recurring = current.get('schedule', {}).get('kind') in {'cron', 'interval'}
+                    if params['request_id'] in (current.get('canonical_completions') or []):
+                        _lift_pause(params['job_id'], record.get('pause_recorded'))
+                        journal.unlink(missing_ok=True)
+                        continue
+                    if recurring:
+                        from cron.occurrences import completed_occurrence
+                        scheduled_instant = (row or {}).get('scheduled_instant')
+                        if scheduled_instant and completed_occurrence(current, scheduled_instant):
+                            _lift_pause(params['job_id'], record.get('pause_recorded'))
+                            journal.unlink(missing_ok=True)
+                            continue
+                        shutdown_error = current.get('last_error') or error
+                        if current.get('shutdown_interrupted_execution') == params['request_id']:
+                            detail = f'Skipped unknown shutdown occurrence: {shutdown_error}'
+                            from cron.executions import mark_unknown_skipped
+                            # False means the unknown row was already settled or pruned.
+                            mark_unknown_skipped(params['request_id'], detail)
+                            if not skip_unknown_occurrence(
+                                    params['job_id'], params['request_id'], shutdown_error,
+                                    pause_recorded=record.get('pause_recorded')):
+                                raise RuntimeError('Could not book skipped unknown cron occurrence')
+                            journal.unlink(missing_ok=True)
+                            continue
+                    if not record.get('pause_recorded'):
+                        reason = 'Canonical cron unknown; no automatic re-execution'
+                        paused = pause_job(params['job_id'], reason=reason)
+                        record['pause_recorded'] = _pause_identity(paused)
+                        atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
+                    if record.get('pause_recorded') and not record.get('pause_notified'):
+                        record['pause_notified'] = True
+                        atomic_json_write(journal, record, mode=0o600, fsync_dir=True)
+                        with _fire_secret_scope():
+                            _deliver_crash_failure(
+                                dict(current, execution_id=params['request_id']), error,
+                                adapters=None, loop=None)
+                        logging.getLogger(__name__).warning(
+                            'CRON_UNKNOWN_OUTCOME_PAUSED: job %s execution %s: %s',
+                            params['job_id'], params['request_id'], error)
+                    continue
             success, output, answer, error = state['result']
             job = dict(state['job'], **(state.get('job_flags') or {}))
             job['execution_id'] = params['request_id']

tests/cron/test_canonical_completion_retention.py

@@ -24,3 +24,20 @@ def test_completion_ids_retire_with_their_journal(tmp_path, monkeypatch):
         # The still-journaled completion replayed by recovery is a no-op, not a second run.
         assert jobs.mark_job_run(job_id, True, execution_id='pending') is True
         assert jobs.get_job(job_id)['repeat']['completed'] == completed
+
+
+def test_duplicate_completion_succeeds_after_fire_owner_changes(tmp_path, monkeypatch):
+    monkeypatch.setenv('HERMES_HOME', str(tmp_path / 'home'))
+    with jobs.use_cron_store(tmp_path / 'cron'):
+        job_id = jobs.create_job(prompt='report', schedule='every 1m', deliver='local')['id']
+        journal = scheduler_authority.journal_path(job_id, 'already-booked')
+        journal.parent.mkdir(parents=True, exist_ok=True)
+        journal.write_text('{}', encoding='utf-8')
+        assert jobs.mark_job_run(job_id, True, execution_id='already-booked')
+        records = jobs.load_jobs()
+        records[0]['fire_claim'] = {'by': 'replacement-owner'}
+        jobs.save_jobs(records)
+        assert jobs.mark_job_run(
+            job_id, True, expected_fire_owner='original-owner',
+            execution_id='already-booked')
+        assert jobs.get_job(job_id)['repeat']['completed'] == 1