fix(tests): accept unavailable_models kwarg in _prompt_model_selection mock

- test_update_check: replace patch("hermes_cli.banner.os.getenv") with
  monkeypatch.setenv("HERMES_HOME") — banner.py no longer imports os
  directly, it uses get_hermes_home() from hermes_constants.

- test_telegram_conflict/approval_buttons: provide real exception classes
  for telegram.error mock (NetworkError, TimedOut, BadRequest) so the
  except clause in connect() doesn't fail with "catching classes that do
  not inherit from BaseException" when xdist pollutes sys.modules.

.github/workflows/tests.yml

@@ -19,6 +19,9 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
 
+      - name: Install system dependencies
+        run: sudo apt-get update && sudo apt-get install -y ripgrep
+
       - name: Install uv
         uses: astral-sh/setup-uv@v5
 

agent/anthropic_adapter.py

@@ -706,29 +706,6 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]:
     }
 
 
-def run_hermes_oauth_login() -> Optional[str]:
-    """Run Hermes-native OAuth PKCE flow for Claude Pro/Max subscription.
-
-    Opens a browser to claude.ai for authorization, prompts for the code,
-    exchanges it for tokens, and stores them in ~/.hermes/.anthropic_oauth.json.
-
-    Returns the access token on success, None on failure.
-    """
-    result = run_hermes_oauth_login_pure()
-    if not result:
-        return None
-
-    access_token = result["access_token"]
-    refresh_token = result["refresh_token"]
-    expires_at_ms = result["expires_at_ms"]
-
-    _save_hermes_oauth_credentials(access_token, refresh_token, expires_at_ms)
-    _write_claude_code_credentials(access_token, refresh_token, expires_at_ms)
-
-    print("Authentication successful!")
-    return access_token
-
-
 def _save_hermes_oauth_credentials(access_token: str, refresh_token: str, expires_at_ms: int) -> None:
     """Save OAuth credentials to ~/.hermes/.anthropic_oauth.json."""
     data = {
@@ -756,38 +733,6 @@ def read_hermes_oauth_credentials() -> Optional[Dict[str, Any]]:
     return None
 
 
-def refresh_hermes_oauth_token() -> Optional[str]:
-    """Refresh the Hermes-managed OAuth token using the stored refresh token.
-
-    Returns the new access token, or None if refresh fails.
-    """
-    creds = read_hermes_oauth_credentials()
-    if not creds or not creds.get("refreshToken"):
-        return None
-
-    try:
-        refreshed = refresh_anthropic_oauth_pure(
-            creds["refreshToken"],
-            use_json=True,
-        )
-        _save_hermes_oauth_credentials(
-            refreshed["access_token"],
-            refreshed["refresh_token"],
-            refreshed["expires_at_ms"],
-        )
-        _write_claude_code_credentials(
-            refreshed["access_token"],
-            refreshed["refresh_token"],
-            refreshed["expires_at_ms"],
-        )
-        logger.debug("Successfully refreshed Hermes OAuth token")
-        return refreshed["access_token"]
-    except Exception as e:
-        logger.debug("Failed to refresh Hermes OAuth token: %s", e)
-
-    return None
-
-
 # ---------------------------------------------------------------------------
 # Message / tool / response format conversion
 # ---------------------------------------------------------------------------
@@ -857,35 +802,6 @@ def _convert_openai_image_part_to_anthropic(part: Dict[str, Any]) -> Optional[Di
     return None
 
 
-def _convert_user_content_part_to_anthropic(part: Any) -> Optional[Dict[str, Any]]:
-    if isinstance(part, dict):
-        ptype = part.get("type")
-        if ptype == "text":
-            block = {"type": "text", "text": part.get("text", "")}
-            if isinstance(part.get("cache_control"), dict):
-                block["cache_control"] = dict(part["cache_control"])
-            return block
-        if ptype == "image_url":
-            return _convert_openai_image_part_to_anthropic(part)
-        if ptype == "image" and part.get("source"):
-            return dict(part)
-        if ptype == "image" and part.get("data"):
-            media_type = part.get("mimeType") or part.get("media_type") or "image/png"
-            return {
-                "type": "image",
-                "source": {
-                    "type": "base64",
-                    "media_type": media_type,
-                    "data": part.get("data", ""),
-                },
-            }
-        if ptype == "tool_result":
-            return dict(part)
-    elif part is not None:
-        return {"type": "text", "text": str(part)}
-    return None
-
-
 def convert_tools_to_anthropic(tools: List[Dict]) -> List[Dict]:
     """Convert OpenAI tool definitions to Anthropic format."""
     if not tools:

agent/builtin_memory_provider.py

@@ -16,6 +16,7 @@ import logging
 from typing import Any, Dict, List
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -92,7 +93,7 @@ class BuiltinMemoryProvider(MemoryProvider):
 
     def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
         """Not used — the memory tool is intercepted in run_agent.py."""
-        return json.dumps({"error": "Built-in memory tool is handled by the agent loop"})
+        return tool_error("Built-in memory tool is handled by the agent loop")
 
     def shutdown(self) -> None:
         """No cleanup needed — files are saved on every write."""

agent/display.py

@@ -986,24 +986,6 @@ def _osc8_link(url: str, text: str) -> str:
     return f"\033]8;;{url}\033\\{text}\033]8;;\033\\"
 
 
-def honcho_session_line(workspace: str, session_name: str) -> str:
-    """One-line session indicator: `Honcho session: <clickable name>`."""
-    url = honcho_session_url(workspace, session_name)
-    linked_name = _osc8_link(url, f"{_SKY_BLUE}{session_name}{_ANSI_RESET}")
-    return f"{_DIM}Honcho session:{_ANSI_RESET} {linked_name}"
-
-
-def write_tty(text: str) -> None:
-    """Write directly to /dev/tty, bypassing stdout capture."""
-    try:
-        fd = os.open("/dev/tty", os.O_WRONLY)
-        os.write(fd, text.encode("utf-8"))
-        os.close(fd)
-    except OSError:
-        sys.stdout.write(text)
-        sys.stdout.flush()
-
-
 # =========================================================================
 # Context pressure display (CLI user-facing warnings)
 # =========================================================================

agent/memory_manager.py

@@ -34,6 +34,7 @@ import re
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -249,7 +250,7 @@ class MemoryManager:
         """
         provider = self._tool_to_provider.get(tool_name)
         if provider is None:
-            return json.dumps({"error": f"No memory provider handles tool '{tool_name}'"})
+            return tool_error(f"No memory provider handles tool '{tool_name}'")
         try:
             return provider.handle_tool_call(tool_name, args, **kwargs)
         except Exception as e:
@@ -257,7 +258,7 @@ class MemoryManager:
                 "Memory provider '%s' handle_tool_call(%s) failed: %s",
                 provider.name, tool_name, e,
             )
-            return json.dumps({"error": f"Memory tool '{tool_name}' failed: {e}"})
+            return tool_error(f"Memory tool '{tool_name}' failed: {e}")
 
     # -- Lifecycle hooks -----------------------------------------------------
 

cli.py

@@ -3536,13 +3536,6 @@ class HermesCLI:
         _cprint(f"  Original session: {parent_session_id}")
         _cprint(f"  Branch session:   {new_session_id}")
 
-    def reset_conversation(self):
-        """Reset the conversation by starting a new session."""
-        # Shut down memory provider before resetting — actual session boundary
-        if hasattr(self, 'agent') and self.agent:
-            self.agent.shutdown_memory_provider(self.conversation_history)
-        self.new_session()
-    
     def save_conversation(self):
         """Save the current conversation to a file."""
         if not self.conversation_history:
@@ -6290,14 +6283,6 @@ class HermesCLI:
             except Exception:
                 pass
 
-    def _clear_current_input(self) -> None:
-        if getattr(self, "_app", None):
-            try:
-                self._app.current_buffer.text = ""
-            except Exception:
-                pass
-
-
     def chat(self, message, images: list = None) -> Optional[str]:
         """
         Send a message to the agent and get a response.

cron/scheduler.py

@@ -158,6 +158,44 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
     }
 
 
+# Media extension sets — keep in sync with gateway/platforms/base.py:_process_message_background
+_AUDIO_EXTS = frozenset({'.ogg', '.opus', '.mp3', '.wav', '.m4a'})
+_VIDEO_EXTS = frozenset({'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'})
+_IMAGE_EXTS = frozenset({'.jpg', '.jpeg', '.png', '.webp', '.gif'})
+
+
+def _send_media_via_adapter(adapter, chat_id: str, media_files: list, metadata: dict | None, loop, job: dict) -> None:
+    """Send extracted MEDIA files as native platform attachments via a live adapter.
+
+    Routes each file to the appropriate adapter method (send_voice, send_image_file,
+    send_video, send_document) based on file extension — mirroring the routing logic
+    in ``BasePlatformAdapter._process_message_background``.
+    """
+    from pathlib import Path
+
+    for media_path, _is_voice in media_files:
+        try:
+            ext = Path(media_path).suffix.lower()
+            if ext in _AUDIO_EXTS:
+                coro = adapter.send_voice(chat_id=chat_id, audio_path=media_path, metadata=metadata)
+            elif ext in _VIDEO_EXTS:
+                coro = adapter.send_video(chat_id=chat_id, video_path=media_path, metadata=metadata)
+            elif ext in _IMAGE_EXTS:
+                coro = adapter.send_image_file(chat_id=chat_id, image_path=media_path, metadata=metadata)
+            else:
+                coro = adapter.send_document(chat_id=chat_id, file_path=media_path, metadata=metadata)
+
+            future = asyncio.run_coroutine_threadsafe(coro, loop)
+            result = future.result(timeout=30)
+            if result and not getattr(result, "success", True):
+                logger.warning(
+                    "Job '%s': media send failed for %s: %s",
+                    job.get("id", "?"), media_path, getattr(result, "error", "unknown"),
+                )
+        except Exception as e:
+            logger.warning("Job '%s': failed to send media %s: %s", job.get("id", "?"), media_path, e)
+
+
 def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> None:
     """
     Deliver job output to the configured target (origin chat, specific platform, etc.).
@@ -246,18 +284,28 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> None:
     if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
         send_metadata = {"thread_id": thread_id} if thread_id else None
         try:
-            future = asyncio.run_coroutine_threadsafe(
-                runtime_adapter.send(chat_id, delivery_content, metadata=send_metadata),
-                loop,
-            )
-            send_result = future.result(timeout=60)
-            if send_result and not getattr(send_result, "success", True):
-                err = getattr(send_result, "error", "unknown")
-                logger.warning(
-                    "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
-                    job["id"], platform_name, chat_id, err,
+            # Send cleaned text (MEDIA tags stripped) — not the raw content
+            text_to_send = cleaned_delivery_content.strip()
+            adapter_ok = True
+            if text_to_send:
+                future = asyncio.run_coroutine_threadsafe(
+                    runtime_adapter.send(chat_id, text_to_send, metadata=send_metadata),
+                    loop,
                 )
-            else:
+                send_result = future.result(timeout=60)
+                if send_result and not getattr(send_result, "success", True):
+                    err = getattr(send_result, "error", "unknown")
+                    logger.warning(
+                        "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
+                        job["id"], platform_name, chat_id, err,
+                    )
+                    adapter_ok = False  # fall through to standalone path
+
+            # Send extracted media files as native attachments via the live adapter
+            if adapter_ok and media_files:
+                _send_media_via_adapter(runtime_adapter, chat_id, media_files, send_metadata, loop, job)
+
+            if adapter_ok:
                 logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
                 return
         except Exception as e:

gateway/delivery.py

@@ -314,38 +314,4 @@ def parse_deliver_spec(
     return deliver
 
 
-def build_delivery_context_for_tool(
-    config: GatewayConfig,
-    origin: Optional[SessionSource] = None
-) -> Dict[str, Any]:
-    """
-    Build context for the unified cronjob tool to understand delivery options.
-    
-    This is passed to the tool so it can validate and explain delivery targets.
-    """
-    connected = config.get_connected_platforms()
-    
-    options = {
-        "origin": {
-            "description": "Back to where this job was created",
-            "available": origin is not None,
-        },
-        "local": {
-            "description": "Save to local files only",
-            "available": True,
-        }
-    }
-    
-    for platform in connected:
-        home = config.get_home_channel(platform)
-        options[platform.value] = {
-            "description": f"{platform.value.title()} home channel",
-            "available": True,
-            "home_channel": home.to_dict() if home else None,
-        }
-    
-    return {
-        "origin": origin.to_dict() if origin else None,
-        "options": options,
-        "always_log_local": config.always_log_local,
-    }
+

gateway/platforms/feishu.py

@@ -387,10 +387,6 @@ def _coerce_required_int(value: Any, default: int, min_value: int = 0) -> int:
     return default if parsed is None else parsed
 
 
-def _is_loop_ready(loop: Optional[asyncio.AbstractEventLoop]) -> bool:
-    return loop is not None and not bool(getattr(loop, "is_closed", lambda: False)())
-
-
 # ---------------------------------------------------------------------------
 # Post payload builders and parsers
 # ---------------------------------------------------------------------------

gateway/platforms/slack.py

@@ -87,6 +87,14 @@ class SlackAdapter(BasePlatformAdapter):
         # Track pending approval message_ts → resolved flag to prevent
         # double-clicks on approval buttons.
         self._approval_resolved: Dict[str, bool] = {}
+        # Track timestamps of messages sent by the bot so we can respond
+        # to thread replies even without an explicit @mention.
+        self._bot_message_ts: set = set()
+        self._BOT_TS_MAX = 5000  # cap to avoid unbounded growth
+        # Track threads where the bot has been @mentioned — once mentioned,
+        # respond to ALL subsequent messages in that thread automatically.
+        self._mentioned_threads: set = set()
+        self._MENTIONED_THREADS_MAX = 5000
 
     async def connect(self) -> bool:
         """Connect to Slack via Socket Mode."""
@@ -268,9 +276,22 @@ class SlackAdapter(BasePlatformAdapter):
 
                 last_result = await self._get_client(chat_id).chat_postMessage(**kwargs)
 
+            # Track the sent message ts so we can auto-respond to thread
+            # replies without requiring @mention.
+            sent_ts = last_result.get("ts") if last_result else None
+            if sent_ts:
+                self._bot_message_ts.add(sent_ts)
+                # Also register the thread root so replies-to-my-replies work
+                if thread_ts:
+                    self._bot_message_ts.add(thread_ts)
+                if len(self._bot_message_ts) > self._BOT_TS_MAX:
+                    excess = len(self._bot_message_ts) - self._BOT_TS_MAX // 2
+                    for old_ts in list(self._bot_message_ts)[:excess]:
+                        self._bot_message_ts.discard(old_ts)
+
             return SendResult(
                 success=True,
-                message_id=last_result.get("ts") if last_result else None,
+                message_id=sent_ts,
                 raw_response=last_result,
             )
 
@@ -778,48 +799,61 @@ class SlackAdapter(BasePlatformAdapter):
         else:
             thread_ts = event.get("thread_ts") or ts  # ts fallback for channels
 
-        # In channels, only respond if bot is mentioned OR if this is a
-        # reply in a thread where the bot has an active session.
+        # In channels, respond if:
+        #   1. The bot is @mentioned in this message, OR
+        #   2. The message is a reply in a thread the bot started/participated in, OR
+        #   3. The message is in a thread where the bot was previously @mentioned, OR
+        #   4. There's an existing session for this thread (survives restarts)
         bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id)
         is_mentioned = bot_uid and f"<@{bot_uid}>" in text
-        
+        event_thread_ts = event.get("thread_ts")
+        is_thread_reply = bool(event_thread_ts and event_thread_ts != ts)
+
         if not is_dm and bot_uid and not is_mentioned:
-            # Check if this is a thread reply (thread_ts exists and differs from ts)
-            event_thread_ts = event.get("thread_ts")
-            is_thread_reply = event_thread_ts and event_thread_ts != ts
-            
-            if is_thread_reply and self._has_active_session_for_thread(
-                channel_id=channel_id,
-                thread_ts=event_thread_ts,
-                user_id=user_id,
-            ):
-                # Allow thread replies without mention if there's an active session
-                pass
-            else:
-                # Not a thread reply or no active session - ignore
+            reply_to_bot_thread = (
+                is_thread_reply and event_thread_ts in self._bot_message_ts
+            )
+            in_mentioned_thread = (
+                event_thread_ts is not None
+                and event_thread_ts in self._mentioned_threads
+            )
+            has_session = (
+                is_thread_reply
+                and self._has_active_session_for_thread(
+                    channel_id=channel_id,
+                    thread_ts=event_thread_ts,
+                    user_id=user_id,
+                )
+            )
+            if not reply_to_bot_thread and not in_mentioned_thread and not has_session:
                 return
-        
+
         if is_mentioned:
             # Strip the bot mention from the text
             text = text.replace(f"<@{bot_uid}>", "").strip()
+            # Register this thread so all future messages auto-trigger the bot
+            if event_thread_ts:
+                self._mentioned_threads.add(event_thread_ts)
+                if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX:
+                    to_remove = list(self._mentioned_threads)[:self._MENTIONED_THREADS_MAX // 2]
+                    for t in to_remove:
+                        self._mentioned_threads.discard(t)
 
-            # When first mentioned in an existing thread, fetch thread context
-            # so the agent understands the conversation it's joining.
-            event_thread_ts = event.get("thread_ts")
-            is_thread_reply = event_thread_ts and event_thread_ts != ts
-            if is_thread_reply and not self._has_active_session_for_thread(
+        # When entering a thread for the first time (no existing session),
+        # fetch thread context so the agent understands the conversation.
+        if is_thread_reply and not self._has_active_session_for_thread(
+            channel_id=channel_id,
+            thread_ts=event_thread_ts,
+            user_id=user_id,
+        ):
+            thread_context = await self._fetch_thread_context(
                 channel_id=channel_id,
                 thread_ts=event_thread_ts,
-                user_id=user_id,
-            ):
-                thread_context = await self._fetch_thread_context(
-                    channel_id=channel_id,
-                    thread_ts=event_thread_ts,
-                    current_ts=ts,
-                    team_id=team_id,
-                )
-                if thread_context:
-                    text = thread_context + text
+                current_ts=ts,
+                team_id=team_id,
+            )
+            if thread_context:
+                text = thread_context + text
 
         # Determine message type
         msg_type = MessageType.TEXT

gateway/run.py

@@ -4551,6 +4551,7 @@ class GatewayRunner:
                     provider_data_collection=pr.get("data_collection"),
                     session_id=task_id,
                     platform=platform_key,
+                    user_id=source.user_id,
                     session_db=self._session_db,
                     fallback_model=self._fallback_model,
                 )
@@ -6645,6 +6646,7 @@ class GatewayRunner:
                     provider_data_collection=pr.get("data_collection"),
                     session_id=session_id,
                     platform=platform_key,
+                    user_id=source.user_id,
                     session_db=self._session_db,
                     fallback_model=self._fallback_model,
                 )

hermes_cli/callbacks.py

@@ -63,47 +63,6 @@ def clarify_callback(cli, question, choices):
     )
 
 
-def sudo_password_callback(cli) -> str:
-    """Prompt for sudo password through the TUI.
-
-    Sets up a password input area and blocks until the user responds.
-    """
-    timeout = 45
-    response_queue = queue.Queue()
-
-    cli._sudo_state = {"response_queue": response_queue}
-    cli._sudo_deadline = _time.monotonic() + timeout
-
-    if hasattr(cli, "_app") and cli._app:
-        cli._app.invalidate()
-
-    while True:
-        try:
-            result = response_queue.get(timeout=1)
-            cli._sudo_state = None
-            cli._sudo_deadline = 0
-            if hasattr(cli, "_app") and cli._app:
-                cli._app.invalidate()
-            if result:
-                cprint(f"\n{_DIM}  ✓ Password received (cached for session){_RST}")
-            else:
-                cprint(f"\n{_DIM}  ⏭ Skipped{_RST}")
-            return result
-        except queue.Empty:
-            remaining = cli._sudo_deadline - _time.monotonic()
-            if remaining <= 0:
-                break
-            if hasattr(cli, "_app") and cli._app:
-                cli._app.invalidate()
-
-    cli._sudo_state = None
-    cli._sudo_deadline = 0
-    if hasattr(cli, "_app") and cli._app:
-        cli._app.invalidate()
-    cprint(f"\n{_DIM}  ⏱ Timeout — continuing without sudo{_RST}")
-    return ""
-
-
 def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict:
     """Prompt for a secret value through the TUI (e.g. API keys for skills).
 

hermes_cli/clipboard.py

@@ -1,4 +1,4 @@
-"""Clipboard image extraction for macOS, Linux, and WSL2.
+"""Clipboard image extraction for macOS, Windows, Linux, and WSL2.
 
 Provides a single function `save_clipboard_image(dest)` that checks the
 system clipboard for image data, saves it to *dest* as PNG, and returns
@@ -6,9 +6,10 @@ True on success.  No external Python dependencies — uses only OS-level
 CLI tools that ship with the platform (or are commonly installed).
 
 Platform support:
-  macOS  — osascript (always available), pngpaste (if installed)
-  WSL2   — powershell.exe via .NET System.Windows.Forms.Clipboard
-  Linux  — wl-paste (Wayland), xclip (X11)
+  macOS   — osascript (always available), pngpaste (if installed)
+  Windows — PowerShell via .NET System.Windows.Forms.Clipboard
+  WSL2    — powershell.exe via .NET System.Windows.Forms.Clipboard
+  Linux   — wl-paste (Wayland), xclip (X11)
 """
 
 import base64
@@ -32,6 +33,8 @@ def save_clipboard_image(dest: Path) -> bool:
     dest.parent.mkdir(parents=True, exist_ok=True)
     if sys.platform == "darwin":
         return _macos_save(dest)
+    if sys.platform == "win32":
+        return _windows_save(dest)
     return _linux_save(dest)
 
 
@@ -42,6 +45,8 @@ def has_clipboard_image() -> bool:
     """
     if sys.platform == "darwin":
         return _macos_has_image()
+    if sys.platform == "win32":
+        return _windows_has_image()
     if _is_wsl():
         return _wsl_has_image()
     if os.environ.get("WAYLAND_DISPLAY"):
@@ -112,6 +117,104 @@ def _macos_osascript(dest: Path) -> bool:
     return False
 
 
+# ── Shared PowerShell scripts (native Windows + WSL2) ─────────────────────
+
+# .NET System.Windows.Forms.Clipboard — used by both native Windows (powershell)
+# and WSL2 (powershell.exe) paths.
+_PS_CHECK_IMAGE = (
+    "Add-Type -AssemblyName System.Windows.Forms;"
+    "[System.Windows.Forms.Clipboard]::ContainsImage()"
+)
+
+_PS_EXTRACT_IMAGE = (
+    "Add-Type -AssemblyName System.Windows.Forms;"
+    "Add-Type -AssemblyName System.Drawing;"
+    "$img = [System.Windows.Forms.Clipboard]::GetImage();"
+    "if ($null -eq $img) { exit 1 }"
+    "$ms = New-Object System.IO.MemoryStream;"
+    "$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);"
+    "[System.Convert]::ToBase64String($ms.ToArray())"
+)
+
+
+# ── Native Windows ────────────────────────────────────────────────────────
+
+# Native Windows uses ``powershell`` (Windows PowerShell 5.1, always present)
+# or ``pwsh`` (PowerShell 7+, optional).  Discovery is cached per-process.
+
+
+def _find_powershell() -> str | None:
+    """Return the first available PowerShell executable, or None."""
+    for name in ("powershell", "pwsh"):
+        try:
+            r = subprocess.run(
+                [name, "-NoProfile", "-NonInteractive", "-Command", "echo ok"],
+                capture_output=True, text=True, timeout=5,
+            )
+            if r.returncode == 0 and "ok" in r.stdout:
+                return name
+        except FileNotFoundError:
+            continue
+        except Exception:
+            continue
+    return None
+
+
+# Cache the resolved PowerShell executable (checked once per process)
+_ps_exe: str | None | bool = False  # False = not yet checked
+
+
+def _get_ps_exe() -> str | None:
+    global _ps_exe
+    if _ps_exe is False:
+        _ps_exe = _find_powershell()
+    return _ps_exe
+
+
+def _windows_has_image() -> bool:
+    """Check if the Windows clipboard contains an image."""
+    ps = _get_ps_exe()
+    if ps is None:
+        return False
+    try:
+        r = subprocess.run(
+            [ps, "-NoProfile", "-NonInteractive", "-Command", _PS_CHECK_IMAGE],
+            capture_output=True, text=True, timeout=5,
+        )
+        return r.returncode == 0 and "True" in r.stdout
+    except Exception as e:
+        logger.debug("Windows clipboard image check failed: %s", e)
+    return False
+
+
+def _windows_save(dest: Path) -> bool:
+    """Extract clipboard image on native Windows via PowerShell → base64 PNG."""
+    ps = _get_ps_exe()
+    if ps is None:
+        logger.debug("No PowerShell found — Windows clipboard image paste unavailable")
+        return False
+    try:
+        r = subprocess.run(
+            [ps, "-NoProfile", "-NonInteractive", "-Command", _PS_EXTRACT_IMAGE],
+            capture_output=True, text=True, timeout=15,
+        )
+        if r.returncode != 0:
+            return False
+
+        b64_data = r.stdout.strip()
+        if not b64_data:
+            return False
+
+        png_bytes = base64.b64decode(b64_data)
+        dest.write_bytes(png_bytes)
+        return dest.exists() and dest.stat().st_size > 0
+
+    except Exception as e:
+        logger.debug("Windows clipboard image extraction failed: %s", e)
+        dest.unlink(missing_ok=True)
+    return False
+
+
 # ── Linux ────────────────────────────────────────────────────────────────
 
 def _is_wsl() -> bool:
@@ -142,24 +245,7 @@ def _linux_save(dest: Path) -> bool:
 
 
 # ── WSL2 (powershell.exe) ────────────────────────────────────────────────
-
-# PowerShell script: get clipboard image as base64-encoded PNG on stdout.
-# Using .NET System.Windows.Forms.Clipboard — always available on Windows.
-_PS_CHECK_IMAGE = (
-    "Add-Type -AssemblyName System.Windows.Forms;"
-    "[System.Windows.Forms.Clipboard]::ContainsImage()"
-)
-
-_PS_EXTRACT_IMAGE = (
-    "Add-Type -AssemblyName System.Windows.Forms;"
-    "Add-Type -AssemblyName System.Drawing;"
-    "$img = [System.Windows.Forms.Clipboard]::GetImage();"
-    "if ($null -eq $img) { exit 1 }"
-    "$ms = New-Object System.IO.MemoryStream;"
-    "$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);"
-    "[System.Convert]::ToBase64String($ms.ToArray())"
-)
-
+# Reuses _PS_CHECK_IMAGE / _PS_EXTRACT_IMAGE defined above.
 
 def _wsl_has_image() -> bool:
     """Check if Windows clipboard has an image (via powershell.exe)."""

hermes_cli/config.py

@@ -1881,6 +1881,24 @@ def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]:
 
 
 
+def read_raw_config() -> Dict[str, Any]:
+    """Read ~/.hermes/config.yaml as-is, without merging defaults or migrating.
+
+    Returns the raw YAML dict, or ``{}`` if the file doesn't exist or can't
+    be parsed.  Use this for lightweight config reads where you just need a
+    single value and don't want the overhead of ``load_config()``'s deep-merge
+    + migration pipeline.
+    """
+    try:
+        config_path = get_config_path()
+        if config_path.exists():
+            with open(config_path, encoding="utf-8") as f:
+                return yaml.safe_load(f) or {}
+    except Exception:
+        pass
+    return {}
+
+
 def load_config() -> Dict[str, Any]:
     """Load configuration from ~/.hermes/config.yaml."""
     import copy

hermes_cli/models.py

@@ -1131,10 +1131,6 @@ def _payload_items(payload: Any) -> list[dict[str, Any]]:
     return []
 
 
-def _extract_model_ids(payload: Any) -> list[str]:
-    return [item.get("id", "") for item in _payload_items(payload) if item.get("id")]
-
-
 def copilot_default_headers() -> dict[str, str]:
     """Standard headers for Copilot API requests.
 

hermes_cli/providers.py

@@ -344,18 +344,6 @@ def get_label(provider_id: str) -> str:
     return canonical
 
 
-# Build LABELS dict for backward compat
-def _build_labels() -> Dict[str, str]:
-    """Build labels dict from overlays + overrides. Lazy, cached."""
-    labels: Dict[str, str] = {}
-    for pid in HERMES_OVERLAYS:
-        labels[pid] = get_label(pid)
-    labels.update(_LABEL_OVERRIDES)
-    return labels
-
-# Lazy-built on first access
-_labels_cache: Optional[Dict[str, str]] = None
-
 # For direct import compat, expose as module-level dict
 # Built on demand by get_label() calls
 LABELS: Dict[str, str] = {

hermes_cli/setup.py

@@ -42,18 +42,6 @@ def _model_config_dict(config: Dict[str, Any]) -> Dict[str, Any]:
     return {}
 
 
-def _set_model_provider(
-    config: Dict[str, Any], provider_id: str, base_url: str = ""
-) -> None:
-    model_cfg = _model_config_dict(config)
-    model_cfg["provider"] = provider_id
-    if base_url:
-        model_cfg["base_url"] = base_url.rstrip("/")
-    else:
-        model_cfg.pop("base_url", None)
-    config["model"] = model_cfg
-
-
 def _set_default_model(config: Dict[str, Any], model_name: str) -> None:
     if not model_name:
         return
@@ -326,16 +314,6 @@ def _setup_provider_model_selection(config, provider_id, current_model, prompt_c
         config["model"] = model_cfg
 
 
-def _sync_model_from_disk(config: Dict[str, Any]) -> None:
-    disk_model = load_config().get("model")
-    if isinstance(disk_model, dict):
-        model_cfg = _model_config_dict(config)
-        model_cfg.update(disk_model)
-        config["model"] = model_cfg
-    elif isinstance(disk_model, str) and disk_model.strip():
-        _set_default_model(config, disk_model.strip())
-
-
 # Import config helpers
 from hermes_cli.config import (
     get_hermes_home,

hermes_cli/tools_config.py

@@ -61,22 +61,6 @@ def _prompt(question: str, default: str = None, password: bool = False) -> str:
         print()
         return default or ""
 
-def _prompt_yes_no(question: str, default: bool = True) -> bool:
-    default_str = "Y/n" if default else "y/N"
-    while True:
-        try:
-            value = input(color(f"{question} [{default_str}]: ", Colors.YELLOW)).strip().lower()
-        except (KeyboardInterrupt, EOFError):
-            print()
-            return default
-        if not value:
-            return default
-        if value in ('y', 'yes'):
-            return True
-        if value in ('n', 'no'):
-            return False
-
-
 # ─── Toolset Registry ─────────────────────────────────────────────────────────
 
 # Toolsets shown in the configurator, grouped for display.

hermes_cli/uninstall.py

@@ -23,10 +23,6 @@ def log_success(msg: str):
 def log_warn(msg: str):
     print(f"{color('⚠', Colors.YELLOW)} {msg}")
 
-def log_error(msg: str):
-    print(f"{color('✗', Colors.RED)} {msg}")
-
-
 def get_project_root() -> Path:
     """Get the project installation directory."""
     return Path(__file__).parent.parent.resolve()

plugins/memory/byterover/__init__.py

@@ -27,6 +27,7 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -320,7 +321,7 @@ class ByteRoverMemoryProvider(MemoryProvider):
             return self._tool_curate(args)
         elif tool_name == "brv_status":
             return self._tool_status()
-        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        return tool_error(f"Unknown tool: {tool_name}")
 
     def shutdown(self) -> None:
         if self._sync_thread and self._sync_thread.is_alive():
@@ -331,7 +332,7 @@ class ByteRoverMemoryProvider(MemoryProvider):
     def _tool_query(self, args: dict) -> str:
         query = args.get("query", "")
         if not query:
-            return json.dumps({"error": "query is required"})
+            return tool_error("query is required")
 
         result = _run_brv(
             ["query", "--", query.strip()[:5000]],
@@ -339,7 +340,7 @@ class ByteRoverMemoryProvider(MemoryProvider):
         )
 
         if not result["success"]:
-            return json.dumps({"error": result.get("error", "Query failed")})
+            return tool_error(result.get("error", "Query failed"))
 
         output = result.get("output", "").strip()
         if not output or len(output) < _MIN_OUTPUT_LEN:
@@ -354,7 +355,7 @@ class ByteRoverMemoryProvider(MemoryProvider):
     def _tool_curate(self, args: dict) -> str:
         content = args.get("content", "")
         if not content:
-            return json.dumps({"error": "content is required"})
+            return tool_error("content is required")
 
         result = _run_brv(
             ["curate", "--", content],
@@ -362,14 +363,14 @@ class ByteRoverMemoryProvider(MemoryProvider):
         )
 
         if not result["success"]:
-            return json.dumps({"error": result.get("error", "Curate failed")})
+            return tool_error(result.get("error", "Curate failed"))
 
         return json.dumps({"result": "Memory curated successfully."})
 
     def _tool_status(self) -> str:
         result = _run_brv(["status"], timeout=15, cwd=self._cwd)
         if not result["success"]:
-            return json.dumps({"error": result.get("error", "Status check failed")})
+            return tool_error(result.get("error", "Status check failed"))
         return json.dumps({"status": result.get("output", "")})
 
 

plugins/memory/hindsight/__init__.py

@@ -26,6 +26,7 @@ import threading
 from typing import Any, Dict, List
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -433,12 +434,12 @@ class HindsightMemoryProvider(MemoryProvider):
             client = self._get_client()
         except Exception as e:
             logger.warning("Hindsight client init failed: %s", e)
-            return json.dumps({"error": f"Hindsight client unavailable: {e}"})
+            return tool_error(f"Hindsight client unavailable: {e}")
 
         if tool_name == "hindsight_retain":
             content = args.get("content", "")
             if not content:
-                return json.dumps({"error": "Missing required parameter: content"})
+                return tool_error("Missing required parameter: content")
             context = args.get("context")
             try:
                 _run_sync(client.aretain(
@@ -447,12 +448,12 @@ class HindsightMemoryProvider(MemoryProvider):
                 return json.dumps({"result": "Memory stored successfully."})
             except Exception as e:
                 logger.warning("hindsight_retain failed: %s", e)
-                return json.dumps({"error": f"Failed to store memory: {e}"})
+                return tool_error(f"Failed to store memory: {e}")
 
         elif tool_name == "hindsight_recall":
             query = args.get("query", "")
             if not query:
-                return json.dumps({"error": "Missing required parameter: query"})
+                return tool_error("Missing required parameter: query")
             try:
                 resp = _run_sync(client.arecall(
                     bank_id=self._bank_id, query=query, budget=self._budget
@@ -463,12 +464,12 @@ class HindsightMemoryProvider(MemoryProvider):
                 return json.dumps({"result": "\n".join(lines)})
             except Exception as e:
                 logger.warning("hindsight_recall failed: %s", e)
-                return json.dumps({"error": f"Failed to search memory: {e}"})
+                return tool_error(f"Failed to search memory: {e}")
 
         elif tool_name == "hindsight_reflect":
             query = args.get("query", "")
             if not query:
-                return json.dumps({"error": "Missing required parameter: query"})
+                return tool_error("Missing required parameter: query")
             try:
                 resp = _run_sync(client.areflect(
                     bank_id=self._bank_id, query=query, budget=self._budget
@@ -476,9 +477,9 @@ class HindsightMemoryProvider(MemoryProvider):
                 return json.dumps({"result": resp.text or "No relevant memories found."})
             except Exception as e:
                 logger.warning("hindsight_reflect failed: %s", e)
-                return json.dumps({"error": f"Failed to reflect: {e}"})
+                return tool_error(f"Failed to reflect: {e}")
 
-        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        return tool_error(f"Unknown tool: {tool_name}")
 
     def shutdown(self) -> None:
         global _loop, _loop_thread

plugins/memory/holographic/__init__.py

@@ -23,6 +23,7 @@ import re
 from typing import Any, Dict, List
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 from .store import MemoryStore
 from .retrieval import FactRetriever
 
@@ -230,7 +231,7 @@ class HolographicMemoryProvider(MemoryProvider):
             return self._handle_fact_store(args)
         elif tool_name == "fact_feedback":
             return self._handle_fact_feedback(args)
-        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        return tool_error(f"Unknown tool: {tool_name}")
 
     def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
         if not self._config.get("auto_extract", False):
@@ -296,7 +297,7 @@ class HolographicMemoryProvider(MemoryProvider):
             elif action == "reason":
                 entities = args.get("entities", [])
                 if not entities:
-                    return json.dumps({"error": "reason requires 'entities' list"})
+                    return tool_error("reason requires 'entities' list")
                 results = retriever.reason(
                     entities,
                     category=args.get("category"),
@@ -334,12 +335,12 @@ class HolographicMemoryProvider(MemoryProvider):
                 return json.dumps({"facts": facts, "count": len(facts)})
 
             else:
-                return json.dumps({"error": f"Unknown action: {action}"})
+                return tool_error(f"Unknown action: {action}")
 
         except KeyError as exc:
-            return json.dumps({"error": f"Missing required argument: {exc}"})
+            return tool_error(f"Missing required argument: {exc}")
         except Exception as exc:
-            return json.dumps({"error": str(exc)})
+            return tool_error(str(exc))
 
     def _handle_fact_feedback(self, args: dict) -> str:
         try:
@@ -348,9 +349,9 @@ class HolographicMemoryProvider(MemoryProvider):
             result = self._store.record_feedback(fact_id, helpful=helpful)
             return json.dumps(result)
         except KeyError as exc:
-            return json.dumps({"error": f"Missing required argument: {exc}"})
+            return tool_error(f"Missing required argument: {exc}")
         except Exception as exc:
-            return json.dumps({"error": str(exc)})
+            return tool_error(str(exc))
 
     # -- Auto-extraction (on_session_end) ------------------------------------
 

plugins/memory/honcho/__init__.py

@@ -21,6 +21,7 @@ import threading
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -216,6 +217,12 @@ class HonchoMemoryProvider(MemoryProvider):
                 logger.debug("Honcho not configured — plugin inactive")
                 return
 
+            # Override peer_name with gateway user_id for per-user memory scoping.
+            # CLI sessions won't have user_id, so the config default is preserved.
+            _gw_user_id = kwargs.get("user_id")
+            if _gw_user_id:
+                cfg.peer_name = _gw_user_id
+
             self._config = cfg
 
             # ----- B1: recall_mode from config -----
@@ -632,15 +639,15 @@ class HonchoMemoryProvider(MemoryProvider):
     def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
         """Handle a Honcho tool call, with lazy session init for tools-only mode."""
         if self._cron_skipped:
-            return json.dumps({"error": "Honcho is not active (cron context)."})
+            return tool_error("Honcho is not active (cron context).")
 
         # Port #1957: ensure session is initialized for tools-only mode
         if not self._session_initialized:
             if not self._ensure_session():
-                return json.dumps({"error": "Honcho session could not be initialized."})
+                return tool_error("Honcho session could not be initialized.")
 
         if not self._manager or not self._session_key:
-            return json.dumps({"error": "Honcho is not active for this session."})
+            return tool_error("Honcho is not active for this session.")
 
         try:
             if tool_name == "honcho_profile":
@@ -652,7 +659,7 @@ class HonchoMemoryProvider(MemoryProvider):
             elif tool_name == "honcho_search":
                 query = args.get("query", "")
                 if not query:
-                    return json.dumps({"error": "Missing required parameter: query"})
+                    return tool_error("Missing required parameter: query")
                 max_tokens = min(int(args.get("max_tokens", 800)), 2000)
                 result = self._manager.search_context(
                     self._session_key, query, max_tokens=max_tokens
@@ -664,7 +671,7 @@ class HonchoMemoryProvider(MemoryProvider):
             elif tool_name == "honcho_context":
                 query = args.get("query", "")
                 if not query:
-                    return json.dumps({"error": "Missing required parameter: query"})
+                    return tool_error("Missing required parameter: query")
                 peer = args.get("peer", "user")
                 result = self._manager.dialectic_query(
                     self._session_key, query, peer=peer
@@ -674,17 +681,17 @@ class HonchoMemoryProvider(MemoryProvider):
             elif tool_name == "honcho_conclude":
                 conclusion = args.get("conclusion", "")
                 if not conclusion:
-                    return json.dumps({"error": "Missing required parameter: conclusion"})
+                    return tool_error("Missing required parameter: conclusion")
                 ok = self._manager.create_conclusion(self._session_key, conclusion)
                 if ok:
                     return json.dumps({"result": f"Conclusion saved: {conclusion}"})
-                return json.dumps({"error": "Failed to save conclusion."})
+                return tool_error("Failed to save conclusion.")
 
-            return json.dumps({"error": f"Unknown tool: {tool_name}"})
+            return tool_error(f"Unknown tool: {tool_name}")
 
         except Exception as e:
             logger.error("Honcho tool %s failed: %s", tool_name, e)
-            return json.dumps({"error": f"Honcho {tool_name} failed: {e}"})
+            return tool_error(f"Honcho {tool_name} failed: {e}")
 
     def shutdown(self) -> None:
         for t in (self._prefetch_thread, self._sync_thread):

plugins/memory/mem0/__init__.py

@@ -23,6 +23,7 @@ import time
 from typing import Any, Dict, List
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -202,7 +203,9 @@ class Mem0MemoryProvider(MemoryProvider):
     def initialize(self, session_id: str, **kwargs) -> None:
         self._config = _load_config()
         self._api_key = self._config.get("api_key", "")
-        self._user_id = self._config.get("user_id", "hermes-user")
+        # Prefer gateway-provided user_id for per-user memory scoping;
+        # fall back to config/env default for CLI (single-user) sessions.
+        self._user_id = kwargs.get("user_id") or self._config.get("user_id", "hermes-user")
         self._agent_id = self._config.get("agent_id", "hermes")
         self._rerank = self._config.get("rerank", True)
 
@@ -303,7 +306,7 @@ class Mem0MemoryProvider(MemoryProvider):
         try:
             client = self._get_client()
         except Exception as e:
-            return json.dumps({"error": str(e)})
+            return tool_error(str(e))
 
         if tool_name == "mem0_profile":
             try:
@@ -315,12 +318,12 @@ class Mem0MemoryProvider(MemoryProvider):
                 return json.dumps({"result": "\n".join(lines), "count": len(lines)})
             except Exception as e:
                 self._record_failure()
-                return json.dumps({"error": f"Failed to fetch profile: {e}"})
+                return tool_error(f"Failed to fetch profile: {e}")
 
         elif tool_name == "mem0_search":
             query = args.get("query", "")
             if not query:
-                return json.dumps({"error": "Missing required parameter: query"})
+                return tool_error("Missing required parameter: query")
             rerank = args.get("rerank", False)
             top_k = min(int(args.get("top_k", 10)), 50)
             try:
@@ -337,12 +340,12 @@ class Mem0MemoryProvider(MemoryProvider):
                 return json.dumps({"results": items, "count": len(items)})
             except Exception as e:
                 self._record_failure()
-                return json.dumps({"error": f"Search failed: {e}"})
+                return tool_error(f"Search failed: {e}")
 
         elif tool_name == "mem0_conclude":
             conclusion = args.get("conclusion", "")
             if not conclusion:
-                return json.dumps({"error": "Missing required parameter: conclusion"})
+                return tool_error("Missing required parameter: conclusion")
             try:
                 client.add(
                     [{"role": "user", "content": conclusion}],
@@ -353,9 +356,9 @@ class Mem0MemoryProvider(MemoryProvider):
                 return json.dumps({"result": "Fact stored."})
             except Exception as e:
                 self._record_failure()
-                return json.dumps({"error": f"Failed to store: {e}"})
+                return tool_error(f"Failed to store: {e}")
 
-        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        return tool_error(f"Unknown tool: {tool_name}")
 
     def shutdown(self) -> None:
         for t in (self._prefetch_thread, self._sync_thread):

plugins/memory/openviking/__init__.py

@@ -31,6 +31,7 @@ import threading
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -461,7 +462,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
 
     def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
         if not self._client:
-            return json.dumps({"error": "OpenViking server not connected"})
+            return tool_error("OpenViking server not connected")
 
         try:
             if tool_name == "viking_search":
@@ -474,9 +475,9 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 return self._tool_remember(args)
             elif tool_name == "viking_add_resource":
                 return self._tool_add_resource(args)
-            return json.dumps({"error": f"Unknown tool: {tool_name}"})
+            return tool_error(f"Unknown tool: {tool_name}")
         except Exception as e:
-            return json.dumps({"error": str(e)})
+            return tool_error(str(e))
 
     def shutdown(self) -> None:
         # Wait for background threads to finish
@@ -493,7 +494,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def _tool_search(self, args: dict) -> str:
         query = args.get("query", "")
         if not query:
-            return json.dumps({"error": "query is required"})
+            return tool_error("query is required")
 
         payload: Dict[str, Any] = {"query": query}
         mode = args.get("mode", "auto")
@@ -530,7 +531,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def _tool_read(self, args: dict) -> str:
         uri = args.get("uri", "")
         if not uri:
-            return json.dumps({"error": "uri is required"})
+            return tool_error("uri is required")
 
         level = args.get("level", "overview")
         # Map our level names to OpenViking GET endpoints
@@ -582,7 +583,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def _tool_remember(self, args: dict) -> str:
         content = args.get("content", "")
         if not content:
-            return json.dumps({"error": "content is required"})
+            return tool_error("content is required")
 
         # Store as a session message that will be extracted during commit.
         # The category hint helps OpenViking's extraction classify correctly.
@@ -606,7 +607,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def _tool_add_resource(self, args: dict) -> str:
         url = args.get("url", "")
         if not url:
-            return json.dumps({"error": "url is required"})
+            return tool_error("url is required")
 
         payload: Dict[str, Any] = {"path": url}
         if args.get("reason"):

plugins/memory/retaindb/__init__.py

@@ -34,6 +34,7 @@ from typing import Any, Dict, List
 from urllib.parse import quote
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -649,11 +650,11 @@ class RetainDBMemoryProvider(MemoryProvider):
 
     def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
         if not self._client:
-            return json.dumps({"error": "RetainDB not initialized"})
+            return tool_error("RetainDB not initialized")
         try:
             return json.dumps(self._dispatch(tool_name, args))
         except Exception as exc:
-            return json.dumps({"error": str(exc)})
+            return tool_error(str(exc))
 
     def _dispatch(self, tool_name: str, args: dict) -> Any:
         c = self._client

plugins/memory/supermemory/__init__.py

@@ -18,6 +18,7 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
+from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -587,7 +588,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
     def _tool_store(self, args: dict) -> str:
         content = str(args.get("content") or "").strip()
         if not content:
-            return json.dumps({"error": "content is required"})
+            return tool_error("content is required")
         metadata = args.get("metadata") or {}
         if not isinstance(metadata, dict):
             metadata = {}
@@ -598,12 +599,12 @@ class SupermemoryMemoryProvider(MemoryProvider):
             preview = content[:80] + ("..." if len(content) > 80 else "")
             return json.dumps({"saved": True, "id": result.get("id", ""), "preview": preview})
         except Exception as exc:
-            return json.dumps({"error": f"Failed to store memory: {exc}"})
+            return tool_error(f"Failed to store memory: {exc}")
 
     def _tool_search(self, args: dict) -> str:
         query = str(args.get("query") or "").strip()
         if not query:
-            return json.dumps({"error": "query is required"})
+            return tool_error("query is required")
         try:
             limit = max(1, min(20, int(args.get("limit", 5) or 5)))
         except Exception:
@@ -621,20 +622,20 @@ class SupermemoryMemoryProvider(MemoryProvider):
                 formatted.append(entry)
             return json.dumps({"results": formatted, "count": len(formatted)})
         except Exception as exc:
-            return json.dumps({"error": f"Search failed: {exc}"})
+            return tool_error(f"Search failed: {exc}")
 
     def _tool_forget(self, args: dict) -> str:
         memory_id = str(args.get("id") or "").strip()
         query = str(args.get("query") or "").strip()
         if not memory_id and not query:
-            return json.dumps({"error": "Provide either id or query"})
+            return tool_error("Provide either id or query")
         try:
             if memory_id:
                 self._client.forget_memory(memory_id)
                 return json.dumps({"forgotten": True, "id": memory_id})
             return json.dumps(self._client.forget_by_query(query))
         except Exception as exc:
-            return json.dumps({"error": f"Forget failed: {exc}"})
+            return tool_error(f"Forget failed: {exc}")
 
     def _tool_profile(self, args: dict) -> str:
         query = str(args.get("query") or "").strip() or None
@@ -651,11 +652,11 @@ class SupermemoryMemoryProvider(MemoryProvider):
                 "dynamic_count": len(profile["dynamic"]),
             })
         except Exception as exc:
-            return json.dumps({"error": f"Profile failed: {exc}"})
+            return tool_error(f"Profile failed: {exc}")
 
     def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
         if not self._active or not self._client:
-            return json.dumps({"error": "Supermemory is not configured"})
+            return tool_error("Supermemory is not configured")
         if tool_name == "supermemory_store":
             return self._tool_store(args)
         if tool_name == "supermemory_search":
@@ -664,7 +665,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
             return self._tool_forget(args)
         if tool_name == "supermemory_profile":
             return self._tool_profile(args)
-        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+        return tool_error(f"Unknown tool: {tool_name}")
 
 
 def register(ctx):

run_agent.py

@@ -526,6 +526,7 @@ class AIAgent:
         reasoning_config: Dict[str, Any] = None,
         prefill_messages: List[Dict[str, Any]] = None,
         platform: str = None,
+        user_id: str = None,
         skip_context_files: bool = False,
         skip_memory: bool = False,
         session_db=None,
@@ -590,6 +591,7 @@ class AIAgent:
         self.quiet_mode = quiet_mode
         self.ephemeral_system_prompt = ephemeral_system_prompt
         self.platform = platform  # "cli", "telegram", "discord", "whatsapp", etc.
+        self._user_id = user_id  # Platform user identifier (gateway sessions)
         # Pluggable print function — CLI replaces this with _cprint so that
         # raw ANSI status lines are routed through prompt_toolkit's renderer
         # instead of going directly to stdout where patch_stdout's StdoutProxy
@@ -1092,6 +1094,9 @@ class AIAgent:
                             "hermes_home": str(_ghh()),
                             "agent_context": "primary",
                         }
+                        # Thread gateway user identity for per-user memory scoping
+                        if self._user_id:
+                            _init_kwargs["user_id"] = self._user_id
                         # Profile identity for per-profile provider scoping
                         try:
                             from hermes_cli.profiles import get_active_profile_name
@@ -1500,10 +1505,6 @@ class AIAgent:
         """Return True when the base URL targets OpenRouter."""
         return "openrouter" in self._base_url_lower
 
-    def _is_anthropic_url(self) -> bool:
-        """Return True when the base URL targets Anthropic (native or /anthropic proxy path)."""
-        return "api.anthropic.com" in self._base_url_lower or self._base_url_lower.rstrip("/").endswith("/anthropic")
-
     def _max_tokens_param(self, value: int) -> dict:
         """Return the correct max tokens kwarg for the current provider.
         
@@ -1689,74 +1690,6 @@ class AIAgent:
         
         return None
 
-    def _classify_empty_content_response(
-        self,
-        assistant_message,
-        *,
-        finish_reason: Optional[str],
-        approx_tokens: int,
-        api_messages: List[Dict[str, Any]],
-        conversation_history: Optional[List[Dict[str, Any]]],
-    ) -> Dict[str, Any]:
-        """Classify think-only/empty responses so we can retry, compress, or salvage.
-
-        We intentionally do NOT short-circuit all structured-reasoning responses.
-        Prior discussion/PR history shows some models recover on retry. Instead we:
-        - compress immediately when the pattern looks like implicit context pressure
-        - salvage reasoning early when the same reasoning-only payload repeats
-        - otherwise preserve the normal retry path
-        """
-        reasoning_text = self._extract_reasoning(assistant_message)
-        has_structured_reasoning = bool(
-            getattr(assistant_message, "reasoning", None)
-            or getattr(assistant_message, "reasoning_content", None)
-            or getattr(assistant_message, "reasoning_details", None)
-        )
-        content = getattr(assistant_message, "content", None) or ""
-        stripped_content = self._strip_think_blocks(content).strip()
-        signature = (
-            content,
-            reasoning_text or "",
-            bool(has_structured_reasoning),
-            finish_reason or "",
-        )
-        repeated_signature = signature == getattr(self, "_last_empty_content_signature", None)
-
-        compressor = getattr(self, "context_compressor", None)
-        ctx_len = getattr(compressor, "context_length", 0) or 0
-        threshold_tokens = getattr(compressor, "threshold_tokens", 0) or 0
-        is_large_session = bool(
-            (ctx_len and approx_tokens >= max(int(ctx_len * 0.4), threshold_tokens))
-            or len(api_messages) > 80
-        )
-        is_local_custom = is_local_endpoint(getattr(self, "base_url", "") or "")
-        is_resumed = bool(conversation_history)
-        context_pressure_signals = any(
-            [
-                finish_reason == "length",
-                getattr(compressor, "_context_probed", False),
-                is_large_session,
-                is_resumed,
-            ]
-        )
-        should_compress = bool(
-            self.compression_enabled
-            and is_local_custom
-            and context_pressure_signals
-            and not stripped_content
-        )
-
-        self._last_empty_content_signature = signature
-        return {
-            "reasoning_text": reasoning_text,
-            "has_structured_reasoning": has_structured_reasoning,
-            "repeated_signature": repeated_signature,
-            "should_compress": should_compress,
-            "is_local_custom": is_local_custom,
-            "is_large_session": is_large_session,
-            "is_resumed": is_resumed,
-        }
-    
     def _cleanup_task_resources(self, task_id: str) -> None:
         """Clean up VM and browser resources for a given task."""
         try:
@@ -5808,6 +5741,7 @@ class AIAgent:
                 api_msg.pop("reasoning", None)
                 api_msg.pop("finish_reason", None)
                 api_msg.pop("_flush_sentinel", None)
+                api_msg.pop("_thinking_prefill", None)
                 if _needs_sanitize:
                     self._sanitize_tool_calls_for_strict_api(api_msg)
                 api_messages.append(api_msg)
@@ -6731,7 +6665,7 @@ class AIAgent:
             api_messages = []
             for msg in messages:
                 api_msg = msg.copy()
-                for internal_field in ("reasoning", "finish_reason"):
+                for internal_field in ("reasoning", "finish_reason", "_thinking_prefill"):
                     api_msg.pop(internal_field, None)
                 if _needs_sanitize:
                     self._sanitize_tool_calls_for_strict_api(api_msg)
@@ -6923,6 +6857,7 @@ class AIAgent:
         self._empty_content_retries = 0
         self._incomplete_scratchpad_retries = 0
         self._codex_incomplete_retries = 0
+        self._thinking_prefill_retries = 0
         self._last_content_with_tools = None
         self._mute_post_response = False
         self._surrogate_sanitized = False
@@ -7268,6 +7203,8 @@ class AIAgent:
                 # Remove finish_reason - not accepted by strict APIs (e.g. Mistral)
                 if "finish_reason" in api_msg:
                     api_msg.pop("finish_reason")
+                # Strip internal thinking-prefill marker
+                api_msg.pop("_thinking_prefill", None)
                 # Strip Codex Responses API fields (call_id, response_item_id) for
                 # strict providers like Mistral, Fireworks, etc. that reject unknown fields.
                 # Uses new dicts so the internal messages list retains the fields
@@ -8802,6 +8739,15 @@ class AIAgent:
                             if clean:
                                 self._vprint(f"  ┊ 💬 {clean}")
                     
+                    # Pop thinking-only prefill message(s) before appending
+                    # (tool-call path — same rationale as the final-response path).
+                    while (
+                        messages
+                        and isinstance(messages[-1], dict)
+                        and messages[-1].get("_thinking_prefill")
+                    ):
+                        messages.pop()
+
                     messages.append(assistant_msg)
 
                     # Close any open streaming display (response box, reasoning
@@ -8915,11 +8861,36 @@ class AIAgent:
                             self._response_was_previewed = True
                             break
 
-                        # Reasoning-only response: the model produced thinking
-                        # but no visible content.  This is a valid response —
-                        # keep reasoning in its own field and set content to
-                        # "(empty)" so every provider accepts the message.
-                        # No retries needed.
+                        # ── Thinking-only prefill continuation ──────────
+                        # The model produced structured reasoning (via API
+                        # fields) but no visible text content.  Rather than
+                        # giving up, append the assistant message as-is and
+                        # continue — the model will see its own reasoning
+                        # on the next turn and produce the text portion.
+                        # Inspired by clawdbot's "incomplete-text" recovery.
+                        _has_structured = bool(
+                            getattr(assistant_message, "reasoning", None)
+                            or getattr(assistant_message, "reasoning_content", None)
+                            or getattr(assistant_message, "reasoning_details", None)
+                        )
+                        if _has_structured and self._thinking_prefill_retries < 2:
+                            self._thinking_prefill_retries += 1
+                            self._vprint(
+                                f"{self.log_prefix}↻ Thinking-only response — "
+                                f"prefilling to continue "
+                                f"({self._thinking_prefill_retries}/2)"
+                            )
+                            interim_msg = self._build_assistant_message(
+                                assistant_message, "incomplete"
+                            )
+                            interim_msg["_thinking_prefill"] = True
+                            messages.append(interim_msg)
+                            self._session_messages = messages
+                            self._save_session_log(messages)
+                            continue
+
+                        # Exhausted prefill attempts or no structured
+                        # reasoning — fall through to "(empty)" terminal.
                         reasoning_text = self._extract_reasoning(assistant_message)
                         assistant_msg = self._build_assistant_message(assistant_message, finish_reason)
                         assistant_msg["content"] = "(empty)"
@@ -8938,6 +8909,7 @@ class AIAgent:
                     if hasattr(self, '_empty_content_retries'):
                         self._empty_content_retries = 0
                     self._last_empty_content_signature = None
+                    self._thinking_prefill_retries = 0
 
                     if (
                         self.api_mode == "codex_responses"
@@ -8976,7 +8948,18 @@ class AIAgent:
                     final_response = self._strip_think_blocks(final_response).strip()
                     
                     final_msg = self._build_assistant_message(assistant_message, finish_reason)
-                    
+
+                    # Pop thinking-only prefill message(s) before appending
+                    # the final response.  This avoids consecutive assistant
+                    # messages which break strict-alternation providers
+                    # (Anthropic Messages API) and keeps history clean.
+                    while (
+                        messages
+                        and isinstance(messages[-1], dict)
+                        and messages[-1].get("_thinking_prefill")
+                    ):
+                        messages.pop()
+
                     messages.append(final_msg)
                     
                     if not self.quiet_mode:

tests/agent/test_auxiliary_config_bridge.py

@@ -13,7 +13,7 @@ from unittest.mock import patch, MagicMock
 import pytest
 import yaml
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
 
 
 def _run_auxiliary_bridge(config_dict, monkeypatch):
@@ -199,7 +199,7 @@ class TestGatewayBridgeCodeParity:
 
     def test_gateway_has_auxiliary_bridge(self):
         """The gateway config bridge must include auxiliary.* bridging."""
-        gateway_path = Path(__file__).parent.parent / "gateway" / "run.py"
+        gateway_path = Path(__file__).parent.parent.parent / "gateway" / "run.py"
         content = gateway_path.read_text()
         # Check for key patterns that indicate the bridge is present
         assert "AUXILIARY_VISION_PROVIDER" in content
@@ -213,7 +213,7 @@ class TestGatewayBridgeCodeParity:
 
     def test_gateway_no_compression_env_bridge(self):
         """Gateway should NOT bridge compression config to env vars (config-only)."""
-        gateway_path = Path(__file__).parent.parent / "gateway" / "run.py"
+        gateway_path = Path(__file__).parent.parent.parent / "gateway" / "run.py"
         content = gateway_path.read_text()
         assert "CONTEXT_COMPRESSION_PROVIDER" not in content
         assert "CONTEXT_COMPRESSION_MODEL" not in content

tests/agent/test_memory_user_id.py

@@ -0,0 +1,289 @@
+"""Tests for per-user memory scoping via user_id threading.
+
+Verifies that gateway user_id flows from AIAgent -> MemoryManager -> plugins,
+so each gateway user gets their own memory bucket instead of sharing a static one.
+"""
+
+import json
+import os
+import pytest
+from unittest.mock import MagicMock, patch
+
+from agent.memory_provider import MemoryProvider
+from agent.memory_manager import MemoryManager
+
+
+# ---------------------------------------------------------------------------
+# Concrete test provider that records init kwargs
+# ---------------------------------------------------------------------------
+
+
+class RecordingProvider(MemoryProvider):
+    """Minimal provider that records what initialize() receives."""
+
+    def __init__(self, name="recording"):
+        self._name = name
+        self._init_kwargs = {}
+        self._init_session_id = None
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def is_available(self) -> bool:
+        return True
+
+    def initialize(self, session_id: str, **kwargs) -> None:
+        self._init_session_id = session_id
+        self._init_kwargs = dict(kwargs)
+
+    def system_prompt_block(self) -> str:
+        return ""
+
+    def prefetch(self, query: str, *, session_id: str = "") -> str:
+        return ""
+
+    def sync_turn(self, user_content, assistant_content, *, session_id=""):
+        pass
+
+    def get_tool_schemas(self):
+        return []
+
+    def handle_tool_call(self, tool_name, args, **kwargs):
+        return json.dumps({})
+
+    def shutdown(self):
+        pass
+
+
+# ---------------------------------------------------------------------------
+# MemoryManager user_id threading tests
+# ---------------------------------------------------------------------------
+
+
+class TestMemoryManagerUserIdThreading:
+    """Verify user_id reaches providers via initialize_all."""
+
+    def test_user_id_forwarded_to_provider(self):
+        mgr = MemoryManager()
+        p = RecordingProvider()
+        mgr.add_provider(p)
+
+        mgr.initialize_all(
+            session_id="sess-123",
+            platform="telegram",
+            user_id="tg_user_42",
+        )
+
+        assert p._init_kwargs.get("user_id") == "tg_user_42"
+        assert p._init_kwargs.get("platform") == "telegram"
+        assert p._init_session_id == "sess-123"
+
+    def test_no_user_id_when_cli(self):
+        """CLI sessions should not have user_id in kwargs."""
+        mgr = MemoryManager()
+        p = RecordingProvider()
+        mgr.add_provider(p)
+
+        mgr.initialize_all(
+            session_id="sess-456",
+            platform="cli",
+        )
+
+        assert "user_id" not in p._init_kwargs
+        assert p._init_kwargs.get("platform") == "cli"
+
+    def test_user_id_none_not_forwarded(self):
+        """Explicit None user_id should not appear in kwargs."""
+        mgr = MemoryManager()
+        p = RecordingProvider()
+        mgr.add_provider(p)
+
+        # Simulates what happens when AIAgent passes user_id=None
+        # (the agent code only adds user_id to kwargs when it's truthy)
+        mgr.initialize_all(
+            session_id="sess-789",
+            platform="discord",
+        )
+
+        assert "user_id" not in p._init_kwargs
+
+    def test_multiple_providers_all_receive_user_id(self):
+        from agent.builtin_memory_provider import BuiltinMemoryProvider
+
+        mgr = MemoryManager()
+        # Use builtin + one external (MemoryManager only allows one external)
+        builtin = BuiltinMemoryProvider()
+        ext = RecordingProvider("external")
+        mgr.add_provider(builtin)
+        mgr.add_provider(ext)
+
+        mgr.initialize_all(
+            session_id="sess-multi",
+            platform="slack",
+            user_id="slack_U12345",
+        )
+
+        assert ext._init_kwargs.get("user_id") == "slack_U12345"
+        assert ext._init_kwargs.get("platform") == "slack"
+
+
+# ---------------------------------------------------------------------------
+# Mem0 provider user_id tests
+# ---------------------------------------------------------------------------
+
+
+class TestMem0UserIdScoping:
+    """Verify Mem0 plugin uses gateway user_id when provided."""
+
+    def test_gateway_user_id_overrides_default(self):
+        """When user_id is passed via kwargs, it should override the config default."""
+        from plugins.memory.mem0 import Mem0MemoryProvider
+
+        provider = Mem0MemoryProvider()
+        # Mock _load_config to return a config with default user_id
+        with patch("plugins.memory.mem0._load_config", return_value={
+            "api_key": "test-key",
+            "user_id": "hermes-user",
+            "agent_id": "hermes",
+            "rerank": True,
+        }):
+            provider.initialize(session_id="test-sess", user_id="tg_user_99")
+
+        assert provider._user_id == "tg_user_99"
+
+    def test_no_user_id_falls_back_to_config(self):
+        """Without user_id in kwargs, should use config default."""
+        from plugins.memory.mem0 import Mem0MemoryProvider
+
+        provider = Mem0MemoryProvider()
+        with patch("plugins.memory.mem0._load_config", return_value={
+            "api_key": "test-key",
+            "user_id": "custom-default",
+            "agent_id": "hermes",
+            "rerank": True,
+        }):
+            provider.initialize(session_id="test-sess")
+
+        assert provider._user_id == "custom-default"
+
+    def test_no_user_id_no_config_uses_hermes_user(self):
+        """Without user_id or config override, should default to 'hermes-user'."""
+        from plugins.memory.mem0 import Mem0MemoryProvider
+
+        provider = Mem0MemoryProvider()
+        with patch("plugins.memory.mem0._load_config", return_value={
+            "api_key": "test-key",
+            "agent_id": "hermes",
+            "rerank": True,
+        }):
+            provider.initialize(session_id="test-sess")
+
+        assert provider._user_id == "hermes-user"
+
+    def test_different_users_get_different_ids(self):
+        """Two providers initialized with different user_ids should be scoped differently."""
+        from plugins.memory.mem0 import Mem0MemoryProvider
+
+        p1 = Mem0MemoryProvider()
+        p2 = Mem0MemoryProvider()
+
+        with patch("plugins.memory.mem0._load_config", return_value={
+            "api_key": "test-key",
+            "user_id": "hermes-user",
+            "agent_id": "hermes",
+            "rerank": True,
+        }):
+            p1.initialize(session_id="sess-1", user_id="alice_123")
+            p2.initialize(session_id="sess-2", user_id="bob_456")
+
+        assert p1._user_id == "alice_123"
+        assert p2._user_id == "bob_456"
+        assert p1._user_id != p2._user_id
+
+
+# ---------------------------------------------------------------------------
+# Honcho provider user_id tests
+# ---------------------------------------------------------------------------
+
+
+class TestHonchoUserIdScoping:
+    """Verify Honcho plugin uses gateway user_id for peer_name when provided."""
+
+    def test_gateway_user_id_overrides_peer_name(self):
+        """When user_id is in kwargs, cfg.peer_name should be overridden."""
+        from plugins.memory.honcho import HonchoMemoryProvider
+
+        provider = HonchoMemoryProvider()
+
+        # Create a mock config with a static peer_name
+        mock_cfg = MagicMock()
+        mock_cfg.enabled = True
+        mock_cfg.api_key = "test-key"
+        mock_cfg.base_url = None
+        mock_cfg.peer_name = "static-user"
+        mock_cfg.recall_mode = "tools"  # Use tools mode to defer session init
+
+        with patch(
+            "plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
+            return_value=mock_cfg,
+        ):
+            provider.initialize(
+                session_id="test-sess",
+                user_id="discord_user_789",
+                platform="discord",
+            )
+
+        # The config's peer_name should have been overridden with the user_id
+        assert mock_cfg.peer_name == "discord_user_789"
+
+    def test_no_user_id_preserves_config_peer_name(self):
+        """Without user_id, the config peer_name should be preserved."""
+        from plugins.memory.honcho import HonchoMemoryProvider
+
+        provider = HonchoMemoryProvider()
+
+        mock_cfg = MagicMock()
+        mock_cfg.enabled = True
+        mock_cfg.api_key = "test-key"
+        mock_cfg.base_url = None
+        mock_cfg.peer_name = "my-custom-peer"
+        mock_cfg.recall_mode = "tools"
+
+        with patch(
+            "plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
+            return_value=mock_cfg,
+        ):
+            provider.initialize(
+                session_id="test-sess",
+                platform="cli",
+            )
+
+        # peer_name should not have been overridden
+        assert mock_cfg.peer_name == "my-custom-peer"
+
+
+# ---------------------------------------------------------------------------
+# AIAgent user_id propagation test
+# ---------------------------------------------------------------------------
+
+
+class TestAIAgentUserIdPropagation:
+    """Verify AIAgent stores user_id and passes it to memory init kwargs."""
+
+    def test_user_id_stored_on_agent(self):
+        """AIAgent should store user_id as instance attribute."""
+        with patch.dict(os.environ, {"HERMES_HOME": "/tmp/test_hermes"}):
+            from run_agent import AIAgent
+            agent = object.__new__(AIAgent)
+            # Manually set the attribute as __init__ does
+            agent._user_id = "test_user_42"
+            assert agent._user_id == "test_user_42"
+
+    def test_user_id_none_by_default(self):
+        """AIAgent should have None user_id when not provided (CLI mode)."""
+        with patch.dict(os.environ, {"HERMES_HOME": "/tmp/test_hermes"}):
+            from run_agent import AIAgent
+            agent = object.__new__(AIAgent)
+            agent._user_id = None
+            assert agent._user_id is None

tests/cli/test_cli_provider_resolution.py

@@ -330,7 +330,7 @@ def test_model_flow_nous_prints_subscription_guidance_without_mutating_explicit_
         "hermes_cli.auth.fetch_nous_models",
         lambda *args, **kwargs: ["claude-opus-4-6"],
     )
-    monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="", pricing=None: "claude-opus-4-6")
+    monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="", pricing=None, **kw: "claude-opus-4-6")
     monkeypatch.setattr("hermes_cli.auth._save_model_choice", lambda model: None)
     monkeypatch.setattr("hermes_cli.auth._update_config_for_provider", lambda provider, url: None)
     monkeypatch.setattr(
@@ -368,7 +368,7 @@ def test_model_flow_nous_applies_managed_tts_default_when_unconfigured(monkeypat
         "hermes_cli.auth.fetch_nous_models",
         lambda *args, **kwargs: ["claude-opus-4-6"],
     )
-    monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="", pricing=None: "claude-opus-4-6")
+    monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="", pricing=None, **kw: "claude-opus-4-6")
     monkeypatch.setattr("hermes_cli.auth._save_model_choice", lambda model: None)
     monkeypatch.setattr("hermes_cli.auth._update_config_for_provider", lambda provider, url: None)
     monkeypatch.setattr(

tests/cli/test_cli_retry.py

@@ -1,6 +1,6 @@
 """Regression tests for CLI /retry history replacement semantics."""
 
-from tests.test_cli_init import _make_cli
+from tests.cli.test_cli_init import _make_cli
 
 
 def test_retry_last_truncates_history_before_requeueing_message():

tests/cron/test_scheduler.py

@@ -7,7 +7,7 @@ from unittest.mock import AsyncMock, patch, MagicMock
 
 import pytest
 
-from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, run_job, SILENT_MARKER, _build_job_prompt
+from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, _send_media_via_adapter, run_job, SILENT_MARKER, _build_job_prompt
 
 
 class TestResolveOrigin:
@@ -277,6 +277,188 @@ class TestDeliverResultWrapping:
         # Media files should be forwarded separately
         assert kwargs["media_files"] == [("/tmp/test-voice.ogg", False)]
 
+    def test_live_adapter_sends_media_as_attachments(self):
+        """When a live adapter is available, MEDIA files should be sent as native
+        platform attachments (e.g., Discord voice, Telegram audio) rather than
+        as literal 'MEDIA:/path' text."""
+        from gateway.config import Platform
+        from concurrent.futures import Future
+
+        adapter = AsyncMock()
+        adapter.send.return_value = MagicMock(success=True)
+        adapter.send_voice.return_value = MagicMock(success=True)
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.DISCORD: pconfig}
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+
+        # run_coroutine_threadsafe returns concurrent.futures.Future (has timeout kwarg)
+        def fake_run_coro(coro, _loop):
+            future = Future()
+            future.set_result(MagicMock(success=True))
+            coro.close()
+            return future
+
+        job = {
+            "id": "tts-job",
+            "deliver": "origin",
+            "origin": {"platform": "discord", "chat_id": "9876"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "Here is TTS\nMEDIA:/tmp/cron-voice.mp3",
+                adapters={Platform.DISCORD: adapter},
+                loop=loop,
+            )
+
+        # Text should be sent without the MEDIA tag
+        adapter.send.assert_called_once()
+        text_sent = adapter.send.call_args[0][1]
+        assert "MEDIA:" not in text_sent
+        assert "Here is TTS" in text_sent
+
+        # Audio file should be sent as a voice attachment
+        adapter.send_voice.assert_called_once()
+        voice_call = adapter.send_voice.call_args
+        assert voice_call[1]["audio_path"] == "/tmp/cron-voice.mp3"
+
+    def test_live_adapter_routes_image_to_send_image_file(self):
+        """Image MEDIA files should be routed to send_image_file, not send_voice."""
+        from gateway.config import Platform
+        from concurrent.futures import Future
+
+        adapter = AsyncMock()
+        adapter.send.return_value = MagicMock(success=True)
+        adapter.send_image_file.return_value = MagicMock(success=True)
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.DISCORD: pconfig}
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+
+        def fake_run_coro(coro, _loop):
+            future = Future()
+            future.set_result(MagicMock(success=True))
+            coro.close()
+            return future
+
+        job = {
+            "id": "img-job",
+            "deliver": "origin",
+            "origin": {"platform": "discord", "chat_id": "1234"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "Chart attached\nMEDIA:/tmp/chart.png",
+                adapters={Platform.DISCORD: adapter},
+                loop=loop,
+            )
+
+        adapter.send_image_file.assert_called_once()
+        assert adapter.send_image_file.call_args[1]["image_path"] == "/tmp/chart.png"
+        adapter.send_voice.assert_not_called()
+
+    def test_live_adapter_media_only_no_text(self):
+        """When content is ONLY a MEDIA tag with no text, media should still be sent."""
+        from gateway.config import Platform
+        from concurrent.futures import Future
+
+        adapter = AsyncMock()
+        adapter.send_voice.return_value = MagicMock(success=True)
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+
+        def fake_run_coro(coro, _loop):
+            future = Future()
+            future.set_result(MagicMock(success=True))
+            coro.close()
+            return future
+
+        job = {
+            "id": "voice-only",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "999"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "MEDIA:/tmp/voice.ogg",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        # Text send should NOT be called (no text after stripping MEDIA tag)
+        adapter.send.assert_not_called()
+        # Audio should still be delivered
+        adapter.send_voice.assert_called_once()
+
+    def test_live_adapter_sends_cleaned_text_not_raw(self):
+        """The live adapter path must send cleaned text (MEDIA tags stripped),
+        not the raw delivery_content with embedded MEDIA: tags."""
+        from gateway.config import Platform
+        from concurrent.futures import Future
+
+        adapter = AsyncMock()
+        adapter.send.return_value = MagicMock(success=True)
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+
+        def fake_run_coro(coro, _loop):
+            future = Future()
+            future.set_result(MagicMock(success=True))
+            coro.close()
+            return future
+
+        job = {
+            "id": "img-job",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "555"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "Report\nMEDIA:/tmp/chart.png",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        text_sent = adapter.send.call_args[0][1]
+        assert "MEDIA:" not in text_sent
+        assert "Report" in text_sent
+
     def test_no_mirror_to_session_call(self):
         """Cron deliveries should NOT mirror into the gateway session."""
         from gateway.config import Platform
@@ -862,3 +1044,57 @@ class TestTickAdvanceBeforeRun:
         adv_mock.assert_called_once_with("test-advance")
         # advance must happen before run
         assert call_order == [("advance", "test-advance"), ("run", "test-advance")]
+
+
+class TestSendMediaViaAdapter:
+    """Unit tests for _send_media_via_adapter — routes files to typed adapter methods."""
+
+    @staticmethod
+    def _run_with_loop(adapter, chat_id, media_files, metadata, job):
+        """Helper: run _send_media_via_adapter with a real running event loop."""
+        import asyncio
+        import threading
+
+        loop = asyncio.new_event_loop()
+        t = threading.Thread(target=loop.run_forever, daemon=True)
+        t.start()
+        try:
+            _send_media_via_adapter(adapter, chat_id, media_files, metadata, loop, job)
+        finally:
+            loop.call_soon_threadsafe(loop.stop)
+            t.join(timeout=5)
+            loop.close()
+
+    def test_video_dispatched_to_send_video(self):
+        adapter = MagicMock()
+        adapter.send_video = AsyncMock()
+        media_files = [("/tmp/clip.mp4", False)]
+        self._run_with_loop(adapter, "123", media_files, None, {"id": "j1"})
+        adapter.send_video.assert_called_once()
+        assert adapter.send_video.call_args[1]["video_path"] == "/tmp/clip.mp4"
+
+    def test_unknown_ext_dispatched_to_send_document(self):
+        adapter = MagicMock()
+        adapter.send_document = AsyncMock()
+        media_files = [("/tmp/report.pdf", False)]
+        self._run_with_loop(adapter, "123", media_files, None, {"id": "j2"})
+        adapter.send_document.assert_called_once()
+        assert adapter.send_document.call_args[1]["file_path"] == "/tmp/report.pdf"
+
+    def test_multiple_media_files_all_delivered(self):
+        adapter = MagicMock()
+        adapter.send_voice = AsyncMock()
+        adapter.send_image_file = AsyncMock()
+        media_files = [("/tmp/voice.mp3", False), ("/tmp/photo.jpg", False)]
+        self._run_with_loop(adapter, "123", media_files, None, {"id": "j3"})
+        adapter.send_voice.assert_called_once()
+        adapter.send_image_file.assert_called_once()
+
+    def test_single_failure_does_not_block_others(self):
+        adapter = MagicMock()
+        adapter.send_voice = AsyncMock(side_effect=RuntimeError("network error"))
+        adapter.send_image_file = AsyncMock()
+        media_files = [("/tmp/voice.ogg", False), ("/tmp/photo.png", False)]
+        self._run_with_loop(adapter, "123", media_files, None, {"id": "j4"})
+        adapter.send_voice.assert_called_once()
+        adapter.send_image_file.assert_called_once()

tests/gateway/test_slack_approval_buttons.py

@@ -371,3 +371,56 @@ class TestSessionKeyFix:
             channel_id="C1", thread_ts="1000.0", user_id="U123"
         )
         assert result is False
+
+
+# ===========================================================================
+# Thread engagement — bot-started threads & mentioned threads
+# ===========================================================================
+
+class TestThreadEngagement:
+    """Test _bot_message_ts and _mentioned_threads tracking."""
+
+    @pytest.mark.asyncio
+    async def test_send_tracks_bot_message_ts(self):
+        """Bot's sent messages are tracked so thread replies work without @mention."""
+        adapter = _make_adapter()
+        mock_client = adapter._team_clients["T1"]
+        mock_client.chat_postMessage = AsyncMock(return_value={"ts": "9000.1"})
+
+        await adapter.send(chat_id="C1", content="Hello!", metadata={"thread_id": "8000.0"})
+
+        assert "9000.1" in adapter._bot_message_ts
+        # Thread root should also be tracked
+        assert "8000.0" in adapter._bot_message_ts
+
+    @pytest.mark.asyncio
+    async def test_bot_message_ts_cap(self):
+        """Verify memory is bounded when many messages are sent."""
+        adapter = _make_adapter()
+        adapter._BOT_TS_MAX = 10  # low cap for testing
+        mock_client = adapter._team_clients["T1"]
+
+        for i in range(20):
+            mock_client.chat_postMessage = AsyncMock(return_value={"ts": f"{i}.0"})
+            await adapter.send(chat_id="C1", content=f"msg {i}")
+
+        assert len(adapter._bot_message_ts) <= 10
+
+    def test_mentioned_threads_populated_on_mention(self):
+        """When bot is @mentioned in a thread, that thread is tracked."""
+        adapter = _make_adapter()
+        # Simulate what _handle_slack_message does on mention
+        adapter._mentioned_threads.add("1000.0")
+        assert "1000.0" in adapter._mentioned_threads
+
+    def test_mentioned_threads_cap(self):
+        """Verify _mentioned_threads is bounded."""
+        adapter = _make_adapter()
+        adapter._MENTIONED_THREADS_MAX = 10
+        for i in range(15):
+            adapter._mentioned_threads.add(f"{i}.0")
+            if len(adapter._mentioned_threads) > adapter._MENTIONED_THREADS_MAX:
+                to_remove = list(adapter._mentioned_threads)[:adapter._MENTIONED_THREADS_MAX // 2]
+                for t in to_remove:
+                    adapter._mentioned_threads.discard(t)
+        assert len(adapter._mentioned_threads) <= 10

tests/gateway/test_telegram_approval_buttons.py

@@ -33,8 +33,15 @@ def _ensure_telegram_mock():
     mod.constants.ChatType.GROUP = "group"
     mod.constants.ChatType.SUPERGROUP = "supergroup"
     mod.constants.ChatType.CHANNEL = "channel"
-    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request", "telegram.error"):
+    # Provide real exception classes so ``except (NetworkError, ...)`` in
+    # connect() doesn't blow up under xdist when this mock leaks.
+    mod.error.NetworkError = type("NetworkError", (OSError,), {})
+    mod.error.TimedOut = type("TimedOut", (OSError,), {})
+    mod.error.BadRequest = type("BadRequest", (Exception,), {})
+
+    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
         sys.modules.setdefault(name, mod)
+    sys.modules.setdefault("telegram.error", mod.error)
 
 
 _ensure_telegram_mock()

tests/gateway/test_telegram_conflict.py

@@ -20,8 +20,16 @@ def _ensure_telegram_mock():
     telegram_mod.constants.ChatType.CHANNEL = "channel"
     telegram_mod.constants.ChatType.PRIVATE = "private"
 
+    # Provide real exception classes so ``except (NetworkError, ...)`` in
+    # connect() doesn't blow up with "catching classes that do not inherit
+    # from BaseException" when another xdist worker pollutes sys.modules.
+    telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {})
+    telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {})
+    telegram_mod.error.BadRequest = type("BadRequest", (Exception,), {})
+
     for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
         sys.modules.setdefault(name, telegram_mod)
+    sys.modules.setdefault("telegram.error", telegram_mod.error)
 
 
 _ensure_telegram_mock()

tests/hermes_cli/test_update_check.py

@@ -15,7 +15,7 @@ def test_version_string_no_v_prefix():
     assert not __version__.startswith("v"), f"__version__ should not start with 'v', got {__version__!r}"
 
 
-def test_check_for_updates_uses_cache(tmp_path):
+def test_check_for_updates_uses_cache(tmp_path, monkeypatch):
     """When cache is fresh, check_for_updates should return cached value without calling git."""
     from hermes_cli.banner import check_for_updates
 
@@ -27,15 +27,15 @@ def test_check_for_updates_uses_cache(tmp_path):
     cache_file = tmp_path / ".update_check"
     cache_file.write_text(json.dumps({"ts": time.time(), "behind": 3}))
 
-    with patch("hermes_cli.banner.os.getenv", return_value=str(tmp_path)):
-        with patch("hermes_cli.banner.subprocess.run") as mock_run:
-            result = check_for_updates()
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    with patch("hermes_cli.banner.subprocess.run") as mock_run:
+        result = check_for_updates()
 
     assert result == 3
     mock_run.assert_not_called()
 
 
-def test_check_for_updates_expired_cache(tmp_path):
+def test_check_for_updates_expired_cache(tmp_path, monkeypatch):
     """When cache is expired, check_for_updates should call git fetch."""
     from hermes_cli.banner import check_for_updates
 
@@ -49,15 +49,15 @@ def test_check_for_updates_expired_cache(tmp_path):
 
     mock_result = MagicMock(returncode=0, stdout="5\n")
 
-    with patch("hermes_cli.banner.os.getenv", return_value=str(tmp_path)):
-        with patch("hermes_cli.banner.subprocess.run", return_value=mock_result) as mock_run:
-            result = check_for_updates()
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    with patch("hermes_cli.banner.subprocess.run", return_value=mock_result) as mock_run:
+        result = check_for_updates()
 
     assert result == 5
     assert mock_run.call_count == 2  # git fetch + git rev-list
 
 
-def test_check_for_updates_no_git_dir(tmp_path):
+def test_check_for_updates_no_git_dir(tmp_path, monkeypatch):
     """Returns None when .git directory doesn't exist anywhere."""
     import hermes_cli.banner as banner
 
@@ -66,19 +66,15 @@ def test_check_for_updates_no_git_dir(tmp_path):
     fake_banner.parent.mkdir(parents=True, exist_ok=True)
     fake_banner.touch()
 
-    original = banner.__file__
-    try:
-        banner.__file__ = str(fake_banner)
-        with patch("hermes_cli.banner.os.getenv", return_value=str(tmp_path)):
-            with patch("hermes_cli.banner.subprocess.run") as mock_run:
-                result = banner.check_for_updates()
-        assert result is None
-        mock_run.assert_not_called()
-    finally:
-        banner.__file__ = original
+    monkeypatch.setattr(banner, "__file__", str(fake_banner))
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    with patch("hermes_cli.banner.subprocess.run") as mock_run:
+        result = banner.check_for_updates()
+    assert result is None
+    mock_run.assert_not_called()
 
 
-def test_check_for_updates_fallback_to_project_root():
+def test_check_for_updates_fallback_to_project_root(tmp_path, monkeypatch):
     """Dev install: falls back to Path(__file__).parent.parent when HERMES_HOME has no git repo."""
     import hermes_cli.banner as banner
 
@@ -87,14 +83,12 @@ def test_check_for_updates_fallback_to_project_root():
         pytest.skip("Not running from a git checkout")
 
     # Point HERMES_HOME at a temp dir with no hermes-agent/.git
-    import tempfile
-    with tempfile.TemporaryDirectory() as td:
-        with patch("hermes_cli.banner.os.getenv", return_value=td):
-            with patch("hermes_cli.banner.subprocess.run") as mock_run:
-                mock_run.return_value = MagicMock(returncode=0, stdout="0\n")
-                result = banner.check_for_updates()
-        # Should have fallen back to project root and run git commands
-        assert mock_run.call_count >= 1
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    with patch("hermes_cli.banner.subprocess.run") as mock_run:
+        mock_run.return_value = MagicMock(returncode=0, stdout="0\n")
+        result = banner.check_for_updates()
+    # Should have fallen back to project root and run git commands
+    assert mock_run.call_count >= 1
 
 
 def test_prefetch_non_blocking():