From b58f84e8e1b65bcd4db711d67b296b3a966bca44 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:43:34 +0530 Subject: [PATCH 1/4] feat(chat): support Antigravity (Gemini 4 Argon) as Steward and Chat executor Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/antigravity_adapter.py | 689 ++++++++++++++++++ .../native_chat/model_settings.py | 31 +- .../steward_executor/machine_defaults.py | 2 + loopx/chat_endpoint_catalog.py | 28 + loopx/chat_endpoints.py | 1 + loopx/chat_manager.py | 2 + loopx/chat_runtime.py | 50 +- .../collaboration/conversation_binding.ts | 2 +- tests/test_antigravity_steward_adapter.py | 202 +++++ 9 files changed, 998 insertions(+), 9 deletions(-) create mode 100644 loopx/capabilities/native_chat/antigravity_adapter.py create mode 100644 tests/test_antigravity_steward_adapter.py diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py new file mode 100644 index 0000000000..0b4e22ad29 --- /dev/null +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -0,0 +1,689 @@ +"""Antigravity Language Server adapter for LoopX Native Chat and Steward.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import re +import shutil +import subprocess +import time +from typing import Any, Callable, Mapping +import urllib.error +import urllib.request +import uuid + +from ...chat import ( + VisibleResponseStreamFilter, + parse_agent_response, +) +from ...chat_agent import CodexChatAgentError, _turn_prompt + + +ANTIGRAVITY_DEFAULT_MODEL = "gemini-4-argon-high" +ANTIGRAVITY_DEFAULT_REASONING_EFFORT = "high" + +ANTIGRAVITY_MODEL_PLACEHOLDERS: dict[str, str] = { + "gemini-4-argon-high": "MODEL_PLACEHOLDER_M417", + "gemini-4-argon-medium": "MODEL_PLACEHOLDER_M418", + "gemini-4-argon-low": "MODEL_PLACEHOLDER_M419", +} + +ANTIGRAVITY_MODEL_OPTIONS: tuple[dict[str, Any], ...] = ( + { + "id": "gemini-4-argon-high", + "display_name": "Gemini 4 Argon (High)", + "description": "Antigravity Gemini 4 Argon with high reasoning effort", + "is_default": True, + "default_reasoning_effort": "high", + "supported_reasoning_efforts": [ + {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, + {"reasoning_effort": "medium", "description": "Balanced thinking"}, + {"reasoning_effort": "high", "description": "Deep reasoning"}, + ], + }, + { + "id": "gemini-4-argon-medium", + "display_name": "Gemini 4 Argon (Medium)", + "description": "Antigravity Gemini 4 Argon with medium reasoning effort", + "is_default": False, + "default_reasoning_effort": "medium", + "supported_reasoning_efforts": [ + {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, + {"reasoning_effort": "medium", "description": "Balanced thinking"}, + {"reasoning_effort": "high", "description": "Deep reasoning"}, + ], + }, + { + "id": "gemini-4-argon-low", + "display_name": "Gemini 4 Argon (Low)", + "description": "Antigravity Gemini 4 Argon with low reasoning effort", + "is_default": False, + "default_reasoning_effort": "low", + "supported_reasoning_efforts": [ + {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, + {"reasoning_effort": "medium", "description": "Balanced thinking"}, + {"reasoning_effort": "high", "description": "Deep reasoning"}, + ], + }, + { + "id": "gemini-4-argon", + "display_name": "Gemini 4 Argon", + "description": "Antigravity Gemini 4 Argon (selects tier from reasoning effort)", + "is_default": False, + "default_reasoning_effort": "high", + "supported_reasoning_efforts": [ + {"reasoning_effort": "low", "description": "Maps to Gemini 4 Argon (Low)"}, + {"reasoning_effort": "medium", "description": "Maps to Gemini 4 Argon (Medium)"}, + {"reasoning_effort": "high", "description": "Maps to Gemini 4 Argon (High)"}, + ], + }, +) + +_NO_PROXY_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) + + +def _rpc_post( + base_url: str, + csrf_token: str, + method: str, + payload: Mapping[str, Any], + *, + timeout: float = 15.0, +) -> dict[str, Any]: + url = f"{base_url.rstrip('/')}/exa.language_server_pb.LanguageServerService/{method}" + req = urllib.request.Request( + url, + data=json.dumps(dict(payload)).encode("utf-8"), + headers={ + "Content-Type": "application/json", + "Connect-Protocol-Version": "1", + "x-codeium-csrf-token": csrf_token, + }, + method="POST", + ) + try: + with _NO_PROXY_OPENER.open(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8", errors="replace") + data = json.loads(raw) if raw.strip() else {} + return data if isinstance(data, dict) else {} + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace")[:400] + raise CodexChatAgentError( + f"Antigravity LanguageServerService/{method} failed with HTTP {exc.code}: {body}", + gate=None, + error_code="antigravity_rpc_error", + ) from exc + except Exception as exc: + raise CodexChatAgentError( + f"Antigravity LanguageServerService/{method} transport failed: {exc}", + gate=None, + error_code="antigravity_transport_error", + ) from exc + + +def discover_antigravity_language_server( + env: Mapping[str, str] | None = None, +) -> dict[str, Any] | None: + """Discover a running local Antigravity language_server endpoint and CSRF token.""" + source_env = os.environ if env is None else env + explicit_url = str(source_env.get("ANTIGRAVITY_LS_URL") or "").strip() + explicit_token = str(source_env.get("ANTIGRAVITY_CSRF_TOKEN") or "").strip() + if explicit_url and explicit_token: + try: + _rpc_post( + explicit_url, + explicit_token, + "GetCascadeModelConfigData", + {}, + timeout=2.5, + ) + return { + "url": explicit_url.rstrip("/"), + "csrf_token": explicit_token, + "pid": None, + "source": "environment", + } + except CodexChatAgentError: + pass + + try: + ps_out = subprocess.check_output( + ["ps", "-ax", "-o", "pid=,command="], + text=True, + stderr=subprocess.DEVNULL, + timeout=3.0, + ) + except Exception: + return None + + candidates: list[tuple[int, str]] = [] + for line in ps_out.splitlines(): + if "language_server" not in line or "--csrf_token" not in line: + continue + match = re.match(r"^\s*(\d+)\s+(.*)$", line) + if not match: + continue + pid = int(match.group(1)) + cmd = match.group(2) + token_match = re.search(r"--csrf_token(?:=|\s+)([A-Za-z0-9._-]+)", cmd) + if token_match: + candidates.append((pid, token_match.group(1))) + + for pid, token in candidates: + try: + lsof_out = subprocess.check_output( + ["lsof", "-nP", "-a", "-p", str(pid), "-iTCP", "-sTCP:LISTEN"], + text=True, + stderr=subprocess.DEVNULL, + timeout=3.0, + ) + except Exception: + continue + ports: list[int] = [] + for line in lsof_out.splitlines(): + port_match = re.search(r":(\d+)\s+\(LISTEN\)", line) + if port_match: + ports.append(int(port_match.group(1))) + for port in ports: + url = f"http://127.0.0.1:{port}" + try: + _rpc_post(url, token, "GetCascadeModelConfigData", {}, timeout=1.5) + return { + "url": url, + "csrf_token": token, + "pid": pid, + "source": "process_discovery", + } + except CodexChatAgentError: + continue + return None + + +def probe_antigravity_endpoint( + env: Mapping[str, str] | None = None, +) -> tuple[bool, str, str]: + """Return `(available, command_display, detail)` for `/api/chat/capabilities`.""" + discovered = discover_antigravity_language_server(env=env) + if discovered is not None: + return ( + True, + "antigravity-ls", + f"Antigravity Language Server ({discovered['url']}, default model {ANTIGRAVITY_DEFAULT_MODEL})", + ) + app_ls = Path("/Applications/Antigravity.app/Contents/Resources/bin/language_server_macos_arm") + agy_bin = shutil.which("agy") + if app_ls.exists() or agy_bin: + return ( + False, + str(agy_bin or app_ls), + "Antigravity is installed, but no active local Antigravity language_server process was found", + ) + return ( + False, + "antigravity", + "Antigravity Language Server not found", + ) + + +def resolve_antigravity_model_placeholder( + model_id: str | None, + reasoning_effort: str | None = None, + *, + dynamic_configs: list[dict[str, Any]] | None = None, +) -> tuple[str, str]: + """Return `(canonical_model_id, model_placeholder)` for Antigravity.""" + raw_model = str(model_id or "").strip() + effort = str(reasoning_effort or "").strip().lower() + lowered = raw_model.lower() + + if not lowered or lowered in {"argon", "gemini-4-argon"}: + if effort == "low": + return "gemini-4-argon-low", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-low"] + if effort == "medium": + return "gemini-4-argon-medium", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-medium"] + return "gemini-4-argon-high", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-high"] + + if lowered in ANTIGRAVITY_MODEL_PLACEHOLDERS: + if lowered.startswith("gemini-4-argon-") and effort in {"low", "medium", "high"}: + candidate = f"gemini-4-argon-{effort}" + if candidate in ANTIGRAVITY_MODEL_PLACEHOLDERS and lowered == ANTIGRAVITY_DEFAULT_MODEL: + return candidate, ANTIGRAVITY_MODEL_PLACEHOLDERS[candidate] + return lowered, ANTIGRAVITY_MODEL_PLACEHOLDERS[lowered] + + if raw_model.startswith("MODEL_PLACEHOLDER_"): + for canonical, placeholder in ANTIGRAVITY_MODEL_PLACEHOLDERS.items(): + if placeholder == raw_model: + return canonical, placeholder + return raw_model, raw_model + + for cfg in dynamic_configs or []: + label = str(cfg.get("label") or "").strip() + placeholder = str((cfg.get("modelOrAlias") or {}).get("model") or "").strip() + if not placeholder: + continue + if label.lower() == lowered or placeholder.lower() == lowered: + return raw_model, placeholder + + return ANTIGRAVITY_DEFAULT_MODEL, ANTIGRAVITY_MODEL_PLACEHOLDERS[ANTIGRAVITY_DEFAULT_MODEL] + + +class AntigravityAdapter: + """Session adapter backed by Antigravity's local LanguageServerService.""" + + adapter_kind = "antigravity_ls" + + def __init__( + self, + *, + work_dir: Path, + objective: str, + cascade_id: str, + ls_url: str, + csrf_token: str, + model_id: str, + model_placeholder: str, + reasoning_effort: str | None = None, + runtime_profile: str = "restricted", + team_plan_context: dict[str, Any] | None = None, + initial_history: list[dict[str, str]] | None = None, + resumed: bool = False, + ) -> None: + self.work_dir = work_dir.resolve() + self.objective = objective + self.cascade_id = cascade_id + self.upstream_thread_id = cascade_id + self.ls_url = ls_url + self.csrf_token = csrf_token + self.model_id = model_id + self.model_placeholder = model_placeholder + self.reasoning_effort = reasoning_effort + self.runtime_profile = runtime_profile + self.team_plan_context = team_plan_context + self._pending_history: list[dict[str, str]] = ( + list(initial_history) if initial_history and not resumed else [] + ) + self._system_primed = resumed + self._closed = False + + @property + def current_model_id(self) -> str: + return self.model_id + + @property + def model_settings(self) -> dict[str, str | None]: + return { + "agent_reasoning_effort": self.reasoning_effort, + "agent_service_tier": None, + } + + @classmethod + def start( + cls, + *, + work_dir: Path, + objective: str, + resume_thread_id: str | None = None, + history: list[dict[str, str]] | None = None, + runtime_profile: str = "restricted", + team_plan_context: dict[str, Any] | None = None, + executor_model: Mapping[str, Any] | None = None, + env: Mapping[str, str] | None = None, + ) -> "AntigravityAdapter": + discovered = discover_antigravity_language_server(env=env) + if discovered is None: + raise CodexChatAgentError( + "Antigravity local language_server is not running or reachable", + gate=None, + error_code="antigravity_unavailable", + ) + ls_url = str(discovered["url"]) + csrf_token = str(discovered["csrf_token"]) + + dynamic_configs: list[dict[str, Any]] = [] + try: + model_data = _rpc_post(ls_url, csrf_token, "GetCascadeModelConfigData", {}, timeout=3.0) + raw_configs = model_data.get("clientModelConfigs") + if isinstance(raw_configs, list): + dynamic_configs = [item for item in raw_configs if isinstance(item, dict)] + except CodexChatAgentError: + pass + + requested_model = ( + str(executor_model.get("model") or "").strip() + if isinstance(executor_model, Mapping) + else None + ) + requested_effort = ( + str(executor_model.get("reasoning_effort") or "").strip() + if isinstance(executor_model, Mapping) + else None + ) + model_id, model_placeholder = resolve_antigravity_model_placeholder( + requested_model, + requested_effort, + dynamic_configs=dynamic_configs, + ) + + cascade_id = str(resume_thread_id or "").strip() + resumed = False + if cascade_id: + try: + traj = _rpc_post( + ls_url, + csrf_token, + "GetCascadeTrajectory", + {"cascadeId": cascade_id}, + timeout=3.0, + ) + if isinstance(traj, dict) and traj.get("trajectory"): + resumed = True + except CodexChatAgentError: + resumed = False + + if not resumed: + started = _rpc_post( + ls_url, + csrf_token, + "StartCascade", + { + "source": "CORTEX_TRAJECTORY_SOURCE_CASCADE_CLIENT", + "workspaceUris": [work_dir.resolve().as_uri()], + }, + timeout=10.0, + ) + cascade_id = str(started.get("cascadeId") or "").strip() + if not cascade_id: + raise CodexChatAgentError( + f"StartCascade did not return a cascadeId: {started}", + gate=None, + error_code="antigravity_start_failed", + ) + + return cls( + work_dir=work_dir, + objective=objective, + cascade_id=cascade_id, + ls_url=ls_url, + csrf_token=csrf_token, + model_id=model_id, + model_placeholder=model_placeholder, + reasoning_effort=requested_effort or ANTIGRAVITY_DEFAULT_REASONING_EFFORT, + runtime_profile=runtime_profile, + team_plan_context=team_plan_context, + initial_history=history, + resumed=resumed, + ) + + def _refresh_ls_connection_if_needed(self) -> None: + try: + _rpc_post(self.ls_url, self.csrf_token, "GetCascadeModelConfigData", {}, timeout=2.0) + return + except CodexChatAgentError: + discovered = discover_antigravity_language_server() + if discovered is not None: + self.ls_url = str(discovered["url"]) + self.csrf_token = str(discovered["csrf_token"]) + + def _build_turn_prompt(self, message: str) -> str: + parts: list[str] = [] + if not self._system_primed: + if self.objective.strip(): + parts.append(self.objective.strip()) + if self._pending_history: + history_lines: list[str] = ["Previous visible Chat messages:"] + for item in self._pending_history[-16:]: + role = str(item.get("role") or "user") + content = str(item.get("content") or "").strip() + if content: + history_lines.append(f"{role}: {content}") + if len(history_lines) > 1: + parts.append("\n".join(history_lines)) + self._pending_history.clear() + self._system_primed = True + parts.append( + _turn_prompt( + message, + runtime_profile=self.runtime_profile, + ) + ) + return "\n\n".join(parts) + + def _build_cascade_config(self) -> tuple[dict[str, Any], dict[str, Any]]: + user_config: dict[str, Any] = {} + user_agent_spec: dict[str, Any] = {} + try: + traj = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectory", + {"cascadeId": self.cascade_id}, + timeout=4.0, + ) + metadata = (traj.get("trajectory") or {}).get("generatorMetadata") or [] + if isinstance(metadata, list): + for entry in reversed(metadata): + if not isinstance(entry, dict): + continue + if not user_config and isinstance(entry.get("userConfig"), dict): + user_config = json.loads(json.dumps(entry["userConfig"])) + if not user_agent_spec and isinstance(entry.get("userAgentSpec"), dict): + user_agent_spec = json.loads(json.dumps(entry["userAgentSpec"])) + if user_config and user_agent_spec: + break + except CodexChatAgentError: + pass + + planner_cfg = user_config.setdefault("plannerConfig", {}) + if not isinstance(planner_cfg, dict): + planner_cfg = {} + user_config["plannerConfig"] = planner_cfg + planner_cfg["planModel"] = self.model_placeholder + planner_cfg["requestedModel"] = {"model": self.model_placeholder} + conversational = planner_cfg.setdefault("conversational", {}) + if isinstance(conversational, dict): + conversational.setdefault("plannerMode", "CONVERSATIONAL_PLANNER_MODE_DEFAULT") + conversational.setdefault("agenticMode", True) + return user_config, user_agent_spec + + def start_turn( + self, + message: str, + event_sink: Callable[[str, dict[str, Any]], None] | None = None, + ) -> dict[str, Any]: + if self._closed: + raise CodexChatAgentError( + "Antigravity session is closed", + gate=None, + error_code="session_closed", + ) + + self._refresh_ls_connection_if_needed() + turn_id = f"agy_turn_{uuid.uuid4().hex[:12]}" + if event_sink is not None: + event_sink( + "turn.started", + {"upstream_turn_id": turn_id, "threadId": self.cascade_id}, + ) + + before_traj = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectory", + {"cascadeId": self.cascade_id}, + timeout=5.0, + ) + start_step_index = int(before_traj.get("numTotalSteps") or 0) + user_config, user_agent_spec = self._build_cascade_config() + full_prompt = self._build_turn_prompt(message) + + send_payload: dict[str, Any] = { + "cascadeId": self.cascade_id, + "items": [{"text": full_prompt}], + "cascadeConfig": user_config, + } + if user_agent_spec: + send_payload["userAgentSpec"] = user_agent_spec + + _rpc_post( + self.ls_url, + self.csrf_token, + "SendUserCascadeMessage", + send_payload, + timeout=15.0, + ) + + stream_filter = VisibleResponseStreamFilter() + emitted_visible_delta = False + emitted_text_by_step: dict[int, str] = {} + announced_phases: set[tuple[int, str]] = set() + latest_response_text = "" + deadline = time.monotonic() + 300.0 + saw_running_or_new_step = False + + while time.monotonic() < deadline: + if self._closed: + raise CodexChatAgentError( + "Antigravity turn was interrupted", + gate=None, + error_code="turn_interrupted", + ) + time.sleep(0.25) + traj = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectory", + {"cascadeId": self.cascade_id}, + timeout=6.0, + ) + status = str(traj.get("status") or "") + total_steps = int(traj.get("numTotalSteps") or 0) + if status == "CASCADE_RUN_STATUS_RUNNING" or total_steps > start_step_index: + saw_running_or_new_step = True + + if total_steps > start_step_index: + steps_resp = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectorySteps", + { + "cascadeId": self.cascade_id, + "startIndex": start_step_index, + "endIndex": total_steps, + }, + timeout=8.0, + ) + steps = steps_resp.get("steps") or [] + if isinstance(steps, list): + for offset, step in enumerate(steps): + if not isinstance(step, dict): + continue + step_idx = start_step_index + offset + step_type = str(step.get("type") or "") + if step_type == "CORTEX_STEP_TYPE_PLANNER_RESPONSE": + pr = step.get("plannerResponse") or {} + if not isinstance(pr, dict): + continue + thinking = str(pr.get("thinking") or "").strip() + if thinking and (step_idx, "thinking") not in announced_phases: + announced_phases.add((step_idx, "thinking")) + if event_sink is not None: + event_sink( + "agent.phase", + { + "phase": "thinking", + "label": f"Gemini 4 Argon 思考中 ({self.model_id})", + }, + ) + tool_calls = pr.get("toolCalls") or [] + if isinstance(tool_calls, list) and tool_calls and (step_idx, "tool") not in announced_phases: + announced_phases.add((step_idx, "tool")) + first_tool = tool_calls[0] if isinstance(tool_calls[0], dict) else {} + tool_name = str(first_tool.get("name") or "tool") + if event_sink is not None: + event_sink( + "agent.phase", + { + "phase": "tool_call", + "label": f"正在调用 {tool_name}", + }, + ) + text = str(pr.get("modifiedResponse") or pr.get("response") or "") + if text: + prev_text = emitted_text_by_step.get(step_idx, "") + if text.startswith(prev_text): + delta_chunk = text[len(prev_text) :] + else: + delta_chunk = text + emitted_text_by_step[step_idx] = text + latest_response_text = text + if delta_chunk and event_sink is not None: + visible_delta = stream_filter.feed(delta_chunk) + if visible_delta: + emitted_visible_delta = True + event_sink("answer.delta", {"text": visible_delta}) + elif step_type == "CORTEX_STEP_TYPE_ERROR_MESSAGE": + err_obj = step.get("errorMessage") or {} + err_msg = str( + (err_obj.get("error") if isinstance(err_obj, dict) else None) + or err_obj + or "Antigravity cascade step error" + ) + raise CodexChatAgentError( + err_msg, + gate=None, + error_code="antigravity_step_error", + ) + + if saw_running_or_new_step and status != "CASCADE_RUN_STATUS_RUNNING" and latest_response_text.strip(): + break + + if event_sink is not None: + tail = stream_filter.finish() + if tail: + emitted_visible_delta = True + event_sink("answer.delta", {"text": tail}) + + if not latest_response_text.strip(): + raise CodexChatAgentError( + "Antigravity cascade finished without a planner response", + gate=None, + error_code="empty_agent_response", + ) + + parsed = parse_agent_response( + latest_response_text, + protected_paths=[self.work_dir], + team_plan_context=self.team_plan_context, + ) + if event_sink is not None: + if not emitted_visible_delta and str(parsed.get("message") or "").strip(): + event_sink("answer.delta", {"text": str(parsed["message"])}) + event_sink("answer.final", {"response": parsed}) + return parsed + + def interrupt_turn(self, turn_id: str | None = None) -> None: + del turn_id + try: + _rpc_post( + self.ls_url, + self.csrf_token, + "CancelCascadeInvocation", + {"cascadeId": self.cascade_id}, + timeout=5.0, + ) + except CodexChatAgentError: + pass + + def close_session(self) -> None: + self._closed = True + + def healthcheck(self) -> bool: + if self._closed: + return False + try: + _rpc_post(self.ls_url, self.csrf_token, "GetCascadeModelConfigData", {}, timeout=2.0) + return True + except CodexChatAgentError: + return False diff --git a/loopx/capabilities/native_chat/model_settings.py b/loopx/capabilities/native_chat/model_settings.py index 69e1a8238c..9e1bc4edcd 100644 --- a/loopx/capabilities/native_chat/model_settings.py +++ b/loopx/capabilities/native_chat/model_settings.py @@ -114,6 +114,32 @@ def inspect_agent_model_options(controller: ChatRuntimeController, agent_id: str ], }, } + if agent_id == "antigravity": + from .antigravity_adapter import ANTIGRAVITY_DEFAULT_MODEL, ANTIGRAVITY_MODEL_OPTIONS + return { + "agent_id": agent_id, + "supported": True, + "model": { + "id": "model", + "label": "Model", + "selected": ANTIGRAVITY_DEFAULT_MODEL, + "options": [ + { + "value": str(item["id"]), + "label": str(item["display_name"]), + "description": str(item.get("description") or ""), + "is_default": bool(item.get("is_default")), + "default_reasoning_effort": str(item.get("default_reasoning_effort") or "high"), + "reasoning_efforts": [ + str(effort["reasoning_effort"]) + for effort in (item.get("supported_reasoning_efforts") or []) + if isinstance(effort, dict) and effort.get("reasoning_effort") + ], + } + for item in ANTIGRAVITY_MODEL_OPTIONS + ], + }, + } if str(capability.get("adapter_kind")) != "acp": return {"agent_id": agent_id, "model": None, "supported": False} adapter = controller._start_adapter( @@ -174,10 +200,11 @@ def configure_session_model(controller: ChatRuntimeController, session_id: str, def observed_model_settings(adapter: ChatRuntimeAdapter, requested_effort: str | None = None, requested_model: str | None = None) -> dict[str, Any]: from ...chat_acp import ACPStdioAdapter + from .antigravity_adapter import AntigravityAdapter from .codex_adapter import CodexAppServerAdapter from ...chat_agent import CodexChatAgentError - settings = adapter.model_settings if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)) else {} - if requested_model and isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)) and adapter.current_model_id != requested_model: + settings = adapter.model_settings if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter, AntigravityAdapter)) else {} + if requested_model and isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter, AntigravityAdapter)) and adapter.current_model_id != requested_model: raise CodexChatAgentError("The Agent did not confirm the requested model; no message was sent.", error_code="unsupported_configuration", gate=None) if requested_effort and settings.get("agent_reasoning_effort") != requested_effort: diff --git a/loopx/capabilities/steward_executor/machine_defaults.py b/loopx/capabilities/steward_executor/machine_defaults.py index c12068071a..9f86a45323 100644 --- a/loopx/capabilities/steward_executor/machine_defaults.py +++ b/loopx/capabilities/steward_executor/machine_defaults.py @@ -47,6 +47,7 @@ # ``loopx.chat_manager`` re-exports every name unchanged. MANAGER_ENDPOINT_MANAGED = MANAGED_TURN_HOST MANAGER_ENDPOINT_INDIVIDUAL = "codex" +MANAGER_ENDPOINT_ANTIGRAVITY = "antigravity" # Executor kinds name where this channel's model work is billed and bounded # rather than which adapter is launched, and they use the same vocabulary as the # governed Turn surface: an individual executor runs on one person's own CLI @@ -57,6 +58,7 @@ MANAGER_EXECUTOR_KIND_MANAGED = EXECUTOR_KIND_MANAGED MANAGER_ENDPOINT_KINDS = { MANAGER_ENDPOINT_INDIVIDUAL: MANAGER_EXECUTOR_KIND_INDIVIDUAL, + MANAGER_ENDPOINT_ANTIGRAVITY: MANAGER_EXECUTOR_KIND_INDIVIDUAL, # The managed host is billed to the operator's own endpoint, not to one # person's CLI login, so it is reached only by selecting it. MANAGER_ENDPOINT_MANAGED: MANAGER_EXECUTOR_KIND_MANAGED, diff --git a/loopx/chat_endpoint_catalog.py b/loopx/chat_endpoint_catalog.py index 4d0331f324..4672febdf8 100644 --- a/loopx/chat_endpoint_catalog.py +++ b/loopx/chat_endpoint_catalog.py @@ -85,6 +85,22 @@ def _environment_readiness(variable: str) -> dict[str, Any]: "unavailable_reason": "api_key_not_set"} +def _antigravity_readiness() -> dict[str, Any]: + """Live local probe for Antigravity's LanguageServerService.""" + + from .capabilities.native_chat.antigravity_adapter import probe_antigravity_endpoint + + available, executable, detail = probe_antigravity_endpoint() + if available: + return {"available": True, "executable": executable, "detail": detail} + return { + "available": False, + "executable": executable, + "unavailable_reason": "executable_not_found", + "detail": detail, + } + + def builtin_chat_endpoints( *, codex_bin: str, @@ -113,6 +129,18 @@ def builtin_chat_endpoints( "trust_scope": "read_only", "source": "builtin", }, + { + "agent_id": "antigravity", + "display_name": "Antigravity (Gemini 4 Argon)", + "adapter_kind": "antigravity_ls", + **_antigravity_readiness(), + "streaming": True, + "resume": True, + "interrupt": True, + "tool_calls": True, + "trust_scope": "read_only", + "source": "builtin", + }, { "agent_id": "claude-code", "display_name": "Claude Code", diff --git a/loopx/chat_endpoints.py b/loopx/chat_endpoints.py index 88e755291c..ca6254f509 100644 --- a/loopx/chat_endpoints.py +++ b/loopx/chat_endpoints.py @@ -22,6 +22,7 @@ # already advertises. RESERVED_AGENT_IDS = { "codex", + "antigravity", "claude-code", "anthropic-api", "openai-api", diff --git a/loopx/chat_manager.py b/loopx/chat_manager.py index 7c56fe67e0..548a899377 100644 --- a/loopx/chat_manager.py +++ b/loopx/chat_manager.py @@ -869,6 +869,8 @@ def manager_model_resolution( if MANAGER_ENDPOINT_KINDS.get(resolved_endpoint) == MANAGER_EXECUTOR_KIND_MANAGED: profile = managed_execution_profile(environ) return str(profile["model"]), MANAGER_MODEL_SOURCE_MANAGED_PROFILE + if resolved_endpoint == "antigravity": + return "gemini-4-argon-high", MANAGER_MODEL_SOURCE_VENDOR_DEFAULT return MANAGER_MODEL_DEFAULT, MANAGER_MODEL_SOURCE_VENDOR_DEFAULT diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 4551c200a2..9f93518747 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -357,10 +357,10 @@ def _start_adapter( if ( manager_runtime is not None and manager_runtime.get("runtime_profile") == "trusted_owner" - and agent_id != "codex" + and agent_id not in {"codex", "antigravity"} ): raise CodexChatAgentError( - "The trusted owner manager profile requires the Codex endpoint.", + "The trusted owner manager profile requires the Codex or Antigravity endpoint.", error_code="manager_runtime_endpoint_unsupported", gate={ "kind": "host_tool_gate", @@ -369,10 +369,48 @@ def _start_adapter( "runtime profile." ), "next_action": ( - "Select the Codex Agent or change Manager runtime to restricted." + "Select the Codex or Antigravity Agent or change Manager runtime to restricted." ), }, ) + if agent_id == "antigravity": + from .capabilities.native_chat.antigravity_adapter import AntigravityAdapter + manager_profile = ( + dict(manager_runtime or self.manager_runtime_profile()) + if goal_id == MANAGER_AGENT_GOAL_ID + else None + ) + if manager_profile is not None: + objective = manager_agent_objective( + str(manager_profile["runtime_profile"]) + ) + model_config = ( + executor_model or manager_model_config( + endpoint=agent_id, + machine_defaults=self.steward_executor_defaults(), + ) + if goal_id == MANAGER_AGENT_GOAL_ID and not execution_mode + else executor_model or {} + ) + if agent_config and agent_config.get("model"): + model_config = {**model_config, "model": agent_config["model"]} + if agent_config and agent_config.get("reasoning_effort"): + model_config = {**model_config, "reasoning_effort": agent_config["reasoning_effort"]} + return AntigravityAdapter.start( + work_dir=work_dir, + objective=self._session_objective( + goal_id=goal_id, objective=objective, + project_coordination=project_coordination and not execution_mode, + ), + resume_thread_id=resume_thread_id, + history=history, + runtime_profile=( + str(manager_profile["runtime_profile"]) + if manager_profile is not None + else "restricted" + ), + executor_model=model_config, + ) if agent_id == "codex": from .capabilities.manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL from .chat_loopx_mode import TOOL as COLLABORATION_TOOL @@ -526,12 +564,12 @@ def open_session( if agent_model_id: agent_model_id = normalize_agent_model_id(agent_model_id) capability = next((item for item in self.capabilities() if item["agent_id"] == agent_id), None) - if capability and capability.get("available") and agent_model_id and agent_id not in {"codex", "claude-code"} and capability.get("adapter_kind") != "acp": + if capability and capability.get("available") and agent_model_id and agent_id not in {"codex", "antigravity", "claude-code"} and capability.get("adapter_kind") != "acp": raise CodexChatAgentError("This Agent does not offer model selection.", error_code="unsupported_configuration", gate=None) if agent_reasoning_effort is not None: agent_reasoning_effort = normalize_agent_reasoning_effort(agent_reasoning_effort) - if capability and capability.get("available") and agent_id != "codex" and capability.get("adapter_kind") != "acp": + if capability and capability.get("available") and agent_id not in {"codex", "antigravity"} and capability.get("adapter_kind") != "acp": raise CodexChatAgentError("This Agent does not offer reasoning choices.", error_code="unsupported_configuration", gate=None) if mode not in {"resume_latest", "new"}: @@ -693,7 +731,7 @@ def open_session( codex_home=str(codex_context.home_for_project(selected_codex_home, project_context)) if agent_id == "codex" else None, project_context=project_context, steward_context=steward_context, - agent_model_id=adapter.current_model_id if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)) else None, + agent_model_id=getattr(adapter, "current_model_id", None) if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)) or getattr(adapter, "adapter_kind", None) == "antigravity_ls" else None, requested_agent_model_id=agent_model_id, requested_agent_reasoning_effort=agent_reasoning_effort, codex_chat_configuration=codex_defaults or None, diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 845cd4fd8c..ab9f7760ce 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -156,7 +156,7 @@ function binding(value: unknown): JsonObject { throw new EffectRuntimeRequestError("ambiguous Agent target grant"); } } - if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || row.executor_endpoint_id !== "codex")) { + if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || !["codex", "antigravity"].includes(String(row.executor_endpoint_id)))) { throw new EffectRuntimeRequestError("managed identity requires a private native steward"); } return { diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py new file mode 100644 index 0000000000..83a4433b3a --- /dev/null +++ b/tests/test_antigravity_steward_adapter.py @@ -0,0 +1,202 @@ +"""Focused tests for Antigravity (Gemini 4 Argon) Steward & Chat adapter.""" + +from __future__ import annotations + +from http.server import BaseHTTPRequestHandler, HTTPServer +import json +from pathlib import Path +import threading +from typing import Any + +from loopx.capabilities.native_chat.antigravity_adapter import ( + ANTIGRAVITY_DEFAULT_MODEL, + AntigravityAdapter, + probe_antigravity_endpoint, + resolve_antigravity_model_placeholder, +) +from loopx.capabilities.steward_executor import ( + STEWARD_EXECUTOR_SCHEMA, + effective_steward_executor_defaults, + normalize_steward_executor_machine_defaults, + steward_executor_endpoints, +) +from loopx.chat_manager import ( + MANAGER_MODEL_SOURCE_MACHINE_CONFIGURATION, + MANAGER_MODEL_SOURCE_VENDOR_DEFAULT, + manager_channel_binding, + manager_model_resolution, +) + + +def test_antigravity_is_registered_in_steward_executor_vocabulary() -> None: + assert "antigravity" in steward_executor_endpoints() + + normalized = normalize_steward_executor_machine_defaults( + { + "schema_version": STEWARD_EXECUTOR_SCHEMA, + "selection_policy": "preferred", + "executor_endpoint": "antigravity", + "eligible_endpoints": [], + "executor_model": "gemini-4-argon-high", + "executor_reasoning_effort": "high", + } + ) + assert normalized["executor_endpoint"] == "antigravity" + assert normalized["executor_model"] == "gemini-4-argon-high" + + model, source = manager_model_resolution({}, endpoint="antigravity") + assert model == ANTIGRAVITY_DEFAULT_MODEL == "gemini-4-argon-high" + assert source == MANAGER_MODEL_SOURCE_VENDOR_DEFAULT + + binding = manager_channel_binding( + {}, + machine_defaults=effective_steward_executor_defaults( + { + "schema_version": "loopx_machine_configuration_v0", + "namespaces": {"steward_executor": normalized}, + } + ), + ) + assert binding["executor_endpoint"] == "antigravity" + assert binding["model"] == "gemini-4-argon-high" + assert binding["model_source"] == MANAGER_MODEL_SOURCE_MACHINE_CONFIGURATION + + +def test_antigravity_model_placeholder_resolution() -> None: + assert resolve_antigravity_model_placeholder("gemini-4-argon-high") == ( + "gemini-4-argon-high", + "MODEL_PLACEHOLDER_M417", + ) + assert resolve_antigravity_model_placeholder("gemini-4-argon-medium") == ( + "gemini-4-argon-medium", + "MODEL_PLACEHOLDER_M418", + ) + assert resolve_antigravity_model_placeholder("gemini-4-argon-low") == ( + "gemini-4-argon-low", + "MODEL_PLACEHOLDER_M419", + ) + assert resolve_antigravity_model_placeholder("argon", "medium") == ( + "gemini-4-argon-medium", + "MODEL_PLACEHOLDER_M418", + ) + assert resolve_antigravity_model_placeholder("gemini-4-argon", "low") == ( + "gemini-4-argon-low", + "MODEL_PLACEHOLDER_M419", + ) + + +def test_antigravity_adapter_executes_turn_and_streams_response(tmp_path: Path) -> None: + recorded_calls: list[tuple[str, dict[str, Any]]] = [] + csrf_token = "test-csrf-token-argon" + + class FakeLSHandler(BaseHTTPRequestHandler): + def log_message(self, format: str, *args: Any) -> None: + return + + def do_POST(self) -> None: + if self.headers.get("x-codeium-csrf-token") != csrf_token: + self.send_response(403) + self.end_headers() + return + length = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(length).decode("utf-8") or "{}") + method = self.path.rsplit("/", 1)[-1] + recorded_calls.append((method, body)) + + if method == "GetCascadeModelConfigData": + resp: dict[str, Any] = { + "clientModelConfigs": [ + { + "label": "Gemini 4 Argon (High)", + "modelOrAlias": {"model": "MODEL_PLACEHOLDER_M417"}, + } + ] + } + elif method == "StartCascade": + assert body.get("source") == "CORTEX_TRAJECTORY_SOURCE_CASCADE_CLIENT" + resp = {"cascadeId": "cascade-argon-001"} + elif method == "GetCascadeTrajectory": + sent = any(m == "SendUserCascadeMessage" for m, _ in recorded_calls) + resp = { + "trajectory": {"cascadeId": "cascade-argon-001", "trajectoryId": "traj-001"}, + "status": "CASCADE_RUN_STATUS_IDLE", + "numTotalSteps": 3 if sent else 1, + } + elif method == "SendUserCascadeMessage": + resp = {} + elif method == "GetCascadeTrajectorySteps": + resp = { + "steps": [ + { + "type": "CORTEX_STEP_TYPE_USER_INPUT", + "status": "CORTEX_STEP_STATUS_DONE", + }, + { + "type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", + "status": "CORTEX_STEP_STATUS_DONE", + "plannerResponse": { + "thinking": "Checking LoopX portfolio state.", + "response": ( + "" + '{"message": "Hello from Gemini 4 Argon Steward!", ' + '"proposals": [], "context_handoff": [], "gate": null}' + "" + ), + }, + }, + ] + } + else: + resp = {} + + payload = json.dumps(resp).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + server = HTTPServer(("127.0.0.1", 0), FakeLSHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + ls_url = f"http://127.0.0.1:{server.server_port}" + env = { + "ANTIGRAVITY_LS_URL": ls_url, + "ANTIGRAVITY_CSRF_TOKEN": csrf_token, + } + available, executable, detail = probe_antigravity_endpoint(env=env) + assert available is True + assert executable == "antigravity-ls" + assert ls_url in detail + + adapter = AntigravityAdapter.start( + work_dir=tmp_path, + objective="Act as LoopX Steward.", + executor_model={"model": "gemini-4-argon-high", "reasoning_effort": "high"}, + env=env, + ) + assert adapter.upstream_thread_id == "cascade-argon-001" + assert adapter.current_model_id == "gemini-4-argon-high" + assert adapter.model_placeholder == "MODEL_PLACEHOLDER_M417" + + events: list[tuple[str, dict[str, Any]]] = [] + result = adapter.start_turn( + "Summarize active goals", + lambda kind, data: events.append((kind, data)), + ) + assert result["message"] == "Hello from Gemini 4 Argon Steward!" + assert any(kind == "turn.started" for kind, _ in events) + assert any(kind == "agent.phase" and data.get("phase") == "thinking" for kind, data in events) + assert any(kind == "answer.delta" for kind, _ in events) + assert any(kind == "answer.final" for kind, _ in events) + + send_calls = [body for method, body in recorded_calls if method == "SendUserCascadeMessage"] + assert len(send_calls) == 1 + planner_cfg = send_calls[0]["cascadeConfig"]["plannerConfig"] + assert planner_cfg["planModel"] == "MODEL_PLACEHOLDER_M417" + assert planner_cfg["requestedModel"] == {"model": "MODEL_PLACEHOLDER_M417"} + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) From fd3019ec6ba808d64b07f2df43d1e7ef0df7d3e2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:18:04 +0530 Subject: [PATCH 2/4] fix(chat): qualify Argon cancellation, resume and tool permissions Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/manager-runtime-profile-v0.md | 15 +- .../rfcs/manager-runtime-profile-v0.zh-CN.md | 12 +- .../native_chat/antigravity_adapter.py | 401 +++++++++--------- .../native_chat/model_settings.py | 6 +- loopx/chat_agent.py | 3 + loopx/chat_endpoint_catalog.py | 2 +- loopx/chat_runtime.py | 8 +- .../collaboration/conversation_binding.ts | 2 +- tests/test_antigravity_steward_adapter.py | 312 ++++++++++---- 9 files changed, 466 insertions(+), 295 deletions(-) diff --git a/docs/architecture/rfcs/manager-runtime-profile-v0.md b/docs/architecture/rfcs/manager-runtime-profile-v0.md index 283453555f..0fc01bc157 100644 --- a/docs/architecture/rfcs/manager-runtime-profile-v0.md +++ b/docs/architecture/rfcs/manager-runtime-profile-v0.md @@ -43,8 +43,19 @@ These defaults do not change ordinary Goal Chat models or grant other Agents too still bound the work. Protected merge, release, deploy, delete, and payment operations retain their typed contracts. Provider permission, audience, and durable LoopX state ownership do not change. -- Codex is currently the only endpoint that enforces `trusted_owner`. Other endpoints fail with - an actionable typed error instead of pretending to provide the selected profile. +- Codex and the local Antigravity Language Server support `trusted_owner`. Antigravity uses + its signed-in host's existing tool permissions and native approvals; LoopX never automatically + accepts those approvals. A waiting interaction returns an actionable gate. Other endpoints + fail with an actionable typed error instead of pretending to provide the selected profile. +- Antigravity's `restricted` Chat is text-only: each invocation explicitly disables native tool + calling and uses the read-only planner mode rather than inheriting a previous invocation's + tool configuration. This is a provider tool restriction, not a Codex or operating-system + sandbox. Trusted-owner remains limited to the existing verified owner audience. Governed + Task execution, workspace filesystem grants and native managed steward identity preparation + retain their existing Codex owner; selecting Antigravity does not widen those contracts. + Argon tiers must be offered by the local account and match the requested reasoning effort. + A missing original cascade fails resume without silently starting a replacement. Stop waits + for the native cascade to become idle; timeout or partial output is not successful completion. - `trusted_owner` applies to the local private owner-manager and a native private steward binding freshly verified by the existing Core conversation owner. That owner checks the App, owner principal, private source, current resource grant and frozen Session identity. diff --git a/docs/architecture/rfcs/manager-runtime-profile-v0.zh-CN.md b/docs/architecture/rfcs/manager-runtime-profile-v0.zh-CN.md index 196e14f044..cacb419a3b 100644 --- a/docs/architecture/rfcs/manager-runtime-profile-v0.zh-CN.md +++ b/docs/architecture/rfcs/manager-runtime-profile-v0.zh-CN.md @@ -36,8 +36,16 @@ LoopX 管家最初只有受限的规划会话：它可以读取 LoopX 提供的 - `trusted_owner` 不是通用提权。用户请求与既有 standing grant 仍限定工作范围； merge、release、deploy、delete、payment 等受保护操作继续走各自的 typed contract； 外部 provider 权限、受众边界和 LoopX durable state owner 不被改写。 -- 当前只有 Codex endpoint 能执行 `trusted_owner`。选择其他 endpoint 时必须返回可恢复的 - typed error，不能把受限执行伪装成已开放。 +- Codex 与本地 Antigravity Language Server 支持 `trusted_owner`。Antigravity 沿用已登录 + 宿主的现有工具权限和原生审批；LoopX 不会自动接受这些审批。等待交互时返回可恢复的 + gate。选择其他 endpoint 时必须返回可恢复的 typed error，不能把受限执行伪装成已开放。 +- Antigravity 的 `restricted` Chat 仅处理文本：每次调用显式禁用原生工具调用并使用只读 + planner 模式，不继承上一回合的工具配置。这是 provider 工具限制，不是 Codex 或操作 + 系统 sandbox。强能力模式仍限于既有核验过的本人受众。受治理 Task 执行、工作区文件 + 系统授权和原生受管理管家身份准备仍由既有 Codex owner 承接；选择 Antigravity 不扩展 + 这些契约。Argon 档位必须由本地账号提供，并与请求的思考深度一致。原始 cascade 丢失 + 时续接失败，不静默创建替代会话。停止等待原生 cascade 回到 idle；超时或部分输出 + 不代表成功完成。 - `trusted_owner` 适用于本地私有 Owner 管家和由既有 Core conversation owner 重新核验的 原生本人管家私聊。Core 核验 App、Owner 身份、私聊来源、当前资源授权和冻结的 Session 身份；受众证明本身不授予工具，仍按实际机器策略执行，包括显式受限选择。其他外部 diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 0b4e22ad29..39ba6bdc18 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -2,16 +2,19 @@ from __future__ import annotations +import ipaddress import json import os from pathlib import Path import re import shutil import subprocess +import threading import time from typing import Any, Callable, Mapping import urllib.error import urllib.request +from urllib.parse import urlsplit import uuid from ...chat import ( @@ -30,58 +33,42 @@ "gemini-4-argon-low": "MODEL_PLACEHOLDER_M419", } -ANTIGRAVITY_MODEL_OPTIONS: tuple[dict[str, Any], ...] = ( +ANTIGRAVITY_MODEL_OPTIONS: tuple[dict[str, Any], ...] = tuple( { - "id": "gemini-4-argon-high", - "display_name": "Gemini 4 Argon (High)", - "description": "Antigravity Gemini 4 Argon with high reasoning effort", - "is_default": True, - "default_reasoning_effort": "high", - "supported_reasoning_efforts": [ - {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, - {"reasoning_effort": "medium", "description": "Balanced thinking"}, - {"reasoning_effort": "high", "description": "Deep reasoning"}, - ], - }, - { - "id": "gemini-4-argon-medium", - "display_name": "Gemini 4 Argon (Medium)", - "description": "Antigravity Gemini 4 Argon with medium reasoning effort", - "is_default": False, - "default_reasoning_effort": "medium", - "supported_reasoning_efforts": [ - {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, - {"reasoning_effort": "medium", "description": "Balanced thinking"}, - {"reasoning_effort": "high", "description": "Deep reasoning"}, - ], - }, - { - "id": "gemini-4-argon-low", - "display_name": "Gemini 4 Argon (Low)", - "description": "Antigravity Gemini 4 Argon with low reasoning effort", - "is_default": False, - "default_reasoning_effort": "low", - "supported_reasoning_efforts": [ - {"reasoning_effort": "low", "description": "Fast lower-latency thinking"}, - {"reasoning_effort": "medium", "description": "Balanced thinking"}, - {"reasoning_effort": "high", "description": "Deep reasoning"}, - ], - }, - { - "id": "gemini-4-argon", - "display_name": "Gemini 4 Argon", - "description": "Antigravity Gemini 4 Argon (selects tier from reasoning effort)", - "is_default": False, - "default_reasoning_effort": "high", - "supported_reasoning_efforts": [ - {"reasoning_effort": "low", "description": "Maps to Gemini 4 Argon (Low)"}, - {"reasoning_effort": "medium", "description": "Maps to Gemini 4 Argon (Medium)"}, - {"reasoning_effort": "high", "description": "Maps to Gemini 4 Argon (High)"}, - ], - }, + "id": f"gemini-4-argon-{effort}", + "display_name": f"Gemini 4 Argon ({effort.title()})", + "is_default": effort == "high", + "default_reasoning_effort": effort, + "supported_reasoning_efforts": [{"reasoning_effort": effort}], + } + for effort in ("high", "medium", "low") ) -_NO_PROXY_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) + +class _LocalRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req: Any, fp: Any, code: int, msg: str, + headers: Any, newurl: str) -> None: + # A local endpoint's CSRF token must never follow an HTTP redirect. + return None + + +_NO_PROXY_OPENER = urllib.request.build_opener( + urllib.request.ProxyHandler({}), _LocalRedirectHandler(), +) + + +def _local_url(value: str) -> str: + parsed = urlsplit(value) + host = parsed.hostname or "" + try: + loopback = host == "localhost" or ipaddress.ip_address(host).is_loopback + except ValueError: + loopback = False + if (parsed.scheme != "http" or not loopback or not parsed.port + or parsed.username or parsed.password or parsed.query or parsed.fragment + or parsed.path not in {"", "/"}): + raise ValueError("Antigravity requires an HTTP loopback endpoint with a port") + return value.rstrip("/") def _rpc_post( @@ -92,7 +79,7 @@ def _rpc_post( *, timeout: float = 15.0, ) -> dict[str, Any]: - url = f"{base_url.rstrip('/')}/exa.language_server_pb.LanguageServerService/{method}" + url = f"{_local_url(base_url)}/exa.language_server_pb.LanguageServerService/{method}" req = urllib.request.Request( url, data=json.dumps(dict(payload)).encode("utf-8"), @@ -109,15 +96,14 @@ def _rpc_post( data = json.loads(raw) if raw.strip() else {} return data if isinstance(data, dict) else {} except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8", errors="replace")[:400] raise CodexChatAgentError( - f"Antigravity LanguageServerService/{method} failed with HTTP {exc.code}: {body}", + f"Antigravity LanguageServerService/{method} failed with HTTP {exc.code}", gate=None, error_code="antigravity_rpc_error", ) from exc except Exception as exc: raise CodexChatAgentError( - f"Antigravity LanguageServerService/{method} transport failed: {exc}", + f"Antigravity LanguageServerService/{method} transport failed ({type(exc).__name__})", gate=None, error_code="antigravity_transport_error", ) from exc @@ -130,8 +116,11 @@ def discover_antigravity_language_server( source_env = os.environ if env is None else env explicit_url = str(source_env.get("ANTIGRAVITY_LS_URL") or "").strip() explicit_token = str(source_env.get("ANTIGRAVITY_CSRF_TOKEN") or "").strip() - if explicit_url and explicit_token: + if explicit_url or explicit_token: + if not explicit_url or not explicit_token: + return None try: + explicit_url = _local_url(explicit_url) _rpc_post( explicit_url, explicit_token, @@ -145,12 +134,12 @@ def discover_antigravity_language_server( "pid": None, "source": "environment", } - except CodexChatAgentError: - pass + except (CodexChatAgentError, ValueError): + return None try: ps_out = subprocess.check_output( - ["ps", "-ax", "-o", "pid=,command="], + ["ps", "-u", str(os.getuid()), "-o", "pid=,command="], text=True, stderr=subprocess.DEVNULL, timeout=3.0, @@ -160,7 +149,8 @@ def discover_antigravity_language_server( candidates: list[tuple[int, str]] = [] for line in ps_out.splitlines(): - if "language_server" not in line or "--csrf_token" not in line: + if ("language_server" not in line or "antigravity" not in line.lower() + or "--csrf_token" not in line): continue match = re.match(r"^\s*(\d+)\s+(.*)$", line) if not match: @@ -171,13 +161,16 @@ def discover_antigravity_language_server( if token_match: candidates.append((pid, token_match.group(1))) + deadline = time.monotonic() + 4.0 for pid, token in candidates: + if time.monotonic() >= deadline: + break try: lsof_out = subprocess.check_output( ["lsof", "-nP", "-a", "-p", str(pid), "-iTCP", "-sTCP:LISTEN"], text=True, stderr=subprocess.DEVNULL, - timeout=3.0, + timeout=max(0.1, deadline - time.monotonic()), ) except Exception: continue @@ -187,9 +180,12 @@ def discover_antigravity_language_server( if port_match: ports.append(int(port_match.group(1))) for port in ports: + remaining = deadline - time.monotonic() + if remaining <= 0: + break url = f"http://127.0.0.1:{port}" try: - _rpc_post(url, token, "GetCascadeModelConfigData", {}, timeout=1.5) + _rpc_post(url, token, "GetCascadeModelConfigData", {}, timeout=min(1.5, remaining)) return { "url": url, "csrf_token": token, @@ -212,7 +208,7 @@ def probe_antigravity_endpoint( "antigravity-ls", f"Antigravity Language Server ({discovered['url']}, default model {ANTIGRAVITY_DEFAULT_MODEL})", ) - app_ls = Path("/Applications/Antigravity.app/Contents/Resources/bin/language_server_macos_arm") + app_ls = Path("/Applications/Antigravity.app/Contents/Resources/bin/language_server") agy_bin = shutil.which("agy") if app_ls.exists() or agy_bin: return ( @@ -234,43 +230,38 @@ def resolve_antigravity_model_placeholder( dynamic_configs: list[dict[str, Any]] | None = None, ) -> tuple[str, str]: """Return `(canonical_model_id, model_placeholder)` for Antigravity.""" - raw_model = str(model_id or "").strip() + raw_model = str(model_id or "").strip().lower() effort = str(reasoning_effort or "").strip().lower() - lowered = raw_model.lower() - - if not lowered or lowered in {"argon", "gemini-4-argon"}: - if effort == "low": - return "gemini-4-argon-low", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-low"] - if effort == "medium": - return "gemini-4-argon-medium", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-medium"] - return "gemini-4-argon-high", ANTIGRAVITY_MODEL_PLACEHOLDERS["gemini-4-argon-high"] - - if lowered in ANTIGRAVITY_MODEL_PLACEHOLDERS: - if lowered.startswith("gemini-4-argon-") and effort in {"low", "medium", "high"}: - candidate = f"gemini-4-argon-{effort}" - if candidate in ANTIGRAVITY_MODEL_PLACEHOLDERS and lowered == ANTIGRAVITY_DEFAULT_MODEL: - return candidate, ANTIGRAVITY_MODEL_PLACEHOLDERS[candidate] - return lowered, ANTIGRAVITY_MODEL_PLACEHOLDERS[lowered] - - if raw_model.startswith("MODEL_PLACEHOLDER_"): - for canonical, placeholder in ANTIGRAVITY_MODEL_PLACEHOLDERS.items(): - if placeholder == raw_model: - return canonical, placeholder - return raw_model, raw_model - - for cfg in dynamic_configs or []: - label = str(cfg.get("label") or "").strip() - placeholder = str((cfg.get("modelOrAlias") or {}).get("model") or "").strip() - if not placeholder: - continue - if label.lower() == lowered or placeholder.lower() == lowered: - return raw_model, placeholder - - return ANTIGRAVITY_DEFAULT_MODEL, ANTIGRAVITY_MODEL_PLACEHOLDERS[ANTIGRAVITY_DEFAULT_MODEL] + if effort and effort not in {"low", "medium", "high"}: + raise CodexChatAgentError("Choose low, medium, or high for Antigravity.", + gate=None, error_code="unsupported_configuration") + if not raw_model or raw_model in {"argon", "gemini-4-argon"}: + canonical = f"gemini-4-argon-{effort or 'high'}" + else: + canonical = raw_model + if (canonical not in ANTIGRAVITY_MODEL_PLACEHOLDERS + or effort and canonical != f"gemini-4-argon-{effort}"): + raise CodexChatAgentError("Choose a supported Argon model and its matching reasoning effort.", + gate=None, error_code="unsupported_configuration") + placeholder = ANTIGRAVITY_MODEL_PLACEHOLDERS[canonical] + if dynamic_configs is not None: + match = next((row for row in dynamic_configs + if row.get("modelId") == canonical + or (row.get("modelOrAlias") or {}).get("model") == placeholder), None) + if match is None: + raise CodexChatAgentError("The local Antigravity account did not offer this Argon model.", + gate=None, error_code="unsupported_configuration") + placeholder = str((match.get("modelOrAlias") or {}).get("model") or placeholder) + return canonical, placeholder class AntigravityAdapter: - """Session adapter backed by Antigravity's local LanguageServerService.""" + """Local text Chat and trusted-owner Steward over LanguageServerService. + + Restricted Chat disables native tools. Trusted-owner uses Antigravity's + existing approvals; LoopX never accepts those approvals on the user's behalf. + Governed Tasks and workspace grants retain their existing executor owner. + """ adapter_kind = "antigravity_ls" @@ -289,6 +280,7 @@ def __init__( team_plan_context: dict[str, Any] | None = None, initial_history: list[dict[str, str]] | None = None, resumed: bool = False, + hard_timeout_sec: float = 300.0, ) -> None: self.work_dir = work_dir.resolve() self.objective = objective @@ -304,8 +296,13 @@ def __init__( self._pending_history: list[dict[str, str]] = ( list(initial_history) if initial_history and not resumed else [] ) - self._system_primed = resumed + self._system_primed = False self._closed = False + self.hard_timeout_sec = hard_timeout_sec + self._interrupted = threading.Event() + self._send_lock = threading.Lock() + self._active_turn_id: str | None = None + self._cancel_acknowledged = False @property def current_model_id(self) -> str: @@ -330,7 +327,13 @@ def start( team_plan_context: dict[str, Any] | None = None, executor_model: Mapping[str, Any] | None = None, env: Mapping[str, str] | None = None, + hard_timeout_sec: float = 300.0, ) -> "AntigravityAdapter": + if runtime_profile not in {"restricted", "trusted_owner"}: + raise ValueError("unsupported Antigravity Chat runtime profile") + # Validate user intent before creating an upstream cascade. + requested = dict(executor_model or {}) + resolve_antigravity_model_placeholder(requested.get("model"), requested.get("reasoning_effort")) discovered = discover_antigravity_language_server(env=env) if discovered is None: raise CodexChatAgentError( @@ -341,14 +344,9 @@ def start( ls_url = str(discovered["url"]) csrf_token = str(discovered["csrf_token"]) - dynamic_configs: list[dict[str, Any]] = [] - try: - model_data = _rpc_post(ls_url, csrf_token, "GetCascadeModelConfigData", {}, timeout=3.0) - raw_configs = model_data.get("clientModelConfigs") - if isinstance(raw_configs, list): - dynamic_configs = [item for item in raw_configs if isinstance(item, dict)] - except CodexChatAgentError: - pass + model_data = _rpc_post(ls_url, csrf_token, "GetCascadeModelConfigData", {}, timeout=3.0) + raw_configs = model_data.get("clientModelConfigs") + dynamic_configs = [item for item in raw_configs if isinstance(item, dict)] if isinstance(raw_configs, list) else [] requested_model = ( str(executor_model.get("model") or "").strip() @@ -382,6 +380,12 @@ def start( except CodexChatAgentError: resumed = False + if cascade_id and not resumed: + raise CodexChatAgentError( + "The original Antigravity conversation is unavailable. Restore it in Antigravity or start a new Chat explicitly.", + gate=None, error_code="resume_failed", + ) + if not resumed: started = _rpc_post( ls_url, @@ -409,23 +413,14 @@ def start( csrf_token=csrf_token, model_id=model_id, model_placeholder=model_placeholder, - reasoning_effort=requested_effort or ANTIGRAVITY_DEFAULT_REASONING_EFFORT, + reasoning_effort=model_id.rsplit("-", 1)[-1], runtime_profile=runtime_profile, team_plan_context=team_plan_context, initial_history=history, resumed=resumed, + hard_timeout_sec=hard_timeout_sec, ) - def _refresh_ls_connection_if_needed(self) -> None: - try: - _rpc_post(self.ls_url, self.csrf_token, "GetCascadeModelConfigData", {}, timeout=2.0) - return - except CodexChatAgentError: - discovered = discover_antigravity_language_server() - if discovered is not None: - self.ls_url = str(discovered["url"]) - self.csrf_token = str(discovered["csrf_token"]) - def _build_turn_prompt(self, message: str) -> str: parts: list[str] = [] if not self._system_primed: @@ -440,8 +435,6 @@ def _build_turn_prompt(self, message: str) -> str: history_lines.append(f"{role}: {content}") if len(history_lines) > 1: parts.append("\n".join(history_lines)) - self._pending_history.clear() - self._system_primed = True parts.append( _turn_prompt( message, @@ -450,44 +443,42 @@ def _build_turn_prompt(self, message: str) -> str: ) return "\n\n".join(parts) - def _build_cascade_config(self) -> tuple[dict[str, Any], dict[str, Any]]: - user_config: dict[str, Any] = {} - user_agent_spec: dict[str, Any] = {} + def _build_cascade_config(self) -> dict[str, Any]: + # Never inherit an old cascade's tool permissions or userAgentSpec. + # Restricted Chat is text-only; trusted_owner retains native host approval. + restricted = self.runtime_profile == "restricted" + return {"plannerConfig": { + "planModel": self.model_placeholder, + "requestedModel": {"model": self.model_placeholder}, + "conversational": { + "plannerMode": ("CONVERSATIONAL_PLANNER_MODE_READ_ONLY" if restricted + else "CONVERSATIONAL_PLANNER_MODE_DEFAULT"), + "agenticMode": not restricted, + }, + **({"toolConfig": {"forceDisableToolCalling": True}, + "customizationDiscoveryConfig": {"hooks": {"inheritUser": False}}} + if restricted else {}), + }} + + def capabilities(self) -> dict[str, Any]: + return {"adapter_kind": self.adapter_kind, "streaming": True, + "resume": True, "interrupt": True, "steering": False} + + def start_turn(self, message: str, + event_sink: Callable[[str, dict[str, Any]], None] | None = None) -> dict[str, Any]: + self._interrupted.clear() + self._cancel_acknowledged = False try: - traj = _rpc_post( - self.ls_url, - self.csrf_token, - "GetCascadeTrajectory", - {"cascadeId": self.cascade_id}, - timeout=4.0, - ) - metadata = (traj.get("trajectory") or {}).get("generatorMetadata") or [] - if isinstance(metadata, list): - for entry in reversed(metadata): - if not isinstance(entry, dict): - continue - if not user_config and isinstance(entry.get("userConfig"), dict): - user_config = json.loads(json.dumps(entry["userConfig"])) - if not user_agent_spec and isinstance(entry.get("userAgentSpec"), dict): - user_agent_spec = json.loads(json.dumps(entry["userAgentSpec"])) - if user_config and user_agent_spec: - break - except CodexChatAgentError: - pass - - planner_cfg = user_config.setdefault("plannerConfig", {}) - if not isinstance(planner_cfg, dict): - planner_cfg = {} - user_config["plannerConfig"] = planner_cfg - planner_cfg["planModel"] = self.model_placeholder - planner_cfg["requestedModel"] = {"model": self.model_placeholder} - conversational = planner_cfg.setdefault("conversational", {}) - if isinstance(conversational, dict): - conversational.setdefault("plannerMode", "CONVERSATIONAL_PLANNER_MODE_DEFAULT") - conversational.setdefault("agenticMode", True) - return user_config, user_agent_spec - - def start_turn( + return self._run_turn(message, event_sink) + except BaseException: + # A timeout/transport or permission failure is not a successful turn. + # Retain an unconfirmed cancellation as an error for the caller fence. + self.interrupt_turn() + raise + finally: + self._active_turn_id = None + + def _run_turn( self, message: str, event_sink: Callable[[str, dict[str, Any]], None] | None = None, @@ -499,14 +490,7 @@ def start_turn( error_code="session_closed", ) - self._refresh_ls_connection_if_needed() turn_id = f"agy_turn_{uuid.uuid4().hex[:12]}" - if event_sink is not None: - event_sink( - "turn.started", - {"upstream_turn_id": turn_id, "threadId": self.cascade_id}, - ) - before_traj = _rpc_post( self.ls_url, self.csrf_token, @@ -514,8 +498,13 @@ def start_turn( {"cascadeId": self.cascade_id}, timeout=5.0, ) + if before_traj.get("status") != "CASCADE_RUN_STATUS_IDLE": + raise CodexChatAgentError( + "The original Antigravity conversation is still active. Stop it in Antigravity before retrying.", + gate=None, error_code="resume_failed", + ) start_step_index = int(before_traj.get("numTotalSteps") or 0) - user_config, user_agent_spec = self._build_cascade_config() + user_config = self._build_cascade_config() full_prompt = self._build_turn_prompt(message) send_payload: dict[str, Any] = { @@ -523,39 +512,33 @@ def start_turn( "items": [{"text": full_prompt}], "cascadeConfig": user_config, } - if user_agent_spec: - send_payload["userAgentSpec"] = user_agent_spec - - _rpc_post( - self.ls_url, - self.csrf_token, - "SendUserCascadeMessage", - send_payload, - timeout=15.0, - ) + with self._send_lock: + self._check_interrupted() + self._active_turn_id = turn_id + if event_sink is not None: + event_sink("turn.started", {"upstream_turn_id": turn_id, "threadId": self.cascade_id}) + _rpc_post(self.ls_url, self.csrf_token, "SendUserCascadeMessage", send_payload, timeout=15.0) + self._system_primed = True + self._pending_history.clear() stream_filter = VisibleResponseStreamFilter() emitted_visible_delta = False emitted_text_by_step: dict[int, str] = {} announced_phases: set[tuple[int, str]] = set() latest_response_text = "" - deadline = time.monotonic() + 300.0 + deadline = time.monotonic() + self.hard_timeout_sec saw_running_or_new_step = False while time.monotonic() < deadline: - if self._closed: - raise CodexChatAgentError( - "Antigravity turn was interrupted", - gate=None, - error_code="turn_interrupted", - ) - time.sleep(0.25) + self._check_interrupted() + self._interrupted.wait(0.25) + self._check_interrupted() traj = _rpc_post( self.ls_url, self.csrf_token, "GetCascadeTrajectory", {"cascadeId": self.cascade_id}, - timeout=6.0, + timeout=2.0, ) status = str(traj.get("status") or "") total_steps = int(traj.get("numTotalSteps") or 0) @@ -563,6 +546,7 @@ def start_turn( saw_running_or_new_step = True if total_steps > start_step_index: + self._check_interrupted() steps_resp = _rpc_post( self.ls_url, self.csrf_token, @@ -572,7 +556,7 @@ def start_turn( "startIndex": start_step_index, "endIndex": total_steps, }, - timeout=8.0, + timeout=2.0, ) steps = steps_resp.get("steps") or [] if isinstance(steps, list): @@ -581,6 +565,14 @@ def start_turn( continue step_idx = start_step_index + offset step_type = str(step.get("type") or "") + step_status = str(step.get("status") or "") + if step_status == "CORTEX_STEP_STATUS_WAITING": + raise CodexChatAgentError( + "Antigravity needs your response in its application; resolve it there and retry.", + gate={"kind": "host_tool_gate", "summary": "Antigravity is waiting for input.", + "next_action": "Open Antigravity and resolve the pending interaction."}, + error_code="host_tool_gate", + ) if step_type == "CORTEX_STEP_TYPE_PLANNER_RESPONSE": pr = step.get("plannerResponse") or {} if not isinstance(pr, dict): @@ -597,6 +589,9 @@ def start_turn( }, ) tool_calls = pr.get("toolCalls") or [] + if tool_calls and self.runtime_profile == "restricted": + raise CodexChatAgentError("Antigravity did not honor text-only Chat.", + gate=None, error_code="unsupported_configuration") if isinstance(tool_calls, list) and tool_calls and (step_idx, "tool") not in announced_phases: announced_phases.add((step_idx, "tool")) first_tool = tool_calls[0] if isinstance(tool_calls[0], dict) else {} @@ -636,8 +631,12 @@ def start_turn( error_code="antigravity_step_error", ) - if saw_running_or_new_step and status != "CASCADE_RUN_STATUS_RUNNING" and latest_response_text.strip(): + self._check_interrupted() + if saw_running_or_new_step and status == "CASCADE_RUN_STATUS_IDLE": break + else: + raise CodexChatAgentError("Antigravity did not finish before the turn deadline.", + gate=None, error_code="response_timeout") if event_sink is not None: tail = stream_filter.finish() @@ -663,21 +662,37 @@ def start_turn( event_sink("answer.final", {"response": parsed}) return parsed + def _check_interrupted(self) -> None: + if self._closed or self._interrupted.is_set(): + raise CodexChatAgentError("Antigravity turn was interrupted.", + gate=None, error_code="turn_interrupted") + def interrupt_turn(self, turn_id: str | None = None) -> None: - del turn_id - try: - _rpc_post( - self.ls_url, - self.csrf_token, - "CancelCascadeInvocation", - {"cascadeId": self.cascade_id}, - timeout=5.0, - ) - except CodexChatAgentError: - pass + active = self._active_turn_id + if active is not None and turn_id is not None and turn_id != active: + raise ValueError("the active Antigravity Turn changed") + self._interrupted.set() + if active is None: + return + with self._send_lock: + if not self._cancel_acknowledged: + _rpc_post(self.ls_url, self.csrf_token, "CancelCascadeInvocation", + {"cascadeId": self.cascade_id}, timeout=5.0) + deadline = time.monotonic() + 5.0 + while True: + trajectory = _rpc_post(self.ls_url, self.csrf_token, "GetCascadeTrajectory", + {"cascadeId": self.cascade_id}, timeout=2.0) + if trajectory.get("status") == "CASCADE_RUN_STATUS_IDLE": + break + if time.monotonic() >= deadline: + raise CodexChatAgentError("Antigravity has not confirmed stop; retry stopping it.", + gate=None, error_code="response_timeout") + time.sleep(0.05) + self._cancel_acknowledged = True def close_session(self) -> None: self._closed = True + self.interrupt_turn() def healthcheck(self) -> bool: if self._closed: diff --git a/loopx/capabilities/native_chat/model_settings.py b/loopx/capabilities/native_chat/model_settings.py index 9e1bc4edcd..fc6732a003 100644 --- a/loopx/capabilities/native_chat/model_settings.py +++ b/loopx/capabilities/native_chat/model_settings.py @@ -204,7 +204,11 @@ def observed_model_settings(adapter: ChatRuntimeAdapter, requested_effort: str | from .codex_adapter import CodexAppServerAdapter from ...chat_agent import CodexChatAgentError settings = adapter.model_settings if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter, AntigravityAdapter)) else {} - if requested_model and isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter, AntigravityAdapter)) and adapter.current_model_id != requested_model: + confirmed_model = requested_model + if isinstance(adapter, AntigravityAdapter) and requested_model: + from .antigravity_adapter import resolve_antigravity_model_placeholder + confirmed_model, _ = resolve_antigravity_model_placeholder(requested_model, requested_effort) + if confirmed_model and isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter, AntigravityAdapter)) and adapter.current_model_id != confirmed_model: raise CodexChatAgentError("The Agent did not confirm the requested model; no message was sent.", error_code="unsupported_configuration", gate=None) if requested_effort and settings.get("agent_reasoning_effort") != requested_effort: diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index d86fc37e4d..1d72040990 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -58,6 +58,9 @@ def _host_tool_gate(summary: str, next_action: str) -> dict[str, str]: # endpoint that is listed as a capability refused the session. AGENT_ENDPOINT_UNAVAILABLE = "agent_endpoint_unavailable" AGENT_ENDPOINT_NEXT_ACTIONS = { + "antigravity_unavailable": ( + "Open Antigravity and sign in to its local Language Server, then retry this session." + ), "dsh_runtime_unavailable": ( "The LoopX Chat service interpreter cannot import deepseek_harness. " "Run `loopx doctor` in that service environment and check python.executable; " diff --git a/loopx/chat_endpoint_catalog.py b/loopx/chat_endpoint_catalog.py index 4672febdf8..ef52c1fd03 100644 --- a/loopx/chat_endpoint_catalog.py +++ b/loopx/chat_endpoint_catalog.py @@ -96,7 +96,7 @@ def _antigravity_readiness() -> dict[str, Any]: return { "available": False, "executable": executable, - "unavailable_reason": "executable_not_found", + "unavailable_reason": "antigravity_unavailable", "detail": detail, } diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 9f93518747..aeeae5698b 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -375,6 +375,11 @@ def _start_adapter( ) if agent_id == "antigravity": from .capabilities.native_chat.antigravity_adapter import AntigravityAdapter + if execution_mode or project_context is not None or read_only_review: + raise CodexChatAgentError( + "Antigravity supports Chat and Steward; use Codex for governed workspace execution.", + gate=None, error_code="unsupported_configuration", + ) manager_profile = ( dict(manager_runtime or self.manager_runtime_profile()) if goal_id == MANAGER_AGENT_GOAL_ID @@ -410,6 +415,7 @@ def _start_adapter( else "restricted" ), executor_model=model_config, + hard_timeout_sec=self.hard_timeout_sec or 300.0, ) if agent_id == "codex": from .capabilities.manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL @@ -1042,7 +1048,7 @@ def _ensure_adapter_locked( self.loopx_mode.commit_recovery(session_id, native_recovery) with self.lock: self.adapters[session_id] = adapter - if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)): + if isinstance(adapter, (ACPStdioAdapter, CodexAppServerAdapter)) or getattr(adapter, "adapter_kind", None) == "antigravity_ls": self.store.update_session(session_id, agent_model_id=adapter.current_model_id, **observed_settings) if session.get("agent_id") == "codex" and session.get("codex_home") is None: diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index ab9f7760ce..845cd4fd8c 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -156,7 +156,7 @@ function binding(value: unknown): JsonObject { throw new EffectRuntimeRequestError("ambiguous Agent target grant"); } } - if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || !["codex", "antigravity"].includes(String(row.executor_endpoint_id)))) { + if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || row.executor_endpoint_id !== "codex")) { throw new EffectRuntimeRequestError("managed identity requires a private native steward"); } return { diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index 83a4433b3a..19d4c0076c 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -2,10 +2,11 @@ from __future__ import annotations -from http.server import BaseHTTPRequestHandler, HTTPServer +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json -from pathlib import Path import threading + +import pytest from typing import Any from loopx.capabilities.native_chat.antigravity_adapter import ( @@ -85,118 +86,241 @@ def test_antigravity_model_placeholder_resolution() -> None: ) -def test_antigravity_adapter_executes_turn_and_streams_response(tmp_path: Path) -> None: - recorded_calls: list[tuple[str, dict[str, Any]]] = [] - csrf_token = "test-csrf-token-argon" +@pytest.fixture +def language_server(monkeypatch): + """Actual HTTP boundary; only the external model/provider is synthetic.""" + state: dict[str, Any] = {"calls": [], "sent_count": 0, "mode": "complete", "active": False, + "sent": threading.Event(), "cancelled": threading.Event()} + token = "synthetic-csrf-token" - class FakeLSHandler(BaseHTTPRequestHandler): - def log_message(self, format: str, *args: Any) -> None: - return + class Handler(BaseHTTPRequestHandler): + def log_message(self, format, *args): + pass - def do_POST(self) -> None: - if self.headers.get("x-codeium-csrf-token") != csrf_token: - self.send_response(403) + def do_POST(self): + assert self.headers.get("x-codeium-csrf-token") == token + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + method = self.path.rsplit("/", 1)[-1] + state["calls"].append((method, body)) + if state.get("redirect"): + self.send_response(302) + self.send_header("Location", state["redirect"]) self.end_headers() return - length = int(self.headers.get("Content-Length") or 0) - body = json.loads(self.rfile.read(length).decode("utf-8") or "{}") - method = self.path.rsplit("/", 1)[-1] - recorded_calls.append((method, body)) - if method == "GetCascadeModelConfigData": - resp: dict[str, Any] = { - "clientModelConfigs": [ - { - "label": "Gemini 4 Argon (High)", - "modelOrAlias": {"model": "MODEL_PLACEHOLDER_M417"}, - } - ] - } + response = {"clientModelConfigs": [ + {"modelId": f"gemini-4-argon-{tier}", + "modelOrAlias": {"model": f"MODEL_PLACEHOLDER_M{number}"}} + for tier, number in (("high", 417), ("medium", 418), ("low", 419))]} elif method == "StartCascade": - assert body.get("source") == "CORTEX_TRAJECTORY_SOURCE_CASCADE_CLIENT" - resp = {"cascadeId": "cascade-argon-001"} + response = {"cascadeId": "synthetic-cascade"} elif method == "GetCascadeTrajectory": - sent = any(m == "SendUserCascadeMessage" for m, _ in recorded_calls) - resp = { - "trajectory": {"cascadeId": "cascade-argon-001", "trajectoryId": "traj-001"}, - "status": "CASCADE_RUN_STATUS_IDLE", - "numTotalSteps": 3 if sent else 1, - } + response = {"trajectory": {"cascadeId": "synthetic-cascade"}, + "numTotalSteps": state["sent_count"] * 2, + "status": ("CASCADE_RUN_STATUS_RUNNING" if state["mode"] in {"running", "waiting"} + and state["active"] else "CASCADE_RUN_STATUS_IDLE")} + if state.get("missing_resume"): + response.pop("trajectory") elif method == "SendUserCascadeMessage": - resp = {} + if state.get("send_failure"): + self.send_response(503) + self.end_headers() + return + state["sent_count"] += 1 + state["active"] = True + state["cancelled"].clear() + state["sent"].set() + response = {} elif method == "GetCascadeTrajectorySteps": - resp = { - "steps": [ - { - "type": "CORTEX_STEP_TYPE_USER_INPUT", - "status": "CORTEX_STEP_STATUS_DONE", - }, - { - "type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", - "status": "CORTEX_STEP_STATUS_DONE", - "plannerResponse": { - "thinking": "Checking LoopX portfolio state.", - "response": ( - "" - '{"message": "Hello from Gemini 4 Argon Steward!", ' - '"proposals": [], "context_handoff": [], "gate": null}' - "" - ), - }, - }, - ] - } + response = {"steps": []} + for index in range(body["startIndex"] // 2, state["sent_count"]): + response["steps"].extend([ + {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE"}, + {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", + "status": "CORTEX_STEP_STATUS_WAITING" if state["mode"] == "waiting" else "CORTEX_STEP_STATUS_DONE", + "plannerResponse": {"thinking": "synthetic status", + "response": '' + json.dumps({"message": f"Argon answer {index + 1}", + "proposals": [], "context_handoff": [], "gate": None}) + ''}}]) + elif method == "CancelCascadeInvocation": + state["active"] = False + state["cancelled"].set() + response = {} else: - resp = {} - - payload = json.dumps(resp).encode("utf-8") + raise AssertionError(method) + payload = json.dumps(response).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) - server = HTTPServer(("127.0.0.1", 0), FakeLSHandler) + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() + state["url"] = f"http://127.0.0.1:{server.server_port}" + monkeypatch.setenv("ANTIGRAVITY_LS_URL", state["url"]) + monkeypatch.setenv("ANTIGRAVITY_CSRF_TOKEN", token) try: - ls_url = f"http://127.0.0.1:{server.server_port}" - env = { - "ANTIGRAVITY_LS_URL": ls_url, - "ANTIGRAVITY_CSRF_TOKEN": csrf_token, - } - available, executable, detail = probe_antigravity_endpoint(env=env) - assert available is True - assert executable == "antigravity-ls" - assert ls_url in detail - - adapter = AntigravityAdapter.start( - work_dir=tmp_path, - objective="Act as LoopX Steward.", - executor_model={"model": "gemini-4-argon-high", "reasoning_effort": "high"}, - env=env, - ) - assert adapter.upstream_thread_id == "cascade-argon-001" - assert adapter.current_model_id == "gemini-4-argon-high" - assert adapter.model_placeholder == "MODEL_PLACEHOLDER_M417" - - events: list[tuple[str, dict[str, Any]]] = [] - result = adapter.start_turn( - "Summarize active goals", - lambda kind, data: events.append((kind, data)), - ) - assert result["message"] == "Hello from Gemini 4 Argon Steward!" - assert any(kind == "turn.started" for kind, _ in events) - assert any(kind == "agent.phase" and data.get("phase") == "thinking" for kind, data in events) - assert any(kind == "answer.delta" for kind, _ in events) - assert any(kind == "answer.final" for kind, _ in events) - - send_calls = [body for method, body in recorded_calls if method == "SendUserCascadeMessage"] - assert len(send_calls) == 1 - planner_cfg = send_calls[0]["cascadeConfig"]["plannerConfig"] - assert planner_cfg["planModel"] == "MODEL_PLACEHOLDER_M417" - assert planner_cfg["requestedModel"] == {"model": "MODEL_PLACEHOLDER_M417"} + yield state finally: server.shutdown() server.server_close() thread.join(timeout=2) + + +def adapter_for(tmp_path, **kwargs): + return AntigravityAdapter.start(work_dir=tmp_path, objective="Act as LoopX Steward.", **kwargs) + + +def test_antigravity_adapter_executes_repeated_turns_and_streams_response(tmp_path, language_server): + available, executable, _ = probe_antigravity_endpoint() + assert available and executable == "antigravity-ls" + adapter = adapter_for(tmp_path) + events = [] + for index in range(2): + result = adapter.start_turn("Summarize synthetic goals", lambda kind, data: events.append((kind, data))) + assert result["message"] == f"Argon answer {index + 1}" + assert adapter.current_model_id == "gemini-4-argon-high" + assert {kind for kind, _ in events} >= {"turn.started", "agent.phase", "answer.delta", "answer.final"} + calls = language_server["calls"] + assert len([1 for method, _ in calls if method == "StartCascade"]) == 1 + payload = next(body for method, body in calls if method == "SendUserCascadeMessage") + config = payload["cascadeConfig"]["plannerConfig"] + assert config["toolConfig"]["forceDisableToolCalling"] is True + assert config["conversational"]["agenticMode"] is False + assert "userAgentSpec" not in payload + + +@pytest.mark.parametrize("model,effort", [("unknown", None), ("gemini-4-argon-low", "high"), ("argon", "xhigh")]) +def test_unsupported_model_intent_has_no_upstream_effect(tmp_path, language_server, model, effort): + from loopx.chat_agent import CodexChatAgentError + with pytest.raises(CodexChatAgentError) as error: + adapter_for(tmp_path, executor_model={"model": model, "reasoning_effort": effort}) + assert error.value.error_code == "unsupported_configuration" + assert language_server["calls"] == [] + + +def test_canonical_model_and_effort_readback_agree(tmp_path, language_server): + from loopx.capabilities.native_chat.model_settings import observed_model_settings + adapter = adapter_for(tmp_path, executor_model={"model": "argon", "reasoning_effort": "low"}) + assert adapter.current_model_id == "gemini-4-argon-low" + assert observed_model_settings(adapter, "low", "argon")["agent_reasoning_effort"] == "low" + + +def test_failed_resume_does_not_create_a_hidden_new_conversation(tmp_path, language_server): + from loopx.chat_agent import CodexChatAgentError + language_server["missing_resume"] = True + with pytest.raises(CodexChatAgentError) as error: + adapter_for(tmp_path, resume_thread_id="missing-cascade") + assert error.value.error_code == "resume_failed" + assert not any(method == "StartCascade" for method, _ in language_server["calls"]) + + +def test_resume_retains_cascade_and_refreshes_current_guidance(tmp_path, language_server): + adapter = adapter_for(tmp_path) + adapter.start_turn("first") + adapter.close_session() + resumed = adapter_for(tmp_path, resume_thread_id="synthetic-cascade") + assert resumed.start_turn("second")["message"] == "Argon answer 2" + sends = [body for method, body in language_server["calls"] if method == "SendUserCascadeMessage"] + assert "Act as LoopX Steward." in sends[1]["items"][0]["text"] + assert len([1 for method, _ in language_server["calls"] if method == "StartCascade"]) == 1 + + +def test_timeout_with_partial_response_never_reports_success(tmp_path, language_server): + from loopx.chat_agent import CodexChatAgentError + language_server["mode"] = "running" + adapter = adapter_for(tmp_path, hard_timeout_sec=0.05) + events = [] + with pytest.raises(CodexChatAgentError) as error: + adapter.start_turn("long turn", lambda kind, data: events.append((kind, data))) + assert error.value.error_code == "response_timeout" + assert language_server["cancelled"].is_set() + assert not any(kind == "answer.final" for kind, _ in events) + + +def test_pending_native_approval_is_actionable_and_not_auto_accepted(tmp_path, language_server): + from loopx.chat_agent import CodexChatAgentError + language_server["mode"] = "waiting" + with pytest.raises(CodexChatAgentError) as error: + adapter_for(tmp_path, runtime_profile="trusted_owner").start_turn("requires native approval") + assert error.value.gate["kind"] == "host_tool_gate" + assert language_server["cancelled"].is_set() + + +def test_failed_send_keeps_priming_for_the_next_turn(tmp_path, language_server): + from loopx.chat_agent import CodexChatAgentError + adapter = adapter_for(tmp_path) + language_server["send_failure"] = True + with pytest.raises(CodexChatAgentError): + adapter.start_turn("failed") + language_server["send_failure"] = False + assert adapter.start_turn("retry")["message"] == "Argon answer 1" + sends = [body for method, body in language_server["calls"] if method == "SendUserCascadeMessage"] + assert "Act as LoopX Steward." in sends[1]["items"][0]["text"] + + +def test_explicit_remote_endpoint_and_redirect_never_forward_token(language_server, monkeypatch): + from loopx.capabilities.native_chat.antigravity_adapter import discover_antigravity_language_server + monkeypatch.setenv("ANTIGRAVITY_LS_URL", "https://example.invalid") + assert discover_antigravity_language_server() is None + assert language_server["calls"] == [] + monkeypatch.setenv("ANTIGRAVITY_LS_URL", language_server["url"]) + language_server["redirect"] = language_server["url"] + "/redirect-target" + assert discover_antigravity_language_server() is None + assert len(language_server["calls"]) == 1 + + +def test_stop_acknowledges_worker_and_next_turn_progresses(tmp_path, language_server): + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + store = ChatSessionStore(tmp_path / "runtime" / "chat") + controller = ChatRuntimeController(store=store, codex_bin="unused-codex", hard_timeout_sec=10) + session, _ = controller.open_session(goal_id="synthetic-goal", agent_id="antigravity", work_dir=tmp_path, + objective="Synthetic conversation", mode="new", channel_id="synthetic-argon-chat") + sid = session["session_id"] + assert session["agent_model_id"] == "gemini-4-argon-high" + language_server["mode"] = "running" + try: + turn, _ = controller.submit_turn(session_id=sid, client_turn_id="first", message="long request", work_dir=tmp_path, + objective="Synthetic conversation") + assert language_server["sent"].wait(5) + adapter = controller.adapters[sid] + with pytest.raises(ValueError, match="Turn changed"): + adapter.interrupt_turn("unrelated-turn") + assert not language_server["cancelled"].is_set() + stopped = controller.interrupt_turn(session_id=sid, turn_id=turn["turn_id"]) + assert stopped["status"] == "interrupted" + assert language_server["cancelled"].is_set() + language_server["mode"] = "complete" + next_turn, _ = controller.submit_turn(session_id=sid, client_turn_id="second", message="continue", work_dir=tmp_path, + objective="Synthetic conversation") + result = controller.wait_for_turn(session_id=sid, turn_id=next_turn["turn_id"], timeout_sec=5) + assert result["status"] == "completed" + assert result["response"]["message"] == "Argon answer 2" + finally: + controller.close_session(sid) + + +def test_deadline_expiry_after_partial_text_fails_closed(tmp_path, language_server, monkeypatch): + from loopx.capabilities.native_chat import antigravity_adapter as provider + from loopx.chat_agent import CodexChatAgentError + language_server["mode"] = "running" + adapter = adapter_for(tmp_path) + ticks = iter([0.0, 0.0, 301.0]) + monkeypatch.setattr(provider.time, "monotonic", lambda: next(ticks, 301.0)) + with pytest.raises(CodexChatAgentError) as error: + adapter.start_turn("long request") + assert error.value.error_code == "response_timeout" + assert language_server["cancelled"].is_set() + + +def test_busy_native_conversation_is_not_sent_to_or_cancelled(tmp_path, language_server): + from loopx.chat_agent import CodexChatAgentError + language_server.update(mode="running", active=True) + adapter = adapter_for(tmp_path, resume_thread_id="synthetic-cascade") + with pytest.raises(CodexChatAgentError) as error: + adapter.start_turn("must not race the native owner") + assert error.value.error_code == "resume_failed" + assert not any(method in {"SendUserCascadeMessage", "CancelCascadeInvocation"} + for method, _ in language_server["calls"]) From 815502466c3baab1e74f97f870c3aa11088de149 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:25:23 +0530 Subject: [PATCH 3/4] refactor(chat): share native provider configuration before dispatch Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/canary/module_metric_baseline.json | 2 +- .../native_chat/model_settings.py | 20 ++++- loopx/chat_runtime.py | 74 ++++--------------- 3 files changed, 34 insertions(+), 62 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index e31401d0c4..3041c487a6 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -31,7 +31,7 @@ "dict_any_count": 0 }, "loopx/chat_agent.py": { - "lines": 2007 + "lines": 2008 }, "loopx/chat_runtime.py": { "any_count": 37, diff --git a/loopx/capabilities/native_chat/model_settings.py b/loopx/capabilities/native_chat/model_settings.py index fc6732a003..b8133ca7f4 100644 --- a/loopx/capabilities/native_chat/model_settings.py +++ b/loopx/capabilities/native_chat/model_settings.py @@ -5,7 +5,7 @@ from __future__ import annotations from pathlib import Path -from typing import Any, TYPE_CHECKING +from typing import Any, Mapping, TYPE_CHECKING if TYPE_CHECKING: from ...chat_runtime import ChatRuntimeController, ChatRuntimeAdapter @@ -15,6 +15,24 @@ MODEL_SELECTION_UNSET = object() +def native_adapter_configuration(controller: ChatRuntimeController, *, agent_id: str, + goal_id: str | None, objective: str, manager_runtime: Mapping[str, Any] | None, + executor_model: Mapping[str, Any] | None, agent_config: Mapping[str, str] | None, + execution_mode: bool) -> tuple[dict[str, Any] | None, str, dict[str, Any]]: + """Share native-provider selection; the existing manager owner decides policy.""" + from ...chat_manager import MANAGER_AGENT_GOAL_ID, manager_agent_objective, manager_model_config + profile = (dict(manager_runtime or controller.manager_runtime_profile()) + if goal_id == MANAGER_AGENT_GOAL_ID else None) + if profile is not None: + objective = manager_agent_objective(str(profile["runtime_profile"])) + model = (executor_model or manager_model_config( + endpoint=agent_id, machine_defaults=controller.steward_executor_defaults()) + if goal_id == MANAGER_AGENT_GOAL_ID and not execution_mode else executor_model or {}) + return profile, objective, {**model, **{ + key: agent_config[key] for key in ("model", "reasoning_effort", "service_tier") + if agent_config and agent_config.get(key)}} + + def session_model_settings_supported(session: dict[str, Any]) -> bool: """Project the same current-Session boundary used by the save operation. diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index aeeae5698b..76dc0d81c7 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -373,6 +373,18 @@ def _start_adapter( ), }, ) + if agent_id in {"codex", "antigravity"}: + from .capabilities.native_chat.model_settings import native_adapter_configuration + manager_profile, objective, model_config = native_adapter_configuration( + self, agent_id=agent_id, goal_id=goal_id, objective=objective, + manager_runtime=manager_runtime, executor_model=executor_model, + agent_config=agent_config, execution_mode=execution_mode, + ) + native_context = { + "objective": self._session_objective(goal_id=goal_id, objective=objective, + project_coordination=project_coordination and not execution_mode), + "runtime_profile": str(manager_profile["runtime_profile"]) if manager_profile else "restricted", + } if agent_id == "antigravity": from .capabilities.native_chat.antigravity_adapter import AntigravityAdapter if execution_mode or project_context is not None or read_only_review: @@ -380,87 +392,29 @@ def _start_adapter( "Antigravity supports Chat and Steward; use Codex for governed workspace execution.", gate=None, error_code="unsupported_configuration", ) - manager_profile = ( - dict(manager_runtime or self.manager_runtime_profile()) - if goal_id == MANAGER_AGENT_GOAL_ID - else None - ) - if manager_profile is not None: - objective = manager_agent_objective( - str(manager_profile["runtime_profile"]) - ) - model_config = ( - executor_model or manager_model_config( - endpoint=agent_id, - machine_defaults=self.steward_executor_defaults(), - ) - if goal_id == MANAGER_AGENT_GOAL_ID and not execution_mode - else executor_model or {} - ) - if agent_config and agent_config.get("model"): - model_config = {**model_config, "model": agent_config["model"]} - if agent_config and agent_config.get("reasoning_effort"): - model_config = {**model_config, "reasoning_effort": agent_config["reasoning_effort"]} return AntigravityAdapter.start( work_dir=work_dir, - objective=self._session_objective( - goal_id=goal_id, objective=objective, - project_coordination=project_coordination and not execution_mode, - ), + **native_context, resume_thread_id=resume_thread_id, history=history, - runtime_profile=( - str(manager_profile["runtime_profile"]) - if manager_profile is not None - else "restricted" - ), executor_model=model_config, hard_timeout_sec=self.hard_timeout_sec or 300.0, ) if agent_id == "codex": from .capabilities.manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL from .chat_loopx_mode import TOOL as COLLABORATION_TOOL - manager_profile = ( - dict(manager_runtime or self.manager_runtime_profile()) - if goal_id == MANAGER_AGENT_GOAL_ID - else None - ) - if manager_profile is not None: - objective = manager_agent_objective( - str(manager_profile["runtime_profile"]) - ) - model_config = ( - executor_model or manager_model_config( - endpoint=agent_id, - machine_defaults=self.steward_executor_defaults(), - ) - if goal_id == MANAGER_AGENT_GOAL_ID and not execution_mode - else executor_model or {} - ) - if agent_config and agent_config.get("model"): - model_config = {**model_config, "model": agent_config["model"]} - if agent_config: - model_config = {**model_config, **{key: agent_config[key] for key in ("reasoning_effort", "service_tier") if agent_config.get(key)}} return CodexAppServerAdapter.start( codex_bin=self.codex_bin, codex_home=codex_home_override or self.codex_home, work_dir=work_dir, goal_id=goal_id, - objective=self._session_objective( - goal_id=goal_id, objective=objective, - project_coordination=project_coordination and not execution_mode, - ), + **native_context, history=history, resume_thread_id=resume_thread_id, startup_timeout_sec=self.startup_timeout_sec, idle_timeout_sec=self.idle_timeout_sec, hard_timeout_sec=self.hard_timeout_sec, execution_mode=execution_mode, - runtime_profile=( - str(manager_profile["runtime_profile"]) - if manager_profile is not None - else "restricted" - ), sandbox=( str(manager_profile["sandbox"]) if manager_profile is not None From 4ed408d39b49180c0a03e3116e8dd9caf9418ad1 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:34:01 +0530 Subject: [PATCH 4/4] fix(chat): make cancelled native permission recovery explicit Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/native_chat/antigravity_adapter.py | 6 +++--- tests/test_antigravity_steward_adapter.py | 2 ++ 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 39ba6bdc18..3a5714a026 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -568,9 +568,9 @@ def _run_turn( step_status = str(step.get("status") or "") if step_status == "CORTEX_STEP_STATUS_WAITING": raise CodexChatAgentError( - "Antigravity needs your response in its application; resolve it there and retry.", - gate={"kind": "host_tool_gate", "summary": "Antigravity is waiting for input.", - "next_action": "Open Antigravity and resolve the pending interaction."}, + "Antigravity requested native permission; this invocation was cancelled. Review its permissions, then retry.", + gate={"kind": "host_tool_gate", "summary": "Antigravity requires native permission.", + "next_action": "Review permissions in Antigravity, then resend this request in LoopX."}, error_code="host_tool_gate", ) if step_type == "CORTEX_STEP_TYPE_PLANNER_RESPONSE": diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index 19d4c0076c..995840b2e9 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -245,6 +245,8 @@ def test_pending_native_approval_is_actionable_and_not_auto_accepted(tmp_path, l with pytest.raises(CodexChatAgentError) as error: adapter_for(tmp_path, runtime_profile="trusted_owner").start_turn("requires native approval") assert error.value.gate["kind"] == "host_tool_gate" + assert "permissions" in error.value.gate["next_action"] + assert "cancelled" in str(error.value) assert language_server["cancelled"].is_set()