delegate.py

"""Drive the other CLI coding agents headlessly and capture their output.

pit becomes an orchestrator: it can hand a self-contained task to Claude Code,
Codex, or Gemini, wait, and read back the result — the same way a parent agent
spawns subagents. Each runs as its own process with ITS OWN permissions, so a
delegated agent is NOT covered by pit's deletion guard (see caveat in the tool
description). `yolo=False` runs the agent in its safest headless mode; yolo=True
grants it full auto (edit/run without prompting) so it can actually finish work.
"""
import os
import shutil
import subprocess

# agent -> (argv builder). prompt is passed as the final positional arg.
# safe = the most autonomous mode that still won't edit/delete without asking
# where the CLI supports it; yolo = full auto.
_AGENTS = {
    # Claude Code: -p print/headless, text output. yolo skips all permission
    # prompts; safe uses plan mode (reads/analyzes, does not modify).
    "claude": lambda p, yolo: (
        ["claude", "-p", p, "--output-format", "text"]
        + (["--dangerously-skip-permissions"] if yolo
           else ["--permission-mode", "plan"])),
    # Codex: exec = non-interactive. --skip-git-repo-check so it runs outside a
    # git repo. yolo = --full-auto (workspace-write); safe = read-only sandbox.
    "codex": lambda p, yolo: (
        ["codex", "exec", "--skip-git-repo-check"]
        + (["--full-auto"] if yolo else ["-s", "read-only"]) + [p]),
    # Gemini: -o text headless. yolo = --yolo (auto-approve actions).
    "gemini": lambda p, yolo: (
        ["gemini", "-o", "text"] + (["--yolo"] if yolo else []) + ["-p", p]),
}


def available() -> dict:
    """Which delegate agents are on PATH."""
    return {name: bool(shutil.which(name)) for name in _AGENTS}


def delegate(agent: str, prompt: str, cwd: str = ".",
             yolo: bool = False, timeout: int = 600) -> str:
    agent = (agent or "").strip().lower()
    if agent not in _AGENTS:
        return (f"[error] unknown agent '{agent}'. Choose one of: "
                f"{', '.join(_AGENTS)}")
    exe = shutil.which(agent)
    if not exe:
        return f"[error] '{agent}' is not installed / not on PATH."
    if not os.path.isdir(cwd):
        return f"[error] cwd is not a directory: {cwd}"
    argv = _AGENTS[agent](prompt, bool(yolo))
    argv[0] = exe                                # full resolved path — Windows
    # .cmd/.bat shims (codex, gemini) won't launch from a bare name without a
    # shell; run those through the shell so the shim resolves.
    use_shell = exe.lower().endswith((".cmd", ".bat"))
    timeout = max(1, min(int(timeout), 1800))
    try:
        p = subprocess.run(
            subprocess.list2cmdline(argv) if use_shell else argv,
            cwd=cwd, capture_output=True, timeout=timeout,
            text=True, encoding="utf-8", errors="replace", shell=use_shell,
        )
    except subprocess.TimeoutExpired:
        return (f"[error] {agent} timed out after {timeout}s (it may need "
                "yolo=true to proceed non-interactively, or a smaller task)")
    except Exception as e:                       # noqa: BLE001
        return f"[error] failed to launch {agent}: {e}"
    out = ((p.stdout or "") + (p.stderr or "")).strip() or "(no output)"
    header = f"[{agent}{' yolo' if yolo else ' safe'}, exit {p.returncode}]\n"
    return header + out