From d1afa16053a3777849c2b5465d59a0147b2172f9 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:56:25 -0700 Subject: [PATCH] fix(cron): retain completed one-shot jobs instead of deleting them on completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 126 +++++++++++++++++++++++++++++++++++++--- tests/cron/test_jobs.py | 125 ++++++++++++++++++++++++++++++++++++--- 2 files changed, 237 insertions(+), 14 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 904b8aa346269..95020ae659ed9 100644 --- a/cron/jobs.py +++ b/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 / diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index e9402208e3cc4..b0121daed1e0c 100644 --- a/tests/cron/test_jobs.py +++ b/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