diff --git a/scripts/agent_exec_adapters.py b/scripts/agent_exec_adapters.py new file mode 100644 index 00000000..61290b55 --- /dev/null +++ b/scripts/agent_exec_adapters.py @@ -0,0 +1,486 @@ +"""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 = "" + +_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=("--print",), + run_everything_argv=( + "--dangerously-skip-permissions", + "--mode", + "auto", + ), + workspace_flag=None, + model_flag="--model", + output_argv=("--output-format", "stream-json"), + prompt_transport="positional", + ), + "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) + 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", +] diff --git a/tests/test_agent_exec_adapters.py b/tests/test_agent_exec_adapters.py new file mode 100644 index 00000000..c836eefd --- /dev/null +++ b/tests/test_agent_exec_adapters.py @@ -0,0 +1,511 @@ +"""Adapter argv, probe, and custom-expansion contract for agent-exec.""" +from __future__ import annotations + +import os +import re +import stat +import tempfile +import unittest +from pathlib import Path + +from scripts.agent_exec_adapters import ( + MAX_UNIT_PACKET_BYTES, + AdapterProbe, + Invocation, + build_invocation, + probe_adapter, +) +from scripts.agent_exec_config import ( + CustomAdapter, + ExecutionConfig, + ExecutionPolicy, + Profile, +) + + +UNIT_PACKET_MARKER = "" + + +def _policy() -> ExecutionPolicy: + return ExecutionPolicy( + fresh_process_per_unit=True, + permission="run-everything", + fallback_to_host="all-unavailable", + max_parallel=1, + timeout_seconds=60, + summary_bytes=1024, + raw_output_bytes=4096, + ) + + +def _config( + profiles: dict[str, Profile], + custom_adapters: dict[str, CustomAdapter] | None = None, +) -> ExecutionConfig: + names = tuple(profiles) + return ExecutionConfig( + schema="agent-execution/v1", + execution=_policy(), + roles={ + "implementation": names, + "repair": names[:1], + "test": names[:1], + "review": names[:1], + }, + profiles=profiles, + custom_adapters=custom_adapters or {}, + ) + + +def _write_fake_cli( + directory: Path, + name: str, + *, + help_text: str, + version: str = "0.0.1-test", +) -> Path: + path = directory / name + # Shebang must start at column 0; keep help body out of textwrap.dedent. + path.write_text( + "#!/bin/sh\n" + 'if [ "$1" = "--version" ] || [ "$1" = "-v" ] || [ "$1" = "-V" ]; then\n' + f' echo "{version}"\n' + " exit 0\n" + "fi\n" + "cat <<'EOF'\n" + f"{help_text.rstrip()}\n" + "EOF\n", + encoding="utf-8", + ) + path.chmod(path.stat().st_mode | stat.S_IEXEC) + return path + + +def _assert_bounded_prompt(test: unittest.TestCase, prompt: str, packet_body: str) -> None: + test.assertIn(packet_body, prompt) + test.assertNotEqual(prompt, packet_body) + lowered = prompt.lower() + test.assertRegex(lowered, r"implement exactly|bounded") + test.assertRegex(lowered, r"workspace") + test.assertRegex(lowered, r"uncommitted") + test.assertRegex(lowered, r"scope") + test.assertRegex(lowered, r"structured|final result") + + +CURSOR_HELP = """ +Usage: agent [options] [prompt...] + -p, --print + --model + -f, --force + --yolo + --sandbox choices: enabled, disabled + --trust + --workspace + --output-format text | json | stream-json +""" + +AGY_HELP = """ +Usage of agy: + --print + --dangerously-skip-permissions + --mode accept-edits, auto, plan + --model + --output-format text, json, stream-json +""" + +CLAUDE_HELP = """ +Usage: claude [options] [prompt] + -p, --print + --permission-mode bypassPermissions, manual, dontAsk + --model + --output-format text, json, stream-json +""" + +CODEX_HELP = """ +Usage: codex exec [OPTIONS] [PROMPT] + exec + -s, --sandbox read-only, workspace-write, danger-full-access + -c, --config + -m, --model + -C, --cd + --json + -o, --output-last-message +""" + +PI_HELP = """ +Usage: pi [options] + --print, -p + --approve, -a + --model + --mode text, json, or rpc +""" + +KIMI_HELP = """ +Usage: kimi [options] + -p, --prompt + --auto + -m, --model + --output-format text, stream-json +""" + + +class AgentExecAdapterTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-adapters-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.bindir = self.root / "bin" + self.bindir.mkdir() + self.workspace = self.root / "workspace" + self.workspace.mkdir() + self.unit_packet = self.root / "unit.md" + self.packet_body = "# unit\nsecret-token-should-not-leak-in-display\n" + self.unit_packet.write_text(self.packet_body, encoding="utf-8") + self.result_path = self.root / "result.json" + self._old_path = os.environ.get("PATH", "") + os.environ["PATH"] = f"{self.bindir}{os.pathsep}{self._old_path}" + self.addCleanup(self._restore_path) + + def _restore_path(self) -> None: + os.environ["PATH"] = self._old_path + + def test_cursor_run_everything_and_auto_model_argv(self) -> None: + _write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP) + profile = Profile( + name="cursor", + adapter="cursor", + executable="cursor-agent", + model=None, + enabled=True, + ) + cfg = _config({"cursor": profile}) + probe = probe_adapter(profile, cfg) + self.assertTrue(probe.available, probe.reason) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertIsInstance(inv, Invocation) + self.assertEqual(inv.cwd, self.workspace) + self.assertEqual( + inv.argv[:-1], + ( + "cursor-agent", + "-p", + "--force", + "--sandbox", + "enabled", + "--trust", + "--workspace", + str(self.workspace), + "--output-format", + "stream-json", + ), + ) + _assert_bounded_prompt(self, inv.argv[-1], self.packet_body) + self.assertNotIn(str(self.unit_packet), inv.argv) + self.assertNotIn("--model", inv.argv) + self.assertIsNone(inv.stdin) + self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER) + self.assertNotIn(self.packet_body, " ".join(inv.sanitized_display)) + self.assertNotIn(inv.argv[-1], inv.sanitized_display) + + def test_antigravity_run_everything_argv(self) -> None: + _write_fake_cli(self.bindir, "agy", help_text=AGY_HELP) + profile = Profile( + name="antigravity", + adapter="antigravity", + executable="agy", + model=None, + enabled=True, + ) + cfg = _config({"antigravity": profile}) + self.assertTrue(probe_adapter(profile, cfg).available) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertEqual( + inv.argv[:-1], + ( + "agy", + "--print", + "--dangerously-skip-permissions", + "--mode", + "auto", + "--output-format", + "stream-json", + ), + ) + _assert_bounded_prompt(self, inv.argv[-1], self.packet_body) + self.assertEqual(inv.cwd, self.workspace) + self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER) + + def test_claude_run_everything_argv(self) -> None: + _write_fake_cli(self.bindir, "claude", help_text=CLAUDE_HELP) + profile = Profile( + name="claude", + adapter="claude", + executable="claude", + model=None, + enabled=True, + ) + cfg = _config({"claude": profile}) + self.assertTrue(probe_adapter(profile, cfg).available) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertEqual( + inv.argv[:-1], + ( + "claude", + "-p", + "--permission-mode", + "bypassPermissions", + "--output-format", + "stream-json", + ), + ) + _assert_bounded_prompt(self, inv.argv[-1], self.packet_body) + self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER) + + def test_codex_uses_noninteractive_workspace_sandbox(self) -> None: + _write_fake_cli(self.bindir, "codex", help_text=CODEX_HELP) + profile = Profile( + name="codex", + adapter="codex", + executable="codex", + model=None, + enabled=True, + ) + cfg = _config({"codex": profile}) + self.assertTrue(probe_adapter(profile, cfg).available) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertEqual( + inv.argv[:-1], + ( + "codex", + "exec", + "--sandbox", + "workspace-write", + "-c", + "approval_policy=never", + "--cd", + str(self.workspace), + "--json", + "--output-last-message", + str(self.result_path), + ), + ) + _assert_bounded_prompt(self, inv.argv[-1], self.packet_body) + self.assertNotIn("--dangerously-bypass-approvals-and-sandbox", inv.argv) + self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER) + + def test_pi_and_kimi_auto_model_omit_model_flag(self) -> None: + _write_fake_cli(self.bindir, "pi", help_text=PI_HELP) + _write_fake_cli(self.bindir, "kimi", help_text=KIMI_HELP) + pi = Profile( + name="pi", adapter="pi", executable="pi", model=None, enabled=True + ) + kimi = Profile( + name="kimi", adapter="kimi", executable="kimi", model=None, enabled=True + ) + cfg = _config({"pi": pi, "kimi": kimi}) + pi_inv = build_invocation( + pi, cfg, self.workspace, self.unit_packet, self.result_path + ) + kimi_inv = build_invocation( + kimi, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertEqual( + pi_inv.argv[:-1], + ( + "pi", + "-p", + "--approve", + "--mode", + "json", + ), + ) + _assert_bounded_prompt(self, pi_inv.argv[-1], self.packet_body) + self.assertNotIn("--model", pi_inv.argv) + self.assertEqual( + kimi_inv.argv[:-1], + ( + "kimi", + "--auto", + "--output-format", + "stream-json", + "--prompt", + ), + ) + _assert_bounded_prompt(self, kimi_inv.argv[-1], self.packet_body) + self.assertNotIn("--model", kimi_inv.argv) + self.assertNotIn("-m", kimi_inv.argv) + self.assertEqual(kimi_inv.sanitized_display[-1], UNIT_PACKET_MARKER) + self.assertNotIn(self.packet_body, " ".join(kimi_inv.sanitized_display)) + self.assertNotIn(kimi_inv.argv[-1], kimi_inv.sanitized_display) + + def test_missing_opencode_is_unavailable(self) -> None: + profile = Profile( + name="opencode", + adapter="opencode", + executable="opencode", + model=None, + enabled=True, + ) + cfg = _config({"opencode": profile}) + # Ensure no opencode on PATH from bindir; real PATH may also lack it. + probe = probe_adapter(profile, cfg) + self.assertIsInstance(probe, AdapterProbe) + self.assertFalse(probe.available) + self.assertIsNotNone(probe.reason) + + def test_fixed_model_uses_only_adapter_model_flag(self) -> None: + _write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP) + profile = Profile( + name="cursor", + adapter="cursor", + executable="cursor-agent", + model="gpt-5", + enabled=True, + ) + cfg = _config({"cursor": profile}) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertIn("--model", inv.argv) + model_idx = inv.argv.index("--model") + self.assertEqual(inv.argv[model_idx + 1], "gpt-5") + self.assertEqual(inv.argv.count("--model"), 1) + self.assertNotIn("auto", inv.argv) + _assert_bounded_prompt(self, inv.argv[-1], self.packet_body) + + def test_custom_adapter_builds_direct_argv_without_shell(self) -> None: + _write_fake_cli( + self.bindir, + "mycli", + help_text="Usage: mycli --yes --cwd --model --output-format", + ) + profile = Profile( + name="mycustom", + adapter="custom", + executable="mycli", + model="fixed-model", + enabled=True, + ) + custom = CustomAdapter( + name="mycustom", + base_argv=("run", "${unit_packet}"), + run_everything_argv=("--yes",), + workspace_argv=("--cwd", "${workspace}"), + model_argv=("--model", "${model}"), + prompt_transport="stdin", + output_format="json", + ) + cfg = _config({"mycustom": profile}, {"mycustom": custom}) + inv = build_invocation( + profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertEqual( + inv.argv, + ( + "mycli", + "run", + str(self.unit_packet), + "--yes", + "--cwd", + str(self.workspace), + "--model", + "fixed-model", + "--output-format", + "json", + ), + ) + self.assertEqual(inv.stdin, self.unit_packet.read_bytes()) + self.assertNotIn("&&", " ".join(inv.argv)) + self.assertNotIn("|", " ".join(inv.argv)) + # Path placeholder is fine; body must not appear in display. + self.assertNotIn(self.packet_body.strip(), " ".join(inv.sanitized_display)) + # auto model must omit the whole model_argv pair + auto_profile = Profile( + name="mycustom", + adapter="custom", + executable="mycli", + model=None, + enabled=True, + ) + auto_inv = build_invocation( + auto_profile, cfg, self.workspace, self.unit_packet, self.result_path + ) + self.assertNotIn("--model", auto_inv.argv) + self.assertNotIn("auto", auto_inv.argv) + + # Positional custom transport may embed the bounded prompt, but display + # must use the stable marker. + positional = CustomAdapter( + name="mycustom", + base_argv=("run",), + run_everything_argv=("--yes",), + workspace_argv=("--cwd", "${workspace}"), + model_argv=("--model", "${model}"), + prompt_transport="positional", + output_format="json", + ) + pos_cfg = _config({"mycustom": profile}, {"mycustom": positional}) + pos_inv = build_invocation( + profile, pos_cfg, self.workspace, self.unit_packet, self.result_path + ) + _assert_bounded_prompt(self, pos_inv.argv[-1], self.packet_body) + self.assertEqual(pos_inv.sanitized_display[-1], UNIT_PACKET_MARKER) + self.assertNotIn(pos_inv.argv[-1], pos_inv.sanitized_display) + + def test_unverified_run_everything_capability_is_unavailable(self) -> None: + _write_fake_cli( + self.bindir, + "cursor-agent", + help_text="Usage: agent\n -p, --print\n --model \n", + ) + profile = Profile( + name="cursor", + adapter="cursor", + executable="cursor-agent", + model=None, + enabled=True, + ) + cfg = _config({"cursor": profile}) + probe = probe_adapter(profile, cfg) + self.assertFalse(probe.available) + self.assertIsNotNone(probe.reason) + self.assertRegex(probe.reason or "", r"unavailable|missing|capability|flag", re.I) + + def test_rejects_oversized_and_invalid_utf8_unit_packet(self) -> None: + _write_fake_cli(self.bindir, "cursor-agent", help_text=CURSOR_HELP) + profile = Profile( + name="cursor", + adapter="cursor", + executable="cursor-agent", + model=None, + enabled=True, + ) + cfg = _config({"cursor": profile}) + + oversized = self.root / "oversized.md" + oversized.write_bytes(b"x" * (MAX_UNIT_PACKET_BYTES + 1)) + with self.assertRaises(ValueError) as oversized_ctx: + build_invocation( + profile, cfg, self.workspace, oversized, self.result_path + ) + self.assertRegex(str(oversized_ctx.exception).lower(), r"size|large|bound|limit") + + invalid = self.root / "invalid.md" + invalid.write_bytes(b"\xff\xfe not utf-8") + with self.assertRaises(ValueError) as utf8_ctx: + build_invocation( + profile, cfg, self.workspace, invalid, self.result_path + ) + self.assertRegex(str(utf8_ctx.exception).lower(), r"utf-8|utf8|encoding|decode") + + +if __name__ == "__main__": + unittest.main()