feat(acp): support permission mode switching

pyproject.toml

@@ -5,7 +5,7 @@ description = "Kimi Code CLI is your next CLI agent."
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
-    "agent-client-protocol==0.8.0",
+    "agent-client-protocol==0.10.0",
     "aiofiles>=24.0,<26.0",
     "aiohttp==3.13.3",
     "typer==0.21.1",

src/kimi_cli/acp/AGENTS.md

@@ -31,7 +31,9 @@
   - MCP servers passed by ACP are converted via `acp_mcp_servers_to_mcp_config`.
 - `session/load`
   - Multi-session only: loads by `Session.find`, then builds `KimiCLI` and `ACPSession`.
-  - No history replay yet (TODO).
+  - Replays persisted `wire.jsonl` history as ACP `session/update` notifications, falls back to
+    context text history for older sessions without wire history, and returns initial modes/models
+    state.
   - Single-session: not implemented.
 - `session/list`
   - Multi-session only: lists sessions via `Session.list`, no pagination.
@@ -71,10 +73,14 @@
     then releases the terminal handle.
 - Approval requests in the core tool system are bridged to ACP
   `session/request_permission` with allow-once/allow-always/reject options.
+- Permission mode switching is exposed through ACP `session/set_mode`.
+  - `default` asks for permissions as needed.
+  - `yolo` toggles Kimi's persisted yolo approval state and resolves pending approvals.
 
 ## Current gaps / not implemented
 - `authenticate` method (not used by current Zed ACP client).
-- `session/set_mode` and `session/set_model` (no multi-mode/model switching in kimi-cli).
+- `session/set_model` currently supports configured Kimi models, but there is no agent/model
+  provider discovery beyond the local config.
 - `ext_method` / `ext_notification` for custom ACP extensions are stubbed.
 - Single-session server does not implement `session/load` or `session/list`.
 

src/kimi_cli/acp/server.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import sys
 import time
 from datetime import datetime
 from pathlib import Path
@@ -26,14 +25,31 @@
 from kimi_cli.soul.toolset import KimiToolset
 from kimi_cli.utils.logging import logger
 
+_DEFAULT_MODE_ID = "default"
+_YOLO_MODE_ID = "yolo"
+_SESSION_MODES = [
+    acp.schema.SessionMode(
+        id=_DEFAULT_MODE_ID,
+        name="Default",
+        description="Ask for permission before running actions that require approval.",
+    ),
+    acp.schema.SessionMode(
+        id=_YOLO_MODE_ID,
+        name="Yolo",
+        description="Automatically approve actions for this session.",
+    ),
+]
+
 
 class ACPServer:
     def __init__(self) -> None:
         self.client_capabilities: acp.schema.ClientCapabilities | None = None
         self.conn: acp.Client | None = None
         self.sessions: dict[str, tuple[ACPSession, _ModelIDConv]] = {}
         self.negotiated_version: ACPVersionSpec | None = None
-        self._auth_methods: list[acp.schema.AuthMethod] = []
+        self._auth_methods: list[
+            acp.schema.EnvVarAuthMethod | acp.schema.TerminalAuthMethod | acp.schema.AuthMethodAgent
+        ] = []
 
     def on_connect(self, conn: acp.Client) -> None:
         logger.info("ACP client connected")
@@ -66,32 +82,18 @@ async def initialize(
                 version=getattr(client_info, "version", None),
             )
 
-        # get command and args of current process for terminal-auth
-        command = sys.argv[0]
-        args: list[str] = []
-
-        # Build terminal auth data for error response
-        terminal_args = args + ["login"]
-
         # Build and cache auth methods for reuse in AUTH_REQUIRED errors
         self._auth_methods = [
-            acp.schema.AuthMethod(
+            acp.schema.TerminalAuthMethod(
                 id="login",
+                type="terminal",
                 name="Login with Kimi account",
                 description=(
                     "Run `kimi login` command in the terminal, "
                     "then follow the instructions to finish login."
                 ),
-                # Store auth data in field_meta for building AUTH_REQUIRED error
-                field_meta={
-                    "terminal-auth": {
-                        "command": command,
-                        "args": terminal_args,
-                        "label": "Kimi Code Login",
-                        "env": {},
-                        "type": "terminal",
-                    }
-                },
+                args=["login"],
+                env={},
             ),
         ]
 
@@ -129,20 +131,9 @@ def _check_auth(self) -> None:
         """Check if Kimi Code authentication is complete. Raise AUTH_REQUIRED if not."""
         reason = self._check_token_usable()
         if reason:
-            auth_methods_data: list[dict[str, Any]] = []
-            for m in self._auth_methods:
-                if m.field_meta and "terminal-auth" in m.field_meta:
-                    terminal_auth = m.field_meta["terminal-auth"]
-                    auth_methods_data.append(
-                        {
-                            "id": m.id,
-                            "name": m.name,
-                            "description": m.description,
-                            "type": terminal_auth.get("type", "terminal"),
-                            "args": terminal_auth.get("args", []),
-                            "env": terminal_auth.get("env", {}),
-                        }
-                    )
+            auth_methods_data = [
+                m.model_dump(by_alias=True, exclude_none=True) for m in self._auth_methods
+            ]
 
             logger.warning("Authentication required, {reason}", reason=reason)
             raise acp.RequestError.auth_required({"authMethods": auth_methods_data})
@@ -195,16 +186,7 @@ async def new_session(
         )
         return acp.NewSessionResponse(
             session_id=session.id,
-            modes=acp.schema.SessionModeState(
-                available_modes=[
-                    acp.schema.SessionMode(
-                        id="default",
-                        name="Default",
-                        description="The default mode.",
-                    ),
-                ],
-                current_mode_id="default",
-            ),
+            modes=_session_mode_state(acp_session),
             models=acp.schema.SessionModelState(
                 available_models=_expand_llm_models(config.models),
                 current_model_id=model_id_conv.to_acp_model_id(),
@@ -255,18 +237,33 @@ async def _setup_session(
 
     async def load_session(
         self, cwd: str, session_id: str, mcp_servers: list[MCPServer] | None = None, **kwargs: Any
-    ) -> None:
+    ) -> acp.schema.LoadSessionResponse:
         logger.info("Loading session: {id} for working directory: {cwd}", id=session_id, cwd=cwd)
 
         if session_id in self.sessions:
             logger.warning("Session already loaded: {id}", id=session_id)
-            return
+            acp_session, model_id_conv = self.sessions[session_id]
+        else:
+            # Check authentication before loading session
+            self._check_auth()
 
-        # Check authentication before loading session
-        self._check_auth()
+            acp_session, model_id_conv = await self._setup_session(cwd, session_id, mcp_servers)
 
-        await self._setup_session(cwd, session_id, mcp_servers)
-        # TODO: replay session history?
+        replayed_updates = await acp_session.replay_history()
+        logger.info(
+            "Replayed {count} ACP history updates for session: {id}",
+            count=replayed_updates,
+            id=session_id,
+        )
+
+        config = acp_session.cli.soul.runtime.config
+        return acp.schema.LoadSessionResponse(
+            modes=_session_mode_state(acp_session),
+            models=acp.schema.SessionModelState(
+                available_models=_expand_llm_models(config.models),
+                current_model_id=model_id_conv.to_acp_model_id(),
+            ),
+        )
 
     async def resume_session(
         self, cwd: str, session_id: str, mcp_servers: list[MCPServer] | None = None, **kwargs: Any
@@ -279,16 +276,7 @@ async def resume_session(
         acp_session, model_id_conv = self.sessions[session_id]
         config = acp_session.cli.soul.runtime.config
         return acp.schema.ResumeSessionResponse(
-            modes=acp.schema.SessionModeState(
-                available_modes=[
-                    acp.schema.SessionMode(
-                        id="default",
-                        name="Default",
-                        description="The default mode.",
-                    ),
-                ],
-                current_mode_id="default",
-            ),
+            modes=_session_mode_state(acp_session),
             models=acp.schema.SessionModelState(
                 available_models=_expand_llm_models(config.models),
                 current_model_id=model_id_conv.to_acp_model_id(),
@@ -321,8 +309,41 @@ async def list_sessions(
             next_cursor=None,
         )
 
-    async def set_session_mode(self, mode_id: str, session_id: str, **kwargs: Any) -> None:
-        assert mode_id == "default", "Only default mode is supported"
+    async def set_session_mode(
+        self, mode_id: str, session_id: str, **kwargs: Any
+    ) -> acp.schema.SetSessionModeResponse:
+        logger.info(
+            "Setting session mode to {mode_id} for session: {id}",
+            mode_id=mode_id,
+            id=session_id,
+        )
+        if session_id not in self.sessions:
+            logger.error("Session not found: {id}", id=session_id)
+            raise acp.RequestError.invalid_params({"session_id": "Session not found"})
+        if mode_id not in {_DEFAULT_MODE_ID, _YOLO_MODE_ID}:
+            logger.error("Mode not found: {mode_id}", mode_id=mode_id)
+            raise acp.RequestError.invalid_params({"mode_id": "Mode not found"})
+
+        acp_session, _ = self.sessions[session_id]
+        runtime = acp_session.cli.soul.runtime
+        enable_yolo = mode_id == _YOLO_MODE_ID
+        if runtime.approval.is_yolo() != enable_yolo:
+            runtime.approval.set_yolo(enable_yolo)
+
+        if enable_yolo and runtime.approval_runtime is not None:
+            for request in runtime.approval_runtime.list_pending():
+                runtime.approval_runtime.resolve(request.id, "approve")
+
+        if self.conn is not None:
+            await self.conn.session_update(
+                session_id=session_id,
+                update=acp.schema.CurrentModeUpdate(
+                    session_update="current_mode_update",
+                    current_mode_id=_current_mode_id(acp_session),
+                ),
+            )
+
+        return acp.schema.SetSessionModeResponse()
 
     async def set_session_model(self, model_id: str, session_id: str, **kwargs: Any) -> None:
         logger.info(
@@ -390,7 +411,10 @@ async def authenticate(self, method_id: str, **kwargs: Any) -> acp.AuthenticateR
                 raise acp.RequestError.auth_required(
                     {
                         "message": "Please complete login in terminal first",
-                        "authMethods": self._auth_methods,
+                        "authMethods": [
+                            m.model_dump(by_alias=True, exclude_none=True)
+                            for m in self._auth_methods
+                        ],
                     }
                 )
 
@@ -466,3 +490,16 @@ def _expand_llm_models(models: dict[str, LLMModel]) -> list[acp.schema.ModelInfo
                     )
                 )
     return expanded_models
+
+
+def _current_mode_id(acp_session: ACPSession) -> str:
+    if acp_session.cli.soul.runtime.approval.is_yolo():
+        return _YOLO_MODE_ID
+    return _DEFAULT_MODE_ID
+
+
+def _session_mode_state(acp_session: ACPSession) -> acp.schema.SessionModeState:
+    return acp.schema.SessionModeState(
+        available_modes=_SESSION_MODES,
+        current_mode_id=_current_mode_id(acp_session),
+    )