rpi_plugin_sdk/
provider.rs

1//! Provider packages use the same C ABI as tools. This bridge owns a runtime
2//! per request by default; opt-in contexts reuse resources until lifecycle shutdown.
3//!
4//! Available with the optional `provider` feature. Use [`crate::export_provider!`]
5//! for single-provider plugins, or [`register`] and [`drive`] when registering
6//! multiple providers through one plugin entrypoint.
7use crate::{
8    PluginApi, ProviderRequestFn, ProviderStreamControl, ProviderStreamEmitFn,
9    ProviderStreamRequestFn, StbString, StbStringRef,
10};
11use rpi_ai::{
12    AssistantMessage, AssistantMessageEvent, Context, Model, Provider, SimpleStreamOptions,
13    StopReason,
14};
15use std::{ffi::c_void, sync::Arc, time::Duration};
16
17mod context;
18pub use context::{drive_with_context, register_with_context, ContextFactory, ProviderContext};
19
20pub type Factory = fn(&Model, &SimpleStreamOptions) -> Arc<dyn Provider>;
21
22/// Reuse the maintained OpenAI-compatible protocol without importing any
23/// vendor credentials from the host process environment. Client construction
24/// errors propagate without bypassing the configured transport.
25pub fn completions(
26    model: &Model,
27    options: &SimpleStreamOptions,
28    anonymous: bool,
29) -> Result<Arc<dyn Provider>, String> {
30    let client = rpi_ai::http::build_client(&model.base_url, None, None)?;
31    if anonymous && options.api_key.is_none() {
32        Ok(Arc::new(rpi_ai::providers::openai_completions::OpenAiCompletionsProvider::with_models_without_auth(model.provider.clone(), client, vec![model.clone()])))
33    } else {
34        Ok(Arc::new(rpi_ai::providers::openai_completions::OpenAiCompletionsProvider::with_models_without_env_api_key(model.provider.clone(), options.api_key.clone(), client, vec![model.clone()])))
35    }
36}
37
38/// Declare a catalog and register one implementation with a one-shot fallback.
39pub fn register(
40    api: &PluginApi,
41    id: &str,
42    base_url: &str,
43    style: &str,
44    manifest: serde_json::Value,
45    request: ProviderRequestFn,
46    stream: ProviderStreamRequestFn,
47) -> i32 {
48    if let Some(declare) = api.declare {
49        let json = serde_json::json!({"providers":[manifest]}).to_string();
50        let rc = declare(StbStringRef::from_str(&json));
51        if rc != 0 {
52            return rc;
53        }
54    }
55    let free = crate::callbacks::free_string;
56    if let Some(register) = api.register_streaming_provider {
57        register(
58            StbStringRef::from_str(id),
59            StbStringRef::from_str(base_url),
60            StbStringRef::from_str(style),
61            stream,
62            free,
63            std::ptr::null_mut(),
64        )
65    } else if let Some(register) = api.register_provider {
66        register(
67            StbStringRef::from_str(id),
68            StbStringRef::from_str(base_url),
69            StbStringRef::from_str(style),
70            request,
71            free,
72            std::ptr::null_mut(),
73        )
74    } else {
75        1
76    }
77}
78
79#[derive(serde::Deserialize)]
80struct Request {
81    model: Model,
82    context: Context,
83    options: serde_json::Value,
84}
85
86fn options(value: &serde_json::Value) -> SimpleStreamOptions {
87    let mut options = SimpleStreamOptions::default();
88    options.api_key = value["apiKey"].as_str().map(str::to_string);
89    options.timeout = value["timeoutMs"].as_u64().map(Duration::from_millis);
90    options.stream_idle_timeout = value["streamIdleTimeoutMs"]
91        .as_u64()
92        .map(Duration::from_millis);
93    options.max_retry_delay = value["maxRetryDelayMs"].as_u64().map(Duration::from_millis);
94    options.max_retries = value["maxRetries"].as_u64().and_then(|n| n.try_into().ok());
95    options.max_tokens = value["maxTokens"].as_u64();
96    options.temperature = value["temperature"].as_f64();
97    options.session_id = value["sessionId"].as_str().map(str::to_string);
98    options.headers = serde_json::from_value(value["headers"].clone()).ok();
99    options.metadata = serde_json::from_value(value["metadata"].clone()).ok();
100    options.reasoning = value["reasoning"]
101        .as_str()
102        .and_then(|s| serde_json::from_value(serde_json::json!(s.to_ascii_lowercase())).ok());
103    options.thinking_budgets = serde_json::from_value(value["thinkingBudgets"].clone()).ok();
104    options.cache_retention = match value["cacheRetention"].as_str() {
105        Some("None") => rpi_ai::CacheRetention::None,
106        Some("Long") => rpi_ai::CacheRetention::Long,
107        _ => rpi_ai::CacheRetention::Short,
108    };
109    options
110}
111
112/// Compare every field without constructing a candidate message or JSON tree.
113fn reconstructs_text(
114    previous: &AssistantMessage,
115    next: &AssistantMessage,
116    index: usize,
117    delta: &str,
118) -> bool {
119    let AssistantMessage {
120        role,
121        content,
122        api,
123        provider,
124        model,
125        response_model,
126        response_id,
127        usage,
128        stop_reason,
129        deferred,
130        error_message,
131        raw_stop_reason,
132        end_turn,
133        timestamp,
134    } = previous;
135    if role != &next.role
136        || api != &next.api
137        || provider != &next.provider
138        || model != &next.model
139        || response_model != &next.response_model
140        || response_id != &next.response_id
141        || usage != &next.usage
142        || stop_reason != &next.stop_reason
143        || deferred != &next.deferred
144        || error_message != &next.error_message
145        || raw_stop_reason != &next.raw_stop_reason
146        || end_turn != &next.end_turn
147        || timestamp != &next.timestamp
148        || content.len() != next.content.len()
149    {
150        return false;
151    }
152    let (Some(rpi_ai::Content::Text(before)), Some(rpi_ai::Content::Text(after))) =
153        (content.get(index), next.content.get(index))
154    else {
155        return false;
156    };
157    let rpi_ai::TextContent {
158        kind,
159        text,
160        text_signature,
161    } = before;
162    kind == &after.kind
163        && text_signature == &after.text_signature
164        && after.text.strip_prefix(text) == Some(delta)
165        && content
166            .iter()
167            .zip(&next.content)
168            .enumerate()
169            .all(|(i, (a, b))| i == index || a == b)
170}
171
172fn event_json(
173    event: &AssistantMessageEvent,
174    previous: &mut Option<(usize, Arc<AssistantMessage>)>,
175) -> serde_json::Value {
176    let compact = match (event, previous.as_ref()) {
177        (
178            AssistantMessageEvent::TextDelta {
179                content_index,
180                delta,
181                partial,
182            },
183            Some((index, before)),
184        ) => index == content_index && reconstructs_text(before, partial, *index, delta),
185        _ => false,
186    };
187    *previous = match event {
188        AssistantMessageEvent::TextDelta {
189            content_index,
190            partial,
191            ..
192        } => Some((*content_index, Arc::clone(partial))),
193        _ => None,
194    };
195    if compact {
196        let AssistantMessageEvent::TextDelta {
197            content_index,
198            delta,
199            ..
200        } = event
201        else {
202            unreachable!()
203        };
204        return serde_json::json!({"type":"text_delta","contentIndex":content_index,"delta":delta});
205    }
206    let mut chunk = serde_json::to_value(event).expect("provider event JSON");
207    // The SDK uses flat tool-call identity fields.
208    let tool = chunk.get("toolCall").cloned().or_else(|| {
209        if event.type_tag() != "toolcall_start" {
210            return None;
211        }
212        let index = chunk["contentIndex"].as_u64()? as usize;
213        chunk.get("partial")?.get("content")?.get(index).cloned()
214    });
215    if let Some(tool) = tool {
216        for field in ["id", "name", "arguments"] {
217            if let Some(value) = tool.get(field) {
218                chunk[field] = value.clone();
219            }
220        }
221    }
222    chunk
223}
224
225async fn drive_stream(
226    provider: Arc<dyn Provider>,
227    request: &Request,
228    options: &SimpleStreamOptions,
229    emit: Option<ProviderStreamEmitFn>,
230    emit_data: *mut c_void,
231    control: Option<&ProviderStreamControl>,
232) -> AssistantMessage {
233    let mut stream = provider
234        .stream_simple(&request.model, &request.context, options)
235        .await;
236    let mut partial = Arc::new(AssistantMessage::empty(
237        request.model.api.clone(),
238        request.model.provider.clone(),
239        request.model.id.clone(),
240        0,
241    ));
242    let mut previous_text = None;
243    let mut poll = tokio::time::interval(Duration::from_millis(10));
244    loop {
245        // Check synchronously too: an already-buffered provider can
246        // finish before the interval's initial tick becomes ready.
247        if options.signal.is_cancelled()
248            || control.is_some_and(|control| (control.is_cancelled)(control.user_data, 0) != 0)
249        {
250            options.signal.cancel();
251            let mut partial = (*partial).clone();
252            partial.stop_reason = StopReason::Aborted;
253            partial.error_message = Some("provider request was cancelled".into());
254            break partial;
255        }
256        tokio::select! {
257            biased;
258            _ = poll.tick() => {}
259            event = stream.next() => {
260                let Some(event) = event else {
261                    break stream.result().await.unwrap_or_else(|error| AssistantMessage::terminal(
262                        request.model.api.clone(), request.model.provider.clone(), request.model.id.clone(), StopReason::Error, error.to_string(), 0));
263                };
264                match &event {
265                    AssistantMessageEvent::Done { message, .. } => break message.clone(),
266                    AssistantMessageEvent::Error { error, .. } => break error.clone(),
267                    _ => {},
268                }
269                partial = match &event {
270                    AssistantMessageEvent::Start { partial }
271                    | AssistantMessageEvent::TextStart { partial, .. }
272                    | AssistantMessageEvent::TextDelta { partial, .. }
273                    | AssistantMessageEvent::TextEnd { partial, .. }
274                    | AssistantMessageEvent::ThinkingStart { partial, .. }
275                    | AssistantMessageEvent::ThinkingDelta { partial, .. }
276                    | AssistantMessageEvent::ThinkingEnd { partial, .. }
277                    | AssistantMessageEvent::ToolCallStart { partial, .. }
278                    | AssistantMessageEvent::ToolCallDelta { partial, .. }
279                    | AssistantMessageEvent::ToolCallEnd { partial, .. } => Arc::clone(partial),
280                    AssistantMessageEvent::Done { .. } | AssistantMessageEvent::Error { .. } => unreachable!(),
281                };
282                let Some(emit) = emit else { continue; };
283                let chunk = event_json(&event, &mut previous_text);
284                if emit(StbString::from_string(chunk.to_string()), emit_data) != 0 {
285                    options.signal.cancel();
286                    let mut partial = (*partial).clone();
287                    partial.stop_reason = StopReason::Aborted;
288                    break partial;
289                }
290            }
291        }
292    }
293}
294
295/// Drive a synchronous plugin request without retaining any borrowed ABI data.
296///
297/// # Safety
298/// All arguments must satisfy the SDK request/callback lifetime contract. `out`
299/// must be writable; any `control` and callback context must outlive this call.
300pub unsafe fn drive(
301    factory: Factory,
302    request: StbStringRef,
303    emit: Option<ProviderStreamEmitFn>,
304    emit_data: *mut c_void,
305    control: *const ProviderStreamControl,
306    out: *mut StbString,
307) -> i32 {
308    crate::guard_or(1, || {
309        if out.is_null() {
310            return 2;
311        }
312        let Ok(request) = serde_json::from_str::<Request>(unsafe { request.as_str() }) else {
313            return 2;
314        };
315        let options = options(&request.options);
316        let Ok(runtime) = tokio::runtime::Builder::new_current_thread()
317            .enable_all()
318            .build()
319        else {
320            return 1;
321        };
322        let control = unsafe { control.as_ref() };
323        let message = runtime.block_on(async {
324            let provider = factory(&request.model, &options);
325            drive_stream(provider, &request, &options, emit, emit_data, control).await
326        });
327        // Convert via Value to normalize tagged Content's field representation.
328        let json = serde_json::to_value(message)
329            .expect("assistant message JSON")
330            .to_string();
331        unsafe {
332            *out = StbString::from_string(json);
333        }
334        0
335    })
336}
337
338/// Export a provider package using an ordinary Rust Provider factory.
339///
340/// Requires the `provider` feature. For a multi-provider package, use one
341/// [`crate::export_plugin!`] entrypoint and call [`register`] for each provider
342/// instead of exporting this macro multiple times.
343#[macro_export]
344macro_rules! export_provider {
345    ($id:expr, $base:expr, $api:expr, $manifest:expr, $factory:path) => {
346        extern "C" fn request(
347            req: $crate::StbStringRef,
348            out: *mut $crate::StbString,
349            _: *mut std::ffi::c_void,
350        ) -> i32 {
351            unsafe {
352                $crate::provider::drive(
353                    $factory,
354                    req,
355                    None,
356                    std::ptr::null_mut(),
357                    std::ptr::null(),
358                    out,
359                )
360            }
361        }
362        extern "C" fn stream(
363            req: $crate::StbStringRef,
364            emit: Option<$crate::ProviderStreamEmitFn>,
365            data: *mut std::ffi::c_void,
366            control: *const $crate::ProviderStreamControl,
367            _: Option<$crate::ProviderStreamCompletionFn>,
368            _: *mut std::ffi::c_void,
369            out: *mut $crate::StbString,
370            _: *mut std::ffi::c_void,
371        ) -> i32 {
372            unsafe { $crate::provider::drive($factory, req, emit, data, control, out) }
373        }
374        $crate::export_plugin!(|api| $crate::provider::register(
375            api, $id, $base, $api, $manifest, request, stream
376        ));
377    };
378}
379
380#[cfg(test)]
381mod bridge_tests;
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    fn factory(_: &Model, _: &SimpleStreamOptions) -> Arc<dyn Provider> {
387        rpi_ai::providers::faux::FauxProvider::new(
388            rpi_ai::providers::faux::FauxScript::new()
389                .with_tool_call("echo", serde_json::json!({"value":"hello"})),
390        )
391    }
392    extern "C" fn collect(raw: StbString, data: *mut c_void) -> i32 {
393        let string = unsafe {
394            crate::callbacks::OwnedString::from_raw(raw, Some(crate::callbacks::free_string))
395        };
396        let value = serde_json::from_str(&string.to_string_lossy()).unwrap();
397        // The bridge calls emit synchronously on this request thread only.
398        unsafe { &mut *data.cast::<Vec<serde_json::Value>>() }.push(value);
399        0
400    }
401    fn request() -> String {
402        serde_json::json!({"model":Model::new("m","M",rpi_ai::Api::Faux,"probe","http://localhost"),
403            "context":Context::new(vec![]),"options":{}})
404        .to_string()
405    }
406    #[test]
407    fn transport_construction_errors_return_terminal_output_with_or_without_emit() {
408        fn factory(
409            _: &ProviderContext,
410            model: &Model,
411            options: &SimpleStreamOptions,
412        ) -> Result<Arc<dyn Provider>, String> {
413            completions(model, options, false)
414        }
415        let model = Model::new(
416            "m",
417            "M",
418            rpi_ai::Api::OpenaiCompletions,
419            "probe",
420            ":invalid",
421        );
422        assert!(completions(&model, &SimpleStreamOptions::default(), false).is_err());
423        let request =
424            serde_json::json!({"model":model, "context":Context::new(vec![]), "options":{}})
425                .to_string();
426        for with_emit in [false, true] {
427            let mut chunks = Vec::<serde_json::Value>::new();
428            let mut out = StbString::empty();
429            assert_eq!(
430                unsafe {
431                    drive_with_context(
432                        factory,
433                        std::ptr::null_mut(),
434                        StbStringRef::from_str(&request),
435                        with_emit.then_some(collect as ProviderStreamEmitFn),
436                        (&mut chunks as *mut Vec<serde_json::Value>).cast(),
437                        std::ptr::null(),
438                        &mut out,
439                    )
440                },
441                0
442            );
443            let output = unsafe {
444                crate::callbacks::OwnedString::from_raw(out, Some(crate::callbacks::free_string))
445            };
446            let message: AssistantMessage =
447                serde_json::from_str(&output.to_string_lossy()).unwrap();
448            assert_eq!(message.stop_reason, StopReason::Error);
449            assert!(message
450                .error_message
451                .as_deref()
452                .unwrap()
453                .contains("invalid target URL"));
454            assert!(chunks.is_empty(), "terminal errors use the output slot");
455        }
456    }
457
458    #[test]
459    fn bridge_emits_tool_identity_and_valid_terminal_json() {
460        let mut chunks = Vec::<serde_json::Value>::new();
461        let mut out = StbString::empty();
462        let rc = unsafe {
463            drive(
464                factory,
465                StbStringRef::from_str(&request()),
466                Some(collect),
467                (&mut chunks as *mut Vec<serde_json::Value>).cast(),
468                std::ptr::null(),
469                &mut out,
470            )
471        };
472        assert_eq!(rc, 0);
473        let output = unsafe {
474            crate::callbacks::OwnedString::from_raw(out, Some(crate::callbacks::free_string))
475        };
476        let message: AssistantMessage = serde_json::from_str(&output.to_string_lossy()).unwrap();
477        assert_eq!(message.stop_reason, StopReason::ToolUse);
478        for kind in ["toolcall_start", "toolcall_end"] {
479            let chunk = chunks
480                .iter()
481                .find(|chunk| chunk["type"] == kind)
482                .expect(kind);
483            assert_eq!(chunk["name"], "echo");
484            assert!(!chunk["id"].as_str().unwrap().is_empty());
485        }
486        let end = chunks
487            .iter()
488            .find(|chunk| chunk["type"] == "toolcall_end")
489            .unwrap();
490        assert_eq!(end["arguments"]["value"], "hello");
491        assert!(!chunks
492            .iter()
493            .any(|chunk| chunk["type"] == "done" || chunk["type"] == "error"));
494    }
495    #[test]
496    fn bridge_honors_the_host_control_surface() {
497        extern "C" fn cancelled(_: *mut c_void, _: u32) -> i32 {
498            1
499        }
500        let control = ProviderStreamControl {
501            is_cancelled: cancelled,
502            wait_cancelled_ms: cancelled,
503            user_data: std::ptr::null_mut(),
504        };
505        let mut out = StbString::empty();
506        assert_eq!(
507            unsafe {
508                drive(
509                    factory,
510                    StbStringRef::from_str(&request()),
511                    None,
512                    std::ptr::null_mut(),
513                    &control,
514                    &mut out,
515                )
516            },
517            0
518        );
519        let output = unsafe {
520            crate::callbacks::OwnedString::from_raw(out, Some(crate::callbacks::free_string))
521        };
522        let message: AssistantMessage = serde_json::from_str(&output.to_string_lossy()).unwrap();
523        assert_eq!(message.stop_reason, StopReason::Aborted);
524    }
525    #[test]
526    fn options_preserve_provider_controls() {
527        let options = options(
528            &serde_json::json!({"apiKey":"key","timeoutMs":42,"streamIdleTimeoutMs":11,
529            "maxRetries":3,"maxRetryDelayMs":17,"reasoning":"Xhigh","maxTokens":80,
530            "headers":{"x-custom":"value"},"cacheRetention":"Long","sessionId":"s"}),
531        );
532        assert_eq!(options.api_key.as_deref(), Some("key"));
533        assert_eq!(options.timeout, Some(Duration::from_millis(42)));
534        assert_eq!(options.stream_idle_timeout, Some(Duration::from_millis(11)));
535        assert_eq!(options.max_retries, Some(3));
536        assert_eq!(options.max_retry_delay, Some(Duration::from_millis(17)));
537        assert_eq!(options.reasoning, Some(rpi_ai::ThinkingLevel::Xhigh));
538        assert_eq!(options.max_tokens, Some(80));
539        assert_eq!(options.session_id.as_deref(), Some("s"));
540        assert_eq!(options.headers.unwrap()["x-custom"], "value");
541    }
542}
rpi_plugin_sdk/provider.rs

rpi_plugin_sdk/
provider.rs
