mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-22 16:25:58 +00:00
321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""Durable cron execution-ledger behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
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.db")
|
|
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_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)
|
|
|
|
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")
|
|
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"
|
|
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_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
|
|
|
|
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"
|