Merge branch 'bb/gui' into bb/gui-glass

Brings in main (via bb/gui) plus the bb/gui-only changes since the
last sync, so a future bb/gui-glass → bb/gui merge is conflict-free.

Conflicts resolved:
- apps/desktop/src/app/chat/composer/focus.ts (add/add): keep the
  glass version. It is a strict superset of the bb/gui original —
  same focus API (`requestComposerFocus`, `onComposerFocusRequest`,
  `markActiveComposer`) plus the insert bus
  (`requestComposerInsert`, `onComposerInsertRequest`,
  `focusComposerInput`) that the glass composer / right-rail
  preview / use-composer-actions already depend on.
- apps/desktop/src/app/skills/index.tsx: keep the glass rewrite
  built on `PageSearchShell` + `Codicon` + `TextTab` — bb/gui's
  older `titlebarHeaderBaseClass` + ad-hoc `Input`/`Search`/`X`
  layout is the version this PR was meant to replace.

`npm run type-check` in apps/desktop passes against the merged tree.

tests/agent/test_compressor_historical_media.py

@@ -0,0 +1,266 @@
+"""Tests for post-compression historical-media stripping.
+
+Port of Kilo-Org/kilocode#9434 (adapted for OpenAI-style message lists).
+Without this pass, tail messages keep their original multi-MB base-64 image
+payloads after context compression, and every subsequent request re-ships
+them — sometimes breaching provider body-size limits and wedging the
+session.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from agent.context_compressor import (
+    ContextCompressor,
+    _content_has_images,
+    _is_image_part,
+    _strip_historical_media,
+    _strip_images_from_content,
+)
+
+
+IMG_URL = {
+    "type": "image_url",
+    "image_url": {"url": "data:image/png;base64," + ("A" * 1024)},
+}
+INPUT_IMG = {
+    "type": "input_image",
+    "image_url": "data:image/png;base64," + ("B" * 1024),
+}
+ANTHROPIC_IMG = {
+    "type": "image",
+    "source": {"type": "base64", "media_type": "image/png", "data": "C" * 1024},
+}
+TEXT = {"type": "text", "text": "hi"}
+INPUT_TEXT = {"type": "input_text", "text": "hi"}
+
+
+class TestIsImagePart:
+    def test_openai_chat_shape(self):
+        assert _is_image_part(IMG_URL) is True
+
+    def test_openai_responses_shape(self):
+        assert _is_image_part(INPUT_IMG) is True
+
+    def test_anthropic_native_shape(self):
+        assert _is_image_part(ANTHROPIC_IMG) is True
+
+    def test_text_part_is_not_image(self):
+        assert _is_image_part(TEXT) is False
+        assert _is_image_part(INPUT_TEXT) is False
+
+    def test_non_dict_rejected(self):
+        assert _is_image_part("image") is False
+        assert _is_image_part(None) is False
+        assert _is_image_part(42) is False
+
+
+class TestContentHasImages:
+    def test_string_content(self):
+        assert _content_has_images("a string") is False
+
+    def test_empty_list(self):
+        assert _content_has_images([]) is False
+
+    def test_text_only_list(self):
+        assert _content_has_images([TEXT, TEXT]) is False
+
+    def test_list_with_image(self):
+        assert _content_has_images([TEXT, IMG_URL]) is True
+
+    def test_none(self):
+        assert _content_has_images(None) is False
+
+
+class TestStripImagesFromContent:
+    def test_string_passthrough(self):
+        assert _strip_images_from_content("hello") == "hello"
+
+    def test_none_passthrough(self):
+        assert _strip_images_from_content(None) is None
+
+    def test_text_only_passthrough(self):
+        parts = [TEXT, {"type": "text", "text": "world"}]
+        assert _strip_images_from_content(parts) == parts
+
+    def test_replaces_image_with_placeholder(self):
+        parts = [TEXT, IMG_URL]
+        out = _strip_images_from_content(parts)
+        assert len(out) == 2
+        assert out[0] == TEXT
+        assert out[1] == {
+            "type": "text",
+            "text": "[Attached image — stripped after compression]",
+        }
+
+    def test_does_not_mutate_input(self):
+        parts = [IMG_URL, TEXT]
+        _ = _strip_images_from_content(parts)
+        assert parts[0] is IMG_URL  # original list untouched
+        assert parts[1] is TEXT
+
+    def test_handles_all_three_shapes(self):
+        parts = [IMG_URL, INPUT_IMG, ANTHROPIC_IMG, TEXT]
+        out = _strip_images_from_content(parts)
+        assert sum(1 for p in out if p.get("type") == "text") == 4
+        assert not any(_is_image_part(p) for p in out)
+
+
+class TestStripHistoricalMedia:
+    def test_empty_passthrough(self):
+        assert _strip_historical_media([]) == []
+
+    def test_no_images_anywhere(self):
+        msgs = [
+            {"role": "user", "content": "hi"},
+            {"role": "assistant", "content": "hey"},
+            {"role": "user", "content": "bye"},
+        ]
+        assert _strip_historical_media(msgs) is msgs  # identity — no copy
+
+    def test_single_image_user_only_first_message(self):
+        # Only image-bearing user is the first message — nothing before it.
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "assistant", "content": "ok"},
+        ]
+        out = _strip_historical_media(msgs)
+        assert out is msgs  # no-op
+        # Image still there.
+        assert _content_has_images(out[0]["content"])
+
+    def test_strips_older_user_image_keeps_newest(self):
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},     # old — strip
+            {"role": "assistant", "content": "looked at it"},
+            {"role": "user", "content": [TEXT, INPUT_IMG]},   # newest — keep
+        ]
+        out = _strip_historical_media(msgs)
+        assert out is not msgs  # new list
+        # First message's image was replaced
+        assert not _content_has_images(out[0]["content"])
+        # Newest user still has its image
+        assert _content_has_images(out[2]["content"])
+
+    def test_strips_assistant_and_tool_images_before_anchor(self):
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},          # old user
+            {"role": "assistant", "content": [TEXT, IMG_URL]},     # old assistant
+            {"role": "tool", "content": [TEXT, IMG_URL], "tool_call_id": "t1"},
+            {"role": "user", "content": [TEXT, IMG_URL]},          # newest user — keep
+        ]
+        out = _strip_historical_media(msgs)
+        for i in range(3):
+            assert not _content_has_images(out[i]["content"]), f"msg {i} still has image"
+        assert _content_has_images(out[3]["content"])
+
+    def test_text_only_newest_user_still_strips_older_images(self):
+        # The anchor is "newest user WITH images". If the newest user is
+        # text-only, we fall back to the previous image-bearing user turn.
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": [TEXT, IMG_URL]},  # anchor
+            {"role": "assistant", "content": "done"},
+            {"role": "user", "content": "follow-up text only"},
+        ]
+        out = _strip_historical_media(msgs)
+        # First image-bearing user (index 0) was stripped — it was before the
+        # newest image-bearing user (index 2).
+        assert not _content_has_images(out[0]["content"])
+        # Anchor (index 2) keeps its image.
+        assert _content_has_images(out[2]["content"])
+
+    def test_no_image_bearing_user_is_noop(self):
+        msgs = [
+            {"role": "user", "content": "first"},
+            {"role": "assistant", "content": [TEXT, IMG_URL]},  # assistant image only
+            {"role": "user", "content": "second"},
+        ]
+        out = _strip_historical_media(msgs)
+        # No image-bearing user anchor → no stripping.
+        assert out is msgs
+        assert _content_has_images(out[1]["content"])
+
+    def test_does_not_mutate_input_messages(self):
+        msg0 = {"role": "user", "content": [TEXT, IMG_URL]}
+        msg1 = {"role": "user", "content": [TEXT, IMG_URL]}
+        msgs = [msg0, msg1]
+        _ = _strip_historical_media(msgs)
+        # Originals untouched
+        assert _content_has_images(msg0["content"])
+        assert _content_has_images(msg1["content"])
+
+    def test_idempotent(self):
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "assistant", "content": "k"},
+            {"role": "user", "content": [TEXT, IMG_URL]},
+        ]
+        first = _strip_historical_media(msgs)
+        second = _strip_historical_media(first)
+        # Second pass is a no-op — no images left before the anchor.
+        assert second is first
+
+    def test_non_dict_messages_pass_through(self):
+        msgs = [
+            "not-a-dict",  # shouldn't crash
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+        assert out[0] == "not-a-dict"
+        # Image-bearing user at index 1 is before the anchor (index 3) → stripped.
+        assert not _content_has_images(out[1]["content"])
+
+
+class TestCompressIntegration:
+    """Verify the stripping runs inside ContextCompressor.compress()."""
+
+    @pytest.fixture
+    def compressor(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
+            c = ContextCompressor(
+                model="test/model",
+                threshold_percent=0.50,
+                protect_first_n=1,
+                protect_last_n=2,
+                quiet_mode=True,
+            )
+            return c
+
+    def test_compress_strips_historical_images(self, compressor):
+        # Enough messages to trigger the summarize path. protect_first_n=1 +
+        # protect_last_n=2 + a middle window of at least 3 with a summary.
+        msgs = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": [TEXT, IMG_URL]},           # old image-bearing user
+            {"role": "assistant", "content": "looked at it"},
+            {"role": "user", "content": "follow-up"},
+            {"role": "assistant", "content": "ack"},
+            {"role": "user", "content": "more"},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": [TEXT, IMG_URL]},           # newest image-bearing user (tail)
+            {"role": "assistant", "content": "done"},
+        ]
+        # Bypass the real LLM summary — return a stub so compress() proceeds.
+        with patch.object(compressor, "_generate_summary", return_value="SUMMARY TEXT"):
+            out = compressor.compress(msgs, current_tokens=60_000)
+
+        # Newest user turn with image should still have it (it's in the tail).
+        user_imgs = [m for m in out if m.get("role") == "user" and _content_has_images(m.get("content"))]
+        assert len(user_imgs) == 1, (
+            "Expected exactly one user message with images after compression "
+            f"(the newest one); got {len(user_imgs)}"
+        )
+        # No assistant or tool messages should carry images either.
+        for m in out:
+            if m is user_imgs[0]:
+                continue
+            assert not _content_has_images(m.get("content")), (
+                f"Stale image in {m.get('role')!r} message after compression"
+            )

tests/cli/test_cli_background_status_indicator.py

@@ -0,0 +1,104 @@
+"""Tests for the /background indicator in the CLI status bar.
+
+The classic prompt_toolkit status bar shows `▶ N` when N tasks launched via
+`/background` are still running. Source of truth is `self._background_tasks`
+(a Dict[str, threading.Thread]); entries are removed in the task thread's
+finally block, so len() reflects truly-running tasks.
+"""
+
+import threading
+from datetime import datetime
+
+from cli import HermesCLI
+
+
+def _stub_thread() -> threading.Thread:
+    """Return a Thread instance that's never started — pure dict-value stand-in."""
+    return threading.Thread(target=lambda: None)
+
+
+def _make_cli():
+    """Bare-metal HermesCLI for snapshot/build tests (no __init__ side effects)."""
+    cli_obj = HermesCLI.__new__(HermesCLI)
+    cli_obj.model = "anthropic/claude-opus-4.6"
+    cli_obj.agent = None
+    cli_obj._background_tasks = {}
+    # The snapshot reads session_start to compute duration; supply a stub.
+    cli_obj.session_start = datetime.now()
+    return cli_obj
+
+
+def test_snapshot_reports_zero_when_no_background_tasks():
+    cli_obj = _make_cli()
+    snap = cli_obj._get_status_bar_snapshot()
+    assert snap["active_background_tasks"] == 0
+
+
+def test_snapshot_counts_live_background_tasks():
+    cli_obj = _make_cli()
+    cli_obj._background_tasks = {"bg_a": _stub_thread(), "bg_b": _stub_thread()}
+    snap = cli_obj._get_status_bar_snapshot()
+    assert snap["active_background_tasks"] == 2
+
+
+def test_snapshot_safe_when_background_tasks_attr_missing():
+    """Older HermesCLI instances (tests with __new__, etc.) may lack the attr."""
+    cli_obj = HermesCLI.__new__(HermesCLI)
+    cli_obj.model = "x"
+    cli_obj.agent = None
+    cli_obj.session_start = datetime.now()
+    # No _background_tasks at all — must not raise.
+    snap = cli_obj._get_status_bar_snapshot()
+    assert snap["active_background_tasks"] == 0
+
+
+def test_plain_text_status_omits_indicator_when_idle():
+    cli_obj = _make_cli()
+    text = cli_obj._build_status_bar_text(width=80)
+    assert "▶" not in text
+
+
+def test_plain_text_status_shows_indicator_when_active():
+    cli_obj = _make_cli()
+    cli_obj._background_tasks = {"bg_a": _stub_thread()}
+    text = cli_obj._build_status_bar_text(width=80)
+    assert "▶ 1" in text
+
+
+def test_plain_text_status_shows_higher_count():
+    cli_obj = _make_cli()
+    cli_obj._background_tasks = {
+        "a": _stub_thread(),
+        "b": _stub_thread(),
+        "c": _stub_thread(),
+    }
+    text = cli_obj._build_status_bar_text(width=80)
+    assert "▶ 3" in text
+
+
+def test_narrow_width_omits_bg_indicator():
+    """The narrow tier (<52) is already cramped — bg is secondary, drop it."""
+    cli_obj = _make_cli()
+    cli_obj._background_tasks = {"bg_a": _stub_thread()}
+    text = cli_obj._build_status_bar_text(width=40)
+    assert "▶" not in text
+
+
+def test_fragments_include_bg_segment_when_active():
+    cli_obj = _make_cli()
+    cli_obj._background_tasks = {"a": _stub_thread(), "b": _stub_thread()}
+    cli_obj._status_bar_visible = True
+    # _get_status_bar_fragments asks _get_tui_terminal_width(); stub it wide.
+    cli_obj._get_tui_terminal_width = lambda: 120  # type: ignore[method-assign]
+    frags = cli_obj._get_status_bar_fragments()
+    rendered = "".join(text for _style, text in frags)
+    assert "▶ 2" in rendered
+
+
+def test_fragments_omit_bg_segment_when_idle():
+    cli_obj = _make_cli()
+    cli_obj._status_bar_visible = True
+    cli_obj._get_tui_terminal_width = lambda: 120  # type: ignore[method-assign]
+    frags = cli_obj._get_status_bar_fragments()
+    rendered = "".join(text for _style, text in frags)
+    assert "▶" not in rendered

tests/cli/test_exit_delete_session.py

@@ -0,0 +1,119 @@
+"""Tests for `/exit --delete` and `/quit --delete` session deletion.
+
+Ports the behavior from google-gemini/gemini-cli#19332: running `/exit` or
+`/quit` with the `--delete` flag arms a one-shot `_delete_session_on_exit`
+flag that the CLI shutdown path uses to remove the current session from
+SQLite + on-disk transcripts before exit.
+"""
+
+from unittest.mock import MagicMock
+
+
+def _make_cli():
+    """Bare HermesCLI suitable for process_command() tests.
+
+    Uses ``__new__`` to skip the heavy __init__; only sets the attributes
+    the /exit branch touches.
+    """
+    from cli import HermesCLI
+    cli = HermesCLI.__new__(HermesCLI)
+    cli.config = {}
+    cli.console = MagicMock()
+    cli.agent = None
+    cli.conversation_history = []
+    cli.session_id = "test-session"
+    cli._delete_session_on_exit = False
+    return cli
+
+
+class TestExitDeleteFlag:
+    def test_plain_exit_does_not_arm_delete(self):
+        cli = _make_cli()
+        result = cli.process_command("/exit")
+        assert result is False
+        assert cli._delete_session_on_exit is False
+
+    def test_plain_quit_does_not_arm_delete(self):
+        cli = _make_cli()
+        result = cli.process_command("/quit")
+        assert result is False
+        assert cli._delete_session_on_exit is False
+
+    def test_exit_delete_arms_flag(self):
+        cli = _make_cli()
+        result = cli.process_command("/exit --delete")
+        assert result is False
+        assert cli._delete_session_on_exit is True
+
+    def test_quit_delete_arms_flag(self):
+        cli = _make_cli()
+        result = cli.process_command("/quit --delete")
+        assert result is False
+        assert cli._delete_session_on_exit is True
+
+    def test_exit_delete_short_form(self):
+        """`-d` is a convenience alias for `--delete`."""
+        cli = _make_cli()
+        result = cli.process_command("/exit -d")
+        assert result is False
+        assert cli._delete_session_on_exit is True
+
+    def test_quit_alias_q_is_not_quit(self):
+        """`/q` is the alias for `/queue`, not `/quit`. This test documents
+        that /q --delete does NOT arm session deletion — it would dispatch
+        to /queue instead."""
+        cli = _make_cli()
+        cli._pending_input = __import__("queue").Queue()
+        # /q with no args shows a usage error and keeps the CLI running.
+        result = cli.process_command("/q")
+        assert result is not False  # queue command doesn't exit
+        assert cli._delete_session_on_exit is False
+
+    def test_delete_flag_is_case_insensitive(self):
+        cli = _make_cli()
+        result = cli.process_command("/exit --DELETE")
+        assert result is False
+        assert cli._delete_session_on_exit is True
+
+    def test_delete_flag_trims_whitespace(self):
+        cli = _make_cli()
+        result = cli.process_command("/exit   --delete   ")
+        assert result is False
+        assert cli._delete_session_on_exit is True
+
+    def test_unknown_exit_argument_does_not_exit(self):
+        """Unrecognised args should NOT exit the CLI — they surface an
+        error message and stay in the session. This prevents accidental
+        session destruction from typos like `/exit -delete`."""
+        cli = _make_cli()
+        result = cli.process_command("/exit --delte")
+        # process_command returns True = keep running
+        assert result is True
+        assert cli._delete_session_on_exit is False
+
+    def test_unknown_exit_argument_prints_help(self):
+        cli = _make_cli()
+        # _cprint goes through module-level print, so capture via console.
+        # We can't patch _cprint directly without import juggling; the
+        # previous assertion already proves the unknown-arg branch is
+        # reached (result True + flag False).
+        result = cli.process_command("/exit garbage")
+        assert result is True
+        assert cli._delete_session_on_exit is False
+
+
+class TestCommandRegistry:
+    def test_quit_command_advertises_delete_flag(self):
+        """The CommandDef args_hint should surface `--delete` in /help and
+        CLI autocomplete."""
+        from hermes_cli.commands import resolve_command
+        cmd = resolve_command("quit")
+        assert cmd is not None
+        assert cmd.args_hint == "[--delete]"
+
+    def test_exit_alias_resolves_to_quit_with_hint(self):
+        from hermes_cli.commands import resolve_command
+        cmd = resolve_command("exit")
+        assert cmd is not None
+        assert cmd.name == "quit"
+        assert cmd.args_hint == "[--delete]"

tests/gateway/test_memory_monitor.py

@@ -0,0 +1,122 @@
+"""Tests for gateway.memory_monitor — periodic process memory logging.
+
+Ported from cline/cline#10343.  The module logs a structured
+``[MEMORY] rss=...MB ...`` line periodically so long-running gateway
+leaks show up as a time series in agent.log / gateway.log.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+
+import pytest
+
+from gateway import memory_monitor as mm
+
+
+@pytest.fixture(autouse=True)
+def _ensure_monitor_stopped():
+    """Every test starts from a clean state and leaves one behind."""
+    mm.stop_memory_monitoring(timeout=1.0)
+    yield
+    mm.stop_memory_monitoring(timeout=1.0)
+
+
+def test_log_memory_usage_emits_memory_line(caplog):
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    mm.log_memory_usage()
+    memory_lines = [r for r in caplog.records if "[MEMORY]" in r.getMessage()]
+    assert memory_lines, "expected at least one [MEMORY] log record"
+
+
+def test_log_memory_usage_has_grep_friendly_format(caplog):
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    mm.log_memory_usage()
+    msg = caplog.records[-1].getMessage()
+    # Grep-friendly contract: line starts with [MEMORY] and carries RSS
+    # (or 'unavailable'), GC counts, thread count, uptime.
+    assert msg.startswith("[MEMORY]"), msg
+    assert "rss=" in msg
+    assert "gc=" in msg
+    assert "threads=" in msg
+    assert "uptime=" in msg
+
+
+def test_log_memory_usage_with_prefix(caplog):
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    mm.log_memory_usage(prefix="baseline")
+    msg = caplog.records[-1].getMessage()
+    assert "[MEMORY] baseline " in msg
+
+
+def test_start_logs_baseline_and_returns_true(caplog):
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    # Large interval so the background timer never fires during the test —
+    # we're only checking the synchronous baseline behavior here.
+    started = mm.start_memory_monitoring(interval_seconds=3600.0)
+    assert started is True
+    assert mm.is_running() is True
+
+    messages = [r.getMessage() for r in caplog.records]
+    assert any("[MEMORY] baseline " in m for m in messages), messages
+    assert any("Periodic memory monitoring started" in m for m in messages), messages
+
+
+def test_double_start_is_noop():
+    assert mm.start_memory_monitoring(interval_seconds=3600.0) is True
+    assert mm.start_memory_monitoring(interval_seconds=3600.0) is False
+    assert mm.is_running() is True
+
+
+def test_stop_logs_shutdown_snapshot(caplog):
+    mm.start_memory_monitoring(interval_seconds=3600.0)
+    caplog.clear()
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    mm.stop_memory_monitoring(timeout=1.0)
+    assert mm.is_running() is False
+
+    messages = [r.getMessage() for r in caplog.records]
+    assert any("[MEMORY] shutdown " in m for m in messages), messages
+    assert any("Periodic memory monitoring stopped" in m for m in messages), messages
+
+
+def test_stop_without_start_is_noop():
+    # Must not raise, must not log shutdown snapshot.
+    mm.stop_memory_monitoring(timeout=0.5)
+    assert mm.is_running() is False
+
+
+def test_periodic_timer_fires(caplog):
+    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
+    # Short interval so we can observe multiple ticks inside the test budget.
+    mm.start_memory_monitoring(interval_seconds=0.1)
+    time.sleep(0.45)
+    mm.stop_memory_monitoring(timeout=1.0)
+
+    periodic = [
+        r for r in caplog.records
+        if r.getMessage().startswith("[MEMORY] rss=") or r.getMessage().startswith("[MEMORY] rss=unavailable")
+    ]
+    # baseline + at least 2 periodic + shutdown — but shutdown has the
+    # "shutdown " prefix so it won't match the strict "[MEMORY] rss=" start.
+    # We expect >= 3 bare "[MEMORY] rss=..." lines.
+    assert len(periodic) >= 3, [r.getMessage() for r in caplog.records]
+
+
+def test_thread_is_daemon():
+    mm.start_memory_monitoring(interval_seconds=3600.0)
+    assert mm._monitor_thread is not None
+    assert mm._monitor_thread.daemon is True, (
+        "memory monitor thread must be daemon so it can never block process exit"
+    )
+
+
+def test_unavailable_rss_warns_and_does_not_start(caplog, monkeypatch):
+    # Force both backends to claim unavailable; start should bail.
+    monkeypatch.setattr(mm, "_get_rss_mb", lambda: None)
+    caplog.set_level(logging.WARNING, logger="gateway.memory_monitor")
+    started = mm.start_memory_monitoring(interval_seconds=3600.0)
+    assert started is False
+    assert mm.is_running() is False
+    assert any("Memory monitoring unavailable" in r.getMessage() for r in caplog.records)

tests/hermes_cli/test_send_cmd.py

@@ -0,0 +1,387 @@
+"""Tests for the ``hermes send`` CLI subcommand.
+
+Covers the argument parsing / stdin / file / list behavior of
+``hermes_cli.send_cmd``. The underlying ``send_message_tool`` is stubbed so
+no network I/O or gateway is required.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import send_cmd
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _parse(argv):
+    """Build the top-level parser and return the parsed args for ``argv``."""
+    import argparse
+
+    parser = argparse.ArgumentParser(prog="hermes")
+    subparsers = parser.add_subparsers(dest="command")
+    send_cmd.register_send_subparser(subparsers)
+    return parser.parse_args(["send", *argv])
+
+
+class _FakeTool:
+    """Replacement for ``tools.send_message_tool.send_message_tool``."""
+
+    def __init__(self, payload):
+        self.payload = payload
+        self.calls = []
+
+    def __call__(self, args, **_kw):
+        self.calls.append(dict(args))
+        return json.dumps(self.payload)
+
+
+@pytest.fixture
+def fake_tool(monkeypatch):
+    """Install a fake send_message_tool and return the stub for inspection."""
+    import sys
+    import types
+
+    fake = _FakeTool({"success": True, "message_id": "m123"})
+
+    mod = types.ModuleType("tools.send_message_tool")
+    mod.send_message_tool = fake
+    # Register the stub so ``from tools.send_message_tool import ...`` inside
+    # cmd_send resolves to our fake. Also patch the parent ``tools`` package
+    # entry so attribute lookup works.
+    monkeypatch.setitem(sys.modules, "tools.send_message_tool", mod)
+    return fake
+
+
+# ---------------------------------------------------------------------------
+# Happy path
+# ---------------------------------------------------------------------------
+
+
+def test_positional_message_success(fake_tool, capsys):
+    args = _parse(["--to", "telegram", "hello world"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls == [
+        {"action": "send", "target": "telegram", "message": "hello world"}
+    ]
+    out = capsys.readouterr()
+    assert "sent" in out.out or out.out == ""  # "sent" is the default success banner
+
+
+def test_stdin_message(fake_tool, monkeypatch, capsys):
+    # Piped stdin (not a tty) should be consumed as the message body.
+    monkeypatch.setattr("sys.stdin", io.StringIO("piped body\n"))
+    # Force isatty to return False so the CLI reads from stdin.
+    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
+    args = _parse(["--to", "discord:#ops"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls[0]["message"] == "piped body\n"
+    assert fake_tool.calls[0]["target"] == "discord:#ops"
+
+
+def test_file_message(fake_tool, tmp_path):
+    body = tmp_path / "msg.txt"
+    body.write_text("from a file\n")
+    args = _parse(["--to", "slack:#eng", "--file", str(body)])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls[0]["message"] == "from a file\n"
+
+
+def test_file_dash_means_stdin(fake_tool, monkeypatch):
+    monkeypatch.setattr("sys.stdin", io.StringIO("dash body"))
+    args = _parse(["--to", "telegram", "--file", "-"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls[0]["message"] == "dash body"
+
+
+def test_subject_prepends_header(fake_tool):
+    args = _parse(["--to", "telegram", "--subject", "[CI]", "body text"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls[0]["message"] == "[CI]\n\nbody text"
+
+
+def test_json_mode_emits_payload(fake_tool, capsys):
+    args = _parse(["--to", "telegram", "--json", "hi"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    out = capsys.readouterr().out
+    payload = json.loads(out)
+    assert payload.get("success") is True
+    assert payload.get("message_id") == "m123"
+
+
+def test_quiet_suppresses_stdout(fake_tool, capsys):
+    args = _parse(["--to", "telegram", "--quiet", "shh"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    out = capsys.readouterr()
+    assert out.out == ""
+
+
+# ---------------------------------------------------------------------------
+# Error paths
+# ---------------------------------------------------------------------------
+
+
+def test_missing_target(fake_tool, capsys, monkeypatch):
+    # Ensure stdin is a tty so the CLI does not try to consume it as a body.
+    monkeypatch.setattr("sys.stdin.isatty", lambda: True)
+    args = _parse(["hello"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "--to" in err
+
+
+def test_missing_message(fake_tool, capsys, monkeypatch):
+    monkeypatch.setattr("sys.stdin.isatty", lambda: True)
+    args = _parse(["--to", "telegram"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "no message" in err.lower()
+
+
+def test_file_not_found_is_usage_error(fake_tool, capsys, monkeypatch):
+    monkeypatch.setattr("sys.stdin.isatty", lambda: True)
+    args = _parse(["--to", "telegram", "--file", "/nonexistent/does-not-exist.txt"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 2
+    err = capsys.readouterr().err
+    assert "cannot read" in err.lower()
+
+
+def test_tool_error_returns_failure_exit(monkeypatch, capsys):
+    import sys as _sys
+    import types as _types
+
+    fake_mod = _types.ModuleType("tools.send_message_tool")
+
+    def _bad_tool(args, **_kw):
+        return json.dumps({"error": "platform blew up"})
+
+    fake_mod.send_message_tool = _bad_tool
+    monkeypatch.setitem(_sys.modules, "tools.send_message_tool", fake_mod)
+
+    args = _parse(["--to", "telegram", "nope"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 1
+    err = capsys.readouterr().err
+    assert "platform blew up" in err
+
+
+def test_skipped_result_is_success(monkeypatch):
+    import sys as _sys
+    import types as _types
+
+    fake_mod = _types.ModuleType("tools.send_message_tool")
+    fake_mod.send_message_tool = lambda args, **_kw: json.dumps(
+        {"success": True, "skipped": True, "reason": "duplicate"}
+    )
+    monkeypatch.setitem(_sys.modules, "tools.send_message_tool", fake_mod)
+
+    args = _parse(["--to", "telegram", "dup"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+
+
+# ---------------------------------------------------------------------------
+# --list
+# ---------------------------------------------------------------------------
+
+
+def test_list_human_output(monkeypatch, capsys):
+    import sys as _sys
+    import types as _types
+
+    fake_dir = _types.ModuleType("gateway.channel_directory")
+    fake_dir.format_directory_for_display = lambda: "Available messaging targets:\n\nTelegram:\n  telegram:-100123\n"
+    fake_dir.load_directory = lambda: {
+        "platforms": {"telegram": [{"id": "-100123", "name": "Test Group"}]}
+    }
+    monkeypatch.setitem(_sys.modules, "gateway.channel_directory", fake_dir)
+
+    args = _parse(["--list"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    out = capsys.readouterr().out
+    assert "Telegram" in out
+
+
+def test_list_json(monkeypatch, capsys):
+    import sys as _sys
+    import types as _types
+
+    fake_dir = _types.ModuleType("gateway.channel_directory")
+    fake_dir.format_directory_for_display = lambda: "(ignored in json mode)"
+    fake_dir.load_directory = lambda: {
+        "platforms": {"telegram": [{"id": "-100123", "name": "Test Group"}]}
+    }
+    monkeypatch.setitem(_sys.modules, "gateway.channel_directory", fake_dir)
+
+    args = _parse(["--list", "--json"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    out = capsys.readouterr().out
+    payload = json.loads(out)
+    assert payload["platforms"]["telegram"][0]["name"] == "Test Group"
+
+
+def test_list_filter_platform(monkeypatch, capsys):
+    import sys as _sys
+    import types as _types
+
+    fake_dir = _types.ModuleType("gateway.channel_directory")
+    fake_dir.format_directory_for_display = lambda: "(should not be called when filter set)"
+    fake_dir.load_directory = lambda: {
+        "platforms": {
+            "telegram": [{"id": "-100123", "name": "TG Chat"}],
+            "discord": [{"id": "555", "name": "bot-home"}],
+        }
+    }
+    monkeypatch.setitem(_sys.modules, "gateway.channel_directory", fake_dir)
+
+    # When --list is set, argparse puts the optional bareword in the
+    # `message` positional slot (where the send-mode body would go).
+    args = _parse(["--list", "telegram"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    out = capsys.readouterr().out
+    assert "telegram" in out.lower()
+    assert "discord" not in out.lower()
+
+
+def test_list_unknown_platform_fails(monkeypatch, capsys):
+    import sys as _sys
+    import types as _types
+
+    fake_dir = _types.ModuleType("gateway.channel_directory")
+    fake_dir.format_directory_for_display = lambda: ""
+    fake_dir.load_directory = lambda: {"platforms": {"telegram": []}}
+    monkeypatch.setitem(_sys.modules, "gateway.channel_directory", fake_dir)
+
+    args = _parse(["--list", "pigeon-post"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 1
+    err = capsys.readouterr().err
+    assert "pigeon-post" in err
+
+
+# ---------------------------------------------------------------------------
+# Parser registration contract
+# ---------------------------------------------------------------------------
+
+
+def test_register_send_subparser_is_reusable():
+    """Sanity check: the registrar returns a parser and wires ``cmd_send``."""
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest="command")
+    send_parser = send_cmd.register_send_subparser(subparsers)
+    assert send_parser is not None
+    args = parser.parse_args(["send", "--to", "telegram", "hi"])
+    assert args.func is send_cmd.cmd_send
+    assert args.to == "telegram"
+    assert args.message == "hi"
+
+
+# ---------------------------------------------------------------------------
+# Env loader
+# ---------------------------------------------------------------------------
+
+
+def test_load_hermes_env_bridges_config_yaml_scalars(tmp_path, monkeypatch):
+    """Top-level config.yaml scalars should be bridged into os.environ.
+
+    This mirrors the gateway/run.py bootstrap behavior: without this, running
+    ``hermes send`` from a fresh shell cannot resolve the home channel
+    because ``TELEGRAM_HOME_CHANNEL`` (saved by ``hermes config set``) lives
+    in config.yaml, not in .env — and the gateway's config loader reads via
+    ``os.getenv(...)``.
+    """
+    import os
+
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    (hermes_home / ".env").write_text("SOME_TOKEN=abc123\n")
+    (hermes_home / "config.yaml").write_text(
+        "TELEGRAM_HOME_CHANNEL: '5550001111'\nnested:\n  ignored: true\n"
+    )
+
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.delenv("TELEGRAM_HOME_CHANNEL", raising=False)
+    monkeypatch.delenv("SOME_TOKEN", raising=False)
+
+    # Force get_hermes_home() to re-resolve under the patched env.
+    from importlib import reload
+
+    import hermes_cli.config as _hc_config
+    reload(_hc_config)
+
+    send_cmd._load_hermes_env()
+
+    assert os.environ.get("SOME_TOKEN") == "abc123"
+    assert os.environ.get("TELEGRAM_HOME_CHANNEL") == "5550001111"
+
+
+def test_load_hermes_env_does_not_override_existing(tmp_path, monkeypatch):
+    """Existing env vars must not be clobbered by config.yaml values."""
+    import os
+
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    (hermes_home / "config.yaml").write_text("TELEGRAM_HOME_CHANNEL: yaml_value\n")
+
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "env_value")
+
+    from importlib import reload
+    import hermes_cli.config as _hc_config
+    reload(_hc_config)
+
+    send_cmd._load_hermes_env()
+
+    assert os.environ.get("TELEGRAM_HOME_CHANNEL") == "env_value"
+
+
+def test_load_hermes_env_handles_missing_files(tmp_path, monkeypatch):
+    """No .env or config.yaml should be a silent no-op, not an exception."""
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    from importlib import reload
+    import hermes_cli.config as _hc_config
+    reload(_hc_config)
+
+    # Should not raise.
+    send_cmd._load_hermes_env()

tests/hermes_cli/test_session_recap.py

@@ -0,0 +1,180 @@
+"""Unit tests for hermes_cli.session_recap."""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from hermes_cli.session_recap import build_recap
+
+
+def _user(text):
+    return {"role": "user", "content": text}
+
+
+def _assistant(text=None, tool_calls=None):
+    msg = {"role": "assistant", "content": text}
+    if tool_calls:
+        msg["tool_calls"] = tool_calls
+    return msg
+
+
+def _tool_call(name, args):
+    return {
+        "id": f"call_{name}",
+        "type": "function",
+        "function": {"name": name, "arguments": json.dumps(args)},
+    }
+
+
+def _tool_result(content="ok"):
+    return {"role": "tool", "content": content}
+
+
+def test_empty_history():
+    out = build_recap([])
+    assert "Session recap" in out
+    assert "nothing to recap" in out
+
+
+def test_header_shows_title_when_provided():
+    out = build_recap([_user("hello")], session_title="Refactor the adapter")
+    assert "Refactor the adapter" in out.splitlines()[0]
+
+
+def test_header_shows_short_id_when_no_title():
+    out = build_recap([_user("hello")], session_id="abcdef1234567890")
+    assert "abcdef12" in out.splitlines()[0]
+
+
+def test_counts_recent_turns():
+    msgs = [
+        _user("one"),
+        _assistant("first reply"),
+        _user("two"),
+        _assistant("second reply"),
+    ]
+    out = build_recap(msgs)
+    assert "2 user turn" in out
+    assert "assistant repl" in out
+
+
+def test_last_ask_and_reply_are_surfaced():
+    msgs = [
+        _user("old question"),
+        _assistant("old answer"),
+        _user("summarise the docs"),
+        _assistant("here is the summary of the docs you asked for"),
+    ]
+    out = build_recap(msgs)
+    assert "summarise the docs" in out
+    assert "summary of the docs" in out
+
+
+def test_tool_counts_and_files():
+    msgs = [
+        _user("edit the readme and run tests"),
+        _assistant(
+            tool_calls=[
+                _tool_call("read_file", {"path": "README.md"}),
+                _tool_call("patch", {"path": "README.md"}),
+            ]
+        ),
+        _tool_result(),
+        _tool_result(),
+        _assistant(
+            tool_calls=[
+                _tool_call("terminal", {"command": "pytest"}),
+            ]
+        ),
+        _tool_result("tests ok"),
+        _assistant("All green."),
+    ]
+    out = build_recap(msgs)
+    assert "patch×1" in out
+    assert "terminal×1" in out
+    assert "read_file×1" in out
+    # README.md should appear (may include cwd-relative prefix stripping).
+    assert "README.md" in out
+
+
+def test_tool_preview_length_truncates_long_user_prompt():
+    long = "x " * 500
+    out = build_recap([_user(long)])
+    ask_line = [l for l in out.splitlines() if "Last ask" in l][0]
+    assert len(ask_line) < 300  # truncated with ellipsis
+    assert "…" in ask_line
+
+
+def test_respects_recent_window():
+    # 30 turns of user+assistant; only the most recent 20 should be summarised.
+    msgs = []
+    for i in range(30):
+        msgs.append(_user(f"question {i}"))
+        msgs.append(_assistant(f"answer {i}"))
+    out = build_recap(msgs)
+    # We scoped to the 20-turn window but show "of 30/30 total".
+    assert "of 30/30 total" in out
+
+
+def test_multimodal_content_blocks_flattened():
+    msgs = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "check this file"},
+                {"type": "image_url", "image_url": {"url": "..."}},
+            ],
+        },
+        _assistant("Looked at your image."),
+    ]
+    out = build_recap(msgs)
+    assert "check this file" in out
+    assert "Looked at your image" in out
+
+
+def test_handles_arguments_as_dict_not_string():
+    # Some providers return arguments already as a dict.
+    msgs = [
+        _user("go"),
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": "patch",
+                        "arguments": {"path": "foo.py"},
+                    },
+                }
+            ],
+        },
+    ]
+    out = build_recap(msgs)
+    assert "patch×1" in out
+    assert "foo.py" in out
+
+
+def test_no_assistant_activity_hint():
+    out = build_recap([_user("just sent my first message")])
+    assert "no assistant activity" in out or "Last ask" in out
+
+
+def test_tool_message_count_reported():
+    msgs = [
+        _user("go"),
+        _assistant(tool_calls=[_tool_call("read_file", {"path": "a"})]),
+        _tool_result(),
+        _tool_result(),
+        _assistant("done"),
+    ]
+    out = build_recap(msgs)
+    assert "2 tool result" in out
+
+
+def test_ignores_non_mapping_entries_gracefully():
+    msgs = [None, "stray", _user("hi"), _assistant("hello")]
+    # Should not raise.
+    out = build_recap(msgs)
+    assert "Session recap" in out

tests/test_package_json_lazy_deps.py

@@ -0,0 +1,85 @@
+"""Invariants for what is eager vs lazy in the root ``package.json``.
+
+The root ``package.json`` is installed by ``hermes update`` on every user,
+including users who never opted into a given browser backend. Anything
+listed in ``dependencies`` therefore runs its npm postinstall script for
+everyone — including binary-fetching backends, on every update.
+
+The contract:
+
+* ``agent-browser`` IS eager. It is the default Chromium-driving backend
+  used whenever the agent makes a browser call without a cloud provider
+  configured, so it must already be installed before any session starts.
+  Its postinstall is also small.
+
+* ``@askjo/camofox-browser`` is NOT eager. It is an explicit opt-in
+  alternative browser backend, selected by the user via
+  ``hermes tools`` → Browser Automation → Camofox, and only used at
+  runtime when ``CAMOFOX_URL`` is set. Its postinstall fetches a ~300MB
+  Firefox-fork binary, which silently blocked ``hermes update`` for
+  multi-minute stretches on slow / network-restricted connections
+  (notably users in China running through a VPN). The package is
+  installed on demand by ``tools_config.py`` ``post_setup_key ==
+  "camofox"`` when the user actually selects Camofox.
+
+If a future PR re-adds Camofox (or any other binary-postinstall package)
+to root ``dependencies``, this test fails — read the lazy-install
+guidance in the ``hermes-agent-dev`` skill before changing the
+expectations.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+
+def _root_package_json() -> dict:
+    with (REPO_ROOT / "package.json").open("r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def test_camofox_is_not_in_root_dependencies() -> None:
+    """Camofox must be opt-in, installed lazily by its post_setup handler."""
+    deps = _root_package_json().get("dependencies", {})
+    assert "@askjo/camofox-browser" not in deps, (
+        "Camofox is a ~300MB binary-postinstall backend that must stay "
+        "out of root package.json dependencies. It belongs in the "
+        "Camofox post_setup handler in hermes_cli/tools_config.py so it "
+        "only installs when the user explicitly selects Camofox via "
+        "`hermes tools` → Browser Automation → Camofox."
+    )
+
+
+def test_agent_browser_stays_eager() -> None:
+    """agent-browser is the default backend; it must remain eager."""
+    deps = _root_package_json().get("dependencies", {})
+    assert "agent-browser" in deps, (
+        "agent-browser is the default browser-tool backend used by every "
+        "session that doesn't have a cloud browser provider configured. "
+        "It must stay in root package.json dependencies so it is present "
+        "after `hermes setup` / `hermes update` without an explicit "
+        "post_setup step."
+    )
+
+
+def test_root_lockfile_has_no_camofox_entries() -> None:
+    """Regenerated lockfiles should not contain Camofox tree entries."""
+    lock_path = REPO_ROOT / "package-lock.json"
+    if not lock_path.exists():
+        # Some CI matrix shards skip lockfile materialization.
+        return
+    text = lock_path.read_text(encoding="utf-8")
+    assert "@askjo/camofox-browser" not in text, (
+        "package-lock.json still references @askjo/camofox-browser. "
+        "Regenerate the lockfile after removing the dep: "
+        "`rm package-lock.json && npm install --package-lock-only "
+        "--ignore-scripts --no-fund --no-audit`."
+    )
+    assert "camoufox-js" not in text, (
+        "package-lock.json still references camoufox-js (transitive of "
+        "@askjo/camofox-browser). Regenerate the lockfile."
+    )

tests/tools/test_mcp_invalid_url.py

@@ -0,0 +1,125 @@
+"""Tests for the MCP remote-URL validator.
+
+Ported from anomalyco/opencode#25019 (``fix: handle invalid mcp urls``).
+
+Previously, a typo in ``config.yaml`` (missing scheme, wrong scheme, empty
+string, dict where a URL was expected) caused the MCP server startup code
+to enter httpx's URL-parsing path and crash inside the transport layer.
+The reconnect-backoff loop would then retry
+``_MAX_INITIAL_CONNECT_RETRIES`` times with doubling backoff — a minute or
+more of pointless retries plus a confusing opaque error message — before
+eventually giving up.
+
+The fix validates the URL once, up front, and fails fast with a specific
+error message identifying the offending server.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from tools.mcp_tool import (
+    InvalidMcpUrlError,
+    _validate_remote_mcp_url,
+)
+
+
+class TestValidUrlsAccepted:
+    """Every valid http(s) URL must pass through untouched (stripped of whitespace)."""
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://localhost:3000/mcp",
+            "https://example.com/mcp",
+            "https://context7.liam.com/mcp",
+            "http://127.0.0.1:8080",
+            "https://api.example.com:443/v1/mcp?session=abc",
+            "http://[::1]:9000/mcp",  # IPv6
+            "https://host.example.com",  # no port, no path
+        ],
+    )
+    def test_accepts_valid_http_url(self, url):
+        assert _validate_remote_mcp_url("test", url) == url
+
+    def test_strips_surrounding_whitespace(self):
+        assert (
+            _validate_remote_mcp_url("test", "  https://example.com/mcp  ")
+            == "https://example.com/mcp"
+        )
+
+
+class TestInvalidUrlsRejected:
+    """Every broken shape must raise ``InvalidMcpUrlError`` with a clear message."""
+
+    def test_none_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="context7.*expected a string"):
+            _validate_remote_mcp_url("context7", None)
+
+    def test_dict_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="expected a string, got dict"):
+            _validate_remote_mcp_url("ctx", {"url": "nested"})
+
+    def test_int_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="expected a string, got int"):
+            _validate_remote_mcp_url("ctx", 8080)
+
+    def test_empty_string_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="empty url"):
+            _validate_remote_mcp_url("ctx", "")
+
+    def test_whitespace_only_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="empty url"):
+            _validate_remote_mcp_url("ctx", "   \t\n")
+
+    def test_missing_scheme_rejected(self):
+        # The most common typo — users copy a host from a web page.
+        with pytest.raises(
+            InvalidMcpUrlError, match="scheme must be http or https"
+        ):
+            _validate_remote_mcp_url("ctx", "example.com/mcp")
+
+    def test_file_scheme_rejected(self):
+        with pytest.raises(
+            InvalidMcpUrlError, match="scheme must be http or https"
+        ):
+            _validate_remote_mcp_url("ctx", "file:///etc/passwd")
+
+    def test_ws_scheme_rejected(self):
+        # WebSocket is not MCP's remote transport.
+        with pytest.raises(
+            InvalidMcpUrlError, match="scheme must be http or https"
+        ):
+            _validate_remote_mcp_url("ctx", "ws://example.com/mcp")
+
+    def test_stdio_scheme_rejected(self):
+        # stdio servers use the ``command`` key, not ``url``.
+        with pytest.raises(
+            InvalidMcpUrlError, match="scheme must be http or https"
+        ):
+            _validate_remote_mcp_url("ctx", "stdio:///node server.js")
+
+    def test_empty_host_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="missing host"):
+            _validate_remote_mcp_url("ctx", "http:///")
+
+    def test_empty_host_with_path_rejected(self):
+        with pytest.raises(InvalidMcpUrlError, match="missing host"):
+            _validate_remote_mcp_url("ctx", "https:///path/only")
+
+    def test_error_mentions_server_name(self):
+        # So users can find the bad entry when there are multiple configured.
+        with pytest.raises(InvalidMcpUrlError, match="my-weird-server"):
+            _validate_remote_mcp_url("my-weird-server", "not a url at all")
+
+
+class TestErrorIsValueError:
+    """InvalidMcpUrlError must be a ValueError for broad downstream catch blocks."""
+
+    def test_is_value_error(self):
+        try:
+            _validate_remote_mcp_url("ctx", "garbage")
+        except ValueError:
+            pass  # expected
+        else:
+            pytest.fail("expected ValueError")