fix(acp): advertise slash commands via ACP protocol

Send AvailableCommandsUpdate on session create/load/resume/fork so ACP
clients (Zed, etc.) can discover /help, /model, /tools, /compact, etc.
Also rewrites /compact to use agent._compress_context() properly with
token estimation and session DB isolation.

Co-authored-by: NexVeridian <NexVeridian@users.noreply.github.com>

acp_adapter/server.py

@@ -13,6 +13,8 @@ from acp.schema import (
     AgentCapabilities,
     AuthenticateResponse,
     AuthMethodAgent,
+    AvailableCommand,
+    AvailableCommandsUpdate,
     ClientCapabilities,
     EmbeddedResourceContentBlock,
     ForkSessionResponse,
@@ -37,6 +39,7 @@ from acp.schema import (
     SessionListCapabilities,
     SessionInfo,
     TextContentBlock,
+    UnstructuredCommandInput,
     Usage,
 )
 
@@ -84,6 +87,48 @@ def _extract_text(
 class HermesACPAgent(acp.Agent):
     """ACP Agent implementation wrapping Hermes AIAgent."""
 
+    _SLASH_COMMANDS = {
+        "help": "Show available commands",
+        "model": "Show or change current model",
+        "tools": "List available tools",
+        "context": "Show conversation context info",
+        "reset": "Clear conversation history",
+        "compact": "Compress conversation context",
+        "version": "Show Hermes version",
+    }
+
+    _ADVERTISED_COMMANDS = (
+        {
+            "name": "help",
+            "description": "List available commands",
+        },
+        {
+            "name": "model",
+            "description": "Show current model and provider, or switch models",
+            "input_hint": "model name to switch to",
+        },
+        {
+            "name": "tools",
+            "description": "List available tools with descriptions",
+        },
+        {
+            "name": "context",
+            "description": "Show conversation message counts by role",
+        },
+        {
+            "name": "reset",
+            "description": "Clear conversation history",
+        },
+        {
+            "name": "compact",
+            "description": "Compress conversation context",
+        },
+        {
+            "name": "version",
+            "description": "Show Hermes version",
+        },
+    )
+
     def __init__(self, session_manager: SessionManager | None = None):
         super().__init__()
         self.session_manager = session_manager or SessionManager()
@@ -219,6 +264,7 @@ class HermesACPAgent(acp.Agent):
         state = self.session_manager.create_session(cwd=cwd)
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("New session %s (cwd=%s)", state.session_id, cwd)
+        self._schedule_available_commands_update(state.session_id)
         return NewSessionResponse(session_id=state.session_id)
 
     async def load_session(
@@ -234,6 +280,7 @@ class HermesACPAgent(acp.Agent):
             return None
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("Loaded session %s", session_id)
+        self._schedule_available_commands_update(session_id)
         return LoadSessionResponse()
 
     async def resume_session(
@@ -249,6 +296,7 @@ class HermesACPAgent(acp.Agent):
             state = self.session_manager.create_session(cwd=cwd)
         await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("Resumed session %s", state.session_id)
+        self._schedule_available_commands_update(state.session_id)
         return ResumeSessionResponse()
 
     async def cancel(self, session_id: str, **kwargs: Any) -> None:
@@ -274,6 +322,8 @@ class HermesACPAgent(acp.Agent):
         if state is not None:
             await self._register_session_mcp_servers(state, mcp_servers)
         logger.info("Forked session %s -> %s", session_id, new_id)
+        if new_id:
+            self._schedule_available_commands_update(new_id)
         return ForkSessionResponse(session_id=new_id)
 
     async def list_sessions(
@@ -411,15 +461,50 @@ class HermesACPAgent(acp.Agent):
 
     # ---- Slash commands (headless) -------------------------------------------
 
-    _SLASH_COMMANDS = {
+    @classmethod
-        "help": "Show available commands",
+    def _available_commands(cls) -> list[AvailableCommand]:
-        "model": "Show or change current model",
+        commands: list[AvailableCommand] = []
-        "tools": "List available tools",
+        for spec in cls._ADVERTISED_COMMANDS:
-        "context": "Show conversation context info",
+            input_hint = spec.get("input_hint")
-        "reset": "Clear conversation history",
+            commands.append(
-        "compact": "Compress conversation context",
+                AvailableCommand(
-        "version": "Show Hermes version",
+                    name=spec["name"],
-    }
+                    description=spec["description"],
+                    input=UnstructuredCommandInput(hint=input_hint)
+                    if input_hint
+                    else None,
+                )
+            )
+        return commands
+
+    async def _send_available_commands_update(self, session_id: str) -> None:
+        """Advertise supported slash commands to the connected ACP client."""
+        if not self._conn:
+            return
+
+        try:
+            await self._conn.session_update(
+                session_id=session_id,
+                update=AvailableCommandsUpdate(
+                    sessionUpdate="available_commands_update",
+                    availableCommands=self._available_commands(),
+                ),
+            )
+        except Exception:
+            logger.warning(
+                "Failed to advertise ACP slash commands for session %s",
+                session_id,
+                exc_info=True,
+            )
+
+    def _schedule_available_commands_update(self, session_id: str) -> None:
+        """Send the command advertisement after the session response is queued."""
+        if not self._conn:
+            return
+        loop = asyncio.get_running_loop()
+        loop.call_soon(
+            asyncio.create_task, self._send_available_commands_update(session_id)
+        )
 
     def _handle_slash_command(self, text: str, state: SessionState) -> str | None:
         """Dispatch a slash command and return the response text.
@@ -539,11 +624,39 @@ class HermesACPAgent(acp.Agent):
             return "Nothing to compress — conversation is empty."
         try:
             agent = state.agent
-            if hasattr(agent, "compress_context"):
+            if not getattr(agent, "compression_enabled", True):
-                agent.compress_context(state.history)
+                return "Context compression is disabled for this agent."
-                self.session_manager.save_session(state.session_id)
+            if not hasattr(agent, "_compress_context"):
-                return f"Context compressed. Messages: {len(state.history)}"
+                return "Context compression not available for this agent."
-            return "Context compression not available for this agent."
+
+            from agent.model_metadata import estimate_messages_tokens_rough
+
+            original_count = len(state.history)
+            approx_tokens = estimate_messages_tokens_rough(state.history)
+            original_session_db = getattr(agent, "_session_db", None)
+
+            try:
+                # ACP sessions must keep a stable session id, so avoid the
+                # SQLite session-splitting side effect inside _compress_context.
+                agent._session_db = None
+                compressed, _ = agent._compress_context(
+                    state.history,
+                    getattr(agent, "_cached_system_prompt", "") or "",
+                    approx_tokens=approx_tokens,
+                    task_id=state.session_id,
+                )
+            finally:
+                agent._session_db = original_session_db
+
+            state.history = compressed
+            self.session_manager.save_session(state.session_id)
+
+            new_count = len(state.history)
+            new_tokens = estimate_messages_tokens_rough(state.history)
+            return (
+                f"Context compressed: {original_count} -> {new_count} messages\n"
+                f"~{approx_tokens:,} -> ~{new_tokens:,} tokens"
+            )
         except Exception as e:
             return f"Compression failed: {e}"
 

tests/acp/test_server.py

@@ -12,6 +12,7 @@ from acp.agent.router import build_agent_router
 from acp.schema import (
     AgentCapabilities,
     AuthenticateResponse,
+    AvailableCommandsUpdate,
     Implementation,
     InitializeResponse,
     ListSessionsResponse,
@@ -113,6 +114,53 @@ class TestSessionOps:
         assert state is not None
         assert state.cwd == "/home/user/project"
 
+    @pytest.mark.asyncio
+    async def test_available_commands_include_help(self, agent):
+        help_cmd = next(
+            (cmd for cmd in agent._available_commands() if cmd.name == "help"),
+            None,
+        )
+
+        assert help_cmd is not None
+        assert help_cmd.description == "List available commands"
+        assert help_cmd.input is None
+
+    @pytest.mark.asyncio
+    async def test_send_available_commands_update(self, agent):
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        await agent._send_available_commands_update("session-123")
+
+        mock_conn.session_update.assert_awaited_once()
+        call = mock_conn.session_update.await_args
+        assert call.kwargs["session_id"] == "session-123"
+        update = call.kwargs["update"]
+        assert isinstance(update, AvailableCommandsUpdate)
+        assert update.session_update == "available_commands_update"
+        assert [cmd.name for cmd in update.available_commands] == [
+            "help",
+            "model",
+            "tools",
+            "context",
+            "reset",
+            "compact",
+            "version",
+        ]
+        model_cmd = next(
+            cmd for cmd in update.available_commands if cmd.name == "model"
+        )
+        assert model_cmd.input is not None
+        assert model_cmd.input.root.hint == "model name to switch to"
+
+    @pytest.mark.asyncio
+    async def test_new_session_schedules_available_commands_update(self, agent):
+        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
+            resp = await agent.new_session(cwd="/home/user/project")
+
+        mock_schedule.assert_called_once_with(resp.session_id)
+
     @pytest.mark.asyncio
     async def test_cancel_sets_event(self, agent):
         resp = await agent.new_session(cwd=".")
@@ -132,6 +180,15 @@ class TestSessionOps:
         load_resp = await agent.load_session(cwd="/tmp", session_id=resp.session_id)
         assert isinstance(load_resp, LoadSessionResponse)
 
+    @pytest.mark.asyncio
+    async def test_load_session_schedules_available_commands_update(self, agent):
+        resp = await agent.new_session(cwd="/tmp")
+        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
+            load_resp = await agent.load_session(cwd="/tmp", session_id=resp.session_id)
+
+        assert isinstance(load_resp, LoadSessionResponse)
+        mock_schedule.assert_called_once_with(resp.session_id)
+
     @pytest.mark.asyncio
     async def test_load_session_not_found_returns_none(self, agent):
         resp = await agent.load_session(cwd="/tmp", session_id="bogus")
@@ -143,6 +200,15 @@ class TestSessionOps:
         resume_resp = await agent.resume_session(cwd="/tmp", session_id=resp.session_id)
         assert isinstance(resume_resp, ResumeSessionResponse)
 
+    @pytest.mark.asyncio
+    async def test_resume_session_schedules_available_commands_update(self, agent):
+        resp = await agent.new_session(cwd="/tmp")
+        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
+            resume_resp = await agent.resume_session(cwd="/tmp", session_id=resp.session_id)
+
+        assert isinstance(resume_resp, ResumeSessionResponse)
+        mock_schedule.assert_called_once_with(resp.session_id)
+
     @pytest.mark.asyncio
     async def test_resume_session_creates_new_if_missing(self, agent):
         resume_resp = await agent.resume_session(cwd="/tmp", session_id="nonexistent")
@@ -170,6 +236,15 @@ class TestListAndFork:
         assert fork_resp.session_id
         assert fork_resp.session_id != new_resp.session_id
 
+    @pytest.mark.asyncio
+    async def test_fork_session_schedules_available_commands_update(self, agent):
+        new_resp = await agent.new_session(cwd="/original")
+        with patch.object(agent, "_schedule_available_commands_update") as mock_schedule:
+            fork_resp = await agent.fork_session(cwd="/forked", session_id=new_resp.session_id)
+
+        assert fork_resp.session_id
+        mock_schedule.assert_called_once_with(fork_resp.session_id)
+
 
 # ---------------------------------------------------------------------------
 # session configuration / model routing
@@ -427,6 +502,55 @@ class TestSlashCommands:
         result = agent._handle_slash_command("/version", state)
         assert HERMES_VERSION in result
 
+    def test_compact_compresses_context(self, agent, mock_manager):
+        state = self._make_state(mock_manager)
+        state.history = [
+            {"role": "user", "content": "one"},
+            {"role": "assistant", "content": "two"},
+            {"role": "user", "content": "three"},
+            {"role": "assistant", "content": "four"},
+        ]
+        state.agent.compression_enabled = True
+        state.agent._cached_system_prompt = "system"
+        original_session_db = object()
+        state.agent._session_db = original_session_db
+
+        def _compress_context(messages, system_prompt, *, approx_tokens, task_id):
+            assert state.agent._session_db is None
+            assert messages == state.history
+            assert system_prompt == "system"
+            assert approx_tokens == 40
+            assert task_id == state.session_id
+            return [{"role": "user", "content": "summary"}], "new-system"
+
+        state.agent._compress_context = MagicMock(side_effect=_compress_context)
+
+        with (
+            patch.object(agent.session_manager, "save_session") as mock_save,
+            patch(
+                "agent.model_metadata.estimate_messages_tokens_rough",
+                side_effect=[40, 12],
+            ),
+        ):
+            result = agent._handle_slash_command("/compact", state)
+
+        assert "Context compressed: 4 -> 1 messages" in result
+        assert "~40 -> ~12 tokens" in result
+        assert state.history == [{"role": "user", "content": "summary"}]
+        assert state.agent._session_db is original_session_db
+        state.agent._compress_context.assert_called_once_with(
+            [
+                {"role": "user", "content": "one"},
+                {"role": "assistant", "content": "two"},
+                {"role": "user", "content": "three"},
+                {"role": "assistant", "content": "four"},
+            ],
+            "system",
+            approx_tokens=40,
+            task_id=state.session_id,
+        )
+        mock_save.assert_called_once_with(state.session_id)
+
     def test_unknown_command_returns_none(self, agent, mock_manager):
         state = self._make_state(mock_manager)
         result = agent._handle_slash_command("/nonexistent", state)
@@ -436,7 +560,8 @@ class TestSlashCommands:
     async def test_slash_command_intercepted_in_prompt(self, agent, mock_manager):
         """Slash commands should be handled without calling the LLM."""
         new_resp = await agent.new_session(cwd="/tmp")
-        mock_conn = AsyncMock(spec=acp.Client)
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
         agent._conn = mock_conn
 
         prompt = [TextContentBlock(type="text", text="/help")]
@@ -449,7 +574,9 @@ class TestSlashCommands:
     async def test_unknown_slash_falls_through_to_llm(self, agent, mock_manager):
         """Unknown /commands should be sent to the LLM, not intercepted."""
         new_resp = await agent.new_session(cwd="/tmp")
-        mock_conn = AsyncMock(spec=acp.Client)
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        mock_conn.request_permission = AsyncMock(return_value=None)
         agent._conn = mock_conn
 
         # Mock run_in_executor to avoid actually running the agent