Merge branch 'main' into rewbs/tool-use-charge-to-subscription

tests/agent/test_auxiliary_client.py

@@ -461,7 +461,7 @@ class TestGetTextAuxiliaryClient:
              patch("agent.auxiliary_client.OpenAI") as mock_openai:
             mock_nous.return_value = {"access_token": "nous-tok"}
             client, model = get_text_auxiliary_client()
-        assert model == "gemini-3-flash"
+        assert model == "google/gemini-3-flash-preview"
 
     def test_custom_endpoint_over_codex(self, monkeypatch, codex_auth_dir):
         monkeypatch.setenv("OPENAI_BASE_URL", "http://localhost:1234/v1")
@@ -693,7 +693,7 @@ class TestVisionClientFallback:
              patch("agent.auxiliary_client.OpenAI"):
             mock_nous.return_value = {"access_token": "nous-tok"}
             client, model = get_vision_auxiliary_client()
-        assert model == "gemini-3-flash"
+        assert model == "google/gemini-3-flash-preview"
         assert client is not None
 
     def test_vision_forced_main_uses_custom_endpoint(self, monkeypatch):
@@ -789,7 +789,7 @@ class TestResolveForcedProvider:
              patch("agent.auxiliary_client.OpenAI"):
             mock_nous.return_value = {"access_token": "nous-tok"}
             client, model = _resolve_forced_provider("nous")
-        assert model == "gemini-3-flash"
+        assert model == "google/gemini-3-flash-preview"
         assert client is not None
 
     def test_forced_nous_not_configured(self, monkeypatch):

tests/agent/test_prompt_builder.py

@@ -460,7 +460,7 @@ class TestBuildContextFilesPrompt:
         with patch("pathlib.Path.home", return_value=fake_home):
             result = build_context_files_prompt(cwd=str(tmp_path))
         assert "Project Context" in result
-        assert "# Hermes ☤" in result
+        assert "Hermes Agent" in result
 
     def test_loads_agents_md(self, tmp_path):
         (tmp_path / "AGENTS.md").write_text("Use Ruff for linting.")

tests/gateway/test_agent_cache.py

@@ -48,6 +48,28 @@ class TestAgentConfigSignature:
         sig2 = GatewayRunner._agent_config_signature("claude-opus-4.6", runtime, ["hermes-telegram"], "")
         assert sig1 != sig2
 
+    def test_same_token_prefix_different_full_token_changes_signature(self):
+        """Tokens sharing a JWT-style prefix must not collide."""
+        from gateway.run import GatewayRunner
+
+        rt1 = {
+            "api_key": "eyJhbGci.token-for-account-a",
+            "base_url": "https://chatgpt.com/backend-api/codex",
+            "provider": "openai-codex",
+            "api_mode": "codex_responses",
+        }
+        rt2 = {
+            "api_key": "eyJhbGci.token-for-account-b",
+            "base_url": "https://chatgpt.com/backend-api/codex",
+            "provider": "openai-codex",
+            "api_mode": "codex_responses",
+        }
+
+        assert rt1["api_key"][:8] == rt2["api_key"][:8]
+        sig1 = GatewayRunner._agent_config_signature("gpt-5.3-codex", rt1, ["hermes-telegram"], "")
+        sig2 = GatewayRunner._agent_config_signature("gpt-5.3-codex", rt2, ["hermes-telegram"], "")
+        assert sig1 != sig2
+
     def test_provider_change_different_signature(self):
         from gateway.run import GatewayRunner
 

tests/gateway/test_background_command.py

@@ -38,6 +38,7 @@ def _make_runner():
     runner._provider_routing = {}
     runner._fallback_model = None
     runner._running_agents = {}
+    runner._background_tasks = set()
 
     mock_store = MagicMock()
     runner.session_store = mock_store

tests/gateway/test_dm_topics.py

@@ -0,0 +1,484 @@
+"""Tests for Telegram DM Private Chat Topics (Bot API 9.4).
+
+Covers:
+- _setup_dm_topics: loading persisted thread_ids from config
+- _setup_dm_topics: creating new topics via API when no thread_id
+- _persist_dm_topic_thread_id: saving thread_id back to config.yaml
+- _get_dm_topic_info: looking up topic config by thread_id
+- _cache_dm_topic_from_message: caching thread_ids from incoming messages
+- _build_message_event: DM topic resolution in message events
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch, mock_open
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_telegram_mock():
+    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
+        return
+
+    telegram_mod = MagicMock()
+    telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
+    telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
+    telegram_mod.constants.ChatType.GROUP = "group"
+    telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
+    telegram_mod.constants.ChatType.CHANNEL = "channel"
+    telegram_mod.constants.ChatType.PRIVATE = "private"
+
+    for name in ("telegram", "telegram.ext", "telegram.constants"):
+        sys.modules.setdefault(name, telegram_mod)
+
+
+_ensure_telegram_mock()
+
+from gateway.platforms.telegram import TelegramAdapter  # noqa: E402
+
+
+def _make_adapter(dm_topics_config=None):
+    """Create a TelegramAdapter with optional DM topics config."""
+    extra = {}
+    if dm_topics_config is not None:
+        extra["dm_topics"] = dm_topics_config
+    config = PlatformConfig(enabled=True, token="***", extra=extra)
+    adapter = TelegramAdapter(config)
+    return adapter
+
+
+# ── _setup_dm_topics: load persisted thread_ids ──
+
+
+@pytest.mark.asyncio
+async def test_setup_dm_topics_loads_persisted_thread_ids():
+    """Topics with thread_id in config should be loaded into cache, not created."""
+    adapter = _make_adapter([
+        {
+            "chat_id": 111,
+            "topics": [
+                {"name": "General", "thread_id": 100},
+                {"name": "Work", "thread_id": 200},
+            ],
+        }
+    ])
+    adapter._bot = AsyncMock()
+
+    await adapter._setup_dm_topics()
+
+    # Both should be in cache
+    assert adapter._dm_topics["111:General"] == 100
+    assert adapter._dm_topics["111:Work"] == 200
+    # create_forum_topic should NOT have been called
+    adapter._bot.create_forum_topic.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_setup_dm_topics_creates_when_no_thread_id():
+    """Topics without thread_id should be created via API."""
+    adapter = _make_adapter([
+        {
+            "chat_id": 222,
+            "topics": [
+                {"name": "NewTopic", "icon_color": 7322096},
+            ],
+        }
+    ])
+    adapter._bot = AsyncMock()
+    mock_topic = SimpleNamespace(message_thread_id=999)
+    adapter._bot.create_forum_topic.return_value = mock_topic
+
+    # Mock the persist method so it doesn't touch the filesystem
+    adapter._persist_dm_topic_thread_id = MagicMock()
+
+    await adapter._setup_dm_topics()
+
+    # Should have been created
+    adapter._bot.create_forum_topic.assert_called_once_with(
+        chat_id=222, name="NewTopic", icon_color=7322096,
+    )
+    # Should be in cache
+    assert adapter._dm_topics["222:NewTopic"] == 999
+    # Should persist
+    adapter._persist_dm_topic_thread_id.assert_called_once_with(222, "NewTopic", 999)
+
+
+@pytest.mark.asyncio
+async def test_setup_dm_topics_mixed_persisted_and_new():
+    """Mix of persisted and new topics should work correctly."""
+    adapter = _make_adapter([
+        {
+            "chat_id": 333,
+            "topics": [
+                {"name": "Existing", "thread_id": 50},
+                {"name": "New", "icon_color": 123},
+            ],
+        }
+    ])
+    adapter._bot = AsyncMock()
+    mock_topic = SimpleNamespace(message_thread_id=777)
+    adapter._bot.create_forum_topic.return_value = mock_topic
+    adapter._persist_dm_topic_thread_id = MagicMock()
+
+    await adapter._setup_dm_topics()
+
+    # Existing loaded from config
+    assert adapter._dm_topics["333:Existing"] == 50
+    # New created via API
+    assert adapter._dm_topics["333:New"] == 777
+    # Only one API call (for "New")
+    adapter._bot.create_forum_topic.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_setup_dm_topics_skips_empty_config():
+    """Empty dm_topics config should be a no-op."""
+    adapter = _make_adapter([])
+    adapter._bot = AsyncMock()
+
+    await adapter._setup_dm_topics()
+
+    adapter._bot.create_forum_topic.assert_not_called()
+    assert adapter._dm_topics == {}
+
+
+@pytest.mark.asyncio
+async def test_setup_dm_topics_no_config():
+    """No dm_topics in config at all should be a no-op."""
+    adapter = _make_adapter()
+    adapter._bot = AsyncMock()
+
+    await adapter._setup_dm_topics()
+
+    adapter._bot.create_forum_topic.assert_not_called()
+
+
+# ── _create_dm_topic: error handling ──
+
+
+@pytest.mark.asyncio
+async def test_create_dm_topic_handles_duplicate_error():
+    """Duplicate topic error should return None gracefully."""
+    adapter = _make_adapter()
+    adapter._bot = AsyncMock()
+    adapter._bot.create_forum_topic.side_effect = Exception("topic_name_duplicate")
+
+    result = await adapter._create_dm_topic(chat_id=111, name="General")
+
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_create_dm_topic_handles_generic_error():
+    """Generic error should return None with warning."""
+    adapter = _make_adapter()
+    adapter._bot = AsyncMock()
+    adapter._bot.create_forum_topic.side_effect = Exception("some random error")
+
+    result = await adapter._create_dm_topic(chat_id=111, name="General")
+
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_create_dm_topic_returns_none_without_bot():
+    """No bot instance should return None."""
+    adapter = _make_adapter()
+    adapter._bot = None
+
+    result = await adapter._create_dm_topic(chat_id=111, name="General")
+
+    assert result is None
+
+
+# ── _persist_dm_topic_thread_id ──
+
+
+def test_persist_dm_topic_thread_id_writes_config(tmp_path):
+    """Should write thread_id into the correct topic in config.yaml."""
+    import yaml
+
+    config_data = {
+        "platforms": {
+            "telegram": {
+                "extra": {
+                    "dm_topics": [
+                        {
+                            "chat_id": 111,
+                            "topics": [
+                                {"name": "General", "icon_color": 123},
+                                {"name": "Work", "icon_color": 456},
+                            ],
+                        }
+                    ]
+                }
+            }
+        }
+    }
+
+    config_file = tmp_path / ".hermes" / "config.yaml"
+    config_file.parent.mkdir(parents=True)
+    with open(config_file, "w") as f:
+        yaml.dump(config_data, f)
+
+    adapter = _make_adapter()
+
+    with patch.object(Path, "home", return_value=tmp_path):
+        adapter._persist_dm_topic_thread_id(111, "General", 999)
+
+    with open(config_file) as f:
+        result = yaml.safe_load(f)
+
+    topics = result["platforms"]["telegram"]["extra"]["dm_topics"][0]["topics"]
+    assert topics[0]["thread_id"] == 999
+    assert "thread_id" not in topics[1]  # "Work" should be untouched
+
+
+def test_persist_dm_topic_thread_id_skips_if_already_set(tmp_path):
+    """Should not overwrite an existing thread_id."""
+    import yaml
+
+    config_data = {
+        "platforms": {
+            "telegram": {
+                "extra": {
+                    "dm_topics": [
+                        {
+                            "chat_id": 111,
+                            "topics": [
+                                {"name": "General", "icon_color": 123, "thread_id": 500},
+                            ],
+                        }
+                    ]
+                }
+            }
+        }
+    }
+
+    config_file = tmp_path / ".hermes" / "config.yaml"
+    config_file.parent.mkdir(parents=True)
+    with open(config_file, "w") as f:
+        yaml.dump(config_data, f)
+
+    adapter = _make_adapter()
+
+    with patch.object(Path, "home", return_value=tmp_path):
+        adapter._persist_dm_topic_thread_id(111, "General", 999)
+
+    with open(config_file) as f:
+        result = yaml.safe_load(f)
+
+    topics = result["platforms"]["telegram"]["extra"]["dm_topics"][0]["topics"]
+    assert topics[0]["thread_id"] == 500  # unchanged
+
+
+# ── _get_dm_topic_info ──
+
+
+def test_get_dm_topic_info_finds_cached_topic():
+    """Should return topic config when thread_id is in cache."""
+    adapter = _make_adapter([
+        {
+            "chat_id": 111,
+            "topics": [
+                {"name": "General", "skill": "my-skill"},
+            ],
+        }
+    ])
+    adapter._dm_topics["111:General"] = 100
+
+    result = adapter._get_dm_topic_info("111", "100")
+
+    assert result is not None
+    assert result["name"] == "General"
+    assert result["skill"] == "my-skill"
+
+
+def test_get_dm_topic_info_returns_none_for_unknown():
+    """Should return None for unknown thread_id."""
+    adapter = _make_adapter([
+        {
+            "chat_id": 111,
+            "topics": [{"name": "General"}],
+        }
+    ])
+    # Mock reload to avoid filesystem access
+    adapter._reload_dm_topics_from_config = lambda: None
+
+    result = adapter._get_dm_topic_info("111", "999")
+
+    assert result is None
+
+
+def test_get_dm_topic_info_returns_none_without_config():
+    """Should return None if no dm_topics config."""
+    adapter = _make_adapter()
+    adapter._reload_dm_topics_from_config = lambda: None
+
+    result = adapter._get_dm_topic_info("111", "100")
+
+    assert result is None
+
+
+def test_get_dm_topic_info_returns_none_for_none_thread():
+    """Should return None if thread_id is None."""
+    adapter = _make_adapter([
+        {"chat_id": 111, "topics": [{"name": "General"}]}
+    ])
+
+    result = adapter._get_dm_topic_info("111", None)
+
+    assert result is None
+
+
+def test_get_dm_topic_info_hot_reloads_from_config(tmp_path):
+    """Should find a topic added to config after startup (hot-reload)."""
+    import yaml
+
+    # Start with empty topics
+    adapter = _make_adapter([
+        {"chat_id": 111, "topics": []}
+    ])
+
+    # Write config with a new topic + thread_id
+    config_data = {
+        "platforms": {
+            "telegram": {
+                "extra": {
+                    "dm_topics": [
+                        {
+                            "chat_id": 111,
+                            "topics": [
+                                {"name": "NewProject", "thread_id": 555},
+                            ],
+                        }
+                    ]
+                }
+            }
+        }
+    }
+    config_file = tmp_path / ".hermes" / "config.yaml"
+    config_file.parent.mkdir(parents=True)
+    with open(config_file, "w") as f:
+        yaml.dump(config_data, f)
+
+    with patch.object(Path, "home", return_value=tmp_path):
+        result = adapter._get_dm_topic_info("111", "555")
+
+    assert result is not None
+    assert result["name"] == "NewProject"
+    # Should now be cached
+    assert adapter._dm_topics["111:NewProject"] == 555
+
+
+# ── _cache_dm_topic_from_message ──
+
+
+def test_cache_dm_topic_from_message():
+    """Should cache a new topic mapping."""
+    adapter = _make_adapter()
+
+    adapter._cache_dm_topic_from_message("111", "100", "General")
+
+    assert adapter._dm_topics["111:General"] == 100
+
+
+def test_cache_dm_topic_from_message_no_overwrite():
+    """Should not overwrite an existing cached topic."""
+    adapter = _make_adapter()
+    adapter._dm_topics["111:General"] = 100
+
+    adapter._cache_dm_topic_from_message("111", "999", "General")
+
+    assert adapter._dm_topics["111:General"] == 100  # unchanged
+
+
+# ── _build_message_event: auto_skill binding ──
+
+
+def _make_mock_message(chat_id=111, chat_type="private", text="hello", thread_id=None,
+                       user_id=42, user_name="Test User", forum_topic_created=None):
+    """Create a mock Telegram Message for _build_message_event tests."""
+    chat = SimpleNamespace(
+        id=chat_id,
+        type=chat_type,
+        title=None,
+    )
+    # Add full_name attribute for DM chats
+    if not hasattr(chat, "full_name"):
+        chat.full_name = user_name
+
+    user = SimpleNamespace(
+        id=user_id,
+        full_name=user_name,
+    )
+
+    msg = SimpleNamespace(
+        chat=chat,
+        from_user=user,
+        text=text,
+        message_thread_id=thread_id,
+        message_id=1001,
+        reply_to_message=None,
+        date=None,
+        forum_topic_created=forum_topic_created,
+    )
+    return msg
+
+
+def test_build_message_event_sets_auto_skill():
+    """When topic has a skill binding, auto_skill should be set on the event."""
+    from gateway.platforms.base import MessageType
+
+    adapter = _make_adapter([
+        {
+            "chat_id": 111,
+            "topics": [
+                {"name": "My Project", "skill": "accessibility-auditor", "thread_id": 100},
+            ],
+        }
+    ])
+    adapter._dm_topics["111:My Project"] = 100
+
+    msg = _make_mock_message(chat_id=111, thread_id=100, text="check this page")
+    event = adapter._build_message_event(msg, MessageType.TEXT)
+
+    assert event.auto_skill == "accessibility-auditor"
+    # chat_topic should be the clean topic name, no [skill: ...] suffix
+    assert event.source.chat_topic == "My Project"
+
+
+def test_build_message_event_no_auto_skill_without_binding():
+    """Topics without skill binding should have auto_skill=None."""
+    from gateway.platforms.base import MessageType
+
+    adapter = _make_adapter([
+        {
+            "chat_id": 111,
+            "topics": [
+                {"name": "General", "thread_id": 200},
+            ],
+        }
+    ])
+    adapter._dm_topics["111:General"] = 200
+
+    msg = _make_mock_message(chat_id=111, thread_id=200)
+    event = adapter._build_message_event(msg, MessageType.TEXT)
+
+    assert event.auto_skill is None
+    assert event.source.chat_topic == "General"
+
+
+def test_build_message_event_no_auto_skill_without_thread():
+    """Regular DM messages (no thread_id) should have auto_skill=None."""
+    from gateway.platforms.base import MessageType
+
+    adapter = _make_adapter()
+    msg = _make_mock_message(chat_id=111, thread_id=None)
+    event = adapter._build_message_event(msg, MessageType.TEXT)
+
+    assert event.auto_skill is None

tests/gateway/test_gateway_shutdown.py

@@ -72,6 +72,7 @@ async def test_gateway_stop_interrupts_running_agents_and_cancels_adapter_tasks(
     runner._exit_reason = None
     runner._pending_messages = {"session": "pending text"}
     runner._pending_approvals = {"session": {"command": "rm -rf /tmp/x"}}
+    runner._background_tasks = set()
     runner._shutdown_all_gateway_honcho = lambda: None
 
     adapter = StubAdapter()

tests/gateway/test_reasoning_command.py

@@ -218,3 +218,112 @@ class TestReasoningCommand:
         assert result["final_response"] == "ok"
         assert _CapturingAgent.last_init is not None
         assert _CapturingAgent.last_init["reasoning_config"] == {"enabled": False}
+
+    def test_run_agent_includes_enabled_mcp_servers_in_gateway_toolsets(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "platform_toolsets:\n"
+            "  cli: [web, memory]\n"
+            "mcp_servers:\n"
+            "  exa:\n"
+            "    url: https://mcp.exa.ai/mcp\n"
+            "  web-search-prime:\n"
+            "    url: https://api.z.ai/api/mcp/web_search_prime/mcp\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+        monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env")
+        monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None)
+        monkeypatch.setattr(
+            gateway_run,
+            "_resolve_runtime_agent_kwargs",
+            lambda: {
+                "provider": "openrouter",
+                "api_mode": "chat_completions",
+                "base_url": "https://openrouter.ai/api/v1",
+                "api_key": "test-key",
+            },
+        )
+        fake_run_agent = types.ModuleType("run_agent")
+        fake_run_agent.AIAgent = _CapturingAgent
+        monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+        _CapturingAgent.last_init = None
+        runner = _make_runner()
+
+        source = SessionSource(
+            platform=Platform.LOCAL,
+            chat_id="cli",
+            chat_name="CLI",
+            chat_type="dm",
+            user_id="user-1",
+        )
+
+        result = asyncio.run(
+            runner._run_agent(
+                message="ping",
+                context_prompt="",
+                history=[],
+                source=source,
+                session_id="session-1",
+                session_key="agent:main:local:dm",
+            )
+        )
+
+        assert result["final_response"] == "ok"
+        assert _CapturingAgent.last_init is not None
+        enabled_toolsets = set(_CapturingAgent.last_init["enabled_toolsets"])
+        assert "web" in enabled_toolsets
+        assert "memory" in enabled_toolsets
+        assert "exa" in enabled_toolsets
+        assert "web-search-prime" in enabled_toolsets
+
+    def test_run_agent_homeassistant_uses_default_platform_toolset(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text("", encoding="utf-8")
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+        monkeypatch.setattr(gateway_run, "_env_path", hermes_home / ".env")
+        monkeypatch.setattr(gateway_run, "load_dotenv", lambda *args, **kwargs: None)
+        monkeypatch.setattr(
+            gateway_run,
+            "_resolve_runtime_agent_kwargs",
+            lambda: {
+                "provider": "openrouter",
+                "api_mode": "chat_completions",
+                "base_url": "https://openrouter.ai/api/v1",
+                "api_key": "test-key",
+            },
+        )
+        fake_run_agent = types.ModuleType("run_agent")
+        fake_run_agent.AIAgent = _CapturingAgent
+        monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+        _CapturingAgent.last_init = None
+        runner = _make_runner()
+
+        source = SessionSource(
+            platform=Platform.HOMEASSISTANT,
+            chat_id="ha",
+            chat_name="Home Assistant",
+            chat_type="dm",
+            user_id="user-1",
+        )
+
+        result = asyncio.run(
+            runner._run_agent(
+                message="ping",
+                context_prompt="",
+                history=[],
+                source=source,
+                session_id="session-1",
+                session_key="agent:main:homeassistant:dm",
+            )
+        )
+
+        assert result["final_response"] == "ok"
+        assert _CapturingAgent.last_init is not None
+        assert "homeassistant" in set(_CapturingAgent.last_init["enabled_toolsets"])

tests/gateway/test_session.py

@@ -386,6 +386,100 @@ class TestLoadTranscriptCorruptLines:
         assert messages[1]["content"] == "b"
 
 
+class TestLoadTranscriptPreferLongerSource:
+    """Regression: load_transcript must return whichever source (SQLite or JSONL)
+    has more messages to prevent silent truncation.  GH-3212."""
+
+    @pytest.fixture()
+    def store_with_db(self, tmp_path):
+        """SessionStore with both SQLite and JSONL active."""
+        from hermes_state import SessionDB
+
+        config = GatewayConfig()
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            s = SessionStore(sessions_dir=tmp_path, config=config)
+        s._db = SessionDB(db_path=tmp_path / "state.db")
+        s._loaded = True
+        return s
+
+    def test_jsonl_longer_than_sqlite_returns_jsonl(self, store_with_db):
+        """Legacy session: JSONL has full history, SQLite has only recent turn."""
+        sid = "legacy_session"
+        store_with_db._db.create_session(session_id=sid, source="gateway", model="m")
+        # JSONL has 10 messages (legacy history — written before SQLite existed)
+        for i in range(10):
+            role = "user" if i % 2 == 0 else "assistant"
+            store_with_db.append_to_transcript(
+                sid, {"role": role, "content": f"msg-{i}"}, skip_db=True,
+            )
+        # SQLite has only 2 messages (recent turn after migration)
+        store_with_db._db.append_message(session_id=sid, role="user", content="new-q")
+        store_with_db._db.append_message(session_id=sid, role="assistant", content="new-a")
+
+        result = store_with_db.load_transcript(sid)
+        assert len(result) == 10
+        assert result[0]["content"] == "msg-0"
+
+    def test_sqlite_longer_than_jsonl_returns_sqlite(self, store_with_db):
+        """Fully migrated session: SQLite has more (JSONL stopped growing)."""
+        sid = "migrated_session"
+        store_with_db._db.create_session(session_id=sid, source="gateway", model="m")
+        # JSONL has 2 old messages
+        store_with_db.append_to_transcript(
+            sid, {"role": "user", "content": "old-q"}, skip_db=True,
+        )
+        store_with_db.append_to_transcript(
+            sid, {"role": "assistant", "content": "old-a"}, skip_db=True,
+        )
+        # SQLite has 4 messages (superset after migration)
+        for i in range(4):
+            role = "user" if i % 2 == 0 else "assistant"
+            store_with_db._db.append_message(session_id=sid, role=role, content=f"db-{i}")
+
+        result = store_with_db.load_transcript(sid)
+        assert len(result) == 4
+        assert result[0]["content"] == "db-0"
+
+    def test_sqlite_empty_falls_back_to_jsonl(self, store_with_db):
+        """No SQLite rows — falls back to JSONL (original behavior preserved)."""
+        sid = "no_db_rows"
+        store_with_db.append_to_transcript(
+            sid, {"role": "user", "content": "hello"}, skip_db=True,
+        )
+        store_with_db.append_to_transcript(
+            sid, {"role": "assistant", "content": "hi"}, skip_db=True,
+        )
+
+        result = store_with_db.load_transcript(sid)
+        assert len(result) == 2
+        assert result[0]["content"] == "hello"
+
+    def test_both_empty_returns_empty(self, store_with_db):
+        """Neither source has data — returns empty list."""
+        result = store_with_db.load_transcript("nonexistent")
+        assert result == []
+
+    def test_equal_length_prefers_sqlite(self, store_with_db):
+        """When both have same count, SQLite wins (has richer fields like reasoning)."""
+        sid = "equal_session"
+        store_with_db._db.create_session(session_id=sid, source="gateway", model="m")
+        # Write 2 messages to JSONL only
+        store_with_db.append_to_transcript(
+            sid, {"role": "user", "content": "jsonl-q"}, skip_db=True,
+        )
+        store_with_db.append_to_transcript(
+            sid, {"role": "assistant", "content": "jsonl-a"}, skip_db=True,
+        )
+        # Write 2 different messages to SQLite only
+        store_with_db._db.append_message(session_id=sid, role="user", content="db-q")
+        store_with_db._db.append_message(session_id=sid, role="assistant", content="db-a")
+
+        result = store_with_db.load_transcript(sid)
+        assert len(result) == 2
+        # Should be the SQLite version (equal count → prefers SQLite)
+        assert result[0]["content"] == "db-q"
+
+
 class TestWhatsAppDMSessionKeyConsistency:
     """Regression: all session-key construction must go through build_session_key
     so DMs are isolated by chat_id across platforms."""

tests/gateway/test_session_race_guard.py

@@ -39,6 +39,7 @@ def _make_runner():
     runner._pending_messages = {}
     runner._pending_approvals = {}
     runner._voice_mode = {}
+    runner._background_tasks = set()
     runner._is_user_authorized = lambda _source: True
     return runner
 

tests/gateway/test_telegram_network_reconnect.py

@@ -0,0 +1,154 @@
+"""
+Tests for Telegram polling network error recovery.
+
+Specifically tests the fix for #3173 — when start_polling() fails after a
+network error, the adapter must self-reschedule the next reconnect attempt
+rather than silently leaving polling dead.
+"""
+
+import asyncio
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_telegram_mock():
+    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
+        return
+
+    telegram_mod = MagicMock()
+    telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
+    telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
+    telegram_mod.constants.ChatType.GROUP = "group"
+    telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
+    telegram_mod.constants.ChatType.CHANNEL = "channel"
+    telegram_mod.constants.ChatType.PRIVATE = "private"
+
+    for name in ("telegram", "telegram.ext", "telegram.constants"):
+        sys.modules.setdefault(name, telegram_mod)
+
+
+_ensure_telegram_mock()
+
+from gateway.platforms.telegram import TelegramAdapter  # noqa: E402
+
+
+def _make_adapter() -> TelegramAdapter:
+    return TelegramAdapter(PlatformConfig(enabled=True, token="test-token"))
+
+
+@pytest.mark.asyncio
+async def test_reconnect_self_schedules_on_start_polling_failure():
+    """
+    When start_polling() raises during a network error retry, the adapter must
+    schedule a new _handle_polling_network_error task — otherwise polling stays
+    dead with no further error callbacks to trigger recovery.
+
+    Regression test for #3173: gateway becomes unresponsive after Telegram 502.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 1
+
+    mock_updater = MagicMock()
+    mock_updater.running = True
+    mock_updater.stop = AsyncMock()
+    mock_updater.start_polling = AsyncMock(side_effect=Exception("Timed out"))
+
+    mock_app = MagicMock()
+    mock_app.updater = mock_updater
+    adapter._app = mock_app
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        await adapter._handle_polling_network_error(Exception("Bad Gateway"))
+
+    # A retry task must have been added to _background_tasks
+    pending = [t for t in adapter._background_tasks if not t.done()]
+    assert len(pending) >= 1, (
+        "Expected at least one self-rescheduled retry task in _background_tasks "
+        f"after start_polling failure, got {len(pending)}"
+    )
+
+    # Clean up — cancel the pending retry so it doesn't run after the test
+    for t in pending:
+        t.cancel()
+        try:
+            await t
+        except (asyncio.CancelledError, Exception):
+            pass
+
+
+@pytest.mark.asyncio
+async def test_reconnect_does_not_self_schedule_when_fatal_error_set():
+    """
+    When a fatal error is already set, the failed reconnect should NOT create
+    another retry task — the gateway is already shutting down this adapter.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 1
+    adapter._set_fatal_error("telegram_network_error", "already fatal", retryable=True)
+
+    mock_updater = MagicMock()
+    mock_updater.running = True
+    mock_updater.stop = AsyncMock()
+    mock_updater.start_polling = AsyncMock(side_effect=Exception("Timed out"))
+
+    mock_app = MagicMock()
+    mock_app.updater = mock_updater
+    adapter._app = mock_app
+
+    initial_count = len(adapter._background_tasks)
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        await adapter._handle_polling_network_error(Exception("Timed out"))
+
+    assert len(adapter._background_tasks) == initial_count, (
+        "Should not schedule a retry when a fatal error is already set"
+    )
+
+
+@pytest.mark.asyncio
+async def test_reconnect_success_resets_error_count():
+    """
+    When start_polling() succeeds, _polling_network_error_count should reset to 0.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 3
+
+    mock_updater = MagicMock()
+    mock_updater.running = True
+    mock_updater.stop = AsyncMock()
+    mock_updater.start_polling = AsyncMock()  # succeeds
+
+    mock_app = MagicMock()
+    mock_app.updater = mock_updater
+    adapter._app = mock_app
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        await adapter._handle_polling_network_error(Exception("Bad Gateway"))
+
+    assert adapter._polling_network_error_count == 0
+
+
+@pytest.mark.asyncio
+async def test_reconnect_triggers_fatal_after_max_retries():
+    """
+    After MAX_NETWORK_RETRIES attempts, the adapter should set a fatal error
+    rather than retrying forever.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 10  # MAX_NETWORK_RETRIES
+
+    fatal_handler = AsyncMock()
+    adapter.set_fatal_error_handler(fatal_handler)
+
+    mock_app = MagicMock()
+    adapter._app = mock_app
+
+    await adapter._handle_polling_network_error(Exception("still failing"))
+
+    assert adapter.has_fatal_error
+    assert adapter.fatal_error_code == "telegram_network_error"
+    fatal_handler.assert_called_once()

tests/gateway/test_verbose_command.py

@@ -0,0 +1,146 @@
+"""Tests for gateway /verbose command (config-gated tool progress cycling)."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import yaml
+
+import gateway.run as gateway_run
+from gateway.config import Platform
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionSource
+
+
+def _make_event(text="/verbose", platform=Platform.TELEGRAM, user_id="12345", chat_id="67890"):
+    """Build a MessageEvent for testing."""
+    source = SessionSource(
+        platform=platform,
+        user_id=user_id,
+        chat_id=chat_id,
+        user_name="testuser",
+    )
+    return MessageEvent(text=text, source=source)
+
+
+def _make_runner():
+    """Create a bare GatewayRunner without calling __init__."""
+    runner = object.__new__(gateway_run.GatewayRunner)
+    runner.adapters = {}
+    runner._ephemeral_system_prompt = ""
+    runner._prefill_messages = []
+    runner._reasoning_config = None
+    runner._show_reasoning = False
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._running_agents = {}
+    runner.hooks = MagicMock()
+    runner.hooks.emit = AsyncMock()
+    runner.hooks.loaded_hooks = []
+    runner._session_db = None
+    runner._get_or_create_gateway_honcho = lambda session_key: (None, None)
+    return runner
+
+
+class TestVerboseCommand:
+    """Tests for _handle_verbose_command in the gateway."""
+
+    @pytest.mark.asyncio
+    async def test_disabled_by_default(self, tmp_path, monkeypatch):
+        """When tool_progress_command is false, /verbose returns an info message."""
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text("display:\n  tool_progress: all\n", encoding="utf-8")
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+
+        runner = _make_runner()
+        result = await runner._handle_verbose_command(_make_event())
+
+        assert "not enabled" in result.lower()
+        assert "tool_progress_command" in result
+
+    @pytest.mark.asyncio
+    async def test_enabled_cycles_mode(self, tmp_path, monkeypatch):
+        """When enabled, /verbose cycles tool_progress mode."""
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "display:\n  tool_progress_command: true\n  tool_progress: all\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+
+        runner = _make_runner()
+        result = await runner._handle_verbose_command(_make_event())
+
+        # all -> verbose
+        assert "VERBOSE" in result
+
+        # Verify config was saved
+        saved = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+        assert saved["display"]["tool_progress"] == "verbose"
+
+    @pytest.mark.asyncio
+    async def test_cycles_through_all_modes(self, tmp_path, monkeypatch):
+        """Calling /verbose repeatedly cycles through all four modes."""
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "display:\n  tool_progress_command: true\n  tool_progress: 'off'\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+        runner = _make_runner()
+
+        # off -> new -> all -> verbose -> off
+        expected = ["new", "all", "verbose", "off"]
+        for mode in expected:
+            result = await runner._handle_verbose_command(_make_event())
+            saved = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+            assert saved["display"]["tool_progress"] == mode, \
+                f"Expected {mode}, got {saved['display']['tool_progress']}"
+
+    @pytest.mark.asyncio
+    async def test_defaults_to_all_when_no_tool_progress_set(self, tmp_path, monkeypatch):
+        """When tool_progress is not in config, defaults to 'all' then cycles to verbose."""
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "display:\n  tool_progress_command: true\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+
+        runner = _make_runner()
+        result = await runner._handle_verbose_command(_make_event())
+
+        # default "all" -> verbose
+        assert "VERBOSE" in result
+        saved = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+        assert saved["display"]["tool_progress"] == "verbose"
+
+    @pytest.mark.asyncio
+    async def test_no_config_file_returns_disabled(self, tmp_path, monkeypatch):
+        """When config.yaml doesn't exist, command reports disabled."""
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+        # No config.yaml
+
+        monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home)
+
+        runner = _make_runner()
+        result = await runner._handle_verbose_command(_make_event())
+        assert "not enabled" in result.lower()
+
+    def test_verbose_is_in_gateway_known_commands(self):
+        """The /verbose command is recognized by the gateway dispatch."""
+        from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS
+        assert "verbose" in GATEWAY_KNOWN_COMMANDS

tests/hermes_cli/test_commands.py

@@ -134,12 +134,19 @@ class TestDerivedDicts:
 # ---------------------------------------------------------------------------
 
 class TestGatewayKnownCommands:
-    def test_excludes_cli_only(self):
+    def test_excludes_cli_only_without_config_gate(self):
         for cmd in COMMAND_REGISTRY:
-            if cmd.cli_only:
+            if cmd.cli_only and not cmd.gateway_config_gate:
                 assert cmd.name not in GATEWAY_KNOWN_COMMANDS, \
                     f"cli_only command '{cmd.name}' should not be in GATEWAY_KNOWN_COMMANDS"
 
+    def test_includes_config_gated_cli_only(self):
+        """Commands with gateway_config_gate are always in GATEWAY_KNOWN_COMMANDS."""
+        for cmd in COMMAND_REGISTRY:
+            if cmd.gateway_config_gate:
+                assert cmd.name in GATEWAY_KNOWN_COMMANDS, \
+                    f"config-gated command '{cmd.name}' should be in GATEWAY_KNOWN_COMMANDS"
+
     def test_includes_gateway_commands(self):
         for cmd in COMMAND_REGISTRY:
             if not cmd.cli_only:
@@ -160,11 +167,11 @@ class TestGatewayHelpLines:
         lines = gateway_help_lines()
         assert len(lines) > 10
 
-    def test_excludes_cli_only_commands(self):
+    def test_excludes_cli_only_commands_without_config_gate(self):
         lines = gateway_help_lines()
         joined = "\n".join(lines)
         for cmd in COMMAND_REGISTRY:
-            if cmd.cli_only:
+            if cmd.cli_only and not cmd.gateway_config_gate:
                 assert f"`/{cmd.name}" not in joined, \
                     f"cli_only command /{cmd.name} should not be in gateway help"
 
@@ -188,10 +195,10 @@ class TestTelegramBotCommands:
         for name, _ in telegram_bot_commands():
             assert "-" not in name, f"Telegram command '{name}' contains a hyphen"
 
-    def test_excludes_cli_only(self):
+    def test_excludes_cli_only_without_config_gate(self):
         names = {name for name, _ in telegram_bot_commands()}
         for cmd in COMMAND_REGISTRY:
-            if cmd.cli_only:
+            if cmd.cli_only and not cmd.gateway_config_gate:
                 tg_name = cmd.name.replace("-", "_")
                 assert tg_name not in names
 
@@ -211,13 +218,84 @@ class TestSlackSubcommandMap:
         assert "bg" in mapping
         assert "reset" in mapping
 
-    def test_excludes_cli_only(self):
+    def test_excludes_cli_only_without_config_gate(self):
         mapping = slack_subcommand_map()
         for cmd in COMMAND_REGISTRY:
-            if cmd.cli_only:
+            if cmd.cli_only and not cmd.gateway_config_gate:
                 assert cmd.name not in mapping
 
 
+# ---------------------------------------------------------------------------
+# Config-gated gateway commands
+# ---------------------------------------------------------------------------
+
+class TestGatewayConfigGate:
+    """Tests for the gateway_config_gate mechanism on CommandDef."""
+
+    def test_verbose_has_config_gate(self):
+        cmd = resolve_command("verbose")
+        assert cmd is not None
+        assert cmd.cli_only is True
+        assert cmd.gateway_config_gate == "display.tool_progress_command"
+
+    def test_verbose_in_gateway_known_commands(self):
+        """Config-gated commands are always recognized by the gateway."""
+        assert "verbose" in GATEWAY_KNOWN_COMMANDS
+
+    def test_config_gate_excluded_from_help_when_off(self, tmp_path, monkeypatch):
+        """When the config gate is falsy, the command should not appear in help."""
+        # Write a config with the gate off (default)
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: false\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        lines = gateway_help_lines()
+        joined = "\n".join(lines)
+        assert "`/verbose" not in joined
+
+    def test_config_gate_included_in_help_when_on(self, tmp_path, monkeypatch):
+        """When the config gate is truthy, the command should appear in help."""
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: true\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        lines = gateway_help_lines()
+        joined = "\n".join(lines)
+        assert "`/verbose" in joined
+
+    def test_config_gate_excluded_from_telegram_when_off(self, tmp_path, monkeypatch):
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: false\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        names = {name for name, _ in telegram_bot_commands()}
+        assert "verbose" not in names
+
+    def test_config_gate_included_in_telegram_when_on(self, tmp_path, monkeypatch):
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: true\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        names = {name for name, _ in telegram_bot_commands()}
+        assert "verbose" in names
+
+    def test_config_gate_excluded_from_slack_when_off(self, tmp_path, monkeypatch):
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: false\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        mapping = slack_subcommand_map()
+        assert "verbose" not in mapping
+
+    def test_config_gate_included_in_slack_when_on(self, tmp_path, monkeypatch):
+        config_file = tmp_path / "config.yaml"
+        config_file.write_text("display:\n  tool_progress_command: true\n")
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        mapping = slack_subcommand_map()
+        assert "verbose" in mapping
+
+
 # ---------------------------------------------------------------------------
 # Autocomplete (SlashCommandCompleter)
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_skills_hub.py

@@ -3,7 +3,7 @@ from io import StringIO
 import pytest
 from rich.console import Console
 
-from hermes_cli.skills_hub import do_check, do_list, do_update, handle_skills_slash
+from hermes_cli.skills_hub import do_check, do_install, do_list, do_update, handle_skills_slash
 
 
 class _DummyLockFile:
@@ -177,3 +177,57 @@ def test_do_update_reinstalls_outdated_skills(monkeypatch):
 
     assert installs == [("skills-sh/example/repo/hub-skill", "category", True)]
     assert "Updated 1 skill" in output
+
+
+def test_do_install_scans_with_resolved_identifier(monkeypatch, tmp_path, hub_env):
+    import tools.skills_guard as guard
+    import tools.skills_hub as hub
+
+    canonical_identifier = "skills-sh/anthropics/skills/frontend-design"
+
+    class _ResolvedSource:
+        def inspect(self, identifier):
+            return type("Meta", (), {
+                "extra": {},
+                "identifier": canonical_identifier,
+            })()
+
+        def fetch(self, identifier):
+            return type("Bundle", (), {
+                "name": "frontend-design",
+                "files": {"SKILL.md": "# Frontend Design"},
+                "source": "skills.sh",
+                "identifier": canonical_identifier,
+                "trust_level": "trusted",
+                "metadata": {},
+            })()
+
+    q_path = tmp_path / "skills" / ".hub" / "quarantine" / "frontend-design"
+    q_path.mkdir(parents=True)
+    (q_path / "SKILL.md").write_text("# Frontend Design")
+
+    scanned = {}
+
+    def _scan_skill(skill_path, source="community"):
+        scanned["source"] = source
+        return guard.ScanResult(
+            skill_name="frontend-design",
+            source=source,
+            trust_level="trusted",
+            verdict="safe",
+        )
+
+    monkeypatch.setattr(hub, "ensure_hub_dirs", lambda: None)
+    monkeypatch.setattr(hub, "create_source_router", lambda auth: [_ResolvedSource()])
+    monkeypatch.setattr(hub, "quarantine_bundle", lambda bundle: q_path)
+    monkeypatch.setattr(hub, "HubLockFile", lambda: type("Lock", (), {"get_installed": lambda self, name: None})())
+    monkeypatch.setattr(guard, "scan_skill", _scan_skill)
+    monkeypatch.setattr(guard, "format_scan_report", lambda result: "scan ok")
+    monkeypatch.setattr(guard, "should_allow_install", lambda result, force=False: (False, "stop after scan"))
+
+    sink = StringIO()
+    console = Console(file=sink, force_terminal=False, color_system=None)
+
+    do_install("skils-sh/anthropics/skills/frontend-design", console=console, skip_confirm=True)
+
+    assert scanned["source"] == canonical_identifier

tests/hermes_cli/test_tools_config.py

@@ -39,6 +39,39 @@ def test_platform_toolset_summary_uses_explicit_platform_list():
     assert summary["cli"] == _get_platform_tools(config, "cli")
 
 
+def test_get_platform_tools_includes_enabled_mcp_servers_by_default():
+    config = {
+        "mcp_servers": {
+            "exa": {"url": "https://mcp.exa.ai/mcp"},
+            "web-search-prime": {"url": "https://api.z.ai/api/mcp/web_search_prime/mcp"},
+            "disabled-server": {"url": "https://example.com/mcp", "enabled": False},
+        }
+    }
+
+    enabled = _get_platform_tools(config, "cli")
+
+    assert "exa" in enabled
+    assert "web-search-prime" in enabled
+    assert "disabled-server" not in enabled
+
+
+def test_get_platform_tools_keeps_enabled_mcp_servers_with_explicit_builtin_selection():
+    config = {
+        "platform_toolsets": {"cli": ["web", "memory"]},
+        "mcp_servers": {
+            "exa": {"url": "https://mcp.exa.ai/mcp"},
+            "web-search-prime": {"url": "https://api.z.ai/api/mcp/web_search_prime/mcp"},
+        },
+    }
+
+    enabled = _get_platform_tools(config, "cli")
+
+    assert "web" in enabled
+    assert "memory" in enabled
+    assert "exa" in enabled
+    assert "web-search-prime" in enabled
+
+
 def test_toolset_has_keys_for_vision_accepts_codex_auth(tmp_path, monkeypatch):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))
     (tmp_path / "auth.json").write_text(

tests/hermes_cli/test_tools_disable_enable.py

@@ -134,6 +134,23 @@ class TestToolsMixedTargets:
         assert "web" not in saved["platform_toolsets"]["cli"]
         assert "create_issue" in saved["mcp_servers"]["github"]["tools"]["exclude"]
 
+    def test_builtin_toggle_does_not_persist_implicit_mcp_defaults(self):
+        config = {
+            "platform_toolsets": {"cli": ["web", "memory"]},
+            "mcp_servers": {"exa": {"url": "https://mcp.exa.ai/mcp"}},
+        }
+        with patch("hermes_cli.tools_config.load_config", return_value=config), \
+             patch("hermes_cli.tools_config.save_config") as mock_save:
+            tools_disable_enable_command(Namespace(
+                tools_action="disable",
+                names=["web"],
+                platform="cli",
+            ))
+        saved = mock_save.call_args[0][0]
+        assert "web" not in saved["platform_toolsets"]["cli"]
+        assert "memory" in saved["platform_toolsets"]["cli"]
+        assert "exa" not in saved["platform_toolsets"]["cli"]
+
 
 # ── List output ──────────────────────────────────────────────────────────────
 

tests/test_anthropic_error_handling.py

@@ -273,6 +273,9 @@ def test_401_credential_refresh_recovers(monkeypatch):
                 return _anthropic_response("Auth refreshed")
 
             self._interruptible_api_call = _fake_api_call
+            # Also patch streaming path — run_conversation now prefers
+            # streaming for health checking even without stream consumers.
+            self._interruptible_streaming_api_call = lambda api_kwargs, **kw: _fake_api_call(api_kwargs)
             return super().run_conversation(
                 user_message, conversation_history=conversation_history, task_id=task_id
             )

tests/test_exit_cleanup_interrupt.py

@@ -0,0 +1,105 @@
+"""Tests for KeyboardInterrupt handling in exit cleanup paths.
+
+``except Exception`` does not catch ``KeyboardInterrupt`` (which inherits
+from ``BaseException``).  A second Ctrl+C during exit cleanup must not
+abort remaining cleanup steps.  These tests exercise the actual production
+code paths — not a copy of the try/except pattern.
+"""
+
+import atexit
+import weakref
+from unittest.mock import MagicMock, patch, call
+
+import pytest
+
+
+class TestHonchoAtexitFlush:
+    """run_agent.py — _register_honcho_exit_hook atexit handler."""
+
+    def test_keyboard_interrupt_during_flush_does_not_propagate(self):
+        """The atexit handler must swallow KeyboardInterrupt from flush_all()."""
+        mock_manager = MagicMock()
+        mock_manager.flush_all.side_effect = KeyboardInterrupt
+
+        # Capture functions passed to atexit.register
+        registered_fns = []
+        original_register = atexit.register
+
+        def capturing_register(fn, *args, **kwargs):
+            registered_fns.append(fn)
+            # Don't actually register — we don't want side effects
+
+        with patch("atexit.register", side_effect=capturing_register):
+            from run_agent import AIAgent
+            agent = object.__new__(AIAgent)
+            agent._honcho = mock_manager
+            agent._honcho_exit_hook_registered = False
+            agent._register_honcho_exit_hook()
+
+        # Our handler is the last one registered
+        assert len(registered_fns) >= 1, "atexit handler was not registered"
+        flush_handler = registered_fns[-1]
+
+        # Invoke the registered handler — must not raise
+        flush_handler()
+        mock_manager.flush_all.assert_called_once()
+
+
+class TestCronJobCleanup:
+    """cron/scheduler.py — end_session + close in the finally block."""
+
+    def test_keyboard_interrupt_in_end_session_does_not_skip_close(self):
+        """If end_session raises KeyboardInterrupt, close() must still run."""
+        mock_db = MagicMock()
+        mock_db.end_session.side_effect = KeyboardInterrupt
+
+        from cron import scheduler
+
+        job = {
+            "id": "test-job-1",
+            "name": "test cleanup",
+            "prompt": "hello",
+            "schedule": "0 9 * * *",
+            "model": "test/model",
+        }
+
+        with patch("hermes_state.SessionDB", return_value=mock_db), \
+             patch.object(scheduler, "_build_job_prompt", return_value="hello"), \
+             patch.object(scheduler, "_resolve_origin", return_value=None), \
+             patch.object(scheduler, "_resolve_delivery_target", return_value=None), \
+             patch("dotenv.load_dotenv", return_value=None), \
+             patch("run_agent.AIAgent") as MockAgent:
+            # Make the agent raise immediately so we hit the finally block
+            MockAgent.return_value.run_conversation.side_effect = RuntimeError("boom")
+            scheduler.run_job(job)
+
+        mock_db.end_session.assert_called_once()
+        mock_db.close.assert_called_once()
+
+    def test_keyboard_interrupt_in_close_does_not_propagate(self):
+        """If close() raises KeyboardInterrupt, it must not escape run_job."""
+        mock_db = MagicMock()
+        mock_db.close.side_effect = KeyboardInterrupt
+
+        from cron import scheduler
+
+        job = {
+            "id": "test-job-2",
+            "name": "test close interrupt",
+            "prompt": "hello",
+            "schedule": "0 9 * * *",
+            "model": "test/model",
+        }
+
+        with patch("hermes_state.SessionDB", return_value=mock_db), \
+             patch.object(scheduler, "_build_job_prompt", return_value="hello"), \
+             patch.object(scheduler, "_resolve_origin", return_value=None), \
+             patch.object(scheduler, "_resolve_delivery_target", return_value=None), \
+             patch("dotenv.load_dotenv", return_value=None), \
+             patch("run_agent.AIAgent") as MockAgent:
+            MockAgent.return_value.run_conversation.side_effect = RuntimeError("boom")
+            # Must not raise
+            scheduler.run_job(job)
+
+        mock_db.end_session.assert_called_once()
+        mock_db.close.assert_called_once()

tests/test_hermes_state.py

@@ -1102,6 +1102,89 @@ class TestListSessionsRich:
         assert "Line one Line two" in sessions[0]["preview"]
 
 
+# =========================================================================
+# Session source exclusion (--source flag for third-party isolation)
+# =========================================================================
+
+class TestExcludeSources:
+    """Tests for exclude_sources on list_sessions_rich and search_messages."""
+
+    def test_list_sessions_rich_excludes_tool_source(self, db):
+        db.create_session("s1", "cli")
+        db.create_session("s2", "tool")
+        db.create_session("s3", "telegram")
+        sessions = db.list_sessions_rich(exclude_sources=["tool"])
+        ids = [s["id"] for s in sessions]
+        assert "s1" in ids
+        assert "s3" in ids
+        assert "s2" not in ids
+
+    def test_list_sessions_rich_no_exclusion_returns_all(self, db):
+        db.create_session("s1", "cli")
+        db.create_session("s2", "tool")
+        sessions = db.list_sessions_rich()
+        ids = [s["id"] for s in sessions]
+        assert "s1" in ids
+        assert "s2" in ids
+
+    def test_list_sessions_rich_source_and_exclude_combined(self, db):
+        """When source= is explicit, exclude_sources should not conflict."""
+        db.create_session("s1", "cli")
+        db.create_session("s2", "tool")
+        db.create_session("s3", "telegram")
+        # Explicit source filter: only tool sessions, no exclusion
+        sessions = db.list_sessions_rich(source="tool")
+        ids = [s["id"] for s in sessions]
+        assert ids == ["s2"]
+
+    def test_list_sessions_rich_exclude_multiple_sources(self, db):
+        db.create_session("s1", "cli")
+        db.create_session("s2", "tool")
+        db.create_session("s3", "cron")
+        db.create_session("s4", "telegram")
+        sessions = db.list_sessions_rich(exclude_sources=["tool", "cron"])
+        ids = [s["id"] for s in sessions]
+        assert "s1" in ids
+        assert "s4" in ids
+        assert "s2" not in ids
+        assert "s3" not in ids
+
+    def test_search_messages_excludes_tool_source(self, db):
+        db.create_session("s1", "cli")
+        db.append_message("s1", "user", "Python deployment question")
+        db.create_session("s2", "tool")
+        db.append_message("s2", "user", "Python automated question")
+        results = db.search_messages("Python", exclude_sources=["tool"])
+        sources = [r["source"] for r in results]
+        assert "cli" in sources
+        assert "tool" not in sources
+
+    def test_search_messages_no_exclusion_returns_all_sources(self, db):
+        db.create_session("s1", "cli")
+        db.append_message("s1", "user", "Rust deployment question")
+        db.create_session("s2", "tool")
+        db.append_message("s2", "user", "Rust automated question")
+        results = db.search_messages("Rust")
+        sources = [r["source"] for r in results]
+        assert "cli" in sources
+        assert "tool" in sources
+
+    def test_search_messages_source_include_and_exclude(self, db):
+        """source_filter (include) and exclude_sources can coexist."""
+        db.create_session("s1", "cli")
+        db.append_message("s1", "user", "Golang test")
+        db.create_session("s2", "telegram")
+        db.append_message("s2", "user", "Golang test")
+        db.create_session("s3", "tool")
+        db.append_message("s3", "user", "Golang test")
+        # Include cli+tool, but exclude tool → should only return cli
+        results = db.search_messages(
+            "Golang", source_filter=["cli", "tool"], exclude_sources=["tool"]
+        )
+        sources = [r["source"] for r in results]
+        assert sources == ["cli"]
+
+
 class TestResolveSessionByNameOrId:
     """Tests for the main.py helper that resolves names or IDs."""
 
@@ -1116,3 +1199,66 @@ class TestResolveSessionByNameOrId:
         db.set_session_title("s1", "my project")
         result = db.resolve_session_by_title("my project")
         assert result == "s1"
+
+
+# =========================================================================
+# Concurrent write safety / lock contention fixes (#3139)
+# =========================================================================
+
+class TestConcurrentWriteSafety:
+    def test_create_session_insert_or_ignore_is_idempotent(self, db):
+        """create_session with the same ID twice must not raise (INSERT OR IGNORE)."""
+        db.create_session(session_id="dup-1", source="cli", model="m")
+        # Second call should be silent — no IntegrityError
+        db.create_session(session_id="dup-1", source="gateway", model="m2")
+        session = db.get_session("dup-1")
+        # Row should exist (first write wins with OR IGNORE)
+        assert session is not None
+        assert session["source"] == "cli"
+
+    def test_ensure_session_creates_missing_row(self, db):
+        """ensure_session must create a minimal row when the session doesn't exist."""
+        assert db.get_session("orphan-session") is None
+        db.ensure_session("orphan-session", source="gateway", model="test-model")
+        row = db.get_session("orphan-session")
+        assert row is not None
+        assert row["source"] == "gateway"
+        assert row["model"] == "test-model"
+
+    def test_ensure_session_is_idempotent(self, db):
+        """ensure_session on an existing row must be a no-op (no overwrite)."""
+        db.create_session(session_id="existing", source="cli", model="original-model")
+        db.ensure_session("existing", source="gateway", model="overwrite-model")
+        row = db.get_session("existing")
+        # First write wins — ensure_session must not overwrite
+        assert row["source"] == "cli"
+        assert row["model"] == "original-model"
+
+    def test_ensure_session_allows_append_message_after_failed_create(self, db):
+        """Messages can be flushed even when create_session failed at startup.
+
+        Simulates the #3139 scenario: create_session raises (lock), then
+        ensure_session is called during flush, then append_message succeeds.
+        """
+        # Simulate failed create_session — row absent
+        db.ensure_session("late-session", source="gateway", model="gpt-4")
+        db.append_message(
+            session_id="late-session",
+            role="user",
+            content="hello after lock",
+        )
+        msgs = db.get_messages("late-session")
+        assert len(msgs) == 1
+        assert msgs[0]["content"] == "hello after lock"
+
+    def test_sqlite_timeout_is_at_least_30s(self, db):
+        """Connection timeout should be >= 30s to survive CLI/gateway contention."""
+        # Access the underlying connection timeout via sqlite3 introspection.
+        # There is no public API, so we check the kwarg via the module default.
+        import sqlite3
+        import inspect
+        from hermes_state import SessionDB as _SessionDB
+        src = inspect.getsource(_SessionDB.__init__)
+        assert "30" in src, (
+            "SQLite timeout should be at least 30s to handle CLI/gateway lock contention"
+        )

tests/test_provider_parity.py

@@ -556,7 +556,7 @@ class TestAuxiliaryClientProviderPriority:
         with patch("agent.auxiliary_client._read_nous_auth", return_value={"access_token": "nous-tok"}), \
              patch("agent.auxiliary_client.OpenAI") as mock:
             client, model = get_text_auxiliary_client()
-        assert model == "gemini-3-flash"
+        assert model == "google/gemini-3-flash-preview"
 
     def test_custom_endpoint_when_no_nous(self, monkeypatch):
         monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)

tests/tools/test_approval.py

@@ -512,3 +512,73 @@ class TestGatewayProtection:
         dangerous, key, desc = detect_dangerous_command(cmd)
         assert dangerous is False
 
+
+class TestNormalizationBypass:
+    """Obfuscation techniques must not bypass dangerous command detection."""
+
+    def test_fullwidth_unicode_rm(self):
+        """Fullwidth Unicode 'ｒｍ -ｒｆ /' must be caught after NFKC normalization."""
+        cmd = "\uff52\uff4d -\uff52\uff46 /"  # ｒｍ -ｒｆ /
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True, f"Fullwidth 'rm -rf /' was not detected: {cmd!r}"
+
+    def test_fullwidth_unicode_dd(self):
+        """Fullwidth 'ｄｄ if=/dev/zero' must be caught."""
+        cmd = "\uff44\uff44 if=/dev/zero of=/dev/sda"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True
+
+    def test_fullwidth_unicode_chmod(self):
+        """Fullwidth 'ｃｈｍｏｄ 777' must be caught."""
+        cmd = "\uff43\uff48\uff4d\uff4f\uff44 777 /tmp/test"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True
+
+    def test_ansi_csi_wrapped_rm(self):
+        """ANSI CSI color codes wrapping 'rm' must be stripped and caught."""
+        cmd = "\x1b[31mrm\x1b[0m -rf /"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True, f"ANSI-wrapped 'rm -rf /' was not detected"
+
+    def test_ansi_osc_embedded_rm(self):
+        """ANSI OSC sequences embedded in command must be stripped."""
+        cmd = "\x1b]0;title\x07rm -rf /"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True
+
+    def test_ansi_8bit_c1_wrapped_rm(self):
+        """8-bit C1 CSI (0x9b) wrapping 'rm' must be stripped and caught."""
+        cmd = "\x9b31mrm\x9b0m -rf /"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True, "8-bit C1 CSI bypass was not caught"
+
+    def test_null_byte_in_rm(self):
+        """Null bytes injected into 'rm' must be stripped and caught."""
+        cmd = "r\x00m -rf /"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True, f"Null-byte 'rm' was not detected: {cmd!r}"
+
+    def test_null_byte_in_dd(self):
+        """Null bytes in 'dd' must be stripped."""
+        cmd = "d\x00d if=/dev/sda"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True
+
+    def test_mixed_fullwidth_and_ansi(self):
+        """Combined fullwidth + ANSI obfuscation must still be caught."""
+        cmd = "\x1b[1m\uff52\uff4d\x1b[0m -rf /"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is True
+
+    def test_safe_command_after_normalization(self):
+        """Normal safe commands must not be flagged after normalization."""
+        cmd = "ls -la /tmp"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is False
+
+    def test_fullwidth_safe_command_not_flagged(self):
+        """Fullwidth 'ｌｓ -ｌａ' is safe and must not be flagged."""
+        cmd = "\uff4c\uff53 -\uff4c\uff41 /tmp"
+        dangerous, key, desc = detect_dangerous_command(cmd)
+        assert dangerous is False
+

tests/tools/test_delegate_toolset_scope.py

@@ -0,0 +1,66 @@
+"""Tests for delegate_tool toolset scoping.
+
+Verifies that subagents cannot gain tools that the parent does not have.
+The LLM controls the `toolsets` parameter — without intersection with the
+parent's enabled_toolsets, it can escalate privileges by requesting
+arbitrary toolsets.
+"""
+
+from unittest.mock import MagicMock, patch
+from types import SimpleNamespace
+
+from tools.delegate_tool import _strip_blocked_tools
+
+
+class TestToolsetIntersection:
+    """Subagent toolsets must be a subset of parent's enabled_toolsets."""
+
+    def test_requested_toolsets_intersected_with_parent(self):
+        """LLM requests toolsets parent doesn't have — extras are dropped."""
+        parent = SimpleNamespace(enabled_toolsets=["terminal", "file"])
+
+        # Simulate the intersection logic from _build_child_agent
+        parent_toolsets = set(parent.enabled_toolsets)
+        requested = ["terminal", "file", "web", "browser", "rl"]
+        scoped = [t for t in requested if t in parent_toolsets]
+
+        assert sorted(scoped) == ["file", "terminal"]
+        assert "web" not in scoped
+        assert "browser" not in scoped
+        assert "rl" not in scoped
+
+    def test_all_requested_toolsets_available_on_parent(self):
+        """LLM requests subset of parent tools — all pass through."""
+        parent = SimpleNamespace(enabled_toolsets=["terminal", "file", "web", "browser"])
+
+        parent_toolsets = set(parent.enabled_toolsets)
+        requested = ["terminal", "web"]
+        scoped = [t for t in requested if t in parent_toolsets]
+
+        assert sorted(scoped) == ["terminal", "web"]
+
+    def test_no_toolsets_requested_inherits_parent(self):
+        """When toolsets is None/empty, child inherits parent's set."""
+        parent_toolsets = ["terminal", "file", "web"]
+        child = _strip_blocked_tools(parent_toolsets)
+        assert "terminal" in child
+        assert "file" in child
+        assert "web" in child
+
+    def test_strip_blocked_removes_delegation(self):
+        """Blocked toolsets (delegation, clarify, etc.) are always removed."""
+        child = _strip_blocked_tools(["terminal", "delegation", "clarify", "memory"])
+        assert "delegation" not in child
+        assert "clarify" not in child
+        assert "memory" not in child
+        assert "terminal" in child
+
+    def test_empty_intersection_yields_empty_toolsets(self):
+        """If parent has no overlap with requested, child gets nothing extra."""
+        parent = SimpleNamespace(enabled_toolsets=["terminal"])
+
+        parent_toolsets = set(parent.enabled_toolsets)
+        requested = ["web", "browser"]
+        scoped = [t for t in requested if t in parent_toolsets]
+
+        assert scoped == []

tests/tools/test_session_search.py

@@ -8,6 +8,7 @@ from tools.session_search_tool import (
     _format_timestamp,
     _format_conversation,
     _truncate_around_matches,
+    _HIDDEN_SESSION_SOURCES,
     MAX_SESSION_CHARS,
     SESSION_SEARCH_SCHEMA,
 )
@@ -17,6 +18,17 @@ from tools.session_search_tool import (
 # Tool schema guidance
 # =========================================================================
 
+class TestHiddenSessionSources:
+    """Verify the _HIDDEN_SESSION_SOURCES constant used for third-party isolation."""
+
+    def test_tool_source_is_hidden(self):
+        assert "tool" in _HIDDEN_SESSION_SOURCES
+
+    def test_standard_sources_not_hidden(self):
+        for src in ("cli", "telegram", "discord", "slack", "cron"):
+            assert src not in _HIDDEN_SESSION_SOURCES
+
+
 class TestSessionSearchSchema:
     def test_keeps_cross_session_recall_guidance_without_current_session_nudge(self):
         description = SESSION_SEARCH_SCHEMA["description"]

tests/tools/test_skills_guard.py

@@ -55,6 +55,13 @@ class TestResolveTrustLevel:
         assert _resolve_trust_level("anthropics/skills") == "trusted"
         assert _resolve_trust_level("openai/skills/some-skill") == "trusted"
 
+    def test_skills_sh_wrapped_trusted_repos(self):
+        assert _resolve_trust_level("skills-sh/openai/skills/skill-creator") == "trusted"
+        assert _resolve_trust_level("skills-sh/anthropics/skills/frontend-design") == "trusted"
+
+    def test_common_skills_sh_prefix_typo_still_maps_to_trusted_repo(self):
+        assert _resolve_trust_level("skils-sh/anthropics/skills/frontend-design") == "trusted"
+
     def test_community_default(self):
         assert _resolve_trust_level("random-user/my-skill") == "community"
         assert _resolve_trust_level("") == "community"

tests/tools/test_skills_hub.py

@@ -179,6 +179,24 @@ class TestSkillsShSource:
         assert bundle.identifier == "skills-sh/vercel-labs/agent-skills/vercel-react-best-practices"
         mock_fetch.assert_called_once_with("vercel-labs/agent-skills/vercel-react-best-practices")
 
+    @patch.object(GitHubSource, "fetch")
+    def test_fetch_accepts_common_skills_sh_prefix_typo(self, mock_fetch):
+        expected_identifier = "anthropics/skills/frontend-design"
+        mock_fetch.side_effect = lambda identifier: SkillBundle(
+            name="frontend-design",
+            files={"SKILL.md": "# Frontend Design"},
+            source="github",
+            identifier=expected_identifier,
+            trust_level="trusted",
+        ) if identifier == expected_identifier else None
+
+        bundle = self._source().fetch("skils-sh/anthropics/skills/frontend-design")
+
+        assert bundle is not None
+        assert bundle.source == "skills.sh"
+        assert bundle.identifier == "skills-sh/anthropics/skills/frontend-design"
+        assert mock_fetch.call_args_list[0] == ((expected_identifier,), {})
+
     @patch("tools.skills_hub._write_index_cache")
     @patch("tools.skills_hub._read_index_cache", return_value=None)
     @patch("tools.skills_hub.httpx.get")
@@ -213,6 +231,26 @@ class TestSkillsShSource:
         assert meta.extra["security_audits"]["socket"] == "Pass"
         mock_inspect.assert_called_once_with("vercel-labs/agent-skills/vercel-react-best-practices")
 
+    @patch.object(GitHubSource, "inspect")
+    def test_inspect_accepts_common_skills_sh_prefix_typo(self, mock_inspect):
+        expected_identifier = "anthropics/skills/frontend-design"
+        mock_inspect.side_effect = lambda identifier: SkillMeta(
+            name="frontend-design",
+            description="Distinctive frontend interfaces.",
+            source="github",
+            identifier=expected_identifier,
+            trust_level="trusted",
+            repo="anthropics/skills",
+            path="frontend-design",
+        ) if identifier == expected_identifier else None
+
+        meta = self._source().inspect("skils-sh/anthropics/skills/frontend-design")
+
+        assert meta is not None
+        assert meta.source == "skills.sh"
+        assert meta.identifier == "skills-sh/anthropics/skills/frontend-design"
+        assert mock_inspect.call_args_list[0] == ((expected_identifier,), {})
+
     @patch.object(GitHubSource, "_list_skills_in_repo")
     @patch.object(GitHubSource, "inspect")
     def test_inspect_falls_back_to_repo_skill_catalog_when_slug_differs(self, mock_inspect, mock_list_skills):
@@ -307,6 +345,39 @@ class TestSkillsShSource:
         assert bundle.files["SKILL.md"] == "# react"
         assert mock_get.called
 
+    @patch("tools.skills_hub._write_index_cache")
+    @patch("tools.skills_hub._read_index_cache", return_value=None)
+    @patch.object(SkillsShSource, "_discover_identifier")
+    @patch.object(SkillsShSource, "_fetch_detail_page")
+    @patch.object(GitHubSource, "fetch")
+    def test_fetch_downloads_only_the_resolved_identifier(
+        self,
+        mock_fetch,
+        mock_detail,
+        mock_discover,
+        _mock_read_cache,
+        _mock_write_cache,
+    ):
+        resolved_identifier = "owner/repo/product-team/product-designer"
+        mock_detail.return_value = {"repo": "owner/repo", "install_skill": "product-designer"}
+        mock_discover.return_value = resolved_identifier
+        resolved_bundle = SkillBundle(
+            name="product-designer",
+            files={"SKILL.md": "# Product Designer"},
+            source="github",
+            identifier=resolved_identifier,
+            trust_level="community",
+        )
+        mock_fetch.side_effect = lambda identifier: resolved_bundle if identifier == resolved_identifier else None
+
+        bundle = self._source().fetch("skills-sh/owner/repo/product-designer")
+
+        assert bundle is not None
+        assert bundle.identifier == "skills-sh/owner/repo/product-designer"
+        # All candidate identifiers are tried before falling back to discovery
+        assert mock_fetch.call_args_list[-1] == ((resolved_identifier,), {})
+        assert mock_fetch.call_args_list[0] == (("owner/repo/product-designer",), {})
+
     @patch("tools.skills_hub._write_index_cache")
     @patch("tools.skills_hub._read_index_cache", return_value=None)
     @patch("tools.skills_hub.httpx.get")
@@ -369,6 +440,36 @@ class TestSkillsShSource:
         # Verify the tree-resolved identifier was used for the final GitHub fetch
         mock_fetch.assert_any_call("owner/repo/cli-tool/components/skills/development/my-skill")
 
+    @patch.object(GitHubSource, "_find_skill_in_repo_tree")
+    @patch.object(GitHubSource, "_list_skills_in_repo")
+    @patch("tools.skills_hub.httpx.get")
+    def test_discover_identifier_uses_tree_search_before_root_scan(
+        self,
+        mock_get,
+        mock_list_skills,
+        mock_find_in_tree,
+    ):
+        root_url = "https://api.github.com/repos/owner/repo/contents/"
+        mock_list_skills.return_value = []
+        mock_find_in_tree.return_value = "owner/repo/product-team/product-designer"
+
+        def _httpx_get_side_effect(url, **kwargs):
+            resp = MagicMock()
+            if url == root_url:
+                resp.status_code = 200
+                resp.json = lambda: []
+                return resp
+            resp.status_code = 404
+            return resp
+
+        mock_get.side_effect = _httpx_get_side_effect
+
+        result = self._source()._discover_identifier("owner/repo/product-designer")
+
+        assert result == "owner/repo/product-team/product-designer"
+        requested_urls = [call.args[0] for call in mock_get.call_args_list]
+        assert root_url not in requested_urls
+
 
 class TestFindSkillInRepoTree:
     """Tests for GitHubSource._find_skill_in_repo_tree."""