diff --git a/docs/release-qualification.md b/docs/release-qualification.md index 71308444..21ff7c56 100644 --- a/docs/release-qualification.md +++ b/docs/release-qualification.md @@ -117,10 +117,10 @@ Local CLI providers (`local_cli` driver) only run automatically if a `campaign_a |---|---|---| | `codex` | `codex` | `codex exec` with stdin (`-`), `--ephemeral`, `--approve-for-me`, a Code Mower-owned root-deny/workspace-write profile, keyring-only auth, `--skip-git-repo-check`, `--json`, `--output-schema`, `--output-last-message`, `-C` (codex-cli 0.147.0) | | `claude_audit` | `claude` | `claude --print` with stdin, `--output-format json`, strict OS sandbox and PyPI allowlist, `--json-schema` (Claude Code 2.1.258) | -| `antigravity_cli` | `antigravity` | `agy --print` with a prompt file, `--sandbox`, noninteractive permission approval, `--add-dir`, `--print-timeout` (agy 1.1.26) | +| `antigravity_cli` | `antigravity` | `agy --print` with a prompt file, `--sandbox`, noninteractive permission approval, `--new-project`, `--add-dir`, `--print-timeout` | | `muse_cli` | `muse` | `muse exec` with `--json`, `--prompt-file`, `--workspace` (Muse Code 1.0.3) | -Newer CLIs keep working while the flags exist; a removed flag fails closed. Prompts travel on stdin or a prompt file, never through a shell. Provider children receive an allowlisted environment with home/config locations for their stored login, but no ambient GitHub, Code Mower cloud, or provider API keys; Muse's explicit API key travels on stdin only. Provider stdout/stderr are parsed transiently and never persisted: only a closed, validated `code_mower.adoptionResult.v1` document is written to `{output}`. Codex gets outbound network access inside its ephemeral workspace-write sandbox for package downloads; its CLI does not provide a domain allowlist here. Claude requires its OS sandbox, disables the unsandboxed escape hatch, denies Bash access to the operator's home directory, and allows package-download network access only to PyPI. Antigravity runs with its CLI sandbox retained and noninteractive permission prompts auto-approved inside that sandbox; this is required because headless `agy --print` cannot answer command prompts. Antigravity/Muse refuse without their ambient-home opt-in (`ANTIGRAVITY_CLI_USE_AMBIENT_HOME` / `MUSE_CLI_USE_AMBIENT_HOME`) or a provider key, mirroring the audit wrappers. +Newer CLIs keep working while the flags exist; a removed flag fails closed. Prompts travel on stdin or a prompt file, never through a shell. Provider children receive an allowlisted environment with home/config locations for their stored login, but no ambient GitHub, Code Mower cloud, or provider API keys; Muse's explicit API key travels on stdin only. Provider stdout/stderr are parsed transiently and never persisted: only a closed, validated `code_mower.adoptionResult.v1` document is written to `{output}`. Codex gets outbound network access inside its ephemeral workspace-write sandbox for package downloads; its CLI does not provide a domain allowlist here. Claude requires its OS sandbox, disables the unsandboxed escape hatch, denies Bash access to the operator's home directory, and allows package-download network access only to PyPI. Antigravity runs with its CLI sandbox retained, `--new-project` passed on every campaign run for project boundary isolation (strictly excluding continue or resume semantics: `--continue`, `-c`, `--conversation`, `--resume`, `-i`, `--prompt-interactive`), and noninteractive permission prompts auto-approved inside that sandbox; this is required because headless `agy --print` cannot answer command prompts. Before writing prompt input or invoking provider work, the adapter checks installed `agy` CLI capability via `--help` for `--new-project` support and fails closed if unsupported. Antigravity/Muse refuse without their ambient-home opt-in (`ANTIGRAVITY_CLI_USE_AMBIENT_HOME` / `MUSE_CLI_USE_AMBIENT_HOME`) or a provider key, mirroring the audit wrappers. Timeout model: each maintained lane sets `campaign_adapter_timeout_seconds: 900`. The campaign passes the outer timeout minus `ADAPTER_INNER_TIMEOUT_MARGIN_SECONDS` (30s) as the adapter's `{adapter_timeout}`, so the adapter's own provider budget always fires first. `{python}` resolves to the running interpreter (launching the adapter); `{target_python}` to the deterministically resolved Python 3.12+ runtime binary passed to `--python-bin`; `{command}` to the installed provider CLI (the campaign refuses to run when it is missing). @@ -158,6 +158,7 @@ An installed CLI and a valid argv contract do not prove the isolated home the ad - **Codex isolated auth probe**: Where Codex exposes `codex login status` (20s budget), doctor probes the isolated home. Authenticated passes; confirmed logged-out produces an actionable warning and removes Codex from `ready_providers`. Timeouts or probe errors degrade safely to a skip. - **Antigravity & Muse ambient-home opt-ins**: Antigravity and Muse require trusted ambient-home opt-ins (`ANTIGRAVITY_CLI_USE_AMBIENT_HOME=1`, `MUSE_CLI_USE_AMBIENT_HOME=1` or `META_API_KEY`/`META_API_KEY_FILE`). Doctor models these requirements directly: missing opt-ins produce an actionable warning and prevent the provider from being reported `campaign-ready`. +- **Antigravity campaign isolation & --new-project capability**: Every maintained Antigravity campaign invocation includes `--new-project` and excludes continue/resume flags (`--continue`, `-c`, `--conversation`, `--resume`, `-i`, `--prompt-interactive`) to guarantee session isolation inside a fresh project boundary. Both the adapter and doctor campaign readiness call the same bounded capability check (`agy --help`) before writing prompts, invoking provider work, or declaring Antigravity ready. When `--new-project` is unsupported, doctor excludes Antigravity from `ready_providers` with bounded actionable metadata and remediation instructing to upgrade to a version whose `--help` exposes `--new-project` (without claiming an unverified minimum version). Probe output, prompts, paths, and secrets are strictly excluded from evidence. - **Structured-result capability**: Doctor distinguishes executable/auth readiness from structured-result capability using a bounded offline fixture (zero token spend, zero network). `doctor.campaign.readiness` detail breaks down `command`, `auth`, and `structured_result` per provider without leaking paths or command output. - **Deterministic Python 3.12+ runtime resolution**: Before invoking local provider adapters, the campaign runner resolves a supported Python 3.12+ executable (probing `CODE_MOWER_PYTHON`, running interpreter, and versioned `python3.12+` binaries on PATH) and passes exact `--python-bin` and `--target-runtime` arguments. Providers must not pick ambient `python3`. If no supported runtime exists, the dispatch fails closed with `python_runtime_unavailable` and actionable remediation. Result validators enforce `runtime_class >= python_3.12`. diff --git a/src/code_mower/campaign_adapters.py b/src/code_mower/campaign_adapters.py index bf6e771f..69570882 100644 --- a/src/code_mower/campaign_adapters.py +++ b/src/code_mower/campaign_adapters.py @@ -34,9 +34,9 @@ ``--json``, schema/last-message output, ``-C`` ``claude`` Claude Code 2.1.258 ``--print`` with stdin, ``--output-format json``, explicit tool/permission controls, ``--json-schema`` -``antigravity`` agy 1.1.26 ``--print`` with a prompt file, ``--sandbox``, - noninteractive approval, ``--add-dir``, - ``--print-timeout`` +``antigravity`` runtime-probed ``--print`` with a prompt file, ``--sandbox``, + noninteractive approval, ``--new-project``, + ``--add-dir``, ``--print-timeout`` ``muse`` Muse Code 1.0.3 ``exec`` with ``--json``, ``--prompt-file``, ``--workspace`` ================= ==================== ==================================================== @@ -96,7 +96,6 @@ VERIFIED_CLI_VERSIONS = { "codex": "codex-cli 0.147.0", "claude": "Claude Code 2.1.258", - "antigravity": "agy 1.1.26", "muse": "Muse Code 1.0.3", } @@ -478,18 +477,21 @@ def build_antigravity_argv( timeout_seconds: int, model: str = "", ) -> list[str]: - """Argv for ``agy --print`` with prompt-file transport. + """Argv for ``agy --print`` with prompt-file transport and project isolation. Mirrors the Antigravity audit wrapper (via the Gemini CLI wrapper): the prompt lives in a file inside the workspace and the agent is pointed at it with a short instruction, sandboxed to that workspace. Headless agy cannot prompt for command permission, so permission checks are auto-approved only - inside that retained sandbox. + inside that retained sandbox. Every qualification run passes + ``--new-project`` so the session executes inside a fresh project boundary + and never inherits active conversations or resume semantics. """ argv = [ agy_bin, "--sandbox", "--dangerously-skip-permissions", + "--new-project", "--add-dir", workspace_dir, "--print-timeout", @@ -668,6 +670,124 @@ def check_structured_result_capability(provider: str) -> bool: return False +def check_antigravity_readiness( + agy_bin: str, + *, + runner: Callable[..., subprocess.CompletedProcess[str]] | None = None, + timeout_seconds: int = 10, + child_env: Mapping[str, str] | None = None, +) -> dict[str, Any]: + """Detect whether installed agy CLI supports --new-project for campaign isolation. + + Fails closed with bounded actionable metadata (never leaking stdout/stderr, + prompts, paths, auth details, or secrets) when the installed agy lacks the flag. + """ + if not agy_bin: + return { + "ready": False, + "provider": "antigravity", + "capability": "new_project", + "required_flag": "--new-project", + "error": "command_not_found", + "actionable": True, + "message": "antigravity CLI is not installed", + "remediation": "Install agy CLI on PATH or specify the executable in code-mower.yml.", + } + + probe_env = dict(child_env) if child_env is not None else build_adapter_child_env("antigravity") + probe_error = "" + try: + if runner is not None: + try: + completed = runner( + [agy_bin, "--help"], + timeout=timeout_seconds, + env=probe_env, + ) + except TypeError: + completed = runner([agy_bin, "--help"]) + else: + completed = subprocess.run( + [agy_bin, "--help"], + capture_output=True, + text=True, + check=False, + timeout=timeout_seconds, + env=probe_env, + ) + output = (completed.stdout or "") + (completed.stderr or "") + if completed.returncode != 0: + probe_error = "capability_probe_failed" + has_new_project = not probe_error and ("--new-project" in output.split()) + except subprocess.TimeoutExpired: + probe_error = "capability_probe_timeout" + has_new_project = False + except OSError: + probe_error = "capability_probe_failed" + has_new_project = False + + if probe_error: + timed_out = probe_error == "capability_probe_timeout" + return { + "ready": False, + "provider": "antigravity", + "capability": "new_project", + "required_flag": "--new-project", + "error": probe_error, + "actionable": True, + "message": ( + "antigravity CLI capability probe timed out" + if timed_out + else "antigravity CLI capability probe failed" + ), + "remediation": ( + "Retry the bounded agy --help probe; if it continues to time out, " + "repair the CLI installation before running a campaign." + if timed_out + else "Verify that agy --help runs successfully, then retry campaign readiness." + ), + } + + if not has_new_project: + return { + "ready": False, + "provider": "antigravity", + "capability": "new_project", + "required_flag": "--new-project", + "error": "missing_new_project_capability", + "actionable": True, + "message": "installed agy CLI lacks required --new-project flag for campaign isolation", + "remediation": "Upgrade agy CLI to a version whose --help exposes --new-project.", + } + + return { + "ready": True, + "provider": "antigravity", + "capability": "new_project", + "required_flag": "--new-project", + "error": "", + "actionable": False, + "message": "agy CLI supports --new-project campaign isolation", + "remediation": "", + } + + +def check_antigravity_new_project_capability( + agy_bin: str, + *, + runner: Callable[..., subprocess.CompletedProcess[str]] | None = None, + timeout_seconds: int = 10, +) -> bool: + """Return True if installed agy supports --new-project, False otherwise.""" + return bool( + check_antigravity_readiness( + agy_bin, + runner=runner, + timeout_seconds=timeout_seconds, + ).get("ready") + ) + + def validate_bound_result( candidate: Any, *, @@ -865,6 +985,7 @@ def run_campaign_adapter( python_bin: str = "", target_runtime: str = "", provider_runner: ProviderRunner = run_provider_command, + capability_runner: Callable[..., subprocess.CompletedProcess[str]] | None = None, ) -> int: """Run one maintained provider adapter. Returns a process exit code. @@ -986,6 +1107,13 @@ def run_campaign_adapter( ) candidate = _extract_claude_result(completed.stdout) elif provider == "antigravity": + readiness = check_antigravity_readiness( + resolved_bin, + runner=capability_runner, + child_env=child_env, + ) + if not readiness["ready"]: + return _fail(provider, readiness["message"]) prompt_path = workspace_dir / "campaign.prompt-input.txt" prompt_path.write_text(prompt, encoding="utf-8") agy_model = model or _first_env_value(ANTIGRAVITY_MODEL_ENV_NAMES) diff --git a/src/code_mower/doctor_checks/adoption.py b/src/code_mower/doctor_checks/adoption.py index 5990b2e4..cdfd1910 100644 --- a/src/code_mower/doctor_checks/adoption.py +++ b/src/code_mower/doctor_checks/adoption.py @@ -577,6 +577,18 @@ def _resolve_adapter_config_for_lane( return _resolve_campaign_adapter_config(lane, repo_root) +def _is_maintained_antigravity_adapter( + lane: Any, + argv_template: Sequence[str] | None, +) -> bool: + if not argv_template: + return False + maintained = lane.provider_config.get("campaign_adapter_argv") + if maintained and tuple(argv_template) == tuple(maintained): + return True + return any("code_mower.campaign_adapters" in str(token) for token in argv_template) + + def check_adoption_campaign_readiness( *, config: Mapping[str, Any] | None, @@ -587,12 +599,16 @@ def check_adoption_campaign_readiness( which_fn: Callable[[str], str | None] = shutil.which, command_runner: Any = None, auth_probe_runner: Any = None, + capability_runner: Any = None, token_dir: Path | None = None, providers: Sequence[str] = DEFAULT_CAMPAIGN_PROVIDERS, ) -> tuple[DoctorCheck, ...]: """Validate release campaign readiness across configured providers and storage.""" from code_mower import lane_status - from code_mower.campaign_adapters import check_structured_result_capability + from code_mower.campaign_adapters import ( + check_antigravity_readiness, + check_structured_result_capability, + ) from code_mower.cloud import resolve_cloud_token from code_mower.release_campaigns import ( _check_credentials, @@ -804,6 +820,53 @@ def check_adoption_campaign_readiness( ) ) else: + capability_check = None + if canonical == "antigravity" and ( + capability_runner is not None + or _is_maintained_antigravity_adapter(lane, argv_template) + ): + cap_runner = capability_runner if capability_runner is not None else command_runner + cap_result = check_antigravity_readiness( + which_fn(cmd) or cmd, + runner=cap_runner, + ) + if not cap_result.get("ready"): + capability_check = cap_result + + if capability_check is not None: + detail = { + "provider": canonical, + "lane": lane.lane_id, + "driver": lane.driver, + "command": command_name, + "command_found": True, + "adapter_configured": True, + "capability": capability_check.get("capability", "new_project"), + "required_flag": capability_check.get("required_flag", "--new-project"), + "error": capability_check.get("error", "missing_new_project_capability"), + "enabled": is_enabled, + "actionable": is_enabled, + "optional": not is_enabled, + } + if is_enabled: + detail["owner_action"] = True + provider_readiness[canonical] = { + "command": True, + "auth": "unknown", + "structured_result": structured_capability, + } + checks.append( + DoctorCheck( + name="doctor.campaign.adapter", + status=STATUS_WARN, + lane=canonical, + message=str(capability_check.get("message")), + detail=detail, + remediation=str(capability_check.get("remediation")), + ) + ) + continue + checks.append( DoctorCheck( name="doctor.campaign.adapter", diff --git a/tests/test_campaign_adapters.py b/tests/test_campaign_adapters.py index 8944cc1e..d95ed030 100644 --- a/tests/test_campaign_adapters.py +++ b/tests/test_campaign_adapters.py @@ -59,7 +59,16 @@ def _adoption_result(provider: str = "codex", **overrides: Any) -> dict[str, Any def _fake_bin(tmp: Path, name: str = "provider-bin") -> str: bin_path = tmp / name - bin_path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + bin_path.write_text( + "#!/bin/sh\n" + "if [ \"$1\" = \"--help\" ]; then\n" + " echo \" --new-project Create a new project\"\n" + " exit 0\n" + "fi\n" + "exit 0\n", + encoding="utf-8", + ) + bin_path.chmod(0o755) return str(bin_path) @@ -179,11 +188,38 @@ def test_antigravity_argv_uses_prompt_file_and_timeout(self) -> None: self.assertIn("--print", argv) self.assertIn("--sandbox", argv) self.assertIn("--dangerously-skip-permissions", argv) + self.assertIn("--new-project", argv) self.assertIn("--print-timeout", argv) self.assertIn("870s", argv) self.assertIn("--model", argv) self.assertIn("/tmp/work/ws/campaign.prompt-input.txt", argv[-1]) + def test_antigravity_argv_isolation_contract_no_continue_or_resume_semantics(self) -> None: + argv = campaign_adapters.build_antigravity_argv( + agy_bin="/bin/agy", + workspace_dir="/tmp/work/ws", + prompt_file="/tmp/work/ws/campaign.prompt-input.txt", + timeout_seconds=870, + model="gemini-3", + ) + self.assertEqual(argv[0], "/bin/agy") + self.assertIn("--sandbox", argv) + self.assertIn("--dangerously-skip-permissions", argv) + self.assertIn("--new-project", argv) + add_dir_idx = argv.index("--add-dir") + self.assertEqual(argv[add_dir_idx + 1], "/tmp/work/ws") + timeout_idx = argv.index("--print-timeout") + self.assertEqual(argv[timeout_idx + 1], "870s") + self.assertIn("--print", argv) + + # Proving every run starts a new project and does not use continue or resume semantics: + self.assertNotIn("--continue", argv) + self.assertNotIn("-c", argv) + self.assertNotIn("--conversation", argv) + self.assertNotIn("--resume", argv) + self.assertNotIn("--prompt-interactive", argv) + self.assertNotIn("-i", argv) + def test_shared_prompt_pins_virtualenv_interpreter_commands(self) -> None: prompt = campaign_adapters.build_qualification_prompt( provider="codex", @@ -564,6 +600,145 @@ def runner( self.assertEqual(seen["timeout"], 870) self.assertTrue(output.is_file()) + def test_check_antigravity_readiness_passes_with_new_project_in_help(self) -> None: + def fake_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + self.assertEqual(argv, ["/bin/agy", "--help"]) + return subprocess.CompletedProcess(argv, 0, stdout=" --new-project Run inside fresh project\n", stderr="") + + res = campaign_adapters.check_antigravity_readiness("/bin/agy", runner=fake_runner) + self.assertTrue(res["ready"]) + self.assertEqual(res["provider"], "antigravity") + self.assertEqual(res["capability"], "new_project") + self.assertEqual(res["required_flag"], "--new-project") + self.assertFalse(res["actionable"]) + self.assertTrue(campaign_adapters.check_antigravity_new_project_capability("/bin/agy", runner=fake_runner)) + + def test_check_antigravity_readiness_fails_without_new_project(self) -> None: + def fake_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(argv, 0, stdout=" --continue Resume session\n", stderr="") + + res = campaign_adapters.check_antigravity_readiness("/bin/agy", runner=fake_runner) + self.assertFalse(res["ready"]) + self.assertEqual(res["provider"], "antigravity") + self.assertEqual(res["capability"], "new_project") + self.assertEqual(res["required_flag"], "--new-project") + self.assertEqual(res["error"], "missing_new_project_capability") + self.assertTrue(res["actionable"]) + self.assertEqual( + res["remediation"], + "Upgrade agy CLI to a version whose --help exposes --new-project.", + ) + self.assertNotIn("1.1.26", res["remediation"]) + # Privacy: help text, paths, prompts, secrets must not leak in readiness dict: + serialized = json.dumps(res) + self.assertNotIn("--continue", serialized) + self.assertNotIn("/bin/agy", serialized) + self.assertFalse(campaign_adapters.check_antigravity_new_project_capability("/bin/agy", runner=fake_runner)) + + def test_check_antigravity_readiness_rejects_flag_name_substrings(self) -> None: + def fake_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + argv, + 0, + stdout=" --new-project-template Select a project template\n", + stderr="", + ) + + result = campaign_adapters.check_antigravity_readiness( + "/bin/agy", + runner=fake_runner, + ) + self.assertFalse(result["ready"]) + + def test_check_antigravity_readiness_production_probe_uses_allowlisted_env(self) -> None: + seen: dict[str, Any] = {} + + def fake_run(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + seen.update(kwargs) + return subprocess.CompletedProcess( + argv, + 0, + stdout=" --new-project Create a new project\n", + stderr="", + ) + + with mock.patch.dict( + os.environ, + { + "GITHUB_TOKEN": "secret-github-token", + "CODE_MOWER_CLOUD_TOKEN": "secret-cloud-token", + "META_API_KEY": "secret-provider-key", + }, + clear=False, + ), mock.patch.object(campaign_adapters.subprocess, "run", side_effect=fake_run): + result = campaign_adapters.check_antigravity_readiness("/bin/agy") + + self.assertTrue(result["ready"]) + probe_env = seen["env"] + self.assertNotIn("GITHUB_TOKEN", probe_env) + self.assertNotIn("CODE_MOWER_CLOUD_TOKEN", probe_env) + self.assertNotIn("META_API_KEY", probe_env) + + def test_check_antigravity_readiness_fails_on_empty_or_nonexistent_command(self) -> None: + res_empty = campaign_adapters.check_antigravity_readiness("") + self.assertFalse(res_empty["ready"]) + self.assertEqual(res_empty["error"], "command_not_found") + self.assertIn("Install agy CLI on PATH", res_empty["remediation"]) + + def failing_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + raise OSError("No such file or directory") + + res_oserror = campaign_adapters.check_antigravity_readiness("/bin/nonexistent-agy", runner=failing_runner) + self.assertFalse(res_oserror["ready"]) + self.assertEqual(res_oserror["error"], "capability_probe_failed") + self.assertIn("agy --help runs successfully", res_oserror["remediation"]) + + def test_check_antigravity_readiness_distinguishes_probe_timeout(self) -> None: + def timeout_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + raise subprocess.TimeoutExpired(argv, 10) + + result = campaign_adapters.check_antigravity_readiness( + "/bin/agy", + runner=timeout_runner, + ) + self.assertFalse(result["ready"]) + self.assertEqual(result["error"], "capability_probe_timeout") + self.assertIn("timed out", result["message"]) + self.assertNotIn("upgrade", result["remediation"].lower()) + + def test_run_campaign_adapter_antigravity_fails_before_prompt_or_provider_call(self) -> None: + def failing_cap_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(argv, 0, stdout="help without new project", stderr="") + + provider_invoked = False + + def provider_runner(*args: Any, **kwargs: Any) -> Any: + nonlocal provider_invoked + provider_invoked = True + raise AssertionError("provider_runner should not have been called") + + with tempfile.TemporaryDirectory() as tmp_str: + tmp = Path(tmp_str) + output = tmp / "result.json" + env = {campaign_adapters.ANTIGRAVITY_AMBIENT_HOME_ENV: "1"} + with mock.patch.dict(os.environ, env, clear=False): + code = campaign_adapters.run_campaign_adapter( + provider="antigravity", + provider_bin="/bin/agy", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + qualification_context="cold_install", + starting_version="", + output_path=output, + provider_runner=provider_runner, + capability_runner=failing_cap_runner, + ) + self.assertEqual(code, 1) + self.assertFalse(provider_invoked) + # Ensure prompt file was never written + prompt_files = list(tmp.glob("campaign.prompt-input.txt")) + self.assertEqual(len(prompt_files), 0) + def test_muse_reads_jsonl_events_with_prompt_file(self) -> None: seen: dict[str, Any] = {} diff --git a/tests/test_doctor_campaign_readiness.py b/tests/test_doctor_campaign_readiness.py index 76b49a66..2ee59b63 100644 --- a/tests/test_doctor_campaign_readiness.py +++ b/tests/test_doctor_campaign_readiness.py @@ -2,9 +2,12 @@ from __future__ import annotations +from collections.abc import Sequence +import json from pathlib import Path import subprocess import tempfile +from typing import Any import unittest from unittest import mock @@ -846,6 +849,9 @@ def fake_capability(provider: str) -> bool: with mock.patch( "code_mower.campaign_adapters.check_structured_result_capability", side_effect=fake_capability, + ), mock.patch( + "code_mower.campaign_adapters.check_antigravity_readiness", + return_value={"ready": True, "provider": "antigravity", "capability": "new_project"}, ): checks = check_adoption_campaign_readiness( config=config, @@ -1026,6 +1032,183 @@ def test_disabled_auth_probe_preserves_aggregate_readiness_for_ambient_providers ] self.assertEqual(warn_checks, []) + def test_antigravity_campaign_readiness_passes_with_new_project_capability(self) -> None: + """When agy --help contains --new-project, Antigravity passes adapter check and is campaign ready.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "antigravity_cli": { + "enabled": True, + } + } + } + + def cap_runner(argv: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + list(argv), 0, stdout=" --new-project Create new project\n", stderr="" + ) + + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "agy" else None, + capability_runner=cap_runner, + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + providers=["antigravity"], + ) + + adapter_check = next( + c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "antigravity" + ) + self.assertEqual(adapter_check.status, STATUS_PASS) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness.status, STATUS_PASS) + self.assertEqual(readiness.detail.get("ready_providers"), ["antigravity"]) + self.assertEqual(readiness.detail.get("actionable_providers"), []) + + def test_antigravity_campaign_readiness_fails_without_new_project_capability(self) -> None: + """When agy lacks --new-project, Antigravity is excluded from ready_providers with actionable warning.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "antigravity_cli": { + "enabled": True, + } + } + } + + def cap_runner(argv: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + list(argv), 0, stdout=" --continue Resume existing session\n", stderr="" + ) + + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "agy" else None, + capability_runner=cap_runner, + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + providers=["antigravity"], + ) + + adapter_check = next( + c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "antigravity" + ) + self.assertEqual(adapter_check.status, STATUS_WARN) + self.assertEqual( + adapter_check.message, + "installed agy CLI lacks required --new-project flag for campaign isolation", + ) + self.assertEqual( + adapter_check.remediation, + "Upgrade agy CLI to a version whose --help exposes --new-project.", + ) + self.assertNotIn("1.1.26", adapter_check.remediation) + + detail = adapter_check.detail + self.assertEqual(detail["provider"], "antigravity") + self.assertEqual(detail["lane"], "antigravity_cli") + self.assertEqual(detail["command"], "agy") + self.assertTrue(detail["command_found"]) + self.assertTrue(detail["adapter_configured"]) + self.assertEqual(detail["capability"], "new_project") + self.assertEqual(detail["required_flag"], "--new-project") + self.assertEqual(detail["error"], "missing_new_project_capability") + self.assertTrue(detail["enabled"]) + self.assertTrue(detail["actionable"]) + self.assertFalse(detail["optional"]) + self.assertTrue(detail.get("owner_action")) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness.status, STATUS_WARN) + self.assertNotIn("antigravity", readiness.detail.get("ready_providers", [])) + self.assertIn("antigravity", readiness.detail.get("actionable_providers", [])) + + def test_antigravity_capability_check_preserves_evidence_privacy(self) -> None: + """Doctor Antigravity capability check strictly keeps help text, paths, prompts, secrets out of evidence.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "antigravity_cli": { + "enabled": True, + } + } + } + + fake_help = ( + "Usage: /mock/path/custom/agy [options]\n" + "Help text with unshared-arg-canary and user-prompt-canary\n" + ) + + def cap_runner(argv: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(list(argv), 0, stdout=fake_help, stderr="") + + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: "/mock/path/custom/agy" if cmd == "agy" else None, + capability_runner=cap_runner, + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + providers=["antigravity"], + ) + + adapter_check = next( + c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "antigravity" + ) + serialized = json.dumps(adapter_check.as_dict()) + + self.assertNotIn("/mock/path/custom", serialized) + self.assertNotIn("unshared-arg-canary", serialized) + self.assertNotIn("user-prompt-canary", serialized) + self.assertEqual(adapter_check.detail.get("command"), "agy") + + def test_antigravity_capability_failure_disabled_optional_provider(self) -> None: + """When optional/disabled Antigravity fails capability check, it does not become actionable or flip aggregate readiness.""" + with tempfile.TemporaryDirectory() as tmp: + # Only claude enabled + config = { + "lanes": { + "claude": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + } + } + } + + def cap_runner(argv: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(list(argv), 0, stdout="help without new project", stderr="") + + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + return_value=True, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd in {"claude", "agy"} else None, + capability_runner=cap_runner, + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + providers=["claude", "antigravity"], + ) + + adapter_check = next( + c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "antigravity" + ) + self.assertEqual(adapter_check.status, STATUS_WARN) + self.assertFalse(adapter_check.detail.get("actionable")) + self.assertTrue(adapter_check.detail.get("optional")) + self.assertNotIn("owner_action", adapter_check.detail) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness.status, STATUS_PASS) + self.assertIn("claude", readiness.detail.get("ready_providers", [])) + self.assertNotIn("antigravity", readiness.detail.get("ready_providers", [])) + self.assertIn("antigravity", readiness.detail.get("optional_providers", [])) + self.assertNotIn("antigravity", readiness.detail.get("actionable_providers", [])) + if __name__ == "__main__": unittest.main()