mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
fix(cron): a headless tick holds agent jobs while the gateway is starting or unreachable
`hermes cron tick` from a crontab held its agent jobs only when discovery said `absent`. With
the gateway `starting` (lock held, no socket yet — every restart window), `draining`, or
`inaccessible` (control socket accepts but never answers, or no gateway serves this profile),
the job fell through to run_canonical_job -> connect_gateway -> ensure_gateway_runtime, waited
its full 30 s deadline, and was then booked as a FAILED run ("GatewayClientError: Gateway
starting: deadline") with next_run_at advanced — the slot consumed by a restart, not by the
job. `-p alt cron tick` with no gateway serving alt booked "RuntimeError: Connection error."
the same way.
The gate now holds on every outcome but `ready`, through the same skip path as `absent` (no
run row, due instant kept, one warning per tick, an actionable last_fire_error that names the
observed state), and probes with a 5 s budget so a tick decides well inside its crontab minute.
This commit is contained in:
1 parent
b36858fd9c
commit
6fda3a044a
2 files changed
+90
-10
No files matched your search
@@ -7,6 +7,11 @@ would leave a stray gateway behind. Headless surfaces refuse rather than spawn (
|
||||
approvals), so a headless tick skips its agent jobs while the gateway is down and lets them fire
|
||||
on the next tick once it is up — no run is recorded and the due instant does not move. The
|
||||
in-process ticker never reaches this gate: it runs inside the gateway by definition.
|
||||
|
||||
"Down" is every outcome but ``ready``: a gateway still ``starting`` (or draining, or one whose
|
||||
control socket does not answer) would otherwise send the tick into ``ensure_gateway_runtime``'s
|
||||
30 s wait and book the deadline as a FAILED run, consuming the slot — the crontab minute is the
|
||||
budget, and a slot lost to a restart window is a missed fire, not a job failure.
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
@@ -17,17 +22,27 @@ GATEWAY_DOWN_HINT = (
|
||||
"the gateway is not running, so agent jobs cannot fire; they run at the next tick once it is "
|
||||
"up. Start it with `hermes gateway start` (or `hermes gateway install` for a service)."
|
||||
)
|
||||
# A crontab tick has a minute; a gateway still starting answers within a few seconds or not at all.
|
||||
_HEADLESS_PROBE_TIMEOUT_S = 5.0
|
||||
|
||||
|
||||
def _gateway_absent(home) -> bool:
|
||||
"""True in exactly the state where ``ensure_gateway_runtime`` would spawn a daemon. Every
|
||||
other state (starting, ready, draining, incompatible, ...) is left to ``connect_gateway``,
|
||||
which awaits or reports it without spawning."""
|
||||
def _gateway_unavailable(home) -> str | None:
|
||||
"""Why a headless tick must hold its agent jobs, or None when a ready owner serves *home*.
|
||||
Everything short of ``ready`` — absent (``ensure`` would spawn), starting/draining (``ensure``
|
||||
would wait a whole deadline then fail the run), inaccessible/incompatible (no owner answers) —
|
||||
is held, never booked as a run."""
|
||||
from hermes_cli.gateway_runtime import discover_gateway_endpoint
|
||||
|
||||
if os.environ.get("HERMES_TUI_GATEWAY_URL", "").strip():
|
||||
return False # explicit remote gateway: connect_gateway never spawns for it
|
||||
return discover_gateway_endpoint(home).state == "absent"
|
||||
return None # explicit remote gateway: connect_gateway never spawns for it
|
||||
observed = discover_gateway_endpoint(home, timeout=_HEADLESS_PROBE_TIMEOUT_S)
|
||||
if observed.state == "ready":
|
||||
return None
|
||||
if observed.state == "absent":
|
||||
return GATEWAY_DOWN_HINT
|
||||
reason = observed.state + (f" ({observed.reason_code})" if observed.reason_code else "")
|
||||
return (f"the gateway is {reason}, so agent jobs cannot fire; they run at the next tick once it "
|
||||
"is ready. Check it with `hermes gateway status`.")
|
||||
|
||||
|
||||
def _note_fire_refused(job_ids, detail: str) -> None:
|
||||
@@ -50,18 +65,19 @@ def _note_fire_refused(job_ids, detail: str) -> None:
|
||||
|
||||
def refuse_agent_jobs_without_gateway(due_jobs: list) -> list:
|
||||
"""Return the due jobs a headless tick may dispatch; agent jobs are held back (skipped, not
|
||||
failed) when no gateway serves this home. Logs once per tick, not once per job."""
|
||||
failed) when no ready gateway serves this home. Logs once per tick, not once per job."""
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
agent_jobs = [job for job in due_jobs if not job.get("no_agent")]
|
||||
if not agent_jobs:
|
||||
return due_jobs
|
||||
home = get_hermes_home().resolve()
|
||||
if not _gateway_absent(home):
|
||||
hint = _gateway_unavailable(home)
|
||||
if hint is None:
|
||||
return due_jobs
|
||||
ids = [str(job["id"]) for job in agent_jobs]
|
||||
_note_fire_refused(ids, f"Skipped: {GATEWAY_DOWN_HINT}")
|
||||
_note_fire_refused(ids, f"Skipped: {hint}")
|
||||
logger.warning(
|
||||
"Cron tick for %s skipped %d agent job(s) (%s): %s",
|
||||
home, len(ids), ", ".join(ids), GATEWAY_DOWN_HINT)
|
||||
home, len(ids), ", ".join(ids), hint)
|
||||
return [job for job in due_jobs if job.get("no_agent")]
|
||||
@@ -6,6 +6,7 @@ unmanaged daemon when nothing owns the profile. Headless surfaces refuse rather
|
||||
the agent job is skipped (not failed), keeps its due instant, and one warning names the fix.
|
||||
"""
|
||||
from datetime import timedelta
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -69,6 +70,69 @@ def test_tick_without_gateway_skips_agent_jobs_without_spawning_or_drift(cron_ho
|
||||
assert agent["id"] in refusals[0].getMessage() and "hermes gateway start" in refusals[0].getMessage()
|
||||
|
||||
|
||||
@pytest.mark.linux_only
|
||||
@pytest.mark.parametrize("stand_in", ["starting_lock", "silent_socket"])
|
||||
def test_tick_holds_agent_jobs_while_gateway_is_starting_or_silent(cron_home, monkeypatch, caplog, stand_in):
|
||||
"""A gateway mid-restart (lock held, no socket) or one whose control socket accepts but never
|
||||
answers is 'not ready', not 'failed': the tick must hold the slot exactly like ``absent`` —
|
||||
no run row, next_run_at kept — and decide well inside the crontab minute."""
|
||||
import fcntl
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
|
||||
from hermes_cli import gateway_runtime_start
|
||||
|
||||
os.chmod(cron_home, 0o700)
|
||||
spawned = []
|
||||
monkeypatch.setattr(gateway_runtime_start, "spawn_unmanaged_gateway", lambda *a, **k: spawned.append(a))
|
||||
keep = []
|
||||
if stand_in == "starting_lock":
|
||||
lock = open(cron_home / "gateway.lock", "w")
|
||||
os.chmod(cron_home / "gateway.lock", 0o600)
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
keep.append(lock)
|
||||
else:
|
||||
from gateway.control_socket import resolve_server_socket_path
|
||||
bind_path, pointer = resolve_server_socket_path(cron_home) # same sun_path fallback as the daemon
|
||||
bind_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
server.bind(str(bind_path))
|
||||
os.chmod(bind_path, 0o600)
|
||||
if pointer is not None:
|
||||
pointer.write_text(str(bind_path), encoding="utf-8")
|
||||
os.chmod(pointer, 0o600)
|
||||
server.listen(4)
|
||||
keep.append(server)
|
||||
|
||||
def hold(): # accept and hold every peer open: an owner that never answers identify
|
||||
while True:
|
||||
try:
|
||||
keep.append(server.accept()[0])
|
||||
except OSError:
|
||||
return
|
||||
threading.Thread(target=hold, daemon=True).start()
|
||||
|
||||
agent = J.create_job(prompt="summarise inbox", schedule="every 1h", name="agent", deliver="local")
|
||||
_make_due(agent["id"])
|
||||
due_before = J.load_jobs()[0]["next_run_at"]
|
||||
started = time.monotonic()
|
||||
with caplog.at_level("WARNING", logger="cron.scheduler_gateway_gate"):
|
||||
ran = S.tick(verbose=False, headless=True)
|
||||
elapsed = time.monotonic() - started
|
||||
for handle in keep:
|
||||
handle.close()
|
||||
|
||||
assert ran == 0 and spawned == []
|
||||
assert elapsed < 20, f"headless probe must decide inside the crontab minute, took {elapsed:.1f}s"
|
||||
stored = J.load_jobs()[0]
|
||||
assert stored["last_run_at"] is None and stored["last_status"] is None
|
||||
assert stored["next_run_at"] == due_before, "the slot is held, not consumed"
|
||||
assert "pending_slot" not in stored
|
||||
assert "hermes gateway status" in stored["last_fire_error"]["detail"]
|
||||
assert len([r for r in caplog.records if "skipped" in r.getMessage()]) == 1
|
||||
|
||||
|
||||
def test_no_agent_jobs_still_fire_without_a_gateway(cron_home, monkeypatch):
|
||||
counter = cron_home / "fires.txt"
|
||||
script = cron_home / "scripts" / "fire.sh"
|
||||
|
||||
Reference in new issue
Block a user