diff --git a/gateway/slash_commands_model.py b/gateway/slash_commands_model.py index 3e28c999c9..1a42095bfa 100644 --- a/gateway/slash_commands_model.py +++ b/gateway/slash_commands_model.py @@ -2,32 +2,42 @@ /model, /codex-runtime, /reasoning, /fast, /personality. Split out of ``gateway/slash_commands.py``; bound onto ``GatewayRunner`` through -``GatewaySlashCommandsMixin``. Origin internals are imported lazily (``from gateway.slash_commands -import ...``) inside the bodies to avoid the import cycle. +``GatewaySlashCommandsMixin``. Origin internals are imported lazily inside the bodies to avoid +the import cycle. """ from __future__ import annotations -import logging import asyncio -from typing import Optional +import contextlib +import dataclasses +import logging +from typing import Any, Optional from agent.i18n import t from gateway.platforms.base import MessageEvent from hermes_cli.config import atomic_config_write, clear_model_endpoint_credentials from utils import base_url_host_matches -# Log-record parity with gateway/run.py and the origin module. -logger = logging.getLogger("gateway.run") +logger = logging.getLogger("gateway.run") # log-record parity with gateway/run.py + +# /fast argument -> (service tier, persisted value, i18n label key; None = value.upper()). +_FAST_SELECTIONS = { + "fast": ("priority", "fast", "gateway.fast.label_fast"), + "on": ("priority", "fast", "gateway.fast.label_fast"), + "normal": (None, "normal", "gateway.fast.label_normal"), + "off": (None, "normal", "gateway.fast.label_normal"), + "auto": ("auto", "auto", None), + "cold": ("cold", "cold", None), +} + +# /reasoning display-toggle arguments -> show_reasoning value. +_REASONING_DISPLAY_TOGGLES = {"show": True, "on": True, "hide": False, "off": False} def _model_switch_skew_guard() -> Optional[str]: - """Refuse a model switch when the gateway is running stale code. - - A long-lived gateway keeps boot-time modules in memory; if the checkout changed underneath it, - a first-time lazy import on a new code path can crash on a stale cached dependency. Detect the - drift and ask for a restart. Scoped to model switching only (the highest-risk trigger). - """ + """Refuse a model switch when the gateway is running stale code: a first-time lazy import on + a new code path can crash on a stale cached dependency. Scoped to the highest-risk trigger.""" from gateway.code_skew import detect_code_skew skew = detect_code_skew() @@ -47,12 +57,9 @@ def _model_switch_skew_guard() -> Optional[str]: async def _persist_model_switch_to_config(result, config_path) -> None: """Write-through a resolved /model switch to ``config_path`` (model.default/provider/base_url). - Write-back round-trip: raw read is correct (merged defaults must not be persisted back to the - user's file). A scalar/None ``model:`` is coerced into a dict first — otherwise - ``cfg.setdefault("model", {})`` returns the existing scalar and the next assignment raises - ``TypeError``. Named providers re-resolve base_url/api_mode fresh, so leftovers are cleared - unconditionally; custom providers have no registry entry to re-derive from, so they need an - explicit set-or-clear (a lone ``if base_url:`` leaves stale values). + Raw read: merged defaults must not be persisted back. A scalar/None ``model:`` is coerced to a + dict first. Named providers re-resolve base_url/api_mode, so leftovers are cleared; custom + providers have no registry entry to re-derive from and need an explicit set-or-clear. """ from hermes_cli.config import read_user_config_raw, save_config @@ -66,17 +73,13 @@ async def _persist_model_switch_to_config(result, config_path) -> None: model_cfg = cfg["model"] = {} try: from hermes_cli.route_identity import should_clear_context_pin_async - - if await should_clear_context_pin_async( - model_cfg.get("default") or model_cfg.get("model"), - result.new_model, - model_cfg.get("base_url"), - result.base_url, - model_cfg.get("provider"), - result.target_provider, - ): - model_cfg.pop("context_length", None) + clear_pin = await should_clear_context_pin_async( + model_cfg.get("default") or model_cfg.get("model"), result.new_model, + model_cfg.get("base_url"), result.base_url, model_cfg.get("provider"), result.target_provider, + ) except Exception: + clear_pin = True + if clear_pin: model_cfg.pop("context_length", None) model_cfg["default"] = result.new_model model_cfg["provider"] = result.target_provider @@ -85,46 +88,65 @@ async def _persist_model_switch_to_config(result, config_path) -> None: model_cfg["base_url"] = result.base_url elif is_custom_target: model_cfg.pop("base_url", None) - if is_custom_target: - if result.api_mode: - model_cfg["api_mode"] = result.api_mode - else: - model_cfg.pop("api_mode", None) - else: + if not is_custom_target: clear_model_endpoint_credentials(model_cfg, clear_base_url=True) + elif result.api_mode: + model_cfg["api_mode"] = result.api_mode + else: + model_cfg.pop("api_mode", None) save_config(cfg) -def _read_model_command_config(config_path): - """Current (model, provider, base_url, user_providers, custom_providers, excluded) for /model. +@dataclasses.dataclass +class _ModelSwitchContext: + """Everything a /model switch needs beyond the target: current route + persistence policy.""" - Fail-open: any config read error yields the defaults (``provider="openrouter"``). - """ - from gateway.run import _load_gateway_config + session_key: str + source: Any + config_path: Any + persist_global: bool + one_turn: bool = False + restore_snapshot: Optional[dict] = None + current_model: str = "" + current_provider: str = "openrouter" + current_base_url: str = "" + current_api_key: str = "" + user_provs: Any = None + custom_provs: Any = None + excluded_provs: list = dataclasses.field(default_factory=list) - current_model, current_provider, current_base_url = "", "openrouter", "" - user_provs = custom_provs = None - excluded_provs: list = [] - try: - cfg = _load_gateway_config(config_path=config_path) - if cfg: + def read_config(self) -> None: + """Fill the current route from ``config_path``; fail-open to the defaults.""" + from gateway.run import _load_gateway_config + try: + cfg = _load_gateway_config(config_path=self.config_path) + if not cfg: + return model_cfg = cfg.get("model", {}) if isinstance(model_cfg, dict): - current_model = model_cfg.get("default", "") - current_provider = model_cfg.get("provider", current_provider) - current_base_url = model_cfg.get("base_url", "") - user_provs = cfg.get("providers") + self.current_model = model_cfg.get("default", "") + self.current_provider = model_cfg.get("provider", self.current_provider) + self.current_base_url = model_cfg.get("base_url", "") + self.user_provs = cfg.get("providers") try: from hermes_cli.config import get_compatible_custom_providers - custom_provs = get_compatible_custom_providers(cfg) + self.custom_provs = get_compatible_custom_providers(cfg) except Exception: - custom_provs = cfg.get("custom_providers") - _excl = cfg.get("model_catalog", {}).get("excluded_providers") - if isinstance(_excl, list): - excluded_provs = _excl - except Exception: - pass - return current_model, current_provider, current_base_url, user_provs, custom_provs, excluded_provs + self.custom_provs = cfg.get("custom_providers") + excl = cfg.get("model_catalog", {}).get("excluded_providers") + if isinstance(excl, list): + self.excluded_provs = excl + except Exception: + pass + + def apply_override(self, override: dict) -> None: + """A session /model override supersedes the configured route.""" + if override: + self.current_model = override.get("model", self.current_model) + self.current_provider = override.get("provider", self.current_provider) + self.current_base_url = override.get("base_url", self.current_base_url) + self.current_api_key = override.get("api_key", self.current_api_key) + def _model_provider_listing_lines(providers) -> list[str]: @@ -135,7 +157,8 @@ def _model_provider_listing_lines(providers) -> list[str]: lines.append(f"**{p['name']}** `--provider {p['slug']}`{tag}:") if p["models"]: model_strs = ", ".join(f"`{m}`" for m in p["models"]) - extra = t("gateway.model.more_models_suffix", count=p["total_models"] - len(p["models"])) if p["total_models"] > len(p["models"]) else "" + hidden = p["total_models"] - len(p["models"]) + extra = t("gateway.model.more_models_suffix", count=hidden) if hidden > 0 else "" lines.append(f" {model_strs}{extra}") elif p.get("api_url"): lines.append(f" `{p['api_url']}`") @@ -146,232 +169,168 @@ def _model_provider_listing_lines(providers) -> list[str]: class GatewayModelCommandsMixin: """Model-route slash commands (/model, /codex-runtime, /reasoning, /fast, /personality).""" + # ----------------------------------------------------------------- /model + async def _perform_model_switch( - self, - switch_model, - *, - raw_input: str, - explicit_provider, - session_key: str, - source, - current_model, - current_provider, - current_base_url, - current_api_key, - persist_global: bool, - user_provs, - custom_provs, + self, ctx: _ModelSwitchContext, raw_input: str, explicit_provider, source ): """Resolve a /model switch off-loop. Returns ``(result, None)`` or ``(None, error_text)``.""" from gateway.run import _load_gateway_config + from hermes_cli.model_switch import switch_model skew_error = _model_switch_skew_guard() if skew_error: return None, skew_error - # Offload the switch off the event loop — switch_model() can fall through to a synchronous - # models.dev HTTP fetch (requests.get, 15s timeout) on a cold/expired cache, which freezes - # the gateway otherwise. + # Off-loop: switch_model() can hit a synchronous models.dev fetch (15s) on a cold cache. result = await asyncio.to_thread( - switch_model, - raw_input=raw_input, - current_provider=current_provider, - current_model=current_model, - current_base_url=current_base_url, - current_api_key=current_api_key, - is_global=persist_global, - explicit_provider=explicit_provider, - user_providers=user_provs, - custom_providers=custom_provs, + switch_model, raw_input=raw_input, current_provider=ctx.current_provider, + current_model=ctx.current_model, current_base_url=ctx.current_base_url, + current_api_key=ctx.current_api_key, is_global=ctx.persist_global, + explicit_provider=explicit_provider, user_providers=ctx.user_provs, + custom_providers=ctx.custom_provs, ) if not result.success: return None, t("gateway.model.error_prefix", error=result.error_message) try: from hermes_cli.context_switch_guard import enrich_model_switch_warnings_for_gateway - - # Offload: merge_preflight_compression_warning() calls the sync - # resolve_display_context_length() provider probe ladder — must not run on the loop. + # Off-loop: merge_preflight_compression_warning() runs the sync provider probe ladder. await asyncio.to_thread( - enrich_model_switch_warnings_for_gateway, - result, - self, - session_key=session_key, - source=source, - custom_providers=custom_provs, + enrich_model_switch_warnings_for_gateway, result, self, session_key=ctx.session_key, + source=source, custom_providers=ctx.custom_provs, load_gateway_config=_load_gateway_config, ) except Exception as exc: logger.debug("preflight-compression switch warning failed: %s", exc) return result, None - async def _commit_model_switch( - self, - result, - *, - session_key: str, - source, - current_model, - current_base_url, - current_api_key, - custom_provs, - persist_global: bool, - config_path, - one_turn: bool = False, - restore_snapshot=None, - picker: bool = False, - ) -> str: - """Apply a resolved switch (cached agent, session, config) and build the confirmation. + def _switch_cached_agent_model(self, result, ctx: _ModelSwitchContext, picker: bool) -> Optional[str]: + """In-place swap on the cached agent; returns the error reply when it failed. - Shared by the typed ``/model `` path and the picker callback (``picker=True``). + The agent rolls back to the OLD model/client and re-raises; the commit (DB, override, + eviction, config) is aborted so the next message doesn't rebuild a broken agent. """ - from gateway.run import _load_gateway_config - from hermes_cli.model_switch import format_model_for_display, resolve_display_context_length_async + cached_agent = self._cached_agent_for(ctx.session_key) + if cached_agent is None: + return None + try: + cached_agent.switch_model( + new_model=result.new_model, new_provider=result.target_provider, + api_key=result.api_key, base_url=result.base_url, api_mode=result.api_mode, + capabilities=getattr(result, "runtime_capabilities", None), + ) + except Exception as exc: + logger.warning( + "%s model switch failed for cached agent: %s", "Picker" if picker else "In-place", exc + ) + return t( + "gateway.model.error_prefix", + error=f"Model switch to {result.new_model} failed ({exc}); staying on {ctx.current_model}.", + ) + return None - # If there's a cached agent, update it in-place - cached_agent = self._cached_agent_for(session_key) - if cached_agent is not None: - try: - cached_agent.switch_model( - new_model=result.new_model, - new_provider=result.target_provider, - api_key=result.api_key, - base_url=result.base_url, - api_mode=result.api_mode, - capabilities=getattr(result, "runtime_capabilities", None), - ) - except Exception as exc: - # In-place swap rolled back to the OLD working model/client and re-raised. Abort the - # commit (DB persist, session override, cache eviction, config write) so a failed switch - # is a no-op — otherwise the next message rebuilds a broken agent from the override. - logger.warning( - "%s model switch failed for cached agent: %s", "Picker" if picker else "In-place", exc - ) - return t( - "gateway.model.error_prefix", - error=f"Model switch to {result.new_model} failed ({exc}); staying on {current_model}.", - ) + async def _record_model_switch( + self, result, ctx: _ModelSwitchContext, *, source, one_turn: bool, picker: bool + ) -> None: + """Persist a committed switch: session DB, next-turn note, override map, config write-through.""" + from hermes_cli.model_switch import format_model_for_display - # Persist the new model to the session DB so the dashboard shows the updated model. _sess_db = getattr(self, "_session_db", None) - if _sess_db is not None: + if _sess_db is not None: # so the dashboard shows the updated model try: _sess_entry = await self.async_session_store.get_or_create_session(source) - # Typed path: if this session was auto-reset, consume the flag so the next regular - # message's cleanup does not wipe the model override just stored below. + # Typed path: consume the auto-reset flag so the next message's cleanup does not + # wipe the override stored below. if not picker and getattr(_sess_entry, "was_auto_reset", False): _sess_entry.was_auto_reset = False await _sess_db.update_session_model( - _sess_entry.session_id, result.new_model, - provider=result.target_provider, + _sess_entry.session_id, result.new_model, provider=result.target_provider, ) except Exception as exc: logger.debug("Failed to persist model switch to DB: %s", exc) - - # Store a note to prepend to the next user message so the model knows about the switch - # (avoids system messages mid-history). Display form strips opaque Palantir RID - # prefixes; the override map below keeps the full ID for the wire. + # Prepended to the next user message (no system messages mid-history). Display form strips + # opaque Palantir RID prefixes; the override map keeps the full ID for the wire. if not hasattr(self, "_pending_model_notes"): self._pending_model_notes = {} - self._pending_model_notes[session_key] = ( - f"[Note: model was just switched from {format_model_for_display(current_model)} to " + self._pending_model_notes[ctx.session_key] = ( + f"[Note: model was just switched from {format_model_for_display(ctx.current_model)} to " f"{format_model_for_display(result.new_model)} " f"via {result.provider_label or result.target_provider}. " f"{'This override applies to the next turn only. ' if one_turn else ''}" f"Adjust your self-identification accordingly.]" ) - - # Store session override so next agent creation uses the new model - self._session_model_overrides[session_key] = { - "model": result.new_model, - "provider": result.target_provider, - "api_key": result.api_key, - "base_url": result.base_url, - "api_mode": result.api_mode, + self._session_model_overrides[ctx.session_key] = { + "model": result.new_model, "provider": result.target_provider, "api_key": result.api_key, + "base_url": result.base_url, "api_mode": result.api_mode, "request_overrides": dict(result.request_overrides or {}), "capabilities": dict(result.runtime_capabilities or {}), } if one_turn: if not hasattr(self, "_pending_one_turn_model_restores"): self._pending_one_turn_model_restores = {} - self._pending_one_turn_model_restores[session_key] = ( - restore_snapshot or {"had_override": False, "override": None} - ) + snapshot = ctx.restore_snapshot or {"had_override": False, "override": None} + self._pending_one_turn_model_restores[ctx.session_key] = snapshot elif not picker and hasattr(self, "_pending_one_turn_model_restores"): - self._pending_one_turn_model_restores.pop(session_key, None) - - # Write-through the non-secret parts (model/provider/base_url) so the override survives a - # restart; api_key/api_mode are never persisted (re-resolved on rehydration). /model --once is - # EXCLUDED: a one-turn override must not outlive a restart; the pre-once value stays persisted. + self._pending_one_turn_model_restores.pop(ctx.session_key, None) + # Non-secret write-through so the override survives a restart (api_key/api_mode are + # re-resolved on rehydration); a --once override must NOT outlive a restart. if not one_turn: try: await self.async_session_store.set_model_override( - session_key, self._session_model_overrides[session_key] + ctx.session_key, self._session_model_overrides[ctx.session_key] ) except Exception: logger.debug("Failed to persist session model override", exc_info=True) - - # Evict cached agent so the next turn creates a fresh agent from the - # override rather than relying on cache signature mismatch detection. - self._evict_cached_agent(session_key) - - # Persist to config (default) unless --session opted out - if persist_global: + self._evict_cached_agent(ctx.session_key) # next turn builds fresh from the override + if ctx.persist_global: try: - await _persist_model_switch_to_config(result, config_path) + await _persist_model_switch_to_config(result, ctx.config_path) except Exception as e: logger.warning("Failed to persist model switch: %s", e) - # Build confirmation message with full metadata. Display form shortens opaque Palantir - # IDs (ri.language-model-service..*) to their trailing slug. - provider_label = result.provider_label or result.target_provider - lines = [t("gateway.model.switched", model=format_model_for_display(result.new_model))] - lines.append(t("gateway.model.provider_label", provider=provider_label)) + async def _model_switch_confirmation( + self, result, ctx: _ModelSwitchContext, *, one_turn: bool, picker: bool + ) -> str: + """Confirmation text with full metadata (display form shortens opaque Palantir IDs).""" + from gateway.run import _load_gateway_config + from hermes_cli.model_switch import format_model_for_display, resolve_display_context_length_async - # Context: always resolve via the provider-aware chain so Codex OAuth, - # Copilot, and Nous-enforced caps win over the raw models.dev entry. + lines = [ + t("gateway.model.switched", model=format_model_for_display(result.new_model)), + t("gateway.model.provider_label", provider=result.provider_label or result.target_provider), + ] + # Provider-aware chain: Codex OAuth, Copilot and Nous caps win over the raw models.dev entry. mi = result.model_info - _sw_config_ctx = None - _sw_model_cfg = {} - try: - _sw_model_cfg = _load_gateway_config().get("model", {}) - if isinstance(_sw_model_cfg, dict): - _sw_raw = _sw_model_cfg.get("context_length") - if _sw_raw is not None: - _sw_config_ctx = int(_sw_raw) - except Exception: - pass - if not isinstance(_sw_model_cfg, dict): - _sw_model_cfg = {} - ctx = await resolve_display_context_length_async( - result.new_model, - result.target_provider, - base_url=result.base_url or current_base_url or "", - api_key=result.api_key or current_api_key or "", - model_info=mi, - custom_providers=custom_provs, - config_context_length=_sw_config_ctx, - configured_model=_sw_model_cfg.get("default") or _sw_model_cfg.get("model"), - configured_provider=_sw_model_cfg.get("provider"), - configured_base_url=_sw_model_cfg.get("base_url"), + model_cfg: dict = {} + config_ctx = None + with contextlib.suppress(Exception): # fail-open on config read errors + model_cfg = _load_gateway_config().get("model", {}) + if isinstance(model_cfg, dict) and model_cfg.get("context_length") is not None: + config_ctx = int(model_cfg["context_length"]) + if not isinstance(model_cfg, dict): + model_cfg = {} + ctx_len = await resolve_display_context_length_async( + result.new_model, result.target_provider, + base_url=result.base_url or ctx.current_base_url or "", + api_key=result.api_key or ctx.current_api_key or "", model_info=mi, + custom_providers=ctx.custom_provs, config_context_length=config_ctx, + configured_model=model_cfg.get("default") or model_cfg.get("model"), + configured_provider=model_cfg.get("provider"), + configured_base_url=model_cfg.get("base_url"), ) - if ctx: - lines.append(t("gateway.model.context_label", tokens=f"{ctx:,}")) + if ctx_len: + lines.append(t("gateway.model.context_label", tokens=f"{ctx_len:,}")) + if mi and mi.max_output: + lines.append(t("gateway.model.max_output_label", tokens=f"{mi.max_output:,}")) if mi: - if mi.max_output: - lines.append(t("gateway.model.max_output_label", tokens=f"{mi.max_output:,}")) lines.append(t("gateway.model.capabilities_label", capabilities=mi.format_capabilities())) - - if not picker: - cache_enabled = ( - (base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower()) - or result.api_mode == "anthropic_messages" - ) - if cache_enabled: - lines.append(t("gateway.model.prompt_caching_enabled")) - + openrouter_claude = ( + base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower() + ) + if not picker and (openrouter_claude or result.api_mode == "anthropic_messages"): + lines.append(t("gateway.model.prompt_caching_enabled")) if result.warning_message: lines.append(t("gateway.model.warning_prefix", warning=result.warning_message)) - - if persist_global: + if ctx.persist_global: lines.append(t("gateway.model.saved_global")) elif one_turn: lines.append(" (next turn only — restores after one response)") @@ -379,17 +338,23 @@ class GatewayModelCommandsMixin: lines.append(t("gateway.model.session_only_hint")) return "\n".join(lines) + async def _commit_model_switch( + self, result, ctx: _ModelSwitchContext, *, source, picker: bool = False + ) -> str: + """Apply a resolved switch (cached agent, session, config) and build the confirmation; shared + by the typed path and the picker callback (``picker=True`` never carries --once).""" + one_turn = False if picker else ctx.one_turn + error = self._switch_cached_agent_model(result, ctx, picker) + if error is not None: + return error + await self._record_model_switch(result, ctx, source=source, one_turn=one_turn, picker=picker) + return await self._model_switch_confirmation(result, ctx, one_turn=one_turn, picker=picker) + async def _send_model_picker(self, event: MessageEvent, source, adapter, session_key: str, listing_kwargs: dict, on_model_selected) -> bool: - """Send the interactive /model picker; False when nothing was sent (text fallback). - - *source* is the session-key-normalized source (Telegram topic recovery), so the picker's - thread metadata lands where the next turn reads. - """ + """Send the interactive /model picker; False when nothing was sent (text fallback). *source* + is session-key-normalized so the picker's thread metadata lands where the next turn reads.""" from hermes_cli.model_switch import list_picker_providers - - try: - # Offload blocking provider-listing (can fall through to a synchronous urllib HTTP fetch - # on a stale cache) off the event loop so the gateway doesn't freeze. See #41289. + try: # off-loop: listing can hit a synchronous HTTP fetch on a stale cache providers = await asyncio.to_thread( list_picker_providers, max_models=50, include_moa=True, **listing_kwargs ) @@ -398,254 +363,160 @@ class GatewayModelCommandsMixin: if not providers: return False result = await adapter.send_model_picker( - chat_id=source.chat_id, - providers=providers, + chat_id=source.chat_id, providers=providers, current_model=listing_kwargs["current_model"], - current_provider=listing_kwargs["current_provider"], - session_key=session_key, + current_provider=listing_kwargs["current_provider"], session_key=session_key, on_model_selected=on_model_selected, metadata=self._thread_metadata_for_source(source, self._reply_anchor_for_event(event)), ) return bool(result.success) + async def _model_listing_reply( + self, event: MessageEvent, ctx: _ModelSwitchContext, profile_home + ) -> Optional[str]: + """``/model`` with no args: interactive picker where supported, else the text list.""" + from hermes_cli.model_switch import list_authenticated_providers + from hermes_cli.providers import get_label + + listing_kwargs = dict( + current_provider=ctx.current_provider, current_base_url=ctx.current_base_url, + current_model=ctx.current_model, user_providers=ctx.user_provs, + custom_providers=ctx.custom_provs, excluded_providers=ctx.excluded_provs, + ) + adapter = self._adapter_for_source(ctx.source) + if adapter is not None and getattr(type(adapter), "send_model_picker", None) is not None: + async def _picker_switch(model_id: str, provider_slug: str) -> str: + # The picker callback binds the raw event source (pre-normalization). + result, error = await self._perform_model_switch(ctx, model_id, provider_slug, event.source) + if error is not None: + return error + return await self._commit_model_switch(result, ctx, source=event.source, picker=True) + + async def _on_model_selected(_chat_id: str, model_id: str, provider_slug: str) -> str: + if profile_home is None: + return await _picker_switch(model_id, provider_slug) + from gateway.run import _profile_runtime_scope + with _profile_runtime_scope(profile_home): + return await _picker_switch(model_id, provider_slug) + + if await self._send_model_picker(event, ctx.source, adapter, ctx.session_key, listing_kwargs, _on_model_selected): + return None # Picker sent — adapter handles the response + + lines = [t("gateway.model.current_label", model=ctx.current_model or "unknown", provider=get_label(ctx.current_provider)), ""] + try: # off-loop: listing can hit a stale-cache HTTP fetch + providers = await asyncio.to_thread(list_authenticated_providers, max_models=5, **listing_kwargs) + lines.extend(_model_provider_listing_lines(providers)) + except Exception: + pass + lines.append(t("gateway.model.usage_switch_model")) + lines.append(t("gateway.model.usage_switch_provider")) + lines.append(t("gateway.model.usage_persist")) + return "\n".join(lines) + + async def _model_selection_guard_reply( + self, event: MessageEvent, ctx: _ModelSwitchContext, result + ) -> tuple[bool, Optional[str]]: + """Selection-guard confirmation for the typed path (pickers confirm via their own UI). + + The unified registry (cost + data-policy guards) runs off-loop — pricing lookups may hit + models.dev on a cache miss. Returns ``(fired, reply)``; the reply is None when the platform + rendered confirm buttons itself. + """ + try: + from hermes_cli.model_selection_guards import combined_selection_warning + warning = await asyncio.to_thread( + combined_selection_warning, result.new_model, provider=result.target_provider, + base_url=result.base_url or ctx.current_base_url or "", + api_key=result.api_key or ctx.current_api_key or "", model_info=result.model_info, + ) + except Exception: + warning = None + if warning is None: + return False, None + + async def _on_cost_confirm(choice: str) -> str: + if choice == "cancel": + return f"🟡 Model switch cancelled. Current model unchanged ({ctx.current_model or 'unknown'})." + # "once" and "always" both proceed — selection guards have no persistent opt-out. + return await self._commit_model_switch(result, ctx, source=ctx.source) + + _p = self._typed_command_prefix_for(event.source.platform) + message = ( + f"⚠️ **{warning.title}**\n\n{warning.message}\n\n" + f"_Text fallback: reply `{_p}approve` to switch or `{_p}cancel` to keep the current model._" + ) + return True, await self._request_slash_confirm( + event=event, command="model", title=warning.title, message=message, handler=_on_cost_confirm, + ) + async def _handle_model_command(self, event: MessageEvent) -> Optional[str]: """Handle /model command — switch model.""" from gateway.run import _hermes_home - from hermes_cli.model_switch import ( - switch_model as _switch_model, parse_model_switch_args, - resolve_persist_behavior, - list_authenticated_providers, - ) - from hermes_cli.providers import get_label + from hermes_cli.model_switch import parse_model_switch_args, resolve_persist_behavior - raw_args = event.get_command_args().strip() - source = event.source - _command_profile_home = None + profile_home = None if getattr(getattr(self, "config", None), "multiplex_profiles", False): - _command_profile_home = self._resolve_profile_home_for_source(source) - - # Parse --provider, --global, --session, --once, and --refresh flags - # via the shared single-owner parser (hermes_cli.model_switch). - request = parse_model_switch_args(raw_args) - model_input = request.target - explicit_provider = request.explicit_provider - is_global_flag = request.is_global - force_refresh = request.force_refresh - is_session = request.is_session - one_turn = request.is_once + profile_home = self._resolve_profile_home_for_source(event.source) + request = parse_model_switch_args(event.get_command_args().strip()) # single-owner parser if request.errors: - # Gateway decoration: "❌ " prefix over the canonical error copy. - return f"❌ {request.error_messages()[0]}" - persist_global = resolve_persist_behavior( - is_global_flag, - is_session, - is_once=one_turn, - explicit_provider=explicit_provider, - ) - - # --refresh: bust the disk cache so the picker shows live data. - if force_refresh: - try: + return f"❌ {request.error_messages()[0]}" # gateway decoration over canonical copy + if request.force_refresh: # bust the disk cache so the picker shows live data + with contextlib.suppress(Exception): from hermes_cli.models import clear_provider_models_cache clear_provider_models_cache() - except Exception: - pass - - # Read current model/provider from config - config_path = (_command_profile_home or _hermes_home) / "config.yaml" - current_model, current_provider, current_base_url, user_provs, custom_provs, excluded_provs = ( - _read_model_command_config(config_path) - ) - current_api_key = "" - - # Check for session override. Normalize the source the same way a normal message turn does - # (Telegram DM topic recovery) before deriving the override key, so the override is stored - # under the key the next message turn reads. - source = await asyncio.to_thread(self._normalize_source_for_session_key, source) + # Normalize like a message turn (Telegram DM topic recovery) before deriving the override + # key, so the override lands under the key the next turn reads. + source = await asyncio.to_thread(self._normalize_source_for_session_key, event.source) session_key = self._session_key_for_source(source) - override = self._session_model_overrides.get(session_key, {}) - restore_snapshot = ( - self._snapshot_session_model_override(session_key) if one_turn else None + ctx = _ModelSwitchContext( + session_key=session_key, + source=source, + config_path=(profile_home or _hermes_home) / "config.yaml", + persist_global=resolve_persist_behavior( + request.is_global, request.is_session, is_once=request.is_once, + explicit_provider=request.explicit_provider, + ), + one_turn=request.is_once, + restore_snapshot=self._snapshot_session_model_override(session_key) if request.is_once else None, ) - if override: - current_model = override.get("model", current_model) - current_provider = override.get("provider", current_provider) - current_base_url = override.get("base_url", current_base_url) - current_api_key = override.get("api_key", current_api_key) - - async def perform_switch(model_id: str, provider_slug, *, src=source): - return await self._perform_model_switch( - _switch_model, - raw_input=model_id, - explicit_provider=provider_slug, - session_key=session_key, - source=src, - current_model=current_model, - current_provider=current_provider, - current_base_url=current_base_url, - current_api_key=current_api_key, - persist_global=persist_global, - user_provs=user_provs, - custom_provs=custom_provs, - ) - - async def commit_switch(result, *, picker: bool = False, src=source) -> str: - """Apply the resolved switch (agent, session, config) and build the reply.""" - return await self._commit_model_switch( - result, - session_key=session_key, - source=src, - current_model=current_model, - current_base_url=current_base_url, - current_api_key=current_api_key, - custom_provs=custom_provs, - persist_global=persist_global, - config_path=config_path, - one_turn=False if picker else one_turn, - restore_snapshot=None if picker else restore_snapshot, - picker=picker, - ) - - async def switch_and_commit(model_id: str, provider_slug, *, picker: bool) -> str: - # The picker callback binds the raw event source (pre-normalization), as it always has. - src = event.source if picker else source - result, error = await perform_switch(model_id, provider_slug, src=src) - if error is not None: - return error - return await commit_switch(result, picker=picker, src=src) - - # No args: show interactive picker (Telegram/Discord) or text list - if not model_input and not explicit_provider: - listing_kwargs = dict( - current_provider=current_provider, - current_base_url=current_base_url, - current_model=current_model, - user_providers=user_provs, - custom_providers=custom_provs, - excluded_providers=excluded_provs, - ) - # Try interactive picker if the platform supports it - adapter = self._adapter_for_source(source) - if adapter is not None and getattr(type(adapter), "send_model_picker", None) is not None: - async def _on_model_selected(_chat_id: str, model_id: str, provider_slug: str) -> str: - """Perform the model switch and return confirmation text.""" - if _command_profile_home is None: - return await switch_and_commit(model_id, provider_slug, picker=True) - from gateway.run import _profile_runtime_scope - - with _profile_runtime_scope(_command_profile_home): - return await switch_and_commit(model_id, provider_slug, picker=True) - - if await self._send_model_picker(event, source, adapter, session_key, listing_kwargs, _on_model_selected): - return None # Picker sent — adapter handles the response - - # Fallback: text list (for platforms without picker or if picker failed) - lines = [t("gateway.model.current_label", model=current_model or "unknown", provider=get_label(current_provider)), ""] - try: - # Offload blocking provider-listing off the event loop so the - # gateway doesn't freeze on a stale-cache HTTP fetch. See #41289. - providers = await asyncio.to_thread(list_authenticated_providers, max_models=5, **listing_kwargs) - lines.extend(_model_provider_listing_lines(providers)) - except Exception: - pass - lines.append(t("gateway.model.usage_switch_model")) - lines.append(t("gateway.model.usage_switch_provider")) - lines.append(t("gateway.model.usage_persist")) - return "\n".join(lines) - - # Perform the switch - result, error = await perform_switch(model_input, explicit_provider) + ctx.read_config() + ctx.apply_override(self._session_model_overrides.get(session_key, {})) + if not request.target and not request.explicit_provider: + return await self._model_listing_reply(event, ctx, profile_home) + result, error = await self._perform_model_switch(ctx, request.target, request.explicit_provider, source) if error is not None: return error + guard_fired, guard_reply = await self._model_selection_guard_reply(event, ctx, result) + if guard_fired: + return guard_reply + return await self._commit_model_switch(result, ctx, source=source) - # Selection-guard confirmation for the typed /model path (pickers confirm via their own - # UI). Runs the unified registry (cost + data-policy guards); pricing lookups may hit - # models.dev or a /models endpoint on a cache miss, so run it off the event loop. - _cost_warning = None - try: - from hermes_cli.model_selection_guards import combined_selection_warning - - _cost_warning = await asyncio.to_thread( - combined_selection_warning, - result.new_model, - provider=result.target_provider, - base_url=result.base_url or current_base_url or "", - api_key=result.api_key or current_api_key or "", - model_info=result.model_info, - ) - except Exception: - _cost_warning = None - if _cost_warning is not None: - async def _on_cost_confirm(choice: str) -> str: - if choice == "cancel": - return ( - f"🟡 Model switch cancelled. Current model unchanged " - f"({current_model or 'unknown'})." - ) - # "once" and "always" both proceed — there is no persistent - # opt-out for selection guards (each guarded switch should be - # an explicit decision). - return await commit_switch(result) - - _p = self._typed_command_prefix_for(event.source.platform) - return await self._request_slash_confirm( - event=event, - command="model", - title=_cost_warning.title, - message=( - f"⚠️ **{_cost_warning.title}**\n\n{_cost_warning.message}\n\n" - f"_Text fallback: reply `{_p}approve` to switch or `{_p}cancel` to keep " - "the current model._" - ), - handler=_on_cost_confirm, - ) - - return await commit_switch(result) + # -------------------------------------------------- /codex-runtime, /personality async def _handle_codex_runtime_command(self, event: MessageEvent) -> str: - """Handle /codex-runtime command in the gateway. - - On change the cached agent is evicted so the next message builds a fresh AIAgent with the - new api_mode (avoids prompt-cache invalidation mid-session). - """ + """Handle /codex-runtime; a real change evicts the cached agent so the new api_mode applies + on the next message (avoids prompt-cache invalidation mid-session).""" from hermes_cli import codex_runtime_switch as crs - raw_args = event.get_command_args().strip() if event else "" - new_value, errors = crs.parse_args(raw_args) + new_value, errors = crs.parse_args(event.get_command_args().strip() if event else "") if errors: return "❌ " + "\n❌ ".join(errors) - - # Load + persist via the same helpers used for /model and /yolo try: from hermes_cli.config import load_config, save_config except Exception as exc: return f"❌ Could not load config: {exc}" - cfg = load_config() - result = crs.apply( - cfg, - new_value, - persist_callback=(save_config if new_value is not None else None), + load_config(), new_value, persist_callback=(save_config if new_value is not None else None), ) - - # On a real change, evict the cached agent so the new runtime takes - # effect on the next message rather than waiting for cache TTL. if result.success and new_value is not None and result.requires_new_session: try: - session_key = self._session_key_for_source(event.source) - self._evict_cached_agent(session_key) + self._evict_cached_agent(self._session_key_for_source(event.source)) except Exception: - logger.debug("could not evict cached agent after codex-runtime change", - exc_info=True) - - prefix = "✓" if result.success else "✗" - return f"{prefix} {result.message}" + logger.debug("could not evict cached agent after codex-runtime change", exc_info=True) + return f"{'✓' if result.success else '✗'} {result.message}" async def _handle_personality_command(self, event: MessageEvent) -> str: - """Handle /personality command - list or set a personality. - - All resolution/persistence goes through hermes_cli.personality, the single owner of state. - """ + """Handle /personality — list or set a personality (hermes_cli.personality owns the state).""" from gateway.run import _load_gateway_config from hermes_cli.personality import ( active_personality_name, @@ -656,56 +527,44 @@ class GatewayModelCommandsMixin: ) args = event.get_command_args().strip() - try: config = _load_gateway_config() except Exception: config = {} personalities = available_personalities(config) - if not args: current = active_personality_name(config) - lines = [t("gateway.personality.header")] - lines.append(t("gateway.personality.none_option")) + lines = [t("gateway.personality.header"), t("gateway.personality.none_option")] for name, prompt in personalities.items(): marker = " ✓" if name == current else "" lines.append( - t( - "gateway.personality.item", - name=f"{name}{marker}", - preview=describe_personality(prompt), - ) + t("gateway.personality.item", name=f"{name}{marker}", preview=describe_personality(prompt)) ) lines.append(t("gateway.personality.usage")) return "\n".join(lines) - try: name, _new_prompt = resolve_personality(args, config) except ValueError: available = "`none`, " + ", ".join(f"`{n}`" for n in personalities) return t("gateway.personality.unknown", name=args.lower(), available=available) - - # Persist the selection only — hermes_cli.personality never writes agent.system_prompt (user- - # owned overlay). persist_personality writes get_hermes_home()/config.yaml (the routed profile - # under multiplex) and the next turn re-resolves the prompt from it: no process-global state. + # Persists the selection only (never agent.system_prompt, a user-owned overlay) into the + # routed profile's config.yaml; the next turn re-resolves the prompt — no process-global state. if not persist_personality(name): return t("gateway.personality.save_failed", error="config write failed") - if not name: return t("gateway.personality.cleared") return t("gateway.personality.set_to", name=name) + # ----------------------------------------------------------- /reasoning, /fast + def _save_gateway_config_key(self, key_path: str, value) -> bool: - """Save a dot-separated key to config.yaml (shared by /reasoning, /fast - and their interactive pickers).""" + """Save a dot-separated key to config.yaml (shared by /reasoning, /fast and their pickers).""" from gateway.slash_commands import _nested_dict from gateway.run import _gateway_config_home from hermes_cli.config import read_user_config_raw config_path = _gateway_config_home() / "config.yaml" try: - # Write-back round-trip: raw read is correct (merged defaults must - # not be persisted back to the user's file). - user_config = read_user_config_raw(config_path) + user_config = read_user_config_raw(config_path) # raw: never persist merged defaults *parents, leaf = key_path.split(".") _nested_dict(user_config, *parents)[leaf] = value atomic_config_write(config_path, user_config) @@ -714,35 +573,24 @@ class GatewayModelCommandsMixin: logger.error("Failed to save config key %s: %s", key_path, e) return False - def _apply_reasoning_selection( - self, - session_key: str, - platform_key: str, - value: str, - persist_global: bool = False, - ) -> str: - """Apply a /reasoning argument (typed or picked) and return the reply. + def _set_reasoning_override(self, session_key: str, value) -> None: + """Store (or clear with None) the session reasoning override and drop the cached agent.""" + self._set_session_reasoning_override(session_key, value) + self._evict_cached_agent(session_key) - Single path shared by `/reasoning ` and the choice picker so both match the parser. - """ + def _apply_reasoning_selection( + self, session_key: str, platform_key: str, value: str, persist_global: bool = False, + ) -> str: + """Apply a /reasoning argument (typed or picked) and return the reply.""" from hermes_constants import parse_reasoning_effort value = (value or "").strip().lower() - - # Display toggle (per-platform) - if value in {"show", "on"}: - self._show_reasoning = True - self._save_gateway_config_key( - f"display.platforms.{platform_key}.show_reasoning", True - ) - return t("gateway.reasoning.display_set_on", platform=platform_key) - if value in {"hide", "off"}: - self._show_reasoning = False - self._save_gateway_config_key( - f"display.platforms.{platform_key}.show_reasoning", False - ) - return t("gateway.reasoning.display_set_off", platform=platform_key) - + show = _REASONING_DISPLAY_TOGGLES.get(value) + if show is not None: # per-platform display toggle + self._show_reasoning = show + self._save_gateway_config_key(f"display.platforms.{platform_key}.show_reasoning", show) + key = "gateway.reasoning.display_set_on" if show else "gateway.reasoning.display_set_off" + return t(key, platform=platform_key) if value == "reset": if persist_global: return t("gateway.reasoning.reset_global_unsupported") @@ -754,62 +602,28 @@ class GatewayModelCommandsMixin: parsed = parse_reasoning_effort(value) if parsed is None: return t("gateway.reasoning.unknown_arg", arg=value) - self._reasoning_config = parsed if persist_global: if self._save_gateway_config_key("agent.reasoning_effort", value): - self._set_session_reasoning_override(session_key, None) - self._evict_cached_agent(session_key) + self._set_reasoning_override(session_key, None) return t("gateway.reasoning.set_global", effort=value) - self._set_session_reasoning_override(session_key, parsed) - self._evict_cached_agent(session_key) + self._set_reasoning_override(session_key, parsed) return t("gateway.reasoning.set_global_save_failed", effort=value) - - self._set_session_reasoning_override(session_key, parsed) - self._evict_cached_agent(session_key) + self._set_reasoning_override(session_key, parsed) return t("gateway.reasoning.set_session", effort=value) - def _reasoning_picker_choices(self, current_effort: str) -> list: - """Build the choice list for the interactive /reasoning picker.""" - from hermes_constants import VALID_REASONING_EFFORTS - - choices = [{"value": "none", "label": t("gateway.reasoning.choice_none"), "is_current": current_effort == "none"}] - choices.extend({"value": level, "label": level, "is_current": level == current_effort} for level in VALID_REASONING_EFFORTS) - choices.extend( - {"value": v, "label": t(f"gateway.reasoning.choice_{v}"), "is_current": False} - for v in ("reset", "show", "hide") - ) - return choices - async def _try_send_choice_picker( - self, - event: MessageEvent, - session_key: str, - title: str, - choices: list, - on_choice_selected, + self, event: MessageEvent, session_key: str, title: str, choices: list, on_choice_selected, ) -> bool: - """Send an interactive choice picker when the platform supports it. - - Mirrors the `/model` gate: capability is detected on the adapter *type* - (``send_choice_picker``); a failed send returns False (text fallback) instead of erroring. - """ + """Send an interactive choice picker when the adapter *type* supports it (the /model gate); + a failed send returns False (text fallback) instead of erroring.""" adapter = self._adapter_for_source(event.source) - has_picker = ( - adapter is not None - and getattr(type(adapter), "send_choice_picker", None) is not None - ) - if not has_picker: + if adapter is None or getattr(type(adapter), "send_choice_picker", None) is None: return False try: - metadata = self._reply_metadata(event) result = await adapter.send_choice_picker( - chat_id=event.source.chat_id, - title=title, - choices=choices, - session_key=session_key, - on_choice_selected=on_choice_selected, - metadata=metadata, + chat_id=event.source.chat_id, title=title, choices=choices, session_key=session_key, + on_choice_selected=on_choice_selected, metadata=self._reply_metadata(event), ) return bool(getattr(result, "success", False)) except Exception as e: @@ -819,163 +633,101 @@ class GatewayModelCommandsMixin: async def _handle_reasoning_command(self, event: MessageEvent) -> Optional[str]: """Handle /reasoning command — manage reasoning effort and display toggle.""" from gateway.run import _platform_config_key + from hermes_constants import VALID_REASONING_EFFORTS raw_args = event.get_command_args().strip() args, persist_global = self._parse_reasoning_command_args(raw_args) - # Normalize the source (Telegram DM topic recovery) before deriving - # the override key so storage matches the key the next message turn - # reads — same fix as /model (#30479). + # Normalize (Telegram DM topic recovery) so the override key matches the next turn's. _reasoning_source = await asyncio.to_thread(self._normalize_source_for_session_key, event.source) session_key = self._session_key_for_source(_reasoning_source) self._show_reasoning = self._load_show_reasoning() - # Use the session's effective model (session /model override wins over - # config default) so per-model reasoning_overrides display correctly. + # Effective model (session /model override wins) so per-model reasoning_overrides display. _session_model = str( ((getattr(self, "_session_model_overrides", {}) or {}).get(session_key) or {}).get("model") or "" ) self._reasoning_config = self._resolve_session_reasoning_config( - source=event.source, - session_key=session_key, - model=_session_model, + source=event.source, session_key=session_key, model=_session_model, ) - - if not raw_args: - # Show current state - rc = self._reasoning_config - if rc is None: - level = t("gateway.reasoning.level_default") - current_effort = "medium" - elif rc.get("enabled") is False: - level = t("gateway.reasoning.level_disabled") - current_effort = "none" - else: - level = rc.get("effort", "medium") - current_effort = level - display_state = ( - t("gateway.reasoning.display_on") - if self._show_reasoning - else t("gateway.reasoning.display_off") - ) - has_session_override = session_key in (getattr(self, "_session_reasoning_overrides", {}) or {}) - scope = ( - t("gateway.reasoning.scope_session") - if has_session_override - else t("gateway.reasoning.scope_global") - ) - - # Interactive picker on platforms that support it (parity with the - # /model picker). Falls through to the text status card otherwise. - _picker_platform_key = _platform_config_key(event.source.platform) - - async def _on_reasoning_choice(_chat_id: str, value: str) -> str: - return self._apply_reasoning_selection( - session_key, _picker_platform_key, value - ) - - picker_sent = await self._try_send_choice_picker( - event, - session_key, - title=t( - "gateway.reasoning.picker_title", - level=level, - scope=scope, - display=display_state, - ), - choices=self._reasoning_picker_choices(current_effort), - on_choice_selected=_on_reasoning_choice, - ) - if picker_sent: - return None # Picker sent — adapter handles the response - - return t( - "gateway.reasoning.status", - level=level, - scope=scope, - display=display_state, - ) - - # Typed argument path — same applier the picker uses. platform_key = _platform_config_key(event.source.platform) - return self._apply_reasoning_selection( - session_key, platform_key, args, persist_global=persist_global + if raw_args: # typed path — same applier the picker uses + return self._apply_reasoning_selection(session_key, platform_key, args, persist_global=persist_global) + rc = self._reasoning_config + if rc is None: + level, current_effort = t("gateway.reasoning.level_default"), "medium" + elif rc.get("enabled") is False: + level, current_effort = t("gateway.reasoning.level_disabled"), "none" + else: + level = current_effort = rc.get("effort", "medium") + display_state = t("gateway.reasoning.display_on") if self._show_reasoning else t("gateway.reasoning.display_off") + has_session_override = session_key in (getattr(self, "_session_reasoning_overrides", {}) or {}) + scope = t("gateway.reasoning.scope_session") if has_session_override else t("gateway.reasoning.scope_global") + + async def _on_reasoning_choice(_chat_id: str, value: str) -> str: + return self._apply_reasoning_selection(session_key, platform_key, value) + + picker_sent = await self._try_send_choice_picker( + event, + session_key, + title=t("gateway.reasoning.picker_title", level=level, scope=scope, display=display_state), + choices=[ + {"value": "none", "label": t("gateway.reasoning.choice_none"), "is_current": current_effort == "none"}, + *({"value": lv, "label": lv, "is_current": lv == current_effort} for lv in VALID_REASONING_EFFORTS), + *({"value": v, "label": t(f"gateway.reasoning.choice_{v}"), "is_current": False} + for v in ("reset", "show", "hide")), + ], + on_choice_selected=_on_reasoning_choice, ) + if picker_sent: + return None # Picker sent — adapter handles the response + return t("gateway.reasoning.status", level=level, scope=scope, display=display_state) + + def _apply_fast_selection(self, session_key: str, value: str, persist: bool = False) -> str: + """Apply a /fast argument (typed or picked) and return the reply.""" + selection = _FAST_SELECTIONS.get(value) + if selection is None: + return t("gateway.fast.unknown_arg", arg=value) + tier, saved_value, label_key = selection + label = t(label_key) if label_key else value.upper() + self._service_tier = tier + if persist and self._save_gateway_config_key("agent.service_tier", saved_value): + self._set_session_service_tier_override(session_key, None, clear=True) # global wins + self._evict_cached_agent(session_key) + return t("gateway.fast.saved", label=label) + # Session override — also the fallback after a failed config write (as /reasoning --global). + self._set_session_service_tier_override(session_key, tier) + self._evict_cached_agent(session_key) + return t("gateway.fast.session_only", label=label) async def _handle_fast_command(self, event: MessageEvent) -> Optional[str]: - """Handle /fast — mirror the CLI Priority Processing toggle in gateway chats. - - Session-scoped by default; ``--global`` persists agent.service_tier (parity with /model). - """ + """Handle /fast — the CLI Priority Processing toggle; session-scoped unless ``--global`` + (persists agent.service_tier, parity with /model).""" from gateway.run import _load_gateway_config, _resolve_gateway_model from hermes_cli.models import model_supports_fast_mode - raw_args = event.get_command_args().strip().lower() - # Reuse the /reasoning arg parser: strips --global (any position), - # normalizes unicode dashes. - args, persist_global = self._parse_reasoning_command_args(raw_args) + # The /reasoning parser strips --global (any position) and normalizes unicode dashes. + args, persist_global = self._parse_reasoning_command_args(event.get_command_args().strip().lower()) session_key = self._session_key_for_source(event.source) - self._service_tier = self._resolve_session_service_tier( - session_key=session_key - ) - - user_config = _load_gateway_config() - model = _resolve_gateway_model(user_config) - if not model_supports_fast_mode(model): + self._service_tier = self._resolve_session_service_tier(session_key=session_key) + if not model_supports_fast_mode(_resolve_gateway_model(_load_gateway_config())): return t("gateway.fast.not_supported") + if args and args != "status": + return self._apply_fast_selection(session_key, args, persist=persist_global) + mode = "fast" if self._service_tier == "priority" else (self._service_tier or "normal") + status = {"fast": t("gateway.fast.status_fast"), "normal": t("gateway.fast.status_normal")}.get(mode, mode) - def _apply_fast_selection(value: str, persist: bool = False) -> str: - """Apply a /fast argument (typed or picked) and return the reply.""" - if value in {"fast", "on"}: - tier = "priority" - saved_value = "fast" - label = t("gateway.fast.label_fast") - elif value in {"normal", "off"}: - tier = None - saved_value = "normal" - label = t("gateway.fast.label_normal") - elif value in {"auto", "cold"}: - tier = saved_value = value - label = value.upper() - else: - return t("gateway.fast.unknown_arg", arg=value) - self._service_tier = tier - if persist: - if self._save_gateway_config_key("agent.service_tier", saved_value): - # Global write supersedes any session override. - self._set_session_service_tier_override( - session_key, None, clear=True - ) - self._evict_cached_agent(session_key) - return t("gateway.fast.saved", label=label) - # Config write failed — fall back to a session override so the - # user's choice still applies (mirrors /reasoning --global). - self._set_session_service_tier_override(session_key, tier) - self._evict_cached_agent(session_key) - return t("gateway.fast.session_only", label=label) - self._set_session_service_tier_override(session_key, tier) - self._evict_cached_agent(session_key) - return t("gateway.fast.session_only", label=label) + async def _on_fast_choice(_chat_id: str, value: str) -> str: + return self._apply_fast_selection(session_key, value, persist=persist_global) - if not args or args == "status": - is_fast = self._service_tier == "priority" - mode = "fast" if is_fast else (self._service_tier or "normal") - status = {"fast": t("gateway.fast.status_fast"), "normal": t("gateway.fast.status_normal")}.get(mode, mode) - - async def _on_fast_choice(_chat_id: str, value: str) -> str: - return _apply_fast_selection(value, persist=persist_global) - - picker_sent = await self._try_send_choice_picker( - event, - session_key, - title=t("gateway.fast.picker_title", mode=status), - choices=[ - {"value": v, "label": t(f"gateway.fast.choice_{v}"), "is_current": mode == v} - for v in ("fast", "normal", "auto", "cold") - ], - on_choice_selected=_on_fast_choice, - ) - if picker_sent: - return None # Picker sent — adapter handles the response - - return t("gateway.fast.status", mode=status) - - return _apply_fast_selection(args, persist=persist_global) + picker_sent = await self._try_send_choice_picker( + event, + session_key, + title=t("gateway.fast.picker_title", mode=status), + choices=[ + {"value": v, "label": t(f"gateway.fast.choice_{v}"), "is_current": mode == v} + for v in ("fast", "normal", "auto", "cold") + ], + on_choice_selected=_on_fast_choice, + ) + if picker_sent: + return None # Picker sent — adapter handles the response + return t("gateway.fast.status", mode=status) diff --git a/gateway/slash_commands_session.py b/gateway/slash_commands_session.py index 6283161176..7591d3cacd 100644 --- a/gateway/slash_commands_session.py +++ b/gateway/slash_commands_session.py @@ -2,16 +2,16 @@ /new, /resume, /sessions, /branch, /title, /save, /undo, /retry, /topic, /compress. Split out of ``gateway/slash_commands.py``; bound onto ``GatewayRunner`` through -``GatewaySlashCommandsMixin``. Origin internals are imported lazily (``from gateway.slash_commands -import ...``) inside the bodies to avoid the import cycle. +``GatewaySlashCommandsMixin``. Origin internals are imported lazily inside the bodies to avoid +the import cycle. """ from __future__ import annotations -import logging import asyncio import contextlib import dataclasses +import logging import os import shlex from typing import Optional, Union @@ -22,22 +22,31 @@ from gateway.config import Platform from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType from gateway.session import SessionSource, build_session_key, is_shared_multi_user_session -# Log-record parity with gateway/run.py and the origin module. -logger = logging.getLogger("gateway.run") +logger = logging.getLogger("gateway.run") # log-record parity with gateway/run.py -# Upper bound on the off-loop agent-resource cleanup during a /new or /reset (see -# _handle_reset_command). A stuck teardown must not block the event loop; past this the reset -# proceeds and the cleanup is left to finish (or leak) in its worker thread. +# Bound on the off-loop agent cleanup during /new; past it the reset proceeds and the teardown +# finishes (or leaks) in its worker thread rather than blocking the event loop. _RESET_CLEANUP_TIMEOUT_S = 30.0 +# chat_type values whose session key is per-user (DM-like), incl. the unknown/blank case. +_DM_CHAT_TYPES = {"dm", "direct", "private", ""} + +_BRANCH_COPIED_FIELDS = ( + "content", "tool_calls", "tool_call_id", "finish_reason", "reasoning", "reasoning_content", + "reasoning_details", "codex_reasoning_items", "codex_message_items", "timestamp", +) + + +def _sattr(obj, name: str) -> str: + """``str(getattr(obj, name) or "")`` — normalized identity field for origin comparisons.""" + return str(getattr(obj, name, "") or "") def _manual_compression_reply_lines(summary: dict, compressor, focus_topic) -> list[str]: - """Lines for the manual /compress confirmation, surfacing summariser/aux-model failures. + """Manual /compress confirmation lines, surfacing summariser/aux-model failures. - ``_last_compress_aborted`` = no usable summary, messages unchanged (force=True bypasses any - cooldown). Provider exception text is force-redacted at this UI boundary even when global - redaction is off. A configured aux model that failed and was recovered via main is an info - note so the user can fix their config. + ``_last_compress_aborted`` = no usable summary, messages unchanged. Provider exception text is + force-redacted at this UI boundary even when global redaction is off; an aux model recovered + via main is an info note so the user can fix their config. """ lines = [f"🗜️ {summary['headline']}"] if focus_topic: @@ -53,13 +62,11 @@ def _manual_compression_reply_lines(summary: dict, compressor, focus_topic) -> l if getattr(compressor, "_last_compress_aborted", False): lines.append(t("gateway.compress.aborted", error=(summary_err or "unknown error"))) elif aux_fail_model: - lines.append( - t( - "gateway.compress.aux_failed", - model=aux_fail_model, - error=(getattr(compressor, "_last_aux_model_failure_error", None) or "unknown error"), - ) - ) + lines.append(t( + "gateway.compress.aux_failed", + model=aux_fail_model, + error=(getattr(compressor, "_last_aux_model_failure_error", None) or "unknown error"), + )) return lines @@ -84,26 +91,17 @@ def _compress_preview_reply(history, partial: bool, keep_last, focus_topic, agg_ def _reset_process_scoped_tool_state() -> None: """Drop env-passthrough and credential-file state at a conversation boundary (best-effort).""" - try: + with contextlib.suppress(Exception): from tools.env_passthrough import clear_env_passthrough clear_env_passthrough() - except Exception: - pass - try: + with contextlib.suppress(Exception): from tools.credential_files import clear_credential_files clear_credential_files() - except Exception: - pass - -_BRANCH_COPIED_FIELDS = ( - "content", "tool_calls", "tool_call_id", "finish_reason", "reasoning", "reasoning_content", - "reasoning_details", "codex_reasoning_items", "codex_message_items", "timestamp", -) def _branch_row(msg: dict) -> dict: - """Transcript row copied into a /branch child. Keeps the api_content sidecar so the branch's - first turn replays the parent's exact wire bytes (warm provider prompt cache), not a cold prefill.""" + """/branch child row; keeps the api_content sidecar so the branch's first turn replays the + parent's exact wire bytes (warm provider prompt cache) instead of a cold prefill.""" row = {k: msg.get(k) for k in _BRANCH_COPIED_FIELDS} row["role"] = msg.get("role", "user") row["tool_name"] = msg.get("tool_name") or msg.get("name") @@ -111,64 +109,94 @@ def _branch_row(msg: dict) -> dict: return row +def _strip_resume_name(parts: list[str]) -> str: + """Join the non-flag /resume tokens; strip literal ``<...>``/``[...]``/quotes typed from the + usage hint (mirrors the CLI).""" + name = " ".join(p for p in parts if p not in {"--all", "--cross-room"}).strip() + if len(name) >= 2 and (name[0], name[-1]) in {("<", ">"), ("[", "]"), ('"', '"'), ("'", "'")}: + name = name[1:-1].strip() + return name + + class GatewaySessionCommandsMixin: """Session-transcript slash commands (/new, /resume, /sessions, /branch, /title, /save, /undo, /retry, /topic, /compress).""" + # ------------------------------------------------------------------ /new, /reset + + async def _cleanup_old_agent_for_reset(self, session_key: str) -> None: + """Close the old agent's tool resources (sandboxes, browsers, subprocesses) before eviction. + + Blocking work on the event loop (confirm-button click) → offloaded with a bounded timeout. + wait_for cancels the await, not the worker thread: a wedged teardown keeps running (or + leaks); the reset proceeds either way. + """ + _old_agent = self._cached_agent_for(session_key) + if _old_agent is None: + return + try: + await asyncio.wait_for( + self._run_in_executor_with_context(self._cleanup_agent_resources, _old_agent), + timeout=_RESET_CLEANUP_TIMEOUT_S, + ) + except asyncio.TimeoutError: + logger.warning( + "Agent resource cleanup for session %s exceeded %ss during /new reset; proceeding with " + "reset (the worker thread is left to finish on its own). (#35994)", + session_key, _RESET_CLEANUP_TIMEOUT_S, + ) + except Exception as cleanup_exc: + logger.warning( + "Agent resource cleanup for session %s failed during /new reset: %s (#35994)", + session_key, cleanup_exc, + ) + + async def _fire_session_reset_hooks( + self, source: SessionSource, session_key: str, old_sid, new_sid + ) -> None: + """Session-boundary hooks: plugin finalize (off-loop + bounded — trace exports can block + arbitrarily), then session:end and session:reset.""" + platform_value = source.platform.value if source.platform else "" + with contextlib.suppress(Exception): + await self._finalize_session_off_loop( + session_id=old_sid, platform=platform_value, reason="new_session", + old_session_id=old_sid, new_session_id=new_sid, + ) + hook_payload = {"platform": platform_value, "user_id": source.user_id, "session_key": session_key} + await self.hooks.emit("session:end", dict(hook_payload)) + await self.hooks.emit("session:reset", dict(hook_payload)) + + def _invoke_session_reset_lifecycle_hook(self, source: SessionSource, old_sid, new_sid) -> None: + """Plugin on_session_reset hook (new session guaranteed to exist); best-effort.""" + try: + from hermes_cli.lifecycle import invoke_hook as _invoke_hook + _invoke_hook( + "on_session_reset", session_id=new_sid, + platform=source.platform.value if source.platform else "", reason="new_session", + old_session_id=old_sid, new_session_id=new_sid, + ) + except Exception: + pass + async def _handle_reset_command(self, event: MessageEvent) -> Union[str, EphemeralReply]: """Handle /new or /reset command.""" source = event.source - - # Get existing session key session_key = self._session_key_for_source(source) self._invalidate_session_run_generation(session_key, reason="session_reset") # Evict the running-agent slot now that the generation is bumped: the in-flight run's own # guarded release (old generation) returns False and would leave a zombie slot that silently # drops all later messages. Idempotent, so the run's finally calling it again is harmless. self._release_running_agent_state(session_key) - - # Snapshot the old entry so on_session_finalize can report the - # expiring session id before reset_session() rotates it. + # Snapshot the old entry so on_session_finalize can report the expiring session id. old_entry = self.session_store._entries.get(session_key) - - # Close the old agent's tool resources (sandboxes, browser daemons, subprocesses) before - # evicting it; getattr-guarded since test fixtures may skip __init__. _cleanup_agent_resources - # is blocking and this handler runs ON the event loop (confirm-button click), so an inline - # call wedges the loop — offload to a worker thread with a bounded timeout. - _old_agent = self._cached_agent_for(session_key) - if _old_agent is not None: - try: - await asyncio.wait_for( - self._run_in_executor_with_context(self._cleanup_agent_resources, _old_agent), - timeout=_RESET_CLEANUP_TIMEOUT_S, - ) - except asyncio.TimeoutError: - # wait_for cancels the await, but the worker thread cannot be cancelled — a wedged - # teardown keeps running (or leaks) for the gateway's lifetime. The reset proceeds. - logger.warning( - "Agent resource cleanup for session %s exceeded %ss during " - "/new reset; proceeding with reset (the worker thread is left " - "to finish on its own). (#35994)", - session_key, _RESET_CLEANUP_TIMEOUT_S, - ) - except Exception as cleanup_exc: - logger.warning( - "Agent resource cleanup for session %s failed during /new " - "reset: %s (#35994)", - session_key, cleanup_exc, - ) + await self._cleanup_old_agent_for_reset(session_key) self._evict_cached_agent(session_key) - - # Conversation boundary: clear ALL conversation-scoped per-session state (model/reasoning - # overrides, one-turn restores, model notes, last-resolved cache, /queue overflow) + - # security state in one funnel call. See _CONVERSATION_SCOPED_STATE in gateway/run.py. + # Conversation boundary: ALL conversation-scoped per-session state + security state in one + # funnel call (see _CONVERSATION_SCOPED_STATE in gateway/run.py). self._clear_conversation_scope(session_key, reason="session_reset") - - # The old conversation's in-flight async delegations end WITH it: once the session id rotates - # their completions have no live owner (orphaned payload on the shared queue, wasted tokens). - # Interrupt by expiring durable session id (parent_session_id), routing key as legacy fallback. + # In-flight async delegations end WITH the conversation: once the id rotates their + # completions have no live owner. Expire by durable id, routing key as legacy fallback. try: from tools.async_delegation import interrupt_for_session - interrupt_for_session( session_key=session_key, parent_session_id=str(getattr(old_entry, "session_id", "") or ""), @@ -178,91 +206,46 @@ class GatewaySessionCommandsMixin: pass _reset_process_scoped_tool_state() - # Reset the session new_entry = await self.async_session_store.reset_session(session_key) - - # (Conversation-scoped overrides + security state were already - # cleared via _clear_conversation_scope above.) - _old_sid = old_entry.session_id if old_entry else None - platform_value = source.platform.value if source.platform else "" - - # Fire plugin on_session_finalize hook (session boundary). Off-loop + bounded: finalize - # hooks can block arbitrarily (observability trace exports) and this handler runs on the - # gateway event loop (see GatewayRunner._finalize_session_off_loop). - with contextlib.suppress(Exception): - await self._finalize_session_off_loop( - session_id=_old_sid, - platform=platform_value, - reason="new_session", - old_session_id=_old_sid, - new_session_id=new_entry.session_id if new_entry else None, - ) - - # Emit session:end (session is ending) then session:reset hooks. - hook_payload = {"platform": platform_value, "user_id": source.user_id, "session_key": session_key} - await self.hooks.emit("session:end", dict(hook_payload)) - await self.hooks.emit("session:reset", dict(hook_payload)) - - # Resolve session config info to surface to the user, scoped to the - # profile serving this source so a multiplexed /reset //new banner - # reports the profile's model, not the base config's (#59003). + await self._fire_session_reset_hooks( + source, session_key, _old_sid, new_entry.session_id if new_entry else None + ) + # Scoped to the profile serving this source so a multiplexed /new banner reports the + # profile's model, not the base config's. try: - session_info = await asyncio.to_thread( - self._reset_notice_session_info, source - ) + session_info = await asyncio.to_thread(self._reset_notice_session_info, source) except Exception: session_info = "" - if new_entry: - header = await asyncio.to_thread(self._telegram_topic_new_header, source) or t("gateway.reset.header_default") - else: - # No existing session, just create one + default_header = t("gateway.reset.header_default") + else: # no existing session: create one new_entry = await self.async_session_store.get_or_create_session(source, force_new=True) - header = await asyncio.to_thread(self._telegram_topic_new_header, source) or t("gateway.reset.header_new") - - # Set session title if provided with /new + default_header = t("gateway.reset.header_new") + header = await asyncio.to_thread(self._telegram_topic_new_header, source) or default_header _title_arg = event.get_command_args().strip() if _title_arg and self._session_db and new_entry: header = await self._reset_titled_header(header, new_entry.session_id, _title_arg) - - # When /new runs inside a Telegram DM topic lane, rewrite the (chat_id, thread_id) → - # session_id binding so the next message uses the freshly-created session. Otherwise the - # binding-lookup at the top of _handle_message_with_agent switches right back to the old one. + # Telegram DM topic lane: rebind (chat_id, thread_id) → session_id so the next message uses + # the fresh session instead of switching back to the old one. if await asyncio.to_thread(self._is_telegram_topic_lane, source) and new_entry is not None: try: await asyncio.to_thread(self._record_telegram_topic_binding, source, new_entry) except Exception: logger.debug("Failed to rebind Telegram topic after /new", exc_info=True) - - # Fire plugin on_session_reset hook (new session guaranteed to exist) - try: - from hermes_cli.lifecycle import invoke_hook as _invoke_hook - _new_sid = new_entry.session_id if new_entry else None - _invoke_hook( - "on_session_reset", - session_id=_new_sid, - platform=platform_value, - reason="new_session", - old_session_id=_old_sid, - new_session_id=_new_sid, - ) - except Exception: - pass - - # Append a random tip to the reset message + self._invoke_session_reset_lifecycle_hook( + source, _old_sid, new_entry.session_id if new_entry else None + ) try: from hermes_cli.tips import get_random_tip _tip_line = t("gateway.reset.tip", tip=get_random_tip()) except Exception: _tip_line = "" - - if session_info: - return EphemeralReply(f"{header}\n\n{session_info}{_tip_line}") - return EphemeralReply(f"{header}{_tip_line}") + body = f"{header}\n\n{session_info}" if session_info else header + return EphemeralReply(f"{body}{_tip_line}") async def _reset_titled_header(self, header: str, session_id: str, title_arg: str) -> str: - """Apply ``/new <title>``: titled header on success, else the header plus a rejection note.""" + """``/new <title>``: titled header on success, else the header plus a rejection note.""" from hermes_state import SessionDB note = "" try: @@ -278,107 +261,80 @@ class GatewaySessionCommandsMixin: note = t("gateway.reset.title_error_untitled", error=str(e)) except Exception: pass - elif not note: - # sanitize_title returned empty (whitespace-only / unprintable) + elif not note: # sanitize_title returned empty (whitespace-only / unprintable) note = t("gateway.reset.title_empty_untitled") return header + note + # ------------------------------------------------------- origin / ownership guards + def _gateway_session_origin_for_id(self, session_id: str) -> Optional[SessionSource]: """Best-effort origin lookup for gateway session IDs.""" lookup = getattr(type(self.session_store), "lookup_by_session_id", None) if callable(lookup): entry = lookup(self.session_store, session_id) return getattr(entry, "origin", None) if entry is not None else None - - # Test doubles and older stores may not expose the public lookup helper. - # Keep the Matrix resume guard fail-closed if no origin can be resolved. + # Test doubles / older stores lack the public lookup; fail closed when nothing resolves. entries = getattr(self.session_store, "_entries", {}) or {} - for entry in entries.values(): - if getattr(entry, "session_id", None) == session_id: - return getattr(entry, "origin", None) - return None + return next( + (getattr(e, "origin", None) for e in entries.values() if getattr(e, "session_id", None) == session_id), + None, + ) @staticmethod def _same_matrix_room(current: SessionSource, origin: Optional[SessionSource]) -> bool: + # thread_id is part of the session key, so another thread of the SAME room is a DIFFERENT + # session; non-threaded rooms compare "" == "". return ( origin is not None and origin.platform == Platform.MATRIX and current.platform == Platform.MATRIX and origin.chat_id == current.chat_id - # thread_id is part of the session key (build_session_key appends it for every chat - # type when present) and Matrix scopes a turn to the current room/thread, so a live - # session in another thread of the SAME room is a DIFFERENT session: thread A must not - # resume/enumerate a target from thread B. Non-threaded rooms compare "" == "" unchanged. - and str(getattr(current, "thread_id", "") or "") - == str(getattr(origin, "thread_id", "") or "") + and _sattr(current, "thread_id") == _sattr(origin, "thread_id") ) def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSource]) -> bool: """Platform-agnostic counterpart to ``_same_matrix_room``. - Per-participant sessions (``build_session_key`` with the default ``group_sessions_per_user``) - must be participant-scoped here too, else a co-member could resume another member's live - session (IDOR). Only an explicitly shared group/thread (``is_shared_multi_user_session``) shares. + Per-participant sessions must be participant-scoped here too, else a co-member could resume + another member's live session (IDOR); only an explicitly shared group/thread shares. """ if origin is None or current is None: return False - if origin.platform != current.platform: + if origin.platform != current.platform or origin.chat_id != current.chat_id: return False - if origin.chat_id != current.chat_id: + # thread_id is part of every session key: threads of one chat are DIFFERENT sessions. + if _sattr(current, "thread_id") != _sattr(origin, "thread_id"): return False - # thread_id is part of the session key for every chat type (build_session_key appends it - # unconditionally), so threads of the same parent chat are DIFFERENT sessions. - # is_shared_multi_user_session only decides sharing WITHIN a thread — require thread equality - # before any sharing logic so a live origin in thread A cannot match a caller in thread B. - if str(getattr(current, "thread_id", "") or "") != str( - getattr(origin, "thread_id", "") or "" - ): - return False - chat_type = (getattr(current, "chat_type", "") or "").lower() - # DM-like chats are always per-user. - if chat_type in {"dm", "direct", "private", ""}: - # chat_id was already required equal above and, when present, IS the DM session key, so - # an equal non-empty chat_id suffices. build_session_key falls back to the participant - # (``user_id_alt or user_id`` — Signal/Feishu key on user_id_alt) only when there is NO - # chat_id; mirror that and fail closed on a missing/different participant so two - # no-chat_id DM origins are never conflated. - if str(getattr(current, "chat_id", "") or ""): + if _sattr(current, "chat_type").lower() in _DM_CHAT_TYPES: + # An equal non-empty chat_id IS the DM key. build_session_key falls back to the + # participant (``user_id_alt or user_id`` — Signal/Feishu key on alt) only without a + # chat_id; mirror that and fail closed on a missing/different participant. + if _sattr(current, "chat_id"): return True cur_pid = str(current.user_id_alt or current.user_id or "") org_pid = str(origin.user_id_alt or origin.user_id or "") return bool(cur_pid) and cur_pid == org_pid - # Non-DM: scope by participant whenever the session key for this source - # is per-user. is_shared_multi_user_session mirrors build_session_key's - # isolation rules exactly, so the guard stays in lock-step with the key. + # Non-DM: a shared key is one session for everyone; a per-user key compares the participant + # it is built from, failing closed when either side lacks one. if self._is_shared_session_source(current): return True - # Per-user key: compare the participant id the key is actually built - # from (user_id_alt or user_id — Signal/Feishu key on user_id_alt). cur_pid = current.user_id_alt or current.user_id org_pid = origin.user_id_alt or origin.user_id - if cur_pid and org_pid: - return cur_pid == org_pid - # Per-user key but a participant id is missing on one side: cannot prove - # the same owner — fail closed. - return False + return bool(cur_pid and org_pid) and cur_pid == org_pid def _is_shared_session_source(self, source: SessionSource) -> bool: - """Whether *source*'s session key is shared by every participant (not per-user). - - Mirrors build_session_key's isolation rules exactly, so the guards stay in lock-step with the key. - """ + """Whether *source*'s session key is shared by every participant (not per-user); mirrors + build_session_key's isolation rules so the guards stay in lock-step with the key.""" return is_shared_multi_user_session( - source, - group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), + source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), ) def _resume_caller_is_admin(self, source: SessionSource) -> bool: - """Whether *source* is an EXPLICITLY-configured admin allowed cross-origin /resume or /sessions. + """Whether *source* is an EXPLICITLY-configured admin (cross-origin /resume, /sessions). - Stricter than ``SlashAccessPolicy.is_admin()``, which returns True for every allowed caller - when slash gating is DISABLED; cross-origin DATA ACCESS needs a real configured admin, else - the default (no admin list) config would make every caller cross-origin-capable (IDOR). + Stricter than ``SlashAccessPolicy.is_admin()``, which is True for every caller when slash + gating is DISABLED — the default config would make everyone cross-origin-capable (IDOR). """ try: from gateway.slash_access import policy_for_source @@ -388,124 +344,83 @@ class GatewaySessionCommandsMixin: except Exception: return False + def _persisted_row_proves_owner(self, source: SessionSource, row: dict) -> bool: + """Whether a persisted (inactive) session *row* provably belongs to *source*'s session key. + + Rows once stored only source + user_id, so the persisted chat/thread origin is compared too + and legacy NULL rows fail closed. The table has no user_id_alt column, so an alt-keyed + (Signal/Feishu) caller is never proven by user_id alone (CWE-639). + """ + caller_src = source.platform.value if source.platform else None + row_src = row.get("source") + caller_uid = _sattr(source, "user_id") + if not caller_uid: + return False + row_thread = str(row.get("thread_id") or "") + if not (row_src and caller_src and str(row_src) == str(caller_src) + and row_thread == _sattr(source, "thread_id")): + return False # blank/legacy source cannot prove the platform; other thread = other session + row_uid = str(row.get("user_id") or "") + row_chat = str(row.get("chat_id") or "") + caller_chat = _sattr(source, "chat_id") + caller_keys_on_alt = bool(_sattr(source, "user_id_alt")) + if _sattr(source, "chat_type").lower() in _DM_CHAT_TYPES: + # A no-chat_id DM is keyed PURELY on the participant (alt-keyed caller fails closed). + if caller_keys_on_alt and not (row_chat and caller_chat): + return False + return bool(row_uid) and row_uid == caller_uid and row_chat == caller_chat + # Non-DM: both sides must carry chat_id and match (legacy NULL-chat rows fail closed). + if not (row_chat and caller_chat and row_chat == caller_chat): + return False + # A SHARED group/thread is one session for everyone: same-chat proof suffices (a user-id + # check would block co-members). A per-user key still requires the same owner. + if self._is_shared_session_source(source): + return True + if caller_keys_on_alt: + return False + return bool(row_uid) and row_uid == caller_uid + async def _resume_target_allowed( self, source: SessionSource, target_id: str, allow_override: bool = False ) -> bool: - """Whether *source* may resume the persisted session *target_id*. + """Whether *source* may resume session *target_id* (IDOR guard for every adapter). - Generalizes the Matrix-only room guard to every adapter so a caller cannot bind to another - user's/room's session (IDOR). Uses the live origin when the target is active, else the DB - row's source + user_id; the row must PROVE ownership or fail closed. Admin ``--all`` bypasses. + The live origin decides when the target is active; otherwise the DB row must PROVE + ownership or fail closed. Admin ``--all`` bypasses. """ if allow_override and self._resume_caller_is_admin(source): return True - # Use the live origin only when it resolves to a real SessionSource; a - # store that can't resolve it (or an unexpected lookup error) must not - # silently allow/deny — fall through to the deterministic DB scoping. + # Only a real SessionSource origin decides; unresolvable/error falls through to DB scoping. try: origin = self._gateway_session_origin_for_id(target_id) except Exception: origin = None if isinstance(origin, SessionSource): return self._same_origin_chat(source, origin) - # Inactive/persisted-only: best-effort scope by DB row source + user. try: row = await self._session_db.get_session(target_id) or {} except Exception: return False - caller_src = source.platform.value if source.platform else None - row_src = row.get("source") - if row_src and caller_src and str(row_src) != str(caller_src): - return False # different platform / source - caller_uid = str(getattr(source, "user_id", "") or "") - row_uid = str(row.get("user_id") or "") - # Chat/thread origin recorded at session creation. Rows once stored only source + user_id, - # so a same-user row could belong to a DIFFERENT chat; comparing the persisted origin closes - # that gap. Legacy rows (NULL) fail closed — resume via a live session or an admin override. - caller_chat = str(getattr(source, "chat_id", "") or "") - row_chat = str(row.get("chat_id") or "") - caller_thread = str(getattr(source, "thread_id", "") or "") - row_thread = str(row.get("thread_id") or "") - chat_type = (getattr(source, "chat_type", "") or "").lower() - caller_is_dm = chat_type in {"dm", "direct", "private", ""} - # build_session_key keys the participant on ``user_id_alt or user_id``, but the sessions table - # has no user_id_alt column, so a row cannot prove the canonical participant for an alt-keyed - # (Signal/Feishu) caller: per-user row_uid == caller_uid checks must fail closed (CWE-639). - caller_keys_on_alt = bool(str(getattr(source, "user_id_alt", "") or "")) - if caller_uid: - # Identity-bearing caller: the row must PROVE the same owner AND platform AND chat/thread. - # A blank/legacy source can't prove the platform (row_src above only rejects a *mismatching* - # non-blank one); a different thread is a different session. Any gap fails closed. - origin_ok = ( - bool(row_src) and bool(caller_src) - and str(row_src) == str(caller_src) - and row_thread == caller_thread - ) - if not origin_ok: - return False - if caller_is_dm: - # DMs are keyed on user_id; require the same owner. A no-chat_id DM is keyed PURELY on - # the participant (so an alt-keyed caller fails closed); when both sides carry chat_id, - # equality is the DM key and suffices, and a mismatching chat_id is rejected. - if caller_keys_on_alt and not (bool(row_chat) and bool(caller_chat)): - return False - return ( - bool(row_uid) and row_uid == caller_uid - and row_chat == caller_chat - ) - # Non-DM (group/channel/forum/thread): build_session_key includes chat_id, so a row (or - # caller) with NO chat provenance cannot prove same-chat. Require both non-blank and - # equal — a legacy NULL-chat row fails closed even when both normalize to "". (CWE-639) - if not (bool(row_chat) and bool(caller_chat) and row_chat == caller_chat): - return False - # Same non-DM chat/thread: mirror build_session_key's participant scoping. A SHARED - # group/thread session (group_sessions_per_user=False, or a shared thread) is one session - # for every participant, so the same-chat proof suffices — do NOT also require user-id - # equality (it would block co-members). A per-user session still requires the same owner. - if self._is_shared_session_source(source): - return True - # Per-user non-DM: the session key includes the participant (``user_id_alt or - # user_id``). If the caller keys on user_id_alt, the persisted row (user_id only) cannot - # prove the canonical participant, so fail closed rather than matching on user_id alone. - if caller_keys_on_alt: - return False - return bool(row_uid) and row_uid == caller_uid - # No caller identity: the row carries only source + user_id, so a same-platform row can belong - # to a DIFFERENT chat or user — same platform alone is NOT ownership proof; fail closed - # (CWE-639). Same-chat resume of an ACTIVE session still works via the live-origin branch. - return False + return self._persisted_row_proves_owner(source, row) async def _resume_row_visible( self, source: SessionSource, row: dict, allow_all: bool ) -> bool: - """Whether a titled-session listing *row* belongs to the caller's origin. - - Prevents cross-origin enumeration of session ids/previews via the numbered /resume list; - keeps Matrix room-scoping, scopes every other platform to the caller unless admin ``--all``. - """ - sid = str(row.get("id") or "") - if source.platform == Platform.MATRIX: - # Cross-room enumeration is cross-ORIGIN data access: gate the ``--all`` short-circuit - # behind a real configured admin, exactly like the non-Matrix branch below. - if allow_all and self._resume_caller_is_admin(source): - return True - return self._same_matrix_room(source, self._gateway_session_origin_for_id(sid)) + """Whether a listing *row* belongs to the caller's origin (blocks cross-origin enumeration of + ids/previews); Matrix is room-scoped, ``--all`` needs a configured admin everywhere.""" if allow_all and self._resume_caller_is_admin(source): return True + sid = str(row.get("id") or "") + if source.platform == Platform.MATRIX: + return self._same_matrix_room(source, self._gateway_session_origin_for_id(sid)) return await self._resume_target_allowed(source, sid, allow_override=False) + # ------------------------------------------------------------------ /retry, /undo + async def _handle_retry_command(self, event: MessageEvent) -> str: """Handle /retry command - re-send the last user message.""" - source = event.source - session_entry = await self.async_session_store.get_or_create_session(source) - history = await self.async_session_store.load_transcript(session_entry.session_id) - - # Find the last *real* user message. Timeline bookkeeping rows carry role=user + - # display_kind (model_switch / async_delegation_complete / auto_continue / hidden); clients - # never count them as user turns. - last_user_idx = None - # The canonical projection excludes bookkeeping and pure handoffs while - # still recognizing a real ask embedded in a compaction carrier. + # The canonical projection skips bookkeeping rows (role=user + display_kind) and pure + # handoffs while still recognizing a real ask embedded in a compaction carrier. from agent.context_compressor import ( history_before_user_originated_turn, retryable_user_text, @@ -513,353 +428,232 @@ class GatewaySessionCommandsMixin: user_originated_turn_view, ) - for i in range(len(history) - 1, -1, -1): - msg = history[i] - if user_originated_turn_view(msg) is not None: - last_user_idx = i - break - + source = event.source + session_entry = await self.async_session_store.get_or_create_session(source) + history = await self.async_session_store.load_transcript(session_entry.session_id) + last_user_idx = next( + (i for i in range(len(history) - 1, -1, -1) if user_originated_turn_view(history[i]) is not None), + None, + ) if last_user_idx is None: return t("gateway.retry.no_previous") - - # Resolve the live text and the scaffold-preserving prefix before any - # transcript write. Messaging retries cannot reconstruct attachments; - # reject media/unknown content without truncating the session. + # Resolve text + scaffold-preserving prefix BEFORE any write; messaging retries cannot + # reconstruct attachments, so media/unknown content is rejected with the session untouched. try: - truncated, live_view = history_before_user_originated_turn( - history, last_user_idx - ) + truncated, live_view = history_before_user_originated_turn(history, last_user_idx) last_user_msg = retryable_user_text(live_view.get("content")) handoff, _ = split_user_originated_turn(history[last_user_idx]) except ValueError as exc: return f"Cannot retry that message safely: {exc}" if handoff is not None: - # A composite carrier is one physical row containing both the retained summary and the - # live ask. Let the carrier-aware rewind archive that row/tail and insert its pure - # scaffold atomically. + # Composite carrier (one row = retained summary + live ask): the carrier-aware rewind + # archives it and inserts the pure scaffold atomically, reselecting the latest carrier + # on the same snapshot so a concurrent newer turn is never removed for stale text. try: rewind_result = await self.async_session_store.rewind_session( - session_entry.session_id, - 1, - require_retryable_composite=True, + session_entry.session_id, 1, require_retryable_composite=True, ) except ValueError as exc: return f"Cannot retry that message safely: {exc}" if rewind_result is None: return "Retry failed; transcript was not changed." - # The store reselects and validates the latest carrier on the same - # snapshot used by the atomic rewind. A concurrent newer turn can - # therefore never be removed while this handler resends stale text. last_user_msg = rewind_result["target_text"] - else: - # After in-place compaction the pre-compaction transcript lives on as - # active=0/compacted=1 rows under this session id. active_only preserves that archive; a - # separate existence probe could fail open or race with the write. - if not await self.async_session_store.rewrite_transcript( - session_entry.session_id, - truncated, - active_only=True, - reject_active_turn_lease=True, - ): - return "Retry failed; transcript was not changed." - # Reset stored token count — transcript was truncated - session_entry.last_prompt_tokens = 0 - - # Re-send by creating a fake text event with the old message + # active_only preserves the active=0/compacted=1 archive left by in-place compaction. + elif not await self.async_session_store.rewrite_transcript( + session_entry.session_id, truncated, active_only=True, reject_active_turn_lease=True, + ): + return "Retry failed; transcript was not changed." + session_entry.last_prompt_tokens = 0 # transcript was truncated retry_event = MessageEvent( - text=last_user_msg, - message_type=MessageType.TEXT, - source=source, - raw_message=event.raw_message, - channel_prompt=event.channel_prompt, + text=last_user_msg, message_type=MessageType.TEXT, source=source, + raw_message=event.raw_message, channel_prompt=event.channel_prompt, ) - - # Let the normal message handler process it return await self._handle_message(retry_event) async def _handle_undo_command(self, event: MessageEvent) -> str: """Handle /undo [N] — back up N user turns (default 1), soft-deleting the truncated rows and - echoing the backed-up text. Evicts the cached agent so the next message rebuilds context - from the active-only transcript (gateway analogue of the CLI's history surgery). - """ + echoing the backed-up text; evicts the cached agent so the next turn rebuilds from the + active-only transcript.""" source = event.source - - # Parse optional turn count: "/undo" → 1, "/undo 3" → 3. n = 1 raw_args = event.get_command_args().strip() if raw_args: try: - n = int(raw_args.split()[0]) - except (ValueError, IndexError): + n = max(1, int(raw_args.split()[0])) + except ValueError: return t("gateway.undo.invalid_count", arg=raw_args.split()[0]) - if n < 1: - n = 1 - session_entry = await self.async_session_store.get_or_create_session(source) result = await self.async_session_store.rewind_session(session_entry.session_id, n) - if result is None: return t("gateway.undo.nothing") - - # Reset stored token count — transcript was truncated. - session_entry.last_prompt_tokens = 0 - # Evict the cached agent so the next turn rebuilds from the active-only - # transcript and memory providers refresh their per-session caches. + session_entry.last_prompt_tokens = 0 # transcript was truncated try: - session_key = build_session_key(source) - self._evict_cached_agent(session_key) + self._evict_cached_agent(build_session_key(source)) except Exception as e: logger.debug("undo: cached-agent eviction skipped: %s", e) - target_text = result["target_text"] preview = target_text[:200] + "..." if len(target_text) > 200 else target_text - return t( - "gateway.undo.removed", - turns=result["turns_undone"], - count=result["rewound_count"], - preview=preview, - ) + return t("gateway.undo.removed", turns=result["turns_undone"], count=result["rewound_count"], preview=preview) + + # --------------------------------------------------------------------- /compress async def _handle_compress_command(self, event: MessageEvent) -> str: - """Profile-scoping wrapper around manual /compress. - - Multiplexed gateways resolve credentials through the fail-closed per-profile secret scope; - slash dispatch (unlike ``_run_agent``) does not install it, so an unscoped /compress would - raise ``UnscopedSecretError``. Single-profile gateways skip this. - """ + """Profile-scoping wrapper around manual /compress: multiplexed gateways resolve credentials + through the fail-closed per-profile secret scope, which slash dispatch (unlike ``_run_agent``) + does not install — unscoped, /compress would raise ``UnscopedSecretError``.""" if not getattr(getattr(self, "config", None), "multiplex_profiles", False): return await self._handle_compress_command_inner(event) - from gateway.run import _profile_runtime_scope - - profile_home = self._resolve_profile_home_for_source(event.source) - with _profile_runtime_scope(profile_home): + with _profile_runtime_scope(self._resolve_profile_home_for_source(event.source)): return await self._handle_compress_command_inner(event) - async def _compress_codex_app_server_session( - self, session_key: str, session_id: str - ) -> str: - """Manual /compress for codex_app_server sessions. - - Compacts the LIVE cached agent's app-server thread (``thread/compact/start``, ``force=True`` - bypasses the ``codex_app_server_auto`` gate) and keeps the agent cached. Never builds a - temporary agent or rewrites the mirror: neither can shrink the server-side thread. - """ + async def _compress_codex_app_server_session(self, session_key: str, session_id: str) -> str: + """Manual /compress for codex_app_server sessions: compacts the LIVE cached agent's + app-server thread (``force=True`` bypasses the ``codex_app_server_auto`` gate) and keeps it + cached. A temporary agent or a mirror rewrite cannot shrink the server-side thread.""" from gateway.run import _AGENT_PENDING_SENTINEL agent = self._cached_agent_for(session_key) - if ( - agent is None - or agent is _AGENT_PENDING_SENTINEL - or getattr(agent, "_codex_session", None) is None - ): + if agent is None or agent is _AGENT_PENDING_SENTINEL or getattr(agent, "_codex_session", None) is None: return ( - "🗜️ Nothing to compact: this session runs on the Codex " - "app-server runtime, whose context lives in a Codex-owned " - "thread that only exists while the agent is active. Send a " - "message first, then /compress — or /reset to start fresh." + "🗜️ Nothing to compact: this session runs on the Codex app-server runtime, whose " + "context lives in a Codex-owned thread that only exists while the agent is active. " + "Send a message first, then /compress — or /reset to start fresh." ) - compressor = getattr(agent, "context_compressor", None) count_before = getattr(compressor, "compression_count", 0) try: - await self._run_in_executor_with_context( - lambda: agent._compress_context( - [], "", force=True, - ) - ) + await self._run_in_executor_with_context(lambda: agent._compress_context([], "", force=True)) except Exception as exc: return t("gateway.compress.failed", error=exc) - count_after = getattr(compressor, "compression_count", 0) - if count_after > count_before: + if getattr(compressor, "compression_count", 0) > count_before: return ( - "🗜️ Codex app-server thread compacted (thread/compact). " - "The transcript mirror is unchanged by design — the " - "app-server now carries the compacted context." + "🗜️ Codex app-server thread compacted (thread/compact). The transcript mirror is " + "unchanged by design — the app-server now carries the compacted context." ) return ( - "⚠️ Codex app-server compaction did not complete — the thread " - "is unchanged. Check the app-server logs, retry /compress, or " - "/reset for a clean session." + "⚠️ Codex app-server compaction did not complete — the thread is unchanged. Check the " + "app-server logs, retry /compress, or /reset for a clean session." ) async def _handle_compress_command_inner(self, event: MessageEvent) -> str: - """Handle /compress command -- manually compress conversation context. + """Handle /compress -- manually compress conversation context; ``/compress <focus>`` tells + the summariser what to preserve.""" + from hermes_cli.partial_compress import extract_compress_flags, parse_partial_compress_args - Optional ``/compress <focus>`` tells the summariser what to preserve, discarding the rest. - """ source = event.source session_entry = await self.async_session_store.get_or_create_session(source) history = await self.async_session_store.load_transcript(session_entry.session_id) - if not history or len(history) < 4: return t("gateway.compress.not_enough") - - # Parse args: either a focus topic (full compress) or the - # boundary-aware "here [N]" form (partial compress). - from hermes_cli.partial_compress import ( - extract_compress_flags, - parse_partial_compress_args, - rejoin_compressed_head_and_tail, - split_history_for_partial_compress, - ) - from agent.conversation_compression import ( - finalize_context_engine_compression_notification, - ) - _raw_args = (event.get_command_args() or "").strip() - # Strip --preview/--dry-run/--aggressive before positional parsing - # so the flags coexist with 'here [N]' / focus-topic forms. - _raw_args, _preview, _aggressive = extract_compress_flags(_raw_args) + # Flags are stripped before positional parsing so they coexist with the boundary-aware + # "here [N]" (partial) and focus-topic (full) forms. + _raw_args, _preview, _aggressive = extract_compress_flags((event.get_command_args() or "").strip()) partial, keep_last, focus_topic = parse_partial_compress_args(_raw_args) - _agg_note = "" if _aggressive: - # LLM-free hard truncation is not supported on this surface — it would need its own - # transcript-persistence branch outside the guarded _compress_context rotation machinery. + # LLM-free hard truncation would need its own persistence branch outside the guarded + # _compress_context rotation machinery — unsupported on this surface. _agg_note = t("gateway.compress.aggressive_unsupported") if not _preview: return _agg_note - if _preview: return _compress_preview_reply(history, partial, keep_last, focus_topic, _agg_note) - try: - from agent.manual_compression_feedback import summarize_manual_compression - from agent.model_metadata import estimate_request_tokens_rough - from gateway.run import _platform_config_key - - session_key = self._session_key_for_source(source) - # Preserve the platform + stable gateway session identity of a normal turn so external - # context engines bind this agent to the original conversation, not a default "cli" host. - platform_key = ( - _platform_config_key(source.platform) if source.platform else None + return await self._run_manual_compression( + source, session_entry, history, partial, keep_last, focus_topic ) - model, runtime_kwargs = self._resolve_session_agent_runtime( - source=source, - session_key=session_key, - ) - if str(runtime_kwargs.get("api_mode") or "").lower() == "codex_app_server": - # codex app-server: the model's context is the server-side thread owned by the LIVE - # cached agent; a temporary agent has none (and finally-eviction would destroy the - # real context). Compact the live thread and KEEP the agent cached; no mirror fallback. - return await self._compress_codex_app_server_session( - session_key, session_entry.session_id - ) - if not runtime_kwargs.get("api_key"): - return t("gateway.compress.no_provider") - - # Pass the FULL transcript (tool results included), like auto-compress: user/assistant- - # only starves tool-result pruning and can trip the protect-first/last early-return. - msgs = [ - m for m in history - if m.get("role") in {"user", "assistant", "tool"} - ] - - # Boundary-aware split: only the head is summarized; the most recent `keep_last` - # exchanges are preserved verbatim. The split snaps the tail to a user-turn start so the - # rejoined transcript keeps role alternation valid. - tail: list = [] - head = msgs - if partial: - head, tail = split_history_for_partial_compress(msgs, keep_last) - if not tail: - # Degenerate split — fall back to full compression. - partial = False - head = msgs - - # Bind the temporary compression agent to the source's platform + stable gateway session - # key. Assign directly (not setdefault: a resolver value would be a stale placeholder, - # and it avoids duplicate-kwarg TypeError); platform only when known so None -> "cli" holds. - if platform_key is not None: - runtime_kwargs["platform"] = platform_key - runtime_kwargs["gateway_session_key"] = session_key - - tmp_agent = await self._build_manual_compression_agent( - session_entry.session_id, model, runtime_kwargs - ) - try: - # Estimate with system prompt + tool schemas included so the figure reflects real - # request pressure, not a transcript-only underestimate. Must be computed after - # tmp_agent is built so _cached_system_prompt/tools are populated. - _sys_prompt = getattr(tmp_agent, "_cached_system_prompt", "") or "" - _tools = getattr(tmp_agent, "tools", None) or None - approx_tokens = estimate_request_tokens_rough( - msgs, system_prompt=_sys_prompt, tools=_tools - ) - - compressor = tmp_agent.context_compressor - if not compressor.has_content_to_compress(head): - return t("gateway.compress.nothing_to_do") - - # Not a bare run_in_executor: the profile secret scope is a contextvar and the - # default-executor hop would drop it, making the compressor's aux-client credential - # resolution fail closed under multiplexing. - compressed, _ = await self._run_in_executor_with_context( - lambda: tmp_agent._compress_context( - head, - "", - approx_tokens=approx_tokens, - focus_topic=focus_topic, - force=True, - defer_context_engine_notification=True, - ) - ) - - # If _compress_context returned unchanged because a concurrent compression lock is - # held, tell the user clearly instead of showing the misleading "No changes from - # compression" no-op text. - _lock_skipped = getattr(tmp_agent, "_compression_skipped_due_to_lock", None) - if _lock_skipped is True or isinstance(_lock_skipped, str): - from agent.manual_compression_feedback import ( - describe_compression_lock_skip, - ) - return describe_compression_lock_skip(_lock_skipped) - - if partial and tail: - compressed = rejoin_compressed_head_and_tail(compressed, tail) - - await self._persist_manual_compression(tmp_agent, session_entry, source, compressed) - finalize_context_engine_compression_notification( - tmp_agent, - committed=True, - ) - new_tokens = estimate_request_tokens_rough( - compressed, system_prompt=_sys_prompt, tools=_tools - ) - summary = summarize_manual_compression( - msgs, - compressed, - approx_tokens, - new_tokens, - compression_state=compressor, - ) - finally: - finalize_context_engine_compression_notification( - tmp_agent, - committed=False, - ) - # Evict cached agent so next turn rebuilds system prompt - # from current files (SOUL.md, memory, etc.). - self._evict_cached_agent(session_key) - # Off-loop + bounded: temporary-agent teardown can block on - # subprocess/network/SQLite work. - await self._cleanup_agent_resources_off_loop( - tmp_agent, context="manual compression" - ) - return "\n".join(_manual_compression_reply_lines(summary, compressor, focus_topic)) except Exception as e: logger.warning("Manual compress failed: %s", e) return t("gateway.compress.failed", error=e) + async def _run_manual_compression( + self, source, session_entry, history: list, partial: bool, keep_last, focus_topic + ) -> str: + """Build a temporary agent, compress the transcript, persist, and describe the outcome.""" + from agent.conversation_compression import finalize_context_engine_compression_notification + from agent.manual_compression_feedback import summarize_manual_compression + from agent.model_metadata import estimate_request_tokens_rough + from gateway.run import _platform_config_key + from hermes_cli.partial_compress import ( + rejoin_compressed_head_and_tail, + split_history_for_partial_compress, + ) + + session_key = self._session_key_for_source(source) + # Platform + stable gateway session key bind this agent (for external context engines) to + # the original conversation, not a default "cli" host. + platform_key = _platform_config_key(source.platform) if source.platform else None + model, runtime_kwargs = self._resolve_session_agent_runtime(source=source, session_key=session_key) + if str(runtime_kwargs.get("api_mode") or "").lower() == "codex_app_server": + # Context lives in the server-side thread of the LIVE cached agent; a temporary agent + # has none (and finally-eviction would destroy the real context). + return await self._compress_codex_app_server_session(session_key, session_entry.session_id) + if not runtime_kwargs.get("api_key"): + return t("gateway.compress.no_provider") + # FULL transcript (tool results included), like auto-compress: user/assistant-only starves + # tool-result pruning and can trip the protect-first/last early-return. + msgs = [m for m in history if m.get("role") in {"user", "assistant", "tool"}] + # Partial: only the head is summarized; the tail snaps to a user-turn start so role + # alternation holds after rejoin. + head, tail = msgs, [] + if partial: + head, tail = split_history_for_partial_compress(msgs, keep_last) + if not tail: # degenerate split — fall back to full compression + partial, head = False, msgs + # Assign, not setdefault (a resolver value would be a stale placeholder); platform only when + # known so None -> "cli" holds. + if platform_key is not None: + runtime_kwargs["platform"] = platform_key + runtime_kwargs["gateway_session_key"] = session_key + + tmp_agent = await self._build_manual_compression_agent(session_entry.session_id, model, runtime_kwargs) + try: + # Estimate with system prompt + tool schemas (real request pressure); needs the built agent. + _sys_prompt = getattr(tmp_agent, "_cached_system_prompt", "") or "" + _tools = getattr(tmp_agent, "tools", None) or None + approx_tokens = estimate_request_tokens_rough(msgs, system_prompt=_sys_prompt, tools=_tools) + compressor = tmp_agent.context_compressor + if not compressor.has_content_to_compress(head): + return t("gateway.compress.nothing_to_do") + # Not a bare run_in_executor: the profile secret scope is a contextvar the default + # executor hop would drop, failing aux-client credential resolution closed. + compressed, _ = await self._run_in_executor_with_context( + lambda: tmp_agent._compress_context( + head, "", approx_tokens=approx_tokens, focus_topic=focus_topic, force=True, + defer_context_engine_notification=True, + ) + ) + # A held compression lock returns unchanged; say so instead of the misleading no-op text. + _lock_skipped = getattr(tmp_agent, "_compression_skipped_due_to_lock", None) + if _lock_skipped is True or isinstance(_lock_skipped, str): + from agent.manual_compression_feedback import describe_compression_lock_skip + return describe_compression_lock_skip(_lock_skipped) + if partial and tail: + compressed = rejoin_compressed_head_and_tail(compressed, tail) + await self._persist_manual_compression(tmp_agent, session_entry, source, compressed) + finalize_context_engine_compression_notification(tmp_agent, committed=True) + new_tokens = estimate_request_tokens_rough(compressed, system_prompt=_sys_prompt, tools=_tools) + summary = summarize_manual_compression( + msgs, compressed, approx_tokens, new_tokens, compression_state=compressor, + ) + finally: + finalize_context_engine_compression_notification(tmp_agent, committed=False) + self._evict_cached_agent(session_key) # next turn rebuilds the prompt from current files + # Off-loop + bounded: teardown can block on subprocess/network/SQLite. + await self._cleanup_agent_resources_off_loop(tmp_agent, context="manual compression") + return "\n".join(_manual_compression_reply_lines(summary, compressor, focus_topic)) + async def _build_manual_compression_agent(self, session_id: str, model, runtime_kwargs: dict): """Build the throwaway AIAgent that performs a manual /compress rewrite of *session_id*.""" from run_agent import AIAgent from gateway.run import _GATEWAY_HYGIENE_PLATFORM, _seed_hygiene_system_prompt + from hermes_cli.config import load_config as _load_cfg + from utils import is_truthy_value as _is_truthy - # The manual compression helper runs outside the live session's fully initialized prompt - # environment and _compress_context may persist its cached system prompt — restore the - # exact live-session prompt so provider blocks are retained. + # _compress_context may persist its cached system prompt, and this agent runs outside the + # live session's prompt environment — restore the exact live prompt so provider blocks stay. session_row = None get_session = getattr(self._session_db, "get_session", None) if callable(get_session): @@ -867,56 +661,38 @@ class GatewaySessionCommandsMixin: session_row = await get_session(session_id) except Exception as exc: logger.warning( - "Manual compression could not restore the system prompt " - "for session %s: %s. Preserving an empty prompt so the " - "live turn rebuilds it with its configured providers.", - session_id, - exc, - exc_info=True, + "Manual compression could not restore the system prompt for session %s: %s. " + "Preserving an empty prompt so the live turn rebuilds it with its configured " + "providers.", session_id, exc, exc_info=True, ) - # This agent performs a lossy rewrite. When compression.checkpoint_required is on, the - # memory provider must be loaded so _compress_context() can write the pre-compression - # checkpoint; otherwise keep the historical fast path (no provider init). - from hermes_cli.config import load_config as _load_cfg - from utils import is_truthy_value as _is_truthy - + # compression.checkpoint_required needs the memory provider loaded so _compress_context() + # can write the pre-compression checkpoint; otherwise keep the fast path (no provider init). _checkpoint_required = _is_truthy( ((_load_cfg() or {}).get("compression") or {}).get("checkpoint_required"), default=False, ) tmp_agent = AIAgent( - **runtime_kwargs, - model=model, - max_iterations=4, - quiet_mode=True, - skip_memory=not _checkpoint_required, - enabled_toolsets=["memory"], - session_id=session_id, - session_db=getattr(self._session_db, "_db", self._session_db), + **runtime_kwargs, model=model, max_iterations=4, quiet_mode=True, + skip_memory=not _checkpoint_required, enabled_toolsets=["memory"], + session_id=session_id, session_db=getattr(self._session_db, "_db", self._session_db), ) _seed_hygiene_system_prompt(tmp_agent, session_row) - # Keep the real source platform during construction so external context engines bind - # correctly. If compression has to rebuild the prompt, stamp that provider-less fallback - # as stale for the next real gateway turn. + # Real platform during construction (context engines bind correctly); afterwards a prompt + # rebuilt by compression is stamped as the provider-less fallback, stale for the next turn. tmp_agent.platform = _GATEWAY_HYGIENE_PLATFORM tmp_agent._print_fn = lambda *a, **kw: None - # Prevent close() from ending the newly rotated session — the gateway session entry now - # points at the new id and must remain open for the next user turn. + # close() must not end the rotated session the gateway entry now points at. tmp_agent._end_session_on_close = False return tmp_agent async def _persist_manual_compression(self, tmp_agent, session_entry, source, compressed) -> None: """Commit a manual /compress result to the session store. - _compress_context either rotated (new continuation id — write compressed messages into the - NEW session so the original stays searchable) or compacted in place (compression.in_place: - same id, transcript replaced). Persist BEFORE repointing the live session: repoint first + - failed DB write would leave the entry on an empty session while reporting success; a failed - write is fatal so old history stays reachable. Only rewrite when rotation produced a NEW id: - in-place compaction already archived + inserted rows and rewrite_transcript() - (active_only=False) would DELETE the archived turns; an unchanged id without in-place means - rotation FAILED and a rewrite would leave only the summary. + Rotation (new continuation id) writes the compressed messages into the NEW session so the + original stays searchable; persist BEFORE repointing so a failed write is fatal and old + history stays reachable. In-place compaction already archived + inserted rows, and a rewrite + would DELETE the archive; an unchanged id without in-place means rotation FAILED. """ new_session_id = tmp_agent.session_id if new_session_id != session_entry.session_id: @@ -927,19 +703,17 @@ class GatewaySessionCommandsMixin: session_entry.session_id = new_session_id await self.async_session_store._save() await asyncio.to_thread( - self._sync_telegram_topic_binding, - source, session_entry, reason="compress-command", + self._sync_telegram_topic_binding, source, session_entry, reason="compress-command", ) elif not getattr(tmp_agent, "_last_compaction_in_place", False): logger.warning( - "Manual /compress: session rotation did not occur " - "(session_id unchanged) and in-place mode is off — " - "preserving original transcript instead of overwriting " - "it (#44794)." + "Manual /compress: session rotation did not occur (session_id unchanged) and in-place " + "mode is off — preserving original transcript instead of overwriting it (#44794)." ) - # Reset stored token count — transcript changed, old value is stale await self.async_session_store.update_session(session_entry.session_key, last_prompt_tokens=0) + # ------------------------------------------------------------------------ /topic + async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> str: """Handle /topic for Telegram DM user-managed topic sessions.""" source = event.source @@ -948,27 +722,18 @@ class GatewaySessionCommandsMixin: if not self._session_db: return self._session_db_unavailable_reply() - # Authorization: /topic activates multi-session mode and mutates SQLite side tables. - # Unauthorized senders (not in allowlist) must not be able to do that. Gateway routes - # already authorize the message before reaching here, but defense in depth. - auth_fn = getattr(self, "_is_user_authorized", None) - if callable(auth_fn): - try: - if not auth_fn(source): - return t("gateway.topic.unauthorized") - except Exception: - logger.debug("Topic auth check failed", exc_info=True) + # Defense in depth: /topic mutates SQLite side tables, so re-check the allowlist here. + try: + if not self._is_user_authorized(source): + return t("gateway.topic.unauthorized") + except Exception: + logger.debug("Topic auth check failed", exc_info=True) args = event.get_command_args().strip() - - # /topic help — inline usage without leaving the bot. if args.lower() in {"help", "?", "-h", "--help"}: return self._telegram_topic_help_text() - - # /topic off — clean disable path so users don't have to edit the DB. if args.lower() in {"off", "disable", "stop"}: return await self._disable_telegram_topic_mode_for_chat(source) - if args: if not source.thread_id: return t("gateway.topic.restore_needs_topic") @@ -976,22 +741,20 @@ class GatewaySessionCommandsMixin: capabilities = await self._get_telegram_topic_capabilities(source) if capabilities.get("checked"): + blocked_key = None if capabilities.get("has_topics_enabled") is False: - # Debounce the BotFather screenshot: don't re-send on every - # /topic while threads are still disabled. + blocked_key = "gateway.topic.topics_disabled" + elif capabilities.get("allows_users_to_create_topics") is False: + blocked_key = "gateway.topic.topics_user_disallowed" + if blocked_key: # debounced BotFather screenshot if self._should_send_telegram_capability_hint(source): await self._send_telegram_topic_setup_image(source) - return t("gateway.topic.topics_disabled") - if capabilities.get("allows_users_to_create_topics") is False: - if self._should_send_telegram_capability_hint(source): - await self._send_telegram_topic_setup_image(source) - return t("gateway.topic.topics_user_disallowed") + return t(blocked_key) + profile_name = self._telegram_topic_profile_name(source) try: await self._session_db.enable_telegram_topic_mode( - chat_id=str(source.chat_id), - user_id=str(source.user_id), - profile_name=self._telegram_topic_profile_name(source), + chat_id=str(source.chat_id), user_id=str(source.user_id), profile_name=profile_name, has_topics_enabled=capabilities.get("has_topics_enabled"), allows_users_to_create_topics=capabilities.get("allows_users_to_create_topics"), ) @@ -1001,36 +764,32 @@ class GatewaySessionCommandsMixin: if not source.thread_id: await self._ensure_telegram_system_topic(source) - - if source.thread_id: - try: - binding = await self._session_db.get_telegram_topic_binding( - chat_id=str(source.chat_id), - thread_id=str(source.thread_id), - profile_name=self._telegram_topic_profile_name(source), - ) - except Exception: - logger.debug("Failed to read Telegram topic binding", exc_info=True) - binding = None - if binding: - session_id = str(binding.get("session_id") or "") - title = None - try: - title = await self._session_db.get_session_title(session_id) - except Exception: - title = None - session_label = title or t("gateway.topic.untitled_session") - return t( - "gateway.topic.bound_status", - label=session_label, - session_id=session_id, - ) + return await self._telegram_topic_root_status_message(source) + try: + binding = await self._session_db.get_telegram_topic_binding( + chat_id=str(source.chat_id), thread_id=str(source.thread_id), + profile_name=profile_name, + ) + except Exception: + logger.debug("Failed to read Telegram topic binding", exc_info=True) + binding = None + if not binding: return t("gateway.topic.thread_ready") + session_id = str(binding.get("session_id") or "") + try: + title = await self._session_db.get_session_title(session_id) + except Exception: + title = None + return t( + "gateway.topic.bound_status", label=title or t("gateway.topic.untitled_session"), + session_id=session_id, + ) - return await self._telegram_topic_root_status_message(source) + # ------------------------------------------------------------------ /save, /title async def _handle_save_command(self, event: MessageEvent) -> str: """Handle /save — export the current session and send it as a document.""" + import tempfile from hermes_cli.session_export import ( SAVE_USAGE, default_save_filename, @@ -1039,15 +798,11 @@ class GatewaySessionCommandsMixin: ) parts = event.get_command_args().split() + redact = bool(parts) and parts[-1].lower() in ("redact", "--redact") + if redact: + parts = parts[:-1] if not parts: return SAVE_USAGE - redact = False - if parts[-1].lower() in ("redact", "--redact"): - redact = True - parts = parts[:-1] - if not parts: - return SAVE_USAGE - try: fmt = normalize_save_format(parts[0]) except ValueError as e: @@ -1056,182 +811,104 @@ class GatewaySessionCommandsMixin: source = event.source session_entry = await self.async_session_store.get_or_create_session(source) session_id = session_entry.session_id - if not self._session_db: return "Session database not available." + # Never trust path separators from chat input; the filename is only echoed to the platform. filename = parts[1] if len(parts) > 1 else default_save_filename(session_id, fmt) - # The filename is echoed to the platform only — never trust path - # separators from chat input. filename = os.path.basename(filename) or default_save_filename(session_id, fmt) - - # self._session_db is an AsyncSessionDB — every forwarded call is - # offloaded to a thread and must be awaited. export_data = await self._session_db.export_session(session_id) if not export_data: return f"No stored messages found for this session ({session_id})." - if redact: from hermes_cli.session_export_md import redact_session_data - export_data = redact_session_data(export_data) - - import tempfile - temp_dir = tempfile.mkdtemp(prefix="hermes_save_") temp_path = os.path.join(temp_dir, filename) try: - # Off-loop: rendering a long session and writing it to disk are CPU/disk-bound and scale - # with transcript size (multi-MB for long sessions). Inline they stall every other chat - # on the gateway event loop (Pattern A). One thread hop covers both. + # Off-loop: render + write scale with transcript size (multi-MB) and would stall the loop. def _render_and_write() -> None: rendered = render_session_for_save(export_data, fmt) with open(temp_path, "w", encoding="utf-8") as f: f.write(rendered) await asyncio.to_thread(_render_and_write) - adapter = self.get_adapter(source.platform) - if adapter: - await adapter.send_document( - chat_id=source.chat_id, - file_path=temp_path, - caption=f"Session export: {filename}", - file_name=filename, - ) - return "Export complete." - return "Platform adapter not found to send the document." + if not adapter: + return "Platform adapter not found to send the document." + await adapter.send_document( + chat_id=source.chat_id, file_path=temp_path, caption=f"Session export: {filename}", + file_name=filename, + ) + return "Export complete." except Exception as e: logger.warning("Session /save failed: %s", e) return f"Error exporting session: {e}" finally: - try: + with contextlib.suppress(Exception): os.remove(temp_path) os.rmdir(temp_dir) - except Exception: - pass async def _handle_title_command(self, event: MessageEvent) -> str: """Handle /title command — set or show the current session's title.""" source = event.source session_entry = await self.async_session_store.get_or_create_session(source) session_id = session_entry.session_id - if not self._session_db: return self._session_db_unavailable_reply() - # Ensure session exists in SQLite DB (it may only exist in session_store - # if this is the first command in a new session) - existing_title = await self._session_db.get_session_title(session_id) - if existing_title is None: - # Session doesn't exist in DB yet — create it - try: + # The session may only exist in session_store so far (first command in a new session). + # The messaging origin is persisted so a later /resume of this titled-but-inactive session + # can prove it belongs to the caller's chat/thread (IDOR scoping). + if await self._session_db.get_session_title(session_id) is None: + with contextlib.suppress(Exception): # may already exist await self._session_db.create_session( session_id=session_id, source=source.platform.value if source.platform else "unknown", - user_id=source.user_id, - # Persist the messaging origin so a later /resume of this - # titled-but-now-inactive session can prove it belongs to the - # caller's chat/thread (IDOR scoping). - chat_id=source.chat_id, - chat_type=source.chat_type, + user_id=source.user_id, chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id, ) - except Exception: - pass # Session might already exist, ignore errors - title_arg = event.get_command_args().strip() - if title_arg: - # Sanitize the title before setting - try: - from hermes_state import SessionDB - sanitized = SessionDB.sanitize_title(title_arg) - except ValueError as e: - return t("gateway.shared.warn_passthrough", error=e) - if not sanitized: - return t("gateway.title.empty_after_clean") - # Set the title - try: - if await self._session_db.set_session_title(session_id, sanitized): - # Propagate the user-chosen title to the visible Telegram forum topic name too. - # Auto-generated titles already rename the topic; without this, /title only - # updated the DB title and the topic kept its auto-assigned name. - schedule_rename = getattr( - self, "_schedule_telegram_topic_title_rename", None - ) - if callable(schedule_rename): - try: - await asyncio.to_thread(schedule_rename, source, session_id, sanitized) - except Exception: - logger.debug( - "Failed to rename Telegram topic from /title", - exc_info=True, - ) - return t("gateway.title.set_to", title=sanitized) - else: - return t("gateway.title.not_found") - except ValueError as e: - return t("gateway.shared.warn_passthrough", error=e) - else: - # Show the current title and session ID + if not title_arg: title = await self._session_db.get_session_title(session_id) if title: return t("gateway.title.current_with_title", session_id=session_id, title=title) - else: - return t("gateway.title.current_no_title", session_id=session_id) - - async def _handle_resume_command(self, event: MessageEvent) -> str: - """Handle /resume command — list or switch to a previous session.""" - if not self._session_db: - return self._session_db_unavailable_reply() - - source = await asyncio.to_thread( - self._normalize_source_for_session_key, event.source - ) - session_key = self._session_key_for_source(source) - raw_args = event.get_command_args().strip() + return t("gateway.title.current_no_title", session_id=session_id) try: - parts = shlex.split(raw_args) - except ValueError as exc: - return t("gateway.resume.parse_error", error=exc) - allow_all = "--all" in parts - allow_cross_room = "--cross-room" in parts - name = " ".join(p for p in parts if p not in {"--all", "--cross-room"}).strip() + from hermes_state import SessionDB + sanitized = SessionDB.sanitize_title(title_arg) + except ValueError as e: + return t("gateway.shared.warn_passthrough", error=e) + if not sanitized: + return t("gateway.title.empty_after_clean") + try: + if not await self._session_db.set_session_title(session_id, sanitized): + return t("gateway.title.not_found") + except ValueError as e: + return t("gateway.shared.warn_passthrough", error=e) + # Mirror the title onto the Telegram forum topic name (auto titles already do this). + try: + await asyncio.to_thread(self._schedule_telegram_topic_title_rename, source, session_id, sanitized) + except Exception: + logger.debug("Failed to rename Telegram topic from /title", exc_info=True) + return t("gateway.title.set_to", title=sanitized) - # Strip common outer brackets/quotes users may type literally from the - # usage hint (e.g. ``/resume <abc123>``). Mirrors the CLI behavior. - if len(name) >= 2 and ( - (name[0] == "<" and name[-1] == ">") - or (name[0] == "[" and name[-1] == "]") - or (name[0] == '"' and name[-1] == '"') - or (name[0] == "'" and name[-1] == "'") - ): - name = name[1:-1].strip() + # -------------------------------------------------------------- /resume, /sessions - async def _list_titled_sessions() -> list[dict]: - """Titled sessions visible to the caller (origin-scoped unless admin ``--all``).""" - user_source = source.platform.value if source.platform else None - widen = allow_all and self._resume_caller_is_admin(source) - sessions = await self._session_db.list_sessions_rich( - source=user_source, - session_key=None if widen else session_key, - limit=10, - ) - titled = [s for s in sessions if s.get("title")][:10] - return [s for s in titled if await self._resume_row_visible(source, s, allow_all)] + async def _list_titled_sessions(self, source, session_key: str, allow_all: bool) -> list[dict]: + """Titled sessions visible to the caller (origin-scoped unless admin ``--all``).""" + widen = allow_all and self._resume_caller_is_admin(source) + sessions = await self._session_db.list_sessions_rich( + source=source.platform.value if source.platform else None, + session_key=None if widen else session_key, limit=10, + ) + titled = [s for s in sessions if s.get("title")][:10] + return [s for s in titled if await self._resume_row_visible(source, s, allow_all)] - if not name: - # List recent titled sessions for this user/platform - try: - titled = await _list_titled_sessions() - return self._resume_listing_reply(source, titled, allow_all) - except Exception as e: - logger.debug("Failed to list titled sessions: %s", e) - return t("gateway.resume.list_failed", error=e) - - # Resolve a numbered choice or a title to a session ID. + async def _resolve_resume_target(self, source, session_key: str, name: str, allow_all: bool): + """``(target_id, name)`` for a numbered choice, session id or title; else the error reply.""" if name.isdigit(): try: - titled = await _list_titled_sessions() + titled = await self._list_titled_sessions(source, session_key, allow_all) except Exception as e: logger.debug("Failed to list titled sessions for numeric resume: %s", e) return t("gateway.resume.list_failed", error=e) @@ -1239,80 +916,86 @@ class GatewaySessionCommandsMixin: if index < 1 or index > len(titled): return t("gateway.resume.out_of_range", index=index) target = titled[index - 1] - target_id = target.get("id") - name = target.get("title") or name - else: - # Try direct session ID lookup first (so `/resume <session_id>` - # works in the gateway, not just `/resume <title>`). + target_id, name = target.get("id"), target.get("title") or name + else: # session id first, then title session = await self._session_db.get_session(name) - if session: - target_id = session["id"] - else: - target_id = await self._session_db.resolve_session_by_title(name) + target_id = session["id"] if session else await self._session_db.resolve_session_by_title(name) if not target_id: return t("gateway.resume.not_found", name=name) - # Compression creates child continuations that hold the live transcript. - # Follow that chain so gateway /resume matches CLI behavior (#15000). + # Follow compression continuations to the live transcript (matches CLI /resume). try: target_id = await self._session_db.resolve_resume_session_id(target_id) except Exception as e: logger.debug("Failed to resolve resume continuation for %s: %s", target_id, e) + return target_id, name + async def _resume_access_denied_reply( + self, source, target_id: str, name: str, allow_all: bool, allow_cross_room: bool + ) -> Optional[str]: + """IDOR guard: a session id/title is a routing handle, not authority — bind /resume to the + caller's own room (Matrix) or platform/user/chat (other adapters).""" if source.platform == Platform.MATRIX: target_origin = self._gateway_session_origin_for_id(target_id) - if not self._same_matrix_room(source, target_origin) and not allow_cross_room: - if target_origin is None: - return t("gateway.resume.matrix_blocked_no_origin", name=name) - return t( - "gateway.resume.matrix_blocked_other_room", - room=target_origin.chat_name or target_origin.chat_id, - name=name, - ) - elif not await self._resume_target_allowed( - source, target_id, allow_override=(allow_all or allow_cross_room) - ): - # IDOR guard: a session id/title is a routing handle, not authority. Bind /resume to the - # caller's own platform/user/chat on every non-Matrix adapter so one user can't attach - # to another's persisted transcript. - return t("gateway.resume.blocked_not_owner", name=name) + if self._same_matrix_room(source, target_origin) or allow_cross_room: + return None + if target_origin is None: + return t("gateway.resume.matrix_blocked_no_origin", name=name) + return t( + "gateway.resume.matrix_blocked_other_room", + room=target_origin.chat_name or target_origin.chat_id, name=name, + ) + if await self._resume_target_allowed(source, target_id, allow_override=(allow_all or allow_cross_room)): + return None + return t("gateway.resume.blocked_not_owner", name=name) - # Check if already on that session + async def _handle_resume_command(self, event: MessageEvent) -> str: + """Handle /resume command — list or switch to a previous session.""" + if not self._session_db: + return self._session_db_unavailable_reply() + source = await asyncio.to_thread(self._normalize_source_for_session_key, event.source) + session_key = self._session_key_for_source(source) + try: + parts = shlex.split(event.get_command_args().strip()) + except ValueError as exc: + return t("gateway.resume.parse_error", error=exc) + allow_all = "--all" in parts + allow_cross_room = "--cross-room" in parts + name = _strip_resume_name(parts) + if not name: + try: + titled = await self._list_titled_sessions(source, session_key, allow_all) + return self._resume_listing_reply(source, titled, allow_all) + except Exception as e: + logger.debug("Failed to list titled sessions: %s", e) + return t("gateway.resume.list_failed", error=e) + + resolved = await self._resolve_resume_target(source, session_key, name, allow_all) + if isinstance(resolved, str): + return resolved + target_id, name = resolved + denied = await self._resume_access_denied_reply(source, target_id, name, allow_all, allow_cross_room) + if denied is not None: + return denied current_entry = await self.async_session_store.get_or_create_session(source) if current_entry.session_id == target_id: return t("gateway.resume.already_on", name=name) - - # Clear any running agent for this session key self._release_running_agent_state(session_key) - - # Switch the session entry to point at the old session new_entry = await self.async_session_store.switch_session(session_key, target_id) if not new_entry: return t("gateway.resume.switch_failed") - - # Conversation boundary: clear ALL conversation-scoped per-session state (model/reasoning - # overrides #10702, one-turn restores, model notes, last-resolved cache #58403, /queue - # overflow) + security state in one funnel call. + # Conversation boundary: all conversation-scoped state + security state in one funnel call. self._clear_conversation_scope(session_key, reason="resume") - - # Evict any cached agent for this session so the next message rebuilds with the correct - # session_id end-to-end — mirrors /branch and /reset. Otherwise the cached AIAgent (and its - # memory provider, which cached _session_id at initialize()) keeps writing to the wrong session. + # Evict so the next turn rebuilds with the right session_id — the cached AIAgent's memory + # provider cached _session_id at initialize() and would keep writing to the wrong session. self._evict_cached_agent(session_key) - - # Get the title for confirmation title = await self._session_db.get_session_title(target_id) or name - - # Count messages for context history = await self.async_session_store.load_transcript(target_id) msg_count = len([m for m in history if m.get("role") == "user"]) if history else 0 - msg_part = f" ({msg_count} message{'s' if msg_count != 1 else ''})" if msg_count else "" - if source.platform == Platform.MATRIX and allow_cross_room: + msg_part = f" ({msg_count} message{'s' if msg_count != 1 else ''})" if msg_count else "" return t( - "gateway.resume.matrix_cross_room_success", - title=title, - room=source.chat_name or source.chat_id, - msg_part=msg_part, + "gateway.resume.matrix_cross_room_success", title=title, + room=source.chat_name or source.chat_id, msg_part=msg_part, ) if not msg_count: return t("gateway.resume.resumed_no_count", title=title) @@ -1321,13 +1004,11 @@ class GatewaySessionCommandsMixin: return t("gateway.resume.resumed_many", title=title, count=msg_count) def _resume_listing_reply(self, source, titled: list[dict], allow_all: bool) -> str: - """Numbered /resume list. A non-admin ``--all`` silently falls back to same-origin scoping; - say so instead of rendering an unexplained narrower list (sibling of the /sessions notice).""" - scope_note = ( - t("gateway.resume.all_requires_admin") - if allow_all and not self._resume_caller_is_admin(source) - else None - ) + """Numbered /resume list; a non-admin ``--all`` falls back to same-origin scoping and says so + (sibling of the /sessions notice).""" + scope_note = None + if allow_all and not self._resume_caller_is_admin(source): + scope_note = t("gateway.resume.all_requires_admin") if not titled: if source.platform == Platform.MATRIX and not allow_all: return t("gateway.resume.matrix_no_named_sessions") @@ -1352,46 +1033,30 @@ class GatewaySessionCommandsMixin: """Handle /sessions — list previous sessions for gateway chats.""" if not self._session_db: return self._session_db_unavailable_reply() - from hermes_cli.session_listing import ( format_gateway_session_listing, parse_session_listing_args, query_session_listing, ) - raw_args = event.get_command_args().strip() try: - include_all, include_unnamed, target, search_query = ( - parse_session_listing_args(raw_args) + include_all, include_unnamed, target, search_query = parse_session_listing_args( + event.get_command_args().strip() ) except ValueError as exc: return t("gateway.resume.parse_error", error=exc) - if search_query == "": return "Usage: `/sessions search <query>`" - if target: - resume_event = dataclasses.replace(event, text=f"/resume {target}") - return await self._handle_resume_command(resume_event) - - source = await asyncio.to_thread( - self._normalize_source_for_session_key, event.source - ) + return await self._handle_resume_command(dataclasses.replace(event, text=f"/resume {target}")) + source = await asyncio.to_thread(self._normalize_source_for_session_key, event.source) session_key = self._session_key_for_source(source) - - # A cross-origin listing (`/sessions all`) is honored only for an admin, mirroring the - # `/resume --all` override. `all` is just a parsed user argument; ungated, any caller could - # enumerate other origins' session ids/titles/previews — the enumeration half of the IDOR. + # `/sessions all` is admin-only like `/resume --all` (else any caller could enumerate other + # origins' ids/titles/previews); a non-admin gets explicit feedback, not a silent narrowing. cross_origin = include_all and self._resume_caller_is_admin(source) - # Don't silently no-op a requested widening: a non-admin `/sessions all` - # used to render the same scoped list with zero feedback, which reads - # as "my session vanished" (community report, Aug 2026). scope_notice = None if include_all and not cross_origin: - scope_notice = ( - "_Note: `all` (cross-chat listing) requires a configured admin; " - "showing this chat's sessions only._" - ) + scope_notice = "_Note: `all` (cross-chat listing) requires a configured admin; showing this chat's sessions only._" current_entry = await self.async_session_store.get_or_create_session(source) rows = await asyncio.to_thread( query_session_listing, @@ -1403,83 +1068,49 @@ class GatewaySessionCommandsMixin: include_all_sources=cross_origin, include_unnamed=include_unnamed, search_query=search_query, - # Search filters at SQL level, so over-fetch before the visibility - # cut: origin-invisible matches would otherwise consume the page. + # Search filters in SQL: over-fetch so origin-invisible matches don't consume the page. limit=50 if search_query else 10, exclude_sources=["tool"], ) if not cross_origin: - # Scope the listing to the caller's own origin on every adapter so - # session ids/previews from other users/rooms aren't enumerable. - rows = [ - row for row in rows - if await self._resume_row_visible(source, row, allow_all=False) - ] + rows = [row for row in rows if await self._resume_row_visible(source, row, allow_all=False)] rows = rows[:10] if search_query: title = f"Sessions matching “{search_query}”" else: title = "Sessions" if include_unnamed else "Named Sessions" - return format_gateway_session_listing( - rows, - include_source=cross_origin, - title=title, - notice=scope_notice, - ) + return format_gateway_session_listing(rows, include_source=cross_origin, title=title, notice=scope_notice) + + # ----------------------------------------------------------------------- /branch async def _handle_branch_command(self, event: MessageEvent) -> str: - """Handle /branch [name] — fork the current session into a new independent copy so the - user can explore a different approach without losing the original. - """ + """Handle /branch [name] — fork the current session into an independent copy.""" + import json as _json import uuid as _uuid + from datetime import datetime as _dt if not self._session_db: return self._session_db_unavailable_reply() - source = event.source session_key = self._session_key_for_source(source) - - # Load the current session and its transcript current_entry = await self.async_session_store.get_or_create_session(source) history = await self.async_session_store.load_transcript(current_entry.session_id) if not history: return t("gateway.branch.no_conversation") - - branch_name = event.get_command_args().strip() - - # Generate the new session ID - from datetime import datetime as _dt - now = _dt.now() - timestamp_str = now.strftime("%Y%m%d_%H%M%S") - short_uuid = _uuid.uuid4().hex[:6] - new_session_id = f"{timestamp_str}_{short_uuid}" - - # Determine branch title - if branch_name: - branch_title = branch_name - else: + new_session_id = f"{_dt.now().strftime('%Y%m%d_%H%M%S')}_{_uuid.uuid4().hex[:6]}" + branch_title = event.get_command_args().strip() + if not branch_title: current_title = await self._session_db.get_session_title(current_entry.session_id) - base = current_title or "branch" - branch_title = await self._session_db.get_next_title_in_lineage(base) - + branch_title = await self._session_db.get_next_title_in_lineage(current_title or "branch") parent_session_id = current_entry.session_id - - # Serialize the parent's full origin (same shape as the reset path's db_create_kwargs in - # gateway/session.py, #82633) so the branch row carries complete identity from birth. Prefer - # the live entry's origin (it may hold richer metadata than the triggering event's source). - _branch_origin = current_entry.origin or source + # Full parent origin (same shape as the reset path in gateway/session.py); the live entry's + # origin may hold richer metadata than the triggering event's source. _branch_origin_json = None - if _branch_origin is not None: - try: - import json as _json - - _branch_origin_json = _json.dumps(_branch_origin.to_dict()) - except Exception: - _branch_origin_json = None - - # Create the new session with parent link. Persist a stable ``_branched_from`` marker in - # model_config so list_sessions_rich() keeps the branch visible in /resume and /sessions - # even after the parent is reopened and re-ended with a different end_reason. + with contextlib.suppress(Exception): + _branch_origin_json = _json.dumps((current_entry.origin or source).to_dict()) + # ``_branched_from`` keeps the branch visible in /resume and /sessions after the parent is + # reopened and re-ended. ALL routing columns go in at CREATE time: a crash before + # switch_session() records the peer would otherwise leave the branch unroutable. try: await self._session_db.create_session( session_id=new_session_id, @@ -1487,12 +1118,6 @@ class GatewaySessionCommandsMixin: model=(self.config.get("model", {}) or {}).get("default") if isinstance(self.config, dict) else None, model_config={"_branched_from": parent_session_id}, parent_session_id=parent_session_id, - # Forward ALL gateway routing columns at CREATE time: otherwise they're NULL until - # switch_session() calls _record_gateway_session_peer(), and a crash in between (each - # append_message is best-effort) leaves the branch unroutable — by chat/thread lookup - # and by /resume's IDOR guard. user_id feeds the full-peer-tuple fallback lookup; - # origin_json/display_name complete the identity (same shape as session.py's reset - # path) so state.db consumers see a fully formed row with no backfill gap. user_id=source.user_id, session_key=session_key, chat_id=source.chat_id, @@ -1505,29 +1130,18 @@ class GatewaySessionCommandsMixin: logger.error("Failed to create branch session: %s", e) return t("gateway.branch.create_failed", error=e) - # Copy conversation history to the new session in bounded-chunk transactions: one txn per - # row was the removed write-amplification pattern, and a history can be hundreds of rows. - # Best-effort like the old loop — a failed copy still yields a usable (partial) branch. - try: + # Chunked transactions; best-effort — a failed copy still yields a usable (partial) branch. + with contextlib.suppress(Exception): await self._session_db.append_messages_batch( new_session_id, [_branch_row(msg) for msg in history], chunk_rows=500, ) - except Exception: - pass # Best-effort copy - - # Set title with contextlib.suppress(Exception): await self._session_db.set_session_title(new_session_id, branch_title) - - # Switch the session store entry to the new session new_entry = await self.async_session_store.switch_session(session_key, new_session_id) if not new_entry: return t("gateway.branch.switch_failed") self._clear_session_boundary_security_state(session_key) - - # Evict any cached agent for this session self._evict_cached_agent(session_key) - msg_count = len([m for m in history if m.get("role") == "user"]) key = "gateway.branch.branched_one" if msg_count == 1 else "gateway.branch.branched_many" return t(key, title=branch_title, count=msg_count, parent=parent_session_id, new=new_session_id)