#!/usr/bin/env python3 """ Kimi Copilot - PTY wrapper for Kimi Code CLI. Captures user input and Kimi output, forwarding both to the brain module for context tracking and side-model analysis. """ import fcntl import os import pty import queue import select import struct import sys import termios import threading import tty class KimiCopilot: def __init__(self, command=None): self.command = command or ["kimi"] self.master_fd = None self.child_pid = None self.running = False # Queues used to ship I/O to the brain thread. self.user_input_queue = queue.Queue() self.kimi_output_queue = queue.Queue() # Queue for brain notifications so they are serialized with the main # forwarding loop instead of racing with child PTY output. self.notification_queue = queue.Queue() @staticmethod def _get_terminal_size(fd): """Return (rows, cols) for the given terminal fd.""" s = struct.pack("HHHH", 0, 0, 0, 0) try: x = fcntl.ioctl(fd, termios.TIOCGWINSZ, s) except OSError: return 24, 80 return struct.unpack("HHHH", x)[:2] @staticmethod def _set_terminal_size(fd, rows, cols): """Propagate terminal size to the PTY.""" s = struct.pack("HHHH", rows, cols, 0, 0) try: fcntl.ioctl(fd, termios.TIOCSWINSZ, s) except OSError: pass def _on_window_resize(self, signum, frame): """Propagate terminal resize events to the child PTY.""" rows, cols = self._get_terminal_size(sys.stdin.fileno()) if self.master_fd is not None: self._set_terminal_size(self.master_fd, rows, cols) def spawn(self): """Fork a PTY and exec the Kimi Code CLI process.""" pid, master_fd = pty.fork() if pid == 0: # Child process: replace with Kimi Code CLI. os.execvp(self.command[0], self.command) self.child_pid = pid self.master_fd = master_fd # Make sure the child PTY matches the user's terminal size. rows, cols = self._get_terminal_size(sys.stdin.fileno()) self._set_terminal_size(master_fd, rows, cols) # Handle terminal resize. import signal signal.signal(signal.SIGWINCH, self._on_window_resize) def _set_raw_mode(self): """Switch local terminal to raw mode and return old settings.""" if not os.isatty(sys.stdin.fileno()): return None old_tty = termios.tcgetattr(sys.stdin) tty.setraw(sys.stdin.fileno()) return old_tty def _restore_tty(self, old_tty): if old_tty is None: return termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_tty) def _emit_notifications(self): """Flush queued brain notifications to stdout in raw-terminal mode. In raw mode ``\n`` does not perform a carriage return, so we rewrite line endings as ``\r\n`` to keep iTerm2 display clean. We also clear each line before writing because the child CLI may have left a status line (context / help hints) at the bottom of the screen; without clearing, Copilot messages get concatenated with that status text. """ while True: try: msg = self.notification_queue.get_nowait() except queue.Empty: break if not msg: continue # Ensure the message ends with a newline. if not msg.endswith("\n"): msg += "\n" # Rewrite every line so it starts at column 0 with the current line # erased. This prevents interleaving with Kimi CLI status lines. lines = msg.split("\n") parts = [] for i, line in enumerate(lines): if i == len(lines) - 1 and line == "": continue parts.append(f"\r\x1b[K{line}") if i < len(lines) - 1: parts.append("\r\n") msg = "".join(parts) if not msg.endswith("\r\n"): msg += "\r\n" try: os.write(sys.stdout.fileno(), msg.encode("utf-8", errors="ignore")) except OSError: break def _forward_loop(self): """Forward bytes between user stdin/stdout and the child PTY.""" old_tty = self._set_raw_mode() try: while self.running: readable, _, _ = select.select( [sys.stdin.fileno(), self.master_fd], [], [], 0.1 ) if sys.stdin.fileno() in readable: try: data = os.read(sys.stdin.fileno(), 4096) except OSError: break if not data: break os.write(self.master_fd, data) text = data.decode("utf-8", errors="ignore") self.user_input_queue.put(text) if self.master_fd in readable: try: data = os.read(self.master_fd, 4096) except OSError: break if not data: # Child exited. break os.write(sys.stdout.fileno(), data) self.kimi_output_queue.put( data.decode("utf-8", errors="ignore") ) # Serialize brain notifications with child output to avoid # interleaved lines and iTerm2 display corruption. self._emit_notifications() finally: self._restore_tty(old_tty) def _brain_loop(self): """Background thread: feed captured I/O to the brain.""" # Import here to avoid circular imports and allow independent testing. try: from brain import Brain except ImportError: Brain = None if Brain is None: # Brain not ready yet; just drain queues. while self.running: try: self.user_input_queue.get(timeout=0.5) except queue.Empty: pass try: self.kimi_output_queue.get(timeout=0.5) except queue.Empty: pass return def notify_callback(msg: str): self.notification_queue.put(msg) brain = Brain(notify_callback=notify_callback) def process_one_item(): try: text = self.user_input_queue.get(timeout=0.2) brain.feed_user_input(text) except queue.Empty: pass try: text = self.kimi_output_queue.get(timeout=0.2) brain.feed_kimi_output(text) except queue.Empty: pass # Active phase: process while the forward loop is running. while self.running: process_one_item() # Drain phase: forward loop has finished, keep processing remaining # queued chunks until empty. This is essential because Kimi may flush # its final output right before exiting. while True: try: self.user_input_queue.get_nowait() except queue.Empty: break while True: try: text = self.kimi_output_queue.get_nowait() brain.feed_kimi_output(text) except queue.Empty: break def run(self): self.spawn() self.running = True brain_thread = threading.Thread(target=self._brain_loop, daemon=True) brain_thread.start() try: self._forward_loop() finally: self.running = False # Give the brain enough time to drain the queue and finish any # side-model API call / RAG write that gets triggered. brain_thread.join(timeout=60.0) # Flush any notifications produced during the drain phase. self._emit_notifications() if self.child_pid is not None: try: os.waitpid(self.child_pid, 0) except ChildProcessError: pass def main(): # Allow overriding the command for testing, e.g.: # python3 kimi_copilot.py /bin/bash if "--panel" in sys.argv: sys.argv.remove("--panel") from copilot_panel import HintPanel import time open_browser = "--no-browser" not in sys.argv if not open_browser: sys.argv.remove("--no-browser") panel = HintPanel() url = panel.open(open_browser=open_browser) print(f"\n[Copilot] 提示面板：{url}") if open_browser: print("[Copilot] 正在打开浏览器...\n") else: print() try: while True: time.sleep(1) except KeyboardInterrupt: return command = sys.argv[1:] or ["kimi"] copilot = KimiCopilot(command=command) copilot.run() if __name__ == "__main__": main()