Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions contributors/emails/seraphine@Seraphines-Mac-Studio.local
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Bartok9
# Seraphine Mac Studio local email on Bartok9 PR tips (per-PR attribution; Teknium/Daniel 2026-08-01)
104 changes: 104 additions & 0 deletions tests/tools/test_cronjob_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,13 @@

import json
import pytest
from unittest.mock import patch

from tools.cronjob_tools import (
_scan_cron_prompt,
check_cronjob_requirements,
cronjob,
_check_cron_ticker_warning,
)


Expand Down Expand Up @@ -246,6 +248,59 @@ def test_create_and_list(self):
assert listing["jobs"][0]["name"] == "Server Check"
assert listing["jobs"][0]["state"] == "scheduled"

def test_create_warns_when_gateway_not_running(self):
"""Direct create-response regression (keep_open on #51790): builtin
scheduler + no gateway must surface the ticker warning on create.message.
"""
with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
side_effect=Exception("no external scheduler configured"),
),
patch("gateway.status.get_running_pid", return_value=None),
):
created = json.loads(
cronjob(action="create", prompt="Check", schedule="every 1h")
)

assert created["success"] is True
assert "gateway is not running" in created["message"].lower(), created["message"]
assert "will NOT fire automatically" in created["message"]

def test_create_does_not_warn_when_gateway_running(self):
"""Warning must not appear when get_running_pid returns a live pid."""
with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
side_effect=Exception("no external scheduler configured"),
),
patch("gateway.status.get_running_pid", return_value=12345),
):
created = json.loads(
cronjob(action="create", prompt="Check", schedule="every 1h")
)

assert created["success"] is True
assert "gateway is not running" not in created["message"].lower(), created["message"]

def test_create_does_not_warn_for_non_builtin_scheduler(self):
"""External scheduler (Chronos) + no gateway PID must stay silent."""
from types import SimpleNamespace

with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
return_value=SimpleNamespace(name="chronos"),
),
patch("gateway.status.get_running_pid", return_value=None),
):
created = json.loads(
cronjob(action="create", prompt="Check", schedule="every 1h")
)

assert created["success"] is True
assert "gateway is not running" not in created["message"].lower(), created["message"]

def test_list_handles_partial_legacy_job_records(self):
from cron.jobs import save_jobs

Expand Down Expand Up @@ -615,3 +670,52 @@ def test_benign_text_mentioning_key_types_allowed(self):
assert _scan_cron_prompt(
"generate a keypair and explain id_rsa vs id_ed25519"
) == ""

class TestCronTickerWarning:
"""Unit coverage for _check_cron_ticker_warning (helper + provider gate)."""

def test_builtin_no_gateway_warns(self):
with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
side_effect=Exception("no external scheduler configured"),
),
patch("gateway.status.get_running_pid", return_value=None),
):
msg = _check_cron_ticker_warning()
assert msg is not None
assert "gateway is not running" in msg.lower()
assert "hermes cron tick" in msg

def test_builtin_gateway_running_silent(self):
with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
side_effect=Exception("no external scheduler configured"),
),
patch("gateway.status.get_running_pid", return_value=12345),
):
assert _check_cron_ticker_warning() is None

def test_non_builtin_no_gateway_silent(self):
from types import SimpleNamespace

with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
return_value=SimpleNamespace(name="chronos"),
),
patch("gateway.status.get_running_pid", return_value=None),
):
assert _check_cron_ticker_warning() is None

def test_status_probe_error_is_best_effort(self):
with (
patch(
"cron.scheduler_provider.resolve_cron_scheduler",
side_effect=Exception("no external scheduler configured"),
),
patch("gateway.status.get_running_pid", side_effect=RuntimeError("boom")),
):
assert _check_cron_ticker_warning() is None

43 changes: 43 additions & 0 deletions tools/cronjob_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,46 @@ def _origin_from_env() -> Optional[Dict[str, str]]:
return None


def _check_cron_ticker_warning() -> Optional[str]:
"""Return a warning when the gateway (and thus the cron ticker) is not running.

Cron jobs only fire automatically while the gateway is up; in CLI-only mode
nothing advances ``next_run_at`` until the user runs ``hermes cron tick``.
Surfacing this at create time stops the silent "I scheduled it but it never
ran" trap. Best-effort: any failure to determine gateway status returns
``None`` (no warning) rather than blocking job creation.

Only the built-in ticker depends on a live gateway process. An external
provider (e.g. Chronos) fires jobs via its own managed scheduler while the
gateway is scaled to zero, so a missing gateway PID does NOT mean those jobs
won't fire — warning there would be a false alarm. Mirror the CLI contract
(``hermes_cli/cron.py``) and stay silent for any non-builtin scheduler.
"""
try:
try:
from cron.scheduler_provider import resolve_cron_scheduler

scheduler_name = resolve_cron_scheduler().name or "builtin"
except Exception:
# Fall back to the historical ticker-based check on any failure.
scheduler_name = "builtin"

if scheduler_name != "builtin":
return None

from gateway.status import get_running_pid

if get_running_pid() is None:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This must first confirm that the resolved scheduler is builtin. Current main deliberately permits Chronos to fire jobs with no running gateway PID; see hermes_cli/cron.py:66-89 and its Chronos regression test.

return (
"⚠️ The gateway is not running — this cron job will NOT fire "
"automatically. Start it with 'hermes gateway run', or trigger "
"jobs manually with 'hermes cron tick'."
)
except Exception:
pass
return None


def _local_delivery_notice(job: Dict[str, Any], user_deliver: Optional[str]) -> Optional[str]:
"""Return an informational notice when a created job won't deliver anywhere.

Expand Down Expand Up @@ -726,6 +766,9 @@ def cronjob(
_local_notice = _local_delivery_notice(job, _normalize_deliver_param(deliver))
if _local_notice:
_create_message = f"{_create_message} {_local_notice}"
_ticker_warning = _check_cron_ticker_warning()
if _ticker_warning:
_create_message = f"{_create_message} {_ticker_warning}"
return json.dumps(
{
"success": True,
Expand Down
Loading