ferrin_core/realtime/
mod.rs

1//! Realtime (WebSocket) sessions.
2//!
3//! A [`RealtimeSession`] connects to a provider's realtime endpoint through
4//! the provider's [`RealtimeModel`](ferrin_spec::RealtimeModel): the model
5//! issues the client secret, describes the WebSocket handshake and
6//! translates between wire messages and the standardized
7//! [`RealtimeClientEvent`] / [`RealtimeServerEvent`] sets. The session owns
8//! the connection, forwards server events as a stream, and executes local
9//! function tools when the model calls them.
10//!
11//! Design: `docs/01-architecture/11-other-modalities.md` §9.
12
13mod session;
14mod tools;
15
16use std::future::IntoFuture;
17use std::pin::Pin;
18use std::sync::Arc;
19use std::task::Context;
20use std::task::Poll;
21use std::time::Duration;
22
23use ferrin_spec::JsonValue;
24use ferrin_spec::RealtimeModelRef;
25use ferrin_spec::realtime_model::ClientSecretOptions;
26use ferrin_tool::ToolSet;
27use futures_core::Stream;
28use tokio::sync::mpsc;
29use tokio::task::JoinSet;
30use tokio_util::sync::CancellationToken;
31
32pub use ferrin_spec::realtime_model::ClientSecret;
33pub use ferrin_spec::realtime_model::ConversationItem;
34pub use ferrin_spec::realtime_model::ConversationRole;
35pub use ferrin_spec::realtime_model::Modality;
36pub use ferrin_spec::realtime_model::RealtimeClientEvent;
37pub use ferrin_spec::realtime_model::RealtimeServerEvent;
38pub use ferrin_spec::realtime_model::RealtimeSessionConfig;
39pub use ferrin_spec::realtime_model::RealtimeToolDefinition;
40pub use ferrin_spec::realtime_model::ResponseCreateOptions;
41pub use ferrin_spec::realtime_model::TranscriptionConfig;
42pub use ferrin_spec::realtime_model::TurnDetection;
43pub use ferrin_spec::realtime_model::TurnDetectionKind;
44pub use session::RealtimeHandle;
45pub use tools::realtime_tool_definitions;
46
47use crate::error::Error;
48use crate::registry::ProviderRegistry;
49use crate::registry::default::resolve_model;
50
51/// Default capacity of the server event buffer.
52const DEFAULT_EVENT_BUFFER: usize = 256;
53
54/// How long [`RealtimeSession::close`] waits for the connection task.
55const CLOSE_TIMEOUT: Duration = Duration::from_secs(5);
56
57/// Items of the event stream: standardized server events, or local failures
58/// (transport errors, unparsable messages, tool execution errors).
59pub type RealtimeEvent = Result<RealtimeServerEvent, Error>;
60
61/// Starts building a realtime session with `model`.
62///
63/// Call [`RealtimeSessionBuilder::connect`] (or `.await` the builder) to open
64/// the connection.
65#[must_use]
66pub fn realtime_session(model: impl Into<RealtimeModelRef>) -> RealtimeSessionBuilder {
67    RealtimeSessionBuilder {
68        model: model.into(),
69        client_secret: None,
70        expires_after_seconds: None,
71        config: RealtimeSessionConfig::default(),
72        tools: ToolSet::new(),
73        tools_context: None,
74        cancellation: CancellationToken::new(),
75        event_buffer: DEFAULT_EVENT_BUFFER,
76    }
77}
78
79/// Configuration of a realtime session before it connects.
80#[derive(Debug)]
81pub struct RealtimeSessionBuilder {
82    model: RealtimeModelRef,
83    client_secret: Option<ClientSecret>,
84    expires_after_seconds: Option<u64>,
85    config: RealtimeSessionConfig,
86    tools: ToolSet,
87    tools_context: Option<JsonValue>,
88    cancellation: CancellationToken,
89    event_buffer: usize,
90}
91
92impl RealtimeSessionBuilder {
93    /// Uses an existing client secret instead of creating one through the
94    /// model.
95    #[must_use]
96    pub fn client_secret(mut self, secret: ClientSecret) -> Self {
97        self.client_secret = Some(secret);
98        self
99    }
100
101    /// Requested lifetime of the client secret created on connect.
102    #[must_use]
103    pub fn expires_after_seconds(mut self, seconds: u64) -> Self {
104        self.expires_after_seconds = Some(seconds);
105        self
106    }
107
108    /// Sets the initial session configuration. Tool definitions derived from
109    /// [`tools`](Self::tools) are appended to `config.tools` on connect.
110    #[must_use]
111    pub fn config(mut self, config: RealtimeSessionConfig) -> Self {
112        self.config = config;
113        self
114    }
115
116    /// Sets the system instructions of the session.
117    #[must_use]
118    pub fn instructions(mut self, instructions: impl Into<String>) -> Self {
119        self.config.instructions = Some(instructions.into());
120        self
121    }
122
123    /// Sets the voice used for audio output.
124    #[must_use]
125    pub fn voice(mut self, voice: impl Into<String>) -> Self {
126        self.config.voice = Some(voice.into());
127        self
128    }
129
130    /// Local tools. Function and dynamic tools are advertised to the model;
131    /// executable ones run inside the session and their outputs are sent
132    /// back automatically. Tools without an executor are advertised only;
133    /// the application answers them with [`RealtimeHandle::add_tool_output`].
134    #[must_use]
135    pub fn tools(mut self, tools: ToolSet) -> Self {
136        self.tools = tools;
137        self
138    }
139
140    /// Context passed to dynamic tool descriptions and executions.
141    #[must_use]
142    pub fn tools_context(mut self, context: JsonValue) -> Self {
143        self.tools_context = Some(context);
144        self
145    }
146
147    /// Cancellation token; cancelling it closes the session.
148    #[must_use]
149    pub fn cancellation(mut self, cancellation: CancellationToken) -> Self {
150        self.cancellation = cancellation;
151        self
152    }
153
154    /// Capacity of the server event buffer (default 256). Reading from the
155    /// connection pauses while the buffer is full.
156    #[must_use]
157    pub fn event_buffer(mut self, capacity: usize) -> Self {
158        self.event_buffer = capacity.max(1);
159        self
160    }
161
162    /// Opens the connection and sends the initial `session-update`.
163    ///
164    /// # Errors
165    ///
166    /// Fails when the client secret cannot be created, the WebSocket
167    /// handshake fails, a tool context is invalid or the initial event
168    /// cannot be serialized.
169    pub async fn connect(self) -> Result<RealtimeSession, Error> {
170        let model = resolve_model(&self.model, ProviderRegistry::realtime_model)?;
171        let mut config = self.config;
172        let definitions =
173            realtime_tool_definitions(&self.tools, self.tools_context.as_ref()).await?;
174        config.tools.extend(definitions);
175
176        let secret = match self.client_secret {
177            Some(secret) => secret,
178            None => model
179                .do_create_client_secret(ClientSecretOptions {
180                    expires_after_seconds: self.expires_after_seconds,
181                    session_config: Some(config.clone()),
182                })
183                .await
184                .map_err(Error::from)?,
185        };
186
187        let (events_tx, events_rx) = mpsc::channel(self.event_buffer);
188        let mut tasks = JoinSet::new();
189        let handle = session::start(
190            session::StartOptions {
191                model,
192                secret,
193                config,
194                tools: Arc::new(self.tools),
195                tools_context: self.tools_context,
196                cancellation: self.cancellation,
197                events: events_tx,
198            },
199            &mut tasks,
200        )
201        .await?;
202        Ok(RealtimeSession {
203            handle,
204            events: events_rx,
205            tasks,
206        })
207    }
208}
209
210impl IntoFuture for RealtimeSessionBuilder {
211    type Output = Result<RealtimeSession, Error>;
212    type IntoFuture = futures_util::future::BoxFuture<'static, Self::Output>;
213
214    fn into_future(self) -> Self::IntoFuture {
215        Box::pin(self.connect())
216    }
217}
218
219/// An open realtime session.
220///
221/// The session is a [`Stream`] of [`RealtimeEvent`]s. Sending happens through
222/// [`RealtimeSession::send`] or a cloned [`RealtimeHandle`], which stays
223/// usable while the session is being polled from another task. The stream
224/// ends when the connection closes.
225#[derive(Debug)]
226pub struct RealtimeSession {
227    handle: RealtimeHandle,
228    events: mpsc::Receiver<RealtimeEvent>,
229    tasks: JoinSet<()>,
230}
231
232impl RealtimeSession {
233    /// Returns a cloneable handle for sending events and closing the session.
234    #[must_use]
235    pub fn handle(&self) -> RealtimeHandle {
236        self.handle.clone()
237    }
238
239    /// Sends a client event.
240    ///
241    /// # Errors
242    ///
243    /// Fails when the event cannot be serialized or the session is closed.
244    pub async fn send(&self, event: RealtimeClientEvent) -> Result<(), Error> {
245        self.handle.send(event).await
246    }
247
248    /// Sends a user text message and requests a response.
249    ///
250    /// # Errors
251    ///
252    /// See [`RealtimeSession::send`].
253    pub async fn send_text(&self, text: impl Into<String>) -> Result<(), Error> {
254        self.handle.send_text(text).await
255    }
256
257    /// Submits the output of a tool call the application executed itself.
258    ///
259    /// # Errors
260    ///
261    /// See [`RealtimeSession::send`].
262    pub async fn add_tool_output(&self, call_id: &str, output: &JsonValue) -> Result<(), Error> {
263        self.handle.add_tool_output(call_id, output).await
264    }
265
266    /// Returns the next event, or `None` once the connection is closed.
267    pub async fn next_event(&mut self) -> Option<RealtimeEvent> {
268        self.events.recv().await
269    }
270
271    /// Borrows the session as a stream of events.
272    pub fn events(&mut self) -> impl Stream<Item = RealtimeEvent> + Send + '_ {
273        self
274    }
275
276    /// Returns `true` once the connection task has finished.
277    #[must_use]
278    pub fn is_closed(&self) -> bool {
279        self.handle.is_closed()
280    }
281
282    /// Closes the connection and waits (bounded) for the connection task.
283    ///
284    /// # Errors
285    ///
286    /// Returns [`Error::Timeout`] when the connection task does not finish
287    /// within five seconds; the task is aborted in that case.
288    pub async fn close(mut self) -> Result<(), Error> {
289        self.handle.close();
290        let started = tokio::time::Instant::now();
291        let deadline = started + CLOSE_TIMEOUT;
292        loop {
293            match tokio::time::timeout_at(deadline, self.tasks.join_next()).await {
294                Ok(None) => return Ok(()),
295                Ok(Some(_)) => {}
296                Err(_) => {
297                    self.tasks.abort_all();
298                    return Err(Error::Timeout {
299                        scope: crate::timeout::TimeoutScope::Total,
300                        elapsed: started.elapsed(),
301                    });
302                }
303            }
304        }
305    }
306}
307
308impl Stream for RealtimeSession {
309    type Item = RealtimeEvent;
310
311    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
312        self.events.poll_recv(cx)
313    }
314}
ferrin_core/realtime/mod.rs

ferrin_core/realtime/
mod.rs
