Merge branch 'main' of github.com:NousResearch/hermes-agent into feat/ink-refactor

tests/agent/test_memory_provider.py

@@ -396,6 +396,108 @@ class TestPluginMemoryDiscovery:
         assert load_memory_provider("nonexistent_provider") is None
 
 
+class TestUserInstalledProviderDiscovery:
+    """Memory providers installed to $HERMES_HOME/plugins/ should be found.
+
+    Regression test for issues #4956 and #9099: load_memory_provider() and
+    discover_memory_providers() only scanned the bundled plugins/memory/
+    directory, ignoring user-installed plugins.
+    """
+
+    def _make_user_memory_plugin(self, tmp_path, name="myprovider"):
+        """Create a minimal user memory provider plugin."""
+        plugin_dir = tmp_path / "plugins" / name
+        plugin_dir.mkdir(parents=True)
+        (plugin_dir / "__init__.py").write_text(
+            "from agent.memory_provider import MemoryProvider\n"
+            "class MyProvider(MemoryProvider):\n"
+            f"    @property\n"
+            f"    def name(self): return {name!r}\n"
+            "    def is_available(self): return True\n"
+            "    def initialize(self, **kw): pass\n"
+            "    def sync_turn(self, *a, **kw): pass\n"
+            "    def get_tool_schemas(self): return []\n"
+            "    def handle_tool_call(self, *a, **kw): return '{}'\n"
+        )
+        (plugin_dir / "plugin.yaml").write_text(
+            f"name: {name}\ndescription: Test user provider\n"
+        )
+        return plugin_dir
+
+    def test_discover_finds_user_plugins(self, tmp_path, monkeypatch):
+        """discover_memory_providers() includes user-installed plugins."""
+        from plugins.memory import discover_memory_providers, _get_user_plugins_dir
+        self._make_user_memory_plugin(tmp_path, "myexternal")
+        monkeypatch.setattr(
+            "plugins.memory._get_user_plugins_dir",
+            lambda: tmp_path / "plugins",
+        )
+        providers = discover_memory_providers()
+        names = [n for n, _, _ in providers]
+        assert "myexternal" in names
+        assert "holographic" in names  # bundled still found
+
+    def test_load_user_plugin(self, tmp_path, monkeypatch):
+        """load_memory_provider() can load from $HERMES_HOME/plugins/."""
+        from plugins.memory import load_memory_provider
+        self._make_user_memory_plugin(tmp_path, "myexternal")
+        monkeypatch.setattr(
+            "plugins.memory._get_user_plugins_dir",
+            lambda: tmp_path / "plugins",
+        )
+        p = load_memory_provider("myexternal")
+        assert p is not None
+        assert p.name == "myexternal"
+        assert p.is_available()
+
+    def test_bundled_takes_precedence(self, tmp_path, monkeypatch):
+        """Bundled provider wins when user plugin has the same name."""
+        from plugins.memory import load_memory_provider, discover_memory_providers
+        # Create user plugin named "holographic" (same as bundled)
+        plugin_dir = tmp_path / "plugins" / "holographic"
+        plugin_dir.mkdir(parents=True)
+        (plugin_dir / "__init__.py").write_text(
+            "from agent.memory_provider import MemoryProvider\n"
+            "class Fake(MemoryProvider):\n"
+            "    @property\n"
+            "    def name(self): return 'holographic-FAKE'\n"
+            "    def is_available(self): return True\n"
+            "    def initialize(self, **kw): pass\n"
+            "    def sync_turn(self, *a, **kw): pass\n"
+            "    def get_tool_schemas(self): return []\n"
+            "    def handle_tool_call(self, *a, **kw): return '{}'\n"
+        )
+        monkeypatch.setattr(
+            "plugins.memory._get_user_plugins_dir",
+            lambda: tmp_path / "plugins",
+        )
+        # Load should return bundled (name "holographic"), not user (name "holographic-FAKE")
+        p = load_memory_provider("holographic")
+        assert p is not None
+        assert p.name == "holographic"  # bundled wins
+
+        # discover should not duplicate
+        providers = discover_memory_providers()
+        holo_count = sum(1 for n, _, _ in providers if n == "holographic")
+        assert holo_count == 1
+
+    def test_non_memory_user_plugins_excluded(self, tmp_path, monkeypatch):
+        """User plugins that don't reference MemoryProvider are skipped."""
+        from plugins.memory import discover_memory_providers
+        plugin_dir = tmp_path / "plugins" / "notmemory"
+        plugin_dir.mkdir(parents=True)
+        (plugin_dir / "__init__.py").write_text(
+            "def register(ctx):\n    ctx.register_tool('foo', 'bar', {}, lambda: None)\n"
+        )
+        monkeypatch.setattr(
+            "plugins.memory._get_user_plugins_dir",
+            lambda: tmp_path / "plugins",
+        )
+        providers = discover_memory_providers()
+        names = [n for n, _, _ in providers]
+        assert "notmemory" not in names
+
+
 # ---------------------------------------------------------------------------
 # Sequential dispatch routing tests
 # ---------------------------------------------------------------------------
@@ -736,3 +838,104 @@ class TestCommitMemorySessionRouting:
         mgr.add_provider(bad)
 
         mgr.on_session_end([])  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# on_memory_write bridge — must fire from both concurrent AND sequential paths
+# ---------------------------------------------------------------------------
+
+
+class TestOnMemoryWriteBridge:
+    """Verify that MemoryManager.on_memory_write is called when built-in
+    memory writes happen.  This is a regression test for #10174 where the
+    sequential tool execution path (_execute_tool_calls_sequential) was
+    missing the bridge call, so single memory tool calls never notified
+    external memory providers.
+    """
+
+    def test_on_memory_write_add(self):
+        """on_memory_write fires for 'add' actions."""
+        mgr = MemoryManager()
+        p = FakeMemoryProvider("ext")
+        mgr.add_provider(p)
+
+        mgr.on_memory_write("add", "memory", "new fact")
+        assert p.memory_writes == [("add", "memory", "new fact")]
+
+    def test_on_memory_write_replace(self):
+        """on_memory_write fires for 'replace' actions."""
+        mgr = MemoryManager()
+        p = FakeMemoryProvider("ext")
+        mgr.add_provider(p)
+
+        mgr.on_memory_write("replace", "user", "updated pref")
+        assert p.memory_writes == [("replace", "user", "updated pref")]
+
+    def test_on_memory_write_remove_not_bridged(self):
+        """The bridge intentionally skips 'remove' — only add/replace notify."""
+        # This tests the contract that run_agent.py checks:
+        #   function_args.get("action") in ("add", "replace")
+        mgr = MemoryManager()
+        p = FakeMemoryProvider("ext")
+        mgr.add_provider(p)
+
+        # Manager itself doesn't filter — run_agent.py does.
+        # But providers should handle remove gracefully.
+        mgr.on_memory_write("remove", "memory", "old fact")
+        assert p.memory_writes == [("remove", "memory", "old fact")]
+
+    def test_memory_manager_tool_injection_deduplicates(self):
+        """Memory manager tools already in self.tools (from plugin registry)
+        must not be appended again.  Duplicate function names cause 400 errors
+        on providers that enforce unique names (e.g. Xiaomi MiMo via Nous Portal).
+
+        Regression test for: duplicate mnemosyne_recall / mnemosyne_remember /
+        mnemosyne_stats in tools array → 400 from Nous Portal.
+        """
+        mgr = MemoryManager()
+        p = FakeMemoryProvider("ext", tools=[
+            {"name": "ext_recall", "description": "Recall", "parameters": {}},
+            {"name": "ext_remember", "description": "Remember", "parameters": {}},
+        ])
+        mgr.add_provider(p)
+
+        # Simulate self.tools already containing one of the plugin tools
+        # (as if it was registered via ctx.register_tool → get_tool_definitions)
+        existing_tools = [
+            {"type": "function", "function": {"name": "ext_recall", "description": "Recall (from registry)", "parameters": {}}},
+            {"type": "function", "function": {"name": "web_search", "description": "Search", "parameters": {}}},
+        ]
+
+        # Apply the same dedup logic from run_agent.py __init__
+        _existing_names = {
+            t.get("function", {}).get("name")
+            for t in existing_tools
+            if isinstance(t, dict)
+        }
+        for _schema in mgr.get_all_tool_schemas():
+            _tname = _schema.get("name", "")
+            if _tname and _tname in _existing_names:
+                continue
+            existing_tools.append({"type": "function", "function": _schema})
+            if _tname:
+                _existing_names.add(_tname)
+
+        # ext_recall should NOT be duplicated; ext_remember should be added
+        tool_names = [t["function"]["name"] for t in existing_tools]
+        assert tool_names.count("ext_recall") == 1, f"ext_recall duplicated: {tool_names}"
+        assert tool_names.count("ext_remember") == 1
+        assert tool_names.count("web_search") == 1
+        assert len(existing_tools) == 3  # web_search + ext_recall + ext_remember
+
+    def test_on_memory_write_tolerates_provider_failure(self):
+        """If a provider's on_memory_write raises, others still get notified."""
+        mgr = MemoryManager()
+        bad = FakeMemoryProvider("builtin")
+        bad.on_memory_write = MagicMock(side_effect=RuntimeError("boom"))
+        good = FakeMemoryProvider("good")
+        mgr.add_provider(bad)
+        mgr.add_provider(good)
+
+        mgr.on_memory_write("add", "user", "test")
+        # Good provider still received the call despite bad provider crashing
+        assert good.memory_writes == [("add", "user", "test")]

tests/gateway/test_discord_slash_commands.py

@@ -134,6 +134,57 @@ async def test_registers_native_restart_slash_command(adapter):
     )
 
 
+# ------------------------------------------------------------------
+# Auto-registration from COMMAND_REGISTRY
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_auto_registers_missing_gateway_commands(adapter):
+    """Commands in COMMAND_REGISTRY that aren't explicitly registered should
+    be auto-registered by the dynamic catch-all block."""
+    adapter._run_simple_slash = AsyncMock()
+    adapter._register_slash_commands()
+
+    tree_names = set(adapter._client.tree.commands.keys())
+
+    # These commands are gateway-available but were not in the original
+    # hardcoded registration list — they should be auto-registered.
+    expected_auto = {"debug", "yolo", "reload", "profile"}
+    for name in expected_auto:
+        assert name in tree_names, f"/{name} should be auto-registered on Discord"
+
+
+@pytest.mark.asyncio
+async def test_auto_registered_command_dispatches_correctly(adapter):
+    """Auto-registered commands should dispatch via _run_simple_slash."""
+    adapter._run_simple_slash = AsyncMock()
+    adapter._register_slash_commands()
+
+    # /debug has no args — test parameterless dispatch
+    debug_cmd = adapter._client.tree.commands["debug"]
+    interaction = SimpleNamespace()
+    adapter._run_simple_slash.reset_mock()
+    await debug_cmd.callback(interaction)
+    adapter._run_simple_slash.assert_awaited_once_with(interaction, "/debug")
+
+
+@pytest.mark.asyncio
+async def test_auto_registered_command_with_args(adapter):
+    """Auto-registered commands with args_hint should accept an optional args param."""
+    adapter._run_simple_slash = AsyncMock()
+    adapter._register_slash_commands()
+
+    # /branch has args_hint="[name]" — test dispatch with args
+    branch_cmd = adapter._client.tree.commands["branch"]
+    interaction = SimpleNamespace()
+    adapter._run_simple_slash.reset_mock()
+    await branch_cmd.callback(interaction, args="my-branch")
+    adapter._run_simple_slash.assert_awaited_once_with(
+        interaction, "/branch my-branch"
+    )
+
+
 # ------------------------------------------------------------------
 # _handle_thread_create_slash — success, session dispatch, failure
 # ------------------------------------------------------------------

tests/gateway/test_duplicate_reply_suppression.py

@@ -232,9 +232,72 @@ class TestAlreadySentWithoutResponsePreviewed:
 
 
 # ===================================================================
-# Test 3: run.py queued-message path — _already_streamed detection
+# Test 2b: run.py — empty response never suppressed (#10xxx)
 # ===================================================================
 
+class TestEmptyResponseNotSuppressed:
+    """When the model returns '(empty)' after tool calls (e.g. mimo-v2-pro
+    going silent after web_search), the gateway must NOT suppress delivery
+    even if the stream consumer sent intermediate text earlier.
+
+    Without this fix, the user sees partial streaming text ('Let me search
+    for that') and then silence — the '(empty)' sentinel is swallowed by
+    already_sent=True."""
+
+    def _make_mock_stream_consumer(self, already_sent=False, final_response_sent=False):
+        return SimpleNamespace(
+            already_sent=already_sent,
+            final_response_sent=final_response_sent,
+        )
+
+    def _apply_suppression_logic(self, response, sc):
+        """Reproduce the fixed logic from gateway/run.py return path."""
+        if sc and isinstance(response, dict) and not response.get("failed"):
+            _final = response.get("final_response") or ""
+            _is_empty_sentinel = not _final or _final == "(empty)"
+            if not _is_empty_sentinel and (
+                getattr(sc, "final_response_sent", False)
+                or getattr(sc, "already_sent", False)
+            ):
+                response["already_sent"] = True
+
+    def test_empty_sentinel_not_suppressed_with_already_sent(self):
+        """'(empty)' final_response should NOT be suppressed even when
+        streaming sent intermediate content."""
+        sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=True)
+        response = {"final_response": "(empty)"}
+        self._apply_suppression_logic(response, sc)
+        assert "already_sent" not in response
+
+    def test_empty_string_not_suppressed_with_already_sent(self):
+        """Empty string final_response should NOT be suppressed."""
+        sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=True)
+        response = {"final_response": ""}
+        self._apply_suppression_logic(response, sc)
+        assert "already_sent" not in response
+
+    def test_none_response_not_suppressed_with_already_sent(self):
+        """None final_response should NOT be suppressed."""
+        sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=True)
+        response = {"final_response": None}
+        self._apply_suppression_logic(response, sc)
+        assert "already_sent" not in response
+
+    def test_real_response_still_suppressed_with_already_sent(self):
+        """Normal non-empty response should still be suppressed when
+        streaming delivered content."""
+        sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=False)
+        response = {"final_response": "Here are the search results..."}
+        self._apply_suppression_logic(response, sc)
+        assert response.get("already_sent") is True
+
+    def test_failed_empty_response_never_suppressed(self):
+        """Failed responses are never suppressed regardless of content."""
+        sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=True)
+        response = {"final_response": "(empty)", "failed": True}
+        self._apply_suppression_logic(response, sc)
+        assert "already_sent" not in response
+
 class TestQueuedMessageAlreadyStreamed:
     """The queued-message path should detect that the first response was
     already streamed (already_sent=True) even without response_previewed."""

tests/gateway/test_message_deduplicator.py

@@ -0,0 +1,89 @@
+"""Tests for MessageDeduplicator TTL enforcement (#10306).
+
+Previously, is_duplicate() returned True for any previously seen ID without
+checking its age — expired entries were only purged when cache size exceeded
+max_size.  Normal workloads never overflowed, so messages stayed "duplicate"
+forever.
+
+The fix checks TTL at query time: if the entry's timestamp plus TTL is in
+the past, the entry is treated as expired and the message is allowed through.
+"""
+
+import time
+from unittest.mock import patch
+
+from gateway.platforms.helpers import MessageDeduplicator
+
+
+class TestMessageDeduplicatorTTL:
+    """TTL-based expiration must work regardless of cache size."""
+
+    def test_duplicate_within_ttl(self):
+        """Same message within TTL window is duplicate."""
+        dedup = MessageDeduplicator(ttl_seconds=60)
+        assert dedup.is_duplicate("msg-1") is False
+        assert dedup.is_duplicate("msg-1") is True
+
+    def test_not_duplicate_after_ttl_expires(self):
+        """Same message AFTER TTL expires should NOT be duplicate."""
+        dedup = MessageDeduplicator(ttl_seconds=5)
+        assert dedup.is_duplicate("msg-1") is False
+
+        # Fast-forward time past TTL
+        dedup._seen["msg-1"] = time.time() - 10  # 10s ago, TTL is 5s
+        assert dedup.is_duplicate("msg-1") is False, \
+            "Expired entry should not be treated as duplicate"
+
+    def test_expired_entry_gets_refreshed(self):
+        """After an expired entry is allowed through, it should be re-tracked."""
+        dedup = MessageDeduplicator(ttl_seconds=5)
+        assert dedup.is_duplicate("msg-1") is False
+
+        # Expire the entry
+        dedup._seen["msg-1"] = time.time() - 10
+
+        # Should be allowed through (expired)
+        assert dedup.is_duplicate("msg-1") is False
+        # Now should be duplicate again (freshly tracked)
+        assert dedup.is_duplicate("msg-1") is True
+
+    def test_different_messages_not_confused(self):
+        """Different message IDs are independent."""
+        dedup = MessageDeduplicator(ttl_seconds=60)
+        assert dedup.is_duplicate("msg-1") is False
+        assert dedup.is_duplicate("msg-2") is False
+        assert dedup.is_duplicate("msg-1") is True
+        assert dedup.is_duplicate("msg-2") is True
+
+    def test_empty_id_never_duplicate(self):
+        """Empty/None message IDs are never treated as duplicate."""
+        dedup = MessageDeduplicator(ttl_seconds=60)
+        assert dedup.is_duplicate("") is False
+        assert dedup.is_duplicate("") is False
+
+    def test_max_size_eviction_prunes_expired(self):
+        """Cache pruning on overflow removes expired entries."""
+        dedup = MessageDeduplicator(max_size=5, ttl_seconds=60)
+        # Add 6 entries, with the first 3 expired
+        now = time.time()
+        for i in range(3):
+            dedup._seen[f"old-{i}"] = now - 120  # expired (2 min ago, TTL 60s)
+        for i in range(3):
+            dedup.is_duplicate(f"new-{i}")
+        # Now we have 6 entries. Next insert triggers pruning.
+        dedup.is_duplicate("trigger")
+        # The 3 expired entries should be gone, leaving 4 fresh ones
+        assert len(dedup._seen) == 4
+        assert "old-0" not in dedup._seen
+        assert "new-0" in dedup._seen
+
+    def test_ttl_zero_means_no_dedup(self):
+        """With TTL=0, all entries expire immediately."""
+        dedup = MessageDeduplicator(ttl_seconds=0)
+        assert dedup.is_duplicate("msg-1") is False
+        # Entry was just added at time.time(), and TTL is 0,
+        # so now - seen_time >= 0 = ttl, meaning it's expired
+        # But time.time() might be the exact same float, so
+        # the check is `now - ts < ttl` which is `0 < 0` = False
+        # This means TTL=0 effectively disables dedup
+        assert dedup.is_duplicate("msg-1") is False

tests/gateway/test_session_env.py

@@ -1,6 +1,8 @@
 import asyncio
 import os
 
+import pytest
+
 from gateway.config import Platform
 from gateway.run import GatewayRunner
 from gateway.session import SessionContext, SessionSource
@@ -8,9 +10,26 @@ from gateway.session_context import (
     get_session_env,
     set_session_vars,
     clear_session_vars,
+    _VAR_MAP,
+    _UNSET,
 )
 
 
+@pytest.fixture(autouse=True)
+def _reset_contextvars():
+    """Reset all session contextvars to _UNSET between tests.
+
+    In production each asyncio.Task gets a fresh context copy where the
+    defaults are _UNSET.  In tests all functions share the same thread
+    context, so a clear_session_vars() from test A (which sets vars to "")
+    would leak into test B.  This fixture ensures each test starts clean.
+    """
+    yield
+    for var in _VAR_MAP.values():
+        # Can't use var.reset() without a token; just set back to sentinel.
+        var.set(_UNSET)
+
+
 def test_set_session_env_sets_contextvars(monkeypatch):
     """_set_session_env should populate contextvars, not os.environ."""
     runner = object.__new__(GatewayRunner)
@@ -98,9 +117,11 @@ def test_get_session_env_falls_back_to_os_environ(monkeypatch):
     tokens = set_session_vars(platform="telegram")
     assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
 
-    # Restore — should fall back to os.environ again
+    # After clear — should return "" (explicitly cleared), NOT fall back
+    # to os.environ.  This is the fix for #10304: stale os.environ values
+    # must not leak through after a gateway session is cleaned up.
     clear_session_vars(tokens)
-    assert get_session_env("HERMES_SESSION_PLATFORM") == "discord"
+    assert get_session_env("HERMES_SESSION_PLATFORM") == ""
 
 
 def test_get_session_env_default_when_nothing_set(monkeypatch):
@@ -164,9 +185,9 @@ def test_session_key_falls_back_to_os_environ(monkeypatch):
     tokens = set_session_vars(session_key="ctx-session-456")
     assert get_session_env("HERMES_SESSION_KEY") == "ctx-session-456"
 
-    # Restore — should fall back to os.environ
+    # After clear — should return "" (explicitly cleared), not os.environ (#10304)
     clear_session_vars(tokens)
-    assert get_session_env("HERMES_SESSION_KEY") == "env-session-123"
+    assert get_session_env("HERMES_SESSION_KEY") == ""
 
 
 def test_set_session_env_includes_session_key():

tests/hermes_cli/test_debug.py

@@ -428,7 +428,9 @@ class TestRunDebug:
         run_debug(args)
 
         out = capsys.readouterr().out
-        assert "hermes debug share" in out
+        assert "hermes debug" in out
+        assert "share" in out
+        assert "delete" in out
 
     def test_share_subcommand_routes(self, hermes_home):
         from hermes_cli.debug import run_debug
@@ -459,3 +461,187 @@ class TestArgparseIntegration:
         args = MagicMock()
         args.debug_command = None
         cmd_debug(args)
+
+
+# ---------------------------------------------------------------------------
+# Delete / auto-delete
+# ---------------------------------------------------------------------------
+
+class TestExtractPasteId:
+    def test_paste_rs_url(self):
+        from hermes_cli.debug import _extract_paste_id
+        assert _extract_paste_id("https://paste.rs/abc123") == "abc123"
+
+    def test_paste_rs_trailing_slash(self):
+        from hermes_cli.debug import _extract_paste_id
+        assert _extract_paste_id("https://paste.rs/abc123/") == "abc123"
+
+    def test_http_variant(self):
+        from hermes_cli.debug import _extract_paste_id
+        assert _extract_paste_id("http://paste.rs/xyz") == "xyz"
+
+    def test_non_paste_rs_returns_none(self):
+        from hermes_cli.debug import _extract_paste_id
+        assert _extract_paste_id("https://dpaste.com/ABCDEF") is None
+
+    def test_empty_returns_none(self):
+        from hermes_cli.debug import _extract_paste_id
+        assert _extract_paste_id("") is None
+
+
+class TestDeletePaste:
+    def test_delete_sends_delete_request(self):
+        from hermes_cli.debug import delete_paste
+
+        mock_resp = MagicMock()
+        mock_resp.status = 200
+        mock_resp.__enter__ = lambda s: s
+        mock_resp.__exit__ = MagicMock(return_value=False)
+
+        with patch("hermes_cli.debug.urllib.request.urlopen",
+                    return_value=mock_resp) as mock_open:
+            result = delete_paste("https://paste.rs/abc123")
+
+        assert result is True
+        req = mock_open.call_args[0][0]
+        assert req.method == "DELETE"
+        assert "paste.rs/abc123" in req.full_url
+
+    def test_delete_rejects_non_paste_rs(self):
+        from hermes_cli.debug import delete_paste
+
+        with pytest.raises(ValueError, match="only paste.rs"):
+            delete_paste("https://dpaste.com/something")
+
+
+class TestScheduleAutoDelete:
+    def test_spawns_detached_process(self):
+        from hermes_cli.debug import _schedule_auto_delete
+
+        with patch("subprocess.Popen") as mock_popen:
+            _schedule_auto_delete(
+                ["https://paste.rs/abc", "https://paste.rs/def"],
+                delay_seconds=10,
+            )
+
+        mock_popen.assert_called_once()
+        call_args = mock_popen.call_args
+        # Verify detached
+        assert call_args[1]["start_new_session"] is True
+        # Verify the script references both URLs
+        script = call_args[0][0][2]  # [python, -c, script]
+        assert "paste.rs/abc" in script
+        assert "paste.rs/def" in script
+        assert "time.sleep(10)" in script
+
+    def test_skips_non_paste_rs_urls(self):
+        from hermes_cli.debug import _schedule_auto_delete
+
+        with patch("subprocess.Popen") as mock_popen:
+            _schedule_auto_delete(["https://dpaste.com/something"])
+
+        mock_popen.assert_not_called()
+
+    def test_handles_popen_failure_gracefully(self):
+        from hermes_cli.debug import _schedule_auto_delete
+
+        with patch("subprocess.Popen",
+                    side_effect=OSError("no such file")):
+            # Should not raise
+            _schedule_auto_delete(["https://paste.rs/abc"])
+
+
+class TestRunDebugDelete:
+    def test_deletes_valid_url(self, capsys):
+        from hermes_cli.debug import run_debug_delete
+
+        args = MagicMock()
+        args.urls = ["https://paste.rs/abc"]
+
+        with patch("hermes_cli.debug.delete_paste", return_value=True):
+            run_debug_delete(args)
+
+        out = capsys.readouterr().out
+        assert "Deleted" in out
+        assert "paste.rs/abc" in out
+
+    def test_handles_delete_failure(self, capsys):
+        from hermes_cli.debug import run_debug_delete
+
+        args = MagicMock()
+        args.urls = ["https://paste.rs/abc"]
+
+        with patch("hermes_cli.debug.delete_paste",
+                    side_effect=Exception("network error")):
+            run_debug_delete(args)
+
+        out = capsys.readouterr().out
+        assert "Could not delete" in out
+
+    def test_no_urls_shows_usage(self, capsys):
+        from hermes_cli.debug import run_debug_delete
+
+        args = MagicMock()
+        args.urls = []
+
+        run_debug_delete(args)
+
+        out = capsys.readouterr().out
+        assert "Usage" in out
+
+
+class TestShareIncludesAutoDelete:
+    """Verify that run_debug_share schedules auto-deletion and prints TTL."""
+
+    def test_share_schedules_auto_delete(self, hermes_home, capsys):
+        from hermes_cli.debug import run_debug_share
+
+        args = MagicMock()
+        args.lines = 50
+        args.expire = 7
+        args.local = False
+
+        with patch("hermes_cli.dump.run_dump"), \
+             patch("hermes_cli.debug.upload_to_pastebin",
+                    return_value="https://paste.rs/test1"), \
+             patch("hermes_cli.debug._schedule_auto_delete") as mock_sched:
+            run_debug_share(args)
+
+        # auto-delete was scheduled with the uploaded URLs
+        mock_sched.assert_called_once()
+        urls_arg = mock_sched.call_args[0][0]
+        assert "https://paste.rs/test1" in urls_arg
+
+        out = capsys.readouterr().out
+        assert "auto-delete" in out
+
+    def test_share_shows_privacy_notice(self, hermes_home, capsys):
+        from hermes_cli.debug import run_debug_share
+
+        args = MagicMock()
+        args.lines = 50
+        args.expire = 7
+        args.local = False
+
+        with patch("hermes_cli.dump.run_dump"), \
+             patch("hermes_cli.debug.upload_to_pastebin",
+                    return_value="https://paste.rs/test"), \
+             patch("hermes_cli.debug._schedule_auto_delete"):
+            run_debug_share(args)
+
+        out = capsys.readouterr().out
+        assert "public paste service" in out
+
+    def test_local_no_privacy_notice(self, hermes_home, capsys):
+        from hermes_cli.debug import run_debug_share
+
+        args = MagicMock()
+        args.lines = 50
+        args.expire = 7
+        args.local = True
+
+        with patch("hermes_cli.dump.run_dump"):
+            run_debug_share(args)
+
+        out = capsys.readouterr().out
+        assert "public paste service" not in out

tests/run_agent/test_async_httpx_del_neuter.py

@@ -103,7 +103,7 @@ class TestCleanupStaleAsyncClients:
         mock_client._client = MagicMock()
         mock_client._client.is_closed = False
 
-        key = ("test_stale", True, "", "", id(loop))
+        key = ("test_stale", True, "", "", "", ())
         with _client_cache_lock:
             _client_cache[key] = (mock_client, "test-model", loop)
 
@@ -127,7 +127,7 @@ class TestCleanupStaleAsyncClients:
         loop = asyncio.new_event_loop()  # NOT closed
 
         mock_client = MagicMock()
-        key = ("test_live", True, "", "", id(loop))
+        key = ("test_live", True, "", "", "", ())
         with _client_cache_lock:
             _client_cache[key] = (mock_client, "test-model", loop)
 
@@ -149,7 +149,7 @@ class TestCleanupStaleAsyncClients:
         )
 
         mock_client = MagicMock()
-        key = ("test_sync", False, "", "", 0)
+        key = ("test_sync", False, "", "", "", ())
         with _client_cache_lock:
             _client_cache[key] = (mock_client, "test-model", None)
 
@@ -160,3 +160,131 @@ class TestCleanupStaleAsyncClients:
         finally:
             with _client_cache_lock:
                 _client_cache.pop(key, None)
+
+
+# ---------------------------------------------------------------------------
+# Cache bounded growth (#10200)
+# ---------------------------------------------------------------------------
+
+class TestClientCacheBoundedGrowth:
+    """Verify the cache stays bounded when loops change (fix for #10200).
+
+    Previously, loop_id was part of the cache key, so every new event loop
+    created a new entry for the same provider config.  Now loop identity is
+    validated at hit time and stale entries are replaced in-place.
+    """
+
+    def test_same_key_replaces_stale_loop_entry(self):
+        """When the loop changes, the old entry should be replaced, not duplicated."""
+        from agent.auxiliary_client import (
+            _client_cache,
+            _client_cache_lock,
+            _get_cached_client,
+        )
+
+        key = ("test_replace", True, "", "", "", ())
+
+        # Simulate a stale entry from a closed loop
+        old_loop = asyncio.new_event_loop()
+        old_loop.close()
+        old_client = MagicMock()
+        old_client._client = MagicMock()
+        old_client._client.is_closed = False
+
+        with _client_cache_lock:
+            _client_cache[key] = (old_client, "old-model", old_loop)
+
+        try:
+            # Now call _get_cached_client — should detect stale loop and evict
+            with patch("agent.auxiliary_client.resolve_provider_client") as mock_resolve:
+                mock_resolve.return_value = (MagicMock(), "new-model")
+                client, model = _get_cached_client(
+                    "test_replace", async_mode=True,
+                )
+            # The old entry should have been replaced
+            with _client_cache_lock:
+                assert key in _client_cache, "Key should still exist (replaced)"
+                entry = _client_cache[key]
+                assert entry[1] == "new-model", "Should have the new model"
+        finally:
+            with _client_cache_lock:
+                _client_cache.pop(key, None)
+
+    def test_different_loops_do_not_grow_cache(self):
+        """Multiple event loops for the same provider should NOT create multiple entries."""
+        from agent.auxiliary_client import (
+            _client_cache,
+            _client_cache_lock,
+        )
+
+        key = ("test_no_grow", True, "", "", "", ())
+
+        loops = []
+        try:
+            for i in range(5):
+                loop = asyncio.new_event_loop()
+                loops.append(loop)
+                mock_client = MagicMock()
+                mock_client._client = MagicMock()
+                mock_client._client.is_closed = False
+
+                # Close previous loop entries (simulating worker thread recycling)
+                if i > 0:
+                    loops[i - 1].close()
+
+                with _client_cache_lock:
+                    # Simulate what _get_cached_client does: replace on loop mismatch
+                    if key in _client_cache:
+                        old_entry = _client_cache[key]
+                        del _client_cache[key]
+                    _client_cache[key] = (mock_client, f"model-{i}", loop)
+
+            # Only one entry should exist for this key
+            with _client_cache_lock:
+                count = sum(1 for k in _client_cache if k == key)
+                assert count == 1, f"Expected 1 entry, got {count}"
+        finally:
+            for loop in loops:
+                if not loop.is_closed():
+                    loop.close()
+            with _client_cache_lock:
+                _client_cache.pop(key, None)
+
+    def test_max_cache_size_eviction(self):
+        """Cache should not exceed _CLIENT_CACHE_MAX_SIZE."""
+        from agent.auxiliary_client import (
+            _client_cache,
+            _client_cache_lock,
+            _CLIENT_CACHE_MAX_SIZE,
+        )
+
+        # Save existing cache state
+        with _client_cache_lock:
+            saved = dict(_client_cache)
+            _client_cache.clear()
+
+        try:
+            # Fill to max + 5
+            for i in range(_CLIENT_CACHE_MAX_SIZE + 5):
+                mock_client = MagicMock()
+                mock_client._client = MagicMock()
+                mock_client._client.is_closed = False
+                key = (f"evict_test_{i}", False, "", "", "", ())
+                with _client_cache_lock:
+                    # Inline the eviction logic (same as _get_cached_client)
+                    while len(_client_cache) >= _CLIENT_CACHE_MAX_SIZE:
+                        evict_key = next(iter(_client_cache))
+                        del _client_cache[evict_key]
+                    _client_cache[key] = (mock_client, f"model-{i}", None)
+
+            with _client_cache_lock:
+                assert len(_client_cache) <= _CLIENT_CACHE_MAX_SIZE, \
+                    f"Cache size {len(_client_cache)} exceeds max {_CLIENT_CACHE_MAX_SIZE}"
+                # The earliest entries should have been evicted
+                assert ("evict_test_0", False, "", "", "", ()) not in _client_cache
+                # The latest entries should be present
+                assert (f"evict_test_{_CLIENT_CACHE_MAX_SIZE + 4}", False, "", "", "", ()) in _client_cache
+        finally:
+            with _client_cache_lock:
+                _client_cache.clear()
+                _client_cache.update(saved)

tests/run_agent/test_interrupt_propagation.py

@@ -28,7 +28,8 @@ class TestInterruptPropagationToChild(unittest.TestCase):
         agent = AIAgent.__new__(AIAgent)
         agent._interrupt_requested = False
         agent._interrupt_message = None
-        agent._execution_thread_id = None  # defaults to current thread in set_interrupt
+        agent._execution_thread_id = None
+        agent._interrupt_thread_signal_pending = False
         agent._active_children = []
         agent._active_children_lock = threading.Lock()
         agent.quiet_mode = True
@@ -46,15 +47,17 @@ class TestInterruptPropagationToChild(unittest.TestCase):
         assert parent._interrupt_requested is True
         assert child._interrupt_requested is True
         assert child._interrupt_message == "new user message"
-        assert is_interrupted() is True
+        assert is_interrupted() is False
+        assert parent._interrupt_thread_signal_pending is True
 
     def test_child_clear_interrupt_at_start_clears_thread(self):
         """child.clear_interrupt() at start of run_conversation clears the
-        per-thread interrupt flag for the current thread.
+        bound execution thread's interrupt flag.
         """
         child = self._make_bare_agent()
         child._interrupt_requested = True
         child._interrupt_message = "msg"
+        child._execution_thread_id = threading.current_thread().ident
 
         # Interrupt for current thread is set
         set_interrupt(True)
@@ -128,6 +131,36 @@ class TestInterruptPropagationToChild(unittest.TestCase):
         child_thread.join(timeout=1)
         set_interrupt(False)
 
+    def test_prestart_interrupt_binds_to_execution_thread(self):
+        """An interrupt that arrives before startup should bind to the agent thread."""
+        agent = self._make_bare_agent()
+        barrier = threading.Barrier(2)
+        result = {}
+
+        agent.interrupt("stop before start")
+        assert agent._interrupt_requested is True
+        assert agent._interrupt_thread_signal_pending is True
+        assert is_interrupted() is False
+
+        def run_thread():
+            from tools.interrupt import set_interrupt as _set_interrupt_for_test
+
+            agent._execution_thread_id = threading.current_thread().ident
+            _set_interrupt_for_test(False, agent._execution_thread_id)
+            if agent._interrupt_requested:
+                _set_interrupt_for_test(True, agent._execution_thread_id)
+                agent._interrupt_thread_signal_pending = False
+            barrier.wait(timeout=5)
+            result["thread_interrupted"] = is_interrupted()
+
+        t = threading.Thread(target=run_thread)
+        t.start()
+        barrier.wait(timeout=5)
+        t.join(timeout=2)
+
+        assert result["thread_interrupted"] is True
+        assert agent._interrupt_thread_signal_pending is False
+
 
 class TestPerThreadInterruptIsolation(unittest.TestCase):
     """Verify that interrupting one agent does NOT affect another agent's thread.

tests/tools/test_browser_camofox.py

@@ -37,6 +37,18 @@ class TestCamofoxMode:
         monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
         assert is_camofox_mode() is True
 
+    def test_cdp_override_takes_priority(self, monkeypatch):
+        """When BROWSER_CDP_URL is set (via /browser connect), CDP takes priority over Camofox."""
+        monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
+        monkeypatch.setenv("BROWSER_CDP_URL", "http://127.0.0.1:9222")
+        assert is_camofox_mode() is False
+
+    def test_cdp_override_blank_does_not_disable_camofox(self, monkeypatch):
+        """Empty/whitespace BROWSER_CDP_URL should not suppress Camofox."""
+        monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
+        monkeypatch.setenv("BROWSER_CDP_URL", "  ")
+        assert is_camofox_mode() is True
+
     def test_health_check_unreachable(self, monkeypatch):
         monkeypatch.setenv("CAMOFOX_URL", "http://localhost:19999")
         assert check_camofox_available() is False

tests/tools/test_managed_modal_environment.py

@@ -296,7 +296,7 @@ def test_managed_modal_execute_times_out_and_cancels(monkeypatch):
     modal_common = sys.modules["tools.environments.modal_utils"]
 
     calls = []
-    monotonic_values = iter([0.0, 12.5])
+    monotonic_values = iter([0.0, 0.0, 0.0, 12.5, 12.5])
 
     def fake_request(method, url, headers=None, json=None, timeout=None):
         calls.append((method, url, json, timeout))

tests/tools/test_session_search.py

@@ -290,6 +290,63 @@ class TestSessionSearch:
         assert result["results"] == []
         assert result["sessions_searched"] == 0
 
+    def test_limit_none_coerced_to_default(self):
+        """Model sends limit=null → should fall back to 3, not TypeError."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = []
+
+        result = json.loads(session_search(
+            query="test", db=mock_db, limit=None,
+        ))
+        assert result["success"] is True
+
+    def test_limit_type_object_coerced_to_default(self):
+        """Model sends limit as a type object → should fall back to 3, not TypeError."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = []
+
+        result = json.loads(session_search(
+            query="test", db=mock_db, limit=int,
+        ))
+        assert result["success"] is True
+
+    def test_limit_string_coerced(self):
+        """Model sends limit as string '2' → should coerce to int."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = []
+
+        result = json.loads(session_search(
+            query="test", db=mock_db, limit="2",
+        ))
+        assert result["success"] is True
+
+    def test_limit_clamped_to_range(self):
+        """Negative or zero limit should be clamped to 1."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = []
+
+        result = json.loads(session_search(
+            query="test", db=mock_db, limit=-5,
+        ))
+        assert result["success"] is True
+
+        result = json.loads(session_search(
+            query="test", db=mock_db, limit=0,
+        ))
+        assert result["success"] is True
+
     def test_current_root_session_excludes_child_lineage(self):
         """Delegation child hits should be excluded when they resolve to the current root session."""
         from unittest.mock import MagicMock