feat(security): Add shell command security analysis to approval workflow

src/kimi_cli/acp/convert.py

@@ -12,6 +12,8 @@
     TextPart,
     ToolReturnValue,
 )
+from kosong.tooling import BriefDisplayBlock
+from kimi_cli.tools.display import ShellDisplayBlock
 
 
 def acp_blocks_to_content_parts(prompt: list[ACPContentBlock]) -> list[ContentPart]:
@@ -52,7 +54,7 @@ def acp_blocks_to_content_parts(prompt: list[ACPContentBlock]) -> list[ContentPa
 
 def display_block_to_acp_content(
     block: DisplayBlock,
-) -> acp.schema.FileEditToolCallContent | None:
+) -> acp.schema.FileEditToolCallContent | acp.schema.ContentToolCallContent | None:
     if isinstance(block, DiffDisplayBlock):
         return acp.schema.FileEditToolCallContent(
             type="diff",
@@ -61,6 +63,18 @@ def display_block_to_acp_content(
             new_text=block.new_text,
         )
 
+    if isinstance(block, ShellDisplayBlock):
+        return acp.schema.ContentToolCallContent(
+            type="content",
+            content=acp.schema.TextContentBlock(type="text", text=block.command),
+        )
+
+    if isinstance(block, BriefDisplayBlock) and block.text:
+        return acp.schema.ContentToolCallContent(
+            type="content",
+            content=acp.schema.TextContentBlock(type="text", text=block.text),
+        )
+
     return None
 
 
@@ -106,6 +120,12 @@ def _to_text_block(text: str) -> acp.schema.ContentToolCallContent:
             # return early to indicate no output should be shown
             return []
 
+        if isinstance(block, BriefDisplayBlock):
+            # Brief messages are UI-only summaries ("Plan approved", etc.);
+            # they were previously dropped here and should not be forwarded
+            # to ACP clients as content blocks.
+            continue
+
         content = display_block_to_acp_content(block)
         if content is not None:
             contents.append(content)

src/kimi_cli/exception.py

@@ -1,28 +1,97 @@
 from __future__ import annotations
 
+from typing import Any
+
 
 class KimiCLIException(Exception):
-    """Base exception class for Kimi Code CLI."""
+    """Base exception class for Kimi Code CLI.
 
-    pass
+    Provides structured error context for better debugging and user feedback.
+    """
 
+    def __init__(self, message: str, *, context: dict[str, Any] | None = None):
+        super().__init__(message)
+        self.message = message
+        self.context = context or {}
 
-class ConfigError(KimiCLIException, ValueError):
-    """Configuration error."""
+    def __str__(self) -> str:
+        if self.context:
+            context_str = ", ".join(f"{k}={v!r}" for k, v in self.context.items())
+            return f"{self.message} ({context_str})"
+        return self.message
 
-    pass
 
+class ConfigError(KimiCLIException, ValueError):
+    """Configuration error.
+
+    Attributes:
+        config_path: Path to the configuration file that caused the error, if applicable.
+        field: Specific configuration field that failed validation, if known.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        config_path: str | None = None,
+        field: str | None = None,
+        context: dict[str, Any] | None = None,
+    ):
+        ctx = context or {}
+        if config_path:
+            ctx["config_path"] = config_path
+        if field:
+            ctx["field"] = field
+        super().__init__(message, context=ctx)
+        self.config_path = config_path
+        self.field = field
 
-class AgentSpecError(KimiCLIException, ValueError):
-    """Agent specification error."""
 
-    pass
+class AgentSpecError(KimiCLIException, ValueError):
+    """Agent specification error.
+
+    Attributes:
+        agent_file: Path to the agent specification file, if applicable.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        agent_file: str | None = None,
+        context: dict[str, Any] | None = None,
+    ):
+        ctx = context or {}
+        if agent_file:
+            ctx["agent_file"] = agent_file
+        super().__init__(message, context=ctx)
+        self.agent_file = agent_file
 
 
 class InvalidToolError(KimiCLIException, ValueError):
-    """Invalid tool error."""
-
-    pass
+    """Invalid tool error.
+
+    Attributes:
+        tool_name: Name of the invalid tool.
+        reason: Specific reason why the tool is invalid.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        tool_name: str | None = None,
+        reason: str | None = None,
+        context: dict[str, Any] | None = None,
+    ):
+        ctx = context or {}
+        if tool_name:
+            ctx["tool_name"] = tool_name
+        if reason:
+            ctx["reason"] = reason
+        super().__init__(message, context=ctx)
+        self.tool_name = tool_name
+        self.reason = reason
 
 
 class SystemPromptTemplateError(KimiCLIException, ValueError):
@@ -32,12 +101,47 @@ class SystemPromptTemplateError(KimiCLIException, ValueError):
 
 
 class MCPConfigError(KimiCLIException, ValueError):
-    """MCP config error."""
-
-    pass
+    """MCP config error.
+
+    Attributes:
+        server_name: Name of the MCP server with configuration issues.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        server_name: str | None = None,
+        context: dict[str, Any] | None = None,
+    ):
+        ctx = context or {}
+        if server_name:
+            ctx["server_name"] = server_name
+        super().__init__(message, context=ctx)
+        self.server_name = server_name
 
 
 class MCPRuntimeError(KimiCLIException, RuntimeError):
-    """MCP runtime error."""
-
-    pass
+    """MCP runtime error.
+
+    Attributes:
+        server_name: Name of the MCP server that encountered an error.
+        exit_code: Process exit code if the server process terminated.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        server_name: str | None = None,
+        exit_code: int | None = None,
+        context: dict[str, Any] | None = None,
+    ):
+        ctx = context or {}
+        if server_name:
+            ctx["server_name"] = server_name
+        if exit_code is not None:
+            ctx["exit_code"] = exit_code
+        super().__init__(message, context=ctx)
+        self.server_name = server_name
+        self.exit_code = exit_code

src/kimi_cli/tools/shell/__init__.py

@@ -5,7 +5,7 @@
 
 import kaos
 from kaos import AsyncReadable
-from kosong.tooling import CallableTool2, ToolReturnValue
+from kosong.tooling import BriefDisplayBlock, CallableTool2, ToolReturnValue
 from pydantic import BaseModel, Field, model_validator
 
 from kimi_cli.background import TaskView, format_task
@@ -14,6 +14,7 @@
 from kimi_cli.soul.toolset import get_current_tool_call_or_none
 from kimi_cli.tools.display import BackgroundTaskDisplayBlock, ShellDisplayBlock
 from kimi_cli.tools.utils import ToolResultBuilder, load_desc
+from kimi_cli.utils.command_security import analyze_command, format_security_notes
 from kimi_cli.utils.environment import Environment
 from kimi_cli.utils.subprocess_env import get_noninteractive_env
 
@@ -82,16 +83,22 @@ async def __call__(self, params: Params) -> ToolReturnValue:
         if params.run_in_background:
             return await self._run_in_background(params)
 
+        # Analyze command for security-relevant patterns
+        security_notes = analyze_command(params.command)
+        display: list[ShellDisplayBlock | BriefDisplayBlock] = [
+            ShellDisplayBlock(
+                language="powershell" if self._is_powershell else "bash",
+                command=params.command,
+            )
+        ]
+        if security_notes:
+            display.append(BriefDisplayBlock(text=format_security_notes(security_notes)))
+
         result = await self._approval.request(
             self.name,
             "run command",
             f"Run command `{params.command}`",
-            display=[
-                ShellDisplayBlock(
-                    language="powershell" if self._is_powershell else "bash",
-                    command=params.command,
-                )
-            ],
+            display=display,
         )
         if not result:
             return result.rejection_error()
@@ -130,16 +137,22 @@ async def _run_in_background(self, params: Params) -> ToolReturnValue:
                 brief="No tool call context",
             )
 
+        # Analyze command for security-relevant patterns
+        security_notes = analyze_command(params.command)
+        bg_display: list[ShellDisplayBlock | BriefDisplayBlock] = [
+            ShellDisplayBlock(
+                language="powershell" if self._is_powershell else "bash",
+                command=params.command,
+            )
+        ]
+        if security_notes:
+            bg_display.append(BriefDisplayBlock(text=format_security_notes(security_notes)))
+
         result = await self._approval.request(
             self.name,
             "run background command",
             f"Run background command `{params.command}`",
-            display=[
-                ShellDisplayBlock(
-                    language="powershell" if self._is_powershell else "bash",
-                    command=params.command,
-                )
-            ],
+            display=bg_display,
         )
         if not result:
             return result.rejection_error()
@@ -154,7 +167,7 @@ async def _run_in_background(self, params: Params) -> ToolReturnValue:
                 shell_path=str(self._shell_path),
                 cwd=str(self._runtime.session.work_dir),
             )
-        except Exception as exc:
+        except (OSError, RuntimeError) as exc:
             builder = ToolResultBuilder()
             return builder.error(f"Failed to start background task: {exc}", brief="Start failed")
 

src/kimi_cli/tools/utils.py

@@ -1,4 +1,5 @@
 import re
+from io import StringIO
 from pathlib import Path
 
 from jinja2 import Environment, Undefined
@@ -54,6 +55,10 @@ def truncate_line(line: str, max_length: int, marker: str = "...") -> str:
 class ToolResultBuilder:
     """
     Builder for tool results with character and line limits.
+
+    This builder efficiently accumulates tool output while enforcing
+    character and line length limits. It uses StringIO for memory-efficient
+    string building.
     """
 
     def __init__(
@@ -66,7 +71,7 @@ def __init__(
         self._marker = "[...truncated]"
         if max_line_length is not None:
             assert max_line_length > len(self._marker)
-        self._buffer: list[str] = []
+        self._buffer = StringIO()
         self._n_chars = 0
         self._n_lines = 0
         self._truncation_happened = False
@@ -91,6 +96,9 @@ def n_lines(self) -> int:
     def write(self, text: str) -> int:
         """
         Write text to the output buffer.
+
+        Text is truncated if it exceeds max_chars or if individual lines
+        exceed max_line_length.
 
         Returns:
             int: Number of characters actually written
@@ -119,7 +127,7 @@ def write(self, text: str) -> int:
             if line != original_line:
                 self._truncation_happened = True
 
-            self._buffer.append(line)
+            self._buffer.write(line)
             chars_written += len(line)
             self._n_chars += len(line)
             if line.endswith("\n"):
@@ -139,7 +147,7 @@ def extras(self, **extras: JsonType) -> None:
 
     def ok(self, message: str = "", *, brief: str = "") -> ToolReturnValue:
         """Create a ToolReturnValue with is_error=False and the current output."""
-        output = "".join(self._buffer)
+        output = self._buffer.getvalue()
 
         final_message = message
         if final_message and not final_message.endswith("."):
@@ -160,7 +168,7 @@ def ok(self, message: str = "", *, brief: str = "") -> ToolReturnValue:
 
     def error(self, message: str, *, brief: str) -> ToolReturnValue:
         """Create a ToolReturnValue with is_error=True and the current output."""
-        output = "".join(self._buffer)
+        output = self._buffer.getvalue()
 
         final_message = message
         if self._truncation_happened:

src/kimi_cli/ui/shell/__init__.py

@@ -537,8 +537,10 @@ def _handler():
                     kwargs["stderr"] = stderr
                 proc = await asyncio.create_subprocess_shell(command, env=get_clean_env(), **kwargs)
                 await proc.wait()
-        except Exception as e:
-            logger.exception("Failed to run shell command:")
+        except asyncio.CancelledError:
+            raise
+        except OSError as e:
+            logger.error("Failed to run shell command: {error}", error=e)
             console.print(f"[red]Failed to run shell command: {e}[/red]")
         finally:
             remove_sigint()