""" Cron job scheduler - executes due jobs. Provides tick() which checks for due jobs and runs them. The gateway calls this every 60 seconds from a background thread. Uses a file-based lock (~/.hermes/cron/.tick.lock) so only one tick runs at a time if multiple processes overlap. """ import asyncio import atexit import concurrent.futures import contextlib import contextvars import errno import json import logging import os import re import shutil import signal import subprocess import sys import threading import time import uuid from datetime import datetime, timezone # fcntl is Unix-only; on Windows use msvcrt for file locking try: import fcntl except ImportError: fcntl = None try: import msvcrt except ImportError: msvcrt = None from pathlib import Path from typing import Any, Callable, List, Optional, Protocol # Add parent directory to path for imports BEFORE repo-level imports. # Without this, standalone invocations (e.g. after `hermes update` reloads # the module) fail with ModuleNotFoundError for hermes_time et al. sys.path.insert(0, str(Path(__file__).parent.parent)) from hermes_constants import get_hermes_home from hermes_cli._subprocess_compat import windows_hide_flags from hermes_cli.config import ( _expand_env_vars, cron_model_drift_axes, cron_model_drift_guard_enabled, load_config, resolve_cron_model_drift_defaults, ) from hermes_cli.fallback_config import get_fallback_chain from hermes_time import now as _hermes_now from agent.interrupt_compat import request_hard_interrupt from agent.delegation_context import ( enter_non_dispatcher_owned_context, exit_non_dispatcher_owned_context, ) logger = logging.getLogger(__name__) def _close_late_session_db_result(future: "concurrent.futures.Future") -> None: """Done-callback: close a SessionDB whose constructor finished after run_job's timeout. When ``run_job``'s SessionDB init times out, the worker thread is abandoned (``shutdown(wait=False)``) so the job can proceed without a session store. If the constructor later completes inside that abandoned worker, the Future's result — an open SessionDB holding .db / WAL / SHM file handles — would be orphaned and never closed, leaking descriptors until EMFILE (#72782). This callback retrieves and closes that eventual late result. """ try: db = future.result() if db is not None: from hermes_state import release_or_close release_or_close(db) except Exception: pass def _set_cron_session_title(session_db, session_id, base_title): """Robustly title a finished cron session before it is closed. Centralizes the title write so the cron finally block can guarantee a non-blank, unique title is persisted before end_session()/close() tear the connection down (issues #50535, #50536, #50537): - #50535: never leaves the session blank. base_title already carries a cron-id fallback for nameless jobs; this also guards a failed write. - #50537: a duplicate title makes set_session_title raise ValueError (the unique-title index). Recover by appending a #N suffix via get_next_title_in_lineage() when supported, instead of swallowing the error and ending up untitled. If lineage dedup is unavailable, raise. - #50536: this runs synchronously in the cron finally block ahead of the session close, so no in-flight title write can race the close. Returns the title actually persisted, or None if nothing could be set. """ if not session_db or not session_id: return None title = (base_title or "").strip() if not title: return None try: session_db.set_session_title(session_id, title) return title except ValueError: # Title collision against the unique-title index. Fall back to the # next title in the lineage (base #2, base #3, ...) when supported. next_title_fn = getattr(session_db, "get_next_title_in_lineage", None) if next_title_fn is None: raise deduped = next_title_fn(title) if not deduped or deduped == title: raise session_db.set_session_title(session_id, deduped) return deduped def _fallback_chain_phrase() -> str: """Wording for the fallback-chain clause of a provider-failure message. "Fallback chain was exhausted or unavailable." used to fire unconditionally on every provider failure, which implies a fallback was attempted and failed. Most installs have fallback_providers: [] (no chain configured at all), so that wording was actively misleading: it sent the operator looking for why a fallback "failed" when none was ever attempted. Distinguish the two cases explicitly. Fails open to the original ambiguous-but-safe wording if config can't be read (e.g. mid-shutdown, permissions) -- never let a lookup error crash failure-message generation itself. """ try: cfg = load_config() or {} chain = get_fallback_chain(cfg) except Exception: return "Fallback chain was exhausted or unavailable." if chain: return "Fallback chain was exhausted or unavailable." return ( "No fallback chain configured — add one with `hermes fallback add`, " "or set a cron fleet default via `cron.model` + `cron.model_provider` " "in config.yaml." ) def _failure_streak_nudge(job: dict) -> str: """Return a review nudge when a recurring job keeps failing, else "". Inspired by Poke (poke.com), which "encourages users to review recurring automations that haven't been acted upon": once a recurring job has failed several runs in a row, the per-run failure ping stops being information and starts being noise — the useful message is "this automation needs your attention (fix, pause, or remove it)". The streak counter (``failure_streak``) is persisted by ``cron.jobs.mark_job_run`` and reset on any successful run. Because the failure message is delivered BEFORE ``mark_job_run`` records this run, the prospective streak for the current failure is stored+1. Threshold config: ``cron.failure_nudge_threshold`` (default 3, ``0`` disables the nudge). One-shot jobs never nudge — they don't recur. """ schedule_kind = (job.get("schedule") or {}).get("kind") if schedule_kind not in {"cron", "interval"}: return "" try: cfg = load_config() or {} threshold = int( ((cfg.get("cron") or {}) if isinstance(cfg, dict) else {}).get( "failure_nudge_threshold", 3 ) ) except Exception: threshold = 3 if threshold <= 0: return "" streak = int(job.get("failure_streak") or 0) + 1 # +1 = this run if streak < threshold: return "" job_ref = job.get("name") or job.get("id") or "this job" return ( f"\nThis job has failed {streak} runs in a row — worth a review. " f"Fix its prompt/config, or pause it with `hermes cron pause {job_ref}` " "(resume/remove also available) to stop the noise." ) def _detect_gateway_code_skew() -> tuple[str, str] | None: """Boot-vs-disk revision skew for THIS process, or None. Thin wrapper over ``gateway.code_skew.detect_code_skew`` so the failure summarizer stays a pure function under test (monkeypatch this seam) and a broken import can never take the delivery path down with it. """ try: from gateway.code_skew import detect_code_skew return detect_code_skew() except Exception: return None class CronTickYielded(RuntimeError): """A stale-code ticker yielded this tick to a fresh gateway. Raised by ``tick()`` BEFORE the tick lock is acquired when the process is provably running stale code (boot fingerprint ≠ disk), it does NOT own the gateway runtime lock, and that lock is held by another (fresh) process. Fresh code picks the job up within one tick interval, so the stale process must stay out of the dispatch race entirely — including lock contention, which would otherwise starve the fresh ticker on a busy minute. Raised instead of returned so the provider loops (``cron/scheduler_provider.py``) record it via ``record_ticker_error`` and mark the heartbeat ``success=False``: a yielded tick is NOT a healthy tick (``hermes cron status`` must not show green while jobs only fire from the other process). Liveness stays visible — the loop keeps beating and keeps yielding; if the fresh gateway dies, its lock releases and the stale ticker's next tick proceeds normally (self-healing, no restart needed). Skew detection returning ``None`` (non-git install, no boot fingerprint — e.g. a one-shot CLI tick, or any probe failure) never yields: yield only on certainty, fail open otherwise. """ def __init__(self, boot_rev: str, disk_rev: str) -> None: self.boot_rev = boot_rev self.disk_rev = disk_rev super().__init__( f"Cron tick yielded to a fresh gateway process (stale code: " f"booted on {boot_rev}, disk is at {disk_rev})" ) # Log the yield at most once per episode: a stale ticker that keeps yielding # for hours must not spam the error log every interval. Reset when the # condition clears (proceeds without yielding) or the skew changes. _YIELD_LOG_INTERVAL_SECONDS = 3600.0 _last_yield_log: dict[str, object] = {} def _should_yield_tick_to_fresh_gateway() -> tuple[str, str] | None: """Decide whether this tick must yield to a fresher gateway process. Returns the ``(boot_rev, disk_rev)`` skew labels when ALL of: this process has a boot fingerprint that differs from the checkout on disk (code skew), it does not own the gateway runtime lock, and some other process currently holds that lock — i.e. a fresh gateway is alive and will dispatch due jobs itself. Returns ``None`` otherwise. Every probe failure returns ``None``: the gateway-status import, the lock probe, and skew detection are each individually fail-open. Yielding is a certainty claim, never a guess. """ skew = _detect_gateway_code_skew() if skew is None: return None try: from gateway import status as _gateway_status except Exception: return None try: if _gateway_status.owns_gateway_runtime_lock(): return None if not _gateway_status.is_gateway_runtime_lock_active(): return None except Exception: return None return skew def _log_tick_yield_once(reason: str) -> None: """Log the yield at error level once per episode (skew signature).""" global _last_yield_log now = time.monotonic() last_reason = _last_yield_log.get("reason") last_at = _last_yield_log.get("at", 0.0) if last_reason != reason or (now - float(last_at)) >= _YIELD_LOG_INTERVAL_SECONDS: logger.error( "Cron tick yielded: this process is running stale code (%s) and a " "fresher gateway owns the runtime lock — jobs will fire from that " "process. Restart this one to reclaim its ticks.", reason, ) _last_yield_log = {"reason": reason, "at": now} def _summarize_cron_failure_for_delivery(job: dict, error: str | None) -> str: """Return a compact one-line failure message for chat delivery. Full details stay in the cron output directory and the logs. Chat should show the operator what broke without dumping provider JSON, retry noise, or stack traces into the delivery channel. """ job_name = job.get("name") or job.get("id") or "cron job" text = (error or "unknown error").strip() lower = text.lower() if "skipped to prevent unintended spend: global inference config drifted" in lower: if "finite one-shot job is consumed" in lower: remediation = ( "This finite one-shot is consumed; create a new one-shot job at " "a future time with an explicit provider and model." ) else: job_id = job.get("id") or "" remediation = ( "On the host running Hermes, pin it explicitly: " f"`hermes cron edit {job_id} --provider " "--model `." ) return ( f"⚠️ Cron '{job_name}' skipped before inference to prevent " f"unintended spend. {remediation}" ) # A no_agent job IS its script — run_job short-circuits it before any model # is reached ("no LLM involvement", see the no_agent branch in run_job). So # provider timeouts, rate limits, auth errors and fallback chains are not # merely unlikely for these jobs, they are structurally impossible. Classify # on the job's MODE before pattern-matching its prose. # # Without this gate the branches below classify by substring, so a script's # own wording decides which subsystem gets blamed. _run_job_script reports a # timeout as "Script timed out after {n}s: {path}" — that contains "timed # out", so it matched the provider branch and the operator was told # "provider timeout. Fallback chain was exhausted or unavailable." for a job # that never opened a socket. "429" or "authentication" appearing anywhere # in a script's output misfires the same way. # # A delivery line that names the wrong subsystem is worse than no line at # all: it does not merely fail to inform, it sends the reader to the wrong # place. # # Falling through leaves the generic cleaner below to report what actually # happened, naming the script. No new message text is needed. provider_reachable = not job.get("no_agent") # Script execution happens outside the LLM/provider path (also for # agent-backed jobs that run a context script). Check the script runner's # explicit error contract ("Script timed out after {n}s: {path}") before # generic timeout matching so a script timeout never claims a provider # fallback was attempted (#82460 @jbagdonas, #78503 @daxro). if lower.startswith("script timed out"): return ( f"⚠️ Cron '{job_name}' failed: script timed out. " "No model was invoked. Full details saved in cron output." ) # Provider/API failures are the common noisy path. Keep these short. # Match 429 as a whole token (#83188 @cation98): bare substring matching # let identifiers containing those digits (job ids, ports, hashes) trip # a false "provider rate limit" alert. if provider_reachable and ( re.search(r"\b429\b", text) or "rate limit" in lower or "usage limit" in lower ): reason = "rate limit" if "weekly usage limit" in lower: reason = "weekly usage limit" elif "quota" in lower: reason = "quota limit" return ( f"⚠️ Cron '{job_name}' failed: provider {reason}. " f"{_fallback_chain_phrase()} " "Full details saved in cron output." ) # The scheduler's own inactivity watchdog (see the TimeoutError raised # above at "Cron job '{job_name}' idle for {secs}s (limit {limit}s) — # last activity: {desc}") produces a message that contains the substring # "timed out"/"timeout" nowhere, but DOES contain "idle for ... (limit # ...)" — however older/other call sites can still phrase an inactivity # abort using "timed out" wording, so match on the "idle for Ns (limit" # shape specifically (case-insensitive) BEFORE the generic provider- # timeout branch below. Without this, an inactivity timeout — the job's # OWN tool call/turn going quiet, no provider or fallback chain ever # involved — gets rewritten into a misleading "provider timeout / # fallback chain exhausted" message, sending the operator to debug the # wrong system entirely (field-reported: a stuck `terminal` tool call # tripped the 600s inactivity limit and was reported as a # provider/fallback failure). Mirrors the same reordering fix # upstream issue #59549 applied for script timeouts vs provider timeouts # — check the more specific, deterministic signature first. if re.search(r"idle for \d+s\s*\(limit \d+s\)", lower): return ( f"⚠️ Cron '{job_name}' failed: the job itself stalled — no tool/API " "activity for the configured inactivity window. Not a provider or " "fallback-chain issue; check what the job was doing when it went " "quiet. Full details saved in cron output." ) if provider_reachable and ( "readtimeout" in lower or "timed out" in lower or "timeout" in lower ): return ( f"⚠️ Cron '{job_name}' failed: provider timeout. " f"{_fallback_chain_phrase()} " "Full details saved in cron output." ) # Match authentication/authorization wording at a word boundary and the # 401/403 status codes as whole tokens, so "oauth", "4015" and similar do # not trip a misleading auth message. if provider_reachable and ( re.search(r"authenticat|authoriz", lower) or re.search(r"\b(401|403)\b", text) ): return ( f"⚠️ Cron '{job_name}' failed: provider authentication error. " "Full details saved in cron output." ) # Strip common exception wrappers and collapse provider payloads. Bound # the input first so a multi-KB provider blob cannot slow the # substitutions. cleaned = re.sub( r"^(RuntimeError|Exception|ValueError|HTTPStatusError):\s*", "", text[:2000], ) cleaned = re.sub(r"\s+", " ", cleaned).strip() if len(cleaned) > 180: cleaned = cleaned[:177].rstrip() + "..." message = f"⚠️ Cron '{job_name}' failed: {cleaned}" # Import-class failures (#95294 part 3): a long-lived gateway whose # checkout was updated underneath it (interrupted `hermes update`, manual # git pull) serves MIXED modules — old entries frozen in sys.modules, # new files loaded by lazy imports — and every agent cron job then dies # with `cannot import name X` / ModuleNotFoundError. The error itself # reads like a code bug, so operators debug the wrong thing (2 days on # the reporting incident, 15 missed jobs). This process knows its own # boot fingerprint: when boot SHA differs from disk HEAD, APPEND the # cause and the one-command fix — never replacing the raw error text, # which carries the failing symbol name. # # Fail-safe by construction: skew detection returns None on non-git # installs and in processes without a boot fingerprint (no false # accusations — message delivered unchanged), the probe seam swallows # every exception, and no_agent script jobs are excluded via the same # mode-gate as the provider branches (a fresh subprocess resolves # imports consistently against disk; its ImportError is the script's # own problem, and blaming gateway skew would send the reader to the # wrong place). if provider_reachable and re.search( r"cannot import name|modulenotfounderror|importerror", lower ): try: skew = _detect_gateway_code_skew() except Exception: skew = None # delivery must never die on a diagnostics probe if skew is not None: boot_rev, disk_rev = skew message += ( f" Likely cause: the gateway is running stale code (booted " f"on {boot_rev}, disk is at {disk_rev}) — run " "`hermes gateway restart` to fix it." ) return message def _upsert_incident_for_failure( job: dict, error: str, *, output_file: Optional[Any] = None ) -> tuple[bool, Optional[str]]: """Record a durable failure incident for this run. The incident store groups "same job + same error signature" across runs so an operator-acked failure stops re-pinging every run. Returns ``(acked, incident_id)``: ``acked`` is True when the incident for this exact signature is already ``closed`` (acked) — the per-run failure ping should be suppressed. ``incident_id`` lets the caller mark the incident ``alerted`` after the ping actually goes out. The streak nudge and ``_summarize_cron_failure_for_delivery`` text stay intact for un-acked failures. Best-effort: an incident-store error must never break the cron run or the delivery path — failures are logged at debug and the caller delivers as if no incident existed. """ try: from cron.incidents import get_incident, upsert_incident incident_id, _is_new = upsert_incident( job["id"], str(error or ""), job_name=job.get("name"), output_file=output_file, ) incident = get_incident(incident_id) acked = bool(incident and incident.get("state") == "closed") return acked, incident_id except Exception as exc: logger.debug( "Incident store unavailable for job %s (delivery unaffected): %s", job["id"], exc, ) return False, None def _mark_incident_alerted(incident_id: Optional[str]) -> None: """Record that a failure ping for this incident reached delivery. Best-effort like the upsert: bookkeeping never breaks the cron run. ``set_incident_state`` is a no-op for closed incidents, so this can never resurrect an acked signature. """ if not incident_id: return try: from cron.incidents import set_incident_state set_incident_state(incident_id, "alerted") except Exception as exc: logger.debug("Failed marking incident %s alerted: %s", incident_id, exc) class CronPromptInjectionBlocked(Exception): """Raised by _build_job_prompt when the fully-assembled prompt trips the injection scanner. Caught in run_job so the operator sees a clean "job blocked" delivery instead of the scheduler crashing. Assembled-prompt scanning (including loaded skill content) plugs the gap from #3968: create-time scanning only covers the user-supplied prompt field; skill content loaded at runtime was never scanned, so a malicious skill could carry an injection payload that reached the non-interactive (auto-approve) cron agent. """ def _resolve_cron_disabled_toolsets(cfg: dict) -> list[str]: """Toolsets a cron-spawned agent must never receive. Two toolsets are always disabled in cron context regardless of config: - ``messaging`` — interactive, needs a live gateway session - ``clarify`` — interactive, blocks waiting for user input ``cronjob`` is policy-denied by default (loop prevention, not a security boundary) and config-gated: setting ``cron.allow_agent_scheduling: true`` in config.yaml drops it from the base denylist so cron-spawned agents may manage the user's cron table. The gate only removes the built-in policy denial — it never overrides the user denylist below. User-level ``agent.disabled_toolsets`` from config.yaml is layered on top so per-job ``enabled_toolsets`` cannot bypass policy that applies to ordinary agent runs (#25752 — LLM-supplied enabled_toolsets was widening past config.yaml's denylist). """ cron_cfg = (cfg or {}).get("cron") or {} if cron_cfg.get("allow_agent_scheduling"): disabled = ["messaging", "clarify"] else: disabled = ["cronjob", "messaging", "clarify"] agent_cfg = (cfg or {}).get("agent") or {} from agent.skill_utils import parse_config_string_list user_disabled = parse_config_string_list(agent_cfg.get("disabled_toolsets")) for name in user_disabled: name = str(name).strip() if name and name not in disabled: disabled.append(name) return disabled def _merge_mcp_into_per_job_toolsets(per_job: list[str], cfg: dict) -> list[str]: """Layer enabled MCP servers onto a per-job ``enabled_toolsets`` allowlist. A per-job list scopes the *native* toolsets, but on its own it silently drops every MCP server: ``discover_mcp_tools()`` registers the tools into the global registry, yet ``get_tool_definitions(enabled_toolsets=...)`` only keeps toolsets named in the list. The agent then rejects every ``mcp_*`` call with "Unknown tool". This restores parity with ``_get_platform_tools`` MCP semantics: * ``no_mcp`` sentinel present -> no MCP servers (sentinel stripped) * one or more MCP server names already listed -> treat as an allowlist, add nothing further (the user named exactly the servers they want) * otherwise -> union in every globally-enabled MCP server """ result = [t for t in per_job if t != "no_mcp"] if "no_mcp" in per_job: return result # lazy import: avoid heavy hermes_cli import at cron module load (matches # _resolve_cron_enabled_toolsets' fallback) and share one MCP-membership # computation with the gateway/CLI platform resolver. from hermes_cli.tools_config import enabled_mcp_server_names enabled_mcp = enabled_mcp_server_names(cfg) if set(result) & enabled_mcp: return result for name in sorted(enabled_mcp): if name not in result: result.append(name) return result def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None: """Resolve the toolset list for a cron job. Precedence: 1. Per-job ``enabled_toolsets`` (set via ``cronjob`` tool on create/update). Keeps the agent's job-scoped toolset override intact — #6130. Enabled MCP servers are layered on per ``_merge_mcp_into_per_job_toolsets`` so a native-toolset allowlist does not silently strip MCP tools. 2. Per-platform ``hermes tools`` config for the ``cron`` platform. Mirrors gateway behavior (``_get_platform_tools(cfg, platform_key)``) so users can gate cron toolsets globally without recreating every job. 3. ``None`` on any lookup failure — AIAgent loads the full default set (legacy behavior before this change, preserved as the safety net). _DEFAULT_OFF_TOOLSETS ({moa, homeassistant, rl}) are removed by ``_get_platform_tools`` for unconfigured platforms, so fresh installs get cron WITHOUT ``moa`` by default (issue reported by Norbert — surprise $4.63 run). """ per_job = job.get("enabled_toolsets") if per_job: return _merge_mcp_into_per_job_toolsets(list(per_job), cfg or {}) try: from hermes_cli.tools_config import _get_platform_tools # lazy: avoid heavy import at cron module load return sorted(_get_platform_tools(cfg or {}, "cron")) except Exception as exc: logger.warning( "Cron toolset resolution failed, falling back to full default toolset: %s", exc, ) return None def _resolve_job_reasoning_config(job: dict, cfg: dict, model: str) -> dict | None: """Resolve the effective reasoning config for a cron run. Precedence: per-job ``reasoning_effort`` pin (validated at the store choke point, ``cron/jobs.py::_normalize_reasoning_effort``) wins outright over config resolution — both the global ``agent.reasoning_effort`` and per-model ``agent.reasoning_overrides``. The pin is model-independent by design: it also governs an auth-fallback model swap, and capability clamping for the model that actually runs stays owned by the provider transports at send time (exactly like config-set effort). A value that no longer parses (hand-edited jobs.json) logs a warning and falls back to config resolution — a bad pin must degrade the run's thinking level, never kill the tick. Absent/None pin returns ``resolve_reasoning_config(cfg, model)`` byte-identical, preserving pre-feature behavior. """ from hermes_constants import parse_reasoning_effort, resolve_reasoning_config pinned = job.get("reasoning_effort") if pinned is not None: parsed = parse_reasoning_effort(pinned) if parsed is not None: logger.info( "Job '%s': using per-job reasoning_effort '%s'", job.get("id", "?"), pinned, ) return parsed logger.warning( "Job '%s': invalid stored reasoning_effort %r — ignoring the pin " "and falling back to config resolution. Fix with `cronjob " "action=update job_id=%s reasoning_effort=` (valid: none, " "minimal, low, medium, high, xhigh, max, ultra).", job.get("id", "?"), pinned, job.get("id", "?"), ) return resolve_reasoning_config(cfg if isinstance(cfg, dict) else {}, str(model)) # Valid delivery platforms — used to validate user-supplied platform names # in cron delivery targets, preventing env var enumeration via crafted names. _KNOWN_DELIVERY_PLATFORMS = frozenset({ "telegram", "discord", "slack", "whatsapp", "signal", "matrix", "mattermost", "homeassistant", "dingtalk", "feishu", "wecom", "wecom_callback", "weixin", "sms", "email", "webhook", "bluebubbles", "qqbot", "yuanbao", }) # Platforms that support a configured cron/notification home target, mapped to # the environment variable used by gateway setup/runtime config. _HOME_TARGET_ENV_VARS = { "matrix": "MATRIX_HOME_ROOM", "telegram": "TELEGRAM_HOME_CHANNEL", "discord": "DISCORD_HOME_CHANNEL", "slack": "SLACK_HOME_CHANNEL", "signal": "SIGNAL_HOME_CHANNEL", "mattermost": "MATTERMOST_HOME_CHANNEL", "sms": "SMS_HOME_CHANNEL", "email": "EMAIL_HOME_ADDRESS", "dingtalk": "DINGTALK_HOME_CHANNEL", "feishu": "FEISHU_HOME_CHANNEL", "wecom": "WECOM_HOME_CHANNEL", "weixin": "WEIXIN_HOME_CHANNEL", "bluebubbles": "BLUEBUBBLES_HOME_CHANNEL", "qqbot": "QQBOT_HOME_CHANNEL", "whatsapp": "WHATSAPP_HOME_CHANNEL", "whatsapp_cloud": "WHATSAPP_CLOUD_HOME_CHANNEL", } # Legacy env var names kept for back-compat. Each entry is the current # primary env var → the previous name. _get_home_target_chat_id falls # back to the legacy name if the primary is unset, so users who set the # old name before the rename keep working until they migrate. _LEGACY_HOME_TARGET_ENV_VARS = { "QQBOT_HOME_CHANNEL": "QQ_HOME_CHANNEL", } from cron.jobs import ( _ensure_cron_dir, advance_next_runs, claim_dispatch, claim_job_for_fire, fire_claim_fence, clear_run_claim, get_due_jobs, heartbeat_fire_claim, heartbeat_run_claim, mark_job_run, save_job_output, use_cron_store, ) from cron.executions import ( _TERMINAL_STATES, create_execution, finish_execution, get_execution, mark_execution_handoff_pending, mark_execution_running, recover_interrupted_executions, ) # 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 # locally for audit. SILENT_MARKER = "[SILENT]" # Canonical silence tokens recognized in cron output. Cron's contract is # intentionally looser than the gateway's exact-whole-response rule: the cron # system prompt *instructs* the agent to emit "[SILENT]", and real agents often # bracket it with a short note or trailing newline. We therefore suppress when # a marker is the entire response OR appears as its own first/last line — but # NOT when a token merely appears mid-sentence in a genuine report (e.g. # "I considered staying [SILENT] but here is the summary…" must deliver). # The actual matcher is shared with the webhook lane — # gateway.response_filters.is_autonomous_silence_response — so the two # autonomous lanes cannot drift apart. def _is_cron_silence_response(text: str) -> bool: """Return True when a cron final response should suppress delivery. Recognizes the bracketed ``[SILENT]`` sentinel (whole-response, first line, or last line) plus the bracketless ``SILENT`` / ``NO_REPLY`` / ``NO REPLY`` variants the model emits when it drops the brackets (#51438, #46917). Whitespace-trimmed and case-insensitive. A token buried mid-sentence is treated as real content and delivered. Delegates to the shared autonomous-lane matcher in :mod:`gateway.response_filters` (also used by the webhook adapter). """ from gateway.response_filters import is_autonomous_silence_response return is_autonomous_silence_response(text) # --------------------------------------------------------------------------- # Persistent thread pool for parallel cron jobs. # The tick function submits jobs here and returns immediately so the ticker # thread is never blocked by long-running jobs (e.g. the fixer running 15+ min). # --------------------------------------------------------------------------- _parallel_pool: Optional[concurrent.futures.ThreadPoolExecutor] = None _parallel_pool_max_workers: Optional[int] = None _running_job_ids: set = set() _running_fire_owners: dict[str, dict[object, tuple[Optional[str], Path]]] = {} # Parent gateway threads synchronously waiting on restart-safe scope workers. # Shutdown must not misclassify these as ownerless in-process runs: the tool # process sweep cannot reach the worker's transient scope. _restart_safe_waiter_job_ids: set[str] = set() _running_lock = threading.Lock() # Wall-clock (time.time()) instant each in-flight job id was claimed by # ``_submit_with_guard``, plus the future that owns its release (a pending # sentinel until ``pool.submit`` returns). Together these bound the # in-flight set: an id whose claim is older than its allowance AND has no # live future can only be a leak — the release path never ran — so the # stale-sweep force-releases it instead of letting every later tick # short-circuit on "already running" until the whole gateway process # restarts (incident: jarvis board-pm-triage-* jobs, 2026-08-02; recurring # router/watchdog no_agent jobs, 2026-08-14 t_20e23f84). _running_since: dict = {} _running_futures: dict = {} # Sentinel installed in ``_running_futures`` at claim time, before # ``pool.submit`` has returned a real future. This closes the race the # stale sweep previously had: a sweep landing between the claim critical # section and the future-record section saw ``missing`` and could (in # principle) release a claim that was about to get its future. With the # sentinel there is never a window where a claim has neither an age nor a # future marker — it is ``_FUTURE_PENDING`` until the real future lands. _FUTURE_PENDING = object() # Countable signal for unified-health: how many stale claims this process has # force-released, and the most recent ones. Exposed via # ``get_inflight_guard_stats()`` and mirrored to a JSONL under the cron dir so # an out-of-process probe can catch a wedge in-cycle. _forced_release_count: int = 0 _forced_releases: list = [] _FORCED_RELEASE_HISTORY = 20 # Floor for the stale allowance, in minutes. Effective allowance per job is # max(2 * interval, this) so a slow-but-healthy hourly job is never clipped. _INFLIGHT_MIN_ALLOWANCE_MINUTES = 30.0 # Execution tokens (``object()`` identity keys from ``_running_fire_owners``) # of runs the shutdown path force-interrupted — see # ``mark_running_jobs_interrupted`` below. ``run_one_job``'s own completion # path checks its OWN token before writing ``last_status`` so a cron agent # thread that keeps running in-process after its tool was killed out from # under it — and produces a plausible-looking final response from truncated # output — can never overwrite the interrupted status with a false "ok" # (#60432). Token keying keeps an interruption scoped to that exact # execution: a later run of the same job ID (recurring jobs reuse the ID # every fire) must not inherit the stale flag. Legacy dispatch paths without # a registered fire owner fall back to storing the bare job ID. _interrupted_job_ids: set = set() class _CancelEventLike(Protocol): """Structural type for cancellation sources (``threading.Event`` and ``_CombinedCancelEvent`` both satisfy it).""" def is_set(self) -> bool: ... def set(self) -> None: ... class _CombinedCancelEvent: """Duck-typed ``threading.Event`` that ORs several cancellation sources. ``run_one_job`` already derives a ``lost_ownership`` event from the fire-claim heartbeat; transports (dashboard webhook drain, API server shutdown) contribute their own per-task event. The worker only ever calls ``is_set()`` / ``set()``, so a tiny wrapper beats a pump thread. """ def __init__(self, *events: Optional["_CancelEventLike"]) -> None: self._events = [event for event in events if event is not None] def is_set(self) -> bool: return any(event.is_set() for event in self._events) def set(self) -> None: for event in self._events: event.set() def get_running_job_ids() -> "frozenset[str]": """Thread-safe snapshot of cron job IDs currently executing. A job ID is a member from the moment ``_submit_with_guard`` dispatches it onto the parallel/sequential pool until ``_process_job`` returns — i.e. for the job's *entire* run, tool calls included, not just the ticker's dispatch instant. The gateway shutdown path (``gateway/run.py::GatewayRunner. _drain_active_agents``) reads this to treat in-flight cron work as active the same way it already treats in-flight chat sessions via ``_running_agents`` — cron jobs run through their own thread pool here, entirely outside that dict, so without this the drain is structurally blind to them (#60432). """ with _running_lock: return frozenset(_running_job_ids | _running_fire_owners.keys()) def try_register_running_job(job_id: str) -> bool: """Atomically add ``job_id`` to the in-flight running set. Returns False (without registering) when the job is already mid-run — the caller must skip the fire. This is the single dedupe owner shared by the ticker's ``_submit_with_guard`` and manual runs (``tools/cronjob_tools``): the fire claim alone cannot prevent a double-fire because its TTL (300s) is routinely outlived by real jobs, after which a manual ``cronjob(action='run')`` would claim successfully and run the same job concurrently (idea from #53395 by @izumi0uu). Registration also makes the run visible to ``get_running_job_ids`` (the gateway shutdown drain, #60432) and ``mark_running_jobs_interrupted``. Callers MUST pair a successful registration with ``release_running_job`` in a ``finally`` block. """ with _running_lock: if job_id in _running_job_ids: return False _running_job_ids.add(job_id) # Claim timestamp + pending-future sentinel are recorded in the SAME # critical section as the add, so there is never a window where an # id is in-flight without an age the stale sweep can bound it by # (t_3778a491). The sentinel is replaced by the real owning future # once ``pool.submit`` returns. _running_since[job_id] = time.time() _running_futures[job_id] = _FUTURE_PENDING return True def release_running_job(job_id: str) -> None: """Remove ``job_id`` from the in-flight running set (idempotent).""" with _running_lock: _running_job_ids.discard(job_id) _running_since.pop(job_id, None) _running_futures.pop(job_id, None) def _inflight_min_allowance_minutes() -> float: """Floor for the stale in-flight allowance, in minutes. Effective allowance per job is ``max(2 * interval, this)``, so a slow-but-healthy long-interval job is never clipped by the sweep. Reads ``cron.inflight_max_minutes`` from config.yaml; the ``HERMES_CRON_INFLIGHT_MAX_MINUTES`` env var is kept as an internal escape hatch only. """ try: _ucfg = load_config() or {} _cfg_val = ( _ucfg.get("cron", {}) if isinstance(_ucfg, dict) else {} ).get("inflight_max_minutes") if _cfg_val is not None: val = float(_cfg_val) if val > 0: return val except Exception: pass raw = os.getenv("HERMES_CRON_INFLIGHT_MAX_MINUTES", "").strip() if raw: try: val = float(raw) if val > 0: return val except (ValueError, TypeError): logger.warning( "Invalid HERMES_CRON_INFLIGHT_MAX_MINUTES=%r; using default %s", raw, _INFLIGHT_MIN_ALLOWANCE_MINUTES, ) return _INFLIGHT_MIN_ALLOWANCE_MINUTES # Cache for cron expression interval computation (expression → minutes). # A cron expression's cadence never changes, so computing it once per expr # avoids repeated croniter evaluation on every 60s tick. _cron_interval_cache: dict = {} def _cron_interval_minutes(expr: str) -> Optional[float]: """Approximate the natural interval of a cron expression, in minutes. The persisted job store keeps ``schedule`` as an already-parsed dict (``{"kind": "cron", "expr": "0 9 * * 1"}``), so the stale allowance for a cron job cannot be derived from a schedule *string* — it must come from the expression itself. We measure the gap between the next two fire times with croniter; that is the job's cadence, and the sweep's allowance becomes ``max(2 * cadence, floor)`` exactly like interval jobs. Falls back to ``None`` (→ floor allowance) if croniter is missing or the expression cannot be evaluated. """ if expr in _cron_interval_cache: return _cron_interval_cache[expr] result = None try: from cron.jobs import _ensure_croniter if _ensure_croniter(): from cron.jobs import croniter as _croniter from datetime import datetime base = datetime.now() it = _croniter(expr, base) first = it.get_next(datetime) second = it.get_next(datetime) gap = (second - first).total_seconds() / 60.0 result = gap if gap > 0 else None except Exception: pass _cron_interval_cache[expr] = result return result def _job_interval_minutes(job: dict) -> Optional[float]: """Best-effort interval length for a job, in minutes (None if unknown). Reads the PERSISTED schedule shape first: the job store keeps ``schedule`` as an already-parsed dict (``{"kind": "interval", "minutes": N}`` or ``{"kind": "cron", "expr": "..."}``), NOT the string form that ``parse_schedule`` consumes. The string path is kept only as a defensive fallback for programmatic callers that still build string schedules (and for tests that exercise that shape). ``kind == "once"`` (one-shot) has no recurring interval — returns None, so the sweep uses the documented floor allowance. """ try: schedule = job.get("schedule") if isinstance(schedule, str) and schedule.strip(): from cron.jobs import parse_schedule schedule = parse_schedule(schedule) or {} if isinstance(schedule, dict): kind = schedule.get("kind") if kind == "interval": minutes = schedule.get("minutes") return float(minutes) if minutes else None if kind == "cron": return _cron_interval_minutes(str(schedule.get("expr") or "")) except Exception: pass return None def get_inflight_guard_stats() -> dict: """Probe-visible snapshot of the in-flight guard. ``forced_releases`` is a monotonic counter of stale claims this process has force-released; any non-zero value means a cron job wedged and was recovered without a gateway restart. """ now = time.time() with _running_lock: return { "running": sorted(_running_job_ids), "running_ages_seconds": { jid: round(now - started, 1) for jid, started in _running_since.items() }, "forced_releases": _forced_release_count, "recent_forced_releases": list(_forced_releases), } def _record_forced_release(job_id: str, name: str, age_seconds: float, allowance_seconds: float) -> None: """Persist a countable signal for one forced release (best-effort).""" entry = { "job_id": job_id, "name": name, "age_seconds": round(age_seconds, 1), "allowance_seconds": round(allowance_seconds, 1), "at": _hermes_now().isoformat(), } with _running_lock: _forced_releases.append(entry) del _forced_releases[:-_FORCED_RELEASE_HISTORY] try: path = _get_hermes_home() / "cron" / "inflight_forced_releases.jsonl" _ensure_cron_dir(path.parent) with open(path, "a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n") except Exception as e: # never let telemetry break a tick logger.debug("Could not append forced-release record: %s", e) def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list: """Force-release in-flight claims that can no longer be making progress. A claim is stale when it is older than ``max(2 * interval, floor)`` AND either has no live future at all (the wedge class: the claim was taken but the release path was never installed — e.g. a hang in the submit path before ``pool.submit`` returned) or has a future that already finished without discarding the id. Every release logs a WARNING with the countable ``event=forced_release`` signal, bumps a probe-visible counter (``get_inflight_guard_stats()``), mirrors a JSONL row under the cron dir, and writes ``last_error`` on the job so the wedge surfaces on the job row instead of being invisible until a downstream liveness key goes dead hours later. A forced release never consumes a finite-repeat job's budget (see below). Returns the list of released job ids. """ global _forced_release_count by_id = {j.get("id"): j for j in (due_jobs or []) if isinstance(j, dict)} floor_seconds = _inflight_min_allowance_minutes() * 60.0 now = time.time() stale: list = [] # Latest durable execution per RELEASABLE-LOOKING in-flight job id, loaded # in one indexed query. Used for the persisted-state reconciliation below # (t_8b5480b3): an in-memory claim whose OWN run's execution row is # terminal cannot represent a live run — the durable ledger proves that # run already ended — so the claim is stale by construction, regardless of # its in-memory age. A leaked claim is then recoverable without # force-run/resume. Two-phase so the healthy steady state pays no DB # work: a claim with a live future is never released, so the query only # covers claims whose future is missing/pending/done (the snapshot is # taken under _running_lock; iterating a set concurrently mutated by # try_register/release_running_job can raise RuntimeError). A claim that # becomes releasable between the snapshot and the sweep loop simply waits # for the next tick's query. from cron.executions import _TERMINAL_STATES as _terminal_states with _running_lock: _claim_futures = { job_id: _running_futures.get(job_id) for job_id in _running_job_ids } _ledger_candidates = [ job_id for job_id, fut in _claim_futures.items() if fut is None or fut is _FUTURE_PENDING or fut.done() ] _latest: dict = {} if _ledger_candidates: try: from cron.executions import latest_executions as _latest_execs _latest = _latest_execs(_ledger_candidates) except Exception: _latest = {} def _row_belongs_to_claim(row: dict, claim_started: float) -> bool: """True when the ledger row was claimed at/after this in-memory claim. The latest terminal row proves THIS claim's run ended only if it was created by this claim's dispatch (create_execution runs moments AFTER try_register_running_job). A terminal row older than the in-memory claim is the PREVIOUS run's outcome — for a recurring job that is the common case in the window between try_register and create_execution, and releasing on it would double-dispatch a healthy fresh claim. Unparseable timestamps fail closed (row treated as previous-run; the age-based path below still bounds the claim). """ claimed_at = row.get("claimed_at") if not claimed_at: return False try: from cron.jobs import _ensure_aware as _ensure_aware_ts row_ts = _ensure_aware_ts(datetime.fromisoformat(claimed_at)) return row_ts.timestamp() >= claim_started except (ValueError, TypeError, OSError): return False # Precompute job intervals OUTSIDE _running_lock so croniter evaluation # does not block try_register/release_running_job for other jobs. _intervals = {jid: _job_interval_minutes(j) for jid, j in by_id.items()} with _running_lock: for job_id in list(_running_job_ids): started = _running_since.get(job_id) if started is None: # Claim predates this guard (or was injected directly) — adopt # it now so it becomes sweepable one allowance from here. _running_since[job_id] = now continue age = now - started interval_minutes = _intervals.get(job_id) allowance = floor_seconds if interval_minutes: allowance = max(allowance, 2.0 * interval_minutes * 60.0) fut = _running_futures.get(job_id) if fut is _FUTURE_PENDING: # The claim is past its allowance and the owning future still # has not been installed — the submit path itself (SessionDB # init, agent import, config load) hung before ``pool.submit`` # returned. That is exactly the wedge class; release it. pass elif fut is not None and not fut.done(): continue # genuinely still executing # Persisted-state reconciliation: if the durable executions ledger # shows THIS claim's run reached a terminal state, the claim is # provably stale even if it is still inside its in-memory age # allowance (or was adopted fresh this tick). Release it now so # the job re-dispatches on the next tick without force-run/resume # (t_8b5480b3 — the 2026-08-14 recurring-router wedge where the # in-memory age bound alone could not see a run the ledger had # already finished). The row must belong to THIS claim # (claimed_at >= claim registration): for a recurring job the # latest terminal row is usually the PREVIOUS run's outcome — # a fresh claim in the try_register→create_execution window, or a # finished run whose worker finally hasn't released yet, would # otherwise be force-released and double-dispatched. Reaching # here implies the future is missing/pending/done (the live-future # case continued above), so every claim in this branch was a # ledger-query candidate. latest = _latest.get(job_id) if ( latest is not None and latest.get("status") in _terminal_states and _row_belongs_to_claim(latest, started) ): reason = "ledger-terminal" elif age >= allowance: reason = "age" else: continue _running_job_ids.discard(job_id) _running_since.pop(job_id, None) _running_futures.pop(job_id, None) _forced_release_count += 1 stale.append((job_id, age, allowance, fut, reason)) for job_id, age, allowance, fut, _reason in stale: job = by_id.get(job_id) or {} name = job.get("name") or job_id if fut is _FUTURE_PENDING: future_state = "pending" elif fut is None: future_state = "missing" else: future_state = "finished" logger.warning( "cron.inflight.forced_release event=forced_release reason=%s job='%s' " "id=%s age=%.0fs allowance=%.0fs future=%s — stale in-flight claim " "released; the job was skipping every fire with 'already running'", _reason, name, job_id, age, allowance, future_state, ) _record_forced_release(job_id, name, age, allowance) # A ledger-terminal release is authoritative: the durable executions # ledger ALREADY records how the last run ended (completed/failed/ # unknown), so we must NOT call mark_job_run here — doing so would # clobber an honest completed/ok status with a synthetic failure, or # double-write an already-recorded failure. We only release the claim # so the job re-dispatches on its next due tick; the ledger is the # record of record for the outcome. The age-based release below keeps # the original wedge-surfacing mark_job_run behaviour (an age-release # may have no ledger row at all, so surfacing last_error is the only # way the wedge becomes visible). if _reason == "ledger-terminal": continue # Finite-repeat guard: a forced release is NOT a real run, so it must # not consume a finite one-shot's repeat budget or let mark_job_run # auto-delete the row (completed >= times). The claim is released and # the row is left untouched, so the job re-fires normally on its next # due tick (self-heal) instead of being deleted. repeat = job.get("repeat") or {} if isinstance(repeat, dict) and repeat.get("times") is not None: logger.warning( "cron.inflight.forced_release.job_untouched job='%s' id=%s — " "finite-repeat job released without mark_job_run (repeat budget " "preserved); row left in place so it re-fires normally", name, job_id, ) continue try: mark_job_run( job_id, False, f"Stale in-flight claim force-released after {age / 60:.1f}m " f"(allowance {allowance / 60:.1f}m); previous run never released " f"the scheduler in-flight guard", ) except Exception as e: logger.warning("Could not record forced release for job %s: %s", job_id, e) return [s[0] for s in stale] def mark_running_jobs_interrupted( reason: str, *, only_owners: Optional[set] = None, ) -> list: """Best-effort: mark every currently in-flight cron job interrupted. Called by the gateway shutdown path immediately after it force-kills tool subprocesses (``process_registry.kill_all()``). A job whose tool subprocess was just killed out from under it must never be allowed to report success — even though its agent thread is still alive in this same process and may go on to produce a plausible-looking final response from the now-truncated tool output. Records the job IDs in ``_interrupted_job_ids`` BEFORE writing ``last_status`` so ``run_one_job``'s own eventual completion for the same job (racing in its own thread) sees the flag and skips its normal write instead of clobbering this one — see the check near the end of ``run_one_job``. This does not attempt to correlate the killed subprocess PID to a specific job ID (the process registry tracks PIDs, not cron job IDs); any job still dispatched at the moment of a forced kill is treated as interrupted, matching the coarser precedent already set by ``GatewayRunner._interrupt_running_agents``, which interrupts every entry in ``_running_agents`` on a drain timeout without per-agent correlation either. ``only_owners``: optional set of ``(job_id, fire_owner)`` pairs. When given (dashboard webhook drain), ONLY those exact executions are marked — unrelated runs sharing the process (e.g. the desktop ticker's own jobs) are left untouched. Interruption flags are recorded per execution token, so a later run of the same job ID never consumes a stale flag that targeted its dead predecessor. Returns the list of job IDs marked, for the caller to log. """ with _running_lock: restart_safe_waiters = set(_restart_safe_waiter_job_ids) active_fires = [ (token, job_id, owner, profile_home) for job_id, executions in _running_fire_owners.items() if job_id not in restart_safe_waiters for token, (owner, profile_home) in executions.items() ] if only_owners is not None: active_fires = [ fire for fire in active_fires if (fire[1], fire[2]) in only_owners ] registered_ids = {job_id for _t, job_id, _o, _p in active_fires} if only_owners is None: active_fires.extend( (None, job_id, None, _get_hermes_home()) for job_id in ( _running_job_ids - registered_ids - restart_safe_waiters ) ) _interrupted_job_ids.update( token if token is not None else job_id for token, job_id, _owner, _profile_home in active_fires ) marked = [] for _token, job_id, fire_owner, profile_home in active_fires: if not fire_owner: logger.warning( "Job '%s' interrupted before its durable fire owner was registered; " "leaving persisted state untouched", job_id, ) # Still report the interruption to the caller: the gateway # shutdown path uses the returned IDs to send the # interrupted-cron notice while adapters are still connected # (#82232). The in-memory interrupt flag WAS recorded above — # only the persisted last_status write is skipped here. marked.append(job_id) continue try: with use_cron_store(profile_home): if mark_job_run( job_id, False, reason, expected_fire_owner=fire_owner, ): marked.append(job_id) except Exception as e: logger.warning("Failed to mark job %s interrupted: %s", job_id, e) return marked def _is_interrupted(job_id: str, token: Optional[object] = None) -> bool: """Non-destructive peek at whether the shutdown path has marked THIS execution interrupted (see ``mark_running_jobs_interrupted``). Called by ``run_one_job`` BEFORE it decides what to deliver — a job whose tool subprocess was killed mid-flight may still produce a plausible-looking ``final_response`` from the truncated output, and that must not go out to the user as if it were a normal result. Unlike ``_consume_interrupted_flag`` below, this does not clear the flag: the later, authoritative check (right before ``last_status`` is written) still needs to see it. ``token`` scopes the check to one exact execution: owner-registered runs are matched by token, so a fresh run reusing the same job ID is not poisoned by a flag that targeted its dead predecessor. The bare job ID is only ever stored for legacy dispatch paths with no registered fire owner. """ with _running_lock: if token is not None and token in _interrupted_job_ids: return True return job_id in _interrupted_job_ids def _consume_interrupted_flag(job_id: str, token: Optional[object] = None) -> bool: """Return True and clear the flag if the shutdown path already marked THIS execution interrupted (see ``mark_running_jobs_interrupted``). Called by ``run_one_job`` right before it would otherwise write its own ``last_status``. Consuming (discarding) rather than just checking keeps the flag from leaking across a later, unrelated run of the same job ID (recurring jobs reuse their ID every fire).""" with _running_lock: hit = False if token is not None and token in _interrupted_job_ids: _interrupted_job_ids.discard(token) hit = True if job_id in _interrupted_job_ids: _interrupted_job_ids.discard(job_id) hit = True return hit def _inactivity_watchdog_loop( *, get_idle_seconds: Callable[[], float], limit_s: float, poll_s: float, stop: threading.Event, future_done: Callable[[], bool], ) -> bool: """Poll job idle time until the limit, stop, or the watched future completes. Driven by ``threading.Event.wait`` (a kernel timeout), not asyncio, so a blocked event-loop / ``run_job`` thread cannot disable this watchdog the way ``asyncio.sleep`` / ``wait_for`` would (family A of #94285 — the 4118s-idle-on-a-600s-limit cron hang). Returns True when *limit_s* of inactivity was observed. """ while not stop.wait(poll_s): if future_done(): return False try: idle = float(get_idle_seconds() or 0.0) except Exception: idle = 0.0 if idle >= limit_s: return True return False def _cron_inactivity_seconds() -> float: """Parse HERMES_CRON_TIMEOUT (seconds). 0 = unlimited; bad input = 600. Shared by run_job's inactivity monitor (which maps 0 to "no limit") and the cwd-lock bound below (which keeps the wait bounded regardless) so the two sites cannot drift apart — the lock bound must stay at or above the inactivity limit or waiters would fail while a healthy holder runs. """ raw = os.getenv("HERMES_CRON_TIMEOUT", "").strip() if not raw: return 600.0 try: return float(raw) except (ValueError, TypeError): logger.warning("Invalid HERMES_CRON_TIMEOUT=%r; using default 600s", raw) return 600.0 def _get_parallel_pool(max_workers: Optional[int]) -> concurrent.futures.ThreadPoolExecutor: """Return (or create) the persistent parallel pool.""" global _parallel_pool, _parallel_pool_max_workers if _parallel_pool is None or _parallel_pool_max_workers != max_workers: if _parallel_pool is not None: _parallel_pool.shutdown(wait=False, cancel_futures=False) _parallel_pool = concurrent.futures.ThreadPoolExecutor( max_workers=max_workers, thread_name_prefix="cron-parallel", ) _parallel_pool_max_workers = max_workers return _parallel_pool def _shutdown_parallel_pool() -> None: """Shut down the persistent pool on process exit.""" global _parallel_pool, _parallel_pool_max_workers if _parallel_pool is not None: _parallel_pool.shutdown(wait=True, cancel_futures=False) _parallel_pool = None _parallel_pool_max_workers = None atexit.register(_shutdown_parallel_pool) # Per-fire usage audit log for cron token spend instrumentation. # Resolves through _get_hermes_home() so profile-scoped paths work correctly. def _usage_audit_path() -> Path: return _get_hermes_home() / "cron" / "usage_audit.jsonl" def _utcnow_iso_ms() -> str: """RFC3339 UTC timestamp with millisecond precision and 'Z' suffix.""" now = datetime.now(timezone.utc) # %f gives microseconds; trim to milliseconds. return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z" def _write_usage_audit(record: dict) -> None: """Append a single JSONL line to ~/.hermes/cron/usage_audit.jsonl. NEVER raises — a logger bug must not break cron jobs. Wraps the entire write (path resolve, mkdir, json.dumps, file append) in a single try. """ try: path = _usage_audit_path() _ensure_cron_dir(path.parent) line = json.dumps(record, ensure_ascii=False) with open(path, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception as e: logger.warning("usage_audit write failed: %s", e) def _interpreter_shutting_down(exc: Optional[BaseException] = None) -> bool: """True when the Python interpreter is finalizing. A cron tick can fire while the gateway is tearing down — SIGTERM from ``hermes update`` / ``hermes gateway stop`` / systemd restart, or an OOM-kill. Once finalization starts, ``concurrent.futures`` refuses new work with ``RuntimeError: cannot schedule new futures after interpreter shutdown`` and asyncio's default executor is gone, so *any* attempt to schedule delivery (live-adapter, ``asyncio.run``, or a fresh pool) is doomed and only pollutes ``errors.log`` with a traceback. Callers use this to skip gracefully with a warning instead of crashing (#58720, #55924). ``exc`` lets a caller also treat an already-raised scheduling error as a shutdown signal: the ``concurrent.futures`` module-global flag can be set a hair before ``sys.is_finalizing()`` flips, so matching the error text is a safe fallback for that race. Thin wrapper — the predicate itself lives in ``tools.interpreter_shutdown.interpreter_shutting_down`` (shared with the conversation loop and the concurrent tool executor) so the shutdown-race bug class is fixed in one place. Kept as a module symbol because tests and callers throughout this file reference it by this name. """ from tools.interpreter_shutdown import interpreter_shutting_down return interpreter_shutting_down(exc) # Backward-compatible module override used by tests and emergency monkeypatches. _hermes_home: Path | None = None def _get_hermes_home() -> Path: """Resolve Hermes home dynamically while preserving test monkeypatch hooks. Cron is per-profile by design (#4707): the in-process ticker runs inside a profile-scoped gateway, so resolving the active HERMES_HOME at call time means a profile's jobs are stored AND executed under that profile's home (its .env, config.yaml, scripts, skills). Do not freeze this at import or anchor it at the shared default root — either re-breaks profile isolation. """ return _hermes_home or get_hermes_home() def _get_lock_paths() -> tuple[Path, Path]: """Resolve cron lock paths at call time so profile/env changes are honored.""" hermes_home = _get_hermes_home() lock_dir = hermes_home / "cron" return lock_dir, lock_dir / ".tick.lock" # Errnos that mean "another ticker (or manual tick) holds the tick lock", # as opposed to a real failure opening/locking the file. Everything else — # most importantly EMFILE/ENFILE (fd exhaustion, #87644) and EACCES on # open() — must be surfaced, never swallowed as lock contention. def _is_lock_contention_errno(err: OSError) -> bool: """Return True when *err* from the lock syscall means the lock is held. - POSIX: ``flock(LOCK_EX|LOCK_NB)`` reports EWOULDBLOCK/EAGAIN when another process holds the lock (EACCES on some NFS implementations). - Windows: ``msvcrt.locking(LK_NBLCK)`` reports EACCES/EDEADLK. """ if err.errno is None: return False if fcntl is not None: return err.errno in (errno.EWOULDBLOCK, errno.EAGAIN, errno.EACCES) if msvcrt is not None: return err.errno in (errno.EACCES, errno.EDEADLK) return False def _is_fd_exhaustion_text(text: str) -> bool: """Text-level half of :func:`_is_fd_exhaustion` (shared with the CLI hint).""" lowered = text.lower() return "too many open files" in lowered or "emfile" in lowered def _is_fd_exhaustion(exc: BaseException) -> bool: """Return True when *exc* indicates file-descriptor exhaustion. Recognizes EMFILE/ENFILE by errno, and the "Too many open files" wording for wrapped exceptions (``load_jobs`` wraps the raw OSError in a RuntimeError with that message, #87644). """ if isinstance(exc, OSError) and exc.errno in (errno.EMFILE, errno.ENFILE): return True return _is_fd_exhaustion_text(str(exc)) def _reclaim_fds_best_effort() -> None: """Best-effort attempt to free leaked file descriptors. The cron FD-leak family (#60859, #79742, #80792) leaks descriptors from abandoned workers/sessions. Two safe, idempotent levers: 1. ``gc.collect()`` — closes file-like objects held only in reference cycles (the classic unclosed-file leak shape), which CPython would otherwise never finalize. 2. ``apply_nofile_soft_limit()`` — raise RLIMIT_NOFILE's soft limit toward the configured target when the hard limit allows, giving the process headroom to keep serving even before every leak is freed. Never raises: a reclamation attempt must not make the ticker worse. """ try: import gc gc.collect() except Exception: pass try: from hermes_cli.resource_limits import apply_nofile_soft_limit apply_nofile_soft_limit(None) except Exception: pass def _resolve_cron_surface_mode(pconfig, logical_platform_name: str) -> str: """Resolve the continuable-cron delivery surface for a platform config. Returns ``"in_channel"`` or ``"thread"`` (default). Two config shapes: - Native adapter: the flat key ``platforms.

.extra.cron_continuable_surface`` (shipped shape, unchanged). - Relay-fronted: ``platforms.relay.extra..cron_continuable_surface`` — the same per-logical-platform sub-block the relay's documented Slack knobs use (``reply_in_thread``, ``dm_top_level_threads_as_sessions``; see RelayAdapter._relay_slack_extra). The sub-block wins over a flat key when both exist, matching _relay_slack_extra precedence, and is scoped to its logical platform so a ``slack:`` block cannot leak onto another fronted platform. Precedence nuance vs _relay_slack_extra: that helper is all-or-nothing (a sub-dict REPLACES the flat extra entirely), while this one falls back to the flat key when the sub-block exists but omits the knob. The difference is deliberate — the flat key is the legacy staging shape and must keep working — but note a flat ``cron_continuable_surface`` then applies to EVERY platform this relay fronts; only the per-platform D6 capability gate contains it. Scope the knob under the sub-block on multi-platform relays. Field gap (2026-08-18): the scheduler read only the flat key, so on the relay lane — where pconfig is platforms.relay — operators had NO working location for the knob and briefs always threaded. """ try: extra = getattr(pconfig, "extra", None) or {} sub = extra.get(str(logical_platform_name or "").lower()) if isinstance(sub, dict) and sub.get("cron_continuable_surface") is not None: raw = sub.get("cron_continuable_surface") else: raw = extra.get("cron_continuable_surface") if raw is not None and str(raw).strip().lower() == "in_channel": return "in_channel" except Exception: pass return "thread" def _resolve_origin(job: dict) -> Optional[dict]: """Extract origin info from a job, preserving any extra routing metadata. Treats non-dict origins (free-form provenance strings, ints, lists from migration scripts or hand-edited jobs.json) as missing instead of crashing with ``AttributeError`` on ``origin.get(...)``. Without this guard, a job tagged with e.g. ``"combined-digest-replaces-x-and-y"`` crashed every fire attempt with ``'str' object has no attribute 'get'`` — ``mark_job_run`` recorded the failure, but the next tick re-loaded the same poisoned origin and crashed identically until the field was patched manually (#18722). """ origin = job.get("origin") if not isinstance(origin, dict): return None platform = origin.get("platform") chat_id = origin.get("chat_id") if platform and chat_id: return origin return None def _cron_mirror_delivery_enabled(job: dict, cfg: Optional[dict] = None) -> bool: """Whether a cron delivery should also be mirrored into the target chat's gateway session transcript. Default OFF — preserves the historical isolation guarantee (cron deliveries live only in the cron job's own session, never the target chat's history) byte-for-byte for everyone who does not opt in. CARVE-OUT: the ``in_channel`` continuable surface seeds its target session independently of this knob (see ``_deliver_result`` / ``_seed_cron_channel_session``). in_channel is itself opt-in (``cron_continuable_surface: in_channel`` + the adapter capability bit), and the seed IS the feature — a continuable flat brief without its seed is a brief the next reply can't see. This knob keeps governing the SEPARATE default/thread-surface transcript mirror only. Precedence (first decisive value wins): 1. Per-job ``attach_to_session`` (bool) — set via the ``cronjob`` tool, lets one briefing job opt in without flipping global behaviour. 2. Global ``cron.mirror_delivery`` (bool) in config.yaml. 3. False. When enabled, the cron's final output is appended to the target session as an assistant turn via the existing ``gateway.mirror.mirror_to_session`` — the same primitive ``send_message`` uses — so the next user reply in that chat sees the brief in context (no "what is Task #2?" amnesia). This is alternation- and cache-safe: the append lands at a turn boundary between user turns, never mid-loop, and never mutates the cached system prompt. """ per_job = job.get("attach_to_session") if isinstance(per_job, bool): return per_job try: if cfg is None: cfg = load_config() or {} return bool((cfg.get("cron", {}) or {}).get("mirror_delivery", False)) except Exception: return False def _target_matches_origin(origin: dict, platform_name: str, chat_id: str, thread_id: Optional[str]) -> bool: """True when a delivery target is the job's own origin conversation. Mirroring is scoped to the origin session by design (see ``_maybe_mirror_cron_delivery``). A job created from a live gateway chat stamps that chat as ``origin`` (``cronjob_tools._origin_from_env``), and that session is guaranteed to exist — it is the very conversation the user was in when they scheduled the job. Fan-out targets (``deliver=all``, explicit ``platform:chat_id`` to some *other* chat, or a home-channel fallback for an origin-less API/script job) are deliberately NOT mirrored: they are broadcasts, not a continuation of a conversation, and may point at a chat the user never opened an agent session in. This makes the historical "cold-start" worry a non-case: when the mirror semantically applies (target == origin) the session always exists; when no session exists, the target was never the origin conversation, so we simply do not mirror. """ if not origin: return False if str(origin.get("platform", "")).lower() != str(platform_name).lower(): return False if str(origin.get("chat_id", "")) != str(chat_id): return False # thread_id must match when the origin pins one (topic-scoped chats); a # target that lost the thread_id is not the same conversation lane. origin_thread = origin.get("thread_id") if origin_thread is not None and str(origin_thread) != str(thread_id or ""): return False return True # Resolution-provenance ranking for the dedup OR-merge in # _resolve_delivery_targets: higher rank = stronger mirror claim. Broadcast # expansions rank 0 so "origin,all"/"all,origin" hitting the same chat keeps # the origin(-fallback) tag regardless of token order. _MIRROR_PROVENANCE_RANK = { "origin": 3, "origin_fallback": 2, "explicit": 1, } def _target_mirror_eligible( job: dict, target: dict, *, global_mirror: bool, origin_match: Optional[bool] = None, ) -> bool: """Whether a resolved delivery target may receive the transcript mirror. The June origin-scoping refactor gated mirroring on target == origin, which correctly excluded broadcasts but also silenced two legitimate conversation shapes — both hit by script-provisioned ("managed") crons, which never capture an origin (``_origin_from_env`` only fires for jobs created from a live gateway chat): - ``origin_fallback``: ``deliver=origin`` with no captured origin resolves to the home channel — the user's primary conversation standing in for the origin, not a broadcast. Eligible under the same flags as a true origin target. (Field report 2026-08-17: brief delivered to the Slack DM, mirror silently skipped, reply hit a context-less session.) - ``explicit``: a ``platform:chat_id`` target is eligible ONLY when the job itself opts in via ``attach_to_session: true`` — the job author declaring this target a conversation (managed per-user DM briefings). The global ``cron.mirror_delivery`` flag never activates explicit targets: it must not start writing transcript entries into arbitrary explicitly-addressed chats (shared channels, other users' DMs). Broadcast expansions (``all``, bare-platform home targets) carry no provenance tag and are never eligible — unchanged invariant. ``origin_match`` lets the caller pass a precomputed ``_target_matches_origin`` result (``_deliver_result`` already computes it for the same target); when ``None`` it is computed here so tests and future callers stay self-contained. """ if origin_match is None: origin = _resolve_origin(job) or {} origin_match = _target_matches_origin( origin, target.get("platform", ""), target.get("chat_id", ""), target.get("thread_id"), ) if origin_match: return True resolved_from = target.get("_resolved_from") if resolved_from == "origin_fallback": # Same activation rules as an origin target: per-job attach wins, # else the global flag. This deliberately restates the precedence # _cron_mirror_delivery_enabled encodes (keep the two in sync): the # sole production caller pre-merges it into `global_mirror`, but the # helper must stay correct standalone — a per-job False must beat a # raw global True for any caller that does not pre-merge. per_job = job.get("attach_to_session") if isinstance(per_job, bool): return per_job return bool(global_mirror) if resolved_from == "explicit": return job.get("attach_to_session") is True return False def _inchannel_seed_allowed(*, is_dm: bool, user_id: Optional[str]) -> bool: """Whether the flat in_channel session seed may run for a target. Group-channel session keys are user-isolated (``…:group::`` — see _seed_cron_channel_session); a seed without a real user_id would create an orphan session that no inbound reply ever resolves to, which is worse than no seed (the plain mirror can still land if a session exists). DM keys don't embed user_id, so DM targets are always seedable. Origin-captured jobs carry the scheduler's user_id; origin-less managed jobs typically don't, and their group-channel targets must fall back to the plain mirror. """ return bool(is_dm or user_id) def _maybe_mirror_cron_delivery( job: dict, platform_name: str, chat_id: str, mirror_text: str, thread_id: Optional[str] = None, user_id: Optional[str] = None, *, enabled: bool = False, ) -> None: """Best-effort mirror of a cron delivery into the origin chat's session. No-op unless ``enabled`` (resolved once by the caller, and already scoped to the origin target — see ``_target_matches_origin``). Reuses the shipped ``mirror_to_session`` so cron rides exactly the same path that interactive ``send_message`` mirroring already uses, including passing ``user_id`` so a per-user-isolated group chat resolves to the exact member who scheduled the job (parity with ``send_message``). All failures are swallowed — a delivery that succeeded must never be reported as failed because the transcript mirror hit a problem. Because the caller only enables this for the target that equals the job's origin conversation, the session is expected to exist (the job was born in that session). A missing session therefore indicates an origin-less / fan-out delivery that should not have been mirrored anyway, and is treated as a silent no-op — never a synthetic session is created. """ if not enabled: return text = (mirror_text or "").strip() if not text: return try: from gateway.mirror import mirror_to_session # Mirror as a USER turn with a labelled prefix, NOT an assistant turn. # The brief is not the agent speaking; an assistant-role mirror lands as # assistant→assistant after the agent's last turn and breaks strict # alternation (issue #2221, the exact failure #2313 removed). A # user-role turn collapses safely via repair_message_sequence's # consecutive-user merge on every provider, and the prefix preserves the # "this came from cron" context that the dropped SQLite mirror metadata # would otherwise lose on replay. ok = mirror_to_session( platform_name, str(chat_id), f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", source_label="cron", thread_id=thread_id, user_id=user_id, role="user", ) if ok: logger.info( "Job '%s': mirrored delivery into %s:%s session transcript", job.get("id", "?"), platform_name, chat_id, ) else: logger.debug( "Job '%s': delivery mirror skipped for %s:%s " "(no matching gateway session — cold start)", job.get("id", "?"), platform_name, chat_id, ) except Exception as e: logger.debug( "Job '%s': delivery mirror failed for %s:%s: %s", job.get("id", "?"), platform_name, chat_id, e, ) def _open_continuable_cron_thread( job: dict, adapter, chat_id: str, loop, ) -> Optional[str]: """Open a dedicated thread for a continuable cron job (thread-preferred). Returns the new ``thread_id`` on success, or ``None`` when the platform has no thread primitive (WhatsApp/Signal/SMS) or creation failed — the ``None`` return is the caller's signal to fall back to the origin-DM mirror, the same open-thread-or-fallback shape as ``GatewayRunner._process_handoff``. Reuses the shipped ``adapter.create_handoff_thread``; no new adapter surface. """ create_thread = getattr(adapter, "create_handoff_thread", None) if not callable(create_thread) or loop is None: return None task_name = job.get("name") or job.get("id", "cron") thread_name = f"Hermes — {task_name}" try: from agent.async_utils import safe_schedule_threadsafe coro = create_thread(str(chat_id), thread_name) future = safe_schedule_threadsafe(coro, loop) # type: ignore[arg-type] if future is None: return None new_thread_id = future.result(timeout=30) return str(new_thread_id) if new_thread_id else None except Exception as e: logger.debug( "Job '%s': create_handoff_thread failed on %s — falling back to " "DM-session mirror: %s", job.get("id", "?"), getattr(adapter, "name", "?"), e, ) return None def _seed_cron_thread_session( job: dict, adapter, platform_name: str, chat_id: str, thread_id: str, mirror_text: str, chat_name: Optional[str] = None, is_dm: bool = False, scope_id: Optional[str] = None, ) -> None: """Seed the freshly-opened cron thread's session with the brief. Without this the brief is *visible* in the new thread but absent from any transcript, so the user's first reply in-thread would hit a session with no record of it ("what is Task #2?"). We create the thread-keyed session (the same key the user's reply will resolve to — ``build_session_key`` keys threads as participant-shared, so no ``user_id`` is needed) and append the brief as an assistant turn via the shipped ``mirror_to_session``. ``scope_id`` is the workspace/server scope (Slack team id). ``build_session_key`` embeds it in every Slack key, so a scoped reply's key carries it — the seed must reproduce it or the seeded row is unreachable (the scope-less flat-seed sibling of the is_dm keying bug). Best-effort None for platforms without scope. ``is_dm`` selects the seeded ``chat_type``: a thread under a DM must seed ``chat_type="dm"`` because the user's in-thread DM reply arrives with chat_type="dm" and ``build_session_key`` routes DM threads through the DM arm (``...:dm::``) — a "thread"-typed seed lands in ``...:thread::``, a row no DM reply ever resolves to (continuation amnesia, Alice live 2026-08-20, job 8e21a957b77b). Channel threads keep ``chat_type="thread"`` (their replies really do arrive as threads). Same sibling-lane class as the flat seed's ``is_dm`` (dcca9d8cfe). Mirrors ``GatewayRunner._process_handoff``'s seed step, but standalone: cron reaches the live ``SessionStore`` through the adapter's ``_session_store`` handle rather than the gateway object. Best-effort — a delivery that already succeeded is never failed by a seeding problem. """ text = (mirror_text or "").strip() if not text: return try: from gateway.config import Platform from gateway.session import SessionSource seeded_session_id: Optional[str] = None session_store = getattr(adapter, "_session_store", None) if session_store is not None: try: platform_enum = Platform(platform_name.lower()) except (ValueError, KeyError): platform_enum = None if platform_enum is not None: # Discord thread destinations must key on the thread's OWN id # to match how the Discord adapter keys organic in-thread # messages (chat_id == thread_id). Other platforms (Slack, # Telegram) use chat_id == parent_channel for thread messages, # so the parent chat_id is correct for them. See the matching # guard in GatewayRunner._process_handoff. if platform_enum == Platform.DISCORD: seed_chat_id = str(thread_id) else: seed_chat_id = str(chat_id) dest_source = SessionSource( platform=platform_enum, chat_id=seed_chat_id, chat_name=chat_name, # DM threads key through the DM arm (see docstring); the # reply's chat_type is what the seed must reproduce. chat_type="dm" if is_dm else "thread", user_id="system:cron", user_name="Cron", thread_id=str(thread_id), scope_id=str(scope_id) if scope_id else None, ) # Ensure the thread-keyed session row exists so the mirror has # a target and the user's later reply joins the same session. # Capture the exact id — the mirror writes into THIS row, not # an origin-heuristic rediscovery (which bails on populated # chats; same class as the flat-seed live failure 2026-08-19). _entry = session_store.get_or_create_session(dest_source) seeded_session_id = getattr(_entry, "session_id", None) from gateway.mirror import mirror_to_session # User-role + labelled prefix (see _maybe_mirror_cron_delivery): the # seeded brief must not read as an assistant turn, or the user's first # in-thread reply produces assistant→user→... off a phantom assistant # message. Pass the seed user_id so the mirror resolves the exact # thread-keyed session row we just created. ok = mirror_to_session( platform_name, str(chat_id), f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", source_label="cron", thread_id=str(thread_id), user_id="system:cron", role="user", session_id=seeded_session_id, ) if ok: logger.info( "Job '%s': opened continuable thread %s on %s:%s and seeded the brief", job.get("id", "?"), thread_id, platform_name, chat_id, ) else: logger.warning( "Job '%s': thread seed did NOT land on %s:%s thread=%s — an " "in-thread reply will not see this brief", job.get("id", "?"), platform_name, chat_id, thread_id, ) except Exception as e: # WARNING, not debug: a silent seed failure IS the continuation- # amnesia bug (Alice 2026-08-19) — it must be visible in production. logger.warning( "Job '%s': seeding cron thread session failed for %s:%s:%s: %s", job.get("id", "?"), platform_name, chat_id, thread_id, e, ) def _seed_cron_channel_session( job: dict, adapter, platform_name: str, chat_id: str, mirror_text: str, *, is_dm: bool, user_id: Optional[str], chat_name: Optional[str] = None, scope_id: Optional[str] = None, ) -> bool: """Seed the FLAT (thread_id=None) session for an ``in_channel`` cron delivery. The ``in_channel`` surface (D1/D2) delivers the brief flat into the channel with no thread, so the continuation surface is the whole-channel / whole-DM session keyed ``thread_id=None`` — the same bucket ``reply_in_thread: false`` routes an inbound plain reply to. Unlike the thread path, the shipped delivery-mirror alone is NOT sufficient here: ``mirror_to_session`` only APPENDS to a session that already EXISTS (``_find_session_id`` → no-op when none matches), and a flat channel ``(…, None)`` row is only created when a human posts a top-level message the bot processes — a ``chat_postMessage`` cron delivery never goes through the inbound handler, so the row is usually absent and the mirror silently drops the brief (verified live: the brief never landed, the reply had no context). So we CREATE the flat session row first, exactly like ``_seed_cron_thread_session`` does for threads, then mirror into it. The session KEY must match what the user's later inbound reply resolves to (``build_session_key``): - **Channel** (``chat_type="group"``): key is ``…:group::`` — user-isolated — so the seed MUST carry the **origin's real ``user_id``** (the member who scheduled the job), NOT a synthetic ``system:cron`` id, or the reply keys to a different session. - **1:1 DM** (``chat_type="dm"``): the key is ``…:dm:`` and does NOT embed ``user_id``, so any ``user_id`` resolves to the same session. ``chat_type`` mirrors the inbound handler's own choice (``"dm" if is_dm else "group"``, ``adapter.py``), so the seeded key is byte-identical to the reply's key. Returns True if a seed row was created and the brief mirrored, else False (caller falls back to the plain mirror). Best-effort — a delivery that already succeeded is never failed by a seeding problem. """ text = (mirror_text or "").strip() if not text: return False try: from gateway.config import Platform from gateway.session import SessionSource chat_type = "dm" if is_dm else "group" session_store = getattr(adapter, "_session_store", None) seeded_session_id: Optional[str] = None if session_store is not None: try: platform_enum = Platform(platform_name.lower()) except (ValueError, KeyError): platform_enum = None if platform_enum is not None: dest_source = SessionSource( platform=platform_enum, chat_id=str(chat_id), chat_name=chat_name, chat_type=chat_type, user_id=str(user_id) if user_id else None, thread_id=None, # flat — the whole-channel/DM session # Workspace scope: build_session_key embeds it in every # Slack key, so a scoped reply only resolves to this row # when the seed carries it too (see thread-seed docstring). scope_id=str(scope_id) if scope_id else None, ) # Create the flat session row so the mirror has a target and the # user's later plain reply joins the SAME session. Capture the # exact session id: the mirror must write into THIS row, not # re-discover it via origin heuristics (which bail out on # populated chats where the flat session coexists with # per-message thread sessions — live failure, Alice 2026-08-19). _entry = session_store.get_or_create_session(dest_source) seeded_session_id = getattr(_entry, "session_id", None) from gateway.mirror import mirror_to_session ok = mirror_to_session( platform_name, str(chat_id), f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", source_label="cron", thread_id=None, user_id=str(user_id) if user_id else None, session_id=seeded_session_id, role="user", ) if ok: logger.info( "Job '%s': seeded flat in_channel session on %s:%s (chat_type=%s)", job.get("id", "?"), platform_name, chat_id, chat_type, ) return bool(ok) except Exception as e: # WARNING, not debug: a silent seed failure IS the "agent has no idea # about its own brief" bug (Alice 2026-08-19) — it must be visible in # production logs. logger.warning( "Job '%s': seeding in_channel session failed for %s:%s: %s", job.get("id", "?"), platform_name, chat_id, e, ) return False def _cron_job_origin_log_suffix(job: dict) -> str: """Return safe provenance details for security warnings about a cron job. The scheduler normally has no live HTTP request object when it detects a bad stored ``context_from`` reference. Including the job's saved origin makes future probe logs actionable without exposing secrets: platform/chat metadata for gateway-created jobs, and optional source-IP fields for API surfaces that persist them in origin metadata. """ origin = job.get("origin") if not isinstance(origin, dict): return "" fields = [] for key in ("platform", "chat_id", "thread_id", "source_ip", "remote", "forwarded_for"): value = origin.get(key) if value is None: continue text = str(value).replace("\r", " ").replace("\n", " ").strip() if text: fields.append(f"origin_{key}={text[:200]!r}") return " " + " ".join(fields) if fields else "" def _plugin_cron_env_var(platform_name: str) -> str: """Return the cron home-channel env var registered by a plugin platform. Falls through the platform registry so plugins that set ``cron_deliver_env_var`` on their ``PlatformEntry`` get cron delivery support without editing this module. """ try: from hermes_cli.plugins import discover_plugins discover_plugins() # idempotent from gateway.platform_registry import platform_registry entry = platform_registry.get(platform_name.lower()) if entry and entry.cron_deliver_env_var: return entry.cron_deliver_env_var except Exception: pass return "" def _is_known_delivery_platform(platform_name: str) -> bool: """Whether ``platform_name`` is a valid cron delivery target. Hardcoded built-ins in ``_KNOWN_DELIVERY_PLATFORMS`` are checked first; plugin platforms registered via ``PlatformEntry`` are accepted if they provide a ``cron_deliver_env_var``. """ name = platform_name.lower() if name in _KNOWN_DELIVERY_PLATFORMS: return True return bool(_plugin_cron_env_var(name)) def _resolve_home_env_var(platform_name: str) -> str: """Return the env var name for a platform's cron home channel. Built-in platforms are in ``_HOME_TARGET_ENV_VARS``; plugin platforms are resolved from the platform registry. """ name = platform_name.lower() env_var = _HOME_TARGET_ENV_VARS.get(name) if env_var: return env_var return _plugin_cron_env_var(name) def _get_config_home_channel(platform_name: str): """Return the persisted ``HomeChannel`` for a platform from gateway config. ``/sethome`` declares ``config.yaml`` canonical (it is the only store that survives for relay-fronted logical platforms, whose adapters are not natively enabled) and mirrors the value into the legacy ``_HOME_CHANNEL`` env var only as a best-effort compatibility shim. Cron historically read ONLY the env mirror, so a home channel that existed solely in config.yaml — e.g. Discord fronted by the relay connector, where no ``DISCORD_HOME_CHANNEL`` was ever exported — was invisible and jobs silently fell back to local-only. Reading the canonical store here fixes that for every relay-fronted platform at once. """ try: from gateway.config import load_gateway_config, Platform config = load_gateway_config() platform = Platform(platform_name.lower()) return config.get_home_channel(platform) except Exception: logger.debug( "config home_channel lookup failed for platform %r", platform_name, exc_info=True, ) return None def _env_home_target_chat_id(platform_name: str) -> str: """Return the home chat id from the legacy env mirror only (no config). Reads through ``get_secret`` (not raw ``os.getenv``) so a profile-scoped secret scope wins in a multiplex gateway. ``DISCORD_HOME_CHANNEL`` lives in each profile's ``.env``; in a multiplex process the winning cron tick runs with the job-owning profile's scope installed (run_one_job sets it), so reading via ``get_secret`` resolves the OWNING profile's chat id rather than the host process's ``os.environ`` (#83182, chat-id leg — the token leg was fixed earlier; chat id / thread id resolve through the same leak). """ env_var = _resolve_home_env_var(platform_name) if not env_var: return "" try: from agent.secret_scope import get_secret except Exception: get_secret = None # type: ignore if get_secret is not None: value = get_secret(env_var, "") if not value: legacy = _LEGACY_HOME_TARGET_ENV_VARS.get(env_var) if legacy: value = get_secret(legacy, "") return value or "" value = os.getenv(env_var, "") if not value: legacy = _LEGACY_HOME_TARGET_ENV_VARS.get(env_var) if legacy: value = os.getenv(legacy, "") return value def _get_home_target_chat_id(platform_name: str) -> str: """Return the configured home target chat/room ID for a delivery platform. Resolution order: platform env var (legacy mirror, kept first so an operator override keeps winning) → legacy env var name → the canonical ``home_channel`` block persisted in config.yaml by ``/sethome``. """ value = _env_home_target_chat_id(platform_name) if value: return value home = _get_config_home_channel(platform_name) if home is not None and home.chat_id: return str(home.chat_id) return "" def _get_home_target_thread_id(platform_name: str) -> Optional[str]: """Return the optional thread/topic ID for a platform home target. Telegram-only override: ``TELEGRAM_CRON_THREAD_ID`` takes precedence over ``TELEGRAM_HOME_CHANNEL_THREAD_ID`` for cron delivery. When topic mode is enabled, deliveries that land in the root DM (thread_id unset) end up in the system-only lobby where the user cannot reply — the gateway returns the lobby reminder and drops ``reply_to_message_id`` (#24409). Pointing cron at a dedicated topic via this env var lets replies work as expected without changing the lobby invariant. """ env_var = _resolve_home_env_var(platform_name) try: from agent.secret_scope import get_secret except Exception: get_secret = None # type: ignore def _scope_get(name: str) -> str: if get_secret is None: return "" v = get_secret(name, "") return v if v is not None else "" if platform_name.lower() == "telegram": cron_thread = _scope_get("TELEGRAM_CRON_THREAD_ID").strip() if cron_thread: return cron_thread if get_secret is not None: value = _scope_get(f"{env_var}_THREAD_ID").strip() if env_var else "" if not value and env_var: legacy = _LEGACY_HOME_TARGET_ENV_VARS.get(env_var) if legacy: value = _scope_get(f"{legacy}_THREAD_ID").strip() else: value = os.getenv(f"{env_var}_THREAD_ID", "").strip() if env_var else "" if not value and env_var: legacy = _LEGACY_HOME_TARGET_ENV_VARS.get(env_var) if legacy: value = os.getenv(f"{legacy}_THREAD_ID", "").strip() if value: return value # Canonical config.yaml fallback — same rationale as # _get_home_target_chat_id, and thread affinity only applies when the # chat itself resolved from the same config block (an env-provided chat # id keeps its env-provided thread semantics). if not _env_home_target_chat_id(platform_name): home = _get_config_home_channel(platform_name) if home is not None and home.thread_id: return str(home.thread_id) return None def _iter_home_target_platforms(): """Iterate built-in + plugin platform names that expose a home channel. Used by the ``deliver=origin`` fallback when the job has no origin. """ for name in _HOME_TARGET_ENV_VARS: yield name try: from hermes_cli.plugins import discover_plugins discover_plugins() # idempotent from gateway.platform_registry import platform_registry for entry in platform_registry.plugin_entries(): if entry.cron_deliver_env_var and entry.name not in _HOME_TARGET_ENV_VARS: yield entry.name except Exception: pass def _relay_fronted_delivery_platforms(connected: set) -> set: """Logical platforms deliverable through a connected relay connector. ``get_connected_platforms()`` only sees NATIVELY configured platforms. On a relay-fronted deployment (relay in ``config.platforms``, the real platform credential living in the connector) the fronted platforms are absent from that set although fire-time routing delivers to them via ``resolve_delivery_transport`` + ``RelayAdapter.fronts_platform``. This keeps validation symmetric with routing by consulting the same env-derived deploy stamp (``GATEWAY_RELAY_PLATFORMS``) the live adapter's identity set is seeded from. No relay connected -> empty set, so native topologies keep the strict credential check unchanged. """ if "relay" not in connected: return set() try: from gateway.relay import relay_fronted_platforms return relay_fronted_platforms() except Exception: logger.debug("relay fronted-platform lookup failed", exc_info=True) return set() def cron_delivery_targets() -> list[dict]: """Return the platforms a cron job can auto-deliver to. Single source of truth for any UI (dashboard dropdown, etc.) that lets a user pick a cron delivery target. A platform is included when it is a valid cron delivery platform AND its gateway is configured (enabled + credentials present). Each entry reports whether the platform's home target (the room/channel cron posts to) is set — a platform can be configured for interactive use but still lack the home target an unattended cron job needs. Returns a list of dicts: ``{"id", "name", "home_target_set", "home_env_var"}`` ordered by the gateway's canonical platform order. Callers should always prepend the implicit ``local`` option themselves — it needs no config. """ targets: list[dict] = [] try: from gateway.config import load_gateway_config gateway_config = load_gateway_config() connected = {p.value for p in gateway_config.get_connected_platforms()} connected |= _relay_fronted_delivery_platforms(connected) except Exception: logger.debug("cron_delivery_targets: gateway config unavailable", exc_info=True) connected = set() for name in _iter_home_target_platforms(): if name not in connected: continue if not _is_known_delivery_platform(name): continue env_var = _resolve_home_env_var(name) targets.append( { "id": name, "name": name.replace("_", " ").title(), "home_target_set": bool(_get_home_target_chat_id(name)), "home_env_var": env_var or None, } ) # Bot Chat targets: one per local profile. Machine-local by design (the # scheduler delivers via a local chat subprocess), so the names listed # here are exactly the names that resolve at fire time — no gateway # config, no home channel needed. try: from hermes_cli.profiles import list_profile_names for profile_name in list_profile_names(): targets.append( { "id": f"{BOT_CHAT_PLATFORM}:{profile_name}", "name": f"Bot Chat ({profile_name})", "home_target_set": True, "home_env_var": None, } ) except Exception: logger.debug("cron_delivery_targets: profile listing unavailable", exc_info=True) return targets def _origin_thread_is_stale(origin: dict) -> bool: """True when a Slack origin's thread is a stale creation-turn artifact. Relay-fronted Slack in thread-per-message mode stamps each top-level message's own id as the session thread (a session KEY, not a durable location). Jobs persisted before origin capture learned to drop that stamp carry it as ``origin.thread_id`` forever. Heuristic that repairs them at fire time without touching genuine threads: when the origin chat IS the configured Slack home chat (the ``/sethome`` conversation), a pinned origin thread is the creation-message artifact — the user's delivery expectation for their home conversation is top-level (or the home target's own configured thread). Non-home chats keep their threads: a job deliberately created inside a working thread stays there. """ if str(origin.get("platform") or "").lower() != "slack": return False if not origin.get("thread_id"): return False home_chat = _get_home_target_chat_id("slack") return bool(home_chat) and str(origin.get("chat_id")) == str(home_chat) def _origin_delivery_thread(origin: dict): """The thread a deliver=origin job should use, stale stamps dropped.""" if _origin_thread_is_stale(origin): home_thread = _get_home_target_thread_id("slack") return home_thread if home_thread else None return origin.get("thread_id") def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[dict]: """Resolve one concrete auto-delivery target for a cron job.""" origin = _resolve_origin(job) if deliver_value == "local": return None # bot-chat[:] — checked before the generic platform:chat_id # split below so the profile-name argument is never misparsed as a # chat_id on an unknown platform. bot_chat_profile = parse_bot_chat_deliver_token(deliver_value) if bot_chat_profile is not None: return _resolve_bot_chat_target(job, bot_chat_profile) if deliver_value == "origin": if origin: return { "platform": origin["platform"], "chat_id": str(origin["chat_id"]), "thread_id": _origin_delivery_thread(origin), # Resolution provenance for mirror eligibility (see # _target_mirror_eligible): this IS the origin conversation. "_resolved_from": "origin", } # Origin missing (e.g. job created via API/script) — try each # platform's home channel as a fallback instead of silently dropping. for platform_name in _iter_home_target_platforms(): chat_id = _get_home_target_chat_id(platform_name) if chat_id: logger.info( "Job '%s' has deliver=origin but no origin; falling back to %s home channel", job.get("name", job.get("id", "?")), platform_name, ) return { "platform": platform_name, "chat_id": chat_id, "thread_id": _get_home_target_thread_id(platform_name), # The fallback stands in for the user's primary # conversation (NOT a broadcast) — mirror-eligible so # continuable crons work for script-provisioned jobs # that never captured an origin. "_resolved_from": "origin_fallback", } return None if ":" in deliver_value: platform_name, rest = deliver_value.split(":", 1) platform_key = platform_name.lower() from tools.send_message_tool import ( prepare_send_message_platforms, resolve_send_target, ) prepare_send_message_platforms() # pass_unresolved_references: stored jobs have no model in the loop to react # to a resolution error, and a target the directory doesn't know # (fresh install, platform-native id) used to be handed to the # adapter as written. Dropping it here silently loses the job's # output. chat_id, thread_id, resolution_error = resolve_send_target( platform_key, rest, pass_unresolved_references=True ) if resolution_error: logger.warning( "Invalid cron delivery target '%s': %s", deliver_value, resolution_error, ) return None if ( thread_id is None and platform_key == "slack" and origin and str(origin.get("platform") or "").lower() == platform_key and str(origin.get("chat_id")) == str(chat_id) and origin.get("thread_id") and not _origin_thread_is_stale(origin) ): thread_id = origin.get("thread_id") return { "platform": platform_name, "chat_id": chat_id, "thread_id": thread_id, # Explicit platform:chat target — mirror-eligible only under the # job's own attach_to_session opt-in (see _target_mirror_eligible). "_resolved_from": "explicit", } platform_name = deliver_value if origin and origin.get("platform") == platform_name: chat_id = _get_home_target_chat_id(platform_name) if chat_id: return { "platform": platform_name, "chat_id": chat_id, "thread_id": _get_home_target_thread_id(platform_name), } return { "platform": platform_name, "chat_id": str(origin["chat_id"]), "thread_id": origin.get("thread_id"), } if not _is_known_delivery_platform(platform_name): return None chat_id = _get_home_target_chat_id(platform_name) if not chat_id: return None return { "platform": platform_name, "chat_id": chat_id, "thread_id": _get_home_target_thread_id(platform_name), } def _get_bot_chat_delivery_timeout() -> int: """Timeout for one bot-chat delivery turn (the target bot runs a full agent turn on the injected output, so this is minutes, not seconds). ``cron.bot_chat_delivery_timeout_seconds`` in config.yaml; default 600. """ try: cfg = load_config() value = int(cfg.get("cron", {}).get("bot_chat_delivery_timeout_seconds", 600)) return value if value > 0 else 600 except Exception: return 600 def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]: """Deliver job output into a profile's canonical Bot Chat as an inbound turn. Runs ``hermes [-p ] chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file `` — the exact lane Bot Mode agent-to-agent messages use, so the adopt-before-mint canonical-session rules apply and the target bot receives the output as a real user-role message it can act on. Alternation-safe by construction: this is an inbound turn on the chat command lane, not a transcript splice. ``profile`` is ``""`` for the job's own profile (subprocess inherits this scheduler's HERMES_HOME) or a validated local profile name. Returns None on success or an error string for ``last_delivery_error``. """ import shutil as _shutil import tempfile job_id = job.get("id", "?") job_name = job.get("name", job_id) hermes_bin = _shutil.which("hermes") if hermes_bin: argv = [hermes_bin] else: try: import importlib.util as _ilu if _ilu.find_spec("hermes_cli") is not None: argv = [sys.executable, "-m", "hermes_cli.main"] else: return "bot-chat delivery failed: hermes CLI not resolvable" except Exception: return "bot-chat delivery failed: hermes CLI not resolvable" env = os.environ.copy() if profile: argv += ["-p", profile] # -p owns profile resolution in the child; a leftover HERMES_HOME # from THIS scheduler's profile must not shadow it. env.pop("HERMES_HOME", None) # The prefix tells the receiving bot this is scheduled output, not the # human typing — mirrors the Bot Mode sender-attribution convention. message = ( f'[Cronjob "{job_name}" output — scheduled job, not the user. ' f"Review it, act on anything that needs action, and summarize " f"for the chat.]\n\n{content}" ) query_file = None try: with tempfile.NamedTemporaryFile( "w", encoding="utf-8", suffix=".txt", prefix="hermes-cron-botchat-", delete=False, ) as fh: fh.write(message) query_file = fh.name argv += [ "chat", "--in", "~", "-c", "Bot Chat", "--create-if-missing", "-Q", "--query-file", query_file, ] result = subprocess.run( argv, capture_output=True, text=True, timeout=_get_bot_chat_delivery_timeout(), env=env, creationflags=windows_hide_flags(), ) if result.returncode != 0: tail = (result.stderr or result.stdout or "").strip()[-500:] msg = ( f"bot-chat delivery to profile " f"'{profile or '(own)'}' failed (exit {result.returncode})" + (f": {tail}" if tail else "") ) logger.warning("Job '%s': %s", job_id, msg) return msg logger.info( "Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile or "(own)", ) return None except subprocess.TimeoutExpired: msg = ( f"bot-chat delivery to profile '{profile or '(own)'}' timed out " f"after {_get_bot_chat_delivery_timeout()}s (the bot's turn may " "still complete; raise cron.bot_chat_delivery_timeout_seconds if " "this recurs)" ) logger.warning("Job '%s': %s", job_id, msg) return msg except Exception as e: msg = f"bot-chat delivery failed: {str(e) or type(e).__name__}" logger.warning("Job '%s': %s", job_id, msg, exc_info=True) return msg finally: if query_file: try: os.unlink(query_file) except OSError: pass def _normalize_deliver_value(deliver) -> str: """Normalize a stored/submitted ``deliver`` value to its canonical string form. The contract is that ``deliver`` is a string (``"local"``, ``"origin"``, ``"telegram"``, ``"telegram:-1001:17"``, or comma-separated combinations). Historically some callers — MCP clients passing an array, direct edits of ``jobs.json``, or stale code paths — have stored a list/tuple like ``["telegram"]``. ``str(["telegram"])`` would serialize to the literal string ``"['telegram']"``, which is not a known platform and fails resolution silently. Flatten lists/tuples into a comma-separated string so both forms work. Returns ``"local"`` for anything falsy. """ if deliver is None or deliver == "": return "local" if isinstance(deliver, (list, tuple)): parts = [str(p).strip() for p in deliver if str(p).strip()] return ",".join(parts) if parts else "local" return str(deliver) # Routing intent tokens — resolved at fire time, not create time, so a # job created before Telegram was wired up will pick up Telegram once it # comes online. ``all`` expands into the set of connected platforms # (those with a configured home chat_id) in _expand_routing_tokens. _ROUTING_TOKENS = frozenset({"all"}) # Pseudo-platform for delivering job output INTO a profile's canonical # "Bot Chat" session as a real inbound turn (the bot sees it, runs a turn, # and can respond — Bot Mode's agent-to-agent lane, not a transcript # mirror). ``bot-chat`` targets the job's own profile; ``bot-chat:`` # targets a named profile on THIS machine. Deliberately excluded from the # ``all`` routing token: ``all`` fans out to messaging home channels, and a # bot-chat delivery costs a full agent turn. BOT_CHAT_PLATFORM = "bot-chat" def parse_bot_chat_deliver_token(part: str) -> Optional[str]: """Return the target profile for a ``bot-chat[:]`` deliver token. Returns ``""`` for the bare token (the job's own profile), the profile name for the explicit form, or ``None`` when ``part`` is not a bot-chat token at all. Case-insensitive on the token; the profile name is normalized by the profile layer at resolve time. """ raw = (part or "").strip() lowered = raw.lower() if lowered == BOT_CHAT_PLATFORM: return "" prefix = BOT_CHAT_PLATFORM + ":" if lowered.startswith(prefix): return raw[len(prefix):].strip() return None def _resolve_bot_chat_target(job: dict, profile_arg: str) -> Optional[dict]: """Resolve a bot-chat deliver token to a concrete delivery target. ``profile_arg`` is ``""`` for the job's own profile (the HERMES_HOME this scheduler runs under — machine-local and self-referential, so no ``-p`` flag is needed at send time) or an explicit profile name that must exist in THIS machine's profile root. Cross-machine delivery is intentionally unsupported: names resolve only against the local ``~/.hermes/profiles/`` tree, so same-named profiles on other gateways can never be targeted by accident. """ if not profile_arg: # Own profile: chat subprocess inherits HERMES_HOME, no name needed. return {"platform": BOT_CHAT_PLATFORM, "chat_id": "", "thread_id": None} try: from hermes_cli.profiles import normalize_profile_name, profile_exists canon = normalize_profile_name(profile_arg) if not profile_exists(canon): logger.warning( "Job '%s': bot-chat delivery profile '%s' not found on this " "machine — skipping target", job.get("id", "?"), profile_arg, ) return None return {"platform": BOT_CHAT_PLATFORM, "chat_id": canon, "thread_id": None} except Exception: logger.warning( "Job '%s': failed to resolve bot-chat profile '%s'", job.get("id", "?"), profile_arg, exc_info=True, ) return None def _expand_routing_tokens(part: str) -> List[str]: """Expand a routing-intent token to concrete platform names. ``all`` expands to every platform in ``_iter_home_target_platforms()`` that has a configured home chat_id right now. Unknown / non-token values pass through unchanged as a single-element list, so the caller can treat every token uniformly. """ token = part.lower() if token not in _ROUTING_TOKENS: return [part] expanded: List[str] = [] for platform_name in _iter_home_target_platforms(): if _get_home_target_chat_id(platform_name): expanded.append(platform_name) return expanded def _delivery_lane_value(job: dict, *, for_failure: bool = False): """Raw deliver-lane value for a run outcome: the failure lane when ``for_failure`` and the job overrides it, else ``deliver``. Keeps delivery bookkeeping (outcome classification, unresolved-origin, incident 'alerted' marking) reading the SAME lane the notice was actually routed through (NS-788 review finding B1).""" if for_failure: failure_deliver = job.get("failure_deliver") if failure_deliver is not None and str(failure_deliver).strip(): return failure_deliver return job.get("deliver", "local") def _resolve_delivery_targets(job: dict, *, for_failure: bool = False) -> List[dict]: """Resolve all concrete auto-delivery targets for a cron job. Accepts the legacy comma-separated ``deliver`` string plus the ``all`` routing-intent token, which expands to every platform with a configured home channel. Tokens may be combined with explicit targets: ``origin,all`` and ``all,telegram:-100:17`` both work. Duplicate (platform, chat_id, thread_id) tuples are collapsed by the existing dedup pass. ``for_failure=True`` resolves failure-category engine notices (failure summaries, interrupted-run notices, drift/preflight alerts): when the job carries a ``failure_deliver`` value, targets resolve from it INSTEAD of ``deliver`` — ``failure_deliver: local`` is the structural opt-out for shared channels (NS-788, Coatue). Absent ``failure_deliver``, failure delivery follows ``deliver`` exactly as before. """ deliver_raw = _delivery_lane_value(job, for_failure=for_failure) deliver = _normalize_deliver_value(deliver_raw) if deliver == "local": return [] raw_parts = [p.strip() for p in deliver.split(",") if p.strip()] # Expand routing intents. parts: List[str] = [] for raw in raw_parts: parts.extend(_expand_routing_tokens(raw)) seen = {} targets = [] for part in parts: target = _resolve_single_delivery_target(job, part) if target: key = (target["platform"].lower(), str(target["chat_id"]), target.get("thread_id")) if key not in seen: seen[key] = target targets.append(target) else: # OR-merge resolution provenance on dedup: "origin,all" (either # order) resolving to the same chat must keep the # origin/origin_fallback tag — a mirror-eligible token must not # lose eligibility to token order (see _target_mirror_eligible). kept = seen[key] if _MIRROR_PROVENANCE_RANK.get(str(target.get("_resolved_from") or ""), 0) > \ _MIRROR_PROVENANCE_RANK.get(str(kept.get("_resolved_from") or ""), 0): kept["_resolved_from"] = target.get("_resolved_from") return targets def _resolve_delivery_target(job: dict) -> Optional[dict]: """Resolve the concrete auto-delivery target for a cron job, if any.""" targets = _resolve_delivery_targets(job) return targets[0] if targets else None # Media extension sets — audio routing is centralized in gateway.platforms.base # via should_send_media_as_audio() so Telegram-specific rules stay in one place. _VIDEO_EXTS = frozenset({'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'}) _IMAGE_EXTS = frozenset({'.jpg', '.jpeg', '.png', '.webp', '.gif'}) def _send_media_via_adapter( adapter, chat_id: str, media_files: list, metadata: dict | None, loop, job: dict, platform=None, ) -> list: """Send extracted MEDIA files as native platform attachments via a live adapter. Routes each file to the appropriate adapter method (send_voice, send_image_file, send_video, send_document) based on file extension — mirroring the routing logic in ``BasePlatformAdapter._process_message_background``. Returns a list of per-file error strings (empty when every attachment delivered). Callers surface these into the job's delivery errors so a dropped attachment is visible in ``last_error``/run status instead of only in the gateway log (the silent-drop half of the manual-run attachment bug: text delivered, file vanished, job marked ok). """ from pathlib import Path from gateway.platforms.base import BasePlatformAdapter, should_send_media_as_audio errors: list = [] requested = [(str(p), v) for p, v in (media_files or [])] media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) # Report paths the safety filter dropped: the model referenced them in # MEDIA: tags but they will never be sent (missing file, denied prefix, # or strict-mode policy miss). kept = {p for p, _ in media_files} for raw_path, _v in requested: try: from gateway.platforms.base import validate_media_delivery_path if validate_media_delivery_path(raw_path) not in kept: errors.append( f"attachment dropped by media path policy: {raw_path}" ) except Exception: errors.append(f"attachment dropped by media path policy: {raw_path}") for media_path, _is_voice in media_files: try: ext = Path(media_path).suffix.lower() route_platform = platform if platform is not None else getattr(adapter, "platform", None) if should_send_media_as_audio(route_platform, ext, is_voice=_is_voice): coro = adapter.send_voice(chat_id=chat_id, audio_path=media_path, metadata=metadata) elif ext in _VIDEO_EXTS: coro = adapter.send_video(chat_id=chat_id, video_path=media_path, metadata=metadata) elif ext in _IMAGE_EXTS: coro = adapter.send_image_file(chat_id=chat_id, image_path=media_path, metadata=metadata) else: coro = adapter.send_document(chat_id=chat_id, file_path=media_path, metadata=metadata) from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe(coro, loop) if future is None: msg = f"cannot send media {media_path}: gateway loop unavailable" logger.warning("Job '%s': %s", job.get("id", "?"), msg) errors.append(msg) return errors try: # Large attachments (long TTS audio, concatenated recordings, # big exports) can legitimately exceed a fixed 30s upload # window. Configurable, matching the other cron timeouts # (cron.media_send_timeout_seconds in config.yaml, or the # HERMES_CRON_MEDIA_SEND_TIMEOUT env override). result = future.result(timeout=_get_media_send_timeout()) except TimeoutError: future.cancel() raise if result and not getattr(result, "success", True): msg = ( f"media send failed for {media_path}: " f"{getattr(result, 'error', 'unknown')}" ) logger.warning("Job '%s': %s", job.get("id", "?"), msg) errors.append(msg) except Exception as e: # Argument-less exceptions (notably TimeoutError, the most likely # failure on this path) have an empty str(), which would render # the reason as nothing at all. Fall back to the class name. msg = ( f"failed to send media {media_path}: {str(e) or type(e).__name__}" ) logger.warning("Job '%s': %s", job.get("id", "?"), msg) errors.append(msg) return errors def _confirm_adapter_delivery(send_result, job_id: str = "?", unverified: Optional[list] = None) -> bool: """Return True only if ``send_result`` unambiguously confirms delivery. A live adapter that returns ``None`` (e.g. a swallowed exception, a busy platform, or a code path that returns early without producing a ``SendResult``) must NOT be treated as success — doing so causes the scheduler to log ``"delivered to via live adapter"`` while the gateway never actually sees the message (#47056). Likewise, a result carrying no ``success`` at all (a partial mock, or a ``dict`` from a code path that never reached the adapter) is a contract violation: it does not actually tell us whether the send succeeded. Require an explicit, truthy ``success`` to count as confirmed. Both shapes are inspected the same way, because ``_deliver_to_platform`` returns either a ``SendResult`` object or a plain ``dict``: * ``delivered is False`` is a REJECTION even when ``success`` is truthy. The silence-narration filter returns ``{"success": True, "delivered": False}`` — a successfully *dropped* message, not a delivered one. Reading only ``success`` there is how a cron brief was logged as delivered while the user got nothing (#77763). * No ``message_id`` and no ``raw_response`` means we have no positive evidence of a send. That is not proof of failure either (some adapters legitimately return a bare success), so it is still accepted — but logged at WARNING so an UNVERIFIED delivery is visible in the log instead of masquerading as a confirmed one. Telegram ``SendResult`` objects carry ``message_id``; the dict-filter shape does not. """ if send_result is None: return False if isinstance(send_result, dict): if "success" not in send_result: return False success = bool(send_result.get("success")) delivered = send_result.get("delivered") message_id = send_result.get("message_id") raw_response = send_result.get("raw_response") else: if not hasattr(send_result, "success"): return False success = bool(getattr(send_result, "success")) delivered = getattr(send_result, "delivered", None) message_id = getattr(send_result, "message_id", None) raw_response = getattr(send_result, "raw_response", None) if not success or delivered is False: return False if message_id is None and not raw_response: logger.warning( "Job '%s': live adapter reported success with no delivery evidence " "(no message_id, no raw_response) — treating as delivered but " "UNVERIFIED", job_id, ) if unverified is not None: unverified.append(True) return True def _is_channel_dm_topic( runtime_adapter: Any, chat_id: Any, loop: Any, job_id: str, ) -> bool: """Decide whether an (already-ambiguous) Telegram topic target is a genuine Bot API *channel* Direct-Messages topic (route via ``direct_messages_topic_id``) rather than a forum-style topic in a private chat (route via ``message_thread_id``). Callers gate this on the ambiguous shape first (``telegram::``) — that shape is identical for both cases, so shape alone cannot decide (this was the #52060 regression). The real signal is the chat *type*: a genuine channel DM topic lives on a ``channel`` chat. Probe the live adapter's ``get_chat_info`` once and only return True when the chat is a channel. Fails SAFE to ``message_thread_id`` (returns False) for adapters without a probe, or any probe error/timeout — that is the pre-#22773 behaviour and the correct default for the common forum-topic case. """ # Resolve on the CLASS, not the instance (general pitfall #11): a MagicMock # instance auto-creates a truthy ``get_chat_info`` attribute, so an # instance-level probe would misclassify test doubles. Real adapters expose # the coroutine on the class regardless. get_chat_info = getattr(type(runtime_adapter), "get_chat_info", None) if not callable(get_chat_info): return False try: from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe( get_chat_info(runtime_adapter, str(chat_id)), loop, # type: ignore[arg-type] ) if future is None: return False # Lighter than a send (metadata-only Bot API call), so a shorter bound # than the 30s/60s send waits elsewhere in this file is intentional. info = future.result(timeout=10) except Exception: logger.debug( "Job '%s': get_chat_info probe failed for chat=%s — " "defaulting to message_thread_id routing", job_id, chat_id, exc_info=True, ) return False is_channel = isinstance(info, dict) and str(info.get("type") or "").lower() == "channel" if is_channel: logger.info( "Job '%s': chat=%s is a channel — routing via direct_messages_topic_id", job_id, chat_id, ) return is_channel def _cron_delivery_notify_enabled(cfg: Optional[dict]) -> bool: """Resolve ``cron.delivery.notify`` (config.yaml). Default True. Only an explicit boolean ``False`` (or a YAML ``false``/``off`` that parses to it) disables the push notification; a missing/malformed section keeps the default so a typo can never silently make cron briefs silent. """ try: cron_cfg = (cfg or {}).get("cron") if not isinstance(cron_cfg, dict): return True delivery_cfg = cron_cfg.get("delivery") if not isinstance(delivery_cfg, dict): return True return delivery_cfg.get("notify", True) is not False except Exception: return True def _record_delivery_verification(job: dict, unverified_targets: list) -> None: """Persist the UNVERIFIED-delivery marker on the job record. ``last_delivery_unverified`` is a list of ``platform:chat_id`` targets whose live adapter acked the send with no message_id/raw_response, or ``None`` once a run delivered with positive evidence (or to no live target). Skips the write when nothing changed so the common verified path costs no jobs.json save. Never raises — status bookkeeping must not fail a delivery. """ new_value = list(unverified_targets) or None if (job.get("last_delivery_unverified") or None) == new_value: return try: from cron.jobs import update_job update_job(job["id"], {"last_delivery_unverified": new_value}) except Exception as exc: # pragma: no cover - defensive logger.debug( "Job '%s': could not record delivery verification: %s", job.get("id"), exc, ) def _deliver_result( job: dict, content: str, adapters=None, loop=None, *, for_failure: bool = False ) -> Optional[str]: """ Deliver job output to the configured target(s) (origin chat, specific platform, etc.). When ``adapters`` and ``loop`` are provided (gateway is running), tries to use the live adapter first — this supports E2EE rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. Falls back to standalone send if the adapter path fails or is unavailable. ``for_failure=True`` routes failure-category engine notices through the job's ``failure_deliver`` override when present (NS-788). Returns None on success, or an error string on failure. """ targets = _resolve_delivery_targets(job, for_failure=for_failure) if not targets: deliver_value = _normalize_deliver_value( _delivery_lane_value(job, for_failure=for_failure) ) if deliver_value == "local": return None # local-only jobs don't deliver — not a failure # deliver=origin with no resolvable origin and no configured home # channels: treat as local rather than reporting an error. CLI-created # jobs never capture a {platform, chat_id} origin, so failing here would # make every CLI `deliver=origin` (or auto-detect) job emit a spurious # "no delivery target resolved" error on every run (#43014). The output # is still persisted in last_output for `cron list`/resume. if deliver_value == "origin": logger.info( "Job '%s': deliver=origin but no origin or home channels — " "skipping delivery (output saved in last_output)", job.get("name", job.get("id", "?")), ) return None msg = f"no delivery target resolved for deliver={deliver_value}" logger.warning("Job '%s': %s", job["id"], msg) return msg # Restart-safe workers intentionally have no live gateway adapter objects. # Hand the send back through a durable queue so the current or replacement # gateway performs it with relay/E2EE parity. The execution id is the # idempotency key; the queue never retries an uncertain claimed send. # Match on this job's own attempt: a worker's script may itself dispatch # another job in-process (``hermes cron run``), and that nested delivery # must not be keyed under the outer execution id. external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER", "") if ( external_execution and adapters is None and external_execution == str(job.get("execution_id") or "") ): from cron.delivery_queue import enqueue_and_wait return enqueue_and_wait( external_execution, job, content, for_failure=for_failure, ) from tools.send_message_tool import _send_to_platform from gateway.config import load_gateway_config, Platform # Optionally wrap the content with a header/footer so the user knows this # is a cron delivery. Wrapping is on by default; set cron.wrap_response: false # in config.yaml for clean output. wrap_response = True user_cfg = None try: user_cfg = load_config() wrap_response = user_cfg.get("cron", {}).get("wrap_response", True) except Exception: pass # cron.delivery.notify (default True): mark live-adapter cron sends as # FINAL notifications so the platform pushes them (Telegram's "important" # mode otherwise sends with disable_notification=True). Configurable so # operators who prefer silent briefs can opt back out. notify_delivery = _cron_delivery_notify_enabled(user_cfg) # Set when a live adapter acked a send with NO delivery evidence (no # message_id / raw_response — the Slack/Matrix/Mattermost bare # SendResult(success=True) shape). Persisted on the job as # ``last_delivery_unverified`` so `hermes cron list` shows the state # instead of it living only in a WARNING log line. unverified_targets: list = [] if wrap_response: task_name = job.get("name", job["id"]) job_id = job.get("id", "") delivery_content = ( f"Cronjob Response: {task_name}\n" f"(job_id: {job_id})\n" f"-------------\n\n" f"{content}\n\n" f"To stop or manage this job, send me a new message (e.g. \"stop reminder {task_name}\")." ) else: delivery_content = content # Extract MEDIA: tags so attachments are forwarded as files, not raw text from gateway.platforms.base import BasePlatformAdapter # Bridge gateway media-policy config (strict / allow_dirs / trust_recent) # into the env vars the path validator reads. Gateway startup does this # at boot; a standalone process (manual `hermes cron run` from the CLI, # a cron tick without the gateway) historically did NOT — so manual runs # filtered attachment paths under a DIFFERENT policy than scheduled runs # and silently dropped files the gateway would deliver. Idempotent, # env-wins, never raises. from gateway.media_policy import apply_media_policy_env apply_media_policy_env(user_cfg) media_files, cleaned_delivery_content = BasePlatformAdapter.extract_media(delivery_content) requested_media = [(str(p), v) for p, v in media_files] media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) # Attachments the policy filter dropped will never be sent on ANY lane — # record them up front so the run status says so (previously one # stderr WARNING was the only trace: text delivered, file vanished). _policy_dropped = len(requested_media) - len(media_files) policy_drop_errors = ( [ f"{_policy_dropped} media attachment(s) dropped by media path " "policy (missing file, denied prefix, or strict-mode miss); " "see gateway.strict / media_delivery_allow_dirs in config.yaml" ] if _policy_dropped > 0 else [] ) # Resolve the delivery-mirror gate ONCE (default off). When on, each # successful delivery is also appended to the target chat's gateway session # transcript so a user reply in that chat sees the cron output in context. # Mirror the CLEAN, unwrapped output (not the cron header/footer). try: mirror_enabled = _cron_mirror_delivery_enabled(job, user_cfg) except Exception: mirror_enabled = False # Keep the cleaned delivery text available independently of the optional # transcript-mirror knob. Continuable surfaces (notably in_channel) must # seed their target session even when attach_to_session=false and # cron.mirror_delivery=false; gating this value on mirror_enabled makes # the seed receive an empty string and return False, which is exactly the # live failure reproduced three times on Alice (job ef7bd2869d15). _, mirror_text = BasePlatformAdapter.extract_media(content) mirror_text = (mirror_text or "").strip() try: config = load_gateway_config() except Exception as e: msg = f"failed to load gateway config: {e}" logger.error("Job '%s': %s", job["id"], msg) return msg delivery_errors = [] for target in targets: platform_name = target["platform"] chat_id = target["chat_id"] thread_id = target.get("thread_id") # bot-chat targets don't ride a gateway adapter: the output becomes a # real inbound turn in the target profile's canonical Bot Chat via the # chat CLI lane (the same one Bot Mode agent-to-agent sends use). The # bot runs a turn and can respond — handled before the Platform enum # below, which knows nothing about this pseudo-platform. if platform_name == BOT_CHAT_PLATFORM: bot_chat_error = _deliver_to_bot_chat(job, content, chat_id) if bot_chat_error: delivery_errors.append(bot_chat_error) continue # Diagnostic: log thread_id for topic-aware delivery debugging origin = _resolve_origin(job) or {} origin_thread = origin.get("thread_id") if origin_thread and not thread_id: logger.warning( "Job '%s': origin has thread_id=%s but delivery target lost it " "(deliver=%s, target=%s)", job["id"], origin_thread, job.get("deliver", "local"), target, ) elif thread_id: logger.debug( "Job '%s': delivering to %s:%s thread_id=%s", job["id"], platform_name, chat_id, thread_id, ) # Mirror scope: the origin conversation, the home-channel FALLBACK for # an origin-less deliver=origin job (a script-provisioned managed cron # standing in for the user's primary conversation — not a broadcast), # or an explicit target the job opted into via attach_to_session. # Broadcast/fan-out targets are never mirrored (_target_mirror_eligible). origin_target = _target_matches_origin(origin, platform_name, chat_id, thread_id) mirror_this_target = mirror_enabled and _target_mirror_eligible( job, target, global_mirror=mirror_enabled, origin_match=origin_target, ) # Pass the origin's user_id so a per-user-isolated group chat resolves to # the exact member who scheduled the job — parity with send_message. # Resolved for ANY origin-matching target (not just mirror-enabled): # the in_channel seed below needs it too, and it must not depend on # the attach_to_session/mirror opt-in. origin_user_id = origin.get("user_id") if origin_target else None # DM shape of this target, needed by BOTH the in_channel flatten gate # below and the seed/_seed_cron_channel_session chat_type further down: # a 1:1 DM keys as ``dm`` (Slack DM channel ids start with "D"; or the # origin says so), everything else as ``group``. origin_chat_type = str(origin.get("chat_type") or "").lower() is_dm_target = origin_chat_type == "dm" or ( not origin_chat_type and str(chat_id).startswith("D") ) # Shared continuable-target gate for the in_channel surface. The # thread-flatten and the flat-session seed MUST use the SAME gate — # if they drift, the brief and its continuation session land in # different places (the split-surface bug the flatten exists to # prevent). Origin targets qualify unconditionally (independent of the # attach_to_session / mirror opt-in — see 3c52d3589f); non-origin # mirror-eligible targets (origin_fallback / opted-in explicit) # qualify only when the seed can actually create a resolvable session # (_inchannel_seed_allowed: DM-shaped, or a known user_id for # user-isolated group keys). inchannel_continuable = origin_target or ( mirror_this_target and _inchannel_seed_allowed(is_dm=is_dm_target, user_id=origin_user_id) ) # Built-in names resolve to their enum member; plugin platform names # create dynamic members via Platform._missing_(). try: platform = Platform(platform_name.lower()) except (ValueError, KeyError): msg = f"unknown platform '{platform_name}'" logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) continue from gateway.delivery import resolve_delivery_transport target_adapters = adapters if isinstance(adapters, SharedRouteAdapters): # Credentialless satellite: the primary adapter is a valid # transport for THIS target only when an exact primary route maps # it to this profile (#101113). Miss → fail closed below. shared = adapters.get(platform, target) target_adapters = {platform: shared} if shared is not None else {} transport = resolve_delivery_transport(platform, config, target_adapters) if transport is not None: pconfig = transport.config runtime_adapter = transport.adapter else: # No live transport. A relay-fronted platform's ONLY sender is the # gateway's live relay adapter — there is no standalone fallback # (the connector owns the credential). A manual in-process run # (`hermes cron run`) has no live relay adapter, so surface the # accurate remediation instead of the native configured/enabled # gate, which misdiagnoses relay-fronted deployments. from gateway.relay import relay_fronted_platforms if platform_name in relay_fronted_platforms(): msg = ( f"platform '{platform_name}' is relay-fronted and has no " "live gateway transport; start the gateway (its ticker " "owns relay-fronted delivery and will fire the job on " "schedule)" ) logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) continue # Preserve the existing standalone delivery path, which uses the # logical platform's configured credential. pconfig = config.platforms.get(platform) runtime_adapter = None if transport is not None and transport.is_relay: # A relay transport carries the RELAY adapter's config, and # resolve_delivery_transport already applied relay's enablement # rule (config block absent OR enabled). The logical platform is # deliberately NOT natively enabled in a relay-fronted deployment # (its credential lives in the connector), so the native # configured/enabled gate below must not apply — it used to # reject exactly the targets the relay was resolved to serve. if pconfig is None: from gateway.config import PlatformConfig pconfig = PlatformConfig(enabled=True) elif not pconfig or not pconfig.enabled: msg = f"platform '{platform_name}' not configured/enabled" logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) continue # Prefer the resolved live transport when the gateway is running. This # supports E2EE native adapters and relay-fronted logical platforms. # The live-send path (which SEEDS the flat in_channel continuation # session via _seed_cron_channel_session) needs not just a live adapter # but a running event loop to schedule the async send onto. Compute that # gate ONCE so the in_channel thread_id clear below stays in lockstep # with the live-send/seed block further down (they used to drift): an # adapter can be present while the loop is absent/not-running, in which # case the live-send block is skipped and delivery falls through to the # standalone path — which cannot seed the flat session (r3609147550). live_adapter_ready = ( runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)() ) delivered = False target_errors = [] # Continuable cron surface (D1/D2/D6): resolve the delivery surface for # this platform generically from its config ``extra``. Default "thread" # (today's behaviour, byte-identical). "in_channel" delivers the brief # FLAT into the channel (no dedicated thread) so a plain channel reply # continues the job in-context via the shared-channel session # ``(platform, chat_id, None)`` — the same bucket ``reply_in_thread: # false`` routes inbound channel messages to. The key is read # generically here (any platform); the ``in_channel`` branch is gated on # the adapter capability flag ``supports_inchannel_continuable`` so an # unsupported platform fails SAFE to "thread" (Slack is the first # consumer; "first consumer ≠ definition"). surface_mode = _resolve_cron_surface_mode(pconfig, platform_name) in_channel_surface = surface_mode == "in_channel" if in_channel_surface and runtime_adapter is not None: # Per-platform capability first: one RelayAdapter fronts N # platforms and the connector advertises the bit per platform at # handshake — the scalar attr only carries the PRIMARY identity's # bit. Native adapters (no per-platform query) keep the class # attribute path unchanged. per_platform_check = getattr( runtime_adapter, "supports_inchannel_continuable_for_platform", None, ) if callable(per_platform_check): try: surface_supported = bool(per_platform_check(platform_name)) except Exception: surface_supported = False else: surface_supported = bool(getattr( runtime_adapter, "supports_inchannel_continuable", False )) if not surface_supported: # Fail safe (D6): platform has no in_channel continuation # primitive. logger.debug( "Job '%s': cron_continuable_surface=in_channel not supported on " "%s, using thread", job.get("id", "?"), platform_name, ) in_channel_surface = False if in_channel_surface and inchannel_continuable and live_adapter_ready: # Force flat delivery (D2): the continuable-channel target must # ignore any inherited origin/target thread_id, or the flat # continuable session seeded below (thread_id=None, via # _seed_cron_channel_session) never matches where the brief is # actually delivered — route_thread_id further down in this loop # reads `thread_id` and would otherwise route into the origin # thread instead of flat into the channel. # # Gated on `inchannel_continuable` (the SAME gate as the seed # below), NOT `mirror_this_target` alone: for origin targets the # seed fires on origin-match alone (in_channel is the # continuation surface, independent of the attach_to_session / # mirror opt-in), so the flatten must use the SAME gate — with # the default knobs off, a mirror-gated flatten kept delivering # into the origin thread while the flat session got seeded, # leaving the brief and its continuation surface in different # places. # Gated on `live_adapter_ready` (adapter present AND a running loop) # so the clear fires ONLY on the live-send path that actually seeds # the flat session — the SAME condition as the live-send block # below. `runtime_adapter is not None` alone is broader than that # path: an adapter can be present while the event loop is absent or # not running, in which case the live-send/seed block is skipped and # delivery falls through to the standalone path. Clearing thread_id # there would flatten a brief into a channel with NO seeded # continuable session behind it (and bypass the D6 capability # check), so the standalone fallback must keep the origin thread # (review r3609147550). # # Fan-out / broadcast / explicit-thread targets keep their thread_id # (they are not continuable and are never seeded). Placed AFTER # mirror_this_target / origin_user_id are computed above — those # need the ORIGINAL thread_id to match the origin conversation. thread_id = None # For an in_channel delivery the flat continuation session is created # explicitly below (the shipped mirror only APPENDS to an existing # session, and the flat channel row is otherwise absent for a # chat_postMessage delivery). ``is_dm_target`` (computed above with # origin_user_id) selects the session chat_type so the seeded key # matches the inbound reply's key. ``inchannel_seeded`` suppresses the # generic mirror below so the brief is not double-written. inchannel_seeded = False # Continuable cron (thread-preferred): when mirroring is enabled for the # origin target and the gateway is live, try to open a DEDICATED thread # for this job and deliver the brief into it. On thread-capable # platforms (Telegram/Discord/Slack) the brief + the user's replies live # in their own scrollback; the thread-keyed session is seeded so a reply # continues with full context. On DM-only platforms (WhatsApp/Signal) # create_handoff_thread returns None and we fall back to mirroring into # the origin DM session (handled after delivery). Cf. _process_handoff. # # in_channel surface (D2): SKIP thread creation entirely — leave # thread_id=None so the delivery posts flat, then # ``_seed_cron_channel_session`` (below) CREATES the shared-channel # session and mirrors the brief into it. The shipped mirror alone is # NOT enough here: ``mirror_to_session`` only APPENDS to an existing # session and a flat ``(platform, chat_id, None)`` row is otherwise # absent for a ``chat_postMessage`` delivery, so the seed must create # the row first (F5). thread_seeded = False opened_thread_id: Optional[str] = None if ( mirror_this_target and not in_channel_surface and runtime_adapter is not None and loop is not None and not thread_id # never override an explicit origin thread/topic ): new_thread_id = _open_continuable_cron_thread( job, runtime_adapter, chat_id, loop, ) if new_thread_id: # Route THIS delivery into the new thread now (the send needs the # thread_id), but defer seeding the thread session until the # delivery actually succeeds — otherwise an open-succeeds / # deliver-fails case leaves a seeded brief the user never saw, # and (worse) suppresses the DM-fallback mirror via thread_seeded. thread_id = new_thread_id opened_thread_id = new_thread_id if live_adapter_ready: # Telegram topic routing (#22773, regression fixed #52060): a # ``telegram::`` cron target is # ambiguous — a forum-style topic in a private chat and a genuine # Bot API channel Direct-Messages topic share the same shape and # need OPPOSITE routing. Disambiguate at delivery time via # ``_is_channel_dm_topic`` (see its docstring for the full # rationale); ``thread_id`` goes in ``route_metadata`` so the # anchorless cron send bypasses the DeliveryRouter's private-chat # reply-anchor requirement. Compute the routed metadata ONCE so both # the text send (via DeliveryRouter) and the media send agree. from gateway.delivery import ( DeliveryRouter, DeliveryTarget, _looks_like_int, looks_like_telegram_private_chat_id, ) is_ambiguous_telegram_topic = ( platform == Platform.TELEGRAM and thread_id is not None and looks_like_telegram_private_chat_id(str(chat_id)) and _looks_like_int(str(thread_id)) ) route_via_dm_topic = is_ambiguous_telegram_topic and _is_channel_dm_topic( runtime_adapter, chat_id, loop, job["id"], ) if route_via_dm_topic: # Genuine Bot API channel Direct-Messages topic (#22773 mode 2): # routed via direct_messages_topic_id, no bare thread_id. route_thread_id = None route_metadata = { "direct_messages_topic_id": str(thread_id), "job_id": job["id"], "notify": notify_delivery, } # Media metadata mirrors the text routing so attachments land in # the same DM topic instead of the General lane (#22773). media_metadata = { "direct_messages_topic_id": str(thread_id), "notify": notify_delivery, } else: # Forum-style topic (private chat / supergroup) or non-topic # target: route via message_thread_id (#52060). Put thread_id in # *route_metadata* (not just the DeliveryTarget) deliberately — # the DeliveryRouter's private-chat topic detection # (gateway/delivery.py) demands a reply anchor when thread_id is # absent from metadata; cron deliveries have no inbound reply # anchor, so the metadata key bypasses that check and lets the # adapter route via a plain message_thread_id. route_thread_id = str(thread_id) if thread_id is not None else None route_metadata = {"job_id": job["id"], "notify": notify_delivery} if route_thread_id: route_metadata["thread_id"] = route_thread_id media_metadata = {"notify": notify_delivery} if thread_id: media_metadata["thread_id"] = thread_id # Relay egress needs a tenant discriminator on the frame: the # connector's fail-closed guard resolves the workspace/guild from # metadata.scope_id, and after a gateway restart the RelayAdapter's # per-chat scope cache is COLD (learned only from inbound), while # DeliveryRouter stamps scope only for the configured HOME channel # (gateway/delivery.py). A scoped origin that is not the home chat # therefore egressed with no scope_id at all and could be rejected # before delivery — the delivery-leg sibling of the seed-key scope # fix. Origin-matching targets only: a fan-out/broadcast target's # tenant is NOT the origin's, and stamping the wrong scope is worse # than none (the router/home path handles fan-out home targets). if origin_target and origin.get("scope_id"): route_metadata.setdefault("scope_id", str(origin["scope_id"])) media_metadata = dict(media_metadata or {}) media_metadata.setdefault("scope_id", str(origin["scope_id"])) try: # Send cleaned text (MEDIA tags stripped) — not the raw content. # Route through the gateway's DeliveryRouter so the live send # gets the same platform-specific routing as live messages — # in particular Telegram's three-mode topic routing. The # standalone cron path lacked this, so DM-topic cron deliveries # landed in the General topic or were rejected by Bot API 10.0 # (#22773). text_to_send = cleaned_delivery_content.strip() adapter_ok = True timed_out = False delivered_message_id = None if not text_to_send and not media_files: # Nothing to hand the adapter at all. This used to fall # straight through to the `if adapter_ok:` branch below and # log "delivered to via live adapter" for a send that # never happened (#77763). Fail closed so the run reports # the empty payload instead. msg = ( f"live adapter send skipped (empty text and no media) " f"for {platform_name}:{chat_id}" ) logger.warning("Job '%s': %s", job["id"], msg) target_errors.append(msg) adapter_ok = False elif text_to_send: from agent.async_utils import safe_schedule_threadsafe router = DeliveryRouter(config, target_adapters) route_target = DeliveryTarget( platform=platform, chat_id=str(chat_id), thread_id=route_thread_id, is_explicit=True, ) # Pass thread routing via the target (not a bare metadata # "thread_id"): the router only applies its Telegram DM-topic # detection when "thread_id"/"message_thread_id" are absent # from metadata, deriving the routing from target.thread_id # or the explicit direct_messages_topic_id above. future = safe_schedule_threadsafe( router._deliver_to_platform( route_target, text_to_send, route_metadata, ), loop, ) if future is None: adapter_ok = False target_errors.append("live adapter event loop scheduling failed") else: send_result = None timeout_handled = False try: send_result = future.result(timeout=60) except TimeoutError: # #38922: a slow confirmation does NOT necessarily # mean the send failed — but we must distinguish two # cases via future.cancel()'s return value: # # cancel() == False -> the coroutine was already # running on the gateway loop when the timeout # fired; the request is in flight on the wire and # cannot be un-sent. Re-sending via standalone # would be a guaranteed DUPLICATE, so treat it as # delivered (assume-delivered). # # cancel() == True -> the scheduled callback never # started executing (loop wedged/backlogged for # the full 60s), so nothing was sent. We MUST # fall through to the standalone path or the # message is silently dropped (worse than a # duplicate). cancelled = future.cancel() if cancelled: msg = ( f"live adapter send to {platform_name}:{chat_id} " "timed out before the coroutine was dispatched" ) logger.warning( "Job '%s': %s, falling back to standalone", job["id"], msg, ) target_errors.append(msg) adapter_ok = False # fall through to standalone path timeout_handled = True else: timed_out = True timeout_handled = True logger.warning( "Job '%s': live adapter send to %s:%s timed out " "after 60s; already dispatched (in flight), " "assuming delivered (skipping standalone fallback " "to avoid duplicate)", job["id"], platform_name, chat_id, ) except Exception as ex: # A real send error (not a slow confirmation) — fall # through to the standalone path so the message is # still delivered. target_errors.append(f"live adapter send failed: {ex}") raise if timeout_handled: # The timeout branch above already decided the # outcome (assume-delivered if in flight, or # adapter_ok=False to fall through if never # dispatched). send_result is None, so skip the # confirmation/thread-fallback inspection below. pass else: # _deliver_to_platform returns either a SendResult # (.success attr) or, when the silence-narration # filter drops the message, a plain dict # {"success": True, "delivered": False, ...}. # Normalize both shapes so a getattr default doesn't # misread a dict, and so a None / success-less object # is NOT counted as delivered (#47056). The # confirmation itself handles both shapes: a truthy # `success` with `delivered: False` is a drop, not a # delivery (#77763). if isinstance(send_result, dict): send_raw_response = send_result.get("raw_response") delivered_message_id = send_result.get("message_id") else: send_raw_response = getattr(send_result, "raw_response", None) delivered_message_id = getattr(send_result, "message_id", None) _evidence_gap: list = [] send_success = _confirm_adapter_delivery( send_result, job["id"], _evidence_gap, ) if send_success and _evidence_gap: unverified_targets.append(f"{platform_name}:{chat_id}") if not send_success: if isinstance(send_result, dict): # A filtered drop carries no "error" — name # the filter instead of reporting "unknown". err = ( send_result.get("error") or send_result.get("filtered") or "unknown" ) shape = "dict" elif send_result is not None: err = getattr(send_result, "error", None) shape = type(send_result).__name__ else: err = "no response from adapter" shape = "None" msg = ( f"live adapter send to {platform_name}:{chat_id} " f"returned unconfirmed result ({shape}, error={err})" ) if transport is not None and transport.is_relay: logger.warning("Job '%s': %s", job["id"], msg) else: logger.warning( "Job '%s': %s, falling back to standalone", job["id"], msg, ) target_errors.append(msg) adapter_ok = False # fall through to standalone path elif ( send_raw_response and thread_id and send_raw_response.get("thread_fallback") ): requested_thread_id = send_raw_response.get("requested_thread_id") or thread_id msg = ( f"configured thread_id {requested_thread_id} for " f"{platform_name}:{chat_id} was not found; delivered without thread_id" ) logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) # Send extracted media files as native attachments via the live # adapter, using the same DM-topic-aware routing as the text send # (#22773 — media previously used a bare thread_id and landed in # the General lane for private DM topics). Skip on an in-flight # confirmation timeout: the gateway loop is contended, so each # media send would also block its 30s budget, and the text # payload is already assumed delivered (#38922). Record the # skipped attachments so the drop is visible rather than silently # lost. if adapter_ok and not timed_out and media_files: routed_media_metadata = dict(media_metadata or {}) if transport is not None and transport.is_relay: routed_media_metadata["_relay_logical_platform"] = platform.value logical_home = config.get_home_channel(platform) if logical_home is not None and logical_home.chat_id == chat_id: if logical_home.user_id: routed_media_metadata["user_id"] = logical_home.user_id if logical_home.scope_id: routed_media_metadata["scope_id"] = logical_home.scope_id _media_errors = _send_media_via_adapter( runtime_adapter, chat_id, media_files, routed_media_metadata or None, loop, job, platform=platform, ) # Surface per-file failures into the run status (parity # with the standalone lane): text delivered but an # attachment didn't is a visible partial failure, not ok. for _me in _media_errors: _msg = f"{_me} (target {platform_name}:{chat_id})" delivery_errors.append(_msg) elif timed_out and media_files: msg = ( f"{len(media_files)} media attachment(s) not delivered to " f"{platform_name}:{chat_id} (live adapter confirmation timed out)" ) logger.warning("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) if adapter_ok: # Log WHERE it went, not just that it went: a ghost delivery # that landed in the wrong lane (General topic instead of the # routed thread) is indistinguishable from a real one without # the routing identity (#77763). logger.info( "Job '%s': delivered to %s:%s via live adapter thread=%s message_id=%s", job["id"], platform_name, chat_id, route_thread_id if route_thread_id is not None else "-", delivered_message_id if delivered_message_id is not None else "-", ) delivered = True # Seed the thread session only now that delivery into it # succeeded (deferred from thread-open above). if opened_thread_id and not thread_seeded: _seed_cron_thread_session( job, runtime_adapter, platform_name, chat_id, opened_thread_id, mirror_text, chat_name=origin.get("chat_name"), is_dm=is_dm_target, scope_id=origin.get("scope_id"), ) thread_seeded = True # in_channel surface: CREATE + seed the flat channel/DM # session (the shipped mirror only appends to an existing # session — the flat row is otherwise absent for a # chat_postMessage delivery, so the brief would be lost). # Gated on `inchannel_continuable` — the SHARED gate with # the thread-flatten above (they must not drift, or the # brief and its continuation session land in different # places). Origin targets seed without requiring the # mirror opt-in: in_channel IS the continuation surface — # a continuable flat cron without its seed is a brief the # next reply can't see (the bug Victor hit live # 2026-08-19: agent had "no idea about the delivery # message"). Mirror-eligible NON-origin targets # (origin_fallback / opted-in explicit — see # _target_mirror_eligible) also seed, guarded by # _inchannel_seed_allowed inside the gate: group-channel # keys are user-isolated, so a seed without a user_id # (origin-less managed cron into a shared channel) would # create an orphan session no reply resolves to — those # fall back to the plain mirror instead. if in_channel_surface and inchannel_continuable and not thread_seeded: inchannel_seeded = _seed_cron_channel_session( job, runtime_adapter, platform_name, chat_id, mirror_text, is_dm=is_dm_target, user_id=origin_user_id, chat_name=origin.get("chat_name"), scope_id=origin.get("scope_id"), ) if not inchannel_seeded: logger.warning( "Job '%s': in_channel seed did NOT land on %s:%s " "— a plain reply will not see this brief", job["id"], platform_name, chat_id, ) # Companion THREAD-surface seed (live gap, Alice # 2026-08-19): a flat brief is still a Slack message # the user can reply to IN ITS THREAD — the natural # mobile/desktop affordance — and that reply keys to # (chat, thread=), a session the flat seed # never touches. Seed it too so BOTH reply surfaces # continue the job. Uses the delivered message id as # the thread anchor; best-effort like every seed. if delivered_message_id: _seed_cron_thread_session( job, runtime_adapter, platform_name, chat_id, str(delivered_message_id), mirror_text, chat_name=origin.get("chat_name"), is_dm=is_dm_target, scope_id=origin.get("scope_id"), ) elif in_channel_surface and not inchannel_continuable: logger.warning( "Job '%s': in_channel delivery to %s:%s is not a " "continuable target (origin=%s:%s thread=%s; not the " "origin conversation, and not a mirror-eligible " "fallback/opted-in target the seed can key) — seed " "skipped; the plain mirror below may still apply", job["id"], platform_name, chat_id, origin.get("platform"), origin.get("chat_id"), origin.get("thread_id"), ) _maybe_mirror_cron_delivery( job, platform_name, chat_id, mirror_text, thread_id=thread_id, user_id=origin_user_id, enabled=mirror_this_target and not thread_seeded and not inchannel_seeded, ) except Exception as e: err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}" if not any(err_msg in err for err in target_errors): target_errors.append(err_msg) if transport is not None and transport.is_relay: logger.warning("Job '%s': %s", job["id"], err_msg) else: logger.warning( "Job '%s': %s, falling back to standalone", job["id"], err_msg, ) if not delivered: if transport is not None and transport.is_relay: # Relay owns the logical destination and its connector owns the # platform credential. A native retry could duplicate delivery # and cannot be authenticated correctly, so fail closed. if not target_errors: target_errors.append( f"relay delivery to {platform_name}:{chat_id} failed" ) delivery_errors.extend(target_errors) continue # If the interpreter is finalizing (gateway SIGTERM / restart / # OOM), scheduling any new delivery is futile — asyncio.run and a # fresh ThreadPoolExecutor both raise "cannot schedule new futures # after interpreter shutdown". Skip gracefully with a warning # rather than emitting an ERROR traceback on every restart-race # (#58720, #55924). if _interpreter_shutting_down(): msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down" logger.warning("Job '%s': %s", job["id"], msg) target_errors.append(msg) delivery_errors.extend(target_errors) continue # The live lane already failed closed on an empty payload; the # standalone senders do not. The Telegram adapter returns # SendResult(success=True) for empty content WITHOUT an API call, # so falling through here turns a phantom live delivery into a # phantom standalone one and logs it as delivered (#77763). Both # _send_to_platform call sites below are reached through this # point, so one guard closes the lane. if not cleaned_delivery_content.strip() and not media_files: msg = ( f"standalone send skipped (empty text and no media) " f"for {platform_name}:{chat_id}" ) logger.warning("Job '%s': %s", job["id"], msg) target_errors.append(msg) delivery_errors.extend(target_errors) continue # Standalone path: run the async send in a fresh event loop (safe from any thread) coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files) try: result = asyncio.run(coro) except RuntimeError as run_err: # asyncio.run() checks for a running loop before awaiting the coroutine; # when it raises, the original coro was never started — close it to # prevent "coroutine was never awaited" RuntimeWarning, then retry in a # fresh thread that has no running loop. coro.close() # If the RuntimeError is the interpreter-finalization signal, # the fresh-thread fallback would fail identically — skip # gracefully instead of logging a shutdown-race traceback. if _interpreter_shutting_down(run_err): msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down" logger.warning("Job '%s': %s", job["id"], msg) target_errors.append(msg) delivery_errors.extend(target_errors) continue # The thread-pool fallback can itself raise (SMTP ConnectionError, # future.result timeout, etc.). An exception raised inside this # `except RuntimeError` block is NOT caught by the sibling # `except Exception` below — it would escape _deliver_result() # and crash the whole delivery loop, silently skipping every # remaining target (#47163). Wrap the fallback in its own # try/except so a per-target failure is logged and the loop # continues to the next target. try: pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) try: # The fallback worker is a fresh thread: it does NOT # inherit the multiplexed profile ContextVars (home # override + secret scope). Run inside a copy of the # active context so the standalone sender reads THIS # profile's bot token, not the process default's # (#100489) — same pattern as the session-db and # heartbeat workers in this module. _fallback_context = contextvars.copy_context() future = pool.submit( _fallback_context.run, asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files), ) result = future.result(timeout=30) finally: pool.shutdown(wait=False) except Exception as e: # A shutdown-race here is expected during teardown; downgrade # to a warning so it doesn't read as a genuine failure. if _interpreter_shutting_down(e): msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down" logger.warning("Job '%s': %s", job["id"], msg) target_errors.append(msg) delivery_errors.extend(target_errors) continue msg = f"delivery to {platform_name}:{chat_id} failed: {e}" logger.error("Job '%s': %s", job["id"], msg, exc_info=True) target_errors.extend([msg]) delivery_errors.extend(target_errors) continue except Exception as e: msg = f"delivery to {platform_name}:{chat_id} failed: {e}" logger.error("Job '%s': %s", job["id"], msg, exc_info=True) target_errors.extend([msg]) delivery_errors.extend(target_errors) continue if result and result.get("error"): # Include target context (platform/chat) so a bare error string # like "Discord send failed: TimeoutError: " is attributable. # Not inside an except block — the error comes from the send # result dict, so there is no traceback to attach. msg = f"delivery error: {result['error']} (target {platform_name}:{chat_id})" logger.error("Job '%s': %s", job["id"], msg) target_errors.extend([msg]) delivery_errors.extend(target_errors) continue # Standalone senders report per-file attachment failures in # ``warnings`` while still returning success (the text leg # delivered). Surface them: a cron whose PDF/image silently # vanished used to mark the run ok with no trace — the exact # "manual run delivers text but no attachment" field report. _sender_warnings = ( result.get("warnings") if isinstance(result, dict) else None ) or [] for _w in _sender_warnings: msg = f"delivery warning: {_w} (target {platform_name}:{chat_id})" logger.error("Job '%s': %s", job["id"], msg) delivery_errors.append(msg) logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id) _maybe_mirror_cron_delivery( job, platform_name, chat_id, mirror_text, thread_id=thread_id, user_id=origin_user_id, enabled=mirror_this_target and not thread_seeded, ) if policy_drop_errors: # Filter-time drops apply to every target; report them once. delivery_errors.extend(policy_drop_errors) _record_delivery_verification(job, unverified_targets) if delivery_errors: return "; ".join(delivery_errors) return None def drain_delivery_queue(adapters, loop) -> int: """Send queued worker results through this gateway's live adapters.""" from cron.delivery_queue import _path, drain # Only restart-safe workers create the queue file. Every gateway (macOS, # Windows, launchd, Docker) runs this housekeeping tick, so skip the sqlite # open/create entirely until a worker has actually queued something. if not _path().exists(): return 0 return drain( lambda queued_job, queued_content, queued_for_failure: _deliver_result( queued_job, queued_content, adapters=adapters, loop=loop, for_failure=queued_for_failure, ) ) _DEFAULT_SCRIPT_TIMEOUT = 3600 # seconds (1 hour) # Backward-compatible module override used by tests and emergency monkeypatches. _SCRIPT_TIMEOUT = _DEFAULT_SCRIPT_TIMEOUT _RUN_CLAIM_HEARTBEAT_SECONDS = 60.0 _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS = _RUN_CLAIM_HEARTBEAT_SECONDS * 3 def _get_script_timeout() -> int: """Resolve cron pre-run script timeout from module/env/config with a safe default.""" if _SCRIPT_TIMEOUT != _DEFAULT_SCRIPT_TIMEOUT: try: timeout = int(float(_SCRIPT_TIMEOUT)) if timeout > 0: return timeout except Exception: logger.warning("Invalid patched _SCRIPT_TIMEOUT=%r; using env/config/default", _SCRIPT_TIMEOUT) env_value = os.getenv("HERMES_CRON_SCRIPT_TIMEOUT", "").strip() if env_value: try: timeout = int(float(env_value)) if timeout > 0: return timeout except Exception: logger.warning("Invalid HERMES_CRON_SCRIPT_TIMEOUT=%r; using config/default", env_value) try: cfg = load_config() or {} cron_cfg = cfg.get("cron", {}) if isinstance(cfg, dict) else {} configured = cron_cfg.get("script_timeout_seconds") if configured is not None: timeout = int(float(configured)) if timeout > 0: return timeout except Exception as exc: logger.debug("Failed to load cron script timeout from config: %s", exc) return _DEFAULT_SCRIPT_TIMEOUT _DEFAULT_MEDIA_SEND_TIMEOUT = 300 def _get_media_send_timeout() -> int: """Resolve the per-attachment media-send timeout from env/config. Mirrors the ``script_timeout_seconds`` resolution pattern: the HERMES_CRON_MEDIA_SEND_TIMEOUT env var wins, then ``cron.media_send_timeout_seconds`` in config.yaml, then the default (300s — large attachments like long TTS audio can legitimately exceed the old fixed 30s upload window). """ env_value = os.getenv("HERMES_CRON_MEDIA_SEND_TIMEOUT", "").strip() if env_value: try: timeout = int(float(env_value)) if timeout > 0: return timeout except Exception: logger.warning( "Invalid HERMES_CRON_MEDIA_SEND_TIMEOUT=%r; using config/default", env_value, ) try: cfg = load_config() or {} cron_cfg = cfg.get("cron", {}) if isinstance(cfg, dict) else {} configured = cron_cfg.get("media_send_timeout_seconds") if configured is not None: timeout = int(float(configured)) if timeout > 0: return timeout except Exception as exc: logger.debug("Failed to load cron media-send timeout from config: %s", exc) return _DEFAULT_MEDIA_SEND_TIMEOUT def _get_session_db_timeout() -> float: """Resolve the bound on run_job's SessionDB init from env/config. Mirrors the ``script_timeout_seconds`` resolution pattern: the HERMES_CRON_SESSION_DB_TIMEOUT env var wins, then ``cron.session_db_timeout_seconds`` in config.yaml (present in DEFAULT_CONFIG, so ``load_config()``'s deep-merge supplies it), then 10s. Unlike the sibling timeouts, 0 is meaningful (unlimited — legacy behavior, opt-in for debugging), so values are passed through untouched. """ env_value = os.getenv("HERMES_CRON_SESSION_DB_TIMEOUT", "").strip() if env_value: try: return float(env_value) except (ValueError, TypeError): logger.warning( "Invalid HERMES_CRON_SESSION_DB_TIMEOUT=%r; using config/default", env_value, ) try: cfg = load_config() or {} cron_cfg = cfg.get("cron", {}) if isinstance(cfg, dict) else {} configured = cron_cfg.get("session_db_timeout_seconds") if configured is not None: return float(configured) except Exception as exc: logger.debug( "Failed to load cron.session_db_timeout_seconds from config: %s", exc ) return 10.0 def _read_windows_pyvenv_cfg(venv_dir: Path) -> dict[str, str]: cfg_path = venv_dir / "pyvenv.cfg" try: lines = cfg_path.read_text(encoding="utf-8").splitlines() except OSError: return {} parsed: dict[str, str] = {} for raw in lines: if "=" not in raw: continue key, value = raw.split("=", 1) parsed[key.strip().lower()] = value.strip() return parsed def _windows_cron_python_invocation(python_exe: str) -> tuple[str, dict[str, str]]: """Return an output-capable hidden Python invocation for Windows scripts. Cron scripts capture stdout/stderr, so using ``pythonw.exe`` directly can lose script output. uv-created venv ``python.exe`` launchers are also a problem: even with CREATE_NO_WINDOW, the launcher can re-exec the base console interpreter and flash a visible window. For uv venvs, bypass the launcher and run the base ``python.exe`` directly with the venv paths overlaid in the environment. """ if sys.platform != "win32": return python_exe, {} interpreter = Path(python_exe) venv_dir = interpreter.parent.parent env_overlay: dict[str, str] = {} if interpreter.name.lower() == "pythonw.exe": sibling = interpreter.with_name("python.exe") if sibling.exists(): interpreter = sibling cfg = _read_windows_pyvenv_cfg(venv_dir) home = cfg.get("home", "") site_packages = venv_dir / "Lib" / "site-packages" if "uv" in cfg and home: base_python = Path(home) / "python.exe" if base_python.exists() and site_packages.exists(): interpreter = base_python env_overlay["VIRTUAL_ENV"] = str(venv_dir) pythonpath_entries = [ str(Path(__file__).resolve().parents[1]), str(site_packages), ] existing_pythonpath = os.environ.get("PYTHONPATH", "") if existing_pythonpath: pythonpath_entries.append(existing_pythonpath) env_overlay["PYTHONPATH"] = os.pathsep.join(pythonpath_entries) return str(interpreter), env_overlay def _terminate_cron_script_process(proc: subprocess.Popen) -> None: """Best-effort hard stop of a cron script and every child it spawned.""" if proc.poll() is not None: return if sys.platform == "win32": try: subprocess.run( ["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True, timeout=10, creationflags=windows_hide_flags(), check=False, ) except (OSError, subprocess.TimeoutExpired): proc.kill() else: try: process_group: Optional[int] = os.getpgid(proc.pid) except (ProcessLookupError, OSError): process_group = None if process_group is not None: try: os.killpg(process_group, signal.SIGTERM) # windows-footgun: ok — POSIX-only branch (win32 handled above) except (ProcessLookupError, PermissionError, OSError): process_group = None if process_group is not None: try: proc.wait(timeout=1.0) except subprocess.TimeoutExpired: pass # Escalate whenever ANY group member survived the TERM: a # TERM-ignoring descendant keeps the stdio pipe write ends # open, and the caller's communicate() would then block on # EOF forever. killpg(pgid, 0) probes group liveness. try: os.killpg(process_group, 0) # windows-footgun: ok — POSIX-only branch except (ProcessLookupError, OSError): process_group = None if process_group is not None: try: os.killpg(process_group, getattr(signal, "SIGKILL", signal.SIGTERM)) except (ProcessLookupError, PermissionError, OSError): pass try: proc.wait(timeout=1.0) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=1.0) def _terminate_cron_script_tree(proc: subprocess.Popen) -> None: """Terminate a script tree, then fall back to the local process-group path.""" if proc.poll() is not None: # Already exited (e.g. finished right at the deadline): nothing to # signal, and calling kill_process_tree on a reaped pid would log a # spurious "no signal" warning. Mirrors _terminate_cron_script_process. return pid = getattr(proc, "pid", None) if not isinstance(pid, int) or pid <= 0: logger.warning( "Cron script tree-kill received invalid pid %r; " "falling back to process-group termination", pid, ) _terminate_cron_script_process(proc) return try: # Function-local so tests can monkeypatch agent.deadline.kill_process_tree; # separate from the kill try below so a packaging/import problem # surfaces as what it is instead of masquerading as a kill failure. from agent.deadline import kill_process_tree except Exception: logger.warning( "agent.deadline.kill_process_tree unavailable; " "falling back to process-group termination", exc_info=True, ) _terminate_cron_script_process(proc) return try: if kill_process_tree(pid): return logger.warning( "Cron script tree-kill reported no signal for pid %s; " "falling back to process-group termination", pid, ) except Exception: logger.warning( "Cron script tree-kill failed for pid %s; " "falling back to process-group termination", pid, exc_info=True, ) _terminate_cron_script_process(proc) def _drain_script_pipes(proc: subprocess.Popen) -> None: """Reap a terminated script process without ever blocking indefinitely. A descendant that survived the tree kill can hold the pipe write ends open, so a bare ``communicate()`` would wait for EOF forever. Bound the drain, then abandon the pipes — the caller only needs the process reaped and the worker thread unblocked, not the output. """ try: proc.communicate(timeout=5.0) return except subprocess.TimeoutExpired: pass try: proc.kill() except OSError: pass for stream in (proc.stdout, proc.stderr): try: if stream is not None: stream.close() except OSError: pass try: proc.wait(timeout=5.0) except subprocess.TimeoutExpired: # Truly wedged — leave the zombie to the OS reaper rather than # blocking the cron worker thread forever. pass def _windows_cron_bootstrap_argv( python_exe: str, env_overlay: dict[str, str], script_path: str, ) -> list[str]: """Bootstrap a cron script under the base interpreter with ``.pth`` support. The uv-venv overlay mode runs the base ``python.exe`` (to avoid the launcher re-execing a console interpreter and flashing a window) and re-attaches the venv via ``PYTHONPATH``. But ``PYTHONPATH`` entries are plain ``sys.path`` additions — Python's site initialization never processes ``.pth`` files for them (only ``site.addsitedir()`` does) — so editable installs (``pip install -e``, ``__editable__*.pth`` links) are invisible to cron script jobs. Bootstrap with ``site.addsitedir()`` on the venv ``site-packages``, then exec the script as ``__main__``. ``runpy.run_path`` keeps ``__file__`` correct; ``sys.path[0]`` is set to the script's directory to preserve the ``python script.py`` import semantics. Note: ``runpy`` does not set ``__package__``/``__spec__`` the way a direct invocation does, so package-relative imports (``from . import x``) may behave differently. Falls back to a plain invocation if the venv layout is unresolvable — the pre-existing PYTHONPATH behaviour is strictly better than failing to run at all. """ site_packages = Path(env_overlay.get("VIRTUAL_ENV", "")) / "Lib" / "site-packages" if not site_packages.is_dir(): # Silent here would make the "editable installs invisible" failure # undiagnosable; the pre-existing PYTHONPATH-only behaviour applies. logger.warning( "Windows cron script: venv site-packages %s not found; running " "without .pth processing (editable installs may be unimportable)", site_packages, ) return [python_exe, script_path] bootstrap = ( "import os, runpy, site, sys;" f"site.addsitedir({str(site_packages)!r});" "script = sys.argv[1];" "sys.argv = [script] + sys.argv[2:];" "sys.path.insert(0, os.path.dirname(os.path.abspath(script)));" "runpy.run_path(script, run_name='__main__')" ) return [python_exe, "-c", bootstrap, script_path] def _run_job_script( script_path: str, workdir: Optional[str] = None, cancel_event: Optional[_CancelEventLike] = None, ) -> tuple[bool, str]: """Execute a cron job's data-collection script and capture its output. Scripts must reside within HERMES_HOME/scripts/. Both relative and absolute paths are resolved and validated against this directory to prevent arbitrary script execution via path traversal or absolute path injection. Supported interpreters (chosen by file extension): * ``.sh`` / ``.bash`` — run with ``/bin/bash`` * anything else — run with the current Python interpreter (``sys.executable``), preserving the original behaviour for Python-based pre-check and data-collection scripts. Shell support lets ``no_agent=True`` jobs ship classic bash watchdogs (the `memory-watchdog.sh` pattern) without wrapping them in Python. Subprocess environment is passed through ``_sanitize_subprocess_env`` so provider credentials and other Hermes-managed secrets are not inherited (SECURITY.md §2.3), matching terminal and MCP child processes. Args: script_path: Path to the script. Relative paths are resolved against HERMES_HOME/scripts/. Absolute and ~-prefixed paths are also validated to ensure they stay within the scripts dir. workdir: Optional absolute path to use as the script's cwd. When set, the subprocess runs in this directory instead of the scripts-dir parent. The Python process cwd is NEVER mutated, avoiding the global-side-effect bug where a cron job's ``os.chdir()`` leaks into concurrent gateway sessions (#69396). Returns: (success, output) — on failure *output* contains the error message so the LLM can report the problem to the user. """ scripts_dir = _get_hermes_home() / "scripts" _ensure_cron_dir(scripts_dir) scripts_dir_resolved = scripts_dir.resolve() # Same ingestion contract as cron.lifecycle_guard._expand_candidate_path: # a NUL-bearing value can never name a real script, and on Windows the # Path operations raise ValueError *after* expanduser (expanduser never # expands "~user" there, so the try below never fires) — reject eagerly # so both platforms fail cleanly instead of crashing the scheduler. # str() first so the guard itself can never raise TypeError on a # non-str script_path (e.g. a Path passed by a future caller) — the # guard must be crash-proof even though every current call site # passes a plain str (#86832 review). if "\x00" in str(script_path): return False, f"Blocked: script path contains a NUL byte: {script_path!r}" try: raw = Path(script_path).expanduser() except (ValueError, RuntimeError, OSError): # Same ingestion contract as cron.lifecycle_guard: a NUL-bearing # value (ValueError) or an unexpandable ``~`` (RuntimeError with no # resolvable HOME) can never name a real script. The creation-time # guard tolerates such values as "nothing to scan", so they can # reach fire time — fail the run with a report instead of crashing # the scheduler with an unhandled exception. return False, f"Blocked: script path is not a valid filesystem path: {script_path!r}" if raw.is_absolute(): path = raw.resolve() else: path = (scripts_dir / raw).resolve() # Guard against path traversal, absolute path injection, and symlink # escape — scripts MUST reside within HERMES_HOME/scripts/. try: path.relative_to(scripts_dir_resolved) except ValueError: return False, ( f"Blocked: script path resolves outside the scripts directory " f"({scripts_dir_resolved}): {script_path!r}" ) if not path.exists(): return False, f"Script not found: {path}" if not path.is_file(): return False, f"Script path is not a file: {path}" script_timeout = _get_script_timeout() # Pick an interpreter by extension. Bash for .sh/.bash, Python for # everything else. We deliberately do NOT honour the file's own # shebang: the scripts dir is trusted, but keeping the interpreter # choice explicit here keeps the allowed surface small and auditable. suffix = path.suffix.lower() if suffix in {".sh", ".bash"}: # Resolve bash dynamically so Windows (Git Bash) and Linux/macOS # all work. On native Windows without Git for Windows installed # shutil.which returns None — fall back to a clear error rather # than a FileNotFoundError with a confusing "[WinError 2]" # traceback. _bash = shutil.which("bash") or ( "/bin/bash" if os.path.isfile("/bin/bash") else None ) if _bash is None: return False, ( f"Cannot run .sh/.bash script {path.name!r}: bash not found on PATH. " "On Windows, install Git for Windows (which ships Git Bash) " "or rewrite the script as Python (.py)." ) argv = [_bash, str(path)] env_overlay: dict[str, str] = {} else: python_exe, env_overlay = _windows_cron_python_invocation(sys.executable) if env_overlay: # Overlay mode (Windows uv venv): PYTHONPATH alone cannot make # editable installs importable — .pth processing needs # site.addsitedir() (see _windows_cron_bootstrap_argv). argv = _windows_cron_bootstrap_argv(python_exe, env_overlay, str(path)) else: argv = [python_exe, str(path)] try: from tools.environments.local import build_subprocess_env popen_kwargs: dict[str, Any] = {"start_new_session": True} if sys.platform == "win32": popen_kwargs = { "creationflags": windows_hide_flags() | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0), "encoding": "utf-8", "errors": "replace", } env = build_subprocess_env() env.update(env_overlay) # Use the job's workdir as the subprocess cwd when configured, # otherwise default to the scripts-dir parent (back-compat). # NEVER mutate the Python process cwd — that would leak into # concurrent gateway sessions (#69396). _script_cwd = workdir or str(path.parent) proc = subprocess.Popen( argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=_script_cwd, env=env, **popen_kwargs, ) deadline = time.monotonic() + script_timeout while True: if cancel_event is not None and cancel_event.is_set(): # Same bug class as the timeout site below: a cancelled fire # must not orphan own-session grandchildren either. _terminate_cron_script_tree(proc) _drain_script_pipes(proc) return False, "Script cancelled because cron fire ownership was lost" remaining = deadline - time.monotonic() if remaining <= 0: # Phase 4a (#85125): a script timeout must leave ZERO living # descendants. killpg only reaches the script's own process # group — a grandchild that called setsid (backgrounded # shell jobs, watchdogs) escapes it and keeps running after # the job reports failure (#71148 / #59549). # agent.deadline.kill_process_tree snapshots the descendant # set via psutil BEFORE signalling, so own-session # grandchildren are reached too — the unified deadline # layer's tree-kill (#85147, d6a5cb9725). _terminate_cron_script_tree(proc) _drain_script_pipes(proc) return False, f"Script timed out after {script_timeout}s: {path}" try: stdout_raw, stderr_raw = proc.communicate(timeout=min(0.1, remaining)) break except subprocess.TimeoutExpired: continue stdout = (stdout_raw or "").strip() stderr = (stderr_raw or "").strip() # Redact secrets from both stdout and stderr before any return path. try: from agent.redact import redact_sensitive_text stdout = redact_sensitive_text(stdout) stderr = redact_sensitive_text(stderr) except Exception as e: logger.warning("Failed to redact sensitive text from output: %s", e) stdout = "[REDACTED - redaction failed]" stderr = "[REDACTED - redaction failed]" if proc.returncode != 0: parts = [f"Script exited with code {proc.returncode}"] if stderr: parts.append(f"stderr:\n{stderr}") if stdout: parts.append(f"stdout:\n{stdout}") return False, "\n".join(parts) return True, stdout except Exception as exc: return False, f"Script execution failed: {exc}" def _run_job_script_with_claim_heartbeat( job: dict, script_path: str, workdir: Optional[str] = None, cancel_event: Optional[_CancelEventLike] = None, ) -> tuple[bool, str]: """Run a cron script while keeping its owned one-shot claim fresh. Script execution is synchronous and may legitimately outlive the stale claim TTL. Without a concurrent heartbeat, another scheduler process can mistake the live run for a dead owner and dispatch the same one-shot again. Recurring jobs and unclaimed/manual runs have no durable one-shot claim and therefore use the ordinary script path without starting a thread. The claim owner is captured from the dispatched job and never re-read from storage. ``heartbeat_run_claim`` compares that stable owner before every refresh, so a stale runner cannot extend a replacement owner's claim. """ schedule = job.get("schedule") claim = job.get("run_claim") owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" if not ( isinstance(schedule, dict) and schedule.get("kind") == "once" and owner ): return _run_job_script(script_path, workdir=workdir, cancel_event=cancel_event) job_id = str(job.get("id") or "") stop = threading.Event() heartbeat_context = contextvars.copy_context() def _heartbeat_loop() -> None: while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: heartbeat_run_claim(job_id, expected_owner=owner) except Exception: logger.debug( "Job '%s': script run_claim heartbeat failed", job_id, exc_info=True, ) heartbeat_thread = threading.Thread( target=heartbeat_context.run, args=(_heartbeat_loop,), name="cron-script-claim-heartbeat", daemon=True, ) try: heartbeat_thread.start() except Exception: logger.debug( "Job '%s': could not start script run_claim heartbeat", job_id, exc_info=True, ) return _run_job_script(script_path, workdir=workdir, cancel_event=cancel_event) try: return _run_job_script(script_path, workdir=workdir, cancel_event=cancel_event) finally: stop.set() # Event.wait() wakes immediately. Keep completion bounded if the # heartbeat is already waiting on another process's jobs-file lock. heartbeat_thread.join(timeout=1.0) def _parse_wake_gate(script_output: str) -> bool: """Parse the last non-empty stdout line of a cron job's pre-check script as a wake gate. The convention (ported from nanoclaw #1232): if the last stdout line is JSON like ``{"wakeAgent": false}``, the agent is skipped entirely — no LLM run, no delivery. Any other output (non-JSON, missing flag, gate absent, or ``wakeAgent: true``) means wake the agent normally. Returns True if the agent should wake, False to skip. """ if not script_output: return True stripped_lines = [line for line in script_output.splitlines() if line.strip()] if not stripped_lines: return True last_line = stripped_lines[-1].strip() try: gate = json.loads(last_line) except (json.JSONDecodeError, ValueError): return True if not isinstance(gate, dict): return True return gate.get("wakeAgent", True) is not False def _build_job_prompt( job: dict, prerun_script: Optional[tuple] = None, extra_prompt: Optional[str] = None, ) -> str: """Build the effective prompt for a cron job, optionally loading one or more skills first. Args: job: The cron job dict. prerun_script: Optional ``(success, stdout)`` from a script that has already been executed by the caller (e.g. for a wake-gate check). When provided, the script is not re-executed and the cached result is used for prompt injection. When omitted, the script (if any) runs inline as before. extra_prompt: Optional per-run context (from ``cronjob(action='run')``, #57331 — salvaged from #57342 by @liuhao1024). Appended to the stored prompt under a ``## Run Context`` header for this single fire only — never persisted to the job definition. """ user_prompt = str(job.get("prompt") or "") if extra_prompt: user_prompt = f"{user_prompt}\n\n## Run Context\n{extra_prompt}" prompt = user_prompt skills = job.get("skills") # True when runtime-collected DATA (script stdout, upstream-job output) # has been injected into the prompt. Data content legitimately quotes # command-shape strings (a triage feed ingesting a bug report that # pastes `rm -rf /`), so it must not be scanned with the strict # user-prompt pattern set — see _scan_assembled_cron_prompt. has_injected_data = False # Run data-collection script if configured, inject output as context. script_path = job.get("script") if script_path: if prerun_script is not None: success, script_output = prerun_script else: success, script_output = _run_job_script(script_path) if success: if script_output: prompt = ( "## Script Output\n" "The following data was collected by a pre-run script. " "Use it as context for your analysis.\n\n" f"```\n{script_output}\n```\n\n" f"{prompt}" ) has_injected_data = True else: # Script produced no output — nothing to report, skip AI call. return None else: prompt = ( "## Script Error\n" "The data-collection script failed. Report this to the user.\n\n" f"```\n{script_output}\n```\n\n" f"{prompt}" ) has_injected_data = True # Inject output from referenced cron jobs as context. context_from = job.get("context_from") if context_from: from cron.jobs import get_cron_output_dir output_dir = get_cron_output_dir() if isinstance(context_from, str): context_from = [context_from] for source_job_id in context_from: # "self" resolves to the job's own id: the job wakes up with its # most recent output injected, giving recurring jobs continuity # across runs (dedupe against what was already reported, continue # where the last run left off) without touching session history. is_self = False if isinstance(source_job_id, str) and source_job_id.strip().lower() == "self": source_job_id = str(job.get("id") or "") is_self = True elif source_job_id == job.get("id"): is_self = True # Guard against path traversal — valid job IDs are 12-char hex strings if not source_job_id or not all(c in "0123456789abcdef" for c in source_job_id): logger.warning( "context_from: skipping invalid job_id %r for job_id=%r name=%r%s", source_job_id, job.get("id"), job.get("name"), _cron_job_origin_log_suffix(job), ) continue try: job_output_dir = output_dir / source_job_id if not job_output_dir.exists(): continue # silent skip — no output yet output_files = sorted( job_output_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True, ) if not output_files: continue # silent skip — no output yet latest_output = output_files[0].read_text(encoding="utf-8").strip() # Truncate to 8K characters to avoid prompt bloat _MAX_CONTEXT_CHARS = 8000 if len(latest_output) > _MAX_CONTEXT_CHARS: latest_output = latest_output[:_MAX_CONTEXT_CHARS] + "\n\n[... output truncated ...]" if latest_output: if is_self: prompt = ( "## Your previous run's output\n" "The following is this job's most recent output from its " "previous run. Use it for continuity: avoid repeating what " "was already reported, and continue where the last run " "left off.\n\n" f"```\n{latest_output}\n```\n\n" f"{prompt}" ) else: prompt = ( f"## Output from job '{source_job_id}'\n" "The following is the most recent output from a preceding " "cron job. Use it as context for your analysis.\n\n" f"```\n{latest_output}\n```\n\n" f"{prompt}" ) has_injected_data = True else: continue # silent skip — empty output except (OSError, PermissionError) as e: logger.warning("context_from: failed to read output for job %r: %s", source_job_id, e) # silent skip — do not pollute the prompt with error messages # Inject the job's durable notepad (per-job KV scratchpad surviving # scheduled wake-ups). Empty notepad renders as "" so jobs that never # use the feature get a byte-identical prompt. from cron import notepad as cron_notepad notepad_section = cron_notepad.render_notepad_section(str(job.get("id") or "")) if notepad_section: prompt = f"{notepad_section}{prompt}" has_injected_data = True # Always prepend cron execution guidance so the agent knows how # delivery works and can suppress delivery when appropriate. cron_hint = ( "[IMPORTANT: You are running as a scheduled cron job. " "DELIVERY: Your final response will be automatically delivered " "to the user — do NOT use send_message or try to deliver " "the output yourself. Just produce your report/output as your " "final response and the system handles the rest. " "SILENT: If there is genuinely nothing new to report, respond " "with exactly \"[SILENT]\" (nothing else) to suppress delivery. " "Never combine [SILENT] with content — either report your " "findings normally, or say [SILENT] and nothing more.]\n\n" ) prompt = cron_hint + prompt if skills is None: legacy = job.get("skill") skills = [legacy] if legacy else [] elif isinstance(skills, str): skills = [skills] skill_names = [str(name).strip() for name in skills if str(name).strip()] if not skill_names: return _scan_assembled_cron_prompt( prompt, job, has_skills=False, has_injected_data=has_injected_data, user_prompt=user_prompt, ) from tools.skills_tool import skill_view from tools.skill_usage import bump_use from agent.skill_bundles import build_bundle_invocation_message, resolve_bundle_command_key from agent.skill_utils import normalize_skill_lookup_name parts = [] skipped: list[str] = [] for skill_name in skill_names: # Cron jobs historically accepted only skill names here, but the CLI/gateway # slash-command path lets bundles shadow skills with the same slug. Mirror # that behavior so `skills: ["my-bundle"]` expands bundle members instead # of being treated as a missing skill. bundle_key = resolve_bundle_command_key(skill_name.lstrip("/")) if bundle_key: bundle_payload = build_bundle_invocation_message( bundle_key, user_instruction="", task_id=str(job.get("id") or "") or None, ) if bundle_payload: bundle_message, _loaded_bundle_skills, _missing_bundle_skills = bundle_payload if parts: parts.append("") parts.append(bundle_message) continue logger.warning( "Cron job '%s': bundle '%s' could not load any skills, skipping", job.get("name", job.get("id")), skill_name, ) skipped.append(skill_name) continue try: loaded = json.loads(skill_view(normalize_skill_lookup_name(skill_name))) except (json.JSONDecodeError, TypeError): logger.warning("Cron job '%s': skill '%s' returned invalid JSON, skipping", job.get("name", job.get("id")), skill_name) skipped.append(skill_name) continue if not loaded.get("success"): error = loaded.get("error") or f"Failed to load skill '{skill_name}'" logger.warning("Cron job '%s': skill not found, skipping — %s", job.get("name", job.get("id")), error) skipped.append(skill_name) continue # Bump usage so the curator sees this skill as actively used. try: bump_use(skill_name, task_id=str(job.get("id") or "") or None) except Exception: logger.debug("Cron job: failed to bump skill usage for '%s'", skill_name, exc_info=True) content = str(loaded.get("content") or "").strip() if parts: parts.append("") parts.extend( [ f'[IMPORTANT: The user has invoked the "{skill_name}" skill, indicating they want you to follow its instructions. The full skill content is loaded below.]', "", content, ] ) if skipped: notice = ( f"[IMPORTANT: The following skill(s) were listed for this job but could not be found " f"and were skipped: {', '.join(skipped)}. " f"Start your response with a brief notice so the user is aware, e.g.: " f"'⚠️ Skill(s) not found and skipped: {', '.join(skipped)}']" ) parts.insert(0, notice) stable_prefix = None if prompt: from agent.skill_commands import append_user_instruction parts.append("") # The skill blocks (and any skipped-skill notice) above are stable per # job config; the appended instruction carries the volatile per-run # data (cron hint + prompt + script output + run context). Declare # that boundary for the Anthropic cache planner (#81867). stable_prefix = append_user_instruction(parts, prompt) assembled = _scan_assembled_cron_prompt("\n".join(parts), job, has_skills=True) if stable_prefix and len(assembled) > len(stable_prefix) and assembled.startswith(stable_prefix): # Guarded because the injection scanner may sanitize (mutate) the # assembled bytes; a mismatch simply falls back to whole-message # caching. from agent.prompt_cache_boundary import register_stable_prefix register_stable_prefix(stable_prefix) return assembled def _scan_assembled_cron_prompt( assembled: str, job: dict, *, has_skills: bool = False, has_injected_data: bool = False, user_prompt: Optional[str] = None, ) -> str: """Scan the fully-assembled cron prompt for injection patterns. Raises ``CronPromptInjectionBlocked`` when a match fires so ``run_job`` can surface a clear refusal to the operator. Plugs the #3968 gap: ``_scan_cron_prompt`` runs on the user-supplied prompt at create/update, but skill content is loaded from disk at runtime and was never scanned. Since cron runs non-interactively (auto-approves tool calls), a malicious skill carrying an injection payload bypassed every gate. Two pattern tiers, selected by what the assembled prompt CONTAINS, not just whether skills are attached: - When the assembled prompt is essentially the user prompt + the cron hint (no skills, no injected data), the STRICT ``_scan_cron_prompt`` patterns apply: a bare ``rm -rf /`` in a small directive prompt is a smoking gun, not prose. - When the assembled prompt includes runtime-loaded content — skill markdown (``has_skills=True``) or DATA injected from a job script's stdout / an upstream job's output (``has_injected_data=True``) — the LOOSER ``_scan_cron_skill_assembled`` pattern set is used: only unambiguous prompt-injection directives block; command-shape patterns are dropped and invisible unicode is sanitized (stripped + logged) rather than blocked, to avoid false-positives that permanently kill a job. Skill bodies are vetted at install time by ``skills_guard.py``; script output is produced by operator-authored code, the same trust class — and data feeds (e.g. a triage bot ingesting bug reports) legitimately quote dangerous commands. When the looser tier is selected because of injected data only, ``user_prompt`` (the raw, pre-assembly prompt) is additionally scanned with the STRICT set so the user-authored surface keeps the full create/update-time guarantee at runtime (defense-in-depth for legacy jobs that predate the create-time scanner). """ from tools.cronjob_tools import _scan_cron_prompt, _scan_cron_skill_assembled if has_skills or has_injected_data: # Runtime-loaded content (vetted skill markdown and/or data from # operator-authored scripts) legitimately contains command-shape # strings. Invisible unicode is sanitized (not blocked) so a stray # zero-width space can't permanently kill the job; the cleaned # prompt is what actually runs. cleaned, scan_error = _scan_cron_skill_assembled(assembled) assembled = cleaned if not scan_error and not has_skills and user_prompt: # Data-injection path: keep the strict guarantee on the # user-authored prompt itself. scan_error = _scan_cron_prompt(user_prompt) else: scan_error = _scan_cron_prompt(assembled) if scan_error: job_label = job.get("name") or job.get("id") or "" logger.warning( "Cron job '%s': assembled prompt blocked by injection scanner — %s", job_label, scan_error, ) raise CronPromptInjectionBlocked(scan_error) return assembled def _guard_job_credential_exfil(job: dict) -> None: """Fail closed if a job's stored provider/base_url pair would exfiltrate a credential (F8 runtime backstop; CWE-200/CWE-522). The model-callable cron tool validates this on create/update, but a job persisted before that guard — or written directly to the jobs store — reaches the scheduler's provider-resolution sink unchecked. Re-validate the EFFECTIVE stored pair with the same guard the tool uses, so a named provider's stored key is never paired with an off-host base_url at fire time. Raises ``RuntimeError`` (caught by the run_job failure path → the run is aborted and reported) when the pair is unsafe; returns ``None`` otherwise. Fallback providers come from operator config, not the model-callable job, so they are trusted and validated by the caller, not here. """ try: from tools.cronjob_tools import _validate_cron_base_url err = _validate_cron_base_url(job.get("provider"), job.get("base_url")) except Exception as exc: # Fail CLOSED: this is the last guard before provider resolution, so an # unexpected validator/import error must not silently allow an unvetted # pair through. A job that carries no base_url override cannot exfiltrate # a stored credential via this path (there is nothing to validate, and # the validator would return None), so it still runs — that keeps the # overwhelmingly-common no-override jobs from wedging on an unrelated # error. But any job that DID set a base_url is refused until the # validator can actually vet the pair. Operator fallback providers come # from config, not the job, so they are unaffected. if job.get("base_url"): err = ( f"could not validate provider/base_url pair " f"({exc.__class__.__name__}: {exc}); refusing to run a job with " "an unverified base_url override" ) else: err = None if err: job_id = job.get("id") logger.error( "Job '%s': refusing to run — unsafe provider/base_url pair could " "exfiltrate a stored credential: %s", job_id, err, ) raise RuntimeError(f"Cron job '{job_id}' blocked for safety: {err}") def _block_and_pause_job( job_id: str, job_name: str, reason: str ) -> tuple[bool, str, str, Optional[str]]: """Fail a run closed and pause the job so it stops being scheduled. Used for job shapes that can never run (a5e29e688dc0). Returning an error alone is not enough — an unrunnable job that stays enabled re-fires on every tick forever. Pausing writes ``paused_at``/``paused_reason``, giving an auditable record of why the scheduler stopped it. """ from cron.jobs import pause_job logger.error("Job '%s': %s", job_id, reason) try: pause_job(job_id, f"Auto-paused by scheduler: {reason}") except Exception: logger.exception("Job '%s': failed to auto-pause unrunnable job", job_id) now_iso = _hermes_now().strftime("%Y-%m-%d %H:%M:%S") doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {now_iso}\n" f"**Status:** blocked (unrunnable job) — auto-paused\n\n" f"{reason}\n" ) alert = f"⚠ Cron job '{job_name}' was auto-paused\n\n{reason}" return False, doc, alert, reason # Marker prefix stamped into the error string returned by ``run_job`` when the # pre-dispatch configuration validation (T1-26) refuses to run the agent. # ``run_one_job`` keys off it to record ``last_status='blocked_config'`` and to # apply the alert-once dedup. The ``:silent`` variant means "already alerted on # a previous tick — do not deliver again". BLOCKED_CONFIG_MARKER = "[blocked_config]" BLOCKED_CONFIG_SILENT_MARKER = "[blocked_config:silent]" # Marker prefix for a #44585 drift-guard skip. Same alert-once contract as # blocked_config: run_one_job keys off it to record last_status and the # ``:silent`` variant means "already alerted on a previous tick — do not # deliver again" (the drift_alerted bit on the job record, #73506 shape). DRIFT_SKIP_MARKER = "[drift_skip]" DRIFT_SKIP_SILENT_MARKER = "[drift_skip:silent]" def _is_transient_provider_resolve_error(exc: BaseException) -> bool: """True when primary provider resolution failed for a transient network reason. Agent crons resolve OAuth credentials (token refresh / discovery) before the agent loop starts. A short DNS outage (Cloudflare WARP / macOS resolver blip) surfaces as httpx/httpcore ConnectError or raw OSError errno 8 ("nodename nor servname provided") and must be eligible for ``fallback_providers`` the same way AuthError already is — otherwise a healthy XAI_API_KEY / Anthropic rung never gets tried and the whole job dies before the first model call. """ # Walk the cause chain; scheduler wraps raw transport errors. seen: set[int] = set() cur: Optional[BaseException] = exc while cur is not None and id(cur) not in seen: seen.add(id(cur)) name = type(cur).__name__ module = type(cur).__module__ or "" msg = str(cur).lower() # Explicit transport classes from httpx/httpcore/aiohttp. if name in { "ConnectError", "ConnectTimeout", "ReadTimeout", "WriteTimeout", "PoolTimeout", "NetworkError", "TimeoutException", "ClientConnectorError", "ClientConnectorDNSError", "ServerTimeoutError", "ClientOSError", }: return True if "httpx" in module or "httpcore" in module or "aiohttp" in module: if any( needle in msg for needle in ( "nodename nor servname", "name or service not known", "temporary failure in name resolution", "failed to resolve", "connection refused", "network is unreachable", "timed out", "timeout", ) ): return True if isinstance(cur, OSError): # Platform-safe classification (the raw-literal set {8, 7, 11, ...} # from the first revision mixed macOS getaddrinfo constants with # errno values and does not hold on Linux — see PR review). # socket.gaierror carries getaddrinfo codes (EAI_*), plain OSError # carries errno; compare each against its own constant namespace. import errno as _errno import socket as _socket if isinstance(cur, _socket.gaierror): _eai_transient = { getattr(_socket, _n) for _n in ("EAI_NONAME", "EAI_AGAIN", "EAI_FAIL", "EAI_NODATA") if hasattr(_socket, _n) } if cur.errno in _eai_transient: return True else: err_no = getattr(cur, "errno", None) if err_no in { _errno.ECONNREFUSED, _errno.ECONNRESET, _errno.EHOSTUNREACH, _errno.ENETUNREACH, _errno.ENETDOWN, _errno.ETIMEDOUT, _errno.EAGAIN, }: return True if any( needle in msg for needle in ( "nodename nor servname", "name or service not known", "temporary failure in name resolution", "network is unreachable", ) ): return True # Bare RuntimeError/Exception that already carries the DNS text # (format_runtime_provider_error sometimes surfaces the raw message). if "nodename nor servname" in msg or "name or service not known" in msg: return True cur = cur.__cause__ or cur.__context__ return False def _cron_preflight_enabled(cfg: dict) -> bool: """Whether cron pre-dispatch configuration validation is enabled. Default ON; only the literal boolean ``false`` under ``cron.preflight`` opts out (mirrors ``cron_model_drift_guard_enabled`` semantics). """ cron_cfg = (cfg or {}).get("cron") if not isinstance(cron_cfg, dict): return True return cron_cfg.get("preflight", True) is not False def _preflight_check_provider_key(job: dict, cfg: dict) -> Optional[str]: """READ-ONLY probe: would provider resolution fail for lack of a key? Mirrors the effective requested-provider computation from run_job's resolution block without any side effects on the run. When a fallback chain is configured the check is skipped entirely — the existing auth-fallback path may legitimately rescue a missing primary key, so blocking here would break that contract (and burning zero LLM calls is already guaranteed by the fallback resolution being config-local). """ try: if get_fallback_chain(cfg): return None except Exception: return None # fail-open: never block on a preflight-internal error _cron_cfg = cfg.get("cron") if isinstance(cfg.get("cron"), dict) else {} requested = ( job.get("provider") or str((_cron_cfg or {}).get("model_provider") or "").strip() or None ) model = job.get("model") or os.getenv("HERMES_MODEL") or "" from hermes_cli.auth import AuthError try: from hermes_cli.runtime_provider import resolve_runtime_provider kwargs = {"requested": requested, "target_model": model} if job.get("base_url"): kwargs["explicit_base_url"] = job.get("base_url") resolve_runtime_provider(**kwargs) except AuthError as exc: return ( f"provider credential missing: {exc}. " "Set the provider API key in .env (or `hermes setup`), or pin a " "working provider via `hermes cron edit " f"{job.get('id')} --provider

`." ) except Exception: # Non-auth resolution errors (bad config shapes, network probes, # import issues) are NOT a missing-credential condition — let the # real resolution path handle and report them as before. return None return None def _primary_profile_routes_for_current_home() -> list: """Primary gateway ``profile_routes`` that target the profile currently being served, or ``[]`` (also when this IS the primary home). Under ``gateway.multiplex_profiles`` a satellite profile's cron jobs are ticked by the primary gateway's in-process ticker (#69377) and delivered through the primary gateway's live adapters — the satellite home never holds the platform credentials itself (giving it a token of its own is a ``duplicate_credential`` fatal). Reads the primary config.yaml directly (both the top-level and nested ``gateway.`` forms) instead of ``load_gateway_config()`` so no primary platform config leaks into this process's environment. Shared by the preflight rescue (#97476) and the delivery-time shared-transport resolver (#101113) so route semantics cannot drift between the two halves. """ try: from hermes_constants import get_default_hermes_root, get_hermes_home primary_home = get_default_hermes_root() current_home = Path(get_hermes_home()) if ( primary_home.expanduser().resolve(strict=False) == current_home.expanduser().resolve(strict=False) ): return [] # this IS the primary home — nothing to consult config_path = primary_home.expanduser() / "config.yaml" if not config_path.exists(): return [] import yaml with open(config_path, encoding="utf-8") as f: raw = yaml.safe_load(f) or {} routes_raw = raw.get("profile_routes") if routes_raw is None and isinstance(raw.get("gateway"), dict): routes_raw = raw["gateway"].get("profile_routes") if not isinstance(routes_raw, list): return [] from gateway.profile_routing import parse_profile_routes from hermes_cli.profiles import profile_matches_home return [ route for route in parse_profile_routes(routes_raw) if route.enabled and profile_matches_home(route.profile) ] except Exception: logger.debug( "primary-gateway profile-route lookup unavailable", exc_info=True, ) return [] def _delivery_platform_routed_from_primary_gateway(platform_name: str) -> bool: """True when the primary gateway routes this platform to the profile the scheduler is currently serving (preflight rescue, #97476).""" platform_key = platform_name.lower() return any( str(route.platform).lower() == platform_key for route in _primary_profile_routes_for_current_home() ) class SharedRouteAdapters: """Read-only adapter map for a credentialless satellite profile (#101113). A satellite under ``gateway.profile_routes`` owns no bot credential and so has no adapter map of its own; its inbound traffic arrives on the PRIMARY adapter and is routed to it by an exact route. Its cron output must go back out the same transport — but ONLY for targets an enabled primary route maps to this profile. ``get(platform, target)`` resolves the primary adapter iff the route matcher used by inbound routing (``ProfileRoute.matches``) accepts the target's ``chat_id``/``thread_id``; every other lookup is a miss, so an unmatched target, a disabled route, or a route naming another profile still fails closed (never the default bot). A plain ``get(platform)`` (no target) is always a miss: routing is per-target, not per-platform. """ def __init__(self, primary_adapters, routes) -> None: self._primary = dict(primary_adapters or {}) self._routes = list(routes or []) def __bool__(self) -> bool: return bool(self._primary) and bool(self._routes) def get(self, platform, target=None, default=None): if not target: return default adapter = self._primary.get(platform) if adapter is None: return default platform_key = str(getattr(platform, "value", platform)).lower() chat_id = str(target.get("chat_id") or "") or None thread_id = target.get("thread_id") thread_id = str(thread_id) if thread_id else None for route in self._routes: if str(route.platform).lower() != platform_key: continue if not (route.chat_id or route.thread_id): continue # guild-only routes are not target-exact if route.matches(str(route.platform), chat_id=chat_id, thread_id=thread_id): return adapter return default return False def _preflight_check_delivery(job: dict) -> Optional[str]: """Check the job's delivery target(s) resolve to configured platforms. ``local``/``origin`` (and the ``all`` routing token) need no gateway credentials and are never checked — a deliver=local job must not pay a gateway-config load. For concrete platform targets, an unknown platform always blocks; a known platform additionally blocks when the gateway config is loadable and reports it unconnected (enabled + credentials — the same source `cron_delivery_targets` uses). Gateway-config load failures fail OPEN so a transient config hiccup never wedges delivery that would have worked. ``failure_deliver`` is checked with the same rules: a typo'd failure platform would otherwise only surface when a failure occurs — exactly when the notice must not be lost (NS-788 follow-up). """ deliver_value = _normalize_deliver_value(job.get("deliver", "local")) failure_deliver_value = _normalize_deliver_value( _delivery_lane_value(job, for_failure=True) ) lane_values = [deliver_value] if failure_deliver_value != deliver_value: lane_values.append(failure_deliver_value) platform_parts: list[str] = [] for lane_value in lane_values: for part in lane_value.split(","): part = part.strip() if not part or part.lower() in {"local", "origin", "all"}: continue # bot-chat targets need no gateway credentials — they deliver via a # local chat subprocess. Unknown-profile failures surface per run in # last_delivery_error (and are validated at create time). if parse_bot_chat_deliver_token(part) is not None: continue platform_parts.append(part.split(":", 1)[0].strip()) if not platform_parts: return None connected: Optional[set] = None for platform_name in platform_parts: if not _is_known_delivery_platform(platform_name): return ( f"delivery platform '{platform_name}' is not a known cron " "delivery target. Fix the job's `deliver` value or configure " "the platform's gateway credentials." ) if connected is None: try: from gateway.config import load_gateway_config gateway_config = load_gateway_config() connected = { p.value for p in gateway_config.get_connected_platforms() } connected |= _relay_fronted_delivery_platforms(connected) except Exception: logger.debug( "preflight: gateway config unavailable — skipping " "delivery credential check", exc_info=True, ) return None # fail-open if platform_name.lower() not in connected: # Multiplex escape hatch: a satellite profile whose deliveries # are routed by the primary gateway's profile_routes is served # by the primary's adapters, so its own unconnected reading is # a false block (#97476). if _delivery_platform_routed_from_primary_gateway(platform_name): continue return ( f"delivery platform '{platform_name}' has no gateway " "credentials configured (not connected). Configure it via " "`hermes setup` or change the job's `deliver` target." ) return None def _preflight_check_skills(job: dict) -> Optional[str]: """Check attached skills report ready (no missing required env/commands). Consults the same ``readiness_status`` payload ``skill_view`` computes for interactive use. Skills that fail to load at all are left to the existing skipped-skill handling in ``_build_job_prompt`` (fail-open): this check only blocks on an affirmative "setup needed" verdict, i.e. the skill exists but its required environment is missing — a run that is guaranteed to misfire. """ skills = job.get("skills") if skills is None: legacy = job.get("skill") skills = [legacy] if legacy else [] elif isinstance(skills, str): skills = [skills] skill_names = [str(name).strip() for name in skills if str(name).strip()] if not skill_names: return None from tools.skills_tool import skill_view for skill_name in skill_names: try: payload = json.loads(skill_view(skill_name)) except Exception: continue # unreadable/missing skill → existing skip handling if not isinstance(payload, dict) or not payload.get("success"): continue if ( payload.get("setup_needed") or payload.get("readiness_status") == "setup_needed" ): missing = [ f"env ${name}" for name in payload.get( "missing_required_environment_variables" ) or [] ] missing += [ f"command '{name}'" for name in payload.get("missing_required_commands") or [] ] missing += [ f"credential file {name}" for name in payload.get("missing_credential_files") or [] ] detail = ", ".join(missing) or "required setup incomplete" return ( f"attached skill '{skill_name}' is not ready: missing " f"{detail}. Provide the missing prerequisites or detach the " "skill from this job." ) return None def _preflight_job_config(job: dict, cfg: dict) -> Optional[str]: """Pre-dispatch configuration validation (T1-26). Returns a human-readable reason when the job's configuration cannot produce a successful run — missing provider API key, unconfigured delivery platform, or an attached skill with missing required env — so the caller can refuse the run BEFORE any agent machinery is constructed and no LLM call is burned. Returns ``None`` when the configuration validates (or when a check cannot be evaluated: every check fails open, so preflight can only ever block on an affirmative misconfiguration verdict). Same fail-before-spend spirit as the #44585 drift guard and the fail-loud-on-hidden-tools direction in #27948; alert dedup follows the alert-once pattern from the dead-pin auto-pause (#73506). """ for name, check in ( ("provider_key", lambda: _preflight_check_provider_key(job, cfg)), ("skills", lambda: _preflight_check_skills(job)), ("delivery", lambda: _preflight_check_delivery(job)), ): try: reason = check() except Exception: logger.debug( "preflight check %s raised — failing open", name, exc_info=True ) continue if reason: return reason return None def _cron_cleanup_timeout_seconds() -> float: """Return the wall-clock bound for cron post-run cleanup.""" default = 10.0 try: from hermes_cli.config import load_config cfg = load_config() or {} cron_cfg = cfg.get("cron", {}) if isinstance(cfg, dict) else {} configured = cron_cfg.get("cleanup_timeout_seconds") if configured is not None: timeout = float(configured) if timeout >= 0: return timeout except Exception as exc: logger.debug("Failed to load cron cleanup timeout from config: %s", exc) return default def _run_cron_cleanup_with_timeout( cleanup, *, job_id: str, label: str, timeout_seconds: Optional[float] = None, ) -> bool: """Run fallible post-run cleanup without permanently wedging a cron ID.""" timeout = ( _cron_cleanup_timeout_seconds() if timeout_seconds is None else float(timeout_seconds) ) if timeout <= 0: try: cleanup() return True except (Exception, KeyboardInterrupt) as exc: logger.debug("Job '%s': %s failed: %s", job_id, label, exc) return False done = threading.Event() error: list[BaseException] = [] def _runner() -> None: try: cleanup() except BaseException as exc: error.append(exc) finally: done.set() # A daemon thread is deliberate: unlike ThreadPoolExecutor workers it is # not joined by Python's interpreter-exit hook if the cleanup target never # returns. The scheduler can release its dispatch guard and the gateway can # still shut down normally. worker = threading.Thread( target=_runner, name=f"cron-cleanup-{job_id}", daemon=True, ) worker.start() if not done.wait(timeout): logger.error( "Job '%s': %s exceeded %.1fs; abandoning cleanup so future runs remain dispatchable", job_id, label, timeout, ) return False if error: logger.debug("Job '%s': %s failed: %s", job_id, label, error[0]) return False return True class _BoundedCronSessionDB: """Proxy SessionDB cleanup calls through the cron cleanup timeout. After the first failed or timed-out operation the proxy fails subsequent calls immediately. A damaged SQLite connection should leak at most one abandoned cleanup worker, not one worker per finalization step. """ def __init__(self, session_db, job_id: str): self._session_db = session_db self._job_id = job_id self._disabled = False def __getattr__(self, name): target = getattr(self._session_db, name) if not callable(target): return target def _bounded(*args, **kwargs): if self._disabled: raise RuntimeError("session finalization disabled after prior cleanup failure") result = {} def _call(): try: result["value"] = target(*args, **kwargs) except BaseException as exc: result["error"] = exc raise ok = _run_cron_cleanup_with_timeout( _call, job_id=self._job_id, label=f"session finalization ({name})", ) if not ok: error = result.get("error") if error is not None: raise error # No exception reached the caller and the operation still did # not complete: this is the timeout path. Disable the damaged # connection so later finalization steps fail immediately. self._disabled = True raise TimeoutError(f"session finalization method {name} timed out") return result.get("value") return _bounded def run_job( job: dict, *, defer_agent_teardown: Optional[list] = None, extra_prompt: Optional[str] = None, cancel_event: Optional[_CancelEventLike] = None, execution_id: Optional[str] = None, ) -> tuple[bool, str, str, Optional[str]]: """ Execute a single cron job. ``defer_agent_teardown``: when a caller passes a list, ``run_job`` skips the agent's async-resource teardown (``agent.close()`` + ``cleanup_stale_async_clients()``) in its ``finally`` block and instead appends the live agent to that list. The caller is then responsible for calling ``_teardown_cron_agent(agent)`` AFTER it has delivered the result. This closes the ordering window in #58720 where delivery ran against a torn-down async client (defense-in-depth alongside the interpreter-shutdown guard). When ``None`` (the default) teardown happens inline as before, so every existing caller is unchanged. ``extra_prompt``: optional per-run context from ``cronjob(action='run', prompt=...)`` (#57331). Appended to the stored prompt for this fire only — never persisted to the job definition. Returns: Tuple of (success, full_output_doc, final_response, error_message) """ job_id = job["id"] job_name = str(job.get("name") or job.get("prompt") or job_id or "cron job") # Fail closed on a corrupt config.yaml before any agent-driven work # (issue #81952): a cron fire is fully non-interactive, and continuing # on built-in defaults lets provider auto-detection adopt .env # credentials the config never named, billing a provider the user did # not choose. no_agent script jobs are exempt — they never construct an # AIAgent or spend tokens. Escape hatch: HERMES_IGNORE_USER_CONFIG=1. if not job.get("no_agent"): from hermes_cli.config import ( InvalidUserConfigError, require_parseable_user_config, ) try: require_parseable_user_config() except InvalidUserConfigError as exc: logger.error("Job '%s': refusing to run — %s", job_id, exc) return (False, f"# Cron Job: {job_name}\n\nError: {exc}\n", "", str(exc)) # --------------------------------------------------------------- # no_agent short-circuit — the script IS the job, no LLM involvement. # --------------------------------------------------------------- # This mirrors the classic "run a bash script on a timer, send its # stdout to telegram" watchdog pattern. The agent path is skipped # entirely: no AIAgent, no prompt, no tool loop, no token spend. # # We check this BEFORE importing run_agent / constructing SessionDB so # a pure-script tick never pays for the agent machinery it isn't going # to use. Keep this block self-contained. # # Semantics: # - script stdout (trimmed) → delivered verbatim as the final message # - empty stdout → silent run (no delivery, success=True) # - non-zero exit / timeout → delivered as an error alert, success=False # - wakeAgent=false gate → treated like empty stdout (silent), since # the whole point of no_agent is that there # is no agent to wake if job.get("no_agent"): # Load .env before the script runs so auto-delivery can resolve home # channels. A standalone cron tick process typically starts WITHOUT # TELEGRAM_HOME_CHANNEL/DISCORD_HOME_CHANNEL in its environment, and # the agent path's per-run dotenv reload below never executes for # no_agent jobs — every deliver=telegram/all script job failed with # "no delivery target resolved". load_hermes_dotenv does not override # already-set vars, so the gateway's in-process tick is unaffected. try: from hermes_cli.env_loader import load_hermes_dotenv load_hermes_dotenv(hermes_home=_get_hermes_home()) except Exception: logger.debug( "Job '%s': no_agent .env reload failed", job_id, exc_info=True ) script_path = job.get("script") # Legacy/hand-edited records can still carry no_agent with a missing or # whitespace-only script. Erroring alone left the job enabled, so it # re-fired every tick — pause it instead (a5e29e688dc0). if not str(script_path or "").strip(): from cron.jobs import NO_AGENT_WITHOUT_SCRIPT_ERROR return _block_and_pause_job( job_id, job_name, NO_AGENT_WITHOUT_SCRIPT_ERROR, ) # Apply workdir if configured — lets scripts use predictable relative # paths. For no_agent jobs this is passed as the subprocess cwd so the # Python process cwd is NEVER mutated — avoiding the global-side-effect # bug where os.chdir() leaks into concurrent gateway sessions (#69396). _job_workdir = (job.get("workdir") or "").strip() or None if _job_workdir and not Path(_job_workdir).is_dir(): logger.warning( "Job '%s': configured workdir %r no longer exists — running without it", job_id, _job_workdir, ) _job_workdir = None try: ok, output = _run_job_script_with_claim_heartbeat( job, script_path, workdir=_job_workdir, cancel_event=cancel_event, ) except Exception as exc: logger.exception( "Job '%s': script execution raised unexpectedly", job_id, ) ok, output = False, f"Script execution failed: {exc}" now_iso = _hermes_now().strftime("%Y-%m-%d %H:%M:%S") if not ok: # Script crashed / timed out / exited non-zero. Deliver the # error so the user knows the watchdog itself broke — silent # failure for an alerting job is the worst-case outcome. alert = ( f"⚠ Cron watchdog '{job_name}' script failed\n\n" f"{output}\n\n" f"Time: {now_iso}" ) doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {now_iso}\n" f"**Mode:** no_agent (script)\n" f"**Status:** script failed\n\n" f"{output}\n" ) return False, doc, alert, output # Honour the wakeAgent gate as a silent signal — `wakeAgent: false` # means "nothing to report this tick", same as empty stdout. if not _parse_wake_gate(output): logger.info( "Job '%s' (no_agent): wakeAgent=false gate — silent run", job_id ) silent_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {now_iso}\n" f"**Mode:** no_agent (script)\n" f"**Status:** silent (wakeAgent=false)\n" ) return True, silent_doc, SILENT_MARKER, None if not output.strip(): logger.info("Job '%s' (no_agent): empty stdout — silent run", job_id) silent_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {now_iso}\n" f"**Mode:** no_agent (script)\n" f"**Status:** silent (empty output)\n" ) return True, silent_doc, SILENT_MARKER, None doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {now_iso}\n" f"**Mode:** no_agent (script)\n\n" f"---\n\n" f"{output}\n" ) return True, doc, output, None # --------------------------------------------------------------- # Fail-closed guard for legacy / hand-edited agent jobs that have nothing # to run: blank prompt, no script, no skills (a5e29e688dc0). create_job / # update_job now reject this shape, but jobs.json records written before # that guard — or edited by hand since — can still reach here and would # otherwise wake the LLM with an empty instruction on every fire. Pause # the job so it stops being scheduled, and never construct the agent. # --------------------------------------------------------------- from cron.jobs import EMPTY_PAYLOAD_ERROR, job_payload_is_empty if job_payload_is_empty(job): return _block_and_pause_job( job_id, job_name, EMPTY_PAYLOAD_ERROR, ) # --------------------------------------------------------------- # Monitor gate — hash-suppressed change detection (see cron/monitor.py). # Runs BEFORE any agent machinery is constructed so an unchanged tick # costs one cheap source run + one hash, no LLM, no delivery. # --------------------------------------------------------------- from cron.monitor import check_monitor, job_has_monitor _monitor_context: Optional[str] = None if job_has_monitor(job): _mon = check_monitor(job) _mon_now = _hermes_now().strftime("%Y-%m-%d %H:%M:%S") if not _mon.ok: # Source failure is an ERROR, never a change: alert the user so # a broken monitor can't silently stop watching. Stored hash is # untouched (check_monitor persists nothing on failure). logger.error("Job '%s': monitor source failed: %s", job_id, _mon.error) _mon_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {_mon_now}\n" f"**Mode:** monitor\n" f"**Status:** monitor source failed\n\n" f"{_mon.error}\n" ) _mon_alert = ( f"⚠ Cron monitor '{job_name}' source failed\n\n" f"{_mon.error}\n\n" f"Time: {_mon_now}" ) return False, _mon_doc, _mon_alert, _mon.error if not _mon.changed: # Unchanged output — suppress the agent run entirely. Recorded # as a silent no_change tick (visible in the executions ledger # via this doc; SILENT_MARKER blocks delivery). logger.info( "Job '%s': monitor output unchanged — suppressing agent run", job_id, ) _mon_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {_mon_now}\n" f"**Mode:** monitor\n" f"**Status:** no_change (agent run suppressed)\n" ) return True, _mon_doc, SILENT_MARKER, None # Changed (or first run): inject the monitor context into the prompt # through the existing per-run context seam and fall through to a # normal agent run. _monitor_context = _mon.context_block if _monitor_context: extra_prompt = ( f"{_monitor_context}\n\n{extra_prompt}" if extra_prompt else _monitor_context ) # --------------------------------------------------------------- # Default (LLM) path — import and construct the agent machinery now # that we know we actually need it. Doing these imports here instead of # at module top keeps no_agent ticks from paying for AIAgent / SessionDB # construction costs. # --------------------------------------------------------------- from run_agent import AIAgent # NOTE: the SQLite session store used to be initialized here, BEFORE the # wake-gate and prompt-validation early returns below. Every gated run # (``wakeAgent: false``, blocked prompt) opened state.db and returned # without reaching the finally that closes it, relying on GC to release # the handle. Init now happens inside the main try, right before the # agent is constructed — after every early-return path (#96290). # Wake-gate: if this job has a pre-check script, run it BEFORE building # the prompt so a ``{"wakeAgent": false}`` response can short-circuit # the whole agent run. We pass the result into _build_job_prompt so # the script is only executed once. prerun_script = None script_path = job.get("script") if script_path: prerun_script = _run_job_script_with_claim_heartbeat( job, script_path, cancel_event=cancel_event, ) _ran_ok, _script_output = prerun_script if _ran_ok and not _parse_wake_gate(_script_output): logger.info( "Job '%s' (ID: %s): wakeAgent=false, skipping agent run", job_name, job_id, ) silent_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" "Script gate returned `wakeAgent=false` — agent skipped.\n" ) return True, silent_doc, SILENT_MARKER, None try: prompt = _build_job_prompt( job, prerun_script=prerun_script, extra_prompt=extra_prompt ) except CronPromptInjectionBlocked as block_exc: # Assembled prompt (user prompt + loaded skill content) tripped the # injection scanner. Refuse to run the agent this tick and surface # a clear failure to the operator so they see WHY the scheduled job # didn't run and can audit the offending skill. logger.warning( "Job '%s' (ID: %s): blocked by prompt-injection scanner — %s", job_name, job_id, block_exc, ) blocked_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"**Status:** BLOCKED\n\n" "The assembled prompt (user prompt + loaded skill content) tripped " "the cron injection scanner and the agent was NOT run.\n\n" f"**Scanner result:** {block_exc}\n\n" "Audit the skill(s) attached to this job for prompt-injection " "payloads or invisible-unicode markers. If the skill is legitimate " "and the match is a false positive, rephrase the content to avoid " "the threat pattern (`tools/cronjob_tools.py::_CRON_THREAT_PATTERNS`)." ) return False, blocked_doc, "", str(block_exc) if prompt is None: logger.info("Job '%s': script produced no output, skipping AI call.", job_name) return True, "", SILENT_MARKER, None _cron_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}" logger.info("Running job '%s' (ID: %s)", job_name, job_id) logger.info("Prompt: %s", prompt[:100]) agent = None # Use ContextVars for per-job session/delivery state so parallel jobs # don't clobber each other's targets (os.environ is process-global). from gateway.session_context import set_session_vars, clear_session_vars, _VAR_MAP # Cron execution is an internal scheduler context, not a live inbound # gateway message. Do not seed HERMES_SESSION_* contextvars from the # stored ``origin`` (which is delivery routing metadata, not a sender # identity). Several tool consumers branch on these vars during job # execution and would otherwise behave as if a real user from the # origin chat was driving the agent: # - tools/terminal_tool.py: background-process notification routing # (notify_on_complete / watch_patterns) reads HERMES_SESSION_PLATFORM # and HERMES_SESSION_CHAT_ID to populate watcher_platform / chat_id, # which would route completion notifications to the origin chat # instead of via HERMES_CRON_AUTO_DELIVER_* below. # - tools/tts_tool.py: picks Opus vs MP3 based on # HERMES_SESSION_PLATFORM == "telegram". # - tools/skills_tool.py + agent/prompt_builder.py: per-platform # skill-disable lists and the system-prompt cache key both consume # HERMES_SESSION_PLATFORM. # - tools/send_message_tool.py: mirror source labelling and the # send_message gate read HERMES_SESSION_PLATFORM. # Cron output delivery itself reads job["origin"] directly via # _resolve_origin(job) and the HERMES_CRON_AUTO_DELIVER_* vars set # below, so clearing HERMES_SESSION_* here does not affect delivery. # Resolve workdir BEFORE set_session_vars so we can pass it as cwd=, # letting set_session_vars handle the _SESSION_CWD ContextVar set/clear # via its existing machinery (clear_session_vars calls clear_session_cwd # internally). This avoids a separate import/set/clear dance (#69396). _job_workdir = (job.get("workdir") or "").strip() or None if _job_workdir and not Path(_job_workdir).is_dir(): logger.warning( "Job '%s': configured workdir %r no longer exists — running without it", job_id, _job_workdir, ) _job_workdir = None _ctx_tokens = set_session_vars( platform="", chat_id="", chat_name="", # A cron job cannot receive a completion after its turn ends. We clear the # HERMES_SESSION_* routing keys just below, so an async delegation's # completion event carries session_key="" — _enrich_async_delegation_routing # cannot resolve it and _inject_watch_notification drops it ("no routing # metadata"). And by the time a child finishes, run_job has already shipped # the job's final response via _deliver_result; there is no turn left to # re-enter. (Worse, get_current_session_key() can fall back to the ambient # os.environ HERMES_SESSION_KEY, which risks routing a cron subagent's output # into an unrelated user chat.) # # Declaring the channel stateless routes delegate_task to its existing # inline/synchronous path, so results return within the job's own turn. # See declare_stateless_channel(). Upstream: #53027, #63142. async_delivery=False, cwd=_job_workdir or "", ) _cron_delivery_vars = ( "HERMES_CRON_AUTO_DELIVER_PLATFORM", "HERMES_CRON_AUTO_DELIVER_CHAT_ID", "HERMES_CRON_AUTO_DELIVER_THREAD_ID", ) for _var_name in _cron_delivery_vars: _VAR_MAP[_var_name].set("") # Tool calls are keyed by a per-run task id. Bind the cron workdir to that # identity instead of mutating process-global TERMINAL_CWD. The session CWD # record is the tool-layer authority for terminal/file/code-exec/delegation, # while the _SESSION_CWD ContextVar above remains the prompt/context-file # authority. Both are isolated across concurrent cron runs. _cron_task_id = ( f"cron:{job_id}:" f"{execution_id or job.get('execution_id') or uuid.uuid4().hex}" ) from tools.terminal_tool import clear_session_cwd as _clear_tool_session_cwd from tools.terminal_tool import record_session_cwd as _record_tool_session_cwd if _job_workdir: _record_tool_session_cwd(_cron_task_id, _job_workdir) _cron_session_var = _VAR_MAP["HERMES_CRON_SESSION"] _cron_session_token = None _non_dispatcher_token = None _session_db = None try: # Scope cron approval policy to this job. Keep the token so the finally # restores the pre-job state instead of pinning an explicit empty value, # which would suppress the legacy os.environ fallback used by standalone # cron entrypoints and tests. _cron_session_token = _cron_session_var.set("1") # Mark this job as NOT the dispatcher-owned kanban worker. # # A kanban worker is a normal `hermes chat -q` CLI agent whose default # toolset includes `cronjob`, running with HERMES_KANBAN_TASK # legitimately in its own env; `cronjob(action="run")` calls # run_one_job() -> run_job() right here in that process. Without this # marker the cron agent is misread as that worker: the kanban toolset is # force-added, the worker protocol is injected into its system prompt, # and kanban_complete defaults task_id to $HERMES_KANBAN_TASK -- letting # an unrelated cron job close the worker's task and overwrite real # results. # # A ContextVar, NOT an os.environ clear: the env is process-global and # shared with the worker's own claim heartbeat (run_agent._touch_activity # -> heartbeat_current_worker_from_env, which would starve and let the # dispatcher reclaim a live task), the gateway's kanban watchers, and # concurrent cron jobs on the parallel pool. contextvars.copy_context() # at the run_conversation hop carries this into the agent thread. _non_dispatcher_token = enter_non_dispatcher_owned_context() if _job_workdir: logger.info("Job '%s': using task-scoped workdir %s", job_id, _job_workdir) # Re-read .env and config.yaml fresh every run so provider/key # changes take effect without a gateway restart. Route through # load_hermes_dotenv (not a bare load_dotenv) and reset the secret- # source cache first: startup already applied external secrets and # recorded this HERMES_HOME in _APPLIED_HOMES, so a naive reload would # re-apply only the .env placeholder and never re-resolve a Bitwarden/ # BSM-backed secret — leaving cron jobs 401'ing on the placeholder # (#33465). Clearing the cache forces the re-pull; the resolved secret # overrides the placeholder only when secrets.bitwarden.override_existing # is set (mirrors startup), and the Bitwarden value-cache keeps the # forced re-pull off the network. load_hermes_dotenv also handles the # utf-8/latin-1 encoding fallback internally. from hermes_cli.env_loader import ( load_hermes_dotenv, reset_secret_source_cache, ) reset_secret_source_cache() load_hermes_dotenv(hermes_home=_get_hermes_home()) delivery_target = _resolve_delivery_target(job) if delivery_target: _VAR_MAP["HERMES_CRON_AUTO_DELIVER_PLATFORM"].set(delivery_target["platform"]) _VAR_MAP["HERMES_CRON_AUTO_DELIVER_CHAT_ID"].set(str(delivery_target["chat_id"])) _VAR_MAP["HERMES_CRON_AUTO_DELIVER_THREAD_ID"].set( "" if delivery_target.get("thread_id") is None else str(delivery_target["thread_id"]) ) # Model resolution precedence: per-job override > cron.model (the # cron-fleet default) > HERMES_MODEL env > config.yaml ``model:`` # (string or ``{default: ...}``). The per-job value is intentionally # re-read from storage every tick so a ``hermes cron edit --model`` # after a failed run takes effect on the next tick — there is no # in-memory cache. model = job.get("model") or os.getenv("HERMES_MODEL") or "" # cron.model / cron.model_provider: a deliberate cron-fleet default # so unattended jobs stop shadowing chat `/model` switches. When an # axis resolves from here, the #44585 drift guard is skipped for that # axis — following cron.model is explicit, not drift. _cron_default_model = "" _cron_default_provider = "" # Load config.yaml for model, reasoning, prefill, toolsets, provider routing _cfg = {} _model_cfg = {} try: from hermes_cli.config import read_user_config_raw _cfg_path = str(_get_hermes_home() / "config.yaml") if os.path.exists(_cfg_path): _cfg = read_user_config_raw(Path(_cfg_path)) # Managed scope: a scheduled job must honor administrator-pinned # model / reasoning / toolsets / provider_routing too. This loader # builds its own dict, so overlay managed values via the shared # helper (fail-open, no-op when no managed scope). try: from hermes_cli import managed_scope _cfg = managed_scope.apply_managed_overlay(_cfg) except Exception: pass _cfg = _expand_env_vars(_cfg) # Coerce null/missing to {} so a falsy default never # clobbers an already-resolved env value with ``None``. _model_cfg = _cfg.get("model") or {} _cron_cfg_for_model = _cfg.get("cron") or {} if isinstance(_cron_cfg_for_model, dict): _cron_default_model = str( _cron_cfg_for_model.get("model") or "" ).strip() _cron_default_provider = str( _cron_cfg_for_model.get("model_provider") or "" ).strip() if not job.get("model"): if _cron_default_model: # Cron-fleet default beats the global chat model: it is # the user's explicit "cron runs on this" setting. model = _cron_default_model else: # Shared with Desktop's post-save impact summary so both # paths compare snapshots against the same global model. _, _global_model = resolve_cron_model_drift_defaults(_cfg) if _global_model: model = _global_model except Exception as e: logger.warning("Job '%s': failed to load config.yaml, using defaults: %s", job_id, e) # Fail fast if no model resolved from job / env / config.yaml: an empty # model otherwise reaches the provider as an opaque 400 (#23979). if not (isinstance(model, str) and model.strip()): raise RuntimeError( f"Cron job '{job_name}' has no model configured " f"(job.model={job.get('model')!r}, " f"HERMES_MODEL={os.getenv('HERMES_MODEL', '')!r}, " "config.yaml model.default missing or empty). " f"Set a per-job model via " f"`hermes cron edit {job_id} --model ` or set a " "default with `hermes model `." ) # Apply IPv4 preference if configured. try: from hermes_constants import apply_ipv4_preference _net_cfg = _cfg.get("network", {}) if isinstance(_net_cfg, dict) and _net_cfg.get("force_ipv4"): apply_ipv4_preference(force=True) except Exception: pass # Reasoning config is resolved after provider authentication so an auth # fallback can first replace the primary model with its configured model. # Resolution itself happens via _resolve_job_reasoning_config below # (per-job pin > agent.reasoning_overrides > agent.reasoning_effort). # Prefill messages from env or config.yaml. The top-level # prefill_messages_file key is canonical; agent.prefill_messages_file is # retained as a legacy fallback for older CLI/godmode configs. prefill_messages = None agent_cfg = _cfg.get("agent", {}) if isinstance(_cfg.get("agent", {}), dict) else {} prefill_file = ( os.getenv("HERMES_PREFILL_MESSAGES_FILE", "") or _cfg.get("prefill_messages_file", "") or agent_cfg.get("prefill_messages_file", "") ) if prefill_file: pfpath = Path(prefill_file).expanduser() if not pfpath.is_absolute(): pfpath = _get_hermes_home() / pfpath if pfpath.exists(): try: with open(pfpath, "r", encoding="utf-8") as _pf: prefill_messages = json.load(_pf) if not isinstance(prefill_messages, list): prefill_messages = None except Exception as e: logger.warning("Job '%s': failed to parse prefill messages file '%s': %s", job_id, pfpath, e) prefill_messages = None # Max iterations — resolved through resolve_turn_limit() so that # agent.max_turns: none / unlimited → sys.maxsize sentinel, and # explicit 0 / null / "none" are honored instead of skipped by `or`. from hermes_cli.config import resolve_turn_limit as _resolve_turn_limit _mt = _cfg.get("agent", {}).get("max_turns") if _mt is None: _mt = _cfg.get("max_turns") max_iterations = _resolve_turn_limit(_mt) # Provider routing pr = _cfg.get("provider_routing") or {} from hermes_cli.runtime_provider import ( resolve_runtime_provider, format_runtime_provider_error, ) from hermes_cli.auth import AuthError # F8 runtime backstop: never resolve a stored provider/base_url pair that # would ship a named provider's stored credential to an off-host endpoint # (CWE-200/CWE-522). The cron tool validates this on create/update, but a # job persisted before that guard — or written directly to the jobs store # — reaches this sink unchecked. Fail closed before resolution so no # off-host call is ever made with a stored key. _guard_job_credential_exfil(job) # --------------------------------------------------------------- # Pre-dispatch configuration validation (T1-26). # # A job whose configuration cannot possibly produce a successful # run — missing provider API key (no fallback chain), unready # attached skill, unconfigured delivery platform — is refused HERE, # before AIAgent is constructed and before the resolution below can # feed a doomed runtime into it, so a misconfigured job never burns # an LLM call. run_one_job keys off the BLOCKED_CONFIG_MARKER in # the returned error to record last_status='blocked_config' and # alert exactly once (dedup persisted via the job's # `preflight_alerted` bit — the #73506 alert-once shape). # Runs after the wake-gate/prompt build so silent script ticks stay # silent. Opt-out: `cron.preflight: false` in config.yaml. # --------------------------------------------------------------- _pf_reason = None try: if _cron_preflight_enabled(_cfg): _pf_reason = _preflight_job_config(job, _cfg) if not _pf_reason and job.get("preflight_alerted"): # Configuration validates again — clear the alert-once # marker so a FUTURE config break re-alerts. try: from cron.jobs import clear_preflight_alerted clear_preflight_alerted(job_id) except Exception: pass except Exception: # The validator must never take down a runnable job — fail open. logger.debug( "Job '%s': preflight validation errored — failing open", job_id, exc_info=True, ) _pf_reason = None if _pf_reason: logger.warning( "Job '%s' (ID: %s): BLOCKED by pre-dispatch config " "validation — %s (no LLM call was made)", job_name, job_id, _pf_reason, ) already_alerted = False try: from cron.jobs import mark_preflight_alerted already_alerted = mark_preflight_alerted(job_id) except Exception: logger.debug( "Job '%s': could not persist preflight alert marker", job_id, exc_info=True, ) marker = ( BLOCKED_CONFIG_SILENT_MARKER if already_alerted else BLOCKED_CONFIG_MARKER ) blocked_doc = ( f"# Cron Job: {job_name}\n\n" f"**Job ID:** {job_id}\n" f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"**Status:** BLOCKED (configuration)\n\n" "Pre-dispatch validation found a configuration problem and " "the agent was NOT run (no tokens spent).\n\n" f"**Reason:** {_pf_reason}\n\n" "The job will stay blocked (without re-alerting) until the " "configuration is fixed; the next healthy run clears this " "state. Set `cron.preflight: false` in config.yaml to " "disable this validation." ) return False, blocked_doc, "", f"{marker} {_pf_reason}" primary_model_for_drift = model configured_provider_for_drift = ( str(_model_cfg.get("provider") or "").strip().lower() if isinstance(_model_cfg, dict) else "" ) primary_provider_for_drift = ( str(job.get("provider") or "").strip().lower() or configured_provider_for_drift or None ) try: # Do not inject HERMES_INFERENCE_PROVIDER here. resolve_runtime_provider() # already prefers persisted config over stale shell/env overrides when # no explicit provider is requested. Passing the env var here short- # circuits that precedence and can resurrect old providers (for # example DeepSeek) for cron jobs that do not pin provider/model. runtime_kwargs = { # Per-job user pin wins; otherwise the cron-fleet default # provider (cron.model_provider); otherwise resolve from # persisted global config. "requested": job.get("provider") or _cron_default_provider or None, # Derive provider-specific api_mode from the model this job # will actually run (per-job pin > env > config default), not # the stale persisted default — mirrors the fallback path # below, which already passes its fb_model. "target_model": model, } if job.get("base_url"): runtime_kwargs["explicit_base_url"] = job.get("base_url") runtime = resolve_runtime_provider(**runtime_kwargs) primary_provider_for_drift = ( str(runtime.get("provider") or "").strip().lower() or primary_provider_for_drift ) except Exception as resolve_exc: # Primary provider resolution failed. Walk fallback_providers for: # 1) AuthError (missing/expired credential) # 2) Transient network/DNS failures during OAuth refresh or # discovery (e.g. macOS morning DNS blip → httpx.ConnectError # "[Errno 8] nodename nor servname provided"). # Previously only AuthError tried the chain; a ConnectError during # xai-oauth token refresh killed agent crons even when XAI_API_KEY # / Anthropic fallbacks were healthy (Daily Focus Kickoff 2026-08-11). # Keeping provider+model atomic still applies — never swap only the # provider while retaining a paid primary model. is_auth = isinstance(resolve_exc, AuthError) is_transient_net = _is_transient_provider_resolve_error(resolve_exc) if not (is_auth or is_transient_net): raise RuntimeError(format_runtime_provider_error(resolve_exc)) from resolve_exc primary_provider_for_drift = ( str(getattr(resolve_exc, "provider", "") or "").strip().lower() or primary_provider_for_drift ) reason = "auth" if is_auth else "transient network" logger.warning( "Job '%s': primary provider resolve failed (%s: %s), trying fallback", job_id, reason, resolve_exc, ) fb_list = get_fallback_chain(_cfg) runtime = None for entry in fb_list: if not isinstance(entry, dict): continue fb_provider = str(entry.get("provider") or "").strip() fb_model = str(entry.get("model") or "").strip() if not fb_provider or not fb_model: continue try: from hermes_cli.fallback_config import resolve_entry_api_key fb_kwargs = { "requested": fb_provider, "target_model": fb_model, } if entry.get("base_url"): fb_kwargs["explicit_base_url"] = entry["base_url"] fb_api_key = resolve_entry_api_key(entry) if fb_api_key: fb_kwargs["explicit_api_key"] = fb_api_key runtime = resolve_runtime_provider(**fb_kwargs) model = fb_model logger.info( "Job '%s': fallback resolved to %s model %s", job_id, runtime.get("provider"), fb_model, ) break except Exception as fb_exc: logger.debug("Job '%s': fallback %s failed: %s", job_id, fb_provider, fb_exc) if runtime is None: raise RuntimeError(format_runtime_provider_error(resolve_exc)) from resolve_exc reasoning_config = _resolve_job_reasoning_config( job, _cfg if isinstance(_cfg, dict) else {}, str(model) ) # Provider/model-drift fail-closed guard (#44585). # # An UNPINNED job (no explicit job["provider"]/["model"]) follows the # global default, which can change after the job was created — a switch # to a paid PROVIDER (e.g. nous) OR a paid MODEL on the same provider # (e.g. claude-fable-5 on openrouter). Without a guard the job would # silently inherit that change and spend real money on every tick — the # $7.73 incident named BOTH a provider and a model. # # create_job() snapshots whatever resolution would have picked at # creation for each unpinned axis (job["provider_snapshot"] / # job["model_snapshot"]). Here, for each axis that (a) has a snapshot and # (b) is unpinned and (c) currently resolves to a DIFFERENT value, we # fail closed: skip this run, make NO paid call, and deliver a loud, # actionable alert telling the user to pin the axis explicitly. # # Back-compat: an axis with no snapshot (pre-existing jobs, no_agent, or # any axis whose creation-time resolution failed) behaves exactly as # before — the guard never engages for it. Pinned axes are unaffected. # # cron.model / cron.model_provider: an axis resolved from the explicit # cron-fleet default is NOT drift — the user deliberately routed # unpinned cron jobs there, so the guard is skipped for that axis. if cron_model_drift_guard_enabled(_cfg): _drift: list[str] = [] _current_provider = str( primary_provider_for_drift or runtime.get("provider") or "" ).strip().lower() _current_model = str(primary_model_for_drift or "").strip().lower() for _axis in cron_model_drift_axes( job, current_provider=_current_provider, current_model=_current_model, config=_cfg, ): _snapshot = str(job.get(f"{_axis}_snapshot") or "").strip().lower() _current = _current_provider if _axis == "provider" else _current_model _drift.append(f"{_axis} '{_snapshot}' -> '{_current}'") if _drift: _changes = "; ".join(_drift) # Lifecycle-aware remediation (#72056, @sashmatash): a finite # one-shot is consumed by this attempted dispatch — telling an # operator to edit a spent job is a dead end. Recurring and # repeatable jobs get the pin command instead. _repeat = job.get("repeat") if isinstance(job.get("repeat"), dict) else {} _finite_oneshot = ( isinstance(job.get("schedule"), dict) and job["schedule"].get("kind") == "once" and _repeat.get("times") == 1 ) if _finite_oneshot: _remediation = ( "This finite one-shot job is consumed by this attempted run; " "create a new one-shot job at a future time with an explicit " "provider and model." ) else: _remediation = ( "To run on the new config, on the host running Hermes " "pin it explicitly: " f"`hermes cron edit {job_id} --provider " "--model ` (or pin the original values to keep " "them)." ) logger.warning( "Job '%s': SKIPPED — global inference config drifted since " "creation (%s) and this job is unpinned. Skipped to prevent " "unintended spend. %s", job_id, _changes, _remediation, ) # Alert-once (#73506 shape): persist the drift_alerted bit so # only the FIRST drifted tick delivers; run_one_job suppresses # delivery on the silent marker. mark_job_run clears the bit # when a run succeeds (drift healed), re-arming the alert. _drift_already_alerted = False try: from cron.jobs import mark_drift_alerted _drift_already_alerted = mark_drift_alerted(job_id) except Exception: pass # fail open: better a duplicate alert than none _drift_marker = ( DRIFT_SKIP_SILENT_MARKER if _drift_already_alerted else DRIFT_SKIP_MARKER ) raise RuntimeError( f"{_drift_marker} Skipped to prevent unintended spend: global " f"inference config drifted since this job was created " f"({_changes}), and this job is unpinned. No inference call " f"was made. {_remediation} " f"This alert is sent once; the job stays skipped until the " f"config is pinned or restored. See #44585." ) fallback_model = get_fallback_chain(_cfg) or None credential_pool = None runtime_provider = str(runtime.get("provider") or "").strip().lower() if runtime_provider: try: from agent.credential_pool import load_pool pool = load_pool(runtime_provider) if pool.has_credentials(): credential_pool = pool logger.info( "Job '%s': loaded credential pool for provider %s with %d entries", job_id, runtime_provider, len(pool.entries()), ) except Exception as e: logger.debug("Job '%s': failed to load credential pool for %s: %s", job_id, runtime_provider, e) # Initialize MCP servers so configured mcp_servers are available to # the agent's tool registry before AIAgent is constructed. Without # this, cron jobs never saw any MCP tools — only the gateway / CLI # paths called discover_mcp_tools() at startup. Idempotent: subsequent # ticks short-circuit on already-connected servers inside # register_mcp_servers(). Non-fatal on failure: a broken MCP server # shouldn't kill an otherwise-working cron job. See #4219. try: from tools.mcp_tool import discover_mcp_tools _mcp_tools = discover_mcp_tools() if _mcp_tools: logger.info( "Job '%s': %d MCP tool(s) available", job_id, len(_mcp_tools), ) except Exception as _mcp_exc: logger.warning( "Job '%s': MCP initialization failed (non-fatal): %s", job_id, _mcp_exc, ) # Initialize the SQLite session store so cron job messages are # persisted and discoverable via session_search (same pattern as # gateway/run.py) — only now, after every early-return path # (wake-gate, prompt validation, drift skip) has passed, so a gated # run never opens state.db just to abandon the handle (#96290). # # Bounded with its own timeout (separate from HERMES_CRON_TIMEOUT, # which only watches the agent's run_conversation below): # SessionDB.__init__ opens/migrates state.db synchronously and has no # timeout of its own against a wedged sqlite3.connect (e.g. a stale # flock left by a crashed sibling process). An unbounded hang here # would wedge the job's worker thread, so the init is bounded and a # timeout proceeds without a session store instead of blocking the # run forever. _session_db_timeout = _get_session_db_timeout() try: from hermes_state import get_shared_session_db if _session_db_timeout > 0: _session_db_pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) # The timeout worker is a second thread, so it does not inherit # the multiplexed profile ContextVar automatically. Run the # constructor inside a copy of the active context so a profile # cron run resolves ITS OWN home (and state.db) instead of # silently falling back to the process-global default. _session_db_context = contextvars.copy_context() _session_db_future = _session_db_pool.submit( _session_db_context.run, get_shared_session_db ) try: _session_db = _session_db_future.result(timeout=_session_db_timeout) except concurrent.futures.TimeoutError: # The worker is abandoned (shutdown below doesn't wait for # it). If SessionDB() later completes inside it, the # future's result would be orphaned and its SQLite FDs # (.db, WAL, SHM) leak until process exit. Register a # done-callback that retrieves and closes any eventual # late result (#72782). _session_db_future.add_done_callback(_close_late_session_db_result) raise finally: # Don't wait for a wedged connect() to unwind — abandon the # worker thread (same pattern as the agent inactivity # timeout further down) rather than blocking shutdown on # it too. _session_db_pool.shutdown(wait=False) else: # 0 = unlimited (legacy behavior, opt-in for debugging) _session_db = get_shared_session_db() except concurrent.futures.TimeoutError: logger.error( "Job '%s': SessionDB init did not return within %.0fs — proceeding " "without a session store for this run instead of blocking it " "forever", job.get("id", "?"), _session_db_timeout, ) except Exception as e: logger.debug("Job '%s': SQLite session store not available: %s", job.get("id", "?"), e) agent = AIAgent( model=model, api_key=runtime.get("api_key"), base_url=runtime.get("base_url"), provider=runtime.get("provider"), requested_provider=runtime.get("requested_provider"), api_mode=runtime.get("api_mode"), request_overrides=runtime.get("request_overrides"), acp_command=runtime.get("command"), acp_args=runtime.get("args"), max_iterations=max_iterations, reasoning_config=reasoning_config, prefill_messages=prefill_messages, fallback_model=fallback_model, credential_pool=credential_pool, providers_allowed=pr.get("only"), providers_ignored=pr.get("ignore"), providers_order=pr.get("order"), provider_sort=pr.get("sort"), openrouter_min_coding_score=(_cfg.get("openrouter") or {}).get("min_coding_score"), enabled_toolsets=_resolve_cron_enabled_toolsets(job, _cfg), disabled_toolsets=_resolve_cron_disabled_toolsets(_cfg), quiet_mode=True, # Cron jobs should always inherit the user's SOUL.md identity from # HERMES_HOME. When a workdir is configured, also inject project # context files (AGENTS.md / CLAUDE.md / .cursorrules) from there. # Without a workdir, keep cwd context discovery disabled. skip_context_files=not bool(_job_workdir), load_soul_identity=True, # Memory is enabled for cron agents like any other agent run: # MEMORY.md / USER.md load into the system prompt and the memory # tool follows normal toolset resolution, so jobs benefit from # (and can update) the user's persistent memory. skip_memory=False, skip_background_review=True, # Cron has no human-in-the-loop need for skill/memory review forks (~30K tok/event) platform="cron", session_id=_cron_session_id, session_db=_session_db, ) # Run the agent with an *inactivity*-based timeout: the job can run # for hours if it's actively calling tools / receiving stream tokens, # but a hung API call or stuck tool with no activity for the configured # duration is caught and killed. Default 600s (10 min inactivity); # override via HERMES_CRON_TIMEOUT env var. 0 = unlimited. # # Uses the agent's built-in activity tracker (updated by # _touch_activity() on every tool call, API call, and stream delta). _cron_timeout = _cron_inactivity_seconds() _cron_inactivity_limit = _cron_timeout if _cron_timeout > 0 else None _POLL_INTERVAL = 5.0 # Keep the one-shot run_claim fresh while the run is alive (#62002): # the claim TTL is a dead-owner detector, but without a heartbeat a # run that legitimately outlives it (stream stall, laptop asleep # mid-run) is indistinguishable from a dead tick — another process # re-dispatches it and get_due_jobs stale-removes the job record out # from under the live run. Refreshing the claim from this monitor # keeps "expired claim" meaning "owner died". _job_schedule = job.get("schedule") _is_oneshot = ( isinstance(_job_schedule, dict) and _job_schedule.get("kind") == "once" ) _run_claim = job.get("run_claim") _run_claim_owner = ( str(_run_claim.get("by") or "") if isinstance(_run_claim, dict) else "" ) _last_claim_heartbeat = time.monotonic() def _abort_if_fire_claim_lost() -> None: if cancel_event is None or not cancel_event.is_set(): return if agent is not None and hasattr(agent, "interrupt"): agent.interrupt("Cron fire claim ownership was lost") raise RuntimeError( f"Cron job '{job_name}' lost its durable fire claim ownership" ) def _heartbeat_run_claim_if_due(): nonlocal _last_claim_heartbeat if not _is_oneshot or not _run_claim_owner: return _mono = time.monotonic() if _mono - _last_claim_heartbeat < _RUN_CLAIM_HEARTBEAT_SECONDS: return _last_claim_heartbeat = _mono try: heartbeat_run_claim(job_id, expected_owner=_run_claim_owner) except Exception: logger.debug( "Job '%s': run_claim heartbeat failed", job_name, exc_info=True ) _cron_pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) # Preserve scheduler-scoped ContextVar state (for example skill-declared # env passthrough registrations) when the cron run hops into the worker # thread used for inactivity timeout monitoring. _cron_context = contextvars.copy_context() # Tag this fire and time the run_conversation call for the usage_audit.jsonl entry. _audit_fire_id = uuid.uuid4().hex _audit_t_start = time.monotonic() _cron_future = _cron_pool.submit( _cron_context.run, agent.run_conversation, prompt, task_id=_cron_task_id, ) _inactivity_timeout = False _watch_stop = threading.Event() def _idle_seconds() -> float: if not hasattr(agent, "get_activity_summary"): return 0.0 try: _act = agent.get_activity_summary() return float(_act.get("seconds_since_activity", 0.0) or 0.0) except Exception: return 0.0 def _watch_inactivity() -> None: nonlocal _inactivity_timeout if _cron_inactivity_limit is None: return if _inactivity_watchdog_loop( get_idle_seconds=_idle_seconds, limit_s=_cron_inactivity_limit, poll_s=_POLL_INTERVAL, stop=_watch_stop, future_done=_cron_future.done, ): _inactivity_timeout = True _watch_thread = threading.Thread( target=_watch_inactivity, name=f"cron-inactivity-{str(job_id)[:8]}", daemon=True, ) try: if _cron_inactivity_limit is not None: # Daemon thread: kernel ``Event.wait`` timeout, independent of # the ``run_job`` thread. A blocked loop / hung # ``get_activity_summary`` on this thread can no longer keep # the 600s inactivity limit from firing (#94285). _watch_thread.start() if _cron_inactivity_limit is None and not _is_oneshot and cancel_event is None: result = _cron_future.result() else: result = None while True: done, _ = concurrent.futures.wait( {_cron_future}, timeout=_POLL_INTERVAL, ) if done: _abort_if_fire_claim_lost() result = _cron_future.result() break if _inactivity_timeout: break _abort_if_fire_claim_lost() _heartbeat_run_claim_if_due() except Exception: _cron_pool.shutdown(wait=False, cancel_futures=True) raise finally: _watch_stop.set() _cron_pool.shutdown(wait=False, cancel_futures=True) if _inactivity_timeout: # Build diagnostic summary from the agent's activity tracker. _activity = {} if hasattr(agent, "get_activity_summary"): try: _activity = agent.get_activity_summary() except Exception: pass _last_desc = _activity.get("last_activity_desc", "unknown") _secs_ago = _activity.get("seconds_since_activity", 0) _cur_tool = _activity.get("current_tool") _iter_n = _activity.get("api_call_count", 0) _iter_max = _activity.get("max_iterations", 0) logger.error( "Job '%s' idle for %.0fs (inactivity limit %.0fs) " "| last_activity=%s | iteration=%s/%s | tool=%s", job_name, _secs_ago, _cron_inactivity_limit, _last_desc, _iter_n, _iter_max, _cur_tool or "none", ) request_hard_interrupt(agent, "Cron job timed out (inactivity)") raise TimeoutError( f"Cron job '{job_name}' idle for " f"{int(_secs_ago)}s (limit {int(_cron_inactivity_limit)}s) " f"— last activity: {_last_desc}" ) # Guard against non-dict returns from run_conversation under error conditions if not isinstance(result, dict): raise RuntimeError( f"agent.run_conversation returned {type(result).__name__} instead of dict: {result!r}" ) # If the agent itself reported failure (e.g. all retries exhausted on # API errors, model abort, mid-run interrupt), do not silently mark the # job as successful. run_agent populates `failed=True`/`completed=False` # on these paths and may put the error into `final_response`, which # would otherwise be delivered as if it were the agent's reply and the # job's `last_status` set to "ok". Raise so the except handler below # builds the proper failure tuple. (issue #17855) turn_exit_reason = str(result.get("turn_exit_reason") or "") final_response_text = (result.get("final_response") or "").strip() max_iteration_summary = ( result.get("failed") is not True and result.get("completed") is False and turn_exit_reason.startswith("max_iterations_reached(") and bool(final_response_text) ) if result.get("failed") is True or (result.get("completed") is False and not max_iteration_summary): _err_text = ( result.get("error") or final_response_text or "agent reported failure" ) raise RuntimeError(_err_text) if max_iteration_summary: logger.warning( "Job '%s' reached the iteration limit but produced a final fallback response; " "delivering the response instead of failing the cron run", job_name, ) final_response = result.get("final_response", "") or "" # Recover model-mangled computer_use screenshot paths before delivery # media extraction (same repair as the gateway turn/background paths). # Cron runs start a fresh conversation, so history_offset=0. The # helper is fail-open and no-ops without a MEDIA: directive. if final_response: from gateway.media_repair import ( repair_explicit_computer_use_media_paths, ) final_response = repair_explicit_computer_use_media_paths( final_response, result.get("messages", []), ) # Strip leaked placeholder text that upstream may inject on empty completions. if final_response.strip() == "(No response generated)": final_response = "" # Cron silence on abnormal empty turns. The turn-completion explainer # (#34452) replaces a blank/empty model turn with a "⚠️ No reply: …" # string so interactive surfaces (CLI/gateway) explain why the box is # empty. In a cron context that turns a previously-silent empty turn # into a delivered warning (Manfredi's Telegram symptom). Detect the # explainer text deterministically (via the same formatter that # produced it) and treat it as empty so the empty-response suppression # and soft-failure marking below apply — restoring pre-#34452 silence # for scheduled jobs without disabling the explainer everywhere. if final_response.strip() and turn_exit_reason: # The formatter's wording varies by persistence cause (locked / # disk / unknown), so render every variant — matching only the # one-argument render would let cause-refined explainer text slip # through and be delivered as a cron warning. _explainer_variants = [] try: from hermes_state import PERSISTENCE_ERROR_CAUSES as _causes except Exception: _causes = ("locked", "disk", "unknown") for _cause in (None, *_causes): try: _variant = AIAgent._format_turn_completion_explanation( turn_exit_reason, _cause ) except TypeError: # Older single-argument formatter (or a test double). try: _variant = AIAgent._format_turn_completion_explanation( turn_exit_reason ) except Exception: _variant = "" except Exception: _variant = "" if _variant: _explainer_variants.append(_variant.strip()) if final_response.strip() in _explainer_variants: logger.info( "Job '%s': abnormal empty turn (%s) — suppressing explainer for cron delivery", job_id, turn_exit_reason, ) final_response = "" # Use a separate variable for log display; keep final_response clean # for delivery logic (empty response = no delivery). logged_response = final_response if final_response else "(No response generated)" output = f"""# Cron Job: {job_name} **Job ID:** {job_id} **Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')} **Schedule:** {job.get('schedule_display', 'N/A')} ## Prompt {prompt} ## Response {logged_response} """ logger.info("Job '%s' completed successfully", job_name) # Emit one JSONL line per fire for usage audit. _audit_duration_ms = int((time.monotonic() - _audit_t_start) * 1000) _audit_response_silent = _is_cron_silence_response(final_response or "") _write_usage_audit({ "ts": _utcnow_iso_ms(), "job_id": job_id, "fire_id": _audit_fire_id, "prompt_tokens": result.get("prompt_tokens"), "completion_tokens": result.get("completion_tokens"), "total_tokens": result.get("total_tokens"), "response_silent": _audit_response_silent, "deliver_target": job.get("deliver"), "model": model or None, "duration_ms": _audit_duration_ms, "error": None, }) return True, output, final_response, None except Exception as e: error_msg = f"{type(e).__name__}: {str(e)}" logger.exception("Job '%s' failed: %s", job_name, error_msg) # Best-effort audit write on failure path. _audit_fire_id # may be unset if the exception fired before submit() — guard # with a None check so the audit write itself never raises. if "_audit_fire_id" in locals(): _audit_duration_ms = int((time.monotonic() - _audit_t_start) * 1000) _write_usage_audit({ "ts": _utcnow_iso_ms(), "job_id": job_id, "fire_id": _audit_fire_id, "prompt_tokens": None, "completion_tokens": None, "total_tokens": None, "response_silent": False, "deliver_target": job.get("deliver"), "model": model or None, "duration_ms": _audit_duration_ms, "error": error_msg, }) output = f"""# Cron Job: {job_name} (FAILED) **Job ID:** {job_id} **Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')} **Schedule:** {job.get('schedule_display', 'N/A')} ## Prompt {prompt} ## Error ``` {error_msg} ``` """ return False, output, "", error_msg finally: _clear_tool_session_cwd(_cron_task_id) # Clean up ContextVar session/delivery state for this job. # clear_session_vars also clears _SESSION_CWD internally, so no # separate clear_session_cwd() call is needed. clear_session_vars(_ctx_tokens) if _cron_session_token is not None: _cron_session_var.reset(_cron_session_token) if _non_dispatcher_token is not None: exit_non_dispatcher_owned_context(_non_dispatcher_token) for _var_name in _cron_delivery_vars: _VAR_MAP[_var_name].set("") if _session_db: # The agent turn has already returned. Bound every subsequent DB # operation so storage failure cannot hold the dispatch guard. _session_db = _BoundedCronSessionDB(_session_db, job_id) # Compression can rotate the live agent onto a continuation while # this run is in flight. Finalize that continuation, not the stale # cron id captured before AIAgent started. SessionDB is the source # of truth for the lineage; agent.session_id is only a fail-safe # when the lookup itself is unavailable. _final_cron_session_id = _cron_session_id try: _compression_tip = _session_db.get_compression_tip( _cron_session_id ) if _compression_tip: _final_cron_session_id = _compression_tip except (Exception, KeyboardInterrupt) as e: try: _agent_session_id = getattr(agent, "session_id", None) if _agent_session_id: _final_cron_session_id = _agent_session_id except (Exception, KeyboardInterrupt): pass logger.debug( "Job '%s': failed to resolve cron compression tip: %s", job_id, e, ) # Title the cron session from the job (name -> id) and PERSIST it # BEFORE end_session()/close() tear the connection down, so the # close can never run over an in-flight title write (#50536). The # run-time suffix keeps it unique against the sessions.title index # across runs; _set_cron_session_title dedupes (#50537) and the # except-fallback below guarantees a non-blank title (#50535). try: _title_base = " ".join(job_name.split())[:60].strip() or f"cron {job_id}" _cron_title = f"{_title_base} · {_hermes_now().strftime('%b %d %H:%M')}" if not _set_cron_session_title( _session_db, _final_cron_session_id, _cron_title ): # Helper returned None (blank base) -> use the id fallback. _set_cron_session_title( _session_db, _final_cron_session_id, f"cron {job_id}" ) except (Exception, KeyboardInterrupt) as e: logger.debug( "Job '%s': failed to set cron session title: %s", job_id, e ) # Last-resort: never leave the session blank (#50535). Try the # next free title in the lineage, then a bare id-stamped title. for _fallback in ( getattr(_session_db, "get_next_title_in_lineage", lambda b: b)( f"cron {job_id}" ), f"cron {job_id} {_final_cron_session_id[-6:]}", ): try: if _set_cron_session_title( _session_db, _final_cron_session_id, _fallback ): break except (Exception, KeyboardInterrupt): continue # Verified completion booking (#93820): the run may only be # recorded as cron_complete when the session's LAST message row is # a real assistant reply — a plain answer or the [SILENT] sentinel # (both are assistant-text rows, so both classify as 'complete'). # A turn that died after a tool call, mid-API-wait, or without any # assistant text leaves the last row as a tool result / pending # call / user prompt and must not surface as a healthy run. # session_lifecycle_statuses is the existing cost-bounded # classifier for exactly this shape. Only a POSITIVELY recognized # pathological status (see the status vocabulary in # hermes_state's session_lifecycle_statuses docstring — keep the # tuple below in sync when it grows) downgrades the booking: an # unknown value (newer classifier shape, test doubles) keeps the # historical reason, and so does a failed probe — the booking # itself is FAIL-OPEN on probe errors, because classification is # best-effort metadata and must not mislabel a healthy run. _end_reason = "cron_complete" try: _statuses = _session_db.session_lifecycle_statuses( [_final_cron_session_id] ) _lifecycle = _statuses.get(_final_cron_session_id) if _lifecycle in ("interrupted", "error", "empty"): _end_reason = "cron_incomplete_no_output" logger.warning( "Job '%s': session ended without a final assistant " "message (lifecycle=%s) — booking run as %s", job_id, _lifecycle, _end_reason, ) except (Exception, KeyboardInterrupt) as e: logger.debug( "Job '%s': session lifecycle classification failed: %s", job_id, e, ) try: _session_db.end_session( _final_cron_session_id, _end_reason ) except (Exception, KeyboardInterrupt) as e: logger.debug("Job '%s': failed to end session: %s", job_id, e) try: from hermes_state import release_or_close release_or_close(_session_db) except (Exception, KeyboardInterrupt) as e: logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e) # Release subprocesses, terminal sandboxes, browser daemons, and the # main OpenAI/httpx client held by this ephemeral cron agent. Without # this, a gateway that ticks cron every N minutes leaks fds per job # until it hits EMFILE (#10200 / "too many open files"). # # When the caller opted to defer teardown (passed a list), hand the live # agent back instead of closing it here — delivery must run against a # live async client, and the caller tears down afterwards (#58720). if defer_agent_teardown is not None: if agent is not None: defer_agent_teardown.append(agent) else: _teardown_cron_agent(agent, job_id) def _teardown_cron_agent( agent, job_id: str, *, timeout_seconds: Optional[float] = None ) -> None: """Release an ephemeral cron agent's async resources within a hard bound. Split out of ``run_job``'s ``finally`` so a caller that defers teardown (to deliver first — #58720) can invoke the identical cleanup AFTER delivery. The timeout matters because this executes after ``run_conversation`` has returned, outside the agent inactivity watchdog. """ def _cleanup_agent() -> None: try: if agent is not None: agent.close() except (Exception, KeyboardInterrupt) as e: logger.debug("Job '%s': failed to close agent resources: %s", job_id, e) # Each cron run spins up a short-lived worker thread whose event loop # dies as soon as the ``ThreadPoolExecutor`` shuts down. Any async # httpx clients cached under that loop are now unusable — reap them. try: from agent.auxiliary_client import cleanup_stale_async_clients cleanup_stale_async_clients() except Exception as e: logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e) _run_cron_cleanup_with_timeout( _cleanup_agent, job_id=job_id, label="agent resource teardown", timeout_seconds=timeout_seconds, ) def _run_with_fire_claim_heartbeat(job: dict, run) -> bool: """Run ``run`` while keeping this job's owned durable fire claim fresh.""" claim = job.get("fire_claim") owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" if not owner: return run(None) job_id = str(job.get("id") or "") stop = threading.Event() lost_ownership = threading.Event() heartbeat_context = contextvars.copy_context() def _finish_unstarted(error: str) -> None: execution_id = job.get("execution_id") if not execution_id: return try: finish_execution(execution_id, success=False, error=error) except Exception: logger.warning( "Job '%s': failed to close unstarted execution ledger row", job_id, exc_info=True, ) try: owns_fire_claim = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: logger.warning( "Job '%s': initial fire_claim validation failed", job_id, exc_info=True, ) _finish_unstarted( "Fire claim ownership could not be validated before execution started." ) return True if owns_fire_claim is False: logger.warning( "Job '%s': fire claim ownership was already lost before execution", job_id, ) _finish_unstarted("Fire claim ownership lost before execution started.") return True def _heartbeat_loop() -> None: last_confirmed = time.monotonic() while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: if not heartbeat_fire_claim(job_id, expected_owner=owner): lost_ownership.set() logger.warning( "Job '%s': fire claim ownership lost; interrupting stale run", job_id, ) return last_confirmed = time.monotonic() except Exception: logger.debug( "Job '%s': fire_claim heartbeat failed", job_id, exc_info=True, ) if ( time.monotonic() - last_confirmed >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS ): lost_ownership.set() logger.warning( "Job '%s': fire_claim could not be renewed within %.1fs; " "interrupting uncertain run", job_id, _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, ) return heartbeat_thread = threading.Thread( target=heartbeat_context.run, args=(_heartbeat_loop,), name="cron-fire-claim-heartbeat", daemon=True, ) try: heartbeat_thread.start() except Exception: logger.warning( "Job '%s': could not start fire_claim heartbeat", job_id, exc_info=True, ) _finish_unstarted( "Fire claim heartbeat could not be started; execution was not run." ) return True try: return run(lost_ownership) finally: stop.set() heartbeat_thread.join(timeout=1.0) def run_one_job( job: dict, *, adapters=None, loop=None, verbose: bool = False, extra_prompt: Optional[str] = None, cancel_event: Optional[_CancelEventLike] = None, ) -> bool: """Run ONE due job end-to-end: execute → save output → deliver → mark. This is the shared firing body extracted from ``tick``'s per-job closure so that BOTH the built-in ticker and an external provider's ``fire_due`` (e.g. Chronos) run the identical sequence — no duplicated correctness. It does NOT decide whether the job is due or acquire the initial claim — both the ticker and external providers use the same store CAS before calling it. It does keep an acquired claim alive for the full execution. Returns True if the job was processed (even if the job itself failed — failure is recorded via ``mark_job_run``), False only if processing raised. ``cancel_event``: optional transport-level cancellation source (dashboard webhook drain, API server shutdown). It is OR-combined with the internal fire-claim heartbeat's lost-ownership event, so either trigger stops the run cooperatively — agent interruption AND script process-tree kill — through the single fenced completion path. """ # Every gateway path (built-in scheduler, external providers, and direct # API fires) crosses this seam. Ensure the detached worker has a durable # attempt to adopt before any launch can occur. if not job.get("execution_id"): execution = create_execution(job["id"], source="direct") job["execution_id"] = execution["id"] execution_id = str(job["execution_id"]) external_owner = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") == execution_id if not external_owner: try: if _launch_external_cron_worker(job): return True except Exception as handoff_error: error = f"Restart-safe cron worker dispatch failed: {handoff_error}" logger.error("Job '%s': %s", job["id"], error) claim = job.get("fire_claim") owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" try: mark_job_run( job["id"], False, error, **({"expected_fire_owner": owner} if owner else {}), ) finally: finish_execution(execution_id, success=False, error=error) return True if extra_prompt is None: # A gateway-forwarded manual run (`hermes cron run --prompt` / # cronjob(action='run', prompt=...) on a relay-fronted target) stamps # its transient context on the job via trigger_job; the ticker/Chronos # fire that consumes the manual occurrence carries it here. Single-fire: # mark_job_run clears the field after the run. _stamped = job.get("manual_run_prompt") if _stamped and job.get("manual_run_at"): extra_prompt = str(_stamped) claim = job.get("fire_claim") fire_owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" execution_token = object() profile_home = _get_hermes_home().resolve() with _running_lock: _running_fire_owners.setdefault(job["id"], {})[execution_token] = ( fire_owner or None, profile_home, ) try: return _run_with_fire_claim_heartbeat( job, lambda lost_ownership: _run_one_job_body( job, adapters=adapters, loop=loop, verbose=verbose, extra_prompt=extra_prompt, fire_claim_lost=( _CombinedCancelEvent(lost_ownership, cancel_event) if cancel_event is not None else lost_ownership ), execution_token=execution_token, ), ) finally: with _running_lock: executions = _running_fire_owners.get(job["id"]) if executions is not None: executions.pop(execution_token, None) if not executions: _running_fire_owners.pop(job["id"], None) def _run_one_job_body( job: dict, *, adapters=None, loop=None, verbose: bool = False, extra_prompt: Optional[str] = None, fire_claim_lost: Optional[_CancelEventLike] = None, execution_token: Optional[object] = None, ) -> bool: claim = job.get("fire_claim") fire_owner = str(claim.get("by") or "") if isinstance(claim, dict) else None class _FireClaimLostDuringSideEffect(Exception): pass def _side_effect_fence(): if fire_owner is None: return contextlib.nullcontext(True) return fire_claim_fence(job["id"], expected_owner=fire_owner) def _fire_claim_ownership_lost() -> bool: if fire_claim_lost is not None and fire_claim_lost.is_set(): return True if fire_owner is None: return False try: if heartbeat_fire_claim(job["id"], expected_owner=fire_owner): return False except Exception: logger.debug( "Job '%s': fire_claim ownership validation failed", job["id"], exc_info=True, ) return False if fire_claim_lost is not None: fire_claim_lost.set() return True execution_id = job.get("execution_id") if not execution_id: execution_id = create_execution(job["id"], source="direct")["id"] delivery_attempted = False delivery_error = None # Durable failure-incident bookkeeping for this run (see cron.incidents): # set on the failure paths below; consumed by the delivery_outcome # computation and the post-delivery "alerted" transition. incident_acked = False failure_incident_id = None from agent.secret_scope import ( build_profile_secret_scope, reset_secret_scope, set_secret_scope, ) _scope_token = None _terminal_scope_token = None 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"]), ) finish_execution( execution_id, success=False, error="Dispatch claim rejected; execution was not started.", ) return True # not an error — already handled/removed # The attempt is claimed durably before executor/provider dispatch and # becomes running only immediately before the actual run. # Detached workers atomically transition the attempt to running while # adopting it. In-process paths must win the claimed->running CAS # here before any user script or agent side effect may begin. external_owner = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") == execution_id if not external_owner and mark_execution_running(execution_id) is None: logger.warning( "Cron job %s lost execution ownership before start; skipping", job["id"], ) return True # Run and deliver under the profile's secret scope. get_secret() fails # closed outside a scope once profile isolation is active, and cron # fires from a ticker thread with no per-turn scope. Delivery adapters # can also resolve credentials, so resetting after run_job would leave # _deliver_result unscoped. Mirrors gateway/run.py's per-turn pattern. _scope_token = set_secret_scope( build_profile_secret_scope(_get_hermes_home()) ) # Same isolation for terminal settings (third profile seam; see # gateway/run.py _profile_runtime_scope): installs the firing # profile's COMPLETE terminal policy for this fire — run, delivery, # and bookkeeping — resetting in this function's finally alongside # the secret scope. Without it the ticker thread reads the # process-global TERMINAL_* env vars a concurrent profile's turn may # have pinned (#68559). Resolution failure installs a refusal scope: # terminal execution inside the fire raises instead of falling back # to the launch process's ambient policy. from tools.terminal_scope import ( install_profile_terminal_scope, ) _terminal_scope_token = install_profile_terminal_scope( _get_hermes_home() ) # Defer the cron agent's async-resource teardown until AFTER delivery. # run_job normally closes the agent (and reaps stale async clients) in # its finally block; doing that before _deliver_result runs means the # live send races a torn-down async client (#58720). Passing a holder # list makes run_job hand the agent back instead, and we tear it down # below once delivery is done. Defense-in-depth alongside the # interpreter-shutdown guard in _deliver_result. _deferred_agents: list = [] try: if fire_claim_lost is None: success, output, final_response, error = run_job( job, defer_agent_teardown=_deferred_agents, extra_prompt=extra_prompt, execution_id=execution_id, ) else: success, output, final_response, error = run_job( job, defer_agent_teardown=_deferred_agents, extra_prompt=extra_prompt, cancel_event=fire_claim_lost, execution_id=execution_id, ) except BaseException: # run_job's finally still hands back the agent when it raises; tear # it down here so a failed run never leaks its async resources # (#10200), then re-raise into the outer handler. BaseException # (not just Exception) so a KeyboardInterrupt/SystemExit mid-run # still triggers teardown before propagating. for _deferred_agent in _deferred_agents: _teardown_cron_agent(_deferred_agent, job["id"]) raise # The outer finally resets the scope after delivery and bookkeeping. if _fire_claim_ownership_lost(): for _deferred_agent in _deferred_agents: _teardown_cron_agent(_deferred_agent, job["id"]) # Distinguish a real ownership loss (TTL expiry / replacement # claim) from a transport-level cancel (dashboard drain): in the # latter case WE still own the claim, and silently discarding # would leave fire_claim lingering until TTL and last_status # stale. Probe ownership once; if still ours, record the # interruption through the owner-fenced terminal write. if fire_owner is not None and heartbeat_fire_claim( job["id"], expected_owner=fire_owner, ): mark_job_run( job["id"], False, "Interrupted by shutdown before terminal completion.", expected_fire_owner=fire_owner, ) finish_execution( execution_id, success=False, error="Interrupted by shutdown before terminal completion.", ) else: finish_execution( execution_id, success=False, error="Fire claim ownership lost; stale result was discarded.", ) return True # Everything from here through delivery runs with the agent still live # (deferred teardown). Wrap it ALL in a try/finally so that if any step # between run_job returning and delivery — save_job_output, the [SILENT] # / empty-response computation, or _deliver_result itself — raises, the # deferred agent is still torn down. Otherwise the outer `except` would # swallow the error and leak the agent's subprocesses/clients (#10200). blocked_config = False side_effect_ownership_lost = False try: with _side_effect_fence() as owns_output: if not owns_output: raise _FireClaimLostDuringSideEffect output_file = save_job_output(job["id"], output) if verbose: logger.info("Output saved to: %s", output_file) # If the gateway shutdown killed this job's tool subprocess # mid-flight (#60432), the agent may still have produced a # plausible-looking final_response from the truncated output -- # force the failure path so the delivered message is an honest # "this run was interrupted" summary instead of that response. # Peek-only: the flag stays set for the authoritative check # right before mark_job_run below. if success and _is_interrupted(job["id"], execution_token): success = False error = ( "Interrupted by gateway shutdown before the run finished " "(tool subprocess was killed mid-flight)." ) # Deliver the final response to the origin/target chat. # If the agent responded with [SILENT], skip delivery (but # output is already saved above). Failed jobs always deliver. # # Exception: a run blocked by pre-dispatch config validation # (T1-26) alerts exactly ONCE — the silent marker means the # operator was already told on a previous tick, so re-delivering # the same alert every tick would be spam (#73506 alert-once # shape). blocked_config_silent = ( bool(error) and BLOCKED_CONFIG_SILENT_MARKER in str(error) ) blocked_config = blocked_config_silent or ( bool(error) and BLOCKED_CONFIG_MARKER in str(error) ) # Drift-guard skip (#44585): same alert-once contract as # blocked_config — the silent marker means the operator already # got the alert on a previous tick. drift_skip_silent = ( bool(error) and DRIFT_SKIP_SILENT_MARKER in str(error) ) drift_skip = drift_skip_silent or ( bool(error) and DRIFT_SKIP_MARKER in str(error) ) if blocked_config and not success: # Blocked-config alert: bypass the generic failure summarizer # (whose auth/timeout heuristics would mislabel this as a # provider runtime failure) — say plainly that config # validation blocked the run and nothing was spent. _pf_text = re.sub( r"\[blocked_config[^\]]*\]\s*", "", str(error) ).strip() deliver_content = ( f"⛔ Cron '{job.get('name') or job['id']}' blocked by " f"configuration validation (no LLM call was made): " f"{_pf_text} " "This alert is sent once; the job stays blocked until " "the configuration is fixed." ) else: if success: deliver_content = final_response else: # Durable failure incident: record this job+error # signature once and, when the operator already acked it, # suppress the per-run failure ping (the streak nudge and # the failure summarizer stay intact for un-acked # failures). Best-effort — an incident-store error never # breaks the delivery path (see _upsert_incident_for_failure). incident_acked, failure_incident_id = _upsert_incident_for_failure( job, error or "", output_file=output_file ) if incident_acked and not drift_skip: deliver_content = "" else: deliver_content = ( _summarize_cron_failure_for_delivery(job, error) + _failure_streak_nudge(job) ) if drift_skip and not success: # Drift-skip alert: bypass the generic summarizer's # 180-char truncation (it would eat the remediation # command) and strip the internal marker — deliver the # guard's own actionable message intact. # Deliberately NOT gated on incident ack: a drift skip # means the run was never attempted and the message # carries the remediation command — acking the failure # signature silences failure pings, not drift alerts # (which already alert once via the drift_alerted marker). _drift_text = re.sub( r"\[drift_skip[^\]]*\]\s*", "", str(error) ).strip() deliver_content = ( f"⚠️ Cron '{job.get('name') or job['id']}' skipped: " f"{_drift_text}" ) # Treat whitespace-only final responses the same as empty # responses: do not deliver a blank message, and let the # empty-response guard below mark the run as a soft failure. should_deliver = bool(deliver_content.strip()) if blocked_config_silent or drift_skip_silent: should_deliver = False unresolved_origin = False # Cron silence suppression — see _is_cron_silence_response. Replaces the # old `SILENT_MARKER in ...upper()` substring check, which both leaked # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed # a real report that merely quoted "[SILENT]" mid-sentence (#51438, # #46917). Keeps the intentional bracketed-prefix / trailing-line # tolerance the cron contract relies on. if should_deliver and success and _is_cron_silence_response(deliver_content): logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) should_deliver = False if should_deliver and _fire_claim_ownership_lost(): should_deliver = False logger.warning( "Job '%s': skipping delivery after fire claim ownership loss", job["id"], ) if should_deliver: unresolved_origin = ( _normalize_deliver_value(_delivery_lane_value(job, for_failure=not success)) == "origin" and not _resolve_delivery_targets(job, for_failure=not success) ) try: with _side_effect_fence() as owns_delivery: if not owns_delivery: raise _FireClaimLostDuringSideEffect delivery_attempted = True delivery_error = _deliver_result( job, deliver_content, adapters=adapters, loop=loop, # Failure summaries (and drift/blocked-config alerts # composed into deliver_content on the failure path) # honor the job's failure_deliver override (NS-788). for_failure=not success, ) except Exception as de: if isinstance(de, _FireClaimLostDuringSideEffect): raise delivery_error = str(de) logger.error("Delivery failed for job %s: %s", job["id"], de) except _FireClaimLostDuringSideEffect: side_effect_ownership_lost = True finally: # Tear down the deferred agent(s) now that save + delivery have run # (or raised). Must happen on every path so cron agents never leak # their subprocesses/clients (#10200). for _deferred_agent in _deferred_agents: _teardown_cron_agent(_deferred_agent, job["id"]) if side_effect_ownership_lost or _fire_claim_ownership_lost(): # Same transport-cancel distinction as the pre-side-effect path: # if WE still own the claim, record the interruption instead of # discarding silently (lingering claim + stale last_status). if fire_owner is not None and heartbeat_fire_claim( job["id"], expected_owner=fire_owner, ): mark_job_run( job["id"], False, "Interrupted by shutdown before terminal completion.", expected_fire_owner=fire_owner, ) finish_execution( execution_id, success=False, error="Interrupted by shutdown before terminal completion.", ) else: finish_execution( execution_id, success=False, error="Fire claim ownership lost; stale result was discarded.", ) return True # Treat empty final_response as a soft failure so last_status # is not "ok" — the agent ran but produced nothing useful. # (issue #8585) if success and not final_response.strip(): success = False error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)" interrupted = _consume_interrupted_flag(job["id"], execution_token) if interrupted: if delivery_error: # The gateway shutdown already wrote last_status for this run, # so mark_job_run is skipped below — but it could not know that # the notice we just tried to send never left the process (the # adapters were torn down first, #82232). Record the delivery # failure on its own via update_job: mark_job_run also advances # next_run_at and the repeat counter, and running that a second # time for one run would skip a fire or auto-delete the job # early. try: from cron.jobs import update_job update_job(job["id"], {"last_delivery_error": delivery_error}) except Exception as _rec_err: logger.debug( "Failed recording delivery_error for interrupted job %s: %s", job["id"], _rec_err, ) finish_execution( execution_id, success=False, error="Interrupted by gateway shutdown before terminal completion.", ) return True mark_kwargs = {"delivery_error": delivery_error} if fire_owner is not None: mark_kwargs["expected_fire_owner"] = fire_owner if blocked_config: mark_kwargs["status"] = "blocked_config" marked = mark_job_run(job["id"], success, error, **mark_kwargs) if fire_owner is not None and not marked: finish_execution( execution_id, success=False, error="Fire claim ownership lost before terminal completion.", ) return True normalized_deliver = _normalize_deliver_value( _delivery_lane_value(job, for_failure=not success) ) if delivery_error: delivery_outcome = "failed" elif should_deliver and unresolved_origin: delivery_outcome = "not_configured" elif should_deliver and normalized_deliver != "local": delivery_outcome = "delivered" elif incident_acked and not success: # Distinct from plain "suppressed" (silence marker / local jobs): # the failure ping was withheld because the operator acked this # exact signature via `hermes cron incidents ack`. delivery_outcome = "suppressed_acked" else: delivery_outcome = "suppressed" if delivery_outcome in ("delivered", "not_configured") and not success: # The failure ping left the process (or was composed for a # configured target) — record it on the incident so the CLI # distinguishes "failure seen" from "operator was pinged". _mark_incident_alerted(failure_incident_id) finish_execution( execution_id, success=success, error=error, delivery_outcome=delivery_outcome, ) return True except BaseException as e: # noqa: BLE001 — deliberate: see below # BaseException, not Exception (#73973): the inner run_job handler # re-raises CancelledError / KeyboardInterrupt / SystemExit after agent # teardown, and none of those are Exception subclasses. If they escape # without mark_job_run(False), a finite one-shot is left wedged — # claim_dispatch() already consumed repeat.completed, but last_run_at # is never written, so the job sits in state "scheduled" until the # run-claim TTL expires and the dispatch-limit guard removes it with # no output and no error. Record the failure first, then re-raise # anything that isn't a plain Exception. Owner fencing still applies: # a stale worker must not record over a replacement claim owner. _err_text = str(e) or type(e).__name__ logger.error( "Error processing job %s: %s", job["id"], _err_text, exc_info=(type(e), e, e.__traceback__), ) delivery_outcome = "suppressed" # Owner fencing: a stale worker whose fire claim was taken over (or a # transport-cancelled worker) must not send a failure alert on top of # the replacement run's own delivery — fall through silently and let # the fenced bookkeeping below decide what (if anything) to record. if ( isinstance(e, Exception) and not delivery_attempted and not isinstance(e, _FireClaimLostDuringSideEffect) and not _fire_claim_ownership_lost() ): normalized_deliver = _normalize_deliver_value( _delivery_lane_value(job, for_failure=True) ) unresolved_origin = False # Durable failure incident: same ack gate as the normal failure # delivery above — an acked signature stays silent on this path # too, so the retry-path alert cannot re-ping after acknowledgment. incident_acked, failure_incident_id = _upsert_incident_for_failure( job, _err_text ) if incident_acked: delivery_outcome = "suppressed_acked" else: try: delivery_attempted = True delivery_error = _deliver_result( job, # Composed exactly like the normal failure delivery above. # mark_job_run below records THIS run in failure_streak # whichever layer failed, so a job that fails before the # run body every tick builds a streak nobody is ever told # about: its alerts only ever leave through here, and the # nudge only ever left through there (#88655). _summarize_cron_failure_for_delivery(job, _err_text) + _failure_streak_nudge(job), adapters=adapters, loop=loop, for_failure=True, ) except Exception as delivery_exc: delivery_error = str(delivery_exc) logger.error( "Delivery failed for job %s: %s", job["id"], delivery_exc ) if not delivery_error and normalized_deliver == "origin": unresolved_origin = not _resolve_delivery_targets( job, for_failure=True ) if delivery_error: delivery_outcome = "failed" elif unresolved_origin: delivery_outcome = "not_configured" elif normalized_deliver != "local": delivery_outcome = "delivered" if delivery_outcome in ("delivered", "not_configured"): _mark_incident_alerted(failure_incident_id) try: if not _consume_interrupted_flag(job["id"], execution_token): mark_kwargs = {} if fire_owner is not None: mark_kwargs["expected_fire_owner"] = fire_owner if isinstance(e, Exception): mark_kwargs["delivery_error"] = delivery_error mark_job_run(job["id"], False, _err_text, **mark_kwargs) except Exception as record_err: # Never let bookkeeping mask the original interruption. logger.error( "Failed to record interrupted run for job %s: %s", job["id"], record_err, ) try: finish_execution( execution_id, success=False, error=_err_text, delivery_outcome=delivery_outcome, ) except Exception as record_err: logger.error( "Failed to finish execution record for job %s: %s", job["id"], record_err, ) if not isinstance(e, Exception): raise return False finally: # Function-level on purpose: this must scope delivery, deferred-agent # teardown, claim-loss handling and bookkeeping, not just run_job. # An earlier revision reset inside the run block's finally, which left # _deliver_result unscoped — do not move it back in a tidy-up. if _scope_token is not None: reset_secret_scope(_scope_token) if _terminal_scope_token is not None: from tools.terminal_scope import reset_terminal_scope reset_terminal_scope(_terminal_scope_token) def _wait_for_external_cron_worker_body( process: subprocess.Popen, *, execution_id: str, ) -> bool: """Preserve ``run_one_job``'s synchronous contract after handoff. The worker owns the durable execution and survives this gateway process. The caller nevertheless waits while it remains alive so manual/background callers do not release their in-process guard or report stale job state. A gateway replacement may kill this waiter; it does not kill the scoped worker or change its ledger ownership. """ def _is_terminal() -> bool: current = get_execution(execution_id) return bool(current and current.get("status") in _TERMINAL_STATES) # The worker commits its terminal row before its process exits, so exit is # the correct wakeup. Each ledger read opens a connection and re-runs # schema init; polling it at 50ms for an hours-long agent run is ~72k # opens/hour of pure contention with the worker's own writes. while True: try: returncode = process.wait(timeout=1.0) except subprocess.TimeoutExpired: if _is_terminal(): return True continue # The worker can commit its terminal row and exit between the first # read and wait(). Re-read the exact attempt before declaring that # it died without terminalizing. if _is_terminal(): return True # If the adopted worker died without terminalizing, its owner is # now provably gone. Recover to ``unknown`` rather than routing the # exception through the pre-handoff dispatch-failure path, which # would falsely assert that no side effect could have happened. recover_interrupted_executions() if _is_terminal(): return True raise RuntimeError( "cron external worker exited before durable recovery could " f"terminalize its execution state (exit {returncode})" ) def _wait_for_external_cron_worker( process: subprocess.Popen, *, execution_id: str, job_id: Optional[str] = None, handoff_files: tuple[Path, ...] = (), ) -> bool: try: return _wait_for_external_cron_worker_body( process, execution_id=execution_id ) finally: if job_id is not None: with _running_lock: _restart_safe_waiter_job_ids.discard(job_id) # The execution is terminal or its worker is dead: nobody will read a # payload or acknowledgement left behind by a late/unread handoff. for stale in handoff_files: try: stale.unlink(missing_ok=True) except OSError: pass def _launch_external_cron_worker(job: dict) -> bool: """Launch *job* outside a managed gateway cgroup when required. Returns ``False`` when the caller is not a managed systemd gateway and the existing in-process path should be used. In managed topology, failure to establish the transient scope raises: falling back would recreate the restart interruption this handoff exists to prevent. """ execution_id = str(job["execution_id"]) job_id = str(job["id"]) handoff_dir = _get_hermes_home() / "cron" / "external-workers" payload_path = handoff_dir / f"{execution_id}.json" ack_path = handoff_dir / f"{execution_id}.ready" command = [ sys.executable, "-m", "cron.scheduler", "--external-worker-file", str(payload_path), "--ack-file", str(ack_path), ] from agent.secret_scope import is_multiplex_active from tools.environments.local import build_subprocess_env from tools.process_registry import restart_safe_gateway_child_argv multiplex_active = is_multiplex_active() scoped_command = restart_safe_gateway_child_argv( command, unit_suffix=f"cron-{job_id}-exec-{execution_id}", ) if scoped_command == command: return False if mark_execution_handoff_pending(execution_id) is None: raise RuntimeError( "cron execution claim changed before external worker handoff" ) _ensure_cron_dir(handoff_dir) try: handoff_dir.chmod(0o700) except OSError: pass fd = os.open(payload_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "w", encoding="utf-8") as payload_file: json.dump( { "job": job, "profile_home": str(_get_hermes_home().resolve()), "multiplex_active": multiplex_active, }, payload_file, ) payload_file.flush() os.fsync(payload_file.fileno()) except BaseException: payload_path.unlink(missing_ok=True) raise worker_env = build_subprocess_env( scrub_secrets=multiplex_active, inherit_profile_home=True, extra={"HERMES_HOME": str(_get_hermes_home().resolve())}, ) try: process = subprocess.Popen( scoped_command, cwd=str(Path(__file__).resolve().parent.parent), env=worker_env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, creationflags=windows_hide_flags(), ) except BaseException: payload_path.unlink(missing_ok=True) raise with _running_lock: _restart_safe_waiter_job_ids.add(job_id) deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: if ack_path.exists(): try: acknowledgement = json.loads(ack_path.read_text(encoding="utf-8")) except Exception: logger.exception( "Cron external worker %s published an unreadable acknowledgement; " "treating handoff as ownership-uncertain", execution_id, ) return _wait_for_external_cron_worker( process, execution_id=execution_id, job_id=job_id, handoff_files=(payload_path,), ) finally: ack_path.unlink(missing_ok=True) if ( not isinstance(acknowledgement, dict) or acknowledgement.get("execution_id") != execution_id ): logger.error( "Cron external worker acknowledgement mismatch for %s; " "treating handoff as ownership-uncertain", execution_id, ) return _wait_for_external_cron_worker( process, execution_id=execution_id, job_id=job_id, handoff_files=(payload_path,), ) logger.info( "Cron job '%s' handed to restart-safe worker pid=%s execution=%s", job_id, acknowledgement.get("pid"), execution_id, ) return _wait_for_external_cron_worker( process, execution_id=execution_id, job_id=job_id, handoff_files=(payload_path,), ) returncode = process.poll() if returncode is not None: with _running_lock: _restart_safe_waiter_job_ids.discard(job_id) payload_path.unlink(missing_ok=True) raise RuntimeError( f"cron external worker exited before ownership acknowledgement " f"(exit {returncode})" ) time.sleep(0.05) # The child may have adopted the durable row just before publishing its # acknowledgement. Never fall back to in-process execution on an uncertain # handoff: that could duplicate side effects. The execution owner/dead-owner # recovery ledger remains the authority. logger.warning( "Cron external worker for job '%s' did not acknowledge within 5s; " "leaving the durable execution claim untouched", job_id, ) return _wait_for_external_cron_worker( process, execution_id=execution_id, job_id=job_id, handoff_files=(payload_path, ack_path), ) def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool: """Adopt and execute one gateway-dispatched cron payload. The execution row is created by the gateway before spawn, then transferred here before the ready acknowledgement is published. No side effect runs unless that durable ownership transfer succeeds. """ try: payload = json.loads(payload_path.read_text(encoding="utf-8")) job = payload["job"] profile_home = Path(payload["profile_home"]).resolve() execution_id = str(job["execution_id"]) except Exception: logger.exception("Cron external worker could not load payload %s", payload_path) return False finally: try: payload_path.unlink(missing_ok=True) except OSError: pass from agent.secret_scope import ( build_profile_secret_scope, is_multiplex_active, reset_secret_scope, set_multiplex_active, set_secret_scope, ) from cron.executions import adopt_claimed_execution from hermes_cli.env_loader import hydrate_profile_secret_sources from hermes_constants import ( reset_hermes_home_override, set_hermes_home_override, ) home_token = set_hermes_home_override(profile_home) previous_multiplex = is_multiplex_active() multiplex_active = bool(payload.get("multiplex_active", False)) set_multiplex_active(multiplex_active) hydrate_profile_secret_sources(profile_home) secret_token = set_secret_scope(build_profile_secret_scope(profile_home)) try: with use_cron_store(profile_home): if adopt_claimed_execution(execution_id) is None: logger.error( "Cron external worker refused execution %s: durable ownership " "could not be established", execution_id, ) return False try: ack_path.parent.mkdir(parents=True, exist_ok=True) fd = os.open(ack_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as ack_file: json.dump({"pid": os.getpid(), "execution_id": execution_id}, ack_file) ack_file.flush() os.fsync(ack_file.fileno()) except Exception: logger.exception( "Cron external worker could not publish ready acknowledgement for %s", execution_id, ) return False old_external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") os.environ["_HERMES_CRON_EXTERNAL_WORKER"] = execution_id try: return run_one_job(job, adapters=None, loop=None, verbose=False) finally: if old_external_execution is None: os.environ.pop("_HERMES_CRON_EXTERNAL_WORKER", None) else: os.environ["_HERMES_CRON_EXTERNAL_WORKER"] = old_external_execution finally: reset_secret_scope(secret_token) set_multiplex_active(previous_multiplex) reset_hermes_home_override(home_token) def _notify_provider_jobs_changed() -> None: """Best-effort: tell the active scheduler provider the job set changed. Called by the consumer surfaces (model tool / CLI / REST) AFTER a successful store mutation (create/update/remove/pause/resume) so an external provider (Chronos) can re-provision/cancel the affected one-shot via NAS. No-op for the built-in (it re-reads jobs.json each tick), so the default path is unchanged. Lives here (not in cron/jobs.py) to keep the store free of provider imports — avoids an import cycle and keeps jobs.py low-coupling. Never raises into the caller. """ try: from cron.scheduler_provider import resolve_cron_scheduler resolve_cron_scheduler().on_jobs_changed() except Exception as e: logger.debug("on_jobs_changed notify failed: %s", e) class CronSchedulerRegistrationError(RuntimeError): """A job was persisted but its first external trigger was not registered.""" def __init__(self, job: dict, cause: Exception) -> None: self.job = job self.cause = cause super().__init__( f"Cron job '{job['id']}' was saved, but its first scheduler " f"registration failed ({type(cause).__name__}). Do not create a " "duplicate. Pause/resume or update the job to retry registration." ) def user_message(self) -> str: """Human-facing variant for chat/CLI surfaces (no exception class name).""" label = self.job.get("name") or self.job["id"] return ( f"Saved cron job '{label}', but couldn't register it with the " "external scheduler yet. The job is kept — don't re-create it; " "pause/resume or edit it (e.g. via /cron) to retry registration." ) def to_dict(self) -> dict: """Return the public partial-failure contract without provider details.""" return { "error": str(self), "job_id": self.job["id"], "job_saved": True, "scheduler_registered": False, "retry_create": False, } def create_job_with_scheduler_registration(**kwargs) -> dict: """Persist one job and register its first trigger with the active provider.""" from cron.jobs import create_job from cron.scheduler_provider import resolve_cron_scheduler job = create_job(**kwargs) try: resolve_cron_scheduler().register_job(job) except Exception as exc: raise CronSchedulerRegistrationError(job, exc) from exc return job # Dead-owner claim reclaim throttle (#86721): recover_interrupted_executions # opens the executions ledger, so the per-tick reap is rate-limited rather # than run on every idle 60s cycle. Tests may reset _last_dead_owner_reap_at # to None to force a reap on the next tick. _DEAD_OWNER_REAP_INTERVAL_SECONDS = 300.0 _last_dead_owner_reap_at: Optional[float] = None # Worktree maintenance throttle: the startup pruner historically ran only on # `hermes -w` launches, so on gateway-driven boxes (where sessions arrive via # Telegram/Discord and nobody launches the CLI for days) merged scratch trees # accumulated into tens of GB. The cron tick is the one reliably periodic # process on every install, so it owns a low-frequency sweep too. Tests may # reset _last_worktree_maintenance_at to None to force a sweep next tick. _WORKTREE_MAINTENANCE_INTERVAL_SECONDS = 6 * 3600.0 _last_worktree_maintenance_at: Optional[float] = None _worktree_maintenance_lock = threading.Lock() def _worktree_maintenance_repos() -> List[str]: """Repos whose ``.worktrees/`` this scheduler should keep pruned. Candidates: the hermes install checkout itself (where ``hermes -w`` sessions on dev boxes create trees) and every configured job workdir's repo root. Only repos that actually have a ``.worktrees/`` dir survive — everything else costs nothing. """ repos: set = set() # The hermes source checkout (editable/git installs). Wheel installs have # no .git here and are skipped. try: install_root = Path(__file__).resolve().parent.parent if (install_root / ".git").exists(): repos.add(str(install_root)) except Exception: pass # Job workdirs may point inside other repos the agent works on. try: from cron.jobs import load_jobs for job in load_jobs(): workdir = str(job.get("workdir") or "").strip() if not workdir or not Path(workdir).is_dir(): continue try: probe = subprocess.run( ["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, cwd=workdir, ) if probe.returncode == 0 and probe.stdout.strip(): repos.add(probe.stdout.strip()) except Exception: continue except Exception: pass return [r for r in sorted(repos) if (Path(r) / ".worktrees").is_dir()] def _maybe_run_worktree_maintenance() -> None: """Throttled, threaded worktree prune from the cron tick. Runs ``cli._prune_stale_worktrees`` (the same conservative pruner the ``hermes -w`` startup path uses — dirty/unpushed/live-locked trees are never touched) against every candidate repo, on a daemon thread so the tick itself never waits on git. Errors never propagate: worktree GC is hygiene, not scheduling. """ global _last_worktree_maintenance_at now = time.monotonic() with _worktree_maintenance_lock: if ( _last_worktree_maintenance_at is not None and now - _last_worktree_maintenance_at < _WORKTREE_MAINTENANCE_INTERVAL_SECONDS ): return _last_worktree_maintenance_at = now def _run() -> None: try: repos = _worktree_maintenance_repos() if not repos: return from cli import _prune_stale_worktrees for repo in repos: try: _prune_stale_worktrees(repo) except Exception: logger.debug( "Cron worktree maintenance failed for %s", repo, exc_info=True, ) except Exception: logger.debug("Cron worktree maintenance skipped", exc_info=True) threading.Thread( target=_run, name="cron-worktree-prune", daemon=True ).start() def tick( verbose: bool = True, adapters=None, loop=None, sync: bool = True, *, can_dispatch=None, ): """ Check and run all due jobs. Uses a file lock so only one tick runs at a time, even if the gateway's in-process ticker and a standalone daemon or manual tick overlap. Args: verbose: Whether to print status messages adapters: Optional dict mapping Platform → live adapter (from gateway) loop: Optional asyncio event loop (from gateway) for live adapter sends can_dispatch: Optional synchronous gate; false leaves due jobs untouched for the next allowed tick Returns: Number of jobs executed (0 if another tick is already running) """ # Stale-code yield gate — BEFORE the lock race (#stale-tick-preemption). # A long-lived process whose checkout was updated underneath it (hot # ``git pull``, interrupted ``hermes update``) serves MIXED sys.modules: # every agent job it dispatches can die on ImportErrors whose real cause # is staleness. When this process is provably stale AND a fresher # process holds the gateway runtime lock, that process's own ticker # dispatches due jobs — this one must not even enter the lock race and # preempt dispatch on a busy minute. When no fresh holder exists # (desktop-standalone users), yielding would silently kill the user's # only ticker, so the tick proceeds and job failures surface through the # delivery path's stale-code hint instead. _skew = _should_yield_tick_to_fresh_gateway() if _skew is not None: _log_tick_yield_once(f"boot={_skew[0]} disk={_skew[1]}") raise CronTickYielded(_skew[0], _skew[1]) lock_dir, lock_file = _get_lock_paths() _ensure_cron_dir(lock_dir) # Cross-platform file locking: fcntl on Unix, msvcrt on Windows. # Only genuine lock contention (another ticker holds the lock) skips the # tick silently. A real OSError — most importantly EMFILE/ENFILE from fd # exhaustion — must NOT be swallowed as "another instance holds the # lock": that previously made the scheduler appear healthy (tick returned # 0, heartbeat recorded success) while no job ever ran again (#87644). lock_fd = None try: lock_fd = open(lock_file, "w", encoding="utf-8") if fcntl: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) elif msvcrt: msvcrt.locking(lock_fd.fileno(), msvcrt.LK_NBLCK, 1) except OSError as exc: if lock_fd is not None and _is_lock_contention_errno(exc): logger.debug("Tick skipped — another instance holds the lock") try: lock_fd.close() except OSError: pass return 0 # Real failure: log loudly, attempt fd reclamation, and let the # caller (ticker loop) see a FAILED tick so liveness degrades # instead of reporting healthy-while-stalled. if lock_fd is not None: try: lock_fd.close() except OSError: pass if _is_fd_exhaustion(exc): # Reclamation is owned by the ticker loop's except handler # (scheduler_provider.py) — it classifies the raised error and # runs _reclaim_fds_best_effort exactly once per failed tick. # Calling it here too would double the gc.collect() pause. logger.error( "Cron tick could not acquire tick lock: %s — scheduler will " "attempt fd reclamation and retry with backoff", exc, ) else: logger.error("Cron tick could not acquire tick lock: %s", exc) raise try: # Global emergency stop (`hermes pause`): skip dispatch entirely while # the ESTOP sentinel exists. Never touches in-flight runs — due jobs # simply wait for the next tick after `hermes resume`. Logged once per # engagement (not every tick) by check_paused. try: from agent.estop import check_paused as _estop_check_paused if _estop_check_paused("cron", logger): return 0 except ImportError: pass if can_dispatch is not None and not can_dispatch(): logger.debug("Cron dispatch paused while gateway drains existing work") return 0 # Dead-owner claim reclaim (#86721): execution rows carry their owner # pid + process start time, but recovery previously ran only at # scheduler STARTUP. A one-shot `hermes cron run` that claimed a job # and died mid-run (its runner thread lived in the exiting CLI # process) left the row 'claimed' forever while the long-lived # gateway ticker kept running — blocking every future run of that # job. Reap provably-dead owners periodically so stale claims # auto-clear without a gateway restart. Only rows whose exact owner # process is proved gone are touched (see _owner_is_live), so live # runs in other processes are never rewritten. Throttled so idle # 60s ticks don't pay a ledger connection every cycle (#33612). global _last_dead_owner_reap_at _reap_now = time.monotonic() if ( _last_dead_owner_reap_at is None or _reap_now - _last_dead_owner_reap_at >= _DEAD_OWNER_REAP_INTERVAL_SECONDS ): _last_dead_owner_reap_at = _reap_now try: from cron.executions import recover_interrupted_executions _reclaimed = recover_interrupted_executions() if _reclaimed: logger.warning( "Reclaimed %d cron execution(s) whose owner process died " "before reaching a terminal state (marked unknown)", _reclaimed, ) except Exception as _reap_exc: logger.debug("Dead-owner execution reclaim failed: %s", _reap_exc) # Periodic worktree GC (throttled to every 6h, threaded): gateway-only # boxes never hit the `hermes -w` startup pruner, so this is the only # sweep they get. Same conservative pruner, same guards. try: _maybe_run_worktree_maintenance() except Exception as _wt_exc: logger.debug("Worktree maintenance dispatch failed: %s", _wt_exc) due_jobs = get_due_jobs() # Bound the in-flight set BEFORE the dedup guard is consulted, so a # leaked claim is force-released in-cycle rather than silently eating # every subsequent fire until the gateway process restarts. Skips the # extra load_jobs when there are no in-flight claims (the common idle # tick) and reuses due_jobs when they already cover the in-flight set # (get_due_jobs calls load_jobs internally, so this avoids a redundant # second file read on every active tick). if _running_job_ids: _sweep_jobs = due_jobs try: _inflight_ids = set(_running_job_ids) _due_ids = {j.get("id") for j in due_jobs if isinstance(j, dict)} if not _inflight_ids <= _due_ids: from cron.jobs import load_jobs as _load_all_jobs _sweep_jobs = _load_all_jobs() except Exception: pass try: sweep_stale_inflight(_sweep_jobs) except Exception as e: logger.warning("Stale in-flight sweep failed: %s", e) if not due_jobs: # Idle tick: skip config load + pool partitioning entirely # (#33612 — the gateway ticker calls tick(verbose=False) every # 60s, so idle ticks previously fell through to load_config()). # Still run the post-tick MCP orphan sweep: main intentionally # sweeps on idle ticks so orphaned stdio children from crashed # jobs are reaped even when nothing is due. if verbose: logger.info("%s - No jobs due", _hermes_now().strftime('%H:%M:%S')) try: from tools.mcp_tool import _kill_orphaned_mcp_children _kill_orphaned_mcp_children() except Exception as _e: logger.debug("Post-tick MCP orphan cleanup failed: %s", _e) return 0 if verbose: logger.info("%s - %s job(s) due", _hermes_now().strftime('%H:%M:%S'), len(due_jobs)) # Advance next_run_at for all recurring jobs FIRST, under the file lock, # before any execution begins. This preserves at-most-once semantics. # For parallel jobs that are already running, the advance keeps # bumping next_run_at forward so the grace window never expires. # mark_job_run() overwrites next_run_at on completion. # Batched: one load + one save for the whole due set, not one per job. # Composes with the claim-time advance in claim_job_for_fire: for # cron-kind jobs both compute the same next occurrence; interval jobs # re-anchor from their own "now" at claim time (harmless for # at-most-once — mark_job_run re-anchors at completion regardless). advance_next_runs([job["id"] for job in due_jobs]) # Resolve max parallel workers: env var > config.yaml > unbounded. # Set HERMES_CRON_MAX_PARALLEL=1 to restore old serial behaviour. _max_workers: Optional[int] = None try: _env_par = os.getenv("HERMES_CRON_MAX_PARALLEL", "").strip() if _env_par: _max_workers = int(_env_par) or None except (ValueError, TypeError): logger.warning("Invalid HERMES_CRON_MAX_PARALLEL value; defaulting to unbounded") if _max_workers is None: try: _ucfg = load_config() or {} _cfg_par = ( _ucfg.get("cron", {}) if isinstance(_ucfg, dict) else {} ).get("max_parallel_jobs") if _cfg_par is not None: _max_workers = int(_cfg_par) or None except Exception: pass if verbose: logger.info( "Running %d job(s) in parallel (max_workers=%s)", len(due_jobs), _max_workers if _max_workers else "unbounded", ) def _process_job(job: dict) -> bool: """Run one due job end-to-end. Thin wrapper around the shared module-level ``run_one_job`` so ``tick`` and external providers (Chronos ``fire_due``) use the identical execute→save→deliver→mark body.""" # Acquire the durable claim only when this worker actually starts, # not while it may wait behind other work in an executor queue. # This prevents a queued lease from expiring before execution. claimed = claim_job_for_fire(job["id"], return_job=True) if not claimed: finish_execution( job["execution_id"], success=False, error="Fire claim lost; execution was not started.", ) return True # Production CAS returns the exact persisted record with its unique # owner. Bool fallback keeps older test doubles/API overrides # compatible; real callers using return_job=True never take it. claimed_job = dict(claimed) if isinstance(claimed, dict) else dict(job) claimed_job["execution_id"] = job["execution_id"] return run_one_job( claimed_job, adapters=adapters, loop=loop, verbose=verbose, ) # Workdir is task-scoped, so every job uses the normal parallel lane. parallel_jobs = due_jobs _results: list = [] _all_futures: list = [] def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor): """Submit a job fire-and-forget with the in-flight dedup guard. Returns the future, or None if the job was skipped because a prior tick's run of the same job is still in flight. The running-set membership is released in the worker's finally block. """ job_id = job["id"] def _clear_run_claim_best_effort() -> None: """Best-effort claim cleanup on the dispatch-failure paths. Only one-shot jobs carry a ``run_claim`` (stamped by get_due_jobs, #59229), so recurring jobs skip the call entirely — clear_run_claim acquires _jobs_lock (blocking cross-process flock) and does a full load_jobs read, and the dispatch-failure paths fire exactly when the process can least afford N pointless lock/read round-trips (interpreter shutdown, EMFILE). clear_run_claim itself does load_jobs/save_jobs file I/O; on those degraded paths it can raise, and these early-exits exist precisely to skip cleanly — a stale claim expiring at the TTL is a better outcome than crashing the tick (#86522). """ _schedule = job.get("schedule") if not (isinstance(_schedule, dict) and _schedule.get("kind") == "once"): return try: clear_run_claim(job_id) except Exception as claim_err: logger.warning( "Could not clear run_claim for job '%s' after dispatch " "failure: %s (claim will expire at TTL)", job.get("name", job_id), claim_err, ) # A tick can race gateway teardown: once the interpreter is # finalizing, ``pool.submit`` raises "cannot schedule new futures # after interpreter shutdown" and crashes the tick. Skip cleanly — # the job stays due and will fire on the next healthy tick # (#58720, #55924). if _interpreter_shutting_down(): logger.warning( "Job '%s' not dispatched — interpreter is shutting down", job.get("name", job_id), ) _clear_run_claim_best_effort() return None if not try_register_running_job(job_id): logger.info("Job '%s' already running — skipping", job.get("name", job_id)) return None # Record the attempt before executor dispatch. Recovery classifies # abandoned records as unknown; it never automatically retries them. try: execution = create_execution(job_id, source="builtin") dispatched_job = dict(job, execution_id=execution["id"]) _ctx = contextvars.copy_context() except Exception as execution_err: # Init/creation failure between the claim and the submit — # release the in-flight claim immediately so the next tick can # retry instead of wedging on 'already running' forever (the # audit requirement: every add is paired with guaranteed # cleanup). release_running_job(job_id) _clear_run_claim_best_effort() logger.exception( "Job '%s' not dispatched: execution creation failed: %s", job.get("name", job_id), execution_err, ) return None def _run_and_release(j=dispatched_job, ctx=_ctx): try: return ctx.run(_process_job, j) finally: release_running_job(j["id"]) try: fut = pool.submit(_run_and_release) except Exception as submit_err: release_running_job(job_id) _clear_run_claim_best_effort() finish_execution( execution["id"], success=False, error=f"Executor dispatch failed: {submit_err}", ) # Interpreter began finalizing between the guard above and the # submit — release the in-flight claim we just took and skip. if isinstance(submit_err, RuntimeError) and _interpreter_shutting_down(submit_err): logger.warning( "Job '%s' not dispatched — interpreter is shutting down", job.get("name", job_id), ) return None logger.error( "Job '%s' not dispatched: %s", job.get("name", job_id), submit_err, ) return None # Record the owning future so the stale sweep can distinguish # "still executing" from "claim leaked before/after the future". with _running_lock: if job_id in _running_job_ids: _running_futures[job_id] = fut return fut # Parallel pass — persistent pool, non-blocking dispatch. # Jobs that are already running (from a previous tick) are skipped. # mark_job_run() updates next_run_at on completion, so the next tick # after completion finds the job due again naturally. No catch-up # queue needed. if parallel_jobs: pool = _get_parallel_pool(_max_workers) for job in parallel_jobs: fut = _submit_with_guard(job, pool) if fut is None: continue _all_futures.append(fut) if not sync: _results.append(True) # optimistically counted # Best-effort sweep of MCP stdio subprocesses that survived their # session teardown. Must run AFTER jobs finish so active sessions # (including live user chats) are never touched — only PIDs explicitly # detected as orphans in tools.mcp_tool._run_stdio's finally block are # reaped. def _sweep_mcp_orphans() -> None: try: from tools.mcp_tool import _kill_orphaned_mcp_children _kill_orphaned_mcp_children() except Exception as _e: logger.debug("Post-tick MCP orphan cleanup failed: %s", _e) if sync: # Sync mode (tests / manual ticks): wait for all dispatched jobs, # collect results, then sweep once. for f in concurrent.futures.as_completed(_all_futures): try: _results.append(f.result()) except Exception as exc: logger.error("Cron job future failed: %s", exc) _results.append(False) _sweep_mcp_orphans() return sum(_results) # Async (gateway ticker) mode: don't block. Sweep orphans via a # done-callback fired after the LAST dispatched job completes, so the # sweep still happens after jobs finish without stalling the tick. if _all_futures: _remaining = [len(_all_futures)] def _on_done(_f: concurrent.futures.Future) -> None: _remaining[0] -= 1 try: _exc = _f.exception() if _exc is not None: logger.error("Cron job future failed in async mode: %s", _exc, exc_info=(type(_exc), _exc, _exc.__traceback__)) except Exception: pass if _remaining[0] <= 0: _sweep_mcp_orphans() for _f in _all_futures: _f.add_done_callback(_on_done) else: # Nothing dispatched (all skipped / no due jobs) — sweep inline. _sweep_mcp_orphans() return sum(_results) finally: if fcntl: try: fcntl.flock(lock_fd, fcntl.LOCK_UN) except (OSError, IOError): pass elif msvcrt: try: msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1) except (OSError, IOError): pass lock_fd.close() if __name__ == "__main__": if "--external-worker-file" in sys.argv: import argparse parser = argparse.ArgumentParser(add_help=False) parser.add_argument("--external-worker-file", type=Path, required=True) parser.add_argument("--ack-file", type=Path, required=True) args = parser.parse_args() # The gateway spawns this worker with stdout/stderr on DEVNULL; without # a handler every adoption/ack failure below would be invisible. try: from hermes_logging import setup_logging setup_logging(hermes_home=_get_hermes_home(), mode="cron") except Exception: pass raise SystemExit( 0 if _run_external_worker_payload(args.external_worker_file, args.ack_file) else 1 ) tick(verbose=True)