feat(clink): add GitHub Copilot CLI integration

clink/agents/__init__.py

@@ -7,12 +7,14 @@
 from .base import AgentOutput, BaseCLIAgent, CLIAgentError
 from .claude import ClaudeAgent
 from .codex import CodexAgent
+from .copilot import CopilotAgent
 from .gemini import GeminiAgent
 
 _AGENTS: dict[str, type[BaseCLIAgent]] = {
     "gemini": GeminiAgent,
     "codex": CodexAgent,
     "claude": ClaudeAgent,
+    "copilot": CopilotAgent,
 }
 
 

clink/agents/copilot.py

@@ -0,0 +1,147 @@
+"""GitHub Copilot-specific CLI agent hooks."""
+
+from __future__ import annotations
+
+import asyncio
+import shutil
+import time
+from collections.abc import Sequence
+
+from clink.constants import DEFAULT_STREAM_LIMIT
+from clink.models import ResolvedCLIClient, ResolvedCLIRole
+from clink.parsers.base import ParserError
+
+from .base import AgentOutput, BaseCLIAgent, CLIAgentError
+
+
+class CopilotAgent(BaseCLIAgent):
+    """Copilot CLI agent that passes the prompt via -p argument."""
+
+    def __init__(self, client: ResolvedCLIClient):
+        super().__init__(client)
+
+    async def run(
+        self,
+        *,
+        role: ResolvedCLIRole,
+        prompt: str,
+        system_prompt: str | None = None,
+        files: Sequence[str],
+        images: Sequence[str],
+    ) -> AgentOutput:
+        _ = (files, images)
+        command = self._build_command(role=role, system_prompt=system_prompt)
+        # Copilot uses -p <text> for non-interactive mode instead of stdin.
+        command.extend(["-p", prompt])
+        env = self._build_environment()
+
+        executable_name = command[0]
+        resolved_executable = shutil.which(executable_name)
+        if resolved_executable is None:
+            raise CLIAgentError(
+                f"Executable '{executable_name}' not found in PATH for CLI '{self.client.name}'. "
+                f"Ensure the command is installed and accessible."
+            )
+        command[0] = resolved_executable
+
+        sanitized_command = list(command)
+        cwd = str(self.client.working_dir) if self.client.working_dir else None
+        limit = DEFAULT_STREAM_LIMIT
+        start_time = time.monotonic()
+
+        self._logger.debug("Executing CLI command: %s", " ".join(sanitized_command))
+
+        try:
+            process = await asyncio.create_subprocess_exec(
+                *command,
+                stdin=asyncio.subprocess.DEVNULL,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                cwd=cwd,
+                limit=limit,
+                env=env,
+            )
+        except FileNotFoundError as exc:
+            raise CLIAgentError(f"Executable not found for CLI '{self.client.name}': {exc}") from exc
+
+        try:
+            stdout_bytes, stderr_bytes = await asyncio.wait_for(
+                process.communicate(),
+                timeout=self.client.timeout_seconds,
+            )
+        except asyncio.TimeoutError as exc:
+            process.kill()
+            await process.communicate()
+            raise CLIAgentError(
+                f"CLI '{self.client.name}' timed out after {self.client.timeout_seconds} seconds",
+                returncode=None,
+            ) from exc
+
+        duration = time.monotonic() - start_time
+        return_code = process.returncode
+        stdout_text = stdout_bytes.decode("utf-8", errors="replace")
+        stderr_text = stderr_bytes.decode("utf-8", errors="replace")
+
+        if return_code != 0:
+            recovered = self._recover_from_error(
+                returncode=return_code,
+                stdout=stdout_text,
+                stderr=stderr_text,
+                sanitized_command=sanitized_command,
+                duration_seconds=duration,
+                output_file_content=None,
+            )
+            if recovered is not None:
+                return recovered
+            raise CLIAgentError(
+                f"CLI '{self.client.name}' exited with status {return_code}",
+                returncode=return_code,
+                stdout=stdout_text,
+                stderr=stderr_text,
+            )
+
+        try:
+            parsed = self._parser.parse(stdout_text, stderr_text)
+        except ParserError as exc:
+            raise CLIAgentError(
+                f"Failed to parse output from CLI '{self.client.name}': {exc}",
+                returncode=return_code,
+                stdout=stdout_text,
+                stderr=stderr_text,
+            ) from exc
+
+        return AgentOutput(
+            parsed=parsed,
+            sanitized_command=sanitized_command,
+            returncode=return_code,
+            stdout=stdout_text,
+            stderr=stderr_text,
+            duration_seconds=duration,
+            parser_name=self._parser.name,
+        )
+
+    def _recover_from_error(
+        self,
+        *,
+        returncode: int,
+        stdout: str,
+        stderr: str,
+        sanitized_command: list[str],
+        duration_seconds: float,
+        output_file_content: str | None,
+    ) -> AgentOutput | None:
+        try:
+            parsed = self._parser.parse(stdout, stderr)
+        except ParserError:
+            return None
+
+        return AgentOutput(
+            parsed=parsed,
+            sanitized_command=sanitized_command,
+            returncode=returncode,
+            stdout=stdout,
+            stderr=stderr,
+            duration_seconds=duration_seconds,
+            parser_name=self._parser.name,
+            output_file_content=output_file_content,
+        )

clink/constants.py

@@ -45,4 +45,10 @@ class CLIInternalDefaults:
         default_role_prompt="systemprompts/clink/default.txt",
         runner="claude",
     ),
+    "copilot": CLIInternalDefaults(
+        parser="copilot_jsonl",
+        additional_args=["--output-format", "json"],
+        default_role_prompt="systemprompts/clink/default.txt",
+        runner="copilot",
+    ),
 }

clink/parsers/__init__.py

@@ -5,12 +5,14 @@
 from .base import BaseParser, ParsedCLIResponse, ParserError
 from .claude import ClaudeJSONParser
 from .codex import CodexJSONLParser
+from .copilot import CopilotJSONLParser
 from .gemini import GeminiJSONParser
 
 _PARSER_CLASSES: dict[str, type[BaseParser]] = {
     CodexJSONLParser.name: CodexJSONLParser,
     GeminiJSONParser.name: GeminiJSONParser,
     ClaudeJSONParser.name: ClaudeJSONParser,
+    CopilotJSONLParser.name: CopilotJSONLParser,
 }
 
 

clink/parsers/copilot.py

@@ -0,0 +1,60 @@
+"""Parser for GitHub Copilot CLI JSONL output."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from .base import BaseParser, ParsedCLIResponse, ParserError
+
+
+class CopilotJSONLParser(BaseParser):
+    """Parse JSONL stdout emitted by `copilot -p ... --output-format json`."""
+
+    name = "copilot_jsonl"
+
+    def parse(self, stdout: str, stderr: str) -> ParsedCLIResponse:
+        lines = [line.strip() for line in (stdout or "").splitlines() if line.strip()]
+        events: list[dict[str, Any]] = []
+        assistant_messages: list[str] = []
+        result_event: dict[str, Any] | None = None
+
+        for line in lines:
+            if not line.startswith("{"):
+                continue
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+
+            events.append(event)
+            event_type = event.get("type")
+
+            if event_type == "assistant.message":
+                data = event.get("data") or {}
+                content = data.get("content")
+                if isinstance(content, str) and content.strip():
+                    assistant_messages.append(content.strip())
+
+            elif event_type == "result":
+                result_event = event
+
+        if not assistant_messages:
+            raise ParserError("Copilot CLI JSONL output did not include an assistant.message event")
+
+        content = "\n\n".join(assistant_messages).strip()
+        metadata: dict[str, Any] = {"events": events}
+
+        if result_event:
+            metadata["result"] = result_event
+            session_id = result_event.get("sessionId")
+            if isinstance(session_id, str):
+                metadata["session_id"] = session_id
+            usage = result_event.get("usage")
+            if isinstance(usage, dict):
+                metadata["usage"] = usage
+
+        if stderr and stderr.strip():
+            metadata["stderr"] = stderr.strip()
+
+        return ParsedCLIResponse(content=content, metadata=metadata)

conf/cli_clients/copilot.json

@@ -0,0 +1,22 @@
+{
+  "name": "copilot",
+  "command": "copilot",
+  "additional_args": [
+    "--allow-all-tools"
+  ],
+  "env": {},
+  "roles": {
+    "default": {
+      "prompt_path": "systemprompts/clink/default.txt",
+      "role_args": []
+    },
+    "planner": {
+      "prompt_path": "systemprompts/clink/default_planner.txt",
+      "role_args": []
+    },
+    "codereviewer": {
+      "prompt_path": "systemprompts/clink/default_codereviewer.txt",
+      "role_args": []
+    }
+  }
+}