agentplane 0.56.1

//! Amazon Bedrock Runtime through the provider-neutral Converse API.
//!
//! `ConverseStream` is the default so partial generation can be classified and
//! metered. Buffered mode is an explicit opt-out.

use std::collections::HashMap;
use std::time::Duration;

use async_trait::async_trait;
use aws_sdk_bedrockruntime::Client;
use aws_sdk_bedrockruntime::config::Region;
use aws_sdk_bedrockruntime::types::{
    CachePointBlock, CachePointType, CacheTtl as WireCacheTtl, ContentBlock, ConversationRole,
    DocumentBlock, DocumentFormat, DocumentSource, ImageBlock, ImageFormat, ImageSource,
    InferenceConfiguration, JsonSchemaDefinition, Message, OutputConfig, OutputFormat,
    OutputFormatStructure, OutputFormatType, ReasoningContentBlock, ReasoningTextBlock,
    SpecificToolChoice, StopReason, SystemContentBlock, Tool, ToolChoice, ToolConfiguration,
    ToolInputSchema, ToolResultBlock, ToolResultContentBlock, ToolResultStatus, ToolSpecification,
    ToolUseBlock,
};
use aws_smithy_types::{Blob, Document, Number};
use serde_json::{Value, json};

use super::{
    AnthropicBudget, CacheTtl, Completion, ModelError, ModelId, ModelProvider,
    ProviderContinuation, Request, SchemaMode, ToolDeclaration, ToolExchange, Usage,
};

const RESPOND_TOOL: &str = "__agentplane_respond";

fn decoded_media(
    block: &Value,
    expected_type: &str,
    model: &ModelId,
) -> Result<(String, Vec<u8>), ModelError> {
    let actual = block
        .get("type")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if actual != expected_type {
        return Err(Bedrock::refused(
            model,
            format!("unsupported Bedrock content block '{actual}'"),
        ));
    }
    let media_type = block
        .get("media_type")
        .and_then(Value::as_str)
        .ok_or_else(|| Bedrock::refused(model, "Bedrock media block has no media_type"))?;
    let data = block
        .get("data")
        .and_then(Value::as_str)
        .ok_or_else(|| Bedrock::refused(model, "Bedrock media block has no inline base64 data"))?;
    let bytes = crate::core::b64::decode(data)
        .ok_or_else(|| Bedrock::refused(model, "Bedrock media data is not canonical base64"))?;
    Ok((media_type.to_owned(), bytes))
}

pub(crate) fn content_from_prompt_json(
    block: &Value,
    model: &ModelId,
) -> Result<ContentBlock, ModelError> {
    if let Some(text) = block.get("text").and_then(Value::as_str) {
        return Ok(ContentBlock::Text(text.to_owned()));
    }
    match block.get("type").and_then(Value::as_str) {
        Some("image") => {
            let (media_type, bytes) = decoded_media(block, "image", model)?;
            let format = match media_type.as_str() {
                "image/gif" => ImageFormat::Gif,
                "image/jpeg" => ImageFormat::Jpeg,
                "image/png" => ImageFormat::Png,
                "image/webp" => ImageFormat::Webp,
                other => {
                    return Err(Bedrock::refused(
                        model,
                        format!("Bedrock does not support image media type '{other}'"),
                    ));
                }
            };
            ImageBlock::builder()
                .format(format)
                .source(ImageSource::Bytes(Blob::new(bytes)))
                .build()
                .map(ContentBlock::Image)
                .map_err(|error| Bedrock::refused(model, error.to_string()))
        }
        Some("document") => {
            let (media_type, bytes) = decoded_media(block, "document", model)?;
            let format = match media_type.as_str() {
                "text/csv" => DocumentFormat::Csv,
                "application/msword" => DocumentFormat::Doc,
                "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => {
                    DocumentFormat::Docx
                }
                "text/html" => DocumentFormat::Html,
                "text/markdown" => DocumentFormat::Md,
                "application/pdf" => DocumentFormat::Pdf,
                "text/plain" => DocumentFormat::Txt,
                "application/vnd.ms-excel" => DocumentFormat::Xls,
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => {
                    DocumentFormat::Xlsx
                }
                other => {
                    return Err(Bedrock::refused(
                        model,
                        format!("Bedrock does not support document media type '{other}'"),
                    ));
                }
            };
            DocumentBlock::builder()
                .format(format)
                // Neutral and constant: Bedrock warns that document names are
                // prompt-injection-bearing model input.
                .name("document")
                .source(DocumentSource::Bytes(Blob::new(bytes)))
                .build()
                .map(ContentBlock::Document)
                .map_err(|error| Bedrock::refused(model, error.to_string()))
        }
        Some(other) => Err(Bedrock::refused(
            model,
            format!("unsupported Bedrock content block '{other}'"),
        )),
        None => Err(Bedrock::refused(
            model,
            "Bedrock content block has neither text nor a supported type",
        )),
    }
}

/// Amazon Bedrock Runtime's Converse driver.
///
/// # This driver has no egress allowlist, and that is a gap rather than a design
///
/// Every other model driver here takes an [`Egress`](crate::core::Egress) and
/// refuses a base URL the deployment never granted. This one cannot: it is
/// handed a built AWS `Client`, and the SDK exposes no way to read back the
/// endpoint that client will actually reach. A check against the endpoint
/// *derived* from the region would pass while an endpoint override sent every
/// call somewhere else — a control that looks like one and is not, which is the
/// same reason this crate ships no `Egress::allow_all`.
///
/// What stands in its place is narrower and worth knowing exactly: the region
/// is on the record. It is read from the client rather than accepted beside it,
/// and it enters [`ModelProvider::request_profile`], so it is digest-covered
/// and a change of region is replay divergence. That says *which service the
/// client was configured for*. It does not say the call went there, and nothing
/// in this driver does. A deployment that needs the destination constrained
/// constrains it where the SDK does — a VPC endpoint, an egress proxy, or an
/// IAM policy — not here.
#[derive(Clone)]
pub struct Bedrock {
    client: Client,
    region: String,
    timeout: Duration,
    schema_mode: SchemaMode,
    stream: bool,
    /// The deployment's Bedrock guardrail, when it has one.
    ///
    /// Passed through rather than reimplemented. Content classification is a
    /// specialist's job — this crate ships no policy evaluator and no tracing
    /// exporter for the same reason — and a deployment on Bedrock already owns
    /// a guardrail, versioned and administered where its compliance people can
    /// see it. What the runtime owns is everything *around* it: the choice is
    /// in the request profile, so it is digest-covered and replay-visible, and
    /// an intervention is a **metered refusal** rather than an answer.
    guardrail: Option<Guardrail>,
    /// Which model family's reasoning dialect this driver instance speaks.
    ///
    /// Absent by default, and absent means *refuse* — see
    /// [`ReasoningDialect`].
    reasoning: Option<ReasoningDialect>,
}

/// How a Bedrock model family spells "think harder".
///
/// Converse is a **provider-neutral** envelope over a model zoo that does not
/// agree on this: Anthropic takes adaptive thinking, Nova 2 takes a
/// `reasoningConfig` in `additionalModelRequestFields`, and several families
/// have no such control at all. There is therefore no mapping this driver can
/// derive from a Converse request, which is why an undeclared dialect refuses
/// rather than guessing — silently sending a different effort than the one the
/// manifest declared would make a digest-covered control advisory.
///
/// It is **declared, never sniffed from the model id**. Cross-region inference
/// profiles prefix the id (`us.amazon.nova-2-lite-v1:0`), a substring match
/// would bind behaviour to a naming convention AWS owns, and this crate already
/// refuses to decide a control by matching on a string elsewhere. One driver
/// instance therefore serves one family, exactly as a guardrail does.
///
/// The choice is in the **request profile**, so it is effect identity: moving a
/// deployment from one dialect to another is replay divergence rather than a
/// quiet change in what governed the call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ReasoningDialect {
    /// Amazon Nova 2's extended thinking.
    ///
    /// Rendered as `additionalModelRequestFields.reasoningConfig`, with
    /// `type: "enabled"` and `maxReasoningEffort` of `low`, `medium` or `high`
    /// — the three levels AWS documents and the only three it accepts.
    ///
    /// [`ReasoningEffort::None`](crate::model::ReasoningEffort::None) sends
    /// `type: "disabled"`, which is Nova's own default and is *not* the same as
    /// sending nothing: it is a deployment stating that this call must not
    /// reason, and stating it puts the fact in the journal.
    ///
    /// `Minimal`, `XHigh` and `Max` are **refused**, because Nova has no
    /// counterpart and collapsing them into `low` or `high` is the silent
    /// substitution this seam exists to prevent. That is the same rule the
    /// Anthropic driver follows for the levels adaptive thinking cannot
    /// express.
    ///
    /// One constraint AWS documents is satisfied here by construction: Nova
    /// refuses `temperature`, `topP` and `topK` alongside `maxReasoningEffort:
    /// "high"`, and this driver has never sent any of them — sampling
    /// parameters are absent from the request profile, so they cannot be part
    /// of effect identity, so they are not sent.
    Nova,
}

/// Which Bedrock guardrail to apply, and whether to ask for its trace.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Guardrail {
    pub identifier: String,
    /// Pinned, never `DRAFT`-by-default: a guardrail version is part of what
    /// governed this call, and a floating pointer would make two runs under
    /// "the same" configuration mean different things.
    pub version: String,
    /// Ask Bedrock to return why it intervened.
    ///
    /// Off by default. The trace names the policy and the matched category,
    /// which is exactly the classification the gate protects — useful in a
    /// journal an operator reads, and a map for a prober if it ever reaches a
    /// model. It never does: an intervention leaves this driver as an error.
    pub trace: bool,
}

impl Guardrail {
    #[must_use]
    pub fn new(identifier: impl Into<String>, version: impl Into<String>) -> Self {
        Self {
            identifier: identifier.into(),
            version: version.into(),
            trace: false,
        }
    }

    /// Ask for the intervention trace, which is journaled and never shown to a
    /// model.
    #[must_use]
    pub const fn with_trace(mut self) -> Self {
        self.trace = true;
        self
    }

    /// The streaming form.
    ///
    /// `SYNCHRONOUS` deliberately, which is Bedrock's *blocking* mode: the
    /// service assesses each chunk before releasing it. The alternative,
    /// `ASYNCHRONOUS`, streams first and intervenes afterwards — lower
    /// latency, and it means blocked content has already reached the caller
    /// by the time the guardrail objects. A control that arrives after the
    /// thing it was installed to prevent is not one.
    fn stream_config(&self) -> aws_sdk_bedrockruntime::types::GuardrailStreamConfiguration {
        use aws_sdk_bedrockruntime::types::{
            GuardrailStreamConfiguration, GuardrailStreamProcessingMode, GuardrailTrace,
        };
        GuardrailStreamConfiguration::builder()
            .guardrail_identifier(&self.identifier)
            .guardrail_version(&self.version)
            .trace(if self.trace {
                GuardrailTrace::Enabled
            } else {
                GuardrailTrace::Disabled
            })
            .stream_processing_mode(GuardrailStreamProcessingMode::Sync)
            .build()
    }

    fn config(&self) -> aws_sdk_bedrockruntime::types::GuardrailConfiguration {
        use aws_sdk_bedrockruntime::types::{GuardrailConfiguration, GuardrailTrace};
        GuardrailConfiguration::builder()
            .guardrail_identifier(&self.identifier)
            .guardrail_version(&self.version)
            .trace(if self.trace {
                GuardrailTrace::Enabled
            } else {
                GuardrailTrace::Disabled
            })
            .build()
    }
}

impl std::fmt::Debug for Bedrock {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Bedrock")
            .field("region", &self.region)
            .field("timeout", &self.timeout)
            .field("schema_mode", &self.schema_mode)
            .field("stream", &self.stream)
            .field("guardrail", &self.guardrail)
            .finish_non_exhaustive()
    }
}

impl Bedrock {
    /// Load the standard AWS credential chain for this region.
    ///
    /// The chain covers every way Bedrock authenticates: static keys and
    /// profiles, `credential_process`, IAM Identity Center SSO, assume-role
    /// and web-identity (IRSA), container and `IMDSv2` instance roles, and
    /// `aws login` console-session profiles. A Bedrock API key in
    /// `AWS_BEARER_TOKEN_BEDROCK` is honoured too, and **overrides** `SigV4`
    /// for the whole client; the SDK reads it from the environment only,
    /// never from a profile. For a key supplied programmatically, build the
    /// client with `Config::builder().bearer_token(..)` and hand it to
    /// [`from_client`](Self::from_client).
    ///
    /// # Errors
    ///
    /// If the region is blank, or the client it builds carries none.
    pub async fn from_env(region: impl Into<String>) -> Result<Self, String> {
        let region = region.into();
        if region.trim().is_empty() {
            return Err("an AWS region is required for Bedrock".to_owned());
        }
        let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
            .region(Region::new(region))
            .load()
            .await;
        Self::from_client(Client::new(&config))
    }

    /// Build from an already configured AWS client.
    ///
    /// The region is read **from the client**, not accepted beside it. It is
    /// what [`ModelProvider::request_profile`] attests, and that profile is
    /// effect identity — so a region taken as a separate argument is a second
    /// copy of a fact the client already holds, free to disagree with the
    /// endpoint the SDK actually reaches. The journal would then attest a
    /// destination the calls never went to, and a replay would compare against
    /// it, with nothing anywhere able to notice.
    ///
    /// # Errors
    ///
    /// If the client carries no region, or a blank one. A client that cannot
    /// name its region cannot resolve a Bedrock endpoint either, so this is a
    /// wiring mistake refused where it is written rather than at the far end of
    /// a run — and recording an empty region would put a fiction on the record
    /// instead.
    pub fn from_client(client: Client) -> Result<Self, String> {
        let region = client
            .config()
            .region()
            .map(|region| region.as_ref().trim().to_owned())
            .filter(|region| !region.is_empty())
            .ok_or_else(|| {
                "the Bedrock client carries no region, so there is nothing true to put in the \
                 request profile — build it from a config with `.region(..)` set"
                    .to_owned()
            })?;
        Ok(Self {
            client,
            region,
            timeout: Duration::from_mins(5),
            schema_mode: SchemaMode::Native,
            stream: true,
            guardrail: None,
            reasoning: None,
        })
    }

    /// Apply a Bedrock guardrail to every call this driver makes.
    ///
    /// Passed through rather than reimplemented: content classification is a
    /// specialist's job, and a deployment on Bedrock already owns a guardrail
    /// administered where its compliance people can see it. What the runtime
    /// owns is everything around it — the choice enters
    /// [`ModelProvider::request_profile`], so it is digest-covered and
    /// replay-visible, and an intervention is a **metered refusal** rather
    /// than an answer.
    #[must_use]
    pub fn guardrail(mut self, guardrail: Guardrail) -> Self {
        self.guardrail = Some(guardrail);
        self
    }

    /// Declare which model family's reasoning dialect this driver speaks.
    ///
    /// Without it, a request carrying a `reasoning_effort` is **refused**:
    /// Converse is one envelope over families that spell reasoning differently
    /// or not at all, so there is nothing for this driver to derive. See
    /// [`ReasoningDialect`] for why it is declared rather than read off the
    /// model id, and for which efforts each dialect can carry faithfully.
    ///
    /// The dialect enters [`ModelProvider::request_profile`], so changing it is
    /// replay divergence rather than a quiet change in what governed the call.
    #[must_use]
    pub const fn reasoning(mut self, dialect: ReasoningDialect) -> Self {
        self.reasoning = Some(dialect);
        self
    }

    /// The `additionalModelRequestFields` this call's reasoning effort needs.
    ///
    /// `None` when the request asked for no reasoning at all, which is the
    /// ordinary case and must send nothing — an empty `reasoningConfig` is not
    /// the same request as no `reasoningConfig`.
    ///
    /// Built once and handed to **both** request paths. The guardrail above is
    /// the precedent and the reason: a control rendered separately on the
    /// buffered and streaming paths is one where only the half nobody exercises
    /// is wrong, and a `stream: true` deployment loses it silently.
    ///
    /// # Errors
    ///
    /// If no dialect is declared, or the declared one cannot carry this effort
    /// faithfully.
    fn reasoning_config(
        &self,
        model: &ModelId,
        effort: Option<super::ReasoningEffort>,
    ) -> Result<Option<Document>, ModelError> {
        use super::ReasoningEffort as E;

        let Some(effort) = effort else {
            return Ok(None);
        };
        let Some(dialect) = self.reasoning else {
            return Err(Self::refused(
                model,
                "Bedrock Converse has no provider-neutral reasoning-effort mapping, because one \
                 envelope covers families that spell reasoning differently or not at all. \
                 Declare which one this driver speaks — `Bedrock::from_client(..).reasoning(\
                 ReasoningDialect::Nova)` — rather than having the driver guess and silently \
                 change the effort the manifest declared",
            ));
        };
        match dialect {
            ReasoningDialect::Nova => {
                // Refused rather than collapsed. Nova documents exactly three
                // levels, and folding `Max` into `high` would answer a request
                // for the most thorough reasoning available with the third of
                // three — a substitution nothing downstream could see, on a
                // value that is digest-covered and therefore claims to describe
                // what actually governed the call.
                let level = match effort {
                    E::None => {
                        return Ok(Some(document_from_json(
                            &json!({ "reasoningConfig": { "type": "disabled" } }),
                        )));
                    }
                    E::Low => "low",
                    E::Medium => "medium",
                    E::High => "high",
                    E::Minimal | E::XHigh | E::Max => {
                        return Err(Self::refused(
                            model,
                            format!(
                                "Amazon Nova extended thinking has no counterpart for reasoning \
                                 effort '{}' — it accepts low, medium and high. Declare one of \
                                 those, or none to disable reasoning explicitly",
                                effort.as_str()
                            ),
                        ));
                    }
                };
                Ok(Some(document_from_json(&json!({
                    "reasoningConfig": { "type": "enabled", "maxReasoningEffort": level },
                }))))
            }
        }
    }

    /// Bound the complete Converse operation.
    #[must_use]
    pub const fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Choose native constrained output or the forced-tool compatibility path.
    #[must_use]
    pub const fn structured_via(mut self, mode: SchemaMode) -> Self {
        self.schema_mode = mode;
        self
    }

    /// Use buffered Converse instead of `ConverseStream`.
    #[must_use]
    pub const fn buffered(mut self) -> Self {
        self.stream = false;
        self
    }

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

    fn messages(
        model: &ModelId,
        prompt: &Value,
        exchanges: &[ToolExchange],
        told: &[String],
        continuation: Option<&ProviderContinuation>,
    ) -> Result<Vec<Message>, ModelError> {
        let mut messages = match crate::model::prompt::turns(prompt, &["messages", "input"]) {
            crate::model::prompt::Turns::Native(items) => items
                .iter()
                .map(|item| Self::message_from_value(model, item))
                .collect::<Result<Vec<_>, _>>()?,
            crate::model::prompt::Turns::Parts(parts) => {
                let mut content = Vec::with_capacity(parts.len());
                for part in &parts {
                    content.push(ContentBlock::Text(part.text.clone().into_owned()));
                    if let Some(ttl) = part.cache {
                        content.push(ContentBlock::CachePoint(cache_point(ttl, model)?));
                    }
                }
                vec![
                    Message::builder()
                        .role(ConversationRole::User)
                        .set_content(Some(content))
                        .build()
                        .map_err(|error| Self::refused(model, error.to_string()))?,
                ]
            }
        };

        if exchanges.is_empty() && told.is_empty() {
            return Ok(messages);
        }
        let prompt_len = messages.len();

        if let Some(state) = continuation {
            if state.provider != "bedrock" {
                return Err(Self::refused(
                    model,
                    format!(
                        "a '{}' continuation cannot be sent to Bedrock",
                        state.provider
                    ),
                ));
            }
            let transcript = state
                .state
                .as_array()
                .ok_or_else(|| Self::refused(model, "the Bedrock continuation is not an array"))?
                .iter()
                .map(|message| continuation_message_from_json(message, model))
                .collect::<Result<Vec<_>, _>>()?;
            messages.extend(transcript);
        } else if !exchanges.is_empty() {
            let assistant = exchanges
                .iter()
                .map(|exchange| {
                    Ok(ContentBlock::ToolUse(
                        ToolUseBlock::builder()
                            .tool_use_id(exchange.call.id.clone())
                            .name(exchange.call.name.clone())
                            .input(document_from_json(&exchange.call.arguments))
                            .build()
                            .map_err(|error| Self::refused(model, error.to_string()))?,
                    ))
                })
                .collect::<Result<Vec<_>, ModelError>>()?;
            messages.push(
                Message::builder()
                    .role(ConversationRole::Assistant)
                    .set_content(Some(assistant))
                    .build()
                    .map_err(|error| Self::refused(model, error.to_string()))?,
            );
        }

        messages.push(continuation_message_from_json(
            &user_turn(exchanges, told),
            model,
        )?);
        join_at_the_prompt(messages, prompt_len, model)
    }

    fn message_from_value(model: &ModelId, value: &Value) -> Result<Message, ModelError> {
        let object = value
            .as_object()
            .ok_or_else(|| Self::refused(model, "a Bedrock message must be an object"))?;
        let role = match object.get("role").and_then(Value::as_str) {
            Some("user") => ConversationRole::User,
            Some("assistant") => ConversationRole::Assistant,
            Some(other) => {
                return Err(Self::refused(
                    model,
                    format!("unsupported Bedrock message role '{other}'"),
                ));
            }
            None => return Err(Self::refused(model, "a Bedrock message has no role")),
        };
        let content = object
            .get("content")
            .ok_or_else(|| Self::refused(model, "a Bedrock message has no content"))?;
        match content {
            Value::String(text) => Self::text_message(role, text, model),
            Value::Array(blocks) => {
                let blocks = blocks
                    .iter()
                    .map(|block| content_from_prompt_json(block, model))
                    .collect::<Result<Vec<_>, ModelError>>()?;
                Message::builder()
                    .role(role)
                    .set_content(Some(blocks))
                    .build()
                    .map_err(|error| Self::refused(model, error.to_string()))
            }
            _ => Err(Self::refused(
                model,
                "Bedrock message content must be text or text blocks",
            )),
        }
    }

    fn text_message(
        role: ConversationRole,
        text: &str,
        model: &ModelId,
    ) -> Result<Message, ModelError> {
        Message::builder()
            .role(role)
            .content(ContentBlock::Text(text.to_owned()))
            .build()
            .map_err(|error| Self::refused(model, error.to_string()))
    }

    fn tool_config(
        model: &ModelId,
        schema: Option<&Value>,
        tools: &[ToolDeclaration],
        mode: SchemaMode,
    ) -> Result<Option<ToolConfiguration>, ModelError> {
        if schema.is_some() && mode == SchemaMode::ForcedTool && !tools.is_empty() {
            return Err(Self::refused(
                model,
                "Bedrock Converse obtains structured output by forcing a synthetic tool, so a \
                 response schema cannot be combined with callable tools",
            ));
        }
        let mut builder = ToolConfiguration::builder();
        // Converse holds a strict tool to the same subset and per-request
        // ceilings Anthropic does; a schema outside them is sent non-strict
        // rather than turning the request into a 400.
        let mut budget = AnthropicBudget::default();
        if let Some(schema) = schema.filter(|_| mode == SchemaMode::ForcedTool) {
            builder = builder
                .tools(tool(
                    RESPOND_TOOL,
                    "Return the answer in the required shape.",
                    schema,
                    budget.tool(schema),
                    model,
                )?)
                .tool_choice(ToolChoice::Tool(
                    SpecificToolChoice::builder()
                        .name(RESPOND_TOOL)
                        .build()
                        .map_err(|error| Self::refused(model, error.to_string()))?,
                ));
        } else {
            for declaration in tools {
                builder = builder.tools(tool(
                    &declaration.name,
                    &declaration.description,
                    &declaration.parameters,
                    budget.tool(&declaration.parameters),
                    model,
                )?);
                if let Some(ttl) = declaration.cache {
                    builder = builder.tools(Tool::CachePoint(cache_point(ttl, model)?));
                }
            }
        }
        if (schema.is_none() || mode == SchemaMode::Native) && tools.is_empty() {
            Ok(None)
        } else {
            builder
                .build()
                .map(Some)
                .map_err(|error| Self::refused(model, error.to_string()))
        }
    }

    fn output_config(
        model: &ModelId,
        schema: Option<&Value>,
        mode: SchemaMode,
    ) -> Result<Option<OutputConfig>, ModelError> {
        let Some(schema) = schema.filter(|_| mode == SchemaMode::Native) else {
            return Ok(None);
        };
        let definition = JsonSchemaDefinition::builder()
            .schema(schema.to_string())
            .name("agentplane_response")
            .build()
            .map_err(|error| Self::refused(model, error.to_string()))?;
        let format = OutputFormat::builder()
            .r#type(OutputFormatType::JsonSchema)
            .structure(OutputFormatStructure::JsonSchema(definition))
            .build()
            .map_err(|error| Self::refused(model, error.to_string()))?;
        Ok(Some(OutputConfig::builder().text_format(format).build()))
    }

    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
    async fn complete_streamed(
        &self,
        model: &ModelId,
        prompt: &Value,
        messages: Vec<Message>,
        max_tokens: i32,
        tools: Option<ToolConfiguration>,
        output_config: Option<OutputConfig>,
        reasoning_config: Option<Document>,
        schema: Option<&Value>,
        observer: Option<(&dyn super::ModelStreamObserver, &crate::core::Label)>,
    ) -> Result<Completion, ModelError> {
        let deadline = tokio::time::Instant::now() + self.timeout;
        let mut operation = self
            .client
            .converse_stream()
            .model_id(&model.model)
            .set_messages(Some(messages))
            .inference_config(
                InferenceConfiguration::builder()
                    .max_tokens(max_tokens)
                    .build(),
            )
            .set_tool_config(tools)
            .set_output_config(output_config)
            // Both paths, or the control is one a `stream: true` deployment
            // silently loses — the same rule written twice is the shape where
            // only the half nobody exercises is wrong.
            .set_guardrail_config(self.guardrail.as_ref().map(Guardrail::stream_config))
            .set_additional_model_request_fields(reasoning_config);
        let system = system_blocks(prompt, model)?;
        if !system.is_empty() {
            operation = operation.set_system(Some(system));
        }
        let output = tokio::time::timeout_at(deadline, operation.send())
            .await
            .map_err(|_| ModelError::Unavailable {
                model: model.clone(),
                detail: "Bedrock did not start the stream before the request deadline".to_owned(),
            })?
            .map_err(|error| {
                classify_sdk(model, &error, |service, detail| {
                    classify_service(
                        model,
                        detail,
                        aws_sdk_bedrockruntime::error::ProvideErrorMetadata::code(service),
                        service.is_throttling_exception() || service.is_model_not_ready_exception(),
                        service.is_access_denied_exception()
                            || service.is_resource_not_found_exception()
                            || service.is_validation_exception(),
                    )
                })
            })?;

        let mut stream = output.stream;
        let mut accumulator = super::bedrock_stream::Accumulator::new();
        loop {
            let event = tokio::time::timeout_at(deadline, stream.recv())
                .await
                .map_err(|_| severed_stream(model, &accumulator, "the request deadline elapsed"))?
                .map_err(|error| {
                    let detail = error.to_string();
                    classify_stream_event(model, &accumulator, error.as_service_error(), &detail)
                })?;
            let Some(event) = event else {
                break;
            };
            if let aws_sdk_bedrockruntime::types::ConverseStreamOutput::ContentBlockDelta(delta) =
                &event
                && let Some(aws_sdk_bedrockruntime::types::ContentBlockDelta::Text(text)) =
                    delta.delta()
                && let Some((observer, label)) = observer
            {
                observer.event(crate::core::Tainted::with_label(
                    super::ModelStreamEvent::TextDelta(text.clone()),
                    label.clone(),
                ));
            }
            accumulator.event(event);
        }

        if accumulator.stop_reason()
            == Some(&aws_sdk_bedrockruntime::types::StopReason::GuardrailIntervened)
        {
            // Metered and landed, exactly as on the buffered path: the model
            // was invoked and the assessment was billed. A streamed
            // intervention has usually already emitted deltas to a live
            // observer — advisory by contract, and the journal keeps the
            // refusal that is canonical.
            return Err(ModelError::Unusable {
                model: model.clone(),
                // Whatever the stream reported before it was stopped. A
                // guardrail assessed the call, and an assessment is billed —
                // reporting zero would tell the ceiling it was free.
                usage: accumulator.usage().unwrap_or_default(),
                detail: "a Bedrock guardrail intervened on this call".to_owned(),
            });
        }
        let Some(stop_reason) = accumulator.stop_reason().cloned() else {
            return Err(severed_stream(
                model,
                &accumulator,
                "the stream ended before messageStop",
            ));
        };
        let Some(usage) = accumulator.usage() else {
            return Err(severed_stream(
                model,
                &accumulator,
                "the stream ended without usage metadata",
            ));
        };
        if let Some((observer, label)) = observer {
            observer.event(crate::core::Tainted::with_label(
                super::ModelStreamEvent::Usage(usage),
                label.clone(),
            ));
        }
        let blocks = accumulator
            .finish()
            .map_err(|detail| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail,
            })?;
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .set_content(Some(blocks))
            .build()
            .map_err(|error| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: error.to_string(),
            })?;
        let token_usage = aws_sdk_bedrockruntime::types::TokenUsage::builder()
            .input_tokens(i32::try_from(usage.input_tokens).unwrap_or(i32::MAX))
            .output_tokens(i32::try_from(usage.output_tokens).unwrap_or(i32::MAX))
            .total_tokens(
                i32::try_from(usage.input_tokens.saturating_add(usage.output_tokens))
                    .unwrap_or(i32::MAX),
            )
            .cache_write_input_tokens(i32::try_from(usage.cache_write_tokens).unwrap_or(i32::MAX))
            .cache_read_input_tokens(i32::try_from(usage.cache_read_tokens).unwrap_or(i32::MAX))
            .build()
            .map_err(|error| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: error.to_string(),
            })?;
        let assembled = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(stop_reason)
            .usage(token_usage)
            .build()
            .map_err(|error| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: error.to_string(),
            })?;
        Self::interpret(model, schema, self.schema_mode, &assembled)
    }

    #[allow(clippy::too_many_lines)]
    fn interpret(
        model: &ModelId,
        schema: Option<&Value>,
        mode: SchemaMode,
        output: &aws_sdk_bedrockruntime::operation::converse::ConverseOutput,
    ) -> Result<Completion, ModelError> {
        // Converse reports the cache counters *beside* `inputTokens` — the
        // Anthropic convention — so the shared fold is what makes
        // `Usage::input_tokens` mean everything the provider processed here
        // as on every other driver.
        let usage = output.usage().map_or_else(Usage::default, usage_from);
        // An allowlist. `end_turn`, `tool_use` and `stop_sequence` are an
        // answer; `max_tokens` and a full context window are a typed
        // truncation; everything else ended without one. A guardrail
        // intervention passed through would be the guardrail's canned message
        // wearing a successful answer. The malformed stops are the provider
        // disowning what reached the content blocks — a fragment passed
        // through would hand a tool loop a call the provider itself could not
        // parse — and `content_filtered` or a reason this SDK reports as
        // unknown is no more an answer. Metered, because the tokens were
        // generated and billed either way.
        let truncated = match output.stop_reason() {
            StopReason::EndTurn | StopReason::ToolUse | StopReason::StopSequence => false,
            StopReason::MaxTokens | StopReason::ModelContextWindowExceeded => true,
            other => {
                return Err(ModelError::Unusable {
                    model: model.clone(),
                    usage,
                    detail: format!(
                        "Bedrock stopped with '{}', which is not an answer: no fragment of \
                         what the model emitted is usable output",
                        other.as_str()
                    ),
                });
            }
        };
        let message = output
            .output()
            .and_then(|value| value.as_message().ok())
            .ok_or_else(|| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: "Bedrock returned no message output".to_owned(),
            })?;

        let mut text = String::new();
        let mut calls = Vec::new();
        let mut state = Vec::new();
        let mut forced = None;
        for block in message.content() {
            state.push(content_to_json(block, model, usage)?);
            match block {
                ContentBlock::Text(value) => text.push_str(value),
                ContentBlock::ToolUse(value) if value.name() == RESPOND_TOOL => {
                    forced = Some(json_from_document(value.input()));
                }
                ContentBlock::ToolUse(value) => calls.push(super::ToolCall {
                    id: value.tool_use_id().to_owned(),
                    name: value.name().to_owned(),
                    arguments: json_from_document(value.input()),
                }),
                ContentBlock::ReasoningContent(_) => {}
                _ => {
                    return Err(ModelError::Unusable {
                        model: model.clone(),
                        usage,
                        detail: "Bedrock returned a content block this driver cannot preserve"
                            .to_owned(),
                    });
                }
            }
        }

        let emulating = schema.is_some() && mode == SchemaMode::ForcedTool;
        let structured = if emulating {
            let value = forced.ok_or_else(|| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: "Bedrock did not honor the forced structured-output tool".to_owned(),
            })?;
            if let Some(schema) = schema {
                super::validate_schema(schema, &value).map_err(|detail| ModelError::Unusable {
                    model: model.clone(),
                    usage,
                    detail,
                })?;
            }
            text = value.to_string();
            Some(value)
        } else if schema.is_some() && calls.is_empty() {
            let value = serde_json::from_str(&text).map_err(|error| ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: format!("Bedrock native structured output was not JSON: {error}"),
            })?;
            super::validate_schema(schema.expect("checked above"), &value).map_err(|detail| {
                ModelError::Unusable {
                    model: model.clone(),
                    usage,
                    detail,
                }
            })?;
            Some(value)
        } else {
            None
        };
        if text.is_empty() && calls.is_empty() && structured.is_none() {
            return Err(ModelError::Unusable {
                model: model.clone(),
                usage,
                detail: "Bedrock returned neither text nor a tool call".to_owned(),
            });
        }
        let stop_reason = output.stop_reason().as_str().to_owned();
        // Every answer leaves the turn it was, so a person can speak after it
        // — except the forced tool that shapes a schema answer.
        let continuation = (!emulating).then(|| {
            ProviderContinuation::new(
                "bedrock",
                json!([{ "role": "assistant", "content": state }]),
            )
        });
        Ok(Completion {
            text,
            tool_calls: calls,
            // `Converse` names no model on the way back — the request's
            // `modelId` is all there is, and repeating it here would report a
            // substitution had been ruled out when nothing looked.
            model: None,
            usage,
            stop_reason: Some(stop_reason),
            truncated,
            structured,
            continuation,
            reasoning: None,
            commentary: None,
            compacted: None,
        })
    }
}

#[async_trait]
impl ModelProvider for Bedrock {
    fn gen_ai_provider(&self) -> Option<&str> {
        Some("aws.bedrock")
    }
    fn request_profile(&self, _model: &ModelId) -> Value {
        json!({
            "driver": "aws-bedrock-converse/v1",
            "region": self.region,
            "stream": self.stream,
            "schema_mode": match self.schema_mode {
                SchemaMode::Native => "native",
                SchemaMode::ForcedTool => "forced-tool",
            },
            // Identity, not decoration: turning a guardrail on, off, or to
            // another version changes what governed the call, so a replay of
            // history written under the old configuration reports divergence
            // rather than answering under the new one. The identifier and
            // version only — a guardrail id is configuration, not a secret,
            // and the trace flag changes no request semantics.
            "guardrail": self.guardrail.as_ref().map(|g| json!({
                "id": g.identifier,
                "version": g.version,
            })),
            // Identity for the same reason the guardrail is. The dialect
            // decides how a declared `reasoning_effort` is rendered, so moving
            // a deployment from one to another changes what governed the call
            // — and a replay of history written under the old one reports
            // divergence rather than answering under the new.
            "reasoning_dialect": self.reasoning.map(|d| match d {
                ReasoningDialect::Nova => "nova",
            }),
        })
    }

    async fn complete(&self, request: Request<'_>) -> Result<Completion, ModelError> {
        let Request {
            model,
            prompt,
            max_output_tokens,
            reasoning_effort,
            // Converse has no field that asks for one, and Nova 2 returns its
            // reasoning as `[REDACTED]`.
            reasoning_summary: _,
            schema,
            tools,
            exchanges,
            told,
            continuation,
            compaction,
            stream,
        } = request;
        super::refuse_provider_side_media(prompt, model)?;
        super::refuse_in_thread_instructions(prompt, model)?;
        // Converse offers no compaction; Anthropic's lives on the Messages
        // API, which this driver does not speak.
        if compaction.is_some() {
            return Err(Self::refused(
                model,
                "Bedrock Converse has no server-side compaction",
            ));
        }
        let reasoning_config = self.reasoning_config(model, reasoning_effort)?;
        let max_tokens = i32::try_from(max_output_tokens)
            .map_err(|_| Self::refused(model, "max_output_tokens exceeds Bedrock's i32 limit"))?;
        let messages = Self::messages(model, prompt, exchanges, told, continuation)?;
        let tools = Self::tool_config(model, schema, tools, self.schema_mode)?;
        let output_config = Self::output_config(model, schema, self.schema_mode)?;
        if self.stream {
            let streamed = self
                .complete_streamed(
                    model,
                    prompt,
                    messages,
                    max_tokens,
                    tools,
                    output_config,
                    reasoning_config,
                    schema,
                    stream,
                )
                .await?;
            let mut completion = streamed;
            accumulate_continuation(&mut completion, continuation, exchanges, told);
            return Ok(completion);
        }
        let mut operation = self
            .client
            .converse()
            .model_id(&model.model)
            .set_messages(Some(messages))
            .inference_config(
                InferenceConfiguration::builder()
                    .max_tokens(max_tokens)
                    .build(),
            )
            .set_tool_config(tools)
            .set_output_config(output_config)
            // Both paths, or the control is one a `stream: true` deployment
            // silently loses — the same rule written twice is the shape where
            // only the half nobody exercises is wrong.
            .set_guardrail_config(self.guardrail.as_ref().map(Guardrail::config))
            .set_additional_model_request_fields(reasoning_config);
        let system = system_blocks(prompt, model)?;
        if !system.is_empty() {
            operation = operation.set_system(Some(system));
        }

        let result = tokio::time::timeout(self.timeout, operation.send())
            .await
            .map_err(|_| ModelError::Unavailable {
                model: model.clone(),
                detail: format!("Bedrock did not complete within {:?}", self.timeout),
            })?;
        let output = result.map_err(|error| {
            classify_sdk(model, &error, |service, detail| {
                classify_service(
                    model,
                    detail,
                    aws_sdk_bedrockruntime::error::ProvideErrorMetadata::code(service),
                    service.is_throttling_exception() || service.is_model_not_ready_exception(),
                    service.is_access_denied_exception()
                        || service.is_resource_not_found_exception()
                        || service.is_validation_exception(),
                )
            })
        })?;
        let mut completion = Self::interpret(model, schema, self.schema_mode, &output)?;
        if let Some((observer, label)) = stream {
            observer.event(crate::core::Tainted::with_label(
                super::ModelStreamEvent::Usage(completion.usage),
                label.clone(),
            ));
        }
        accumulate_continuation(&mut completion, continuation, exchanges, told);
        Ok(completion)
    }
}

fn tool_result_json(exchange: &ToolExchange) -> Value {
    json!({
        "type": "tool_result",
        "tool_use_id": exchange.call.id,
        "output": exchange.output,
        "failed": exchange.failed,
    })
}

/// Converse refuses two user messages in a row, and a person's turn right
/// after the prompt — on a first call, or first in the history — is the
/// second: it joins the prompt's last message.
fn join_at_the_prompt(
    mut messages: Vec<Message>,
    prompt_len: usize,
    model: &ModelId,
) -> Result<Vec<Message>, ModelError> {
    let user = |m: &Message| m.role() == &ConversationRole::User;
    if prompt_len == 0
        || prompt_len >= messages.len()
        || !user(&messages[prompt_len - 1])
        || !user(&messages[prompt_len])
    {
        return Ok(messages);
    }
    let next = messages.remove(prompt_len);
    let last = &messages[prompt_len - 1];
    let mut content = last.content().to_vec();
    content.extend(next.content().iter().cloned());
    messages[prompt_len - 1] = Message::builder()
        .role(ConversationRole::User)
        .set_content(Some(content))
        .build()
        .map_err(|error| Bedrock::refused(model, error.to_string()))?;
    Ok(messages)
}

/// The user turn that follows an answer: the tool results, then what a
/// person said, as one message in the continuation's JSON shape.
fn user_turn(exchanges: &[ToolExchange], told: &[String]) -> Value {
    json!({
        "role": "user",
        "content": exchanges
            .iter()
            .map(tool_result_json)
            .chain(told.iter().map(|text| json!({ "type": "text", "text": text })))
            .collect::<Vec<_>>(),
    })
}

fn accumulate_continuation(
    completion: &mut Completion,
    prior: Option<&ProviderContinuation>,
    exchanges: &[ToolExchange],
    told: &[String],
) {
    let Some(current) = completion.continuation.as_mut() else {
        return;
    };
    let mut transcript = prior
        .and_then(|value| value.state.as_array())
        .cloned()
        .unwrap_or_default();
    if !exchanges.is_empty() || !told.is_empty() {
        transcript.push(user_turn(exchanges, told));
    }
    if let Some(messages) = current.state.as_array() {
        transcript.extend(messages.iter().cloned());
    }
    current.state = Value::Array(transcript);
}

/// A breakpoint, as Converse spells one: a block of its own after what it
/// closes.
fn cache_point(ttl: CacheTtl, model: &ModelId) -> Result<CachePointBlock, ModelError> {
    CachePointBlock::builder()
        .r#type(CachePointType::Default)
        .ttl(match ttl {
            CacheTtl::FiveMinutes => WireCacheTtl::FiveMinutes,
            CacheTtl::OneHour => WireCacheTtl::OneHour,
        })
        .build()
        .map_err(|error| Bedrock::refused(model, error.to_string()))
}

/// The instruction as Converse system blocks, each breakpoint a block of its
/// own after the part it closes.
fn system_blocks(prompt: &Value, model: &ModelId) -> Result<Vec<SystemContentBlock>, ModelError> {
    let Some(system) = crate::model::prompt::system(prompt) else {
        return Ok(Vec::new());
    };
    let crate::model::prompt::System::Parts(parts) = system else {
        return Err(Bedrock::refused(
            model,
            "Bedrock's system instruction must be a string or a list of text parts",
        ));
    };
    let mut blocks = Vec::with_capacity(parts.len());
    for part in parts {
        blocks.push(SystemContentBlock::Text(part.text.into_owned()));
        if let Some(ttl) = part.cache {
            blocks.push(SystemContentBlock::CachePoint(cache_point(ttl, model)?));
        }
    }
    Ok(blocks)
}

/// Converse's usage, folded as every driver folds it, with the one-hour share
/// of the cache writes read from its per-tier detail.
pub(crate) fn usage_from(usage: &aws_sdk_bedrockruntime::types::TokenUsage) -> Usage {
    let count = |n: i32| u64::try_from(n).unwrap_or_default();
    let one_hour = usage
        .cache_details()
        .iter()
        .filter(|detail| *detail.ttl() == WireCacheTtl::OneHour)
        .map(|detail| count(detail.input_tokens()))
        .sum::<u64>();
    Usage::with_cache_beside_input(
        count(usage.input_tokens()),
        count(usage.output_tokens()),
        count(usage.cache_write_input_tokens().unwrap_or_default()),
        count(usage.cache_read_input_tokens().unwrap_or_default()),
    )
    .with_cache_write_1h(one_hour)
}

fn tool(
    name: &str,
    description: &str,
    schema: &Value,
    strict: bool,
    model: &ModelId,
) -> Result<Tool, ModelError> {
    let specification = ToolSpecification::builder()
        .name(name)
        .description(description)
        .input_schema(ToolInputSchema::Json(document_from_json(schema)))
        .strict(strict)
        .build()
        .map_err(|error| Bedrock::refused(model, error.to_string()))?;
    Ok(Tool::ToolSpec(specification))
}

/// Classify a failed SDK call by how far it got.
///
/// Only a request the SDK could not build is known never to have reached AWS;
/// a dispatch failure may have been sent, and a timeout or an unreadable
/// response certainly was, so each of those is `Unavailable` — reached, with
/// nothing said about whether it generated. The exception is a credential that
/// could not be resolved, which is decided before anything is sent: see
/// [`unresolvable_identity`].
///
/// A service error's detail is AWS's own code, message and request id — what
/// an operator quotes to AWS. Anything else carries the SDK's whole source
/// chain, because its own `Display` says only "dispatch failure" and the cause
/// is underneath.
fn classify_sdk<E, R>(
    model: &ModelId,
    error: &aws_sdk_bedrockruntime::error::SdkError<E, R>,
    service: impl FnOnce(&E, String) -> ModelError,
) -> ModelError
where
    E: std::error::Error
        + aws_sdk_bedrockruntime::error::ProvideErrorMetadata
        + aws_sdk_bedrockruntime::operation::RequestId
        + 'static,
    R: std::fmt::Debug + 'static,
{
    use aws_sdk_bedrockruntime::error::{DisplayErrorContext, SdkError};
    if let SdkError::ServiceError(context) = error {
        let err = context.err();
        return service(err, service_detail(err));
    }
    let detail = DisplayErrorContext(error).to_string();
    if unresolvable_identity(error) {
        return ModelError::Refused {
            account: true,
            model: model.clone(),
            code: None,
            detail,
        };
    }
    match error {
        SdkError::ConstructionFailure(_) => ModelError::Unreachable {
            model: model.clone(),
            detail,
        },
        _ => ModelError::Unavailable {
            model: model.clone(),
            detail,
        },
    }
}

/// `<code>: <message> (request <id>)`, from AWS's error metadata.
fn service_detail<E>(error: &E) -> String
where
    E: std::error::Error
        + aws_sdk_bedrockruntime::error::ProvideErrorMetadata
        + aws_sdk_bedrockruntime::operation::RequestId,
{
    let code = error.code().unwrap_or("ServiceError");
    let message = error
        .message()
        .map_or_else(|| error.to_string(), str::to_owned);
    match error.request_id() {
        Some(id) => format!("{code}: {message} (request {id})"),
        None => format!("{code}: {message}"),
    }
}

/// Whether the call failed resolving credentials its configuration cannot
/// produce, which refuses the account before anything reached Bedrock.
///
/// `InvalidConfiguration`, and `Unhandled`, which is where aws-config puts an
/// expired `aws login` token and an unreadable token file: a person has to
/// sign in or fix the profile. Not the rest, which an outage can cause and a
/// retry can fix: `ProviderTimedOut`, `ProviderError` (an STS 503, a token
/// refresh that could not connect — and an expired SSO session, which
/// aws-config reports the same way) and `CredentialsNotLoaded` (which an
/// unreachable instance metadata service also ends in).
fn unresolvable_identity(error: &(dyn std::error::Error + 'static)) -> bool {
    use aws_credential_types::provider::error::CredentialsError;
    let mut cause = Some(error);
    while let Some(inner) = cause {
        if let Some(credentials) = inner.downcast_ref::<CredentialsError>() {
            return matches!(
                credentials,
                CredentialsError::InvalidConfiguration(_) | CredentialsError::Unhandled(_)
            );
        }
        cause = inner.source();
    }
    false
}

/// The service codes that refuse the **account**: the credential is missing a
/// permission, unknown, expired or signed wrongly. Matched by code rather than
/// by typed variant, because only `AccessDeniedException` is modelled — the
/// rest arrive as unmodelled errors that carry nothing but their code.
const ACCOUNT_CODES: [&str; 4] = [
    "AccessDeniedException",
    "UnrecognizedClientException",
    "InvalidSignatureException",
    "ExpiredTokenException",
];

/// Classify a Converse service error, buffered or at stream start.
///
/// A throttle is `RateLimited` with no window: the SDK models it as a typed
/// error and does not surface the response headers at this seam, and the AWS
/// client applies its own adaptive retry beneath this call, so a second
/// schedule stacked on it would multiply rather than add. A refused
/// credential ([`ACCOUNT_CODES`]) refuses the account; lookup and validation
/// failures refuse the request. Anything else — a timeout, a model or service
/// fault — may have generated, and Converse exposes no partial usage for it,
/// so it is `Unavailable` rather than free or absent.
fn classify_service(
    model: &ModelId,
    detail: String,
    code: Option<&str>,
    throttled: bool,
    refused: bool,
) -> ModelError {
    let account = code.is_some_and(|code| ACCOUNT_CODES.contains(&code));
    if throttled {
        ModelError::RateLimited {
            model: model.clone(),
            detail,
            retry_after: None,
        }
    } else if refused || account {
        ModelError::Refused {
            account,
            model: model.clone(),
            code: code.and_then(|code| super::bounded_code(code.to_owned())),
            detail,
        }
    } else {
        ModelError::Unavailable {
            model: model.clone(),
            detail,
        }
    }
}

fn severed_stream(
    model: &ModelId,
    accumulator: &super::bedrock_stream::Accumulator,
    detail: &str,
) -> ModelError {
    if let Some(usage) = accumulator.usage() {
        return ModelError::Interrupted {
            model: model.clone(),
            usage,
            detail: detail.to_owned(),
        };
    }
    if accumulator.generated() {
        return ModelError::Unaccounted {
            model: model.clone(),
            detail: detail.to_owned(),
        };
    }
    ModelError::Unavailable {
        model: model.clone(),
        detail: detail.to_owned(),
    }
}

fn classify_stream_event(
    model: &ModelId,
    accumulator: &super::bedrock_stream::Accumulator,
    error: Option<&aws_sdk_bedrockruntime::types::error::ConverseStreamOutputError>,
    detail: &str,
) -> ModelError {
    if accumulator.started() || accumulator.generated() {
        return severed_stream(model, accumulator, detail);
    }
    if error.is_some_and(
        aws_sdk_bedrockruntime::types::error::ConverseStreamOutputError::is_throttling_exception,
    ) {
        ModelError::RateLimited {
            model: model.clone(),
            detail: detail.to_owned(),
            // See `classify_service`: the SDK does not surface response
            // headers here, and it retries throttles itself beneath this call.
            retry_after: None,
        }
    } else if error.is_some_and(
        aws_sdk_bedrockruntime::types::error::ConverseStreamOutputError::is_validation_exception,
    ) {
        ModelError::Refused {
            account: false,
            model: model.clone(),
            code: None,
            detail: detail.to_owned(),
        }
    } else {
        ModelError::Unavailable {
            model: model.clone(),
            detail: detail.to_owned(),
        }
    }
}

pub(super) fn document_from_json(value: &Value) -> Document {
    match value {
        Value::Null => Document::Null,
        Value::Bool(value) => Document::Bool(*value),
        Value::String(value) => Document::String(value.clone()),
        Value::Array(values) => Document::Array(values.iter().map(document_from_json).collect()),
        Value::Object(values) => Document::Object(
            values
                .iter()
                .map(|(key, value)| (key.clone(), document_from_json(value)))
                .collect::<HashMap<_, _>>(),
        ),
        Value::Number(value) => {
            if let Some(value) = value.as_u64() {
                Document::Number(Number::PosInt(value))
            } else if let Some(value) = value.as_i64() {
                Document::Number(Number::NegInt(value))
            } else {
                Document::Number(Number::Float(
                    value.as_f64().expect("JSON number is finite"),
                ))
            }
        }
    }
}

pub(super) fn json_from_document(value: &Document) -> Value {
    match value {
        Document::Null => Value::Null,
        Document::Bool(value) => Value::Bool(*value),
        Document::String(value) => Value::String(value.clone()),
        Document::Array(values) => Value::Array(values.iter().map(json_from_document).collect()),
        Document::Object(values) => Value::Object(
            values
                .iter()
                .map(|(key, value)| (key.clone(), json_from_document(value)))
                .collect(),
        ),
        Document::Number(Number::PosInt(value)) => json!(value),
        Document::Number(Number::NegInt(value)) => json!(value),
        Document::Number(Number::Float(value)) => json!(value),
    }
}

fn content_to_json(
    block: &ContentBlock,
    model: &ModelId,
    usage: Usage,
) -> Result<Value, ModelError> {
    match block {
        ContentBlock::Text(text) => Ok(json!({"type": "text", "text": text})),
        ContentBlock::ToolUse(tool) => Ok(json!({
            "type": "tool_use",
            "id": tool.tool_use_id(),
            "name": tool.name(),
            "input": json_from_document(tool.input()),
        })),
        ContentBlock::ReasoningContent(ReasoningContentBlock::ReasoningText(reasoning)) => {
            Ok(json!({
                "type": "reasoning",
                "text": reasoning.text(),
                "signature": reasoning.signature(),
            }))
        }
        ContentBlock::ReasoningContent(ReasoningContentBlock::RedactedContent(bytes)) => {
            Ok(json!({
                "type": "redacted_reasoning",
                "data": crate::core::b64::encode(bytes.as_ref()),
            }))
        }
        _ => Err(ModelError::Unusable {
            model: model.clone(),
            usage,
            detail: "Bedrock returned an unknown or unsupported continuation content block"
                .to_owned(),
        }),
    }
}

fn content_from_json(value: &Value, model: &ModelId) -> Result<ContentBlock, ModelError> {
    let kind = value
        .get("type")
        .and_then(Value::as_str)
        .ok_or_else(|| Bedrock::refused(model, "a Bedrock continuation block has no type"))?;
    match kind {
        "text" => Ok(ContentBlock::Text(
            value
                .get("text")
                .and_then(Value::as_str)
                .ok_or_else(|| Bedrock::refused(model, "a text block has no text"))?
                .to_owned(),
        )),
        "tool_use" => Ok(ContentBlock::ToolUse(
            ToolUseBlock::builder()
                .tool_use_id(
                    value
                        .get("id")
                        .and_then(Value::as_str)
                        .ok_or_else(|| Bedrock::refused(model, "a tool block has no id"))?,
                )
                .name(
                    value
                        .get("name")
                        .and_then(Value::as_str)
                        .ok_or_else(|| Bedrock::refused(model, "a tool block has no name"))?,
                )
                .input(document_from_json(
                    value.get("input").unwrap_or(&Value::Null),
                ))
                .build()
                .map_err(|error| Bedrock::refused(model, error.to_string()))?,
        )),
        "reasoning" => {
            let mut builder = ReasoningTextBlock::builder().text(
                value
                    .get("text")
                    .and_then(Value::as_str)
                    .ok_or_else(|| Bedrock::refused(model, "a reasoning block has no text"))?,
            );
            if let Some(signature) = value.get("signature").and_then(Value::as_str) {
                builder = builder.signature(signature);
            }
            Ok(ContentBlock::ReasoningContent(
                ReasoningContentBlock::ReasoningText(
                    builder
                        .build()
                        .map_err(|error| Bedrock::refused(model, error.to_string()))?,
                ),
            ))
        }
        "redacted_reasoning" => {
            let encoded = value
                .get("data")
                .and_then(Value::as_str)
                .ok_or_else(|| Bedrock::refused(model, "redacted reasoning has no data"))?;
            let bytes = crate::core::b64::decode(encoded).ok_or_else(|| {
                Bedrock::refused(model, "redacted reasoning data is not canonical base64")
            })?;
            Ok(ContentBlock::ReasoningContent(
                ReasoningContentBlock::RedactedContent(Blob::new(bytes)),
            ))
        }
        other => Err(Bedrock::refused(
            model,
            format!("unsupported Bedrock continuation block '{other}'"),
        )),
    }
}

fn tool_result_from_json(value: &Value, model: &ModelId) -> Result<ContentBlock, ModelError> {
    let id = value
        .get("tool_use_id")
        .and_then(Value::as_str)
        .ok_or_else(|| Bedrock::refused(model, "a tool result has no tool_use_id"))?;
    let status = if value.get("failed").and_then(Value::as_bool) == Some(true) {
        ToolResultStatus::Error
    } else {
        ToolResultStatus::Success
    };
    Ok(ContentBlock::ToolResult(
        ToolResultBlock::builder()
            .tool_use_id(id)
            .content(tool_result_content(
                value.get("output").unwrap_or(&Value::Null),
            ))
            .status(status)
            .build()
            .map_err(|error| Bedrock::refused(model, error.to_string()))?,
    ))
}

/// Converse takes only an object as a JSON tool result and refuses the whole
/// request for anything else, so every other output — a list, a number, a
/// failed tool's message — goes as text, the way the other wires send it.
fn tool_result_content(output: &Value) -> ToolResultContentBlock {
    match output {
        Value::Object(_) => ToolResultContentBlock::Json(document_from_json(output)),
        Value::String(text) => ToolResultContentBlock::Text(text.clone()),
        other => ToolResultContentBlock::Text(other.to_string()),
    }
}

fn continuation_message_from_json(value: &Value, model: &ModelId) -> Result<Message, ModelError> {
    let role = value
        .get("role")
        .and_then(Value::as_str)
        .ok_or_else(|| Bedrock::refused(model, "a continuation message has no role"))?;
    let blocks = value
        .get("content")
        .and_then(Value::as_array)
        .ok_or_else(|| Bedrock::refused(model, "a continuation message has no content array"))?;
    let (role, content) = match role {
        "assistant" => (
            ConversationRole::Assistant,
            blocks
                .iter()
                .map(|block| content_from_json(block, model))
                .collect::<Result<Vec<_>, _>>()?,
        ),
        "user" => (
            ConversationRole::User,
            blocks
                .iter()
                .map(|block| {
                    if block.get("type").and_then(Value::as_str) == Some("text") {
                        content_from_json(block, model)
                    } else {
                        tool_result_from_json(block, model)
                    }
                })
                .collect::<Result<Vec<_>, _>>()?,
        ),
        other => {
            return Err(Bedrock::refused(
                model,
                format!("unsupported continuation role '{other}'"),
            ));
        }
    };
    Message::builder()
        .role(role)
        .set_content(Some(content))
        .build()
        .map_err(|error| Bedrock::refused(model, error.to_string()))
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Converse reports the cache counters **beside** `inputTokens`.
    ///
    /// `Usage::input_tokens` means everything the provider processed, cached
    /// tokens included — copied verbatim, the count under-reports by the whole
    /// cached prefix and the token ceiling reads a fraction of the real spend,
    /// surfacing as a bill nobody can reconcile rather than as a failure.
    #[test]
    fn cache_tokens_are_folded_into_the_input_count() {
        let model = ModelId::new("bedrock", "test");
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .content(ContentBlock::Text("ok".to_owned()))
            .build()
            .expect("message");
        let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(aws_sdk_bedrockruntime::types::StopReason::EndTurn)
            .usage(
                aws_sdk_bedrockruntime::types::TokenUsage::builder()
                    .input_tokens(10)
                    .output_tokens(4)
                    .total_tokens(14)
                    .cache_read_input_tokens(90)
                    .cache_write_input_tokens(6)
                    .build()
                    .expect("usage"),
            )
            .build()
            .expect("converse output");

        let completion =
            Bedrock::interpret(&model, None, SchemaMode::Native, &output).expect("an answer");
        assert_eq!(
            completion.usage.input_tokens, 106,
            "10 uncached + 90 cache-read + 6 cache-write"
        );
        assert_eq!(completion.usage.cache_read_tokens, 90);
        assert_eq!(completion.usage.cache_write_tokens, 6);
    }

    /// A guardrail intervention is a metered refusal on the buffered path too.
    ///
    /// The intervention replaces the content with the guardrail's canned
    /// message; returned as a completion, that message wears a successful
    /// answer and the one path a streaming-by-default deployment never
    /// exercises is the one that lets it through.
    #[test]
    fn a_buffered_guardrail_intervention_is_a_refusal_not_an_answer() {
        let model = ModelId::new("bedrock", "test");
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .content(ContentBlock::Text(
                "Sorry, I cannot answer that.".to_owned(),
            ))
            .build()
            .expect("message");
        let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(aws_sdk_bedrockruntime::types::StopReason::GuardrailIntervened)
            .usage(
                aws_sdk_bedrockruntime::types::TokenUsage::builder()
                    .input_tokens(7)
                    .output_tokens(3)
                    .total_tokens(10)
                    .build()
                    .expect("usage"),
            )
            .build()
            .expect("converse output");

        let error = Bedrock::interpret(&model, None, SchemaMode::Native, &output)
            .expect_err("an intervention is a refusal, not an answer");
        match error {
            ModelError::Unusable { usage, .. } => {
                assert_eq!(usage.input_tokens, 7, "the assessment was billed");
            }
            other => panic!("expected a metered Unusable, got {other:?}"),
        }
    }

    /// A tool call the provider disowned never reaches the caller's tool loop.
    ///
    /// `malformed_tool_use` means Converse could not parse what the model
    /// emitted; what survives in the content blocks is a fragment. A driver
    /// that returns it hands the runtime a plausible-looking call the provider
    /// itself refused to stand behind — and the failure is metered, because
    /// the tokens were generated whatever shape they arrived in.
    #[test]
    fn a_malformed_tool_use_stop_is_unusable_not_an_answer() {
        let model = ModelId::new("bedrock", "test");
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .content(ContentBlock::Text("{\"partial\": ".to_owned()))
            .build()
            .expect("message");
        let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(aws_sdk_bedrockruntime::types::StopReason::MalformedToolUse)
            .usage(
                aws_sdk_bedrockruntime::types::TokenUsage::builder()
                    .input_tokens(7)
                    .output_tokens(3)
                    .total_tokens(10)
                    .build()
                    .expect("usage"),
            )
            .build()
            .expect("converse output");

        let error = Bedrock::interpret(&model, None, SchemaMode::Native, &output)
            .expect_err("a disowned turn is not an answer");
        match error {
            ModelError::Unusable { usage, .. } => {
                assert_eq!(usage.output_tokens, 3, "the generation was billed");
            }
            other => panic!("expected a metered Unusable, got {other:?}"),
        }
    }

    /// Completeness is an allowlist: a filtered stop, or one this SDK names
    /// only as unknown, carries text that reads like an answer and is not one.
    #[test]
    fn a_filtered_or_unknown_stop_is_unusable_not_an_answer() {
        let model = ModelId::new("bedrock", "test");
        for reason in [
            aws_sdk_bedrockruntime::types::StopReason::ContentFiltered,
            aws_sdk_bedrockruntime::types::StopReason::from("a_reason_not_yet_invented"),
        ] {
            let message = Message::builder()
                .role(ConversationRole::Assistant)
                .content(ContentBlock::Text("looks whole".to_owned()))
                .build()
                .expect("message");
            let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
                .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                    message,
                ))
                .stop_reason(reason.clone())
                .usage(
                    aws_sdk_bedrockruntime::types::TokenUsage::builder()
                        .input_tokens(7)
                        .output_tokens(3)
                        .total_tokens(10)
                        .build()
                        .expect("usage"),
                )
                .build()
                .expect("converse output");
            let error = Bedrock::interpret(&model, None, SchemaMode::Native, &output)
                .expect_err(reason.as_str());
            match error {
                ModelError::Unusable { usage, .. } => {
                    assert_eq!(usage.output_tokens, 3, "the generation was billed");
                }
                other => panic!(
                    "{}: expected a metered Unusable, got {other:?}",
                    reason.as_str()
                ),
            }
        }
    }

    #[test]
    fn smithy_documents_round_trip_json() {
        let value = json!({"s": "x", "n": -2, "u": 3, "f": 1.5, "a": [true, null]});
        assert_eq!(json_from_document(&document_from_json(&value)), value);
    }

    #[test]
    fn governed_images_and_documents_become_inline_bedrock_blocks() {
        let model = ModelId::new("bedrock", "test");
        let image = content_from_prompt_json(
            &json!({
                "type": "image",
                "media_type": "image/png",
                "data": crate::core::b64::encode(b"png")
            }),
            &model,
        )
        .expect("image");
        assert_eq!(
            image.as_image().expect("image block").format(),
            &ImageFormat::Png
        );

        let document = content_from_prompt_json(
            &json!({
                "type": "document",
                "media_type": "application/pdf",
                "data": crate::core::b64::encode(b"pdf")
            }),
            &model,
        )
        .expect("document");
        let document = document.as_document().expect("document block");
        assert_eq!(document.format(), &DocumentFormat::Pdf);
        assert_eq!(document.name(), "document");
    }

    #[test]
    fn bedrock_media_never_accepts_a_remote_source() {
        let model = ModelId::new("bedrock", "test");
        assert!(
            content_from_prompt_json(
                &json!({
                    "type": "image",
                    "media_type": "image/png",
                    "url": "https://example.test/image.png"
                }),
                &model,
            )
            .is_err()
        );
    }

    #[test]
    fn signed_and_redacted_reasoning_round_trip() {
        let model = ModelId::new("bedrock", "test");
        for value in [
            json!({"type": "reasoning", "text": "private", "signature": "sig"}),
            json!({"type": "redacted_reasoning", "data": "AQID"}),
        ] {
            let block = content_from_json(&value, &model).expect("decode block");
            assert_eq!(
                content_to_json(&block, &model, Usage::default()).unwrap(),
                value
            );
        }
    }

    /// A guardrail is part of what governed the call, so it is part of
    /// identity — and its two forms agree about which guardrail they name.
    #[test]
    fn a_guardrail_is_effect_identity_and_both_paths_send_the_same_one() {
        let config = aws_sdk_bedrockruntime::Config::builder()
            .region(Region::new("eu-west-1"))
            .behavior_version_latest()
            .http_client(crate::model::canned_http(200, None, ""))
            .build();
        let client = Client::from_conf(config);
        let plain = Bedrock::from_client(client.clone()).expect("a client with a region");
        let model = ModelId::new("bedrock", "m");
        let guarded = Bedrock::from_client(client)
            .expect("a client with a region")
            .guardrail(Guardrail::new("gr-7", "3"));

        assert_eq!(plain.request_profile(&model)["guardrail"], Value::Null);
        assert_eq!(
            guarded.request_profile(&model)["guardrail"],
            json!({ "id": "gr-7", "version": "3" }),
            "a guardrail that does not enter the request profile can be turned \
             off between a run and its replay with nothing on the record"
        );
        assert_ne!(
            plain.request_profile(&model),
            guarded.request_profile(&model),
            "guarded and unguarded calls shared one effect identity"
        );

        // The buffered and streaming forms must name the same guardrail: a
        // control applied on one path only is one a `stream: true` deployment
        // silently loses.
        let g = Guardrail::new("gr-7", "3");
        assert_eq!(g.config().guardrail_identifier(), "gr-7");
        assert_eq!(g.stream_config().guardrail_identifier(), "gr-7");
        assert_eq!(g.config().guardrail_version(), "3");
        assert_eq!(g.stream_config().guardrail_version(), "3");
        // Synchronous: the alternative streams blocked content to the caller
        // and objects afterwards, which is a control arriving after the thing
        // it exists to prevent.
        assert_eq!(
            g.stream_config().stream_processing_mode().as_str(),
            "sync",
            "an asynchronous guardrail releases content before assessing it"
        );

        // Trace is off by default and reaches both request forms when asked for,
        // because the intervention reason it returns is journaled — and a
        // control the streaming path drops is one a `stream: true` deployment
        // loses.
        assert_eq!(g.config().trace().as_str(), "disabled");
        assert_eq!(g.stream_config().trace().as_str(), "disabled");
        let traced = Guardrail::new("gr-7", "3").with_trace();
        assert_eq!(traced.config().trace().as_str(), "enabled");
        assert_eq!(
            traced.stream_config().trace().as_str(),
            "enabled",
            "with_trace must reach the streaming path too"
        );
    }

    #[test]
    fn request_profile_commits_to_region_and_buffering() {
        let config = aws_sdk_bedrockruntime::Config::builder()
            .region(Region::new("eu-west-1"))
            .behavior_version_latest()
            // A stub client, so no TLS provider is built: the default one
            // eagerly reads the OS trust store, and under a fully parallel
            // suite the macOS keychain read can transiently yield zero roots
            // — a panic inside aws-smithy, in a test that asserts a JSON
            // profile and never opens a connection.
            .http_client(crate::model::canned_http(200, None, ""))
            .build();
        let driver =
            Bedrock::from_client(Client::from_conf(config)).expect("a client with a region");
        assert_eq!(
            driver.request_profile(&ModelId::new("bedrock", "m")),
            json!({
                "driver": "aws-bedrock-converse/v1",
                "region": "eu-west-1",
                "stream": true,
                "schema_mode": "native",
                // Present and null when unguarded: an absent key and a null
                // one are the same to a reader and different to a digest, so
                // the profile states the answer rather than omitting it.
                "guardrail": Value::Null,
                "reasoning_dialect": Value::Null,
            })
        );

        assert_eq!(
            driver.request_profile(&ModelId::new("bedrock", "m")),
            driver
                .clone()
                .timeout(Duration::from_secs(7))
                .request_profile(&ModelId::new("bedrock", "m")),
            "a timeout is how long this plane waits, not what it asks — it must not \
             enter the effect identity"
        );

        assert_ne!(
            driver.request_profile(&ModelId::new("bedrock", "m")),
            driver
                .buffered()
                .request_profile(&ModelId::new("bedrock", "m")),
            "buffered and streamed Bedrock requests reused one effect identity"
        );
    }

    /// **The attested region is the client's, and there is no second copy of it
    /// to disagree.**
    ///
    /// `region` is what `request_profile` puts on the record, and that profile
    /// is effect identity — so a replay is judged against it. Taken as an
    /// argument beside the client it is a fact stored twice, and the copy the
    /// journal attests is the one nothing checks: a driver built with a
    /// `us-east-1` client and the string `"eu-west-1"` would send every call to
    /// Virginia and swear to Ireland, on a record built to be evidence.
    ///
    /// The positive half is what does the work here. That a blank region is
    /// refused proves only that a constructor validates its input; that the
    /// profile *follows the client* is the property, and it is the half that
    /// fails if the parameter ever comes back.
    #[test]
    fn the_attested_region_is_the_one_the_client_will_reach() {
        let stub = |region: Option<&str>| {
            let mut config = aws_sdk_bedrockruntime::Config::builder()
                .behavior_version_latest()
                .http_client(crate::model::canned_http(200, None, ""));
            if let Some(region) = region {
                config = config.region(Region::new(region.to_owned()));
            }
            Bedrock::from_client(Client::from_conf(config.build()))
        };

        for region in ["us-east-1", "eu-west-1"] {
            let driver = stub(Some(region)).expect("a client carrying a region");
            assert_eq!(
                driver.request_profile(&ModelId::new("bedrock", "m"))["region"],
                json!(region),
                "the profile attested a region other than the one the client \
                 resolves its endpoint from"
            );
        }

        // A client with no region cannot name where it went, and an empty
        // string on the record is worse than a refusal: it is a destination
        // that reads as answered.
        let err = stub(None).expect_err("a client with no region was accepted");
        assert!(
            err.contains("region"),
            "the refusal must name the missing region, got: {err}"
        );
        let err = stub(Some("   ")).expect_err("a blank region was accepted");
        assert!(
            err.contains("region"),
            "the refusal must name the blank region, got: {err}"
        );
    }

    /// A Bedrock driver whose every answer is `status` with `body`, through
    /// the real SDK's error parsing, signing with `credentials`.
    fn answering(
        status: u16,
        body: &'static str,
        credentials: aws_sdk_bedrockruntime::config::SharedCredentialsProvider,
        buffered: bool,
    ) -> Bedrock {
        let config = aws_sdk_bedrockruntime::Config::builder()
            .region(Region::new("eu-west-1"))
            .behavior_version_latest()
            .credentials_provider(credentials)
            .retry_config(aws_sdk_bedrockruntime::config::retry::RetryConfig::disabled())
            .http_client(crate::model::canned_http(
                status,
                Some("application/json"),
                body,
            ))
            .build();
        let driver = Bedrock::from_client(Client::from_conf(config)).expect("a region");
        if buffered { driver.buffered() } else { driver }
    }

    async fn refusal_of(driver: &Bedrock) -> ModelError {
        let model = ModelId::new("bedrock", "anthropic.claude-x");
        let prompt = json!("hello");
        driver
            .complete(Request {
                model: &model,
                prompt: &prompt,
                max_output_tokens: 64,
                reasoning_effort: None,
                reasoning_summary: false,
                schema: None,
                tools: &[],
                exchanges: &[],
                told: &[],
                continuation: None,
                compaction: None,
                stream: None,
            })
            .await
            .expect_err("every answer is an error")
    }

    fn signing() -> aws_sdk_bedrockruntime::config::SharedCredentialsProvider {
        aws_sdk_bedrockruntime::config::SharedCredentialsProvider::new(
            aws_sdk_bedrockruntime::config::Credentials::for_tests(),
        )
    }

    /// **A refused credential refuses the account, on both paths.** Only
    /// `AccessDeniedException` is modelled; the others arrive as unmodelled
    /// errors with nothing but a code, and were `Unavailable` — retried, and
    /// never the account's. A lookup or validation failure stays the
    /// request's. The detail is AWS's code, message and request id, not the
    /// SDK's dump of the raw response.
    #[tokio::test]
    async fn a_refused_credential_refuses_the_account() {
        const EXPIRED: &str = r#"{"__type":"ExpiredTokenException","message":"The security token included in the request is expired"}"#;
        const UNKNOWN: &str = r#"{"__type":"UnrecognizedClientException","message":"The security token included in the request is invalid."}"#;
        const DENIED: &str = r#"{"__type":"AccessDeniedException","message":"You don't have access to the model with the specified model ID."}"#;
        const INVALID: &str = r#"{"__type":"ValidationException","message":"The provided model identifier is invalid."}"#;
        for buffered in [true, false] {
            for (status, body, code) in [
                (403, EXPIRED, "ExpiredTokenException"),
                (403, UNKNOWN, "UnrecognizedClientException"),
                (403, DENIED, "AccessDeniedException"),
            ] {
                let refused = refusal_of(&answering(status, body, signing(), buffered)).await;
                assert!(
                    refused.refuses_the_account(),
                    "{code} (buffered: {buffered}) is the account's: {refused:?}"
                );
                assert_eq!(refused.code().as_deref(), Some(code));
                let text = refused.to_string();
                assert!(text.contains(&format!("{code}: ")), "{text}");
                assert!(!text.contains("raw: Response"), "the SDK's dump: {text}");
            }
            let request = refusal_of(&answering(400, INVALID, signing(), buffered)).await;
            assert!(
                matches!(request, ModelError::Refused { account: false, .. }),
                "a validation failure is the request's: {request:?}"
            );
        }
    }

    /// **A credential that cannot be resolved is decided before anything is
    /// sent.** An expired `aws login` token reaches the client as
    /// `CredentialsError::Unhandled` and a broken profile as
    /// `InvalidConfiguration`: the account's, so no retry is spent on them. A
    /// provider error — an STS 503, a refresh that could not connect — stays
    /// an outage, retried.
    #[tokio::test]
    async fn a_credential_that_cannot_be_resolved_refuses_the_account() {
        use aws_credential_types::provider::error::CredentialsError;
        use aws_sdk_bedrockruntime::config::SharedCredentialsProvider;
        fn failing(error: fn() -> CredentialsError) -> SharedCredentialsProvider {
            SharedCredentialsProvider::new(
                aws_credential_types::credential_fn::provide_credentials_fn(move || async move {
                    Err(error())
                }),
            )
        }
        for (error, account) in [
            (
                (|| CredentialsError::unhandled("the login token has expired"))
                    as fn() -> CredentialsError,
                true,
            ),
            (
                || CredentialsError::invalid_configuration("no such profile"),
                true,
            ),
            (
                || CredentialsError::provider_error("STS answered 503"),
                false,
            ),
            (|| CredentialsError::not_loaded("nothing configured"), false),
        ] {
            let refused = refusal_of(&answering(200, "", failing(error), true)).await;
            if account {
                assert!(
                    matches!(
                        refused,
                        ModelError::Refused {
                            account: true,
                            code: None,
                            ..
                        }
                    ),
                    "{refused:?}"
                );
            } else {
                assert!(
                    matches!(refused, ModelError::Unavailable { .. }),
                    "{refused:?}"
                );
            }
        }
    }

    fn stub_driver() -> Bedrock {
        let config = aws_sdk_bedrockruntime::Config::builder()
            .region(Region::new("eu-west-1"))
            .behavior_version_latest()
            .http_client(crate::model::canned_http(200, None, ""))
            .build();
        Bedrock::from_client(Client::from_conf(config)).expect("a client with a region")
    }

    /// **Nova's reasoning dialect, rendered exactly as AWS documents it.**
    ///
    /// The shape is a known answer taken from Amazon's own Nova 2 extended-thinking
    /// page, not from a round trip through this driver: `reasoningConfig` with a
    /// `type` and a `maxReasoningEffort` of `low`, `medium` or `high`. A round
    /// trip would agree with itself under any spelling at all, including a wrong
    /// one, and Bedrock answers a misspelled `additionalModelRequestFields` key
    /// by **ignoring it** — so the failure mode this pins is a deployment that
    /// declared `high` effort, was billed for none, and had nothing on the record
    /// to say so.
    #[test]
    fn nova_reasoning_effort_is_rendered_the_way_aws_documents_it() {
        use super::super::ReasoningEffort as E;

        let nova = stub_driver().reasoning(ReasoningDialect::Nova);
        let model = ModelId::new("bedrock", "us.amazon.nova-2-lite-v1:0");

        for (effort, expected) in [(E::Low, "low"), (E::Medium, "medium"), (E::High, "high")] {
            let document = nova
                .reasoning_config(&model, Some(effort))
                .expect("a documented level maps")
                .expect("an effort produces a config");
            assert_eq!(
                json_from_document(&document),
                json!({ "reasoningConfig": { "type": "enabled", "maxReasoningEffort": expected } }),
                "the {} level did not render as AWS documents it",
                effort.as_str()
            );
        }

        // `None` is a *statement*, not an omission: a deployment saying this
        // call must not reason. Sending `disabled` puts that on the wire and in
        // the journal; sending nothing would be indistinguishable from a
        // deployment that never considered the question.
        assert_eq!(
            json_from_document(
                &nova
                    .reasoning_config(&model, Some(E::None))
                    .expect("none maps")
                    .expect("none is still a config")
            ),
            json!({ "reasoningConfig": { "type": "disabled" } })
        );

        // No effort asked for sends nothing at all. An empty `reasoningConfig`
        // is a different request from no `reasoningConfig`.
        assert!(
            nova.reasoning_config(&model, None)
                .expect("no effort is not an error")
                .is_none()
        );
    }

    /// **An effort Nova cannot carry is refused, never collapsed.**
    ///
    /// The positive half above would pass with a mapping that folded every
    /// unknown level into `high` — and that mapping is the defect, because
    /// `reasoning_effort` is digest-covered: it claims to describe what governed
    /// the call. Answering a request for `max` with the third of three levels is
    /// a substitution nothing downstream can see, on the one value that exists
    /// to be seen.
    ///
    /// Same rule as the Anthropic driver, which refuses the levels adaptive
    /// thinking cannot express rather than approximating them.
    #[test]
    fn nova_refuses_an_effort_it_has_no_counterpart_for() {
        use super::super::ReasoningEffort as E;

        let nova = stub_driver().reasoning(ReasoningDialect::Nova);
        let model = ModelId::new("bedrock", "us.amazon.nova-2-lite-v1:0");

        for effort in [E::Minimal, E::XHigh, E::Max] {
            let error = nova
                .reasoning_config(&model, Some(effort))
                .expect_err("an effort Nova cannot express must be refused");
            let text = error.to_string();
            assert!(
                text.contains(effort.as_str()) && text.contains("low, medium and high"),
                "the refusal must name the effort and the levels that exist: {text}"
            );
        }
    }

    /// **An undeclared dialect still refuses, and the message says what to do.**
    ///
    /// Converse is one envelope over families that spell reasoning differently
    /// or not at all, so a driver with no declared dialect has nothing to
    /// derive and guessing is the failure. The refusal must also *teach*:
    /// its message names `.reasoning(..)`, the one call that answers it.
    #[test]
    fn an_undeclared_dialect_refuses_rather_than_guessing() {
        let plain = stub_driver();
        let model = ModelId::new("bedrock", "m");
        let text = plain
            .reasoning_config(&model, Some(super::super::ReasoningEffort::Medium))
            .expect_err("no dialect must refuse")
            .to_string();
        assert!(
            text.contains("ReasoningDialect::Nova"),
            "a refusal that does not name the remedy leaves the reader stuck: {text}"
        );
    }

    /// **The dialect is effect identity, on both request paths.**
    ///
    /// Two properties, and the guardrail above is the precedent for each. The
    /// dialect decides how a declared effort is rendered, so switching it
    /// changes what governed the call and a replay of older history must report
    /// divergence rather than answering under the new rendering. And the config
    /// is built once and handed to `converse` and `converse_stream` alike,
    /// because a control applied on one path only is one a `stream: true`
    /// deployment silently loses.
    #[test]
    fn the_reasoning_dialect_is_effect_identity_and_reaches_both_paths() {
        let model = ModelId::new("bedrock", "us.amazon.nova-2-lite-v1:0");
        let plain = stub_driver();
        let nova = stub_driver().reasoning(ReasoningDialect::Nova);

        assert_eq!(
            plain.request_profile(&model)["reasoning_dialect"],
            Value::Null
        );
        assert_eq!(nova.request_profile(&model)["reasoning_dialect"], "nova");
        assert_ne!(
            plain.request_profile(&model),
            nova.request_profile(&model),
            "a driver that renders reasoning differently shared one effect identity"
        );

        // Buffered and streamed differ only in `stream`, so the dialect must
        // survive that difference — the streaming path takes the config as an
        // argument rather than rebuilding it, and this is what says so.
        assert_eq!(
            nova.clone().buffered().request_profile(&model)["reasoning_dialect"],
            nova.request_profile(&model)["reasoning_dialect"],
        );
    }

    #[test]
    fn native_schema_can_coexist_with_tools_but_forced_fallback_cannot() {
        let model = ModelId::new("bedrock", "test");
        let schema = json!({"type": "object"});
        let tools = vec![ToolDeclaration::new(
            "lookup",
            "Look something up.",
            json!({"type": "object"}),
        )];
        assert!(
            Bedrock::tool_config(&model, Some(&schema), &tools, SchemaMode::Native)
                .expect("native schema and tools")
                .is_some()
        );
        assert!(
            Bedrock::tool_config(&model, Some(&schema), &tools, SchemaMode::ForcedTool).is_err(),
            "forced-tool structured output consumed the same tool channel as caller tools"
        );
    }

    /// Reasoning + `toolUse` + no sibling text is a working tool turn.
    ///
    /// With extended thinking on, choosing a tool IS the answer: Converse
    /// returns `reasoningContent` and `toolUse` blocks with no text beside
    /// them. This test pins two things at once. First, that the emptiness
    /// guard in `interpret` requires *both* no text and no tool call before
    /// declaring the answer unusable — weaken it to text alone and this fails.
    /// Second, that the continuation carries the reasoning block byte-for-byte
    /// beside the tool call, signature included, because Bedrock rejects a
    /// follow-up turn whose reasoning does not return its signature and a
    /// driver that rebuilt the turn from the fields it understands cannot
    /// return what it never kept. What this does NOT cover is the streaming
    /// accumulator's reassembly of those blocks — that has its own tests in
    /// `bedrock_stream.rs`; both paths funnel into this same `interpret`.
    #[test]
    fn a_reasoning_tool_turn_with_no_text_is_an_answer_not_an_empty_one() {
        let model = ModelId::new("bedrock", "test");
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .content(ContentBlock::ReasoningContent(
                ReasoningContentBlock::ReasoningText(
                    ReasoningTextBlock::builder()
                        .text("private reasoning")
                        .signature("sig-bytes")
                        .build()
                        .expect("reasoning block"),
                ),
            ))
            .content(ContentBlock::ToolUse(
                ToolUseBlock::builder()
                    .tool_use_id("tool_1")
                    .name("lookup")
                    .input(document_from_json(&json!({ "id": "AC-1" })))
                    .build()
                    .expect("tool block"),
            ))
            .build()
            .expect("assistant message");
        let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(aws_sdk_bedrockruntime::types::StopReason::ToolUse)
            .build()
            .expect("converse output");

        let completion = Bedrock::interpret(&model, None, SchemaMode::Native, &output)
            .expect("a tool call with empty sibling text is a normal tool turn");
        assert_eq!(completion.tool_calls.len(), 1);
        assert_eq!(completion.tool_calls[0].name, "lookup");
        let state = completion
            .continuation
            .expect("a tool turn must carry its provider continuation")
            .state;
        assert_eq!(
            state,
            json!([{ "role": "assistant", "content": [
                { "type": "reasoning", "text": "private reasoning", "signature": "sig-bytes" },
                { "type": "tool_use", "id": "tool_1", "name": "lookup", "input": { "id": "AC-1" } },
            ]}]),
            "the continuation must carry the provider turn verbatim, reasoning \
             block and signature included"
        );
    }

    /// The twin of the test above: no text AND no tool call stays unusable.
    ///
    /// Together the pair pins the guard to exactly `text.is_empty() &&
    /// calls.is_empty()` (with no structured answer either) — remove either
    /// conjunct and one of the two fails.
    #[test]
    fn an_answer_with_neither_text_nor_tool_calls_is_unusable() {
        let model = ModelId::new("bedrock", "test");
        let message = Message::builder()
            .role(ConversationRole::Assistant)
            .content(ContentBlock::ReasoningContent(
                ReasoningContentBlock::ReasoningText(
                    ReasoningTextBlock::builder()
                        .text("reasoned and then said nothing")
                        .build()
                        .expect("reasoning block"),
                ),
            ))
            .build()
            .expect("assistant message");
        let output = aws_sdk_bedrockruntime::operation::converse::ConverseOutput::builder()
            .output(aws_sdk_bedrockruntime::types::ConverseOutput::Message(
                message,
            ))
            .stop_reason(aws_sdk_bedrockruntime::types::StopReason::EndTurn)
            .build()
            .expect("converse output");
        let error = Bedrock::interpret(&model, None, SchemaMode::Native, &output)
            .expect_err("reasoning with neither text nor a tool call is not usable");
        assert!(
            matches!(error, ModelError::Unusable { .. }),
            "wrong classification: {error}"
        );
    }

    /// A person's turn follows the history as one user message after the
    /// assistant's answer, as text beside any tool results.
    #[test]
    fn a_persons_turn_follows_the_history() {
        let model = ModelId::new("bedrock", "test");
        let continuation = ProviderContinuation::new(
            "bedrock",
            json!([{ "role": "assistant", "content": [{ "type": "text", "text": "42." }] }]),
        );
        let messages = Bedrock::messages(
            &model,
            &json!({"input": "balance?"}),
            &[],
            &["and in euros?".to_owned()],
            Some(&continuation),
        )
        .expect("a person's turn after an answer");
        assert_eq!(messages.len(), 3, "question, answer, the person's turn");
        assert_eq!(messages[2].role(), &ConversationRole::User);
        assert_eq!(
            messages[2].content(),
            &[ContentBlock::Text("and in euros?".to_owned())]
        );
    }

    /// A person's first turn joins the prompt's user message: Converse
    /// refuses two user messages in a row.
    #[test]
    fn a_first_persons_turn_joins_the_prompts_message() {
        let model = ModelId::new("bedrock", "test");
        let messages = Bedrock::messages(
            &model,
            &json!({"input": "balance?"}),
            &[],
            &["in euros".to_owned()],
            None,
        )
        .expect("a first turn");
        assert_eq!(messages.len(), 1, "two user messages in a row");
        assert_eq!(messages[0].content().len(), 2);
    }

    /// A failed tool's message is a string, so this covers every failure too.
    #[test]
    fn a_tool_result_is_json_only_when_it_is_an_object() {
        let model = ModelId::new("bedrock", "test");
        let call = super::super::ToolCall {
            id: "tool_1".to_owned(),
            name: "list_files".to_owned(),
            arguments: json!({}),
        };
        let content = |exchange: ToolExchange| {
            let messages =
                Bedrock::messages(&model, &json!({"input": "files?"}), &[exchange], &[], None)
                    .expect("a tool turn");
            messages
                .iter()
                .flat_map(Message::content)
                .find_map(|block| block.as_tool_result().ok())
                .expect("a tool result")
                .content()[0]
                .clone()
        };
        let text = |text: &str| ToolResultContentBlock::Text(text.to_owned());

        assert!(content(ToolExchange::ok(call.clone(), json!({"files": []}))).is_json());
        for (output, sent) in [
            (json!(["a.rs", "b.rs"]), r#"["a.rs","b.rs"]"#),
            (json!("a.rs"), "a.rs"),
            (json!(3), "3"),
            (json!(true), "true"),
            (Value::Null, "null"),
        ] {
            assert_eq!(content(ToolExchange::ok(call.clone(), output)), text(sent));
        }
        assert_eq!(
            content(ToolExchange::failed(call, "permission denied")),
            text("permission denied"),
        );
    }

    #[test]
    fn cumulative_bedrock_transcript_decodes_every_prior_turn() {
        let model = ModelId::new("bedrock", "test");
        let continuation = ProviderContinuation::new(
            "bedrock",
            json!([
                {"role": "assistant", "content": [
                    {"type": "reasoning", "text": "private", "signature": "sig"},
                    {"type": "tool_use", "id": "tool_1", "name": "lookup", "input": {}}
                ]},
                {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "tool_1", "output": {"v": 1}, "failed": false}
                ]}
            ]),
        );
        let messages = Bedrock::messages(
            &model,
            &json!({"input": "question"}),
            &[ToolExchange::ok(
                super::super::ToolCall {
                    id: "tool_2".to_owned(),
                    name: "lookup".to_owned(),
                    arguments: json!({}),
                },
                json!({"v": 2}),
            )],
            &[],
            Some(&continuation),
        )
        .expect("decode cumulative transcript");
        assert_eq!(messages.len(), 4, "question plus three continuation turns");
        assert!(messages[1].content()[0].is_reasoning_content());
        assert!(messages[2].content()[0].is_tool_result());
        assert!(messages[3].content()[0].is_tool_result());
    }
}

#[cfg(test)]
mod prompt_tests {
    use aws_sdk_bedrockruntime::types::{
        CacheDetail, CacheTtl as WireCacheTtl, ContentBlock, SystemContentBlock, TokenUsage, Tool,
    };
    use serde_json::json;

    use super::{Bedrock, ModelId, SchemaMode, ToolDeclaration};
    use crate::core::Tainted;
    use crate::model::{CacheTtl, Prompt};

    fn model() -> ModelId {
        ModelId::new("bedrock", "anthropic.claude-x")
    }

    /// **A breakpoint is a `cachePoint` after the block it closes**, in the
    /// instruction and in the question — Converse caches nothing unasked.
    #[test]
    fn a_prompt_part_carries_its_breakpoint() {
        let prompt = Prompt::new()
            .system_cached(Tainted::trusted("policy".to_owned()), CacheTtl::OneHour)
            .input_cached(
                Tainted::trusted("context".to_owned()),
                CacheTtl::FiveMinutes,
            )
            .input(Tainted::trusted("task".to_owned()))
            .build();
        let system = super::system_blocks(prompt.peek(), &model()).expect("system");
        assert!(matches!(&system[0], SystemContentBlock::Text(t) if t == "policy"));
        assert!(
            matches!(&system[1], SystemContentBlock::CachePoint(p) if p.ttl() == Some(&WireCacheTtl::OneHour)),
            "{system:?}"
        );
        let messages =
            Bedrock::messages(&model(), prompt.peek(), &[], &[], None).expect("messages");
        let content = messages[0].content();
        assert!(matches!(&content[0], ContentBlock::Text(t) if t == "context"));
        assert!(
            matches!(&content[1], ContentBlock::CachePoint(p) if p.ttl() == Some(&WireCacheTtl::FiveMinutes)),
            "{content:?}"
        );
        assert!(matches!(&content[2], ContentBlock::Text(t) if t == "task"));
    }

    /// **A tool breakpoint is honoured, and a tool outside the strict subset
    /// is sent non-strict** rather than turning the request into a 400.
    #[test]
    fn a_tool_is_strict_only_where_the_subset_holds_it() {
        let tools = [
            ToolDeclaration::new(
                "closed",
                "d",
                json!({ "type": "object", "properties": {}, "required": [], "additionalProperties": false }),
            )
            .cache_breakpoint(CacheTtl::OneHour),
            ToolDeclaration::new("open", "d", json!({ "type": "object" })),
        ];
        let config = Bedrock::tool_config(&model(), None, &tools, SchemaMode::Native)
            .expect("tool config")
            .expect("tools were declared");
        let strict: Vec<Option<bool>> = config
            .tools()
            .iter()
            .filter_map(|tool| match tool {
                Tool::ToolSpec(spec) => Some(spec.strict()),
                _ => None,
            })
            .collect();
        assert_eq!(strict, [Some(true), Some(false)]);
        assert!(
            matches!(&config.tools()[1], Tool::CachePoint(p) if p.ttl() == Some(&WireCacheTtl::OneHour)),
            "{config:?}"
        );
    }

    /// **The one-hour share of the cache writes is read from the per-tier
    /// detail**, so a priced run bills it at its own rate.
    #[test]
    fn the_one_hour_writes_are_counted() {
        let usage = TokenUsage::builder()
            .input_tokens(10)
            .output_tokens(5)
            .total_tokens(15)
            .cache_write_input_tokens(300)
            .cache_details(
                CacheDetail::builder()
                    .ttl(WireCacheTtl::OneHour)
                    .input_tokens(200)
                    .build()
                    .expect("detail"),
            )
            .build()
            .expect("usage");
        let usage = super::usage_from(&usage);
        assert_eq!(usage.cache_write_tokens, 300);
        assert_eq!(usage.cache_write_1h_tokens, 200);
    }
}