diff --git a/gateway/run.py b/gateway/run.py index e87739c14f01..59127859e9ad 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -44,7 +44,7 @@ from contextvars import copy_context from pathlib import Path from datetime import datetime -from typing import Callable, Dict, Optional, Any, List, Union +from typing import Dict, Optional, Any, List, Union # account_usage imports the OpenAI SDK chain (~230 ms). Only needed by # /usage; we still import it at module top in the gateway because test @@ -77,6 +77,7 @@ r"|configured\s+compression\s+model\s+.+\s+failed" r"|no\s+auxiliary\s+llm\s+provider\s+configured" r"|auto-lowered\s+compression\s+threshold" + r"|codex\s+gpt-5\.5\s+caps\s+context\s+at\s+272k.*auto-compaction\s+was\s+raised" r"|compacting\s+context\s+[—-]\s+summarizing\s+earlier\s+conversation" r"|preflight\s+compression" r"|session\s+compressed\s+\d+\s+times" @@ -147,7 +148,6 @@ def _gateway_surface_passes_raw_text(platform: Any) -> bool: _GATEWAY_SECRET_PATTERNS = ( re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"), re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), - re.compile(r"\bxapp-\d+-[A-Za-z0-9\-]{20,}\b"), re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{20,}\b"), re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"), re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"), @@ -618,19 +618,20 @@ def _coerce_gateway_timestamp(value: Any) -> Optional[float]: def _auto_continue_freshness_window() -> float: """Return the configured auto-continue freshness window in seconds. - Thin wrapper that delegates to the canonical implementation in - ``gateway.session`` (the single source of truth shared with the - routing-time zombie gate in ``get_or_create_session``). Reads - ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from ``config.yaml`` - ``agent.gateway_auto_continue_freshness`` at gateway startup, same - pattern as ``HERMES_AGENT_TIMEOUT``). Falls back to the module default - when unset or malformed. Non-positive values disable the freshness gate - (restores the pre-fix "always fresh" behaviour for users who want to opt - out). Kept here so existing call sites and test patches importing it - from ``gateway.run`` continue to work. + Reads ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from + ``config.yaml`` ``agent.gateway_auto_continue_freshness`` at gateway + startup, same pattern as ``HERMES_AGENT_TIMEOUT``). Falls back to the + module default when unset or malformed. Non-positive values disable + the freshness gate (restores the pre-fix "always fresh" behaviour for + users who want to opt out). """ - from gateway.session import auto_continue_freshness_window - return auto_continue_freshness_window() + raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS") + if raw is None or raw == "": + return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) + try: + return float(raw) + except (TypeError, ValueError): + return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) def _float_env(name: str, default: float) -> float: @@ -1676,7 +1677,7 @@ def _profile_runtime_scope(profile_home: "Path"): build_session_key, is_shared_multi_user_session, ) -from gateway.delivery import DeliveryRouter, looks_like_telegram_private_chat_id +from gateway.delivery import DeliveryRouter from gateway.authz_mixin import GatewayAuthorizationMixin from gateway.kanban_watchers import GatewayKanbanWatchersMixin from gateway.slash_commands import GatewaySlashCommandsMixin @@ -1706,48 +1707,6 @@ def _profile_runtime_scope(profile_home: "Path"): logger = logging.getLogger(__name__) -_OWN_POLICY_OPEN_ENV = { - Platform.WECOM: ("WECOM_DM_POLICY", "WECOM_GROUP_POLICY", "WECOM_ALLOW_ALL_USERS"), - Platform.WEIXIN: ("WEIXIN_DM_POLICY", "WEIXIN_GROUP_POLICY", "WEIXIN_ALLOW_ALL_USERS"), - Platform.YUANBAO: ("YUANBAO_DM_POLICY", "YUANBAO_GROUP_POLICY", "YUANBAO_ALLOW_ALL_USERS"), - Platform.QQBOT: (None, None, "QQ_ALLOW_ALL_USERS"), - Platform.WHATSAPP: ("WHATSAPP_DM_POLICY", "WHATSAPP_GROUP_POLICY", "WHATSAPP_ALLOW_ALL_USERS"), -} - - -def _own_policy_open_startup_violation(config) -> Optional[str]: - """Return a startup-abort reason when open policy lacks allow-all opt-in.""" - for platform, platform_config in getattr(config, "platforms", {}).items(): - if not getattr(platform_config, "enabled", False): - continue - open_env = _OWN_POLICY_OPEN_ENV.get(platform) - if not open_env: - continue - dm_env, group_env, allow_all_env = open_env - extra = getattr(platform_config, "extra", None) or {} - dm_policy = str( - extra.get("dm_policy") - or (os.getenv(dm_env, "pairing") if dm_env else "pairing") - ).strip().lower() - group_policy = str( - extra.get("group_policy") - or (os.getenv(group_env, "pairing") if group_env else "pairing") - ).strip().lower() - if dm_policy != "open" and group_policy != "open": - continue - gateway_allow_all = os.getenv( - "GATEWAY_ALLOW_ALL_USERS", "" - ).lower() in {"true", "1", "yes"} - platform_opted_in = gateway_allow_all or ( - allow_all_env - and os.getenv(allow_all_env, "").lower() in {"true", "1", "yes"} - ) - if platform_opted_in: - continue - return f"{platform.value}: open policy without allow-all opt-in" - return None - - # Sentinel placed into _running_agents immediately when a session starts # processing, *before* any await. Prevents a second message for the same # session from bypassing the "already running" guard during the async gap @@ -2695,16 +2654,6 @@ def __init__(self, config: Optional[GatewayConfig] = None): self._restart_via_service = False self._detached_restart_helper_started = False self._restart_command_source: Optional[SessionSource] = None - # Monotonic-ish wall clock of when this GatewayRunner was constructed. - # Used by the /restart redelivery guard to bound the window in which a - # missing dedup marker is treated as a stale redelivery. - self._startup_time: float = time.time() - # Set True at startup when this process booted as the result of a - # chat-originated /restart (i.e. .restart_notify.json existed on boot). - # A one-shot signal consumed by _is_stale_restart_redelivery so the - # marker-missing fallback only suppresses a /restart when we KNOW we - # just came out of a restart cycle — never on a genuine fresh boot. - self._booted_from_restart: bool = False self._stop_task: Optional[asyncio.Task] = None self._restart_task: Optional[asyncio.Task] = None self._executor_lock = threading.Lock() @@ -3690,77 +3639,12 @@ def _resolve_turn_agent_config(self, user_message: str, model: str, runtime_kwar route["request_overrides"] = overrides or {} return route - def _sync_session_model_from_agent(self, session_id: str, agent: Any) -> None: - """Persist the runtime model/provider actually used by a gateway turn. - - Provider fallback can switch ``agent.model``/``agent.provider`` after the - session row was created. Keep the session DB metadata in sync so session - lists, desktop/dashboard details, and follow-up session tooling report the - backend that actually answered the latest turn. - - Called from the ``run_sync`` closure, which executes off the event loop - in the executor thread — so the synchronous ``SessionDB`` (``_db``) is - used directly rather than awaiting the AsyncSessionDB forwarder. - """ - if not session_id or agent is None or self._session_db is None: - return - model = getattr(agent, "model", None) - if not model: - return - runtime = { - "provider": getattr(agent, "provider", None), - "base_url": getattr(agent, "base_url", None), - "api_mode": getattr(agent, "api_mode", None), - "fallback_active": bool(getattr(agent, "_fallback_activated", False)), - } - runtime = {k: v for k, v in runtime.items() if v not in (None, "")} - - try: - db = self._session_db._db - row = db.get_session(session_id) - if not row: - return - current_model = row.get("model") - raw_config = row.get("model_config") - try: - config = json.loads(raw_config) if raw_config else {} - except Exception: - config = {} - if not isinstance(config, dict): - config = {} - gateway_runtime = dict(config.get("gateway_runtime") or {}) - if current_model == model and all( - gateway_runtime.get(k) == v for k, v in runtime.items() - ): - return - config["gateway_runtime"] = runtime - db.update_session_meta(session_id, json.dumps(config), model=model) - except Exception: - logger.debug("Failed to sync gateway session model metadata", exc_info=True) - async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> None: """React to an adapter failure after startup. If the error is retryable (e.g. network blip, DNS failure), queue the platform for background reconnection instead of giving up permanently. """ - # Snapshot the current owner of this platform slot before doing - # anything else. If it's neither this adapter nor empty, a different - # adapter has already taken over (e.g. this is a delayed notification - # from a background retry chain that raced with, and lost to, a - # reconnect that already succeeded). Acting on a stale notification - # would overwrite an already-healthy platform's runtime status and - # incorrectly re-queue it for reconnection, so bail out before any of - # that happens. - existing = self.adapters.get(adapter.platform) - if existing is not None and existing is not adapter: - logger.debug( - "Ignoring stale fatal error from a superseded %s adapter instance: %s", - adapter.platform.value, - adapter.fatal_error_code or "unknown", - ) - return - logger.error( "Fatal %s adapter error (%s): %s", adapter.platform.value, @@ -3784,15 +3668,13 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non error_message=adapter.fatal_error_message, ) + existing = self.adapters.get(adapter.platform) if existing is adapter: - # Claim this adapter for teardown before awaiting disconnect() — - # a second fatal-error notification for the same adapter (e.g. - # from a concurrent recovery path) would otherwise still see - # itself as "existing" during the await below and disconnect() - # the same object twice. - self.adapters.pop(adapter.platform, None) - self.delivery_router.adapters = self.adapters - await adapter.disconnect() + try: + await adapter.disconnect() + finally: + self.adapters.pop(adapter.platform, None) + self.delivery_router.adapters = self.adapters # Queue retryable failures for background reconnection if adapter.fatal_error_retryable: @@ -3890,29 +3772,6 @@ def _scale_to_zero_idle_timeout_seconds(self) -> float: raw = None return parse_idle_timeout_seconds(raw) - def _restart_loop_guard_config(self) -> tuple: - """Return ``(max_restarts, window_seconds)`` for the auto-resume - restart-loop breaker (#30719, defense-3), read from - ``gateway.restart_loop_guard`` in config.yaml with the module defaults - as fallback. ``max_restarts <= 0`` disables the breaker. - """ - from gateway import restart_loop_guard as _rlg - - max_restarts = _rlg.DEFAULT_MAX_RESTARTS - window_seconds = _rlg.DEFAULT_WINDOW_SECONDS - try: - user_cfg = _load_gateway_config() - gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None - rlg = gw.get("restart_loop_guard") if isinstance(gw, dict) else None - if isinstance(rlg, dict): - if isinstance(rlg.get("max_restarts"), int): - max_restarts = rlg["max_restarts"] - if isinstance(rlg.get("window_seconds"), int) and rlg["window_seconds"] > 0: - window_seconds = rlg["window_seconds"] - except Exception: # noqa: BLE001 - pass - return max_restarts, window_seconds - def _scale_to_zero_should_arm(self) -> bool: """Whether to start the idle watcher (D1/D11/§3.4(1)).""" from gateway.relay import relay_wake_url @@ -4794,36 +4653,6 @@ def _agent_has_active_subagents(running_agent: Any) -> bool: except Exception: return False - def _session_has_compression_in_flight(self, session_key: str) -> bool: - """Return True when a compression lock is held for this session's id. - - Context compression is interrupt-protected (#23975) but gateway - ``interrupt`` busy-input mode can still start a follow-up turn against - the pre-rotation parent while compression is mid-flight, producing - orphaned compression siblings (#56391). Callers demote interrupt to - queue when this returns True. - """ - session_store = getattr(self, "session_store", None) - if not session_key or session_store is None: - return False - try: - with session_store._lock: # noqa: SLF001 — snapshot entry under lock - session_store._ensure_loaded_locked() # noqa: SLF001 - entry = session_store._entries.get(session_key) # noqa: SLF001 - session_id = getattr(entry, "session_id", None) if entry is not None else None - if not session_id: - return False - except Exception: - return False - session_db = getattr(self, "_session_db", None) - if session_db is None: - return False - db = getattr(session_db, "_db", session_db) - try: - return bool(db.get_compression_lock_holder(str(session_id))) - except Exception: - return False - # Hard cap on per-session pending follow-ups for busy_input_mode=queue # (and the draining/steer-fallback/subagent-demotion paths that share # this entry point). Without a cap, a stuck agent + a rapid-fire user @@ -4833,7 +4662,7 @@ def _session_has_compression_in_flight(self, session_key: str) -> bool: _BUSY_QUEUE_MAX_PENDING = 32 def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None: - adapter = self._adapter_for_source(event.source) + adapter = self.adapters.get(event.source.platform) if not adapter: return # #28503 — Previously this called ``merge_pending_message_event`` @@ -4890,7 +4719,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session # --- Draining case (gateway restarting/stopping) --- if self._draining: - adapter = self._adapter_for_source(event.source) + adapter = self.adapters.get(event.source.platform) if not adapter: return True @@ -4992,7 +4821,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session ) # Normal busy case (agent actively running a task) - adapter = self._adapter_for_source(event.source) + adapter = self.adapters.get(event.source.platform) if not adapter: return False # let default path handle it @@ -5044,17 +4873,6 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session session_key, ) effective_mode = "queue" - demoted_for_compression = ( - effective_mode == "interrupt" - and self._session_has_compression_in_flight(session_key) - ) - if demoted_for_compression: - logger.info( - "Demoting busy_input_mode 'interrupt' to 'queue' for session %s " - "because context compression is in flight (#56391)", - session_key, - ) - effective_mode = "queue" steered = False if effective_mode == "steer": steer_text = (event.text or "").strip() @@ -5168,11 +4986,6 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session f"⏳ Subagent working{status_detail} — your message is queued for " f"when it finishes (use /stop to cancel everything)." ) - elif is_queue_mode and demoted_for_compression: - message = ( - f"⏳ Compressing context{status_detail} — your message is queued for " - f"when it finishes (use /stop to cancel everything)." - ) elif is_queue_mode: message = ( f"⏳ Queued for the next turn{status_detail}. " @@ -5905,51 +5718,34 @@ def _launch_systemd_restart_shortcut(self) -> None: service_name = "hermes-gateway" current_pid = os.getpid() - - # Detect whether the gateway unit is registered as a system or - # user service. Daemon-style deployments are typically system - # units (e.g. /etc/systemd/system/hermes-gateway.service), while - # `hermes setup` under a non-root account may register a user - # unit. Hard-coding ``--user`` broke system-unit deployments: - # systemctl returned an empty MainPID, the PID-equality check - # below failed, and the planned-restart helper was never - # launched — leaving the gateway dead until a manual reboot. - def _query_pid(scope_flags): - try: - out = subprocess.run( - [systemctl, *scope_flags, "show", service_name, - "--property=MainPID", "--value"], - capture_output=True, text=True, timeout=2, - ) - return (out.stdout or "").strip() - except Exception: - return "" - - system_pid = _query_pid([]) - user_pid = _query_pid(["--user"]) - if str(current_pid) == system_pid: - scope_flags = [] - systemctl_scope = "systemctl" - elif str(current_pid) == user_pid: - scope_flags = ["--user"] - systemctl_scope = "systemctl --user" - else: - # MainPID does not match in either scope — likely invoked - # outside of systemd or the unit was renamed. Bail out - # rather than restart the wrong unit. + show = subprocess.run( + [ + systemctl, + "--user", + "show", + service_name, + "--property=MainPID", + "--value", + ], + capture_output=True, + text=True, + timeout=2, + ) + if (show.stdout or "").strip() != str(current_pid): return + systemctl_user = "systemctl --user" service_arg = shlex.quote(service_name) shell_cmd = ( f"while kill -0 {current_pid} 2>/dev/null; do sleep 0.2; done; " - f"{systemctl_scope} reset-failed {service_arg}; " - f"{systemctl_scope} restart {service_arg}" + f"{systemctl_user} reset-failed {service_arg}; " + f"{systemctl_user} restart {service_arg}" ) unit_name = f"{service_name}-planned-restart-{current_pid}".replace(".", "-") subprocess.Popen( [ systemd_run, - *scope_flags, + "--user", "--collect", "--unit", unit_name, @@ -5962,10 +5758,9 @@ def _query_pid(scope_flags): start_new_session=True, ) logger.info( - "Launched systemd planned-restart helper for %s (pid=%s, scope=%s)", + "Launched systemd planned-restart helper for %s (pid=%s)", service_name, current_pid, - "user" if scope_flags else "system", ) except Exception as e: logger.debug("Failed to launch systemd planned-restart helper: %s", e) @@ -6138,26 +5933,6 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int: logger.warning("Failed to enumerate resume-pending sessions: %s", exc) return 0 - # Defense-3 (#30719): break the SIGTERM-respawn loop. Only count this - # boot when there are restart-interrupted sessions to resume — a clean - # boot must not accrue toward the breaker. If too many such boots have - # happened in the configured window, skip auto-resume for THIS boot: - # the gateway still comes up and serves real inbound messages, it just - # stops replaying the session that keeps killing it. The session stays - # resume_pending, so a real user message can still continue it (a human - # is now in the loop). Defenses 1-2 cover the cron/CLI/terminal paths; - # this catches every other SIGTERM source (e.g. a raw `terminal( - # "launchctl kickstart ai.hermes.gateway")`). - if candidates: - try: - from gateway import restart_loop_guard as _rlg - - _max_restarts, _window = self._restart_loop_guard_config() - if _rlg.check_and_record(_max_restarts, _window): - return 0 - except Exception as exc: # noqa: BLE001 — breaker must fail OPEN - logger.debug("Restart-loop guard check skipped: %s", exc) - now = datetime.now() scheduled = 0 for entry in candidates: @@ -6180,27 +5955,6 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int: ) continue - # Validate the session owner against the current allowlist - # before auto-resuming. A session created before - # TELEGRAM_ALLOWED_USERS (or equivalent) was configured, or - # before the owner was removed from it, must not silently - # receive a full agent response on gateway restart just - # because it has a resume-pending marker (issue #23778). - try: - if not self._is_user_authorized(source): - logger.warning( - "Skipping auto-resume for %s: session owner is no " - "longer authorized under the current allowlist", - entry.session_key, - ) - continue - except Exception as exc: - logger.warning( - "Skipping auto-resume for %s: authorization check failed: %s", - entry.session_key, exc, - ) - continue - # Claim the session slot *before* spawning the task so that an # inbound message arriving between task creation and the task's # first await (where _process_message_background sets the real @@ -6440,34 +6194,10 @@ async def start(self) -> bool: ) if not _any_allowlist and not _allow_all: logger.warning( - "No env user allowlists configured. Messaging platforms default to " - "pairing/allowlist policies and will deny unknown senders unless you " - "configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id) " - "or explicitly opt in with GATEWAY_ALLOW_ALL_USERS=true plus " - "dm_policy/group_policy: open on the platform." - ) - - reason = _own_policy_open_startup_violation(self.config) - if reason: - platform_value = reason.split(":", 1)[0] - allow_all_env = None - for platform, open_env in _OWN_POLICY_OPEN_ENV.items(): - if platform.value == platform_value: - allow_all_env = open_env[2] - break - logger.error( - "Refusing to start: %s has dm_policy/group_policy set to 'open' " - "but neither GATEWAY_ALLOW_ALL_USERS nor %s is enabled.", - platform_value, - allow_all_env or "a platform allow-all flag", + "No user allowlists configured. All unauthorized users will be denied. " + "Set GATEWAY_ALLOW_ALL_USERS=true in ~/.hermes/.env to allow open access, " + "or configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id)." ) - try: - from gateway.status import write_runtime_status - write_runtime_status(gateway_state="startup_failed", exit_reason=reason) - except Exception: - pass - self._request_clean_exit(reason) - return True # Discover Python plugins before shell hooks so plugin block # decisions take precedence in tie cases. The CLI startup path @@ -6625,7 +6355,6 @@ async def start(self) -> bool: adapter.set_session_store(self.session_store) adapter.set_busy_session_handler(self._handle_active_session_busy_message) adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id) - adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform)) adapter._busy_text_mode = self._busy_text_mode # Try to connect @@ -6850,13 +6579,6 @@ async def start(self) -> bool: # Notify the chat that initiated /restart that the gateway is back. planned_restart_notification_pending = _planned_restart_notification_pending() - # Capture, before _send_restart_notification() unlinks the marker, - # whether this process booted from a chat-originated /restart. Used as - # a one-shot signal by the /restart redelivery guard so a missing - # dedup marker only suppresses a /restart when we KNOW we just came out - # of a restart cycle (see _is_stale_restart_redelivery). - if _restart_notification_pending() or planned_restart_notification_pending: - self._booted_from_restart = True await self._send_restart_notification() # Broadcast a lightweight "gateway is back" message to configured home @@ -7075,37 +6797,26 @@ async def _process_handoff(self, row: Dict[str, Any]) -> None: str(home.thread_id) if home.thread_id else None ) - # Determine chat_type/user_id for the destination source. - # - # Telegram private-chat DM topics are represented differently from - # group/forum threads by the inbound adapter. A handoff-created topic - # in a positive Telegram chat_id must therefore use the same DM-topic - # source shape as the user's next real message; otherwise the synthetic - # handoff turn binds a generic `thread` session key while real replies - # arrive on a `dm` session key. - home_chat_id = str(home.chat_id) - is_telegram_private_chat = ( - platform == Platform.TELEGRAM - and looks_like_telegram_private_chat_id(home_chat_id) - ) - - if new_thread_id and not is_telegram_private_chat: + # Determine chat_type for the destination source. If we created a + # thread, key the session_key as a thread (build_session_key sets + # thread sessions to user-shared by default, which is what we + # want — the synthetic turn and any later real-user message both + # land on the same key without needing a user_id). + if new_thread_id: dest_chat_type = "thread" - dest_user_id = "system:handoff" else: - # No thread — assume DM-style for the home channel. For Telegram - # private-chat topics, use the real user id (same as chat_id) so - # topic-mode checks and binding persistence see the same identity as - # subsequent inbound user messages. + # No thread — assume DM-style for the home channel. For + # group/channel home channels without thread support + # (Matrix/WhatsApp/Signal), the platform's own keying makes + # the synthetic turn shared anyway (single-DM platforms). dest_chat_type = "dm" - dest_user_id = home_chat_id if is_telegram_private_chat else "system:handoff" dest_source = SessionSource( platform=platform, - chat_id=home_chat_id, + chat_id=str(home.chat_id), chat_name=home.name, chat_type=dest_chat_type, - user_id=dest_user_id, + user_id="system:handoff", user_name="Handoff", thread_id=effective_thread_id, ) @@ -7452,7 +7163,6 @@ async def _platform_reconnect_watcher(self) -> None: adapter.set_session_store(self.session_store) adapter.set_busy_session_handler(self._handle_active_session_busy_message) adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id) - adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform)) adapter._busy_text_mode = self._busy_text_mode # Reconnect after an outage: preserve the platform's @@ -7935,25 +7645,17 @@ def _phase_elapsed() -> float: if self._restart_requested and self._restart_via_service: self._launch_systemd_restart_shortcut() - # Always exit with TEMPFAIL (75) on service-managed - # restarts. The shortcut helper above is best-effort and - # commonly fails on real deployments: non-root gateway - # units hit Polkit denials when invoking ``systemd-run - # --system``, headless boxes have no user bus for - # ``--user``, and operator-managed unit files may use - # ``Restart=on-failure`` rather than ``Restart=always``. - # Exit 75 paired with ``RestartForceExitStatus=75`` makes - # systemd treat the planned restart as a controlled - # failure and revive the unit via ``Restart=on-failure``, - # regardless of whether the helper survived. Without - # this, a clean exit (0) on Linux left the gateway dead - # until someone rebooted the host. Only the planned code - # (75) is whitelisted via ``RestartForceExitStatus``; a - # genuine crash exits non-zero-but-not-75, so real crash - # loops are still governed by the unit's normal - # ``Restart=``/``RestartSec`` (and any StartLimit the - # operator sets) rather than force-restarted here. - self._exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE + # systemd units use Restart=always, so a planned restart should + # exit cleanly and still be relaunched. Using TEMPFAIL here + # makes systemd treat the operator-requested restart as a + # failure and can trip stepped restart backoff. launchd's + # KeepAlive.SuccessfulExit=false needs a non-zero exit to + # relaunch, so keep the old code on macOS. + self._exit_code = ( + GATEWAY_SERVICE_RESTART_EXIT_CODE + if sys.platform == "darwin" or not os.environ.get("INVOCATION_ID") + else 0 + ) self._exit_reason = self._exit_reason or "Gateway restart requested" self._draining = False @@ -8066,14 +7768,6 @@ async def _start_one_profile_adapters( with _profile_runtime_scope(profile_home): profile_cfg = load_gateway_config() - violation = _own_policy_open_startup_violation(profile_cfg) - if violation: - raise MultiplexConfigError( - f"Profile '{profile_name}' enables {violation}. " - "Enable GATEWAY_ALLOW_ALL_USERS or the platform allow-all flag " - "for that profile, or change dm_policy/group_policy away from " - "'open'." - ) profile_map = self._profile_adapters.setdefault(profile_name, {}) connected = 0 @@ -8125,7 +7819,6 @@ async def _start_one_profile_adapters( adapter.set_session_store(self.session_store) adapter.set_busy_session_handler(self._handle_active_session_busy_message) adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id) - adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform)) adapter._busy_text_mode = self._busy_text_mode try: @@ -8294,39 +7987,6 @@ def _create_adapter( return None - def _make_adapter_auth_check( - self, - platform: Platform, - ) -> Callable[[str, Optional[str], Optional[str]], bool]: - """Build a platform-bound auth callback for adapter use. - - Adapters that fetch external context (e.g. Slack - ``conversations.replies``) call this through - ``BasePlatformAdapter._is_sender_authorized`` to mark non-allowlisted - senders as unverified in LLM context, mitigating indirect prompt - injection from third parties in shared threads/channels. - - The returned callback delegates to :meth:`_is_user_authorized` so the - full auth chain — platform allowlists, group allowlists, pairing - store, allow-all flags — stays the single source of truth. - """ - def check( - user_id: str, - chat_type: Optional[str] = None, - chat_id: Optional[str] = None, - ) -> bool: - if not user_id: - return False - source = SessionSource( - platform=platform, - chat_id=chat_id or "", - chat_type=chat_type or "group", - user_id=user_id, - ) - return self._is_user_authorized(source) - return check - - @@ -8377,23 +8037,6 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: """ source = event.source - # 🔴 Cross-session leak guard. This handler runs inside a per-message - # asyncio task created via create_task(), which snapshots the spawning - # context with copy_context(). If a *concurrent* message had already - # bound its session via set_session_vars() when this task was created, - # we inherited ITS HERMES_SESSION_* ContextVars. Until we bind our own - # (a few steps down, in _set_session_env), any subprocess spawned here - # would read the foreign session's identity via the subprocess-env - # bridge — the _UNSET-strip guard there can't help because the vars are - # set-to-foreign, not _UNSET. Reset to _UNSET now so that window strips - # safe (no session) instead of leaking the sibling's. See - # gateway/session_context.reset_session_vars + the inheritance test. - try: - from gateway.session_context import reset_session_vars - reset_session_vars() - except Exception: - logger.debug("reset_session_vars failed at handler entry", exc_info=True) - if ( getattr(self, "_startup_restore_in_progress", False) and not getattr(event, "internal", False) @@ -8470,14 +8113,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: elif not self._is_user_authorized(source): logger.warning("Unauthorized user: %s (%s) on %s", source.user_id, source.user_name, source.platform.value) # In DMs: offer pairing code. In groups: silently ignore. - if ( - source.chat_type == "dm" - and self._get_unauthorized_dm_behavior( - source.platform, - profile=source.profile, - ) - == "pair" - ): + if source.chat_type == "dm" and self._get_unauthorized_dm_behavior(source.platform) == "pair": platform_name = source.platform.value if source.platform else "unknown" # Rate-limit ALL pairing responses (code or rejection) to # prevent spamming the user with repeated messages when @@ -8488,7 +8124,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: platform_name, source.user_id, source.user_name or "" ) if code: - adapter = self._adapter_for_source(source) + adapter = self.adapters.get(source.platform) if adapter: await adapter.send( source.chat_id, @@ -8498,7 +8134,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: f"`hermes pairing approve {platform_name} {code}`" ) else: - adapter = self._adapter_for_source(source) + adapter = self.adapters.get(source.platform) if adapter: await adapter.send( source.chat_id, @@ -10044,7 +9680,7 @@ async def _prepare_inbound_message_text( if "@" in message_text: try: from agent.context_references import preprocess_context_references_async - from agent.model_metadata import get_model_context_length_async + from agent.model_metadata import get_model_context_length _msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~")) _msg_runtime = _resolve_runtime_agent_kwargs() @@ -10058,7 +9694,7 @@ async def _prepare_inbound_message_text( _msg_config_ctx = int(_msg_raw_ctx) except Exception: pass - _msg_ctx_len = await get_model_context_length_async( + _msg_ctx_len = get_model_context_length( self._model, base_url=self._base_url or _msg_runtime.get("base_url") or "", api_key=_msg_runtime.get("api_key") or "", @@ -10396,7 +10032,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g if history and len(history) >= 4: from agent.model_metadata import ( estimate_messages_tokens_rough, - get_model_context_length_async, + get_model_context_length, ) # Read model + compression config from config.yaml. @@ -10497,7 +10133,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g pass if _hyg_compression_enabled: - _hyg_context_length = await get_model_context_length_async( + _hyg_context_length = get_model_context_length( _hyg_model, base_url=_hyg_base_url or "", api_key=_hyg_api_key or "", @@ -10961,12 +10597,19 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g _response_time, _api_calls, _resp_len, ) - # NOTE: the cross-process cache-coherence re-baseline - # (_refresh_agent_cache_message_count) is intentionally deferred - # until AFTER this turn's transcript persistence block below — it - # must include the first-turn `session_meta` marker row and the - # compression session_id swap, both of which happen later. See - # the call site after the `update_session(...)` write. + # Re-baseline the cached agent's message_count snapshot now that + # this turn has completed and the agent has flushed its rows to + # the SessionDB. The cross-process coherence guard (#45966) + # snapshots the count at agent-BUILD time (before this turn's own + # writes) and never refreshes it on reuse — so without this, this + # process's own turn would grow the count and the next turn would + # see a mismatch and rebuild the agent every turn, destroying + # prompt caching. Refreshing here makes the guard fire only on a + # DIFFERENT process's writes. Uses the (possibly compaction- + # updated) live session_id. Fail-safe inside the helper. + await self._refresh_agent_cache_message_count( + session_key, session_entry.session_id + ) # Successful turn — clear any stuck-loop counter for this session. # This ensures the counter only accumulates across CONSECUTIVE @@ -11237,15 +10880,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # The agent already persisted these messages to SQLite via # _flush_messages_to_session_db(), so skip the DB write here - # to prevent the duplicate-write bug (#860 / #42039). This holds - # for the codex app-server runtime too: although it early-returns - # and bypasses conversation_loop's per-step flushes, it flushes its - # own projected assistant/tool messages before returning and - # reports agent_persisted=True (see agent/codex_runtime.py). Reading - # the flag (default = self._session_db is not None) keeps the - # persistence contract explicit and lets any future non-persisting - # runtime opt into a gateway-side write by returning False. - agent_persisted = agent_result.get("agent_persisted", self._session_db is not None) + # to prevent the duplicate-write bug (#860 / #42039). + agent_persisted = self._session_db is not None # Find only the NEW messages from this turn (skip history we loaded). # Use the filtered history length (history_offset) that was actually @@ -11359,28 +10995,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), ) - # Re-baseline the cached agent's message_count snapshot now that - # ALL of this turn's transcript writes are done — the agent's - # flushed user/assistant/tool rows AND the first-turn `session_meta` - # marker appended above. The cross-process coherence guard (#45966) - # snapshots the count at agent-BUILD time (before this turn's own - # writes) and never refreshes it on reuse, so without this the - # process's own turn grows message_count and the next turn sees a - # mismatch and rebuilds the agent — destroying prompt caching. - # - # This MUST run after the `session_meta` append: that row also - # increments message_count, so re-baselining before it (the old - # position) left the snapshot one short and the guard mis-fired on - # turn 2 of EVERY fresh gateway conversation, rebuilding the cached - # agent and busting the prompt cache. Running here also uses the - # compaction-updated session_id (the agent_result session_id swap - # above), matching this function's documented contract. Refreshing - # here makes the guard fire only on a DIFFERENT process's writes. - # Fail-safe inside the helper. - await self._refresh_agent_cache_message_count( - session_key, session_entry.session_id - ) - # Intentional silence is a delivery decision, not a transcript # mutation. The agent's [SILENT]/NO_REPLY assistant turn above is # still persisted in session history so later turns keep normal @@ -11782,25 +11396,6 @@ def _is_stale_restart_redelivery(self, event: MessageEvent) -> bool: try: marker_path = _hermes_home / ".restart_last_processed.json" if not marker_path.exists(): - # Belt-and-suspenders for when the dedup marker goes missing - # (manually cleaned up, or the previous cycle's write failed). - # Without a marker the update_id comparison below can't run, so - # a redelivered /restart would sail through and re-restart the - # gateway — an infinite loop (issue #18528). - # - # Suppress ONLY when we can independently confirm we just came - # out of a restart cycle: this process booted from a - # chat-originated /restart (_booted_from_restart) AND is still - # within a short post-boot window. This never swallows a - # genuine first /restart on a fresh boot (no restart marker on - # boot → flag stays False). Consume the flag one-shot so a - # legitimate /restart sent later in the same session is honored. - if ( - getattr(self, "_booted_from_restart", False) - and time.time() - getattr(self, "_startup_time", 0.0) < 60 - ): - self._booted_from_restart = False - return True return False data = json.loads(marker_path.read_text()) except Exception: @@ -15206,15 +14801,6 @@ async def _refresh_agent_cache_message_count( only when the same agent is still cached (no rebuild/eviction raced in between). Fail-safe: any DB error leaves the snapshot as-is, which at worst costs one unnecessary rebuild on the next turn. - - When the cache entry records a ``session_id`` (4-tuple form, #54947) - that differs from the current ``session_id`` — meaning the cache - was built for a DIFFERENT conversation under the same ``session_key`` - — the snapshot is intentionally left untouched. Overwriting it with - the current session's count would corrupt the original conversation's - baseline and cause the next switch back to fire the cross-process - guard spuriously. Fail-safe: the legacy 3-tuple shape (no - ``session_id``) is still re-baselined as before. """ if self._session_db is None or not session_id: return @@ -15239,23 +14825,8 @@ async def _refresh_agent_cache_message_count( and len(cached) > 2 and cached[0] is not _AGENT_PENDING_SENTINEL ): - # If the snapshot was taken for a different session_id - # (same session_key, different conversation), leave the - # snapshot alone — the current session_id's count belongs - # to a different DB row (#54947). - _snapshot_sid = cached[3] if len(cached) > 3 else None - if _snapshot_sid is not None and _snapshot_sid != session_id: - return if cached[2] != _live: - if _snapshot_sid is None: - # Legacy 3-tuple: preserve the original 3-element - # shape so existing entries stay compatible with - # callers that index ``cached[2]`` directly. - _cache[session_key] = (cached[0], cached[1], _live) - else: - _cache[session_key] = ( - cached[0], cached[1], _live, _snapshot_sid, - ) + _cache[session_key] = (cached[0], cached[1], _live) def _evict_cached_agent(self, session_key: str) -> None: """Remove a cached agent for a session (called on /new, /model, etc). @@ -16962,25 +16533,9 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: if cached and cached[1] == _sig: # cached[2] is the message_count at cache time; # stale when a second process appended rows. - # cached[3] (when present) is the session_id the - # snapshot was taken for — used to skip the guard - # when the active session_id differs (#54947). _cached_mc = cached[2] if len(cached) > 2 else None - _cached_sid = cached[3] if len(cached) > 3 else None - # If the snapshot belongs to a different session_id - # (same session_key, different conversation), the - # message_count comparison is meaningless — the - # counts track DIFFERENT DB rows. REUSE the cached - # agent rather than rebuild and bust the prompt cache - # on every session switch (#54947). - _session_id_mismatch = ( - _cached_sid is not None - and session_id is not None - and _cached_sid != session_id - ) if ( - not _session_id_mismatch - and _cached_mc is not None + _cached_mc is not None and _current_msg_count is not None and _current_msg_count != _cached_mc ): @@ -17011,6 +16566,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: _xproc_evicted_agent = _ev_agent else: agent = cached[0] + reused_cached_agent = True # Refresh LRU order so the cap enforcement evicts # truly-oldest entries, not the one we just used. if hasattr(_cache, "move_to_end"): @@ -17023,7 +16579,6 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: # (cached agent may have been created with old config) agent.max_iterations = max_iterations logger.debug("Reusing cached agent for session %s", session_key) - reused_cached_agent = True # Lock released — now schedule cleanup of any cross-process-evicted # agent on a daemon thread so memory-provider shutdown / socket @@ -17081,14 +16636,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: ) if _cache_lock and _cache is not None: with _cache_lock: - # Record the session_id the snapshot was taken for - # alongside the message_count, so the cross-process - # guard can skip the (meaningless) count comparison - # when the active session_id later switches under - # the same session_key (#54947). - _cache[session_key] = ( - agent, _sig, _current_msg_count, session_id, - ) + _cache[session_key] = (agent, _sig, _current_msg_count) self._enforce_agent_cache_cap() logger.debug("Created new agent for session %s (sig=%s)", session_key, _sig) @@ -17482,30 +17030,10 @@ def _approval_notify_sync(approval_data: dict) -> None: _resume_entry = self.session_store._entries.get(session_key) except Exception: _resume_entry = None - - # resume_pending freshness uses a SECOND signal in addition to the - # transcript clock above. The restart watchdog stamps the session - # with ``last_resume_marked_at`` at interrupt time — that is the - # correct "when were we interrupted" signal. The transcript clock - # (_interruption_is_fresh) can be far older: an active thread you - # return to may have its last persisted row hours back, even though - # the interruption itself just happened. Gating resume_pending on - # the transcript clock alone makes the recovery note silently drop, - # and because the startup auto-resume turn carries empty text - # (_schedule_resume_pending_sessions), the model then receives a - # blank user message and replies with confused "the message came - # through blank" noise. Treat the marker as fresh when - # EITHER signal is fresh so the two freshness checks agree. - _resume_mark_is_fresh = False - if _resume_entry is not None and getattr(_resume_entry, "resume_pending", False): - _resume_mark_is_fresh = _is_fresh_gateway_interruption( - getattr(_resume_entry, "last_resume_marked_at", None), - window_secs=_freshness_window, - ) _is_resume_pending = bool( _resume_entry is not None and getattr(_resume_entry, "resume_pending", False) - and (_interruption_is_fresh or _resume_mark_is_fresh) + and _interruption_is_fresh ) _has_fresh_tool_tail = bool( agent_history @@ -17567,42 +17095,6 @@ def _approval_notify_sync(approval_data: dict) -> None: if _srn: message = _srn + "\n\n" + message - # Safety net: a startup auto-resume event carries empty - # text and relies on the resume_pending branch above to supply the - # recovery note. If that branch did not fire for any reason (e.g. - # both freshness signals disagreed, or the marker was cleared - # between scheduling and dispatch) we must NOT hand the model a - # blank user turn — it responds with confused "the message came - # through blank" noise. Restricted to resume_pending sessions so - # legitimately empty user turns (e.g. an image with no caption, - # wrapped as native content below) are untouched. - if ( - isinstance(message, str) - and not message.strip() - and _resume_entry is not None - and getattr(_resume_entry, "resume_pending", False) - ): - _sn_reason = ( - getattr(_resume_entry, "resume_reason", None) or "restart_timeout" - ) - _sn_reason_phrase = ( - "a gateway restart" - if _sn_reason == "restart_timeout" - else "a gateway shutdown" - if _sn_reason == "shutdown_timeout" - else "a gateway interruption" - ) - message = ( - f"[System note: The previous turn was interrupted by " - f"{_sn_reason_phrase}; the gateway is now back online. " - f"Any restart/shutdown command in the history has already " - f"run — do NOT re-execute or verify it. Report to the user " - f"that the session was restored successfully and ask what " - f"they would like to do next. Do NOT re-execute old tool " - f"calls — skip any unfinished work from the conversation " - f"history.]" - ) - _approval_session_key = session_key or "" _approval_session_token = set_current_session_key(_approval_session_key) register_gateway_notify(_approval_session_key, _approval_notify_sync) @@ -17749,7 +17241,6 @@ def _approval_notify_sync(approval_data: dict) -> None: ) effective_session_id = agent_session_id - self._sync_session_model_from_agent(effective_session_id, agent) # history_offset=0 whenever the agent's message list no longer has # the original history prefix — i.e. on rotation (split) OR in-place # compaction. In both cases the returned `messages` is the compacted @@ -17760,14 +17251,9 @@ def _approval_notify_sync(approval_data: dict) -> None: ) if not final_response: - final_response = _normalize_empty_agent_response( - result, final_response or "", history_len=len(agent_history), - ) - final_response = _sanitize_gateway_final_response(source.platform, final_response) - if not final_response: - final_response = f"⚠️ {result['error']}" if result.get("error") else "" + error_msg = f"⚠️ {result['error']}" if result.get("error") else "" return { - "final_response": final_response, + "final_response": error_msg, "messages": result.get("messages", []), "api_calls": result.get("api_calls", 0), "failed": result.get("failed", False), @@ -17889,11 +17375,6 @@ def _title_failure_cb(task: str, exc: BaseException) -> None: "session_id": effective_session_id, "response_previewed": result.get("response_previewed", False), "response_transformed": result.get("response_transformed", False), - # Pass through the agent_persisted flag so the persistence block - # above can correctly determine whether the codex app-server path - # self-persisted (it didn't — see codex_runtime.py). Default - # True preserves the skip-db behaviour for the standard runtime. - "agent_persisted": (result_holder[0].get("agent_persisted", True) if result_holder[0] else True), } # Start progress message sender if enabled. Gate on needs_progress_queue @@ -18624,22 +18105,6 @@ def _stream_confirmed_final_delivery( except Exception: pass - # Re-baseline the cached agent's message_count snapshot before - # recursing into the in-band queued (/queue) follow-up turn. - # The first turn has completed and flushed its own user + - # assistant rows to the SessionDB, so the cross-process - # coherence guard (#45966) — which this recursive _run_agent - # call re-enters — would otherwise see the grown on-disk count - # against the stale build-time snapshot and rebuild the agent - # on THIS process's OWN writes, destroying the prompt-cache - # prefix #46237 was merged to preserve. The existing - # re-baseline in _handle_message_with_agent only runs after the - # whole _run_agent chain unwinds — too late for the in-band - # follow-up. Use the same (session_key, session_id) the - # recursive call runs under so the snapshot matches exactly - # what the follow-up's guard will consult. Fail-safe in helper. - await self._refresh_agent_cache_message_count(session_key, session_id) - followup_result = await self._run_agent( message=next_message, context_prompt=context_prompt, @@ -19593,76 +19058,21 @@ def main(): data = yaml.safe_load(f) or {} config = GatewayConfig.from_dict(data) - # start_gateway() performs the full graceful teardown (adapters - # disconnected, sessions saved + flushed, SQLite closed, cron/MCP stopped, - # PID file + runtime lock released) before it returns OR raises SystemExit - # with an explicit code. Force-exit afterwards so a wedged non-daemon worker - # thread (e.g. a ThreadPoolExecutor tool/LLM call blocked with no timeout) - # cannot block interpreter finalization (Py_FinalizeEx joins all non-daemon - # threads, incl. concurrent.futures' _python_exit) and strand the gateway - # half-shut down with the supervisor unable to restart it (#53107). - # - # SystemExit is caught explicitly: start_gateway raises it on the - # clean-fatal-config (#51228), planned-restart, and service-restart paths, - # all of which complete teardown first. Routing those codes through the - # same os._exit backstop means EVERY exit path is wedge-proof, not just the - # boolean-return ones. - try: - success = asyncio.run(start_gateway(config)) - exit_code = 0 if success else 1 - except SystemExit as e: - # e.code may be None (→ 0), an int, or a str (→ 1, like CPython). - if e.code is None: - exit_code = 0 - elif isinstance(e.code, int): - exit_code = e.code - else: - exit_code = 1 - _exit_after_graceful_shutdown(exit_code) - - -def _exit_after_graceful_shutdown(exit_code: int) -> None: - """Flush stdio, release the PID file + runtime lock, then hard-exit. - - Graceful teardown is already complete by the time this runs, so there is - nothing left that needs a clean interpreter shutdown. We deliberately use - ``os._exit`` (not ``sys.exit``): ``sys.exit`` raises ``SystemExit``, which - triggers ``Py_FinalizeEx`` → ``wait_for_thread_shutdown`` and joins every - non-daemon thread — exactly the hang (#53107) a wedged tool-worker causes. - - ``os._exit`` bypasses ``atexit`` handlers, so we cannot rely on the - ``atexit``-registered ``remove_pid_file`` / ``release_gateway_runtime_lock`` - (registered in ``start_gateway``) to run. The full-shutdown path releases - both explicitly in ``_stop_impl``, but the EARLY exit paths — - clean-fatal-config (#51228) and startup-aborted-before-running — raise - ``SystemExit`` right after ``runner.start()`` without going through - ``_stop_impl``, so on those paths ``atexit`` was the only thing releasing - them. Now that those paths are routed through this backstop (#53107), - release both here explicitly. Both calls are idempotent — - ``remove_pid_file`` only unlinks a PID file that belongs to this process, - and ``release_gateway_runtime_lock`` no-ops when the lock is already - released — so this is a no-op on the normal shutdown path and the actual - cleanup on the early-exit paths. - - Logging is not flushed here: the gateway's handlers are synchronous - ``RotatingFileHandler``s that write each record immediately (no - ``MemoryHandler``/``QueueHandler`` buffering), so there is nothing pending. - Only stdio is buffered, so only stdio is flushed. - """ + # start_gateway() already performs graceful teardown before returning. + # Force-exit afterwards so a wedged non-daemon worker thread cannot block + # interpreter finalization and strand the gateway half-shut down. + success = asyncio.run(start_gateway(config)) + _exit_after_graceful_shutdown(success) + + +def _exit_after_graceful_shutdown(success: bool) -> None: + """Flush stdio and terminate immediately after graceful shutdown.""" for stream in (sys.stdout, sys.stderr): try: stream.flush() except Exception: pass - # Guaranteed cleanup chokepoint: os._exit skips atexit, and the early - # SystemExit exit paths never run _stop_impl, so release here (idempotent). - try: - from gateway.status import remove_pid_file, release_gateway_runtime_lock - remove_pid_file() - release_gateway_runtime_lock() - except Exception: - pass - os._exit(exit_code) + os._exit(0 if success else 1) if __name__ == "__main__": diff --git a/hermes_cli/main.py b/hermes_cli/main.py index b71c59f38353..735fe2602c7a 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -12806,6 +12806,23 @@ def _dispatch_secrets(args): # noqa: ANN001 seen_plugin_commands.add(cmd_info["name"]) discover_plugins() + # Bundled platform plugins are normally deferred so plain `hermes` + # startup doesn't import every platform SDK. If the unknown + # positional token is itself a platform plugin's CLI command (for + # example `hermes photon status`), resolve just that platform now; + # its register() call can then contribute the CLI subparser. + plugin_candidate = _first_positional_argv() + if plugin_candidate: + try: + from gateway.platform_registry import platform_registry + + platform_registry.get(plugin_candidate) + except Exception: + logging.getLogger(__name__).debug( + "Deferred platform CLI discovery failed for %s", + plugin_candidate, + exc_info=True, + ) for cmd_info in get_plugin_manager()._cli_commands.values(): if cmd_info["name"] in seen_plugin_commands: continue diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py index d6e627f667bc..301bab90507f 100644 --- a/plugins/platforms/photon/adapter.py +++ b/plugins/platforms/photon/adapter.py @@ -39,6 +39,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, List, Optional +from hermes_constants import get_hermes_home + if TYPE_CHECKING: # Type checkers see ``httpx`` as the always-imported module, so every use # site type-checks cleanly. The runtime fallback below keeps the optional @@ -85,6 +87,53 @@ _SIDECAR_DIR = Path(__file__).parent / "sidecar" + +def _sidecar_token_file(port: int) -> Path: + """Runtime handoff file for out-of-process senders.""" + return get_hermes_home() / "runtime" / f"photon-sidecar-{port}.json" + + +def _write_sidecar_token_file(port: int, token: str, pid: int) -> None: + """Persist the live sidecar control token with user-only permissions.""" + path = _sidecar_token_file(port) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + f".{os.getpid()}.tmp") + tmp.write_text( + json.dumps({"port": port, "token": token, "pid": pid}), + encoding="utf-8", + ) + try: + os.chmod(tmp, 0o600) + except OSError: + pass + tmp.replace(path) + + +def _read_sidecar_token_file(port: int) -> Optional[str]: + """Read the live gateway sidecar token for `hermes send`/cron fallback.""" + try: + data = json.loads(_sidecar_token_file(port).read_text(encoding="utf-8")) + except Exception: + return None + if int(data.get("port") or 0) != port: + return None + token = data.get("token") + return token if isinstance(token, str) and token else None + + +def _remove_sidecar_token_file(port: int, token: str) -> None: + """Remove our token file without clobbering a newer sidecar's token.""" + path = _sidecar_token_file(port) + try: + data = json.loads(path.read_text(encoding="utf-8")) + if data.get("token") == token: + path.unlink() + except FileNotFoundError: + return + except Exception: + logger.debug("[photon] failed to remove sidecar token file", exc_info=True) + + # Photon / Envoy / spectrum-ts error substrings that indicate a transient # upstream overload rather than a permanent failure. These are not in the # core _RETRYABLE_ERROR_PATTERNS because they are specific to this adapter. @@ -884,6 +933,11 @@ async def _start_sidecar(self) -> None: env=env, start_new_session=(sys.platform != "win32"), ) + _write_sidecar_token_file( + self._sidecar_port, + self._sidecar_token, + int(self._sidecar_proc.pid), + ) # Pump sidecar stderr/stdout into our logger so users see crashes. loop = asyncio.get_event_loop() @@ -983,6 +1037,7 @@ async def _stop_sidecar(self) -> None: except subprocess.TimeoutExpired: proc.kill() finally: + _remove_sidecar_token_file(self._sidecar_port, self._sidecar_token) self._sidecar_proc = None if self._sidecar_supervisor_task is not None: self._sidecar_supervisor_task.cancel() @@ -1584,13 +1639,13 @@ async def _standalone_send( (pconfig.extra or {}).get("sidecar_port") or os.getenv("PHOTON_SIDECAR_PORT"), _DEFAULT_SIDECAR_PORT, ) - token = os.getenv("PHOTON_SIDECAR_TOKEN") + token = os.getenv("PHOTON_SIDECAR_TOKEN") or _read_sidecar_token_file(port) if not token: return { "error": ( "Photon standalone send requires a running sidecar with " - "PHOTON_SIDECAR_TOKEN set in the environment. Cron processes " - "cannot spawn the sidecar themselves." + "a readable runtime token. Start/restart the Hermes gateway " + "so it can launch the Photon sidecar." ) } base = f"http://{_DEFAULT_SIDECAR_BIND}:{port}" diff --git a/tests/gateway/test_telegram_noise_filter.py b/tests/gateway/test_telegram_noise_filter.py index 5ba7c04e359b..23774083d501 100644 --- a/tests/gateway/test_telegram_noise_filter.py +++ b/tests/gateway/test_telegram_noise_filter.py @@ -33,6 +33,7 @@ "🗜️ Compacting context — summarizing earlier conversation so I can continue...", "⚠️ Session compressed 12 times — accuracy may degrade. Consider /new to start fresh.", "⚠ Compression summary failed: upstream error. Inserted a fallback context marker.", + "ℹ Codex gpt-5.5 caps context at 272K, so auto-compaction was raised to 85% (from 50%) to use more of the window before summarizing.\n Opt back out: hermes config set compression.codex_gpt55_autoraise false", "⏱️ Rate limited. Waiting 30.0s (attempt 2/3)...", "⏳ Retrying in 4.2s (attempt 1/3)...", ] diff --git a/tests/plugins/platforms/photon/test_sidecar_lifecycle.py b/tests/plugins/platforms/photon/test_sidecar_lifecycle.py index 31b005c2488f..71ed69e01a9d 100644 --- a/tests/plugins/platforms/photon/test_sidecar_lifecycle.py +++ b/tests/plugins/platforms/photon/test_sidecar_lifecycle.py @@ -8,12 +8,16 @@ """ from __future__ import annotations +import json +import stat import subprocess +from pathlib import Path from typing import Any, Dict, List, Tuple import pytest from gateway.config import PlatformConfig +from hermes_constants import reset_hermes_home_override, set_hermes_home_override from plugins.platforms.photon import adapter as photon_adapter from plugins.platforms.photon.adapter import PhotonAdapter @@ -59,6 +63,61 @@ def _fake_kill(pid: int, sig: int) -> None: return kills +@pytest.fixture() +def isolated_hermes_home(tmp_path: Path): + token = set_hermes_home_override(tmp_path) + try: + yield tmp_path + finally: + reset_hermes_home_override(token) + + +class TestSidecarTokenFile: + def test_write_then_read_round_trip(self, isolated_hermes_home: Path) -> None: + photon_adapter._write_sidecar_token_file(9001, "secret-token", 12345) + + path = isolated_hermes_home / "runtime" / "photon-sidecar-9001.json" + assert path.exists() + assert photon_adapter._read_sidecar_token_file(9001) == "secret-token" + assert json.loads(path.read_text(encoding="utf-8")) == { + "port": 9001, + "token": "secret-token", + "pid": 12345, + } + if hasattr(stat, "S_IMODE"): + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + def test_read_ignores_wrong_port(self, isolated_hermes_home: Path) -> None: + path = isolated_hermes_home / "runtime" / "photon-sidecar-9002.json" + path.parent.mkdir(parents=True) + path.write_text( + json.dumps({"port": 9003, "token": "wrong-port", "pid": 1}), + encoding="utf-8", + ) + + assert photon_adapter._read_sidecar_token_file(9002) is None + + def test_read_missing_or_corrupt_file_returns_none(self, isolated_hermes_home: Path) -> None: + assert photon_adapter._read_sidecar_token_file(9004) is None + + path = isolated_hermes_home / "runtime" / "photon-sidecar-9004.json" + path.parent.mkdir(parents=True) + path.write_text("not json", encoding="utf-8") + + assert photon_adapter._read_sidecar_token_file(9004) is None + + def test_remove_only_deletes_matching_token(self, isolated_hermes_home: Path) -> None: + photon_adapter._write_sidecar_token_file(9005, "newer-token", 222) + path = isolated_hermes_home / "runtime" / "photon-sidecar-9005.json" + + photon_adapter._remove_sidecar_token_file(9005, "older-token") + assert path.exists() + assert photon_adapter._read_sidecar_token_file(9005) == "newer-token" + + photon_adapter._remove_sidecar_token_file(9005, "newer-token") + assert not path.exists() + + @pytest.mark.asyncio async def test_reap_noop_when_port_free(monkeypatch: pytest.MonkeyPatch) -> None: adapter = _make_adapter(monkeypatch) @@ -125,7 +184,7 @@ async def test_reap_raises_for_foreign_listener( @pytest.mark.asyncio async def test_start_sidecar_spawns_with_stdin_pipe( - monkeypatch: pytest.MonkeyPatch, tmp_path + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_hermes_home: Path ) -> None: """The spawn must hold a stdin pipe and enable the sidecar's EOF watch.""" adapter = _make_adapter(monkeypatch) @@ -169,3 +228,4 @@ class _Resp: kwargs = spawned["kwargs"] assert kwargs["stdin"] is subprocess.PIPE assert kwargs["env"]["PHOTON_SIDECAR_WATCH_STDIN"] == "1" + assert photon_adapter._read_sidecar_token_file(adapter._sidecar_port) == adapter._sidecar_token