"""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 json import re import shutil import subprocess from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path from typing import Any 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 = "" TERMINAL_STATUSES = ( "completed", "failed", "blocked", "cancelled", "failed-unknown", "scope-expansion", ) TERMINAL_RECORD_SCHEMA: dict[str, Any] = { "type": "object", "required": [ "terminal_status", "summary", "changed_files", "evidence", "scope_expansion", ], "properties": { "terminal_status": {"enum": list(TERMINAL_STATUSES)}, "summary": {"type": "string"}, "changed_files": {"type": "array", "items": {"type": "string"}}, "evidence": {"type": "array", "items": {"type": "string"}}, "scope_expansion": {"type": ["string", "null"]}, }, } # Bounds the backwards brace scan over one final message. _MAX_TERMINAL_TEXT_CHARS = 262_144 _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: schema = json.dumps(TERMINAL_RECORD_SCHEMA, separators=(",", ":")) 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" "--- TERMINAL RECORD ---\n\n" "End your final message with exactly one terminal record: a single-line " "JSON object that validates against this JSON Schema:\n" f"{schema}\n" "Use terminal_status completed only when the unit is fully implemented. " "Use blocked, cancelled, failed-unknown, failed, or scope-expansion " "otherwise. A missing or invalid terminal record fails the attempt.\n\n" "--- BOUNDED IMPLEMENTATION UNIT PACKET ---\n\n" f"{packet_text}" ) def extract_terminal_record(text: str | None) -> dict[str, Any] | None: """Return the sole terminal record that ends one final message. Only whitespace may follow the record, and no other valid record may precede it; anything else fails closed. """ if not isinstance(text, str) or not text: return None text = text[-_MAX_TERMINAL_TEXT_CHARS:].rstrip() decoder = json.JSONDecoder() def decode_at(pos: int) -> tuple[object, int]: try: return decoder.raw_decode(text, pos) except json.JSONDecodeError: return None, -1 pos = text.rfind("{") while pos != -1: candidate, end = decode_at(pos) if end == len(text) and isinstance(candidate, dict): break pos = text.rfind("{", 0, pos) else: return None record = _valid_terminal_record(candidate) if record is None: return None earlier = text.rfind("{", 0, pos) while earlier != -1: if _valid_terminal_record(decode_at(earlier)[0]) is not None: return None earlier = text.rfind("{", 0, earlier) return record def _valid_terminal_record(obj: object) -> dict[str, Any] | None: if not isinstance(obj, dict): return None status = obj.get("terminal_status") if not isinstance(status, str): return None status = status.strip().replace("_", "-") if status not in TERMINAL_STATUSES: return None if not isinstance(obj.get("summary"), str): return None for key in ("changed_files", "evidence"): items = obj.get(key) if not isinstance(items, list) or any(not isinstance(i, str) for i in items): return None if "scope_expansion" not in obj or not ( obj["scope_expansion"] is None or isinstance(obj["scope_expansion"], str) ): return None record = dict(obj) record["terminal_status"] = status return record def decode_terminal_record( adapter: str, *, event_lines: Sequence[str | None], stdout_tail: str, result_text: str | None, ) -> dict[str, Any] | None: """Decode the worker terminal record from the adapter's real output. ``event_lines`` are recent complete stdout lines that may carry a terminal envelope, oldest first; ``None`` marks an envelope line too large to keep. Event-stream adapters are read only from their final assistant/result envelope, so prompt echoes and tool output never count as a record. """ if adapter in ("cursor", "claude"): return extract_terminal_record(_cursor_final_text(event_lines)) if adapter == "antigravity": return extract_terminal_record(_agy_final_text(event_lines, stdout_tail)) if adapter == "pi": return extract_terminal_record(_pi_final_text(event_lines)) record = extract_terminal_record(result_text) if record is None: record = extract_terminal_record(stdout_tail) return record def _event_objects(event_lines: Sequence[str | None]) -> list[dict[str, Any] | None]: """Newest-first parsed events; ``None`` keeps the oversized-envelope marker.""" out: list[dict[str, Any] | None] = [] for line in reversed(event_lines): if line is None: out.append(None) continue try: obj = json.loads(line) except json.JSONDecodeError: continue if isinstance(obj, dict): out.append(obj) return out def _cursor_final_text(event_lines: Sequence[str | None]) -> str | None: for obj in _event_objects(event_lines): if obj is None: return None if obj.get("type") != "result": continue if obj.get("is_error") is not False or obj.get("subtype") != "success": return None text = obj.get("result") return text if isinstance(text, str) else None return None def _agy_final_text( event_lines: Sequence[str | None], stdout_tail: str ) -> str | None: candidates = _event_objects(event_lines) try: whole = json.loads(stdout_tail.strip()) except json.JSONDecodeError: whole = None if isinstance(whole, dict): candidates.append(whole) for obj in candidates: if obj is None: return None if obj.get("event") == "result": payload = obj.get("result") elif "event" not in obj and "response" in obj: payload = obj else: continue if not isinstance(payload, dict): return None if payload.get("is_error") or payload.get("error"): return None text = payload.get("response") return text if isinstance(text, str) else None return None def _pi_final_text(event_lines: Sequence[str | None]) -> str | None: for obj in _event_objects(event_lines): if obj is None: return None if obj.get("type") != "message_end": continue message = obj.get("message") if not isinstance(message, dict) or message.get("role") != "assistant": continue if message.get("stopReason") in ("error", "aborted"): return None content = message.get("content") if isinstance(content, str): return content if not isinstance(content, list): return None return "".join( block["text"] for block in content if isinstance(block, dict) and block.get("type") == "text" and isinstance(block.get("text"), str) ) return None 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", "TERMINAL_RECORD_SCHEMA", "TERMINAL_STATUSES", "UNIT_PACKET_DISPLAY_MARKER", "AdapterProbe", "AdapterSpec", "BUILTIN_SPECS", "Invocation", "build_invocation", "decode_terminal_record", "extract_terminal_record", "probe_adapter", ]