fix(gateway,cron): guard cron model-tool path + auto-resume loop breaker (#30719)

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:

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 ``<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."
+        )