fix: address review edge cases in shell capture and cd handling

- Windows: fall back to pipe capture when PTY is unavailable
- PTY: catch asyncio.CancelledError alongside KeyboardInterrupt
  to prevent Ctrl+C from tearing down the shell loop
- Security: neutralise <system-reminder> tags in captured output
  to prevent injection via untrusted shell output
- cd: preserve shell expansion for ~, $VAR, and - targets instead
  of quoting them with shlex.quote

Author: n-WN

src/kimi_cli/ui/shell/__init__.py

@@ -4,6 +4,7 @@
 import contextlib
 import os
 import shlex
+import sys
 import time
 from collections import deque
 from collections.abc import Awaitable, Callable, Coroutine
@@ -23,7 +24,11 @@
 from kimi_cli.soul import LLMNotSet, LLMNotSupported, MaxStepsReached, RunCancelled, Soul, run_soul
 from kimi_cli.soul.kimisoul import KimiSoul
 from kimi_cli.ui.shell import update as _update_mod
-from kimi_cli.ui.shell.capture import execute_with_pty_capture, inject_to_context
+from kimi_cli.ui.shell.capture import (
+    execute_with_pipe_capture,
+    execute_with_pty_capture,
+    inject_to_context,
+)
 from kimi_cli.ui.shell.console import console
 from kimi_cli.ui.shell.echo import render_user_echo_text
 from kimi_cli.ui.shell.mcp_status import render_mcp_prompt
@@ -521,9 +526,10 @@ async def _run_shell_command(self, command: str) -> None:
         exit_code: int | None = None
         raw_output: str | None = None
         try:
-            exit_code, raw_output = await execute_with_pty_capture(
-                command, env=get_clean_env(), cwd=os.getcwd()
+            execute = (
+                execute_with_pty_capture if sys.platform != "win32" else execute_with_pipe_capture
             )
+            exit_code, raw_output = await execute(command, env=get_clean_env(), cwd=os.getcwd())
         except Exception as e:
             logger.exception("Failed to run shell command:")
             console.print(f"[red]Failed to run shell command: {e}[/red]")
@@ -544,19 +550,21 @@ async def _handle_cd(self, args: list[str]) -> None:
         """
         target = args[1] if len(args) > 1 else "~"
 
-        # Expand ~ on Python side so shlex.quote won't suppress tilde expansion.
-        if target.startswith("~"):
-            target = os.path.expanduser(target)
-
         # Provide OLDPWD so `cd -` works across invocations.
         env = get_clean_env()
         old_cwd = os.getcwd()
         if "OLDPWD" not in env:
             env["OLDPWD"] = old_cwd
 
-        # Let the shell resolve -, $HOME, CDPATH, etc.
+        # Use shlex.quote for safety, but NOT for targets that need shell
+        # expansion: ~, -, and $VAR references.  For those we let the real
+        # shell handle expansion directly.
+        needs_shell_expansion = target.startswith("~") or target.startswith("$") or target == "-"
+        quoted_target = target if needs_shell_expansion else shlex.quote(target)
+
+        # Let the shell resolve ~, -, $HOME, CDPATH, etc.
         probe = await asyncio.create_subprocess_shell(
-            f"cd {shlex.quote(target)} && pwd",
+            f"cd {quoted_target} && pwd",
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             env=env,

src/kimi_cli/ui/shell/capture.py

@@ -163,14 +163,14 @@ def _on_master_readable() -> None:
         # Drain any remaining buffered output after process exits.
         with contextlib.suppress(TimeoutError):
             await asyncio.wait_for(eof_event.wait(), timeout=0.5)
-    except KeyboardInterrupt:
+    except (KeyboardInterrupt, asyncio.CancelledError):
         # Forward SIGINT to child if it is still running.
-        logger.debug("PTY capture: KeyboardInterrupt, forwarding SIGINT to child")
+        logger.debug("PTY capture: interrupted, forwarding SIGINT to child")
         if proc.returncode is None:
             proc.send_signal(2)  # SIGINT
             try:
                 await asyncio.wait_for(proc.wait(), timeout=3.0)
-            except (TimeoutError, ProcessLookupError):
+            except (TimeoutError, ProcessLookupError, asyncio.CancelledError):
                 logger.debug("PTY capture: child did not exit after SIGINT, killing")
                 proc.kill()
     finally:
@@ -254,6 +254,10 @@ async def inject_to_context(
     """
     output = clean_output(raw_output)
 
+    # Neutralise any system-reminder tags in the output to prevent injection.
+    output = output.replace("<system-reminder>", "&lt;system-reminder&gt;")
+    output = output.replace("</system-reminder>", "&lt;/system-reminder&gt;")
+
     # Truncate, keeping the tail (usually more informative).
     truncated = False
     encoded = output.encode("utf-8")