From c7a5de7d6eff45156f100dd32f120bc4c4c2bc08 Mon Sep 17 00:00:00 2001 From: rjvandeve <138697058+rjvandeve@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:29:35 -0400 Subject: [PATCH] fix(cron): make pause authoritative against half-paused records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pause_job already sets enabled=false atomically with state/paused_at, but get_due_jobs only checked enabled — so a contradictory record (enabled=true + paused_at/state=paused) still fired. That was the 07-30 outage failure mode: list looked frozen, fleet kept merging. - is_job_runnable / effective_job_state: pause markers gate fire; display derives from the scheduler-honoured enabled flag so half-paused never renders as [paused] - get_due_jobs self-heals enabled=false + logs error on contradiction - claim_job_for_fire uses is_job_runnable (paused_at counts too) - list/format paths use effective_job_state - behavioural tests: pause blocks due fire; half-pause self-disables --- cron/jobs.py | 81 ++++++++++++++++++++++++++++++++++++++--- hermes_cli/cron.py | 6 ++- tests/cron/test_jobs.py | 80 ++++++++++++++++++++++++++++++++++++++++ tools/cronjob_tools.py | 5 ++- 4 files changed, 165 insertions(+), 7 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 9d5d44b126f47..377b34c35487a 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -471,14 +471,55 @@ def _normalize_job_record(job: Dict[str, Any]) -> Dict[str, Any]: normalized["name"] = name normalized["schedule_display"] = _schedule_display_for_job(normalized) - state = _coerce_job_text(normalized.get("state")).strip() - if not state: - state = "scheduled" if normalized.get("enabled", True) else "paused" - normalized["state"] = state + # Display state is derived from the scheduler-honoured ``enabled`` flag so a + # half-paused record (enabled=true + state/paused_at) cannot render as + # "paused" while the fleet is still live. See effective_job_state(). + normalized["state"] = effective_job_state(normalized) return normalized +def _has_pause_marker(job: Dict[str, Any]) -> bool: + """True when the record carries any operator-facing pause signal.""" + if _coerce_job_text(job.get("state")).strip() == "paused": + return True + return bool(job.get("paused_at")) + + +def is_job_runnable(job: Dict[str, Any]) -> bool: + """True iff the scheduler may fire this job. + + ``enabled`` is the scheduler-honoured flag. Pause markers (``state`` / + ``paused_at``) are a second gate so a contradictory half-paused record + never fires even before self-heal runs. + """ + if not job.get("enabled", True): + return False + if _has_pause_marker(job): + return False + return True + + +def effective_job_state(job: Dict[str, Any]) -> str: + """Operator-facing state derived from the scheduler-honoured flag. + + A job with ``enabled=true`` must never display as paused — that was the + 07-30 outage failure mode (list looked frozen, fleet kept merging). + Terminal states (completed/error) are preserved regardless of enabled. + """ + stored = _coerce_job_text(job.get("state")).strip() + if stored in {"completed", "error"}: + return stored + if not job.get("enabled", True): + if _has_pause_marker(job) or stored == "paused": + return "paused" + return stored or "paused" + # enabled=true is authoritative: never claim paused + if stored == "paused" or job.get("paused_at"): + return "scheduled" + return stored or "scheduled" + + def _secure_dir(path: Path): """Set directory to owner-only access (0700). No-op on Windows.""" try: @@ -2382,7 +2423,9 @@ def claim_job_for_fire(job_id: str, *, claim_ttl_seconds: int = 300) -> bool: for job in jobs: if job["id"] != job_id: continue - if not job.get("enabled", True) or job.get("state") == "paused": + # enabled + pause markers must both clear — a half-paused record + # (enabled=true, state=paused/paused_at set) must not claim. + if not is_job_runnable(job): return False now = _hermes_now() existing = job.get("fire_claim") @@ -2637,6 +2680,34 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: if not job.get("enabled", True): continue + # Contradiction self-heal: enabled=true with pause markers means the + # operator believes the job is frozen while the scheduler would still + # fire it (07-30 outage). Refuse to run and force enabled=false so + # the next list/report is honest. Log loudly — this should be rare + # after pause_job sets both fields atomically. + if _has_pause_marker(job): + jid = job.get("id") + logger.error( + "Job '%s' (%s) has pause markers while enabled=true; " + "self-disabling so it cannot fire (pause must be authoritative).", + job.get("name", jid), + jid, + ) + for rj in raw_jobs: + if rj.get("id") != jid: + continue + rj["enabled"] = False + rj["state"] = "paused" + if not rj.get("paused_at"): + rj["paused_at"] = now.isoformat() + if not rj.get("paused_reason"): + rj["paused_reason"] = ( + "auto-disabled: enabled+paused contradiction" + ) + needs_save = True + break + continue + # Cross-process running-claim guard (#59229): if another scheduler # process already claimed this one-shot and its run is still in flight # (claim younger than the TTL), skip it — do NOT re-dispatch. The diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index aa3fad4437b6f..7d18124bc2e39 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -113,11 +113,15 @@ def cron_list(show_all: bool = False): print(color("└─────────────────────────────────────────────────────────────────────────┘", Colors.CYAN)) print() + from cron.jobs import effective_job_state + for job in jobs: job_id = job.get("id", "?") name = job.get("name", "(unnamed)") schedule = job.get("schedule_display", job.get("schedule", {}).get("value", "?")) - state = job.get("state", "scheduled" if job.get("enabled", True) else "paused") + # Derive from the scheduler-honoured flag — never show [paused] when + # enabled=true (half-paused contradiction must not look frozen). + state = effective_job_state(job) next_run = job.get("next_run_at", "?") # `repeat` may be present-but-null in the job record (e.g. a one-shot diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index b0121daed1e0c..17b4347be44b6 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -20,9 +20,11 @@ from cron.jobs import ( mark_job_run, advance_next_run, claim_dispatch, + claim_job_for_fire, heartbeat_run_claim, get_due_jobs, save_job_output, + _hermes_now, ) @@ -260,7 +262,85 @@ class TestPauseResumeJob: assert paused["enabled"] is False assert paused["state"] == "paused" assert paused["paused_reason"] == "user paused" + assert paused.get("paused_at") + def test_pause_is_authoritative_due_jobs_do_not_fire(self, tmp_cron_dir): + """Behavioural invariant: after pause, a past-due job must not be due. + + Checks that last_run_at cannot advance via the scheduler path — not + merely that pause() returned success. Regression for the 07-30 outage + where state=paused coexisted with enabled=true and jobs kept firing. + """ + job = create_job(prompt="Must not fire while paused", schedule="every 1h") + past = (_hermes_now() - timedelta(hours=2)).isoformat() + # Force the job overdue, then pause. + updated = update_job(job["id"], {"next_run_at": past}) + assert updated["enabled"] is True + assert job["id"] in {j["id"] for j in get_due_jobs()} + + paused = pause_job(job["id"], reason="outage freeze") + assert paused["enabled"] is False + assert paused["state"] == "paused" + assert paused.get("paused_at") + # Scheduler-honoured flag and pause markers must never contradict. + assert not (paused.get("enabled") and paused.get("paused_at")) + + due_ids = {j["id"] for j in get_due_jobs()} + assert job["id"] not in due_ids + + before = get_job(job["id"]) + assert before["last_run_at"] is None or before["last_run_at"] == job.get("last_run_at") + # claim path also closed + assert claim_job_for_fire(job["id"]) is False + after = get_job(job["id"]) + assert after["last_run_at"] == before.get("last_run_at") + assert after["enabled"] is False + + def test_contradictory_half_pause_self_disables_and_does_not_fire(self, tmp_cron_dir): + """enabled=true + paused_at must not fire; scan heals enabled=false.""" + now = _hermes_now() + job = { + "id": "half-paused-1", + "name": "half-paused", + "prompt": "should never run", + "schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"}, + "schedule_display": "every 5m", + "repeat": {"times": None, "completed": 0}, + # The contradiction from the 07-30 outage: + "enabled": True, + "state": "paused", + "paused_at": (now - timedelta(hours=20)).isoformat(), + "paused_reason": "operator thought this was frozen", + "next_run_at": (now - timedelta(hours=1)).isoformat(), + "last_run_at": None, + "last_status": None, + "last_error": None, + "last_delivery_error": None, + "created_at": (now - timedelta(days=1)).isoformat(), + "deliver": "local", + } + save_jobs([job]) + + # Display must NOT say paused while enabled (honest list). + from cron.jobs import effective_job_state, list_jobs + + assert effective_job_state(job) == "scheduled" + listed = {j["id"]: j for j in list_jobs(include_disabled=True)} + # Honest list: enabled=true half-pause must not render as paused. + assert listed["half-paused-1"]["enabled"] is True + assert listed["half-paused-1"]["state"] != "paused" + + assert claim_job_for_fire("half-paused-1") is False + due = get_due_jobs() + assert "half-paused-1" not in {j["id"] for j in due} + + healed = get_job("half-paused-1") + assert healed is not None + assert healed["enabled"] is False + assert healed["state"] == "paused" + assert healed.get("paused_at") + # Still not due after heal + assert "half-paused-1" not in {j["id"] for j in get_due_jobs()} def test_resume_rejects_past_oneshot(self, tmp_cron_dir, monkeypatch): """Resuming a paused one-shot whose time is now in the past must raise diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index ec827aac4c6f5..659eb8bec10d9 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -552,6 +552,8 @@ def _validate_cron_script_path(script: Optional[str]) -> Optional[str]: def _format_job(job: Dict[str, Any]) -> Dict[str, Any]: + from cron.jobs import effective_job_state + prompt = str(job.get("prompt") or "") skills = _canonical_skills(job.get("skill"), job.get("skills")) job_id = str(job.get("id") or "unknown") @@ -573,7 +575,8 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]: "last_status": job.get("last_status"), "last_delivery_error": job.get("last_delivery_error"), "enabled": job.get("enabled", True), - "state": job.get("state", "scheduled" if job.get("enabled", True) else "paused"), + # Derive from enabled so half-paused records never render as paused. + "state": effective_job_state(job), "paused_at": job.get("paused_at"), "paused_reason": job.get("paused_reason"), }