Add execution CLI adapter registry
Why: Hide provider-specific noninteractive, model, permission, workspace, prompt, and output differences behind one adapter seam. Evidence: Cursor observed red then 22 green config/adapter tests; the host replays them and whitespace checks during integration. Impact: Adds deterministic direct-argv adapters and probes without launching providers or enabling project routing. Refs: docs/superpowers/plans/2026-10-15-project-agent-execution.md
This commit is contained in:
@@ -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 = "<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=("--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",
|
||||
]
|
||||
@@ -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 = "<unit-packet>"
|
||||
|
||||
|
||||
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 <model>
|
||||
-f, --force
|
||||
--yolo
|
||||
--sandbox <mode> choices: enabled, disabled
|
||||
--trust
|
||||
--workspace <path>
|
||||
--output-format <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 <mode> bypassPermissions, manual, dontAsk
|
||||
--model <model>
|
||||
--output-format <format> text, json, stream-json
|
||||
"""
|
||||
|
||||
CODEX_HELP = """
|
||||
Usage: codex exec [OPTIONS] [PROMPT]
|
||||
exec
|
||||
-s, --sandbox <SANDBOX_MODE> read-only, workspace-write, danger-full-access
|
||||
-c, --config <key=value>
|
||||
-m, --model <MODEL>
|
||||
-C, --cd <DIR>
|
||||
--json
|
||||
-o, --output-last-message <FILE>
|
||||
"""
|
||||
|
||||
PI_HELP = """
|
||||
Usage: pi [options]
|
||||
--print, -p
|
||||
--approve, -a
|
||||
--model <pattern>
|
||||
--mode <mode> text, json, or rpc
|
||||
"""
|
||||
|
||||
KIMI_HELP = """
|
||||
Usage: kimi [options]
|
||||
-p, --prompt <prompt>
|
||||
--auto
|
||||
-m, --model <model>
|
||||
--output-format <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 <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()
|
||||
Reference in New Issue
Block a user