feat: add cron doctor health check

This commit is contained in:
joe102084
2026-06-11 03:04:35 +08:00
committed by Teknium
parent 1617ff0254
commit b028fe632e
5 changed files with 168 additions and 5 deletions
+32
View File
@@ -0,0 +1,32 @@
# Cron Doctor Spec
## Problem
Scheduled jobs can silently degrade when a script is moved, a workdir disappears,
a provider run fails, or delivery starts failing. `hermes cron list` shows some of
this inline, but there is no compact read-only health check that can be run from a
terminal, cron job, or CI-style smoke check.
## Goal
Add `hermes cron doctor` as a read-only diagnostic command that summarizes cron
job health and exits non-zero when actionable issues are found.
## Non-goals
- Do not mutate jobs or auto-repair state.
- Do not start/stop the gateway.
- Do not inspect secrets or print credentials.
## Acceptance criteria
- `hermes cron doctor` returns `0` and prints a healthy message when active jobs
have no detected issues.
- It returns `1` and prints grouped job-level issues when any active job has:
- last run failure (`last_status` not `ok`),
- last delivery failure,
- no `next_run_at` while still active,
- `no_agent` enabled without a script,
- script path missing/outside `HERMES_HOME/scripts`, or
- configured workdir path missing.
- Parser, command dispatch, and focused tests cover the new subcommand.
+99 -2
View File
@@ -9,7 +9,7 @@ import json
import re
import sys
from pathlib import Path
from typing import Iterable, List, Optional
from typing import Any, Dict, Iterable, List, Optional
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
sys.path.insert(0, str(PROJECT_ROOT))
@@ -552,6 +552,100 @@ def _print_active_jobs_summary(jobs) -> None:
print(" No active jobs")
def _scripts_dir_for_cron() -> Path:
"""Return the scripts directory used by cron jobs.
Prefer ``cron.jobs.CRON_DIR.parent`` over a fresh ``get_hermes_home()`` call
so tests and profile-aware callers that monkeypatch cron storage inspect the
same Hermes home the jobs were loaded from.
"""
from cron.jobs import CRON_DIR
return CRON_DIR.parent / "scripts"
def _script_health_issue(script: str) -> Optional[str]:
"""Return a human-readable script issue, or ``None`` when the path is OK."""
scripts_dir = _scripts_dir_for_cron().resolve()
raw = Path(script).expanduser()
path = raw.resolve() if raw.is_absolute() else (scripts_dir / raw).resolve()
try:
path.relative_to(scripts_dir)
except ValueError:
return f"script resolves outside HERMES_HOME/scripts: {script!r}"
if not path.exists():
return f"script not found: {path}"
if not path.is_file():
return f"script path is not a file: {path}"
return None
def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]:
issues: List[str] = []
last_status = str(job.get("last_status") or "").strip().lower()
if last_status and last_status != "ok":
err = str(job.get("last_error") or "unknown error").strip()
issues.append(f"last run failed: {err}")
delivery_err = str(job.get("last_delivery_error") or "").strip()
if delivery_err:
issues.append(f"last delivery failed: {delivery_err}")
if job.get("enabled", True) and job.get("state") not in {"paused", "completed"}:
if not job.get("next_run_at"):
issues.append("active job has no next_run_at")
script = str(job.get("script") or "").strip()
if job.get("no_agent") and not script:
issues.append("no-agent job has no script")
if script:
script_issue = _script_health_issue(script)
if script_issue:
issues.append(script_issue)
workdir = str(job.get("workdir") or "").strip()
if workdir and not Path(workdir).expanduser().exists():
issues.append(f"workdir not found: {workdir}")
return issues
def cron_doctor() -> int:
"""Run read-only cron health checks and return a shell-friendly status."""
from cron.jobs import list_jobs
jobs = list_jobs(include_disabled=False)
findings: List[tuple[Dict[str, Any], List[str]]] = []
for job in jobs:
issues = _cron_doctor_issues_for_job(job)
if issues:
findings.append((job, issues))
if not findings:
print(color("✓ Cron doctor found no issues", Colors.GREEN))
if jobs:
print(color(f" Checked {len(jobs)} active job(s).", Colors.DIM))
else:
print(color(" No active jobs configured.", Colors.DIM))
return 0
issue_count = sum(len(issues) for _, issues in findings)
print(color(f"Cron doctor found {issue_count} issue(s) across {len(findings)} job(s):", Colors.YELLOW))
print()
for job, issues in findings:
job_id = job.get("id", "?")
name = job.get("name", "(unnamed)")
print(f" {color(job_id, Colors.YELLOW)} {name}")
for issue in issues:
print(f" - {issue}")
print()
print(color("Next: fix the listed job config, then run `hermes cron doctor` again.", Colors.DIM))
return 1
def cron_create(args):
# The gateway-lifecycle guard lives in cron.jobs.create_job so it fires on
# every job-creation path (this CLI subcommand AND the agent's `cronjob`
@@ -839,6 +933,9 @@ def cron_command(args):
cron_status()
return 0
if subcmd == "doctor":
return cron_doctor()
if subcmd == "tick":
return cron_tick()
@@ -871,5 +968,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|runs|tick]")
print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|runs|doctor|tick]")
sys.exit(1)
+3
View File
@@ -321,6 +321,9 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
cron_notepad.add_argument("key", nargs="?", help="Notepad key (get/set/delete)")
cron_notepad.add_argument("value", nargs="?", help="Value to store (set)")
# cron doctor
cron_subparsers.add_parser("doctor", help="Check scheduled jobs for common health issues")
# 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)
+32 -1
View File
@@ -6,7 +6,7 @@ from types import SimpleNamespace
import pytest
from cron.jobs import create_job, get_job, list_jobs
from cron.jobs import create_job, get_job, list_jobs, load_jobs, save_jobs
from hermes_cli import cron as cron_cli
from hermes_cli.cron import cron_command
from hermes_cli.subcommands.cron import build_cron_parser
@@ -129,6 +129,37 @@ class TestCronCommandLifecycle:
assert jobs[0]["name"] == "Skill combo"
class TestCronDoctor:
def test_doctor_reports_cron_health_issues(self, tmp_cron_dir, capsys):
job = create_job(prompt="Daily digest", schedule="every 1h", script="missing.py")
jobs = load_jobs()
jobs[0]["last_status"] = "error"
jobs[0]["last_error"] = "Provider returned error"
jobs[0]["last_delivery_error"] = "telegram timeout"
save_jobs(jobs)
rc = cron_command(Namespace(cron_command="doctor"))
out = capsys.readouterr().out
assert rc == 1
assert "Cron doctor found 3 issue(s)" in out
assert job["id"] in out
assert "last run failed: Provider returned error" in out
assert "last delivery failed: telegram timeout" in out
assert "script not found" in out
def test_doctor_reports_healthy_jobs(self, tmp_cron_dir, capsys):
scripts_dir = tmp_cron_dir / "scripts"
scripts_dir.mkdir()
(scripts_dir / "ok.py").write_text("print('ok')\n", encoding="utf-8")
create_job(prompt="Daily digest", schedule="every 1h", script="ok.py")
rc = cron_command(Namespace(cron_command="doctor"))
out = capsys.readouterr().out
assert rc == 0
assert "✓ Cron doctor found no issues" in out
class TestGatewayNotRunningWarning:
"""`cron create` / `cron list` must warn when the gateway (and thus the
+2 -2
View File
@@ -25,8 +25,8 @@ def _build():
def test_cron_subactions_present():
parser = _build()
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")
for action in ("list", "create", "edit", "pause", "resume", "run", "remove", "status", "runs", "doctor", "tick"):
ns = parser.parse_args(["cron", action] if action in ("list", "status", "runs", "doctor", "tick")
else ["cron", action, "jobid"] if action in ("pause", "resume", "run", "remove", "edit")
else ["cron", "create", "30m"])
assert ns.command == "cron"