gimme-job/gimme_job/runtime/claude_cli.py

"""Claude Code CLI subprocess wrapper."""
from __future__ import annotations

import json
import subprocess
import time
from pathlib import Path
from typing import Optional

from loguru import logger

from gimme_job.models.runtime import ClaudeRunResult

DEFAULT_LEARN_TOOLS = ["Read", "Edit", "Write", "Bash", "Grep", "Glob"]
DEFAULT_REPAIR_TOOLS = ["Read", "Edit", "Write", "Bash", "Grep", "Glob"]
DEFAULT_TIMEOUT_SECONDS = 600


class ClaudeCodeClient:
    def __init__(self, cwd: Optional[Path] = None, timeout: int = DEFAULT_TIMEOUT_SECONDS):
        from gimme_job.utils.paths import project_root
        self.cwd = str(cwd or project_root())
        self.timeout = timeout

    def run_prompt(
        self,
        prompt: str,
        cwd: Optional[Path] = None,
        allowed_tools: Optional[list[str]] = None,
        max_turns: int = 30,
    ) -> ClaudeRunResult:
        """Run a prompt via `claude -p`. Returns parsed ClaudeRunResult."""
        tools = allowed_tools or DEFAULT_LEARN_TOOLS
        cmd = [
            "claude",
            "-p", prompt,
            "--output-format", "json",
            "--max-turns", str(max_turns),
            "--allowedTools", ",".join(tools),
        ]

        work_dir = str(cwd) if cwd else self.cwd
        start = time.time()

        logger.info(f"Calling Claude Code CLI (cwd={work_dir})")
        logger.debug(f"Command: {' '.join(cmd[:4])}...")

        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                cwd=work_dir,
                timeout=self.timeout,
            )
            duration_ms = int((time.time() - start) * 1000)

            if result.returncode != 0:
                logger.error(f"Claude Code returned non-zero: {result.returncode}")
                logger.debug(f"stderr: {result.stderr[:500]}")
                return ClaudeRunResult(
                    success=False,
                    output=result.stdout,
                    error=result.stderr[:500],
                    duration_ms=duration_ms,
                )

            # Parse JSON output
            parsed = self._parse_output(result.stdout)
            return ClaudeRunResult(
                success=True,
                session_id=parsed.get("session_id"),
                output=parsed.get("result", result.stdout),
                cost_usd=parsed.get("cost_usd"),
                duration_ms=duration_ms,
            )

        except subprocess.TimeoutExpired:
            logger.error(f"Claude Code timed out after {self.timeout}s")
            return ClaudeRunResult(
                success=False,
                error=f"Timed out after {self.timeout}s",
            )
        except FileNotFoundError:
            logger.error("Claude Code CLI not found. Install: npm install -g @anthropic-ai/claude-code")
            return ClaudeRunResult(success=False, error="claude CLI not found")

    def resume(
        self,
        session_id: str,
        prompt: str,
        cwd: Optional[Path] = None,
    ) -> ClaudeRunResult:
        """Resume a previous session via `--resume`."""
        cmd = [
            "claude",
            "-p", prompt,
            "--resume", session_id,
            "--output-format", "json",
        ]

        work_dir = str(cwd) if cwd else self.cwd
        start = time.time()

        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, cwd=work_dir, timeout=self.timeout
            )
            duration_ms = int((time.time() - start) * 1000)
            parsed = self._parse_output(result.stdout)
            return ClaudeRunResult(
                success=result.returncode == 0,
                session_id=parsed.get("session_id"),
                output=parsed.get("result", result.stdout),
                duration_ms=duration_ms,
                error=result.stderr[:500] if result.returncode != 0 else None,
            )
        except Exception as e:
            return ClaudeRunResult(success=False, error=str(e))

    def _parse_output(self, raw: str) -> dict:
        """Parse Claude Code JSON output, handling streaming line format."""
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            pass

        # Try last non-empty line (streaming format)
        lines = [l.strip() for l in raw.strip().splitlines() if l.strip()]
        for line in reversed(lines):
            try:
                return json.loads(line)
            except json.JSONDecodeError:
                continue

        return {"result": raw}