dd00040182
Trim of the #121921 salvage plus the rest of the class: - `_builtin_gateway_liveness` (shared by `cron list/create`, `cron status` and the `cronjob` tool via tools/cronjob_job_args.py): own-home lock/pid, else a fresh ticker heartbeat in THIS home's store. The multiplexer-record rung was redundant once the heartbeat is the evidence (served + no heartbeat was already False), and it wrongly said "dead" when the record excluded the profile or predated `served_profiles` while `hermes serve` / the Desktop backend was ticking the home in-process (#121881) or a shared gateway was firing the profile's jobs (#99631). - `cron status`: the in-process ticker is a rung beside the multiplexer rung instead of a re-indented copy of the "no gateway" block; its restart hint names the Desktop app / serve backend. - The red verdict names the inspected profile and claims only what the probe can know ("No scheduler is serving profile 'X'"): from a named profile it cannot see a gateway alive for other profiles, so "No gateway is running on this host" was a host-wide negative it could not make (#99579). `cron list` names the profile in its banner and its empty-state line, and the list/create warning names it too (idea from #99613 by @webtecnica; its `_active_profile_label` wrapper, banner rewrite and unrelated kanban/script-validation scope were not taken). WHEN L3 (ticker liveness read by a CLI command), WHERE T7 (Desktop/serve in-process ticker) and T2/T3 (`hermes -p X` against the host multiplexer). Tests: the salvaged serve-ticker test keeps the fresh->alive / stale->dead invariant; one named-profile test proves the profile is named on every verdict and that probe's own heartbeat counts when the default record excludes probe. Both red on origin/main (`_builtin_gateway_liveness()` False with a fresh heartbeat; "No scheduled jobs." without the profile). Existing assertions on the old host-wide string were moved to the new one. Co-authored-by: webtecnica <webtecnica@gmail.com>
676 lines
26 KiB
Python
676 lines
26 KiB
Python
"""Tests for hermes_cli.cron command handling."""
|
|
|
|
import argparse
|
|
import time
|
|
from argparse import Namespace
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from cron.jobs import create_job, get_job, list_jobs, load_jobs, pause_job, 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
|
|
|
|
|
|
@pytest.fixture()
|
|
def tmp_cron_dir(tmp_path, monkeypatch):
|
|
monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron")
|
|
monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json")
|
|
monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output")
|
|
return tmp_path
|
|
|
|
|
|
class TestCronCommandLifecycle:
|
|
|
|
def test_edit_persists_user_owned_inference_pins(self, tmp_cron_dir, capsys):
|
|
job = create_job(prompt="Daily report", schedule="every 1h")
|
|
parser = argparse.ArgumentParser(prog="hermes")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
build_cron_parser(subparsers, cmd_cron=cron_command)
|
|
|
|
args = parser.parse_args(
|
|
[
|
|
"cron",
|
|
"edit",
|
|
job["id"],
|
|
"--model",
|
|
"new-model",
|
|
"--provider",
|
|
"nous",
|
|
]
|
|
)
|
|
cron_command(args)
|
|
|
|
updated = get_job(job["id"])
|
|
assert updated["model"] == "new-model"
|
|
assert updated["provider"] == "nous"
|
|
assert "Updated job" in capsys.readouterr().out
|
|
|
|
def test_edit_can_replace_and_clear_skills(self, tmp_cron_dir, capsys):
|
|
job = create_job(
|
|
prompt="Combine skill outputs",
|
|
schedule="every 1h",
|
|
skill="blogwatcher",
|
|
)
|
|
|
|
cron_command(
|
|
Namespace(
|
|
cron_command="edit",
|
|
job_id=job["id"],
|
|
schedule="every 2h",
|
|
prompt="Revised prompt",
|
|
name="Edited Job",
|
|
deliver=None,
|
|
repeat=None,
|
|
skill=None,
|
|
skills=["maps", "blogwatcher"],
|
|
clear_skills=False,
|
|
add_skills=None,
|
|
remove_skills=None,
|
|
script=None,
|
|
workdir=None,
|
|
no_agent=None,
|
|
)
|
|
)
|
|
updated = get_job(job["id"])
|
|
assert updated["skills"] == ["maps", "blogwatcher"]
|
|
assert updated["name"] == "Edited Job"
|
|
assert updated["prompt"] == "Revised prompt"
|
|
assert updated["schedule_display"] == "every 120m"
|
|
|
|
cron_command(
|
|
Namespace(
|
|
cron_command="edit",
|
|
job_id=job["id"],
|
|
schedule=None,
|
|
prompt=None,
|
|
name=None,
|
|
deliver=None,
|
|
repeat=None,
|
|
skill=None,
|
|
skills=None,
|
|
clear_skills=True,
|
|
add_skills=None,
|
|
remove_skills=None,
|
|
script=None,
|
|
workdir=None,
|
|
no_agent=None,
|
|
)
|
|
)
|
|
cleared = get_job(job["id"])
|
|
assert cleared["skills"] == []
|
|
assert cleared["skill"] is None
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Updated job" in out
|
|
|
|
def test_create_with_multiple_skills(self, tmp_cron_dir, capsys):
|
|
cron_command(
|
|
Namespace(
|
|
cron_command="create",
|
|
schedule="every 1h",
|
|
prompt="Use both skills",
|
|
name="Skill combo",
|
|
deliver=None,
|
|
repeat=None,
|
|
skill=None,
|
|
skills=["blogwatcher", "maps"],
|
|
script=None,
|
|
workdir=None,
|
|
no_agent=False,
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "Created job" in out
|
|
|
|
jobs = list_jobs()
|
|
assert len(jobs) == 1
|
|
assert jobs[0]["skills"] == ["blogwatcher", "maps"]
|
|
assert jobs[0]["name"] == "Skill combo"
|
|
|
|
|
|
class TestUnverifiedDeliveryVisibility:
|
|
"""An evidence-free live-adapter ack (Slack/Matrix/Mattermost bare
|
|
``SendResult(success=True)``) is accepted as delivered, but the UNVERIFIED
|
|
state must be visible in ``hermes cron list`` and ``hermes cron doctor``,
|
|
not only in a WARNING log line."""
|
|
|
|
def _seed(self):
|
|
job = create_job(prompt="Nightly brief", schedule="every 1h", deliver="slack:C0123456")
|
|
jobs = load_jobs()
|
|
jobs[0]["last_status"] = "ok"
|
|
jobs[0]["last_delivery_unverified"] = ["slack:C0123456"]
|
|
save_jobs(jobs)
|
|
return job
|
|
|
|
def test_list_shows_unverified_delivery(self, tmp_cron_dir, capsys):
|
|
job = self._seed()
|
|
cron_command(Namespace(cron_command="list", all=True, json=False))
|
|
out = capsys.readouterr().out
|
|
assert job["id"] in out
|
|
assert "Delivery UNVERIFIED" in out
|
|
assert "slack:C0123456" in out
|
|
assert "without message_id/raw_response" in out
|
|
|
|
def test_list_is_quiet_when_delivery_was_verified(self, tmp_cron_dir, capsys):
|
|
create_job(prompt="Nightly brief", schedule="every 1h", deliver="slack:C0123456")
|
|
cron_command(Namespace(cron_command="list", all=True, json=False))
|
|
assert "UNVERIFIED" not in capsys.readouterr().out
|
|
|
|
def test_doctor_reports_unverified_delivery(self, tmp_cron_dir, capsys):
|
|
job = self._seed()
|
|
rc = cron_command(Namespace(cron_command="doctor"))
|
|
out = capsys.readouterr().out
|
|
assert rc == 1
|
|
assert job["id"] in out
|
|
assert "last delivery unverified" in out
|
|
assert "slack:C0123456" in out
|
|
|
|
|
|
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 "was not delivered (telegram timeout)" in out
|
|
assert "hermes cron edit" 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
|
|
|
|
def test_doctor_reports_delivery_failure_once(self, tmp_cron_dir, capsys):
|
|
"""A delivery_failed run is a delivery issue, not a failed agent run.
|
|
|
|
The agent succeeded (last_error is None), so the generic last-run-failed
|
|
line would only ever say "unknown error" — double-reporting the same
|
|
incident (#83993).
|
|
"""
|
|
create_job(prompt="Daily digest", schedule="every 1h")
|
|
jobs = load_jobs()
|
|
jobs[0]["last_status"] = "delivery_failed"
|
|
jobs[0]["last_error"] = None
|
|
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 "was not delivered (telegram timeout)" in out
|
|
assert "hermes cron edit" in out
|
|
assert "last run failed" not in out
|
|
assert "unknown error" not in out
|
|
|
|
def test_doctor_flags_overdue_next_run(self, tmp_cron_dir, capsys):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
create_job(prompt="Hourly ping", schedule="every 1h")
|
|
jobs = load_jobs()
|
|
stale = (datetime.now(timezone.utc) - timedelta(hours=3)).isoformat()
|
|
jobs[0]["next_run_at"] = stale
|
|
save_jobs(jobs)
|
|
|
|
rc = cron_command(Namespace(cron_command="doctor"))
|
|
|
|
out = capsys.readouterr().out
|
|
assert rc == 1
|
|
assert "overdue" in out
|
|
assert "not firing" in out
|
|
|
|
def test_doctor_tolerates_slightly_late_next_run(self, tmp_cron_dir, capsys):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
create_job(prompt="Hourly ping", schedule="every 1h")
|
|
jobs = load_jobs()
|
|
# 5 minutes late is within the ticker grace window — healthy.
|
|
recent = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat()
|
|
jobs[0]["next_run_at"] = recent
|
|
save_jobs(jobs)
|
|
|
|
rc = cron_command(Namespace(cron_command="doctor"))
|
|
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "✓ Cron doctor found no issues" in out
|
|
|
|
|
|
class TestCronListStatusRendering:
|
|
"""`cron list` must never paint an undelivered run as a success (#83993)."""
|
|
|
|
def test_default_list_includes_paused_jobs(self, tmp_cron_dir, capsys, monkeypatch):
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [1])
|
|
job = create_job(prompt="Paused digest", schedule="every 1h")
|
|
pause_job(job["id"])
|
|
|
|
cron_command(Namespace(cron_command="list", all=False))
|
|
|
|
out = capsys.readouterr().out
|
|
assert job["id"] in out
|
|
assert "[paused]" in out
|
|
assert "No scheduled jobs" not in out
|
|
|
|
def test_delivery_failed_is_not_green_ok(self, tmp_cron_dir, capsys, monkeypatch):
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [1])
|
|
# capsys is not a tty, so force colors on to check the paint itself.
|
|
monkeypatch.setattr("hermes_cli.colors.should_use_color", lambda: True)
|
|
create_job(prompt="Daily digest", schedule="every 1h")
|
|
jobs = load_jobs()
|
|
jobs[0]["last_run_at"] = "2026-09-01T09:00:00+00:00"
|
|
jobs[0]["last_status"] = "delivery_failed"
|
|
jobs[0]["last_error"] = None
|
|
jobs[0]["last_delivery_error"] = "telegram timeout"
|
|
save_jobs(jobs)
|
|
|
|
cron_command(Namespace(cron_command="list", all=True))
|
|
|
|
out = capsys.readouterr().out
|
|
last_run_line = next(l for l in out.splitlines() if "Last run:" in l)
|
|
assert "was not delivered" in last_run_line
|
|
assert "telegram timeout" in last_run_line, (
|
|
"the delivery detail lives in last_delivery_error, not last_error"
|
|
)
|
|
assert cron_cli.Colors.GREEN not in last_run_line
|
|
|
|
def test_ok_run_still_green(self, tmp_cron_dir, capsys, monkeypatch):
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [1])
|
|
monkeypatch.setattr("hermes_cli.colors.should_use_color", lambda: True)
|
|
create_job(prompt="Daily digest", schedule="every 1h")
|
|
jobs = load_jobs()
|
|
jobs[0]["last_run_at"] = "2026-09-01T09:00:00+00:00"
|
|
jobs[0]["last_status"] = "ok"
|
|
save_jobs(jobs)
|
|
|
|
cron_command(Namespace(cron_command="list", all=True))
|
|
|
|
out = capsys.readouterr().out
|
|
last_run_line = next(l for l in out.splitlines() if "Last run:" in l)
|
|
assert f"{cron_cli.Colors.GREEN}ok" in last_run_line
|
|
assert "not delivered" not in last_run_line
|
|
|
|
|
|
class TestGatewayNotRunningWarning:
|
|
"""`cron create` / `cron list` must warn when the gateway (and thus the
|
|
cron ticker) isn't running, since jobs only fire inside the gateway.
|
|
Regression guard for #51038 — the most common cron 'jobs never fired'
|
|
report was simply a gateway that was never started.
|
|
"""
|
|
|
|
|
|
def test_list_warns_when_gateway_absent(self, tmp_cron_dir, capsys, monkeypatch):
|
|
create_job(prompt="Daily report", schedule="0 11 * * *")
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [])
|
|
cron_command(Namespace(cron_command="list", all=True))
|
|
out = capsys.readouterr().out
|
|
assert "Scheduler is not ready" in out
|
|
|
|
|
|
class TestExternalCronProviderStatus:
|
|
"""With an external cron provider (e.g. Chronos), jobs fire via a
|
|
NAS-mediated webhook, NOT the in-process ticker. The ticker-heartbeat /
|
|
gateway-process heuristics are meaningless there, so neither
|
|
`cron status` nor the create/list warning must claim the gateway being
|
|
absent means jobs won't fire — that was a false-negative on every healthy
|
|
Chronos instance (the heartbeat is intentionally never written).
|
|
"""
|
|
|
|
def test_status_reports_provider_not_ticker_for_chronos(
|
|
self, tmp_cron_dir, capsys, monkeypatch
|
|
):
|
|
create_job(prompt="Ping", schedule="every 2m")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.cron._active_cron_provider_name", lambda: "chronos"
|
|
)
|
|
# Even with NO gateway process and NO ticker heartbeat, Chronos status
|
|
# must NOT report a stall / "not firing".
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [])
|
|
cron_command(Namespace(cron_command="status"))
|
|
out = capsys.readouterr().out
|
|
assert "chronos" in out
|
|
assert "managed scheduler" in out
|
|
assert "not firing" not in out.lower()
|
|
assert "STALLED" not in out
|
|
assert "No scheduler is serving profile" not in out
|
|
# Still surfaces the active-job summary.
|
|
assert "active job(s)" in out
|
|
|
|
|
|
def test_create_silent_for_chronos_even_without_gateway(
|
|
self, tmp_cron_dir, capsys, monkeypatch
|
|
):
|
|
# The create-time "gateway not running" nag is a ticker-only concern;
|
|
# an external provider doesn't depend on a live in-process ticker.
|
|
monkeypatch.setattr(
|
|
"hermes_cli.cron._active_cron_provider_name", lambda: "chronos"
|
|
)
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [])
|
|
cron_command(
|
|
Namespace(
|
|
cron_command="create",
|
|
schedule="every 2m",
|
|
prompt="Ping",
|
|
name="Ping",
|
|
deliver=None,
|
|
repeat=None,
|
|
skill=None,
|
|
skills=None,
|
|
script=None,
|
|
workdir=None,
|
|
no_agent=False,
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "Created job" in out
|
|
assert "Scheduler is not ready" not in out
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_cron_create_failure_returns_nonzero(monkeypatch, capsys):
|
|
monkeypatch.setattr(cron_cli, "_cron_api", lambda **kwargs: {"success": False, "error": "boom"})
|
|
|
|
args = SimpleNamespace(
|
|
schedule="every day",
|
|
prompt="refresh docs",
|
|
name=None,
|
|
deliver=None,
|
|
repeat=None,
|
|
skill=None,
|
|
skills=None,
|
|
script=None,
|
|
workdir=None,
|
|
no_agent=False,
|
|
)
|
|
|
|
rc = cron_cli.cron_create(args)
|
|
|
|
out = capsys.readouterr().out
|
|
assert rc == 1
|
|
assert "Failed to create job: boom" in out
|
|
|
|
|
|
class TestCronRunBackgroundDispatch:
|
|
"""`hermes cron run` must not report 'failed' when the run was dispatched
|
|
to the background delegation worker.
|
|
|
|
The CLI process inherits the gateway/desktop session env, so a manual run
|
|
can be dispatched to the daemon instead of executing inline. Such
|
|
responses carry execution_mode='background' / delegation_id and the job
|
|
keeps running after the CLI exits — a terminal success/failure verdict
|
|
would be a lie (#83340). The CLI must report the background dispatch
|
|
instead, and leave synchronous runs unchanged.
|
|
"""
|
|
|
|
def _run_cmd(self, capsys):
|
|
rc = cron_command(Namespace(cron_command="run", job_id="job-1"))
|
|
return rc, capsys.readouterr().out
|
|
|
|
def test_background_dispatch_with_delegation_id_does_not_report_failed(
|
|
self, monkeypatch, capsys
|
|
):
|
|
monkeypatch.setattr(
|
|
cron_cli,
|
|
"_cron_api",
|
|
lambda **kwargs: {
|
|
"success": True,
|
|
"job": {
|
|
"id": "job-1",
|
|
"name": "Watchdog",
|
|
"execution_mode": "background",
|
|
"delegation_id": "del-abc123",
|
|
# No execution_success — the inline verdict must not apply.
|
|
"executed": True,
|
|
},
|
|
},
|
|
)
|
|
|
|
rc, out = self._run_cmd(capsys)
|
|
|
|
assert rc == 0
|
|
assert "Running in background (delegation del-abc123)." in out
|
|
assert "failed" not in out.lower()
|
|
assert "Ran now" not in out
|
|
|
|
def test_background_dispatch_without_delegation_id(self, monkeypatch, capsys):
|
|
monkeypatch.setattr(
|
|
cron_cli,
|
|
"_cron_api",
|
|
lambda **kwargs: {
|
|
"success": True,
|
|
"job": {
|
|
"id": "job-1",
|
|
"name": "Watchdog",
|
|
"execution_mode": "background",
|
|
},
|
|
},
|
|
)
|
|
|
|
rc, out = self._run_cmd(capsys)
|
|
|
|
assert rc == 0
|
|
assert "Running in background." in out
|
|
assert "failed" not in out.lower()
|
|
|
|
def test_sync_run_success_unchanged(self, monkeypatch, capsys):
|
|
monkeypatch.setattr(
|
|
cron_cli,
|
|
"_cron_api",
|
|
lambda **kwargs: {
|
|
"success": True,
|
|
"job": {
|
|
"id": "job-1",
|
|
"name": "Watchdog",
|
|
"executed": True,
|
|
"execution_success": True,
|
|
},
|
|
},
|
|
)
|
|
|
|
rc, out = self._run_cmd(capsys)
|
|
|
|
assert rc == 0
|
|
assert "Ran now: succeeded." in out
|
|
|
|
def test_sync_run_failure_still_reported(self, monkeypatch, capsys):
|
|
# A genuine synchronous failure must keep reporting 'failed' — only
|
|
# background-dispatched runs are exempt from the terminal verdict.
|
|
monkeypatch.setattr(
|
|
cron_cli,
|
|
"_cron_api",
|
|
lambda **kwargs: {
|
|
"success": True,
|
|
"job": {
|
|
"id": "job-1",
|
|
"name": "Watchdog",
|
|
"executed": True,
|
|
"execution_success": False,
|
|
},
|
|
},
|
|
)
|
|
|
|
rc, out = self._run_cmd(capsys)
|
|
|
|
assert rc == 0
|
|
assert "Ran now: failed." in out
|
|
|
|
def test_delegation_id_alone_counts_as_background(self, monkeypatch, capsys):
|
|
# Some dispatchers may not set execution_mode but always return the
|
|
# delegation_id — either marker alone must suppress the verdict.
|
|
monkeypatch.setattr(
|
|
cron_cli,
|
|
"_cron_api",
|
|
lambda **kwargs: {
|
|
"success": True,
|
|
"job": {"id": "job-1", "name": "Watchdog", "delegation_id": "del-xyz"},
|
|
},
|
|
)
|
|
|
|
rc, out = self._run_cmd(capsys)
|
|
|
|
assert rc == 0
|
|
assert "Running in background (delegation del-xyz)." in out
|
|
assert "failed" not in out.lower()
|
|
|
|
|
|
class TestSlashCronListLastStatus:
|
|
"""The in-chat ``/cron list`` (cli_commands_mixin) renders every
|
|
``last_status`` literal explicitly — ``delivery_failed`` names the delivery
|
|
reason (last_error is None for those runs) instead of printing the bare
|
|
literal next to a run that looks otherwise fine."""
|
|
|
|
def _run_list(self, tmp_cron_dir, capsys):
|
|
from hermes_cli.cli_commands_mixin import CLICommandsMixin
|
|
|
|
class _Host(CLICommandsMixin):
|
|
pass
|
|
|
|
_Host()._handle_cron_command("/cron list --all")
|
|
return capsys.readouterr().out
|
|
|
|
def test_delivery_failed_names_the_delivery_error(self, tmp_cron_dir, capsys):
|
|
create_job(prompt="Nightly brief", schedule="every 1h", deliver="telegram:1")
|
|
jobs = load_jobs()
|
|
jobs[0]["last_run_at"] = "2026-09-01T07:00:00+00:00"
|
|
jobs[0]["last_status"] = "delivery_failed"
|
|
jobs[0]["last_error"] = None
|
|
jobs[0]["last_delivery_error"] = "telegram: 502 Bad Gateway"
|
|
save_jobs(jobs)
|
|
|
|
out = self._run_list(tmp_cron_dir, capsys)
|
|
assert "Last run: 2026-09-01T07:00:00+00:00 (delivery_failed: telegram: 502 Bad Gateway)" in out
|
|
|
|
|
|
|
|
class TestStatusSurfacesDeadScheduler:
|
|
"""#114309 — with the ticker dead and a job's next_run_at stranded in the past, `cron
|
|
status` / `cron list` must not present the stale timestamp as an upcoming "Next run":
|
|
flag it as overdue and say when the scheduler last ticked."""
|
|
|
|
def _dead_gateway(self, monkeypatch, lock_dir):
|
|
# No gateway owns the HOST role either: point the rendezvous dir at an empty scratch dir
|
|
# so an unrelated host record can never make this profile look served. (Superseded by the
|
|
# tests/conftest.py hook in #118097 once that lands.)
|
|
lock_dir.mkdir(exist_ok=True)
|
|
monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(lock_dir))
|
|
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [])
|
|
monkeypatch.setattr(
|
|
"hermes_cli.gateway.named_profile_served_by_running_multiplexer", lambda: None
|
|
)
|
|
monkeypatch.setattr("gateway.status.is_gateway_runtime_lock_active", lambda: False)
|
|
|
|
def _park_next_run(self, job_id, when):
|
|
jobs = load_jobs()
|
|
jobs[[j["id"] for j in jobs].index(job_id)]["next_run_at"] = when.isoformat()
|
|
save_jobs(jobs)
|
|
|
|
def test_overdue_next_run_and_stale_heartbeat_are_loud(
|
|
self, tmp_cron_dir, capsys, monkeypatch
|
|
):
|
|
job = create_job(prompt="Hourly", schedule="every 60m")
|
|
self._dead_gateway(monkeypatch, tmp_cron_dir / "locks")
|
|
self._park_next_run(job["id"], datetime.now(timezone.utc) - timedelta(hours=7))
|
|
(tmp_cron_dir / "cron" / "ticker_heartbeat").write_text(str(time.time() - 25 * 3600))
|
|
|
|
cron_command(Namespace(cron_command="status"))
|
|
status_out = capsys.readouterr().out
|
|
cron_command(Namespace(cron_command="list", all=False, json=False))
|
|
list_out = capsys.readouterr().out
|
|
|
|
assert "No scheduler is serving profile" in status_out
|
|
assert "Scheduler last ticked" in status_out
|
|
assert "OVERDUE" in status_out and "7h ago" in status_out
|
|
# The stale timestamp must no longer read as an upcoming run on either surface.
|
|
assert "Next run:" not in status_out
|
|
assert "Overdue:" in list_out and "Next run:" not in list_out
|
|
|
|
def test_overdue_within_doctor_grace_stays_plain(self, tmp_cron_dir, capsys, monkeypatch):
|
|
# status shares `cron doctor`'s 15-minute grace (_OVERDUE_GRACE_SECONDS): a job only
|
|
# a few minutes behind the ticker's own cadence is not an outage yet, and status must
|
|
# not flash OVERDUE while doctor calls the same job healthy.
|
|
job = create_job(prompt="Hourly", schedule="every 60m")
|
|
self._dead_gateway(monkeypatch, tmp_cron_dir / "locks")
|
|
self._park_next_run(job["id"], datetime.now(timezone.utc) - timedelta(minutes=5))
|
|
|
|
cron_command(Namespace(cron_command="status"))
|
|
status_out = capsys.readouterr().out
|
|
cron_command(Namespace(cron_command="list", all=False, json=False))
|
|
list_out = capsys.readouterr().out
|
|
|
|
assert "Next run:" in status_out and "OVERDUE" not in status_out
|
|
assert "Scheduler last ticked" not in status_out # no heartbeat file → nothing to date
|
|
assert "Next run:" in list_out and "Overdue:" not in list_out
|
|
|
|
def test_slash_cron_and_list_flag_overdue_but_not_paused(self, tmp_cron_dir, capsys):
|
|
# The in-chat `/cron` overview and `/cron list` (classic CLI + Ink TUI forward to the
|
|
# same handler) read the same rows; a 7h-past stamp must not read as an upcoming run,
|
|
# while a paused job keeps its plain label — pausing is why it did not fire.
|
|
from hermes_cli.cli_commands_mixin import CLICommandsMixin
|
|
|
|
class _Host(CLICommandsMixin):
|
|
pass
|
|
|
|
stale = create_job(prompt="Hourly", schedule="every 60m")
|
|
paused = create_job(prompt="Parked", schedule="every 60m")
|
|
when = datetime.now(timezone.utc) - timedelta(hours=7)
|
|
self._park_next_run(stale["id"], when)
|
|
self._park_next_run(paused["id"], when)
|
|
jobs = load_jobs()
|
|
jobs[[j["id"] for j in jobs].index(paused["id"])]["enabled"] = False
|
|
save_jobs(jobs)
|
|
|
|
_Host()._handle_cron_command("/cron")
|
|
overview_out = capsys.readouterr().out
|
|
_Host()._handle_cron_command("/cron list --all")
|
|
list_out = capsys.readouterr().out
|
|
|
|
for out in (overview_out, list_out):
|
|
assert out.count("Overdue:") == 1 and "7h ago" in out
|
|
assert "Next" not in overview_out # the overview lists enabled jobs only
|
|
assert list_out.count("Next run:") == 1 # only the paused job's stamp stays plain
|
|
assert list_out.index("Overdue:") < list_out.index("Next run:")
|
|
|
|
|
|
class TestSlashCronRunSkipped:
|
|
"""``/cron run`` on a job whose claim is refused (paused here; a live claim held by another
|
|
run is the same shape) must print the refusal, never ``Triggered … next scheduler tick``."""
|
|
|
|
def test_refused_run_prints_reason_not_triggered(self, tmp_cron_dir, capsys):
|
|
from hermes_cli.cli_commands_mixin import CLICommandsMixin
|
|
|
|
class _Host(CLICommandsMixin):
|
|
pass
|
|
|
|
job = create_job(prompt="Nightly brief", schedule="every 1h", deliver="local")
|
|
jobs = load_jobs()
|
|
jobs[0]["enabled"] = False
|
|
save_jobs(jobs)
|
|
|
|
_Host()._handle_cron_command(f"/cron run {job['id']}")
|
|
out = capsys.readouterr().out
|
|
assert "Job is paused/disabled; resume it before running." in out
|
|
assert "Triggered" not in out and "next scheduler tick" not in out
|