Agent-System-Dev/tests/test_agent_exec_adapters.py

"""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[:-2],
            (
                "agy",
                "--dangerously-skip-permissions",
                "--mode",
                "auto",
                "--output-format",
                "stream-json",
            ),
        )
        self.assertEqual(inv.argv[-2], "--print")
        _assert_bounded_prompt(self, inv.argv[-1], self.packet_body)
        self.assertNotIn("--model", inv.argv)
        self.assertIsNone(inv.stdin)
        self.assertEqual(inv.cwd, self.workspace)
        self.assertEqual(inv.sanitized_display[-1], UNIT_PACKET_MARKER)
        self.assertEqual(inv.sanitized_display[-2], "--print")
        self.assertNotIn(self.packet_body, " ".join(inv.sanitized_display))

    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()