rho_coding_agent/providers/
sdk_contract.rs

1//! Shared helpers for exposing application transports through the public SDK
2//! provider contract.
3//!
4//! Built-in providers implement [`rho_sdk::provider::ModelProvider`] directly.
5//! Callback-based stream transports remain an internal detail and are bridged
6//! here into the SDK's bounded async event sender.
7
8use std::{
9    collections::VecDeque,
10    future::Future,
11    sync::{Arc, Mutex},
12};
13
14use rho_sdk::{
15    model::{ModelEvent, ModelResponse},
16    provider::ProviderEventSender,
17    CancellationToken, ProviderError, ProviderErrorKind, Retryability,
18};
19use tokio::sync::Notify;
20
21use crate::model::ModelError;
22
23/// Converts an application [`ModelError`] into a sanitized public [`ProviderError`].
24///
25/// HTTP response bodies and other transport payloads are omitted so credentials
26/// and provider-private content do not enter the SDK error contract.
27pub fn provider_error_from_model_error(error: ModelError) -> ProviderError {
28    match error {
29        ModelError::MissingApiKey
30        | ModelError::MissingCodexAuth
31        | ModelError::MissingAnthropicApiKey
32        | ModelError::MissingGithubCopilotAuth
33        | ModelError::MissingXaiAuth => ProviderError::new(
34            ProviderErrorKind::Authentication,
35            error.to_string(),
36            Retryability::Permanent,
37        ),
38        ModelError::Credentials(_) => ProviderError::new(
39            ProviderErrorKind::Authentication,
40            "credential store operation failed",
41            Retryability::Permanent,
42        ),
43        ModelError::Interrupted => ProviderError::interrupted("provider stream interrupted"),
44        ModelError::StreamIdleTimeout { timeout } => ProviderError::new(
45            ProviderErrorKind::Timeout,
46            format!(
47                "provider stream received no data for {timeout:?}; the connection may be stale"
48            ),
49            Retryability::Retryable,
50        ),
51        ModelError::StreamFailedAfterOutput { message } => ProviderError::new(
52            ProviderErrorKind::InvalidResponse,
53            "provider stream failed after emitting output",
54            Retryability::Permanent,
55        )
56        .with_diagnostic(sanitize_diagnostic(&message)),
57        ModelError::InvalidResponse(details) => ProviderError::new(
58            ProviderErrorKind::InvalidResponse,
59            "provider returned an invalid response",
60            Retryability::Permanent,
61        )
62        .with_diagnostic(sanitize_diagnostic(&details)),
63        ModelError::UnsupportedReasoning {
64            provider,
65            model,
66            requested,
67        } => ProviderError::new(
68            ProviderErrorKind::Other,
69            format!(
70                "provider '{provider}' model '{model}' does not support reasoning level '{requested}'"
71            ),
72            Retryability::Permanent,
73        ),
74        ModelError::UnsupportedProvider(provider) => ProviderError::new(
75            ProviderErrorKind::Other,
76            format!("unsupported provider '{provider}'"),
77            Retryability::Permanent,
78        ),
79        ModelError::HttpStatus { status, body } => {
80            let status_code = status.as_u16();
81            let (kind, retryability) = match status_code {
82                401 | 403 => (ProviderErrorKind::Authentication, Retryability::Permanent),
83                408 | 504 => (ProviderErrorKind::Timeout, Retryability::Retryable),
84                429 => (ProviderErrorKind::RateLimit, Retryability::Retryable),
85                500..=599 => (ProviderErrorKind::Unavailable, Retryability::Retryable),
86                _ => (ProviderErrorKind::Other, Retryability::Permanent),
87            };
88            let error = ProviderError::new(kind, format!("HTTP {status_code}"), retryability);
89            if body.is_empty() {
90                error
91            } else {
92                error.with_diagnostic(sanitize_diagnostic(&body))
93            }
94        }
95        ModelError::Request(_) => ProviderError::new(
96            ProviderErrorKind::Unavailable,
97            "provider request failed",
98            Retryability::Retryable,
99        ),
100        ModelError::Io(_) => ProviderError::new(
101            ProviderErrorKind::Other,
102            "provider I/O failed",
103            Retryability::Retryable,
104        ),
105    }
106}
107
108fn sanitize_diagnostic(value: &str) -> String {
109    const MAX_BYTES: usize = crate::provider_backend::http_error::MAX_ERROR_BODY_BYTES;
110
111    let mut diagnostic = String::new();
112    let mut truncated = false;
113    for character in value.chars() {
114        let escaped = match character {
115            '\n' | '\t' => character.to_string(),
116            character if character.is_control() => character.escape_default().to_string(),
117            character => character.to_string(),
118        };
119        if diagnostic.len() + escaped.len() > MAX_BYTES {
120            truncated = true;
121            break;
122        }
123        diagnostic.push_str(&escaped);
124    }
125    if truncated {
126        diagnostic.push_str("\n[diagnostic truncated]");
127    }
128    diagnostic
129}
130
131/// Shared queue used by [`callback_event_sink`] and [`drive_callback_stream`].
132pub type CallbackEventQueue = Arc<Mutex<VecDeque<ModelEvent>>>;
133
134/// Builds the synchronous callback used by application stream transports.
135///
136/// Events are buffered temporarily because the callback cannot await. The
137/// companion [`drive_callback_stream`] loop drains that buffer through the
138/// SDK's bounded event sender before polling the provider again.
139pub fn callback_event_sink(
140    cancellation: CancellationToken,
141    pending: CallbackEventQueue,
142    notify: Arc<Notify>,
143) -> impl FnMut(ModelEvent) -> Result<(), ModelError> + Send {
144    move |event| {
145        if cancellation.is_cancelled() {
146            return Err(ModelError::Interrupted);
147        }
148        pending
149            .lock()
150            .unwrap_or_else(|poisoned| poisoned.into_inner())
151            .push_back(event);
152        notify.notify_one();
153        Ok(())
154    }
155}
156
157/// Drains buffered callback events through the bounded SDK event channel and
158/// drives the provider future with host backpressure across awaits.
159pub async fn drive_callback_stream<Fut>(
160    cancellation: CancellationToken,
161    events: ProviderEventSender,
162    pending: CallbackEventQueue,
163    notify: Arc<Notify>,
164    provider: Fut,
165) -> Result<ModelResponse, ProviderError>
166where
167    Fut: Future<Output = Result<ModelResponse, ModelError>>,
168{
169    let mut provider = std::pin::pin!(provider);
170    let mut provider_result: Option<Result<ModelResponse, ModelError>> = None;
171
172    loop {
173        loop {
174            let next = pending
175                .lock()
176                .unwrap_or_else(|poisoned| poisoned.into_inner())
177                .pop_front();
178            let Some(event) = next else {
179                break;
180            };
181            if cancellation.is_cancelled() {
182                return Err(ProviderError::interrupted("provider stream interrupted"));
183            }
184            events.send(event).await?;
185        }
186
187        if let Some(result) = provider_result.take() {
188            return result.map_err(provider_error_from_model_error);
189        }
190
191        let notified = notify.notified();
192        let has_pending = !pending
193            .lock()
194            .unwrap_or_else(|poisoned| poisoned.into_inner())
195            .is_empty();
196        if has_pending {
197            continue;
198        }
199
200        tokio::select! {
201            biased;
202            () = notified => {}
203            () = cancellation.cancelled() => {
204                return Err(ProviderError::interrupted("provider stream interrupted"));
205            }
206            result = &mut provider => {
207                provider_result = Some(result);
208            }
209        }
210    }
211}
212
213/// Implements [`rho_sdk::provider::ModelProvider`] for an application transport
214/// that already exposes inherent `model_identity`, `complete_turn`, and
215/// `stream_turn` methods.
216///
217/// Streaming buffers same-poll callback bursts, then applies the SDK event
218/// channel's async backpressure before polling the provider again.
219#[macro_export]
220macro_rules! impl_sdk_model_provider {
221    ($provider:ty) => {
222        impl ::rho_sdk::provider::ModelProvider for $provider {
223            fn identity(&self) -> ::rho_sdk::model::ModelIdentity {
224                self.model_identity()
225            }
226
227            fn send_turn<'a>(
228                &'a self,
229                request: ::rho_sdk::model::ModelRequest<'a>,
230            ) -> ::rho_sdk::provider::ProviderFuture<'a> {
231                ::std::boxed::Box::pin(async move {
232                    self.complete_turn(request)
233                        .await
234                        .map_err($crate::providers::sdk_contract::provider_error_from_model_error)
235                })
236            }
237
238            fn send_turn_stream<'a>(
239                &'a self,
240                request: ::rho_sdk::model::ModelRequest<'a>,
241                events: ::rho_sdk::provider::ProviderEventSender,
242            ) -> ::rho_sdk::provider::ProviderFuture<'a> {
243                ::std::boxed::Box::pin(async move {
244                    let cancellation = request.cancellation.clone();
245                    let pending = ::std::sync::Arc::new(::std::sync::Mutex::new(
246                        ::std::collections::VecDeque::new(),
247                    ));
248                    let notify = ::std::sync::Arc::new(::tokio::sync::Notify::new());
249                    let mut on_event = $crate::providers::sdk_contract::callback_event_sink(
250                        cancellation.clone(),
251                        ::std::sync::Arc::clone(&pending),
252                        ::std::sync::Arc::clone(&notify),
253                    );
254                    let provider = self.stream_turn(request, &mut on_event);
255                    $crate::providers::sdk_contract::drive_callback_stream(
256                        cancellation,
257                        events,
258                        pending,
259                        notify,
260                        provider,
261                    )
262                    .await
263                })
264            }
265        }
266    };
267}
268
269#[cfg(test)]
270#[path = "sdk_contract_tests.rs"]
271mod tests;
rho_coding_agent/providers/sdk_contract.rs

rho_coding_agent/providers/
sdk_contract.rs
