fix(gateway): normalize step_callback prev_tools for backward compat

The PR changed prev_tools from list[str] to list[dict] with name/result
keys.  The gateway's _step_callback_sync passed this directly to hooks
as 'tool_names', breaking user-authored hooks that call
', '.join(tool_names).

Now:
- 'tool_names' always contains strings (backward-compatible)
- 'tools' carries the enriched dicts for hooks that want results

Also adds summary logging to register_mcp_servers() and comprehensive
tests for all three PR changes:
- sanitize_mcp_name_component edge cases
- register_mcp_servers public API
- _register_session_mcp_servers ACP integration
- step_callback result forwarding
- gateway normalization backward compat

tests/acp/test_events.py

@@ -205,6 +205,47 @@ class TestStepCallback:
         assert "read_file" not in tool_call_ids
         mock_rcts.assert_called_once()
 
+    def test_result_passed_to_build_tool_complete(self, mock_conn, event_loop_fixture):
+        """Tool result from prev_tools dict is forwarded to build_tool_complete."""
+        from collections import deque
+
+        tool_call_ids = {"terminal": deque(["tc-xyz789"])}
+        loop = event_loop_fixture
+
+        cb = make_step_cb(mock_conn, "session-1", loop, tool_call_ids)
+
+        with patch("acp_adapter.events.asyncio.run_coroutine_threadsafe") as mock_rcts, \
+             patch("acp_adapter.events.build_tool_complete") as mock_btc:
+            future = MagicMock(spec=Future)
+            future.result.return_value = None
+            mock_rcts.return_value = future
+
+            # Provide a result string in the tool info dict
+            cb(1, [{"name": "terminal", "result": '{"output": "hello"}'}])
+
+        mock_btc.assert_called_once_with(
+            "tc-xyz789", "terminal", result='{"output": "hello"}'
+        )
+
+    def test_none_result_passed_through(self, mock_conn, event_loop_fixture):
+        """When result is None (e.g. first iteration), None is passed through."""
+        from collections import deque
+
+        tool_call_ids = {"web_search": deque(["tc-aaa"])}
+        loop = event_loop_fixture
+
+        cb = make_step_cb(mock_conn, "session-1", loop, tool_call_ids)
+
+        with patch("acp_adapter.events.asyncio.run_coroutine_threadsafe") as mock_rcts, \
+             patch("acp_adapter.events.build_tool_complete") as mock_btc:
+            future = MagicMock(spec=Future)
+            future.result.return_value = None
+            mock_rcts.return_value = future
+
+            cb(1, [{"name": "web_search", "result": None}])
+
+        mock_btc.assert_called_once_with("tc-aaa", "web_search", result=None)
+
 
 # ---------------------------------------------------------------------------
 # Message callback

tests/acp/test_server.py

@@ -505,3 +505,179 @@ class TestSlashCommands:
         assert state.agent.provider == "anthropic"
         assert state.agent.base_url == "https://anthropic.example/v1"
         assert runtime_calls[-1] == "anthropic"
+
+
+# ---------------------------------------------------------------------------
+# _register_session_mcp_servers
+# ---------------------------------------------------------------------------
+
+
+class TestRegisterSessionMcpServers:
+    """Tests for ACP MCP server registration in session lifecycle."""
+
+    @pytest.mark.asyncio
+    async def test_noop_when_no_servers(self, agent, mock_manager):
+        """No-op when mcp_servers is None or empty."""
+        state = mock_manager.create_session(cwd="/tmp")
+        # Should not raise
+        await agent._register_session_mcp_servers(state, None)
+        await agent._register_session_mcp_servers(state, [])
+
+    @pytest.mark.asyncio
+    async def test_registers_stdio_servers(self, agent, mock_manager):
+        """McpServerStdio servers are converted and passed to register_mcp_servers."""
+        from acp.schema import McpServerStdio, EnvVariable
+
+        state = mock_manager.create_session(cwd="/tmp")
+        # Give the mock agent the attributes _register_session_mcp_servers reads
+        state.agent.enabled_toolsets = ["hermes-acp"]
+        state.agent.disabled_toolsets = None
+        state.agent.tools = []
+        state.agent.valid_tool_names = set()
+
+        server = McpServerStdio(
+            name="test-server",
+            command="/usr/bin/test",
+            args=["--flag"],
+            env=[EnvVariable(name="KEY", value="val")],
+        )
+
+        registered_config = {}
+        def capture_register(config_map):
+            registered_config.update(config_map)
+            return ["mcp_test_server_tool1"]
+
+        with patch("tools.mcp_tool.register_mcp_servers", side_effect=capture_register), \
+             patch("model_tools.get_tool_definitions", return_value=[]):
+            await agent._register_session_mcp_servers(state, [server])
+
+        assert "test-server" in registered_config
+        cfg = registered_config["test-server"]
+        assert cfg["command"] == "/usr/bin/test"
+        assert cfg["args"] == ["--flag"]
+        assert cfg["env"] == {"KEY": "val"}
+
+    @pytest.mark.asyncio
+    async def test_registers_http_servers(self, agent, mock_manager):
+        """McpServerHttp servers are converted correctly."""
+        from acp.schema import McpServerHttp, HttpHeader
+
+        state = mock_manager.create_session(cwd="/tmp")
+        state.agent.enabled_toolsets = ["hermes-acp"]
+        state.agent.disabled_toolsets = None
+        state.agent.tools = []
+        state.agent.valid_tool_names = set()
+
+        server = McpServerHttp(
+            name="http-server",
+            url="https://api.example.com/mcp",
+            headers=[HttpHeader(name="Authorization", value="Bearer tok")],
+        )
+
+        registered_config = {}
+        def capture_register(config_map):
+            registered_config.update(config_map)
+            return []
+
+        with patch("tools.mcp_tool.register_mcp_servers", side_effect=capture_register), \
+             patch("model_tools.get_tool_definitions", return_value=[]):
+            await agent._register_session_mcp_servers(state, [server])
+
+        assert "http-server" in registered_config
+        cfg = registered_config["http-server"]
+        assert cfg["url"] == "https://api.example.com/mcp"
+        assert cfg["headers"] == {"Authorization": "Bearer tok"}
+
+    @pytest.mark.asyncio
+    async def test_refreshes_agent_tool_surface(self, agent, mock_manager):
+        """After MCP registration, agent.tools and valid_tool_names are refreshed."""
+        from acp.schema import McpServerStdio
+
+        state = mock_manager.create_session(cwd="/tmp")
+        state.agent.enabled_toolsets = ["hermes-acp"]
+        state.agent.disabled_toolsets = None
+        state.agent.tools = []
+        state.agent.valid_tool_names = set()
+        state.agent._cached_system_prompt = "old prompt"
+
+        server = McpServerStdio(
+            name="srv",
+            command="/bin/test",
+            args=[],
+            env=[],
+        )
+
+        fake_tools = [
+            {"function": {"name": "mcp_srv_search"}},
+            {"function": {"name": "terminal"}},
+        ]
+
+        with patch("tools.mcp_tool.register_mcp_servers", return_value=["mcp_srv_search"]), \
+             patch("model_tools.get_tool_definitions", return_value=fake_tools):
+            await agent._register_session_mcp_servers(state, [server])
+
+        assert state.agent.tools == fake_tools
+        assert state.agent.valid_tool_names == {"mcp_srv_search", "terminal"}
+        # _invalidate_system_prompt should have been called
+        state.agent._invalidate_system_prompt.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_register_failure_logs_warning(self, agent, mock_manager):
+        """If register_mcp_servers raises, warning is logged but no crash."""
+        from acp.schema import McpServerStdio
+
+        state = mock_manager.create_session(cwd="/tmp")
+        server = McpServerStdio(
+            name="bad",
+            command="/nonexistent",
+            args=[],
+            env=[],
+        )
+
+        with patch("tools.mcp_tool.register_mcp_servers", side_effect=RuntimeError("boom")):
+            # Should not raise
+            await agent._register_session_mcp_servers(state, [server])
+
+    @pytest.mark.asyncio
+    async def test_new_session_calls_register(self, agent, mock_manager):
+        """new_session passes mcp_servers to _register_session_mcp_servers."""
+        with patch.object(agent, "_register_session_mcp_servers", new_callable=AsyncMock) as mock_reg:
+            resp = await agent.new_session(cwd="/tmp", mcp_servers=["fake"])
+            assert resp is not None
+            mock_reg.assert_called_once()
+            # Second arg should be the mcp_servers list
+            assert mock_reg.call_args[0][1] == ["fake"]
+
+    @pytest.mark.asyncio
+    async def test_load_session_calls_register(self, agent, mock_manager):
+        """load_session passes mcp_servers to _register_session_mcp_servers."""
+        # Create a session first so load can find it
+        state = mock_manager.create_session(cwd="/tmp")
+        sid = state.session_id
+
+        with patch.object(agent, "_register_session_mcp_servers", new_callable=AsyncMock) as mock_reg:
+            resp = await agent.load_session(cwd="/tmp", session_id=sid, mcp_servers=["fake"])
+            assert resp is not None
+            mock_reg.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_resume_session_calls_register(self, agent, mock_manager):
+        """resume_session passes mcp_servers to _register_session_mcp_servers."""
+        state = mock_manager.create_session(cwd="/tmp")
+        sid = state.session_id
+
+        with patch.object(agent, "_register_session_mcp_servers", new_callable=AsyncMock) as mock_reg:
+            resp = await agent.resume_session(cwd="/tmp", session_id=sid, mcp_servers=["fake"])
+            assert resp is not None
+            mock_reg.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_fork_session_calls_register(self, agent, mock_manager):
+        """fork_session passes mcp_servers to _register_session_mcp_servers."""
+        state = mock_manager.create_session(cwd="/tmp")
+        sid = state.session_id
+
+        with patch.object(agent, "_register_session_mcp_servers", new_callable=AsyncMock) as mock_reg:
+            resp = await agent.fork_session(cwd="/tmp", session_id=sid, mcp_servers=["fake"])
+            assert resp is not None
+            mock_reg.assert_called_once()