fix(cron): retain completed one-shot jobs instead of deleting them on completion

mark_job_run popped a finite one-shot from jobs.json the moment its
repeat limit was reached and returned early — discarding the
last_status / last_error / last_delivery_error it had just written.
Every finished one-shot vanished from `cronjob action=list` with no
inspectable record, and a delivery failure (agent succeeded, platform
send failed) was silently thrown away with it.

Changes:
- mark_job_run now retires a limit-reached one-shot as a terminal
  record (state="completed", enabled=False, next_run_at=None) —
  mirroring the existing next_run_at-is-None terminal branch — so the
  final status and any delivery error persist and surface in the
  cronjob tool's list output (which already emits last_delivery_error
  and defaults to include_disabled=True).
- claim_dispatch's stale-job cleanup marks already-ran jobs completed
  instead of popping them; genuinely wedged claims (last_run_at never
  written) are still removed with the operator-visible diagnostic.
- Retention sweep in the due scan prunes completed one-shot records
  older than cron.completed_retention_days (default 7; non-positive
  disables) so jobs.json cannot grow unboundedly. Recurring jobs and
  non-terminal one-shots are never candidates.

Tests: completion retains record + delivery error, list surfaces it,
completed jobs never re-dispatch, sweep prunes old / keeps recent /
ignores recurring / honors the disable knob; recurring lifecycle
unchanged.

cron/jobs.py

@@ -1738,8 +1738,19 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
 
                     # Check if we've hit the repeat limit
                     if times is not None and times > 0 and completed >= times:
-                        # Remove the job (limit reached)
-                        jobs.pop(i)
+                        # Limit reached: retain the record as a terminal
+                        # completion instead of popping it. Deleting the job
+                        # here discarded the last_status / last_error /
+                        # last_delivery_error written above — a finished
+                        # one-shot vanished from `cronjob list` with no
+                        # inspectable outcome, and a failed delivery was
+                        # invisible. Mirror the terminal shape of the
+                        # next_run_at-is-None branch below; the retention
+                        # sweep prunes these after
+                        # COMPLETED_ONESHOT_RETENTION_DAYS.
+                        job["enabled"] = False
+                        job["state"] = "completed"
+                        job["next_run_at"] = None
                         save_jobs(jobs)
                         return
                 
@@ -1857,13 +1868,29 @@ def claim_dispatch(job_id: str) -> bool:
                 return True  # infinite — always dispatch
             completed = repeat.get("completed", 0)
             if completed >= times:
-                # Already dispatched the max number of times (e.g. a prior
-                # tick claimed then died before mark_job_run could remove it).
-                # Clean up so it stops appearing as due on every tick.
+                # Already dispatched the max number of times.
+                if job.get("last_run_at") is not None:
+                    # A prior run completed normally (e.g. mark_job_run raced
+                    # with this tick). Retain the terminal record — same shape
+                    # as mark_job_run's repeat-limit branch — instead of
+                    # deleting the job and its final status/delivery error.
+                    job["enabled"] = False
+                    job["state"] = "completed"
+                    job["next_run_at"] = None
+                    save_jobs(jobs)
+                    logger.info(
+                        "Job '%s': dispatch limit reached (%d/%d) — marking completed",
+                        job.get("name", job.get("id", "?")),
+                        completed,
+                        times,
+                    )
+                    return False
+                # A prior tick claimed the dispatch then died before the run
+                # completed (#73973) — a genuinely wedged claim. Remove it so
+                # it stops appearing as due, and leave an operator-visible
+                # diagnostic instead of vanishing silently.
                 jobs.pop(i)
                 save_jobs(jobs)
-                # If the claimed run never completed (#73973), leave an
-                # operator-visible diagnostic instead of vanishing silently.
                 _write_wedged_oneshot_diagnostic(job)
                 logger.info(
                     "Job '%s': dispatch limit reached (%d/%d) — removing",
@@ -2049,6 +2076,82 @@ def claim_job_for_fire(job_id: str, *, claim_ttl_seconds: int = 300) -> bool:
         return False
 
 
+# Completed one-shot job records are retained in jobs.json (final status +
+# delivery error stay inspectable via `cronjob list`) instead of being deleted
+# at completion, then pruned by _sweep_completed_oneshots once they age out.
+COMPLETED_ONESHOT_RETENTION_DAYS = 7
+
+
+def _completed_oneshot_retention_days() -> float:
+    """Resolve the completed one-shot retention window from config.
+
+    ``cron.completed_retention_days`` (number, default
+    ``COMPLETED_ONESHOT_RETENTION_DAYS``). A non-positive value disables the
+    sweep, retaining completed one-shot records indefinitely.
+    """
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config() or {}
+        cron_cfg = cfg.get("cron", {}) if isinstance(cfg, dict) else {}
+        return float(
+            cron_cfg.get(
+                "completed_retention_days", COMPLETED_ONESHOT_RETENTION_DAYS
+            )
+        )
+    except Exception:
+        return float(COMPLETED_ONESHOT_RETENTION_DAYS)
+
+
+def _sweep_completed_oneshots(raw_jobs: List[Dict[str, Any]], now: datetime) -> bool:
+    """Prune terminal ``state == "completed"`` one-shot records past retention.
+
+    Mutates *raw_jobs* in place; returns True when anything was removed (the
+    caller persists). Only one-shot (``schedule.kind == "once"``) records in
+    the terminal completed state are candidates; recurring jobs and non-
+    terminal one-shots are never touched. Age is measured from
+    ``last_run_at`` — a completed record without a parseable ``last_run_at``
+    is kept (never guess a record into deletion).
+    """
+    retention_days = _completed_oneshot_retention_days()
+    if retention_days <= 0:
+        return False
+    cutoff = now - timedelta(days=retention_days)
+    removed = False
+    for rj in list(raw_jobs):
+        try:
+            if rj.get("state") != "completed":
+                continue
+            schedule = rj.get("schedule")
+            kind = schedule.get("kind") if isinstance(schedule, dict) else None
+            if kind != "once":
+                continue
+            last_run = rj.get("last_run_at")
+            if not isinstance(last_run, str):
+                continue
+            try:
+                last_run_dt = _ensure_aware(datetime.fromisoformat(last_run))
+            except Exception:
+                continue
+            if last_run_dt >= cutoff:
+                continue
+            raw_jobs.remove(rj)
+            removed = True
+            logger.info(
+                "Job '%s': pruning completed one-shot record "
+                "(finished %s, retention %.1f days)",
+                rj.get("name", rj.get("id", "?")),
+                last_run,
+                retention_days,
+            )
+        except Exception:
+            logger.debug(
+                "Retention sweep skipped malformed job record %r",
+                rj.get("id", "?"),
+                exc_info=True,
+            )
+    return removed
+
+
 def get_due_jobs() -> List[Dict[str, Any]]:
     """Get all jobs that are due to run now.
 
@@ -2168,6 +2271,15 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
     # (derived from HERMES_CRON_TIMEOUT). See _oneshot_run_claim_ttl_seconds.
     _run_claim_ttl = _oneshot_run_claim_ttl_seconds()
 
+    # Retention sweep: completed one-shots are retained (so their final
+    # status / delivery error stay inspectable via `cronjob list`) instead of
+    # being deleted on completion, but they must not accumulate in jobs.json
+    # forever. Prune terminal one-shot records older than the retention
+    # window each scan.
+    if _sweep_completed_oneshots(raw_jobs, now):
+        needs_save = True
+        jobs = [j for j in jobs if any(rj.get("id") == j.get("id") for rj in raw_jobs)]
+
     for job in jobs:
         # Per-job containment (structural guard): one malformed or
         # unexpected job record must never abort the whole scan. The id /

tests/cron/test_jobs.py

@@ -323,11 +323,45 @@ class TestMarkJobRun:
         assert updated["repeat"]["completed"] == 1
         assert updated["last_status"] == "ok"
 
-    def test_repeat_limit_removes_job(self, tmp_cron_dir):
+    def test_repeat_limit_retains_completed_record(self, tmp_cron_dir):
+        """A finished one-shot must stay inspectable, not vanish from the store."""
         job = create_job(prompt="Once", schedule="30m", repeat=1)
         mark_job_run(job["id"], success=True)
-        # Job should be removed after hitting repeat limit
-        assert get_job(job["id"]) is None
+        updated = get_job(job["id"])
+        assert updated is not None, "completed one-shot was deleted from jobs.json"
+        assert updated["state"] == "completed"
+        assert updated["enabled"] is False
+        assert updated["next_run_at"] is None
+        assert updated["last_status"] == "ok"
+
+    def test_repeat_limit_retains_delivery_error(self, tmp_cron_dir):
+        """A one-shot whose delivery failed must keep the error on its record."""
+        job = create_job(prompt="Once", schedule="30m", repeat=1)
+        mark_job_run(
+            job["id"], success=True,
+            delivery_error="platform 'telegram' not configured",
+        )
+        updated = get_job(job["id"])
+        assert updated is not None
+        assert updated["state"] == "completed"
+        assert updated["last_delivery_error"] == "platform 'telegram' not configured"
+
+    def test_completed_oneshot_visible_in_list(self, tmp_cron_dir):
+        """list_jobs(include_disabled=True) surfaces the completed record."""
+        job = create_job(prompt="Once", schedule="30m", repeat=1)
+        mark_job_run(job["id"], success=True, delivery_error="send failed: 502")
+        listed = {j["id"]: j for j in list_jobs(include_disabled=True)}
+        assert job["id"] in listed
+        assert listed[job["id"]]["state"] == "completed"
+        assert listed[job["id"]]["last_delivery_error"] == "send failed: 502"
+        # Default (enabled-only) listing hides it, matching paused/disabled jobs.
+        assert job["id"] not in {j["id"] for j in list_jobs()}
+
+    def test_completed_oneshot_not_due(self, tmp_cron_dir):
+        """A retained completed one-shot must never be dispatched again."""
+        job = create_job(prompt="Once", schedule="30m", repeat=1)
+        mark_job_run(job["id"], success=True)
+        assert job["id"] not in {j["id"] for j in get_due_jobs()}
 
 
     def test_error_status(self, tmp_cron_dir):
@@ -639,9 +673,14 @@ class TestGetDueJobs:
         assert get_job("slowrun") is not None
 
         # Run completes → outcome lands on a record that still exists
-        # (times=1 reached, so mark_job_run retires the job normally).
+        # (times=1 reached, so mark_job_run retires the job as a terminal
+        # completed record instead of deleting it).
         mark_job_run("slowrun", True)
-        assert get_job("slowrun") is None
+        retired = get_job("slowrun")
+        assert retired is not None
+        assert retired["state"] == "completed"
+        assert retired["enabled"] is False
+        assert retired["last_status"] == "ok"
 
 
     def test_heartbeat_run_claim_rejects_replaced_owner(self, tmp_cron_dir):
@@ -900,12 +939,17 @@ class TestClaimDispatch:
 
     def test_mark_job_run_does_not_double_count_preclaimed_oneshot(self, tmp_cron_dir):
         # Full lifecycle: claim bumps completed to times, then mark_job_run must
-        # NOT increment again — it recognizes the pre-claim and removes the job.
+        # NOT increment again — it recognizes the pre-claim and retires the job
+        # as a terminal completed record (retained for inspection, not re-fired).
         save_jobs([self._oneshot(times=1, completed=0)])
         assert claim_dispatch("os1") is True
         assert load_jobs()[0]["repeat"]["completed"] == 1
         mark_job_run("os1", success=True)
-        assert load_jobs() == []  # completed once, removed — not fired twice
+        retired = load_jobs()
+        assert len(retired) == 1  # completed once, retired — not fired twice
+        assert retired[0]["repeat"]["completed"] == 1  # no double count
+        assert retired[0]["state"] == "completed"
+        assert retired[0]["enabled"] is False
 
 
     def test_get_due_jobs_removes_stale_maxed_oneshot(self, tmp_cron_dir):
@@ -1139,3 +1183,70 @@ class TestAdvanceNextRuns:
         assert advance_next_run(rec_ids[0]) is True
         assert advance_next_run(one_ids[0]) is False
         assert advance_next_run("missing-id") is False
+
+
+# =========================================================================
+# Completed one-shot retention sweep
+# =========================================================================
+
+class TestCompletedOneshotRetentionSweep:
+    """Completed one-shots are retained for inspection, then pruned by age."""
+
+    def _completed_oneshot(self, age_days: float):
+        """Create a one-shot, complete it, and backdate its last_run_at."""
+        job = create_job(prompt="Once", schedule="30m", repeat=1)
+        mark_job_run(job["id"], success=True, delivery_error="boom")
+        stamp = (
+            datetime.now(timezone.utc) - timedelta(days=age_days)
+        ).isoformat()
+        jobs = load_jobs()
+        for j in jobs:
+            if j["id"] == job["id"]:
+                j["last_run_at"] = stamp
+        save_jobs(jobs)
+        return job["id"]
+
+    def test_sweep_prunes_old_completed_oneshot(self, tmp_cron_dir):
+        old_id = self._completed_oneshot(age_days=30)
+        get_due_jobs()  # sweep runs as part of the due scan
+        assert get_job(old_id) is None
+
+    def test_sweep_keeps_recent_completed_oneshot(self, tmp_cron_dir):
+        recent_id = self._completed_oneshot(age_days=1)
+        get_due_jobs()
+        kept = get_job(recent_id)
+        assert kept is not None
+        assert kept["state"] == "completed"
+        assert kept["last_delivery_error"] == "boom"
+
+    def test_sweep_ignores_recurring_jobs(self, tmp_cron_dir):
+        """Old recurring jobs are never candidates, whatever their history."""
+        job = create_job(prompt="Recurring", schedule="every 1h")
+        stamp = (
+            datetime.now(timezone.utc) - timedelta(days=365)
+        ).isoformat()
+        jobs = load_jobs()
+        for j in jobs:
+            if j["id"] == job["id"]:
+                j["last_run_at"] = stamp
+        save_jobs(jobs)
+        get_due_jobs()
+        assert get_job(job["id"]) is not None
+
+    def test_sweep_disabled_by_nonpositive_retention(self, tmp_cron_dir, monkeypatch):
+        monkeypatch.setattr(
+            "cron.jobs._completed_oneshot_retention_days", lambda: 0.0
+        )
+        old_id = self._completed_oneshot(age_days=30)
+        get_due_jobs()
+        assert get_job(old_id) is not None
+
+    def test_recurring_jobs_unaffected_by_retention_change(self, tmp_cron_dir):
+        """A recurring job still cycles normally alongside retained one-shots."""
+        recurring = create_job(prompt="Recurring", schedule="every 1h")
+        self._completed_oneshot(age_days=1)
+        mark_job_run(recurring["id"], success=True)
+        updated = get_job(recurring["id"])
+        assert updated["enabled"] is True
+        assert updated["state"] == "scheduled"
+        assert updated["next_run_at"] is not None