Agent-System-Dev/scripts/agent_exec_adapters.py

"""Built-in and controlled-custom CLI adapter registry for agent-exec.

Builds direct argv only (never shell). Probes local executable/version/help;
does not contact providers or launch real agent runs.
"""
from __future__ import annotations

import re
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path

from scripts.agent_exec_config import (
    ALLOWED_PLACEHOLDERS,
    SECRET_VALUE_RE,
    ExecutionConfig,
    Profile,
)

_PROBE_TIMEOUT_SECONDS = 5
# Conservative bound for embedding unit packets into prompts (policy-compatible).
MAX_UNIT_PACKET_BYTES = 262_144
UNIT_PACKET_DISPLAY_MARKER = "<unit-packet>"

_PLACEHOLDER_MAP = {
    "${workspace}": "workspace",
    "${unit_packet}": "unit_packet",
    "${result_path}": "result_path",
    "${model}": "model",
}

_CUSTOM_OUTPUT_ARGV: dict[str, tuple[str, ...]] = {
    "json": ("--output-format", "json"),
    "stream-json": ("--output-format", "stream-json"),
    "jsonl": ("--output-format", "jsonl"),
    "text": (),
}

# Substrings that must appear in --help before a built-in is available.
_REQUIRED_HELP_MARKERS: dict[str, tuple[str, ...]] = {
    "cursor": (
        "--print",
        "--force",
        "--sandbox",
        "--trust",
        "--workspace",
        "--model",
        "--output-format",
    ),
    "antigravity": (
        "--print",
        "--dangerously-skip-permissions",
        "--mode",
        "--model",
        "--output-format",
    ),
    "claude": (
        "--print",
        "--permission-mode",
        "--model",
        "--output-format",
    ),
    "codex": (
        "exec",
        "--sandbox",
        "workspace-write",
        "--config",
        "--model",
        "--cd",
        "--json",
        "--output-last-message",
    ),
    "pi": ("--print", "--approve", "--model", "--mode"),
    "kimi": ("--prompt", "--auto", "--model", "--output-format"),
    "opencode": (
        "--print",
        "--yes",
        "--workspace",
        "--output-format",
    ),
}


@dataclass(frozen=True)
class AdapterSpec:
    name: str
    noninteractive_argv: tuple[str, ...]
    run_everything_argv: tuple[str, ...]
    workspace_flag: str | None
    model_flag: str | None
    output_argv: tuple[str, ...]
    prompt_transport: str


@dataclass(frozen=True)
class AdapterProbe:
    available: bool
    reason: str | None
    executable: str
    observed_version: str | None


@dataclass(frozen=True)
class Invocation:
    argv: tuple[str, ...]
    cwd: Path
    stdin: bytes | None
    sanitized_display: tuple[str, ...]


BUILTIN_SPECS: dict[str, AdapterSpec] = {
    "cursor": AdapterSpec(
        name="cursor",
        noninteractive_argv=("-p",),
        run_everything_argv=("--force", "--sandbox", "enabled", "--trust"),
        workspace_flag="--workspace",
        model_flag="--model",
        output_argv=("--output-format", "stream-json"),
        prompt_transport="positional",
    ),
    "antigravity": AdapterSpec(
        name="antigravity",
        noninteractive_argv=(),
        run_everything_argv=(
            "--dangerously-skip-permissions",
            "--mode",
            "auto",
        ),
        workspace_flag=None,
        model_flag="--model",
        output_argv=("--output-format", "stream-json"),
        prompt_transport="print",
    ),
    "claude": AdapterSpec(
        name="claude",
        noninteractive_argv=("-p",),
        run_everything_argv=("--permission-mode", "bypassPermissions"),
        workspace_flag=None,
        model_flag="--model",
        output_argv=("--output-format", "stream-json"),
        prompt_transport="positional",
    ),
    "codex": AdapterSpec(
        name="codex",
        noninteractive_argv=("exec",),
        run_everything_argv=(
            "--sandbox",
            "workspace-write",
            "-c",
            "approval_policy=never",
        ),
        workspace_flag="--cd",
        model_flag="--model",
        output_argv=("--json", "--output-last-message"),
        prompt_transport="positional",
    ),
    "pi": AdapterSpec(
        name="pi",
        noninteractive_argv=("-p",),
        run_everything_argv=("--approve",),
        workspace_flag=None,
        model_flag="--model",
        output_argv=("--mode", "json"),
        prompt_transport="positional",
    ),
    "kimi": AdapterSpec(
        name="kimi",
        noninteractive_argv=(),
        run_everything_argv=("--auto",),
        workspace_flag=None,
        model_flag="--model",
        output_argv=("--output-format", "stream-json"),
        prompt_transport="flag",
    ),
    "opencode": AdapterSpec(
        name="opencode",
        noninteractive_argv=("--print",),
        run_everything_argv=("--yes",),
        workspace_flag="--workspace",
        model_flag="--model",
        output_argv=("--output-format", "json"),
        prompt_transport="positional",
    ),
}


def probe_adapter(profile: Profile, config: ExecutionConfig) -> AdapterProbe:
    """Local executable/version/help probe; never contacts a provider."""
    del config  # reserved for custom-adapter-aware probes in later units
    executable = profile.executable
    resolved = shutil.which(executable)
    if resolved is None:
        return AdapterProbe(
            available=False,
            reason="executable not found on PATH",
            executable=executable,
            observed_version=None,
        )

    version = _bounded_probe_output(resolved, ["--version"])
    if version is None:
        version = _bounded_probe_output(resolved, ["-v"])
    if version is None:
        version = _bounded_probe_output(resolved, ["-V"])

    help_text = _bounded_probe_output(resolved, ["--help"]) or ""
    help_text += "\n" + (_bounded_probe_output(resolved, ["-h"]) or "")
    # Codex documents exec options under the subcommand.
    if profile.adapter == "codex":
        help_text += "\n" + (
            _bounded_probe_output(resolved, ["exec", "--help"]) or ""
        )

    markers = _REQUIRED_HELP_MARKERS.get(profile.adapter)
    if markers is not None:
        missing = [m for m in markers if not _help_has_marker(help_text, m)]
        if missing:
            return AdapterProbe(
                available=False,
                reason=(
                    "unavailable: missing required capability flags in help: "
                    + ", ".join(missing)
                ),
                executable=resolved,
                observed_version=version,
            )

    if profile.adapter == "custom":
        # Custom adapters are validated at config load; help is advisory only.
        pass

    return AdapterProbe(
        available=True,
        reason=None,
        executable=resolved,
        observed_version=version,
    )


def build_invocation(
    profile: Profile,
    config: ExecutionConfig,
    workspace: Path,
    unit_packet: Path,
    result_path: Path,
) -> Invocation:
    """Compose a direct argv invocation; never uses a shell."""
    workspace = Path(workspace)
    unit_packet = Path(unit_packet)
    result_path = Path(result_path)
    packet_text = _read_unit_packet_text(unit_packet)
    prompt = _build_bounded_unit_prompt(packet_text)

    if profile.adapter == "custom":
        return _build_custom_invocation(
            profile,
            config,
            workspace,
            unit_packet,
            result_path,
            packet_text=packet_text,
            prompt=prompt,
        )

    spec = BUILTIN_SPECS.get(profile.adapter)
    if spec is None:
        raise ValueError(f"unknown adapter: {profile.adapter!r}")

    argv: list[str] = [profile.executable]
    argv.extend(spec.noninteractive_argv)
    argv.extend(spec.run_everything_argv)
    if spec.workspace_flag is not None:
        argv.extend([spec.workspace_flag, str(workspace)])
    if profile.model is not None and spec.model_flag is not None:
        argv.extend([spec.model_flag, profile.model])
    argv.extend(spec.output_argv)
    if spec.name == "codex":
        # --output-last-message requires the result path as the next argv item.
        argv.append(str(result_path))

    stdin: bytes | None = None
    prompt_values: set[str] = set()
    if spec.prompt_transport == "positional":
        argv.append(prompt)
        prompt_values.add(prompt)
    elif spec.prompt_transport == "stdin":
        stdin = packet_text.encode("utf-8")
    elif spec.prompt_transport == "flag":
        argv.extend(["--prompt", prompt])
        prompt_values.add(prompt)
    elif spec.prompt_transport == "print":
        # agy 1.3.5 consumes the argv item immediately after --print.
        argv.extend(["--print", prompt])
        prompt_values.add(prompt)
    else:
        raise ValueError(f"unsupported prompt_transport: {spec.prompt_transport!r}")

    argv_t = tuple(argv)
    return Invocation(
        argv=argv_t,
        cwd=workspace,
        stdin=stdin,
        sanitized_display=_sanitize_display(argv_t, prompt_values=prompt_values),
    )


def _build_custom_invocation(
    profile: Profile,
    config: ExecutionConfig,
    workspace: Path,
    unit_packet: Path,
    result_path: Path,
    *,
    packet_text: str,
    prompt: str,
) -> Invocation:
    custom = config.custom_adapters.get(profile.name)
    if custom is None:
        raise ValueError(
            f"profile {profile.name!r} uses custom adapter without definition"
        )

    values = {
        "workspace": str(workspace),
        # Placeholder expands to the packet path, not the body.
        "unit_packet": str(unit_packet),
        "result_path": str(result_path),
        "model": profile.model or "",
    }

    argv: list[str] = [profile.executable]
    argv.extend(_expand_argv(custom.base_argv, values, omit_model=False))
    argv.extend(_expand_argv(custom.run_everything_argv, values, omit_model=False))
    argv.extend(_expand_argv(custom.workspace_argv, values, omit_model=False))
    if profile.model is not None:
        argv.extend(_expand_argv(custom.model_argv, values, omit_model=False))
    # model is None: omit the complete model flag/value pair (entire model_argv).
    argv.extend(_CUSTOM_OUTPUT_ARGV.get(custom.output_format, ()))

    stdin: bytes | None = None
    prompt_values: set[str] = set()
    if custom.prompt_transport == "stdin":
        stdin = packet_text.encode("utf-8")
    elif custom.prompt_transport == "positional":
        argv.append(prompt)
        prompt_values.add(prompt)
    elif custom.prompt_transport == "flag":
        argv.extend(["--prompt", prompt])
        prompt_values.add(prompt)
    else:
        raise ValueError(
            f"unsupported prompt_transport: {custom.prompt_transport!r}"
        )

    argv_t = tuple(argv)
    return Invocation(
        argv=argv_t,
        cwd=workspace,
        stdin=stdin,
        sanitized_display=_sanitize_display(argv_t, prompt_values=prompt_values),
    )


def _read_unit_packet_text(path: Path) -> str:
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise ValueError(f"unable to read unit packet: {path}") from exc
    if len(raw) > MAX_UNIT_PACKET_BYTES:
        raise ValueError(
            f"unit packet exceeds size limit of {MAX_UNIT_PACKET_BYTES} bytes"
        )
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise ValueError("unit packet is not valid UTF-8") from exc


def _build_bounded_unit_prompt(packet_text: str) -> str:
    return (
        "Implement exactly the supplied bounded implementation unit. "
        "Work only inside the current workspace. "
        "Leave all changes uncommitted. "
        "Do not expand scope beyond the unit packet. "
        "Return a structured final result matching the unit's required schema.\n\n"
        "--- BOUNDED IMPLEMENTATION UNIT PACKET ---\n\n"
        f"{packet_text}"
    )


def _expand_argv(
    items: tuple[str, ...],
    values: dict[str, str],
    *,
    omit_model: bool,
) -> list[str]:
    out: list[str] = []
    for item in items:
        if item in ALLOWED_PLACEHOLDERS:
            key = _PLACEHOLDER_MAP[item]
            if key == "model" and (omit_model or not values.get("model")):
                continue
            out.append(values[key])
            continue
        if "${" in item:
            # Config validation already rejects non-whole-item placeholders.
            raise ValueError(f"placeholder must fill whole argv item: {item!r}")
        out.append(item)
    return out


def _help_has_marker(help_text: str, marker: str) -> bool:
    if marker in help_text:
        return True
    # Accept short forms when long forms are required markers.
    short_aliases = {
        "--print": ("-p",),
        "--force": ("-f",),
        "--model": ("-m",),
        "--cd": ("-C",),
        "--config": ("-c",),
        "--approve": ("-a",),
        "--prompt": ("-p",),
    }
    for alias in short_aliases.get(marker, ()):
        if re.search(rf"(^|\s){re.escape(alias)}(\s|,|$)", help_text):
            return True
    return False


def _bounded_probe_output(executable: str, args: list[str]) -> str | None:
    try:
        completed = subprocess.run(
            [executable, *args],
            check=False,
            capture_output=True,
            text=True,
            timeout=_PROBE_TIMEOUT_SECONDS,
            shell=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    text = (completed.stdout or "") + (completed.stderr or "")
    text = text.strip()
    return text or None


_SENSITIVE_FLAG_RE = re.compile(
    r"^(?:--api-key|--token|--password|--secret|-H|--header|--authorization)$",
    re.IGNORECASE,
)


def _sanitize_display(
    argv: tuple[str, ...],
    *,
    prompt_values: set[str],
) -> tuple[str, ...]:
    out: list[str] = []
    redact_next = False
    for item in argv:
        if item in prompt_values:
            out.append(UNIT_PACKET_DISPLAY_MARKER)
            continue
        if redact_next:
            out.append("***")
            redact_next = False
            continue
        if _SENSITIVE_FLAG_RE.match(item):
            out.append(item)
            redact_next = True
            continue
        if SECRET_VALUE_RE.match(item.strip()):
            out.append("***")
            continue
        out.append(item)
    return tuple(out)


__all__ = [
    "MAX_UNIT_PACKET_BYTES",
    "UNIT_PACKET_DISPLAY_MARKER",
    "AdapterProbe",
    "AdapterSpec",
    "BUILTIN_SPECS",
    "Invocation",
    "build_invocation",
    "probe_adapter",
]