From abc22cdf1a5c0fe30bf1a226bfe3caf489e8316e Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 12 Jul 2026 23:50:37 -0700 Subject: [PATCH] fix(cron): harden execution attempt ledger --- cron/executions.py | 348 +++++++++--------- cron/jobs.py | 16 +- cron/scheduler.py | 25 +- cron/scheduler_provider.py | 9 +- hermes_cli/backup.py | 1 + hermes_cli/cron.py | 24 +- hermes_cli/subcommands/cron.py | 6 + plugins/cron_providers/chronos/__init__.py | 3 +- tests/cron/test_execution_ledger.py | 128 ++++++- tests/hermes_cli/test_cron_parser_builder.py | 8 +- website/docs/user-guide/features/cron.md | 14 + .../current/user-guide/features/cron.md | 8 +- 12 files changed, 387 insertions(+), 203 deletions(-) diff --git a/cron/executions.py b/cron/executions.py index 654d0a81a9a..0abd13b73ef 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -1,222 +1,228 @@ -"""Durable audit ledger for cron execution attempts. +"""Profile-local 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. +Interrupted attempts become ``unknown`` only after their exact owner process is +proved gone. Terminal states are immutable. """ from __future__ import annotations -import contextlib -import copy import json import os -import tempfile +import sqlite3 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" +EXECUTIONS_FILE = get_hermes_home().resolve() / "cron" / "executions.db" +MAX_TERMINAL_EXECUTIONS = 1000 +_TERMINAL_STATES = ("completed", "failed", "unknown") _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: +def _connect() -> sqlite3.Connection: EXECUTIONS_FILE.parent.mkdir(parents=True, exist_ok=True) - fd, tmp_name = tempfile.mkstemp( - dir=str(EXECUTIONS_FILE.parent), prefix=".executions_", suffix=".tmp" + conn = sqlite3.connect(EXECUTIONS_FILE, timeout=5) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout=5000") + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=FULL") + conn.execute( + """CREATE TABLE IF NOT EXISTS executions ( + id TEXT PRIMARY KEY, + job_id TEXT NOT NULL, + source TEXT NOT NULL, + process_id TEXT NOT NULL, + pid INTEGER NOT NULL, + process_started_at INTEGER, + status TEXT NOT NULL CHECK(status IN + ('claimed','running','completed','failed','unknown')), + claimed_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + error TEXT + )""" ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_executions_job_claimed " + "ON executions(job_id, claimed_at DESC, id DESC)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_executions_status_claimed " + "ON executions(status, claimed_at DESC, id DESC)" + ) + return conn + + +def _record(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]: + return dict(row) if row is not None else None + + +def _process_start_time(pid: int) -> Optional[int]: 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 + from gateway.status import get_process_start_time + return get_process_start_time(pid) except Exception: - try: - os.unlink(tmp_name) - except OSError: - pass - raise + return None + + +def _owner_is_live(pid: int, started_at: Optional[int]) -> bool: + try: + from gateway.status import _pid_exists + if not _pid_exists(pid): + return False + except Exception: + return True # fail safe: inability to prove death must not rewrite state + if started_at is None: + return pid == os.getpid() + current = _process_start_time(pid) + return current is not None and current == started_at + + +def _prune_unlocked(conn: sqlite3.Connection) -> None: + limit = max(0, int(MAX_TERMINAL_EXECUTIONS)) + conn.execute( + """DELETE FROM executions WHERE id IN ( + SELECT id FROM executions + WHERE status IN ('completed','failed','unknown') + ORDER BY claimed_at DESC, id DESC LIMIT -1 OFFSET ? + )""", + (limit,), + ) def create_execution(job_id: str, *, source: str) -> Dict[str, Any]: - """Persist a claimed attempt before it is submitted for execution.""" + """Persist a claimed attempt before executor/provider dispatch.""" 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 + execution_id = uuid.uuid4().hex + pid = os.getpid() + with _lock, _connect() as conn: + conn.execute( + """INSERT INTO executions + (id, job_id, source, process_id, pid, process_started_at, + status, claimed_at) + VALUES (?, ?, ?, ?, ?, ?, 'claimed', ?)""", + (execution_id, str(job_id), str(source), _PROCESS_ID, pid, + _process_start_time(pid), now), + ) + row = conn.execute( + "SELECT * FROM executions WHERE id=?", (execution_id,) + ).fetchone() + return _record(row) # type: ignore[return-value] def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: - return _transition( - execution_id, - "running", - started_at=_hermes_now().isoformat(), - ) + """Transition one claimed attempt to running exactly once.""" + now = _hermes_now().isoformat() + with _lock, _connect() as conn: + cur = conn.execute( + """UPDATE executions SET status='running', started_at=? + WHERE id=? AND status='claimed'""", + (now, execution_id), + ) + if cur.rowcount != 1: + return None + return _record(conn.execute( + "SELECT * FROM executions WHERE id=?", (execution_id,) + ).fetchone()) def finish_execution( - execution_id: str, - *, - success: bool, - error: Optional[str] = None, + 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"), - ) + """Write a terminal result once; terminal attempts cannot be rewritten.""" + now = _hermes_now().isoformat() + status = "completed" if success else "failed" + detail = None if success else (str(error) if error else "unknown failure") + with _lock, _connect() as conn: + cur = conn.execute( + """UPDATE executions SET status=?, finished_at=?, error=? + WHERE id=? AND status IN ('claimed','running')""", + (status, now, detail, execution_id), + ) + if cur.rowcount != 1: + return None + _prune_unlocked(conn) + return _record(conn.execute( + "SELECT * FROM executions WHERE id=?", (execution_id,) + ).fetchone()) def recover_interrupted_executions() -> int: - """Classify prior in-flight attempts as unknown; never enqueue a retry.""" + """Mark provably abandoned attempts unknown without scheduling retries.""" 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"}: + with _lock, _connect() as conn: + rows = conn.execute( + """SELECT id, process_id, pid, process_started_at FROM executions + WHERE status IN ('claimed','running')""" + ).fetchall() + for row in rows: + if row["process_id"] == _PROCESS_ID: 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): + if _owner_is_live(int(row["pid"]), row["process_started_at"]): 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." + cur = conn.execute( + """UPDATE executions SET status='unknown', finished_at=?, error=? + WHERE id=? AND status IN ('claimed','running')""", + (now, + "Scheduler restarted after this execution's owner exited before a durable " + "terminal state; whether side effects ran is unknown.", + row["id"]), ) - changed += 1 + changed += cur.rowcount if changed: - _save_unlocked(records) + _prune_unlocked(conn) return changed -def list_executions(*, job_id: Optional[str] = None) -> List[Dict[str, Any]]: - with _ledger_lock(): - records = copy.deepcopy(_load_unlocked()) +def list_executions( + *, job_id: Optional[str] = None, limit: int = 50, + before_claimed_at: Optional[str] = None, +) -> List[Dict[str, Any]]: + """Return indexed, newest-first execution history with cursor pagination.""" + clauses: List[str] = [] + params: List[Any] = [] 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 + clauses.append("job_id=?") + params.append(str(job_id)) + if before_claimed_at is not None: + clauses.append("claimed_at < ?") + params.append(str(before_claimed_at)) + where = " WHERE " + " AND ".join(clauses) if clauses else "" + params.append(max(1, min(int(limit), 500))) + with _lock, _connect() as conn: + rows = conn.execute( + "SELECT * FROM executions" + where + + " ORDER BY claimed_at DESC, id DESC LIMIT ?", + params, + ).fetchall() + return [dict(row) for row in rows] def latest_execution(job_id: str) -> Optional[Dict[str, Any]]: - records = list_executions(job_id=job_id) - return records[0] if records else None + rows = list_executions(job_id=job_id, limit=1) + return rows[0] if rows else None + + +def latest_executions(job_ids: List[str]) -> Dict[str, Dict[str, Any]]: + """Load latest execution for many jobs in one indexed query.""" + clean = [str(job_id) for job_id in dict.fromkeys(job_ids) if job_id] + if not clean: + return {} + placeholders = ",".join("?" for _ in clean) + with _lock, _connect() as conn: + rows = conn.execute( + f"""SELECT e.* FROM executions e + WHERE e.job_id IN ({placeholders}) + AND e.id=(SELECT e2.id FROM executions e2 + WHERE e2.job_id=e.job_id + ORDER BY e2.claimed_at DESC, e2.id DESC LIMIT 1)""", + clean, + ).fetchall() + return {row["job_id"]: dict(row) for row in rows} diff --git a/cron/jobs.py b/cron/jobs.py index 46128c2e312..6c5222110ed 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -455,14 +455,6 @@ 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 @@ -1324,6 +1316,14 @@ def list_jobs(include_disabled: bool = False) -> List[Dict[str, Any]]: jobs = [_normalize_job_record(j) for j in load_jobs()] if not include_disabled: jobs = [j for j in jobs if j.get("enabled", True)] + try: + from cron.executions import latest_executions + + latest = latest_executions([job.get("id", "") for job in jobs]) + except Exception: + latest = {} + for job in jobs: + job["latest_execution"] = latest.get(job.get("id", "")) return jobs diff --git a/cron/scheduler.py b/cron/scheduler.py index 20b36f49f72..0b1395010d8 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3946,23 +3946,28 @@ def tick( try: return pool.submit(_run_and_release) - except RuntimeError as submit_err: + except Exception as submit_err: + with _running_lock: + _running_job_ids.discard(job_id) + finish_execution( + execution["id"], + success=False, + error=f"Executor dispatch failed: {submit_err}", + ) # Interpreter began finalizing between the guard above and the # submit — release the in-flight claim we just took and skip. - 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.", - ) + if isinstance(submit_err, RuntimeError) and _interpreter_shutting_down(submit_err): logger.warning( "Job '%s' not dispatched — interpreter is shutting down", job.get("name", job_id), ) return None - raise + logger.error( + "Job '%s' not dispatched: %s", + job.get("name", job_id), + submit_err, + ) + return None # Sequential pass for env-mutating (workdir) jobs. # Queued to a persistent single-thread pool so they run one at a time diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 29bcaf67947..b17d8341da3 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -82,6 +82,12 @@ class CronScheduler(ABC): Built-in: no-op (it re-reads jobs.json on every tick).""" return None + def recover_interrupted(self) -> int: + """Run profile-local attempt recovery for every provider lifecycle.""" + from cron.executions import recover_interrupted_executions + + return recover_interrupted_executions() + def fire_due(self, job_id: str, *, adapters: Any = None, loop: Any = None) -> bool: """Run a single job NOW via the shared orchestrator. Called by the inbound fire webhook when an external scheduler signals a job is due. @@ -171,11 +177,10 @@ 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() + recovered = self.recover_interrupted() if recovered: logger.warning( "Marked %d interrupted cron execution(s) unknown after restart", diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 33d07169d5b..5f945dbccec 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -769,6 +769,7 @@ _QUICK_STATE_FILES = ( ".env", "auth.json", "cron/jobs.json", + "cron/executions.db", "gateway_state.json", "channel_directory.json", "channel_aliases.json", diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index e383e2dc4b4..bf8a02fe099 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -196,6 +196,24 @@ def cron_tick(): tick(verbose=True) +def cron_runs(job_id: Optional[str] = None, limit: int = 20): + """Show indexed durable cron execution history.""" + from cron.executions import list_executions + + records = list_executions(job_id=job_id, limit=limit) + if not records: + print("No cron execution attempts recorded.") + return + for record in records: + print( + f"{record.get('id', '?')} {record.get('status', '?'):<9} " + f"job={record.get('job_id', '?')} source={record.get('source', '?')} " + f"{record.get('claimed_at', '?')}" + ) + if record.get("error"): + print(f" {record['error']}") + + def cron_status(): """Show cron execution status.""" from cron.jobs import list_jobs @@ -440,6 +458,10 @@ def cron_command(args): cron_tick() return 0 + if subcmd in {"runs", "history"}: + cron_runs(getattr(args, "job_id", None), getattr(args, "limit", 20)) + return 0 + if subcmd in {"create", "add"}: return cron_create(args) @@ -459,5 +481,5 @@ def cron_command(args): return _job_action("remove", args.job_id, "Removed") print(f"Unknown cron command: {subcmd}") - print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|tick]") + print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|runs|tick]") sys.exit(1) diff --git a/hermes_cli/subcommands/cron.py b/hermes_cli/subcommands/cron.py index c50b3401462..cce0c827eda 100644 --- a/hermes_cli/subcommands/cron.py +++ b/hermes_cli/subcommands/cron.py @@ -156,6 +156,12 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None: # cron status cron_subparsers.add_parser("status", help="Check if cron scheduler is running") + cron_runs = cron_subparsers.add_parser( + "runs", aliases=["history"], help="Show durable execution attempts" + ) + cron_runs.add_argument("job_id", nargs="?", help="Optional job ID filter") + cron_runs.add_argument("--limit", type=int, default=20, help="Rows to show (1-500)") + # cron tick (mostly for debugging) cron_tick = cron_subparsers.add_parser("tick", help="Run due jobs once and exit") add_accept_hooks_flag(cron_tick) diff --git a/plugins/cron_providers/chronos/__init__.py b/plugins/cron_providers/chronos/__init__.py index ac0e39e55cb..45e0fcf33e0 100644 --- a/plugins/cron_providers/chronos/__init__.py +++ b/plugins/cron_providers/chronos/__init__.py @@ -109,8 +109,7 @@ class ChronosCronScheduler(CronScheduler): # 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() + self.recover_interrupted() try: self.reconcile() except Exception as e: diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 54812785b7f..7d032b06a85 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import sqlite3 import subprocess import sys from pathlib import Path @@ -12,8 +13,7 @@ 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") + monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db") return executions @@ -39,6 +39,78 @@ def test_execution_transitions_are_durable(monkeypatch, tmp_path): assert persisted == [completed] +def test_terminal_execution_cannot_be_rewritten(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("immutable", source="builtin") + executions.mark_execution_running(record["id"]) + executions.finish_execution(record["id"], success=True) + + assert executions.finish_execution( + record["id"], success=False, error="late writer" + ) is None + assert executions.latest_execution("immutable")["status"] == "completed" + + +def test_retention_bounds_terminal_history_but_preserves_inflight(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + monkeypatch.setattr(executions, "MAX_TERMINAL_EXECUTIONS", 3) + inflight = executions.create_execution("live", source="builtin") + executions.mark_execution_running(inflight["id"]) + for index in range(8): + row = executions.create_execution(f"done-{index}", source="builtin") + executions.finish_execution(row["id"], success=True) + + records = executions.list_executions(limit=100) + assert len([row for row in records if row["status"] == "completed"]) == 3 + assert executions.latest_execution("live")["status"] == "running" + + +def test_corrupt_store_fails_closed_without_overwrite(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + executions.EXECUTIONS_FILE.parent.mkdir(parents=True) + executions.EXECUTIONS_FILE.write_bytes(b"not a sqlite database") + + with __import__("pytest").raises(sqlite3.DatabaseError): + executions.create_execution("new", source="builtin") + assert executions.EXECUTIONS_FILE.read_bytes() == b"not a sqlite database" + + +def test_execution_history_is_paginated(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + ids = [] + for _index in range(5): + row = executions.create_execution("paged", source="builtin") + executions.finish_execution(row["id"], success=True) + ids.append(row["id"]) + + first = executions.list_executions(job_id="paged", limit=2) + second = executions.list_executions( + job_id="paged", limit=2, before_claimed_at=first[-1]["claimed_at"] + ) + assert [row["id"] for row in first] == list(reversed(ids))[:2] + assert set(row["id"] for row in first).isdisjoint(row["id"] for row in second) + + +def test_cron_runs_cli_prints_execution_history(monkeypatch, tmp_path, capsys): + executions = _point_ledger(monkeypatch, tmp_path) + row = executions.create_execution("cli-job", source="builtin") + executions.finish_execution(row["id"], success=False, error="boom") + from hermes_cli.cron import cron_runs + + cron_runs("cli-job", limit=10) + + output = capsys.readouterr().out + assert row["id"] in output + assert "failed" in output + assert "boom" in output + + +def test_quick_backup_includes_execution_ledger(): + from hermes_cli.backup import _QUICK_STATE_FILES + + assert "cron/executions.db" in _QUICK_STATE_FILES + + def test_failed_execution_keeps_error(monkeypatch, tmp_path): executions = _point_ledger(monkeypatch, tmp_path) @@ -61,15 +133,29 @@ def test_recovery_does_not_mark_live_process_execution_unknown(monkeypatch, tmp_ 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})) + with sqlite3.connect(executions.EXECUTIONS_FILE) as conn: + conn.execute( + "UPDATE executions SET process_id=?, pid=? WHERE id=?", + ("another-import", os.getpid(), record["id"]), + ) assert executions.recover_interrupted_executions() == 0 assert executions.latest_execution("other-live")["status"] == "claimed" +def test_recovery_rejects_recycled_pid(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("recycled", source="builtin") + with sqlite3.connect(executions.EXECUTIONS_FILE) as conn: + conn.execute( + "UPDATE executions SET process_id=?, process_started_at=? WHERE id=?", + ("old-import", -1, record["id"]), + ) + + assert executions.recover_interrupted_executions() == 1 + assert executions.latest_execution("recycled")["status"] == "unknown" + + 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" @@ -121,6 +207,36 @@ def test_restart_marks_interrupted_execution_unknown_without_requeue(tmp_path): assert [r["status"] for r in records] == ["unknown"] +def test_generic_submit_failure_finishes_attempt_and_releases_guard(monkeypatch): + import cron.scheduler as scheduler + + class BrokenPool: + def submit(self, _callable): + raise ValueError("executor rejected") + + finished = [] + monkeypatch.setattr( + scheduler, "create_execution", + lambda *_args, **_kwargs: {"id": "exec-submit-fail"}, + ) + monkeypatch.setattr( + scheduler, "finish_execution", + lambda execution_id, **kwargs: finished.append((execution_id, kwargs)), + ) + monkeypatch.setattr(scheduler, "get_due_jobs", lambda: [{"id": "submit-fail"}]) + monkeypatch.setattr(scheduler, "advance_next_run", lambda _job_id: None) + monkeypatch.setattr(scheduler, "_get_parallel_pool", lambda _workers: BrokenPool()) + + assert scheduler.tick(verbose=False, sync=False) == 0 + assert finished == [ + ("exec-submit-fail", { + "success": False, + "error": "Executor dispatch failed: executor rejected", + }) + ] + assert "submit-fail" not in scheduler.get_running_job_ids() + + def test_run_one_job_records_running_then_terminal(monkeypatch): import cron.scheduler as scheduler diff --git a/tests/hermes_cli/test_cron_parser_builder.py b/tests/hermes_cli/test_cron_parser_builder.py index 16be898b1a9..653ffd5f705 100644 --- a/tests/hermes_cli/test_cron_parser_builder.py +++ b/tests/hermes_cli/test_cron_parser_builder.py @@ -25,8 +25,8 @@ def _build(): def test_cron_subactions_present(): parser = _build() - for action in ("list", "create", "edit", "pause", "resume", "run", "remove", "status", "tick"): - ns = parser.parse_args(["cron", action] if action in ("list", "status", "tick") + for action in ("list", "create", "edit", "pause", "resume", "run", "remove", "status", "runs", "tick"): + ns = parser.parse_args(["cron", action] if action in ("list", "status", "runs", "tick") else ["cron", action, "jobid"] if action in ("pause", "resume", "run", "remove", "edit") else ["cron", "create", "30m"]) assert ns.command == "cron" @@ -42,6 +42,10 @@ def test_cron_aliases(): for alias in ("rm", "delete"): ns = parser.parse_args(["cron", alias, "jid"]) assert ns.cron_command == alias + ns = parser.parse_args(["cron", "history", "jid", "--limit", "7"]) + assert ns.cron_command == "history" + assert ns.job_id == "jid" + assert ns.limit == 7 def test_cron_create_options(): diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index 3a07d5bde04..44ad20cf808 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -225,6 +225,20 @@ On each tick Hermes: A file lock at `~/.hermes/cron/.tick.lock` prevents overlapping scheduler ticks from double-running the same job batch. +### Execution history + +Hermes records each claimed cron attempt in the profile-local +`~/.hermes/cron/executions.db` before executor or provider dispatch. Attempts +move through `claimed`, `running`, and one immutable terminal state: +`completed`, `failed`, or `unknown`. After restart, Hermes marks an abandoned +attempt `unknown` only when the original PID and process-start fingerprint prove +that its owner is gone. Unknown attempts are audit records and are never +automatically rerun. + +Inspect recent attempts with `hermes cron runs [job-id] --limit 20` (alias: +`history`). Terminal history is bounded; active attempts are never pruned. The +ledger is included in quick backups. + ## Delivery options When scheduling jobs, you specify where the output goes: diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md index e543f8cfcb2..fb51b436538 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/cron.md @@ -244,7 +244,13 @@ hermes cron status 6. 投递最终响应 7. 更新运行元数据和下次调度时间 -`~/.hermes/cron/.tick.lock` 处的文件锁防止重叠的调度器 tick 重复运行同一批任务。 +`~/.hermes/cron/.tick.lock` 文件锁可防止重叠的调度器 tick 重复运行同一批任务。 + +### 执行历史 + +Hermes 会在执行器或调度提供程序分派之前，将每次已领取的 cron 尝试记录到当前 profile 的 `~/.hermes/cron/executions.db`。尝试会依次进入 `claimed`、`running`，然后进入不可变的终态：`completed`、`failed` 或 `unknown`。重启后，只有原 PID 与进程启动时间指纹能够证明所有者已经消失时，Hermes 才会将遗留尝试标记为 `unknown`。未知尝试仅用于审计，绝不会自动重跑。 + +使用 `hermes cron runs [job-id] --limit 20`（别名：`history`）查看最近的尝试。终态历史有界，活动尝试不会被清理；快速备份也包含该账本。 ## 投递选项