From 86ca0d967d5ef17936cbc3cc8c716e979bc4019c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:59:54 -0700 Subject: [PATCH] refactor(agent/gemini,acp): table-driven generationConfig, folded predicates --- agent/copilot_acp_client.py | 14 +++----- agent/gemini_native_adapter.py | 59 ++++++++++++---------------------- 2 files changed, 25 insertions(+), 48 deletions(-) diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py index f7fd22edfb..2038a10536 100644 --- a/agent/copilot_acp_client.py +++ b/agent/copilot_acp_client.py @@ -164,9 +164,7 @@ def _model_selection_request(session: dict[str, Any], requested_model: str) -> t if not session_id or not requested_model or requested_model == "copilot-acp": return None model_option = next( - (o for o in (session.get("configOptions") or []) - if isinstance(o, dict) and (o.get("category") == "model" or o.get("id") == "model")), - None, + (o for o in (session.get("configOptions") or []) if isinstance(o, dict) and "model" in (o.get("category"), o.get("id"))), None ) if model_option is not None: if requested_model not in _enabled_ids(model_option.get("options"), "value"): @@ -208,9 +206,7 @@ def _render_message_content(content: Any) -> str: if isinstance(content, dict): if "text" in content: return str(content.get("text") or "").strip() - if isinstance(content.get("content"), str): - return content["content"].strip() - return json.dumps(content, ensure_ascii=True) + return content["content"].strip() if isinstance(content.get("content"), str) else json.dumps(content, ensure_ascii=True) if isinstance(content, list): parts = [ item if isinstance(item, str) else item["text"].strip() @@ -225,8 +221,7 @@ def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path: candidate = Path(path_text) if not candidate.is_absolute(): raise PermissionError("ACP file-system paths must be absolute.") - resolved = candidate.resolve() - root = Path(cwd).resolve() + resolved, root = candidate.resolve(), Path(cwd).resolve() try: resolved.relative_to(root) except ValueError as exc: @@ -241,8 +236,7 @@ def _effective_timeout(timeout: Any) -> float: if isinstance(timeout, (int, float)): return float(timeout) candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")] - numeric = [float(v) for v in candidates if isinstance(v, (int, float))] - return max(numeric) if numeric else _DEFAULT_TIMEOUT_SECONDS + return max((float(v) for v in candidates if isinstance(v, (int, float))), default=_DEFAULT_TIMEOUT_SECONDS) def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any: diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py index b565b544d8..ddfb39904c 100644 --- a/agent/gemini_native_adapter.py +++ b/agent/gemini_native_adapter.py @@ -102,24 +102,22 @@ def probe_gemini_tier( base = str(base_url or DEFAULT_GEMINI_BASE_URL).strip().rstrip("/") or DEFAULT_GEMINI_BASE_URL if base.lower().endswith("/openai"): base = base[: -len("/openai")] - payload = {"contents": [{"role": "user", "parts": [{"text": "hi"}]}], "generationConfig": {"maxOutputTokens": 1}} try: with httpx.Client(timeout=timeout) as client: resp = client.post( - f"{base}/models/{model}:generateContent", params={"key": key}, json=payload, + f"{base}/models/{model}:generateContent", params={"key": key}, + json={"contents": [{"role": "user", "parts": [{"text": "hi"}]}], "generationConfig": {"maxOutputTokens": 1}}, headers={"Content-Type": "application/json", "X-Goog-Api-Client": _API_CLIENT}, ) except Exception as exc: logger.debug("probe_gemini_tier: network error: %s", exc) return "unknown" - rpd_header = {k.lower(): v for k, v in resp.headers.items()}.get("x-ratelimit-limit-requests-per-day") - if rpd_header: - # Free-tier daily caps top out at 1000 (flash-lite); Tier 1 starts ~1500+. - try: + try: + if rpd_header: # free-tier daily caps top out at 1000 (flash-lite); Tier 1 starts ~1500+ return "free" if int(rpd_header) <= 1000 else "paid" - except (TypeError, ValueError): - pass + except (TypeError, ValueError): + pass if resp.status_code == 429: return "free" if "free_tier" in _response_text(resp).lower() else "paid" return "paid" if 200 <= resp.status_code < 300 else "unknown" @@ -204,9 +202,7 @@ def _extract_multimodal_parts(content: Any) -> List[Dict[str, Any]]: def _tool_call_extra_signature(tool_call: Dict[str, Any]) -> Optional[str]: extra = tool_call.get("extra_content") or {} - if not isinstance(extra, dict): - return None - google = extra.get("google") or extra.get("thought_signature") + google = (extra.get("google") or extra.get("thought_signature")) if isinstance(extra, dict) else None sig = (google.get("thought_signature") or google.get("thoughtSignature")) if isinstance(google, dict) else google return sig if isinstance(sig, str) and sig else None @@ -343,11 +339,8 @@ def _translate_tools_to_gemini(tools: Any) -> List[Dict[str, Any]]: def _translate_tool_choice_to_gemini(tool_choice: Any) -> Optional[Dict[str, Any]]: if isinstance(tool_choice, str) and tool_choice in _TOOL_CHOICE_MODES: return {"functionCallingConfig": {"mode": _TOOL_CHOICE_MODES[tool_choice]}} - if isinstance(tool_choice, dict): - name = (tool_choice.get("function") or {}).get("name") - if isinstance(name, str) and name: - return {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [name]}} - return None + name = (tool_choice.get("function") or {}).get("name") if isinstance(tool_choice, dict) else None + return {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [name]}} if isinstance(name, str) and name else None # (camelCase key, snake_case alias, accepted types, normalizer) @@ -390,9 +383,7 @@ def _effective_gemini_max_output_tokens(max_tokens: Optional[int], thinking_conf requested = int(max_tokens) except (TypeError, ValueError): requested = 0 - if requested <= 0: - return GEMINI_DEFAULT_MAX_OUTPUT_TOKENS - if _thinking_requests_output_headroom(thinking_config): + if requested <= 0 or _thinking_requests_output_headroom(thinking_config): return max(requested, GEMINI_DEFAULT_MAX_OUTPUT_TOKENS) return requested @@ -405,26 +396,18 @@ def build_gemini_request( # Gemini 3+ both requires tool-call ids and accepts multimodal functionResponse parts. is_gemini3 = gemini_requires_tool_call_ids(model) contents, system_instruction = _build_gemini_contents(messages, include_tool_call_ids=is_gemini3, is_gemini3=is_gemini3) - request: Dict[str, Any] = {"contents": contents} - for key, value in ( - ("systemInstruction", system_instruction), - ("tools", _translate_tools_to_gemini(tools)), + optional = ( + ("systemInstruction", system_instruction), ("tools", _translate_tools_to_gemini(tools)), ("toolConfig", _translate_tool_choice_to_gemini(tool_choice)), - ): - if value: - request[key] = value - - generation_config: Dict[str, Any] = {} - if temperature is not None: - generation_config["temperature"] = temperature - generation_config["maxOutputTokens"] = _effective_gemini_max_output_tokens(max_tokens, thinking_config) - if top_p is not None: - generation_config["topP"] = top_p - if stop: - generation_config["stopSequences"] = stop if isinstance(stop, list) else [str(stop)] - if normalized_thinking := _normalize_thinking_config(thinking_config): - generation_config["thinkingConfig"] = normalized_thinking - request["generationConfig"] = generation_config + ) + request: Dict[str, Any] = {"contents": contents, **{k: v for k, v in optional if v}} + # Key order is part of the wire format (prompt-cache parity): temperature, maxOutputTokens, topP, stop, thinking. + generation = ( + ("temperature", temperature), ("maxOutputTokens", _effective_gemini_max_output_tokens(max_tokens, thinking_config)), + ("topP", top_p), ("stopSequences", (stop if isinstance(stop, list) else [str(stop)]) if stop else None), + ("thinkingConfig", _normalize_thinking_config(thinking_config)), + ) + request["generationConfig"] = {k: v for k, v in generation if v is not None} return request