fix(cron): make pause authoritative against half-paused records

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

@@ -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

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

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

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"),
     }