Meta-Harness/meta_harness

"""Coding-agent adapters for the Meta-Harness ``coding_agent`` proposer.

The reference proposer is a *coding agent* — an LLM wrapped in a tool-use loop with filesystem
access — that BROWSES the on-disk archive D (grep/cat) and writes new candidate files. The
reference uses Claude Code, but any headless coding-agent CLI that can (a) read files under a
working dir and (b) write files there works the same way, because the whole contract is prompt-
driven: "read D, write ``candidates/<name>.py`` + ``pending_eval.json``". This module abstracts the
launch behind a small adapter so the agent is a config choice (``proposer.coding_agent``).

Each adapter knows only three agent-specific things: the binary, how to pass the SKILL.md steering
(a real ``--append-system-prompt`` flag, or — for agents without one — prepended to the prompt),
and the headless/full-access flags. The archive D, the task prompt, and the ``pending_eval.json``
contract are identical across agents.

Adapters:
  * ``claude_code``  — Claude Code CLI (``claude -p``). Subscription-billed, native ``--append-
    system-prompt``. Fully wired; the default. Reuses its native-trace config layout while scoping
    subscription auth to each agent session (API keys stripped so a run can never bill the API).
  * ``codex``        — OpenAI Codex CLI (``codex exec``). Built from the real ``codex exec`` flags
    (``-C`` cwd, ``--sandbox``/``--dangerously-bypass-approvals-and-sandbox``, ``-m`` model, ``-c
    model_reasoning_effort=…``); no system-prompt flag, so the skill is prepended to the prompt.
    EXPERIMENTAL — uses Codex's own auth (``codex login`` / ``OPENAI_API_KEY``); not run-tested here.
  * ``cursor_agent`` — Cursor Agent CLI (``cursor-agent -p``). EXPERIMENTAL and best-effort: the
    binary is not always installed, so selecting it without ``cursor-agent`` on PATH fails fast with
    an install hint. Skill prepended to the prompt.

To add another agent: subclass :class:`CodingAgentAdapter`, implement :meth:`spec`, and register it
in :data:`ADAPTERS`.
"""
from __future__ import annotations

import base64
import json
import logging
import os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path

from ..claude_code.scaffold import _kill_proc_group, _prepare_claude_config_dir

log = logging.getLogger(__name__)

_CLAUDE_MODEL_PREFIXES = ("claude-", "sonnet", "opus", "haiku")
_DISALLOWED_CLAUDE_TOOLS = "AskUserQuestion,EnterPlanMode,ExitPlanMode"
# codex `model_reasoning_effort` accepts these; map max/xhigh down to the top codex level
_CODEX_EFFORT = {"minimal": "minimal", "low": "low", "medium": "medium", "high": "high",
                 "xhigh": "high", "max": "high"}
_CLAUDE_AUTH_FILE_ENV = "GALAPAGOS_CLAUDE_AUTH_FILE"


@dataclass
class LaunchSpec:
    """Everything :func:`run_agent` needs to launch one headless coding-agent session at ``cwd=D``."""
    argv: list[str]
    env: dict = field(default_factory=dict)
    log_name: str = "agent.log"


class CodingAgentAdapter:
    """One headless coding-agent CLI. Subclasses supply :attr:`name`/:attr:`binary`/:meth:`spec`."""

    name: str = "base"
    binary: str = ""
    experimental: bool = False
    install_hint: str = ""

    def resolve_binary(self) -> str:
        import shutil
        path = shutil.which(self.binary)
        if path is None:
            raise RuntimeError(
                f"meta_harness proposer.coding_agent={self.name!r} needs the {self.binary!r} CLI on "
                f"PATH{(' — ' + self.install_hint) if self.install_hint else ''}. "
                "Install it, pick a different proposer.coding_agent, or set proposer.mode: llm.")
        return path

    def spec(self, d_dir: Path, system_text: str, user_prompt: str, *, model: str | None,
             effort: str | None, max_turns: int | None) -> LaunchSpec:
        raise NotImplementedError

    def launch(self, d_dir: Path, system_text: str, user_prompt: str, *, model: str | None,
               effort: str | None, max_turns: int | None, timeout: int,
               log_name: str | None = None) -> int:
        spec = self.spec(d_dir, system_text, user_prompt, model=model, effort=effort,
                         max_turns=max_turns)
        if log_name:
            spec.log_name = log_name
        return run_agent(spec, d_dir, timeout)


def run_agent(spec: LaunchSpec, d_dir: Path, timeout: int) -> int:
    """Run one agent session (stdout+stderr → ``d_dir/<log_name>``), killing the whole process group
    on timeout. Returns the exit code (124 on timeout)."""
    log_path = d_dir / spec.log_name
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with open(log_path, "w") as log_fh:
        proc = subprocess.Popen(spec.argv, stdin=subprocess.DEVNULL, stdout=log_fh,
                                stderr=subprocess.STDOUT, cwd=str(d_dir), env=spec.env,
                                start_new_session=True)
        try:
            return proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            _kill_proc_group(proc)
            proc.wait()
            log.warning("coding-agent session timed out after %ss", timeout)
            return 124


# ---- claude_code (the default; fully wired) ------------------------------------------------------
class ClaudeCodeAdapter(CodingAgentAdapter):
    name = "claude_code"
    binary = "claude"
    install_hint = "npm install -g @anthropic-ai/claude-code (subscription: `claude setup-token`)"

    def __init__(self):
        # One adapter instance belongs to one proposer/run (get_adapter returns a fresh instance).
        # Container auth is consumed from a short-lived bootstrap file once, then retained only in
        # controller memory so later rounds need no persistent secret file or PID-1 environment value.
        self._auth_loaded = False
        self._oauth_token: str | None = None
        self._credential_bytes: bytes | None = None

    def _load_subscription_auth(self, env: dict) -> None:
        """Consume subscription auth once without leaving the bootstrap secret on disk.

        The outer Docker orchestrator stages a mode-0600 JSON file *after* container start. Its path is
        non-secret; its contents are removed immediately after this method reads them. Host mode falls
        back to the normal OAuth env var or ``~/.claude/.credentials.json``.
        """
        if self._auth_loaded:
            return
        bootstrap_raw = env.pop(_CLAUDE_AUTH_FILE_ENV, None)
        if bootstrap_raw:
            bootstrap = Path(bootstrap_raw)
            try:
                payload = json.loads(bootstrap.read_text(encoding="utf-8"))
                kind = payload.get("kind")
                if kind == "oauth-token":
                    self._oauth_token = str(payload.get("value") or "").strip()
                    if not self._oauth_token:
                        raise ValueError("empty OAuth token")
                elif kind == "login-credentials":
                    self._credential_bytes = base64.b64decode(str(payload.get("value") or ""),
                                                               validate=True)
                    if not self._credential_bytes:
                        raise ValueError("empty credentials payload")
                else:
                    raise ValueError(f"unknown auth kind {kind!r}")
            except Exception as exc:
                raise RuntimeError(f"invalid Meta-Harness Claude auth bootstrap: {exc}") from exc
            finally:
                try:
                    bootstrap.unlink()
                except OSError:
                    pass
        else:
            token = (env.pop("CLAUDE_CODE_OAUTH_TOKEN", None) or "").strip()
            if token:
                self._oauth_token = token
            else:
                source_raw = env.pop("GALAPAGOS_CLAUDE_CREDENTIALS_FILE", None)
                source = Path(source_raw) if source_raw else Path.home() / ".claude" / ".credentials.json"
                if source.is_file():
                    self._credential_bytes = source.read_bytes()
                else:
                    raise RuntimeError(
                        "meta_harness proposer.coding_agent='claude_code' needs subscription auth: "
                        "set CLAUDE_CODE_OAUTH_TOKEN (`claude setup-token`) or sign in with "
                        "`claude` + /login")
        self._auth_loaded = True

    def _setup_session_auth(self, env: dict, claude_config_dir: Path) -> None:
        for var in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"):
            env.pop(var, None)
        self._load_subscription_auth(env)
        if self._oauth_token:
            env["CLAUDE_CODE_OAUTH_TOKEN"] = self._oauth_token
            return
        target = claude_config_dir / ".credentials.json"
        target.write_bytes(self._credential_bytes or b"")
        target.chmod(0o600)

    def spec(self, d_dir, system_text, user_prompt, *, model, effort, max_turns) -> LaunchSpec:
        if model and not any(model.startswith(p) for p in _CLAUDE_MODEL_PREFIXES):
            raise ValueError(f"claude_code drives Claude models, got {model!r}; use a "
                             "claude-*/sonnet/opus/haiku name (or null for the CLI default)")
        cli = self.resolve_binary()
        # Harbor-style native traces plus session-scoped subscription auth: the trace directory
        # persists, while the credential file exists only for the duration of this launch.
        claude_config_dir = _prepare_claude_config_dir(d_dir)
        env = dict(os.environ)
        env["CLAUDE_CONFIG_DIR"] = str(claude_config_dir)
        self._setup_session_auth(env, claude_config_dir)
        argv = [cli, "-p", user_prompt,
                "--dangerously-skip-permissions", "--disallowedTools", _DISALLOWED_CLAUDE_TOOLS,
                "--output-format", "stream-json", "--verbose"]
        if system_text:   # no-skill mode passes "" — omit the flag entirely
            argv[3:3] = ["--append-system-prompt", system_text]
        if max_turns is not None:
            argv += ["--max-turns", str(int(max_turns))]
        if model:
            argv += ["--model", model]
        if effort:
            argv += ["--effort", effort]   # claude accepts low|medium|high|xhigh|max
        return LaunchSpec(argv=argv, env=env, log_name="claude.log")

    def launch(self, d_dir: Path, system_text: str, user_prompt: str, *, model: str | None,
               effort: str | None, max_turns: int | None, timeout: int,
               log_name: str | None = None) -> int:
        try:
            return super().launch(
                d_dir, system_text, user_prompt, model=model, effort=effort,
                max_turns=max_turns, timeout=timeout, log_name=log_name)
        finally:
            # Native traces stay in D; subscription credentials never do. This also runs when binary
            # resolution, process launch, timeout handling, or the agent itself fails.
            credentials = d_dir / "claude_sessions" / ".credentials.json"
            try:
                credentials.unlink()
            except OSError:
                pass


# ---- codex (experimental; built from the real `codex exec` flags) --------------------------------
class CodexAdapter(CodingAgentAdapter):
    name = "codex"
    binary = "codex"
    experimental = True
    install_hint = "npm install -g @openai/codex (auth: `codex login` or OPENAI_API_KEY)"

    def spec(self, d_dir, system_text, user_prompt, *, model, effort, max_turns) -> LaunchSpec:
        cli = self.resolve_binary()
        env = dict(os.environ)  # Codex resolves its own auth (~/.codex login or OPENAI_API_KEY)
        # Codex has no system-prompt flag → prepend the skill steering to the prompt.
        prompt = f"{system_text}\n\n{user_prompt}" if system_text else user_prompt
        argv = [cli, "exec", prompt, "-C", str(d_dir), "--skip-git-repo-check",
                "--dangerously-bypass-approvals-and-sandbox"]
        if model:
            argv += ["-m", model]
        if effort:
            argv += ["-c", f'model_reasoning_effort="{_CODEX_EFFORT.get(effort, "high")}"']
        return LaunchSpec(argv=argv, env=env, log_name="codex.log")


# ---- cursor_agent (experimental; gated on availability) ------------------------------------------
class CursorAgentAdapter(CodingAgentAdapter):
    name = "cursor_agent"
    binary = "cursor-agent"
    experimental = True
    install_hint = "install Cursor's agent CLI (`cursor-agent`); auth via `cursor-agent login`"

    def spec(self, d_dir, system_text, user_prompt, *, model, effort, max_turns) -> LaunchSpec:
        cli = self.resolve_binary()
        env = dict(os.environ)
        prompt = f"{system_text}\n\n{user_prompt}" if system_text else user_prompt
        # Cursor Agent headless print mode with full-access (flags may change across versions —
        # isolated here for easy adjustment).
        argv = [cli, "-p", prompt, "--force"]
        if model:
            argv += ["--model", model]
        return LaunchSpec(argv=argv, env=env, log_name="cursor.log")


ADAPTERS: dict[str, type[CodingAgentAdapter]] = {
    adapter.name: adapter for adapter in (ClaudeCodeAdapter, CodexAdapter, CursorAgentAdapter)
}


def get_adapter(name: str) -> CodingAgentAdapter:
    """Resolve a coding-agent adapter by ``proposer.coding_agent`` name (raises on unknown)."""
    try:
        return ADAPTERS[name]()
    except KeyError:
        raise ValueError(
            f"unknown proposer.coding_agent {name!r}; available: {', '.join(sorted(ADAPTERS))}")