flowcat_core/realtime/
service_adapter.rs

1// SPDX-License-Identifier: Apache-2.0
2//
3//! Bridge: [`RealtimeLlmService`] → [`RealtimeLlm`] + [`RealtimeKickoff`].
4//!
5//! flowcat has two realtime trait surfaces: the s2s pipeline
6//! ([`build_s2s_task`](crate::pipeline::build_s2s_task)) is generic over
7//! [`RealtimeLlm`] + [`RealtimeKickoff`] (which [`GeminiLive`](crate::GeminiLive)
8//! implements directly), while the streaming realtime *connectors* (OpenAI, Grok,
9//! Ultravox, … in `flowcat-services`) implement the frozen
10//! [`RealtimeLlmService`]. [`ServiceRealtimeAdapter`] wraps any `RealtimeLlmService`
11//! and presents it as a `RealtimeLlm + RealtimeKickoff`, so a connector drops into the
12//! pipeline with **zero glue in the consuming app** — the realtime equivalent of how
13//! a cascaded `LlmService`/`SttService`/`TtsService` is consumed directly.
14//!
15//! The only structural difference between the two traits is `send_audio`'s argument
16//! (`AudioChunk` on the pipeline side vs `Arc<AudioFrame>` on the service side);
17//! `RealtimeSetup`/`RealtimeEvent`/`ToolDecl` are shared/aliased, so everything else
18//! is a straight delegate. The service trait has no lock-free `event_notify`/
19//! `poll_event`, so the adapter inherits [`RealtimeLlm`]'s blocking defaults (which
20//! the trait documents as the legacy-OK path).
21
22use std::sync::Arc;
23
24use async_trait::async_trait;
25
26use crate::error::FlowcatError;
27use crate::processor::frame::AudioFrame;
28use crate::realtime::{RealtimeKickoff, RealtimeLlm};
29use crate::service::RealtimeLlmService;
30use crate::types::{AudioChunk, RealtimeEvent, RealtimeSetup, ToolDecl};
31
32/// Adapts any [`RealtimeLlmService`] connector into the pipeline's
33/// [`RealtimeLlm`] + [`RealtimeKickoff`] surface. See the module docs.
34pub struct ServiceRealtimeAdapter<S: RealtimeLlmService> {
35    inner: S,
36}
37
38impl<S: RealtimeLlmService> ServiceRealtimeAdapter<S> {
39    /// Wrap a realtime connector.
40    pub fn new(inner: S) -> Self {
41        Self { inner }
42    }
43
44    /// Unwrap back to the underlying connector.
45    pub fn into_inner(self) -> S {
46        self.inner
47    }
48}
49
50#[async_trait]
51impl<S: RealtimeLlmService> RealtimeLlm for ServiceRealtimeAdapter<S> {
52    async fn connect(&mut self, setup: RealtimeSetup) -> Result<(), FlowcatError> {
53        self.inner.connect(setup).await
54    }
55
56    async fn send_audio(&mut self, chunk: AudioChunk) -> Result<(), FlowcatError> {
57        // The only real shape difference: pipeline `AudioChunk` → service
58        // `Arc<AudioFrame>` (`AudioFrame: From<AudioChunk>`).
59        self.inner
60            .send_audio(Arc::new(AudioFrame::from(chunk)))
61            .await
62    }
63
64    async fn update_system(
65        &mut self,
66        prompt: String,
67        tools: Vec<ToolDecl>,
68    ) -> Result<(), FlowcatError> {
69        // `Tool` is a type alias of `ToolDecl`, so the vec passes straight through.
70        self.inner.update_system(prompt, tools).await
71    }
72
73    async fn send_tool_result(
74        &mut self,
75        id: String,
76        result: serde_json::Value,
77    ) -> Result<(), FlowcatError> {
78        self.inner.send_tool_result(id, result).await
79    }
80
81    async fn next_event(&mut self) -> Option<RealtimeEvent> {
82        self.inner.next_event().await
83    }
84
85    fn input_sample_rate(&self) -> u32 {
86        self.inner.input_sample_rate()
87    }
88
89    // Delegate the lock-free event path to the connector. Without this the pipeline
90    // would use the blocking `next_event` (holding the session lock across the idle
91    // wait between turns), which starves `send_audio` and stalls caller audio after
92    // the greeting. A connector that doesn't override these gets the service trait's
93    // blocking defaults (same as before) — no regression.
94    fn event_notify(&self) -> Option<std::sync::Arc<tokio::sync::Notify>> {
95        self.inner.event_notify()
96    }
97
98    async fn poll_event(&mut self) -> crate::realtime::PollEvent {
99        self.inner.poll_event().await
100    }
101}
102
103#[async_trait]
104impl<S: RealtimeLlmService> RealtimeKickoff for ServiceRealtimeAdapter<S> {
105    async fn kickoff(&mut self) -> Result<(), FlowcatError> {
106        self.inner.kickoff().await
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113
114    /// A minimal `RealtimeLlmService` that records what it received, so we can assert
115    /// the adapter delegates + converts correctly without a live socket.
116    #[derive(Default)]
117    struct SpyService {
118        connected: bool,
119        last_audio_rate: Option<u32>,
120        kicked: bool,
121    }
122
123    #[async_trait]
124    impl RealtimeLlmService for SpyService {
125        async fn connect(&mut self, _setup: RealtimeSetup) -> Result<(), FlowcatError> {
126            self.connected = true;
127            Ok(())
128        }
129        async fn send_audio(&mut self, chunk: Arc<AudioFrame>) -> Result<(), FlowcatError> {
130            self.last_audio_rate = Some(chunk.sample_rate);
131            Ok(())
132        }
133        async fn update_system(
134            &mut self,
135            _p: String,
136            _t: Vec<ToolDecl>,
137        ) -> Result<(), FlowcatError> {
138            Ok(())
139        }
140        async fn send_tool_result(
141            &mut self,
142            _id: String,
143            _r: serde_json::Value,
144        ) -> Result<(), FlowcatError> {
145            Ok(())
146        }
147        async fn next_event(&mut self) -> Option<RealtimeEvent> {
148            None
149        }
150        fn input_sample_rate(&self) -> u32 {
151            24_000
152        }
153        async fn kickoff(&mut self) -> Result<(), FlowcatError> {
154            self.kicked = true;
155            Ok(())
156        }
157    }
158
159    #[tokio::test]
160    async fn adapter_delegates_and_converts_audio() {
161        let mut a = ServiceRealtimeAdapter::new(SpyService::default());
162        // input_sample_rate flows from the connector (24k here).
163        assert_eq!(a.input_sample_rate(), 24_000);
164        // send_audio: AudioChunk(24k) → Arc<AudioFrame>(24k) reaches the service.
165        a.send_audio(AudioChunk::new(vec![0i16; 10], 24_000))
166            .await
167            .unwrap();
168        assert_eq!(a.inner.last_audio_rate, Some(24_000));
169        // kickoff routes through to the connector's kickoff.
170        RealtimeKickoff::kickoff(&mut a).await.unwrap();
171        assert!(a.inner.kicked);
172    }
173}
flowcat_core/realtime/service_adapter.rs

flowcat_core/realtime/
service_adapter.rs
