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/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/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py new file mode 100644 index 0000000000..3a5714a026 --- /dev/null +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -0,0 +1,704 @@ +"""Antigravity Language Server adapter for LoopX Native Chat and Steward.""" + +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 ( + 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], ...] = tuple( + { + "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") +) + + +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( + base_url: str, + csrf_token: str, + method: str, + payload: Mapping[str, Any], + *, + timeout: float = 15.0, +) -> dict[str, Any]: + 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"), + 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: + raise CodexChatAgentError( + 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 ({type(exc).__name__})", + 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 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, + "GetCascadeModelConfigData", + {}, + timeout=2.5, + ) + return { + "url": explicit_url.rstrip("/"), + "csrf_token": explicit_token, + "pid": None, + "source": "environment", + } + except (CodexChatAgentError, ValueError): + return None + + try: + ps_out = subprocess.check_output( + ["ps", "-u", str(os.getuid()), "-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 "antigravity" not in line.lower() + 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))) + + 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=max(0.1, deadline - time.monotonic()), + ) + 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: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + url = f"http://127.0.0.1:{port}" + try: + _rpc_post(url, token, "GetCascadeModelConfigData", {}, timeout=min(1.5, remaining)) + 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") + 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().lower() + effort = str(reasoning_effort or "").strip().lower() + 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: + """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" + + 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, + hard_timeout_sec: float = 300.0, + ) -> 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 = 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: + 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, + 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( + "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"]) + + 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() + 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 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, + 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=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 _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)) + parts.append( + _turn_prompt( + message, + runtime_profile=self.runtime_profile, + ) + ) + return "\n\n".join(parts) + + 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: + 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, + ) -> dict[str, Any]: + if self._closed: + raise CodexChatAgentError( + "Antigravity session is closed", + gate=None, + error_code="session_closed", + ) + + turn_id = f"agy_turn_{uuid.uuid4().hex[:12]}" + before_traj = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectory", + {"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 = 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, + } + 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() + self.hard_timeout_sec + saw_running_or_new_step = False + + while time.monotonic() < deadline: + 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=2.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: + self._check_interrupted() + steps_resp = _rpc_post( + self.ls_url, + self.csrf_token, + "GetCascadeTrajectorySteps", + { + "cascadeId": self.cascade_id, + "startIndex": start_step_index, + "endIndex": total_steps, + }, + timeout=2.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 "") + step_status = str(step.get("status") or "") + if step_status == "CORTEX_STEP_STATUS_WAITING": + raise CodexChatAgentError( + "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": + 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 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 {} + 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", + ) + + 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() + 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 _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: + 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: + 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..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. @@ -114,6 +132,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 +218,15 @@ 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 {} + 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/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_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 4d0331f324..ef52c1fd03 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": "antigravity_unavailable", + "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..76dc0d81c7 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,54 +369,52 @@ 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 == "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 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, ) - 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(), + 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: + raise CodexChatAgentError( + "Antigravity supports Chat and Steward; use Codex for governed workspace execution.", + gate=None, error_code="unsupported_configuration", ) - if goal_id == MANAGER_AGENT_GOAL_ID and not execution_mode - else executor_model or {} + return AntigravityAdapter.start( + work_dir=work_dir, + **native_context, + resume_thread_id=resume_thread_id, + history=history, + executor_model=model_config, + hard_timeout_sec=self.hard_timeout_sec or 300.0, ) - 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)}} + if agent_id == "codex": + from .capabilities.manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL + from .chat_loopx_mode import TOOL as COLLABORATION_TOOL 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 @@ -526,12 +524,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 +691,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, @@ -1004,7 +1002,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/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py new file mode 100644 index 0000000000..995840b2e9 --- /dev/null +++ b/tests/test_antigravity_steward_adapter.py @@ -0,0 +1,328 @@ +"""Focused tests for Antigravity (Gemini 4 Argon) Steward & Chat adapter.""" + +from __future__ import annotations + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import threading + +import pytest +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", + ) + + +@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 Handler(BaseHTTPRequestHandler): + def log_message(self, format, *args): + pass + + 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 + if method == "GetCascadeModelConfigData": + 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": + response = {"cascadeId": "synthetic-cascade"} + elif method == "GetCascadeTrajectory": + 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": + 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": + 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: + 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 = 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: + 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 "permissions" in error.value.gate["next_action"] + assert "cancelled" in str(error.value) + 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"])