entelix_core/codecs/

gemini.rs

//! `GeminiCodec` — IR ⇄ Google Gemini `generateContent` API
//! (`POST /v1beta/models/{model}:generateContent`,
//!   `POST /v1beta/models/{model}:streamGenerateContent?alt=sse`).
//!
//! Wire format reference:
//! <https://ai.google.dev/api/rest/v1beta/models/generateContent>.
//!
//! Notable mappings:
//!
//! - IR `messages` → `contents: [{role: "user"|"model", parts: [...]}]`.
//!   Gemini uses `"model"` for assistant turns.
//! - IR `system: Option<String>` + IR `Role::System` → top-level
//!   `systemInstruction: { parts: [{ text }] }`.
//! - IR `Role::Tool` → `contents: [{role: "user", parts: [{
//!   functionResponse: { name, response: { ... } } }]}]`. Gemini does
//!   not roundtrip `tool_use_id`; the codec records the `LossyEncode`.
//! - IR `ContentPart::ToolUse` → `parts: [{ functionCall: { name, args } }]`.
//! - IR `tools` → `tools: [{ functionDeclarations: [...] }]`. Each
//!   declaration's `parameters` is dialect-translated from JSON
//!   Schema 2020-12 (`type: [T, "null"]`, `oneOf`/`anyOf` of `{const}`
//!   alternatives, standalone `{const}` literals) into Gemini's
//!   OpenAPI 3.0 subset (`nullable: true`, `enum`) at encode time —
//!   see [`encode_input_schema`] for the three translation rules.
//! - IR `tool_choice` → `toolConfig: { functionCallingConfig: { mode } }`.
//! - Streaming SSE: `data: {...}\n\n` per chunk; each chunk is a full
//!   `GenerateContentResponse` with delta text in `candidates[0].content`.

#![allow(clippy::cast_possible_truncation)]

use bytes::Bytes;
use futures::StreamExt;
use serde_json::{Map, Value, json};

use crate::codecs::codec::{BoxByteStream, BoxDeltaStream, Codec, EncodedRequest};
use crate::error::{Error, Result};
use crate::ir::{
    Capabilities, CitationSource, ContentPart, MediaSource, ModelRequest, ModelResponse,
    ModelWarning, OutputStrategy, ProviderEchoSnapshot, ReasoningEffort, RefusalReason,
    ResponseFormat, Role, SafetyCategory, SafetyLevel, SafetyRating, StopReason, ToolChoice,
    ToolKind, ToolResultContent, Usage,
};
use crate::stream::StreamDelta;

const DEFAULT_MAX_CONTEXT_TOKENS: u32 = 1_000_000;

/// Provider key for [`GeminiCodec`] (and its [`super::VertexGeminiCodec`]
/// composition wrapper) — identifies this vendor's entries in
/// [`ProviderEchoSnapshot`]. The wire shape is identical across AI
/// Studio and Vertex AI; only the routing differs.
const PROVIDER_KEY: &str = "gemini";

/// Wire field name Gemini uses for the opaque thought-signature
/// round-trip token. Vertex AI strictly rejects camelCase
/// (`thoughtSignature`) on encode and requires snake_case; AI Studio
/// accepts both. Encoding always emits snake_case for portability.
/// Decoder accepts both because Gemini servers have historically
/// emitted both spellings depending on transport and version.
const WIRE_THOUGHT_SIGNATURE: &str = "thought_signature";
const WIRE_THOUGHT_SIGNATURE_LEGACY: &str = "thoughtSignature";

/// Extract a Gemini `thought_signature` from a `Part` (or any JSON
/// object that may carry it as a sibling field), accepting both
/// snake_case and the legacy camelCase spelling. Returns the wrapping
/// `ProviderEchoSnapshot` carrier ready to attach to a `ContentPart`.
fn decode_thought_signature(obj: &Value) -> Option<ProviderEchoSnapshot> {
    let sig = obj
        .get(WIRE_THOUGHT_SIGNATURE)
        .or_else(|| obj.get(WIRE_THOUGHT_SIGNATURE_LEGACY)) // silent-fallback-ok: snake_case + camelCase are both valid vendor spellings of the same field — accepting either is the contract, no default injected.
        .and_then(Value::as_str)?;
    Some(ProviderEchoSnapshot::for_provider(
        PROVIDER_KEY,
        WIRE_THOUGHT_SIGNATURE,
        sig.to_owned(),
    ))
}

/// Look up this codec's `thought_signature` payload from a part's
/// echoes. Returns the raw signature string so the caller can stamp
/// it onto the appropriate wire-format object.
fn encode_thought_signature(echoes: &[ProviderEchoSnapshot]) -> Option<&str> {
    ProviderEchoSnapshot::find_in(echoes, PROVIDER_KEY)
        .and_then(|e| e.payload_str(WIRE_THOUGHT_SIGNATURE))
}

/// Stateless codec for the Gemini `generateContent` family of endpoints.
#[derive(Clone, Copy, Debug, Default)]
pub struct GeminiCodec;

impl GeminiCodec {
    /// Create a fresh codec instance.
    pub const fn new() -> Self {
        Self
    }
}

impl Codec for GeminiCodec {
    fn name(&self) -> &'static str {
        PROVIDER_KEY
    }

    fn capabilities(&self, _model: &str) -> Capabilities {
        Capabilities {
            streaming: true,
            tools: true,
            multimodal_image: true,
            multimodal_audio: true,
            multimodal_video: true,
            multimodal_document: true,
            system_prompt: true,
            structured_output: true,
            prompt_caching: true,
            thinking: true,
            citations: true,
            web_search: true,
            computer_use: false,
            max_context_tokens: DEFAULT_MAX_CONTEXT_TOKENS,
        }
    }

    fn encode(&self, request: &ModelRequest) -> Result<EncodedRequest> {
        let (body, warnings) = build_body(request)?;
        finalize_request(&request.model, &body, warnings, false)
    }

    fn encode_streaming(&self, request: &ModelRequest) -> Result<EncodedRequest> {
        let (body, warnings) = build_body(request)?;
        let mut encoded = finalize_request(&request.model, &body, warnings, true)?;
        encoded.headers.insert(
            http::header::ACCEPT,
            http::HeaderValue::from_static("text/event-stream"),
        );
        Ok(encoded.into_streaming())
    }

    fn decode(&self, body: &[u8], warnings_in: Vec<ModelWarning>) -> Result<ModelResponse> {
        let raw: Value = super::codec::parse_response_body(body, "Gemini")?;
        let mut warnings = warnings_in;
        let id = String::new(); // Gemini one-shot responses lack a top-level id
        let model = str_field(&raw, "modelVersion").to_owned();
        let mut usage = decode_usage(raw.get("usageMetadata"));
        // Lift the candidate-scoped safetyRatings onto Usage so consumers
        // see safety on a single canonical channel.
        if let Some(candidate) = raw
            .get("candidates")
            .and_then(Value::as_array)
            .and_then(|a| a.first())
        {
            usage.safety_ratings = decode_safety_ratings(candidate);
        }
        let (content, stop_reason) = decode_candidate(&raw, &mut warnings);
        Ok(ModelResponse {
            id,
            model,
            stop_reason,
            content,
            usage,
            rate_limit: None,
            warnings,
            provider_echoes: Vec::new(),
        })
    }

    fn decode_stream<'a>(
        &'a self,
        bytes: BoxByteStream<'a>,
        warnings_in: Vec<ModelWarning>,
    ) -> BoxDeltaStream<'a> {
        Box::pin(stream_gemini(bytes, warnings_in))
    }
}

// ── body builders ──────────────────────────────────────────────────────────

fn build_body(request: &ModelRequest) -> Result<(Value, Vec<ModelWarning>)> {
    if request.messages.is_empty() && request.system.is_empty() {
        return Err(Error::invalid_request(
            "Gemini generateContent requires at least one message",
        ));
    }
    let mut warnings = Vec::new();
    let (system_text, contents) = encode_messages(request, &mut warnings);

    let mut body = Map::new();
    body.insert("contents".into(), Value::Array(contents));
    if let Some(text) = system_text {
        body.insert(
            "systemInstruction".into(),
            json!({ "parts": [{ "text": text }] }),
        );
    }

    let mut generation_config = Map::new();
    if let Some(t) = request.max_tokens {
        generation_config.insert("maxOutputTokens".into(), json!(t));
    }
    if let Some(t) = request.temperature {
        generation_config.insert("temperature".into(), json!(t));
    }
    if let Some(p) = request.top_p {
        generation_config.insert("topP".into(), json!(p));
    }
    if let Some(k) = request.top_k {
        generation_config.insert("topK".into(), json!(k));
    }
    if !request.stop_sequences.is_empty() {
        generation_config.insert("stopSequences".into(), json!(request.stop_sequences));
    }
    if let Some(format) = &request.response_format {
        encode_gemini_structured_output(format, &mut generation_config, &mut body, &mut warnings)?;
    }
    if let Some(effort) = &request.reasoning_effort {
        encode_gemini_thinking(
            &request.model,
            effort,
            &mut generation_config,
            &mut warnings,
        );
    }
    if !generation_config.is_empty() {
        body.insert("generationConfig".into(), Value::Object(generation_config));
    }
    if !request.tools.is_empty() {
        body.insert("tools".into(), encode_tools(&request.tools, &mut warnings));
        body.insert(
            "toolConfig".into(),
            encode_tool_choice(&request.tool_choice),
        );
    }
    apply_provider_extensions(request, &mut body, &mut warnings);
    Ok((Value::Object(body), warnings))
}

/// Read [`crate::ir::GeminiExt`] and merge each set field into the
/// wire body. `candidate_count` lands inside `generationConfig`,
/// creating the map if `build_body` did not already emit one.
/// Foreign-vendor extensions surface as
/// [`ModelWarning::ProviderExtensionIgnored`].
fn apply_provider_extensions(
    request: &ModelRequest,
    body: &mut Map<String, Value>,
    warnings: &mut Vec<ModelWarning>,
) {
    let ext = &request.provider_extensions;
    // Gemini has no native parallel-tool toggle on the
    // generateContent surface. Surface the lossy snap so the operator
    // sees their `parallel_tool_calls` setting was dropped on the
    // wire instead of debugging a silently-ignored knob.
    if request.parallel_tool_calls.is_some() {
        warnings.push(ModelWarning::LossyEncode {
            field: "parallel_tool_calls".into(),
            detail: "Gemini exposes no parallel-tool toggle — setting dropped".into(),
        });
    }
    if let Some(gemini) = &ext.gemini {
        if !gemini.safety_settings.is_empty() {
            let arr: Vec<Value> = gemini
                .safety_settings
                .iter()
                .map(|o| {
                    json!({
                        "category": o.category,
                        "threshold": o.threshold,
                    })
                })
                .collect();
            body.insert("safetySettings".into(), Value::Array(arr));
        }
        if let Some(n) = gemini.candidate_count {
            let entry = body
                .entry("generationConfig")
                .or_insert_with(|| Value::Object(Map::new()));
            if let Some(map) = entry.as_object_mut() {
                map.insert("candidateCount".into(), json!(n));
            }
        }
        if let Some(name) = &gemini.cached_content {
            body.insert("cachedContent".into(), Value::String(name.clone()));
        }
        if gemini.url_context.is_some() {
            // Append the parameterless `url_context` tool to the
            // request's tools array (allocate the array if it
            // wasn't populated by `encode_tools`).
            let entry = body
                .entry("tools")
                .or_insert_with(|| Value::Array(Vec::new()));
            if let Some(arr) = entry.as_array_mut() {
                arr.push(json!({ "url_context": {} }));
            }
        }
    }
    if let Some(seed) = request.seed {
        let entry = body
            .entry("generationConfig")
            .or_insert_with(|| Value::Object(Map::new()));
        if let Some(map) = entry.as_object_mut() {
            map.insert("seed".into(), json!(seed));
        }
    }
    if request.end_user_id.is_some() {
        warnings.push(ModelWarning::LossyEncode {
            field: "end_user_id".into(),
            detail: "Gemini has no end-user attribution channel — drop the field".into(),
        });
    }
    if ext.anthropic.is_some() {
        warnings.push(ModelWarning::ProviderExtensionIgnored {
            vendor: "anthropic".into(),
        });
    }
    if ext.openai_chat.is_some() {
        warnings.push(ModelWarning::ProviderExtensionIgnored {
            vendor: "openai_chat".into(),
        });
    }
    if ext.openai_responses.is_some() {
        warnings.push(ModelWarning::ProviderExtensionIgnored {
            vendor: "openai_responses".into(),
        });
    }
    if ext.bedrock.is_some() {
        warnings.push(ModelWarning::ProviderExtensionIgnored {
            vendor: "bedrock".into(),
        });
    }
}

/// Resolve [`OutputStrategy`] and emit the Gemini native
/// `responseJsonSchema` (Native) or a forced-tool surface (Tool).
/// `Auto` resolves to `Native` — Gemini's `responseJsonSchema` is
/// the most direct surface and Gemini 2.5+ always strict-validates.
fn encode_gemini_structured_output(
    format: &ResponseFormat,
    generation_config: &mut Map<String, Value>,
    body: &mut Map<String, Value>,
    warnings: &mut Vec<ModelWarning>,
) -> Result<()> {
    // Operator-supplied `response_format` schemas route through the
    // same strip funnel every advertised tool receives via
    // `SchemaToolAdapter`. Both `responseJsonSchema` (Native, vendor-
    // side validation, but `schemars` envelope keys would still leak
    // operator content to the wire) and the synthetic function
    // declaration's `parameters` (Tool, forced-call) consume the
    // pre-stripped form. The Tool path additionally runs the
    // dialect translation rules so the synthetic decl is no different
    // from a registered function tool on the Gemini wire.
    let stripped = crate::LlmFacingSchema::strip(&format.json_schema.schema);
    let strategy = match format.strategy {
        OutputStrategy::Auto | OutputStrategy::Native => OutputStrategy::Native,
        explicit => explicit,
    };
    match strategy {
        OutputStrategy::Native => {
            generation_config.insert("responseMimeType".into(), json!("application/json"));
            generation_config.insert("responseJsonSchema".into(), stripped);
            if !format.strict {
                warnings.push(ModelWarning::LossyEncode {
                    field: "response_format.strict".into(),
                    detail: "Gemini always strict-validates structured output; \
                         the strict=false request was approximated"
                        .into(),
                });
            }
        }
        OutputStrategy::Tool => {
            // Forced single function call. Gemini wraps tools as
            // `tools[0].functionDeclarations[0]` and `toolConfig`
            // narrows the selection; `mode: "ANY"` +
            // `allowedFunctionNames: [name]` is the canonical
            // forced-call shape.
            let tool_name = format.json_schema.name.clone();
            let parameters =
                encode_input_schema(&stripped, "response_format.json_schema.schema", warnings);
            let synthetic_decl = json!({
                "name": tool_name,
                "description": format!(
                    "Emit the response as a JSON object matching the {tool_name} schema."
                ),
                "parameters": parameters,
            });
            body.insert(
                "tools".into(),
                json!([{
                    "functionDeclarations": [synthetic_decl],
                }]),
            );
            body.insert(
                "toolConfig".into(),
                json!({
                    "functionCallingConfig": {
                        "mode": "ANY",
                        "allowedFunctionNames": [format.json_schema.name],
                    }
                }),
            );
            if !format.strict {
                warnings.push(ModelWarning::LossyEncode {
                    field: "response_format.strict".into(),
                    detail: "Gemini Tool-strategy structured output is always \
                         schema-validated; strict=false was approximated"
                        .into(),
                });
            }
        }
        OutputStrategy::Prompted => {
            return Err(Error::invalid_request(
                "OutputStrategy::Prompted is deferred to entelix 1.1; use \
                 OutputStrategy::Native or OutputStrategy::Tool",
            ));
        }
        OutputStrategy::Auto => unreachable!("Auto resolved above"),
    }
    Ok(())
}

/// Gemini model family detection — 3.x uses `thinkingLevel`
/// (discrete bucket), 2.5 uses `thinkingBudget` (integer token
/// count, with `-1` = auto and `0` = disable on Flash only).
/// Detection by model-string prefix because Gemini's API does not
/// expose a wire signal for "this model accepts which thinking
/// shape".
fn is_gemini_3(model: &str) -> bool {
    model.starts_with("gemini-3")
}

/// Gemini 2.5 Flash accepts `thinkingBudget: 0` to disable thinking;
/// Pro cannot disable. Detection by model-string prefix.
fn is_gemini_25_flash(model: &str) -> bool {
    model.starts_with("gemini-2.5-flash") || model.starts_with("gemini-2.5-flash-lite")
}

/// Translate the cross-vendor [`ReasoningEffort`] knob onto
/// Gemini's `generationConfig.thinkingConfig`. Per:
///
/// 2.5 (`thinkingBudget` integer):
/// - `Off` → `0` (Flash only — Pro emits LossyEncode → `512`)
/// - `Minimal` → `512`
/// - `Low` → `1024`
/// - `Medium` → `8192`
/// - `High` → `24576`
/// - `Auto` → `-1`
/// - `VendorSpecific(s)` — `s` parses as decimal `thinkingBudget`;
///   non-numeric emits LossyEncode → `Medium`.
///
/// 3.x (`thinkingLevel` enum):
/// - `Off` → LossyEncode → `"minimal"` (Gemini 3 cannot disable)
/// - `Minimal/Low/Medium/High` → `"minimal"/"low"/"medium"/"high"`
/// - `Auto` → LossyEncode → `"high"` (no auto bucket)
/// - `VendorSpecific(s)` — literal `thinkingLevel`.
fn encode_gemini_thinking(
    model: &str,
    effort: &ReasoningEffort,
    generation_config: &mut Map<String, Value>,
    warnings: &mut Vec<ModelWarning>,
) {
    let mut thinking_config = Map::new();
    if is_gemini_3(model) {
        let level = match effort {
            ReasoningEffort::Off => {
                warnings.push(ModelWarning::LossyEncode {
                    field: "reasoning_effort".into(),
                    detail: "Gemini 3 cannot disable thinking — snapped to `\"minimal\"`".into(),
                });
                "minimal"
            }
            ReasoningEffort::Minimal => "minimal",
            ReasoningEffort::Low => "low",
            ReasoningEffort::Medium => "medium",
            ReasoningEffort::High => "high",
            ReasoningEffort::Auto => {
                warnings.push(ModelWarning::LossyEncode {
                    field: "reasoning_effort".into(),
                    detail: "Gemini 3 has no `Auto` bucket — snapped to `\"high\"`".into(),
                });
                "high"
            }
            ReasoningEffort::VendorSpecific(literal) => {
                thinking_config.insert("thinkingLevel".into(), Value::String(literal.clone()));
                generation_config.insert("thinkingConfig".into(), Value::Object(thinking_config));
                return;
            }
        };
        thinking_config.insert("thinkingLevel".into(), Value::String(level.into()));
    } else {
        // Gemini 2.5 (default for any non-3.x prefix — falls
        // through cleanly for 2.5 Pro / Flash / Flash-Lite).
        let budget: i32 = match effort {
            ReasoningEffort::Off => {
                if is_gemini_25_flash(model) {
                    0
                } else {
                    warnings.push(ModelWarning::LossyEncode {
                        field: "reasoning_effort".into(),
                        detail: format!(
                            "Gemini 2.5 Pro ({model}) cannot disable thinking — snapped to `512`"
                        ),
                    });
                    512
                }
            }
            ReasoningEffort::Minimal => 512,
            ReasoningEffort::Low => 1024,
            ReasoningEffort::Medium => 8192,
            ReasoningEffort::High => 24576,
            ReasoningEffort::Auto => -1,
            ReasoningEffort::VendorSpecific(literal) => {
                if let Ok(parsed) = literal.parse::<i32>() {
                    parsed
                } else {
                    warnings.push(ModelWarning::LossyEncode {
                        field: "reasoning_effort".into(),
                        detail: format!(
                            "Gemini 2.5 vendor-specific reasoning_effort {literal:?} is not \
                             a numeric thinkingBudget — falling through to `Medium`"
                        ),
                    });
                    8192
                }
            }
        };
        thinking_config.insert("thinkingBudget".into(), json!(budget));
    }
    generation_config.insert("thinkingConfig".into(), Value::Object(thinking_config));
}

fn finalize_request(
    model: &str,
    body: &Value,
    warnings: Vec<ModelWarning>,
    streaming: bool,
) -> Result<EncodedRequest> {
    let bytes = serde_json::to_vec(body)?;
    let path = if streaming {
        format!("/v1beta/models/{model}:streamGenerateContent?alt=sse")
    } else {
        format!("/v1beta/models/{model}:generateContent")
    };
    let mut encoded = EncodedRequest::post_json(path, Bytes::from(bytes));
    encoded.warnings = warnings;
    Ok(encoded)
}

// ── encode helpers ─────────────────────────────────────────────────────────

fn encode_messages(
    request: &ModelRequest,
    warnings: &mut Vec<ModelWarning>,
) -> (Option<String>, Vec<Value>) {
    let mut system_parts: Vec<String> = request
        .system
        .blocks()
        .iter()
        .map(|b| b.text.clone())
        .collect();
    if request.system.any_cached() {
        warnings.push(ModelWarning::LossyEncode {
            field: "system.cache_control".into(),
            detail: "Gemini has no native prompt-cache control on \
                     systemInstruction; block text is concatenated and \
                     the cache directive is dropped"
                .into(),
        });
    }
    let mut contents = Vec::new();

    for (idx, msg) in request.messages.iter().enumerate() {
        match msg.role {
            Role::System => {
                let mut lossy_non_text = false;
                let mut text = String::new();
                for part in &msg.content {
                    if let ContentPart::Text { text: t, .. } = part {
                        text.push_str(t);
                    } else {
                        lossy_non_text = true;
                    }
                }
                if lossy_non_text {
                    warnings.push(ModelWarning::LossyEncode {
                        field: format!("messages[{idx}].content"),
                        detail: "non-text parts dropped from system message (Gemini routes \
                                 system into systemInstruction)"
                            .into(),
                    });
                }
                if !text.is_empty() {
                    system_parts.push(text);
                }
            }
            Role::User => {
                contents.push(json!({
                    "role": "user",
                    "parts": encode_user_parts(&msg.content, warnings, idx),
                }));
            }
            Role::Assistant => {
                contents.push(json!({
                    "role": "model",
                    "parts": encode_assistant_parts(&msg.content, warnings, idx),
                }));
            }
            Role::Tool => {
                contents.push(json!({
                    "role": "user",
                    "parts": encode_tool_response_parts(&msg.content, warnings, idx),
                }));
            }
        }
    }

    let system_text = if system_parts.is_empty() {
        None
    } else {
        Some(system_parts.join("\n\n"))
    };
    (system_text, contents)
}

fn encode_user_parts(
    parts: &[ContentPart],
    warnings: &mut Vec<ModelWarning>,
    msg_idx: usize,
) -> Vec<Value> {
    let mut out = Vec::new();
    for (part_idx, part) in parts.iter().enumerate() {
        let path = || format!("messages[{msg_idx}].content[{part_idx}]");
        match part {
            ContentPart::Text { text, .. } => out.push(json!({ "text": text })),
            ContentPart::Image { source, .. } => out.push(encode_media_gemini(source, "image/*")),
            ContentPart::Audio { source, .. } => out.push(encode_media_gemini(source, "audio/wav")),
            ContentPart::Video { source, .. } => out.push(encode_media_gemini(source, "video/mp4")),
            ContentPart::Document { source, .. } => {
                out.push(encode_media_gemini(source, "application/pdf"));
            }
            ContentPart::Thinking { .. } => warnings.push(ModelWarning::LossyEncode {
                field: path(),
                detail: "Gemini does not accept thinking blocks on input; block dropped".into(),
            }),
            ContentPart::Citation { .. } => warnings.push(ModelWarning::LossyEncode {
                field: path(),
                detail: "Gemini does not echo citations on input; block dropped".into(),
            }),
            ContentPart::ToolUse { .. } | ContentPart::ToolResult { .. } => {
                warnings.push(ModelWarning::LossyEncode {
                    field: path(),
                    detail: "tool_use / tool_result not allowed on user role for Gemini".into(),
                });
            }
            ContentPart::ImageOutput { .. } | ContentPart::AudioOutput { .. } => {
                warnings.push(ModelWarning::LossyEncode {
                    field: path(),
                    detail: "Gemini does not accept assistant-produced image / audio output \
                             as input — block dropped"
                        .into(),
                });
            }
            ContentPart::RedactedThinking { .. } => {
                warnings.push(ModelWarning::LossyEncode {
                    field: path(),
                    detail: "Gemini does not accept redacted_thinking blocks; block dropped".into(),
                });
            }
        }
    }
    out
}

fn encode_media_gemini(source: &MediaSource, fallback_mime: &str) -> Value {
    match source {
        MediaSource::Base64 { media_type, data } => json!({
            "inlineData": { "mimeType": media_type, "data": data },
        }),
        MediaSource::Url { url, media_type } => {
            let mime = media_type.as_deref().unwrap_or(fallback_mime); // silent-fallback-ok: caller-supplied fallback_mime is the typed MediaSource defaulting policy
            json!({
                "fileData": { "mimeType": mime, "fileUri": url },
            })
        }
        MediaSource::FileId { id, media_type } => {
            let mime = media_type.as_deref().unwrap_or(fallback_mime); // silent-fallback-ok: caller-supplied fallback_mime is the typed MediaSource defaulting policy
            json!({
                "fileData": { "mimeType": mime, "fileUri": id },
            })
        }
    }
}

fn encode_assistant_parts(
    parts: &[ContentPart],
    warnings: &mut Vec<ModelWarning>,
    msg_idx: usize,
) -> Vec<Value> {
    let mut out = Vec::new();
    for (part_idx, part) in parts.iter().enumerate() {
        let path = || format!("messages[{msg_idx}].content[{part_idx}]");
        match part {
            ContentPart::Text {
                text,
                provider_echoes,
                ..
            } => {
                let mut o = Map::new();
                o.insert("text".into(), Value::String(text.clone()));
                if let Some(sig) = encode_thought_signature(provider_echoes) {
                    o.insert(WIRE_THOUGHT_SIGNATURE.into(), Value::String(sig.to_owned()));
                }
                out.push(Value::Object(o));
            }
            ContentPart::ToolUse {
                name,
                input,
                provider_echoes,
                ..
            } => {
                // Gemini's wire shape uses the assistant-emitted function name
                // as the round-trip key — there is no separate id field. The
                // `tool_use_id` round-trip is preserved at the IR layer by
                // letting the codec re-derive the id on decode from the same
                // `name + args` shape.
                let mut o = Map::new();
                o.insert(
                    "functionCall".into(),
                    json!({ "name": name, "args": input }),
                );
                if let Some(sig) = encode_thought_signature(provider_echoes) {
                    o.insert(WIRE_THOUGHT_SIGNATURE.into(), Value::String(sig.to_owned()));
                }
                out.push(Value::Object(o));
            }
            ContentPart::Thinking {
                text,
                provider_echoes,
                ..
            } => {
                let mut o = Map::new();
                o.insert("text".into(), Value::String(text.clone()));
                o.insert("thought".into(), Value::Bool(true));
                if let Some(sig) = encode_thought_signature(provider_echoes) {
                    o.insert(WIRE_THOUGHT_SIGNATURE.into(), Value::String(sig.to_owned()));
                }
                out.push(Value::Object(o));
            }
            ContentPart::Citation { snippet, .. } => out.push(json!({ "text": snippet })),
            other => {
                warnings.push(ModelWarning::LossyEncode {
                    field: path(),
                    detail: format!(
                        "{} not supported on model role for Gemini — dropped",
                        debug_part_kind(other)
                    ),
                });
            }
        }
    }
    out
}

fn encode_tool_response_parts(
    parts: &[ContentPart],
    warnings: &mut Vec<ModelWarning>,
    msg_idx: usize,
) -> Vec<Value> {
    let mut out = Vec::new();
    for (part_idx, part) in parts.iter().enumerate() {
        if let ContentPart::ToolResult {
            tool_use_id: _,
            name,
            content,
            is_error,
            ..
        } = part
        {
            let response_value = match content {
                ToolResultContent::Json(v) => v.clone(),
                ToolResultContent::Text(t) => json!({ "text": t }),
            };
            // Gemini's `functionResponse` keys correlation by
            // `name`, not by id. The IR carries the original name on
            // `ContentPart::ToolResult` precisely so this codec can
            // emit it verbatim — no placeholder, no LossyEncode.
            out.push(json!({
                "functionResponse": {
                    "name": name,
                    "response": response_value,
                },
            }));
            if *is_error {
                warnings.push(ModelWarning::LossyEncode {
                    field: format!("messages[{msg_idx}].content[{part_idx}].is_error"),
                    detail: "Gemini has no functionResponse error flag — passing through content"
                        .into(),
                });
            }
        } else {
            warnings.push(ModelWarning::LossyEncode {
                field: format!("messages[{msg_idx}].content[{part_idx}]"),
                detail: "non-tool_result part on Role::Tool dropped".into(),
            });
        }
    }
    out
}

fn encode_tools(tools: &[crate::ir::ToolSpec], warnings: &mut Vec<ModelWarning>) -> Value {
    let mut declarations = Vec::new();
    let mut tool_entries: Vec<Value> = Vec::new();
    for (idx, t) in tools.iter().enumerate() {
        match &t.kind {
            ToolKind::Function { input_schema } => {
                let parameters = encode_input_schema(
                    input_schema,
                    &format!("tools[{idx}].input_schema"),
                    warnings,
                );
                declarations.push(json!({
                    "name": t.name,
                    "description": t.description,
                    "parameters": parameters,
                }));
            }
            ToolKind::WebSearch { .. } => {
                // Gemini's google_search built-in is parameterless — domain
                // restrictions and use caps are not exposed on the wire.
                tool_entries.push(json!({ "google_search": {} }));
            }
            ToolKind::CodeExecution => {
                // Gemini's code_execution built-in: a sandboxed Python REPL
                // the model invokes autonomously when a turn benefits from
                // computation. Parameterless on the wire.
                tool_entries.push(json!({ "code_execution": {} }));
            }
            // Anthropic / OpenAI vendor built-ins have no Gemini equivalent.
            ToolKind::Computer { .. }
            | ToolKind::TextEditor
            | ToolKind::Bash
            | ToolKind::FileSearch { .. }
            | ToolKind::CodeInterpreter
            | ToolKind::ImageGeneration
            | ToolKind::McpConnector { .. }
            | ToolKind::Memory => warnings.push(ModelWarning::LossyEncode {
                field: format!("tools[{idx}]"),
                detail: "Gemini natively ships google_search and code_execution — other \
                         vendor built-ins (computer, text_editor, file_search, …) have no \
                         Gemini equivalent; tool dropped"
                    .into(),
            }),
        }
    }
    if !declarations.is_empty() {
        tool_entries.insert(0, json!({ "functionDeclarations": declarations }));
    }
    Value::Array(tool_entries)
}

fn encode_tool_choice(choice: &ToolChoice) -> Value {
    let mode = match choice {
        ToolChoice::Auto => "AUTO",
        // Gemini's "ANY" forces a tool call; for `Specific` we additionally
        // narrow via `allowedFunctionNames` below.
        ToolChoice::Required | ToolChoice::Specific { .. } => "ANY",
        ToolChoice::None => "NONE",
    };
    let mut config = json!({ "functionCallingConfig": { "mode": mode } });
    if let ToolChoice::Specific { name } = choice
        && let Some(cfg) = config
            .get_mut("functionCallingConfig")
            .and_then(Value::as_object_mut)
    {
        cfg.insert("allowedFunctionNames".into(), json!([name]));
    }
    config
}

/// Translate an IR JSON Schema (a `serde_json::Value` produced by
/// `LlmFacingSchema::strip` over the schemars output) into Gemini's
/// OpenAPI 3.0 subset dialect.
///
/// Gemini's `functionDeclarations[].parameters` does **not** accept
/// three JSON Schema 2020-12 keywords that OpenAI / Anthropic / Bedrock
/// consume natively. The codec rewrites each at encode time:
///
/// - **Rule A — nullable array form.** `type: [T, "null"]` (`schemars`
///   for `Option<T>`) → `type: T` plus a sibling `nullable: true`.
///   Gemini's protobuf schema types `type` as a scalar, so any array
///   on that field fails with `Proto field is not repeating, cannot
///   start list`.
/// - **Rule B — `oneOf`/`anyOf` of `{const}` alternatives.** The shape
///   `schemars` emits for Rust enum variants — bare `{const}` (no
///   doc comment) or `{const, description, type}` (variant with a
///   doc comment, the dominant production shape) → a single schema
///   with `type` + `enum`. Per-variant `description` folds into the
///   parent's `description` as a Markdown bullet list so the model
///   retains per-value docstrings.
/// - **Rule C — standalone `const` literal.** Any schema-walk
///   position carrying `const` outside a `oneOf`/`anyOf` collapse
///   target — typically the discriminator property of an
///   internally-tagged enum (`#[serde(tag = "kind")] enum E { A {…} }`
///   emits each variant's `kind` field as `{"type": "string",
///   "const": "<variant>"}`) → `{type, enum: [const_value]}`.
///
/// Translations are vendor-specific — OpenAI / Anthropic / Bedrock
/// accept the JSON Schema 2020-12 forms verbatim — so they live
/// inside this codec rather than in `LlmFacingSchema::strip` (which
/// stays vendor-agnostic and only strips envelope-meta keys every
/// vendor ignores).
///
/// Inputs Gemini cannot represent (degenerate `type: ["null"]`,
/// multi-type unions, mixed-type const disjunctions, type/const
/// disagreement, …) surface through [`ModelWarning::LossyEncode`]
/// so the operator sees the coercion rather than chasing the
/// vendor's 400 down a stack trace.
///
/// The walk dispatches in three phases per schema node:
/// 1. Rule B collapse takes the whole node when the parent is a pure
///    const disjunction.
/// 2. Rule C takes the whole node when `const` appears alone.
/// 3. Otherwise the default key walk handles `type` (Rule A) and
///    recurses into structural keys via [`walk_schema_key`].
fn encode_input_schema(schema: &Value, path: &str, warnings: &mut Vec<ModelWarning>) -> Value {
    let Some(obj) = schema.as_object() else {
        // Boolean schema (`additionalProperties: false`, `items: true`)
        // or any non-object leaf — clone through unchanged. The
        // schema-walk targets only object nodes.
        return schema.clone();
    };

    if let Some(collapsed) = try_collapse_const_alternatives(obj, path, warnings) {
        return collapsed.emit(obj, path, warnings);
    }

    if obj.contains_key("const") && !obj.contains_key("enum") {
        return translate_const_literal(obj, path, warnings);
    }

    let mut out = Map::new();
    for (key, value) in obj {
        if key == "type" {
            apply_type_rule(value, path, warnings, &mut out);
            continue;
        }
        out.insert(key.clone(), walk_schema_key(key, value, path, warnings));
    }
    Value::Object(out)
}

/// Walk a single non-`type` schema key. Structural keys
/// (`properties`, `items`, `additionalProperties`, `not`, `anyOf` /
/// `oneOf` / `allOf`) recurse into [`encode_input_schema`] so the
/// dialect rules apply at every depth; everything else (`description`,
/// `enum`, `default`, `required`, `format`, bounds, vendor metadata)
/// clones through verbatim because `LlmFacingSchema::strip` already
/// removed envelope-meta the model should not see.
///
/// Shared by `encode_input_schema`'s main walk, `ConstCollapse::emit`'s
/// sibling walk, and `translate_const_literal`'s post-prologue walk —
/// new key support (or a key reclassification when Gemini's dialect
/// expands) lands in one place.
fn walk_schema_key(
    key: &str,
    value: &Value,
    path: &str,
    warnings: &mut Vec<ModelWarning>,
) -> Value {
    match key {
        "properties" => walk_properties(value, path, warnings),
        "items" => match value {
            // `items` is either a single schema (homogeneous array)
            // or an array of schemas (tuple validation — Gemini does
            // not accept this, but the walk preserves whatever was
            // there so the vendor's rejection carries the original
            // shape rather than our re-encoded guess).
            Value::Array(arr) => Value::Array(
                arr.iter()
                    .enumerate()
                    .map(|(i, v)| encode_input_schema(v, &format!("{path}.items[{i}]"), warnings))
                    .collect(),
            ),
            other => encode_input_schema(other, &format!("{path}.items"), warnings),
        },
        "additionalProperties" | "not" => {
            encode_input_schema(value, &format!("{path}.{key}"), warnings)
        }
        "anyOf" | "oneOf" | "allOf" => walk_schema_array(value, path, key, warnings),
        _ => value.clone(),
    }
}

/// Outcome of the `oneOf` / `anyOf` const-collapse rule. Built only
/// when every alternative is a `{const: X}` object — possibly
/// decorated with the per-variant doc keys `schemars 1.x` emits for
/// Rust enum variants (`description`, `title`) and a redundant
/// `type` consistent with the const's scalar shape. Other shapes
/// (struct-variant alternatives with `properties`, mixed-type
/// consts, non-scalar consts) decline collapse.
struct ConstCollapse {
    /// The disjunction keyword (`"oneOf"` or `"anyOf"`) that was
    /// collapsed. Recorded so the emit step knows which input key to
    /// drop on its way to the output map.
    source_key: &'static str,
    /// Inferred Gemini OpenAPI 3.0 scalar type for the collapsed
    /// `enum`. Drives the emitted `type` field.
    scalar_type: &'static str,
    /// The collected `const` values, in source order, ready to land
    /// inside the emitted `enum` array.
    values: Vec<Value>,
    /// Per-alternative `description` field, in source order — `None`
    /// when that alternative omitted the key. Folded into the parent
    /// schema's `description` so the model retains the per-value
    /// docstring `schemars` generated from each Rust enum variant's
    /// doc comment.
    descriptions: Vec<Option<String>>,
}

impl ConstCollapse {
    /// Emit the collapsed schema. Drops the source disjunction key
    /// and the input's sibling `type` (collapse-inferred type wins).
    /// Per-alternative descriptions fold into the parent's
    /// `description`; the original parent description (if any) leads
    /// and the per-value lines follow.
    fn emit(self, obj: &Map<String, Value>, path: &str, warnings: &mut Vec<ModelWarning>) -> Value {
        let parent_desc = obj
            .get("description")
            .and_then(Value::as_str)
            .map(str::to_owned);
        let alt_block = self.fold_alternative_descriptions();

        let mut out = Map::new();
        out.insert("type".into(), Value::String(self.scalar_type.to_owned()));
        out.insert("enum".into(), Value::Array(self.values));
        if let Some(description) = merge_descriptions(parent_desc, alt_block) {
            out.insert("description".into(), Value::String(description));
        }

        for (key, value) in obj {
            if matches!(
                key.as_str(),
                "oneOf" | "anyOf" | "type" | "enum" | "description"
            ) {
                if key == "type"
                    && let Some(sibling) = value.as_str()
                    && sibling != self.scalar_type
                {
                    // Operator-supplied `type` disagrees with the
                    // const-inferred one — surface the discarded
                    // value so an inconsistent schema does not
                    // silently coerce.
                    warnings.push(ModelWarning::LossyEncode {
                        field: format!("{path}.type"),
                        detail: format!(
                            "schema mixes `{source}` of consts (inferred type `{inferred}`) \
                             with a sibling `type: \"{sibling}\"` — the inferred type is \
                             authoritative",
                            source = self.source_key,
                            inferred = self.scalar_type,
                        ),
                    });
                }
                continue;
            }
            out.insert(key.clone(), walk_schema_key(key, value, path, warnings));
        }
        Value::Object(out)
    }

    /// Render the per-alternative descriptions as a single Markdown
    /// bullet block. `None` when every alternative omitted the key
    /// (nothing to fold).
    fn fold_alternative_descriptions(&self) -> Option<String> {
        if self.descriptions.iter().all(Option::is_none) {
            return None;
        }
        let mut lines: Vec<String> = Vec::with_capacity(self.values.len());
        for (value, desc) in self.values.iter().zip(self.descriptions.iter()) {
            let value_str = value
                .as_str()
                .map_or_else(|| value.to_string(), str::to_owned);
            match desc {
                Some(text) => lines.push(format!("- `{value_str}`: {text}")),
                None => lines.push(format!("- `{value_str}`")),
            }
        }
        Some(format!("Values:\n{}", lines.join("\n")))
    }
}

/// Join the parent schema's original `description` (if any) with the
/// collapsed alternatives' folded description block (if any). `None`
/// only when neither source had a description.
fn merge_descriptions(parent: Option<String>, alternatives: Option<String>) -> Option<String> {
    match (parent, alternatives) {
        (Some(p), Some(a)) => Some(format!("{p}\n\n{a}")),
        (Some(p), None) => Some(p),
        (None, Some(a)) => Some(a),
        (None, None) => None,
    }
}

/// Map a JSON value to its Gemini OpenAPI 3.0 scalar `type` name.
/// `Number` splits by `is_i64()` / `is_u64()` so int-only consts emit
/// `"integer"` and float consts emit `"number"`. Non-scalar values
/// (object, array, null) return `None` — Gemini's enum acceptance is
/// not documented for non-scalars, so the caller decides whether to
/// proceed or decline.
fn infer_scalar_type(value: &Value) -> Option<&'static str> {
    match value {
        Value::String(_) => Some("string"),
        Value::Bool(_) => Some("boolean"),
        Value::Number(n) if n.is_i64() || n.is_u64() => Some("integer"),
        Value::Number(_) => Some("number"),
        _ => None,
    }
}

/// Translate a schema carrying a standalone `const` literal into the
/// equivalent single-value `enum` shape. Gemini's OpenAPI 3.0 subset
/// rejects `const`; `enum: [X]` is the lossless equivalent every
/// JSON Schema validator already treats identically.
///
/// Fires on any schema-walk position whose object has `const` and
/// lacks `enum` — typically the discriminator property of an
/// internally-tagged enum (`#[serde(tag = "kind")]` → schemars emits
/// each variant's `kind` field as `{"type":"string","const":"<name>"}`).
///
/// Sibling `type` is honoured when consistent with the const's
/// inferred scalar type; inconsistent sibling types coerce to the
/// inferred type and emit `LossyEncode` so an operator-buggy schema
/// surfaces rather than silently coercing.
fn translate_const_literal(
    obj: &Map<String, Value>,
    path: &str,
    warnings: &mut Vec<ModelWarning>,
) -> Value {
    let const_value = obj["const"].clone();
    let inferred = infer_scalar_type(&const_value);
    let sibling_type = obj.get("type").and_then(Value::as_str);

    let mut out = Map::new();
    match (sibling_type, inferred) {
        (Some(sibling), Some(inferred_type)) if sibling != inferred_type => {
            warnings.push(ModelWarning::LossyEncode {
                field: format!("{path}.type"),
                detail: format!(
                    "standalone `const` inferred type `{inferred_type}` disagrees \
                     with sibling `type: \"{sibling}\"` — using inferred"
                ),
            });
            out.insert("type".into(), Value::String((*inferred_type).to_owned()));
        }
        (Some(sibling), _) => {
            out.insert("type".into(), Value::String(sibling.to_owned()));
        }
        (None, Some(inferred_type)) => {
            out.insert("type".into(), Value::String((*inferred_type).to_owned()));
        }
        (None, None) => {
            // Non-scalar const with no operator-supplied `type` — the
            // codec emits a bare `enum: [<value>]` with no type signal,
            // because there is no scalar type to infer. Surface the
            // missing-type situation so the operator sees what the
            // codec produced; the vendor's response to a typeless
            // enum is the vendor's call to make.
            warnings.push(ModelWarning::LossyEncode {
                field: format!("{path}.const"),
                detail: "non-scalar `const` (object/array/null) lacks an explicit \
                         sibling `type`; emitted as bare `enum: [<value>]` with no \
                         type signal"
                    .into(),
            });
        }
    }
    out.insert("enum".into(), Value::Array(vec![const_value]));

    // Walk the remaining keys — `const`, `type`, and `enum` were
    // absorbed above.
    for (key, value) in obj {
        if matches!(key.as_str(), "const" | "type" | "enum") {
            continue;
        }
        out.insert(key.clone(), walk_schema_key(key, value, path, warnings));
    }
    Value::Object(out)
}

/// Try to collapse a `oneOf` / `anyOf` array of const alternatives
/// into a single `type` + `enum`. Returns `None` when any condition
/// for safe collapse is unmet — the caller then walks the
/// disjunction as a normal schema array.
///
/// Conditions (all must hold):
/// 1. The parent has either `oneOf` or `anyOf` (but not both).
/// 2. The disjunction has at least one alternative.
/// 3. The parent does not carry structural object-shape keys
///    (`properties` / `items` / `additionalProperties` / `not` /
///    `allOf`). Mixing `oneOf:[{const}]` with object-shape keys is
///    an operator anti-pattern that cannot be safely collapsed —
///    the resulting `{type:"string", enum:[…], properties:{…}}`
///    would be structurally contradictory.
/// 4. The parent does not already carry a competing value-spec
///    (`enum` or standalone `const`) that the collapse would
///    silently override — those signal an operator-buggy schema
///    whose original intent must reach the vendor unaltered.
/// 5. Every alternative is an object whose key set is a subset of
///    `{const, description, title, type}`. The `const` key is
///    required; the others are the doc-decoration `schemars 1.x`
///    emits for documented Rust enum variants. Any other key
///    (e.g. `properties` in struct-variant enums) declines.
/// 6. Every `const` value is the same JSON scalar shape (string /
///    integer / float / bool). Mixed-type, object, array, and null
///    consts decline.
/// 7. When an alternative carries a sibling `type`, it must match
///    the const's inferred scalar type. Inconsistency declines —
///    the schema is operator-buggy and should surface as the
///    vendor's rejection rather than a silent coercion.
fn try_collapse_const_alternatives(
    obj: &Map<String, Value>,
    path: &str,
    warnings: &mut Vec<ModelWarning>,
) -> Option<ConstCollapse> {
    if obj.contains_key("enum") || obj.contains_key("const") {
        return None;
    }
    if STRUCTURAL_OBJECT_KEYS.iter().any(|k| obj.contains_key(*k)) {
        return None;
    }
    let (source_key, alternatives) = match (obj.get("oneOf"), obj.get("anyOf")) {
        (Some(Value::Array(arr)), None) => ("oneOf", arr),
        (None, Some(Value::Array(arr))) => ("anyOf", arr),
        _ => return None,
    };
    if alternatives.is_empty() {
        return None;
    }
    let mut values: Vec<Value> = Vec::with_capacity(alternatives.len());
    let mut descriptions: Vec<Option<String>> = Vec::with_capacity(alternatives.len());
    let mut inferred: Option<&'static str> = None;
    for alt in alternatives {
        let alt_obj = alt.as_object()?;
        let const_value = alt_obj.get("const")?;
        let Some(alt_type) = infer_scalar_type(const_value) else {
            // Object / array / null consts: conservative skip.
            return decline_collapse(source_key, path, warnings);
        };
        // Verify the alternative's key set is the allowed subset.
        // Any unexpected key signals a richer shape (struct-variant
        // enum, vendor extension, …) that cannot be safely folded.
        for key in alt_obj.keys() {
            if !matches!(key.as_str(), "const" | "description" | "title" | "type") {
                return decline_collapse(source_key, path, warnings);
            }
        }
        // If the alternative carries a sibling `type`, it must match
        // the const-inferred scalar type — otherwise the operator's
        // schema is internally inconsistent and the collapse would
        // silently pick one side.
        if let Some(sibling_type) = alt_obj.get("type").and_then(Value::as_str)
            && sibling_type != alt_type
        {
            return decline_collapse(source_key, path, warnings);
        }
        match inferred {
            None => inferred = Some(alt_type),
            Some(existing) if existing == alt_type => {}
            // Number promotes integer → number when mixed integer + float.
            Some("integer") if alt_type == "number" => inferred = Some("number"),
            Some("number") if alt_type == "integer" => {}
            Some(_) => return decline_collapse(source_key, path, warnings),
        }
        values.push(const_value.clone());
        descriptions.push(
            alt_obj
                .get("description")
                .and_then(Value::as_str)
                .map(str::to_owned),
        );
    }
    let scalar_type = inferred?;
    Some(ConstCollapse {
        source_key,
        scalar_type,
        values,
        descriptions,
    })
}

/// Keys whose presence on a parent schema means the schema is
/// structurally object-shaped, not enum-shaped — collapsing a
/// `oneOf` of consts in that context would produce a contradictory
/// `{type:"string", enum:[…], properties:{…}}`. Used by
/// [`try_collapse_const_alternatives`].
const STRUCTURAL_OBJECT_KEYS: &[&str] = &[
    "properties",
    "items",
    "additionalProperties",
    "not",
    "allOf",
];

/// Helper used by `try_collapse_const_alternatives` when the
/// disjunction *looks* like a const-only collapse target but a
/// structural condition fails (decorated alternative, type-mixed
/// consts, object const, …). Emits one `LossyEncode` so the operator
/// sees the pattern was recognised and intentionally not collapsed.
/// The warning reports the codec's decision; the disjunction is
/// still walked node-by-node, so individual const alternatives
/// downstream may still translate via Rule C — predicting the
/// vendor's final acceptance is the vendor's job, not the codec's.
fn decline_collapse(
    source_key: &'static str,
    path: &str,
    warnings: &mut Vec<ModelWarning>,
) -> Option<ConstCollapse> {
    warnings.push(ModelWarning::LossyEncode {
        field: format!("{path}.{source_key}"),
        detail: format!(
            "`{source_key}` of consts cannot collapse to a single `type` + `enum` — \
             alternatives carry extra keys, mixed JSON types, or non-scalar \
             const values; walking the disjunction node-by-node instead"
        ),
    });
    None
}

/// Apply Rule A (`type: [T, "null"]` → `type: T` + `nullable: true`)
/// to a `type` field's value, writing the resulting key(s) into the
/// output map. Inputs Gemini cannot represent (`type: ["null"]`,
/// empty array, multi-type unions, non-string/array scalar) surface
/// through `LossyEncode` and are coerced toward the closest Gemini-
/// representable shape; the standalone scalar case passes through.
fn apply_type_rule(
    value: &Value,
    path: &str,
    warnings: &mut Vec<ModelWarning>,
    out: &mut Map<String, Value>,
) {
    match value {
        Value::String(_) => {
            out.insert("type".into(), value.clone());
        }
        Value::Array(types) => {
            let mut non_null: Vec<&str> = Vec::with_capacity(types.len());
            let mut saw_null = false;
            let mut malformed = false;
            for entry in types {
                match entry.as_str() {
                    Some("null") => saw_null = true,
                    Some(other) => non_null.push(other),
                    None => malformed = true,
                }
            }
            if malformed {
                warnings.push(ModelWarning::LossyEncode {
                    field: format!("{path}.type"),
                    detail: "type array contains a non-string entry; Gemini will reject".into(),
                });
                out.insert("type".into(), value.clone());
                return;
            }
            match (non_null.as_slice(), saw_null) {
                ([], false) => {
                    warnings.push(ModelWarning::LossyEncode {
                        field: format!("{path}.type"),
                        detail: "empty `type` array; Gemini will reject".into(),
                    });
                    out.insert("type".into(), value.clone());
                }
                ([], true) => {
                    // Degenerate input — a field whose only valid value
                    // is `null` has no representation in Gemini's
                    // OpenAPI 3.0 subset. Pass the original `type` key
                    // through so Vertex's rejection identifies the
                    // operator's actual field rather than a synthesized
                    // typeless `nullable: true` substitute (which would
                    // itself be invalid).
                    warnings.push(ModelWarning::LossyEncode {
                        field: format!("{path}.type"),
                        detail: "type containing only `null` is degenerate; \
                                 Gemini OpenAPI 3.0 subset has no representation"
                            .into(),
                    });
                    out.insert("type".into(), value.clone());
                }
                ([single], saw_null) => {
                    out.insert("type".into(), Value::String((*single).to_owned()));
                    if saw_null {
                        out.insert("nullable".into(), Value::Bool(true));
                    }
                }
                ([first, rest @ ..], saw_null) => {
                    let detail = if saw_null {
                        format!(
                            "Gemini does not accept multi-type unions; coerced to \
                             nullable first scalar `{first}` (dropped: {rest:?})"
                        )
                    } else {
                        format!(
                            "Gemini does not accept multi-type unions; coerced to \
                             first scalar `{first}` (dropped: {rest:?})"
                        )
                    };
                    warnings.push(ModelWarning::LossyEncode {
                        field: format!("{path}.type"),
                        detail,
                    });
                    out.insert("type".into(), Value::String((*first).to_owned()));
                    if saw_null {
                        out.insert("nullable".into(), Value::Bool(true));
                    }
                }
            }
        }
        _ => {
            warnings.push(ModelWarning::LossyEncode {
                field: format!("{path}.type"),
                detail: "type field is neither a string nor an array; Gemini will reject".into(),
            });
            out.insert("type".into(), value.clone());
        }
    }
}

/// Recurse into a `properties` map — preserve user-named keys
/// verbatim, walk each value as a schema.
fn walk_properties(value: &Value, path: &str, warnings: &mut Vec<ModelWarning>) -> Value {
    let Value::Object(map) = value else {
        return value.clone();
    };
    let walked: Map<String, Value> = map
        .iter()
        .map(|(k, v)| {
            let child_path = format!("{path}.properties.{k}");
            (k.clone(), encode_input_schema(v, &child_path, warnings))
        })
        .collect();
    Value::Object(walked)
}

/// Recurse into a `oneOf` / `anyOf` / `allOf` array — every element
/// is itself a schema. The collapse rule has already declined (the
/// caller is in the non-collapse branch), so this is a vanilla walk.
fn walk_schema_array(
    value: &Value,
    path: &str,
    key: &str,
    warnings: &mut Vec<ModelWarning>,
) -> Value {
    let Value::Array(arr) = value else {
        return value.clone();
    };
    let walked: Vec<Value> = arr
        .iter()
        .enumerate()
        .map(|(i, v)| {
            let child_path = format!("{path}.{key}[{i}]");
            encode_input_schema(v, &child_path, warnings)
        })
        .collect();
    Value::Array(walked)
}

const fn debug_part_kind(part: &ContentPart) -> &'static str {
    match part {
        ContentPart::Text { .. } => "text",
        ContentPart::Image { .. } => "image",
        ContentPart::Audio { .. } => "audio",
        ContentPart::Video { .. } => "video",
        ContentPart::Document { .. } => "document",
        ContentPart::Thinking { .. } => "thinking",
        ContentPart::Citation { .. } => "citation",
        ContentPart::ToolUse { .. } => "tool_use",
        ContentPart::ToolResult { .. } => "tool_result",
        ContentPart::ImageOutput { .. } => "image_output",
        ContentPart::AudioOutput { .. } => "audio_output",
        ContentPart::RedactedThinking { .. } => "redacted_thinking",
    }
}

// ── decode helpers ─────────────────────────────────────────────────────────

fn decode_candidate(
    raw: &Value,
    warnings: &mut Vec<ModelWarning>,
) -> (Vec<ContentPart>, StopReason) {
    let candidate = raw
        .get("candidates")
        .and_then(Value::as_array)
        .and_then(|a| a.first())
        .cloned()
        .unwrap_or(Value::Null); // silent-fallback-ok: response with no candidates array → Null (downstream nested accessors propagate as None)
    let parts_raw = candidate
        .get("content")
        .and_then(|c| c.get("parts"))
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default(); // silent-fallback-ok: candidate with no parts array → empty Vec (downstream loop iterates over zero items)
    let mut parts = Vec::new();
    // Per-response counter of `functionCall` parts seen so far —
    // synthesized into the tool-use id so streaming and
    // non-streaming decoders produce the same id sequence
    // (`{name}#{tool_seq}`) for the same logical sequence of tool
    // calls. Using the part-array index directly would diverge:
    // non-streaming sees `[text, fnCall, text, fnCall]` at indices
    // 1, 3 while streaming would emit them as 0, 1.
    let mut tool_seq: usize = 0;
    for (idx, part) in parts_raw.iter().enumerate() {
        // Thinking blocks: parts marked `thought: true` carry reasoning text.
        if part.get("thought").and_then(Value::as_bool) == Some(true) {
            let text = str_field(part, "text").to_owned();
            let provider_echoes = decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
            parts.push(ContentPart::Thinking {
                text,
                cache_control: None,
                provider_echoes,
            });
            continue;
        }
        if let Some(text) = part.get("text").and_then(Value::as_str)
            && !text.is_empty()
        {
            // Plain `text` parts may also carry `thought_signature`
            // on reasoning turns — preserve it for round-trip.
            let provider_echoes = decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
            parts.push(ContentPart::Text {
                text: text.to_owned(),
                cache_control: None,
                provider_echoes,
            });
            continue;
        }
        if let Some(call) = part.get("functionCall") {
            let name = str_field(call, "name").to_owned();
            let args = call.get("args").cloned().unwrap_or_else(|| json!({})); // silent-fallback-ok: functionCall without args = empty-args call (vendor sometimes omits when schema has no required fields)
            // `thought_signature` rides on the `Part` itself
            // (sibling of `functionCall`), not inside the inner
            // object — Gemini 3.x rejects the next turn with HTTP
            // 400 if the first `functionCall` of a step is missing
            // its echo.
            let provider_echoes = decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
            // Gemini does not round-trip a tool-use id — derive one
            // from `(name, tool_seq)` where `tool_seq` is a per-
            // response counter of function-call parts. Streaming
            // decoder uses the identical counter, so the same
            // logical sequence of tool calls produces the same id
            // sequence regardless of code path.
            parts.push(ContentPart::ToolUse {
                id: format!("{name}#{tool_seq}"),
                name,
                input: args,
                provider_echoes,
            });
            tool_seq = tool_seq.saturating_add(1);
            continue;
        }
        warnings.push(ModelWarning::LossyEncode {
            field: format!("candidates[0].content.parts[{idx}]"),
            detail: "unknown Gemini part type dropped".into(),
        });
    }
    // Grounding metadata → Citation parts.
    if let Some(meta) = candidate.get("groundingMetadata")
        && let Some(chunks) = meta.get("groundingChunks").and_then(Value::as_array)
    {
        for chunk in chunks {
            if let Some(web) = chunk.get("web") {
                let url = str_field(web, "uri").to_owned();
                let title = web.get("title").and_then(Value::as_str).map(str::to_owned);
                if !url.is_empty() {
                    parts.push(ContentPart::Citation {
                        snippet: title.clone().unwrap_or_default(), // silent-fallback-ok: grounding citation without title → snippet "" (the URL is the load-bearing pointer; title is purely descriptive)
                        source: CitationSource::Url { url, title },
                        cache_control: None,
                        provider_echoes: Vec::new(),
                    });
                }
            }
        }
    }
    let stop_reason = decode_finish_reason(
        candidate.get("finishReason").and_then(Value::as_str),
        warnings,
    );
    (parts, stop_reason)
}

fn decode_finish_reason(reason: Option<&str>, warnings: &mut Vec<ModelWarning>) -> StopReason {
    match reason {
        Some("STOP") => StopReason::EndTurn,
        Some("MAX_TOKENS") => StopReason::MaxTokens,
        // Gemini distinguishes safety blocks from copyright
        // (RECITATION) blocks — preserve the distinction in IR so
        // dashboards can split by cause instead of collapsing both
        // into a single refusal bucket.
        Some("SAFETY") => StopReason::Refusal {
            reason: RefusalReason::Safety,
        },
        Some("RECITATION") => StopReason::Refusal {
            reason: RefusalReason::Recitation,
        },
        Some(other) => {
            warnings.push(ModelWarning::UnknownStopReason {
                raw: other.to_owned(),
            });
            StopReason::Other {
                raw: other.to_owned(),
            }
        }
        None => {
            // Invariant #15 — silent EndTurn fallback was masking
            // truncated stream payloads from callers. Record as
            // Other + warning instead.
            warnings.push(ModelWarning::LossyEncode {
                field: "finishReason".into(),
                detail: "Gemini candidate carried no finishReason — \
                         IR records `Other{raw:\"missing\"}`"
                    .into(),
            });
            StopReason::Other {
                raw: "missing".to_owned(),
            }
        }
    }
}

fn decode_usage(usage: Option<&Value>) -> Usage {
    // Cross-vendor `Usage::output_tokens` invariant — *total billable
    // output*. Gemini reports the visible-token slice in
    // `candidatesTokenCount` and bills thinking tokens separately
    // (vendor docs: "response pricing is the sum of output tokens
    // and thinking tokens"). OpenAI / Anthropic already include
    // their reasoning slices inside `output_tokens` / `completion_tokens`,
    // so the codec aligns Gemini to the same shape. `reasoning_tokens`
    // remains as an informational sub-counter operators can isolate
    // for thinking-only cost attribution.
    let visible = u_field(usage, "candidatesTokenCount");
    let thoughts = u_field(usage, "thoughtsTokenCount");
    Usage {
        input_tokens: u_field(usage, "promptTokenCount"),
        output_tokens: visible.saturating_add(thoughts),
        cached_input_tokens: u_field(usage, "cachedContentTokenCount"),
        cache_creation_input_tokens: 0,
        reasoning_tokens: thoughts,
        safety_ratings: Vec::new(),
    }
}

fn decode_safety_ratings(candidate: &Value) -> Vec<SafetyRating> {
    let Some(raw) = candidate.get("safetyRatings").and_then(Value::as_array) else {
        return Vec::new();
    };
    raw.iter()
        .filter_map(|r| {
            let category = match r.get("category").and_then(Value::as_str)? {
                "HARM_CATEGORY_HARASSMENT" => SafetyCategory::Harassment,
                "HARM_CATEGORY_HATE_SPEECH" => SafetyCategory::HateSpeech,
                "HARM_CATEGORY_SEXUALLY_EXPLICIT" => SafetyCategory::SexuallyExplicit,
                "HARM_CATEGORY_DANGEROUS_CONTENT" => SafetyCategory::DangerousContent,
                other => SafetyCategory::Other(other.to_owned()),
            };
            let level = match r.get("probability").and_then(Value::as_str)? {
                "LOW" => SafetyLevel::Low,
                "MEDIUM" => SafetyLevel::Medium,
                "HIGH" => SafetyLevel::High,
                // `"NEGLIGIBLE"` and any unrecognised vendor label collapse
                // to the lowest bucket — the IR's four-bucket scale is the
                // canonical resolution.
                _ => SafetyLevel::Negligible,
            };
            Some(SafetyRating { category, level })
        })
        .collect()
}

fn str_field<'a>(v: &'a Value, key: &str) -> &'a str {
    v.get(key).and_then(Value::as_str).unwrap_or("") // silent-fallback-ok: missing optional string field
}

fn u_field(v: Option<&Value>, key: &str) -> u32 {
    v.and_then(|inner| inner.get(key))
        .and_then(Value::as_u64)
        .map_or(0, |n| u32::try_from(n).unwrap_or(u32::MAX)) // silent-fallback-ok: missing usage metric = 0 (vendor didn't report = unused); u64→u32 saturate
}

// ── SSE streaming parser ───────────────────────────────────────────────────

#[allow(tail_expr_drop_order, clippy::too_many_lines)]
fn stream_gemini(
    bytes: BoxByteStream<'_>,
    warnings_in: Vec<ModelWarning>,
) -> impl futures::Stream<Item = Result<StreamDelta>> + Send + '_ {
    async_stream::stream! {
        let mut bytes = bytes;
        let mut buf: Vec<u8> = Vec::new();
        let mut started = false;
        let mut warnings_emitted = false;
        let mut last_stop = StopReason::EndTurn;
        let mut current_tool_open = false;
        let mut tool_synth_idx: u64 = 0;

        while let Some(chunk) = bytes.next().await {
            match chunk {
                Ok(b) => buf.extend_from_slice(&b),
                Err(e) => {
                    yield Err(e);
                    return;
                }
            }
            if !warnings_emitted {
                warnings_emitted = true;
                for w in &warnings_in {
                    yield Ok(StreamDelta::Warning(w.clone()));
                }
            }
            while let Some(pos) = find_double_newline(&buf) {
                let frame: Vec<u8> = buf.drain(..pos.saturating_add(2)).collect();
                let Ok(frame_str) = std::str::from_utf8(&frame) else {
                    continue;
                };
                let Some(payload) = parse_sse_data(frame_str) else {
                    continue;
                };
                let Ok(event) = serde_json::from_str::<Value>(&payload) else {
                    yield Err(Error::invalid_request(format!(
                        "Gemini stream: malformed chunk: {payload}"
                    )));
                    return;
                };
                if !started {
                    started = true;
                    let model = str_field(&event, "modelVersion").to_owned();
                    yield Ok(StreamDelta::Start {
                        id: String::new(),
                        model,
                        provider_echoes: Vec::new(),
                    });
                }
                if let Some(usage) = event.get("usageMetadata") {
                    yield Ok(StreamDelta::Usage(decode_usage(Some(usage))));
                }
                let Some(candidate) = event
                    .get("candidates")
                    .and_then(Value::as_array)
                    .and_then(|a| a.first())
                else {
                    continue;
                };
                if let Some(reason) = candidate.get("finishReason").and_then(Value::as_str) {
                    last_stop = decode_finish_reason(Some(reason), &mut Vec::new());
                }
                let Some(parts) = candidate
                    .get("content")
                    .and_then(|c| c.get("parts"))
                    .and_then(Value::as_array)
                else {
                    continue;
                };
                for part in parts {
                    // Thinking branches first so a `text` payload marked
                    // `thought: true` routes to ThinkingDelta rather than
                    // TextDelta.
                    if part.get("thought").and_then(Value::as_bool) == Some(true) {
                        if current_tool_open {
                            yield Ok(StreamDelta::ToolUseStop);
                            current_tool_open = false;
                        }
                        let text = part
                            .get("text")
                            .and_then(Value::as_str)
                            .unwrap_or("") // silent-fallback-ok: missing thinking text → empty body; downstream is_empty() guard suppresses the StreamDelta
                            .to_owned();
                        let provider_echoes =
                            decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
                        if !text.is_empty() || !provider_echoes.is_empty() {
                            yield Ok(StreamDelta::ThinkingDelta {
                                text,
                                provider_echoes,
                            });
                        }
                        continue;
                    }
                    if let Some(text) = part.get("text").and_then(Value::as_str)
                        && !text.is_empty()
                    {
                        if current_tool_open {
                            yield Ok(StreamDelta::ToolUseStop);
                            current_tool_open = false;
                        }
                        let provider_echoes =
                            decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
                        yield Ok(StreamDelta::TextDelta {
                            text: text.to_owned(),
                            provider_echoes,
                        });
                        continue;
                    }
                    if let Some(call) = part.get("functionCall") {
                        if current_tool_open {
                            yield Ok(StreamDelta::ToolUseStop);
                        }
                        let name = str_field(call, "name").to_owned();
                        let args = call.get("args").cloned().unwrap_or_else(|| json!({})); // silent-fallback-ok: streaming functionCall without args = empty-args call
                        let synth_id = format!("{name}#{tool_synth_idx}");
                        tool_synth_idx = tool_synth_idx.saturating_add(1);
                        let provider_echoes =
                            decode_thought_signature(part).map_or_else(Vec::new, |e| vec![e]);
                        yield Ok(StreamDelta::ToolUseStart {
                            id: synth_id,
                            name,
                            provider_echoes,
                        });
                        yield Ok(StreamDelta::ToolUseInputDelta {
                            partial_json: args.to_string(),
                        });
                        current_tool_open = true;
                    }
                }
            }
        }
        if current_tool_open {
            yield Ok(StreamDelta::ToolUseStop);
        }
        yield Ok(StreamDelta::Stop {
            stop_reason: last_stop,
        });
    }
}

fn find_double_newline(buf: &[u8]) -> Option<usize> {
    let lf = buf.windows(2).position(|w| w == b"\n\n");
    let crlf = buf.windows(4).position(|w| w == b"\r\n\r\n");
    match (lf, crlf) {
        (Some(a), Some(b)) => Some(a.min(b)),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    }
}

fn parse_sse_data(frame: &str) -> Option<String> {
    let mut out: Option<String> = None;
    for line in frame.lines() {
        if let Some(rest) = line.strip_prefix("data:") {
            let trimmed = rest.strip_prefix(' ').unwrap_or(rest); // silent-fallback-ok: SSE data line may or may not have leading space; idiomatic strip-or-pass-through
            match &mut out {
                Some(existing) => {
                    existing.push('\n');
                    existing.push_str(trimmed);
                }
                None => out = Some(trimmed.to_owned()),
            }
        }
    }
    out
}