feat(cron): add truthful execution ledger

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

cron/jobs.py

@@ -350,6 +350,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
 
 

cron/scheduler.py

@@ -238,6 +238,7 @@ _LEGACY_HOME_TARGET_ENV_VARS = {
 }
 
 from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch
+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
@@ -3361,6 +3362,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
@@ -3374,8 +3378,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
@@ -3482,12 +3495,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
 
 
@@ -3629,9 +3644,13 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
                     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:
@@ -3646,6 +3665,11 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
                 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),

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:
@@ -167,9 +169,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()

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

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:

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"