agentplane 0.34.0

//! A `ModelProvider` for the Google Gemini Developer API.
//!
//! Behind the `gemini` feature. Speaks `generateContent` and
//! `streamGenerateContent` directly rather than going through Google's
//! OpenAI-compatibility endpoint, and the reason is the one thing this wire has
//! that the compatible one cannot carry cleanly.
//!
//! # Why not the compatibility endpoint
//!
//! Gemini is reachable as `.../v1beta/openai/chat/completions`, so the
//! `chat-completions` driver *does* reach it. What it cannot reach is a
//! first-class contract:
//!
//! * **Reasoning is refused there.** The compatible wire has no
//!   model-family-neutral spelling for it, so a declared `reasoning_effort`
//!   is refused rather than silently dropped — which means the control a
//!   manifest declares does not apply to Gemini at all.
//! * **Governed media is refused there**, for the same reason: multimodal
//!   content is a per-server dialect on that wire.
//! * **Structured output falls back to a forced tool**, because whether a
//!   compatible server honours `json_schema` is exactly what cannot be
//!   assumed. Gemini enforces a schema natively during generation.
//!
//! And one thing that is not a matter of degree. Gemini's thinking models
//! attach an encrypted **thought signature** to the parts they emit, and
//! **reject** a follow-up turn that does not carry it back. Through the
//! compatible endpoint it rides in
//! `tool_calls[].extra_content.google.thought_signature` — a place a driver
//! normalising every provider into one shape has nowhere to keep. The
//! ecosystem has the scars: `LiteLLM` ended up smuggling the signature inside the
//! tool-call *id*, which then leaked into requests to other providers and still
//! degenerates multi-turn tool calling when the signature arrives on a thought
//! part rather than a function-call part.
//!
//! This driver avoids that class of bug by never being in a position to have
//! it: the continuation is the model's `content` **verbatim**, opaque to the
//! runtime, exactly as the `OpenAI` driver carries encrypted reasoning items and
//! the Anthropic driver carries signed thinking blocks. There is no field to
//! know about, so there is no field to lose.
//!
//! # Which surface, and why this one
//!
//! Google's **Interactions API** is generally available and is what it
//! recommends for new work. This driver deliberately targets `generateContent`,
//! and the reason is an invariant rather than inertia: the Interactions API's
//! defining feature is server-side conversation state addressed by
//! `previous_interaction_id`, retained for 55 days on the paid tier and **one
//! day** on the free one. Provider-held conversation state cannot be replay
//! truth here — a run replayed after the retention window would have nothing to
//! replay against. A deployment would therefore have to set `store=false`,
//! which removes the API's own advantage and, at the time of writing, is the
//! mode whose thought-signature handling Google does not document.
//!
//! `generateContent` is stateless by construction, carries the signature
//! explicitly for the client to return, and remains fully supported. Revisit
//! when the Interactions API documents stateless multi-turn function calling —
//! not merely when it gains features, because features behind server-side state
//! are features this runtime cannot use.
//!
//! # What this driver deliberately does not send
//!
//! `GenerationConfig` has knobs this driver never touches, and the omissions
//! are decisions rather than gaps:
//!
//! * **`temperature`, `topP`, `topK`, `seed`** — no sampling parameter is sent
//!   by any driver here. They are absent from [`Request`], so they could not
//!   enter the effect key, and a knob that changes what the provider does
//!   without changing effect identity is one a replay cannot account for.
//!   `seed` is the tempting one and is the clearest case: replay here never
//!   calls the model again, so a seed buys nothing, and sending one would imply
//!   a reproducibility guarantee no provider actually makes.
//! * **`candidateCount`** — left at its default of one. The runtime journals
//!   one completion per effect; asking for several answers and picking one
//!   would make the choice an unrecorded step.
//! * **`responseModalities`, `speechConfig`, `mediaResolution`** — image and
//!   audio *output*. A completion is text plus tool calls; generated bytes
//!   would need blob storage, a digest in the chain and a retention unit, which
//!   is a feature rather than a field.
//! * **`stopSequences`** — no seam declares them.
//!
//! Two fields are chosen rather than merely used. Schemas go in
//! **`responseJsonSchema`**, not `responseSchema`: the latter is Gemini's
//! trimmed OpenAPI-subset dialect, and translating a caller's JSON Schema into
//! it would mean the effect key records one shape while the wire carries
//! another — the quiet divergence this crate refuses everywhere. And thinking
//! is requested as **`thinkingLevel`**, the form the Gemini 3 models take,
//! rather than the token-denominated `thinkingBudget` of the 2.5 generation: a
//! level is what [`ReasoningEffort`] means, and converting one into a token
//! count would be this driver inventing a number nobody declared.
//!
//! # The failure table
//!
//! Status classification is shared doctrine, common to every HTTP driver here.
//! What is
//! specific here is the success envelope: a `finishReason` of `MAX_TOKENS` is a
//! truncated answer reported through [`Completion::truncated`] rather than as a
//! silently shortened string; `SAFETY`, `RECITATION` and `PROHIBITED_CONTENT`
//! are **metered** declines, because deciding to stop cost whatever it cost;
//! and a candidate with no parts at all is a loud `Unusable` rather than an
//! empty answer.

use async_trait::async_trait;
use serde_json::{Value, json};

use crate::core::Secret;

use super::wire::{RESPOND_TOOL, classify_status, classify_transport, structured};
use super::{
    Completion, ModelError, ModelId, ModelProvider, ReasoningEffort, Request, SchemaMode, Usage,
    gemini_stream, sse,
};

/// The provider tag a continuation from this driver carries.
pub(crate) const PROVIDER: &str = "gemini";

/// A harm category Gemini can be asked to block.
///
/// Named as an enum rather than taken as a string for the reason every other
/// declared control here is typed: a misspelled category is not an error on
/// this wire, it is a setting that silently governs nothing, and the deployment
/// that wrote it would believe the opposite.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[non_exhaustive]
pub enum HarmCategory {
    Harassment,
    HateSpeech,
    SexuallyExplicit,
    DangerousContent,
    CivicIntegrity,
    Jailbreak,
}

impl HarmCategory {
    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Harassment => "HARM_CATEGORY_HARASSMENT",
            Self::HateSpeech => "HARM_CATEGORY_HATE_SPEECH",
            Self::SexuallyExplicit => "HARM_CATEGORY_SEXUALLY_EXPLICIT",
            Self::DangerousContent => "HARM_CATEGORY_DANGEROUS_CONTENT",
            Self::CivicIntegrity => "HARM_CATEGORY_CIVIC_INTEGRITY",
            Self::Jailbreak => "HARM_CATEGORY_JAILBREAK",
        }
    }
}

/// How much of a category to block.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[non_exhaustive]
pub enum HarmBlockThreshold {
    LowAndAbove,
    MediumAndAbove,
    OnlyHigh,
    /// Block nothing in this category.
    ///
    /// Present because a deployment may genuinely need it — a content-moderation
    /// agent has to be able to read what it moderates — and because expressing
    /// it *explicitly* is better than the alternative. An absent setting is the
    /// provider's default, which Google may change; `None` is a decision, and it
    /// is in the request profile where a reviewer and a replay can both see it.
    None,
}

impl HarmBlockThreshold {
    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::LowAndAbove => "BLOCK_LOW_AND_ABOVE",
            Self::MediumAndAbove => "BLOCK_MEDIUM_AND_ABOVE",
            Self::OnlyHigh => "BLOCK_ONLY_HIGH",
            Self::None => "BLOCK_NONE",
        }
    }
}

/// The deployment's own safety thresholds, passed through.
///
/// The same posture as the Bedrock guardrail and for the same reasons. This
/// crate ships no content classifier — a deployment that needs one already has
/// a better one, administered where its compliance people can see it — and what
/// the runtime owns is everything *around* it:
///
/// * the settings are **effect identity**, so loosening a threshold between a
///   run and its replay is divergence rather than a silent change in what
///   governed the call;
/// * an intervention is a **metered refusal**, never an answer: a prompt
///   blocked before generation is a `Refused` naming its reason, and a
///   generation stopped by `SAFETY` is `Unusable` carrying the tokens it
///   burned, because deciding to stop cost whatever it cost;
/// * both request paths carry it, since a control the streaming path drops is
///   one a `stream: true` deployment loses.
///
/// Deliberately not a default. An empty set means *the provider's defaults*,
/// which is what every deployment had before this existed, and inventing a
/// house policy here would be this crate deciding a question it has no standing
/// to decide.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SafetySettings {
    /// Ordered by category, so two deployments that configured the same
    /// thresholds in a different order produce the same request bytes — and
    /// therefore the same effect identity, rather than a spurious divergence.
    thresholds: std::collections::BTreeMap<HarmCategory, HarmBlockThreshold>,
}

impl SafetySettings {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Set one category's threshold, replacing any previous one.
    #[must_use]
    pub fn block(mut self, category: HarmCategory, threshold: HarmBlockThreshold) -> Self {
        self.thresholds.insert(category, threshold);
        self
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.thresholds.is_empty()
    }

    /// The `safetySettings` array.
    fn wire(&self) -> Value {
        Value::Array(
            self.thresholds
                .iter()
                .map(|(category, threshold)| {
                    json!({ "category": category.as_str(), "threshold": threshold.as_str() })
                })
                .collect(),
        )
    }

    /// The form that enters the request profile.
    ///
    /// The same pairs the wire carries. A profile that recorded only *whether*
    /// safety was configured would let a threshold move from
    /// `BLOCK_LOW_AND_ABOVE` to `BLOCK_NONE` without changing effect identity,
    /// which is precisely the change worth catching.
    fn profile(&self) -> Value {
        self.wire()
    }
}

/// Calls the Gemini Developer API.
///
/// The key is held here and never journaled — transport metadata in exactly the
/// sense a peer credential is, and the same rule applies: a secret in a
/// hash-chained record cannot be redacted afterwards, only discovered.
pub struct Gemini {
    http: reqwest::Client,
    key: Secret,
    base: String,
    version: String,
    default_schema_mode: SchemaMode,
    schema_modes: std::collections::BTreeMap<String, SchemaMode>,
    stream: bool,
    egress: Option<crate::core::Egress>,
    timeout: std::time::Duration,
    /// The deployment's own safety thresholds, if it declared any.
    safety: SafetySettings,
}

impl std::fmt::Debug for Gemini {
    /// Redacts the key. Deriving `Debug` would print it into every log line and
    /// span that touches the provider.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Gemini")
            .field("base", &self.base)
            .field("version", &self.version)
            .field("key", &"<redacted>")
            .finish_non_exhaustive()
    }
}

impl Gemini {
    pub const DEFAULT_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(5);

    /// The default endpoint.
    pub const DEFAULT_BASE: &'static str = "https://generativelanguage.googleapis.com";

    /// The API version this driver speaks.
    ///
    /// Pinned rather than tracking the newest, for the reason every version
    /// here is pinned: a provider that changes its response shape under a
    /// running plane changes what replay reads back.
    pub const VERSION: &'static str = "v1beta";

    /// Build with a key.
    ///
    /// # Errors
    ///
    /// If the HTTP client cannot be built.
    pub fn new(key: impl Into<String>) -> Result<Self, ModelError> {
        let http = crate::netguard::guarded_client(crate::netguard::Reach::Configured)
            .build()
            .map_err(|e| ModelError::Unreachable {
                model: ModelId::new(PROVIDER, "*"),
                detail: format!("could not build an HTTP client: {e}"),
            })?;
        Ok(Self {
            http,
            key: Secret::new(key),
            base: Self::DEFAULT_BASE.to_owned(),
            version: Self::VERSION.to_owned(),
            // Native: Gemini enforces a response schema during generation, so
            // the forced tool is a fallback rather than the honest default it
            // has to be on a wire whose servers only imitate the shape.
            default_schema_mode: SchemaMode::Native,
            schema_modes: std::collections::BTreeMap::new(),
            stream: true,
            egress: None,
            timeout: Self::DEFAULT_TIMEOUT,
            safety: SafetySettings::new(),
        })
    }

    /// Take the key from `GEMINI_API_KEY`, falling back to `GOOGLE_API_KEY`.
    ///
    /// Both are read because both are in wide use and a deployment that
    /// exported the other one would otherwise get an authentication failure
    /// naming neither.
    ///
    /// # Errors
    ///
    /// If neither variable is set, or the HTTP client cannot be built.
    pub fn from_env() -> Result<Self, ModelError> {
        let key = std::env::var("GEMINI_API_KEY")
            .or_else(|_| std::env::var("GOOGLE_API_KEY"))
            .map_err(|_| ModelError::Refused {
                model: ModelId::new(PROVIDER, "*"),
                detail: "neither GEMINI_API_KEY nor GOOGLE_API_KEY is set".to_owned(),
            })?;
        Self::new(key)
    }

    /// Point at another endpoint — a regional host, or a test double.
    #[must_use]
    pub fn base(mut self, base: impl Into<String>) -> Self {
        let mut base = base.into();
        while base.ends_with('/') {
            base.pop();
        }
        self.base = base;
        self
    }

    /// Bound connection, generation, and response streaming as one operation.
    #[must_use]
    pub const fn timeout(mut self, timeout: std::time::Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// How to obtain a schema-conforming answer from every model.
    #[must_use]
    pub fn structured_via(mut self, mode: SchemaMode) -> Self {
        self.default_schema_mode = mode;
        self
    }

    /// How to obtain a schema-conforming answer from **one** model.
    #[must_use]
    pub fn structured_via_for(mut self, model: impl Into<String>, mode: SchemaMode) -> Self {
        self.schema_modes.insert(model.into(), mode);
        self
    }

    /// Apply the deployment's own safety thresholds to every call.
    ///
    /// See [`SafetySettings`] for the posture. In short: this crate ships no
    /// classifier, Google's is configured here, and what the runtime owns is
    /// that the choice is **effect identity** and that an intervention is a
    /// metered refusal rather than an answer.
    #[must_use]
    pub fn safety(mut self, safety: SafetySettings) -> Self {
        self.safety = safety;
        self
    }

    /// Restrict where this driver may connect.
    ///
    /// Deny-by-default once set — see [`Egress`](crate::core::Egress).
    #[must_use]
    pub fn egress(mut self, egress: crate::core::Egress) -> Self {
        self.egress = Some(egress);
        self
    }

    /// Ask for the whole response at once instead of streaming it.
    ///
    /// Streaming is the default for the reason it is elsewhere in this crate:
    /// Gemini reports `usageMetadata` in the stream, so a severed connection can
    /// say what it burned rather than being billed as zero and retried for free.
    #[must_use]
    pub const fn buffered(mut self) -> Self {
        self.stream = false;
        self
    }

    fn mode_for(&self, model: &ModelId) -> SchemaMode {
        self.schema_modes
            .get(&model.model)
            .copied()
            .unwrap_or(self.default_schema_mode)
    }

    fn check_egress(&self, model: &ModelId) -> Result<(), ModelError> {
        let Some(egress) = &self.egress else {
            return Ok(());
        };
        let host = reqwest::Url::parse(&self.base)
            .ok()
            .and_then(|u| u.host_str().map(ToOwned::to_owned));
        egress
            .permits(host.as_deref())
            .map_err(|e| ModelError::Egress {
                model: model.clone(),
                detail: e.to_string(),
            })
    }

    fn refused(model: &ModelId, detail: impl Into<String>) -> ModelError {
        ModelError::Refused {
            model: model.clone(),
            detail: detail.into(),
        }
    }

    /// `thinkingConfig`, or why this effort cannot be asked for.
    ///
    /// Gemini names four levels — `minimal`, `low`, `medium`, `high` — and this
    /// maps the four that match exactly. The rest are **refused rather than
    /// collapsed**: `reasoning_effort` is digest-covered precisely so it
    /// describes what governed a call, and answering a request for `max` with
    /// the highest level that happens to exist is a substitution nothing
    /// downstream could see. The same rule the Anthropic and Bedrock drivers
    /// follow for the levels their providers cannot express.
    ///
    /// `None` is refused rather than mapped to `minimal`, and the distinction
    /// matters: Google documents that thinking **cannot be turned off** on the
    /// Gemini 3 models, so a driver rendering "do not reason" as "reason a
    /// little" would report a control as applied that the provider never
    /// applied.
    fn thinking_config(model: &ModelId, effort: ReasoningEffort) -> Result<Value, ModelError> {
        let level = match effort {
            ReasoningEffort::Minimal => "minimal",
            ReasoningEffort::Low => "low",
            ReasoningEffort::Medium => "medium",
            ReasoningEffort::High => "high",
            ReasoningEffort::None | ReasoningEffort::XHigh | ReasoningEffort::Max => {
                return Err(Self::refused(
                    model,
                    format!(
                        "Gemini has no thinking level for reasoning effort '{}' — it names \
                         minimal, low, medium and high, and thinking cannot be switched off \
                         on the Gemini 3 models",
                        effort.as_str()
                    ),
                ));
            }
        };
        Ok(json!({ "thinkingLevel": level }))
    }

    /// The `contents` array, from whatever shape the caller supplied.
    ///
    /// A bare string is one user turn. An array is already `contents` and is
    /// passed through untouched — which is how governed media reaches this
    /// driver, as `inlineData` parts the caller assembled and the runtime
    /// materialised. An object is read for `messages` only when
    /// `prompt_envelope` says it is an envelope rather than content; anything
    /// else is one user turn carrying the object.
    fn contents(prompt: &Value) -> Value {
        match prompt {
            Value::String(text) => json!([{ "role": "user", "parts": [{ "text": text }] }]),
            Value::Array(_) => prompt.clone(),
            other => crate::model::prompt_envelope(other, "messages")
                .cloned()
                .unwrap_or_else(|| {
                    // `system` is an instruction *about* the content and is lifted
                    // out separately; leaving it here would show the model its own
                    // orders as part of the question.
                    let mut rest = other.clone();
                    if let Some(map) = rest.as_object_mut() {
                        map.remove("system");
                    }
                    json!([{ "role": "user", "parts": [{ "text": rest.to_string() }] }])
                }),
        }
    }

    /// The system instruction, if the caller set one.
    ///
    /// Gemini takes this as a **top-level** `systemInstruction`, not a role
    /// inside `contents`. A driver that left it in the turns would show the
    /// model an instruction it treats as ordinary content — and the instruction
    /// slot is authority-bearing here, so that is not a cosmetic difference.
    fn system_instruction(prompt: &Value) -> Option<Value> {
        let system = prompt.get("system").filter(|s| !s.is_null())?;
        Some(match system {
            Value::String(text) => json!({ "parts": [{ "text": text }] }),
            // Already a `Content` — passed through, so a caller can supply
            // several parts.
            other => other.clone(),
        })
    }

    /// The request body.
    fn body(&self, model: &ModelId, request: &Request<'_>) -> Result<Value, ModelError> {
        let Request {
            prompt,
            max_output_tokens,
            reasoning_effort,
            schema,
            tools,
            exchanges,
            continuation,
            ..
        } = request;

        let mut contents = Self::contents(prompt);
        Self::append_tool_turns(&mut contents, exchanges, *continuation, model)?;

        let mut generation_config = json!({ "maxOutputTokens": max_output_tokens });
        if let Some(effort) = reasoning_effort {
            generation_config["thinkingConfig"] = Self::thinking_config(model, *effort)?;
        }

        let mut body = json!({ "contents": contents });
        // Both request paths get it from here, because both build their body
        // here — the Bedrock guardrail had to say this twice and that is the
        // shape where only the half nobody exercises is wrong.
        if !self.safety.is_empty() {
            body["safetySettings"] = self.safety.wire();
        }
        if let Some(system) = Self::system_instruction(prompt) {
            body["systemInstruction"] = system;
        }

        let mode = self.mode_for(model);
        let mut declarations: Vec<Value> = tools
            .iter()
            .map(|t| {
                json!({
                    "name": t.name,
                    "description": t.description,
                    // `parametersJsonSchema` rather than `parameters`: the
                    // latter is Gemini's own trimmed Schema dialect, and
                    // handing it a full JSON Schema is how a valid declaration
                    // becomes a 400 nobody can read.
                    "parametersJsonSchema": t.parameters,
                })
            })
            .collect();

        if let Some(schema) = schema {
            match mode {
                SchemaMode::Native => {
                    generation_config["responseMimeType"] = json!("application/json");
                    generation_config["responseJsonSchema"] = (*schema).clone();
                }
                SchemaMode::ForcedTool => {
                    if !tools.is_empty() {
                        return Err(Self::refused(
                            model,
                            "forced-tool structured output cannot be combined with declared \
                             tools: the model would be offered a choice between answering and \
                             calling one. Use SchemaMode::Native, which Gemini enforces during \
                             generation",
                        ));
                    }
                    declarations.push(json!({
                        "name": RESPOND_TOOL,
                        "description": "Return the answer in the required shape.",
                        "parametersJsonSchema": (*schema).clone(),
                    }));
                    body["toolConfig"] = json!({
                        "functionCallingConfig": {
                            "mode": "ANY",
                            "allowedFunctionNames": [RESPOND_TOOL],
                        }
                    });
                }
            }
        }

        if !declarations.is_empty() {
            body["tools"] = json!([{ "functionDeclarations": declarations }]);
        }
        body["generationConfig"] = generation_config;
        Ok(body)
    }

    /// Append the model turn that asked for tools, and the results.
    ///
    /// The model turn is the continuation **verbatim**. That is the whole
    /// reason this driver exists rather than deferring to the compatible wire:
    /// a part may carry a `thoughtSignature`, Gemini rejects a follow-up turn
    /// that does not return it, and a driver reconstructing the turn from the
    /// fields it understands cannot return what it never kept.
    fn append_tool_turns(
        contents: &mut Value,
        exchanges: &[super::ToolExchange],
        continuation: Option<&super::ProviderContinuation>,
        model: &ModelId,
    ) -> Result<(), ModelError> {
        if exchanges.is_empty() {
            if continuation.is_some() {
                // Silently dropping it would journal an effect key that
                // records a continuation the wire never carried.
                return Err(Self::refused(
                    model,
                    "a continuation without tool exchanges has no request to follow",
                ));
            }
            return Ok(());
        }
        let Some(array) = contents.as_array_mut() else {
            return Err(Self::refused(
                model,
                "the prompt did not assemble into a `contents` array",
            ));
        };

        match continuation {
            Some(state) if state.provider == PROVIDER => match state.state.as_array() {
                Some(turns) => array.extend(turns.iter().cloned()),
                None => {
                    return Err(Self::refused(
                        model,
                        "the continuation was not a Gemini contents array",
                    ));
                }
            },
            Some(other) => {
                return Err(Self::refused(
                    model,
                    format!(
                        "the continuation was issued by '{}' and this is the Gemini driver — \
                         provider state is opaque and is never valid across providers",
                        other.provider
                    ),
                ));
            }
            // No continuation: a caller assembled the exchanges by hand. The
            // turn is rebuilt, and it is worth naming what that costs, because
            // it is the failure this driver exists to avoid — a rebuilt turn
            // carries no thought signature, and a thinking model will refuse
            // it. Callers who ran the model through this driver always have
            // one.
            None => array.push(json!({
                "role": "model",
                "parts": exchanges
                    .iter()
                    .map(|e| json!({
                        "functionCall": { "name": e.call.name, "args": e.call.arguments }
                    }))
                    .collect::<Vec<_>>(),
            })),
        }

        array.push(Self::tool_responses(exchanges));
        Ok(())
    }

    /// The results turn: one user content of `functionResponse` parts, in the
    /// order the calls were made.
    fn tool_responses(exchanges: &[super::ToolExchange]) -> Value {
        json!({
            "role": "user",
            "parts": exchanges
                .iter()
                .map(|e| {
                    let mut response = json!({ "name": e.call.name, "response": {
                        // Gemini expects an object; a bare value is wrapped
                        // rather than sent as-is, because a string here is a
                        // 400 rather than a smaller answer.
                        "output": e.output,
                    }});
                    if e.failed {
                        response["response"] = json!({ "error": e.output });
                    }
                    json!({ "functionResponse": response })
                })
                .collect::<Vec<_>>(),
        })
    }

    /// Extend the provider transcript after a successful tool-calling
    /// response, exactly as every sibling driver does: prior rounds, then this
    /// round's results, then the turn just answered. The runtime clears
    /// `exchanges` each turn and relies on this being the whole history.
    fn accumulate_continuation(
        completion: &mut Completion,
        prior: Option<&super::ProviderContinuation>,
        exchanges: &[super::ToolExchange],
    ) {
        let Some(current) = completion.continuation.as_mut() else {
            return;
        };
        let mut state = prior
            .and_then(|value| value.state.as_array())
            .cloned()
            .unwrap_or_default();
        if !exchanges.is_empty() {
            state.push(Self::tool_responses(exchanges));
        }
        if let Some(turns) = current.state.as_array() {
            state.extend(turns.iter().cloned());
        }
        current.state = Value::Array(state);
    }

    /// Turn a response envelope into a [`Completion`], or say why it is not one.
    ///
    /// Shared by the buffered and streaming paths, so the two cannot disagree
    /// about what a usable answer is.
    fn interpret(
        &self,
        parsed: &Value,
        model: &ModelId,
        schema: Option<&Value>,
    ) -> Result<Completion, ModelError> {
        let usage = Self::usage(parsed);

        let Some(candidate) = parsed.get("candidates").and_then(|c| c.get(0)) else {
            // No candidate at all: either the prompt was blocked before
            // generating, which the feedback names, or the response is one this
            // driver cannot read. Both are refusals; only the first can say why.
            let detail = parsed
                .get("promptFeedback")
                .and_then(|f| f.get("blockReason"))
                .and_then(Value::as_str)
                .map_or_else(
                    || "the response carried no candidates".to_owned(),
                    |reason| format!("the prompt was blocked before generating: {reason}"),
                );
            return Err(Self::refused(model, detail));
        };

        let finish = candidate
            .get("finishReason")
            .and_then(Value::as_str)
            .map(ToOwned::to_owned);
        let truncated = finish.as_deref() == Some("MAX_TOKENS");

        // It generated and then stopped for a reason that is not an answer.
        // Metered, because deciding cost whatever it cost — the same rule the
        // Anthropic driver applies to a `refusal` stop reason.
        if let Some(reason) = finish.as_deref()
            && matches!(
                reason,
                "SAFETY" | "RECITATION" | "PROHIBITED_CONTENT" | "BLOCKLIST" | "SPII"
            )
        {
            return Err(ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: format!("generation stopped: {reason}"),
            });
        }

        let content = candidate.get("content").cloned().unwrap_or(Value::Null);
        let parts = content
            .get("parts")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();

        let Scanned {
            text,
            calls,
            forced,
        } = Self::scan(&parts);

        let emulating = schema.is_some() && self.mode_for(model) == SchemaMode::ForcedTool;
        if text.is_empty() && calls.is_empty() && forced.is_none() && !truncated {
            return Err(ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: format!("the answer carried no content (finishReason {finish:?})"),
            });
        }

        let (text, structured_value) = if emulating {
            let Some(arguments) = forced else {
                return Err(ModelError::Unusable {
                    model: model.clone(),
                    usage,
                    detail: "a tool call was forced and the answer carried none — the model \
                             did not honour the function-calling config"
                        .to_owned(),
                });
            };
            let raw = arguments.to_string();
            let parsed_schema = structured(schema, &raw, &[], model, usage)?;
            (raw, parsed_schema)
        } else {
            let parsed_schema = structured(schema, &text, &calls, model, usage)?;
            (text, parsed_schema)
        };

        // The model's own turn, byte for byte, so the next request returns
        // every signature exactly where it sat. An array of contents, not the
        // bare turn: the state must carry *every* prior round, or round
        // three's request forgets round one's signed turn — silently, since
        // the model simply re-asks with amnesia.
        let continuation = (!calls.is_empty() && content.is_object())
            .then(|| super::ProviderContinuation::new(PROVIDER, json!([content.clone()])));

        Ok(Completion {
            structured: structured_value,
            tool_calls: calls,
            text,
            usage,
            stop_reason: finish,
            truncated,
            continuation,
        })
    }

    /// Usage, in this crate's terms.
    ///
    /// Two normalisations, both of which cost real money when got wrong.
    /// `thoughtsTokenCount` is billed as **output** and is reported *beside*
    /// `candidatesTokenCount` rather than inside it, so it is added — a
    /// reasoning-heavy run would otherwise under-report its bill by most of it.
    /// `cachedContentTokenCount` is a **subset** of `promptTokenCount`, as
    /// `OpenAI` reports cached input, so it is recorded rather than added.
    fn usage(parsed: &Value) -> Usage {
        let count = |key: &str| {
            parsed
                .get("usageMetadata")
                .and_then(|u| u.get(key))
                .and_then(Value::as_u64)
                .unwrap_or_default()
        };
        Usage {
            input_tokens: count("promptTokenCount"),
            output_tokens: count("candidatesTokenCount") + count("thoughtsTokenCount"),
            cache_read_tokens: count("cachedContentTokenCount"),
            cache_write_tokens: 0,
            minor_units: 0,
        }
    }

    fn url(&self, model: &ModelId) -> String {
        let method = if self.stream {
            "streamGenerateContent?alt=sse"
        } else {
            "generateContent"
        };
        format!(
            "{}/{}/models/{}:{method}",
            self.base, self.version, model.model
        )
    }

    async fn read_buffered(
        &self,
        response: reqwest::Response,
        model: &ModelId,
        schema: Option<&Value>,
    ) -> Result<Completion, ModelError> {
        // Read under this plane's ceiling rather than to end-of-stream: a
        // provider is a counterparty, and a counterparty must not decide how
        // much of this process's memory its answer costs.
        let body = crate::netguard::intake::read(response, crate::netguard::intake::ANSWER)
            .await
            .map_err(|e| super::wire::classify_intake(model, Usage::default(), &e))?;
        let parsed: Value = serde_json::from_slice(&body).map_err(|e| ModelError::Unusable {
            model: model.clone(),
            usage: Usage::default(),
            detail: format!("the response did not parse: {e}"),
        })?;
        self.interpret(&parsed, model, schema)
    }

    async fn read_streamed(
        &self,
        response: reqwest::Response,
        model: &ModelId,
        schema: Option<&Value>,
        observer: Option<(&dyn super::ModelStreamObserver, &crate::core::Label)>,
    ) -> Result<Completion, ModelError> {
        use futures_util::StreamExt;

        let mut decoder = sse::Decoder::new();
        let mut acc = gemini_stream::Accumulator::new();
        let mut body = response.bytes_stream();
        // The same ceiling the buffered path applies, to the same bytes.
        // `sse::Decoder` already bounds one event, which is the unterminated
        // line; this bounds the *number* of them. A stream of well-formed
        // hundred-byte deltas passes every check the decoder makes and grows
        // the accumulator until the process dies.
        let mut meter = crate::netguard::intake::Meter::new(crate::netguard::intake::ANSWER);

        while let Some(chunk) = body.next().await {
            let chunk = match chunk {
                Ok(chunk) => chunk,
                Err(e) => return Err(severed(model, &acc, &e.to_string())),
            };
            // Charged before the chunk is kept: what the ceiling bounds is
            // what this process holds, not what it has already held. The
            // refusal is `Unusable` rather than `severed` — it is this
            // plane's rather than the provider's, and the call generated.
            // This wire reports no usage until the end, so the figure is
            // zero here for the same reason `severed` reports `Unaccounted`:
            // what is unknown is the amount, not whether it happened.
            if let Err(e) = meter.charge(chunk.len()) {
                return Err(super::wire::classify_intake(model, Usage::default(), &e));
            }
            let events = decoder
                .push(&chunk)
                .map_err(|error| severed(model, &acc, &error.to_string()))?;
            for event in events {
                if let Some(delta) = acc.push(&event.data)
                    && let Some((observer, label)) = observer
                {
                    observer.event(crate::core::Tainted::with_label(
                        super::ModelStreamEvent::TextDelta(delta),
                        label.clone(),
                    ));
                }
            }
        }
        if !acc.done() {
            // A blocked prompt is a refusal with a name, not an outage: the
            // stream ends without a finish reason, and classifying that as
            // severed would mark it safe to repeat — a retry loop re-hitting
            // the same block forever.
            if acc.prompt_blocked() {
                return self.interpret(&acc.into_response(), model, schema);
            }
            return Err(severed(
                model,
                &acc,
                "the stream ended before the model said why it stopped",
            ));
        }

        let completion = self.interpret(&acc.into_response(), model, schema)?;
        if let Some((observer, label)) = observer {
            observer.event(crate::core::Tainted::with_label(
                super::ModelStreamEvent::Usage(completion.usage),
                label.clone(),
            ));
        }
        Ok(completion)
    }
}

/// What one pass over a candidate's parts found.
struct Scanned {
    text: String,
    calls: Vec<super::ToolCall>,
    /// The forced-tool answer, when structured output is being emulated.
    forced: Option<Value>,
}

impl Gemini {
    /// Read a candidate's parts into text, tool calls, and a forced answer.
    ///
    /// Split out of `interpret` so the interesting decisions are visible rather
    /// than buried in the middle of a long function: which parts are *not* the
    /// answer, and where a tool call's id comes from when the provider issues
    /// none.
    fn scan(parts: &[Value]) -> Scanned {
        let mut text = String::new();
        let mut calls = Vec::new();
        let mut forced: Option<Value> = None;
        for (index, part) in parts.iter().enumerate() {
            // A part marked `thought` is opaque reasoning. It stays in the
            // continuation and never becomes the answer: exposing it would make
            // the model's internal deliberation read as its conclusion.
            if part.get("thought").and_then(Value::as_bool) == Some(true) {
                continue;
            }
            if let Some(call) = part.get("functionCall") {
                let name = call
                    .get("name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_owned();
                let arguments = call.get("args").cloned().unwrap_or_else(|| json!({}));
                if name == RESPOND_TOOL {
                    forced = Some(arguments);
                    continue;
                }
                calls.push(super::ToolCall {
                    // Gemini's function calls need not carry an id, and the
                    // runtime keys tool results by one. Derived from the
                    // position rather than generated, so it is stable across a
                    // replay of the same recorded response — a random id would
                    // make a replayed completion differ from the one journaled.
                    id: call
                        .get("id")
                        .and_then(Value::as_str)
                        .map_or_else(|| format!("{name}-{index}"), ToOwned::to_owned),
                    name,
                    arguments,
                });
                continue;
            }
            if let Some(chunk) = part.get("text").and_then(Value::as_str) {
                text.push_str(chunk);
            }
        }
        Scanned {
            text,
            calls,
            forced,
        }
    }
}

/// A stream that stopped before the model said why.
///
/// Three rungs, and which one applies is decided by what the wire has already
/// said rather than by how far the answer got:
///
/// 1. **`usageMetadata` seen** — Gemini reports it on the chunks themselves,
///    cumulatively, so a stream cut off mid-answer has already been told what
///    it burned. That is [`ModelError::Interrupted`], the one severed-stream
///    answer that carries a bill, and it is the reason streaming is this
///    driver's default.
/// 2. **content but no usage** — generation happened and the cost is unknown.
///    [`ModelError::Unaccounted`]: never free, never counted.
/// 3. **nothing at all** — no evidence anything reached the model, so the call
///    is safe to repeat and costs nothing.
///
/// Rung 1 is not an optimisation. Without it every severed Gemini stream bills
/// zero, and the token ceiling that exists to bound a runaway provider counts
/// nothing while the provider spends — the failure the ceiling was bought to
/// prevent, in the one driver whose wire makes it avoidable.
fn severed(model: &ModelId, acc: &gemini_stream::Accumulator, detail: &str) -> ModelError {
    // Parsed by the buffered path's own function, so the normalisation that
    // costs money — thought tokens billed as output, cached input a subset of
    // the prompt — has exactly one spelling.
    if let Some(envelope) = acc.usage_envelope() {
        return ModelError::Interrupted {
            model: model.clone(),
            usage: Gemini::usage(&envelope),
            detail: detail.to_owned(),
        };
    }
    if acc.generated() {
        return ModelError::Unaccounted {
            model: model.clone(),
            detail: detail.to_owned(),
        };
    }
    ModelError::Unavailable {
        model: model.clone(),
        detail: detail.to_owned(),
    }
}

/// Fill a rate limit's window from the body's `google.rpc.RetryInfo`.
///
/// The shared classifier reads `Retry-After`, which is the one header every
/// other wire this crate meets uses — and the one this provider does not
/// send. Google names its window *inside* the 429 body, as a `RetryInfo`
/// detail carrying a proto `Duration` (`"40s"`). Discarded, the default
/// policy spends its attempts in milliseconds against a window measured in
/// tens of seconds and reports the provider down — the exact defect the
/// `retry_after` plumbing exists to prevent. Any other classification passes
/// through untouched, and a window the header already named is not replaced.
fn with_retry_info(error: ModelError, body: &str) -> ModelError {
    match error {
        ModelError::RateLimited {
            model,
            detail,
            retry_after: None,
        } => ModelError::RateLimited {
            model,
            detail,
            retry_after: retry_info_seconds(body),
        },
        other => other,
    }
}

/// The `RetryInfo` detail's delay, in whole seconds.
///
/// Whole seconds only, floor of a fractional value, zero reads as no advice —
/// the same conservatisms `core::retry_after_seconds` applies to the header
/// form. The ceiling on believing it (`max_advice`) stays where it always
/// was, in the retry policy.
fn retry_info_seconds(body: &str) -> Option<u64> {
    let parsed: Value = serde_json::from_str(body).ok()?;
    let details = parsed.get("error")?.get("details")?.as_array()?;
    let delay = details
        .iter()
        .find(|d| {
            d.get("@type").and_then(Value::as_str)
                == Some("type.googleapis.com/google.rpc.RetryInfo")
        })?
        .get("retryDelay")?
        .as_str()?;
    let seconds: u64 = delay.strip_suffix('s')?.split('.').next()?.parse().ok()?;
    (seconds > 0).then_some(seconds)
}

#[async_trait]
impl ModelProvider for Gemini {
    fn request_profile(&self, model: &ModelId) -> Value {
        json!({
            "driver": "google-gemini-generatecontent/v1",
            "base": self.base,
            "api_version": self.version,
            "stream": self.stream,
            "schema_mode": match self.mode_for(model) {
                SchemaMode::Native => "native",
                SchemaMode::ForcedTool => "forced-tool",
            },
            "timeout_ms": u64::try_from(self.timeout.as_millis()).unwrap_or(u64::MAX),
            // Identity, not decoration: loosening a threshold changes what
            // governed the call, so a replay of history written under the
            // stricter one reports divergence rather than answering under the
            // looser. Absent when nothing was declared, which is a different
            // request from one declaring the provider defaults explicitly.
            "safety": (!self.safety.is_empty()).then(|| self.safety.profile()),
        })
    }

    async fn complete(&self, request: Request<'_>) -> Result<Completion, ModelError> {
        let model = request.model;
        super::refuse_provider_side_media(request.prompt, model)?;
        super::refuse_in_thread_instructions(request.prompt, model)?;
        self.check_egress(model)?;

        let body = self.body(model, &request)?;
        let response = self
            .http
            .post(self.url(model))
            // The header form rather than a `?key=` query parameter: a URL is
            // logged by proxies, written into traces and echoed in errors, and
            // a credential that reaches any of those is one that cannot be
            // un-leaked.
            .header("x-goog-api-key", self.key.expose())
            .timeout(self.timeout)
            .json(&body)
            .send()
            .await
            .map_err(|e| classify_transport(model, &e))?;

        let status = response.status();
        if !status.is_success() {
            let headers = response.headers().clone();
            // Bounded, and the ceiling is the small one: this body is read
            // only to say *why* the call failed, so an endpoint answering a
            // failure with a gigabyte gets an unexplained failure rather than
            // this process's memory.
            let text =
                crate::netguard::intake::read_text(response, crate::netguard::intake::METADATA)
                    .await
                    .unwrap_or_default();
            return Err(with_retry_info(
                classify_status(model, status.as_u16(), &headers, &text),
                &text,
            ));
        }

        let mut completion = if self.stream {
            self.read_streamed(response, model, request.schema, request.stream)
                .await?
        } else {
            self.read_buffered(response, model, request.schema).await?
        };
        // A buffered call still answers the observer's one guaranteed
        // question — what did this cost — as every driver does on both paths.
        if !self.stream
            && let Some((observer, label)) = request.stream
        {
            observer.event(crate::core::Tainted::with_label(
                super::ModelStreamEvent::Usage(completion.usage),
                label.clone(),
            ));
        }
        Self::accumulate_continuation(&mut completion, request.continuation, request.exchanges);
        Ok(completion)
    }
}