Validate and configure Antigravity repair models

README.md

@@ -604,7 +604,7 @@ payloads into normal public GitHub comments, so raw JSON is not posted.
 
 When a structured plan review, plan revision, PR review, or coder follow-up is
 present but malformed, the loop may run a model-backed repair pass. The default
-backend is Antigravity (`agy`) with the single model `Gemini 3 Flash`. The
+backend is Antigravity (`agy`) with the default model `Gemini 3.5 Flash (Medium)`. The
 repair pass is format-only: it asks the model to re-emit the agent's intent as the
 required JSON object, footer marker, and signature. The repaired response is
 accepted only if it passes the same strict validation as the original response;
@@ -613,10 +613,13 @@ failed repairs remain local protocol errors and are not posted to GitHub.
 Repair runs in a fresh temporary directory with an empty tool allow-list and a
 repair-only `GEMINI.md`; it receives no checkout or repository context. Configure
 it with `--repair-backend antigravity|gemini`, repeat `--repair-model` to define
-an explicit ordered fallback chain, and set `--repair-timeout-seconds`. The
-normal Antigravity coder/reviewer model chain is never inherited. For example:
+an explicit ordered fallback chain, and set `--repair-timeout-seconds`. Explicit
+repair models are tried first, followed by the configured Antigravity chain with
+duplicates removed. The loop validates candidates once with `agy models`, reports
+available choices for stale names, and attempts candidates directly when discovery
+is unavailable. For example:
 
-`--repair-model "Gemini 3 Flash" --repair-model "Gemini 3.1 Pro (High)"`
+`--repair-model "Gemini 3.5 Flash (Medium)" --repair-model "Gemini 3.1 Pro (High)"`
 
 The legacy `gemini --prompt` path is used only with
 `--repair-backend gemini` and requires non-interactive enterprise/API-key/Vertex

docs/local_agent_loop.md

@@ -1177,8 +1177,8 @@ the attempt log path instead of posting that raw output as a review.
 For structured plan reviews, plan revisions, PR reviews, and coder follow-ups,
 a present but malformed structured response may get a repair pass before the
 local failure is raised. By default the repair pass calls Antigravity through
-the existing PTY backend with the single model `Gemini 3 Flash`; the normal
-coder/reviewer fallback chain is not inherited. It uses a fresh temporary
+the existing PTY backend with the default model `Gemini 3.5 Flash (Medium)`;
+explicit repair models are followed by the configured coder/reviewer chain. It uses a fresh temporary
 workdir, empty tool permissions, and repair-only instructions forbidding file
 inspection, tests, mutation, background work, and subagents. The format-repair
 prompt asks it to preserve the agent's intent while emitting
@@ -1345,8 +1345,11 @@ output`. A recovered response is still revalidated before posting; a failed
 repair leaves the run in local failure just like any other protocol error.
 
 Use `--repair-backend`, repeatable `--repair-model`, and
-`--repair-timeout-seconds` to configure this path. Multiple models are the only
-repair fallback chain. `--repair-backend gemini` retains the legacy CLI for
+`--repair-timeout-seconds` to configure this path. The repair chain is the explicit
+prefix followed by `antigravity_models`, with duplicates removed. It queries `agy
+models` once through the same PTY-safe runner as `agy --print`, rejects stale names
+with available-choice guidance, and attempts candidates directly when discovery is
+unavailable. `--repair-backend gemini` retains the legacy CLI for
 enterprise/API-key/Vertex authentication; personal OAuth may require an
 interactive authorization and is not the default. Normal diagnostics include
 backend, model, return code, a sanitized bounded combined-PTY diagnostic, log

docs/skill_mode.md

@@ -158,8 +158,9 @@ never falsely reported `approved`. A malformed-but-content-bearing structured
 review is recovered automatically when possible: safe skill normalization,
 envelope normalization, deterministic unknown-prior-item stripping against the
 complete carried ledger, and then model format repair. Local-loop repair now
-defaults to isolated Antigravity with `Gemini 3 Flash`; explicit repeatable
-repair models define the only fallback chain. Legacy Gemini CLI repair requires
+defaults to isolated Antigravity with `Gemini 3.5 Flash (Medium)`; explicit repeatable
+repair models are tried first, followed by the configured Antigravity chain. Legacy
+Gemini CLI repair requires
 `--repair-backend gemini` and suitable non-interactive authentication. The
 classifier is conservative: only known tooling-failure signatures or
 truly-empty output count as unavailable.

src/coding_review_agent_loop/agents/antigravity.py

@@ -39,7 +39,7 @@
 from ..errors import AgentLoopError
 from ..logging import agent_log_path, log
 from ..protocol import PUBLIC_RESPONSE_MARKER
-from ..runner import Runner
+from ..runner import Runner, strip_ansi
 
 if TYPE_CHECKING:
     from ..config import AgentLoopConfig
@@ -133,6 +133,22 @@ def _strip_public_response_marker(raw: str) -> tuple[str, AgentTextSource]:
     return raw.rsplit(PUBLIC_RESPONSE_MARKER, 1)[1].lstrip("\n"), "stdout_marker"
 
 
+def _parse_model_catalog(raw: str) -> set[str]:
+    """Extract model labels from the human-readable ``agy models`` output."""
+    models: set[str] = set()
+    for line in strip_ansi(raw).splitlines():
+        value = line.strip().lstrip("-*• ").strip()
+        if not value or set(value) <= {"-", "=", "_"}:
+            continue
+        lowered = value.lower().rstrip(":")
+        if lowered in {"models", "available models", "available model", "name", "model"}:
+            continue
+        if lowered.startswith(("error:", "warning:", "usage:", "command:")):
+            continue
+        models.add(value)
+    return models
+
+
 class AntigravityBackend:
     name: AgentName = "antigravity"
     display_name = "Antigravity"
@@ -144,6 +160,35 @@ def workdir(self, config: AgentLoopConfig) -> Path:
     def default_args(self, *, dangerous: bool) -> tuple[str, ...]:
         return ("--dangerously-skip-permissions",) if dangerous else ()
 
+    def discover_models(
+        self, runner: Runner, config: AgentLoopConfig, *, timeout_seconds: float
+    ) -> tuple[set[str] | None, str]:
+        """Query agy's model catalog once, preserving the PTY requirement."""
+        repair_root = Path(tempfile.gettempdir()) / "coding-review-agent-loop" / "repair"
+        repair_root.mkdir(parents=True, exist_ok=True)
+        log_path = agent_log_path(config, "antigravity-repair-models")
+        with tempfile.TemporaryDirectory(prefix="agy-catalog-", dir=repair_root) as temp_dir:
+            try:
+                result = runner.run_with_log(
+                    [config.antigravity_cmd, *config.antigravity_args, "models"],
+                    cwd=Path(temp_dir),
+                    log_path=log_path,
+                    label="Antigravity model catalog",
+                    progress_interval_seconds=config.progress_interval_seconds,
+                    check=False,
+                    env={"AGENT_LOOP_WORKDIR": str(Path(temp_dir).resolve())},
+                    use_pty=True,
+                    timeout_seconds=timeout_seconds,
+                )
+            except Exception as exc:
+                return None, str(exc)
+        if result.returncode != 0:
+            return None, result.stdout or result.stderr or f"agy models exited with {result.returncode}"
+        models = _parse_model_catalog(result.stdout)
+        if not models:
+            return None, "agy models returned no parseable model choices"
+        return models, result.stdout
+
     def run(
         self,
         runner: Runner,

src/coding_review_agent_loop/config.py

@@ -37,7 +37,7 @@
     "Gemini 3.5 Flash (High)",
     "Gemini 3.1 Pro (High)",
 )
-DEFAULT_REPAIR_MODELS: tuple[str, ...] = ("Gemini 3 Flash",)
+DEFAULT_REPAIR_MODELS: tuple[str, ...] = ("Gemini 3.5 Flash (Medium)",)
 
 # Discuss-mode research policy values (#477). The CLI flag choices, the config
 # validation, and the prompt builders all derive from this set.

src/coding_review_agent_loop/repair.py

@@ -1005,7 +1005,8 @@ def attempt_envelope_normalization(raw: str, *, expected_kind: str | None) -> st
 _REPAIR_MODEL = "gemini-3.1-flash-lite"
 _SUPPORTED_EXPECTED_KINDS = {"plan_state", "pr_review", "plan_review", "coder_followup", "plan_revision", "discuss_review", "discuss_answer", "discuss_agenda", "discuss_semantic_comparison", "discuss_answer_confirmation"}
 RepairOutcome = Literal[
-    "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output"
+    "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output",
+    "unavailable_model",
 ]
 
 
@@ -1109,14 +1110,41 @@ def execute_repair(
     """Run the configured repair chain and validate each candidate."""
     prompt = _build_repair_prompt(raw, **prompt_kwargs)
     attempts: list[RepairAttemptResult] = []
-    models = config.repair_models
+    if config.repair_backend == "antigravity":
+        models = tuple(dict.fromkeys((*config.repair_models, *config.antigravity_models)))
+        catalog, catalog_diagnostic = AntigravityBackend().discover_models(
+            runner, config, timeout_seconds=min(float(config.repair_timeout_seconds), 30.0)
+        )
+    else:
+        models = config.repair_models
+        catalog = None
+        catalog_diagnostic = ""
     for index, model in enumerate(models):
         fallback_planned = index + 1 < len(models)
         log_path: Path | None = None
         output = ""
         returncode: int | None = None
         diagnostic = ""
         outcome: RepairOutcome
+        if catalog is not None and model not in catalog:
+            choices = ", ".join(sorted(catalog))
+            diagnostic = _sanitize_diagnostic(
+                f"Repair model {model!r} is not available in agy models. "
+                f"Available choices: {choices}", config=config
+            )
+            attempts.append(RepairAttemptResult(
+                backend="antigravity", model=model, prompt=prompt, output="",
+                returncode=None, outcome="unavailable_model", diagnostic=diagnostic,
+                log_path=None, fallback_planned=fallback_planned,
+            ))
+            if usage_context is not None:
+                usage_context.add_record(
+                    agent="antigravity", session_id=None, returncode=None,
+                    usage=estimate_usage(prompt, ""), role="repair", model=model,
+                    outcome="unavailable_model", log_path=None,
+                    fallback_planned=fallback_planned,
+                )
+            continue
         try:
             if config.repair_backend == "antigravity":
                 log_path = agent_log_path(config, "antigravity-repair", run_id=run_id)
@@ -1175,6 +1203,16 @@ def execute_repair(
                 log_path.parent.mkdir(parents=True, exist_ok=True)
                 log_path.write_text(diagnostic + "\n", encoding="utf-8")
 
+        if config.repair_backend == "antigravity" and catalog is None and catalog_diagnostic:
+            diagnostic = (
+                _sanitize_diagnostic(
+                    "agy model catalog unavailable; attempting configured candidates directly: "
+                    + catalog_diagnostic,
+                    config=config,
+                )
+                + (f"\n{diagnostic}" if diagnostic else "")
+            )
+
         attempt = RepairAttemptResult(
             backend=config.repair_backend,
             model=model,

src/coding_review_agent_loop/usage.py

@@ -92,7 +92,8 @@ class UsageCallRecord:
     role: Literal["repair"] | None = None
     model: str | None = None
     outcome: Literal[
-        "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output"
+        "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output",
+        "unavailable_model",
     ] | None = None
     log_path: str | None = None
     fallback_planned: bool | None = None
@@ -200,7 +201,8 @@ def add_record(
         role: Literal["repair"] | None = None,
         model: str | None = None,
         outcome: Literal[
-            "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output"
+            "succeeded", "nonzero_exit", "empty_output", "timeout", "spawn_error", "invalid_output",
+            "unavailable_model",
         ] | None = None,
         log_path: str | None = None,
         fallback_planned: bool | None = None,

tests/agent_loop_helpers.py

@@ -199,6 +199,7 @@ def __init__(
         codex_outputs=None,
         gemini_outputs=None,
         antigravity_outputs=None,
+        antigravity_catalog_outputs=None,
         issue_payload=None,
         issue_comments=None,
         pr_payload=None,
@@ -243,6 +244,7 @@ def __init__(
         self.codex_outputs = list(codex_outputs or [])
         self.gemini_outputs = list(gemini_outputs or [])
         self.antigravity_outputs = list(antigravity_outputs or [])
+        self.antigravity_catalog_outputs = list(antigravity_catalog_outputs or [])
         self.issue_payload = {
             "number": 56,
             "state": "open",
@@ -594,7 +596,9 @@ def _run_with_log_locked(self, args, *, cwd, log_path, check):
             return CommandResult(cmd, cwd_path, output, "", returncode)
 
         if cmd[:1] == ["agy"]:
-            output = self._next_agent_output(self.antigravity_outputs)
+            output = self._next_agent_output(
+                self.antigravity_catalog_outputs if "models" in cmd else self.antigravity_outputs
+            )
             if isinstance(output, dict):
                 stdout = output.get("stdout", "")
                 returncode = output.get("returncode", 0)

tests/test_repair.py

@@ -804,6 +804,63 @@ def test_execute_repair_uses_configured_legacy_gemini_override(tmp_path):
     assert attempts[0].backend == "gemini"
     assert "gemini-enterprise-flash" in run.call_args.args[0]
 
+
+def test_antigravity_repair_validates_once_and_falls_back_to_catalog_chain(tmp_path, monkeypatch):
+    from coding_review_agent_loop.agents import antigravity as agy_mod
+
+    monkeypatch.setattr(agy_mod, "_antigravity_settings_path", lambda: tmp_path / "settings.json")
+    valid = structured_pr_review(state="approved", reviewer="Google Antigravity")
+    calls = []
+
+    class CatalogRunner(FakeRunner):
+        def run_with_log(self, args, *, use_pty=False, **kwargs):
+            calls.append((list(args), use_pty))
+            return super().run_with_log(args, use_pty=use_pty, **kwargs)
+
+    runner = CatalogRunner(
+        antigravity_catalog_outputs=[("Available models:\nModelB\n", 0)],
+        antigravity_outputs=[(valid, 0)],
+    )
+    config = make_config(
+        tmp_path,
+        repair_models=("StaleModel", "ModelB"),
+        antigravity_models=("ModelB", "ModelC"),
+    )
+    repaired, _, attempts = execute_repair(
+        "malformed", runner=runner, config=config, run_id="catalog",
+        usage_context=None,
+        validate=lambda text: parse_structured_pr_review(text, reviewer="Google Antigravity"),
+        expected_kind="pr_review",
+    )
+
+    assert repaired == valid
+    assert [attempt.model for attempt in attempts] == ["StaleModel", "ModelB"]
+    assert attempts[0].outcome == "unavailable_model"
+    assert "Available choices: ModelB" in attempts[0].diagnostic
+    assert sum("models" in args for args, _ in calls) == 1
+    assert all(use_pty for _, use_pty in calls)
+
+
+def test_antigravity_repair_attempts_directly_when_catalog_fails(tmp_path, monkeypatch):
+    from coding_review_agent_loop.agents import antigravity as agy_mod
+
+    monkeypatch.setattr(agy_mod, "_antigravity_settings_path", lambda: tmp_path / "settings.json")
+    runner = FakeRunner(
+        antigravity_catalog_outputs=[("catalog unavailable", 2)],
+        antigravity_outputs=[("first failed", 1), ("second failed", 1)],
+    )
+    config = make_config(tmp_path, repair_models=("ModelA",), antigravity_models=("ModelB",))
+    _, _, attempts = execute_repair(
+        "malformed", runner=runner, config=config, run_id="catalog-failure",
+        usage_context=None, validate=lambda text: text, expected_kind="pr_review",
+    )
+
+    assert [attempt.model for attempt in attempts] == ["ModelA", "ModelB"]
+    assert all(attempt.outcome == "nonzero_exit" for attempt in attempts)
+    assert all("catalog unavailable" in attempt.diagnostic for attempt in attempts)
+    assert sum("models" in command for command, _ in runner.commands) == 1
+
+
 def test_runner_pty_timeout_is_opt_in_and_retains_combined_log(tmp_path):
     log_path = tmp_path / "logs" / "timeout.log"
     result = Runner().run_with_log(