Update realtime websocket API (#13265)

- migrate the realtime websocket transport to the new session and handoff flow - make the realtime model configurable in config.toml and use API-key auth for the websocket --------- Co-authored-by: Codex <noreply@openai.com>
2026-07-01 00:31:56 +08:00 · 2026-03-02 16:05:40 -08:00
parent d473e8d56d
commit b20b6aa46f
21 changed files with 1449 additions and 507 deletions
@@ -2367,6 +2367,8 @@ impl Session {
        self.send_event_raw(event).await;
        self.maybe_mirror_event_text_to_realtime(&legacy_source)
            .await;
+        self.maybe_clear_realtime_handoff_for_event(&legacy_source)
+            .await;

        let show_raw_agent_reasoning = self.show_raw_agent_reasoning();
        for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) {
@@ -2382,14 +2384,23 @@ impl Session {
        let Some(text) = realtime_text_for_event(msg) else {
            return;
        };
-        if self.conversation.running_state().await.is_none() {
+        if self.conversation.running_state().await.is_none()
+            || self.conversation.active_handoff_id().await.is_none()
+        {
            return;
        }
-        if let Err(err) = self.conversation.text_in(text).await {
+        if let Err(err) = self.conversation.handoff_out(text).await {
            debug!("failed to mirror event text to realtime conversation: {err}");
        }
    }

+    async fn maybe_clear_realtime_handoff_for_event(&self, msg: &EventMsg) {
+        if !matches!(msg, EventMsg::TurnComplete(_)) {
+            return;
+        }
+        self.conversation.clear_active_handoff().await;
+    }
+
    pub(crate) async fn send_event_raw(&self, event: Event) {
        // Record the last known agent status.
        if let Some(status) = agent_status_from_event(&event.msg) {
@@ -435,12 +435,16 @@ pub struct Config {
    pub realtime_audio: RealtimeAudioConfig,

    /// Experimental / do not use. Overrides only the realtime conversation
-    /// websocket transport base URL (the `Op::RealtimeConversation` `/ws`
+    /// websocket transport base URL (the `Op::RealtimeConversation`
+    /// `/v1/realtime`
    /// connection) without changing normal provider HTTP requests.
    pub experimental_realtime_ws_base_url: Option<String>,
+    /// Experimental / do not use. Selects the realtime websocket model/snapshot
+    /// used for the `Op::RealtimeConversation` connection.
+    pub experimental_realtime_ws_model: Option<String>,
    /// Experimental / do not use. Overrides only the realtime conversation
-    /// websocket transport backend prompt (the `Op::RealtimeConversation`
-    /// `/ws` session.create backend_prompt) without changing normal prompts.
+    /// websocket transport instructions (the `Op::RealtimeConversation`
+    /// `/ws` session.update instructions) without changing normal prompts.
    pub experimental_realtime_ws_backend_prompt: Option<String>,
    /// When set, restricts ChatGPT login to a specific workspace identifier.
    pub forced_chatgpt_workspace_id: Option<String>,
@@ -1188,12 +1192,16 @@ pub struct ConfigToml {
    pub audio: Option<RealtimeAudioToml>,

    /// Experimental / do not use. Overrides only the realtime conversation
-    /// websocket transport base URL (the `Op::RealtimeConversation` `/ws`
+    /// websocket transport base URL (the `Op::RealtimeConversation`
+    /// `/v1/realtime`
    /// connection) without changing normal provider HTTP requests.
    pub experimental_realtime_ws_base_url: Option<String>,
+    /// Experimental / do not use. Selects the realtime websocket model/snapshot
+    /// used for the `Op::RealtimeConversation` connection.
+    pub experimental_realtime_ws_model: Option<String>,
    /// Experimental / do not use. Overrides only the realtime conversation
-    /// websocket transport backend prompt (the `Op::RealtimeConversation`
-    /// `/ws` session.create backend_prompt) without changing normal prompts.
+    /// websocket transport instructions (the `Op::RealtimeConversation`
+    /// `/ws` session.update instructions) without changing normal prompts.
    pub experimental_realtime_ws_backend_prompt: Option<String>,
    pub projects: Option<HashMap<String, ProjectConfig>>,

@@ -2182,6 +2190,7 @@ impl Config {
                    speaker: audio.speaker,
                }),
            experimental_realtime_ws_base_url: cfg.experimental_realtime_ws_base_url,
+            experimental_realtime_ws_model: cfg.experimental_realtime_ws_model,
            experimental_realtime_ws_backend_prompt: cfg.experimental_realtime_ws_backend_prompt,
            forced_chatgpt_workspace_id,
            forced_login_method,
@@ -4924,6 +4933,7 @@ model_verbosity = "high"
                chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
                realtime_audio: RealtimeAudioConfig::default(),
                experimental_realtime_ws_base_url: None,
+                experimental_realtime_ws_model: None,
                experimental_realtime_ws_backend_prompt: None,
                base_instructions: None,
                developer_instructions: None,
@@ -5052,6 +5062,7 @@ model_verbosity = "high"
            chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
            realtime_audio: RealtimeAudioConfig::default(),
            experimental_realtime_ws_base_url: None,
+            experimental_realtime_ws_model: None,
            experimental_realtime_ws_backend_prompt: None,
            base_instructions: None,
            developer_instructions: None,
@@ -5178,6 +5189,7 @@ model_verbosity = "high"
            chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
            realtime_audio: RealtimeAudioConfig::default(),
            experimental_realtime_ws_base_url: None,
+            experimental_realtime_ws_model: None,
            experimental_realtime_ws_backend_prompt: None,
            base_instructions: None,
            developer_instructions: None,
@@ -5290,6 +5302,7 @@ model_verbosity = "high"
            chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
            realtime_audio: RealtimeAudioConfig::default(),
            experimental_realtime_ws_base_url: None,
+            experimental_realtime_ws_model: None,
            experimental_realtime_ws_backend_prompt: None,
            base_instructions: None,
            developer_instructions: None,
@@ -6135,6 +6148,34 @@ experimental_realtime_ws_backend_prompt = "prompt from config"
        Ok(())
    }

+    #[test]
+    fn experimental_realtime_ws_model_loads_from_config_toml() -> std::io::Result<()> {
+        let cfg: ConfigToml = toml::from_str(
+            r#"
+experimental_realtime_ws_model = "realtime-test-model"
+"#,
+        )
+        .expect("TOML deserialization should succeed");
+
+        assert_eq!(
+            cfg.experimental_realtime_ws_model.as_deref(),
+            Some("realtime-test-model")
+        );
+
+        let codex_home = TempDir::new()?;
+        let config = Config::load_from_base_config_with_overrides(
+            cfg,
+            ConfigOverrides::default(),
+            codex_home.path().to_path_buf(),
+        )?;
+
+        assert_eq!(
+            config.experimental_realtime_ws_model.as_deref(),
+            Some("realtime-test-model")
+        );
+        Ok(())
+    }
+
    #[test]
    fn realtime_audio_loads_from_config_toml() -> std::io::Result<()> {
        let cfg: ConfigToml = toml::from_str(
@@ -1,5 +1,6 @@
 use crate::CodexAuth;
 use crate::api_bridge::map_api_error;
+use crate::auth::read_openai_api_key_from_env;
 use crate::codex::Session;
 use crate::default_client::default_headers;
 use crate::error::CodexErr;
@@ -24,8 +25,10 @@ use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::RealtimeConversationClosedEvent;
 use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
 use codex_protocol::protocol::RealtimeConversationStartedEvent;
+use codex_protocol::protocol::RealtimeHandoffRequested;
 use http::HeaderMap;
-use serde_json::Value;
+use http::HeaderValue;
+use http::header::AUTHORIZATION;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
@@ -37,17 +40,55 @@ use tracing::info;
 use tracing::warn;

 const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
-const TEXT_IN_QUEUE_CAPACITY: usize = 64;
+const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64;
+const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
 const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256;

 pub(crate) struct RealtimeConversationManager {
    state: Mutex<Option<ConversationState>>,
 }

+#[derive(Clone, Debug)]
+struct RealtimeHandoffState {
+    output_tx: Sender<HandoffOutput>,
+    active_handoff: Arc<Mutex<Option<String>>>,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct HandoffOutput {
+    handoff_id: String,
+    output_text: String,
+}
+
+impl RealtimeHandoffState {
+    fn new(output_tx: Sender<HandoffOutput>) -> Self {
+        Self {
+            output_tx,
+            active_handoff: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    async fn send_output(&self, output_text: String) -> CodexResult<()> {
+        let Some(handoff_id) = self.active_handoff.lock().await.clone() else {
+            return Ok(());
+        };
+
+        self.output_tx
+            .send(HandoffOutput {
+                handoff_id,
+                output_text,
+            })
+            .await
+            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
+        Ok(())
+    }
+}
+
 #[allow(dead_code)]
 struct ConversationState {
    audio_tx: Sender<RealtimeAudioFrame>,
-    text_tx: Sender<String>,
+    user_text_tx: Sender<String>,
+    handoff: RealtimeHandoffState,
    task: JoinHandle<()>,
    realtime_active: Arc<AtomicBool>,
 }
@@ -72,6 +113,7 @@ impl RealtimeConversationManager {
        api_provider: ApiProvider,
        extra_headers: Option<HeaderMap>,
        prompt: String,
+        model: Option<String>,
        session_id: Option<String>,
    ) -> CodexResult<(Receiver<RealtimeEvent>, Arc<AtomicBool>)> {
        let previous_state = {
@@ -84,7 +126,11 @@ impl RealtimeConversationManager {
            let _ = state.task.await;
        }

-        let session_config = RealtimeSessionConfig { prompt, session_id };
+        let session_config = RealtimeSessionConfig {
+            instructions: prompt,
+            model,
+            session_id,
+        };
        let client = RealtimeWebsocketClient::new(api_provider);
        let connection = client
            .connect(
@@ -99,17 +145,30 @@ impl RealtimeConversationManager {
        let events = connection.events();
        let (audio_tx, audio_rx) =
            async_channel::bounded::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
-        let (text_tx, text_rx) = async_channel::bounded::<String>(TEXT_IN_QUEUE_CAPACITY);
+        let (user_text_tx, user_text_rx) =
+            async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
+        let (handoff_output_tx, handoff_output_rx) =
+            async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
        let (events_tx, events_rx) =
            async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);

        let realtime_active = Arc::new(AtomicBool::new(true));
-        let task = spawn_realtime_input_task(writer, events, text_rx, audio_rx, events_tx);
+        let handoff = RealtimeHandoffState::new(handoff_output_tx);
+        let task = spawn_realtime_input_task(
+            writer,
+            events,
+            user_text_rx,
+            handoff_output_rx,
+            audio_rx,
+            events_tx,
+            handoff.clone(),
+        );

        let mut guard = self.state.lock().await;
        *guard = Some(ConversationState {
            audio_tx,
-            text_tx,
+            user_text_tx,
+            handoff,
            task,
            realtime_active: Arc::clone(&realtime_active),
        });
@@ -143,7 +202,7 @@ impl RealtimeConversationManager {
    pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> {
        let sender = {
            let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.text_tx.clone())
+            guard.as_ref().map(|state| state.user_text_tx.clone())
        };

        let Some(sender) = sender else {
@@ -159,6 +218,38 @@ impl RealtimeConversationManager {
        Ok(())
    }

+    pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> {
+        let handoff = {
+            let guard = self.state.lock().await;
+            let Some(state) = guard.as_ref() else {
+                return Err(CodexErr::InvalidRequest(
+                    "conversation is not running".to_string(),
+                ));
+            };
+            state.handoff.clone()
+        };
+
+        handoff.send_output(output_text).await
+    }
+
+    pub(crate) async fn active_handoff_id(&self) -> Option<String> {
+        let handoff = {
+            let guard = self.state.lock().await;
+            guard.as_ref().map(|state| state.handoff.clone())
+        }?;
+        handoff.active_handoff.lock().await.clone()
+    }
+
+    pub(crate) async fn clear_active_handoff(&self) {
+        let handoff = {
+            let guard = self.state.lock().await;
+            guard.as_ref().map(|state| state.handoff.clone())
+        };
+        if let Some(handoff) = handoff {
+            *handoff.active_handoff.lock().await = None;
+        }
+    }
+
    pub(crate) async fn shutdown(&self) -> CodexResult<()> {
        let state = {
            let mut guard = self.state.lock().await;
@@ -181,7 +272,8 @@ pub(crate) async fn handle_start(
 ) -> CodexResult<()> {
    let provider = sess.provider().await;
    let auth = sess.services.auth_manager.auth().await;
-    let mut api_provider = provider.to_api_provider(auth.as_ref().map(CodexAuth::auth_mode))?;
+    let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?;
+    let mut api_provider = provider.to_api_provider(Some(crate::auth::AuthMode::ApiKey))?;
    let config = sess.get_config().await;
    if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url {
        api_provider.base_url = realtime_ws_base_url.clone();
@@ -190,14 +282,23 @@ pub(crate) async fn handle_start(
        .experimental_realtime_ws_backend_prompt
        .clone()
        .unwrap_or(params.prompt);
+    let model = config.experimental_realtime_ws_model.clone();

    let requested_session_id = params
        .session_id
        .or_else(|| Some(sess.conversation_id.to_string()));
+    let extra_headers =
+        realtime_request_headers(requested_session_id.as_deref(), realtime_api_key.as_str())?;
    info!("starting realtime conversation");
    let (events_rx, realtime_active) = match sess
        .conversation
-        .start(api_provider, None, prompt, requested_session_id.clone())
+        .start(
+            api_provider,
+            extra_headers,
+            prompt,
+            model,
+            requested_session_id.clone(),
+        )
        .await
    {
        Ok(events_rx) => events_rx,
@@ -227,8 +328,8 @@ pub(crate) async fn handle_start(
        while let Ok(event) = events_rx.recv().await {
            debug!(conversation_id = %sess_clone.conversation_id, "received realtime conversation event");
            let maybe_routed_text = match &event {
-                RealtimeEvent::ConversationItemAdded(item) => {
-                    realtime_text_from_conversation_item(item)
+                RealtimeEvent::HandoffRequested(handoff) => {
+                    realtime_text_from_handoff_request(handoff)
                }
                _ => None,
            };
@@ -271,26 +372,57 @@ pub(crate) async fn handle_audio(
    }
 }

-fn realtime_text_from_conversation_item(item: &Value) -> Option<String> {
-    match item.get("type").and_then(Value::as_str) {
-        Some("message") => {
-            if item.get("role").and_then(Value::as_str) != Some("assistant") {
-                return None;
-            }
-            let content = item.get("content")?.as_array()?;
-            let text = content
-                .iter()
-                .filter(|entry| entry.get("type").and_then(Value::as_str) == Some("text"))
-                .filter_map(|entry| entry.get("text").and_then(Value::as_str))
-                .collect::<String>();
-            if text.is_empty() { None } else { Some(text) }
-        }
-        Some("spawn_transcript") => item
-            .get("delta_user_transcript")
-            .and_then(Value::as_str)
-            .and_then(|text| (!text.is_empty()).then(|| text.to_string())),
-        Some(_) | None => None,
+fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
+    (!handoff.input_transcript.is_empty()).then(|| handoff.input_transcript.clone())
+}
+
+fn realtime_api_key(
+    auth: Option<&CodexAuth>,
+    provider: &crate::ModelProviderInfo,
+) -> CodexResult<String> {
+    if let Some(api_key) = provider.api_key()? {
+        return Ok(api_key);
    }
+
+    if let Some(token) = provider.experimental_bearer_token.clone() {
+        return Ok(token);
+    }
+
+    if let Some(api_key) = auth.and_then(CodexAuth::api_key) {
+        return Ok(api_key.to_string());
+    }
+
+    // TODO(aibrahim): Remove this temporary fallback once realtime auth no longer
+    // requires API key auth for ChatGPT/SIWC sessions.
+    if provider.is_openai()
+        && let Some(api_key) = read_openai_api_key_from_env()
+    {
+        return Ok(api_key);
+    }
+
+    Err(CodexErr::InvalidRequest(
+        "realtime conversation requires API key auth".to_string(),
+    ))
+}
+
+fn realtime_request_headers(
+    session_id: Option<&str>,
+    api_key: &str,
+) -> CodexResult<Option<HeaderMap>> {
+    let mut headers = HeaderMap::new();
+
+    if let Some(session_id) = session_id
+        && let Ok(session_id) = HeaderValue::from_str(session_id)
+    {
+        headers.insert("x-session-id", session_id);
+    }
+
+    let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| {
+        CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}"))
+    })?;
+    headers.insert(AUTHORIZATION, auth_value);
+
+    Ok(Some(headers))
 }

 pub(crate) async fn handle_text(
@@ -326,14 +458,16 @@ pub(crate) async fn handle_close(sess: &Arc<Session>, sub_id: String) {
 fn spawn_realtime_input_task(
    writer: RealtimeWebsocketWriter,
    events: RealtimeWebsocketEvents,
-    text_rx: Receiver<String>,
+    user_text_rx: Receiver<String>,
+    handoff_output_rx: Receiver<HandoffOutput>,
    audio_rx: Receiver<RealtimeAudioFrame>,
    events_tx: Sender<RealtimeEvent>,
+    handoff_state: RealtimeHandoffState,
 ) -> JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            tokio::select! {
-                text = text_rx.recv() => {
+                text = user_text_rx.recv() => {
                    match text {
                        Ok(text) => {
                            if let Err(err) = writer.send_conversation_item_create(text).await {
@@ -345,9 +479,31 @@ fn spawn_realtime_input_task(
                        Err(_) => break,
                    }
                }
+                handoff_output = handoff_output_rx.recv() => {
+                    match handoff_output {
+                        Ok(HandoffOutput {
+                            handoff_id,
+                            output_text,
+                        }) => {
+                            if let Err(err) = writer
+                                .send_conversation_handoff_append(handoff_id, output_text)
+                                .await
+                            {
+                                let mapped_error = map_api_error(err);
+                                warn!("failed to send handoff output: {mapped_error}");
+                                break;
+                            }
+                        }
+                        Err(_) => break,
+                    }
+                }
                event = events.next_event() => {
                    match event {
                        Ok(Some(event)) => {
+                            if let RealtimeEvent::HandoffRequested(handoff) = &event {
+                                *handoff_state.active_handoff.lock().await =
+                                    Some(handoff.handoff_id.clone());
+                            }
                            let should_stop = matches!(&event, RealtimeEvent::Error(_));
                            if events_tx.send(event).await.is_err() {
                                break;
@@ -414,82 +570,98 @@ async fn send_conversation_error(

 #[cfg(test)]
 mod tests {
-    use super::realtime_text_from_conversation_item;
+    use super::HandoffOutput;
+    use super::RealtimeHandoffState;
+    use super::realtime_text_from_handoff_request;
+    use async_channel::bounded;
+    use codex_protocol::protocol::RealtimeHandoffMessage;
+    use codex_protocol::protocol::RealtimeHandoffRequested;
    use pretty_assertions::assert_eq;
-    use serde_json::json;

    #[test]
-    fn extracts_text_from_assistant_message_items_only() {
-        let assistant = json!({
-            "type": "message",
-            "role": "assistant",
-            "content": [{"type": "text", "text": "hello"}],
-        });
+    fn extracts_text_from_handoff_request_input_transcript() {
+        let handoff = RealtimeHandoffRequested {
+            handoff_id: "handoff_1".to_string(),
+            item_id: "item_1".to_string(),
+            input_transcript: "hello".to_string(),
+            messages: vec![RealtimeHandoffMessage {
+                role: "user".to_string(),
+                text: "hello".to_string(),
+            }],
+        };
        assert_eq!(
-            realtime_text_from_conversation_item(&assistant),
+            realtime_text_from_handoff_request(&handoff),
            Some("hello".to_string())
        );
-
-        let user = json!({
-            "type": "message",
-            "role": "user",
-            "content": [{"type": "text", "text": "world"}],
-        });
-        assert_eq!(realtime_text_from_conversation_item(&user), None);
    }

    #[test]
-    fn extracts_and_concatenates_text_entries_only() {
-        let item = json!({
-            "type": "message",
-            "role": "assistant",
-            "content": [
-                {"type": "text", "text": "a"},
-                {"type": "ignored", "text": "x"},
-                {"type": "text", "text": "b"}
-            ],
-        });
-        assert_eq!(
-            realtime_text_from_conversation_item(&item),
-            Some("ab".to_string())
-        );
+    fn ignores_empty_handoff_request_input_transcript() {
+        let handoff = RealtimeHandoffRequested {
+            handoff_id: "handoff_1".to_string(),
+            item_id: "item_1".to_string(),
+            input_transcript: String::new(),
+            messages: vec![],
+        };
+        assert_eq!(realtime_text_from_handoff_request(&handoff), None);
    }

-    #[test]
-    fn ignores_non_message_or_missing_text() {
-        let non_message = json!({
-            "type": "tool_call",
-            "content": [{"type": "text", "text": "nope"}],
-        });
-        assert_eq!(realtime_text_from_conversation_item(&non_message), None);
+    #[tokio::test]
+    async fn clears_active_handoff_explicitly() {
+        let (tx, _rx) = bounded(1);
+        let state = RealtimeHandoffState::new(tx);

-        let no_text = json!({
-            "type": "message",
-            "role": "assistant",
-            "content": [{"type": "other", "value": 1}],
-        });
-        assert_eq!(realtime_text_from_conversation_item(&no_text), None);
-
-        let empty_spawn_transcript = json!({
-            "type": "spawn_transcript",
-            "delta_user_transcript": "",
-        });
+        *state.active_handoff.lock().await = Some("handoff_1".to_string());
        assert_eq!(
-            realtime_text_from_conversation_item(&empty_spawn_transcript),
-            None
+            state.active_handoff.lock().await.clone(),
+            Some("handoff_1".to_string())
        );
+
+        *state.active_handoff.lock().await = None;
+        assert_eq!(state.active_handoff.lock().await.clone(), None);
    }

-    #[test]
-    fn extracts_text_from_spawn_transcript_items() {
-        let item = json!({
-            "type": "spawn_transcript",
-            "delta_user_transcript": "delegate from transcript",
-            "backend_prompt_messages": [{"role": "user", "content": "delegate from transcript"}],
-        });
+    #[tokio::test]
+    async fn sends_multiple_handoff_outputs_until_cleared() {
+        let (tx, rx) = bounded(4);
+        let state = RealtimeHandoffState::new(tx);
+
+        state
+            .send_output("ignored".to_string())
+            .await
+            .expect("send");
+        assert!(rx.is_empty());
+
+        *state.active_handoff.lock().await = Some("handoff_1".to_string());
+        state.send_output("result".to_string()).await.expect("send");
+        state
+            .send_output("result 2".to_string())
+            .await
+            .expect("send");
+
+        let output_1 = rx.recv().await.expect("recv");
        assert_eq!(
-            realtime_text_from_conversation_item(&item),
-            Some("delegate from transcript".to_string())
+            output_1,
+            HandoffOutput {
+                handoff_id: "handoff_1".to_string(),
+                output_text: "result".to_string(),
+            }
        );
+
+        let output_2 = rx.recv().await.expect("recv");
+        assert_eq!(
+            output_2,
+            HandoffOutput {
+                handoff_id: "handoff_1".to_string(),
+                output_text: "result 2".to_string(),
+            }
+        );
+
+        *state.active_handoff.lock().await = None;
+        state
+            .send_output("ignored after clear".to_string())
+            .await
+            .expect("send");
+        assert!(rx.is_empty());
    }
 }