provider_core/

http_transport.rs

// SPDX-License-Identifier: AGPL-3.0-or-later

//! Transport-independent, injected HTTP seam for provider request specifications.

use crate::anthropic_chat::{
    AnthropicBodyWriteState, AnthropicFailure, AnthropicRequestSpec, AnthropicSseParser,
    AnthropicStreamError, AnthropicStreamEvent, AnthropicTransportFailureSignal,
    normalize_http_response as normalize_anthropic_response,
    normalize_transport_failure as normalize_anthropic_transport_failure,
};
use crate::gemini_chat::{
    GeminiBodyWriteState, GeminiFailure, GeminiRequestSpec, GeminiSseParser, GeminiStreamError,
    GeminiStreamEvent, GeminiTransportFailureSignal,
    normalize_http_response as normalize_gemini_response,
    normalize_transport_failure as normalize_gemini_transport_failure,
};
use crate::local_credential_store::LocalCredentialStore;
use crate::ollama_chat::{
    BodyWriteState, OllamaAuthentication, OllamaFailure, OllamaNdjsonParser, OllamaReplayGuard,
    OllamaRequestSpec, OllamaStreamError, OllamaStreamEvent, ReplayDecision,
    TransportFailureSignal, normalize_http_response as normalize_ollama_response,
    normalize_transport_failure as normalize_ollama_transport_failure,
};
use crate::openai_compat_chat::{
    BodyWriteState as OpenAiCompatBodyWriteState, OpenAiCompatFailure, OpenAiCompatRequestSpec,
    OpenAiCompatSseParser, OpenAiCompatStreamError, OpenAiCompatStreamEvent,
    TransportFailureSignal as OpenAiCompatTransportFailureSignal,
    normalize_http_response as normalize_openai_compat_response,
    normalize_transport_failure as normalize_openai_compat_transport_failure,
};
use crate::openrouter_chat::{
    OpenRouterFailure, OpenRouterRequestSpec, OpenRouterSseParser, OpenRouterStreamError,
    OpenRouterStreamEvent, normalize_http_response as normalize_openrouter_response,
    normalize_transport_failure as normalize_openrouter_transport_failure,
};
use futures_util::StreamExt;
use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE};
use router_protocol::AttemptId;
use std::fmt;
use std::future::Future;
#[cfg(unix)]
use std::path::Path;
use std::path::{Component, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
#[cfg(unix)]
use std::{
    fs::{File, OpenOptions},
    io::{Read, Write},
    os::unix::{
        fs::{DirBuilderExt, MetadataExt, OpenOptionsExt},
        process::CommandExt,
    },
    process::{Child, Command, Stdio},
    sync::atomic::AtomicU64,
    thread,
    time::{Instant, SystemTime, UNIX_EPOCH},
};
use tokio::sync::Notify;
use zeroize::Zeroizing;

pub use crate::ollama_chat::NetworkPolicy;

const MAX_RESPONSE_CHUNKS: usize = 4_096;
const MAX_REQUEST_BODY_BYTES: usize = 4 * 1024 * 1024;
const MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
const MAX_ERROR_BODY_BYTES: usize = 32 * 1024;
const MAX_REQUEST_ID_BYTES: usize = 128;
const MAX_RESOLVED_SECRET_BYTES: usize = 16 * 1024;
const MAX_KEYCHAIN_COMPONENT_BYTES: usize = 255;
#[cfg(unix)]
const MAX_SECRET_HELPER_BYTES: usize = 4 * 1024 * 1024;
const SECRET_HELPER_TIMEOUT: Duration = Duration::from_secs(5);
const SECRET_HELPER_POLL_INTERVAL: Duration = Duration::from_millis(5);
const FIXED_USER_AGENT: &str = "steadymux-provider-core/0.1";
#[cfg(unix)]
static SECRET_HELPER_STAGE_COUNTER: AtomicU64 = AtomicU64::new(0);

/// A resolved credential that can only be exposed deliberately at the dispatch edge.
pub struct ResolvedSecret(Zeroizing<Vec<u8>>);

impl ResolvedSecret {
    pub fn new(value: impl Into<Vec<u8>>) -> Result<Self, SecretResolutionError> {
        Self::from_zeroizing(Zeroizing::new(value.into()))
    }

    fn from_zeroizing(value: Zeroizing<Vec<u8>>) -> Result<Self, SecretResolutionError> {
        if value.is_empty()
            || value.len() > MAX_RESOLVED_SECRET_BYTES
            || value.contains(&b'\r')
            || value.contains(&b'\n')
        {
            return Err(SecretResolutionError);
        }
        Ok(Self(value))
    }

    pub fn expose_for_dispatch(&self) -> &[u8] {
        self.0.as_slice()
    }
}

impl fmt::Debug for ResolvedSecret {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("ResolvedSecret([REDACTED])")
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SecretResolutionError;

impl fmt::Display for SecretResolutionError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("provider credential resolution failed")
    }
}

impl std::error::Error for SecretResolutionError {}

pub trait SecretResolver: Send + Sync {
    fn resolve(&self, secret_ref: &str) -> Result<ResolvedSecret, SecretResolutionError>;
}

enum ParsedSecretReference<'a> {
    MacOsKeychain { service: &'a str, account: &'a str },
    LocalFile,
    Command,
    Environment { name: &'a str },
}

impl fmt::Debug for ParsedSecretReference<'_> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::MacOsKeychain { .. } => formatter.write_str("MacOsKeychain([REDACTED])"),
            Self::LocalFile => formatter.write_str("LocalFile([REDACTED])"),
            Self::Command => formatter.write_str("Command([REDACTED])"),
            Self::Environment { .. } => formatter.write_str("Environment([REDACTED])"),
        }
    }
}

fn parse_secret_reference(
    secret_ref: &str,
) -> Result<ParsedSecretReference<'_>, SecretResolutionError> {
    if let Some(value) = secret_ref.strip_prefix("keychain:") {
        let (service, account) = value.split_once('/').ok_or(SecretResolutionError)?;
        if account.contains('/')
            || !valid_keychain_component(service)
            || !valid_keychain_component(account)
        {
            return Err(SecretResolutionError);
        }
        return Ok(ParsedSecretReference::MacOsKeychain { service, account });
    }

    if let Some(value) = secret_ref.strip_prefix("env:") {
        if !valid_environment_name(value) {
            return Err(SecretResolutionError);
        }
        return Ok(ParsedSecretReference::Environment { name: value });
    }

    if let Some(value) = secret_ref.strip_prefix("local-file-v1:") {
        if value.len() != 32
            || !value
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        {
            return Err(SecretResolutionError);
        }
        return Ok(ParsedSecretReference::LocalFile);
    }

    command_secret_path(secret_ref)?;
    Ok(ParsedSecretReference::Command)
}

fn valid_environment_name(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 256
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
}

fn valid_keychain_component(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_KEYCHAIN_COMPONENT_BYTES
        && value.bytes().all(|byte| {
            byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':' | b'@' | b'+')
        })
}

/// Resolves an approved platform-native credential reference or the existing
/// owner-selected command helper reference. Keychain references fail closed on
/// platforms other than macOS.
#[derive(Clone, Copy, Default)]
pub struct ProductionSecretResolver {
    command: CommandSecretResolver,
}

impl fmt::Debug for ProductionSecretResolver {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProductionSecretResolver")
            .field("credential", &"[REDACTED]")
            .finish()
    }
}

impl ProductionSecretResolver {
    pub fn supports_reference(secret_ref: &str) -> bool {
        match parse_secret_reference(secret_ref) {
            Ok(ParsedSecretReference::MacOsKeychain { .. }) => cfg!(target_os = "macos"),
            Ok(ParsedSecretReference::LocalFile) => cfg!(target_os = "macos"),
            Ok(ParsedSecretReference::Environment { .. }) => true,
            Ok(ParsedSecretReference::Command) => {
                CommandSecretResolver::supports_reference(secret_ref)
            }
            Err(_) => false,
        }
    }

    /// Validates the reference form without reading a credential. Command
    /// references retain their existing filesystem ownership checks.
    pub fn validate_reference(secret_ref: &str) -> Result<(), SecretResolutionError> {
        match parse_secret_reference(secret_ref)? {
            ParsedSecretReference::MacOsKeychain { .. } => {
                #[cfg(target_os = "macos")]
                {
                    Ok(())
                }
                #[cfg(not(target_os = "macos"))]
                {
                    Err(SecretResolutionError)
                }
            }
            ParsedSecretReference::LocalFile => {
                #[cfg(target_os = "macos")]
                {
                    Ok(())
                }
                #[cfg(not(target_os = "macos"))]
                {
                    Err(SecretResolutionError)
                }
            }
            ParsedSecretReference::Environment { .. } => Ok(()),
            ParsedSecretReference::Command => CommandSecretResolver::validate_reference(secret_ref),
        }
    }
}

impl SecretResolver for ProductionSecretResolver {
    fn resolve(&self, secret_ref: &str) -> Result<ResolvedSecret, SecretResolutionError> {
        match parse_secret_reference(secret_ref)? {
            ParsedSecretReference::MacOsKeychain { service, account } => {
                resolve_macos_keychain_secret(service, account)
            }
            ParsedSecretReference::LocalFile => {
                let value = LocalCredentialStore::production()
                    .and_then(|store| store.resolve(secret_ref))
                    .map_err(|_| SecretResolutionError)?;
                ResolvedSecret::from_zeroizing(value)
            }
            ParsedSecretReference::Environment { name } => {
                let value = std::env::var(name).map_err(|_| SecretResolutionError)?;
                ResolvedSecret::from_zeroizing(Zeroizing::new(value.into_bytes()))
            }
            ParsedSecretReference::Command => self.command.resolve(secret_ref),
        }
    }
}

#[cfg(target_os = "macos")]
fn resolve_macos_keychain_secret(
    service: &str,
    account: &str,
) -> Result<ResolvedSecret, SecretResolutionError> {
    use security_framework::item::SearchResult;

    let mut options = local_data_protection_keychain_search(service, account);
    options.load_data(true);
    let mut results = options.search().map_err(|_| SecretResolutionError)?;
    if results.len() != 1 {
        return Err(SecretResolutionError);
    }
    let SearchResult::Data(value) = results.pop().ok_or(SecretResolutionError)? else {
        return Err(SecretResolutionError);
    };
    ResolvedSecret::from_zeroizing(Zeroizing::new(value))
}

#[cfg(not(target_os = "macos"))]
fn resolve_macos_keychain_secret(
    _service: &str,
    _account: &str,
) -> Result<ResolvedSecret, SecretResolutionError> {
    Err(SecretResolutionError)
}

/// Creates or replaces a validated generic-password entry in the local,
/// non-synchronizing macOS Data Protection Keychain. The legacy login
/// Keychain is not searched or mutated, so an older file-Keychain item with
/// the same service/account may coexist until an explicit migration removes
/// it. The credential is accepted only in memory and is never included in
/// diagnostics.
#[cfg(target_os = "macos")]
pub fn upsert_macos_keychain_secret(
    secret_ref: &str,
    value: &[u8],
) -> Result<(), SecretResolutionError> {
    let ParsedSecretReference::MacOsKeychain { service, account } =
        parse_secret_reference(secret_ref)?
    else {
        return Err(SecretResolutionError);
    };
    let value = ResolvedSecret::new(value.to_vec())?;
    security_framework::passwords::set_generic_password_options(
        value.expose_for_dispatch(),
        local_data_protection_password_options(service, account),
    )
    .map_err(|_| SecretResolutionError)
}

/// Verifies that one exact local Data Protection Keychain item can be read
/// without authentication UI. The resolved value remains zeroizing, is
/// dropped immediately, and is never returned to the caller.
#[cfg(target_os = "macos")]
pub fn verify_macos_keychain_secret(secret_ref: &str) -> Result<(), SecretResolutionError> {
    let ParsedSecretReference::MacOsKeychain { service, account } =
        parse_secret_reference(secret_ref)?
    else {
        return Err(SecretResolutionError);
    };
    let secret = resolve_macos_keychain_secret(service, account)?;
    drop(secret);
    Ok(())
}

/// Deletes only the matching local Data Protection Keychain item. A legacy
/// login-Keychain item with the same service/account remains untouched.
/// Missing entries and platform errors remain content-free.
#[cfg(target_os = "macos")]
pub fn delete_macos_keychain_secret(secret_ref: &str) -> Result<(), SecretResolutionError> {
    let ParsedSecretReference::MacOsKeychain { service, account } =
        parse_secret_reference(secret_ref)?
    else {
        return Err(SecretResolutionError);
    };
    local_data_protection_keychain_search(service, account)
        .delete()
        .map_err(|_| SecretResolutionError)
}

#[cfg(target_os = "macos")]
fn local_data_protection_password_options(
    service: &str,
    account: &str,
) -> security_framework::passwords::PasswordOptions {
    let mut options =
        security_framework::passwords::PasswordOptions::new_generic_password(service, account);
    options.use_protected_keychain();
    options.set_access_synchronized(Some(false));
    options
}

#[cfg(target_os = "macos")]
fn local_data_protection_keychain_search(
    service: &str,
    account: &str,
) -> security_framework::item::ItemSearchOptions {
    use security_framework::item::{ItemClass, ItemSearchOptions};

    let mut options = ItemSearchOptions::new();
    options
        .ignore_legacy_keychains()
        .class(ItemClass::generic_password())
        .service(service)
        .account(account)
        .cloud_sync(Some(false))
        // Normal runtime reads are non-interactive. An ACL that would require
        // UI fails closed instead of presenting a recurring system prompt or
        // indefinitely holding an in-flight daemon request.
        .skip_authenticated_items(true);
    options
}

/// Resolves only an owner-selected executable referenced as
/// `command:/absolute/path`. The helper receives no arguments, stdin, stderr,
/// inherited environment, or shell interpretation. All failure diagnostics are
/// deliberately content-free.
#[derive(Clone, Copy)]
pub struct CommandSecretResolver {
    timeout: Duration,
}

impl Default for CommandSecretResolver {
    fn default() -> Self {
        Self {
            timeout: SECRET_HELPER_TIMEOUT,
        }
    }
}

impl fmt::Debug for CommandSecretResolver {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("CommandSecretResolver")
            .field("credential", &"[REDACTED]")
            .field("timeout", &self.timeout)
            .finish()
    }
}

impl CommandSecretResolver {
    pub fn supports_reference(secret_ref: &str) -> bool {
        command_secret_path(secret_ref).is_ok() && cfg!(unix)
    }

    /// Performs the same redacted path/ownership/permission validation used
    /// immediately before execution, without starting the helper.
    pub fn validate_reference(secret_ref: &str) -> Result<(), SecretResolutionError> {
        #[cfg(unix)]
        {
            open_secret_helper_source(secret_ref).map(|_| ())
        }
        #[cfg(not(unix))]
        {
            let _ = secret_ref;
            Err(SecretResolutionError)
        }
    }

    #[cfg(test)]
    const fn with_timeout(timeout: Duration) -> Self {
        Self { timeout }
    }
}

impl SecretResolver for CommandSecretResolver {
    fn resolve(&self, secret_ref: &str) -> Result<ResolvedSecret, SecretResolutionError> {
        #[cfg(unix)]
        {
            resolve_command_secret(secret_ref, self.timeout)
        }
        #[cfg(not(unix))]
        {
            let _ = (secret_ref, self.timeout);
            Err(SecretResolutionError)
        }
    }
}

fn command_secret_path(secret_ref: &str) -> Result<PathBuf, SecretResolutionError> {
    let value = secret_ref
        .strip_prefix("command:")
        .ok_or(SecretResolutionError)?;
    let path = PathBuf::from(value);
    if value.is_empty()
        || value.len() > 4096
        || value.chars().any(char::is_control)
        || !path.is_absolute()
        || path
            .components()
            .any(|component| !matches!(component, Component::RootDir | Component::Normal(_)))
    {
        return Err(SecretResolutionError);
    }
    Ok(path)
}

#[cfg(unix)]
fn resolve_command_secret(
    secret_ref: &str,
    timeout: Duration,
) -> Result<ResolvedSecret, SecretResolutionError> {
    resolve_command_secret_with_hook(secret_ref, timeout, |_| Ok(()))
}

#[cfg(unix)]
fn resolve_command_secret_with_hook<F>(
    secret_ref: &str,
    timeout: Duration,
    after_source_open: F,
) -> Result<ResolvedSecret, SecretResolutionError>
where
    F: FnOnce(&Path) -> Result<(), SecretResolutionError>,
{
    let (source_path, mut source) = open_secret_helper_source(secret_ref)?;
    after_source_open(&source_path)?;
    let source_bytes = read_bounded_source(&mut source)?;
    let staged = StagedSecretHelper::create(&source_path, &source_bytes)?;
    staged.verify(&source_bytes)?;
    let mut command = Command::new(&staged.path);
    command
        .env_clear()
        .stdin(Stdio::null())
        .stderr(Stdio::null())
        .stdout(Stdio::piped())
        .process_group(0);
    let mut child = SecretHelperChild(command.spawn().map_err(|_| SecretResolutionError)?);
    staged.verify(&source_bytes)?;
    let mut stdout = child.0.stdout.take().ok_or(SecretResolutionError)?;
    rustix::fs::fcntl_setfl(&stdout, rustix::fs::OFlags::NONBLOCK)
        .map_err(|_| SecretResolutionError)?;
    let started = Instant::now();
    let mut value = Zeroizing::new(Vec::with_capacity(256));
    let status = loop {
        read_bounded_nonblocking(&mut stdout, &mut value)?;
        if let Some(status) = child.0.try_wait().map_err(|_| SecretResolutionError)? {
            read_bounded_nonblocking(&mut stdout, &mut value)?;
            break status;
        }
        if started.elapsed() >= timeout {
            return Err(SecretResolutionError);
        }
        thread::sleep(SECRET_HELPER_POLL_INTERVAL);
    };
    if !status.success() {
        return Err(SecretResolutionError);
    }
    if value.ends_with(b"\r\n") {
        let trimmed_len = value.len() - 2;
        value.truncate(trimmed_len);
    } else if value.ends_with(b"\n") {
        value.pop();
    }
    drop(stdout);
    drop(child);
    staged.cleanup(&source_bytes)?;
    ResolvedSecret::from_zeroizing(value)
}

#[cfg(unix)]
struct SecretHelperChild(Child);

#[cfg(unix)]
impl Drop for SecretHelperChild {
    fn drop(&mut self) {
        let process_group = rustix::process::Pid::from_child(&self.0);
        let _ = rustix::process::kill_process_group(process_group, rustix::process::Signal::KILL);
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

#[cfg(unix)]
fn validate_safe_ancestry(
    path: &Path,
    final_directory_owner_only: bool,
) -> Result<(), SecretResolutionError> {
    let expected_uid = rustix::process::geteuid().as_raw();
    let mut cursor = PathBuf::new();
    for component in path.components() {
        cursor.push(component.as_os_str());
        let metadata = std::fs::symlink_metadata(&cursor).map_err(|_| SecretResolutionError)?;
        let owner_is_trusted = metadata.uid() == 0 || metadata.uid() == expected_uid;
        if metadata.file_type().is_symlink()
            || !metadata.is_dir()
            || !owner_is_trusted
            || metadata.mode() & 0o022 != 0
            || (cursor == path
                && final_directory_owner_only
                && (metadata.uid() != expected_uid || metadata.mode() & 0o777 != 0o700))
        {
            return Err(SecretResolutionError);
        }
    }
    Ok(())
}

#[cfg(unix)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct SecretHelperIdentity {
    device: u64,
    inode: u64,
    uid: u32,
    mode: u32,
    length: u64,
    modified_seconds: i64,
    modified_nanoseconds: i64,
    changed_seconds: i64,
    changed_nanoseconds: i64,
}

#[cfg(unix)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct StableNodeIdentity {
    device: u64,
    inode: u64,
    uid: u32,
    mode: u32,
}

#[cfg(unix)]
fn stable_node_identity(metadata: &std::fs::Metadata) -> StableNodeIdentity {
    StableNodeIdentity {
        device: metadata.dev(),
        inode: metadata.ino(),
        uid: metadata.uid(),
        mode: metadata.mode(),
    }
}

#[cfg(unix)]
fn metadata_identity(metadata: &std::fs::Metadata) -> SecretHelperIdentity {
    SecretHelperIdentity {
        device: metadata.dev(),
        inode: metadata.ino(),
        uid: metadata.uid(),
        mode: metadata.mode(),
        length: metadata.len(),
        modified_seconds: metadata.mtime(),
        modified_nanoseconds: metadata.mtime_nsec(),
        changed_seconds: metadata.ctime(),
        changed_nanoseconds: metadata.ctime_nsec(),
    }
}

#[cfg(unix)]
fn open_secret_helper_source(secret_ref: &str) -> Result<(PathBuf, File), SecretResolutionError> {
    let path = command_secret_path(secret_ref)?;
    let parent = path.parent().ok_or(SecretResolutionError)?;
    validate_safe_ancestry(parent, false)?;
    let mut options = OpenOptions::new();
    options
        .read(true)
        .custom_flags((rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::CLOEXEC).bits() as i32);
    let file = options.open(&path).map_err(|_| SecretResolutionError)?;
    let metadata = file.metadata().map_err(|_| SecretResolutionError)?;
    validate_source_metadata(&metadata)?;
    Ok((path, file))
}

#[cfg(unix)]
fn validate_source_metadata(metadata: &std::fs::Metadata) -> Result<(), SecretResolutionError> {
    if !metadata.is_file()
        || metadata.uid() != rustix::process::geteuid().as_raw()
        || metadata.nlink() != 1
        || metadata.mode() & 0o7777 != 0o700
        || metadata.len() == 0
        || metadata.len() > MAX_SECRET_HELPER_BYTES as u64
    {
        return Err(SecretResolutionError);
    }
    Ok(())
}

#[cfg(unix)]
fn read_bounded_source(source: &mut File) -> Result<Vec<u8>, SecretResolutionError> {
    let metadata = source.metadata().map_err(|_| SecretResolutionError)?;
    validate_source_metadata(&metadata)?;
    let mut bytes = Vec::with_capacity(metadata.len() as usize);
    Read::by_ref(source)
        .take(MAX_SECRET_HELPER_BYTES as u64 + 1)
        .read_to_end(&mut bytes)
        .map_err(|_| SecretResolutionError)?;
    let final_metadata = source.metadata().map_err(|_| SecretResolutionError)?;
    if bytes.len() > MAX_SECRET_HELPER_BYTES
        || bytes.len() as u64 != metadata.len()
        || metadata.dev() != final_metadata.dev()
        || metadata.ino() != final_metadata.ino()
        || metadata.uid() != final_metadata.uid()
        || metadata.mode() != final_metadata.mode()
        || metadata.len() != final_metadata.len()
        || metadata.mtime() != final_metadata.mtime()
        || metadata.mtime_nsec() != final_metadata.mtime_nsec()
    {
        return Err(SecretResolutionError);
    }
    Ok(bytes)
}

#[cfg(unix)]
#[derive(Debug)]
struct StagedSecretHelper {
    parent: PathBuf,
    directory: PathBuf,
    path: PathBuf,
    directory_identity: StableNodeIdentity,
    file_identity: Option<SecretHelperIdentity>,
}

#[cfg(unix)]
impl StagedSecretHelper {
    fn create(source_path: &Path, bytes: &[u8]) -> Result<Self, SecretResolutionError> {
        let parent = source_path
            .parent()
            .ok_or(SecretResolutionError)?
            .to_owned();
        validate_safe_ancestry(&parent, false)?;
        let counter = SECRET_HELPER_STAGE_COUNTER.fetch_add(1, Ordering::Relaxed);
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| SecretResolutionError)?
            .as_nanos();
        let directory = parent.join(format!(
            ".steadymux-secret-helper-{}-{counter}-{nonce}",
            std::process::id()
        ));
        std::fs::DirBuilder::new()
            .mode(0o700)
            .create(&directory)
            .map_err(|_| SecretResolutionError)?;
        let directory_metadata =
            std::fs::symlink_metadata(&directory).map_err(|_| SecretResolutionError)?;
        let directory_identity = stable_node_identity(&directory_metadata);
        let path = directory.join("helper");
        let mut cleanup = Self {
            parent,
            directory,
            path,
            directory_identity,
            file_identity: None,
        };
        let result = (|| {
            validate_safe_ancestry(&cleanup.directory, true)?;
            sync_directory(&cleanup.parent)?;
            let mut options = OpenOptions::new();
            options
                .write(true)
                .create_new(true)
                .mode(0o700)
                .custom_flags(
                    (rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::CLOEXEC).bits() as i32,
                );
            let mut file = options
                .open(&cleanup.path)
                .map_err(|_| SecretResolutionError)?;
            file.write_all(bytes).map_err(|_| SecretResolutionError)?;
            file.sync_all().map_err(|_| SecretResolutionError)?;
            let metadata = file.metadata().map_err(|_| SecretResolutionError)?;
            cleanup.file_identity = Some(metadata_identity(&metadata));
            drop(file);
            cleanup.verify(bytes)?;
            sync_directory(&cleanup.directory)
        })();
        if result.is_err() {
            drop(cleanup);
            return Err(SecretResolutionError);
        }
        Ok(cleanup)
    }

    fn verify(&self, expected: &[u8]) -> Result<(), SecretResolutionError> {
        validate_safe_ancestry(&self.directory, true)?;
        let directory_metadata =
            std::fs::symlink_metadata(&self.directory).map_err(|_| SecretResolutionError)?;
        if stable_node_identity(&directory_metadata) != self.directory_identity {
            return Err(SecretResolutionError);
        }
        let mut options = OpenOptions::new();
        options.read(true).custom_flags(
            (rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::CLOEXEC).bits() as i32,
        );
        let mut file = options
            .open(&self.path)
            .map_err(|_| SecretResolutionError)?;
        let metadata = file.metadata().map_err(|_| SecretResolutionError)?;
        if Some(metadata_identity(&metadata)) != self.file_identity
            || !metadata.is_file()
            || metadata.uid() != rustix::process::geteuid().as_raw()
            || metadata.mode() & 0o777 != 0o700
            || metadata.nlink() != 1
            || metadata.len() != expected.len() as u64
        {
            return Err(SecretResolutionError);
        }
        let mut observed = Vec::with_capacity(expected.len());
        Read::by_ref(&mut file)
            .take(expected.len() as u64 + 1)
            .read_to_end(&mut observed)
            .map_err(|_| SecretResolutionError)?;
        let final_metadata = file.metadata().map_err(|_| SecretResolutionError)?;
        if observed != expected || Some(metadata_identity(&final_metadata)) != self.file_identity {
            return Err(SecretResolutionError);
        }
        Ok(())
    }

    fn cleanup(mut self, expected: &[u8]) -> Result<(), SecretResolutionError> {
        let identity = self.file_identity.ok_or(SecretResolutionError)?;
        let mut options = OpenOptions::new();
        options.read(true).custom_flags(
            (rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::CLOEXEC).bits() as i32,
        );
        let mut file = options
            .open(&self.path)
            .map_err(|_| SecretResolutionError)?;
        let metadata = file.metadata().map_err(|_| SecretResolutionError)?;
        if !metadata.is_file()
            || metadata.dev() != identity.device
            || metadata.ino() != identity.inode
            || metadata.uid() != identity.uid
            || metadata.mode() != identity.mode
            || metadata.nlink() != 1
            || metadata.len() != identity.length
        {
            return Err(SecretResolutionError);
        }
        let mut observed = Vec::with_capacity(expected.len());
        Read::by_ref(&mut file)
            .take(expected.len() as u64 + 1)
            .read_to_end(&mut observed)
            .map_err(|_| SecretResolutionError)?;
        if observed != expected {
            return Err(SecretResolutionError);
        }
        drop(file);
        let directory_metadata =
            std::fs::symlink_metadata(&self.directory).map_err(|_| SecretResolutionError)?;
        if stable_node_identity(&directory_metadata) != self.directory_identity {
            return Err(SecretResolutionError);
        }
        std::fs::remove_file(&self.path).map_err(|_| SecretResolutionError)?;
        self.file_identity = None;
        sync_directory(&self.directory)?;
        std::fs::remove_dir(&self.directory).map_err(|_| SecretResolutionError)?;
        sync_directory(&self.parent)
    }
}

#[cfg(unix)]
impl Drop for StagedSecretHelper {
    fn drop(&mut self) {
        if std::fs::symlink_metadata(&self.path).is_ok_and(|metadata| {
            !metadata.file_type().is_symlink()
                && metadata.is_file()
                && Some(metadata_identity(&metadata)) == self.file_identity
        }) {
            let _ = std::fs::remove_file(&self.path);
            let _ = sync_directory(&self.directory);
        }
        if std::fs::symlink_metadata(&self.directory).is_ok_and(|metadata| {
            !metadata.file_type().is_symlink()
                && metadata.is_dir()
                && stable_node_identity(&metadata) == self.directory_identity
        }) {
            let _ = std::fs::remove_dir(&self.directory);
            let _ = sync_directory(&self.parent);
        }
    }
}

#[cfg(unix)]
fn sync_directory(path: &Path) -> Result<(), SecretResolutionError> {
    File::open(path)
        .and_then(|directory| directory.sync_all())
        .map_err(|_| SecretResolutionError)
}

#[cfg(unix)]
fn read_bounded_nonblocking(
    stdout: &mut std::process::ChildStdout,
    value: &mut Zeroizing<Vec<u8>>,
) -> Result<(), SecretResolutionError> {
    let mut buffer = [0_u8; 4096];
    loop {
        match stdout.read(&mut buffer) {
            Ok(0) => return Ok(()),
            Ok(read) => {
                if value.len().saturating_add(read) > MAX_RESOLVED_SECRET_BYTES {
                    return Err(SecretResolutionError);
                }
                value.extend_from_slice(&buffer[..read]);
            }
            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => return Ok(()),
            Err(_) => return Err(SecretResolutionError),
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProviderWireProtocol {
    OllamaNdjson,
    OpenRouterSse,
    OpenAiCompatSse,
    AnthropicSse,
    GeminiSse,
}

/// Fully prepared only immediately before the injected executor is called.
/// Debug deliberately excludes the body, secret reference, and resolved secret.
pub struct PreparedRequest {
    pub method: &'static str,
    pub origin: String,
    pub path: String,
    pub content_type: &'static str,
    pub accept: &'static str,
    pub network: NetworkPolicy,
    pub protocol: ProviderWireProtocol,
    body: Vec<u8>,
    authorization: Option<ResolvedSecret>,
    /// When set, the executor writes the resolved secret in this header name
    /// instead of the `Authorization: Bearer` header. The Anthropic
    /// (`x-api-key`) and Gemini (`x-goog-api-key`) protocols require a custom
    /// API-key header.
    pub auth_header_name: Option<&'static str>,
    /// A fixed protocol header the executor writes alongside the credential
    /// header. The Anthropic Messages protocol requires the pinned
    /// `anthropic-version: 2023-06-01` header on every request.
    pub version_header: Option<(&'static str, &'static str)>,
    /// True for the bounded non-streaming completion dispatch (model
    /// intelligence categorization). Bounded requests carry `application/json`
    /// accept and are validated against the bounded allowlist, which accepts
    /// the OpenRouter origin with the OpenAiCompat wire protocol.
    pub bounded: bool,
}

impl PreparedRequest {
    pub fn body_for_dispatch(&self) -> &[u8] {
        &self.body
    }

    pub fn authorization_for_dispatch(&self) -> Option<&[u8]> {
        self.authorization
            .as_ref()
            .map(ResolvedSecret::expose_for_dispatch)
    }
}

impl fmt::Debug for PreparedRequest {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("PreparedRequest")
            .field("method", &self.method)
            .field("origin", &self.origin)
            .field("path", &self.path)
            .field("content_type", &self.content_type)
            .field("accept", &self.accept)
            .field("network", &self.network)
            .field("protocol", &self.protocol)
            .field("body", &"[REDACTED]")
            .field(
                "authorization",
                &self.authorization.as_ref().map(|_| "[REDACTED]"),
            )
            .field("auth_header_name", &self.auth_header_name)
            .field("version_header", &self.version_header)
            .finish()
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ProviderHttpResponse {
    pub status: u16,
    pub content_type: Option<String>,
    pub retry_after: Option<String>,
    pub redirected: bool,
    pub request_id: Option<String>,
    pub chunks: Vec<Vec<u8>>,
}

#[derive(Clone, Debug, Default)]
pub struct DispatchControl {
    pub cancelled_before_dispatch: bool,
    pub forwarded_tool_result: bool,
    pub cancellation: Option<CancellationSignal>,
}

#[derive(Clone, Debug)]
pub struct CancellationSignal {
    inner: Arc<CancellationState>,
}

#[derive(Debug)]
struct CancellationState {
    cancelled: AtomicBool,
    notify: Notify,
}

impl Default for CancellationSignal {
    fn default() -> Self {
        Self::new()
    }
}

impl CancellationSignal {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(CancellationState {
                cancelled: AtomicBool::new(false),
                notify: Notify::new(),
            }),
        }
    }

    pub fn cancel(&self) {
        self.inner.cancelled.store(true, Ordering::Release);
        self.inner.notify.notify_waiters();
    }

    pub fn is_cancelled(&self) -> bool {
        self.inner.cancelled.load(Ordering::Acquire)
    }

    async fn cancelled(&self) {
        loop {
            let notified = self.inner.notify.notified();
            if self.is_cancelled() {
                return;
            }
            notified.await;
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ProviderHttpError {
    pub kind: ProviderHttpErrorKind,
    pub body_write: BodyWriteState,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProviderHttpErrorKind {
    Cancelled,
    Timeout,
    Transport,
    InvalidPreparedRequest,
    ResponseTooLarge,
}

impl fmt::Display for ProviderHttpError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("provider HTTP execution failed")
    }
}

impl std::error::Error for ProviderHttpError {}

pub type ProviderHttpFuture<'a> =
    Pin<Box<dyn Future<Output = Result<ProviderHttpResponse, ProviderHttpError>> + Send + 'a>>;

pub trait ProviderHttpExecutor: Send + Sync {
    fn execute<'a>(
        &'a self,
        request: PreparedRequest,
        control: DispatchControl,
    ) -> ProviderHttpFuture<'a>;
}

#[derive(Clone)]
pub struct ReqwestExecutor {
    client: reqwest::Client,
    absolute_timeout: Duration,
}

impl fmt::Debug for ReqwestExecutor {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ReqwestExecutor")
            .field("absolute_timeout", &self.absolute_timeout)
            .field("redirects", &"disabled")
            .field("proxy", &"disabled")
            .field("tls", &"rustls")
            .finish()
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReqwestExecutorBuildError;

impl fmt::Display for ReqwestExecutorBuildError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("provider HTTP client construction failed")
    }
}

impl std::error::Error for ReqwestExecutorBuildError {}

impl ReqwestExecutor {
    pub fn new(
        connect_timeout: Duration,
        read_timeout: Duration,
        absolute_timeout: Duration,
    ) -> Result<Self, ReqwestExecutorBuildError> {
        if connect_timeout.is_zero()
            || read_timeout.is_zero()
            || absolute_timeout.is_zero()
            || absolute_timeout < connect_timeout
        {
            return Err(ReqwestExecutorBuildError);
        }
        let client = reqwest::Client::builder()
            .use_rustls_tls()
            .redirect(reqwest::redirect::Policy::none())
            .no_proxy()
            .user_agent(FIXED_USER_AGENT)
            .connect_timeout(connect_timeout)
            .read_timeout(read_timeout)
            .build()
            .map_err(|_| ReqwestExecutorBuildError)?;
        Ok(Self {
            client,
            absolute_timeout,
        })
    }

    pub const fn user_agent(&self) -> &'static str {
        FIXED_USER_AGENT
    }

    async fn execute_bounded(
        &self,
        request: PreparedRequest,
        control: DispatchControl,
    ) -> Result<ProviderHttpResponse, ProviderHttpError> {
        validate_prepared_request(&request)?;
        if control.cancelled_before_dispatch
            || control
                .cancellation
                .as_ref()
                .is_some_and(CancellationSignal::is_cancelled)
        {
            return Err(http_error(
                ProviderHttpErrorKind::Cancelled,
                BodyWriteState::NotWritten,
            ));
        }

        let exact_url = format!("{}{}", request.origin, request.path);
        let mut builder = self
            .client
            .post(&exact_url)
            .header(CONTENT_TYPE, request.content_type)
            .header(ACCEPT, request.accept)
            .body(request.body);
        if let Some(authorization) = request.authorization {
            let value =
                sensitive_header_value(authorization.expose_for_dispatch()).map_err(|_| {
                    http_error(
                        ProviderHttpErrorKind::InvalidPreparedRequest,
                        BodyWriteState::NotWritten,
                    )
                })?;
            match request.auth_header_name {
                // The Anthropic and Gemini protocols carry the API key in a
                // custom header rather than `Authorization`.
                Some(name) => builder = builder.header(name, value),
                None => builder = builder.header(AUTHORIZATION, value),
            }
        }
        if let Some((name, value)) = request.version_header {
            builder = builder.header(name, value);
        }

        let cancellation = control.cancellation.as_ref();
        let response = tokio::select! {
            result = builder.send() => result.map_err(|error| {
                let kind = if error.is_timeout() {
                    ProviderHttpErrorKind::Timeout
                } else {
                    ProviderHttpErrorKind::Transport
                };
                http_error(kind, BodyWriteState::Unknown)
            })?,
            () = wait_for_cancellation(cancellation) => {
                return Err(http_error(
                    ProviderHttpErrorKind::Cancelled,
                    BodyWriteState::Unknown,
                ));
            }
        };

        let status = response.status().as_u16();
        let content_type = bounded_header(response.headers(), CONTENT_TYPE.as_str());
        let retry_after = bounded_header(response.headers(), "retry-after");
        let request_id = bounded_request_id(response.headers());
        let redirected = response.url().as_str() != exact_url;
        let success = response.status().as_u16() == 200;
        let mut stream = response.bytes_stream();
        let mut chunks = Vec::new();
        let mut total = 0usize;
        loop {
            let next = tokio::select! {
                next = stream.next() => next,
                () = wait_for_cancellation(cancellation) => {
                    return Err(http_error(
                        ProviderHttpErrorKind::Cancelled,
                        BodyWriteState::Written,
                    ));
                }
            };
            let Some(chunk) = next else {
                break;
            };
            let chunk = chunk.map_err(|error| {
                let kind = if error.is_timeout() {
                    ProviderHttpErrorKind::Timeout
                } else {
                    ProviderHttpErrorKind::Transport
                };
                http_error(kind, BodyWriteState::Written)
            })?;
            total = total.checked_add(chunk.len()).ok_or_else(|| {
                http_error(
                    ProviderHttpErrorKind::ResponseTooLarge,
                    BodyWriteState::Written,
                )
            })?;
            let limit = if success {
                MAX_RESPONSE_BYTES
            } else {
                MAX_ERROR_BODY_BYTES
            };
            if total > limit || (success && chunks.len() >= MAX_RESPONSE_CHUNKS) {
                if success {
                    return Err(http_error(
                        ProviderHttpErrorKind::ResponseTooLarge,
                        BodyWriteState::Written,
                    ));
                }
                break;
            }
            if success {
                chunks.push(chunk.to_vec());
            }
        }

        Ok(ProviderHttpResponse {
            status,
            content_type,
            retry_after,
            redirected,
            request_id,
            chunks,
        })
    }
}

fn sensitive_header_value(bytes: &[u8]) -> Result<reqwest::header::HeaderValue, ()> {
    let mut value = reqwest::header::HeaderValue::from_bytes(bytes).map_err(|_| ())?;
    value.set_sensitive(true);
    Ok(value)
}

impl ProviderHttpExecutor for ReqwestExecutor {
    fn execute<'a>(
        &'a self,
        request: PreparedRequest,
        control: DispatchControl,
    ) -> ProviderHttpFuture<'a> {
        Box::pin(async move {
            tokio::time::timeout(
                self.absolute_timeout,
                self.execute_bounded(request, control),
            )
            .await
            .map_err(|_| http_error(ProviderHttpErrorKind::Timeout, BodyWriteState::Unknown))?
        })
    }
}

async fn wait_for_cancellation(signal: Option<&CancellationSignal>) {
    match signal {
        Some(signal) => signal.cancelled().await,
        None => std::future::pending().await,
    }
}

fn validate_prepared_request(request: &PreparedRequest) -> Result<(), ProviderHttpError> {
    let base_policy = !request.network.allow_proxy && !request.network.follow_redirects;
    let localhost = request.origin == "http://127.0.0.1:11434"
        && request.path == "/api/chat"
        && request.protocol == ProviderWireProtocol::OllamaNdjson
        && !request.network.allow_dns
        && !request.network.require_tls
        && request.authorization.is_none();
    let direct_ollama = request.origin == "https://ollama.com"
        && request.path == "/api/chat"
        && request.protocol == ProviderWireProtocol::OllamaNdjson
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some();
    let openrouter = request.origin == "https://openrouter.ai"
        && request.path == "/api/v1/chat/completions"
        && request.protocol == ProviderWireProtocol::OpenRouterSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some();
    let openai_compat = matches!(
        (request.origin.as_str(), request.path.as_str()),
        (
            router_protocol::DASHSCOPE_ORIGIN,
            router_protocol::DASHSCOPE_CHAT_PATH
        ) | (
            router_protocol::DASHSCOPE_TOKEN_PLAN_ORIGIN,
            router_protocol::DASHSCOPE_CHAT_PATH
        ) | (router_protocol::XAI_ORIGIN, router_protocol::XAI_CHAT_PATH)
            | (router_protocol::ZAI_ORIGIN, router_protocol::ZAI_CHAT_PATH)
            | (
                router_protocol::MOONSHOT_ORIGIN,
                router_protocol::MOONSHOT_CHAT_PATH
            )
            | (
                router_protocol::DEEPSEEK_ORIGIN,
                router_protocol::DEEPSEEK_CHAT_PATH
            )
    ) && request.protocol == ProviderWireProtocol::OpenAiCompatSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some();
    // The Anthropic Messages protocol: static `/v1/messages` path, the API key
    // in the `x-api-key` header, and the pinned `anthropic-version` header.
    let anthropic = request.origin == router_protocol::ANTHROPIC_ORIGIN
        && request.path == router_protocol::ANTHROPIC_CHAT_PATH
        && request.protocol == ProviderWireProtocol::AnthropicSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some()
        && request.auth_header_name == Some(router_protocol::ANTHROPIC_API_KEY_HEADER)
        && request.version_header
            == Some((
                router_protocol::ANTHROPIC_VERSION_HEADER,
                router_protocol::ANTHROPIC_VERSION,
            ));
    // The Gemini generateContent protocol: the API key in the `x-goog-api-key`
    // header and the model id materialized in the path.
    let gemini = request.origin == router_protocol::GEMINI_ORIGIN
        && request.protocol == ProviderWireProtocol::GeminiSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some()
        && request.auth_header_name == Some(router_protocol::GEMINI_API_KEY_HEADER)
        && request.version_header.is_none();
    let accept_matches = match request.protocol {
        ProviderWireProtocol::OpenRouterSse
        | ProviderWireProtocol::OpenAiCompatSse
        | ProviderWireProtocol::AnthropicSse
        | ProviderWireProtocol::GeminiSse => request.accept == "text/event-stream",
        ProviderWireProtocol::OllamaNdjson => {
            matches!(request.accept, "application/x-ndjson" | "application/json")
        }
    };
    // Bounded non-streaming completion dispatch (model intelligence
    // categorization). The request is genuinely non-streaming, so it carries
    // `application/json` accept; the allowlist reuses the streaming endpoint
    // predicates but accepts the OpenRouter origin with the OpenAiCompat wire
    // protocol (the only shape that differs from the streaming allowlist).
    let bounded_openrouter = request.origin == "https://openrouter.ai"
        && request.path == "/api/v1/chat/completions"
        && request.protocol == ProviderWireProtocol::OpenAiCompatSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some();
    // The bounded Anthropic and Gemini calls are non-streaming and carry the
    // API key in the same custom headers as their streaming siblings (plus the
    // pinned `anthropic-version` header for Anthropic). The Gemini path is
    // model-materialized, so like the streaming predicate it is not fixed.
    let bounded_anthropic = request.origin == router_protocol::ANTHROPIC_ORIGIN
        && request.path == router_protocol::ANTHROPIC_CHAT_PATH
        && request.protocol == ProviderWireProtocol::AnthropicSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some()
        && request.auth_header_name == Some(router_protocol::ANTHROPIC_API_KEY_HEADER)
        && request.version_header
            == Some((
                router_protocol::ANTHROPIC_VERSION_HEADER,
                router_protocol::ANTHROPIC_VERSION,
            ));
    let bounded_gemini = request.origin == router_protocol::GEMINI_ORIGIN
        && request.protocol == ProviderWireProtocol::GeminiSse
        && request.network.allow_dns
        && request.network.require_tls
        && request.authorization.is_some()
        && request.auth_header_name == Some(router_protocol::GEMINI_API_KEY_HEADER)
        && request.version_header.is_none();
    let bounded = request.bounded
        && (localhost
            || direct_ollama
            || bounded_openrouter
            || openai_compat
            || bounded_anthropic
            || bounded_gemini);
    if request.method != "POST"
        || request.content_type != "application/json"
        || !(accept_matches || (request.bounded && request.accept == "application/json"))
        || request.body.is_empty()
        || request.body.len() > MAX_REQUEST_BODY_BYTES
        || !base_policy
        || !(localhost
            || direct_ollama
            || openrouter
            || openai_compat
            || anthropic
            || gemini
            || bounded)
    {
        return Err(http_error(
            ProviderHttpErrorKind::InvalidPreparedRequest,
            BodyWriteState::NotWritten,
        ));
    }
    Ok(())
}

fn bounded_header(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
    headers
        .get(name)
        .and_then(|value| value.to_str().ok())
        .filter(|value| value.len() <= MAX_REQUEST_ID_BYTES)
        .map(str::to_owned)
}

fn bounded_request_id(headers: &reqwest::header::HeaderMap) -> Option<String> {
    ["x-request-id", "x-openrouter-request-id"]
        .into_iter()
        .find_map(|name| bounded_header(headers, name))
        .filter(|value| {
            !value.is_empty()
                && value.bytes().all(|byte| {
                    byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':')
                })
        })
}

const fn http_error(kind: ProviderHttpErrorKind, body_write: BodyWriteState) -> ProviderHttpError {
    ProviderHttpError { kind, body_write }
}

#[derive(Clone, Debug, PartialEq)]
pub enum NormalizedProviderEvent {
    Ollama(OllamaStreamEvent),
    OpenRouter(OpenRouterStreamEvent),
    OpenAiCompat(OpenAiCompatStreamEvent),
    Anthropic(AnthropicStreamEvent),
    Gemini(GeminiStreamEvent),
}

#[derive(Clone, Debug, PartialEq)]
pub struct DispatchSuccess {
    pub events: Vec<NormalizedProviderEvent>,
    pub replay: ReplayDecision,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DispatchFailureKind {
    Cancelled,
    SecretResolution,
    ResponseBounds,
    Executor(ProviderHttpErrorKind),
    OllamaHttp(OllamaFailure),
    OpenRouterHttp(OpenRouterFailure),
    OpenAiCompatHttp(OpenAiCompatFailure),
    AnthropicHttp(AnthropicFailure),
    GeminiHttp(GeminiFailure),
    OllamaStream(OllamaStreamError),
    OpenRouterStream(OpenRouterStreamError),
    OpenAiCompatStream(OpenAiCompatStreamError),
    AnthropicStream(AnthropicStreamError),
    GeminiStream(GeminiStreamError),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DispatchFailure {
    pub kind: DispatchFailureKind,
    pub replay: ReplayDecision,
}

impl fmt::Display for DispatchFailure {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("provider dispatch failed")
    }
}

impl std::error::Error for DispatchFailure {}

pub async fn dispatch_ollama(
    spec: OllamaRequestSpec,
    attempt_id: &AttemptId,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
    control: DispatchControl,
) -> Result<DispatchSuccess, DispatchFailure> {
    let mut replay = OllamaReplayGuard::default();
    if control.forwarded_tool_result {
        replay.mark_forwarded_tool_result();
    }
    if control.cancelled_before_dispatch
        || control
            .cancellation
            .as_ref()
            .is_some_and(CancellationSignal::is_cancelled)
    {
        return Err(DispatchFailure {
            kind: DispatchFailureKind::Cancelled,
            replay: replay.decision(),
        });
    }

    let request = prepare_ollama(spec, resolver).map_err(|_| DispatchFailure {
        kind: DispatchFailureKind::SecretResolution,
        replay: replay.decision(),
    })?;
    let response = match executor.execute(request, control).await {
        Ok(response) => response,
        Err(error) => {
            if error.body_write != BodyWriteState::NotWritten {
                replay.mark_body_written();
                replay.mark_transport_outcome_unknown();
            }
            let kind = match error.kind {
                ProviderHttpErrorKind::Cancelled => {
                    DispatchFailureKind::OllamaHttp(normalize_ollama_transport_failure(
                        TransportFailureSignal::Cancelled,
                        error.body_write,
                    ))
                }
                ProviderHttpErrorKind::Timeout | ProviderHttpErrorKind::Transport => {
                    DispatchFailureKind::OllamaHttp(normalize_ollama_transport_failure(
                        TransportFailureSignal::Timeout,
                        error.body_write,
                    ))
                }
                ProviderHttpErrorKind::ResponseTooLarge => DispatchFailureKind::ResponseBounds,
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    DispatchFailureKind::Executor(error.kind)
                }
            };
            return Err(DispatchFailure {
                kind,
                replay: replay.decision(),
            });
        }
    };
    replay.mark_body_written();
    normalize_ollama_response(
        response.status,
        response.content_type.as_deref(),
        response.retry_after.as_deref(),
        response.redirected,
    )
    .map_err(|failure| DispatchFailure {
        kind: DispatchFailureKind::OllamaHttp(failure),
        replay: replay.decision(),
    })?;
    validate_response_bounds(&response.chunks).map_err(|_| {
        replay.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::ResponseBounds,
            replay: replay.decision(),
        }
    })?;

    let mut parser = OllamaNdjsonParser::new(attempt_id);
    let mut events = Vec::new();
    for chunk in response.chunks {
        let parsed = parser.feed(&chunk).map_err(|error| {
            replay.mark_transport_outcome_unknown();
            DispatchFailure {
                kind: DispatchFailureKind::OllamaStream(error),
                replay: replay.decision(),
            }
        })?;
        for event in parsed {
            replay.observe(&event);
            events.push(NormalizedProviderEvent::Ollama(event));
        }
    }
    for event in parser.finish().map_err(|error| {
        replay.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::OllamaStream(error),
            replay: replay.decision(),
        }
    })? {
        replay.observe(&event);
        events.push(NormalizedProviderEvent::Ollama(event));
    }
    Ok(DispatchSuccess {
        events,
        replay: replay.decision(),
    })
}

pub async fn dispatch_openrouter(
    spec: OpenRouterRequestSpec,
    attempt_id: &AttemptId,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
    control: DispatchControl,
) -> Result<DispatchSuccess, DispatchFailure> {
    let mut parser = OpenRouterSseParser::new(attempt_id);
    if control.forwarded_tool_result {
        parser.mark_forwarded_tool_result();
    }
    if control.cancelled_before_dispatch
        || control
            .cancellation
            .as_ref()
            .is_some_and(CancellationSignal::is_cancelled)
    {
        return Err(DispatchFailure {
            kind: DispatchFailureKind::Cancelled,
            replay: parser.replay_decision(),
        });
    }
    let request = prepare_openrouter(spec, resolver).map_err(|_| DispatchFailure {
        kind: DispatchFailureKind::SecretResolution,
        replay: parser.replay_decision(),
    })?;
    let response = match executor.execute(request, control).await {
        Ok(response) => response,
        Err(error) => {
            if error.body_write != BodyWriteState::NotWritten {
                parser.mark_body_written();
                parser.mark_transport_outcome_unknown();
            }
            let signal = match error.kind {
                ProviderHttpErrorKind::Cancelled => {
                    crate::openrouter_chat::TransportFailureSignal::Cancelled
                }
                ProviderHttpErrorKind::Timeout | ProviderHttpErrorKind::Transport => {
                    crate::openrouter_chat::TransportFailureSignal::Timeout
                }
                ProviderHttpErrorKind::ResponseTooLarge => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::ResponseBounds,
                        replay: parser.replay_decision(),
                    });
                }
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::Executor(error.kind),
                        replay: parser.replay_decision(),
                    });
                }
            };
            let body_write = match error.body_write {
                BodyWriteState::NotWritten => crate::openrouter_chat::BodyWriteState::NotWritten,
                BodyWriteState::Written => crate::openrouter_chat::BodyWriteState::Written,
                BodyWriteState::Unknown => crate::openrouter_chat::BodyWriteState::Unknown,
            };
            return Err(DispatchFailure {
                kind: DispatchFailureKind::OpenRouterHttp(normalize_openrouter_transport_failure(
                    signal, body_write,
                )),
                replay: parser.replay_decision(),
            });
        }
    };
    parser.mark_body_written();
    normalize_openrouter_response(
        response.status,
        response.content_type.as_deref(),
        response.retry_after.as_deref(),
        response.redirected,
    )
    .map_err(|failure| DispatchFailure {
        kind: DispatchFailureKind::OpenRouterHttp(failure),
        replay: parser.replay_decision(),
    })?;
    validate_response_bounds(&response.chunks).map_err(|_| {
        parser.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::ResponseBounds,
            replay: parser.replay_decision(),
        }
    })?;

    let mut events = Vec::new();
    for chunk in response.chunks {
        events.extend(
            parser
                .feed(&chunk)
                .map_err(|error| {
                    parser.mark_transport_outcome_unknown();
                    DispatchFailure {
                        kind: DispatchFailureKind::OpenRouterStream(error),
                        replay: parser.replay_decision(),
                    }
                })?
                .into_iter()
                .map(NormalizedProviderEvent::OpenRouter),
        );
    }
    events.extend(
        parser
            .finish()
            .map_err(|error| {
                parser.mark_transport_outcome_unknown();
                DispatchFailure {
                    kind: DispatchFailureKind::OpenRouterStream(error),
                    replay: parser.replay_decision(),
                }
            })?
            .into_iter()
            .map(NormalizedProviderEvent::OpenRouter),
    );
    Ok(DispatchSuccess {
        events,
        replay: parser.replay_decision(),
    })
}

pub async fn dispatch_openai_compat(
    spec: OpenAiCompatRequestSpec,
    attempt_id: &AttemptId,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
    control: DispatchControl,
) -> Result<DispatchSuccess, DispatchFailure> {
    let mut parser = OpenAiCompatSseParser::new(attempt_id);
    if control.forwarded_tool_result {
        parser.mark_forwarded_tool_result();
    }
    if control.cancelled_before_dispatch
        || control
            .cancellation
            .as_ref()
            .is_some_and(CancellationSignal::is_cancelled)
    {
        return Err(DispatchFailure {
            kind: DispatchFailureKind::Cancelled,
            replay: parser.replay_decision(),
        });
    }
    let request = prepare_openai_compat(spec, resolver).map_err(|_| DispatchFailure {
        kind: DispatchFailureKind::SecretResolution,
        replay: parser.replay_decision(),
    })?;
    let response = match executor.execute(request, control).await {
        Ok(response) => response,
        Err(error) => {
            if error.body_write != crate::ollama_chat::BodyWriteState::NotWritten {
                parser.mark_body_written();
                parser.mark_transport_outcome_unknown();
            }
            let signal = match error.kind {
                ProviderHttpErrorKind::Cancelled => OpenAiCompatTransportFailureSignal::Cancelled,
                ProviderHttpErrorKind::Timeout | ProviderHttpErrorKind::Transport => {
                    OpenAiCompatTransportFailureSignal::Timeout
                }
                ProviderHttpErrorKind::ResponseTooLarge => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::ResponseBounds,
                        replay: parser.replay_decision(),
                    });
                }
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::Executor(error.kind),
                        replay: parser.replay_decision(),
                    });
                }
            };
            let body_write = match error.body_write {
                crate::ollama_chat::BodyWriteState::NotWritten => {
                    OpenAiCompatBodyWriteState::NotWritten
                }
                crate::ollama_chat::BodyWriteState::Written => OpenAiCompatBodyWriteState::Written,
                crate::ollama_chat::BodyWriteState::Unknown => OpenAiCompatBodyWriteState::Unknown,
            };
            return Err(DispatchFailure {
                kind: DispatchFailureKind::OpenAiCompatHttp(
                    normalize_openai_compat_transport_failure(signal, body_write),
                ),
                replay: parser.replay_decision(),
            });
        }
    };
    parser.mark_body_written();
    normalize_openai_compat_response(
        response.status,
        response.content_type.as_deref(),
        response.retry_after.as_deref(),
        response.redirected,
    )
    .map_err(|failure| DispatchFailure {
        kind: DispatchFailureKind::OpenAiCompatHttp(failure),
        replay: parser.replay_decision(),
    })?;
    validate_response_bounds(&response.chunks).map_err(|_| {
        parser.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::ResponseBounds,
            replay: parser.replay_decision(),
        }
    })?;

    let mut events = Vec::new();
    for chunk in response.chunks {
        events.extend(
            parser
                .feed(&chunk)
                .map_err(|error| {
                    parser.mark_transport_outcome_unknown();
                    DispatchFailure {
                        kind: DispatchFailureKind::OpenAiCompatStream(error),
                        replay: parser.replay_decision(),
                    }
                })?
                .into_iter()
                .map(NormalizedProviderEvent::OpenAiCompat),
        );
    }
    events.extend(
        parser
            .finish()
            .map_err(|error| {
                parser.mark_transport_outcome_unknown();
                DispatchFailure {
                    kind: DispatchFailureKind::OpenAiCompatStream(error),
                    replay: parser.replay_decision(),
                }
            })?
            .into_iter()
            .map(NormalizedProviderEvent::OpenAiCompat),
    );
    Ok(DispatchSuccess {
        events,
        replay: parser.replay_decision(),
    })
}

pub async fn dispatch_anthropic(
    spec: AnthropicRequestSpec,
    attempt_id: &AttemptId,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
    control: DispatchControl,
) -> Result<DispatchSuccess, DispatchFailure> {
    let mut parser = AnthropicSseParser::new(attempt_id);
    if control.cancelled_before_dispatch
        || control
            .cancellation
            .as_ref()
            .is_some_and(CancellationSignal::is_cancelled)
    {
        return Err(DispatchFailure {
            kind: DispatchFailureKind::Cancelled,
            replay: parser.replay_decision(),
        });
    }
    let request = prepare_anthropic(spec, resolver).map_err(|_| DispatchFailure {
        kind: DispatchFailureKind::SecretResolution,
        replay: parser.replay_decision(),
    })?;
    let response = match executor.execute(request, control).await {
        Ok(response) => response,
        Err(error) => {
            if error.body_write != crate::ollama_chat::BodyWriteState::NotWritten {
                parser.mark_body_written();
                parser.mark_transport_outcome_unknown();
            }
            let signal = match error.kind {
                ProviderHttpErrorKind::Cancelled => AnthropicTransportFailureSignal::Cancelled,
                ProviderHttpErrorKind::Timeout | ProviderHttpErrorKind::Transport => {
                    AnthropicTransportFailureSignal::Timeout
                }
                ProviderHttpErrorKind::ResponseTooLarge => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::ResponseBounds,
                        replay: parser.replay_decision(),
                    });
                }
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::Executor(error.kind),
                        replay: parser.replay_decision(),
                    });
                }
            };
            let body_write = match error.body_write {
                crate::ollama_chat::BodyWriteState::NotWritten => {
                    AnthropicBodyWriteState::NotWritten
                }
                crate::ollama_chat::BodyWriteState::Written => AnthropicBodyWriteState::Written,
                crate::ollama_chat::BodyWriteState::Unknown => AnthropicBodyWriteState::Unknown,
            };
            return Err(DispatchFailure {
                kind: DispatchFailureKind::AnthropicHttp(normalize_anthropic_transport_failure(
                    signal, body_write,
                )),
                replay: parser.replay_decision(),
            });
        }
    };
    parser.mark_body_written();
    normalize_anthropic_response(
        response.status,
        response.content_type.as_deref(),
        response.retry_after.as_deref(),
        response.redirected,
    )
    .map_err(|failure| DispatchFailure {
        kind: DispatchFailureKind::AnthropicHttp(failure),
        replay: parser.replay_decision(),
    })?;
    validate_response_bounds(&response.chunks).map_err(|_| {
        parser.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::ResponseBounds,
            replay: parser.replay_decision(),
        }
    })?;

    let mut events = Vec::new();
    for chunk in response.chunks {
        events.extend(
            parser
                .feed(&chunk)
                .map_err(|error| {
                    parser.mark_transport_outcome_unknown();
                    DispatchFailure {
                        kind: DispatchFailureKind::AnthropicStream(error),
                        replay: parser.replay_decision(),
                    }
                })?
                .into_iter()
                .map(NormalizedProviderEvent::Anthropic),
        );
    }
    events.extend(
        parser
            .finish()
            .map_err(|error| {
                parser.mark_transport_outcome_unknown();
                DispatchFailure {
                    kind: DispatchFailureKind::AnthropicStream(error),
                    replay: parser.replay_decision(),
                }
            })?
            .into_iter()
            .map(NormalizedProviderEvent::Anthropic),
    );
    Ok(DispatchSuccess {
        events,
        replay: parser.replay_decision(),
    })
}

pub async fn dispatch_gemini(
    spec: GeminiRequestSpec,
    attempt_id: &AttemptId,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
    control: DispatchControl,
) -> Result<DispatchSuccess, DispatchFailure> {
    let mut parser = GeminiSseParser::new(attempt_id);
    if control.cancelled_before_dispatch
        || control
            .cancellation
            .as_ref()
            .is_some_and(CancellationSignal::is_cancelled)
    {
        return Err(DispatchFailure {
            kind: DispatchFailureKind::Cancelled,
            replay: parser.replay_decision(),
        });
    }
    let request = prepare_gemini(spec, resolver).map_err(|_| DispatchFailure {
        kind: DispatchFailureKind::SecretResolution,
        replay: parser.replay_decision(),
    })?;
    let response = match executor.execute(request, control).await {
        Ok(response) => response,
        Err(error) => {
            if error.body_write != crate::ollama_chat::BodyWriteState::NotWritten {
                parser.mark_body_written();
                parser.mark_transport_outcome_unknown();
            }
            let signal = match error.kind {
                ProviderHttpErrorKind::Cancelled => GeminiTransportFailureSignal::Cancelled,
                ProviderHttpErrorKind::Timeout | ProviderHttpErrorKind::Transport => {
                    GeminiTransportFailureSignal::Timeout
                }
                ProviderHttpErrorKind::ResponseTooLarge => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::ResponseBounds,
                        replay: parser.replay_decision(),
                    });
                }
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    return Err(DispatchFailure {
                        kind: DispatchFailureKind::Executor(error.kind),
                        replay: parser.replay_decision(),
                    });
                }
            };
            let body_write = match error.body_write {
                crate::ollama_chat::BodyWriteState::NotWritten => GeminiBodyWriteState::NotWritten,
                crate::ollama_chat::BodyWriteState::Written => GeminiBodyWriteState::Written,
                crate::ollama_chat::BodyWriteState::Unknown => GeminiBodyWriteState::Unknown,
            };
            return Err(DispatchFailure {
                kind: DispatchFailureKind::GeminiHttp(normalize_gemini_transport_failure(
                    signal, body_write,
                )),
                replay: parser.replay_decision(),
            });
        }
    };
    parser.mark_body_written();
    normalize_gemini_response(
        response.status,
        response.content_type.as_deref(),
        response.retry_after.as_deref(),
        response.redirected,
    )
    .map_err(|failure| DispatchFailure {
        kind: DispatchFailureKind::GeminiHttp(failure),
        replay: parser.replay_decision(),
    })?;
    validate_response_bounds(&response.chunks).map_err(|_| {
        parser.mark_transport_outcome_unknown();
        DispatchFailure {
            kind: DispatchFailureKind::ResponseBounds,
            replay: parser.replay_decision(),
        }
    })?;

    let mut events = Vec::new();
    for chunk in response.chunks {
        events.extend(
            parser
                .feed(&chunk)
                .map_err(|error| {
                    parser.mark_transport_outcome_unknown();
                    DispatchFailure {
                        kind: DispatchFailureKind::GeminiStream(error),
                        replay: parser.replay_decision(),
                    }
                })?
                .into_iter()
                .map(NormalizedProviderEvent::Gemini),
        );
    }
    events.extend(
        parser
            .finish()
            .map_err(|error| {
                parser.mark_transport_outcome_unknown();
                DispatchFailure {
                    kind: DispatchFailureKind::GeminiStream(error),
                    replay: parser.replay_decision(),
                }
            })?
            .into_iter()
            .map(NormalizedProviderEvent::Gemini),
    );
    Ok(DispatchSuccess {
        events,
        replay: parser.replay_decision(),
    })
}

fn prepare_anthropic(
    spec: AnthropicRequestSpec,
    resolver: &dyn SecretResolver,
) -> Result<PreparedRequest, SecretResolutionError> {
    let body = spec.body_bytes().map_err(|_| SecretResolutionError)?;
    let authorization = resolve_api_key_header(resolver, spec.api_key_secret.as_str())?;
    Ok(PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path: spec.path,
        content_type: spec.content_type,
        accept: spec.accept,
        network: spec.network,
        protocol: ProviderWireProtocol::AnthropicSse,
        body,
        authorization: Some(authorization),
        auth_header_name: Some(router_protocol::ANTHROPIC_API_KEY_HEADER),
        version_header: Some((
            router_protocol::ANTHROPIC_VERSION_HEADER,
            spec.version_header,
        )),
        bounded: false,
    })
}

fn prepare_gemini(
    spec: GeminiRequestSpec,
    resolver: &dyn SecretResolver,
) -> Result<PreparedRequest, SecretResolutionError> {
    let body = spec.body_bytes().map_err(|_| SecretResolutionError)?;
    let authorization = resolve_api_key_header(resolver, spec.api_key_secret.as_str())?;
    Ok(PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path: spec.path,
        content_type: spec.content_type,
        accept: spec.accept,
        network: spec.network,
        protocol: ProviderWireProtocol::GeminiSse,
        body,
        authorization: Some(authorization),
        auth_header_name: Some(router_protocol::GEMINI_API_KEY_HEADER),
        version_header: None,
        bounded: false,
    })
}

fn prepare_openai_compat(
    spec: OpenAiCompatRequestSpec,
    resolver: &dyn SecretResolver,
) -> Result<PreparedRequest, SecretResolutionError> {
    let body = spec.body_bytes().map_err(|_| SecretResolutionError)?;
    let authorization = resolve_bearer_header(resolver, spec.bearer_secret.as_str())?;
    Ok(PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path: spec.path,
        content_type: spec.content_type,
        accept: spec.accept,
        network: spec.network,
        protocol: ProviderWireProtocol::OpenAiCompatSse,
        body,
        authorization: Some(authorization),
        auth_header_name: None,
        version_header: None,
        bounded: false,
    })
}

fn prepare_ollama(
    spec: OllamaRequestSpec,
    resolver: &dyn SecretResolver,
) -> Result<PreparedRequest, SecretResolutionError> {
    let body = spec.body_bytes().map_err(|_| SecretResolutionError)?;
    let localhost = spec.origin == "http://127.0.0.1:11434";
    let authorization = match &spec.authentication {
        OllamaAuthentication::None => None,
        OllamaAuthentication::BearerSecretReference(reference) if !localhost => {
            Some(resolve_bearer_header(resolver, reference.as_str())?)
        }
        OllamaAuthentication::BearerSecretReference(_) => return Err(SecretResolutionError),
    };
    Ok(PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path: spec.path,
        content_type: spec.content_type,
        accept: spec.accept,
        network: spec.network,
        protocol: ProviderWireProtocol::OllamaNdjson,
        body,
        authorization,
        auth_header_name: None,
        version_header: None,
        bounded: false,
    })
}

fn prepare_openrouter(
    spec: OpenRouterRequestSpec,
    resolver: &dyn SecretResolver,
) -> Result<PreparedRequest, SecretResolutionError> {
    let body = spec.body_bytes().map_err(|_| SecretResolutionError)?;
    let authorization = resolve_bearer_header(resolver, spec.bearer_secret.as_str())?;
    Ok(PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path: spec.path,
        content_type: spec.content_type,
        accept: spec.accept,
        network: spec.network,
        protocol: ProviderWireProtocol::OpenRouterSse,
        body,
        authorization: Some(authorization),
        auth_header_name: None,
        version_header: None,
        bounded: false,
    })
}

/// A bounded, non-streaming chat completion used by the model intelligence
/// loop. The response is validated against `max_response_bytes` before parsing,
/// and only the completion text escapes the transport layer.
#[derive(Clone, PartialEq, Eq)]
pub struct BoundedChatCompletionSpec {
    pub origin: String,
    pub path: String,
    pub model: String,
    pub prompt: String,
    pub bearer_secret: Option<String>,
    pub protocol: BoundedChatCompletionProtocol,
    pub max_response_bytes: usize,
}

impl fmt::Debug for BoundedChatCompletionSpec {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("BoundedChatCompletionSpec")
            .field("origin", &self.origin)
            .field("path", &self.path)
            .field("model", &self.model)
            .field("prompt", &"[redacted]")
            .field("protocol", &self.protocol)
            .field("max_response_bytes", &self.max_response_bytes)
            .finish_non_exhaustive()
    }
}

/// The wire protocol for a bounded chat completion. The OpenAI-compatible and
/// Ollama protocols share the chat-completions request shape with
/// `stream: false`; they differ from the Anthropic (`/v1/messages`) and Gemini
/// (`generateContent`) protocols in request shape, response framing, and
/// authentication.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BoundedChatCompletionProtocol {
    Ollama,
    OpenAiCompat,
    Anthropic,
    Gemini,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BoundedChatCompletionErrorKind {
    SecretResolution,
    InvalidPreparedRequest,
    Cancelled,
    Timeout,
    Transport,
    ResponseTooLarge,
    Http { status: u16 },
    InvalidResponse,
    EmptyCompletion,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BoundedChatCompletionError {
    pub kind: BoundedChatCompletionErrorKind,
}

impl fmt::Display for BoundedChatCompletionError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self.kind {
            BoundedChatCompletionErrorKind::SecretResolution => {
                formatter.write_str("bounded chat completion credential resolution failed")
            }
            BoundedChatCompletionErrorKind::InvalidPreparedRequest => {
                formatter.write_str("bounded chat completion request was invalid")
            }
            BoundedChatCompletionErrorKind::Cancelled => {
                formatter.write_str("bounded chat completion was cancelled")
            }
            BoundedChatCompletionErrorKind::Timeout => {
                formatter.write_str("bounded chat completion timed out")
            }
            BoundedChatCompletionErrorKind::Transport => {
                formatter.write_str("bounded chat completion transport failed")
            }
            BoundedChatCompletionErrorKind::ResponseTooLarge => {
                formatter.write_str("bounded chat completion response exceeded its bound")
            }
            BoundedChatCompletionErrorKind::Http { status } => {
                write!(
                    formatter,
                    "bounded chat completion provider returned HTTP {status}"
                )
            }
            BoundedChatCompletionErrorKind::InvalidResponse => {
                formatter.write_str("bounded chat completion response was invalid")
            }
            BoundedChatCompletionErrorKind::EmptyCompletion => {
                formatter.write_str("bounded chat completion returned no text")
            }
        }
    }
}

impl std::error::Error for BoundedChatCompletionError {}

impl From<SecretResolutionError> for BoundedChatCompletionError {
    fn from(_: SecretResolutionError) -> Self {
        Self {
            kind: BoundedChatCompletionErrorKind::SecretResolution,
        }
    }
}

/// Dispatches one bounded, non-streaming chat completion and returns the
/// completion text. The response is bounded before parsing and the returned
/// text is the only content that escapes the transport layer.
pub async fn dispatch_bounded_chat_completion(
    spec: BoundedChatCompletionSpec,
    resolver: &dyn SecretResolver,
    executor: &dyn ProviderHttpExecutor,
) -> Result<String, BoundedChatCompletionError> {
    let body = build_bounded_chat_body(spec.protocol, &spec.model, &spec.prompt)?;
    let authorization = match (&spec.protocol, spec.bearer_secret.as_deref()) {
        (BoundedChatCompletionProtocol::Ollama, None) => None,
        (BoundedChatCompletionProtocol::Ollama, Some(secret_ref))
        | (BoundedChatCompletionProtocol::OpenAiCompat, Some(secret_ref)) => {
            Some(resolve_bearer_header(resolver, secret_ref)?)
        }
        (BoundedChatCompletionProtocol::OpenAiCompat, None) => {
            return Err(BoundedChatCompletionError {
                kind: BoundedChatCompletionErrorKind::SecretResolution,
            });
        }
        // The Anthropic and Gemini protocols carry the raw key in a custom
        // header (`x-api-key`, `x-goog-api-key`), never a Bearer prefix.
        (BoundedChatCompletionProtocol::Anthropic, Some(secret_ref))
        | (BoundedChatCompletionProtocol::Gemini, Some(secret_ref)) => {
            Some(resolve_api_key_header(resolver, secret_ref)?)
        }
        (BoundedChatCompletionProtocol::Anthropic, None)
        | (BoundedChatCompletionProtocol::Gemini, None) => {
            return Err(BoundedChatCompletionError {
                kind: BoundedChatCompletionErrorKind::SecretResolution,
            });
        }
    };
    // The Gemini model id is a path parameter (`models/{model}:generateContent`),
    // so the bounded request materializes the route template with the model id.
    let path = match spec.protocol {
        BoundedChatCompletionProtocol::Gemini => {
            materialize_gemini_bounded_path(&spec.path, &spec.model)
        }
        _ => spec.path,
    };
    // Local Ollama is a loopback daemon: no DNS, no TLS, no authorization.
    // Every other bounded endpoint is a remote provider: DNS and TLS required,
    // authorization required (enforced by the caller for cloud providers).
    let local = spec.origin == "http://127.0.0.1:11434";
    let request = PreparedRequest {
        method: "POST",
        origin: spec.origin,
        path,
        content_type: "application/json",
        accept: "application/json",
        network: NetworkPolicy {
            allow_proxy: false,
            allow_dns: !local,
            follow_redirects: false,
            require_tls: !local,
        },
        protocol: match spec.protocol {
            BoundedChatCompletionProtocol::Ollama => ProviderWireProtocol::OllamaNdjson,
            BoundedChatCompletionProtocol::OpenAiCompat => ProviderWireProtocol::OpenAiCompatSse,
            BoundedChatCompletionProtocol::Anthropic => ProviderWireProtocol::AnthropicSse,
            BoundedChatCompletionProtocol::Gemini => ProviderWireProtocol::GeminiSse,
        },
        body,
        authorization,
        auth_header_name: match spec.protocol {
            BoundedChatCompletionProtocol::Anthropic => {
                Some(router_protocol::ANTHROPIC_API_KEY_HEADER)
            }
            BoundedChatCompletionProtocol::Gemini => Some(router_protocol::GEMINI_API_KEY_HEADER),
            _ => None,
        },
        version_header: match spec.protocol {
            BoundedChatCompletionProtocol::Anthropic => Some((
                router_protocol::ANTHROPIC_VERSION_HEADER,
                router_protocol::ANTHROPIC_VERSION,
            )),
            _ => None,
        },
        bounded: true,
    };
    let response = executor
        .execute(request, DispatchControl::default())
        .await
        .map_err(|error| BoundedChatCompletionError {
            kind: match error.kind {
                ProviderHttpErrorKind::Cancelled => BoundedChatCompletionErrorKind::Cancelled,
                ProviderHttpErrorKind::Timeout => BoundedChatCompletionErrorKind::Timeout,
                ProviderHttpErrorKind::Transport => BoundedChatCompletionErrorKind::Transport,
                ProviderHttpErrorKind::ResponseTooLarge => {
                    BoundedChatCompletionErrorKind::ResponseTooLarge
                }
                ProviderHttpErrorKind::InvalidPreparedRequest => {
                    BoundedChatCompletionErrorKind::InvalidPreparedRequest
                }
            },
        })?;
    if response.redirected || response.status != 200 {
        return Err(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::Http {
                status: response.status,
            },
        });
    }
    let mut total = 0usize;
    for chunk in &response.chunks {
        total = total
            .checked_add(chunk.len())
            .ok_or(BoundedChatCompletionError {
                kind: BoundedChatCompletionErrorKind::ResponseTooLarge,
            })?;
        if total > spec.max_response_bytes {
            return Err(BoundedChatCompletionError {
                kind: BoundedChatCompletionErrorKind::ResponseTooLarge,
            });
        }
    }
    let text = match spec.protocol {
        BoundedChatCompletionProtocol::Ollama => parse_ollama_bounded_completion(&response.chunks)?,
        BoundedChatCompletionProtocol::OpenAiCompat => {
            parse_openai_compat_bounded_completion(&response.chunks)?
        }
        BoundedChatCompletionProtocol::Anthropic => {
            parse_anthropic_bounded_completion(&response.chunks)?
        }
        BoundedChatCompletionProtocol::Gemini => parse_gemini_bounded_completion(&response.chunks)?,
    };
    if text.is_empty() {
        return Err(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::EmptyCompletion,
        });
    }
    Ok(text)
}

fn build_bounded_chat_body(
    protocol: BoundedChatCompletionProtocol,
    model: &str,
    prompt: &str,
) -> Result<Vec<u8>, BoundedChatCompletionError> {
    let body = match protocol {
        // Ollama and the OpenAI-compatible direct APIs share the
        // chat-completions request shape.
        BoundedChatCompletionProtocol::Ollama | BoundedChatCompletionProtocol::OpenAiCompat => {
            serde_json::json!({
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": false,
            })
        }
        // The Anthropic Messages body requires `max_tokens` and carries the
        // model id in the body (the path is static).
        BoundedChatCompletionProtocol::Anthropic => serde_json::json!({
            "model": model,
            "max_tokens": crate::anthropic_chat::DEFAULT_MAX_TOKENS,
            "messages": [{"role": "user", "content": prompt}],
            "stream": false,
        }),
        // The Gemini generateContent body carries the model id in the path,
        // not the body; the generationConfig bounds the response.
        BoundedChatCompletionProtocol::Gemini => serde_json::json!({
            "contents": [{"role": "user", "parts": [{"text": prompt}]}],
            "generationConfig": {
                "maxOutputTokens": crate::gemini_chat::DEFAULT_MAX_OUTPUT_TOKENS,
            },
        }),
    };
    serde_json::to_vec(&body).map_err(|_| BoundedChatCompletionError {
        kind: BoundedChatCompletionErrorKind::InvalidPreparedRequest,
    })
}

/// Percent-encode any byte outside the Gemini model-id safe set and substitute
/// the result into the `{model}` route template, producing the non-streaming
/// generateContent path (`/v1beta/models/{model}:generateContent`).
fn materialize_gemini_bounded_path(template: &str, model_id: &str) -> String {
    let mut encoded = String::with_capacity(model_id.len());
    for byte in model_id.bytes() {
        match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'.' | b'_' | b'-' => {
                encoded.push(byte as char);
            }
            other => {
                encoded.push('%');
                encoded.push_str(&format!("{other:02X}"));
            }
        }
    }
    template.replace("{model}", &encoded)
}

fn parse_openai_compat_bounded_completion(
    chunks: &[Vec<u8>],
) -> Result<String, BoundedChatCompletionError> {
    let mut bytes = Vec::new();
    for chunk in chunks {
        bytes.extend_from_slice(chunk);
    }
    let value: serde_json::Value =
        serde_json::from_slice(&bytes).map_err(|_| BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })?;
    value
        .get("choices")
        .and_then(serde_json::Value::as_array)
        .and_then(|choices| choices.first())
        .and_then(|choice| choice.get("message"))
        .and_then(|message| message.get("content"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })
}

fn parse_ollama_bounded_completion(
    chunks: &[Vec<u8>],
) -> Result<String, BoundedChatCompletionError> {
    let mut bytes = Vec::new();
    for chunk in chunks {
        bytes.extend_from_slice(chunk);
    }
    let value: serde_json::Value =
        serde_json::from_slice(&bytes).map_err(|_| BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })?;
    value
        .get("message")
        .and_then(serde_json::Value::as_object)
        .and_then(|message| message.get("content"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })
}

fn parse_anthropic_bounded_completion(
    chunks: &[Vec<u8>],
) -> Result<String, BoundedChatCompletionError> {
    let mut bytes = Vec::new();
    for chunk in chunks {
        bytes.extend_from_slice(chunk);
    }
    let value: serde_json::Value =
        serde_json::from_slice(&bytes).map_err(|_| BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })?;
    value
        .get("content")
        .and_then(serde_json::Value::as_array)
        .and_then(|blocks| {
            blocks
                .iter()
                .find(|block| block.get("type").and_then(serde_json::Value::as_str) == Some("text"))
        })
        .and_then(|block| block.get("text"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })
}

fn parse_gemini_bounded_completion(
    chunks: &[Vec<u8>],
) -> Result<String, BoundedChatCompletionError> {
    let mut bytes = Vec::new();
    for chunk in chunks {
        bytes.extend_from_slice(chunk);
    }
    let value: serde_json::Value =
        serde_json::from_slice(&bytes).map_err(|_| BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })?;
    value
        .get("candidates")
        .and_then(serde_json::Value::as_array)
        .and_then(|candidates| candidates.first())
        .and_then(|candidate| candidate.get("content"))
        .and_then(|content| content.get("parts"))
        .and_then(serde_json::Value::as_array)
        .and_then(|parts| parts.first())
        .and_then(|part| part.get("text"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or(BoundedChatCompletionError {
            kind: BoundedChatCompletionErrorKind::InvalidResponse,
        })
}

fn resolve_bearer_header(
    resolver: &dyn SecretResolver,
    secret_ref: &str,
) -> Result<ResolvedSecret, SecretResolutionError> {
    let secret = resolver.resolve(secret_ref)?;
    let mut header = Vec::with_capacity(7usize.saturating_add(secret.expose_for_dispatch().len()));
    header.extend_from_slice(b"Bearer ");
    header.extend_from_slice(secret.expose_for_dispatch());
    ResolvedSecret::new(header)
}

/// Resolve a secret for a custom API-key header (`x-api-key`, `x-goog-api-key`).
/// Unlike the bearer helper, the raw secret is the header value, with no
/// scheme prefix.
fn resolve_api_key_header(
    resolver: &dyn SecretResolver,
    secret_ref: &str,
) -> Result<ResolvedSecret, SecretResolutionError> {
    resolver.resolve(secret_ref)
}

fn validate_response_bounds(chunks: &[Vec<u8>]) -> Result<(), ()> {
    if chunks.len() > MAX_RESPONSE_CHUNKS {
        return Err(());
    }
    let mut total = 0usize;
    for chunk in chunks {
        total = total.checked_add(chunk.len()).ok_or(())?;
        if total > MAX_RESPONSE_BYTES {
            return Err(());
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ollama_chat::{self, OllamaFailureClass, OllamaStreamEvent, ReplayBoundary};
    use crate::openrouter_chat::{self, OpenRouterFailureClass, OpenRouterStreamEvent};
    use router_protocol::{
        BillingAuthorityClass, CanonicalRequest, CapabilityRequirements, CatalogSourceKind,
        CredentialSourceKind, EndpointIdentity, ExecutionLocation, HarnessInstanceId,
        M1_WIRE_VERSION, ModelNamespace, PhaseKind, PhasePlacement, ProviderService, RouteIdentity,
        SessionId, ToolChoice, TransportKind, TurnId,
    };
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};

    #[cfg(unix)]
    fn secret_helper(contents: &str) -> (tempfile::TempDir, String) {
        use std::os::unix::fs::PermissionsExt;

        let directory = tempfile::Builder::new()
            .prefix(".steadymux-secret-helper-test-")
            .tempdir_in(std::env::current_dir().expect("current directory"))
            .expect("temporary directory");
        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        let helper = canonical.join("secret-helper");
        std::fs::write(&helper, contents).expect("write helper");
        std::fs::set_permissions(&helper, std::fs::Permissions::from_mode(0o700))
            .expect("owner-only executable helper");
        (directory, format!("command:{}", helper.display()))
    }

    type ObservedRequest = Arc<Mutex<Option<(String, String, bool, String)>>>;

    struct Resolver {
        calls: AtomicUsize,
        value: &'static [u8],
    }

    impl SecretResolver for Resolver {
        fn resolve(&self, _secret_ref: &str) -> Result<ResolvedSecret, SecretResolutionError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            ResolvedSecret::new(self.value)
        }
    }

    struct FakeExecutor {
        response: Mutex<Option<Result<ProviderHttpResponse, ProviderHttpError>>>,
        observed: ObservedRequest,
    }

    #[test]
    fn keychain_reference_parser_is_strict_bounded_and_content_free() {
        let parsed =
            parse_secret_reference("keychain:com.steadymux.openrouter/user+alpha@example.com")
                .expect("valid keychain reference");
        assert!(matches!(
            parsed,
            ParsedSecretReference::MacOsKeychain {
                service: "com.steadymux.openrouter",
                account: "user+alpha@example.com"
            }
        ));

        for invalid in [
            "keychain:",
            "keychain:/account",
            "keychain:service/",
            "keychain:service/account/extra",
            "keychain:service name/account",
            "keychain:service/account name",
            "keychain:service/account?query",
            "keychain:service/accouñt",
            "keychain:service/account\ncanary-secret",
        ] {
            let error = parse_secret_reference(invalid).expect_err("invalid keychain reference");
            let diagnostic = format!("{error:?} {error}");
            assert_eq!(
                diagnostic,
                "SecretResolutionError provider credential resolution failed"
            );
            assert!(!diagnostic.contains(invalid));
            assert!(!diagnostic.contains("canary-secret"));
        }

        let oversized_service = "s".repeat(MAX_KEYCHAIN_COMPONENT_BYTES + 1);
        let oversized_account = "a".repeat(MAX_KEYCHAIN_COMPONENT_BYTES + 1);
        assert!(parse_secret_reference(&format!("keychain:{oversized_service}/account")).is_err());
        assert!(parse_secret_reference(&format!("keychain:service/{oversized_account}")).is_err());
    }

    #[test]
    fn production_secret_resolver_selects_only_supported_reference_forms() {
        let resolver = ProductionSecretResolver::default();
        assert!(matches!(
            parse_secret_reference("keychain:steadymux-openrouter/default"),
            Ok(ParsedSecretReference::MacOsKeychain { .. })
        ));
        #[cfg(unix)]
        assert!(matches!(
            parse_secret_reference("command:/owner/selected/helper"),
            Ok(ParsedSecretReference::Command)
        ));
        assert!(matches!(
            parse_secret_reference("local-file-v1:0123456789abcdef0123456789abcdef"),
            Ok(ParsedSecretReference::LocalFile)
        ));
        assert_eq!(
            ProductionSecretResolver::supports_reference(
                "local-file-v1:0123456789abcdef0123456789abcdef"
            ),
            cfg!(target_os = "macos")
        );
        assert_eq!(
            ProductionSecretResolver::validate_reference(
                "local-file-v1:0123456789abcdef0123456789abcdef"
            )
            .is_ok(),
            cfg!(target_os = "macos")
        );
        for invalid in [
            "local-file-v1:",
            "local-file-v1:0123456789abcdef0123456789abcde",
            "local-file-v1:0123456789abcdef0123456789abcdeg",
            "local-file-v1:0123456789abcdef0123456789abcde/",
        ] {
            assert!(!ProductionSecretResolver::supports_reference(invalid));
            assert!(ProductionSecretResolver::validate_reference(invalid).is_err());
        }
        assert!(matches!(
            parse_secret_reference("env:STEADYMUX_FOLLOW_ON_KEY"),
            Ok(ParsedSecretReference::Environment {
                name: "STEADYMUX_FOLLOW_ON_KEY"
            })
        ));
        assert!(ProductionSecretResolver::supports_reference(
            "env:STEADYMUX_FOLLOW_ON_KEY"
        ));
        ProductionSecretResolver::validate_reference("env:STEADYMUX_FOLLOW_ON_KEY")
            .expect("validated environment reference without reading the variable");
        for invalid_env in ["env:", "env:with space", "env:has-dash"] {
            assert!(!ProductionSecretResolver::supports_reference(invalid_env));
            assert!(ProductionSecretResolver::validate_reference(invalid_env).is_err());
        }
        assert!(!format!("{resolver:?}").contains("steadymux-openrouter"));

        #[cfg(target_os = "macos")]
        {
            assert!(ProductionSecretResolver::supports_reference(
                "keychain:steadymux-openrouter/default"
            ));
            ProductionSecretResolver::validate_reference("keychain:steadymux-openrouter/default")
                .expect("validated native keychain reference without reading the keychain");
        }
    }

    #[cfg(target_os = "macos")]
    #[test]
    fn native_keychain_mutation_entrypoints_reject_before_keychain_io() {
        let valid_reference = "keychain:steadymux-openrouter/default";
        let invalid_reference = "keychain:steadymux-openrouter/default/extra";
        assert!(upsert_macos_keychain_secret(valid_reference, b"").is_err());
        assert!(upsert_macos_keychain_secret(invalid_reference, b"canary-secret").is_err());
        assert!(verify_macos_keychain_secret(invalid_reference).is_err());
        assert!(delete_macos_keychain_secret(invalid_reference).is_err());

        let diagnostic = format!(
            "{:?} {}",
            upsert_macos_keychain_secret(valid_reference, b""),
            SecretResolutionError
        );
        assert!(!diagnostic.contains("steadymux-openrouter"));
        assert!(!diagnostic.contains("canary-secret"));
    }

    #[cfg(target_os = "macos")]
    #[test]
    fn native_keychain_upsert_query_selects_local_data_protection_backend() {
        let options = local_data_protection_password_options("fixture-service", "fixture-account");
        #[allow(deprecated)]
        let fields = options
            .query
            .iter()
            .map(|(key, value)| (key.to_string(), format!("{value:?}")))
            .collect::<Vec<_>>();
        assert_eq!(
            fields
                .iter()
                .map(|(key, _)| key.as_str())
                .collect::<Vec<_>>(),
            ["class", "svce", "acct", "nleg", "sync"]
        );
        assert!(fields[3].1.contains("value = true"));
        assert!(fields[4].1.contains("value = false"));
        assert!(fields.iter().all(|(key, _)| key != "agrp"));
    }

    #[cfg(target_os = "macos")]
    #[test]
    #[cfg(not(target_os = "macos"))]
    #[test]
    fn production_secret_resolver_fails_closed_for_keychain_on_unsupported_platforms() {
        let resolver = ProductionSecretResolver::default();
        let reference = "keychain:steadymux-openrouter/default";
        assert!(!ProductionSecretResolver::supports_reference(reference));
        assert!(ProductionSecretResolver::validate_reference(reference).is_err());
        assert!(resolver.resolve(reference).is_err());
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_is_redacted_bounded_and_argument_free() {
        let (_directory, reference) = secret_helper(
            "#!/bin/sh\n[ \"$#\" -eq 0 ] || exit 7\n[ -z \"${HOME+x}\" ] || exit 8\nif IFS= read -r ignored; then exit 9; fi\nprintf 'credential-canary-value\\n'\n",
        );
        // A generous helper-execution window keeps the success path from
        // racing a load-delayed subprocess exec; the semantic under test is
        // the secret bytes and redaction, not the 5s production default.
        let resolver = CommandSecretResolver::with_timeout(Duration::from_secs(60));
        assert!(CommandSecretResolver::supports_reference(&reference));
        let secret = resolver
            .resolve(&reference)
            .expect("resolved helper secret");
        assert_eq!(secret.expose_for_dispatch(), b"credential-canary-value");
        assert!(!format!("{resolver:?} {secret:?}").contains("credential-canary-value"));

        for unsupported in [
            "keychain:ollama",
            "secret-service:ollama",
            "credential-manager:ollama",
            "command:relative-helper",
        ] {
            assert!(!CommandSecretResolver::supports_reference(unsupported));
            assert!(resolver.resolve(unsupported).is_err());
        }
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_refuses_unsafe_helpers_and_multiple_newlines() {
        use std::os::unix::fs::{PermissionsExt, symlink};

        let (directory, reference) = secret_helper("#!/bin/sh\nprintf 'secret\\n\\n'\n");
        let resolver = CommandSecretResolver::default();
        assert!(resolver.resolve(&reference).is_err());

        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        let helper = canonical.join("secret-helper");
        std::fs::set_permissions(&helper, std::fs::Permissions::from_mode(0o722))
            .expect("unsafe permissions");
        assert!(resolver.resolve(&reference).is_err());

        std::fs::set_permissions(&helper, std::fs::Permissions::from_mode(0o711))
            .expect("non-owner executable permissions");
        assert!(resolver.resolve(&reference).is_err());

        std::fs::set_permissions(&helper, std::fs::Permissions::from_mode(0o700))
            .expect("restore permissions");
        let link = canonical.join("secret-helper-link");
        symlink(&helper, &link).expect("helper symlink");
        assert!(
            resolver
                .resolve(&format!("command:{}", link.display()))
                .is_err()
        );

        let unsafe_parent = canonical.join("unsafe-parent");
        std::fs::create_dir(&unsafe_parent).expect("unsafe parent");
        std::fs::set_permissions(&unsafe_parent, std::fs::Permissions::from_mode(0o777))
            .expect("unsafe parent permissions");
        let nested = unsafe_parent.join("helper");
        std::fs::write(&nested, "#!/bin/sh\nprintf safe\n").expect("nested helper");
        std::fs::set_permissions(&nested, std::fs::Permissions::from_mode(0o700))
            .expect("nested helper permissions");
        assert!(
            resolver
                .resolve(&format!("command:{}", nested.display()))
                .is_err()
        );
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_kills_timeout_and_rejects_oversized_stdout() {
        let (directory, timeout_reference) = secret_helper("#!/bin/sh\nexec /bin/sleep 2\n");
        let resolver = CommandSecretResolver::with_timeout(Duration::from_millis(50));
        let started = Instant::now();
        assert!(resolver.resolve(&timeout_reference).is_err());
        assert!(started.elapsed() < Duration::from_secs(1));
        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        assert!(
            canonical
                .read_dir()
                .expect("stage cleanup inventory")
                .all(|entry| !entry
                    .expect("inventory entry")
                    .file_name()
                    .to_string_lossy()
                    .starts_with(".steadymux-secret-helper-"))
        );

        let (_directory, oversized_reference) = secret_helper(
            "#!/bin/sh\ni=0\nwhile [ \"$i\" -lt 17000 ]; do printf x; i=$((i + 1)); done\n",
        );
        assert!(
            CommandSecretResolver::default()
                .resolve(&oversized_reference)
                .is_err()
        );
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_executes_fd_pinned_bytes_after_source_path_swap() {
        use std::os::unix::fs::PermissionsExt;

        let (directory, reference) = secret_helper("#!/bin/sh\nprintf 'pinned-secret\\n'\n");
        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        let displaced = canonical.join("displaced-helper");
        // Generous window: the fd-pinned secret bytes (not the 5s production
        // default) are the semantic under test, so a load-delayed child exec
        // must not turn this success path into a timeout.
        let secret =
            resolve_command_secret_with_hook(&reference, Duration::from_secs(60), |source_path| {
                std::fs::rename(source_path, &displaced).map_err(|_| SecretResolutionError)?;
                std::fs::write(source_path, "#!/bin/sh\nprintf 'swapped-secret\\n'\n")
                    .map_err(|_| SecretResolutionError)?;
                std::fs::set_permissions(source_path, std::fs::Permissions::from_mode(0o700))
                    .map_err(|_| SecretResolutionError)
            })
            .expect("FD-pinned helper execution");
        assert_eq!(secret.expose_for_dispatch(), b"pinned-secret");
        assert!(!format!("{secret:?}").contains("pinned-secret"));
        assert!(
            canonical
                .read_dir()
                .expect("stage cleanup inventory")
                .all(|entry| !entry
                    .expect("inventory entry")
                    .file_name()
                    .to_string_lossy()
                    .starts_with(".steadymux-secret-helper-"))
        );
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_revalidates_staging_ancestry_after_source_open() {
        use std::os::unix::fs::PermissionsExt;

        let (directory, reference) = secret_helper("#!/bin/sh\nprintf safe\n");
        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        let result = resolve_command_secret_with_hook(&reference, Duration::from_secs(2), |_| {
            std::fs::set_permissions(&canonical, std::fs::Permissions::from_mode(0o777))
                .map_err(|_| SecretResolutionError)
        });
        std::fs::set_permissions(&canonical, std::fs::Permissions::from_mode(0o700))
            .expect("restore staging parent permissions");
        assert!(result.is_err());
        assert_eq!(canonical.read_dir().expect("inventory").count(), 1);
    }

    #[cfg(unix)]
    #[test]
    fn command_secret_resolver_kills_the_helper_process_group_on_timeout() {
        use std::os::unix::fs::PermissionsExt;

        let directory = tempfile::tempdir().expect("temporary directory");
        let canonical = std::fs::canonicalize(directory.path()).expect("canonical temp directory");
        let marker = canonical.join("grandchild-survived");
        let script = format!(
            "#!/bin/sh\n(/bin/sleep 1; /usr/bin/touch '{}') &\nwait\n",
            marker.display()
        );
        let helper = canonical.join("secret-helper");
        std::fs::write(&helper, script).expect("write helper");
        std::fs::set_permissions(&helper, std::fs::Permissions::from_mode(0o700))
            .expect("owner-only executable helper");
        let reference = format!("command:{}", helper.display());
        assert!(
            CommandSecretResolver::with_timeout(Duration::from_millis(100))
                .resolve(&reference)
                .is_err()
        );
        thread::sleep(Duration::from_millis(1_200));
        assert!(
            !marker.exists(),
            "grandchild survived process-group cleanup"
        );
    }

    impl ProviderHttpExecutor for FakeExecutor {
        fn execute<'a>(
            &'a self,
            request: PreparedRequest,
            _control: DispatchControl,
        ) -> ProviderHttpFuture<'a> {
            let observed = Arc::clone(&self.observed);
            let response = self
                .response
                .lock()
                .expect("response lock")
                .take()
                .expect("one call");
            Box::pin(async move {
                *observed.lock().expect("observed lock") = Some((
                    request.origin.clone(),
                    request.path.clone(),
                    request.authorization_for_dispatch().is_some(),
                    format!("{request:?}"),
                ));
                response
            })
        }
    }

    fn request() -> CanonicalRequest {
        CanonicalRequest {
            wire_version: M1_WIRE_VERSION.to_owned(),
            model: "router-auto".to_owned(),
            stream: true,
            harness_instance: HarnessInstanceId::parse("cli").expect("harness"),
            session_id: SessionId::parse("session-transport").expect("session"),
            turn_id: TurnId::parse("turn-transport").expect("turn"),
            input: Vec::new(),
            tools: Vec::new(),
            tool_choice: ToolChoice::None,
            structured_output: None,
            reasoning: Default::default(),
            max_output_tokens: None,
            max_spend_micro_usd: None,
            capabilities: CapabilityRequirements::default(),
        }
    }

    fn ollama_route(transport: TransportKind) -> RouteIdentity {
        let direct = transport == TransportKind::DirectHttps;
        RouteIdentity {
            phase: PhaseKind::OllamaCloud,
            placement: PhasePlacement::Builtin,
            provider: ProviderService::OllamaCloud,
            transport,
            endpoint: EndpointIdentity {
                origin: if direct {
                    "https://ollama.com"
                } else {
                    "http://127.0.0.1:11434"
                }
                .to_owned(),
                path: "/api/chat".to_owned(),
            },
            credential_source: if direct {
                CredentialSourceKind::SecretRef
            } else {
                CredentialSourceKind::OllamaDaemonSession
            },
            credential_ref: direct.then(|| "keychain:ollama-secret-canary".to_owned()),
            credential_refs: Vec::new(),
            catalog_source: CatalogSourceKind::OllamaCloudCatalog,
            catalog_ref: "catalog:ollama-cloud".to_owned(),
            billing_authority: BillingAuthorityClass::OllamaCloud,
            model_namespace: ModelNamespace::OllamaCloud,
            model_id: if direct { "glm-5.2" } else { "glm-5.2:cloud" }.to_owned(),
            execution_location: ExecutionLocation::RemoteCloud,
            account_ref: "account:ollama-cloud".to_owned(),
            capacity_domain_ref: "capacity:ollama-cloud".to_owned(),
            local_execution_proof: false,
            approved_local_inventory: Vec::new(),
            reasoning_modes: vec!["reasoning".to_owned()],
        }
    }

    fn openrouter_route() -> RouteIdentity {
        RouteIdentity {
            phase: PhaseKind::OpenrouterApi,
            placement: PhasePlacement::Builtin,
            provider: ProviderService::Openrouter,
            transport: TransportKind::DirectHttps,
            endpoint: EndpointIdentity {
                origin: "https://openrouter.ai".to_owned(),
                path: "/api/v1/chat/completions".to_owned(),
            },
            credential_source: CredentialSourceKind::SecretRef,
            credential_ref: Some("keychain:openrouter-secret-canary".to_owned()),
            credential_refs: Vec::new(),
            catalog_source: CatalogSourceKind::OpenrouterCatalog,
            catalog_ref: "catalog:openrouter".to_owned(),
            billing_authority: BillingAuthorityClass::OpenrouterMeteredApi,
            model_namespace: ModelNamespace::Openrouter,
            model_id: "anthropic/claude-sonnet-4.5".to_owned(),
            execution_location: ExecutionLocation::RemoteCloud,
            account_ref: "account:openrouter".to_owned(),
            capacity_domain_ref: "capacity:openrouter".to_owned(),
            local_execution_proof: false,
            approved_local_inventory: Vec::new(),
            reasoning_modes: Vec::new(),
        }
    }

    fn executor(
        response: Result<ProviderHttpResponse, ProviderHttpError>,
    ) -> (FakeExecutor, ObservedRequest) {
        let observed = Arc::new(Mutex::new(None));
        (
            FakeExecutor {
                response: Mutex::new(Some(response)),
                observed: Arc::clone(&observed),
            },
            observed,
        )
    }

    #[test]
    fn reqwest_executor_is_fixed_policy_and_rejects_drift_before_io() {
        let executor = ReqwestExecutor::new(
            Duration::from_secs(2),
            Duration::from_secs(3),
            Duration::from_secs(5),
        )
        .expect("client");
        assert_eq!(executor.user_agent(), FIXED_USER_AGENT);
        let debug = format!("{executor:?}");
        assert!(debug.contains("disabled"));
        assert!(debug.contains("rustls"));

        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"unused",
        };
        let spec = ollama_chat::build_request(
            &ollama_route(TransportKind::OllamaSignedInRelay),
            &request(),
        )
        .expect("spec");
        let mut prepared = prepare_ollama(spec, &resolver).expect("prepared");
        validate_prepared_request(&prepared).expect("exact local policy");
        assert_eq!(prepared.origin, "http://127.0.0.1:11434");
        assert!(!prepared.network.allow_dns);
        assert!(prepared.authorization_for_dispatch().is_none());
        prepared.network.allow_dns = true;
        assert_eq!(
            validate_prepared_request(&prepared),
            Err(ProviderHttpError {
                kind: ProviderHttpErrorKind::InvalidPreparedRequest,
                body_write: BodyWriteState::NotWritten,
            })
        );
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn bearer_header_is_marked_sensitive_before_reqwest_receives_it() {
        let value = sensitive_header_value(b"Bearer credential-canary-value").expect("header");
        assert!(value.is_sensitive());
        assert!(!format!("{value:?}").contains("credential-canary-value"));
    }

    #[tokio::test]
    async fn localhost_never_resolves_auth_and_parses_fragmented_ollama() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"unused",
        };
        let spec = ollama_chat::build_request(
            &ollama_route(TransportKind::OllamaSignedInRelay),
            &request(),
        )
        .expect("spec");
        let response = ProviderHttpResponse {
            status: 200,
            content_type: Some("application/x-ndjson".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![
                b"{\"message\":{\"content\":\"ok\"},\"do".to_vec(),
                b"ne\":false}\r\n{\"done\":true}\n".to_vec(),
            ],
        };
        let (executor, observed) = executor(Ok(response));
        let result = dispatch_ollama(
            spec,
            &AttemptId::parse("attempt-local").expect("attempt"),
            &resolver,
            &executor,
            DispatchControl::default(),
        )
        .await
        .expect("dispatch");
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
        assert!(matches!(
            result.events[0],
            NormalizedProviderEvent::Ollama(OllamaStreamEvent::Content { .. })
        ));
        let observed = observed.lock().expect("observed");
        assert_eq!(observed.as_ref().map(|value| value.2), Some(false));
    }

    #[tokio::test]
    async fn remote_secret_is_resolved_jit_and_redacted_from_debug_and_errors() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"credential-canary-value",
        };
        let spec =
            ollama_chat::build_request(&ollama_route(TransportKind::DirectHttps), &request())
                .expect("spec");
        let (executor, observed) = executor(Ok(ProviderHttpResponse {
            status: 401,
            content_type: None,
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: Vec::new(),
        }));
        let error = dispatch_ollama(
            spec,
            &AttemptId::parse("attempt-remote").expect("attempt"),
            &resolver,
            &executor,
            DispatchControl::default(),
        )
        .await
        .expect_err("401");
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 1);
        assert!(matches!(
            error.kind,
            DispatchFailureKind::OllamaHttp(OllamaFailure {
                class: OllamaFailureClass::Authentication,
                ..
            })
        ));
        let debug = observed
            .lock()
            .expect("observed")
            .as_ref()
            .expect("request")
            .3
            .clone();
        assert!(!debug.contains("credential-canary-value"));
        assert!(!format!("{error:?}").contains("credential-canary-value"));
    }

    #[tokio::test]
    async fn openrouter_sse_is_normalized_with_usage_exactly_once() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"openrouter-canary",
        };
        let spec = openrouter_chat::build_request(&openrouter_route(), &request()).expect("spec");
        let transcript = b"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\ndata: {\"choices\":[],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1,\"total_tokens\":3}}\n\ndata: [DONE]\n\n";
        let response = ProviderHttpResponse {
            status: 200,
            content_type: Some("text/event-stream; charset=utf-8".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: transcript.iter().map(|byte| vec![*byte]).collect(),
        };
        let (executor, _) = executor(Ok(response));
        let result = dispatch_openrouter(
            spec,
            &AttemptId::parse("attempt-openrouter").expect("attempt"),
            &resolver,
            &executor,
            DispatchControl::default(),
        )
        .await
        .expect("dispatch");
        assert_eq!(
            result
                .events
                .iter()
                .filter(|event| matches!(
                    event,
                    NormalizedProviderEvent::OpenRouter(OpenRouterStreamEvent::Usage(_))
                ))
                .count(),
            1
        );
        assert!(matches!(
            result.events.last(),
            Some(NormalizedProviderEvent::OpenRouter(
                OpenRouterStreamEvent::Done
            ))
        ));
        assert_eq!(
            result.replay,
            ReplayDecision::Denied(ReplayBoundary::PublicContent)
        );
    }

    #[tokio::test]
    async fn cancellation_before_dispatch_is_safe_and_written_timeout_is_ambiguous() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"credential-canary",
        };
        let cancelled_spec =
            openrouter_chat::build_request(&openrouter_route(), &request()).expect("spec");
        let (unused_executor, _) = executor(Ok(ProviderHttpResponse {
            status: 200,
            content_type: Some("text/event-stream".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: Vec::new(),
        }));
        let cancelled = dispatch_openrouter(
            cancelled_spec,
            &AttemptId::parse("attempt-cancel").expect("attempt"),
            &resolver,
            &unused_executor,
            DispatchControl {
                cancelled_before_dispatch: true,
                forwarded_tool_result: false,
                cancellation: None,
            },
        )
        .await
        .expect_err("cancelled");
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
        assert_eq!(cancelled.replay, ReplayDecision::Allowed);

        let timeout_spec =
            openrouter_chat::build_request(&openrouter_route(), &request()).expect("spec");
        let (executor, _) = executor(Err(ProviderHttpError {
            kind: ProviderHttpErrorKind::Timeout,
            body_write: BodyWriteState::Written,
        }));
        let timeout = dispatch_openrouter(
            timeout_spec,
            &AttemptId::parse("attempt-timeout").expect("attempt"),
            &resolver,
            &executor,
            DispatchControl::default(),
        )
        .await
        .expect_err("timeout");
        assert!(matches!(
            timeout.kind,
            DispatchFailureKind::OpenRouterHttp(OpenRouterFailure {
                class: OpenRouterFailureClass::AmbiguousDispatch,
                ..
            })
        ));
        assert_eq!(
            timeout.replay,
            ReplayDecision::Denied(ReplayBoundary::AmbiguousDispatch)
        );
    }

    fn bounded_spec(protocol: BoundedChatCompletionProtocol) -> BoundedChatCompletionSpec {
        BoundedChatCompletionSpec {
            origin: "https://provider.example".to_owned(),
            path: "/v1/chat/completions".to_owned(),
            model: "provider/categorizer".to_owned(),
            prompt: "categorize these models".to_owned(),
            bearer_secret: Some("env:STEADYMUX_CATEGORIZER_KEY".to_owned()),
            protocol,
            max_response_bytes: 256 * 1024,
        }
    }

    fn openai_compat_response(text: &str) -> ProviderHttpResponse {
        ProviderHttpResponse {
            status: 200,
            content_type: Some("application/json".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![
                serde_json::json!({
                    "choices": [{"message": {"content": text}}],
                    "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
                })
                .to_string()
                .into_bytes(),
            ],
        }
    }

    fn ollama_response(text: &str) -> ProviderHttpResponse {
        ProviderHttpResponse {
            status: 200,
            content_type: Some("application/json".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![
                serde_json::json!({
                    "model": "provider/categorizer",
                    "message": {"role": "assistant", "content": text},
                    "done": true,
                })
                .to_string()
                .into_bytes(),
            ],
        }
    }

    #[tokio::test]
    async fn bounded_chat_completion_openai_compat_returns_text() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let (executor, observed) = executor(Ok(openai_compat_response("categorization-evidence")));
        let text = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat),
            &resolver,
            &executor,
        )
        .await
        .expect("completion");
        assert_eq!(text, "categorization-evidence");
        let (origin, path, authorized, debug) =
            observed.lock().expect("observed").clone().expect("call");
        assert_eq!(origin, "https://provider.example");
        assert_eq!(path, "/v1/chat/completions");
        assert!(authorized);
        assert!(!debug.contains("secret-value"));
        assert!(!debug.contains("categorize these models"));
    }

    #[tokio::test]
    async fn bounded_chat_completion_ollama_returns_text() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let (executor, _) = executor(Ok(ollama_response("ollama-evidence")));
        let text = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::Ollama),
            &resolver,
            &executor,
        )
        .await
        .expect("completion");
        assert_eq!(text, "ollama-evidence");
    }

    #[tokio::test]
    async fn bounded_chat_completion_ollama_without_secret_is_allowed() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"unused",
        };
        let mut spec = bounded_spec(BoundedChatCompletionProtocol::Ollama);
        spec.bearer_secret = None;
        let (executor, observed) = executor(Ok(ollama_response("local-evidence")));
        let text = dispatch_bounded_chat_completion(spec, &resolver, &executor)
            .await
            .expect("completion");
        assert_eq!(text, "local-evidence");
        let (_, _, authorized, _) = observed.lock().expect("observed").clone().expect("call");
        assert!(!authorized);
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn bounded_chat_completion_openai_compat_requires_secret() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"unused",
        };
        let mut spec = bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat);
        spec.bearer_secret = None;
        let (executor, _) = executor(Ok(openai_compat_response("x")));
        let error = dispatch_bounded_chat_completion(spec, &resolver, &executor)
            .await
            .expect_err("secret required");
        assert_eq!(error.kind, BoundedChatCompletionErrorKind::SecretResolution);
        assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn bounded_chat_completion_rejects_oversized_response() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let mut spec = bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat);
        spec.max_response_bytes = 16;
        let (executor, _) = executor(Ok(openai_compat_response("this text is far too long")));
        let error = dispatch_bounded_chat_completion(spec, &resolver, &executor)
            .await
            .expect_err("oversized");
        assert_eq!(error.kind, BoundedChatCompletionErrorKind::ResponseTooLarge);
    }

    #[tokio::test]
    async fn bounded_chat_completion_rejects_non_200() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let mut response = openai_compat_response("x");
        response.status = 429;
        let (executor, _) = executor(Ok(response));
        let error = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat),
            &resolver,
            &executor,
        )
        .await
        .expect_err("rate limited");
        assert_eq!(
            error.kind,
            BoundedChatCompletionErrorKind::Http { status: 429 }
        );
    }

    #[tokio::test]
    async fn bounded_chat_completion_rejects_invalid_response() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let response = ProviderHttpResponse {
            status: 200,
            content_type: Some("application/json".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![b"not-json".to_vec()],
        };
        let (executor, _) = executor(Ok(response));
        let error = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat),
            &resolver,
            &executor,
        )
        .await
        .expect_err("invalid response");
        assert_eq!(error.kind, BoundedChatCompletionErrorKind::InvalidResponse);
    }

    #[tokio::test]
    async fn bounded_chat_completion_rejects_empty_completion() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let (executor, _) = executor(Ok(openai_compat_response("")));
        let error = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat),
            &resolver,
            &executor,
        )
        .await
        .expect_err("empty completion");
        assert_eq!(error.kind, BoundedChatCompletionErrorKind::EmptyCompletion);
    }

    #[tokio::test]
    async fn bounded_chat_completion_anthropic_returns_text_with_native_headers() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let response = ProviderHttpResponse {
            status: 200,
            content_type: Some("application/json".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![
                serde_json::json!({
                    "content": [{"type": "text", "text": "anthropic-evidence"}],
                    "model": "claude-sonnet-5",
                })
                .to_string()
                .into_bytes(),
            ],
        };
        let (executor, observed) = executor(Ok(response));
        let mut spec = bounded_spec(BoundedChatCompletionProtocol::Anthropic);
        spec.origin = router_protocol::ANTHROPIC_ORIGIN.to_owned();
        spec.path = router_protocol::ANTHROPIC_CHAT_PATH.to_owned();
        spec.model = "claude-sonnet-5".to_owned();
        let text = dispatch_bounded_chat_completion(spec, &resolver, &executor)
            .await
            .expect("completion");
        assert_eq!(text, "anthropic-evidence");
        let (origin, path, authorized, debug) =
            observed.lock().expect("observed").clone().expect("call");
        assert_eq!(origin, router_protocol::ANTHROPIC_ORIGIN);
        assert_eq!(path, router_protocol::ANTHROPIC_CHAT_PATH);
        assert!(authorized);
        assert!(
            debug.contains(router_protocol::ANTHROPIC_API_KEY_HEADER),
            "the bounded Anthropic request must carry the x-api-key header"
        );
        assert!(
            debug.contains(router_protocol::ANTHROPIC_VERSION_HEADER),
            "the bounded Anthropic request must carry the pinned anthropic-version header"
        );
        assert!(!debug.contains("secret-value"));
        assert!(!debug.contains("anthropic-evidence"));
    }

    #[tokio::test]
    async fn bounded_chat_completion_gemini_returns_text_and_materializes_model_path() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let response = ProviderHttpResponse {
            status: 200,
            content_type: Some("application/json".to_owned()),
            retry_after: None,
            redirected: false,
            request_id: None,
            chunks: vec![
                serde_json::json!({
                    "candidates": [{
                        "content": {"parts": [{"text": "gemini-evidence"}], "role": "model"},
                        "finishReason": "STOP",
                    }],
                })
                .to_string()
                .into_bytes(),
            ],
        };
        let (executor, observed) = executor(Ok(response));
        let mut spec = bounded_spec(BoundedChatCompletionProtocol::Gemini);
        spec.origin = router_protocol::GEMINI_ORIGIN.to_owned();
        spec.path = router_protocol::GEMINI_GENERATE_CONTENT_PATH.to_owned();
        spec.model = "gemini-3.6-flash".to_owned();
        let text = dispatch_bounded_chat_completion(spec, &resolver, &executor)
            .await
            .expect("completion");
        assert_eq!(text, "gemini-evidence");
        let (origin, path, authorized, debug) =
            observed.lock().expect("observed").clone().expect("call");
        assert_eq!(origin, router_protocol::GEMINI_ORIGIN);
        assert_eq!(path, "/v1beta/models/gemini-3.6-flash:generateContent");
        assert!(authorized);
        assert!(
            debug.contains(router_protocol::GEMINI_API_KEY_HEADER),
            "the bounded Gemini request must carry the x-goog-api-key header"
        );
        assert!(!debug.contains("secret-value"));
        assert!(!debug.contains("gemini-evidence"));
    }

    #[tokio::test]
    async fn bounded_chat_completion_anthropic_and_gemini_require_secret() {
        for protocol in [
            BoundedChatCompletionProtocol::Anthropic,
            BoundedChatCompletionProtocol::Gemini,
        ] {
            let resolver = Resolver {
                calls: AtomicUsize::new(0),
                value: b"unused",
            };
            let mut spec = bounded_spec(protocol);
            spec.bearer_secret = None;
            let (executor, _) = executor(Ok(openai_compat_response("x")));
            let error = dispatch_bounded_chat_completion(spec, &resolver, &executor)
                .await
                .expect_err("secret required");
            assert_eq!(error.kind, BoundedChatCompletionErrorKind::SecretResolution);
            assert_eq!(resolver.calls.load(Ordering::SeqCst), 0);
        }
    }

    #[tokio::test]
    async fn bounded_chat_completion_maps_executor_failures() {
        let resolver = Resolver {
            calls: AtomicUsize::new(0),
            value: b"secret-value",
        };
        let (executor, _) = executor(Err(ProviderHttpError {
            kind: ProviderHttpErrorKind::Timeout,
            body_write: BodyWriteState::NotWritten,
        }));
        let error = dispatch_bounded_chat_completion(
            bounded_spec(BoundedChatCompletionProtocol::OpenAiCompat),
            &resolver,
            &executor,
        )
        .await
        .expect_err("timeout");
        assert_eq!(error.kind, BoundedChatCompletionErrorKind::Timeout);
    }

    fn bounded_request(
        origin: &str,
        path: &str,
        protocol: ProviderWireProtocol,
        local: bool,
        authorization: Option<ResolvedSecret>,
        auth_header_name: Option<&'static str>,
        version_header: Option<(&'static str, &'static str)>,
    ) -> PreparedRequest {
        PreparedRequest {
            method: "POST",
            origin: origin.to_owned(),
            path: path.to_owned(),
            content_type: "application/json",
            accept: "application/json",
            network: NetworkPolicy {
                allow_proxy: false,
                allow_dns: !local,
                follow_redirects: false,
                require_tls: !local,
            },
            protocol,
            body: b"{\"model\":\"categorizer\"}".to_vec(),
            authorization,
            auth_header_name,
            version_header,
            bounded: true,
        }
    }

    #[test]
    fn bounded_requests_pass_validate_prepared_request() {
        // Local Ollama: loopback, no DNS, no TLS, no authorization.
        validate_prepared_request(&bounded_request(
            "http://127.0.0.1:11434",
            "/api/chat",
            ProviderWireProtocol::OllamaNdjson,
            true,
            None,
            None,
            None,
        ))
        .expect("bounded localhost passes");
        // Ollama cloud: DNS and TLS required, authorization required.
        validate_prepared_request(&bounded_request(
            "https://ollama.com",
            "/api/chat",
            ProviderWireProtocol::OllamaNdjson,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        ))
        .expect("bounded ollama cloud passes");
        // OpenRouter origin with the OpenAiCompat wire protocol (the bounded
        // categorization dispatch shape for the OpenAI and OpenRouter phases).
        validate_prepared_request(&bounded_request(
            "https://openrouter.ai",
            "/api/v1/chat/completions",
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        ))
        .expect("bounded openrouter passes");
        // A bounded OpenAI-compatible origin.
        validate_prepared_request(&bounded_request(
            router_protocol::DASHSCOPE_ORIGIN,
            router_protocol::DASHSCOPE_CHAT_PATH,
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        ))
        .expect("bounded openai compat passes");
        // The DeepSeek direct-API origin (PRD-001y S1), also OpenAI-compatible.
        validate_prepared_request(&bounded_request(
            router_protocol::DEEPSEEK_ORIGIN,
            router_protocol::DEEPSEEK_CHAT_PATH,
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        ))
        .expect("bounded deepseek passes");
        // The Qwen token-plan origin (PRD-001y S3 requalification): the
        // operator's Qwen account authenticates only against the token-plan
        // host, and the follow-on relay must be allowed to reach it.
        validate_prepared_request(&bounded_request(
            router_protocol::DASHSCOPE_TOKEN_PLAN_ORIGIN,
            router_protocol::DASHSCOPE_CHAT_PATH,
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        ))
        .expect("bounded qwen token-plan passes");
        // The Anthropic bounded shape (PRD-001y S2): `x-api-key` plus the
        // pinned `anthropic-version` header.
        validate_prepared_request(&bounded_request(
            router_protocol::ANTHROPIC_ORIGIN,
            router_protocol::ANTHROPIC_CHAT_PATH,
            ProviderWireProtocol::AnthropicSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            Some(router_protocol::ANTHROPIC_API_KEY_HEADER),
            Some((
                router_protocol::ANTHROPIC_VERSION_HEADER,
                router_protocol::ANTHROPIC_VERSION,
            )),
        ))
        .expect("bounded anthropic passes");
        // The Gemini bounded shape (PRD-001y S2): `x-goog-api-key`, no version
        // header, and the model-materialized generateContent path.
        validate_prepared_request(&bounded_request(
            router_protocol::GEMINI_ORIGIN,
            "/v1beta/models/gemini-3.6-flash:generateContent",
            ProviderWireProtocol::GeminiSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            Some(router_protocol::GEMINI_API_KEY_HEADER),
            None,
        ))
        .expect("bounded gemini passes");
    }

    #[test]
    fn non_bounded_openrouter_request_requires_streaming_accept_and_protocol() {
        // The same OpenRouter shape with `bounded: false` must be rejected: the
        // streaming allowlist requires the OpenRouterSse wire protocol and a
        // text/event-stream accept header.
        let mut request = bounded_request(
            "https://openrouter.ai",
            "/api/v1/chat/completions",
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        );
        request.bounded = false;
        assert!(validate_prepared_request(&request).is_err());
    }

    #[test]
    fn streaming_deepseek_dispatch_passes_allowlist() {
        // PRD-001y S1 wire shape: the DeepSeek direct path is a streaming
        // OpenAI-compatible SSE dispatch against
        // `https://api.deepseek.com/chat/completions` with bearer auth. The
        // bounded helper builds the shared field shape; flip the two fields the
        // streaming allowlist predicate requires (unbounded, text/event-stream
        // accept) to prove the route is eligible for transport validation.
        let mut request = bounded_request(
            router_protocol::DEEPSEEK_ORIGIN,
            router_protocol::DEEPSEEK_CHAT_PATH,
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        );
        request.bounded = false;
        request.accept = "text/event-stream";
        validate_prepared_request(&request).expect("deepseek streaming dispatch passes");

        // The Qwen token-plan host (PRD-001y S3): the operator's live Qwen key
        // resolves only against `token-plan.ap-southeast-1.maas.aliyuncs.com`,
        // never the generic dashscope-intl origin, so the streaming OpenAI-
        // compatible dispatch must pass transport validation for that origin.
        let mut qwen = bounded_request(
            router_protocol::DASHSCOPE_TOKEN_PLAN_ORIGIN,
            router_protocol::DASHSCOPE_CHAT_PATH,
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        );
        qwen.accept = "text/event-stream";
        validate_prepared_request(&qwen).expect("qwen token-plan streaming dispatch passes");
    }

    #[test]
    fn openai_compat_request_to_unlisted_origin_fails_closed() {
        // An OpenAI-compatible streaming shape against an origin that is NOT in
        // the fail-closed allowlist must be rejected before any wire IO, even
        // though the protocol, authorization, DNS, and TLS predicates all hold.
        let mut request = bounded_request(
            "https://not-allowlisted.example.com",
            "/chat/completions",
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        );
        request.bounded = false;
        request.accept = "text/event-stream";
        assert_eq!(
            validate_prepared_request(&request),
            Err(ProviderHttpError {
                kind: ProviderHttpErrorKind::InvalidPreparedRequest,
                body_write: BodyWriteState::NotWritten,
            })
        );
    }

    #[test]
    fn qwen_token_plan_admit_rejects_non_chat_path() {
        // The PRD-001y S3 admit is scoped to exactly
        // `(DASHSCOPE_TOKEN_PLAN_ORIGIN, DASHSCOPE_CHAT_PATH)`. A request to
        // the token-plan host with any other path must fail closed even though
        // the protocol, DNS, TLS, and authorization predicates all hold.
        let mut request = bounded_request(
            router_protocol::DASHSCOPE_TOKEN_PLAN_ORIGIN,
            "/v1/chat/completions",
            ProviderWireProtocol::OpenAiCompatSse,
            false,
            Some(ResolvedSecret::new(b"secret-value").expect("valid secret")),
            None,
            None,
        );
        request.bounded = false;
        request.accept = "text/event-stream";
        assert_eq!(
            validate_prepared_request(&request),
            Err(ProviderHttpError {
                kind: ProviderHttpErrorKind::InvalidPreparedRequest,
                body_write: BodyWriteState::NotWritten,
            })
        );
    }
}