semantic_operators/
cascade.rs

1//! Uncertainty-only escalation and optional per-call deadlines.
2use crate::{Answers, Error, ErrorKind, Provider, ProviderFuture, Request, Result, ask};
3use std::sync::Arc;
4
5/// Ordered providers; only uncertain successful answers advance to the next hop.
6pub struct Cascade {
7    providers: Vec<Arc<dyn Provider>>,
8    escalate_below: f64,
9}
10impl Cascade {
11    /// Create a cascade of at least two providers with a finite `[0,1]` threshold.
12    ///
13    /// # Errors
14    /// Returns `InvalidArgument` if fewer than two providers are supplied or
15    /// `escalate_below` is nonfinite or outside `[0, 1]`.
16    pub fn new(providers: Vec<Arc<dyn Provider>>, escalate_below: f64) -> Result<Self> {
17        if providers.len() < 2
18            || !escalate_below.is_finite()
19            || !(0.0..=1.0).contains(&escalate_below)
20        {
21            return Err(Error::new(
22                ErrorKind::InvalidArgument,
23                "cascade needs two providers and a threshold from zero to one",
24            ));
25        }
26        Ok(Self {
27            providers,
28            escalate_below,
29        })
30    }
31}
32impl Provider for Cascade {
33    fn ask<'a>(&'a self, request: &'a Request) -> ProviderFuture<'a> {
34        Box::pin(async move {
35            let mut pending = request.questions().clone();
36            let mut results = Answers::new();
37            for provider in &self.providers {
38                let hop = request.with_questions(pending.clone())?;
39                let answers = ask(provider, &hop).await?;
40                pending.retain(|name, _| {
41                    answers.get(name).is_some_and(|answer| {
42                        !answer.decided() || answer.confidence() < self.escalate_below
43                    })
44                });
45                results.extend(answers);
46                if pending.is_empty() {
47                    break;
48                }
49            }
50            crate::provider::validate_answers(request, results)
51        })
52    }
53}
54/// Bound a call with Tokio's timer. Requires a time-enabled Tokio runtime.
55/// Cancellation is local; the server may still finish and bill the request.
56#[cfg(feature = "tokio")]
57#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
58pub struct TimeLimited<P> {
59    provider: P,
60    duration: std::time::Duration,
61}
62#[cfg(feature = "tokio")]
63impl<P: Provider> TimeLimited<P> {
64    /// Reject zero durations; timer starts when the future is polled.
65    ///
66    /// # Errors
67    /// Returns `InvalidArgument` for a zero duration. Runtime/timer availability
68    /// is checked when a call is polled, not during construction.
69    pub fn new(provider: P, duration: std::time::Duration) -> Result<Self> {
70        if duration.is_zero() {
71            return Err(Error::new(
72                ErrorKind::InvalidArgument,
73                "timeout must be positive",
74            ));
75        }
76        Ok(Self { provider, duration })
77    }
78}
79#[cfg(feature = "tokio")]
80impl<P: Provider> Provider for TimeLimited<P> {
81    fn ask<'a>(&'a self, request: &'a Request) -> ProviderFuture<'a> {
82        Box::pin(async move {
83            crate::provider::deadline(self.duration, ask(&self.provider, request)).await?
84        })
85    }
86}
semantic_operators/cascade.rs

semantic_operators/
cascade.rs
