clawdforge/clawdforge/runner.py

"""Wrap `claude -p ... --output-format json` and parse its result."""
import json
import os
import shutil
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path


@dataclass
class RunResult:
    ok: bool
    result: object  # parsed JSON or text
    raw_stdout: str
    raw_stderr: str
    duration_ms: int
    stop_reason: str | None
    error: str | None = None


class Runner:
    def __init__(self, *, claude_bin: str, default_model: str, default_timeout: int, runs_dir: str):
        self.claude_bin = claude_bin
        self.default_model = default_model
        self.default_timeout = default_timeout
        self.runs_dir = runs_dir
        Path(runs_dir).mkdir(parents=True, exist_ok=True)

    def run(
        self,
        *,
        prompt: str,
        model: str | None = None,
        system: str | None = None,
        system_mode: str | None = "append",
        effort: str | None = None,
        files: list[str] | None = None,
        timeout_secs: int | None = None,
    ) -> RunResult:
        # Linux ARG_MAX caps a CLI argument around 128KB-2MB depending on
        # kernel + env. Long prompts (recipe corpora, big curate jobs) blow
        # past it. For prompts > 64KB we pipe via stdin instead.
        ARG_MAX_THRESHOLD = 64 * 1024  # bytes; leaves headroom for env+other args
        use_stdin = len(prompt.encode("utf-8")) > ARG_MAX_THRESHOLD

        cmd = [self.claude_bin, "-p"]
        if not use_stdin:
            cmd.append(prompt)
        cmd += [
            "--output-format", "json",
            "--model", model or self.default_model,
        ]
        if system:
            # "append" = additive on top of claude's defaults; right
            # for tool-using assistants. "replace" = REPLACES the
            # default base prompt entirely; right for personas
            # (fiction authors, chat bots, in-world characters) where
            # claude's defaults would bleed through as friction.
            if system_mode == "replace":
                cmd += ["--system-prompt", system]
            else:
                cmd += ["--append-system-prompt", system]
        if effort:
            cmd += ["--effort", effort]
        if files:
            for f in files:
                cmd += ["--files", f]

        timeout = timeout_secs or self.default_timeout

        # Stage a per-run dir so claude has a clean working directory
        run_id = f"{int(time.time()*1000)}-{os.getpid()}"
        cwd = Path(self.runs_dir) / run_id
        cwd.mkdir(parents=True, exist_ok=True)

        started = time.monotonic()
        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=str(cwd),
                input=prompt if use_stdin else None,
            )
        except subprocess.TimeoutExpired as e:
            duration_ms = int((time.monotonic() - started) * 1000)
            return RunResult(
                ok=False,
                result=None,
                raw_stdout=(e.stdout or b"").decode("utf-8", "replace") if isinstance(e.stdout, bytes) else (e.stdout or ""),
                raw_stderr=(e.stderr or b"").decode("utf-8", "replace") if isinstance(e.stderr, bytes) else (e.stderr or ""),
                duration_ms=duration_ms,
                stop_reason="timeout",
                error=f"timeout after {timeout}s",
            )
        finally:
            shutil.rmtree(cwd, ignore_errors=True)

        duration_ms = int((time.monotonic() - started) * 1000)

        if proc.returncode != 0:
            return RunResult(
                ok=False,
                result=None,
                raw_stdout=proc.stdout,
                raw_stderr=proc.stderr,
                duration_ms=duration_ms,
                stop_reason="error",
                error=f"claude exit {proc.returncode}",
            )

        # claude --output-format json wraps in {"type":"result","result":"...","stop_reason":...}
        parsed: object
        stop_reason: str | None = None
        try:
            outer = json.loads(proc.stdout)
            if isinstance(outer, dict) and "result" in outer:
                stop_reason = outer.get("stop_reason") or outer.get("subtype")
                inner = (outer["result"] or "").strip()
                if inner.startswith("```"):
                    # strip ```lang fenced block
                    parts = inner.split("\n", 1)
                    if len(parts) == 2:
                        inner = parts[1].rsplit("```", 1)[0].strip()
                # Try to JSON-parse the inner; if it's not valid JSON, return as text
                try:
                    parsed = json.loads(inner)
                except json.JSONDecodeError:
                    parsed = inner
            else:
                parsed = outer
        except json.JSONDecodeError:
            # Outer wasn't even JSON. Return raw text.
            parsed = proc.stdout.strip()

        return RunResult(
            ok=True,
            result=parsed,
            raw_stdout=proc.stdout,
            raw_stderr=proc.stderr,
            duration_ms=duration_ms,
            stop_reason=stop_reason,
            error=None,
        )