everruns_contracts/

runtime_provider.rs

//! Runtime providers: service identity, endpoint, authentication, and wire driver.
//!
//! A [`ChatDriver`] implements a wire
//! protocol. A `Provider` is a configured service that speaks that protocol.
//! Keeping credentials and endpoints here lets one driver serve any number of
//! services without adding vendor branches to the runtime.

use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;

use async_trait::async_trait;
use futures::StreamExt;
use serde::{Deserialize, Serialize};

use crate::driver_registry::{BoxedChatDriver, ChatDriver};
use crate::error::Result;

/// Open, normalized identity used by model specifications to select a provider.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ProviderKey(String);

impl ProviderKey {
    pub fn new(id: impl AsRef<str>) -> Self {
        Self(id.as_ref().trim().to_ascii_lowercase())
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for ProviderKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

impl From<&str> for ProviderKey {
    fn from(value: &str) -> Self {
        Self::new(value)
    }
}

impl From<String> for ProviderKey {
    fn from(value: String) -> Self {
        Self::new(value)
    }
}

impl Serialize for ProviderKey {
    fn serialize<S: serde::Serializer>(
        &self,
        serializer: S,
    ) -> std::result::Result<S::Ok, S::Error> {
        serializer.serialize_str(self.as_str())
    }
}

impl<'de> Deserialize<'de> for ProviderKey {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        String::deserialize(deserializer).map(Self::new)
    }
}

/// Immutable request material available to an authentication implementation.
///
/// `body` is the exact serialized payload. This makes the contract suitable
/// for request signatures such as AWS SigV4 as well as ordinary header auth.
pub struct ProviderAuthRequest<'a> {
    pub method: &'a str,
    pub url: &'a str,
    pub headers: &'a [(String, String)],
    pub body: &'a [u8],
}

/// Resolves authentication for each outbound provider request.
#[async_trait]
pub trait ProviderAuth: Send + Sync {
    async fn headers(&self, request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>>;
    fn as_any(&self) -> &dyn std::any::Any;
}

/// Static `Authorization: Bearer …` authentication.
pub struct BearerAuth {
    key: String,
}

impl BearerAuth {
    pub fn new(key: impl Into<String>) -> Self {
        Self { key: key.into() }
    }
}

#[async_trait]
impl ProviderAuth for BearerAuth {
    async fn headers(&self, _request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>> {
        Ok(vec![(
            "authorization".to_string(),
            format!("Bearer {}", self.key),
        )])
    }
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

/// Static authentication carried in one named header.
pub struct StaticHeaderAuth {
    name: String,
    value: String,
}

impl StaticHeaderAuth {
    pub fn new(name: impl Into<String>, value: impl Into<String>) -> Self {
        Self {
            name: name.into().to_ascii_lowercase(),
            value: value.into(),
        }
    }
}

#[async_trait]
impl ProviderAuth for StaticHeaderAuth {
    async fn headers(&self, _request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>> {
        Ok(vec![(self.name.clone(), self.value.clone())])
    }
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

/// Endpoint and authentication policy handed to a wire driver.
///
/// This runtime value is intentionally not serializable. Its `Debug` output
/// exposes only header names and whether authentication is configured.
#[derive(Clone, Default)]
pub struct ProviderEndpoint {
    base_url: Option<String>,
    headers: Vec<(String, String)>,
    auth: Option<Arc<dyn ProviderAuth>>,
}

impl ProviderEndpoint {
    /// Endpoint for a protocol client outside a provider assembly.
    pub fn from_parts(base_url: impl Into<String>, auth: impl ProviderAuth + 'static) -> Self {
        Self {
            base_url: Some(base_url.into().trim_end_matches('/').to_string()),
            headers: Vec::new(),
            auth: Some(Arc::new(auth)),
        }
    }

    pub fn base_url(&self) -> Option<&str> {
        self.base_url.as_deref()
    }

    pub fn url(&self, path: &str) -> Option<String> {
        self.base_url.as_ref().map(|base| {
            let base = base.trim_end_matches('/');
            if path.is_empty() {
                return base.to_string();
            }
            if let Ok(mut url) = url::Url::parse(base) {
                // Operations can carry protocol query parameters (Gemini's
                // alt=sse). Keep them out of set_path, which escapes ? and #.
                let (path, fragment) = path
                    .split_once('#')
                    .map_or((path, None), |(p, f)| (p, Some(f)));
                let (path, query) = path
                    .split_once('?')
                    .map_or((path, None), |(p, q)| (p, Some(q)));
                let suffix = format!("/{}", path.trim_start_matches('/'));
                // Match actual path segments, never a hostname or suffix of
                // another segment; preserve query parameters while appending.
                let existing = url.path().trim_end_matches('/');
                if !existing.ends_with(&suffix) {
                    url.set_path(&format!("{existing}{suffix}"));
                }
                if let Some(query) = query {
                    let combined = match url.query() {
                        Some(existing) if !existing.is_empty() && !query.is_empty() => {
                            format!("{existing}&{query}")
                        }
                        Some(existing) if query.is_empty() => existing.to_string(),
                        _ => query.to_string(),
                    };
                    url.set_query(Some(&combined));
                }
                if let Some(fragment) = fragment {
                    url.set_fragment(Some(fragment));
                }
                url.to_string()
            } else {
                format!("{base}/{}", path.trim_start_matches('/'))
            }
        })
    }

    pub async fn resolve(
        &self,
        method: &str,
        url: impl Into<String>,
        body: &[u8],
    ) -> Result<ResolvedProviderRequest> {
        let url = url.into();
        let mut headers = self.headers.clone();
        if let Some(auth) = &self.auth {
            let auth_headers = auth
                .headers(ProviderAuthRequest {
                    method,
                    url: &url,
                    headers: &headers,
                    body,
                })
                .await?;
            for (name, value) in auth_headers {
                headers.retain(|(existing, _)| !existing.eq_ignore_ascii_case(&name));
                headers.push((name.to_ascii_lowercase(), value));
            }
        }
        Ok(ResolvedProviderRequest { url, headers })
    }

    /// Access a provider-owned authentication implementation needed by a
    /// protocol whose signing stack cannot be represented as header strings.
    pub fn auth<T: ProviderAuth + 'static>(&self) -> Option<&T> {
        self.auth.as_deref()?.as_any().downcast_ref()
    }
}

impl fmt::Debug for ProviderEndpoint {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ProviderEndpoint")
            .field("base_url", &self.base_url.as_ref().map(|_| "<configured>"))
            .field("auth", &self.auth.as_ref().map(|_| "<configured>"))
            .field(
                "headers",
                &self
                    .headers
                    .iter()
                    .map(|(name, _)| name.as_str())
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

/// Fully resolved outbound request metadata.
#[derive(Clone)]
pub struct ResolvedProviderRequest {
    pub url: String,
    pub headers: Vec<(String, String)>,
}

impl fmt::Debug for ResolvedProviderRequest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ResolvedProviderRequest")
            .field("url", &redacted_url(&self.url))
            .field(
                "headers",
                &self
                    .headers
                    .iter()
                    .map(|(name, _)| name.as_str())
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

fn redacted_url(value: &str) -> String {
    let Ok(mut url) = url::Url::parse(value) else {
        return "<configured>".to_string();
    };
    let _ = url.set_username("");
    let _ = url.set_password(None);
    url.set_query(None);
    url.set_fragment(None);
    url.to_string()
}

/// Runtime service assembly over one reusable wire-protocol driver.
#[derive(Clone)]
pub struct RuntimeProvider {
    id: ProviderKey,
    driver: Option<Arc<dyn ChatDriver>>,
    decisions: Option<Arc<dyn crate::decision_driver::DecisionDriver>>,
    embeddings: Option<Arc<dyn crate::driver_registry::EmbeddingsDriver>>,
    endpoint: ProviderEndpoint,
    // Which driver kind this assembly speaks, for lookups that are about the
    // vendor rather than this configured instance (model profiles, catalog
    // enrichment). Optional because the runtime identity and the driver kind
    // are independent: a provider may be keyed `"my-openai"`. Unset falls back
    // to the key, which is the conventional case.
    driver_id: Option<crate::provider::DriverId>,
}

/// Public application-facing name for a runtime provider.
pub type Provider = RuntimeProvider;

impl RuntimeProvider {
    pub fn new(id: impl Into<ProviderKey>, driver: impl ChatDriver + 'static) -> Self {
        Self::from_driver(id, Arc::new(driver))
    }

    pub fn from_driver(id: impl Into<ProviderKey>, driver: Arc<dyn ChatDriver>) -> Self {
        Self {
            id: id.into(),
            driver: Some(driver),
            decisions: None,
            embeddings: None,
            endpoint: ProviderEndpoint::default(),
            driver_id: None,
        }
    }

    pub fn base_url(mut self, url: impl Into<String>) -> Self {
        self.endpoint.base_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    pub fn auth(mut self, auth: impl ProviderAuth + 'static) -> Self {
        self.endpoint.auth = Some(Arc::new(auth));
        self
    }

    pub fn auth_arc(mut self, auth: Arc<dyn ProviderAuth>) -> Self {
        self.endpoint.auth = Some(auth);
        self
    }

    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.endpoint
            .headers
            .push((name.into().to_ascii_lowercase(), value.into()));
        self
    }

    /// Declare the driver kind this provider speaks, when it differs from the
    /// runtime key. Used for model-profile lookups, never for routing.
    pub fn with_driver_id(mut self, driver_id: crate::provider::DriverId) -> Self {
        self.driver_id = Some(driver_id);
        self
    }

    pub fn id(&self) -> &ProviderKey {
        &self.id
    }

    /// The driver kind this provider speaks, falling back to the runtime key
    /// when nothing declared one.
    pub fn driver_id(&self) -> crate::provider::DriverId {
        self.driver_id
            .clone()
            .unwrap_or_else(|| crate::provider::DriverId::external(self.id.as_str()))
    }

    pub fn driver(&self) -> Result<&Arc<dyn ChatDriver>> {
        self.driver
            .as_ref()
            .ok_or_else(|| self.unsupported_service("chat"))
    }

    fn unsupported_service(&self, service: &str) -> crate::error::AgentLoopError {
        crate::error::AgentLoopError::Configuration(format!(
            "Provider '{}' does not implement the {service} service",
            self.id
        ))
    }

    /// Construct a provider whose services are supplied by typed builders.
    pub fn services(id: impl Into<ProviderKey>) -> Self {
        Self {
            id: id.into(),
            driver: None,
            decisions: None,
            embeddings: None,
            endpoint: ProviderEndpoint::default(),
            driver_id: None,
        }
    }

    pub fn with_decisions(
        mut self,
        driver: impl crate::decision_driver::DecisionDriver + 'static,
    ) -> Self {
        self.decisions = Some(Arc::new(driver));
        self
    }

    pub fn with_embeddings(
        mut self,
        driver: impl crate::driver_registry::EmbeddingsDriver + 'static,
    ) -> Self {
        self.embeddings = Some(Arc::new(driver));
        self
    }

    pub fn supports_service(&self, service: crate::ServiceKind) -> bool {
        match service {
            crate::ServiceKind::Chat => self.driver.is_some(),
            crate::ServiceKind::Decisions => self.decisions.is_some(),
            crate::ServiceKind::Embeddings => self.embeddings.is_some(),
            _ => false,
        }
    }

    pub async fn evaluate_decisions(
        &self,
        request: crate::decisions::DecisionRequest,
    ) -> Result<crate::decisions::DecisionOutcome> {
        if request.provider.as_ref().is_some_and(|key| key != &self.id) {
            return Err(crate::error::AgentLoopError::Configuration(
                "A bound provider cannot select another account".into(),
            ));
        }
        let driver = self
            .decisions
            .as_ref()
            .ok_or_else(|| self.unsupported_service("decisions"))?;
        driver.capabilities().check(driver.id(), &request)?;
        driver
            .evaluate(&self.endpoint, request)
            .await
            .map_err(|error| error.with_provider(self.id.as_str()))
    }

    pub async fn embed(
        &self,
        request: crate::driver_registry::EmbedRequest,
    ) -> std::result::Result<
        crate::driver_registry::EmbedResponse,
        crate::driver_registry::EmbeddingsDriverError,
    > {
        let driver = self.embeddings.as_ref().ok_or_else(|| {
            crate::driver_registry::EmbeddingsDriverError::Provider(
                self.unsupported_service("embeddings").to_string(),
            )
        })?;
        driver.embed(&self.endpoint, request).await
    }

    pub fn endpoint(&self) -> &ProviderEndpoint {
        &self.endpoint
    }

    /// A requested response format this driver cannot enforce is a
    /// configuration error, never a silently unconstrained reply (EVE-1116).
    fn check_response_format(&self, config: &crate::driver_registry::LlmCallConfig) -> Result<()> {
        if config.response_format.is_some()
            && !self.driver()?.supports_response_format(&config.model)
        {
            return Err(crate::error::AgentLoopError::Configuration(format!(
                "Structured output (response_format) is not supported by provider '{}' for model '{}'",
                self.id, config.model
            )));
        }
        Ok(())
    }

    pub async fn chat_completion_stream(
        &self,
        messages: Vec<crate::driver_registry::Message>,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Result<crate::driver_registry::LlmResponseStream> {
        self.check_response_format(config)?;
        let id = self.id.to_string();
        let limits = config.limits;
        // Establishing the stream is itself a round trip that can hang, so it
        // runs inside the budget, and what it spends is charged against what
        // the stream then gets.
        let (stream, spent) = crate::turn_collector::connect_within(
            &limits,
            self.driver()?
                .chat_completion_stream(&self.endpoint, messages, config),
        )
        .await
        .map_err(|error| error.with_provider(&id))?;
        let stream: crate::driver_registry::LlmResponseStream =
            Box::pin(stream.map(move |result| result.map_err(|error| error.with_provider(&id))));
        // The call's limits bind whoever consumes the stream, not just the
        // collected path: a caller rendering events itself is exactly the one
        // with no other way to bound a provider that stops sending.
        let stream = crate::turn_collector::limit_stream(stream, limits.after(spent));
        // Drivers use an empty `TextDelta` as filler for wire frames that carry
        // nothing (usage, role, keep-alive). The limits above still see those
        // frames; the caller never does, so every delta it gets has content.
        Ok(Box::pin(stream.filter(|event| {
            std::future::ready(!matches!(event, Ok(crate::driver_registry::LlmStreamEvent::TextDelta(delta)) if delta.is_empty()))
        })))
    }

    pub async fn chat_completion(
        &self,
        messages: Vec<crate::driver_registry::Message>,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Result<crate::driver_registry::LlmResponse> {
        self.check_response_format(config)?;
        self.driver()?
            .chat_completion(&self.endpoint, messages, config)
            .await
            .map_err(|error| error.with_provider(self.id.as_str()))
    }

    pub fn supports_native_non_streaming(&self) -> bool {
        self.driver
            .as_ref()
            .is_some_and(|driver| driver.supports_native_non_streaming())
    }

    pub async fn chat_completion_non_streaming(
        &self,
        messages: Vec<crate::driver_registry::Message>,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Result<crate::driver_registry::LlmResponse> {
        self.check_response_format(config)?;
        if !config.limits.is_unbounded() {
            // Native JSON responses cannot enforce a byte cap before buffering
            // the body. Route bounded calls through the streaming collector so
            // every advertised limit is applied while bytes are arriving.
            let stream = self.chat_completion_stream(messages, config).await?;
            return Ok(
                crate::turn_collector::collect_turn(stream, &config.limits, |_| {})
                    .await?
                    .into_response(),
            );
        }
        self.driver()?
            .chat_completion_non_streaming(&self.endpoint, messages, config)
            .await
            .map_err(|error| error.with_provider(self.id.as_str()))
    }

    pub async fn list_models(
        &self,
    ) -> Result<Option<Vec<crate::driver_registry::DiscoveredModel>>> {
        let Some(driver) = &self.driver else {
            return Ok(None);
        };
        driver
            .list_models(&self.endpoint)
            .await
            .map_err(|error| error.with_provider(self.id.as_str()))
    }

    /// The provider's catalog, ready for display: discovered ids sorted
    /// newest-first and merged with the model-profile registry.
    ///
    /// `Ok(None)` means this provider offers no catalog — callers keep their
    /// curated suggestions rather than treating it as a failure. See
    /// [`list_models`](Self::list_models) for the driver's raw answer.
    pub async fn models(
        &self,
    ) -> Result<Option<Vec<crate::model_discovery::DiscoveredProviderModel>>> {
        let Some(models) = self.list_models().await? else {
            return Ok(None);
        };
        Ok(Some(crate::model_discovery::normalize_and_enrich(
            &self.driver_id(),
            models,
        )))
    }

    pub fn into_boxed_driver(self) -> BoxedChatDriver {
        Box::new(ProviderBoundDriver(self))
    }

    pub fn into_embeddings_driver(
        self,
    ) -> std::result::Result<
        crate::driver_registry::BoxedEmbeddingsDriver,
        crate::driver_registry::EmbeddingsDriverError,
    > {
        if self.embeddings.is_none() {
            return Err(crate::driver_registry::EmbeddingsDriverError::Provider(
                "Provider does not support embeddings".into(),
            ));
        }
        Ok(Box::new(ProviderOwnedEmbeddingsDriver(self)))
    }

    pub fn bind_embeddings(
        self,
        driver: crate::driver_registry::BoxedEmbeddingsDriver,
    ) -> crate::driver_registry::BoxedEmbeddingsDriver {
        Box::new(ProviderBoundEmbeddingsDriver {
            id: self.id,
            endpoint: self.endpoint,
            driver,
        })
    }
}

impl fmt::Debug for RuntimeProvider {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Provider")
            .field("id", &self.id)
            .field("endpoint", &self.endpoint)
            .finish_non_exhaustive()
    }
}

struct ProviderBoundDriver(RuntimeProvider);

struct ProviderBoundEmbeddingsDriver {
    id: ProviderKey,
    endpoint: ProviderEndpoint,
    driver: crate::driver_registry::BoxedEmbeddingsDriver,
}

#[async_trait]
impl crate::driver_registry::EmbeddingsDriver for ProviderBoundEmbeddingsDriver {
    async fn embed(
        &self,
        _endpoint: &ProviderEndpoint,
        request: crate::driver_registry::EmbedRequest,
    ) -> std::result::Result<
        crate::driver_registry::EmbedResponse,
        crate::driver_registry::EmbeddingsDriverError,
    > {
        self.driver
            .embed(&self.endpoint, request)
            .await
            .map_err(|error| {
                crate::driver_registry::EmbeddingsDriverError::Provider(format!(
                    "provider '{}': {error}",
                    self.id
                ))
            })
    }
}

#[async_trait]
impl ChatDriver for ProviderBoundDriver {
    fn native_async_driver(
        &self,
        model: &str,
        tools: std::collections::BTreeMap<String, Option<serde_json::Value>>,
        continuation: Option<crate::native_async::Delivery>,
    ) -> Option<Arc<dyn ChatDriver>> {
        let driver = self
            .0
            .driver
            .as_ref()?
            .native_async_driver(model, tools, continuation)?;
        Some(Arc::new(ProviderBoundDriver(RuntimeProvider {
            id: self.0.id.clone(),
            endpoint: self.0.endpoint.clone(),
            driver: Some(driver),
            decisions: self.0.decisions.clone(),
            embeddings: self.0.embeddings.clone(),
            driver_id: self.0.driver_id.clone(),
        })))
    }
    async fn chat_completion_stream(
        &self,
        _endpoint: &ProviderEndpoint,
        messages: Vec<crate::driver_registry::Message>,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Result<crate::driver_registry::LlmResponseStream> {
        self.0.chat_completion_stream(messages, config).await
    }

    async fn list_models(
        &self,
        _endpoint: &ProviderEndpoint,
    ) -> Result<Option<Vec<crate::driver_registry::DiscoveredModel>>> {
        self.0.list_models().await
    }

    fn supports_native_non_streaming(&self) -> bool {
        self.0.supports_native_non_streaming()
    }

    async fn chat_completion_non_streaming(
        &self,
        _endpoint: &ProviderEndpoint,
        messages: Vec<crate::driver_registry::Message>,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Result<crate::driver_registry::LlmResponse> {
        self.0.chat_completion_non_streaming(messages, config).await
    }

    fn supports_compact(&self) -> bool {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.supports_compact())
            .unwrap_or(false)
    }

    fn supports_stateful_responses(&self) -> bool {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.supports_stateful_responses())
            .unwrap_or(false)
    }

    fn effective_context_window(&self, model: &str) -> Option<usize> {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.effective_context_window(model))
            .unwrap_or(None)
    }

    fn supports_parallel_tool_calls(&self, model: &str) -> bool {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.supports_parallel_tool_calls(model))
            .unwrap_or(false)
    }

    fn supports_response_format(&self, model: &str) -> bool {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.supports_response_format(model))
            .unwrap_or(false)
    }

    fn provider_managed_reduction_option(
        &self,
        _endpoint: &ProviderEndpoint,
        model: &str,
        budget_tokens: usize,
    ) -> Option<(String, serde_json::Value)> {
        self.0.driver.as_ref()?.provider_managed_reduction_option(
            self.0.endpoint(),
            model,
            budget_tokens,
        )
    }

    fn provider_managed_reduction_fallback_reason(
        &self,
        _endpoint: &ProviderEndpoint,
        config: &crate::driver_registry::LlmCallConfig,
    ) -> Option<&'static str> {
        self.0
            .driver
            .as_ref()?
            .provider_managed_reduction_fallback_reason(self.0.endpoint(), config)
    }

    fn validate_provider_opaque_context(
        &self,
        context: &crate::driver_registry::ProviderOpaqueContext,
    ) -> bool {
        self.0
            .driver
            .as_ref()
            .map(|driver| driver.validate_provider_opaque_context(context))
            .unwrap_or(false)
    }

    async fn compact(
        &self,
        _endpoint: &ProviderEndpoint,
        request: crate::compact::CompactRequest,
    ) -> Result<Option<crate::compact::CompactResponse>> {
        self.0
            .driver()?
            .compact(self.0.endpoint(), request)
            .await
            .map_err(|error| error.with_provider(self.0.id.as_str()))
    }
}

/// Runtime provider instances keyed by their open service identity.
#[derive(Clone, Default)]
pub struct RuntimeProviderRegistry {
    providers: HashMap<ProviderKey, Arc<RuntimeProvider>>,
}

/// Public registry name for runtime provider instances.
pub type ProviderRegistry = RuntimeProviderRegistry;

impl RuntimeProviderRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn register(&mut self, provider: RuntimeProvider) -> Result<()> {
        if self.providers.contains_key(provider.id()) {
            return Err(crate::error::AgentLoopError::Configuration(format!(
                "provider '{}' is already registered; use replace() to overwrite intentionally",
                provider.id()
            )));
        }
        self.providers
            .insert(provider.id.clone(), Arc::new(provider));
        Ok(())
    }

    pub fn replace(&mut self, provider: RuntimeProvider) -> Option<Arc<RuntimeProvider>> {
        self.providers
            .insert(provider.id.clone(), Arc::new(provider))
    }

    pub fn get(&self, id: &ProviderKey) -> Option<Arc<RuntimeProvider>> {
        self.providers.get(id).cloned()
    }

    pub fn ids(&self) -> Vec<String> {
        let mut ids = self
            .providers
            .keys()
            .map(ToString::to_string)
            .collect::<Vec<_>>();
        ids.sort();
        ids
    }
}

impl fmt::Debug for RuntimeProviderRegistry {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ProviderRegistry")
            .field("providers", &self.ids())
            .finish()
    }
}

#[async_trait]
impl crate::decisions::DecisionsService for RuntimeProvider {
    fn is_configured(&self) -> bool {
        self.decisions.is_some()
    }
    async fn evaluate(
        &self,
        request: crate::decisions::DecisionRequest,
    ) -> Result<crate::decisions::DecisionOutcome> {
        if request
            .provider
            .as_ref()
            .is_some_and(|provider| provider != &self.id)
        {
            return Err(crate::error::AgentLoopError::Configuration(
                "A bound provider cannot select another account".into(),
            ));
        }
        self.evaluate_decisions(request).await
    }
    fn name(&self) -> &'static str {
        "ProviderDecisions"
    }
}

struct ProviderOwnedEmbeddingsDriver(RuntimeProvider);
#[async_trait]
impl crate::driver_registry::EmbeddingsDriver for ProviderOwnedEmbeddingsDriver {
    async fn embed(
        &self,
        _endpoint: &ProviderEndpoint,
        request: crate::driver_registry::EmbedRequest,
    ) -> std::result::Result<
        crate::driver_registry::EmbedResponse,
        crate::driver_registry::EmbeddingsDriverError,
    > {
        self.0.embed(request).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

    struct Noop;
    #[async_trait]
    impl ChatDriver for Noop {
        async fn chat_completion_stream(
            &self,
            _endpoint: &ProviderEndpoint,
            _messages: Vec<crate::Message>,
            _config: &crate::LlmCallConfig,
        ) -> Result<crate::LlmResponseStream> {
            unreachable!("configuration-only fixture must not execute")
        }
    }

    struct Catalog;
    #[async_trait]
    impl ChatDriver for Catalog {
        async fn chat_completion_stream(
            &self,
            _endpoint: &ProviderEndpoint,
            _messages: Vec<crate::Message>,
            _config: &crate::LlmCallConfig,
        ) -> Result<crate::LlmResponseStream> {
            unreachable!("catalog-only fixture must not execute")
        }

        async fn list_models(
            &self,
            _endpoint: &ProviderEndpoint,
        ) -> Result<Option<Vec<crate::driver_registry::DiscoveredModel>>> {
            Ok(Some(vec![crate::driver_registry::DiscoveredModel {
                model_id: "gpt-5.6-terra".to_string(),
                display_name: None,
                created_at: None,
                owned_by: None,
                capabilities: vec!["chat".to_string()],
                discovered_profile: None,
            }]))
        }
    }

    struct NativeNonStreaming {
        native_called: Arc<AtomicBool>,
        connect_delay: Duration,
        event_delay: Duration,
        response: String,
    }

    #[async_trait]
    impl ChatDriver for NativeNonStreaming {
        fn supports_native_non_streaming(&self) -> bool {
            true
        }

        async fn chat_completion_stream(
            &self,
            _endpoint: &ProviderEndpoint,
            _messages: Vec<crate::Message>,
            _config: &crate::LlmCallConfig,
        ) -> Result<crate::LlmResponseStream> {
            tokio::time::sleep(self.connect_delay).await;
            let events = vec![
                (
                    Duration::ZERO,
                    Ok(crate::LlmStreamEvent::TextDelta(self.response.clone())),
                ),
                (
                    self.event_delay,
                    Ok(crate::LlmStreamEvent::Done(Box::default())),
                ),
            ];
            Ok(Box::pin(futures::stream::iter(events).then(
                |(delay, event)| async move {
                    tokio::time::sleep(delay).await;
                    event
                },
            )))
        }

        async fn chat_completion_non_streaming(
            &self,
            _endpoint: &ProviderEndpoint,
            _messages: Vec<crate::Message>,
            _config: &crate::LlmCallConfig,
        ) -> Result<crate::LlmResponse> {
            self.native_called.store(true, Ordering::SeqCst);
            Ok(crate::LlmResponse {
                text: self.response.clone(),
                reasoning: vec![],
                tool_calls: None,
                metadata: Default::default(),
            })
        }
    }

    #[test]
    fn the_driver_kind_falls_back_to_the_runtime_key() {
        assert_eq!(
            RuntimeProvider::new("openai", Noop).driver_id(),
            crate::provider::DriverId::OpenAI
        );
    }

    #[test]
    fn a_declared_driver_kind_survives_a_caller_chosen_key() {
        let provider = RuntimeProvider::new("my-gateway", Noop)
            .with_driver_id(crate::provider::DriverId::OpenAI);
        assert_eq!(provider.id().as_str(), "my-gateway");
        assert_eq!(provider.driver_id(), crate::provider::DriverId::OpenAI);
    }

    #[tokio::test]
    async fn models_enriches_bare_ids_through_the_declared_driver_kind() {
        let catalog = RuntimeProvider::new("my-gateway", Catalog)
            .with_driver_id(crate::provider::DriverId::OpenAI)
            .models()
            .await
            .expect("catalog request")
            .expect("driver offers a catalog");
        assert_eq!(catalog.len(), 1);
        // The driver returned a bare id; the profile registry named it.
        assert!(catalog[0].display_name.is_some());
    }

    #[tokio::test]
    async fn a_driver_without_a_catalog_reports_no_catalog() {
        assert!(
            RuntimeProvider::new("openai", Noop)
                .models()
                .await
                .expect("catalog request")
                .is_none()
        );
    }

    #[tokio::test]
    async fn bounded_non_streaming_calls_enforce_the_total_timeout_before_headers() {
        let native_called = Arc::new(AtomicBool::new(false));
        let provider = RuntimeProvider::new(
            "bounded",
            NativeNonStreaming {
                native_called: Arc::clone(&native_called),
                connect_delay: Duration::from_millis(100),
                event_delay: Duration::ZERO,
                response: "done".into(),
            },
        );
        let mut config = crate::LlmCallConfig::new("model");
        config.limits =
            crate::turn_collector::TurnLimits::default().with_total(Duration::from_millis(10));

        let error = provider
            .chat_completion_non_streaming(vec![], &config)
            .await
            .expect_err("the call must time out");

        assert!(error.to_string().contains("did not finish"));
        assert!(!native_called.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn bounded_non_streaming_calls_enforce_the_total_timeout_during_the_body() {
        let native_called = Arc::new(AtomicBool::new(false));
        let provider = RuntimeProvider::new(
            "bounded",
            NativeNonStreaming {
                native_called: Arc::clone(&native_called),
                connect_delay: Duration::ZERO,
                event_delay: Duration::from_millis(100),
                response: "partial".into(),
            },
        );
        let mut config = crate::LlmCallConfig::new("model");
        config.limits =
            crate::turn_collector::TurnLimits::default().with_total(Duration::from_millis(10));

        let error = provider
            .chat_completion_non_streaming(vec![], &config)
            .await
            .expect_err("the body must time out");

        assert!(error.to_string().contains("did not finish"));
        assert!(!native_called.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn bounded_non_streaming_calls_reject_oversized_responses() {
        let native_called = Arc::new(AtomicBool::new(false));
        let provider = RuntimeProvider::new(
            "bounded",
            NativeNonStreaming {
                native_called: Arc::clone(&native_called),
                connect_delay: Duration::ZERO,
                event_delay: Duration::ZERO,
                response: "too large".into(),
            },
        );
        let mut config = crate::LlmCallConfig::new("model");
        config.limits = crate::turn_collector::TurnLimits::default().with_max_response_bytes(3);

        let error = provider
            .chat_completion_non_streaming(vec![], &config)
            .await
            .expect_err("the response must exceed the cap");

        assert!(error.to_string().contains("3-byte limit"));
        assert!(!native_called.load(Ordering::SeqCst));
    }

    #[test]
    fn endpoint_deduplicates_only_complete_path_segments() {
        for (base, path, expected) in [
            (
                "https://service.example/v1/",
                "chat",
                "https://service.example/v1/chat",
            ),
            (
                "https://service.example/v1/chat",
                "/chat",
                "https://service.example/v1/chat",
            ),
            (
                "https://service.example/v1/notchat",
                "chat",
                "https://service.example/v1/notchat/chat",
            ),
            ("https://chat", "chat", "https://chat/chat"),
            (
                "https://service.example/v1beta",
                "models/gemini-2.5-flash:streamGenerateContent?alt=sse",
                "https://service.example/v1beta/models/gemini-2.5-flash:streamGenerateContent?alt=sse",
            ),
            (
                "https://service.example/v1?token=a%2Fb#base",
                "chat?alt=sse#operation",
                "https://service.example/v1/chat?token=a%2Fb&alt=sse#operation",
            ),
            (
                "https://service.example/v1/chat?token=x",
                "chat?alt=sse",
                "https://service.example/v1/chat?token=x&alt=sse",
            ),
            (
                "https://service.example/v1?token=x",
                "chat",
                "https://service.example/v1/chat?token=x",
            ),
            (
                "https://service.example/v1/chat?token=x",
                "chat",
                "https://service.example/v1/chat?token=x",
            ),
            (
                "https://service.example/v1/notchat/completions",
                "chat/completions",
                "https://service.example/v1/notchat/completions/chat/completions",
            ),
            (
                "https://service.example/v1/chat/",
                "",
                "https://service.example/v1/chat",
            ),
        ] {
            let endpoint = ProviderEndpoint {
                base_url: Some(base.into()),
                ..Default::default()
            };
            assert_eq!(
                endpoint.url(path).as_deref(),
                Some(expected),
                "{base} + {path}"
            );
        }
        assert_eq!(ProviderEndpoint::default().url("chat"), None);
    }

    #[test]
    fn provider_key_deserialization_is_canonical() {
        let key: ProviderKey = serde_json::from_str(r#"" Gateway-PROD ""#).unwrap();
        assert_eq!(key.as_str(), "gateway-prod");
        assert_eq!(serde_json::to_string(&key).unwrap(), r#""gateway-prod""#);
    }

    #[tokio::test]
    async fn debug_redacts_auth_and_header_values() {
        let provider = RuntimeProvider::new("Gateway", Noop)
            .base_url("https://example.test/")
            .header("x-secret", "hidden-service-value")
            .auth(BearerAuth::new("hidden-key"));
        let debug = format!("{provider:?}");
        assert!(debug.contains("gateway"));
        assert!(debug.contains("x-secret"));
        assert!(!debug.contains("hidden-service-value"));
        assert!(!debug.contains("hidden-key"));
        let request = provider
            .endpoint()
            .resolve(
                "POST",
                "https://user:password@example.test/chat?token=query-secret#fragment-secret",
                b"{}",
            )
            .await
            .unwrap();
        assert_eq!(
            format!("{request:?}"),
            "ResolvedProviderRequest { url: \"https://example.test/chat\", headers: [\"x-secret\", \"authorization\"] }"
        );
    }

    #[test]
    fn duplicate_registration_is_explicit() {
        let mut registry = RuntimeProviderRegistry::new();
        registry.register(RuntimeProvider::new("a", Noop)).unwrap();
        let error = registry
            .register(RuntimeProvider::new("A", Noop).base_url("https://rejected.example"))
            .unwrap_err();
        let crate::AgentLoopError::Configuration(message) = error else {
            panic!("duplicate must be a configuration error")
        };
        assert_eq!(
            message,
            "provider 'a' is already registered; use replace() to overwrite intentionally"
        );
        assert_eq!(registry.ids(), vec!["a"]);
        let original = registry.get(&ProviderKey::new(" A ")).unwrap();
        assert!(
            original.endpoint().base_url().is_none(),
            "rejected duplicate must not replace original"
        );
        let old = registry
            .replace(RuntimeProvider::new("a", Noop).base_url("https://replacement.example"));
        assert!(Arc::ptr_eq(&old.unwrap(), &original));
        assert_eq!(
            registry
                .get(&ProviderKey::new("a"))
                .unwrap()
                .endpoint()
                .base_url(),
            Some("https://replacement.example")
        );
        assert!(
            original.endpoint().base_url().is_none(),
            "existing handle retains old provider"
        );
        assert!(registry.get(&ProviderKey::new("missing")).is_none());
        assert!(registry.replace(RuntimeProvider::new("z", Noop)).is_none());
        assert_eq!(registry.ids(), ["a", "z"]);
    }

    #[tokio::test]
    async fn one_protocol_serves_distinct_provider_identities() {
        let protocol: Arc<dyn ChatDriver> = Arc::new(Noop);
        let first = Provider::from_driver("first", protocol.clone())
            .base_url("https://first.example/v1")
            .header("x-service", "first")
            .auth(BearerAuth::new("first-key"));
        let second = Provider::from_driver("second", protocol.clone())
            .base_url("https://second.example/v1")
            .header("x-service", "second")
            .auth(BearerAuth::new("second-key"));

        assert!(Arc::ptr_eq(
            first.driver().unwrap(),
            second.driver().unwrap()
        ));
        let first_request = first
            .endpoint()
            .resolve("POST", first.endpoint().url("chat").unwrap(), b"{}")
            .await
            .unwrap();
        let second_request = second
            .endpoint()
            .resolve("POST", second.endpoint().url("chat").unwrap(), b"{}")
            .await
            .unwrap();
        assert_eq!(first_request.url, "https://first.example/v1/chat");
        assert_eq!(second_request.url, "https://second.example/v1/chat");
        assert_eq!(
            first_request.headers,
            [
                ("x-service".into(), "first".into()),
                ("authorization".into(), "Bearer first-key".into())
            ]
        );
        assert_eq!(
            second_request.headers,
            [
                ("x-service".into(), "second".into()),
                ("authorization".into(), "Bearer second-key".into())
            ]
        );
    }

    #[tokio::test]
    async fn refreshable_auth_is_resolved_for_each_request() {
        struct Rotating(std::sync::atomic::AtomicUsize);
        #[async_trait]
        impl ProviderAuth for Rotating {
            async fn headers(
                &self,
                request: ProviderAuthRequest<'_>,
            ) -> Result<Vec<(String, String)>> {
                assert_eq!(request.method, "POST");
                assert_eq!(request.url, "https://service.example/chat");
                assert_eq!(
                    request.headers,
                    [
                        ("AUTHORIZATION".into(), "stale".into()),
                        ("x-static".into(), "preserve".into())
                    ]
                );
                let token = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
                Ok(vec![
                    ("authorization".into(), format!("Bearer token-{token}")),
                    (
                        "x-signed-body".into(),
                        String::from_utf8_lossy(request.body).into_owned(),
                    ),
                ])
            }
            fn as_any(&self) -> &dyn std::any::Any {
                self
            }
        }

        let endpoint = ProviderEndpoint {
            base_url: Some("https://service.example".into()),
            headers: vec![
                ("AUTHORIZATION".into(), "stale".into()),
                ("x-static".into(), "preserve".into()),
            ],
            auth: Some(Arc::new(Rotating(std::sync::atomic::AtomicUsize::new(0)))),
        };
        let first = endpoint
            .resolve("POST", "https://service.example/chat", b"one")
            .await
            .unwrap();
        let second = endpoint
            .resolve("POST", "https://service.example/chat", b"two")
            .await
            .unwrap();
        assert_eq!(
            first.headers,
            [
                ("x-static".into(), "preserve".into()),
                ("authorization".into(), "Bearer token-1".into()),
                ("x-signed-body".into(), "one".into())
            ]
        );
        assert_eq!(
            second.headers,
            [
                ("x-static".into(), "preserve".into()),
                ("authorization".into(), "Bearer token-2".into()),
                ("x-signed-body".into(), "two".into())
            ]
        );
    }

    #[tokio::test]
    async fn provider_identity_prefixes_start_and_stream_errors() {
        struct Failing {
            fail_to_start: bool,
        }
        #[async_trait]
        impl ChatDriver for Failing {
            async fn chat_completion_stream(
                &self,
                _endpoint: &ProviderEndpoint,
                _messages: Vec<crate::Message>,
                _config: &crate::LlmCallConfig,
            ) -> Result<crate::LlmResponseStream> {
                if self.fail_to_start {
                    return Err(crate::AgentLoopError::llm("request failed"));
                }
                Ok(Box::pin(futures::stream::once(async {
                    Err(crate::AgentLoopError::llm("stream failed"))
                })))
            }
        }

        let config = crate::LlmCallConfig {
            model: "model".into(),
            ..Default::default()
        };
        let start = Provider::new(
            "customer-gateway",
            Failing {
                fail_to_start: true,
            },
        );
        let error = match start.chat_completion_stream(Vec::new(), &config).await {
            Ok(_) => panic!("the test driver should fail before returning a stream"),
            Err(error) => error,
        };
        let crate::AgentLoopError::Llm(error) = error else {
            panic!("LLM error variant must survive")
        };
        assert_eq!(error.message, "provider 'customer-gateway': request failed");

        let stream = Provider::new(
            "customer-gateway",
            Failing {
                fail_to_start: false,
            },
        );
        let error = stream
            .chat_completion_stream(Vec::new(), &config)
            .await
            .unwrap()
            .next()
            .await
            .unwrap()
            .unwrap_err();
        let crate::AgentLoopError::Llm(error) = error else {
            panic!("LLM error variant must survive")
        };
        assert_eq!(error.message, "provider 'customer-gateway': stream failed");
    }
}

#[cfg(test)]
#[path = "runtime_provider_service_tests.rs"]
mod service_tests;