rein-aharness/agent_harness/adapter.py

84 lines
2.8 KiB
Python
Raw Normal View History

"""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
- an allow-list identical in spirit to binky-control/scripts/rhythm-session.sh:
read/edit tools plus local git add/commit/status/log/diff no push, no
network, no arbitrary shell.
"""
from __future__ import annotations
import subprocess
from pathlib import Path
from llm_connect.claude_code import ClaudeCodeAdapter
from llm_connect.exceptions import LLMSubprocessError, LLMTimeoutError
from llm_connect.models import LLMResponse, RunConfig
ALLOWED_TOOLS = (
"Read,Write,Edit,Glob,Grep,"
"Bash(git add:*),Bash(git commit:*),Bash(git status),"
"Bash(git log:*),Bash(git diff:*),Bash(date:*),Bash(ls:*)"
)
class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
def __init__(self, workdir: Path, **kwargs):
super().__init__(**kwargs)
self._workdir = workdir
def _build_command(self, config: RunConfig) -> list[str]:
cmd = [
self._cli_path,
"--print",
"--permission-mode",
"acceptEdits",
"--allowedTools",
ALLOWED_TOOLS,
]
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
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),
},
)