diff --git a/cron/executions.py b/cron/executions.py new file mode 100644 index 0000000000000..654d0a81a9a32 --- /dev/null +++ b/cron/executions.py @@ -0,0 +1,222 @@ +"""Durable audit ledger for cron execution attempts. + +The ledger records what is known about each attempt; it is not a retry queue. +An attempt left claimed or running when a scheduler process restarts is marked +``unknown`` because the new process cannot prove whether its side effects ran. +""" + +from __future__ import annotations + +import contextlib +import copy +import json +import os +import tempfile +import threading +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional + +try: + import fcntl +except ImportError: # pragma: no cover - non-Unix + fcntl = None +try: + import msvcrt +except ImportError: # pragma: no cover - non-Windows + msvcrt = None + +from hermes_constants import get_hermes_home +from hermes_time import now as _hermes_now +from utils import atomic_replace + +EXECUTIONS_FILE = get_hermes_home().resolve() / "cron" / "executions.json" +EXECUTIONS_LOCK_FILE = get_hermes_home().resolve() / "cron" / ".executions.lock" +_lock = threading.RLock() +_lock_state = threading.local() +_PROCESS_ID = uuid.uuid4().hex + + +@contextlib.contextmanager +def _ledger_lock(): + depth = getattr(_lock_state, "depth", 0) + if depth: + _lock_state.depth = depth + 1 + try: + yield + finally: + _lock_state.depth -= 1 + return + + with _lock: + EXECUTIONS_LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + lock_file = open(EXECUTIONS_LOCK_FILE, "a+", encoding="utf-8") + try: + if fcntl: + fcntl.flock(lock_file, fcntl.LOCK_EX) + elif msvcrt: # pragma: no cover - Windows + lock_file.seek(0) + if not lock_file.read(1): + lock_file.write("0") + lock_file.flush() + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + _lock_state.depth = 1 + yield + finally: + _lock_state.depth = 0 + if fcntl: + fcntl.flock(lock_file, fcntl.LOCK_UN) + elif msvcrt: # pragma: no cover - Windows + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + lock_file.close() + + +def _load_unlocked() -> List[Dict[str, Any]]: + if not EXECUTIONS_FILE.exists(): + return [] + try: + data = json.loads(EXECUTIONS_FILE.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return [] + records = data.get("executions", []) if isinstance(data, dict) else [] + return records if isinstance(records, list) else [] + + +def _save_unlocked(records: List[Dict[str, Any]]) -> None: + EXECUTIONS_FILE.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp( + dir=str(EXECUTIONS_FILE.parent), prefix=".executions_", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump({"version": 1, "executions": records}, handle, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + atomic_replace(tmp_name, EXECUTIONS_FILE) + try: + os.chmod(EXECUTIONS_FILE, 0o600) + except OSError: + pass + except Exception: + try: + os.unlink(tmp_name) + except OSError: + pass + raise + + +def create_execution(job_id: str, *, source: str) -> Dict[str, Any]: + """Persist a claimed attempt before it is submitted for execution.""" + now = _hermes_now().isoformat() + record: Dict[str, Any] = { + "id": uuid.uuid4().hex, + "job_id": str(job_id), + "source": str(source), + "process_id": _PROCESS_ID, + "pid": os.getpid(), + "status": "claimed", + "claimed_at": now, + "started_at": None, + "finished_at": None, + "error": None, + } + with _ledger_lock(): + records = _load_unlocked() + records.append(record) + _save_unlocked(records) + return copy.deepcopy(record) + + +def _transition(execution_id: str, status: str, **updates: Any) -> Optional[Dict[str, Any]]: + with _ledger_lock(): + records = _load_unlocked() + for record in records: + if record.get("id") != execution_id: + continue + record["status"] = status + record.update(updates) + _save_unlocked(records) + return copy.deepcopy(record) + return None + + +def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: + return _transition( + execution_id, + "running", + started_at=_hermes_now().isoformat(), + ) + + +def finish_execution( + execution_id: str, + *, + success: bool, + error: Optional[str] = None, +) -> Optional[Dict[str, Any]]: + return _transition( + execution_id, + "completed" if success else "failed", + finished_at=_hermes_now().isoformat(), + error=None if success else (str(error) if error else "unknown failure"), + ) + + +def recover_interrupted_executions() -> int: + """Classify prior in-flight attempts as unknown; never enqueue a retry.""" + now = _hermes_now().isoformat() + changed = 0 + + def _owner_is_live(record: Dict[str, Any]) -> bool: + try: + owner_pid = int(record.get("pid")) + except (TypeError, ValueError): + return False + if owner_pid <= 0: + return False + if owner_pid == os.getpid(): + return True + try: + os.kill(owner_pid, 0) # windows-footgun: ok -- liveness probe + return True + except OSError: + return False + + with _ledger_lock(): + records = _load_unlocked() + for record in records: + if record.get("status") not in {"claimed", "running"}: + continue + # Multiple scheduler surfaces can start in one process. Their + # startup recovery must not relabel work this same process is + # currently executing. A real restart imports this module in a new + # process and therefore has a distinct process id. + if record.get("process_id") == _PROCESS_ID or _owner_is_live(record): + continue + record["status"] = "unknown" + record["finished_at"] = now + record["error"] = ( + "Scheduler restarted before this execution reached a durable " + "terminal state; whether side effects ran is unknown." + ) + changed += 1 + if changed: + _save_unlocked(records) + return changed + + +def list_executions(*, job_id: Optional[str] = None) -> List[Dict[str, Any]]: + with _ledger_lock(): + records = copy.deepcopy(_load_unlocked()) + if job_id is not None: + records = [record for record in records if record.get("job_id") == job_id] + records.sort(key=lambda record: str(record.get("claimed_at", "")), reverse=True) + return records + + +def latest_execution(job_id: str) -> Optional[Dict[str, Any]]: + records = list_executions(job_id=job_id) + return records[0] if records else None diff --git a/cron/jobs.py b/cron/jobs.py index d3bb624128c36..46128c2e31242 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -455,6 +455,14 @@ def _normalize_job_record(job: Dict[str, Any]) -> Dict[str, Any]: state = "scheduled" if normalized.get("enabled", True) else "paused" normalized["state"] = state + # Expose the independent latest attempt through existing job detail/list + # readers without coupling execution history into schedule persistence. + try: + from cron.executions import latest_execution + normalized["latest_execution"] = latest_execution(job_id) + except Exception: + normalized["latest_execution"] = None + return normalized diff --git a/cron/scheduler.py b/cron/scheduler.py index 8c52377e21347..20b36f49f7273 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -278,6 +278,7 @@ _LEGACY_HOME_TARGET_ENV_VARS = { } from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch, heartbeat_run_claim +from cron.executions import create_execution, finish_execution, mark_execution_running # Sentinel: when a cron agent has nothing new to report, it can start its # response with this marker to suppress delivery. Output is still saved @@ -3635,6 +3636,9 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - Returns True if the job was processed (even if the job itself failed — failure is recorded via ``mark_job_run``), False only if processing raised. """ + execution_id = job.get("execution_id") + if not execution_id: + execution_id = create_execution(job["id"], source="direct")["id"] try: # Pre-run dispatch claim (issue #38758): atomically commit a finite # one-shot's dispatch BEFORE its side effect runs, so a tick that dies @@ -3648,8 +3652,17 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - "Job '%s': one-shot dispatch limit reached — skipping", job.get("name", job["id"]), ) + finish_execution( + execution_id, + success=False, + error="Dispatch claim rejected; execution was not started.", + ) return True # not an error — already handled/removed + # The attempt is claimed durably before executor/provider dispatch and + # becomes running only immediately before the actual run. + mark_execution_running(execution_id) + # Run the job under the profile's secret scope. get_secret() fails # closed outside a scope once profile isolation is in play (multiple # gateway profiles / room→profile multiplexing), and cron fires from @@ -3756,12 +3769,14 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - if not _consume_interrupted_flag(job["id"]): mark_job_run(job["id"], success, error, delivery_error=delivery_error) + finish_execution(execution_id, success=success, error=error) return True except Exception as e: logger.error("Error processing job %s: %s", job['id'], e) if not _consume_interrupted_flag(job["id"]): mark_job_run(job["id"], False, str(e)) + finish_execution(execution_id, success=False, error=str(e)) return False @@ -3916,9 +3931,13 @@ def tick( logger.info("Job '%s' already running — skipping", job.get("name", job_id)) return None _running_job_ids.add(job_id) + # Record the attempt before executor dispatch. Recovery classifies + # abandoned records as unknown; it never automatically retries them. + execution = create_execution(job_id, source="builtin") + dispatched_job = dict(job, execution_id=execution["id"]) _ctx = contextvars.copy_context() - def _run_and_release(j=job, ctx=_ctx): + def _run_and_release(j=dispatched_job, ctx=_ctx): try: return ctx.run(_process_job, j) finally: @@ -3933,6 +3952,11 @@ def tick( if _interpreter_shutting_down(submit_err): with _running_lock: _running_job_ids.discard(job_id) + finish_execution( + execution["id"], + success=False, + error="Interpreter shutdown prevented executor dispatch.", + ) logger.warning( "Job '%s' not dispatched — interpreter is shutting down", job.get("name", job_id), diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 5429f79cf3621..29bcaf6794793 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -95,6 +95,7 @@ class CronScheduler(ABC): was lost (another machine/retry won it) or the job no longer exists. """ from cron.jobs import claim_job_for_fire, get_job + from cron.executions import create_execution from cron.scheduler import run_one_job if not claim_job_for_fire(job_id): @@ -102,6 +103,7 @@ class CronScheduler(ABC): job = get_job(job_id) if job is None: return False # job removed (e.g. repeat-N exhausted) between arm and fire + job["execution_id"] = create_execution(job_id, source=self.name)["id"] return run_one_job(job, adapters=adapters, loop=loop) def reconcile(self) -> None: @@ -169,9 +171,16 @@ class InProcessCronScheduler(CronScheduler): import logging from cron.scheduler import tick as cron_tick from cron.jobs import record_ticker_heartbeat + from cron.executions import recover_interrupted_executions logger = logging.getLogger("cron.scheduler_provider") logger.info("In-process cron scheduler started (interval=%ds)", interval) + recovered = recover_interrupted_executions() + if recovered: + logger.warning( + "Marked %d interrupted cron execution(s) unknown after restart", + recovered, + ) # Heartbeat once before the first sleep so `hermes cron status` sees a # live ticker immediately after startup, not only after the first tick. record_ticker_heartbeat() diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index b0c907326e882..e383e2dc4b498 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -174,6 +174,13 @@ def cron_list(show_all: bool = False): status_display = color(f"{last_status}: {job.get('last_error', '?')}", Colors.RED) print(f" Last run: {last_run} {status_display}") + latest_execution = job.get("latest_execution") + if latest_execution: + print( + f" Execution: {latest_execution.get('status', '?')} " + f"{latest_execution.get('id', '?')}" + ) + delivery_err = job.get("last_delivery_error") if delivery_err: print(f" {color('⚠ Delivery failed:', Colors.YELLOW)} {delivery_err}") diff --git a/plugins/cron_providers/chronos/__init__.py b/plugins/cron_providers/chronos/__init__.py index 6f04dc12fa380..ac0e39e55cba1 100644 --- a/plugins/cron_providers/chronos/__init__.py +++ b/plugins/cron_providers/chronos/__init__.py @@ -106,6 +106,11 @@ class ChronosCronScheduler(CronScheduler): Does NOT block and does NOT spawn a 60s wake (DQ-1) — that is the whole point of scale-to-zero. The machine wakes only on a NAS→agent fire. """ + # A new provider lifecycle cannot prove what an interrupted prior + # process did. Classify those attempts unknown for audit only; do not + # requeue them here. + from cron.executions import recover_interrupted_executions + recover_interrupted_executions() try: self.reconcile() except Exception as e: diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py new file mode 100644 index 0000000000000..54812785b7f60 --- /dev/null +++ b/tests/cron/test_execution_ledger.py @@ -0,0 +1,205 @@ +"""Durable cron execution-ledger behavior.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + + +def _point_ledger(monkeypatch, tmp_path): + import cron.executions as executions + + monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.json") + monkeypatch.setattr(executions, "EXECUTIONS_LOCK_FILE", tmp_path / "cron" / ".executions.lock") + return executions + + +def test_execution_transitions_are_durable(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + + claimed = executions.create_execution("job-1", source="builtin") + assert claimed["status"] == "claimed" + assert claimed["claimed_at"] + assert claimed["started_at"] is None + assert claimed["finished_at"] is None + + running = executions.mark_execution_running(claimed["id"]) + assert running["status"] == "running" + assert running["started_at"] + + completed = executions.finish_execution(claimed["id"], success=True) + assert completed["status"] == "completed" + assert completed["finished_at"] + assert completed["error"] is None + + persisted = executions.list_executions(job_id="job-1") + assert persisted == [completed] + + +def test_failed_execution_keeps_error(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + + record = executions.create_execution("job-2", source="external") + failed = executions.finish_execution(record["id"], success=False, error="provider exploded") + + assert failed["status"] == "failed" + assert failed["error"] == "provider exploded" + + +def test_recovery_does_not_mark_live_process_execution_unknown(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("still-live", source="builtin") + executions.mark_execution_running(record["id"]) + + assert executions.recover_interrupted_executions() == 0 + assert executions.latest_execution("still-live")["status"] == "running" + + +def test_recovery_does_not_mark_other_live_owner_unknown(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("other-live", source="builtin") + records = json.loads(executions.EXECUTIONS_FILE.read_text())["executions"] + records[0]["process_id"] = "another-import" + records[0]["pid"] = os.getpid() + executions.EXECUTIONS_FILE.write_text(json.dumps({"version": 1, "executions": records})) + + assert executions.recover_interrupted_executions() == 0 + assert executions.latest_execution("other-live")["status"] == "claimed" + + +def test_restart_marks_interrupted_execution_unknown_without_requeue(tmp_path): + """Real temp-HERMES_HOME subprocess restart: in-flight is audit-only unknown.""" + home = tmp_path / "home" + repo = Path(__file__).resolve().parents[2] + env = os.environ.copy() + env["HERMES_HOME"] = str(home) + env["PYTHONPATH"] = str(repo) + + create = subprocess.run( + [ + sys.executable, + "-c", + "from cron.executions import create_execution, mark_execution_running; " + "r=create_execution('restart-job', source='builtin'); " + "mark_execution_running(r['id']); print(r['id'])", + ], + cwd=repo, + env=env, + text=True, + capture_output=True, + check=True, + ) + execution_id = create.stdout.strip() + + recover = subprocess.run( + [ + sys.executable, + "-c", + "import json; from cron.executions import recover_interrupted_executions, list_executions; " + "print(recover_interrupted_executions()); " + "print(json.dumps(list_executions(job_id='restart-job'))) ", + ], + cwd=repo, + env=env, + text=True, + capture_output=True, + check=True, + ) + lines = recover.stdout.strip().splitlines() + assert lines[0] == "1" + records = json.loads(lines[1]) + assert len(records) == 1 + assert records[0]["id"] == execution_id + assert records[0]["status"] == "unknown" + assert records[0]["finished_at"] + assert "restart" in records[0]["error"].lower() + # Recovery only classifies the old attempt. It must not manufacture a new + # claimed record (which would imply an automatic retry). + assert [r["status"] for r in records] == ["unknown"] + + +def test_run_one_job_records_running_then_terminal(monkeypatch): + import cron.scheduler as scheduler + + events = [] + monkeypatch.setattr( + scheduler, + "mark_execution_running", + lambda execution_id: events.append(("running", execution_id)), + raising=False, + ) + monkeypatch.setattr( + scheduler, + "finish_execution", + lambda execution_id, **kwargs: events.append(("finish", execution_id, kwargs)), + raising=False, + ) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda job, *, defer_agent_teardown=None: (True, "output", "response", None), + ) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_args: None) + monkeypatch.setattr(scheduler, "_deliver_result", lambda *_args, **_kwargs: None) + monkeypatch.setattr(scheduler, "mark_job_run", lambda *_args, **_kwargs: None) + + assert scheduler.run_one_job({"id": "job-3", "execution_id": "exec-3"}) is True + assert events[0] == ("running", "exec-3") + assert events[-1][0:2] == ("finish", "exec-3") + assert events[-1][2]["success"] is True + + +def test_provider_start_recovers_interrupted_records_before_tick(monkeypatch): + import cron.scheduler_provider as provider + + events = [] + stop = __import__("threading").Event() + stop.set() + monkeypatch.setattr( + "cron.executions.recover_interrupted_executions", + lambda: events.append("recover") or 0, + raising=False, + ) + monkeypatch.setattr("cron.jobs.record_ticker_heartbeat", lambda **_kwargs: events.append("heartbeat")) + + provider.InProcessCronScheduler().start(stop, interval=1) + + assert events[:2] == ["recover", "heartbeat"] + + +def test_external_provider_start_recovers_interrupted_records(monkeypatch): + from plugins.cron_providers.chronos import ChronosCronScheduler + + provider = ChronosCronScheduler() + provider._client = type("Client", (), {"arm": lambda self, **kwargs: None})() + events = [] + monkeypatch.setattr( + "cron.executions.recover_interrupted_executions", + lambda: events.append("recover") or 0, + ) + monkeypatch.setattr(provider, "reconcile", lambda: events.append("reconcile")) + + provider.start(__import__("threading").Event()) + + assert events == ["recover", "reconcile"] + + +def test_job_listing_exposes_latest_execution(monkeypatch, tmp_path): + import cron.jobs as jobs + + monkeypatch.setattr(jobs, "CRON_DIR", tmp_path / "cron") + monkeypatch.setattr(jobs, "JOBS_FILE", tmp_path / "cron" / "jobs.json") + monkeypatch.setattr(jobs, "OUTPUT_DIR", tmp_path / "cron" / "output") + executions = _point_ledger(monkeypatch, tmp_path) + + job = jobs.create_job(prompt="audit me", schedule="every 1h", name="audit") + record = executions.create_execution(job["id"], source="builtin") + executions.mark_execution_running(record["id"]) + + listed = jobs.list_jobs(include_disabled=True) + assert listed[0]["latest_execution"]["id"] == record["id"] + assert listed[0]["latest_execution"]["status"] == "running"