Aether/_deprecated_py_src/services/provider/adapters/antigravity/envelope.py

"""Antigravity v1internal request/response envelope helpers.

Antigravity reuses the `gemini:chat` endpoint signature but wraps the actual
wire format:
- Request: V1InternalRequest (top-level metadata + nested GeminiRequest)
- Response: V1InternalResponse (top-level responseId + nested GeminiResponse)

对齐 Antigravity-Manager wrapper.rs 的处理逻辑：
- Claude model tool ID 注入（request + response）
- Thinking budget capping (Auto cap 24576)
- [undefined] 字符串深度清理
- parametersJsonSchema → parameters 重命名
- JSON Schema 禁止字段清洗
- Antigravity System Instruction 注入
- Signature 错误检测
- Model alias mapping（preview → physical）
- Google Search (grounding) 注入
- thoughtSignature 注入到 functionCall parts
- Image generation config 注入（aspectRatio / imageSize）
"""

from __future__ import annotations

import uuid
from typing import Any

from src.services.provider.adapters.antigravity.constants import (
    ANTIGRAVITY_SYSTEM_INSTRUCTION,
    ASPECT_RATIO_TABLE,
    IMAGE_ASPECT_RATIO_SUFFIXES,
    IMAGE_GEN_UPSTREAM_MODEL,
    MODEL_MAX_OUTPUT_LIMIT,
    NETWORKING_TOOL_KEYWORDS,
    OUTPUT_OVERHEAD,
    OUTPUT_OVERHEAD_IMAGE,
)
from src.services.provider.adapters.antigravity.constants import (
    REQUEST_USER_AGENT as ANTIGRAVITY_REQUEST_USER_AGENT,
)
from src.services.provider.adapters.antigravity.constants import (
    SIGNATURE_ERROR_KEYWORDS,
    STANDARD_ASPECT_RATIOS,
    THINKING_BUDGET_AUTO_CAP,
    THINKING_BUDGET_DEFAULT_INJECT,
    THINKING_MODELS_AUTO_INJECT_KEYWORDS,
    WEB_SEARCH_MODEL,
)
from src.services.provider.adapters.antigravity.url_availability import url_availability
from src.services.provider.request_context import get_selected_base_url

# ---------------------------------------------------------------------------
# Key normalization: snake_case → camelCase
# ---------------------------------------------------------------------------

# The Gemini normalizer (gemini.py) outputs snake_case keys that mirror protobuf
# field names (e.g. "generation_config", "function_declarations").  However, the
# Antigravity v1internal JSON protocol (like the REST Gemini API) uses camelCase
# for all field names.  A mismatch causes downstream processing functions in this
# module to silently skip keys or create duplicate entries.
#
# We normalise once at the entry point of wrap_v1internal_request so that every
# subsequent helper can safely assume camelCase.

_TOP_LEVEL_KEY_RENAMES: dict[str, str] = {
    "system_instruction": "systemInstruction",
    "generation_config": "generationConfig",
    "tool_config": "toolConfig",
    # safety_settings 在 wrap_v1internal_request 入口处已 pop，无需映射
}

_GENERATION_CONFIG_KEY_RENAMES: dict[str, str] = {
    "max_output_tokens": "maxOutputTokens",
    "stop_sequences": "stopSequences",
    "top_p": "topP",
    "top_k": "topK",
    "thinking_config": "thinkingConfig",
    "response_modalities": "responseModalities",
    "response_mime_type": "responseMimeType",
}


def _normalize_to_camel_case(body: dict[str, Any]) -> None:
    """In-place normalise known Gemini snake_case keys to their camelCase form.

    This must be called **before** any other processing so that all helpers
    in this module can consistently use camelCase lookups.
    """
    # 1. Top-level keys
    for snake, camel in _TOP_LEVEL_KEY_RENAMES.items():
        if snake in body and camel not in body:
            body[camel] = body.pop(snake)

    # 2. Inside tools: function_declarations → functionDeclarations
    tools = body.get("tools")
    if isinstance(tools, list):
        for tool in tools:
            if isinstance(tool, dict):
                if "function_declarations" in tool and "functionDeclarations" not in tool:
                    tool["functionDeclarations"] = tool.pop("function_declarations")

    # 3. Inside generationConfig: normalise sub-keys
    gc = body.get("generationConfig")
    if isinstance(gc, dict):
        for snake, camel in _GENERATION_CONFIG_KEY_RENAMES.items():
            if snake in gc and camel not in gc:
                gc[camel] = gc.pop(snake)

    # 4. Inside contents → parts: function_call/function_response → camelCase
    contents = body.get("contents")
    if isinstance(contents, list):
        for content in contents:
            if not isinstance(content, dict):
                continue
            parts = content.get("parts")
            if not isinstance(parts, list):
                continue
            for part in parts:
                if not isinstance(part, dict):
                    continue
                if "function_call" in part and "functionCall" not in part:
                    part["functionCall"] = part.pop("function_call")
                if "function_response" in part and "functionResponse" not in part:
                    part["functionResponse"] = part.pop("function_response")


# ---------------------------------------------------------------------------
# Request body 预处理工具函数（对齐 AM wrapper.rs / common_utils.rs）
# ---------------------------------------------------------------------------


def _deep_clean_undefined(obj: Any) -> None:
    """In-place 递归清理 '[undefined]' 字符串值。

    Cherry Studio 等客户端会在请求中注入 '[undefined]' 字符串，
    可能导致上游 API 解析错误。
    """
    if isinstance(obj, dict):
        keys_to_remove = [k for k, v in obj.items() if v == "[undefined]"]
        for k in keys_to_remove:
            del obj[k]
        for v in obj.values():
            if isinstance(v, (dict, list)):
                _deep_clean_undefined(v)
    elif isinstance(obj, list):
        i = 0
        while i < len(obj):
            if obj[i] == "[undefined]":
                obj.pop(i)
            else:
                if isinstance(obj[i], (dict, list)):
                    _deep_clean_undefined(obj[i])
                i += 1


def _inject_claude_tool_ids_request(inner_request: dict[str, Any], model: str) -> None:
    """为 Claude 模型注入 functionCall/functionResponse 的 id 字段。

    Google v1internal 在目标模型为 Claude 时要求 functionCall 带有 id 字段，
    但标准 Gemini 协议不包含此字段。对齐 AM wrapper.rs #1522。

    算法：
    1. 第一遍：扫描所有 functionCall，为没有 id 的生成 id，并记录 (name, id) 队列
    2. 第二遍：扫描所有 functionResponse，从对应 name 的队列中取出 id 使用
    """
    if "claude" not in model.lower():
        return

    contents = inner_request.get("contents")
    if not isinstance(contents, list):
        return

    # 第一遍：收集所有 functionCall 的 ID（按 name 分组，保持顺序）
    # name -> [id1, id2, ...] 每个调用的 ID 按出现顺序排列
    call_ids_by_name: dict[str, list[str]] = {}
    # 所有 call ID 按原始出现顺序（用于 fallback 匹配）
    all_call_ids_ordered: list[str] = []
    # 用于生成新 ID 的计数器（全局，确保唯一性）
    name_counters: dict[str, int] = {}

    for content in contents:
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue

        for part in parts:
            if not isinstance(part, dict):
                continue

            fc = part.get("functionCall") or part.get("function_call")
            if isinstance(fc, dict):
                name = fc.get("name", "")
                if not isinstance(name, str) or not name:
                    name = "unknown"

                fc_id = fc.get("id")
                if not (isinstance(fc_id, str) and fc_id):
                    # 生成新 ID
                    count = name_counters.get(name, 0)
                    fc_id = f"call_{name}_{count}"
                    fc["id"] = fc_id
                    name_counters[name] = count + 1

                # 记录这个 call 的 ID，供后续 response 使用
                if name not in call_ids_by_name:
                    call_ids_by_name[name] = []
                call_ids_by_name[name].append(fc_id)
                # 同时按原始出现顺序记录（用于 fallback）
                all_call_ids_ordered.append(fc_id)

    # 第二遍：为 functionResponse 分配匹配的 ID
    # 使用索引追踪每个 name 已消费到第几个 ID
    response_index_by_name: dict[str, int] = {}
    # 已使用的 call ID 集合
    used_call_ids: set[str] = set()
    # fallback 用的有序索引
    fallback_index = 0

    for content in contents:
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue

        for part in parts:
            if not isinstance(part, dict):
                continue

            fr = part.get("functionResponse") or part.get("function_response")
            if isinstance(fr, dict) and not (isinstance(fr.get("id"), str) and fr.get("id")):
                name = fr.get("name", "")
                if not isinstance(name, str) or not name:
                    name = "unknown"

                assigned_id: str | None = None

                # 优先：从对应 name 的 call ID 队列中取出下一个 ID
                call_ids = call_ids_by_name.get(name, [])
                idx = response_index_by_name.get(name, 0)
                while idx < len(call_ids):
                    cid = call_ids[idx]
                    idx += 1
                    if cid not in used_call_ids:
                        assigned_id = cid
                        used_call_ids.add(cid)
                        break
                response_index_by_name[name] = idx

                # Fallback：如果 name 是 unknown（原本为空），尝试使用下一个未使用的 call ID（按原始出现顺序）
                if assigned_id is None and name == "unknown":
                    while fallback_index < len(all_call_ids_ordered):
                        cid = all_call_ids_ordered[fallback_index]
                        fallback_index += 1
                        if cid not in used_call_ids:
                            assigned_id = cid
                            used_call_ids.add(cid)
                            break

                if assigned_id is not None:
                    fr["id"] = assigned_id
                else:
                    # 没有匹配的 call ID（异常情况），生成一个新 ID
                    count = name_counters.get(name, 0)
                    fr["id"] = f"call_{name}_{count}"
                    name_counters[name] = count + 1


def _process_thinking_budget(inner_request: dict[str, Any], model: str) -> None:
    """处理 Thinking Budget：自动注入 + Auto Cap + maxOutputTokens 约束。

    对齐 AM wrapper.rs + CLIProxyAPI 混合方案：
    - 对 flash/pro/thinking 模型处理 thinkingConfig
    - 自动注入 thinkingConfig（对已知需要 thinking 的模型）
    - Auto Cap：budget 超过 24576 时裁剪
    - Claude 要求：maxOutputTokens 必须 > thinkingBudget
    - 优先增加 maxOutputTokens；超限时才减少 budget
    """
    lower_model = model.lower()
    if not any(kw in lower_model for kw in ("flash", "pro", "thinking")):
        return

    # 确保 generationConfig 存在
    gen_config = inner_request.setdefault("generationConfig", {})
    if not isinstance(gen_config, dict):
        return

    # 自动注入 thinkingConfig（对已知需要 thinking 的模型）
    if gen_config.get("thinkingConfig") is None:
        should_inject = any(kw in lower_model for kw in THINKING_MODELS_AUTO_INJECT_KEYWORDS)
        if should_inject:
            gen_config["thinkingConfig"] = {
                "includeThoughts": True,
                "thinkingBudget": THINKING_BUDGET_DEFAULT_INJECT,
            }

    # Auto Cap
    thinking_config = gen_config.get("thinkingConfig")
    if not isinstance(thinking_config, dict):
        return

    budget = thinking_config.get("thinkingBudget")
    if not isinstance(budget, int) or budget <= 0:
        return

    # 1. 限制 budget 上限
    if budget > THINKING_BUDGET_AUTO_CAP:
        budget = THINKING_BUDGET_AUTO_CAP
        thinking_config["thinkingBudget"] = budget

    # 2. 确保 maxOutputTokens > thinkingBudget
    current_max = gen_config.get("maxOutputTokens")

    # 根据模型类型选择增量（对齐 Antigravity-Manager）
    overhead = OUTPUT_OVERHEAD_IMAGE if "-image" in lower_model else OUTPUT_OVERHEAD

    # 计算理想的 maxOutputTokens
    ideal_max = budget + overhead

    # 3. 确保不超过模型限制
    if ideal_max > MODEL_MAX_OUTPUT_LIMIT:
        # 超限时：减少 budget 而不是超限（对齐 CLIProxyAPI 策略）
        ideal_max = MODEL_MAX_OUTPUT_LIMIT
        # 确保 budget < ideal_max，保留 overhead 空间给输出
        max_budget = ideal_max - overhead
        if budget > max_budget:
            thinking_config["thinkingBudget"] = max_budget

    # 4. 应用修正
    if current_max is None or (isinstance(current_max, int) and current_max <= budget):
        gen_config["maxOutputTokens"] = ideal_max


def _clean_tool_declarations(inner_request: dict[str, Any]) -> None:
    """清洗工具声明：重命名字段 + 移除禁止的 Schema 字段 + 过滤搜索声明。

    对齐 AM wrapper.rs：
    - parametersJsonSchema → parameters（Gemini CLI 兼容）
    - 移除 Gemini 不支持的 Schema 字段（multipleOf 等）
    - 过滤 web_search / google_search 工具声明
    """
    tools = inner_request.get("tools")
    if not isinstance(tools, list):
        return

    for tool in tools:
        if not isinstance(tool, dict):
            continue
        # 支持 camelCase + snake_case
        decls_key = (
            "functionDeclarations"
            if "functionDeclarations" in tool
            else "function_declarations" if "function_declarations" in tool else None
        )
        if decls_key is None:
            continue
        decls = tool.get(decls_key)
        if not isinstance(decls, list):
            continue

        # 1. 过滤搜索关键字函数（对齐 NETWORKING_TOOL_KEYWORDS）
        decls[:] = [
            d
            for d in decls
            if not (
                isinstance(d, dict)
                and isinstance(d.get("name"), str)
                and d["name"] in NETWORKING_TOOL_KEYWORDS
            )
        ]

        # 2. 重命名 + 清洗
        for decl in decls:
            if not isinstance(decl, dict):
                continue

            # parametersJsonSchema → parameters
            if "parametersJsonSchema" in decl:
                params = decl.pop("parametersJsonSchema")
                if isinstance(params, dict):
                    _clean_json_schema(params)
                decl["parameters"] = params
            elif "parameters" in decl:
                params = decl["parameters"]
                if isinstance(params, dict):
                    _clean_json_schema(params)


def _clean_json_schema(schema: dict[str, Any]) -> None:
    """递归移除 Gemini 不支持的 JSON Schema 字段。"""
    from src.core.api_format.schema_utils import clean_gemini_schema

    clean_gemini_schema(schema)


# ---------------------------------------------------------------------------
# Google Search (Grounding) 检测与注入（对齐 AM common_utils.rs）
# ---------------------------------------------------------------------------


def _detect_networking_tools(inner_request: dict[str, Any]) -> bool:
    """检测请求中是否包含联网/搜索工具声明。

    对齐 AM common_utils.rs detects_networking_tool，支持多种声明风格：
    1. Claude/Anthropic 直发风格: {"name": "web_search"} / {"type": "web_search_20250305"}
    2. OpenAI 嵌套风格: {"type": "function", "function": {"name": "web_search"}}
    3. Gemini 原生风格: {"functionDeclarations": [{"name": "web_search"}]}
    4. Gemini googleSearch 声明: {"googleSearch": {}}
    """
    tools = inner_request.get("tools")
    if not isinstance(tools, list):
        return False

    for tool in tools:
        if not isinstance(tool, dict):
            continue

        # 1. 直发风格: name / type 字段
        name = tool.get("name")
        if isinstance(name, str) and name in NETWORKING_TOOL_KEYWORDS:
            return True
        type_ = tool.get("type")
        if isinstance(type_, str) and type_ in NETWORKING_TOOL_KEYWORDS:
            return True

        # 2. OpenAI 嵌套风格
        func = tool.get("function")
        if isinstance(func, dict):
            fn_name = func.get("name")
            if isinstance(fn_name, str) and fn_name in NETWORKING_TOOL_KEYWORDS:
                return True

        # 3. Gemini functionDeclarations 风格（支持 camelCase + snake_case）
        decls = tool.get("functionDeclarations")
        if decls is None:
            decls = tool.get("function_declarations")
        if isinstance(decls, list):
            for decl in decls:
                if isinstance(decl, dict):
                    decl_name = decl.get("name")
                    if isinstance(decl_name, str) and decl_name in NETWORKING_TOOL_KEYWORDS:
                        return True

        # 4. Gemini googleSearch / googleSearchRetrieval 声明
        if tool.get("googleSearch") is not None or tool.get("googleSearchRetrieval") is not None:
            return True

    return False


def _detect_online_suffix(model: str) -> bool:
    """检测模型名是否含 -online 后缀（联网意图）。"""
    return model.rstrip().endswith("-online")


def _inject_google_search_tool(inner_request: dict[str, Any]) -> None:
    """注入 googleSearch tool 到请求中。

    对齐 AM common_utils.rs inject_google_search_tool：
    - 如果已有 functionDeclarations，跳过（v1internal 不支持混用）
    - 先清理已有的 googleSearch / googleSearchRetrieval
    - 注入 {"googleSearch": {}}
    """
    tools = inner_request.setdefault("tools", [])
    if not isinstance(tools, list):
        inner_request["tools"] = [{"googleSearch": {}}]
        return

    # 如果已有 functionDeclarations，不注入（v1internal 不支持混用 search 和 functions）
    has_functions = any(
        isinstance(t, dict) and ("functionDeclarations" in t or "function_declarations" in t)
        for t in tools
    )
    if has_functions:
        return

    # 清理已存在的 googleSearch / googleSearchRetrieval（避免重复）
    tools[:] = [
        t
        for t in tools
        if not (isinstance(t, dict) and ("googleSearch" in t or "googleSearchRetrieval" in t))
    ]

    # 注入
    tools.append({"googleSearch": {}})


# ---------------------------------------------------------------------------
# thoughtSignature 注入到 functionCall parts（对齐 AM wrapper.rs）
# ---------------------------------------------------------------------------


def _inject_thought_signatures(inner_request: dict[str, Any], session_id: str | None) -> None:
    """Inject thought signatures into functionCall parts.

    Prefer tool-specific signatures (tool_use_id -> thoughtSignature), then fall back
    to a session-level signature when available.
    """

    try:
        from src.core.api_format.conversion.thinking_cache import signature_cache
    except Exception:
        return

    session_sig: str | None = None
    if session_id:
        session_sig = signature_cache.get_session_signature(session_id)

    contents = inner_request.get("contents")
    if not isinstance(contents, list):
        return

    for content in contents:
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue

        for part in parts:
            if not isinstance(part, dict):
                continue

            fc = part.get("functionCall") or part.get("function_call")
            if not isinstance(fc, dict):
                continue

            # Normalize existing signature aliases to thoughtSignature (if present).
            sig_val: str | None = None
            ts = part.get("thoughtSignature")
            if isinstance(ts, str) and ts.strip():
                sig_val = ts.strip()
                if ts != sig_val:
                    part["thoughtSignature"] = sig_val
            else:
                tss = part.get("thought_signature")
                if isinstance(tss, str) and tss.strip():
                    sig_val = tss.strip()
                    part["thoughtSignature"] = sig_val
                    part.pop("thought_signature", None)
                else:
                    legacy = part.get("signature")
                    if isinstance(legacy, str) and legacy.strip():
                        sig_val = legacy.strip()
                        part["thoughtSignature"] = sig_val
                        part.pop("signature", None)

            if sig_val:
                continue

            tool_id = fc.get("id")
            tool_sig: str | None = None
            if isinstance(tool_id, str) and tool_id:
                tool_sig = signature_cache.get_tool_signature(tool_id)

            chosen = tool_sig or session_sig
            if not chosen:
                continue

            part["thoughtSignature"] = chosen
            # Avoid sending duplicate aliases once we inject.
            part.pop("thought_signature", None)
            if part.get("signature") in (None, ""):
                part.pop("signature", None)


# ---------------------------------------------------------------------------
# Image Generation Config（对齐 AM common_utils.rs parse_image_config_with_params）
# ---------------------------------------------------------------------------


def _calculate_aspect_ratio(size: str) -> str:
    """从 "WIDTHxHEIGHT" 或 "W:H" 字符串计算宽高比。

    对齐 AM common_utils.rs calculate_aspect_ratio_from_size：
    1. 先检查是否已是标准比例字符串 (如 "16:9")
    2. 解析 WIDTHxHEIGHT 并容差匹配
    3. 默认返回 "1:1"
    """
    if size in STANDARD_ASPECT_RATIOS:
        return size

    if "x" in size:
        try:
            w_str, h_str = size.split("x", 1)
            width, height = float(w_str), float(h_str)
            if width > 0 and height > 0:
                ratio = width / height
                for target_ratio, label in ASPECT_RATIO_TABLE:
                    if abs(ratio - target_ratio) < 0.05:
                        return label
        except (ValueError, ZeroDivisionError):
            pass

    return "1:1"


def _parse_image_config(
    model: str,
    inner_request: dict[str, Any],
) -> tuple[dict[str, Any], str]:
    """解析图像生成配置，返回 (imageConfig, clean_model_name)。

    对齐 AM common_utils.rs parse_image_config_with_params + resolve_request_config：
    1. 从请求体中提取 OpenAI 风格 size / quality 参数（优先）
    2. 回退到模型后缀解析 (如 -16x9, -4k)
    3. 合并请求体中的 generationConfig.imageConfig（如果存在）
    4. 上游模型固定为 "gemini-3-pro-image"
    """
    # 提取 OpenAI 风格参数（可能由跨格式转换层注入到请求根部）
    size = inner_request.pop("size", None)
    quality = inner_request.pop("quality", None)
    if not isinstance(size, str):
        size = None
    if not isinstance(quality, str):
        quality = None

    # --- 解析 aspectRatio ---
    aspect_ratio = "1:1"
    if size:
        aspect_ratio = _calculate_aspect_ratio(size)
    else:
        lower_model = model.lower()
        for suffix, ratio in IMAGE_ASPECT_RATIO_SUFFIXES.items():
            if suffix in lower_model:
                aspect_ratio = ratio
                break

    config: dict[str, Any] = {"aspectRatio": aspect_ratio}

    # --- 解析 imageSize ---
    if quality:
        q_lower = quality.lower()
        if q_lower in ("hd", "4k"):
            config["imageSize"] = "4K"
        elif q_lower in ("medium", "2k"):
            config["imageSize"] = "2K"
        elif q_lower in ("standard", "1k"):
            config["imageSize"] = "1K"
    else:
        lower_model = model.lower()
        if "-4k" in lower_model or "-hd" in lower_model:
            config["imageSize"] = "4K"
        elif "-2k" in lower_model:
            config["imageSize"] = "2K"

    # --- 合并请求体中已有的 imageConfig（body 可以覆盖除 imageSize 降级外的字段） ---
    gen_config = inner_request.get("generationConfig")
    if isinstance(gen_config, dict):
        body_image_config = gen_config.get("imageConfig")
        if isinstance(body_image_config, dict):
            for key, value in body_image_config.items():
                # 防止 body 降级 inferred imageSize（对齐 AM 的 shield 逻辑）
                if (
                    key == "imageSize"
                    and (value == "1K" or value is None)
                    and "imageSize" in config
                ):
                    continue
                config[key] = value

    return config, IMAGE_GEN_UPSTREAM_MODEL


def _apply_image_gen_config(inner_request: dict[str, Any], image_config: dict[str, Any]) -> None:
    """将 imageConfig 应用到请求的 generationConfig 中。

    对齐 AM wrapper.rs 的图像生成处理：
    - 移除 tools / systemInstruction
    - 确保 contents 中每个 content 有 role 字段
    - 清理 generationConfig 中与图像生成冲突的字段
    - 注入 imageConfig
    - 处理图像思维模式（默认 disabled）
    """
    # 移除不兼容字段（_normalize_to_camel_case 已统一 key，仅需 camelCase）
    for key in ("tools", "toolConfig", "systemInstruction"):
        inner_request.pop(key, None)

    # 确保 contents 中每个 content 有 role 字段
    contents = inner_request.get("contents")
    if isinstance(contents, list):
        for content in contents:
            if isinstance(content, dict) and "role" not in content:
                content["role"] = "user"

    # 清理 generationConfig
    gen_config = inner_request.setdefault("generationConfig", {})
    if not isinstance(gen_config, dict):
        gen_config = {}
        inner_request["generationConfig"] = gen_config

    # 移除与图像生成冲突的字段（_normalize_to_camel_case 已统一 key）
    for key in ("responseMimeType", "responseModalities"):
        gen_config.pop(key, None)

    # 注入 imageConfig
    gen_config["imageConfig"] = image_config

    # 图像思维模式：默认 disabled（对齐 AM wrapper.rs image_thinking_mode）
    gen_config["thinkingConfig"] = {"includeThoughts": False}


def _compact_contents(inner_request: dict[str, Any]) -> None:
    """Strip invalid parts, drop empty contents, merge consecutive same-role.

    Delegates to :func:`compact_gemini_contents` from the Gemini normalizer to
    avoid duplicating the validation logic.

    This function modifies *inner_request* in-place.
    """
    from src.core.api_format.conversion.normalizers.gemini import compact_gemini_contents

    contents = inner_request.get("contents")
    if not isinstance(contents, list):
        return

    result = compact_gemini_contents(contents)
    inner_request["contents"] = result


def _inject_system_instruction(inner_request: dict[str, Any]) -> None:
    """注入 Antigravity 身份系统指令。

    对齐 AM wrapper.rs：
    - 如果已有 systemInstruction：在前面插入（避免重复）
    - 如果没有：创建新的
    - 补全 role: user（Gemini API 要求）

    注意：Gemini API 使用 protobuf oneof，`system_instruction`（snake_case）和
    `systemInstruction`（camelCase）是同一个字段，只能设置其中一个。
    需要先统一到 camelCase 再处理。
    """
    # 统一 snake_case 到 camelCase（避免 oneof 冲突）
    if "system_instruction" in inner_request:
        snake_value = inner_request.pop("system_instruction")
        # 仅当 camelCase 不存在时才迁移
        if "systemInstruction" not in inner_request:
            inner_request["systemInstruction"] = snake_value

    system_instruction = inner_request.get("systemInstruction")

    if isinstance(system_instruction, dict):
        # 补全 role
        if "role" not in system_instruction:
            system_instruction["role"] = "user"

        parts = system_instruction.get("parts")
        if isinstance(parts, list):
            # 检查是否已包含 Antigravity 身份（避免重复注入）
            has_antigravity = False
            if parts and isinstance(parts[0], dict):
                text = parts[0].get("text", "")
                if isinstance(text, str) and "You are Antigravity" in text:
                    has_antigravity = True

            if not has_antigravity:
                parts.insert(0, {"text": ANTIGRAVITY_SYSTEM_INSTRUCTION})
    else:
        # 没有 systemInstruction，创建新的
        inner_request["systemInstruction"] = {
            "role": "user",
            "parts": [{"text": ANTIGRAVITY_SYSTEM_INSTRUCTION}],
        }


# ---------------------------------------------------------------------------
# Response 后处理工具函数
# ---------------------------------------------------------------------------


def _inject_claude_tool_ids_response(response: dict[str, Any], model: str) -> None:
    """为 Claude 模型的响应注入 functionCall 的 id 字段。

    对齐 AM wrapper.rs inject_ids_to_response：
    让下游客户端（如 OpenCode/Vercel AI SDK）能感知 tool call ID，
    并在下一轮对话中原样带回。
    """
    if "claude" not in model.lower():
        return

    candidates = response.get("candidates")
    if not isinstance(candidates, list):
        return

    for candidate in candidates:
        if not isinstance(candidate, dict):
            continue
        content = candidate.get("content")
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue

        name_counters: dict[str, int] = {}
        for part in parts:
            if not isinstance(part, dict):
                continue
            fc = part.get("functionCall") or part.get("function_call")
            if isinstance(fc, dict) and not (isinstance(fc.get("id"), str) and fc.get("id")):
                name = fc.get("name", "unknown")
                if not isinstance(name, str):
                    name = "unknown"
                count = name_counters.get(name, 0)
                fc["id"] = f"call_{name}_{count}"
                name_counters[name] = count + 1


# ---------------------------------------------------------------------------
# Core wrap / unwrap 函数
# ---------------------------------------------------------------------------


def _generate_stable_session_id(inner_request: dict[str, Any]) -> str:
    """生成稳定的 sessionId（对齐 CLIProxyAPI generateStableSessionID）。

    基于请求内容的哈希生成确定性 sessionId，相同内容的请求会得到相同的 sessionId，
    有助于上游维持会话上下文。
    """
    import hashlib
    import json

    try:
        # 使用 contents 字段生成稳定哈希（与 CLIProxyAPI 对齐）
        contents = inner_request.get("contents")
        if contents:
            raw = json.dumps(contents, sort_keys=True, separators=(",", ":"))
        else:
            raw = json.dumps(inner_request, sort_keys=True, separators=(",", ":"))
        digest = hashlib.sha256(raw.encode()).hexdigest()[:32]
        return f"session-{digest}"
    except Exception:
        return f"session-{uuid.uuid4().hex[:32]}"


def wrap_v1internal_request(
    gemini_request: dict[str, Any],
    *,
    project_id: str,
    model: str,
    request_type: str = "agent",
) -> dict[str, Any]:
    """Wrap a GeminiRequest into Antigravity V1InternalRequest.

    处理流程（对齐 AM wrapper.rs + common_utils.rs）：
    1. 移除 model / safetySettings
    2. 深度清理 [undefined] 字符串
    3. 模型别名映射（preview → physical）
    4. 联网检测 + -online 后缀检测
    5. 图像生成检测 + imageConfig 解析
    6. Claude model tool ID 注入
    7. thoughtSignature 注入到 functionCall parts
    8. Thinking budget 处理
    9. 工具声明清洗
    10. Google Search 注入（联网请求）
    11. System Instruction 注入
    12. 清理空 parts 的 contents 并合并连续同角色条目
    13. 注入 sessionId
    14. 构建 v1internal 信封
    """
    from src.core.video_utils import is_image_gen_model

    inner_request = dict(gemini_request)
    inner_request.pop("model", None)
    inner_request.pop("safetySettings", None)
    inner_request.pop("safety_settings", None)

    # 0. 统一 snake_case → camelCase（Gemini normalizer 输出 snake_case，但
    #    v1internal 以及本模块所有 helper 均使用 camelCase）
    _normalize_to_camel_case(inner_request)

    # 1. 深度清理 [undefined]
    _deep_clean_undefined(inner_request)

    # 2. 剥离 -online 后缀（联网意图由 tools 检测，不依赖后缀传递到上游）
    has_online_suffix = _detect_online_suffix(model)
    final_model = model.rstrip().removesuffix("-online")

    # 3. 联网检测（对齐 AM：-online 后缀或客户端声明了联网工具）
    has_networking = has_online_suffix or _detect_networking_tools(inner_request)

    # 4. 图像生成检测 + imageConfig 解析
    is_image_gen = is_image_gen_model(final_model)

    if is_image_gen:
        # 解析 imageConfig 并确定上游模型名
        image_config, final_model = _parse_image_config(final_model, inner_request)
        _apply_image_gen_config(inner_request, image_config)
        request_type = "image_gen"
        # 图像生成不需要联网
        has_networking = False
    else:
        # 5. Claude tool ID 注入
        _inject_claude_tool_ids_request(inner_request, final_model)

        # 6. thoughtSignature 注入到 functionCall parts（对齐 AM wrapper.rs）
        session_id = inner_request.get("sessionId")
        if not isinstance(session_id, str):
            # 提前生成 sessionId 用于 signature 查找
            session_id = _generate_stable_session_id(inner_request)
            inner_request["sessionId"] = session_id
        _inject_thought_signatures(inner_request, session_id)

    # 7. Thinking budget 处理（图像生成和普通请求都需要）
    _process_thinking_budget(inner_request, final_model)

    if not is_image_gen:
        # 8. 工具声明清洗
        _clean_tool_declarations(inner_request)

        # 9. Google Search 注入（对齐 AM common_utils.rs）
        if has_networking:
            # 仅 gemini-2.5-flash 支持 googleSearch（对齐 AM：其他模型降级到 2.5-flash）
            if final_model != WEB_SEARCH_MODEL:
                final_model = WEB_SEARCH_MODEL
            _inject_google_search_tool(inner_request)
            request_type = "web_search"

        # 10. System Instruction 注入
        _inject_system_instruction(inner_request)

    # 11. 清理空 parts 的 contents 并合并连续同角色条目
    #     跨格式转换（如 Responses API reasoning 块）可能产生空 parts 的 content，
    #     Gemini API 要求每个 content 至少有一个有效 part，并且严格交替 user/model 角色。
    _compact_contents(inner_request)

    # 12. 注入 sessionId（如果还没有的话）
    if "sessionId" not in inner_request:
        inner_request["sessionId"] = _generate_stable_session_id(inner_request)

    return {
        "project": project_id,
        "requestId": f"agent-{uuid.uuid4()}",
        "request": inner_request,
        "model": final_model,
        "userAgent": ANTIGRAVITY_REQUEST_USER_AGENT,
        "requestType": request_type,
    }


def unwrap_v1internal_response(response: dict[str, Any]) -> dict[str, Any]:
    """Unwrap Antigravity V1InternalResponse into a GeminiResponse-like dict."""
    inner = response.get("response")
    if isinstance(inner, dict):
        unwrapped = dict(inner)
        resp_id = response.get("responseId")
        if resp_id is not None:
            unwrapped["_v1internal_response_id"] = resp_id
        return unwrapped
    return response


def cache_thought_signatures(model: str, response: dict[str, Any]) -> None:
    """Best-effort cache for Antigravity thought signatures.

    同时缓存到 legacy (text) 层和 tool (Layer 1) 层。
    """
    try:
        from src.core.api_format.conversion.thinking_cache import signature_cache
    except Exception:
        return

    try:
        candidates = response.get("candidates")
        if not isinstance(candidates, list):
            return

        for cand in candidates:
            if not isinstance(cand, dict):
                continue
            content = cand.get("content")
            if not isinstance(content, dict):
                continue
            parts = content.get("parts")
            if not isinstance(parts, list):
                continue

            for part in parts:
                if not isinstance(part, dict):
                    continue

                # 缓存 thinking signature（legacy text 层）
                text = part.get("text")
                sig = (
                    part.get("thoughtSignature")
                    or part.get("thought_signature")
                    or part.get("signature")
                )
                if isinstance(text, str) and text and isinstance(sig, str) and sig:
                    signature_cache.cache(model, text, sig)

                # 缓存 tool call signature（Layer 1）
                fc = part.get("functionCall") or part.get("function_call")
                if isinstance(fc, dict) and isinstance(sig, str) and sig:
                    tool_id = fc.get("id")
                    if isinstance(tool_id, str) and tool_id:
                        signature_cache.cache_tool_signature(tool_id, sig)
    except Exception:
        # Never fail request path due to cache issues.
        return


def is_signature_error(status_code: int, error_body: str) -> bool:
    """检测 400 错误是否为 thinking signature 相关错误。

    对齐 AM handlers/common.rs：用于判断是否应该移除 thinking 配置后重试。
    """
    if status_code != 400:
        return False
    return any(kw in error_body for kw in SIGNATURE_ERROR_KEYWORDS)


# ---------------------------------------------------------------------------
# Envelope 类（Provider Hook 接口）
# ---------------------------------------------------------------------------


class AntigravityV1InternalEnvelope:
    """Provider envelope hooks for Antigravity v1internal wrapper."""

    name = "antigravity:v1internal"

    def extra_headers(self) -> dict[str, str] | None:
        from src.services.provider.adapters.antigravity.constants import (
            get_v1internal_extra_headers,
        )

        return get_v1internal_extra_headers()

    def wrap_request(
        self,
        request_body: dict[str, Any],
        *,
        model: str,
        url_model: str | None,
        decrypted_auth_config: dict[str, Any] | None,
    ) -> tuple[dict[str, Any], str | None]:
        from src.core.logger import logger as _envelope_logger

        project_id = (decrypted_auth_config or {}).get("project_id")
        if not isinstance(project_id, str) or not project_id:
            from src.core.exceptions import ProviderNotAvailableException

            raise ProviderNotAvailableException(
                "Antigravity OAuth 配置缺少 project_id，请重新授权",
                provider_name="antigravity",
                upstream_response="missing auth_config.project_id",
            )

        wrapped = wrap_v1internal_request(
            request_body,
            project_id=project_id,
            model=model,
        )

        # Debug: 打印 v1internal 请求的关键字段（不打印完整 body 避免日志过大）
        _envelope_logger.debug(
            "[Antigravity Envelope] model={}, project_id={}, requestType={}, "
            "userAgent={}, has_sessionId={}, has_systemInstruction={}, "
            "has_contents={}, has_generationConfig={}",
            wrapped.get("model"),
            str(wrapped.get("project", ""))[:8] + "...",
            wrapped.get("requestType"),
            wrapped.get("userAgent"),
            "sessionId" in (wrapped.get("request") or {}),
            "systemInstruction" in (wrapped.get("request") or {}),
            "contents" in (wrapped.get("request") or {}),
            "generationConfig" in (wrapped.get("request") or {}),
        )

        # Antigravity's model lives in the request body, not the URL path.
        return wrapped, None

    def unwrap_response(self, data: Any) -> Any:
        if isinstance(data, dict):
            return unwrap_v1internal_response(data)
        return data

    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:
        if isinstance(data, dict):
            # Claude model: 注入 tool call ID（对齐 AM wrapper.rs inject_ids_to_response）
            _inject_claude_tool_ids_response(data, model)
            cache_thought_signatures(model, data)

    def capture_selected_base_url(self) -> str | None:
        return get_selected_base_url()

    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
        if not base_url:
            return
        if 200 <= status_code < 300:
            url_availability.mark_success(base_url)
        elif status_code in (404, 408, 429) or 500 <= status_code < 600:
            url_availability.mark_unavailable(base_url)

    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:  # noqa: ARG002
        if not base_url:
            return
        url_availability.mark_unavailable(base_url)

    def force_stream_rewrite(self) -> bool:
        # Streaming must be rewritten even when endpoint signature matches, because
        # Antigravity wraps chunks in v1internal envelope.
        return True


antigravity_v1internal_envelope = AntigravityV1InternalEnvelope()


__all__ = [
    "AntigravityV1InternalEnvelope",
    "antigravity_v1internal_envelope",
    "cache_thought_signatures",
    "is_signature_error",
    "unwrap_v1internal_response",
    "wrap_v1internal_request",
]