Class JsonRpcClient

python/copilot/_jsonrpc.py:65–478

Minimal async JSON-RPC 2.0 client for stdio transport Uses threads for blocking IO but provides async interface.

class JsonRpcClient:
    """
    Minimal async JSON-RPC 2.0 client for stdio transport

    Uses threads for blocking IO but provides async interface.
    """

    def __init__(self, process):
        """
        Create client from subprocess.Popen with stdin/stdout pipes

        Args:
            process: subprocess.Popen with stdin=PIPE, stdout=PIPE
        """
        self.process = process
        self.pending_requests: dict[str, asyncio.Future] = {}
        self._pending_inline_callbacks: dict[str, Callable[[Any], None]] = {}
        self.notification_handler: Callable[[str, dict], None] | None = None
        self.request_handlers: dict[str, RequestHandler] = {}
        self._running = False
        self._read_thread: threading.Thread | None = None
        self._stderr_thread: threading.Thread | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._write_lock = threading.Lock()
        self._pending_lock = threading.Lock()
        self._process_exit_error: str | None = None
        self._stderr_output: list[str] = []
        self._stderr_lock = threading.Lock()
        self.on_close: Callable[[], None] | None = None

    def start(self, loop: asyncio.AbstractEventLoop | None = None):
        """Start listening for messages in background thread"""
        if not self._running:
            self._running = True
            # Always use the provided loop or get the running loop
            self._loop = loop or asyncio.get_running_loop()
            self._read_thread = threading.Thread(target=self._read_loop, daemon=True)
            self._read_thread.start()
            # Start stderr reader thread if process has stderr
            if hasattr(self.process, "stderr") and self.process.stderr:
                self._stderr_thread = threading.Thread(target=self._stderr_loop, daemon=True)
                self._stderr_thread.start()

    def _stderr_loop(self):
        """Read stderr in background to capture error messages"""
        try:
            while self._running:
                if not self.process.stderr:
                    break
                line = self.process.stderr.readline()
                if not line:
                    break
                stderr_line = line.decode("utf-8") if isinstance(line, bytes) else line
                logger.warning("[CLI] %s", stderr_line.rstrip())
                with self._stderr_lock:
                    self._stderr_output.append(stderr_line)
        except Exception:
            logger.debug("Error reading Copilot CLI stderr", exc_info=True)