diff --git a/docs/release-qualification.md b/docs/release-qualification.md index 9940261d..71308444 100644 --- a/docs/release-qualification.md +++ b/docs/release-qualification.md @@ -122,9 +122,9 @@ Local CLI providers (`local_cli` driver) only run automatically if a `campaign_a 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. -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; `{command}` to the installed provider CLI (the campaign refuses to run when it is missing). +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). -**Adopters: override or disable per-repo in `code-mower.yml`.** `campaign_adapter_argv` replaces the maintained template wholesale (same placeholders plus `{python}` and `{adapter_timeout}`); `campaign_adapter_enabled: false` disables the lane's adapter so the provider degrades to `unavailable`/manual. Only `campaign_adapter_argv`, `campaign_adapter_timeout_seconds`, and `campaign_adapter_enabled` are read here. +**Adopters: override or disable per-repo in `code-mower.yml`.** `campaign_adapter_argv` replaces the maintained template wholesale (same placeholders plus `{python}`, `{target_python}`, and `{adapter_timeout}`); `campaign_adapter_enabled: false` disables the lane's adapter so the provider degrades to `unavailable`/manual. Only `campaign_adapter_argv`, `campaign_adapter_timeout_seconds`, and `campaign_adapter_enabled` are read here. **Adopters: configure this per-repo in `code-mower.yml` at the repository root.** This is the recommended setup and does not require editing installed Python: @@ -146,23 +146,22 @@ lanes: `campaign_adapter_timeout_seconds` must be a positive integer number of seconds, written as one: a fractional value such as `1.9` is rejected with the bounded `adapter_configuration_invalid` error rather than truncated to a shorter budget than was configured, as are `0`, negatives, non-finite numbers, and booleans. Omitting it uses the built-in default. -Supported placeholders: `{command}` (resolved binary), `{release_tag}`, `{package_spec}`, `{qualification_context}`, `{starting_version}`, `{output}`, `{repo_path}`. The adapter command must write a `code_mower.adoptionResult.v1` JSON document to the `{output}` path whose `provider` and `executor` both match the invoked provider, and whose `release_tag`, `qualification_context`, and `starting_version` fields all match the campaign's -- a cold-install result cannot complete an upgrade campaign, and an upgrade result must match the campaign's exact starting version. Anything else (extra fields, mismatched identity, no file, non-zero exit, or a timeout) leaves the provider `unavailable`/`blocked` with a bounded error code -- never a fabricated pass. A local drop-in result file and `--record-result` are bound the same way. Install the provider's CLI binary on PATH and verify local authentication. +Supported placeholders: `{command}` (resolved binary), `{release_tag}`, `{package_spec}`, `{qualification_context}`, `{starting_version}`, `{output}`, `{repo_path}`, `{python}` (running interpreter), `{target_python}` (resolved Python 3.12+ binary), `{target_runtime}`, `{adapter_timeout}`. The adapter command must write a `code_mower.adoptionResult.v1` JSON document to the `{output}` path whose `provider` and `executor` both match the invoked provider, and whose `release_tag`, `qualification_context`, and `starting_version` fields all match the campaign's -- a cold-install result cannot complete an upgrade campaign, and an upgrade result must match the campaign's exact starting version. Anything else (extra fields, mismatched identity, no file, non-zero exit, or a timeout) leaves the provider `unavailable`/`blocked` with a bounded error code -- never a fabricated pass. A local drop-in result file and `--record-result` are bound the same way. Install the provider's CLI binary on PATH and verify local authentication. Codex campaign runs use an isolated `CODEX_HOME` at `~/.config/code-mower/provider-homes/codex` (override with `CODE_MOWER_CODEX_CAMPAIGN_HOME`). Code Mower creates its non-secret restricted config automatically and refuses a readable `auth.json`. Authenticate that home once with `CODEX_HOME="$HOME/.config/code-mower/provider-homes/codex" codex login --device-auth -c 'cli_auth_credentials_store="keyring"' --enable secret_auth_storage`; the explicit login flags make Codex store that home-specific credential in the OS keyring even before Code Mower has created the config file. The adapter preserves the real OS `HOME` only so the platform keyring can locate the user's login keychain; Codex configuration and state remain isolated under `CODEX_HOME`, ambient token variables are removed, and the root-deny policy lets the agent write only its disposable workspace. Network remains available for package installation. A previous result file is removed before every adapter attempt, and a failed run never leaves stale evidence for a caller to accept. An explicit `--retry-provider` never accepts a pre-existing result file for that provider -- the stale file is removed before the new attempt runs, so a retry can only be satisfied by fresh evidence. A retry advances only the retried provider: every other participant keeps its recorded state, evidence, and attempt, dispatch, and completion timestamps (aggregate campaign fields still recompute), and newly arrived evidence for them waits for the next ordinary resume. The superseded attempt leaves one bounded metadata-only summary per retry (`attempt_history`, most recent 5: timestamps, state, outcome, error code, and elapsed time -- never results, output, paths, or secrets). Retained entries are rebuilt from those allowed scalar fields on every retry, so a malformed or hand-edited stored history is sanitized (unknown/nested fields dropped, malformed entries discarded) rather than copied verbatim. -### Campaign Authentication Readiness +### Campaign Authentication & Runtime Readiness -An installed CLI and a valid argv contract do not prove the isolated home the adapter runs under is authenticated: without this check, a campaign is dispatched and only then fails with a generic adapter error, after paid work has started. Where a provider exposes a safe, read-only login-status command, `code-mower doctor --adoption` runs it before dispatch as `doctor.campaign.auth`. Codex is the first such provider (`codex login status`, 20s budget); providers that expose no safe status command declare none and stay capability-only, so their readiness is never guessed. +An installed CLI and a valid argv contract do not prove the isolated home the adapter runs under is authenticated: without this check, a campaign is dispatched and only then fails with a generic adapter error, after paid work has started. -The probe runs in exactly the environment the adapter builds -- for Codex the isolated `CODEX_HOME` plus the real OS `HOME` needed to reach the platform keyring -- with the same allowlisted environment and no ambient provider, GitHub, or cloud tokens. Outcomes: +- **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`. +- **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`. -- **Authenticated** (probe exits 0): the check passes and the provider stays campaign-ready. -- **Unauthenticated** (a *confirmed* logged-out result): a stable owner-action warning carrying the bounded `campaign_auth_unauthenticated` code, remediation pointing at the one-time login above, and the provider dropped from `ready_providers` in `doctor.campaign.readiness`. A nonzero exit alone is never enough. The provider must declare both an expected logged-out exit code and a narrow allowlist of logged-out output markers (`campaign_auth_logged_out_exit_codes` and `campaign_auth_logged_out_markers`; for Codex, exit `1` plus `not logged in`/`not authenticated`), and the probe result must match both. The marker match happens in memory on a bounded prefix of the output, which is never stored or emitted. -- **Timeout or probe error**: a non-blocking skip with `campaign_auth_probe_timeout` or `campaign_auth_probe_unavailable`. This also covers every nonzero exit *without* the provider's logged-out signature -- an unsupported `login status` subcommand on an older or newer CLI, a broken keyring backend, a transient config error, or unexpected output. Failing to reach or interpret the probe is not evidence of a missing login, so it never becomes an owner action, never removes a usable provider from `ready_providers`, and never fails doctor. - -Probe stdout/stderr is never persisted: only the bounded state word, a registered error code, and non-content output shape reach doctor JSON, so account names, tokens, credential contents, and local paths cannot leak. `hosted-builders` and `orchestrator-only` postures already skip local adapter checks, so no probe runs there. Set `CODE_MOWER_CAMPAIGN_AUTH_PROBE=0` to leave every local adapter capability-only for a doctor run. +Probe stdout/stderr is never persisted: only bounded status tokens, registered error codes, and non-content output shape reach doctor JSON, so account names, tokens, credential contents, and local paths cannot leak. `hosted-builders` and `orchestrator-only` postures already skip local adapter checks, so no probe runs there. Set `CODE_MOWER_CAMPAIGN_AUTH_PROBE=0` to leave every local adapter capability-only for a doctor run. This overlays only `campaign_adapter_argv` and `campaign_adapter_timeout_seconds` onto the matching lane's built-in `provider_config`; every other key in `code-mower.yml` is ignored for this purpose, so it cannot widen the general config contract. A missing `code-mower.yml`, a missing `lanes` key, or a lane with no matching entry is treated as no override. Configure each lane under **exactly one** spelling: `muse` and `muse_cli` (or `claude` and `claude_code`) name one lane with one adapter command, so a config that declares more than one of them is *ambiguous*, not merely redundant -- the entries may carry two different `campaign_adapter_argv` values and there is no correct choice between two adapter commands. That is rejected with the bounded `adapter_configuration_invalid` error code and a detail naming the conflicting spellings (drawn only from the built-in alias table, so the message never echoes config text), rather than silently running whichever spelling happened to be looked up first. An existing repo config that fails to load, or is structurally malformed (a non-mapping `lanes`, lane, or `provider_config` entry), degrades to the safe `adapter_configuration_invalid` error code and an actionable status message, never a crash or a leaked template/traceback. `campaign_adapter_argv` must be a YAML list of non-empty scalar tokens (no shell strings; the adapter still runs with `shell=False`). diff --git a/src/code_mower/campaign_adapters.py b/src/code_mower/campaign_adapters.py index df668c79..bf6e771f 100644 --- a/src/code_mower/campaign_adapters.py +++ b/src/code_mower/campaign_adapters.py @@ -52,6 +52,7 @@ import argparse import json import os +import shlex import shutil import subprocess import sys @@ -278,6 +279,8 @@ def build_qualification_prompt( normalized_version: str, qualification_context: str, starting_version: str, + python_bin: str = "python3", + target_runtime: str = "", ) -> str: """Build the shared release-qualification prompt for one provider run. @@ -286,10 +289,11 @@ def build_qualification_prompt( local checkout paths: the agent works in a fresh disposable directory it creates itself. """ + python_cmd = shlex.quote(python_bin or "python3") if qualification_context == "upgrade": install_plan = ( "1. In the current disposable directory, run " - "`python3 -m venv .venv`. Use only `.venv/bin/python` and installed " + f"`{python_cmd} -m venv .venv`. Use only `.venv/bin/python` and installed " "entry points for the remaining steps.\n" f"2. Install the starting version with `.venv/bin/python -m pip install " f"{package_identity}=={starting_version}` to rehearse " @@ -299,7 +303,7 @@ def build_qualification_prompt( else: install_plan = ( "1. In the current disposable directory, run " - "`python3 -m venv .venv`. Use only `.venv/bin/python` and installed " + f"`{python_cmd} -m venv .venv`. Use only `.venv/bin/python` and installed " "entry points for the remaining steps.\n" f"2. Install the exact release with `.venv/bin/python -m pip install " f'"{package_spec}"`. No other version is acceptable.' @@ -321,8 +325,13 @@ def build_qualification_prompt( f"- package_spec: {package_spec}", f"- qualification_context: {qualification_context}", f"- starting_version: {starting_version if starting_version else '(empty)'}", - "", - "Procedure (measure wall-clock seconds for each step):", + ] + if target_runtime: + lines.append(f"- target_runtime: {target_runtime}") + lines.extend( + [ + "", + "Procedure (measure wall-clock seconds for each step):", install_plan, f"4. Assert the installed {package_identity} version is exactly", f" {normalized_version}, using `.venv/bin/python -c` and", @@ -347,7 +356,11 @@ def build_qualification_prompt( f"ending_version is exactly {normalized_version}, without a leading v,", "on pass/pass_with_warnings (and is empty on an incomplete run);", "host_class is one of local, ci, github_actions, unknown;", - "runtime_class is unknown or python_.; provider and", + ( + f"runtime_class is exactly {target_runtime}; provider and" + if target_runtime + else "runtime_class is python_.; provider and" + ), "executor are lowercase safe identifiers; steps is a non-empty list of", "{id, status, elapsed_seconds, warning_count, owner_action_count} with", "id one of board, doctor, lanes_status, overhead, package_install or a namespaced", @@ -359,7 +372,7 @@ def build_qualification_prompt( "pass_with_warnings if any status is warn or unavailable; otherwise", "pass. In particular, all-pass steps require outcome pass. No extra", "fields, no prose, no fences.", - ] + ]) return "\n".join(lines) + "\n" @@ -549,18 +562,22 @@ def _extract_claude_result(stdout: str) -> dict[str, Any] | None: try: envelope = json.loads(stdout) except json.JSONDecodeError: + envelope = None + if isinstance(envelope, Mapping): + if envelope.get("is_error") is True: + return None + structured = envelope.get("structured_output") + if isinstance(structured, Mapping): + return dict(structured) + result = envelope.get("result") + if isinstance(result, Mapping): + return dict(result) + if isinstance(result, str) and result.strip(): + parsed = code_mower_gemini_cli.parse_response_json(result) + if parsed is not None: + return parsed return None - if not isinstance(envelope, dict) or envelope.get("is_error") is True: - return None - structured = envelope.get("structured_output") - if isinstance(structured, dict): - return structured - result = envelope.get("result") - if isinstance(result, dict): - return result - if isinstance(result, str) and result.strip(): - return code_mower_gemini_cli.parse_response_json(result) - return None + return code_mower_gemini_cli.parse_response_json(stdout) def _extract_antigravity_result(stdout: str) -> dict[str, Any] | None: @@ -587,9 +604,68 @@ def _extract_antigravity_result(stdout: str) -> dict[str, Any] | None: def _extract_muse_result(stdout: str) -> dict[str, Any] | None: """Pull the adoption result out of ``muse exec --json`` JSONL events.""" response_text, _meta = code_mower_muse_cli.muse_jsonl_response(stdout) - if not response_text: - return None - return code_mower_gemini_cli.parse_response_json(response_text) + if response_text: + parsed = code_mower_gemini_cli.parse_response_json(response_text) + if parsed is not None: + return parsed + return code_mower_gemini_cli.parse_response_json(stdout) + + +def check_structured_result_capability(provider: str) -> bool: + """Validate structured-result extraction and schema compliance using an offline fixture. + + Uses a zero-network, zero-token in-memory fixture to verify that Code Mower + can correctly parse and validate adoption results from this provider. + """ + from datetime import datetime, timedelta, timezone + from code_mower.release_qualify import validate_adoption_result_payload + + canonical = provider.lower().replace("-", "_") + recent = datetime.now(timezone.utc) - timedelta(seconds=60) + sample_payload = { + "schema": "code_mower.adoptionResult.v1", + "timestamp_utc": recent.strftime("%Y-%m-%dT%H:%M:%SZ"), + "release_tag": "v1.0.8", + "package_identity": "code-mower", + "normalized_version": "1.0.8", + "qualification_context": "cold_install", + "starting_version": "", + "ending_version": "1.0.8", + "provider": canonical, + "executor": canonical, + "host_class": "local", + "runtime_class": "python_3.12", + "execution_state": "executed", + "elapsed_seconds": 1.0, + "outcome": "pass", + "steps": [ + { + "id": "doctor", + "status": "pass", + "elapsed_seconds": 1.0, + "warning_count": 0, + "owner_action_count": 0, + } + ], + } + payload_str = json.dumps(sample_payload) + try: + if canonical in {"claude_audit", "claude"}: + claude_envelope = json.dumps({"is_error": False, "result": payload_str}) + extracted = _extract_claude_result(claude_envelope) + elif canonical in {"antigravity", "antigravity_cli"}: + extracted = _extract_antigravity_result(payload_str) + elif canonical in {"muse", "muse_cli"}: + muse_event = json.dumps({"payload_type": "run.output.delta", "payload": {"text": payload_str}}) + extracted = _extract_muse_result(muse_event) + else: + extracted = code_mower_gemini_cli.parse_response_json(payload_str) + if not isinstance(extracted, Mapping): + return False + validate_adoption_result_payload(extracted) + return True + except Exception: + return False def validate_bound_result( @@ -600,6 +676,7 @@ def validate_bound_result( package_identity: str, qualification_context: str, starting_version: str, + target_runtime: str = "", ) -> dict[str, Any]: """Closed-validate a transient candidate and bind it to this campaign. @@ -620,6 +697,8 @@ def validate_bound_result( raise ValueError("provider result identity mismatch") if str(candidate.get("starting_version") or "") != starting_version: raise ValueError("provider result identity mismatch") + if target_runtime and candidate.get("runtime_class") != target_runtime: + raise ValueError("provider result identity mismatch") return candidate @@ -783,6 +862,8 @@ def run_campaign_adapter( muse_max_model_steps: int = MUSE_DEFAULT_MAX_MODEL_STEPS, muse_reasoning_effort: str = "", codex_home: Path | None = None, + python_bin: str = "", + target_runtime: str = "", provider_runner: ProviderRunner = run_provider_command, ) -> int: """Run one maintained provider adapter. Returns a process exit code. @@ -824,6 +905,8 @@ def run_campaign_adapter( normalized_version=normalized_version, qualification_context=qualification_context, starting_version=starting_version, + python_bin=python_bin, + target_runtime=target_runtime, ) prepared_codex_home: Path | None = None if provider == "codex": @@ -958,6 +1041,7 @@ def run_campaign_adapter( package_identity=package_identity, qualification_context=qualification_context, starting_version=starting_version, + target_runtime=target_runtime, ) except ValueError as exc: return _fail(provider, str(exc)[:180]) @@ -978,6 +1062,8 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--provider", required=True, choices=SUPPORTED_ADAPTER_PROVIDERS) parser.add_argument("--provider-bin", required=True) + parser.add_argument("--python-bin", default="") + parser.add_argument("--target-runtime", default="") parser.add_argument("--release-tag", required=True) parser.add_argument("--package-spec", required=True) parser.add_argument("--qualification-context", required=True) @@ -1011,6 +1097,8 @@ def main(argv: Sequence[str] | None = None) -> int: claude_max_budget_usd=args.claude_max_budget_usd, muse_max_model_steps=args.muse_max_model_steps, muse_reasoning_effort=args.muse_reasoning_effort, + python_bin=args.python_bin, + target_runtime=args.target_runtime, ) diff --git a/src/code_mower/doctor_checks/adoption.py b/src/code_mower/doctor_checks/adoption.py index 6d186833..5990b2e4 100644 --- a/src/code_mower/doctor_checks/adoption.py +++ b/src/code_mower/doctor_checks/adoption.py @@ -592,6 +592,7 @@ def check_adoption_campaign_readiness( ) -> 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.cloud import resolve_cloud_token from code_mower.release_campaigns import ( _check_credentials, @@ -600,6 +601,7 @@ def check_adoption_campaign_readiness( _safe_error, _validate_adapter_argv_template, _validate_adapter_timeout, + resolve_supported_runtime, resolve_provider_lane, ) @@ -607,6 +609,9 @@ def check_adoption_campaign_readiness( current_env = os.environ if env is None else env runner = lane_status._run_command if command_runner is None else command_runner checks: list[DoctorCheck] = [] + provider_readiness: dict[str, dict[str, Any]] = {} + runtime_resolution_checked = False + supported_runtime: tuple[str, str] | None = None for prov in providers: try: @@ -618,6 +623,11 @@ def check_adoption_campaign_readiness( if lane.driver == "local_cli": if adoption_posture in {"hosted-builders", "orchestrator-only"}: + provider_readiness[canonical] = { + "command": False, + "auth": "skipped", + "structured_result": check_structured_result_capability(canonical), + } checks.append( DoctorCheck( name="doctor.campaign.adapter", @@ -635,6 +645,51 @@ def check_adoption_campaign_readiness( ) continue + if not runtime_resolution_checked: + supported_runtime = resolve_supported_runtime( + environ=current_env, + which_fn=which_fn, + ) + runtime_resolution_checked = True + runtime_ready = supported_runtime is not None + runtime_detail: dict[str, Any] = { + "provider": canonical, + "lane": lane.lane_id, + "driver": lane.driver, + "runtime_available": runtime_ready, + "enabled": is_enabled, + } + if runtime_ready: + runtime_detail["runtime_class"] = supported_runtime[1] + else: + runtime_detail.update( + { + "error": "python_runtime_unavailable", + "actionable": is_enabled, + "optional": not is_enabled, + } + ) + if is_enabled: + runtime_detail["owner_action"] = True + checks.append( + DoctorCheck( + name="doctor.campaign.runtime", + status=STATUS_PASS if runtime_ready else STATUS_WARN, + lane=canonical, + message=( + f"{canonical} campaign Python runtime ready" + if runtime_ready + else f"{canonical} campaign requires Python 3.12+" + ), + detail=runtime_detail, + remediation=( + "" + if runtime_ready + else "Install Python 3.12+ on PATH or set CODE_MOWER_PYTHON to a supported interpreter." + ), + ) + ) + argv_template, timeout_value, config_error, config_detail = _resolve_adapter_config_for_lane( lane, repo_root=root, @@ -651,6 +706,7 @@ def check_adoption_campaign_readiness( cmd = _find_command(lane, which_fn=which_fn) command_name = lane.provider_config.get("command") or lane.provider + structured_capability = check_structured_result_capability(canonical) if config_error: detail = { @@ -667,6 +723,11 @@ def check_adoption_campaign_readiness( } if is_enabled: detail["owner_action"] = True + provider_readiness[canonical] = { + "command": bool(cmd), + "auth": "unknown", + "structured_result": structured_capability, + } checks.append( DoctorCheck( name="doctor.campaign.adapter", @@ -694,6 +755,11 @@ def check_adoption_campaign_readiness( } if is_enabled: detail["owner_action"] = True + provider_readiness[canonical] = { + "command": False, + "auth": "unknown", + "structured_result": structured_capability, + } checks.append( DoctorCheck( name="doctor.campaign.adapter", @@ -718,6 +784,11 @@ def check_adoption_campaign_readiness( } 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", @@ -746,6 +817,7 @@ def check_adoption_campaign_readiness( "command": cmd, "command_found": True, "adapter_configured": True, + "structured_result_capability": structured_capability, "enabled": is_enabled, }, ) @@ -758,8 +830,40 @@ def check_adoption_campaign_readiness( env=current_env, probe_runner=auth_probe_runner, ) + auth_state = "unprobed" if auth_check is not None: checks.append(auth_check) + auth_state = str(auth_check.detail.get("auth_probe", auth_check.status)) + provider_readiness[canonical] = { + "command": True, + "auth": auth_state, + "structured_result": structured_capability, + } + auth_pass = auth_check is None or auth_check.status != STATUS_WARN + if auth_pass and not structured_capability: + detail = { + "provider": canonical, + "lane": lane.lane_id, + "driver": lane.driver, + "structured_result_capability": False, + "enabled": is_enabled, + "actionable": is_enabled, + "optional": not is_enabled, + } + if is_enabled: + detail["owner_action"] = True + checks.append( + DoctorCheck( + name="doctor.campaign.structured_result", + status=STATUS_WARN, + lane=canonical, + message=f"{canonical} campaign structured-result capability probe failed", + detail=detail, + remediation=( + f"Verify {canonical} campaign adapter output parsing and schema compliance." + ), + ) + ) elif lane.driver in {"hosted_bridge", "saas_event"}: has_credentials, missing_var = _check_credentials(lane, env=current_env) @@ -767,6 +871,22 @@ def check_adoption_campaign_readiness( lane, env=current_env ) has_repo = bool(repo_slug) + structured_capability = check_structured_result_capability(canonical) + cmd_ready = bool(has_credentials and has_repo and transport_ready) + if cmd_ready: + auth_state = "ready" + elif not has_credentials: + auth_state = "missing_credentials" + elif not has_repo: + auth_state = "missing_repo" + else: + auth_state = "unverified_transport" + + provider_readiness[canonical] = { + "command": cmd_ready, + "auth": auth_state, + "structured_result": structured_capability, + } if not has_credentials: detail = { @@ -860,6 +980,30 @@ def check_adoption_campaign_readiness( }, ) ) + if not structured_capability: + detail = { + "provider": canonical, + "lane": lane.lane_id, + "driver": lane.driver, + "structured_result_capability": False, + "enabled": is_enabled, + "actionable": is_enabled, + "optional": not is_enabled, + } + if is_enabled: + detail["owner_action"] = True + checks.append( + DoctorCheck( + name="doctor.campaign.structured_result", + status=STATUS_WARN, + lane=canonical, + message=f"{canonical} campaign structured-result capability probe failed", + detail=detail, + remediation=( + f"Verify {canonical} campaign adapter output parsing and schema compliance." + ), + ) + ) # 3. Campaign Storage Writable Check storage_rel = ".code-mower/campaigns" @@ -1052,8 +1196,10 @@ def check_adoption_campaign_readiness( if check.name in { "doctor.campaign.adapter", + "doctor.campaign.runtime", "doctor.campaign.credentials", CAMPAIGN_AUTH_CHECK_NAME, + "doctor.campaign.structured_result", } ] # A provider is ready only when every one of its checks is clean: an @@ -1062,7 +1208,13 @@ def check_adoption_campaign_readiness( check.lane for check in provider_checks if check.status == STATUS_WARN and check.lane } ready_providers = sorted( - {check.lane for check in provider_checks if check.status == STATUS_PASS and check.lane} + { + check.lane + for check in provider_checks + if check.status == STATUS_PASS + and check.lane + and provider_readiness.get(check.lane, {}).get("structured_result") is True + } - warned_providers ) actionable_providers = sorted( @@ -1123,6 +1275,7 @@ def check_adoption_campaign_readiness( "actionable_providers": actionable_providers, "optional_providers": optional_providers, "preview_command": preview_command, + "provider_readiness": provider_readiness, }, remediation=readiness_remediation, ) diff --git a/src/code_mower/doctor_checks/campaign_auth.py b/src/code_mower/doctor_checks/campaign_auth.py index f8db393c..ac4b2863 100644 --- a/src/code_mower/doctor_checks/campaign_auth.py +++ b/src/code_mower/doctor_checks/campaign_auth.py @@ -232,26 +232,132 @@ def check_campaign_auth_readiness( Returns ``None`` when the provider exposes no safe status command, which keeps that lane capability-only instead of guessing it is authenticated. """ - probe_args = campaign_auth_probe_args(lane) - if not probe_args or not command: + if not command: return None - timeout_seconds = campaign_auth_probe_timeout(lane) - if not campaign_auth_probe_requested(env): + current_env = os.environ if env is None else env + if not campaign_auth_probe_requested(current_env): + if canonical in {"antigravity", "muse"} or campaign_auth_probe_args(lane): + timeout_seconds = campaign_auth_probe_timeout(lane) + return DoctorCheck( + name=CAMPAIGN_AUTH_CHECK_NAME, + status=STATUS_SKIP, + lane=canonical, + message=f"skipped {canonical} campaign authentication probe ({CAMPAIGN_AUTH_PROBE_ENV})", + detail=_detail( + canonical=canonical, + lane=lane, + state=AUTH_STATE_SKIPPED, + enabled=enabled, + timeout_seconds=timeout_seconds, + ), + ) + return None + + if canonical == "antigravity": + opted_in = current_env.get("ANTIGRAVITY_CLI_USE_AMBIENT_HOME", "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + if opted_in: + return DoctorCheck( + name=CAMPAIGN_AUTH_CHECK_NAME, + status=STATUS_PASS, + lane=canonical, + message="antigravity ambient-home auth opt-in configured", + detail={ + "provider": canonical, + "lane": getattr(lane, "lane_id", canonical), + "driver": getattr(lane, "driver", "local_cli"), + "auth_probe": "ambient_opt_in", + "auth_ready": True, + "ambient_home_opt_in": True, + "enabled": enabled, + }, + ) + detail = { + "provider": canonical, + "lane": getattr(lane, "lane_id", canonical), + "driver": getattr(lane, "driver", "local_cli"), + "auth_probe": "missing_opt_in", + "auth_ready": False, + "ambient_home_opt_in": False, + "missing_variable": "ANTIGRAVITY_CLI_USE_AMBIENT_HOME", + "enabled": enabled, + "actionable": enabled, + "optional": not enabled, + } + if enabled: + detail["owner_action"] = True return DoctorCheck( name=CAMPAIGN_AUTH_CHECK_NAME, - status=STATUS_SKIP, + status=STATUS_WARN, + lane=canonical, + message="antigravity campaign auth requires ANTIGRAVITY_CLI_USE_AMBIENT_HOME=1 in trusted environments", + detail=detail, + remediation=( + "Set ANTIGRAVITY_CLI_USE_AMBIENT_HOME=1 in trusted environments to " + "allow local OAuth state." + ), + ) + + if canonical == "muse": + from .. import muse_cli_audit_pr as code_mower_muse_cli + + has_key = bool(code_mower_muse_cli.resolve_muse_api_key(current_env)) + has_ambient = current_env.get("MUSE_CLI_USE_AMBIENT_HOME", "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + if has_key or has_ambient: + return DoctorCheck( + name=CAMPAIGN_AUTH_CHECK_NAME, + status=STATUS_PASS, + lane=canonical, + message="muse campaign authentication configured", + detail={ + "provider": canonical, + "lane": getattr(lane, "lane_id", canonical), + "driver": getattr(lane, "driver", "local_cli"), + "auth_probe": "api_key" if has_key else "ambient_opt_in", + "auth_ready": True, + "enabled": enabled, + }, + ) + detail = { + "provider": canonical, + "lane": getattr(lane, "lane_id", canonical), + "driver": getattr(lane, "driver", "local_cli"), + "auth_probe": "missing_auth", + "auth_ready": False, + "enabled": enabled, + "actionable": enabled, + "optional": not enabled, + } + if enabled: + detail["owner_action"] = True + return DoctorCheck( + name=CAMPAIGN_AUTH_CHECK_NAME, + status=STATUS_WARN, lane=canonical, - message=f"skipped {canonical} campaign authentication probe ({CAMPAIGN_AUTH_PROBE_ENV})", - detail=_detail( - canonical=canonical, - lane=lane, - state=AUTH_STATE_SKIPPED, - enabled=enabled, - timeout_seconds=timeout_seconds, + message="muse campaign auth requires META_API_KEY or MUSE_CLI_USE_AMBIENT_HOME=1", + detail=detail, + remediation=( + "Set META_API_KEY or point META_API_KEY_FILE at a local key file, or set " + "MUSE_CLI_USE_AMBIENT_HOME=1 in trusted environments." ), ) + probe_args = campaign_auth_probe_args(lane) + if not probe_args: + return None + + timeout_seconds = campaign_auth_probe_timeout(lane) + provider = str(getattr(lane, "provider", "") or canonical) child_env, env_error = campaign_auth_probe_env(provider) output = "" diff --git a/src/code_mower/muse_cli_audit_pr.py b/src/code_mower/muse_cli_audit_pr.py index 9b4989ca..bc8c0fcf 100644 --- a/src/code_mower/muse_cli_audit_pr.py +++ b/src/code_mower/muse_cli_audit_pr.py @@ -91,20 +91,24 @@ def parse_api_key_file(text: str) -> str: ).value -def resolve_muse_api_key() -> str: +def resolve_muse_api_key(env: Mapping[str, str] | None = None) -> str: + current_env = os.environ if env is None else env for name in MUSE_KEY_ENV_NAMES: - value = os.environ.get(name, "").strip() + value = current_env.get(name, "").strip() if value: return value for name in MUSE_KEY_FILE_ENV_NAMES: - path_text = os.environ.get(name, "").strip() + path_text = current_env.get(name, "").strip() if not path_text: continue try: + path = Path(path_text).expanduser() + if not path.is_file(): + continue value = parse_api_key_file( - Path(path_text).expanduser().read_text(encoding="utf-8") + path.read_text(encoding="utf-8") ) - except OSError: + except (OSError, UnicodeDecodeError): continue if value: return value diff --git a/src/code_mower/provider_registry.py b/src/code_mower/provider_registry.py index ce71d249..feca8d72 100644 --- a/src/code_mower/provider_registry.py +++ b/src/code_mower/provider_registry.py @@ -84,6 +84,10 @@ def _maintained_campaign_adapter_argv(provider: str) -> tuple[str, ...]: provider, "--provider-bin", "{command}", + "--python-bin", + "{target_python}", + "--target-runtime", + "{target_runtime}", "--release-tag", "{release_tag}", "--package-spec", diff --git a/src/code_mower/release_campaigns.py b/src/code_mower/release_campaigns.py index 4c45e7b4..89384d95 100644 --- a/src/code_mower/release_campaigns.py +++ b/src/code_mower/release_campaigns.py @@ -164,6 +164,7 @@ "github_dispatch_failed", "github_poll_unavailable", "hosted_response_timeout", + "python_runtime_unavailable", "unknown_provider", } ) @@ -175,6 +176,7 @@ "adapter_configuration_invalid": "unavailable", "command_not_found": "unavailable", "adapter_timeout": "unavailable", + "python_runtime_unavailable": "unavailable", "adapter_exited_nonzero": "blocked", "adapter_produced_no_result": "blocked", "adapter_result_invalid": "blocked", @@ -460,6 +462,7 @@ def _provider_next_action( dry_run: bool, adapter_configured: bool = True, error: str = "", + error_code: str = "", ) -> tuple[str, str]: if state == "complete": return "none", "" @@ -475,6 +478,14 @@ def _provider_next_action( f"record manual result for {provider}", "no campaign adapter configured", ) + if lane.driver == "local_cli" and ( + error_code == "python_runtime_unavailable" + or error == "python_runtime_unavailable" + ): + return ( + "install Python 3.12+ on PATH or set CODE_MOWER_PYTHON", + error if error != "python_runtime_unavailable" else "supported Python 3.12+ runtime is unavailable", + ) if lane.driver == "local_cli" and not command_available: cmd = lane.provider_config.get("command") or provider return f"install {cmd} CLI on PATH or record manual result", error or f"command not found: {cmd}" @@ -1272,6 +1283,8 @@ def _build_adapter_argv( repo_path: Path, argv_template: Any, adapter_timeout: int | None = None, + python_bin: str = "", + target_runtime: str = "", ) -> list[str]: validated_template = _validate_adapter_argv_template(argv_template) substitutions = { @@ -1282,10 +1295,11 @@ def _build_adapter_argv( "starting_version": starting_version, "output": str(output_path), "repo_path": str(repo_path), - # The running interpreter, for maintained adapters that re-enter Code - # Mower (`{python} -m code_mower.campaign_adapters ...`). Templates - # that do not name it are unaffected: unused substitutions are ignored. + # The running interpreter running Code Mower "python": sys.executable, + # The resolved supported Python executable for target runtime qualification + "target_python": python_bin or sys.executable, + "target_runtime": target_runtime or _detect_runtime_class(), } if adapter_timeout is not None: substitutions["adapter_timeout"] = str(adapter_timeout) @@ -1435,6 +1449,96 @@ def _resolve_campaign_adapter_config( return argv_template, timeout_value, "", "" +_VERSIONED_PYTHON_RE = re.compile(r"^python3\.(\d+)$") + + +def _versioned_python_candidates(path_value: str) -> tuple[str, ...]: + """Return discovered Python 3.12+ command names in newest-first order.""" + minors: set[int] = set() + for raw_dir in path_value.split(os.pathsep): + if not raw_dir: + continue + try: + entries = Path(raw_dir).expanduser().iterdir() + for entry in entries: + match = _VERSIONED_PYTHON_RE.fullmatch(entry.name) + if match and int(match.group(1)) >= 12: + minors.add(int(match.group(1))) + except OSError: + continue + return tuple(f"python3.{minor}" for minor in sorted(minors, reverse=True)) + + +def resolve_supported_runtime( + *, + environ: Mapping[str, str] | None = None, + which_fn: Callable[[str], str | None] | None = None, + runner: Callable[..., subprocess.CompletedProcess[str]] | None = None, +) -> tuple[str, str] | None: + """Resolve one supported Python 3.12+ runtime deterministically. + + Returns (executable, target_runtime) where target_runtime is + f"python_{major}.{minor}", or None when no supported runtime exists. + """ + env = os.environ if environ is None else environ + explicit = env.get("CODE_MOWER_PYTHON", "").strip() + if explicit: + candidates = (explicit,) + else: + discovered = _versioned_python_candidates(str(env.get("PATH", ""))) + candidates = ( + sys.executable, + *discovered, + "python3.12", + "python3.13", + "python3.14", + "python3", + ) + which = shutil.which if which_fn is None else which_fn + run = subprocess.run if runner is None else runner + for cand in candidates: + if not cand: + continue + cand_path = Path(cand).expanduser() + if ( + os.sep in cand + or (os.altsep and os.altsep in cand) + or cand == sys.executable + or (cand == explicit and cand_path.exists()) + ): + if cand_path.exists(): + resolved = str(cand_path.resolve()) if not cand_path.is_absolute() else cand + else: + resolved = None + else: + resolved = which(cand) + if resolved and not Path(resolved).is_absolute(): + resolved = str(Path(resolved).resolve()) + if not resolved: + continue + try: + completed = run( + [resolved, "-c", "import sys; print(f'{sys.version_info[0]}.{sys.version_info[1]}')"], + text=True, + capture_output=True, + check=False, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + continue + if completed.returncode != 0: + continue + out = (completed.stdout or "").strip() + parts = out.split(".") + try: + major, minor = int(parts[0]), int(parts[1]) + except (IndexError, ValueError): + continue + if (major, minor) >= (3, 12): + return resolved, f"python_{major}.{minor}" + return None + + def _invoke_local_adapter( lane: ProviderLane, provider: str, @@ -1447,6 +1551,8 @@ def _invoke_local_adapter( repo_path: Path, which_fn: Callable[[str], str | None], adapter_runner: AdapterRunner, + python_runner: Callable[..., subprocess.CompletedProcess[str]] | None = None, + environ: Mapping[str, str] | None = None, ) -> tuple[dict[str, Any] | None, str, str]: """Invoke a provider's explicit, registry-configured campaign adapter. @@ -1478,6 +1584,15 @@ def _invoke_local_adapter( if not argv_template: return None, _safe_error("no_campaign_adapter_configured"), "no campaign adapter configured" + runtime = resolve_supported_runtime(environ=environ, which_fn=which_fn, runner=python_runner) + if runtime is None: + return ( + None, + _safe_error("python_runtime_unavailable"), + "supported Python 3.12+ runtime is unavailable", + ) + python_bin, target_runtime = runtime + try: output_path.unlink(missing_ok=True) except OSError: @@ -1517,6 +1632,8 @@ def _invoke_local_adapter( repo_path=repo_path, argv_template=argv_template, adapter_timeout=max(1, timeout - ADAPTER_INNER_TIMEOUT_MARGIN_SECONDS), + python_bin=python_bin, + target_runtime=target_runtime, ) except ValueError: return ( @@ -2520,6 +2637,7 @@ def dispatch_or_advance_campaign( repo_path=repo_path, which_fn=which_fn, adapter_runner=adapter_runner, + environ=current_env, ) if result is None: provider_data["state"] = _ADAPTER_ERROR_STATE.get(error_code, "unavailable") @@ -2541,6 +2659,7 @@ def dispatch_or_advance_campaign( dry_run=False, adapter_configured=adapter_configured, error=detail, + error_code=error_code, ) else: provider_data["adoption_result"] = result diff --git a/src/code_mower/release_qualify.py b/src/code_mower/release_qualify.py index dfd69bea..bc0963f6 100644 --- a/src/code_mower/release_qualify.py +++ b/src/code_mower/release_qualify.py @@ -441,10 +441,16 @@ def validate_adoption_result_payload( if result.get("host_class") not in VALID_HOST_CLASSES: raise ValueError(f"unsupported adoption result host_class {result.get('host_class')!r}") runtime_class = str(result.get("runtime_class") or "") - if runtime_class != "unknown" and not RUNTIME_CLASS_PATTERN.match(runtime_class): + if not RUNTIME_CLASS_PATTERN.match(runtime_class): raise ValueError( - "adoption result runtime_class must be 'unknown' or 'python_.'" + "adoption result runtime_class must be 'python_.'" ) + parts = runtime_class.removeprefix("python_").split(".") + try: + if tuple(int(p) for p in parts[:2]) < (3, 12): + raise ValueError("adoption result runtime_class must be >= python_3.12") + except (IndexError, ValueError) as exc: + raise ValueError("adoption result runtime_class must be >= python_3.12") from exc execution_state = result.get("execution_state") if execution_state not in VALID_EXECUTION_STATES: diff --git a/tests/test_campaign_adapters.py b/tests/test_campaign_adapters.py index 1a973d5e..8944cc1e 100644 --- a/tests/test_campaign_adapters.py +++ b/tests/test_campaign_adapters.py @@ -10,6 +10,7 @@ import json import os import re +import shlex import subprocess import sys import tempfile @@ -201,6 +202,47 @@ def test_shared_prompt_pins_virtualenv_interpreter_commands(self) -> None: self.assertIn("without a leading v", prompt) self.assertIn("all-pass steps require outcome pass", prompt) + def test_shared_prompt_quotes_target_interpreter_with_spaces_and_metacharacters(self) -> None: + """Target interpreter path containing spaces or shell metacharacters must be safely quoted.""" + space_py = "/opt/Python 3.12/bin/python3" + prompt_cold = campaign_adapters.build_qualification_prompt( + provider="codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + package_identity="code-mower", + normalized_version="1.0.0", + qualification_context="cold_install", + starting_version="", + python_bin=space_py, + ) + self.assertIn(f"`{shlex.quote(space_py)} -m venv .venv`", prompt_cold) + self.assertIn("'/opt/Python 3.12/bin/python3' -m venv .venv", prompt_cold) + + prompt_upgrade = campaign_adapters.build_qualification_prompt( + provider="codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + package_identity="code-mower", + normalized_version="1.0.0", + qualification_context="upgrade", + starting_version="0.9.0", + python_bin=space_py, + ) + self.assertIn(f"`{shlex.quote(space_py)} -m venv .venv`", prompt_upgrade) + + meta_py = '/opt/py$env;`test`/bin/python3' + prompt_meta = campaign_adapters.build_qualification_prompt( + provider="codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + package_identity="code-mower", + normalized_version="1.0.0", + qualification_context="cold_install", + starting_version="", + python_bin=meta_py, + ) + self.assertIn(f"`{shlex.quote(meta_py)} -m venv .venv`", prompt_meta) + def test_shared_prompt_teaches_step_id_taxonomy(self) -> None: prompt = campaign_adapters.build_qualification_prompt( provider="codex", @@ -469,6 +511,32 @@ def runner( self.assertEqual(json.loads(output.read_text(encoding="utf-8")), _adoption_result("claude")) self.assertNotIn("SECRET-NOISE", output.read_text(encoding="utf-8")) + def test_claude_rejects_is_error_true_envelope_even_with_valid_fenced_result(self) -> None: + """A valid Claude error envelope must fail closed even if result contains fenced JSON.""" + valid_payload = _adoption_result("claude") + fenced_payload = f"Error details:\n```json\n{json.dumps(valid_payload)}\n```" + envelope = json.dumps({ + "is_error": True, + "result": fenced_payload, + }) + extracted = campaign_adapters._extract_claude_result(envelope) + self.assertIsNone(extracted) + + def runner( + argv: Any, prompt_input: Any, timeout: int, workdir: Path, child_env: Any + ) -> Any: + return subprocess.CompletedProcess(list(argv), 0, stdout=envelope, stderr="") + + code, output, _tmp = self._run("claude", runner) + self.assertNotEqual(code, 0) + self.assertFalse(output.exists()) + + def test_claude_fenced_fallback_applies_only_when_not_valid_envelope(self) -> None: + valid_payload = _adoption_result("claude") + raw_fenced = f"```json\n{json.dumps(valid_payload)}\n```" + extracted = campaign_adapters._extract_claude_result(raw_fenced) + self.assertEqual(extracted, valid_payload) + def test_antigravity_uses_prompt_file_not_stdin(self) -> None: seen: dict[str, Any] = {} @@ -791,7 +859,13 @@ def test_four_lanes_carry_maintained_adapter_argv(self) -> None: self.assertIsNotNone(template) assert template is not None joined = " ".join(template) - for placeholder in ("{python}", "{command}", "{adapter_timeout}", "{output}"): + for placeholder in ( + "{python}", + "{target_python}", + "{command}", + "{adapter_timeout}", + "{output}", + ): self.assertIn(placeholder, joined) self.assertIn(adapter_provider, template) timeout = lane.provider_config.get("campaign_adapter_timeout_seconds") @@ -802,5 +876,62 @@ def test_lanes_without_adapters_stay_manual(self) -> None: self.assertNotIn("campaign_adapter_argv", REFERENCE_PROVIDERS["aider"].provider_config) +class StructuredResultCapabilityTests(unittest.TestCase): + """The offline structured-result probe exercises real provider extraction paths.""" + + def test_all_maintained_lanes_pass_structured_result_capability(self) -> None: + for prov in ( + "codex", + "claude_audit", + "claude", + "antigravity_cli", + "antigravity", + "muse_cli", + "muse", + ): + with self.subTest(provider=prov): + self.assertTrue(campaign_adapters.check_structured_result_capability(prov)) + + def test_claude_audit_routes_through_extract_claude_result(self) -> None: + """claude_audit must explicitly route through _extract_claude_result.""" + with mock.patch( + "code_mower.campaign_adapters._extract_claude_result", + wraps=campaign_adapters._extract_claude_result, + ) as mock_extract: + self.assertTrue(campaign_adapters.check_structured_result_capability("claude_audit")) + mock_extract.assert_called_once() + + def test_claude_audit_fails_if_generic_extractor_used_on_envelope(self) -> None: + """Regression: claude_audit envelope fails schema validation under generic extractor. + + If claude_audit falls through to the generic parse_response_json extractor, + the Claude envelope object (with is_error, result) is returned directly + instead of unwrapping the inner adoption result, causing validation to fail. + """ + from code_mower import gemini_cli_audit_pr as code_mower_gemini_cli + + sample_payload = _adoption_result("claude_audit") + claude_envelope = json.dumps({"is_error": False, "result": json.dumps(sample_payload)}) + + # Generic extractor cannot unwrap the envelope + generic_extracted = code_mower_gemini_cli.parse_response_json(claude_envelope) + self.assertIsNotNone(generic_extracted) + with self.assertRaises(ValueError): + release_qualify.validate_adoption_result_payload(generic_extracted) + + # But _extract_claude_result correctly extracts and validates it + extracted = campaign_adapters._extract_claude_result(claude_envelope) + self.assertEqual(extracted, sample_payload) + release_qualify.validate_adoption_result_payload(extracted) + + def test_claude_audit_capability_probe_fails_on_error_envelope(self) -> None: + """If _extract_claude_result returns None (e.g. is_error=True), check_structured_result_capability fails.""" + with mock.patch( + "code_mower.campaign_adapters._extract_claude_result", + return_value=None, + ): + self.assertFalse(campaign_adapters.check_structured_result_capability("claude_audit")) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_doctor_campaign_auth.py b/tests/test_doctor_campaign_auth.py index ba287a84..9ff145c1 100644 --- a/tests/test_doctor_campaign_auth.py +++ b/tests/test_doctor_campaign_auth.py @@ -85,6 +85,45 @@ def _rendered(checks): class CampaignAuthProbeTests(unittest.TestCase): + def test_lane_id_aliases_reach_provider_specific_auth_checks(self) -> None: + cases = ( + ("antigravity_cli", "agy", "antigravity", "missing_opt_in"), + ("muse_cli", "muse", "muse", "missing_auth"), + ) + with tempfile.TemporaryDirectory() as tmp: + for lane_id, command, provider, auth_state in cases: + with self.subTest(lane_id=lane_id): + checks = check_adoption_campaign_readiness( + config={ + "lanes": { + lane_id: { + "provider_config": { + "campaign_adapter_argv": [ + "{command}", + "qualify", + "--output", + "{output}", + ], + } + } + } + }, + repo_root=Path(tmp), + env={}, + which_fn=lambda candidate, expected=command: ( + f"/opt/bin/{expected}" if candidate == expected else None + ), + providers=[lane_id], + ) + + auth = next(c for c in checks if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth.lane, provider) + self.assertEqual(auth.status, STATUS_WARN) + self.assertEqual(auth.detail.get("auth_probe"), auth_state) + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertNotIn(provider, readiness.detail.get("ready_providers", [])) + self.assertIn(provider, readiness.detail.get("actionable_providers", [])) + def test_authenticated_isolated_home_passes_without_exposing_output(self) -> None: recorded: list[tuple[list[str], int, dict]] = [] @@ -195,7 +234,7 @@ def runner(argv, timeout_seconds, child_env): config = { "lanes": { - "muse_cli": { + "claude_review": { "provider_config": { "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], } @@ -207,14 +246,309 @@ def runner(argv, timeout_seconds, child_env): config=config, repo_root=Path(tmp), env={}, - which_fn=lambda cmd: "/opt/bin/muse" if cmd == "muse" else None, + which_fn=lambda cmd: "/opt/bin/claude" if cmd == "claude" else None, auth_probe_runner=runner, - providers=["muse"], + providers=["claude"], ) self.assertEqual(_auth_checks(checks), []) adapter = [c for c in checks if c.name == "doctor.campaign.adapter"] self.assertEqual(adapter[0].status, STATUS_PASS) + def test_antigravity_requires_ambient_home_opt_in(self) -> None: + config = { + "lanes": { + "antigravity_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + } + } + } + with tempfile.TemporaryDirectory() as tmp: + # Missing opt-in warns and is excluded from ready_providers + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={}, + which_fn=lambda cmd: "/opt/bin/agy" if cmd == "agy" else None, + providers=["antigravity"], + ) + auth_check = next(c for c in checks if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_check.status, STATUS_WARN) + self.assertEqual(auth_check.lane, "antigravity") + self.assertEqual(auth_check.detail.get("auth_probe"), "missing_opt_in") + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertNotIn("antigravity", readiness.detail.get("ready_providers", [])) + self.assertEqual( + readiness.detail.get("provider_readiness", {}).get("antigravity"), + {"command": True, "auth": "missing_opt_in", "structured_result": True}, + ) + + # Present opt-in passes and is included in ready_providers + checks_opted_in = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + which_fn=lambda cmd: "/opt/bin/agy" if cmd == "agy" else None, + providers=["antigravity"], + ) + auth_pass = next(c for c in checks_opted_in if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_pass.status, STATUS_PASS) + self.assertEqual(auth_pass.lane, "antigravity") + readiness_pass = next(c for c in checks_opted_in if c.name == "doctor.campaign.readiness") + self.assertIn("antigravity", readiness_pass.detail.get("ready_providers", [])) + self.assertEqual( + readiness_pass.detail.get("provider_readiness", {}).get("antigravity"), + {"command": True, "auth": "ambient_opt_in", "structured_result": True}, + ) + + # Disabled probe skips auth check, emits no warning, and stays in ready_providers + checks_disabled = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={CAMPAIGN_AUTH_PROBE_ENV: "0"}, + which_fn=lambda cmd: "/opt/bin/agy" if cmd == "agy" else None, + providers=["antigravity"], + ) + auth_skip = next(c for c in checks_disabled if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_skip.status, STATUS_SKIP) + self.assertEqual(auth_skip.detail.get("auth_probe"), "skipped") + readiness_skip = next(c for c in checks_disabled if c.name == "doctor.campaign.readiness") + self.assertIn("antigravity", readiness_skip.detail.get("ready_providers", [])) + self.assertEqual( + readiness_skip.detail.get("provider_readiness", {}).get("antigravity"), + {"command": True, "auth": "skipped", "structured_result": True}, + ) + + def test_muse_requires_api_key_or_ambient_home(self) -> None: + config = { + "lanes": { + "muse_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + } + } + } + with tempfile.TemporaryDirectory() as tmp: + # Missing auth warns and drops from ready_providers + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={}, + which_fn=lambda cmd: "/opt/bin/muse" if cmd == "muse" else None, + providers=["muse"], + ) + auth_check = next(c for c in checks if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_check.status, STATUS_WARN) + self.assertEqual(auth_check.detail.get("auth_probe"), "missing_auth") + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness.detail.get("ready_providers", [])) + + # Opted in via ambient home passes + checks_ambient = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"MUSE_CLI_USE_AMBIENT_HOME": "1"}, + which_fn=lambda cmd: "/opt/bin/muse" if cmd == "muse" else None, + providers=["muse"], + ) + auth_pass = next(c for c in checks_ambient if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_pass.status, STATUS_PASS) + readiness_ambient = next(c for c in checks_ambient if c.name == "doctor.campaign.readiness") + self.assertIn("muse", readiness_ambient.detail.get("ready_providers", [])) + + # Opted in via META_API_KEY passes + checks_key = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"META_API_KEY": "dummy-key"}, + which_fn=lambda cmd: "/opt/bin/muse" if cmd == "muse" else None, + providers=["muse"], + ) + auth_key = next(c for c in checks_key if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_key.status, STATUS_PASS) + readiness_key = next(c for c in checks_key if c.name == "doctor.campaign.readiness") + self.assertIn("muse", readiness_key.detail.get("ready_providers", [])) + + # Disabled probe skips auth check, emits no warning, and stays in ready_providers + checks_disabled = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={CAMPAIGN_AUTH_PROBE_ENV: "0"}, + which_fn=lambda cmd: "/opt/bin/muse" if cmd == "muse" else None, + providers=["muse"], + ) + auth_skip = next(c for c in checks_disabled if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_skip.status, STATUS_SKIP) + self.assertEqual(auth_skip.detail.get("auth_probe"), "skipped") + readiness_skip = next(c for c in checks_disabled if c.name == "doctor.campaign.readiness") + self.assertIn("muse", readiness_skip.detail.get("ready_providers", [])) + self.assertEqual( + readiness_skip.detail.get("provider_readiness", {}).get("muse"), + {"command": True, "auth": "skipped", "structured_result": True}, + ) + + def test_muse_meta_api_key_file_validation(self) -> None: + config = { + "lanes": { + "muse_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + } + } + } + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + + def which_fn(cmd: str) -> str | None: + return "/opt/bin/muse" if cmd == "muse" else None + + # 1. Missing file: path does not exist + missing_file = tmp_path / "nonexistent.key" + checks_missing = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(missing_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_missing = next(c for c in checks_missing if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_missing.status, STATUS_WARN) + self.assertEqual(auth_missing.detail.get("auth_probe"), "missing_auth") + readiness_missing = next(c for c in checks_missing if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_missing.detail.get("ready_providers", [])) + self.assertNotIn(str(missing_file), auth_missing.message) + self.assertNotIn(str(missing_file), str(auth_missing.detail)) + + # 2. Non-file (directory) + non_file = tmp_path / "key_dir" + non_file.mkdir() + checks_dir = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(non_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_dir = next(c for c in checks_dir if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_dir.status, STATUS_WARN) + readiness_dir = next(c for c in checks_dir if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_dir.detail.get("ready_providers", [])) + + # 3. Unreadable file where portable + unreadable_file = tmp_path / "unreadable.key" + unreadable_file.write_text("secret-unreadable-token", encoding="utf-8") + try: + unreadable_file.chmod(0o000) + is_unreadable = not os.access(unreadable_file, os.R_OK) + except OSError: + is_unreadable = False + if is_unreadable: + checks_unreadable = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(unreadable_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_unreadable = next(c for c in checks_unreadable if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_unreadable.status, STATUS_WARN) + readiness_unreadable = next(c for c in checks_unreadable if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_unreadable.detail.get("ready_providers", [])) + self.assertNotIn("secret-unreadable-token", str(auth_unreadable.detail)) + self.assertNotIn(str(unreadable_file), str(auth_unreadable.detail)) + unreadable_file.chmod(0o600) + + # 4. Empty and whitespace-only file + empty_file = tmp_path / "empty.key" + empty_file.write_text("", encoding="utf-8") + checks_empty = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(empty_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_empty = next(c for c in checks_empty if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_empty.status, STATUS_WARN) + readiness_empty = next(c for c in checks_empty if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_empty.detail.get("ready_providers", [])) + + ws_file = tmp_path / "whitespace.key" + ws_file.write_text(" \n\t \n", encoding="utf-8") + checks_ws = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(ws_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_ws = next(c for c in checks_ws if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_ws.status, STATUS_WARN) + readiness_ws = next(c for c in checks_ws if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_ws.detail.get("ready_providers", [])) + + # 5. Malformed file (rejected assignment or comments only) + malformed_file = tmp_path / "malformed.key" + malformed_file.write_text("OTHER_API_KEY=token123\n", encoding="utf-8") + checks_malformed = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(malformed_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_malformed = next(c for c in checks_malformed if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_malformed.status, STATUS_WARN) + readiness_malformed = next(c for c in checks_malformed if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_malformed.detail.get("ready_providers", [])) + + comment_file = tmp_path / "comments.key" + comment_file.write_text("# Just a comment\n# Another comment\n", encoding="utf-8") + checks_comment = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(comment_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_comment = next(c for c in checks_comment if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_comment.status, STATUS_WARN) + readiness_comment = next(c for c in checks_comment if c.name == "doctor.campaign.readiness") + self.assertNotIn("muse", readiness_comment.detail.get("ready_providers", [])) + + # 6. Valid file coverage (raw key, shell assignment, export assignment) + valid_secret = "muse-super-secret-key-12345" + for filename, content in ( + ("valid_raw.key", valid_secret), + ("valid_shell.key", f"META_API_KEY={valid_secret}"), + ("valid_export.key", f'export META_API_KEY="{valid_secret}"\n'), + ): + with self.subTest(file_type=filename): + valid_file = tmp_path / filename + valid_file.write_text(content, encoding="utf-8") + checks_valid = check_adoption_campaign_readiness( + config=config, + repo_root=tmp_path, + env={"META_API_KEY_FILE": str(valid_file)}, + which_fn=which_fn, + providers=["muse"], + ) + auth_valid = next(c for c in checks_valid if c.name == CAMPAIGN_AUTH_CHECK_NAME) + self.assertEqual(auth_valid.status, STATUS_PASS) + self.assertEqual(auth_valid.detail.get("auth_probe"), "api_key") + readiness_valid = next(c for c in checks_valid if c.name == "doctor.campaign.readiness") + self.assertIn("muse", readiness_valid.detail.get("ready_providers", [])) + # Do not expose path or contents + self.assertNotIn(valid_secret, auth_valid.message) + self.assertNotIn(valid_secret, str(auth_valid.detail)) + self.assertNotIn(valid_secret, str(readiness_valid.detail)) + self.assertNotIn(str(valid_file), auth_valid.message) + self.assertNotIn(str(valid_file), str(auth_valid.detail)) + self.assertNotIn(str(valid_file), str(readiness_valid.detail)) + def test_only_codex_declares_a_campaign_auth_probe(self) -> None: probing = sorted( lane_id @@ -305,6 +639,47 @@ def test_codex_declares_a_bounded_logged_out_signature(self) -> None: self.assertEqual(campaign_auth_logged_out_exit_codes(capability_only), ()) self.assertEqual(campaign_auth_logged_out_markers(capability_only), ()) + def test_disabled_auth_probe_skips_ambient_home_providers_and_preserves_readiness(self) -> None: + """Disabled auth probing must skip Antigravity and Muse ambient checks without warnings, preserving readiness.""" + config = { + "lanes": { + "antigravity_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + }, + "muse_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + }, + } + } + with tempfile.TemporaryDirectory() as tmp: + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={CAMPAIGN_AUTH_PROBE_ENV: "0"}, + which_fn=lambda cmd: f"/opt/bin/{cmd}" if cmd in {"agy", "muse"} else None, + providers=["antigravity", "muse"], + ) + auth_checks = [c for c in checks if c.name == CAMPAIGN_AUTH_CHECK_NAME] + self.assertEqual(len(auth_checks), 2) + for c in auth_checks: + self.assertEqual(c.status, STATUS_SKIP) + self.assertEqual(c.detail.get("auth_probe"), "skipped") + + warn_checks = [ + c for c in checks if c.status == STATUS_WARN and c.lane in {"antigravity", "muse"} + ] + self.assertEqual(warn_checks, []) + + 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", "muse"]) + self.assertEqual(readiness.detail.get("actionable_providers"), []) + self.assertEqual(readiness.detail.get("optional_providers"), []) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_doctor_campaign_readiness.py b/tests/test_doctor_campaign_readiness.py index 095a85ec..76b49a66 100644 --- a/tests/test_doctor_campaign_readiness.py +++ b/tests/test_doctor_campaign_readiness.py @@ -46,6 +46,97 @@ def test_campaign_adapter_passes_when_command_and_adapter_configured(self) -> No self.assertTrue(check.detail.get("adapter_configured")) self.assertTrue(check.detail.get("command_found")) + def test_campaign_runtime_unavailable_excludes_enabled_local_provider(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "codex": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + } + } + } + with mock.patch( + "code_mower.release_campaigns.resolve_supported_runtime", + return_value=None, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + providers=["codex"], + ) + + runtime = next(c for c in checks if c.name == "doctor.campaign.runtime") + self.assertEqual(runtime.status, STATUS_WARN) + self.assertEqual(runtime.lane, "codex") + self.assertTrue(runtime.detail.get("actionable")) + self.assertTrue(runtime.detail.get("owner_action")) + self.assertEqual(runtime.detail.get("error"), "python_runtime_unavailable") + self.assertNotIn("executable", runtime.detail) + self.assertIn("CODE_MOWER_PYTHON", runtime.remediation) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertNotIn("codex", readiness.detail.get("ready_providers", [])) + self.assertIn("codex", readiness.detail.get("actionable_providers", [])) + + def test_campaign_runtime_available_keeps_local_provider_ready(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with mock.patch( + "code_mower.release_campaigns.resolve_supported_runtime", + return_value=("/private/python", "python_3.13"), + ): + checks = check_adoption_campaign_readiness( + config={ + "lanes": { + "codex": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + } + } + } + }, + repo_root=Path(tmp), + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + providers=["codex"], + ) + + runtime = next(c for c in checks if c.name == "doctor.campaign.runtime") + self.assertEqual(runtime.status, STATUS_PASS) + self.assertEqual(runtime.detail.get("runtime_class"), "python_3.13") + self.assertNotIn("/private/python", repr(runtime.as_dict())) + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertIn( + "codex", + readiness.detail.get("ready_providers", []), + [(c.name, c.status, c.lane, c.detail) for c in checks], + ) + + def test_campaign_runtime_unavailable_is_optional_for_disabled_provider(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with mock.patch( + "code_mower.release_campaigns.resolve_supported_runtime", + return_value=None, + ): + checks = check_adoption_campaign_readiness( + config={"lanes": {"codex": {"enabled": False}}}, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + providers=["codex"], + ) + + runtime = next(c for c in checks if c.name == "doctor.campaign.runtime") + self.assertEqual(runtime.status, STATUS_WARN) + self.assertTrue(runtime.detail.get("optional")) + self.assertFalse(runtime.detail.get("actionable")) + self.assertNotIn("owner_action", runtime.detail) + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertNotIn("codex", readiness.detail.get("ready_providers", [])) + self.assertIn("codex", readiness.detail.get("optional_providers", [])) + def test_campaign_adapter_warns_actionable_when_enabled_missing_adapter(self) -> None: with tempfile.TemporaryDirectory() as tmp: repo_root = Path(tmp) @@ -556,6 +647,7 @@ def test_campaign_readiness_points_to_preview_and_ignores_other_lanes(self) -> N checks = check_adoption_campaign_readiness( config={"lanes": {"gitar": {"enabled": True}}}, repo_root=Path(tmp), + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, which_fn=lambda command: f"/bin/{command}", providers=["codex"], ) @@ -600,6 +692,340 @@ def test_run_doctor_passes_configured_repo_to_campaign_readiness(self) -> None: "codemower-ai/code-mower", ) + def test_structured_result_capability_failure_emits_actionable_warn(self) -> None: + """When command and auth pass but structured-result capability fails, emit actionable WARN.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "codex": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + } + } + } + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + return_value=False, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + providers=["codex"], + ) + + # 1. doctor.campaign.adapter passed + adapter_check = next(c for c in checks if c.name == "doctor.campaign.adapter") + self.assertEqual(adapter_check.status, STATUS_PASS) + + # 2. doctor.campaign.structured_result emitted WARN + structured_check = next(c for c in checks if c.name == "doctor.campaign.structured_result") + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertEqual(structured_check.lane, "codex") + self.assertTrue(structured_check.detail.get("actionable")) + self.assertTrue(structured_check.detail.get("owner_action")) + self.assertIn("capability probe failed", structured_check.message) + self.assertIn("Verify codex campaign adapter output parsing", structured_check.remediation) + + # 3. Provider is in actionable_providers and NOT in ready_providers + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertIn("codex", readiness.detail.get("actionable_providers", [])) + self.assertNotIn("codex", readiness.detail.get("ready_providers", [])) + self.assertEqual(readiness.status, STATUS_WARN) + + def test_structured_result_capability_failure_for_hosted_provider(self) -> None: + """When hosted provider credentials pass but structured-result capability fails, emit actionable WARN.""" + with tempfile.TemporaryDirectory() as tmp: + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + return_value=False, + ): + checks = check_adoption_campaign_readiness( + config={"lanes": {"devin": {"enabled": True}}}, + repo_root=Path(tmp), + env={ + "DEVIN_AUDIT_LABEL_TOKEN": "dummy-token", + "CODE_MOWER_DEVIN_CAMPAIGN_TRANSPORT_READY": "1", + }, + repo_slug="codemower-ai/code-mower", + providers=["devin"], + ) + + cred_check = next(c for c in checks if c.name == "doctor.campaign.credentials") + self.assertEqual(cred_check.status, STATUS_PASS) + + structured_check = next(c for c in checks if c.name == "doctor.campaign.structured_result") + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertEqual(structured_check.lane, "devin") + self.assertTrue(structured_check.detail.get("actionable")) + self.assertFalse(structured_check.detail.get("optional")) + self.assertTrue(structured_check.detail.get("owner_action")) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertIn("devin", readiness.detail.get("actionable_providers", [])) + self.assertNotIn("devin", readiness.detail.get("ready_providers", [])) + self.assertEqual(readiness.status, STATUS_WARN) + + def test_structured_result_capability_failure_disabled_optional_local_cli_provider(self) -> None: + """When disabled/optional local_cli provider fails structured-result probe, do not become actionable or flip aggregate readiness.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "claude": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + "codex": { + "enabled": False, + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + }, + }, + } + } + + def fake_capability(provider: str) -> bool: + return provider == "claude" + + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + side_effect=fake_capability, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, + which_fn=lambda cmd: f"/bin/{cmd}" if cmd in {"claude", "codex"} else None, + providers=["claude", "codex"], + ) + + claude_adapter = next(c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "claude") + self.assertEqual(claude_adapter.status, STATUS_PASS) + + codex_adapter = next(c for c in checks if c.name == "doctor.campaign.adapter" and c.lane == "codex") + self.assertEqual(codex_adapter.status, STATUS_PASS) + + structured_check = next( + c for c in checks if c.name == "doctor.campaign.structured_result" and c.lane == "codex" + ) + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertFalse(structured_check.detail.get("actionable")) + self.assertTrue(structured_check.detail.get("optional")) + self.assertNotIn("owner_action", structured_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("codex", readiness.detail.get("ready_providers", [])) + self.assertIn("codex", readiness.detail.get("optional_providers", [])) + self.assertNotIn("codex", readiness.detail.get("actionable_providers", [])) + + def test_structured_result_capability_failure_disabled_by_default_local_cli_provider(self) -> None: + """When disabled-by-default local_cli provider fails structured-result probe, do not become actionable or flip aggregate readiness.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "claude": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + } + } + + def fake_capability(provider: str) -> bool: + return provider == "claude" + + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + side_effect=fake_capability, + ): + 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, + env={"ANTIGRAVITY_CLI_USE_AMBIENT_HOME": "1"}, + providers=["claude", "antigravity"], + ) + + structured_check = next( + c for c in checks if c.name == "doctor.campaign.structured_result" and c.lane == "antigravity" + ) + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertFalse(structured_check.detail.get("actionable")) + self.assertTrue(structured_check.detail.get("optional")) + self.assertNotIn("owner_action", structured_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", [])) + + def test_structured_result_capability_failure_disabled_optional_hosted_provider(self) -> None: + """When disabled/optional hosted provider fails structured-result probe, do not become actionable or flip aggregate readiness.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "codex": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + } + } + + def fake_capability(provider: str) -> bool: + return provider == "codex" + + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + side_effect=fake_capability, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + env={ + "CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0", + "DEVIN_AUDIT_LABEL_TOKEN": "dummy-token", + "CODE_MOWER_DEVIN_CAMPAIGN_TRANSPORT_READY": "1", + }, + repo_slug="codemower-ai/code-mower", + providers=["codex", "devin"], + ) + + cred_check = next(c for c in checks if c.name == "doctor.campaign.credentials" and c.lane == "devin") + self.assertEqual(cred_check.status, STATUS_PASS) + + structured_check = next( + c for c in checks if c.name == "doctor.campaign.structured_result" and c.lane == "devin" + ) + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertFalse(structured_check.detail.get("actionable")) + self.assertTrue(structured_check.detail.get("optional")) + self.assertNotIn("owner_action", structured_check.detail) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness.status, STATUS_PASS) + self.assertIn("codex", readiness.detail.get("ready_providers", [])) + self.assertNotIn("devin", readiness.detail.get("ready_providers", [])) + self.assertIn("devin", readiness.detail.get("optional_providers", [])) + self.assertNotIn("devin", readiness.detail.get("actionable_providers", [])) + + def test_structured_result_capability_failure_disabled_optional_saas_event_provider(self) -> None: + """When disabled/optional saas_event hosted provider fails structured-result probe, do not become actionable or flip aggregate readiness.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "codex": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + } + } + + def fake_capability(provider: str) -> bool: + return provider == "codex" + + with mock.patch( + "code_mower.campaign_adapters.check_structured_result_capability", + side_effect=fake_capability, + ): + checks = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd == "codex" else None, + env={ + "CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0", + "CURSOR_BUGBOT_AUDIT_LABEL_TOKEN": "dummy-token", + "CODE_MOWER_CURSOR_BUGBOT_CAMPAIGN_TRANSPORT_READY": "1", + }, + repo_slug="codemower-ai/code-mower", + providers=["codex", "cursor_bugbot"], + ) + + cred_check = next(c for c in checks if c.name == "doctor.campaign.credentials" and c.lane == "cursor_bugbot") + self.assertEqual(cred_check.status, STATUS_PASS) + + structured_check = next( + c for c in checks if c.name == "doctor.campaign.structured_result" and c.lane == "cursor_bugbot" + ) + self.assertEqual(structured_check.status, STATUS_WARN) + self.assertFalse(structured_check.detail.get("actionable")) + self.assertTrue(structured_check.detail.get("optional")) + self.assertNotIn("owner_action", structured_check.detail) + + readiness = next(c for c in checks if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness.status, STATUS_PASS) + self.assertIn("codex", readiness.detail.get("ready_providers", [])) + self.assertNotIn("cursor_bugbot", readiness.detail.get("ready_providers", [])) + self.assertIn("cursor_bugbot", readiness.detail.get("optional_providers", [])) + self.assertNotIn("cursor_bugbot", readiness.detail.get("actionable_providers", [])) + + def test_disabled_auth_probe_preserves_aggregate_readiness_for_ambient_providers(self) -> None: + """Setting CODE_MOWER_CAMPAIGN_AUTH_PROBE=0 keeps Antigravity and Muse ready without opt-ins or warnings.""" + with tempfile.TemporaryDirectory() as tmp: + config = { + "lanes": { + "antigravity_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + "muse_cli": { + "provider_config": { + "campaign_adapter_argv": ["{command}", "qualify", "--output", "{output}"], + "campaign_adapter_timeout_seconds": 60, + } + }, + } + } + + # 1. Probing enabled (default) without ambient opt-ins emits warnings and excludes both + checks_probed = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd in {"agy", "muse"} else None, + env={}, + providers=["antigravity", "muse"], + ) + readiness_probed = next(c for c in checks_probed if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness_probed.status, STATUS_WARN) + self.assertEqual(readiness_probed.detail.get("ready_providers"), []) + self.assertIn("antigravity", readiness_probed.detail.get("actionable_providers", [])) + self.assertIn("muse", readiness_probed.detail.get("actionable_providers", [])) + + # 2. Probing disabled via CODE_MOWER_CAMPAIGN_AUTH_PROBE=0 skips auth, no warnings, both ready + checks_disabled = check_adoption_campaign_readiness( + config=config, + repo_root=Path(tmp), + which_fn=lambda cmd: f"/bin/{cmd}" if cmd in {"agy", "muse"} else None, + env={"CODE_MOWER_CAMPAIGN_AUTH_PROBE": "0"}, + providers=["antigravity", "muse"], + ) + readiness_disabled = next(c for c in checks_disabled if c.name == "doctor.campaign.readiness") + self.assertEqual(readiness_disabled.status, STATUS_PASS) + self.assertEqual(readiness_disabled.detail.get("ready_providers"), ["antigravity", "muse"]) + self.assertEqual(readiness_disabled.detail.get("actionable_providers"), []) + self.assertEqual(readiness_disabled.detail.get("optional_providers"), []) + warn_checks = [ + c for c in checks_disabled if c.status == STATUS_WARN and c.lane in {"antigravity", "muse"} + ] + self.assertEqual(warn_checks, []) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_release_campaigns.py b/tests/test_release_campaigns.py index 16fb591d..2122ebe9 100644 --- a/tests/test_release_campaigns.py +++ b/tests/test_release_campaigns.py @@ -11301,5 +11301,409 @@ def test_malformed_or_future_attempted_at_timestamp_degrades_safely(self) -> Non self.assertIn(expected_detail_snippet, card["next_detail"]) +class RuntimeReadinessCampaignTests(unittest.TestCase): + """Tests for deterministic Python 3.12+ runtime resolution and fail-closed readiness.""" + + def test_resolve_supported_runtime_with_env_override(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + custom_py = Path(tmp) / "custom-python" + custom_py.touch() + + def fake_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + if cmd[0] == str(custom_py): + return subprocess.CompletedProcess(cmd, 0, stdout="3.13.1\n", stderr="") + return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") + + res = release_campaigns.resolve_supported_runtime( + environ={"CODE_MOWER_PYTHON": str(custom_py)}, + runner=fake_runner, + ) + self.assertIsNotNone(res) + bin_path, runtime_class = res + self.assertEqual(bin_path, str(custom_py)) + self.assertEqual(runtime_class, "python_3.13") + + def test_resolve_supported_runtime_rejects_unsupported_python(self) -> None: + def fake_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(cmd, 0, stdout="3.9.6\n", stderr="") + + res = release_campaigns.resolve_supported_runtime( + environ={"CODE_MOWER_PYTHON": "/usr/bin/python3.9"}, + runner=fake_runner, + which_fn=lambda _: "/usr/bin/python3.9", + ) + self.assertIsNone(res) + + def test_resolve_supported_runtime_probes_versioned_candidates(self) -> None: + def fake_which(cmd: str) -> str | None: + if cmd == "python3.12": + return "/usr/local/bin/python3.12" + return None + + def fake_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + if cmd[0] == "/usr/local/bin/python3.12": + return subprocess.CompletedProcess(cmd, 0, stdout="3.12.3\n", stderr="") + return subprocess.CompletedProcess(cmd, 0, stdout="3.10.0\n", stderr="") + + res = release_campaigns.resolve_supported_runtime( + environ={}, + runner=fake_runner, + which_fn=fake_which, + ) + self.assertIsNotNone(res) + bin_path, runtime_class = res + self.assertEqual(bin_path, "/usr/local/bin/python3.12") + self.assertEqual(runtime_class, "python_3.12") + + def test_resolve_supported_runtime_discovers_newer_versioned_python(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + bin_dir = Path(tmp) / "bin" + bin_dir.mkdir() + python315 = bin_dir / "python3.15" + python315.touch() + + def fake_which(cmd: str) -> str | None: + return str(python315) if cmd == "python3.15" else None + + def fake_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + if cmd[0] == str(python315): + return subprocess.CompletedProcess(cmd, 0, stdout="3.15.0\n", stderr="") + return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") + + res = release_campaigns.resolve_supported_runtime( + environ={"PATH": str(bin_dir)}, + runner=fake_runner, + which_fn=fake_which, + ) + + self.assertEqual(res, (str(python315), "python_3.15")) + + def test_invoke_local_adapter_fails_closed_when_runtime_unavailable(self) -> None: + invoked: list[str] = [] + + def fake_runner(argv: list[str], timeout: int) -> subprocess.CompletedProcess[str]: + invoked.append("adapter_invoked") + return subprocess.CompletedProcess(argv, 0, stdout="", stderr="") + + lane = _fake_local_cli_lane() + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + output_path = tmp_path / "output.json" + with mock.patch( + "code_mower.release_campaigns.resolve_supported_runtime", + return_value=None, + ): + result, error, detail = release_campaigns._invoke_local_adapter( + lane, + "codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + qualification_context="cold_install", + starting_version="", + output_path=output_path, + repo_path=tmp_path, + adapter_runner=fake_runner, + which_fn=lambda _: "/bin/codex", + ) + + self.assertIsNone(result) + self.assertEqual(error, "python_runtime_unavailable") + self.assertIn("supported Python 3.12+ runtime is unavailable", detail) + # Ensure provider adapter was never invoked + self.assertEqual(len(invoked), 0) + + def test_provider_next_action_remediation_on_runtime_unavailable(self) -> None: + lane = _fake_local_cli_lane() + next_action, next_detail = release_campaigns._provider_next_action( + provider="codex", + lane=lane, + state="unavailable", + command_available=True, + has_credentials=True, + has_issue=True, + dry_run=False, + error_code="python_runtime_unavailable", + error="supported Python 3.12+ runtime is unavailable", + ) + self.assertIn("install Python 3.12+ on PATH or set CODE_MOWER_PYTHON", next_action) + self.assertIn("supported Python 3.12+ runtime is unavailable", next_detail) + + def test_provider_next_action_remediation_keyed_on_error_code_not_message(self) -> None: + """Remediation must key on stable python_runtime_unavailable error code, not free-text detail.""" + lane = _fake_local_cli_lane() + next_action, next_detail = release_campaigns._provider_next_action( + provider="codex", + lane=lane, + state="unavailable", + command_available=True, + has_credentials=True, + has_issue=True, + dry_run=False, + error_code="python_runtime_unavailable", + error="a different arbitrary human error message", + ) + self.assertEqual("install Python 3.12+ on PATH or set CODE_MOWER_PYTHON", next_action) + self.assertEqual("a different arbitrary human error message", next_detail) + + def test_maintained_adapter_argv_includes_python_bin_and_target_runtime(self) -> None: + from code_mower.provider_registry import _maintained_campaign_adapter_argv + + argv = _maintained_campaign_adapter_argv("codex") + self.assertEqual(argv[0], "{python}") + self.assertIn("--python-bin", argv) + python_bin_idx = argv.index("--python-bin") + self.assertEqual(argv[python_bin_idx + 1], "{target_python}") + self.assertIn("--target-runtime", argv) + target_runtime_idx = argv.index("--target-runtime") + self.assertEqual(argv[target_runtime_idx + 1], "{target_runtime}") + + def test_invoke_local_adapter_with_distinct_launcher_and_target_interpreters(self) -> None: + """Launcher must remain sys.executable even when target runtime is a distinct interpreter.""" + captured_argv: list[str] = [] + + def fake_adapter_runner(argv: list[str], timeout: int) -> subprocess.CompletedProcess[str]: + captured_argv.extend(argv) + output_idx = argv.index("--output") + out_file = Path(argv[output_idx + 1]) + valid_payload = { + "schema": "code_mower.adoptionResult.v1", + "timestamp_utc": "2026-09-05T00:00:00Z", + "release_tag": "v1.0.0", + "package_identity": "code-mower", + "normalized_version": "1.0.0", + "qualification_context": "cold_install", + "starting_version": "", + "ending_version": "1.0.0", + "provider": "codex", + "executor": "codex", + "host_class": "local", + "runtime_class": "python_3.12", + "execution_state": "executed", + "elapsed_seconds": 1.0, + "outcome": "pass", + "steps": [ + { + "id": "doctor", + "status": "pass", + "elapsed_seconds": 1.0, + "warning_count": 0, + "owner_action_count": 0, + } + ], + } + out_file.write_text(json.dumps(valid_payload), encoding="utf-8") + return subprocess.CompletedProcess(argv, 0, stdout="", stderr="") + + from code_mower.provider_registry import REFERENCE_PROVIDERS + + distinct_target_python = "/opt/custom_python312/bin/python3.12" + lane = REFERENCE_PROVIDERS["codex"] + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + output_path = tmp_path / "output.json" + with mock.patch( + "code_mower.release_campaigns.resolve_supported_runtime", + return_value=(distinct_target_python, "python_3.12"), + ): + result, error, detail = release_campaigns._invoke_local_adapter( + lane, + "codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + qualification_context="cold_install", + starting_version="", + output_path=output_path, + repo_path=tmp_path, + adapter_runner=fake_adapter_runner, + which_fn=lambda _: "/bin/codex", + ) + + self.assertIsNotNone(result) + self.assertEqual(error, "") + # Launcher must be the interpreter running Code Mower (sys.executable) + self.assertEqual(captured_argv[0], sys.executable) + self.assertNotEqual(captured_argv[0], distinct_target_python) + self.assertEqual(captured_argv[1:3], ["-m", "code_mower.campaign_adapters"]) + # Target Python must be passed to --python-bin + python_bin_idx = captured_argv.index("--python-bin") + self.assertEqual(captured_argv[python_bin_idx + 1], distinct_target_python) + + def test_resolve_supported_runtime_resolves_relative_env_override(self) -> None: + """Relative CODE_MOWER_PYTHON candidate must be resolved to an absolute path.""" + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + custom_dir = tmp_path / "custom_env" + custom_dir.mkdir() + custom_py = custom_dir / "python3" + custom_py.touch() + + def fake_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + if Path(cmd[0]).resolve() == custom_py.resolve(): + return subprocess.CompletedProcess(cmd, 0, stdout="3.12.8\n", stderr="") + return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") + + old_cwd = os.getcwd() + try: + os.chdir(tmp) + for rel_candidate in ("./custom_env/python3", "custom_env/python3"): + with self.subTest(candidate=rel_candidate): + res = release_campaigns.resolve_supported_runtime( + environ={"CODE_MOWER_PYTHON": rel_candidate}, + runner=fake_runner, + ) + self.assertIsNotNone(res) + bin_path, runtime_class = res + self.assertTrue(Path(bin_path).is_absolute(), f"Expected absolute path, got {bin_path}") + self.assertEqual(Path(bin_path).resolve(), custom_py.resolve()) + self.assertEqual(runtime_class, "python_3.12") + finally: + os.chdir(old_cwd) + + def test_invoke_local_adapter_with_relative_code_mower_python_under_different_launch_cwd(self) -> None: + """Relative CODE_MOWER_PYTHON must resolve to absolute path so adapters work under different launch cwd.""" + with tempfile.TemporaryDirectory() as tmp: + repo_root = Path(tmp) / "repo" + repo_root.mkdir() + disposable_cwd = Path(tmp) / "disposable_workdir" + disposable_cwd.mkdir() + + custom_bin_dir = repo_root / "tools" + custom_bin_dir.mkdir() + custom_py = custom_bin_dir / "python3.12" + custom_py.touch() + + def fake_python_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + if Path(cmd[0]).resolve() == custom_py.resolve(): + return subprocess.CompletedProcess(cmd, 0, stdout="3.12.4\n", stderr="") + return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") + + captured_target_python: list[str] = [] + + def fake_adapter_runner(argv: list[str], timeout: int) -> subprocess.CompletedProcess[str]: + python_bin_idx = argv.index("--python-bin") + target_py = argv[python_bin_idx + 1] + captured_target_python.append(target_py) + + # From disposable_cwd, target_py must be valid and exist + old_runner_cwd = os.getcwd() + try: + os.chdir(disposable_cwd) + target_path = Path(target_py) + self.assertTrue(target_path.is_absolute(), f"--python-bin must be absolute: {target_py}") + self.assertTrue(target_path.exists(), f"Target python must exist from disposable cwd: {target_py}") + finally: + os.chdir(old_runner_cwd) + + output_idx = argv.index("--output") + out_file = Path(argv[output_idx + 1]) + valid_payload = { + "schema": "code_mower.adoptionResult.v1", + "timestamp_utc": "2026-09-05T00:00:00Z", + "release_tag": "v1.0.0", + "package_identity": "code-mower", + "normalized_version": "1.0.0", + "qualification_context": "cold_install", + "starting_version": "", + "ending_version": "1.0.0", + "provider": "codex", + "executor": "codex", + "host_class": "local", + "runtime_class": "python_3.12", + "execution_state": "executed", + "elapsed_seconds": 1.0, + "outcome": "pass", + "steps": [ + { + "id": "doctor", + "status": "pass", + "elapsed_seconds": 1.0, + "warning_count": 0, + "owner_action_count": 0, + } + ], + } + out_file.write_text(json.dumps(valid_payload), encoding="utf-8") + return subprocess.CompletedProcess(argv, 0, stdout="", stderr="") + + from code_mower.provider_registry import REFERENCE_PROVIDERS + lane = REFERENCE_PROVIDERS["codex"] + output_path = repo_root / "output.json" + + old_cwd = os.getcwd() + try: + os.chdir(repo_root) + result, error, detail = release_campaigns._invoke_local_adapter( + lane, + "codex", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + qualification_context="cold_install", + starting_version="", + output_path=output_path, + repo_path=repo_root, + adapter_runner=fake_adapter_runner, + python_runner=fake_python_runner, + which_fn=lambda _: "/bin/codex", + environ={"CODE_MOWER_PYTHON": "./tools/python3.12"}, + ) + finally: + os.chdir(old_cwd) + + self.assertIsNotNone(result) + self.assertEqual(error, "") + self.assertEqual(len(captured_target_python), 1) + self.assertEqual(Path(captured_target_python[0]).resolve(), custom_py.resolve()) + + def test_build_adapter_argv_default_and_empty_target_runtime_resolves_bounded_runtime(self) -> None: + """_build_adapter_argv must evaluate bounded target_runtime deterministically rather than NameError.""" + lane = _fake_local_cli_lane() + for empty_rt in (None, ""): + with self.subTest(target_runtime=empty_rt): + kwargs: dict[str, Any] = { + "release_tag": "v1.0.0", + "package_spec": "code-mower==1.0.0", + "qualification_context": "cold_install", + "starting_version": "", + "output_path": Path("/tmp/out.json"), + "repo_path": Path("/tmp/repo"), + "argv_template": ( + "{command}", + "--python", + "{target_python}", + "--runtime", + "{target_runtime}", + ), + } + if empty_rt is not None: + kwargs["target_runtime"] = empty_rt + + argv = release_campaigns._build_adapter_argv( + lane, + "/bin/fake-provider-cli", + **kwargs, + ) + self.assertEqual(argv[0], "/bin/fake-provider-cli") + self.assertEqual(argv[1], "--python") + self.assertTrue(argv[2]) + self.assertEqual(argv[3], "--runtime") + self.assertTrue(re.fullmatch(r"python_\d+\.\d+", argv[4]), f"Expected bounded python_X.Y, got {argv[4]}") + self.assertEqual(argv[4], release_campaigns._detect_runtime_class()) + + # Non-empty target_runtime is preserved + argv_explicit = release_campaigns._build_adapter_argv( + lane, + "/bin/fake-provider-cli", + release_tag="v1.0.0", + package_spec="code-mower==1.0.0", + qualification_context="cold_install", + starting_version="", + output_path=Path("/tmp/out.json"), + repo_path=Path("/tmp/repo"), + argv_template=("{command}", "--runtime", "{target_runtime}"), + target_runtime="python_3.14", + ) + self.assertEqual(argv_explicit, ["/bin/fake-provider-cli", "--runtime", "python_3.14"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_release_qualify.py b/tests/test_release_qualify.py index 6939d485..3e50f48a 100644 --- a/tests/test_release_qualify.py +++ b/tests/test_release_qualify.py @@ -824,6 +824,77 @@ def test_builtin_runner_reports_a_dotted_spec_version_mismatch(self) -> None: ) self.assertIn("Version mismatch", str(ctx.exception)) + def test_validate_adoption_result_payload_requires_python_312_or_higher(self) -> None: + release_qualify.validate_adoption_result_payload(_valid_adoption_result(runtime_class="python_3.12")) + release_qualify.validate_adoption_result_payload(_valid_adoption_result(runtime_class="python_3.13")) + release_qualify.validate_adoption_result_payload(_valid_adoption_result(runtime_class="python_3.14")) + + for unsupported in ("python_3.11", "python_3.10", "python_3.9", "python_3.8"): + with self.subTest(runtime_class=unsupported): + with self.assertRaisesRegex(ValueError, "must be >= python_3.12"): + release_qualify.validate_adoption_result_payload(_valid_adoption_result(runtime_class=unsupported)) + + for non_python in ("unknown", "", "node_22", "ruby_3.2", "python_3", "python_3.12.1", "pypy_3.12"): + with self.subTest(runtime_class=non_python): + with self.assertRaisesRegex(ValueError, "must be 'python_.'"): + release_qualify.validate_adoption_result_payload(_valid_adoption_result(runtime_class=non_python)) + + def test_validate_adoption_result_rejects_unknown_runtime_across_result_types(self) -> None: + """Hosted, manual, and custom results reporting runtime_class unknown must be rejected.""" + # 1. Hosted runner result (e.g. host_class='github_actions') + hosted_unknown = _valid_adoption_result( + host_class="github_actions", + provider="devin", + executor="devin", + runtime_class="unknown", + ) + with self.assertRaisesRegex(ValueError, "must be 'python_.'"): + release_qualify.validate_adoption_result_payload(hosted_unknown) + + hosted_valid = _valid_adoption_result( + host_class="github_actions", + provider="devin", + executor="devin", + runtime_class="python_3.12", + ) + release_qualify.validate_adoption_result_payload(hosted_valid) + + # 2. Manual qualification result + manual_unknown = _valid_adoption_result( + host_class="local", + provider="manual", + executor="manual", + runtime_class="unknown", + ) + with self.assertRaisesRegex(ValueError, "must be 'python_.'"): + release_qualify.validate_adoption_result_payload(manual_unknown) + + manual_valid = _valid_adoption_result( + host_class="local", + provider="manual", + executor="manual", + runtime_class="python_3.12", + ) + release_qualify.validate_adoption_result_payload(manual_valid) + + # 3. Custom-adapter result + custom_unknown = _valid_adoption_result( + host_class="local", + provider="custom_audit", + executor="custom_runner", + runtime_class="unknown", + ) + with self.assertRaisesRegex(ValueError, "must be 'python_.'"): + release_qualify.validate_adoption_result_payload(custom_unknown) + + custom_valid = _valid_adoption_result( + host_class="local", + provider="custom_audit", + executor="custom_runner", + runtime_class="python_3.12", + ) + release_qualify.validate_adoption_result_payload(custom_valid) + if __name__ == "__main__": unittest.main()