"""Agentic Claude Code adapter. llm-connect's ClaudeCodeAdapter is a text-generation adapter (`claude --print`, no working directory, no tool grants). An executor run needs an *agentic* session: file edits and git commits inside the target repo, under a hard tool allow-list. This adapter subclasses it, keeping the llm-connect LLMAdapter interface so a hosted adapter can be swapped in later, and adds: - cwd pinned to the target repo - --permission-mode acceptEdits - allow-list from a named tool profile (default: green-commit-only) - optional real-time per-tool-call audit events (HARNESS-WP-0002-T03) Claude Code executes its own tools internally in `--print` mode — it is not possible for a caller to externally dispatch individual tool calls (that would require abandoning Claude Code's self-contained agent model entirely). What `--output-format stream-json --include-hook-events` does allow: observing each tool_use/tool_result/hook event as it happens. When `on_tool_event` is supplied, this adapter runs in that streaming mode and invokes the callback once per event, in real time, while still returning one aggregate `LLMResponse` at the end for interface compatibility. """ from __future__ import annotations import json import subprocess import threading from pathlib import Path from typing import Any, Callable from llm_connect.claude_code import ClaudeCodeAdapter from llm_connect.exceptions import LLMSubprocessError, LLMTimeoutError from llm_connect.models import LLMResponse, RunConfig from rein_aharness.profiles import ToolProfile, get_profile # Backward-compatible alias for the seed profile allow-list string. ALLOWED_TOOLS = get_profile("green-commit-only").allowed_tools ToolEventCallback = Callable[[dict[str, Any]], None] def _is_tool_event(event: dict[str, Any]) -> bool: """True for tool_use/tool_result content blocks and hook lifecycle events. Deliberately excludes plain assistant text messages — those aren't tool audit events, just conversational output. """ event_type = event.get("type") if event_type == "system" and str(event.get("subtype", "")).startswith("hook_"): return True if event_type in ("assistant", "user"): for block in event.get("message", {}).get("content", []) or []: if isinstance(block, dict) and block.get("type") in ("tool_use", "tool_result"): return True return False class AgenticClaudeCodeAdapter(ClaudeCodeAdapter): def __init__( self, workdir: Path, *, tool_profile: str | ToolProfile = "green-commit-only", on_tool_event: ToolEventCallback | None = None, **kwargs, ): super().__init__(**kwargs) self._workdir = workdir self._on_tool_event = on_tool_event if isinstance(tool_profile, ToolProfile): self._profile = tool_profile else: self._profile = get_profile(tool_profile) @property def tool_profile(self) -> ToolProfile: return self._profile def _build_command(self, config: RunConfig) -> list[str]: cmd = [ self._cli_path, "--print", "--permission-mode", "acceptEdits", "--allowedTools", self._profile.allowed_tools, ] if self._on_tool_event is not None: cmd += ["--output-format", "stream-json", "--include-hook-events", "--verbose"] if self._model: cmd.extend(["--model", self._model]) return cmd def execute_prompt(self, prompt: str, config: RunConfig) -> LLMResponse: self._preflight_budget(config) cmd = self._build_command(config) timeout = config.timeout_seconds or self._config.timeout_seconds if self._on_tool_event is not None: response = self._execute_streaming(cmd, prompt, timeout) else: response = self._execute_blocking(cmd, prompt, timeout) self._consume_budget(config, response) return response def _execute_blocking(self, cmd: list[str], prompt: str, timeout: int) -> LLMResponse: try: result = subprocess.run( cmd, input=prompt, capture_output=True, text=True, timeout=timeout, cwd=self._workdir, ) except subprocess.TimeoutExpired as exc: raise LLMTimeoutError( f"claude CLI timed out after {timeout}s", cause=exc ) from exc if result.returncode != 0: raise LLMSubprocessError( f"claude CLI exited with code {result.returncode}", return_code=result.returncode, stderr=result.stderr, ) return LLMResponse( content=result.stdout, model=self._model or "claude-code-cli", usage={}, finish_reason="stop", metadata={ "provider": "claude-code-agentic", "cli_path": self._cli_path, "workdir": str(self._workdir), "tool_profile": self._profile.name, }, ) def _execute_streaming(self, cmd: list[str], prompt: str, timeout: int) -> LLMResponse: proc = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=self._workdir, ) text_parts: list[str] = [] tool_event_count = 0 def reader() -> None: nonlocal tool_event_count assert proc.stdout is not None for line in proc.stdout: line = line.strip() if not line: continue try: event = json.loads(line) except json.JSONDecodeError: continue self._handle_stream_event(event, text_parts) if _is_tool_event(event): tool_event_count += 1 self._on_tool_event(event) reader_thread = threading.Thread(target=reader, daemon=True) assert proc.stdin is not None proc.stdin.write(prompt) proc.stdin.close() reader_thread.start() try: returncode = proc.wait(timeout=timeout) except subprocess.TimeoutExpired as exc: proc.kill() proc.wait() raise LLMTimeoutError(f"claude CLI timed out after {timeout}s", cause=exc) from exc reader_thread.join(timeout=5) stderr = proc.stderr.read() if proc.stderr else "" if returncode != 0: raise LLMSubprocessError( f"claude CLI exited with code {returncode}", return_code=returncode, stderr=stderr, ) return LLMResponse( content="".join(text_parts), model=self._model or "claude-code-cli", usage={}, finish_reason="stop", metadata={ "provider": "claude-code-agentic", "cli_path": self._cli_path, "workdir": str(self._workdir), "tool_profile": self._profile.name, "tool_event_count": tool_event_count, }, ) @staticmethod def _handle_stream_event(event: dict[str, Any], text_parts: list[str]) -> None: if event.get("type") != "assistant": return for block in event.get("message", {}).get("content", []) or []: if isinstance(block, dict) and block.get("type") == "text": text_parts.append(block["text"])