providers_kit/provider.rs
1//! Root provider contract and opaque provider handle.
2
3use std::fmt;
4use std::future::Future;
5use std::num::NonZeroUsize;
6use std::pin::Pin;
7use std::sync::{Arc, Mutex};
8use std::task::{Context, Poll};
9
10use futures_util::Stream;
11use tokio::sync::Notify;
12use tokio::task::JoinHandle;
13
14use crate::KitError;
15use crate::batch::{BatchJob, BatchListRequest, BatchPage, BatchResult};
16use crate::batch_ownership::BatchRegistry;
17use crate::config::{ProviderConfig, ProviderKind};
18use crate::error::Operation;
19use crate::error::ValidationViolation;
20use crate::files::{FileListRequest, FileObject, FilePage, FileUploadRequest};
21use crate::generation::{GenerateEvent, GenerateEventStream, GenerationContinuation};
22use crate::models::{Model, ModelBackend, ModelId, ModelInner};
23use crate::models::{ModelListRequest, ModelPage, ModelSummary};
24use crate::providers::{
25 anthropic::AnthropicBackend,
26 gemini::GeminiBackend,
27 ollama::OllamaBackend,
28 openai::{OpenAiBackend, OpenAiBatchSubmission},
29};
30use crate::runtime::HealthStatus;
31
32/// Public implementation contract for one provider root.
33///
34/// Provider and Model clones share one lifecycle. Shutdown seals model admission,
35/// drains accepted work, and publishes one terminal result to every waiter; cancelling
36/// a waiter cannot cancel owned cleanup. Models have no independent shutdown.
37/// Caller implementations retain their concrete associated children directly.
38///
39/// # Examples
40///
41/// ```
42///
43/// # #[tokio::main(flavor = "current_thread")]
44/// # async fn main() -> Result<(), providers_kit::KitError> {
45/// use providers_kit::{Provider, ProviderBackend, ProviderConfig, ModelBackend, ModelId, config::{ProviderKind, OllamaConfig, TransportConfig}};
46/// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
47/// let model = provider.model(ModelId::new("exact-model")?)?;
48/// assert_eq!(model.model_id().as_str(), "exact-model");
49/// provider.shutdown().await?;
50///
51/// # Ok(())
52/// # }
53/// ```
54///
55/// # Implement a concrete local hierarchy
56///
57/// This runnable caller implementation is a deliberately limited, in-process text
58/// echo service, **not** a language model or proof of any supplied provider. It
59/// supports one text message with default settings and text-only Sessions. Optional
60/// operations retain their Unsupported defaults. Its work is synchronous and finite:
61/// there is no transport, spawned task or asynchronous resource to drain. A real
62/// network implementation must additionally own admitted work and shutdown cleanup.
63///
64/// The continuation is caller-owned, cloneable, and deliberately has no serialization
65/// contract. Finite streams yield on demand without a forwarding task. Shared closure
66/// seals the root and existing children; consuming Session close releases its local queue.
67///
68/// ```
69/// use std::{collections::VecDeque, sync::{Arc, atomic::{AtomicBool, Ordering}}};
70/// use futures_util::{stream, StreamExt};
71/// use providers_kit::{ProviderBackend, ModelBackend, SessionBackend, ModelId, KitError};
72/// use providers_kit::error::{ErrorKind, ErrorPhase, Operation};
73/// use providers_kit::generation::{ContentPart, FinishReason, GenerateEvent,
74/// GenerateEventStream, GenerateRequest, GenerateResponse, GenerationConfig, Message, Role, Usage};
75/// use providers_kit::session::{SessionConfig, SessionEvent, SessionInput};
76///
77/// #[derive(Clone, Default)]
78/// struct EchoProvider { closed: Arc<AtomicBool> }
79/// #[derive(Clone)]
80/// struct EchoContinuation(Arc<()>);
81/// struct EchoModel { id: ModelId, root: EchoProvider, replay: EchoContinuation }
82/// struct EchoSession { model: ModelId, root: EchoProvider, events: VecDeque<SessionEvent> }
83///
84/// impl EchoProvider {
85/// fn check(&self, operation: Operation) -> Result<(), KitError> {
86/// if self.closed.load(Ordering::SeqCst) {
87/// return Err(KitError::new(ErrorKind::Closed, operation, ErrorPhase::Lifecycle));
88/// }
89/// Ok(())
90/// }
91/// }
92/// impl ProviderBackend for EchoProvider {
93/// type Model = EchoModel;
94/// fn model(&self, id: ModelId) -> Result<EchoModel, KitError> {
95/// self.check(Operation::Generate)?;
96/// if id.as_str() != "local-echo" {
97/// return Err(KitError::new(ErrorKind::Unsupported, Operation::Generate,
98/// ErrorPhase::InputValidation));
99/// }
100/// Ok(EchoModel { id, root: self.clone(), replay: EchoContinuation(Arc::new(())) })
101/// }
102/// async fn shutdown(&self) -> Result<(), KitError> {
103/// // No await: sealing and this resource-free cleanup complete together.
104/// self.closed.store(true, Ordering::SeqCst);
105/// Ok(())
106/// }
107/// }
108/// impl EchoModel {
109/// fn response(&self, request: &GenerateRequest<EchoContinuation>)
110/// -> Result<GenerateResponse<EchoContinuation>, KitError>
111/// {
112/// self.root.check(Operation::Generate)?;
113/// let unsupported = || KitError::new(ErrorKind::Unsupported,
114/// Operation::Generate, ErrorPhase::InputValidation);
115/// if !request.tools().is_empty() || request.reasoning().is_some()
116/// || request.response_format().is_some()
117/// || request.generation() != &GenerationConfig::default()
118/// || request.messages().len() != 1 {
119/// return Err(unsupported());
120/// }
121/// let message = &request.messages()[0];
122/// if message.role() != Role::User { return Err(unsupported()); }
123/// let [ContentPart::Text { text }] = message.content() else {
124/// return Err(unsupported());
125/// };
126/// if text.len() > 64 * 1024 { return Err(unsupported()); }
127/// Ok(GenerateResponse {
128/// model: self.id.clone(),
129/// message: Message::assistant_text(text.clone())?
130/// .with_continuation(self.replay.clone())?,
131/// finish_reason: Some(FinishReason::Stop),
132/// citations: Vec::new(), artifacts: Vec::new(), hosted_tool_outputs: Vec::new(),
133/// usage: Usage::default(), // No token accounting is invented.
134/// })
135/// }
136/// }
137/// impl ModelBackend for EchoModel {
138/// type Continuation = EchoContinuation;
139/// type Session = EchoSession;
140/// fn model_id(&self) -> &ModelId { &self.id }
141/// async fn generate(&self, request: &GenerateRequest<EchoContinuation>)
142/// -> Result<GenerateResponse<EchoContinuation>, KitError> { self.response(request) }
143/// async fn generate_stream(&self, request: &GenerateRequest<EchoContinuation>)
144/// -> Result<GenerateEventStream<EchoContinuation>, KitError>
145/// {
146/// let response = self.response(request)?;
147/// let text: String = response.message.content().iter().filter_map(|part| match part {
148/// ContentPart::Text { text } => Some(text.as_str()), _ => None,
149/// }).collect();
150/// Ok(GenerateEventStream::new(stream::iter([
151/// Ok(GenerateEvent::Started { model: self.id.clone() }),
152/// Ok(GenerateEvent::ContentDelta { delta: text }),
153/// Ok(GenerateEvent::Completed(Box::new(response))),
154/// ])))
155/// }
156/// async fn open_session(&self, config: SessionConfig) -> Result<EchoSession, KitError> {
157/// self.root.check(Operation::OpenSession)?;
158/// if config != SessionConfig::default() {
159/// return Err(KitError::new(ErrorKind::Unsupported, Operation::OpenSession,
160/// ErrorPhase::InputValidation));
161/// }
162/// Ok(EchoSession { model: self.id.clone(), root: self.root.clone(), events: VecDeque::new() })
163/// }
164/// }
165/// impl SessionBackend for EchoSession {
166/// fn id(&self) -> &str { "local-echo-session" }
167/// fn model_id(&self) -> &ModelId { &self.model }
168/// async fn send(&mut self, input: SessionInput) -> Result<(), KitError> {
169/// self.root.check(Operation::SessionSend)?;
170/// if !self.events.is_empty() {
171/// return Err(KitError::new(ErrorKind::InvalidState, Operation::SessionSend,
172/// ErrorPhase::Lifecycle));
173/// }
174/// match input {
175/// SessionInput::Text { text } if !text.trim().is_empty() && text.len() <= 64 * 1024 => {
176/// self.events.push_back(SessionEvent::TextDelta { delta: text });
177/// self.events.push_back(SessionEvent::Completed);
178/// Ok(())
179/// }
180/// _ => Err(KitError::new(ErrorKind::Unsupported, Operation::SessionSend,
181/// ErrorPhase::InputValidation)),
182/// }
183/// }
184/// async fn recv(&mut self) -> Result<SessionEvent, KitError> {
185/// self.root.check(Operation::SessionReceive)?;
186/// self.events.pop_front().ok_or_else(|| KitError::new(ErrorKind::InvalidState,
187/// Operation::SessionReceive, ErrorPhase::Lifecycle))
188/// }
189/// async fn close(self) -> Result<(), KitError> { Ok(()) }
190/// }
191///
192/// # #[tokio::main(flavor = "current_thread")]
193/// # async fn main() -> Result<(), KitError> {
194/// let provider = EchoProvider::default();
195/// let model = provider.model(ModelId::new("local-echo")?)?;
196/// let request = GenerateRequest::new(vec![Message::user_text("hello")?])?;
197/// let response = model.generate(&request).await?;
198/// assert!(matches!(response.message.content(), [ContentPart::Text { text }] if text == "hello"));
199/// assert!(Arc::ptr_eq(&response.message.continuation().unwrap().0, &model.replay.0));
200/// let mut events = model.generate_stream(&request).await?;
201/// assert!(matches!(events.next().await.transpose()?, Some(GenerateEvent::Started { .. })));
202/// assert!(matches!(events.next().await.transpose()?, Some(GenerateEvent::ContentDelta { delta }) if delta == "hello"));
203/// assert!(matches!(events.next().await.transpose()?, Some(GenerateEvent::Completed(_))));
204/// assert!(events.next().await.is_none());
205/// let mut session = model.open_session(SessionConfig::default()).await?;
206/// session.send(SessionInput::Text { text: "hello again".into() }).await?;
207/// assert!(matches!(session.recv().await?, SessionEvent::TextDelta { delta } if delta == "hello again"));
208/// assert_eq!(session.recv().await?, SessionEvent::Completed);
209/// session.close().await?;
210/// provider.shutdown().await?;
211/// assert_eq!(model.generate(&request).await.err().unwrap().kind(), ErrorKind::Closed);
212/// provider.shutdown().await?; // Replays the same resource-free terminal result.
213/// # Ok(())
214/// # }
215/// ```
216///
217pub trait ProviderBackend: Send + Sync + 'static {
218 /// Exact model child returned by this provider.
219 type Model: ModelBackend;
220
221 /// Selects one exact provider-local model.
222 ///
223 /// # Errors
224 ///
225 /// Returns an error when root admission is closed or the selected implementation cannot construct this exact model. Selection never silently falls back to another model.
226 ///
227 /// # Examples
228 ///
229 /// ```
230 ///
231 /// # #[tokio::main(flavor = "current_thread")]
232 /// # async fn main() -> Result<(), providers_kit::KitError> {
233 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, ModelBackend, ModelId, config::{ProviderKind, OllamaConfig, TransportConfig}};
234 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
235 /// let model = provider.model(ModelId::new("exact-model")?)?;
236 /// assert_eq!(model.model_id().as_str(), "exact-model");
237 /// provider.shutdown().await?;
238 ///
239 /// # Ok(())
240 /// # }
241 /// ```
242 ///
243 fn model(&self, id: ModelId) -> Result<Self::Model, KitError>;
244
245 /// Lists provider-visible models.
246 ///
247 /// This example requires a real configured service, valid credentials, and any exact
248 /// resource ids shown. Only delete or cancel resources the caller owns.
249 /// The default trait implementation returns Unsupported.
250 ///
251 /// # Errors
252 ///
253 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
254 ///
255 /// # Examples
256 ///
257 /// ```no_run
258 ///
259 /// # #[tokio::main(flavor = "current_thread")]
260 /// # async fn main() -> Result<(), providers_kit::KitError> {
261 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
262 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
263 /// let result = provider.list_models(providers_kit::models::ModelListRequest::default()).await;
264 /// let cleanup = provider.shutdown().await;
265 /// let _value = result?;
266 /// cleanup?;
267 ///
268 /// # Ok(())
269 /// # }
270 /// ```
271 ///
272 fn list_models(
273 &self,
274 _request: ModelListRequest,
275 ) -> impl Future<Output = Result<ModelPage, KitError>> + Send + '_ {
276 async {
277 Err(KitError::unsupported(
278 ValidationViolation::Unsupported {
279 field: "models.list",
280 rule: "operation is not implemented",
281 },
282 "provider model discovery",
283 )
284 .at_operation(Operation::ListModels))
285 }
286 }
287
288 /// Fetches descriptive metadata for one exact provider-local model.
289 ///
290 /// This example requires a real configured service, valid credentials, and any exact
291 /// resource ids shown. Only delete or cancel resources the caller owns.
292 /// The default trait implementation returns Unsupported.
293 ///
294 /// # Errors
295 ///
296 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
297 ///
298 /// # Examples
299 ///
300 /// ```no_run
301 ///
302 /// # #[tokio::main(flavor = "current_thread")]
303 /// # async fn main() -> Result<(), providers_kit::KitError> {
304 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
305 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
306 /// let result = provider.model_info(providers_kit::ModelId::new("your-exact-model-id")?).await;
307 /// let cleanup = provider.shutdown().await;
308 /// let _value = result?;
309 /// cleanup?;
310 ///
311 /// # Ok(())
312 /// # }
313 /// ```
314 ///
315 fn model_info(
316 &self,
317 _id: ModelId,
318 ) -> impl Future<Output = Result<ModelSummary, KitError>> + Send + '_ {
319 async {
320 Err(KitError::unsupported(
321 ValidationViolation::Unsupported {
322 field: "models.info",
323 rule: "operation is not implemented",
324 },
325 "provider model metadata",
326 )
327 .at_operation(Operation::ModelInfo))
328 }
329 }
330
331 /// Uploads one provider-managed file.
332 ///
333 /// This example requires a real configured service, valid credentials, and any exact
334 /// resource ids shown. Only delete or cancel resources the caller owns.
335 /// The default trait implementation returns Unsupported.
336 ///
337 /// # Errors
338 ///
339 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
340 ///
341 /// # Examples
342 ///
343 /// ```no_run
344 ///
345 /// # #[tokio::main(flavor = "current_thread")]
346 /// # async fn main() -> Result<(), providers_kit::KitError> {
347 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
348 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
349 /// let result = provider.upload_file(providers_kit::files::FileUploadRequest::new("notes.txt", "text/plain", b"notes".to_vec())?).await;
350 /// let deletion = match &result { Ok(file) => provider.delete_file(file.id.clone()).await, Err(_) => Ok(()) };
351 /// let cleanup = provider.shutdown().await;
352 /// let _value = result?;
353 /// deletion?;
354 /// cleanup?;
355 ///
356 /// # Ok(())
357 /// # }
358 /// ```
359 ///
360 fn upload_file(
361 &self,
362 _request: FileUploadRequest,
363 ) -> impl Future<Output = Result<FileObject, KitError>> + Send + '_ {
364 async {
365 Err(KitError::unsupported(
366 ValidationViolation::Unsupported {
367 field: "files.upload",
368 rule: "operation is not implemented",
369 },
370 "provider file upload",
371 )
372 .at_operation(Operation::UploadFile))
373 }
374 }
375
376 /// Lists provider-managed files.
377 ///
378 /// This example requires a real configured service, valid credentials, and any exact
379 /// resource ids shown. Only delete or cancel resources the caller owns.
380 /// The default trait implementation returns Unsupported.
381 ///
382 /// # Errors
383 ///
384 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
385 ///
386 /// # Examples
387 ///
388 /// ```no_run
389 ///
390 /// # #[tokio::main(flavor = "current_thread")]
391 /// # async fn main() -> Result<(), providers_kit::KitError> {
392 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
393 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
394 /// let result = provider.list_files(providers_kit::files::FileListRequest::default()).await;
395 /// let cleanup = provider.shutdown().await;
396 /// let _value = result?;
397 /// cleanup?;
398 ///
399 /// # Ok(())
400 /// # }
401 /// ```
402 ///
403 fn list_files(
404 &self,
405 _request: FileListRequest,
406 ) -> impl Future<Output = Result<FilePage, KitError>> + Send + '_ {
407 async {
408 Err(KitError::unsupported(
409 ValidationViolation::Unsupported {
410 field: "files.list",
411 rule: "operation is not implemented",
412 },
413 "provider file listing",
414 )
415 .at_operation(Operation::ListFiles))
416 }
417 }
418
419 /// Fetches metadata for one provider-managed file.
420 ///
421 /// This example requires a real configured service, valid credentials, and any exact
422 /// resource ids shown. Only delete or cancel resources the caller owns.
423 /// The default trait implementation returns Unsupported.
424 ///
425 /// # Errors
426 ///
427 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
428 ///
429 /// # Examples
430 ///
431 /// ```no_run
432 ///
433 /// # #[tokio::main(flavor = "current_thread")]
434 /// # async fn main() -> Result<(), providers_kit::KitError> {
435 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
436 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
437 /// let result = provider.file_info("owned-file-id".into()).await;
438 /// let cleanup = provider.shutdown().await;
439 /// let _value = result?;
440 /// cleanup?;
441 ///
442 /// # Ok(())
443 /// # }
444 /// ```
445 ///
446 fn file_info(
447 &self,
448 _id: String,
449 ) -> impl Future<Output = Result<FileObject, KitError>> + Send + '_ {
450 async {
451 Err(KitError::unsupported(
452 ValidationViolation::Unsupported {
453 field: "files.info",
454 rule: "operation is not implemented",
455 },
456 "provider file metadata",
457 )
458 .at_operation(Operation::FileInfo))
459 }
460 }
461
462 /// Deletes one provider-managed file.
463 ///
464 /// This example requires a real configured service, valid credentials, and any exact
465 /// resource ids shown. Only delete or cancel resources the caller owns.
466 /// The default trait implementation returns Unsupported.
467 ///
468 /// # Errors
469 ///
470 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
471 ///
472 /// # Examples
473 ///
474 /// ```no_run
475 ///
476 /// # #[tokio::main(flavor = "current_thread")]
477 /// # async fn main() -> Result<(), providers_kit::KitError> {
478 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
479 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
480 /// let result = provider.delete_file("owned-file-id".into()).await;
481 /// let cleanup = provider.shutdown().await;
482 /// let _value = result?;
483 /// cleanup?;
484 ///
485 /// # Ok(())
486 /// # }
487 /// ```
488 ///
489 fn delete_file(&self, _id: String) -> impl Future<Output = Result<(), KitError>> + Send + '_ {
490 async {
491 Err(KitError::unsupported(
492 ValidationViolation::Unsupported {
493 field: "files.delete",
494 rule: "operation is not implemented",
495 },
496 "provider file deletion",
497 )
498 .at_operation(Operation::DeleteFile))
499 }
500 }
501
502 /// Downloads one provider-managed file under the caller's byte limit.
503 ///
504 /// This example requires a real configured service, valid credentials, and any exact
505 /// resource ids shown. Only delete or cancel resources the caller owns.
506 /// The default trait implementation returns Unsupported.
507 ///
508 /// # Errors
509 ///
510 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
511 ///
512 /// # Examples
513 ///
514 /// ```no_run
515 ///
516 /// # #[tokio::main(flavor = "current_thread")]
517 /// # async fn main() -> Result<(), providers_kit::KitError> {
518 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
519 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
520 /// let result = provider.download_file("owned-file-id".into(), std::num::NonZeroUsize::new(1024 * 1024).unwrap()).await;
521 /// let cleanup = provider.shutdown().await;
522 /// let _value = result?;
523 /// cleanup?;
524 ///
525 /// # Ok(())
526 /// # }
527 /// ```
528 ///
529 fn download_file(
530 &self,
531 _id: String,
532 _max_bytes: NonZeroUsize,
533 ) -> impl Future<Output = Result<Vec<u8>, KitError>> + Send + '_ {
534 async {
535 Err(KitError::unsupported(
536 ValidationViolation::Unsupported {
537 field: "files.download",
538 rule: "operation is not implemented",
539 },
540 "provider file download",
541 )
542 .at_operation(Operation::DownloadFile))
543 }
544 }
545
546 /// Fetches one batch job.
547 ///
548 /// Supplied OpenAI status reads opt into the configured bounded transport
549 /// retries; every attempt shares the original absolute request deadline.
550 /// This does not retry decoding or successful-response body failures. The
551 /// lookup used by `batch_results` is excluded. See the [batch workflow](crate::batch).
552 ///
553 /// This example requires a real configured service, valid credentials, and any exact
554 /// resource ids shown. Only delete or cancel resources the caller owns.
555 /// The default trait implementation returns Unsupported.
556 ///
557 /// # Errors
558 ///
559 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
560 ///
561 /// # Examples
562 ///
563 /// ```no_run
564 ///
565 /// # #[tokio::main(flavor = "current_thread")]
566 /// # async fn main() -> Result<(), providers_kit::KitError> {
567 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
568 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
569 /// let result = provider.batch_status("owned-batch-id".into()).await;
570 /// let cleanup = provider.shutdown().await;
571 /// let _value = result?;
572 /// cleanup?;
573 ///
574 /// # Ok(())
575 /// # }
576 /// ```
577 ///
578 fn batch_status(
579 &self,
580 _id: String,
581 ) -> impl Future<Output = Result<BatchJob, KitError>> + Send + '_ {
582 async {
583 Err(KitError::unsupported(
584 ValidationViolation::Unsupported {
585 field: "batch.status",
586 rule: "operation is not implemented",
587 },
588 "provider batch status",
589 )
590 .at_operation(Operation::BatchStatus))
591 }
592 }
593
594 /// Cancels one batch job.
595 ///
596 /// This example requires a real configured service, valid credentials, and any exact
597 /// resource ids shown. Only delete or cancel resources the caller owns.
598 /// The default trait implementation returns Unsupported.
599 ///
600 /// # Errors
601 ///
602 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
603 ///
604 /// # Examples
605 ///
606 /// ```no_run
607 ///
608 /// # #[tokio::main(flavor = "current_thread")]
609 /// # async fn main() -> Result<(), providers_kit::KitError> {
610 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
611 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
612 /// let result = provider.cancel_batch("owned-batch-id".into()).await;
613 /// let cleanup = provider.shutdown().await;
614 /// let _value = result?;
615 /// cleanup?;
616 ///
617 /// # Ok(())
618 /// # }
619 /// ```
620 ///
621 fn cancel_batch(
622 &self,
623 _id: String,
624 ) -> impl Future<Output = Result<BatchJob, KitError>> + Send + '_ {
625 async {
626 Err(KitError::unsupported(
627 ValidationViolation::Unsupported {
628 field: "batch.cancel",
629 rule: "operation is not implemented",
630 },
631 "provider batch cancellation",
632 )
633 .at_operation(Operation::CancelBatch))
634 }
635 }
636
637 /// Downloads and normalizes batch result rows under the caller's byte limit.
638 ///
639 /// This example requires a real configured service, valid credentials, and any exact
640 /// resource ids shown. Only delete or cancel resources the caller owns.
641 /// The default trait implementation returns Unsupported.
642 /// The byte bound covers the downloaded result; each terminal row must be inspected
643 /// independently. Row failures are result data, while download/normalization failures
644 /// are outer errors.
645 ///
646 /// # Errors
647 ///
648 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
649 ///
650 /// # Examples
651 ///
652 /// ```no_run
653 ///
654 /// # #[tokio::main(flavor = "current_thread")]
655 /// # async fn main() -> Result<(), providers_kit::KitError> {
656 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
657 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
658 /// let result = provider.batch_results("owned-batch-id".into(), std::num::NonZeroUsize::new(1024 * 1024).unwrap()).await;
659 /// let cleanup = provider.shutdown().await;
660 /// let _value = result?;
661 /// cleanup?;
662 ///
663 /// # Ok(())
664 /// # }
665 /// ```
666 ///
667 fn batch_results(
668 &self,
669 _id: String,
670 _max_bytes: NonZeroUsize,
671 ) -> impl Future<
672 Output = Result<Vec<BatchResult<<Self::Model as ModelBackend>::Continuation>>, KitError>,
673 > + Send
674 + '_ {
675 async {
676 Err(KitError::unsupported(
677 ValidationViolation::Unsupported {
678 field: "batch.results",
679 rule: "operation is not implemented",
680 },
681 "provider batch results",
682 )
683 .at_operation(Operation::BatchResults))
684 }
685 }
686
687 /// Lists provider batch jobs.
688 ///
689 /// This example requires a real configured service, valid credentials, and any exact
690 /// resource ids shown. Only delete or cancel resources the caller owns.
691 /// The default trait implementation returns Unsupported.
692 ///
693 /// # Errors
694 ///
695 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
696 ///
697 /// # Examples
698 ///
699 /// ```no_run
700 ///
701 /// # #[tokio::main(flavor = "current_thread")]
702 /// # async fn main() -> Result<(), providers_kit::KitError> {
703 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
704 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
705 /// let result = provider.list_batches(providers_kit::batch::BatchListRequest::default()).await;
706 /// let cleanup = provider.shutdown().await;
707 /// let _value = result?;
708 /// cleanup?;
709 ///
710 /// # Ok(())
711 /// # }
712 /// ```
713 ///
714 fn list_batches(
715 &self,
716 _request: BatchListRequest,
717 ) -> impl Future<Output = Result<BatchPage, KitError>> + Send + '_ {
718 async {
719 Err(KitError::unsupported(
720 ValidationViolation::Unsupported {
721 field: "batch.list",
722 rule: "operation is not implemented",
723 },
724 "provider batch listing",
725 )
726 .at_operation(Operation::ListBatches))
727 }
728 }
729
730 /// Probes provider runtime health.
731 ///
732 /// This example requires a real configured service, valid credentials, and any exact
733 /// resource ids shown. Only delete or cancel resources the caller owns.
734 /// The default trait implementation returns Unsupported.
735 ///
736 /// # Errors
737 ///
738 /// Returns [`KitError`] classified as Unsupported when unavailable, Closed after sealing, or the owning input, transport, deadline, service-rejection, or response-validation category. Inspect `kind()` and `context()` rather than diagnostic prose.
739 ///
740 /// # Examples
741 ///
742 /// ```no_run
743 ///
744 /// # #[tokio::main(flavor = "current_thread")]
745 /// # async fn main() -> Result<(), providers_kit::KitError> {
746 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, config::{ApiKey, OpenAiConfig, ProviderKind, TransportConfig}};
747 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::OpenAi(OpenAiConfig::new(ApiKey::new("replace-with-your-key")?)), TransportConfig::default()))?;
748 /// let result = provider.health().await;
749 /// let cleanup = provider.shutdown().await;
750 /// let _value = result?;
751 /// cleanup?;
752 ///
753 /// # Ok(())
754 /// # }
755 /// ```
756 ///
757 fn health(&self) -> impl Future<Output = Result<HealthStatus, KitError>> + Send + '_ {
758 async {
759 Err(KitError::unsupported(
760 ValidationViolation::Unsupported {
761 field: "runtime.health",
762 rule: "operation is not implemented",
763 },
764 "provider runtime health",
765 )
766 .at_operation(Operation::Health))
767 }
768 }
769
770 /// Shuts down backend-owned resources.
771 ///
772 /// Retain cleanup independently of each borrowed waiter. Repeated and concurrent
773 /// calls replay the same terminal result; last-handle drop is not an async cleanup API.
774 ///
775 /// # Errors
776 ///
777 /// Returns the shared backend cleanup failure, if any. A cancelled waiter does not reset shutdown or abandon cleanup.
778 ///
779 /// # Examples
780 ///
781 /// ```
782 ///
783 /// # #[tokio::main(flavor = "current_thread")]
784 /// # async fn main() -> Result<(), providers_kit::KitError> {
785 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, ModelBackend, ModelId, config::{ProviderKind, OllamaConfig, TransportConfig}};
786 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
787 /// let clone = provider.clone();
788 /// provider.shutdown().await?;
789 /// clone.shutdown().await?;
790 /// assert!(clone.model(ModelId::new("exact-model")?).is_err());
791 ///
792 /// # Ok(())
793 /// # }
794 /// ```
795 ///
796 fn shutdown(&self) -> impl Future<Output = Result<(), KitError>> + Send + '_;
797}
798
799enum ProviderInner {
800 OpenAi(OpenAiBackend),
801 Anthropic(AnthropicBackend),
802 Gemini(GeminiBackend),
803 Ollama(OllamaBackend),
804}
805
806impl ProviderBackend for ProviderInner {
807 type Model = ModelInner;
808
809 fn model(&self, id: ModelId) -> Result<ModelInner, KitError> {
810 match self {
811 Self::OpenAi(backend) => backend.model(id).map(ModelInner::OpenAi),
812 Self::Anthropic(backend) => backend.model(id).map(ModelInner::Anthropic),
813 Self::Gemini(backend) => backend.model(id).map(ModelInner::Gemini),
814 Self::Ollama(backend) => backend.model(id).map(ModelInner::Ollama),
815 }
816 }
817
818 async fn list_models(&self, request: ModelListRequest) -> Result<ModelPage, KitError> {
819 match self {
820 Self::OpenAi(backend) => backend.list_models(request).await,
821 Self::Anthropic(backend) => backend.list_models(request).await,
822 Self::Gemini(backend) => backend.list_models(request).await,
823 Self::Ollama(backend) => backend.list_models(request).await,
824 }
825 }
826
827 async fn model_info(&self, id: ModelId) -> Result<ModelSummary, KitError> {
828 match self {
829 Self::OpenAi(backend) => backend.model_info(id).await,
830 Self::Anthropic(backend) => backend.model_info(id).await,
831 Self::Gemini(backend) => backend.model_info(id).await,
832 Self::Ollama(backend) => backend.model_info(id).await,
833 }
834 }
835
836 async fn upload_file(&self, request: FileUploadRequest) -> Result<FileObject, KitError> {
837 match self {
838 Self::OpenAi(backend) => backend.upload_file(request).await,
839 Self::Anthropic(backend) => backend.upload_file(request).await,
840 Self::Gemini(backend) => backend.upload_file(request).await,
841 Self::Ollama(backend) => backend.upload_file(request).await,
842 }
843 }
844
845 async fn list_files(&self, request: FileListRequest) -> Result<FilePage, KitError> {
846 match self {
847 Self::OpenAi(backend) => backend.list_files(request).await,
848 Self::Anthropic(backend) => backend.list_files(request).await,
849 Self::Gemini(backend) => backend.list_files(request).await,
850 Self::Ollama(backend) => backend.list_files(request).await,
851 }
852 }
853
854 async fn file_info(&self, id: String) -> Result<FileObject, KitError> {
855 match self {
856 Self::OpenAi(backend) => backend.file_info(id).await,
857 Self::Anthropic(backend) => backend.file_info(id).await,
858 Self::Gemini(backend) => backend.file_info(id).await,
859 Self::Ollama(backend) => backend.file_info(id).await,
860 }
861 }
862
863 async fn delete_file(&self, id: String) -> Result<(), KitError> {
864 match self {
865 Self::OpenAi(backend) => backend.delete_file(id).await,
866 Self::Anthropic(backend) => backend.delete_file(id).await,
867 Self::Gemini(backend) => backend.delete_file(id).await,
868 Self::Ollama(backend) => backend.delete_file(id).await,
869 }
870 }
871
872 async fn download_file(
873 &self,
874 id: String,
875 max_bytes: NonZeroUsize,
876 ) -> Result<Vec<u8>, KitError> {
877 match self {
878 Self::OpenAi(backend) => backend.download_file(id, max_bytes).await,
879 Self::Anthropic(backend) => backend.download_file(id, max_bytes).await,
880 Self::Gemini(backend) => backend.download_file(id, max_bytes).await,
881 Self::Ollama(backend) => backend.download_file(id, max_bytes).await,
882 }
883 }
884
885 async fn batch_status(&self, id: String) -> Result<BatchJob, KitError> {
886 match self {
887 Self::OpenAi(backend) => backend.batch_status(id).await,
888 Self::Anthropic(backend) => backend.batch_status(id).await,
889 Self::Gemini(backend) => backend.batch_status(id).await,
890 Self::Ollama(backend) => backend.batch_status(id).await,
891 }
892 }
893
894 async fn cancel_batch(&self, id: String) -> Result<BatchJob, KitError> {
895 match self {
896 Self::OpenAi(backend) => backend.cancel_batch(id).await,
897 Self::Anthropic(backend) => backend.cancel_batch(id).await,
898 Self::Gemini(backend) => backend.cancel_batch(id).await,
899 Self::Ollama(backend) => backend.cancel_batch(id).await,
900 }
901 }
902
903 async fn batch_results(
904 &self,
905 id: String,
906 max_bytes: NonZeroUsize,
907 ) -> Result<Vec<BatchResult<GenerationContinuation>>, KitError> {
908 match self {
909 Self::OpenAi(backend) => backend.batch_results(id, max_bytes).await,
910 Self::Anthropic(backend) => backend.batch_results(id, max_bytes).await,
911 Self::Gemini(backend) => backend.batch_results(id, max_bytes).await,
912 Self::Ollama(backend) => backend.batch_results(id, max_bytes).await,
913 }
914 }
915
916 async fn list_batches(&self, request: BatchListRequest) -> Result<BatchPage, KitError> {
917 match self {
918 Self::OpenAi(backend) => backend.list_batches(request).await,
919 Self::Anthropic(backend) => backend.list_batches(request).await,
920 Self::Gemini(backend) => backend.list_batches(request).await,
921 Self::Ollama(backend) => backend.list_batches(request).await,
922 }
923 }
924
925 async fn health(&self) -> Result<HealthStatus, KitError> {
926 match self {
927 Self::OpenAi(backend) => backend.health().await,
928 Self::Anthropic(backend) => backend.health().await,
929 Self::Gemini(backend) => backend.health().await,
930 Self::Ollama(backend) => backend.health().await,
931 }
932 }
933
934 async fn shutdown(&self) -> Result<(), KitError> {
935 match self {
936 Self::OpenAi(backend) => backend.shutdown().await,
937 Self::Anthropic(backend) => backend.shutdown().await,
938 Self::Gemini(backend) => backend.shutdown().await,
939 Self::Ollama(backend) => backend.shutdown().await,
940 }
941 }
942}
943
944/// Cloneable opaque provider handle.
945///
946/// Provider and Model clones share one lifecycle. Shutdown seals model admission,
947/// drains accepted work, and publishes one terminal result to every waiter; cancelling
948/// a waiter cannot cancel owned cleanup. Models have no independent shutdown.
949/// Caller implementations retain their concrete associated children directly.
950///
951/// # Examples
952///
953/// ```
954///
955/// # #[tokio::main(flavor = "current_thread")]
956/// # async fn main() -> Result<(), providers_kit::KitError> {
957/// use providers_kit::{Provider, ProviderBackend, ProviderConfig, ModelBackend, ModelId, config::{ProviderKind, OllamaConfig, TransportConfig}};
958/// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
959/// let model = provider.model(ModelId::new("exact-model")?)?;
960/// assert_eq!(model.model_id().as_str(), "exact-model");
961/// provider.shutdown().await?;
962///
963/// # Ok(())
964/// # }
965/// ```
966///
967#[derive(Clone)]
968pub struct Provider {
969 inner: Arc<ProviderRuntime>,
970}
971
972pub(crate) struct ProviderRuntime {
973 backend: ProviderInner,
974 state: Mutex<ProviderLifecycleState>,
975 batches: Mutex<BatchRegistry>,
976 changed: Notify,
977}
978
979enum ProviderLifecycleState {
980 Open {
981 admitted: usize,
982 },
983 Closing {
984 admitted: usize,
985 task: Option<JoinHandle<Result<(), KitError>>>,
986 },
987 Closed {
988 terminal: Result<(), KitError>,
989 },
990}
991
992pub(crate) struct ProviderAdmission {
993 runtime: Arc<ProviderRuntime>,
994 batch_bytes: Option<usize>,
995}
996
997struct ProviderAdmittedStream {
998 inner: GenerateEventStream,
999 admission: Option<ProviderAdmission>,
1000}
1001
1002struct ShutdownTaskLease<'a> {
1003 runtime: &'a ProviderRuntime,
1004 task: Option<JoinHandle<Result<(), KitError>>>,
1005}
1006
1007impl ProviderRuntime {
1008 fn new(backend: ProviderInner) -> Self {
1009 Self {
1010 backend,
1011 state: Mutex::new(ProviderLifecycleState::Open { admitted: 0 }),
1012 batches: Mutex::new(BatchRegistry::new()),
1013 changed: Notify::new(),
1014 }
1015 }
1016
1017 pub(crate) fn admit(
1018 self: &Arc<Self>,
1019 operation: Operation,
1020 ) -> Result<ProviderAdmission, KitError> {
1021 let mut state = self.state();
1022 match &mut *state {
1023 ProviderLifecycleState::Open { admitted } => {
1024 *admitted = admitted.checked_add(1).ok_or_else(|| {
1025 KitError::invalid_state("provider admission is exhausted")
1026 .at_operation(operation)
1027 })?;
1028 Ok(ProviderAdmission {
1029 runtime: Arc::clone(self),
1030 batch_bytes: None,
1031 })
1032 }
1033 ProviderLifecycleState::Closing { .. } | ProviderLifecycleState::Closed { .. } => {
1034 Err(KitError::closed("provider admission closed").at_operation(operation))
1035 }
1036 }
1037 }
1038
1039 pub(crate) async fn run_owned_batch(
1040 &self,
1041 submission: OpenAiBatchSubmission,
1042 mut admission: ProviderAdmission,
1043 ) -> Result<BatchJob, KitError> {
1044 let handle = tokio::runtime::Handle::try_current().map_err(|_| {
1045 KitError::invalid_state("inline batch ownership requires a Tokio runtime")
1046 .at_operation(Operation::SubmitBatch)
1047 })?;
1048 let bytes = submission.payload_bytes();
1049 let (observer, activate) = {
1050 let mut batches = self.batches();
1051 batches.reserve(bytes)?;
1052 batches.register(submission, &handle)
1053 };
1054 // Preserve the original admission acquired before preparation. Even
1055 // shutdown sealing during rendering cannot harvest tasks until this
1056 // admission is released. Activation must follow registry lock release.
1057 admission.batch_bytes = Some(bytes);
1058 if let Err(admission) = activate.send(admission) {
1059 // The pre-spawn publication guard has already reported worker loss.
1060 // This release also returns the byte charge exactly once.
1061 drop(admission);
1062 }
1063 observer
1064 .wait()
1065 .await
1066 .map_err(|error| error.at_operation(Operation::SubmitBatch))
1067 }
1068
1069 async fn shutdown(self: &Arc<Self>) -> Result<(), KitError> {
1070 self.start_shutdown()?;
1071 self.terminal().await
1072 }
1073
1074 fn start_shutdown(self: &Arc<Self>) -> Result<(), KitError> {
1075 let mut state = self.state();
1076 let admitted = match &*state {
1077 ProviderLifecycleState::Open { admitted } => *admitted,
1078 ProviderLifecycleState::Closing { .. } | ProviderLifecycleState::Closed { .. } => {
1079 return Ok(());
1080 }
1081 };
1082 let handle = tokio::runtime::Handle::try_current()
1083 .map_err(|_| KitError::invalid_state("provider shutdown requires a Tokio runtime"))?;
1084 let runtime = Arc::clone(self);
1085 let task = handle.spawn(async move { runtime.finish_shutdown().await });
1086 *state = ProviderLifecycleState::Closing {
1087 admitted,
1088 task: Some(task),
1089 };
1090 Ok(())
1091 }
1092
1093 async fn finish_shutdown(self: Arc<Self>) -> Result<(), KitError> {
1094 self.drain().await;
1095 // Admissions cover preparation, registration, execution and cleanup.
1096 // Only after draining can no previously admitted call add another task.
1097 let mut tasks = self.batches().take_tasks();
1098 while let Some(result) = tasks.join_next().await {
1099 self.batches().joined(result);
1100 }
1101 let ownership = self.batches().terminal();
1102 let backend = self.backend.shutdown().await;
1103 match (ownership, backend) {
1104 (Ok(()), result) => result,
1105 (Err(primary), Ok(())) => Err(primary),
1106 (Err(primary), Err(cleanup)) => Err(primary.with_supplementary(cleanup)),
1107 }
1108 }
1109
1110 fn batches(&self) -> std::sync::MutexGuard<'_, BatchRegistry> {
1111 self.batches
1112 .lock()
1113 .unwrap_or_else(std::sync::PoisonError::into_inner)
1114 }
1115
1116 fn state(&self) -> std::sync::MutexGuard<'_, ProviderLifecycleState> {
1117 self.state
1118 .lock()
1119 .unwrap_or_else(std::sync::PoisonError::into_inner)
1120 }
1121
1122 fn release(&self) {
1123 let should_notify = {
1124 let mut state = self.state();
1125 match &mut *state {
1126 ProviderLifecycleState::Open { admitted } => {
1127 debug_assert_ne!(*admitted, 0);
1128 *admitted = admitted.saturating_sub(1);
1129 false
1130 }
1131 ProviderLifecycleState::Closing { admitted, .. } => {
1132 debug_assert_ne!(*admitted, 0);
1133 *admitted = admitted.saturating_sub(1);
1134 *admitted == 0
1135 }
1136 ProviderLifecycleState::Closed { .. } => {
1137 debug_assert!(false, "closed provider retained an admission");
1138 false
1139 }
1140 }
1141 };
1142 if should_notify {
1143 self.changed.notify_waiters();
1144 }
1145 }
1146
1147 async fn drain(&self) {
1148 loop {
1149 let notified = self.changed.notified();
1150 let drained = match &*self.state() {
1151 ProviderLifecycleState::Open { admitted }
1152 | ProviderLifecycleState::Closing { admitted, .. } => *admitted == 0,
1153 ProviderLifecycleState::Closed { .. } => true,
1154 };
1155 if drained {
1156 return;
1157 }
1158 notified.await;
1159 }
1160 }
1161
1162 async fn terminal(&self) -> Result<(), KitError> {
1163 loop {
1164 let notified = self.changed.notified();
1165 let mut lease = {
1166 let mut state = self.state();
1167 match &mut *state {
1168 ProviderLifecycleState::Closed { terminal } => return terminal.clone(),
1169 ProviderLifecycleState::Closing { task, .. } => {
1170 task.take().map(|task| ShutdownTaskLease::new(self, task))
1171 }
1172 ProviderLifecycleState::Open { .. } => {
1173 return Err(KitError::invalid_state(
1174 "provider shutdown lost lifecycle ownership",
1175 ));
1176 }
1177 }
1178 };
1179 let Some(lease) = lease.as_mut() else {
1180 notified.await;
1181 continue;
1182 };
1183 let terminal = lease.join().await;
1184 lease.complete(terminal.clone());
1185 return terminal;
1186 }
1187 }
1188}
1189
1190impl<'a> ShutdownTaskLease<'a> {
1191 fn new(runtime: &'a ProviderRuntime, task: JoinHandle<Result<(), KitError>>) -> Self {
1192 Self {
1193 runtime,
1194 task: Some(task),
1195 }
1196 }
1197
1198 async fn join(&mut self) -> Result<(), KitError> {
1199 let Some(task) = self.task.as_mut() else {
1200 return Err(KitError::invalid_state(
1201 "provider shutdown task lease was empty",
1202 ));
1203 };
1204 task.await.unwrap_or_else(|_| {
1205 Err(KitError::invalid_state(
1206 "provider shutdown owner could not be joined",
1207 ))
1208 })
1209 }
1210
1211 fn complete(&mut self, terminal: Result<(), KitError>) {
1212 self.task = None;
1213 let mut state = self.runtime.state();
1214 if matches!(&*state, ProviderLifecycleState::Closing { task: None, .. }) {
1215 *state = ProviderLifecycleState::Closed { terminal };
1216 drop(state);
1217 self.runtime.changed.notify_waiters();
1218 }
1219 }
1220}
1221
1222impl Drop for ShutdownTaskLease<'_> {
1223 fn drop(&mut self) {
1224 let Some(task) = self.task.take() else {
1225 return;
1226 };
1227 let mut state = self.runtime.state();
1228 if let ProviderLifecycleState::Closing { task: slot, .. } = &mut *state
1229 && slot.is_none()
1230 {
1231 *slot = Some(task);
1232 drop(state);
1233 self.runtime.changed.notify_waiters();
1234 return;
1235 }
1236 debug_assert!(false, "shutdown task lease lost lifecycle ownership");
1237 }
1238}
1239
1240impl Drop for ProviderAdmission {
1241 fn drop(&mut self) {
1242 if let Some(bytes) = self.batch_bytes {
1243 self.runtime.batches().release(bytes);
1244 }
1245 self.runtime.release();
1246 }
1247}
1248
1249impl ProviderAdmission {
1250 pub(crate) fn bind(self, stream: GenerateEventStream) -> GenerateEventStream {
1251 GenerateEventStream::new(ProviderAdmittedStream {
1252 inner: stream,
1253 admission: Some(self),
1254 })
1255 }
1256}
1257
1258impl Stream for ProviderAdmittedStream {
1259 type Item = Result<GenerateEvent, KitError>;
1260
1261 fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
1262 let polled = Pin::new(&mut self.inner).poll_next(context);
1263 if matches!(
1264 &polled,
1265 Poll::Ready(None)
1266 | Poll::Ready(Some(Err(_)))
1267 | Poll::Ready(Some(Ok(GenerateEvent::Completed(_))))
1268 ) {
1269 drop(self.admission.take());
1270 }
1271 polled.map(|item| {
1272 item.map(|result| result.map_err(|error| error.at_operation(Operation::GenerateStream)))
1273 })
1274 }
1275}
1276
1277impl Provider {
1278 /// Constructs a built-in provider from validated configuration.
1279 ///
1280 /// Construction creates clients locally and does not probe credentials or model availability.
1281 ///
1282 /// # Errors
1283 ///
1284 /// Returns a configuration or client-construction error for invalid endpoints, transport settings, or HTTP/TLS client setup. Service authentication is checked by subsequent operations.
1285 ///
1286 /// # Examples
1287 ///
1288 /// ```
1289 ///
1290 /// # #[tokio::main(flavor = "current_thread")]
1291 /// # async fn main() -> Result<(), providers_kit::KitError> {
1292 /// use providers_kit::{Provider, ProviderBackend, ProviderConfig, ModelBackend, ModelId, config::{ProviderKind, OllamaConfig, TransportConfig}};
1293 /// let provider = Provider::from_config(ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
1294 /// let model = provider.model(ModelId::new("exact-model")?)?;
1295 /// assert_eq!(model.model_id().as_str(), "exact-model");
1296 /// provider.shutdown().await?;
1297 ///
1298 /// # Ok(())
1299 /// # }
1300 /// ```
1301 ///
1302 pub fn from_config(config: ProviderConfig) -> Result<Self, KitError> {
1303 let (kind, transport) = config.into_parts();
1304 let backend = match kind {
1305 ProviderKind::OpenAi(config) => ProviderInner::OpenAi(
1306 OpenAiBackend::from_config(config, transport)
1307 .map_err(|error| error.at_operation(Operation::Configure))?,
1308 ),
1309 ProviderKind::Anthropic(config) => ProviderInner::Anthropic(
1310 AnthropicBackend::from_config(config, transport)
1311 .map_err(|error| error.at_operation(Operation::Configure))?,
1312 ),
1313 ProviderKind::Gemini(config) => ProviderInner::Gemini(
1314 GeminiBackend::from_config(config, transport)
1315 .map_err(|error| error.at_operation(Operation::Configure))?,
1316 ),
1317 ProviderKind::Ollama(config) => ProviderInner::Ollama(
1318 OllamaBackend::from_config(config, transport)
1319 .map_err(|error| error.at_operation(Operation::Configure))?,
1320 ),
1321 };
1322 Ok(Self {
1323 inner: Arc::new(ProviderRuntime::new(backend)),
1324 })
1325 }
1326}
1327
1328impl ProviderBackend for Provider {
1329 type Model = Model;
1330
1331 fn model(&self, id: ModelId) -> Result<Model, KitError> {
1332 let _admission = self.inner.admit(Operation::SelectModel)?;
1333 let model = self
1334 .inner
1335 .backend
1336 .model(id)
1337 .map_err(|error| error.at_operation(Operation::SelectModel))?;
1338 Ok(Model::new(model, Arc::clone(&self.inner)))
1339 }
1340
1341 async fn list_models(&self, request: ModelListRequest) -> Result<ModelPage, KitError> {
1342 let _admission = self.inner.admit(Operation::ListModels)?;
1343 self.inner
1344 .backend
1345 .list_models(request)
1346 .await
1347 .map_err(|error| error.at_operation(Operation::ListModels))
1348 }
1349
1350 async fn model_info(&self, id: ModelId) -> Result<ModelSummary, KitError> {
1351 let _admission = self.inner.admit(Operation::ModelInfo)?;
1352 self.inner
1353 .backend
1354 .model_info(id)
1355 .await
1356 .map_err(|error| error.at_operation(Operation::ModelInfo))
1357 }
1358
1359 async fn upload_file(&self, request: FileUploadRequest) -> Result<FileObject, KitError> {
1360 let _admission = self.inner.admit(Operation::UploadFile)?;
1361 self.inner
1362 .backend
1363 .upload_file(request)
1364 .await
1365 .map_err(|error| error.at_operation(Operation::UploadFile))
1366 }
1367
1368 async fn list_files(&self, request: FileListRequest) -> Result<FilePage, KitError> {
1369 let _admission = self.inner.admit(Operation::ListFiles)?;
1370 self.inner
1371 .backend
1372 .list_files(request)
1373 .await
1374 .map_err(|error| error.at_operation(Operation::ListFiles))
1375 }
1376
1377 async fn file_info(&self, id: String) -> Result<FileObject, KitError> {
1378 let _admission = self.inner.admit(Operation::FileInfo)?;
1379 self.inner
1380 .backend
1381 .file_info(id)
1382 .await
1383 .map_err(|error| error.at_operation(Operation::FileInfo))
1384 }
1385
1386 async fn delete_file(&self, id: String) -> Result<(), KitError> {
1387 let _admission = self.inner.admit(Operation::DeleteFile)?;
1388 self.inner
1389 .backend
1390 .delete_file(id)
1391 .await
1392 .map_err(|error| error.at_operation(Operation::DeleteFile))
1393 }
1394
1395 async fn download_file(
1396 &self,
1397 id: String,
1398 max_bytes: NonZeroUsize,
1399 ) -> Result<Vec<u8>, KitError> {
1400 let _admission = self.inner.admit(Operation::DownloadFile)?;
1401 self.inner
1402 .backend
1403 .download_file(id, max_bytes)
1404 .await
1405 .map_err(|error| error.at_operation(Operation::DownloadFile))
1406 }
1407
1408 async fn batch_status(&self, id: String) -> Result<BatchJob, KitError> {
1409 let _admission = self.inner.admit(Operation::BatchStatus)?;
1410 self.inner
1411 .backend
1412 .batch_status(id)
1413 .await
1414 .map_err(|error| error.at_operation(Operation::BatchStatus))
1415 }
1416
1417 async fn cancel_batch(&self, id: String) -> Result<BatchJob, KitError> {
1418 let _admission = self.inner.admit(Operation::CancelBatch)?;
1419 self.inner
1420 .backend
1421 .cancel_batch(id)
1422 .await
1423 .map_err(|error| error.at_operation(Operation::CancelBatch))
1424 }
1425
1426 async fn batch_results(
1427 &self,
1428 id: String,
1429 max_bytes: NonZeroUsize,
1430 ) -> Result<Vec<BatchResult<GenerationContinuation>>, KitError> {
1431 let _admission = self.inner.admit(Operation::BatchResults)?;
1432 self.inner
1433 .backend
1434 .batch_results(id, max_bytes)
1435 .await
1436 .map_err(|error| error.at_operation(Operation::BatchResults))
1437 }
1438
1439 async fn list_batches(&self, request: BatchListRequest) -> Result<BatchPage, KitError> {
1440 let _admission = self.inner.admit(Operation::ListBatches)?;
1441 self.inner
1442 .backend
1443 .list_batches(request)
1444 .await
1445 .map_err(|error| error.at_operation(Operation::ListBatches))
1446 }
1447
1448 async fn health(&self) -> Result<HealthStatus, KitError> {
1449 let _admission = self.inner.admit(Operation::Health)?;
1450 self.inner
1451 .backend
1452 .health()
1453 .await
1454 .map_err(|error| error.at_operation(Operation::Health))
1455 }
1456
1457 async fn shutdown(&self) -> Result<(), KitError> {
1458 self.inner
1459 .shutdown()
1460 .await
1461 .map_err(|error| error.at_operation(Operation::Shutdown))
1462 }
1463}
1464
1465impl fmt::Debug for Provider {
1466 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1467 formatter.debug_struct("Provider").finish_non_exhaustive()
1468 }
1469}