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, ProviderRequestEvent},
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::MissingMoonshotApiKey
34        | ModelError::MissingOpenRouterApiKey
35        | ModelError::MissingKimiAuth
36        | ModelError::MissingXaiApiKey
37        | ModelError::MissingXaiAuth => ProviderError::new(
38            ProviderErrorKind::Authentication,
39            error.to_string(),
40            Retryability::Permanent,
41        ),
42        ModelError::Credentials(_) => ProviderError::new(
43            ProviderErrorKind::Authentication,
44            "credential store operation failed",
45            Retryability::Permanent,
46        ),
47        ModelError::Interrupted => ProviderError::interrupted("provider stream interrupted"),
48        ModelError::StreamIdleTimeout { timeout } => ProviderError::new(
49            ProviderErrorKind::Timeout,
50            format!(
51                "provider stream received no data for {timeout:?}; the connection may be stale"
52            ),
53            Retryability::Retryable,
54        ),
55        ModelError::StreamFailedAfterOutput { message } => ProviderError::new(
56            ProviderErrorKind::InvalidResponse,
57            "provider stream failed after emitting output",
58            Retryability::Permanent,
59        )
60        .with_diagnostic(sanitize_diagnostic(&message)),
61        ModelError::InvalidResponse(details) => ProviderError::new(
62            ProviderErrorKind::InvalidResponse,
63            "provider returned an invalid response",
64            Retryability::Permanent,
65        )
66        .with_diagnostic(sanitize_diagnostic(&details)),
67        ModelError::ProviderReported {
68            kind,
69            error_type,
70            message,
71        } => {
72            let (provider_kind, public_message, retryability) = match kind {
73                crate::model::ProviderReportedErrorKind::Timeout => (
74                    ProviderErrorKind::Timeout,
75                    "provider reported a timeout",
76                    Retryability::Retryable,
77                ),
78                crate::model::ProviderReportedErrorKind::RateLimit => (
79                    ProviderErrorKind::RateLimit,
80                    "provider reported a rate limit",
81                    Retryability::Retryable,
82                ),
83                crate::model::ProviderReportedErrorKind::Unavailable => (
84                    ProviderErrorKind::Unavailable,
85                    "provider reported a temporary failure",
86                    Retryability::Retryable,
87                ),
88                crate::model::ProviderReportedErrorKind::InvalidResponse => (
89                    ProviderErrorKind::InvalidResponse,
90                    "provider reported an invalid response",
91                    Retryability::Permanent,
92                ),
93            };
94            ProviderError::new(provider_kind, public_message, retryability).with_diagnostic(
95                sanitize_diagnostic(&format!("{error_type}: {message}")),
96            )
97        }
98        ModelError::UnsupportedReasoning {
99            provider,
100            model,
101            requested,
102        } => ProviderError::new(
103            ProviderErrorKind::Other,
104            format!(
105                "provider '{provider}' model '{model}' does not support reasoning level '{requested}'"
106            ),
107            Retryability::Permanent,
108        ),
109        ModelError::UnsupportedProvider(provider) => ProviderError::new(
110            ProviderErrorKind::Other,
111            format!("unsupported provider '{provider}'"),
112            Retryability::Permanent,
113        ),
114        ModelError::HttpStatus { status, body } => {
115            let status_code = status.as_u16();
116            let (kind, retryability) = match status_code {
117                401 | 403 => (ProviderErrorKind::Authentication, Retryability::Permanent),
118                408 | 504 => (ProviderErrorKind::Timeout, Retryability::Retryable),
119                429 => (ProviderErrorKind::RateLimit, Retryability::Retryable),
120                500..=599 => (ProviderErrorKind::Unavailable, Retryability::Retryable),
121                _ => (ProviderErrorKind::Other, Retryability::Permanent),
122            };
123            let error = ProviderError::new(kind, format!("HTTP {status_code}"), retryability);
124            if body.is_empty() {
125                error
126            } else {
127                error.with_diagnostic(sanitize_diagnostic(&body))
128            }
129        }
130        ModelError::Request(_) => ProviderError::new(
131            ProviderErrorKind::Unavailable,
132            "provider request failed",
133            Retryability::Retryable,
134        ),
135        ModelError::Io(_) => ProviderError::new(
136            ProviderErrorKind::Other,
137            "provider I/O failed",
138            Retryability::Retryable,
139        ),
140    }
141}
142
143fn sanitize_diagnostic(value: &str) -> String {
144    const MAX_BYTES: usize = crate::provider_backend::http_error::MAX_ERROR_BODY_BYTES;
145
146    let mut diagnostic = String::new();
147    let mut truncated = false;
148    for character in value.chars() {
149        let escaped = match character {
150            '\n' | '\t' => character.to_string(),
151            character if character.is_control() => character.escape_default().to_string(),
152            character => character.to_string(),
153        };
154        if diagnostic.len() + escaped.len() > MAX_BYTES {
155            truncated = true;
156            break;
157        }
158        diagnostic.push_str(&escaped);
159    }
160    if truncated {
161        diagnostic.push_str("\n[diagnostic truncated]");
162    }
163    diagnostic
164}
165
166/// Event buffered by the application callback adapter.
167#[doc(hidden)]
168#[derive(Debug)]
169pub enum CallbackEvent {
170    Model(ModelEvent),
171    Request(ProviderRequestEvent),
172}
173
174/// Shared queue used by the callback sinks and [`drive_callback_stream`].
175pub type CallbackEventQueue = Arc<Mutex<VecDeque<CallbackEvent>>>;
176
177/// Builds the synchronous callback used by application stream transports.
178///
179/// Events are buffered temporarily because the callback cannot await. The
180/// companion [`drive_callback_stream`] loop drains that buffer through the
181/// SDK's bounded event sender before polling the provider again.
182pub fn callback_event_sink(
183    cancellation: CancellationToken,
184    pending: CallbackEventQueue,
185    notify: Arc<Notify>,
186) -> impl FnMut(ModelEvent) -> Result<(), ModelError> + Send {
187    move |event| {
188        if cancellation.is_cancelled() {
189            return Err(ModelError::Interrupted);
190        }
191        pending
192            .lock()
193            .unwrap_or_else(|poisoned| poisoned.into_inner())
194            .push_back(CallbackEvent::Model(event));
195        notify.notify_one();
196        Ok(())
197    }
198}
199
200/// Builds the synchronous physical request callback used by application transports.
201pub fn callback_request_event_sink(
202    cancellation: CancellationToken,
203    pending: CallbackEventQueue,
204    notify: Arc<Notify>,
205) -> impl FnMut(ProviderRequestEvent) -> Result<(), ModelError> + Send {
206    move |event| {
207        if cancellation.is_cancelled() {
208            return Err(ModelError::Interrupted);
209        }
210        pending
211            .lock()
212            .unwrap_or_else(|poisoned| poisoned.into_inner())
213            .push_back(CallbackEvent::Request(event));
214        notify.notify_one();
215        Ok(())
216    }
217}
218
219/// Drains buffered callback events through the bounded SDK event channel and
220/// drives the provider future with host backpressure across awaits.
221pub async fn drive_callback_stream<Fut>(
222    cancellation: CancellationToken,
223    events: ProviderEventSender,
224    pending: CallbackEventQueue,
225    notify: Arc<Notify>,
226    provider: Fut,
227) -> Result<ModelResponse, ProviderError>
228where
229    Fut: Future<Output = Result<ModelResponse, ModelError>>,
230{
231    let mut provider = std::pin::pin!(provider);
232    let mut provider_result: Option<Result<ModelResponse, ModelError>> = None;
233
234    loop {
235        loop {
236            let next = pending
237                .lock()
238                .unwrap_or_else(|poisoned| poisoned.into_inner())
239                .pop_front();
240            let Some(event) = next else {
241                break;
242            };
243            match event {
244                CallbackEvent::Model(event) => events.send(event).await?,
245                CallbackEvent::Request(ProviderRequestEvent::RequestAttemptFailed {
246                    kind,
247                    usage,
248                }) => {
249                    events.send_request_attempt_failed(kind, usage).await?;
250                }
251            }
252        }
253
254        if let Some(result) = provider_result.take() {
255            return result.map_err(provider_error_from_model_error);
256        }
257
258        let notified = notify.notified();
259        let has_pending = !pending
260            .lock()
261            .unwrap_or_else(|poisoned| poisoned.into_inner())
262            .is_empty();
263        if has_pending {
264            continue;
265        }
266
267        tokio::select! {
268            biased;
269            () = notified => {}
270            () = cancellation.cancelled() => {
271                return Err(ProviderError::interrupted("provider stream interrupted"));
272            }
273            result = &mut provider => {
274                provider_result = Some(result);
275            }
276        }
277    }
278}
279
280/// Implements [`rho_sdk::provider::ModelProvider`] for an application transport
281/// that already exposes inherent `model_identity`, `complete_turn`, and
282/// `stream_turn` methods.
283///
284/// Streaming buffers same-poll callback bursts, then applies the SDK event
285/// channel's async backpressure before polling the provider again.
286#[macro_export]
287macro_rules! impl_sdk_model_provider {
288    ($provider:ty) => {
289        impl ::rho_sdk::provider::ModelProvider for $provider {
290            fn cancellation_mode(&self) -> ::rho_sdk::provider::ProviderCancellationMode {
291                ::rho_sdk::provider::ProviderCancellationMode::Cooperative
292            }
293
294            fn identity(&self) -> ::rho_sdk::model::ModelIdentity {
295                self.model_identity()
296            }
297
298            fn send_turn<'a>(
299                &'a self,
300                request: ::rho_sdk::model::ModelRequest<'a>,
301            ) -> ::rho_sdk::provider::ProviderFuture<'a> {
302                ::std::boxed::Box::pin(async move {
303                    self.complete_turn(request)
304                        .await
305                        .map_err($crate::providers::sdk_contract::provider_error_from_model_error)
306                })
307            }
308
309            fn send_turn_stream<'a>(
310                &'a self,
311                request: ::rho_sdk::model::ModelRequest<'a>,
312                events: ::rho_sdk::provider::ProviderEventSender,
313            ) -> ::rho_sdk::provider::ProviderFuture<'a> {
314                ::std::boxed::Box::pin(async move {
315                    let cancellation = request.cancellation.clone();
316                    let pending = ::std::sync::Arc::new(::std::sync::Mutex::new(
317                        ::std::collections::VecDeque::new(),
318                    ));
319                    let notify = ::std::sync::Arc::new(::tokio::sync::Notify::new());
320                    let mut on_event = $crate::providers::sdk_contract::callback_event_sink(
321                        cancellation.clone(),
322                        ::std::sync::Arc::clone(&pending),
323                        ::std::sync::Arc::clone(&notify),
324                    );
325                    let mut on_request_event =
326                        $crate::providers::sdk_contract::callback_request_event_sink(
327                            cancellation.clone(),
328                            ::std::sync::Arc::clone(&pending),
329                            ::std::sync::Arc::clone(&notify),
330                        );
331                    let provider = self.stream_turn(request, &mut on_event, &mut on_request_event);
332                    $crate::providers::sdk_contract::drive_callback_stream(
333                        cancellation,
334                        events,
335                        pending,
336                        notify,
337                        provider,
338                    )
339                    .await
340                })
341            }
342        }
343    };
344}
345
346#[cfg(test)]
347#[path = "sdk_contract_tests.rs"]
348mod tests;
rho_coding_agent/providers/sdk_contract.rs

rho_coding_agent/providers/
sdk_contract.rs
