everruns_provider/
runtime_provider.rs

1//! Runtime providers: service identity, endpoint, authentication, and wire driver.
2//!
3//! A [`ChatDriver`](crate::driver_registry::ChatDriver) implements a wire
4//! protocol. A `Provider` is a configured service that speaks that protocol.
5//! Keeping credentials and endpoints here lets one driver serve any number of
6//! services without adding vendor branches to the runtime.
7
8use std::collections::HashMap;
9use std::fmt;
10use std::sync::Arc;
11
12use async_trait::async_trait;
13use futures::StreamExt;
14use serde::{Deserialize, Serialize};
15
16use crate::driver_registry::{BoxedChatDriver, ChatDriver};
17use crate::error::Result;
18
19/// Open, normalized identity used by model specifications to select a provider.
20#[derive(Debug, Clone, PartialEq, Eq, Hash)]
21pub struct ProviderKey(String);
22
23impl ProviderKey {
24    pub fn new(id: impl AsRef<str>) -> Self {
25        Self(id.as_ref().trim().to_ascii_lowercase())
26    }
27
28    pub fn as_str(&self) -> &str {
29        &self.0
30    }
31}
32
33impl fmt::Display for ProviderKey {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        f.write_str(&self.0)
36    }
37}
38
39impl From<&str> for ProviderKey {
40    fn from(value: &str) -> Self {
41        Self::new(value)
42    }
43}
44
45impl From<String> for ProviderKey {
46    fn from(value: String) -> Self {
47        Self::new(value)
48    }
49}
50
51impl Serialize for ProviderKey {
52    fn serialize<S: serde::Serializer>(
53        &self,
54        serializer: S,
55    ) -> std::result::Result<S::Ok, S::Error> {
56        serializer.serialize_str(self.as_str())
57    }
58}
59
60impl<'de> Deserialize<'de> for ProviderKey {
61    fn deserialize<D: serde::Deserializer<'de>>(
62        deserializer: D,
63    ) -> std::result::Result<Self, D::Error> {
64        String::deserialize(deserializer).map(Self::new)
65    }
66}
67
68/// Immutable request material available to an authentication implementation.
69///
70/// `body` is the exact serialized payload. This makes the contract suitable
71/// for request signatures such as AWS SigV4 as well as ordinary header auth.
72pub struct ProviderAuthRequest<'a> {
73    pub method: &'a str,
74    pub url: &'a str,
75    pub headers: &'a [(String, String)],
76    pub body: &'a [u8],
77}
78
79/// Resolves authentication for each outbound provider request.
80#[async_trait]
81pub trait ProviderAuth: Send + Sync {
82    async fn headers(&self, request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>>;
83    fn as_any(&self) -> &dyn std::any::Any;
84}
85
86/// Static `Authorization: Bearer …` authentication.
87pub struct BearerAuth {
88    key: String,
89}
90
91impl BearerAuth {
92    pub fn new(key: impl Into<String>) -> Self {
93        Self { key: key.into() }
94    }
95}
96
97#[async_trait]
98impl ProviderAuth for BearerAuth {
99    async fn headers(&self, _request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>> {
100        Ok(vec![(
101            "authorization".to_string(),
102            format!("Bearer {}", self.key),
103        )])
104    }
105    fn as_any(&self) -> &dyn std::any::Any {
106        self
107    }
108}
109
110/// Static authentication carried in one named header.
111pub struct StaticHeaderAuth {
112    name: String,
113    value: String,
114}
115
116impl StaticHeaderAuth {
117    pub fn new(name: impl Into<String>, value: impl Into<String>) -> Self {
118        Self {
119            name: name.into().to_ascii_lowercase(),
120            value: value.into(),
121        }
122    }
123}
124
125#[async_trait]
126impl ProviderAuth for StaticHeaderAuth {
127    async fn headers(&self, _request: ProviderAuthRequest<'_>) -> Result<Vec<(String, String)>> {
128        Ok(vec![(self.name.clone(), self.value.clone())])
129    }
130    fn as_any(&self) -> &dyn std::any::Any {
131        self
132    }
133}
134
135/// Endpoint and authentication policy handed to a wire driver.
136///
137/// This runtime value is intentionally not serializable. Its `Debug` output
138/// exposes only header names and whether authentication is configured.
139#[derive(Clone, Default)]
140pub struct ProviderEndpoint {
141    base_url: Option<String>,
142    headers: Vec<(String, String)>,
143    auth: Option<Arc<dyn ProviderAuth>>,
144}
145
146impl ProviderEndpoint {
147    pub fn base_url(&self) -> Option<&str> {
148        self.base_url.as_deref()
149    }
150
151    pub fn url(&self, path: &str) -> Option<String> {
152        self.base_url.as_ref().map(|base| {
153            let base = base.trim_end_matches('/');
154            if path.is_empty() || base.ends_with(path) {
155                base.to_string()
156            } else {
157                format!("{base}/{}", path.trim_start_matches('/'))
158            }
159        })
160    }
161
162    pub async fn resolve(
163        &self,
164        method: &str,
165        url: impl Into<String>,
166        body: &[u8],
167    ) -> Result<ResolvedProviderRequest> {
168        let url = url.into();
169        let mut headers = self.headers.clone();
170        if let Some(auth) = &self.auth {
171            let auth_headers = auth
172                .headers(ProviderAuthRequest {
173                    method,
174                    url: &url,
175                    headers: &headers,
176                    body,
177                })
178                .await?;
179            for (name, value) in auth_headers {
180                headers.retain(|(existing, _)| !existing.eq_ignore_ascii_case(&name));
181                headers.push((name.to_ascii_lowercase(), value));
182            }
183        }
184        Ok(ResolvedProviderRequest { url, headers })
185    }
186
187    /// Access a provider-owned authentication implementation needed by a
188    /// protocol whose signing stack cannot be represented as header strings.
189    pub fn auth<T: ProviderAuth + 'static>(&self) -> Option<&T> {
190        self.auth.as_deref()?.as_any().downcast_ref()
191    }
192}
193
194impl fmt::Debug for ProviderEndpoint {
195    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
196        f.debug_struct("ProviderEndpoint")
197            .field("base_url", &self.base_url.as_ref().map(|_| "<configured>"))
198            .field("auth", &self.auth.as_ref().map(|_| "<configured>"))
199            .field(
200                "headers",
201                &self
202                    .headers
203                    .iter()
204                    .map(|(name, _)| name.as_str())
205                    .collect::<Vec<_>>(),
206            )
207            .finish()
208    }
209}
210
211/// Fully resolved outbound request metadata.
212#[derive(Clone)]
213pub struct ResolvedProviderRequest {
214    pub url: String,
215    pub headers: Vec<(String, String)>,
216}
217
218impl fmt::Debug for ResolvedProviderRequest {
219    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
220        f.debug_struct("ResolvedProviderRequest")
221            .field("url", &redacted_url(&self.url))
222            .field(
223                "headers",
224                &self
225                    .headers
226                    .iter()
227                    .map(|(name, _)| name.as_str())
228                    .collect::<Vec<_>>(),
229            )
230            .finish()
231    }
232}
233
234fn redacted_url(value: &str) -> String {
235    let Ok(mut url) = reqwest::Url::parse(value) else {
236        return "<configured>".to_string();
237    };
238    let _ = url.set_username("");
239    let _ = url.set_password(None);
240    url.set_query(None);
241    url.set_fragment(None);
242    url.to_string()
243}
244
245/// Runtime service assembly over one reusable wire-protocol driver.
246#[derive(Clone)]
247pub struct RuntimeProvider {
248    id: ProviderKey,
249    driver: Arc<dyn ChatDriver>,
250    endpoint: ProviderEndpoint,
251}
252
253/// Public application-facing name for a runtime provider.
254pub type Provider = RuntimeProvider;
255
256impl RuntimeProvider {
257    pub fn new(id: impl Into<ProviderKey>, driver: impl ChatDriver + 'static) -> Self {
258        Self::from_driver(id, Arc::new(driver))
259    }
260
261    pub fn from_driver(id: impl Into<ProviderKey>, driver: Arc<dyn ChatDriver>) -> Self {
262        Self {
263            id: id.into(),
264            driver,
265            endpoint: ProviderEndpoint::default(),
266        }
267    }
268
269    pub fn base_url(mut self, url: impl Into<String>) -> Self {
270        self.endpoint.base_url = Some(url.into().trim_end_matches('/').to_string());
271        self
272    }
273
274    pub fn auth(mut self, auth: impl ProviderAuth + 'static) -> Self {
275        self.endpoint.auth = Some(Arc::new(auth));
276        self
277    }
278
279    pub fn auth_arc(mut self, auth: Arc<dyn ProviderAuth>) -> Self {
280        self.endpoint.auth = Some(auth);
281        self
282    }
283
284    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
285        self.endpoint
286            .headers
287            .push((name.into().to_ascii_lowercase(), value.into()));
288        self
289    }
290
291    pub fn id(&self) -> &ProviderKey {
292        &self.id
293    }
294
295    pub fn driver(&self) -> &Arc<dyn ChatDriver> {
296        &self.driver
297    }
298
299    pub fn endpoint(&self) -> &ProviderEndpoint {
300        &self.endpoint
301    }
302
303    pub async fn chat_completion_stream(
304        &self,
305        messages: Vec<crate::driver_registry::LlmMessage>,
306        config: &crate::driver_registry::LlmCallConfig,
307    ) -> Result<crate::driver_registry::LlmResponseStream> {
308        let id = self.id.to_string();
309        let stream = self
310            .driver
311            .chat_completion_stream(&self.endpoint, messages, config)
312            .await
313            .map_err(|error| error.with_provider(&id))?;
314        Ok(Box::pin(stream.map(move |result| {
315            result.map_err(|error| error.with_provider(&id))
316        })))
317    }
318
319    pub async fn chat_completion(
320        &self,
321        messages: Vec<crate::driver_registry::LlmMessage>,
322        config: &crate::driver_registry::LlmCallConfig,
323    ) -> Result<crate::driver_registry::LlmResponse> {
324        self.driver
325            .chat_completion(&self.endpoint, messages, config)
326            .await
327            .map_err(|error| error.with_provider(self.id.as_str()))
328    }
329
330    pub async fn list_models(
331        &self,
332    ) -> Result<Option<Vec<crate::driver_registry::DiscoveredModel>>> {
333        self.driver
334            .list_models(&self.endpoint)
335            .await
336            .map_err(|error| error.with_provider(self.id.as_str()))
337    }
338
339    pub fn into_boxed_driver(self) -> BoxedChatDriver {
340        Box::new(ProviderBoundDriver(self))
341    }
342
343    pub fn bind_embeddings(
344        self,
345        driver: crate::driver_registry::BoxedEmbeddingsDriver,
346    ) -> crate::driver_registry::BoxedEmbeddingsDriver {
347        Box::new(ProviderBoundEmbeddingsDriver {
348            id: self.id,
349            endpoint: self.endpoint,
350            driver,
351        })
352    }
353}
354
355impl fmt::Debug for RuntimeProvider {
356    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
357        f.debug_struct("Provider")
358            .field("id", &self.id)
359            .field("endpoint", &self.endpoint)
360            .finish_non_exhaustive()
361    }
362}
363
364struct ProviderBoundDriver(RuntimeProvider);
365
366struct ProviderBoundEmbeddingsDriver {
367    id: ProviderKey,
368    endpoint: ProviderEndpoint,
369    driver: crate::driver_registry::BoxedEmbeddingsDriver,
370}
371
372#[async_trait]
373impl crate::driver_registry::EmbeddingsDriver for ProviderBoundEmbeddingsDriver {
374    async fn embed(
375        &self,
376        _endpoint: &ProviderEndpoint,
377        request: crate::driver_registry::EmbedRequest,
378    ) -> std::result::Result<
379        crate::driver_registry::EmbedResponse,
380        crate::driver_registry::EmbeddingsDriverError,
381    > {
382        self.driver
383            .embed(&self.endpoint, request)
384            .await
385            .map_err(|error| {
386                crate::driver_registry::EmbeddingsDriverError::Provider(format!(
387                    "provider '{}': {error}",
388                    self.id
389                ))
390            })
391    }
392}
393
394#[async_trait]
395impl ChatDriver for ProviderBoundDriver {
396    async fn chat_completion_stream(
397        &self,
398        _endpoint: &ProviderEndpoint,
399        messages: Vec<crate::driver_registry::LlmMessage>,
400        config: &crate::driver_registry::LlmCallConfig,
401    ) -> Result<crate::driver_registry::LlmResponseStream> {
402        self.0.chat_completion_stream(messages, config).await
403    }
404
405    async fn list_models(
406        &self,
407        _endpoint: &ProviderEndpoint,
408    ) -> Result<Option<Vec<crate::driver_registry::DiscoveredModel>>> {
409        self.0.list_models().await
410    }
411
412    fn supports_compact(&self) -> bool {
413        self.0.driver.supports_compact()
414    }
415
416    fn supports_stateful_responses(&self) -> bool {
417        self.0.driver.supports_stateful_responses()
418    }
419
420    fn effective_context_window(&self, model: &str) -> Option<usize> {
421        self.0.driver.effective_context_window(model)
422    }
423
424    fn supports_parallel_tool_calls(&self, model: &str) -> bool {
425        self.0.driver.supports_parallel_tool_calls(model)
426    }
427
428    async fn compact(
429        &self,
430        _endpoint: &ProviderEndpoint,
431        request: crate::openresponses_protocol::CompactRequest,
432    ) -> Result<Option<crate::openresponses_protocol::CompactResponse>> {
433        self.0
434            .driver
435            .compact(self.0.endpoint(), request)
436            .await
437            .map_err(|error| error.with_provider(self.0.id.as_str()))
438    }
439}
440
441/// Runtime provider instances keyed by their open service identity.
442#[derive(Clone, Default)]
443pub struct RuntimeProviderRegistry {
444    providers: HashMap<ProviderKey, Arc<RuntimeProvider>>,
445}
446
447/// Public registry name for runtime provider instances.
448pub type ProviderRegistry = RuntimeProviderRegistry;
449
450impl RuntimeProviderRegistry {
451    pub fn new() -> Self {
452        Self::default()
453    }
454
455    pub fn register(&mut self, provider: RuntimeProvider) -> Result<()> {
456        if self.providers.contains_key(provider.id()) {
457            return Err(crate::error::AgentLoopError::Configuration(format!(
458                "provider '{}' is already registered; use replace() to overwrite intentionally",
459                provider.id()
460            )));
461        }
462        self.providers
463            .insert(provider.id.clone(), Arc::new(provider));
464        Ok(())
465    }
466
467    pub fn replace(&mut self, provider: RuntimeProvider) -> Option<Arc<RuntimeProvider>> {
468        self.providers
469            .insert(provider.id.clone(), Arc::new(provider))
470    }
471
472    pub fn get(&self, id: &ProviderKey) -> Option<Arc<RuntimeProvider>> {
473        self.providers.get(id).cloned()
474    }
475
476    pub fn ids(&self) -> Vec<String> {
477        let mut ids = self
478            .providers
479            .keys()
480            .map(ToString::to_string)
481            .collect::<Vec<_>>();
482        ids.sort();
483        ids
484    }
485}
486
487impl fmt::Debug for RuntimeProviderRegistry {
488    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
489        f.debug_struct("ProviderRegistry")
490            .field("providers", &self.ids())
491            .finish()
492    }
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498
499    #[test]
500    fn provider_key_deserialization_is_canonical() {
501        let key: ProviderKey = serde_json::from_str(r#"" Gateway-PROD ""#).unwrap();
502        assert_eq!(key.as_str(), "gateway-prod");
503        assert_eq!(serde_json::to_string(&key).unwrap(), r#""gateway-prod""#);
504    }
505
506    #[test]
507    fn debug_redacts_auth_and_header_values() {
508        struct Noop;
509        #[async_trait]
510        impl ChatDriver for Noop {
511            async fn chat_completion_stream(
512                &self,
513                _endpoint: &ProviderEndpoint,
514                _messages: Vec<crate::driver_registry::LlmMessage>,
515                _config: &crate::driver_registry::LlmCallConfig,
516            ) -> Result<crate::driver_registry::LlmResponseStream> {
517                unreachable!()
518            }
519        }
520
521        let provider = RuntimeProvider::new("Gateway", Noop)
522            .base_url("https://example.test/")
523            .header("x-secret", "hidden-service-value")
524            .auth(BearerAuth::new("hidden-key"));
525        let debug = format!("{provider:?}");
526        assert!(debug.contains("gateway"));
527        assert!(debug.contains("x-secret"));
528        assert!(!debug.contains("hidden-service-value"));
529        assert!(!debug.contains("hidden-key"));
530    }
531
532    #[test]
533    fn duplicate_registration_is_explicit() {
534        struct Noop;
535        #[async_trait]
536        impl ChatDriver for Noop {
537            async fn chat_completion_stream(
538                &self,
539                _endpoint: &ProviderEndpoint,
540                _messages: Vec<crate::driver_registry::LlmMessage>,
541                _config: &crate::driver_registry::LlmCallConfig,
542            ) -> Result<crate::driver_registry::LlmResponseStream> {
543                unreachable!()
544            }
545        }
546        let mut registry = RuntimeProviderRegistry::new();
547        registry.register(RuntimeProvider::new("a", Noop)).unwrap();
548        assert!(registry.register(RuntimeProvider::new("A", Noop)).is_err());
549        assert_eq!(registry.ids(), vec!["a"]);
550    }
551
552    #[tokio::test]
553    async fn one_protocol_serves_distinct_provider_identities() {
554        struct Noop;
555        #[async_trait]
556        impl ChatDriver for Noop {
557            async fn chat_completion_stream(
558                &self,
559                _endpoint: &ProviderEndpoint,
560                _messages: Vec<crate::driver_registry::LlmMessage>,
561                _config: &crate::driver_registry::LlmCallConfig,
562            ) -> Result<crate::driver_registry::LlmResponseStream> {
563                unreachable!()
564            }
565        }
566
567        let protocol: Arc<dyn ChatDriver> = Arc::new(Noop);
568        let first = Provider::from_driver("first", protocol.clone())
569            .base_url("https://first.example/v1")
570            .header("x-service", "first")
571            .auth(BearerAuth::new("first-key"));
572        let second = Provider::from_driver("second", protocol.clone())
573            .base_url("https://second.example/v1")
574            .header("x-service", "second")
575            .auth(BearerAuth::new("second-key"));
576
577        assert!(Arc::ptr_eq(first.driver(), second.driver()));
578        let first_request = first
579            .endpoint()
580            .resolve("POST", first.endpoint().url("chat").unwrap(), b"{}")
581            .await
582            .unwrap();
583        let second_request = second
584            .endpoint()
585            .resolve("POST", second.endpoint().url("chat").unwrap(), b"{}")
586            .await
587            .unwrap();
588        assert_ne!(first_request.url, second_request.url);
589        assert_ne!(first_request.headers, second_request.headers);
590    }
591
592    #[tokio::test]
593    async fn refreshable_auth_is_resolved_for_each_request() {
594        struct Rotating(std::sync::atomic::AtomicUsize);
595        #[async_trait]
596        impl ProviderAuth for Rotating {
597            async fn headers(
598                &self,
599                request: ProviderAuthRequest<'_>,
600            ) -> Result<Vec<(String, String)>> {
601                let token = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
602                Ok(vec![
603                    ("authorization".into(), format!("Bearer token-{token}")),
604                    (
605                        "x-signed-body".into(),
606                        String::from_utf8_lossy(request.body).into_owned(),
607                    ),
608                ])
609            }
610            fn as_any(&self) -> &dyn std::any::Any {
611                self
612            }
613        }
614
615        let endpoint = ProviderEndpoint {
616            base_url: Some("https://service.example".into()),
617            headers: Vec::new(),
618            auth: Some(Arc::new(Rotating(std::sync::atomic::AtomicUsize::new(0)))),
619        };
620        let first = endpoint
621            .resolve("POST", "https://service.example/chat", b"one")
622            .await
623            .unwrap();
624        let second = endpoint
625            .resolve("POST", "https://service.example/chat", b"two")
626            .await
627            .unwrap();
628        assert_eq!(first.headers[0].1, "Bearer token-1");
629        assert_eq!(second.headers[0].1, "Bearer token-2");
630        assert_eq!(first.headers[1].1, "one");
631        assert_eq!(second.headers[1].1, "two");
632    }
633
634    #[tokio::test]
635    async fn provider_identity_prefixes_start_and_stream_errors() {
636        struct Failing {
637            fail_to_start: bool,
638        }
639        #[async_trait]
640        impl ChatDriver for Failing {
641            async fn chat_completion_stream(
642                &self,
643                _endpoint: &ProviderEndpoint,
644                _messages: Vec<crate::LlmMessage>,
645                _config: &crate::LlmCallConfig,
646            ) -> Result<crate::LlmResponseStream> {
647                if self.fail_to_start {
648                    return Err(crate::AgentLoopError::llm("request failed"));
649                }
650                Ok(Box::pin(futures::stream::once(async {
651                    Err(crate::AgentLoopError::llm("stream failed"))
652                })))
653            }
654        }
655
656        let config = crate::LlmCallConfig {
657            model: "model".into(),
658            temperature: None,
659            max_tokens: None,
660            tools: Vec::new(),
661            reasoning_effort: None,
662            speed: None,
663            verbosity: None,
664            metadata: std::collections::HashMap::new(),
665            previous_response_id: None,
666            provider_opaque_context: None,
667            tool_search: None,
668            prompt_cache: None,
669            openrouter_routing: None,
670            parallel_tool_calls: None,
671            volatile_suffix_len: 0,
672        };
673        let start = Provider::new(
674            "customer-gateway",
675            Failing {
676                fail_to_start: true,
677            },
678        );
679        let error = match start.chat_completion_stream(Vec::new(), &config).await {
680            Ok(_) => panic!("the test driver should fail before returning a stream"),
681            Err(error) => error,
682        };
683        assert!(error.to_string().contains("provider 'customer-gateway'"));
684
685        let stream = Provider::new(
686            "customer-gateway",
687            Failing {
688                fail_to_start: false,
689            },
690        );
691        let error = stream
692            .chat_completion_stream(Vec::new(), &config)
693            .await
694            .unwrap()
695            .next()
696            .await
697            .unwrap()
698            .unwrap_err();
699        assert!(error.to_string().contains("provider 'customer-gateway'"));
700    }
701}
everruns_provider/runtime_provider.rs

everruns_provider/
runtime_provider.rs
