# NousResearch/hermes-agent · Gateway cron agent restart loop prevention fixes

## Open now

- Nothing open on GitHub.

Multiple fixes addressing scenarios where agents could schedule or trigger gateway restart/kill commands that cause infinite respawn loops, including guards in cron dispatch, terminal blocking, loop detection, and one-shot job commit ordering.

## TL;DR

- 9 member(s): 0 open issue(s), 2 closed issue(s), 0 open PR(s), 3 merged PR(s), 4 PR(s) closed without merging.
- A fix was merged to main in PR #35679. Verify it addresses your case.
- **Recommended:** read PR #35679 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-35679`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
- Members outside that structural core may belong to a different underlying problem.
- Pairwise link density `0.20`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Closed / duplicate issues

- #7818: Two architectural gaps exposed by a launchctl self-kill loop: smart approval auto-approves self-termination, and resume doesn't detect post-crash retry
- #30719: Agent can schedule gateway-restart cron job that kills its own runtime, creating respawn loop with launchctl/systemd KeepAlive

## Prior & competing PRs

- **#35679: fix(gateway,cron): prevent agent gateway-restart loops (salvage #30728)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `hermes_cli/cron.py`, `hermes_cli/gateway.py`, `tests/conftest.py`, `tests/hermes_cli/test_gateway_restart_loop.py`
  - fix hunks:
  - `hermes_cli/cron.py`

    ```diff
    @@ -6,6 +6,7 @@
     """
     
     import json
    +import re
     import sys
     from pathlib import Path
     from typing import Iterable, List, Optional
    ```

    ```diff
    @@ -15,6 +16,24 @@
     
     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+(restart|stop|start)\s+.*hermes)"
    +    r"|(p?kill\s+.*hermes.*gateway)"
    +)
    +
    +
    +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:
    ```

    ```diff
    @@ -166,6 +185,28 @@ def cron_status():
     
     
     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
    +
         result = _cron_api(
             action="create",
             schedule=args.schedule,
    ```
  - `hermes_cli/gateway.py`

    ```diff
    @@ -5422,6 +5422,16 @@ def _gateway_command_inner(args):
                 sys.exit(1)
     
         elif subcmd == "stop":
    +        # Defense: refuse self-targeting gateway stop from inside the gateway.
    +        # Prevents agent-initiated kill loops when combined with supervisor KeepAlive.
    +        if os.getenv("_HERMES_GATEWAY") == "1":
    +            print_error(
    +                "Refusing to stop the gateway from inside the gateway process.\n"
    +                "This command was blocked to prevent restart loops.\n"
    +                "Use `hermes gateway stop` from a shell outside the running gateway."
    +            )
    +            sys.exit(1)
    +
             stop_all = getattr(args, 'all', False)
             system = getattr(args, 'system', False)
     
    ```

    ```diff
    @@ -5497,6 +5507,16 @@ def _gateway_command_inner(args):
                     print(f"✓ Stopped {get_service_name()} service")
         
         elif subcmd == "restart":
    +        # Defense: refuse self-targeting gateway restart from inside the gateway.
    +        # Prevents agent-initiated kill loops when combined with supervisor KeepAlive.
    +        if os.getenv("_HERMES_GATEWAY") == "1":
    +            print_error(
    +                "Refusing to restart the gateway from inside the gateway process.\n"
    +                "This command was blocked to prevent restart loops.\n"
    +                "Use `hermes gateway restart` from a shell outside the running gateway."
    +            )
    +            sys.exit(1)
    +
             # Try service first, fall back to killing and restarting
             service_available = False
             system = getattr(args, 'system', False)
    ```
  - `tests/conftest.py`

    ```diff
    @@ -182,6 +182,7 @@ def _looks_like_credential(name: str) -> bool:
         "HERMES_SESSION_SOURCE",
         "HERMES_SESSION_KEY",
         "HERMES_GATEWAY_SESSION",
    +    "_HERMES_GATEWAY",
         "HERMES_PLATFORM",
         "HERMES_MODEL",
         "HERMES_INFERENCE_MODEL",
    ```
  - `tests/hermes_cli/test_gateway_restart_loop.py`

    ```diff
    @@ -0,0 +1,252 @@
    +"""Tests for gateway restart-loop defenses (#30719).
    +
    +Covers:
    +- Defense 1: gateway stop/restart refuse when _HERMES_GATEWAY=1
    +- Defense 2: cron create rejects prompts containing gateway lifecycle commands
    +- _contains_gateway_lifecycle_command pattern matching
    +"""
    +
    +import os
    +from argparse import Namespace
    +
    +import pytest
    +
    +from hermes_cli.cron import (
    +    _contains_gateway_lifecycle_command,
    +    cron_command,
    +)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Defense 2: _contains_gateway_lifecycle_command pattern tests
    +# ---------------------------------------------------------------------------
    +
    +class TestGatewayLifecyclePattern:
    +    """Verify the regex catches gateway lifecycle commands."""
    +
    +    @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
    +    ])
    +    def test_hermes_gateway_commands(self, text):
    +        assert _contains_gateway_lifecycle_command(text), f"Should match: {text!r}"
    +
    +    @pytest.mark.parametrize("text", [
    +        "launchctl kickstart gui/501/ai.hermes.gateway",
    +        "launchctl unload ~/Library/LaunchAgents/ai.hermes.gateway.plist",
    +        "launchctl stop ai.hermes.gateway",
    +        "systemctl restart hermes-gateway",
    +        "systemctl stop hermes-gateway.service",
    +        "systemctl start hermes-gateway",
    +    ])
    +    def test_service_manager_commands(self, text):
    +        assert _contains_gateway_lifecycle_command(text), f"Should match: {text!r}"
    +
    +    @pytest.mark.parametrize("text", [
    +        "kill hermes gateway process",
    +        "pkill -f hermes.*gateway",
    +    ])
    +    def test_kill_commands(self, text):
    +        assert _contains_gateway_lifecycle_command(text), f"Should match: {text!r}"
    +
    +    @pytest.mark.parametrize("text", [
    +        "restart the server application",
    +        "hermes cron list",
    +        "hermes update",
    +        "hermes config set model claude",
    +        "echo 'just a normal cron job'",
    +        "run the backup script",
    +        "gateway is running fine",
    +        # Regression (#30728 follow-up): legit prompts that merely mention an
    +        # unrelated gateway + a restart must NOT be blocked.
    +        "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",
    +    ])
    +    def test_safe_commands(self, text):
    +        assert not _contains_gateway_lifecycle_command(text), f"Should NOT match: {text!r}"
    +
    +
    +class TestCronCreateLifecycleBlock:
    +    """Verify cron create rejects gateway lifecycle prompts."""
    +
    +    @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_block_hermes_gateway_restart(self, capsys):
    +        args = Namespace(
    +            cron_command="create",
    +            schedule="30m",
    +            prompt="Upgrade hermes then run hermes gateway restart",
    +            name=None,
    +            deliver=None,
    +            repeat=None,
    +            skill=None,
    +            skills=None,
    +            script=None,
    +            workdir=None,
    +            profile=None,
    +            no_agent=False,
    +        )
    +        rc = cron_command(args)
    +        assert rc == 1
    +        out = capsys.readouterr().out
    +        assert "Blocked" in out
    +        assert "#30719" in out
    +
    +    def test_block_launchctl_kickstart(self, capsys):
    +        args = Namespace(
    +            cron_command="create",
    +            schedule="0 9 * * *",
    +            prompt="Run launchctl kickstart -k gui/501/ai.hermes.gateway",
    +            name=None,
    +            deliver=None,
    +            repeat=None,
    +            skill=None,
    +            skills=None,
    +            script=None,
    +            workdir=None,
    +            profile=None,
    +            no_agent=False,
    +        )
    +        rc = cron_command(args)
    +        assert rc == 1
    +        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")
    +        args = Namespace(
    +            cron_command="create",
    +            schedule="1h",
    +            prompt=None,
    +            name=None,
    +            deliver=None,
    +            repeat=None,
    +            skill=None,
    +            skills=None,
    +            script=str(script),
    +            workdir=None,
    +            profile=None,
    +            no_agent=False,
    +        )
    +        rc = cron_command(args)
    +        assert rc == 1
    +        out = capsys.readouterr().out
    +        assert "Blocked" in out
    +
    +    def test_allow_safe_prompt(self, capsys):
    +        args = Namespace(
    +            cron_command="create",
    +            schedule="30m",
    +            prompt="Check server health and report status",
    +            name=None,
    +            deliver=None,
    +            repeat=None,
    +            skill=None,
    +            skills=None,
    +            script=None,
    +            workdir=None,
    +            profile=None,
    +            no_agent=False,
    +        )
    +        rc = cron_command(args)
    +        assert rc == 0
    +        out = capsys.readouterr().out
    +        assert "Created job" in out
    +
    +    def test_allow_empty_prompt(self, capsys):
    +        """Empty prompt (no lifecycle content) should pass the filter — the
    +        API will still reject it for lacking prompt+skill, but that's a
    +        separate validation, not the lifecycle guard."""
    +        args = Namespace(
    +            cron_command="create",
    +            schedule="30m",
    +            prompt=None,
    +            name=None,
    +            deliver=None,
    +            repeat=None,
    +            skill=None,
    +            skills=None,
    +            script=None,
    +            workdir=None,
    +            profile=None,
    +            no_agent=False,
    +        )
    +        rc = cron_command(args)
    +        # The lifecycle guard passes (no gateway command in prompt).
    +        # The API rejects it for "requires prompt or skill" → rc 1, but
    +        # the error message is about prompt/skill, NOT about "Blocked".
    +        out = capsys.readouterr().out
    +        assert "Blocked" not in out
    +
    +
    +# ---------------------------------------------------------------------------
    +# Defense 1: gateway stop/restart refuse inside gateway
    +# ---------------------------------------------------------------------------
    +
    +class TestGatewaySelfTargetingGuard:
    +    """Verify hermes gateway stop/restart refuse when _HERMES_GATEWAY=1."""
    +
    +    def test_stop_refuses_inside_gateway(self, monkeypatch):
    +        monkeypatch.setenv("_HERMES_GATEWAY", "1")
    +        from hermes_cli.gateway import gateway_command
    +        args = Namespace(gateway_command="stop", all=False, system=False)
    +        with pytest.raises(SystemExit) as exc_info:
    +            gateway_command(args)
    +        assert exc_info.value.code == 1
    +
    +    def test_restart_refuses_inside_gateway(self, monkeypatch):
    +        monkeypatch.setenv("_HERMES_GATEWAY", "1")
    +        from hermes_cli.gateway import gateway_command
    +        args = Namespace(gateway_command="restart", all=False, system=False)
    +        with pytest.raises(SystemExit) as exc_info:
    +            gateway_command(args)
    +        assert exc_info.value.code == 1
    +
    +    def test_stop_allows_outside_gateway(self, monkeypatch):
    +        # With the gateway marker unset, the self-targeting guard must NOT
    +        # fire. Prove control reaches the real stop path (rather than driving
    +        # real signal delivery, which would trip the live-system guard) by
    +        # short-circuiting the first downstream call with a sentinel.
    +        monkeypatch.delenv("_HERMES_GATEWAY", raising=False)
    +        import hermes_cli.gateway as gw
    +
    +        class _Reached(Exception):
    +            pass
    +
    +        def _sentinel(*a, **k):
    +            raise _Reached()
    +
    +        monkeypatch.setattr(gw, "_dispatch_via_service_manager_if_s6", _sentinel)
    +        monkeypatch.setattr(gw, "_dispatch_all_via_service_manager_if_s6", _sentinel)
    +        args = Namespace(gateway_command="stop", all=False, system=False)
    +        with pytest.raises(_Reached):
    +            gw.gateway_command(args)
    +
    +    def test_restart_allows_outside_gateway(self, monkeypatch):
    +        # Same as above for restart: guard must not fire when the marker is
    +        # unset. The first thing restart does after the guard is the s6
    +        # dispatch check — sentinel it so we never reach real signal delivery.
    +        monkeypatch.delenv("_HERMES_GATEWAY", raising=False)
    +        import hermes_cli.gateway as gw
    +
    +        class _Reached(Exception):
    +            pass
    +
    +        def _sentinel(*a, **k):
    +            raise _Reached()
    +
    +        monkeypatch.setattr(gw, "_dispatch_via_service_manager_if_s6", _sentinel)
    +        monkeypatch.setattr(gw, "_dispatch_all_via_service_manager_if_s6", _sentinel)
    +        args = Namespace(gateway_command="restart", all=False, system=False)
    +        with pytest.raises(_Reached):
    +            gw.gateway_command(args)
    ```
- **#56177: fix(cron): commit one-shot dispatch before side effect to stop crash re-fire loop (#38758)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `cron/jobs.py`, `cron/scheduler.py`, `tests/cron/test_jobs.py`, `tests/cron/test_scheduler.py`
  - fix hunks:
  - `cron/jobs.py`

    ```diff
    @@ -1249,13 +1249,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)
    ```

    ```diff
    @@ -1300,6 +1314,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.
     
    ```

    ```diff
    @@ -1543,6 +1620,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:
    ```
  - `cron/scheduler.py`

    ```diff
    @@ -236,7 +236,7 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None:
         "QQBOT_HOME_CHANNEL": "QQ_HOME_CHANNEL",
     }
     
    -from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run
    +from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch
     
     # Sentinel: when a cron agent has nothing new to report, it can start its
     # response with this marker to suppress delivery.  Output is still saved
    ```

    ```diff
    @@ -2781,6 +2781,20 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
         failure is recorded via ``mark_job_run``), False only if processing raised.
         """
         try:
    +        # Pre-run dispatch claim (issue #38758): atomically commit a finite
    +        # one-shot's dispatch BEFORE its side effect runs, so a tick that dies
    +        # mid-execution (gateway kill, OOM, segfault, hard-timeout) cannot
    +        # re-fire the job forever on restart. No-op for recurring jobs (they
    +        # use advance_next_run) and infinite/no-repeat jobs. This lives here in
    +        # the shared body so BOTH the built-in ticker and the external provider
    +        # (Chronos fire_due) get at-most-times semantics.
    +        if not claim_dispatch(job["id"]):
    +            logger.info(
    +                "Job '%s': one-shot dispatch limit reached — skipping",
    +                job.get("name", job["id"]),
    +            )
    +            return True  # not an error — already handled/removed
    +
             success, output, final_response, error = run_job(job)
     
             output_file = save_job_output(job["id"], output)
    ```
  - `tests/cron/test_jobs.py`

    ```diff
    @@ -19,6 +19,7 @@
         remove_job,
         mark_job_run,
         advance_next_run,
    +    claim_dispatch,
         get_due_jobs,
         save_job_output,
     )
    ```

    ```diff
    @@ -1314,3 +1315,102 @@ def test_cron_output_keep_defaults_on_bad_config(self, monkeypatch):
                 "hermes_cli.config.load_config", lambda: {"cron": {"output_retention": "oops"}}
             )
             assert jobs._cron_output_keep() == jobs._CRON_OUTPUT_DEFAULT_KEEP
    +
    +
    +# =========================================================================
    +# claim_dispatch — pre-run one-shot crash safety (issue #38758)
    +# =========================================================================
    +
    +class TestClaimDispatch:
    +    """One-shot jobs must commit their dispatch BEFORE the side effect runs, so
    +    a tick that dies mid-execution (gateway kill, OOM, hard-timeout) can re-fire
    +    the job at most ``repeat.times`` times instead of infinitely."""
    +
    +    def _oneshot(self, times=1, completed=0):
    +        return {
    +            "id": "os1",
    +            "name": "one-shot",
    +            "enabled": True,
    +            "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"},
    +            "repeat": {"times": times, "completed": completed},
    +        }
    +
    +    def test_claim_increments_and_persists(self, tmp_cron_dir):
    +        save_jobs([self._oneshot(times=1, completed=0)])
    +        assert claim_dispatch("os1") is True
    +        # Persisted BEFORE any side effect — survives a crash.
    +        assert load_jobs()[0]["repeat"]["completed"] == 1
    +
    +    def test_already_dispatched_oneshot_is_removed(self, tmp_cron_dir):
    +        # A prior tick claimed (completed==times) then died before mark_job_run
    +        # could remove the job.  The next claim must refuse AND clean up.
    +        save_jobs([self._oneshot(times=1, completed=1)])
    +        assert claim_dispatch("os1") is False
    +        assert load_jobs() == []  # removed, will not re-fire
    +
    +    def test_recurring_job_is_not_claimed(self, tmp_cron_dir):
    +        job = {
    +            "id": "rec",
    +            "schedule": {"kind": "interval", "minutes": 5},
    +            "repeat": {"times": 3, "completed": 0},
    +        }
    +        save_jobs([job])
    +        assert claim_dispatch("rec") is True
    +        # Recurring jobs use advance_next_run(); claim must NOT touch completed.
    +        assert load_jobs()[0]["repeat"]["completed"] == 0
    +
    +    def test_infinite_oneshot_not_claimed(self, tmp_cron_dir):
    +        job = self._oneshot(times=0, completed=0)  # times<=0 means infinite
    +        save_jobs([job])
    +        assert claim_dispatch("os1") is True
    +        assert load_jobs()[0]["repeat"]["completed"] == 0
    +
    +    def test_no_repeat_block_not_claimed(self, tmp_cron_dir):
    +        job = {"id": "os1", "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"}}
    +        save_jobs([job])
    +        assert claim_dispatch("os1") is True
    +        assert "repeat" not in load_jobs()[0]
    +
    +    def test_missing_job_proceeds(self, tmp_cron_dir):
    +        # A handed-in job dict not persisted in the store (external provider /
    +        # direct caller) can't be claimed — proceed rather than suppress it.
    +        save_jobs([])
    +        assert claim_dispatch("ghost") is True
    +
    +    def test_mark_job_run_does_not_double_count_preclaimed_oneshot(self, tmp_cron_dir):
    +        # Full lifecycle: claim bumps completed to times, then mark_job_run must
    +        # NOT increment again — it recognizes the pre-claim and removes the job.
    +        save_jobs([self._oneshot(times=1, completed=0)])
    +        assert claim_dispatch("os1") is True
    +        assert load_jobs()[0]["repeat"]["completed"] == 1
    +        mark_job_run("os1", success=True)
    +        assert load_jobs() == []  # completed once, removed — not fired twice
    +
    +    def test_mark_job_run_still_increments_recurring(self, tmp_cron_dir):
    +        # The double-count guard is one-shot-specific; recurring jobs keep the
    +        # legacy post-run increment.
    +        job = {
    +            "id": "rec",
    +            "schedule": {"kind": "interval", "minutes": 5},
    +            "repeat": {"times": 3, "completed": 1},
    +        }
    +        save_jobs([job])
    +        mark_job_run("rec", success=True)
    +        assert load_jobs()[0]["repeat"]["completed"] == 2
    +
    +    def test_get_due_jobs_removes_stale_maxed_oneshot(self, tmp_cron_dir):
    +        # A claimed one-shot whose tick died leaves completed>=times with
    +        # last_run_at still unset, so the recovery helper re-arms it as due.
    +        # get_due_jobs must drop it instead of returning it for another fire.
    +        past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat()
    +        save_jobs([{
    +            "id": "os1",
    +            "name": "one-shot",
    +            "enabled": True,
    +            "schedule": {"kind": "once", "run_at": past},
    +            "repeat": {"times": 1, "completed": 1},
    +            "next_run_at": None,
    +        }])
    +        due = get_due_jobs()
    +        assert due == []
    +        assert load_jobs() == []  # cleaned up
    ```
  - `tests/cron/test_scheduler.py`

    ```diff
    @@ -2542,6 +2542,46 @@ def test_whitespace_only_response_is_marked_failed_not_delivered(self):
             )
     
     
    +class TestOneShotDispatchClaim:
    +    """run_one_job must claim a finite one-shot's dispatch BEFORE run_job so a
    +    tick that dies mid-execution can't re-fire it forever (issue #38758)."""
    +
    +    def _oneshot(self):
    +        return {
    +            "id": "monitor-job",
    +            "name": "monitor",
    +            "deliver": "origin",
    +            "origin": {"platform": "telegram", "chat_id": "123"},
    +            "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"},
    +            "repeat": {"times": 1, "completed": 0},
    +        }
    +
    +    def test_claim_runs_before_run_job(self):
    +        order = []
    +        with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \
    +             patch("cron.scheduler.claim_dispatch", side_effect=lambda _id: order.append("claim") or True), \
    +             patch("cron.scheduler.run_job", side_effect=lambda _j: order.append("run") or (True, "# out", "ok", None)), \
    +             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
    +             patch("cron.scheduler._deliver_result"), \
    +             patch("cron.scheduler.mark_job_run"):
    +            from cron.scheduler import tick
    +            tick(verbose=False)
    +        assert order == ["claim", "run"]  # claim strictly before side effect
    +
    +    def test_refused_claim_skips_run_job(self):
    +        with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \
    +             patch("cron.scheduler.claim_dispatch", return_value=False), \
    +             patch("cron.scheduler.run_job") as run_mock, \
    +             patch("cron.scheduler.save_job_output"), \
    +             patch("cron.scheduler._deliver_result") as deliver_mock, \
    +             patch("cron.scheduler.mark_job_run") as mark_mock:
    +            from cron.scheduler import tick
    +            tick(verbose=False)
    +        run_mock.assert_not_called()
    +        deliver_mock.assert_not_called()
    +        mark_mock.assert_not_called()
    +
    +
     class TestBuildJobPromptSilentHint:
         """Verify _build_job_prompt always injects [SILENT] guidance."""
     
    ```
- **#56240: fix(gateway,cron): guard cron model-tool path + auto-resume loop breaker (#30719)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `cron/jobs.py`, `cron/lifecycle_guard.py`, `gateway/restart_loop_guard.py`, `gateway/run.py`, `hermes_cli/config.py`, `hermes_cli/cron.py`, `tests/hermes_cli/test_gateway_restart_loop.py`
  - fix hunks:
  - `cron/jobs.py`

    ```diff
    @@ -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(
    ```
  - `cron/lifecycle_guard.py`

    ```diff
    @@ -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 ``<HERMES_HOME>/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."
    +        )
    ```
  - `gateway/restart_loop_guard.py`

    ```diff
    @@ -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/<uid>/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 ``<HERMES_HOME>/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
    ```
  - `gateway/run.py`

    ```diff
    @@ -3800,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
    ```

    ```diff
    @@ -5961,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:
    ```
  - `hermes_cli/config.py`

    ```diff
    @@ -2717,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
    ```
  - `hermes_cli/cron.py`

    ```diff
    @@ -6,7 +6,6 @@
     """
     
     import json
    -import re
     import sys
     from pathlib import Path
     from typing import Iterable, List, Optional
    ```

    ```diff
    @@ -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:
    ```

    ```diff
    @@ -301,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,
    ```
  - `tests/hermes_cli/test_gateway_restart_loop.py`

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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}"
    ```

    ```diff
    @@ -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}"
    ```

    ```diff
    @@ -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",
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```
- **#30728: fix(gateway,cron): prevent agent restart loops via self-targeting gateway commands (#30719)**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `#35679`, `#30719`
- **#33395: fix(gateway,cron): prevent agent restart loops — salvage #30728 + correctness follow-ups (#30719)**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#56240`, `#56240`, `#33395`, `#30719`
- **#35815: fix(gateway): replace blanket internal-restart guard with loop detector**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#35668`
- **#59343: fix: prevent no_agent one-shot cron jobs from re-firing on gateway re…**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `84c724d69`, `#56177`

## Fix status

- Merged to main in PR #35679 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-07-01`.
- Subsystem: `cron`