From 5deca9d5efe5ae8efad4715f166df741677ce161 Mon Sep 17 00:00:00 2001 From: qbit-mirror-bot Date: Wed, 1 Jul 2026 11:06:59 +0000 Subject: [PATCH] fix(gateway,cron): guard cron model-tool path + auto-resume loop breaker (#30719) --- cron/jobs.py | 121 ++++++++++- cron/lifecycle_guard.py | 141 +++++++++++++ gateway/restart_loop_guard.py | 150 ++++++++++++++ gateway/run.py | 156 +++++++++++++-- hermes_cli/config.py | 77 ++++++- hermes_cli/cron.py | 59 ++---- tests/hermes_cli/test_gateway_restart_loop.py | 188 +++++++++++++++++- 7 files changed, 814 insertions(+), 78 deletions(-) create mode 100644 cron/lifecycle_guard.py create mode 100644 gateway/restart_loop_guard.py diff --git a/cron/jobs.py b/cron/jobs.py index dd69ef55ef06..fdb994950115 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -960,6 +960,15 @@ def create_job( context_from = None prompt_text = _coerce_job_text(prompt) + + # Reject cron jobs that schedule gateway-lifecycle commands. Prevents + # agent-driven SIGTERM-respawn loops under launchd/systemd KeepAlive + # (#30719). Enforced here (not only in the CLI layer) so the agent's + # `cronjob` model tool — which calls create_job directly — is also + # covered, not just `hermes cron create`. + from cron.lifecycle_guard import check_gateway_lifecycle + check_gateway_lifecycle(prompt_text, normalized_script) + label_source = (prompt_text or (normalized_skills[0] if normalized_skills else None) or (normalized_script if normalized_no_agent else None)) or "cron job" provider_snapshot, model_snapshot = _compute_provider_model_snapshots( @@ -1249,13 +1258,27 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None, # be claimed again on its next fire (Phase 4C CAS). job["fire_claim"] = None - # Increment completed count + # Increment completed count. Finite one-shot jobs are + # pre-claimed by claim_dispatch() BEFORE the side effect runs + # (issue #38758), which already incremented completed — do not + # double-count them here. Recurring jobs and direct callers + # with no pre-run claim still get the legacy increment. if job.get("repeat"): - job["repeat"]["completed"] = job["repeat"].get("completed", 0) + 1 - + repeat = job["repeat"] + times = repeat.get("times") + completed = repeat.get("completed", 0) + kind = job.get("schedule", {}).get("kind") + preclaimed_oneshot = ( + kind == "once" + and times is not None + and times > 0 + and completed > 0 + ) + if not preclaimed_oneshot: + completed += 1 + repeat["completed"] = completed + # Check if we've hit the repeat limit - times = job["repeat"].get("times") - completed = job["repeat"]["completed"] if times is not None and times > 0 and completed >= times: # Remove the job (limit reached) jobs.pop(i) @@ -1300,6 +1323,69 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None, logger.warning("mark_job_run: job_id %s not found, skipping save", job_id) +def claim_dispatch(job_id: str) -> bool: + """Atomically claim a finite one-shot job dispatch BEFORE execution. + + Increments ``repeat.completed`` under the cross-process jobs lock and + persists the claim immediately, so that if the tick dies mid-execution + (gateway kill, OOM, segfault, hard-timeout) the dispatch is not lost. + This converts finite one-shot jobs from *at-least-once* to *at-most-times* + semantics — a job that self-destructs fires at most ``repeat.times`` times + instead of infinitely (issue #38758). + + Returns ``True`` if the caller may proceed to run the job, ``False`` if the + dispatch limit is already reached (in which case the stale job is removed). + + Only claims jobs with ``schedule.kind == "once"`` and ``repeat.times > 0``. + Recurring jobs (they use ``advance_next_run``) and infinite-repeat / no-repeat + jobs are left unchanged and always allowed to proceed. + """ + with _jobs_lock(): + jobs = load_jobs() + for i, job in enumerate(jobs): + if job["id"] != job_id: + continue + if job.get("schedule", {}).get("kind") != "once": + return True # recurring jobs use advance_next_run(), not dispatch claims + repeat = job.get("repeat") + if not repeat: + return True # no repeat limit — always dispatch + times = repeat.get("times") + if times is None or times <= 0: + return True # infinite — always dispatch + completed = repeat.get("completed", 0) + if completed >= times: + # Already dispatched the max number of times (e.g. a prior + # tick claimed then died before mark_job_run could remove it). + # Clean up so it stops appearing as due on every tick. + jobs.pop(i) + save_jobs(jobs) + logger.info( + "Job '%s': dispatch limit reached (%d/%d) — removing", + job.get("name", job["id"]), + completed, + times, + ) + return False + # Claim this dispatch before the side effect runs. + repeat["completed"] = completed + 1 + save_jobs(jobs) + logger.debug( + "Job '%s': claimed dispatch %d/%d", + job.get("name", job["id"]), + repeat["completed"], + times, + ) + return True + + logger.debug( + "claim_dispatch: job_id %s not in store — proceeding without claim " + "(handed-in job dict; nothing to persist a claim against)", + job_id, + ) + return True + + def advance_next_run(job_id: str) -> bool: """Preemptively advance next_run_at for a recurring job before execution. @@ -1543,6 +1629,31 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: break # Fall through to due.append(job) — execute once now + # One-shot dispatch-limit guard (issue #38758): a finite one-shot + # claimed via claim_dispatch() but whose tick died before + # mark_job_run could remove it will have completed >= times while + # still looking due (last_run_at was never written, so the + # recovery helper re-armed it). Remove it instead of re-firing. + if kind == "once": + repeat = job.get("repeat") + if repeat: + times = repeat.get("times") + completed = repeat.get("completed", 0) + if times is not None and times > 0 and completed >= times: + logger.info( + "Job '%s': one-shot dispatch limit reached (%d/%d) " + "— removing stale due entry", + job.get("name", job["id"]), + completed, + times, + ) + for rj in raw_jobs: + if rj["id"] == job["id"]: + raw_jobs.remove(rj) + needs_save = True + break + continue + due.append(job) if needs_save: diff --git a/cron/lifecycle_guard.py b/cron/lifecycle_guard.py new file mode 100644 index 000000000000..6c70c1af8ae8 --- /dev/null +++ b/cron/lifecycle_guard.py @@ -0,0 +1,141 @@ +"""Gateway lifecycle guard for cron job creation (#30719). + +An agent running inside a gateway can schedule a cron job that calls +``hermes gateway restart`` (or ``launchctl kickstart ai.hermes.gateway`` +or ``systemctl restart hermes-gateway``). When the cron fires, the +gateway dies, the supervisor (launchd KeepAlive / systemd Restart=) +revives it, auto-resume picks up the offending session, and the resumed +turn re-runs the same logic — a SIGTERM-respawn loop every ~10 seconds +until manually broken. + +This module rejects cron job specs whose prompt or script contains a +direct shell-level gateway-lifecycle command. It is enforced at +``cron.jobs.create_job`` so it fires on every job-creation path: the +``hermes cron create`` CLI subcommand AND the agent's ``cronjob`` model +tool (which calls ``create_job`` directly, bypassing the CLI layer). + +The pattern is intentionally command-shaped: it anchors on a concrete +command identifier (``hermes gateway``, ``launchctl ... hermes-gateway``, +``systemctl ... hermes-gateway``, ``pkill`` against the gateway) so it +cannot fire on prose. A cron ``prompt`` is fed to a future LLM, not a +shell, so an over-broad substring match on English ("Kong API gateway +autoscaling and restart behavior") would produce a high false-positive +rate without preventing the actual foot-gun, which requires a real +command shape. + +This is a defence-in-depth layer. ``tools/terminal_tool.py`` already +blocks these commands at *execution* time when ``_HERMES_GATEWAY=1``, and +``hermes gateway stop|restart`` refuse to self-target from inside the +gateway. Blocking at *creation* time as well means the agent gets an +immediate, informative rejection instead of scheduling a job that will +only fail (silently) when it fires. +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Optional + + +class GatewayLifecycleBlocked(ValueError): + """Raised when a cron job spec contains a gateway-lifecycle command.""" + + +# Shell-level command shapes that target the gateway lifecycle. Each branch +# is anchored on a concrete command identifier so a match can only fire on +# actual shell-command-shaped strings, not on prose. +_GATEWAY_LIFECYCLE_PATTERN = re.compile( + r"(?i)" + # Branch A: `hermes gateway restart|stop` — the canonical foot-gun. + # `start` is intentionally excluded: starting a gateway from inside a + # gateway is benign (a no-op or "already running" error), and a + # legitimate cron job might start a sibling profile's gateway. + r"(?:hermes\s+gateway\s+(?:restart|stop))" + # Branch B: launchctl ops on a hermes-gateway label. macOS launchd + # labels look like `ai.hermes.gateway` / `hermes-gateway`. Requiring the + # gateway identifier prevents blocking unrelated hermes services (e.g. + # `launchctl unload ai.hermes.update-checker.plist`). + r"|(?:launchctl\s+(?:kickstart|unload|load|stop|restart)\b[^\n]*\bhermes[.\-]?gateway)" + # Branch C: systemctl ops on a hermes-gateway unit. + r"|(?:systemctl\s+(?:-\S+\s+)*(?:restart|stop|start)\b[^\n]*\bhermes[.\-]?gateway)" + # Branch D: pkill / kill targeting the hermes gateway process. Both + # token orders because real reproductions show both. + r"|(?:p?kill\b[^\n]*\bhermes\b[^\n]*\bgateway)" + r"|(?:p?kill\b[^\n]*\bgateway\b[^\n]*\bhermes)" +) + + +def contains_gateway_lifecycle_command(text: str) -> bool: + """Return True if *text* contains a gateway lifecycle command pattern.""" + if not text: + return False + return bool(_GATEWAY_LIFECYCLE_PATTERN.search(text)) + + +def _resolve_script_path(script_path: str) -> Path: + """Resolve a cron ``script`` value the same way the scheduler does. + + The scheduler (``cron.scheduler``) resolves a bare/relative script path + under ``/scripts/`` and only accepts absolute paths as-is. + We MUST mirror that here so the guard scans the file that will actually + run — otherwise a job whose script lives at the scheduler's real location + (``~/.hermes/scripts/restart.sh``) but is passed as the bare name + ``restart.sh`` would read as a nonexistent relative path and silently + scan prompt-only content, letting the command through. + """ + from hermes_constants import get_hermes_home + + raw = Path(script_path).expanduser() + if raw.is_absolute(): + return raw + return get_hermes_home() / "scripts" / raw + + +def _read_script_for_scanning(script_path: str) -> str: + """Read a script file for lifecycle-pattern scanning. + + Decodes with ``errors="replace"`` so binary or non-UTF-8 content does not + silently bypass the check — a plain text-mode read raises + ``UnicodeDecodeError`` on such files, and swallowing that error would let + an attacker hide the command in binary noise. Returns an empty string + only when the file cannot be read at all. + """ + try: + return _resolve_script_path(script_path).read_bytes().decode( + "utf-8", errors="replace" + ) + except OSError: + return "" + + +def check_gateway_lifecycle( + prompt: Optional[str], + script: Optional[str] = None, +) -> None: + """Raise ``GatewayLifecycleBlocked`` if *prompt* or *script* contains a + gateway-lifecycle command pattern. + + ``prompt`` is scanned directly. ``script``, when supplied, is read from + disk and concatenated for the scan. Both are considered together so a + job cannot slip through by splitting the command across the prompt and + the script. + + Callers should let the exception propagate when they want the create to + fail with a ``ValueError``-shaped error (the agent's ``cronjob`` tool + surfaces this as a tool error; the CLI prints it in red and exits 1). + """ + combined = prompt or "" + if script: + script_text = _read_script_for_scanning(script) + if script_text: + combined = f"{combined}\n{script_text}" + + if contains_gateway_lifecycle_command(combined): + raise GatewayLifecycleBlocked( + "Blocked: cron job contains a gateway lifecycle command " + "(restart/stop/kill). This is blocked to prevent agent-driven " + "SIGTERM-respawn loops under launchd/systemd supervision " + "(#30719). Run `hermes gateway restart` from a shell outside " + "the running gateway instead." + ) diff --git a/gateway/restart_loop_guard.py b/gateway/restart_loop_guard.py new file mode 100644 index 000000000000..c17ee8ca42bd --- /dev/null +++ b/gateway/restart_loop_guard.py @@ -0,0 +1,150 @@ +"""Auto-resume restart-loop breaker (#30719, defense-3). + +Defenses 1 and 2 (the ``_HERMES_GATEWAY`` guard on ``hermes gateway +stop|restart`` + ``terminal_tool``, and the cron-creation lifecycle +filter) stop the agent from scheduling its own restart via the cron and +CLI paths. They do NOT cover every SIGTERM source: an agent running a +raw ``terminal("launchctl kickstart -k gui//ai.hermes.gateway")``, +an external monitor with a bad trigger, or any other repeated crash can +still drive the supervisor (launchd ``KeepAlive`` / systemd ``Restart=``) +into a tight respawn loop. On each boot the gateway auto-resumes the +restart-interrupted session, whose next turn re-runs the offending +logic — SIGTERM every ~10 seconds until manually broken. + +This module is the last-resort circuit breaker: it records a timestamp +each time the gateway boots with restart-interrupted sessions pending, +keeps a rolling window of recent boots persisted across processes (each +boot is a fresh process, so in-memory state is useless), and reports the +loop as "tripped" once too many such boots happen inside a short window. +When tripped, the caller SKIPS auto-resume for that boot — the gateway +still starts and serves real inbound messages, it just stops replaying +the session that keeps killing it, which breaks the cycle and puts a +human back in the loop. + +State lives in ``/gateway/restart_loop.json`` so it is +profile-scoped and survives process death. It is intentionally tiny and +best-effort: any read/write failure fails OPEN (no false trip) because a +broken breaker must never wedge a healthy gateway. +""" + +from __future__ import annotations + +import json +import logging +import time +from typing import List, Optional + +from hermes_constants import get_hermes_home + +logger = logging.getLogger("gateway.run") + +# Defaults chosen so a legitimate operator restart (or two) never trips the +# breaker, but the documented ~10s respawn loop does within a few cycles. +DEFAULT_MAX_RESTARTS = 3 +DEFAULT_WINDOW_SECONDS = 60 + + +def _state_path(): + return get_hermes_home() / "gateway" / "restart_loop.json" + + +def _load_boots() -> List[float]: + try: + raw = _state_path().read_text(encoding="utf-8") + data = json.loads(raw) + boots = data.get("boots", []) + return [float(t) for t in boots if isinstance(t, (int, float))] + except (OSError, ValueError, TypeError): + return [] + + +def _save_boots(boots: List[float]) -> None: + try: + path = _state_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"boots": boots}), encoding="utf-8") + except OSError: + pass + + +def record_restart_interrupted_boot( + window_seconds: int = DEFAULT_WINDOW_SECONDS, + *, + now: Optional[float] = None, +) -> List[float]: + """Record that the gateway just booted with restart-interrupted sessions. + + Prunes boots older than ``window_seconds`` and appends the current time. + Returns the pruned+appended list (most recent last). Best-effort — a + persistence failure returns the in-memory list without raising. + """ + ts = time.time() if now is None else now + cutoff = ts - max(1, window_seconds) + boots = [t for t in _load_boots() if t >= cutoff] + boots.append(ts) + _save_boots(boots) + return boots + + +def is_restart_loop_tripped( + max_restarts: int = DEFAULT_MAX_RESTARTS, + window_seconds: int = DEFAULT_WINDOW_SECONDS, + *, + now: Optional[float] = None, +) -> bool: + """Return True if the gateway has restarted ``>= max_restarts`` times with + restart-interrupted sessions inside the last ``window_seconds``. + + Reads the persisted boot log written by + ``record_restart_interrupted_boot`` and counts boots within the window. + Fails OPEN (returns False) on any error — a broken breaker must never + wedge a healthy gateway. + """ + if max_restarts <= 0: + return False + ts = time.time() if now is None else now + cutoff = ts - max(1, window_seconds) + try: + recent = [t for t in _load_boots() if t >= cutoff] + except Exception: # pragma: no cover — _load_boots already guards + return False + return len(recent) >= max_restarts + + +def clear() -> None: + """Remove the persisted boot log (used on clean shutdown / by tests).""" + try: + _state_path().unlink(missing_ok=True) + except OSError: + pass + + +def check_and_record( + max_restarts: int = DEFAULT_MAX_RESTARTS, + window_seconds: int = DEFAULT_WINDOW_SECONDS, + *, + now: Optional[float] = None, +) -> bool: + """Record this restart-interrupted boot and report whether the loop is now + tripped. + + This is the single entry point the gateway calls: it appends the current + boot, then checks whether the (now-updated) window has reached the + threshold. Returns True when auto-resume should be SKIPPED to break the + loop. + """ + boots = record_restart_interrupted_boot(window_seconds, now=now) + tripped = len(boots) >= max_restarts if max_restarts > 0 else False + if tripped: + logger.warning( + "Restart-loop breaker TRIPPED: %d restart-interrupted gateway " + "boots within %ds (threshold %d). Skipping auto-resume to break " + "a suspected SIGTERM-respawn loop (#30719). Restart-interrupted " + "sessions stay resume-pending and will continue on the next real " + "user message. If this is a false positive, delete %s.", + len(boots), + window_seconds, + max_restarts, + _state_path(), + ) + return tripped diff --git a/gateway/run.py b/gateway/run.py index 5f1cd2313c87..60bf26635a1b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -147,7 +147,6 @@ def _gateway_surface_passes_raw_text(platform: Any) -> bool: _GATEWAY_SECRET_PATTERNS = ( re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"), re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), - re.compile(r"\bxapp-\d+-[A-Za-z0-9\-]{20,}\b"), re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{20,}\b"), re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"), re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"), @@ -3801,6 +3800,29 @@ def _scale_to_zero_idle_timeout_seconds(self) -> float: raw = None return parse_idle_timeout_seconds(raw) + def _restart_loop_guard_config(self) -> tuple: + """Return ``(max_restarts, window_seconds)`` for the auto-resume + restart-loop breaker (#30719, defense-3), read from + ``gateway.restart_loop_guard`` in config.yaml with the module defaults + as fallback. ``max_restarts <= 0`` disables the breaker. + """ + from gateway import restart_loop_guard as _rlg + + max_restarts = _rlg.DEFAULT_MAX_RESTARTS + window_seconds = _rlg.DEFAULT_WINDOW_SECONDS + try: + user_cfg = _load_gateway_config() + gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None + rlg = gw.get("restart_loop_guard") if isinstance(gw, dict) else None + if isinstance(rlg, dict): + if isinstance(rlg.get("max_restarts"), int): + max_restarts = rlg["max_restarts"] + if isinstance(rlg.get("window_seconds"), int) and rlg["window_seconds"] > 0: + window_seconds = rlg["window_seconds"] + except Exception: # noqa: BLE001 + pass + return max_restarts, window_seconds + def _scale_to_zero_should_arm(self) -> bool: """Whether to start the idle watcher (D1/D11/§3.4(1)).""" from gateway.relay import relay_wake_url @@ -5962,6 +5984,26 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int: logger.warning("Failed to enumerate resume-pending sessions: %s", exc) return 0 + # Defense-3 (#30719): break the SIGTERM-respawn loop. Only count this + # boot when there are restart-interrupted sessions to resume — a clean + # boot must not accrue toward the breaker. If too many such boots have + # happened in the configured window, skip auto-resume for THIS boot: + # the gateway still comes up and serves real inbound messages, it just + # stops replaying the session that keeps killing it. The session stays + # resume_pending, so a real user message can still continue it (a human + # is now in the loop). Defenses 1-2 cover the cron/CLI/terminal paths; + # this catches every other SIGTERM source (e.g. a raw `terminal( + # "launchctl kickstart ai.hermes.gateway")`). + if candidates: + try: + from gateway import restart_loop_guard as _rlg + + _max_restarts, _window = self._restart_loop_guard_config() + if _rlg.check_and_record(_max_restarts, _window): + return 0 + except Exception as exc: # noqa: BLE001 — breaker must fail OPEN + logger.debug("Restart-loop guard check skipped: %s", exc) + now = datetime.now() scheduled = 0 for entry in candidates: @@ -10680,19 +10722,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g _response_time, _api_calls, _resp_len, ) - # Re-baseline the cached agent's message_count snapshot now that - # this turn has completed and the agent has flushed its rows to - # the SessionDB. The cross-process coherence guard (#45966) - # snapshots the count at agent-BUILD time (before this turn's own - # writes) and never refreshes it on reuse — so without this, this - # process's own turn would grow the count and the next turn would - # see a mismatch and rebuild the agent every turn, destroying - # prompt caching. Refreshing here makes the guard fire only on a - # DIFFERENT process's writes. Uses the (possibly compaction- - # updated) live session_id. Fail-safe inside the helper. - await self._refresh_agent_cache_message_count( - session_key, session_entry.session_id - ) + # NOTE: the cross-process cache-coherence re-baseline + # (_refresh_agent_cache_message_count) is intentionally deferred + # until AFTER this turn's transcript persistence block below — it + # must include the first-turn `session_meta` marker row and the + # compression session_id swap, both of which happen later. See + # the call site after the `update_session(...)` write. # Successful turn — clear any stuck-loop counter for this session. # This ensures the counter only accumulates across CONSECUTIVE @@ -11078,6 +11113,28 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), ) + # Re-baseline the cached agent's message_count snapshot now that + # ALL of this turn's transcript writes are done — the agent's + # flushed user/assistant/tool rows AND the first-turn `session_meta` + # marker appended above. The cross-process coherence guard (#45966) + # snapshots the count at agent-BUILD time (before this turn's own + # writes) and never refreshes it on reuse, so without this the + # process's own turn grows message_count and the next turn sees a + # mismatch and rebuilds the agent — destroying prompt caching. + # + # This MUST run after the `session_meta` append: that row also + # increments message_count, so re-baselining before it (the old + # position) left the snapshot one short and the guard mis-fired on + # turn 2 of EVERY fresh gateway conversation, rebuilding the cached + # agent and busting the prompt cache. Running here also uses the + # compaction-updated session_id (the agent_result session_id swap + # above), matching this function's documented contract. Refreshing + # here makes the guard fire only on a DIFFERENT process's writes. + # Fail-safe inside the helper. + await self._refresh_agent_cache_message_count( + session_key, session_entry.session_id + ) + # Intentional silence is a delivery decision, not a transcript # mutation. The agent's [SILENT]/NO_REPLY assistant turn above is # still persisted in session history so later turns keep normal @@ -14903,6 +14960,15 @@ async def _refresh_agent_cache_message_count( only when the same agent is still cached (no rebuild/eviction raced in between). Fail-safe: any DB error leaves the snapshot as-is, which at worst costs one unnecessary rebuild on the next turn. + + When the cache entry records a ``session_id`` (4-tuple form, #54947) + that differs from the current ``session_id`` — meaning the cache + was built for a DIFFERENT conversation under the same ``session_key`` + — the snapshot is intentionally left untouched. Overwriting it with + the current session's count would corrupt the original conversation's + baseline and cause the next switch back to fire the cross-process + guard spuriously. Fail-safe: the legacy 3-tuple shape (no + ``session_id``) is still re-baselined as before. """ if self._session_db is None or not session_id: return @@ -14927,8 +14993,23 @@ async def _refresh_agent_cache_message_count( and len(cached) > 2 and cached[0] is not _AGENT_PENDING_SENTINEL ): + # If the snapshot was taken for a different session_id + # (same session_key, different conversation), leave the + # snapshot alone — the current session_id's count belongs + # to a different DB row (#54947). + _snapshot_sid = cached[3] if len(cached) > 3 else None + if _snapshot_sid is not None and _snapshot_sid != session_id: + return if cached[2] != _live: - _cache[session_key] = (cached[0], cached[1], _live) + if _snapshot_sid is None: + # Legacy 3-tuple: preserve the original 3-element + # shape so existing entries stay compatible with + # callers that index ``cached[2]`` directly. + _cache[session_key] = (cached[0], cached[1], _live) + else: + _cache[session_key] = ( + cached[0], cached[1], _live, _snapshot_sid, + ) def _evict_cached_agent(self, session_key: str) -> None: """Remove a cached agent for a session (called on /new, /model, etc). @@ -16635,9 +16716,25 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: if cached and cached[1] == _sig: # cached[2] is the message_count at cache time; # stale when a second process appended rows. + # cached[3] (when present) is the session_id the + # snapshot was taken for — used to skip the guard + # when the active session_id differs (#54947). _cached_mc = cached[2] if len(cached) > 2 else None + _cached_sid = cached[3] if len(cached) > 3 else None + # If the snapshot belongs to a different session_id + # (same session_key, different conversation), the + # message_count comparison is meaningless — the + # counts track DIFFERENT DB rows. REUSE the cached + # agent rather than rebuild and bust the prompt cache + # on every session switch (#54947). + _session_id_mismatch = ( + _cached_sid is not None + and session_id is not None + and _cached_sid != session_id + ) if ( - _cached_mc is not None + not _session_id_mismatch + and _cached_mc is not None and _current_msg_count is not None and _current_msg_count != _cached_mc ): @@ -16668,7 +16765,6 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: _xproc_evicted_agent = _ev_agent else: agent = cached[0] - reused_cached_agent = True # Refresh LRU order so the cap enforcement evicts # truly-oldest entries, not the one we just used. if hasattr(_cache, "move_to_end"): @@ -16681,6 +16777,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: # (cached agent may have been created with old config) agent.max_iterations = max_iterations logger.debug("Reusing cached agent for session %s", session_key) + reused_cached_agent = True # Lock released — now schedule cleanup of any cross-process-evicted # agent on a daemon thread so memory-provider shutdown / socket @@ -16738,7 +16835,14 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: ) if _cache_lock and _cache is not None: with _cache_lock: - _cache[session_key] = (agent, _sig, _current_msg_count) + # Record the session_id the snapshot was taken for + # alongside the message_count, so the cross-process + # guard can skip the (meaningless) count comparison + # when the active session_id later switches under + # the same session_key (#54947). + _cache[session_key] = ( + agent, _sig, _current_msg_count, session_id, + ) self._enforce_agent_cache_cap() logger.debug("Created new agent for session %s (sig=%s)", session_key, _sig) @@ -18207,6 +18311,22 @@ def _stream_confirmed_final_delivery( except Exception: pass + # Re-baseline the cached agent's message_count snapshot before + # recursing into the in-band queued (/queue) follow-up turn. + # The first turn has completed and flushed its own user + + # assistant rows to the SessionDB, so the cross-process + # coherence guard (#45966) — which this recursive _run_agent + # call re-enters — would otherwise see the grown on-disk count + # against the stale build-time snapshot and rebuild the agent + # on THIS process's OWN writes, destroying the prompt-cache + # prefix #46237 was merged to preserve. The existing + # re-baseline in _handle_message_with_agent only runs after the + # whole _run_agent chain unwinds — too late for the in-band + # follow-up. Use the same (session_key, session_id) the + # recursive call runs under so the snapshot matches exactly + # what the follow-up's guard will consult. Fail-safe in helper. + await self._refresh_agent_cache_message_count(session_key, session_id) + followup_result = await self._run_agent( message=next_message, context_prompt=context_prompt, diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 8ba3bae113e4..92ec87fbdc8d 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1391,11 +1391,8 @@ def _ensure_hermes_home_managed(home: Path): }, # Anthropic prompt caching (Claude via OpenRouter or native Anthropic API). - # Set enabled: false as an escape hatch for strict providers that reject - # cache_control markers; cache_ttl must be "5m" or "1h" (Anthropic-supported - # tiers), other values are ignored. + # cache_ttl must be "5m" or "1h" (Anthropic-supported tiers); other values are ignored. "prompt_caching": { - "enabled": True, "cache_ttl": "5m", }, @@ -2158,6 +2155,14 @@ def _ensure_hermes_home_managed(home: Path): "moa": { "default_preset": "default", "active_preset": "", + # When true, every MoA turn that runs the reference fan-out writes the + # FULL turn (each reference's exact input messages + output + usage/cost, + # and the aggregator's exact input + output) to a JSONL file at + # /moa-traces/.jsonl. Off by default — turn it + # on to audit / improve MoA behavior from real runs. Set trace_dir to + # override the output directory. + "save_traces": False, + "trace_dir": "", "presets": { "default": { "reference_models": [ @@ -2712,6 +2717,23 @@ def _ensure_hermes_home_managed(home: Path): "idle_timeout_minutes": 5, }, + # Auto-resume restart-loop breaker (#30719, defense-3). When the + # gateway is killed mid-turn (SIGTERM) and revived by a supervisor + # (launchd KeepAlive / systemd Restart=), it auto-resumes the + # restart-interrupted session on the next boot. If the resumed turn + # keeps triggering another kill (e.g. the agent runs a raw + # `launchctl kickstart ai.hermes.gateway` that defenses 1-2 don't + # cover), the result is a tight SIGTERM-respawn loop. This breaker + # counts restart-interrupted boots in a rolling window and, once + # `max_restarts` boots happen within `window_seconds`, SKIPS + # auto-resume for that boot — the gateway still starts and serves + # real inbound messages, it just stops replaying the session that + # keeps killing it. Set `max_restarts` to 0 to disable the breaker. + "restart_loop_guard": { + "max_restarts": 3, + "window_seconds": 60, + }, + # Inject a human-readable timestamp prefix (e.g. # "[Tue 2026-04-28 13:40:53 CEST]") onto user messages IN THE MODEL'S # CONTEXT so the agent has temporal awareness of when each message was @@ -6695,6 +6717,23 @@ def invalidate_env_cache() -> None: _env_cache = None +_STRUCTURED_VALUE_MARKERS = ("://", "?", "&") + + +def _looks_like_structured_value(value: str) -> bool: + """True when ``value`` looks like a URL/query string or holds whitespace. + + Such a value is treated as one opaque secret. An embedded + ``KNOWN_KEY=`` substring inside it (e.g. a webhook URL carrying a query + parameter, or a proxy base URL with an embedded key) is part of the value, + not the start of a second .env entry, so the concatenation splitter must + not break on it. Plain token secrets (API keys) never contain these. + """ + if any(marker in value for marker in _STRUCTURED_VALUE_MARKERS): + return True + return any(ch.isspace() for ch in value) + + def _sanitize_env_lines(lines: list) -> list: """Fix corrupted .env lines before reading or writing. @@ -6743,10 +6782,32 @@ def _sanitize_env_lines(lines: list) -> list: ) }) - if len(split_positions) > 1: - for i, pos in enumerate(split_positions): - end = split_positions[i + 1] if i + 1 < len(split_positions) else len(stripped) - part = stripped[pos:end].strip() + # Only treat the line as a concatenation when it actually begins with a + # known KEY= (split_positions[0] == 0). A first match at a non-zero + # offset means the matches sit inside a value, so splitting there would + # silently drop the leading text — keep the line intact instead. + split_into_entries = False + segments: list[str] = [] + if len(split_positions) > 1 and split_positions[0] == 0: + segments = [ + stripped[pos:( + split_positions[i + 1] if i + 1 < len(split_positions) else len(stripped) + )] + for i, pos in enumerate(split_positions) + ] + # A genuine concatenation has a simple token value in every segment + # that precedes a boundary. If a preceding value looks structured + # (a URL/query string or whitespace), the embedded KNOWN_KEY= is + # part of that value rather than a new entry, so we must not split — + # otherwise we truncate the real secret and fabricate a bogus one. + split_into_entries = all( + not _looks_like_structured_value(seg.split("=", 1)[1]) + for seg in segments[:-1] + ) + + if split_into_entries: + for seg in segments: + part = seg.strip() if part: sanitized.append(part + "\n") else: diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index 1f806050ad91..b0c907326e88 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -6,7 +6,6 @@ """ import json -import re import sys from pathlib import Path from typing import Iterable, List, Optional @@ -16,25 +15,17 @@ from hermes_cli.colors import Colors, color -# Patterns that indicate a cron job targets the gateway lifecycle. -# Matches commands that restart/stop the gateway or its service manager. -# Deliberately specific — a bare "gateway ... restart" catch-all would block -# legitimate prompts that merely mention an unrelated gateway (e.g. "summarize -# the API gateway logs and report restart events"). -_GATEWAY_LIFECYCLE_PATTERNS = re.compile( - r"(?i)" - r"(hermes\s+gateway\s+(restart|stop|start))" - r"|(launchctl\s+(kickstart|unload|load|stop|restart)\s+.*hermes)" - r"|(systemctl\s+(-\S+\s+)*(restart|stop|start)\s+.*hermes)" - r"|(p?kill\s+.*hermes.*gateway)" +# Gateway-lifecycle command detection lives in ``cron.lifecycle_guard`` so it +# can be shared across every job-creation path (CLI + the agent's ``cronjob`` +# model tool via ``cron.jobs.create_job``) without a circular import. Re-export +# ``_contains_gateway_lifecycle_command`` here for back-compat: ``tools/ +# terminal_tool.py`` imports it from this module to hard-block the same +# commands at execution time when ``_HERMES_GATEWAY=1``. +from cron.lifecycle_guard import ( # noqa: F401 (re-exported for terminal_tool) + contains_gateway_lifecycle_command as _contains_gateway_lifecycle_command, ) -def _contains_gateway_lifecycle_command(text: str) -> bool: - """Return True if *text* contains a gateway lifecycle command pattern.""" - return bool(_GATEWAY_LIFECYCLE_PATTERNS.search(text)) - - def _normalize_skills(single_skill=None, skills: Optional[Iterable[str]] = None) -> Optional[List[str]]: if skills is None: if single_skill is None: @@ -137,7 +128,11 @@ def cron_list(show_all: bool = False): repeat_completed = repeat_info.get("completed", 0) repeat_str = f"{repeat_completed}/{repeat_times}" if repeat_times else "∞" - deliver = job.get("deliver", ["local"]) + # `deliver` may be present-but-null in the job record (same pitfall as + # `repeat` above), so coalesce to the default rather than relying on the + # dict-default, which only applies to a missing key. A null value would + # otherwise reach `", ".join(None)` and crash the whole listing (#32896). + deliver = job.get("deliver") or ["local"] if isinstance(deliver, str): deliver = [deliver] deliver_str = ", ".join(deliver) @@ -297,28 +292,12 @@ def _print_active_jobs_summary(jobs) -> None: def cron_create(args): - # Defense: reject cron jobs that contain gateway lifecycle commands. - # Prevents agents from scheduling their own restart/stop, which creates - # SIGTERM-respawn loops under launchd/systemd KeepAlive (#30719). - prompt = getattr(args, "prompt", None) or "" - script = getattr(args, "script", None) - combined = prompt - if script: - try: - script_text = Path(script).read_text(encoding="utf-8") - combined = f"{combined}\n{script_text}" - except (OSError, UnicodeDecodeError): - pass - if _contains_gateway_lifecycle_command(combined): - print(color( - "Blocked: cron job contains a gateway lifecycle command " - "(restart/stop/kill).\n" - "This is blocked to prevent restart loops (#30719).\n" - "Use `hermes gateway restart` from a shell outside the gateway.", - Colors.RED, - )) - return 1 - + # 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` + # model tool, which calls create_job directly). When it blocks, create_job + # raises GatewayLifecycleBlocked, the `cronjob` tool wrapper catches it and + # returns it as result["error"], and the `if not result.get("success")` + # branch below prints it in red and exits 1 — same UX as before. result = _cron_api( action="create", schedule=args.schedule, diff --git a/tests/hermes_cli/test_gateway_restart_loop.py b/tests/hermes_cli/test_gateway_restart_loop.py index 239b8060024a..03ef426ec896 100644 --- a/tests/hermes_cli/test_gateway_restart_loop.py +++ b/tests/hermes_cli/test_gateway_restart_loop.py @@ -28,7 +28,6 @@ class TestGatewayLifecyclePattern: @pytest.mark.parametrize("text", [ "hermes gateway restart", "hermes gateway stop", - "hermes gateway start", "hermes gateway restart", # double spaces "Hermez Gateway Restart".lower().replace("z", "s"), # case handled "HERMES GATEWAY RESTART", # uppercase @@ -50,6 +49,7 @@ def test_service_manager_commands(self, text): @pytest.mark.parametrize("text", [ "kill hermes gateway process", "pkill -f hermes.*gateway", + "pkill -f gateway.*hermes", # inverse token order ]) def test_kill_commands(self, text): assert _contains_gateway_lifecycle_command(text), f"Should match: {text!r}" @@ -62,11 +62,28 @@ def test_kill_commands(self, text): "echo 'just a normal cron job'", "run the backup script", "gateway is running fine", + # `hermes gateway start` is benign — starting a gateway from inside a + # gateway is a no-op / "already running", and a legit cron job may + # start a sibling profile's gateway. Only restart/stop/kill are the + # foot-gun (#30719 lists only those). + "hermes gateway start", + "hermes gateway start --all", + # Tightened launchctl/systemctl branches: ops on NON-gateway hermes + # services must not be falsely blocked (the old `.*hermes` matched any + # hermes token). + "launchctl unload ai.hermes.update-checker.plist", + "launchctl restart ai.hermes.daemon", + "systemctl restart hermes-meta.service", + "systemctl restart hermes-cron-helper", # Regression (#30728 follow-up): legit prompts that merely mention an - # unrelated gateway + a restart must NOT be blocked. + # unrelated gateway + a restart must NOT be blocked. The cron prompt is + # fed to an LLM, not a shell, so substring detection on English text is + # a high-FP no-op — only concrete command shapes trigger the block. "Summarize the API gateway logs and report any restart events from last night", "Check if the payment gateway needs a restart after the deploy", "Monitor the gateway and tell me if a restart is recommended", + "research how the OpenAI API gateway handles restart after rate limiting", + "compare AWS API Gateway vs Cloudflare on restart latency", ]) def test_safe_commands(self, text): assert not _contains_gateway_lifecycle_command(text), f"Should NOT match: {text!r}" @@ -122,9 +139,14 @@ def test_block_launchctl_kickstart(self, capsys): out = capsys.readouterr().out assert "Blocked" in out - def test_block_script_with_lifecycle_command(self, tmp_path, capsys): - script = tmp_path / "restart.sh" - script.write_text("#!/bin/bash\nhermes gateway restart\n") + def test_block_script_with_lifecycle_command(self, tmp_path, capsys, monkeypatch): + # A no_agent job whose script IS the job (the issue's real abuse path: + # restart_hermes_gateway_once.sh). The script must live under + # HERMES_HOME/scripts so the scheduler — and the guard — resolve it. + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + scripts_dir = tmp_path / ".hermes" / "scripts" + scripts_dir.mkdir(parents=True) + (scripts_dir / "restart.sh").write_text("#!/bin/bash\nhermes gateway restart\n") args = Namespace( cron_command="create", schedule="1h", @@ -134,10 +156,10 @@ def test_block_script_with_lifecycle_command(self, tmp_path, capsys): repeat=None, skill=None, skills=None, - script=str(script), + script="restart.sh", workdir=None, profile=None, - no_agent=False, + no_agent=True, ) rc = cron_command(args) assert rc == 1 @@ -357,3 +379,155 @@ def execute(self, command, **kwargs): # approval flow handles it (here mocked as approved). assert result["exit_code"] == 0 assert calls == ["systemctl restart hermes-gateway"] + + +# --------------------------------------------------------------------------- +# cron.lifecycle_guard module — the shared checker create_job/CLI/terminal use +# --------------------------------------------------------------------------- + +class TestLifecycleGuardModule: + """Direct tests for cron.lifecycle_guard.check_gateway_lifecycle.""" + + def test_prompt_with_command_raises(self): + from cron.lifecycle_guard import GatewayLifecycleBlocked, check_gateway_lifecycle + with pytest.raises(GatewayLifecycleBlocked) as exc: + check_gateway_lifecycle("please run hermes gateway restart", None) + assert "#30719" in str(exc.value) + + def test_clean_prompt_does_not_raise(self): + from cron.lifecycle_guard import check_gateway_lifecycle + check_gateway_lifecycle("research the gateway architecture", None) + check_gateway_lifecycle("check server health and restart watchers", None) + + def test_script_with_command_raises(self, tmp_path, monkeypatch): + from cron.lifecycle_guard import GatewayLifecycleBlocked, check_gateway_lifecycle + script = tmp_path / "restart.sh" + script.write_text("#!/bin/bash\nhermes gateway restart\n") + with pytest.raises(GatewayLifecycleBlocked): + check_gateway_lifecycle("clean prompt", str(script)) + + def test_split_across_prompt_and_script_still_blocks(self, tmp_path): + """Concatenated scan prevents splitting the command between prompt and + script to slip through.""" + from cron.lifecycle_guard import GatewayLifecycleBlocked, check_gateway_lifecycle + script = tmp_path / "ops.sh" + script.write_text("hermes gateway stop\n") + with pytest.raises(GatewayLifecycleBlocked): + check_gateway_lifecycle("daily ops job", str(script)) + + def test_binary_script_does_not_silently_bypass(self, tmp_path): + """Non-UTF-8 bytes used to be swallowed by UnicodeDecodeError; now we + decode with errors='replace' so the scan always sees the command.""" + from cron.lifecycle_guard import GatewayLifecycleBlocked, check_gateway_lifecycle + script = tmp_path / "weird.bin" + script.write_bytes(b"\xfehermes gateway restart\xff") + with pytest.raises(GatewayLifecycleBlocked): + check_gateway_lifecycle("", str(script)) + + def test_missing_script_does_not_raise(self, tmp_path): + from cron.lifecycle_guard import check_gateway_lifecycle + check_gateway_lifecycle("clean prompt", str(tmp_path / "nonexistent.sh")) + + def test_relative_script_resolved_under_scripts_dir(self, tmp_path, monkeypatch): + """A bare/relative script name resolves under HERMES_HOME/scripts (the + same place the scheduler runs it from) — otherwise the guard would read + a nonexistent relative path and scan prompt-only content.""" + from cron.lifecycle_guard import GatewayLifecycleBlocked, check_gateway_lifecycle + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + scripts_dir = tmp_path / ".hermes" / "scripts" + scripts_dir.mkdir(parents=True) + (scripts_dir / "restart.sh").write_text( + "launchctl kickstart -k gui/501/ai.hermes.gateway\n" + ) + with pytest.raises(GatewayLifecycleBlocked): + check_gateway_lifecycle("daily", "restart.sh") + + +# --------------------------------------------------------------------------- +# Defense 2 (chokepoint): cron.jobs.create_job blocks the AGENT model-tool path +# --------------------------------------------------------------------------- + +class TestCreateJobBlocksLifecycleCommands: + """The regression the CLI-layer-only guard could not catch: the agent's + `cronjob` model tool calls cron.jobs.create_job directly, bypassing + hermes_cli.cron.cron_create. Enforcing at create_job covers both.""" + + @pytest.fixture(autouse=True) + def _setup_cron_dir(self, 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") + + def test_create_job_blocks_prompt_command(self): + from cron.jobs import create_job + from cron.lifecycle_guard import GatewayLifecycleBlocked + with pytest.raises(GatewayLifecycleBlocked): + create_job(prompt="then run hermes gateway restart", schedule="30m") + + def test_create_job_allows_benign_prompt(self): + from cron.jobs import create_job + job = create_job(prompt="summarize the API gateway logs and note restart events", + schedule="30m") + assert job["id"] + + def test_cronjob_tool_surfaces_block_as_error(self, tmp_path, monkeypatch): + """End-to-end through the model tool: the block comes back as + result['error'] with the #30719 hint, not an unhandled exception.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + (tmp_path / ".hermes").mkdir(parents=True) + from tools.cronjob_tools import cronjob + result = json.loads(cronjob( + action="create", schedule="0 9 * * *", + prompt="please run hermes gateway restart nightly", + )) + assert result.get("success") is False + assert "#30719" in result.get("error", "") + + +# --------------------------------------------------------------------------- +# Defense 3: auto-resume restart-loop breaker +# --------------------------------------------------------------------------- + +class TestRestartLoopGuard: + """gateway.restart_loop_guard trips after >= max_restarts + restart-interrupted boots inside window_seconds, breaking a + SIGTERM-respawn loop that defenses 1-2 don't cover.""" + + @pytest.fixture(autouse=True) + def _isolate_state(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + (tmp_path / ".hermes").mkdir(parents=True) + import gateway.restart_loop_guard as rlg + rlg.clear() + + def test_burst_trips_on_threshold(self): + import gateway.restart_loop_guard as rlg + assert rlg.check_and_record(3, 60, now=1000.0) is False + assert rlg.check_and_record(3, 60, now=1005.0) is False + assert rlg.check_and_record(3, 60, now=1010.0) is True + + def test_spread_boots_never_trip(self): + import gateway.restart_loop_guard as rlg + assert rlg.check_and_record(3, 60, now=1000.0) is False + assert rlg.check_and_record(3, 60, now=1070.0) is False + assert rlg.check_and_record(3, 60, now=1140.0) is False + + def test_disabled_when_max_restarts_zero(self): + import gateway.restart_loop_guard as rlg + for i in range(5): + assert rlg.check_and_record(0, 60, now=1000.0 + i) is False + + def test_is_tripped_reads_without_recording(self): + import gateway.restart_loop_guard as rlg + rlg.record_restart_interrupted_boot(60, now=1000.0) + rlg.record_restart_interrupted_boot(60, now=1001.0) + assert rlg.is_restart_loop_tripped(3, 60, now=1002.0) is False + rlg.record_restart_interrupted_boot(60, now=1002.0) + assert rlg.is_restart_loop_tripped(3, 60, now=1003.0) is True + + def test_clear_resets(self): + import gateway.restart_loop_guard as rlg + rlg.check_and_record(3, 60, now=1000.0) + rlg.check_and_record(3, 60, now=1001.0) + rlg.clear() + assert rlg.check_and_record(3, 60, now=1002.0) is False