tracer-cloud--opensre/integrations/llm_cli/gemini_cli.py

"""Google Gemini CLI adapter (``gemini -p``, non-interactive headless mode).

Env vars
--------
GEMINI_CLI_BIN   Optional explicit path to the ``gemini`` binary.
GEMINI_CLI_MODEL Optional model override passed as ``--model``.
GEMINI_CLI_TIMEOUT_SECONDS Optional invocation timeout override for long prompts.

Auth
----
Gemini CLI supports multiple auth modes (cached login sessions, ``GEMINI_API_KEY``,
Vertex env credentials). Probe classification uses a short headless call and
maps outcomes to ``logged_in``: True / False / None.
"""

from __future__ import annotations

import json
import os
import subprocess

from integrations.llm_cli.base import CLIInvocation, CLIProbe
from integrations.llm_cli.binary_resolver import (
    candidate_binary_names as _candidate_binary_names,
)
from integrations.llm_cli.binary_resolver import (
    default_cli_fallback_paths as _default_cli_fallback_paths,
)
from integrations.llm_cli.binary_resolver import (
    resolve_cli_binary,
)
from integrations.llm_cli.constants import (
    DEFAULT_EXEC_TIMEOUT_SEC as _DEFAULT_EXEC_TIMEOUT_SEC,
)
from integrations.llm_cli.constants import (
    MAX_EXEC_TIMEOUT_SEC as _MAX_EXEC_TIMEOUT_SEC,
)
from integrations.llm_cli.constants import (
    MIN_EXEC_TIMEOUT_SEC as _MIN_EXEC_TIMEOUT_SEC,
)
from integrations.llm_cli.probe_utils import run_version_probe
from integrations.llm_cli.semver_utils import parse_semver_three_part
from integrations.llm_cli.subprocess_env import build_cli_subprocess_env
from integrations.llm_cli.timeout_utils import resolve_timeout_from_env

_PROBE_TIMEOUT_SEC = 20.0
_AUTH_HINT = "Run: gemini (interactive login) or set GEMINI_API_KEY."
# Source: https://developers.googleblog.com/an-important-update-transitioning-gemini-cli-to-antigravity-cli/
_SUNSET_NOTE = (
    " Note: Gemini CLI stops serving Pro/Ultra and free users on 2026-06-18; "
    "paid Gemini Code Assist remains supported. Consider antigravity-cli (agy)."
)


def _resolve_exec_timeout_seconds() -> float:
    return resolve_timeout_from_env(
        env_key="GEMINI_CLI_TIMEOUT_SECONDS",
        default=_DEFAULT_EXEC_TIMEOUT_SEC,
        minimum=_MIN_EXEC_TIMEOUT_SEC,
        maximum=_MAX_EXEC_TIMEOUT_SEC,
    )


def _gemini_auth_env_overrides() -> dict[str, str]:
    """Build Gemini subprocess auth/config overrides used by probe and invoke."""
    env: dict[str, str] = {"NO_COLOR": "1"}
    keys = (
        "GEMINI_API_KEY",
        "GOOGLE_API_KEY",
        "GOOGLE_GENAI_USE_VERTEXAI",
        "GOOGLE_APPLICATION_CREDENTIALS",
        "GOOGLE_CLOUD_PROJECT",
        "GOOGLE_CLOUD_LOCATION",
    )
    for key in keys:
        val = os.environ.get(key, "").strip()
        if val:
            env[key] = val
    return env


def _has_explicit_gemini_auth_env() -> str | None:
    env = _gemini_auth_env_overrides()
    for key in ("GEMINI_API_KEY", "GOOGLE_API_KEY", "GOOGLE_APPLICATION_CREDENTIALS"):
        if env.get(key):
            return key
    if env.get("GOOGLE_GENAI_USE_VERTEXAI") and env.get("GOOGLE_CLOUD_PROJECT"):
        return "GOOGLE_GENAI_USE_VERTEXAI"
    return None


def _classify_gemini_auth(returncode: int, stdout: str, stderr: str) -> tuple[bool | None, str]:
    raw = (stdout or "").strip()
    if raw.startswith("{"):
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            payload = None
        if isinstance(payload, dict):
            err = payload.get("error")
            if isinstance(err, dict):
                message = str(err.get("message", "")).strip()
                code = err.get("code")
                msg_lower = message.lower()
                if (
                    "please set an auth method" in msg_lower
                    or "gemini_api_key" in msg_lower
                    or "not authenticated" in msg_lower
                    or "login required" in msg_lower
                    or code == 41
                ):
                    return False, f"Not authenticated. {_AUTH_HINT}"
    text = (stdout + "\n" + stderr).lower()
    if "not authenticated" in text or ("authentication" in text and "required" in text):
        return False, f"Not authenticated. {_AUTH_HINT}"
    if "login required" in text or "please login" in text:
        return False, f"Not authenticated. {_AUTH_HINT}"
    if "please set an auth method" in text:
        return False, f"Not authenticated. {_AUTH_HINT}"
    if "invalid api key" in text or ("api key" in text and "missing" in text):
        return False, "Gemini API key missing or invalid. Set GEMINI_API_KEY or login via `gemini`."
    if returncode == 0:
        return True, "Authenticated via Gemini CLI."
    if "network" in text or "timeout" in text or "unreachable" in text or "connection" in text:
        return None, "Network error while checking auth; will retry at invocation."
    tail = (stderr or stdout).strip()[:200]
    if tail:
        return None, f"Auth status unclear (exit {returncode}): {tail}"
    return None, f"Auth status unclear (exit {returncode})."


def _fallback_gemini_cli_paths() -> list[str]:
    return _default_cli_fallback_paths("gemini")


class GeminiCLIAdapter:
    """Non-interactive Gemini CLI (``gemini -p`` headless mode)."""

    name = "gemini-cli"
    binary_env_key = "GEMINI_CLI_BIN"
    install_hint = "npm i -g @google/gemini-cli"
    auth_hint = _AUTH_HINT.removesuffix(".")
    min_version: str | None = None
    default_exec_timeout_sec = _DEFAULT_EXEC_TIMEOUT_SEC

    def _resolve_binary(self) -> str | None:
        return resolve_cli_binary(
            explicit_env_key="GEMINI_CLI_BIN",
            binary_names=_candidate_binary_names("gemini"),
            fallback_paths=_fallback_gemini_cli_paths,
        )

    def _probe_binary(self, binary_path: str) -> CLIProbe:
        version_output, version_error = run_version_probe(
            binary_path,
            timeout_sec=_PROBE_TIMEOUT_SEC,
        )
        if version_error:
            return CLIProbe(
                installed=False,
                version=None,
                logged_in=None,
                bin_path=None,
                detail=version_error,
            )

        version = parse_semver_three_part(version_output or "")
        probe_env = build_cli_subprocess_env(_gemini_auth_env_overrides())
        try:
            auth_proc = subprocess.run(
                [binary_path, "-p", "respond with: ok", "--output-format", "json"],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=_PROBE_TIMEOUT_SEC,
                check=False,
                env=probe_env,
            )
        except subprocess.TimeoutExpired:
            logged_in = None
            auth_detail = (
                f"Gemini auth probe timed out after {_PROBE_TIMEOUT_SEC:.0f}s; auth status unknown."
            )
        except OSError as exc:
            logged_in = None
            auth_detail = f"Could not spawn gemini for auth probe: {exc}"
        else:
            logged_in, auth_detail = _classify_gemini_auth(
                auth_proc.returncode, auth_proc.stdout, auth_proc.stderr
            )

        auth_env_source = _has_explicit_gemini_auth_env()
        if logged_in is not True and auth_env_source:
            logged_in = True
            auth_detail = f"Authenticated via {auth_env_source} fallback."

        # Gate the sunset notice on ``logged_in is not False`` so a user mid
        # auth-failure isn't shown the migration ad next to the actual error
        # they need to read. Authenticated (True) and ambiguous (None) probes
        # still surface it — those are the cohorts whose CLI is operational
        # today and who need to plan for 2026-06-18.
        detail = auth_detail + _SUNSET_NOTE if logged_in is not False else auth_detail

        return CLIProbe(
            installed=True,
            version=version,
            logged_in=logged_in,
            bin_path=binary_path,
            detail=detail,
        )

    def detect(self) -> CLIProbe:
        binary = self._resolve_binary()
        if not binary:
            return CLIProbe(
                installed=False,
                version=None,
                logged_in=None,
                bin_path=None,
                detail=(
                    "Gemini CLI not found on PATH or known install locations. "
                    f"Install with: {self.install_hint} or set GEMINI_CLI_BIN."
                ),
            )
        return self._probe_binary(binary)

    def build(
        self,
        *,
        prompt: str,
        model: str | None,
        workspace: str,
        reasoning_effort: str | None = None,
    ) -> CLIInvocation:
        _ = reasoning_effort
        binary = self._resolve_binary()
        if not binary:
            raise RuntimeError(
                f"Gemini CLI not found. {self.install_hint} "
                "or set GEMINI_CLI_BIN to the full binary path."
            )

        argv: list[str] = [binary, "-p", prompt, "--output-format", "json"]
        resolved_model = (model or "").strip()
        if resolved_model:
            argv.extend(["--model", resolved_model])

        ws = (workspace or "").strip()
        cwd = ws or os.getcwd()
        env = _gemini_auth_env_overrides()

        return CLIInvocation(
            argv=tuple(argv),
            stdin=None,
            cwd=cwd,
            env=env,
            timeout_sec=_resolve_exec_timeout_seconds(),
        )

    def parse(self, *, stdout: str, stderr: str, returncode: int) -> str:
        text = (stdout or "").strip()
        if not text:
            raise RuntimeError(
                self.explain_failure(stdout=stdout, stderr=stderr, returncode=returncode)
                + " (empty output)"
            )
        try:
            payload = json.loads(text)
        except json.JSONDecodeError:
            return text
        if isinstance(payload, dict):
            response = payload.get("response")
            if isinstance(response, str):
                return response.strip()
            err = payload.get("error")
            if isinstance(err, dict):
                message = err.get("message")
                if isinstance(message, str) and message.strip():
                    raise RuntimeError(f"Gemini CLI returned an error: {message.strip()}")
        return text

    def explain_failure(self, *, stdout: str, stderr: str, returncode: int) -> str:
        from integrations.llm_cli.failure_explain import explain_cli_failure

        return explain_cli_failure(
            exit_label="gemini -p",
            stdout=stdout,
            stderr=stderr,
            returncode=returncode,
        )