From c18668fb43ed7591c80a1a813a02c8fb4f4c802f Mon Sep 17 00:00:00 2001
From: Ahmed Ibrahim <aibrahim@openai.com>
Date: Thu, 9 Apr 2026 18:10:48 -0700
Subject: [PATCH 1/3] Rename Realtime V2 tool to background_agent

Rename the Realtime V2 delegation tool and parser constant to background_agent, and update the tool description and fixtures to match.

Co-authored-by: Codex <noreply@openai.com>
---
 .../tests/suite/v2/realtime_conversation.rs   | 38 ++++++++++---------
 .../endpoint/realtime_websocket/methods.rs    | 12 +++---
 .../endpoint/realtime_websocket/methods_v2.rs | 10 ++---
 .../realtime_websocket/protocol_v2.rs         |  4 +-
 4 files changed, 33 insertions(+), 31 deletions(-)

diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
index f1f60be45e0a..0dc88d72b4e3 100644
--- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
@@ -165,14 +165,15 @@ struct StartedWebrtcRealtime {
     sdp: ThreadRealtimeSdpNotification,
 }
 
-// Scripted SSE responses for the normal Codex agent loop. Realtime can ask for a delegated
-// Codex turn; that turn talks to this mock `/responses` endpoint and may request ordinary tools.
+// Scripted SSE responses for the normal background agent loop. Realtime can ask for a delegated
+// background agent turn; that turn talks to this mock `/responses` endpoint and may request
+// ordinary tools.
 struct MainLoopResponsesScript {
     responses: Vec<String>,
 }
 
 // Scripted server events for the direct realtime sideband WebSocket. This mock is the realtime
-// session app-server joins after call creation; it is not the main-loop Responses stream.
+// session app-server joins after call creation; it is not the background agent Responses stream.
 struct RealtimeSidebandScript {
     connections: Vec<WebSocketConnectionConfig>,
 }
@@ -425,13 +426,13 @@ fn session_updated(session_id: &str) -> Value {
     })
 }
 
-fn v2_codex_tool_call(call_id: &str, prompt: &str) -> Value {
+fn v2_background_agent_tool_call(call_id: &str, prompt: &str) -> Value {
     json!({
         "type": "conversation.item.done",
         "item": {
             "id": format!("item_{call_id}"),
             "type": "function_call",
-            "name": "codex",
+            "name": "background_agent",
             "call_id": call_id,
             "arguments": json!({ "prompt": prompt }).to_string()
         }
@@ -478,7 +479,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
                 "item": {
                     "id": "item_2",
                     "type": "function_call",
-                    "name": "codex",
+                    "name": "background_agent",
                     "call_id": "handoff_1",
                     "arguments": "{\"input_transcript\":\"delegate now\"}"
                 }
@@ -973,7 +974,7 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
         Some("multipart/form-data; boundary=codex-realtime-call-boundary")
     );
     let body = String::from_utf8(request.body).context("multipart body should be utf-8")?;
-    let session = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"codex","description":"Delegate a request to Codex and return the final result to the user. Use this as the default action. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to Codex."}},"required":["prompt"],"additionalProperties":false}}]}"#;
+    let session = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"background_agent","description":"Send a user request to the background agent. Use this as the default action. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to the background agent."}},"required":["prompt"],"additionalProperties":false}}]}"#;
     assert_eq!(
         body,
         format!(
@@ -1089,7 +1090,7 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
     let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
     assert_eq!(started.started.version, RealtimeConversationVersion::V1);
 
-    // Phase 2: wait for the delegated Codex turn that is launched by the handoff request.
+    // Phase 2: wait for the delegated background agent turn that is launched by the handoff request.
     let turn_started = harness
         .read_notification::<TurnStartedNotification>("turn/started")
         .await?;
@@ -1204,11 +1205,12 @@ async fn webrtc_v2_forwards_audio_and_text_between_client_and_sideband() -> Resu
 }
 
 #[tokio::test]
-async fn webrtc_v2_codex_tool_call_delegates_and_returns_function_output() -> Result<()> {
+async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_output() -> Result<()>
+{
     skip_if_no_network!(Ok(()));
 
-    // Phase 1: script a v2 codex function call and a delegated Responses turn that returns final
-    // assistant text.
+    // Phase 1: script a v2 background agent function call and a delegated Responses turn that
+    // returns final assistant text.
     let mut harness = RealtimeE2eHarness::new(
         RealtimeTestVersion::V2,
         main_loop_responses(vec![create_final_assistant_message_sse_response(
@@ -1217,7 +1219,7 @@ async fn webrtc_v2_codex_tool_call_delegates_and_returns_function_output() -> Re
         realtime_sideband(vec![realtime_sideband_connection(vec![
             vec![
                 session_updated("sess_v2_tool"),
-                v2_codex_tool_call("call_v2", "delegate from v2"),
+                v2_background_agent_tool_call("call_v2", "delegate from v2"),
             ],
             vec![],
         ])]),
@@ -1259,8 +1261,8 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<(
     skip_if_no_network!(Ok(()));
 
     // Phase 1: keep the two mocked OpenAI conversations explicit. The realtime sideband only
-    // calls the `codex` function; the shell command is requested by the delegated main-loop
-    // Responses turn that app-server starts after receiving that function call.
+    // calls the `background_agent` function; the shell command is requested by the delegated
+    // background agent Responses turn that app-server starts after receiving that function call.
     let main_loop = main_loop_responses(vec![
         create_shell_command_sse_response(
             realtime_tool_ok_command(),
@@ -1273,7 +1275,7 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<(
     let realtime = realtime_sideband(vec![realtime_sideband_connection(vec![
         vec![
             session_updated("sess_v2_shell"),
-            v2_codex_tool_call("call_shell", "run shell through delegated turn"),
+            v2_background_agent_tool_call("call_shell", "run shell through delegated turn"),
         ],
         vec![],
     ])]);
@@ -1288,7 +1290,7 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<(
 
     let _ = harness.start_webrtc_realtime("v=offer\r\n").await?;
 
-    // Phase 2: observe the delegated main-loop turn executing the requested shell command.
+    // Phase 2: observe the delegated background agent turn executing the requested shell command.
     let started_command = wait_for_started_command_execution(&mut harness.mcp).await?;
     let ThreadItem::CommandExecution { id, status, .. } = started_command.item else {
         unreachable!("helper returns command execution items");
@@ -1367,7 +1369,7 @@ async fn webrtc_v2_tool_call_does_not_block_sideband_audio() -> Result<()> {
         realtime_sideband(vec![realtime_sideband_connection(vec![
             vec![
                 session_updated("sess_v2_nonblocking"),
-                v2_codex_tool_call("call_audio", "delegate while audio continues"),
+                v2_background_agent_tool_call("call_audio", "delegate while audio continues"),
                 json!({
                     "type": "response.output_audio.delta",
                     "delta": "CQoL",
@@ -1669,7 +1671,7 @@ fn assert_v2_session_update(request: &Value) -> Result<()> {
     );
     assert_eq!(
         request["session"]["tools"][0]["name"].as_str(),
-        Some("codex")
+        Some("background_agent")
     );
     Ok(())
 }
diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
index 745789410fb7..911d3629231c 100644
--- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
@@ -868,7 +868,7 @@ mod tests {
             "item": {
                 "id": "item_123",
                 "type": "function_call",
-                "name": "codex",
+                "name": "background_agent",
                 "call_id": "call_123",
                 "arguments": "{\"prompt\":\"delegate this\"}"
             }
@@ -989,7 +989,7 @@ mod tests {
                 "output": [{
                     "id": "item_123",
                     "type": "function_call",
-                    "name": "codex",
+                    "name": "background_agent",
                     "call_id": "call_123",
                     "arguments": "{\"prompt\":\"delegate from done\"}"
                 }]
@@ -1283,7 +1283,7 @@ mod tests {
             assert_eq!(fourth_json["handoff_id"], "handoff_1");
             assert_eq!(
                 fourth_json["output_text"],
-                "\"Agent Final Message\":\n\nhello from codex"
+                "\"Agent Final Message\":\n\nhello from background agent"
             );
 
             ws.send(Message::Text(
@@ -1407,7 +1407,7 @@ mod tests {
         connection
             .send_conversation_handoff_append(
                 "handoff_1".to_string(),
-                "hello from codex".to_string(),
+                "hello from background agent".to_string(),
             )
             .await
             .expect("send handoff");
@@ -1493,7 +1493,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn realtime_v2_session_update_includes_codex_tool_and_handoff_output_item() {
+    async fn realtime_v2_session_update_includes_background_agent_tool_and_handoff_output_item() {
         let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
         let addr = listener.local_addr().expect("local addr");
 
@@ -1553,7 +1553,7 @@ mod tests {
             );
             assert_eq!(
                 first_json["session"]["tools"][0]["name"],
-                Value::String("codex".to_string())
+                Value::String("background_agent".to_string())
             );
             assert_eq!(
                 first_json["session"]["tools"][0]["parameters"]["required"],
diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
index fd512a902d96..cfca6fce6132 100644
--- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
+++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
@@ -27,8 +27,8 @@ use serde_json::json;
 
 const REALTIME_V2_OUTPUT_MODALITY_AUDIO: &str = "audio";
 const REALTIME_V2_TOOL_CHOICE: &str = "auto";
-const REALTIME_V2_CODEX_TOOL_NAME: &str = "codex";
-const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate a request to Codex and return the final result to the user. Use this as the default action. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
+const REALTIME_V2_BACKGROUND_AGENT_TOOL_NAME: &str = "background_agent";
+const REALTIME_V2_BACKGROUND_AGENT_TOOL_DESCRIPTION: &str = "Send a user request to the background agent. Use this as the default action. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
 
 pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage {
     RealtimeOutboundMessage::ConversationItemCreate {
@@ -93,14 +93,14 @@ pub(super) fn session_update_session(
             },
             tools: Some(vec![SessionFunctionTool {
                 r#type: SessionToolType::Function,
-                name: REALTIME_V2_CODEX_TOOL_NAME.to_string(),
-                description: REALTIME_V2_CODEX_TOOL_DESCRIPTION.to_string(),
+                name: REALTIME_V2_BACKGROUND_AGENT_TOOL_NAME.to_string(),
+                description: REALTIME_V2_BACKGROUND_AGENT_TOOL_DESCRIPTION.to_string(),
                 parameters: json!({
                     "type": "object",
                     "properties": {
                         "prompt": {
                             "type": "string",
-                            "description": "The user request to delegate to Codex."
+                            "description": "The user request to delegate to the background agent."
                         }
                     },
                     "required": ["prompt"],
diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
index 036842d64acc..ea7b4d8117f8 100644
--- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
+++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
@@ -11,7 +11,7 @@ use serde_json::Map as JsonMap;
 use serde_json::Value;
 use tracing::debug;
 
-const CODEX_TOOL_NAME: &str = "codex";
+const BACKGROUND_AGENT_TOOL_NAME: &str = "background_agent";
 const DEFAULT_AUDIO_SAMPLE_RATE: u32 = 24_000;
 const DEFAULT_AUDIO_CHANNELS: u16 = 1;
 const TOOL_ARGUMENT_KEYS: [&str; 5] = ["input_transcript", "input", "text", "prompt", "query"];
@@ -118,7 +118,7 @@ fn parse_conversation_item_done_event(parsed: &Value) -> Option<RealtimeEvent> {
 fn parse_handoff_requested_event(item: &JsonMap<String, Value>) -> Option<RealtimeEvent> {
     let item_type = item.get("type").and_then(Value::as_str);
     let item_name = item.get("name").and_then(Value::as_str);
-    if item_type != Some("function_call") || item_name != Some(CODEX_TOOL_NAME) {
+    if item_type != Some("function_call") || item_name != Some(BACKGROUND_AGENT_TOOL_NAME) {
         return None;
     }
 

From 205e0ba1c3d0144b76e6783a7a58bdf8b4f0dbfa Mon Sep 17 00:00:00 2001
From: Ahmed Ibrahim <aibrahim@openai.com>
Date: Thu, 9 Apr 2026 21:48:19 -0700
Subject: [PATCH 2/3] Update realtime V2 e2e tool fixture

Co-authored-by: Codex <noreply@openai.com>
---
 codex-rs/codex-api/tests/realtime_websocket_e2e.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs
index 9be1733d49ec..9969a96f0978 100644
--- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs
@@ -493,7 +493,7 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() {
                 "item": {
                     "id": "item_123",
                     "type": "function_call",
-                    "name": "codex",
+                    "name": "background_agent",
                     "call_id": "call_123",
                     "arguments": "{\"prompt\":\"delegate now\"}"
                 }

From 047ea26f15202836bc05d8baa3565ced8a7cfe78 Mon Sep 17 00:00:00 2001
From: Ahmed Ibrahim <aibrahim@openai.com>
Date: Thu, 9 Apr 2026 18:20:45 -0700
Subject: [PATCH 3/3] Extract realtime input task handlers

Co-authored-by: Codex <noreply@openai.com>
---
 codex-rs/core/src/realtime_conversation.rs | 573 ++++++++++++---------
 1 file changed, 340 insertions(+), 233 deletions(-)

diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs
index c04dc8eaf46c..4b12aa4fe260 100644
--- a/codex-rs/core/src/realtime_conversation.rs
+++ b/codex-rs/core/src/realtime_conversation.rs
@@ -2,11 +2,14 @@ use crate::client::ModelClient;
 use crate::codex::Session;
 use crate::realtime_context::build_realtime_startup_context;
 use crate::realtime_prompt::prepare_realtime_backend_prompt;
+use anyhow::Context;
 use async_channel::Receiver;
+use async_channel::RecvError;
 use async_channel::Sender;
 use async_channel::TrySendError;
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use codex_api::ApiError;
 use codex_api::Provider as ApiProvider;
 use codex_api::RealtimeAudioFrame;
 use codex_api::RealtimeEvent;
@@ -113,6 +116,11 @@ struct OutputAudioState {
     audio_end_ms: u32,
 }
 
+struct ResponseCreateState {
+    pending_response_create: bool,
+    response_in_progress: bool,
+}
+
 struct RealtimeInputTask {
     writer: RealtimeWebsocketWriter,
     events: RealtimeWebsocketEvents,
@@ -366,15 +374,20 @@ impl RealtimeConversationManager {
         };
 
         *handoff.last_output_text.lock().await = Some(output_text.clone());
-        if matches!(handoff.session_kind, RealtimeSessionKind::V1) {
-            handoff
-                .output_tx
-                .send(HandoffOutput::ImmediateAppend {
-                    handoff_id,
-                    output_text,
-                })
-                .await
-                .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
+        match handoff.session_kind {
+            RealtimeSessionKind::V1 => {
+                handoff
+                    .output_tx
+                    .send(HandoffOutput::ImmediateAppend {
+                        handoff_id,
+                        output_text,
+                    })
+                    .await
+                    .map_err(|_| {
+                        CodexErr::InvalidRequest("conversation is not running".to_string())
+                    })?;
+            }
+            RealtimeSessionKind::V2 => {}
         }
         Ok(())
     }
@@ -387,8 +400,9 @@ impl RealtimeConversationManager {
         let Some(handoff) = handoff else {
             return Ok(());
         };
-        if matches!(handoff.session_kind, RealtimeSessionKind::V1) {
-            return Ok(());
+        match handoff.session_kind {
+            RealtimeSessionKind::V1 => return Ok(()),
+            RealtimeSessionKind::V2 => {}
         }
 
         let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
@@ -698,14 +712,16 @@ async fn handle_start_inner(
             if !fanout_realtime_active.load(Ordering::Relaxed) {
                 break;
             }
-            // if not audio out, log the event
-            if !matches!(event, RealtimeEvent::AudioOut(_)) {
-                info!(
-                    event = ?event,
-                    "received realtime conversation event"
-                );
+            match &event {
+                RealtimeEvent::AudioOut(_) => {}
+                _ => {
+                    info!(
+                        event = ?event,
+                        "received realtime conversation event"
+                    );
+                }
             }
-            if matches!(event, RealtimeEvent::Error(_)) {
+            if let RealtimeEvent::Error(_) = &event {
                 end = RealtimeConversationEnd::Error;
             }
             let maybe_routed_text = match &event {
@@ -731,8 +747,11 @@ async fn handle_start_inner(
                 .await;
         }
         if fanout_realtime_active.swap(false, Ordering::Relaxed) {
-            if matches!(end, RealtimeConversationEnd::TransportClosed) {
-                info!("realtime conversation transport closed");
+            match end {
+                RealtimeConversationEnd::TransportClosed => {
+                    info!("realtime conversation transport closed");
+                }
+                RealtimeConversationEnd::Requested | RealtimeConversationEnd::Error => {}
             }
             sess_clone
                 .conversation
@@ -858,233 +877,321 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> {
     } = input;
 
     tokio::spawn(async move {
-        let mut pending_response_create = false;
-        let mut response_in_progress = false;
+        let mut response_create_state = ResponseCreateState {
+            pending_response_create: false,
+            response_in_progress: false,
+        };
         let mut output_audio_state: Option<OutputAudioState> = None;
 
         loop {
-            tokio::select! {
-                text = user_text_rx.recv() => {
-                    match text {
-                        Ok(text) => {
-                            if let Err(err) = writer.send_conversation_item_create(text).await {
-                                let mapped_error = map_api_error(err);
-                                warn!("failed to send input text: {mapped_error}");
-                                let _ = events_tx
-                                    .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                    .await;
-                                break;
-                            }
-                            if matches!(session_kind, RealtimeSessionKind::V2) {
-                                if response_in_progress {
-                                    pending_response_create = true;
-                                } else if let Err(err) = writer.send_response_create().await {
-                                    let mapped_error = map_api_error(err);
-                                    warn!("failed to send text response.create: {mapped_error}");
-                                    let _ = events_tx
-                                        .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                        .await;
-                                    break;
-                                } else {
-                                    pending_response_create = false;
-                                    response_in_progress = true;
-                                }
-                            }
-                        }
-                        Err(_) => break,
+            let result = tokio::select! {
+                // Text typed by the user that should be sent into realtime.
+                user_text = user_text_rx.recv() => {
+                    handle_user_text_input(
+                        user_text,
+                        &writer,
+                        &events_tx,
+                        session_kind,
+                        &mut response_create_state,
+                    )
+                        .await
+                }
+                // Background agent progress or final output that should be sent back to realtime.
+                background_agent_output = handoff_output_rx.recv() => {
+                    handle_handoff_output(
+                        background_agent_output,
+                        &writer,
+                        &events_tx,
+                        session_kind,
+                        &mut response_create_state,
+                    )
+                        .await
+                }
+                // Events received from the realtime server.
+                realtime_event = events.next_event() => {
+                    handle_realtime_server_event(
+                        realtime_event,
+                        &writer,
+                        &events_tx,
+                        &handoff_state,
+                        session_kind,
+                        &mut output_audio_state,
+                        &mut response_create_state,
+                    )
+                    .await
+                }
+                // Audio frames captured from the user microphone.
+                user_audio_frame = audio_rx.recv() => {
+                    handle_user_audio_input(user_audio_frame, &writer, &events_tx)
+                        .await
+                }
+            };
+            if result.is_err() {
+                break;
+            }
+        }
+    })
+}
+
+async fn handle_user_text_input(
+    text: Result<String, RecvError>,
+    writer: &RealtimeWebsocketWriter,
+    events_tx: &Sender<RealtimeEvent>,
+    session_kind: RealtimeSessionKind,
+    response_create_state: &mut ResponseCreateState,
+) -> anyhow::Result<()> {
+    let text = text.context("user text input channel closed")?;
+
+    if let Err(err) = writer.send_conversation_item_create(text).await {
+        let mapped_error = map_api_error(err);
+        warn!("failed to send input text: {mapped_error}");
+        let _ = events_tx
+            .send(RealtimeEvent::Error(mapped_error.to_string()))
+            .await;
+        return Err(mapped_error.into());
+    }
+    match session_kind {
+        RealtimeSessionKind::V1 => {}
+        RealtimeSessionKind::V2 => {
+            if response_create_state.response_in_progress {
+                response_create_state.pending_response_create = true;
+            } else if let Err(err) = writer.send_response_create().await {
+                let mapped_error = map_api_error(err);
+                warn!("failed to send text response.create: {mapped_error}");
+                let _ = events_tx
+                    .send(RealtimeEvent::Error(mapped_error.to_string()))
+                    .await;
+                return Err(mapped_error.into());
+            } else {
+                response_create_state.pending_response_create = false;
+                response_create_state.response_in_progress = true;
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn handle_handoff_output(
+    handoff_output: Result<HandoffOutput, RecvError>,
+    writer: &RealtimeWebsocketWriter,
+    events_tx: &Sender<RealtimeEvent>,
+    session_kind: RealtimeSessionKind,
+    response_create_state: &mut ResponseCreateState,
+) -> anyhow::Result<()> {
+    let handoff_output = handoff_output.context("handoff output channel closed")?;
+
+    let should_create_response = match handoff_output {
+        HandoffOutput::ImmediateAppend {
+            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}");
+                let _ = events_tx
+                    .send(RealtimeEvent::Error(mapped_error.to_string()))
+                    .await;
+                return Err(mapped_error.into());
+            }
+            false
+        }
+        HandoffOutput::FinalToolCall {
+            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}");
+                let _ = events_tx
+                    .send(RealtimeEvent::Error(mapped_error.to_string()))
+                    .await;
+                return Err(mapped_error.into());
+            }
+            match session_kind {
+                RealtimeSessionKind::V1 => false,
+                RealtimeSessionKind::V2 => true,
+            }
+        }
+    };
+    if should_create_response {
+        if response_create_state.response_in_progress {
+            response_create_state.pending_response_create = true;
+        } else if let Err(err) = writer.send_response_create().await {
+            let mapped_error = map_api_error(err);
+            warn!("failed to send handoff response.create: {mapped_error}");
+            let _ = events_tx
+                .send(RealtimeEvent::Error(mapped_error.to_string()))
+                .await;
+            return Err(mapped_error.into());
+        } else {
+            response_create_state.pending_response_create = false;
+            response_create_state.response_in_progress = true;
+        }
+    }
+    Ok(())
+}
+
+async fn handle_realtime_server_event(
+    event: Result<Option<RealtimeEvent>, ApiError>,
+    writer: &RealtimeWebsocketWriter,
+    events_tx: &Sender<RealtimeEvent>,
+    handoff_state: &RealtimeHandoffState,
+    session_kind: RealtimeSessionKind,
+    output_audio_state: &mut Option<OutputAudioState>,
+    response_create_state: &mut ResponseCreateState,
+) -> anyhow::Result<()> {
+    let event = match event {
+        Ok(Some(event)) => event,
+        Ok(None) => anyhow::bail!("realtime event stream ended"),
+        Err(err) => {
+            let mapped_error = map_api_error(err);
+            if events_tx
+                .send(RealtimeEvent::Error(mapped_error.to_string()))
+                .await
+                .is_err()
+            {
+                return Err(mapped_error.into());
+            }
+            error!("realtime stream closed: {mapped_error}");
+            return Err(mapped_error.into());
+        }
+    };
+
+    let mut forward_event = true;
+    let should_stop = match &event {
+        RealtimeEvent::ConversationItemAdded(item) => {
+            match session_kind {
+                RealtimeSessionKind::V1 => {}
+                RealtimeSessionKind::V2 => {
+                    if let Some("response.created") = item.get("type").and_then(Value::as_str) {
+                        response_create_state.response_in_progress = true;
                     }
                 }
-                handoff_output = handoff_output_rx.recv() => {
-                    match handoff_output {
-                        Ok(handoff_output) => {
-                            match handoff_output {
-                                HandoffOutput::ImmediateAppend {
-                                    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}");
-                                        let _ = events_tx
-                                            .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                            .await;
-                                        break;
-                                    }
-                                }
-                                HandoffOutput::FinalToolCall {
-                                    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}");
-                                        let _ = events_tx
-                                            .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                            .await;
-                                        break;
-                                    }
-                                    if matches!(session_kind, RealtimeSessionKind::V2) {
-                                        if response_in_progress {
-                                            pending_response_create = true;
-                                        } else if let Err(err) = writer.send_response_create().await {
-                                            let mapped_error = map_api_error(err);
-                                            warn!(
-                                                "failed to send handoff response.create: {mapped_error}"
-                                            );
-                                            let _ = events_tx
-                                                .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                                .await;
-                                            break;
-                                        } else {
-                                            pending_response_create = false;
-                                            response_in_progress = true;
-                                        }
-                                    }
-                                }
-                            }
-                        }
-                        Err(_) => break,
+            }
+            false
+        }
+        RealtimeEvent::AudioOut(frame) => {
+            match session_kind {
+                RealtimeSessionKind::V1 => {}
+                RealtimeSessionKind::V2 => {
+                    update_output_audio_state(output_audio_state, frame);
+                }
+            }
+            false
+        }
+        RealtimeEvent::InputAudioSpeechStarted(event) => {
+            match session_kind {
+                RealtimeSessionKind::V1 => {}
+                RealtimeSessionKind::V2 => {
+                    if let Some(output_audio_state) = output_audio_state.take()
+                        && event
+                            .item_id
+                            .as_deref()
+                            .is_none_or(|item_id| item_id == output_audio_state.item_id)
+                        && let Err(err) = writer
+                            .send_payload(
+                                json!({
+                                    "type": "conversation.item.truncate",
+                                    "item_id": output_audio_state.item_id,
+                                    "content_index": 0,
+                                    "audio_end_ms": output_audio_state.audio_end_ms,
+                                })
+                                .to_string(),
+                            )
+                            .await
+                    {
+                        let mapped_error = map_api_error(err);
+                        warn!("failed to truncate realtime audio: {mapped_error}");
                     }
                 }
-                event = events.next_event() => {
-                    match event {
-                        Ok(Some(event)) => {
-                            let mut should_stop = false;
-                            let mut forward_event = true;
-
-                            match &event {
-                                RealtimeEvent::ConversationItemAdded(item) => {
-                                    match item.get("type").and_then(Value::as_str) {
-                                        Some("response.created")
-                                            if matches!(session_kind, RealtimeSessionKind::V2) =>
-                                        {
-                                            response_in_progress = true;
-                                        }
-                                        _ => {}
-                                    }
-                                }
-                                RealtimeEvent::AudioOut(frame) => {
-                                    if matches!(session_kind, RealtimeSessionKind::V2) {
-                                        update_output_audio_state(&mut output_audio_state, frame);
-                                    }
-                                }
-                                RealtimeEvent::InputAudioSpeechStarted(event) => {
-                                    if matches!(session_kind, RealtimeSessionKind::V2)
-                                        && let Some(output_audio_state) =
-                                            output_audio_state.take()
-                                        && event
-                                            .item_id
-                                            .as_deref()
-                                            .is_none_or(|item_id| item_id == output_audio_state.item_id)
-                                        && let Err(err) = writer
-                                            .send_payload(json!({
-                                                "type": "conversation.item.truncate",
-                                                "item_id": output_audio_state.item_id,
-                                                "content_index": 0,
-                                                "audio_end_ms": output_audio_state.audio_end_ms,
-                                            })
-                                            .to_string())
-                                            .await
-                                    {
-                                        let mapped_error = map_api_error(err);
-                                        warn!("failed to truncate realtime audio: {mapped_error}");
-                                    }
-                                }
-                                RealtimeEvent::ResponseCancelled(_) => {
-                                    response_in_progress = false;
-                                    output_audio_state = None;
-                                    if matches!(session_kind, RealtimeSessionKind::V2)
-                                        && pending_response_create
-                                    {
-                                        if let Err(err) = writer.send_response_create().await {
-                                            let mapped_error = map_api_error(err);
-                                            warn!(
-                                                "failed to send deferred response.create after cancellation: {mapped_error}"
-                                            );
-                                            let _ = events_tx
-                                                .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                                .await;
-                                            break;
-                                        }
-                                        pending_response_create = false;
-                                        response_in_progress = true;
-                                    }
-                                }
-                                RealtimeEvent::HandoffRequested(handoff) => {
-                                    *handoff_state.active_handoff.lock().await =
-                                        Some(handoff.handoff_id.clone());
-                                    *handoff_state.last_output_text.lock().await = None;
-                                    response_in_progress = false;
-                                    output_audio_state = None;
-                                }
-                                RealtimeEvent::Error(message)
-                                    if matches!(session_kind, RealtimeSessionKind::V2)
-                                        && message.starts_with(ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX) =>
-                                {
-                                    warn!(
-                                        "realtime rejected response.create because a response is already in progress; deferring follow-up response.create"
-                                    );
-                                    pending_response_create = true;
-                                    response_in_progress = true;
-                                    forward_event = false;
-                                }
-                                RealtimeEvent::Error(_) => {
-                                    should_stop = true;
-                                }
-                                RealtimeEvent::SessionUpdated { .. }
-                                | RealtimeEvent::InputTranscriptDelta(_)
-                                | RealtimeEvent::OutputTranscriptDelta(_)
-                                | RealtimeEvent::ConversationItemDone { .. } => {}
-                            }
-                            if forward_event && events_tx.send(event).await.is_err() {
-                                break;
-                            }
-                            if should_stop {
-                                error!("realtime stream error event received");
-                                break;
-                            }
-                        }
-                        Ok(None) => {
-                            break;
-                        }
-                        Err(err) => {
+            }
+            false
+        }
+        RealtimeEvent::ResponseCancelled(_) => {
+            response_create_state.response_in_progress = false;
+            *output_audio_state = None;
+            match session_kind {
+                RealtimeSessionKind::V1 => {}
+                RealtimeSessionKind::V2 => {
+                    if response_create_state.pending_response_create {
+                        if let Err(err) = writer.send_response_create().await {
                             let mapped_error = map_api_error(err);
-                            if events_tx
+                            warn!(
+                                "failed to send deferred response.create after cancellation: {mapped_error}"
+                            );
+                            let _ = events_tx
                                 .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                .await
-                                .is_err()
-                            {
-                                break;
-                            }
-                            error!("realtime stream closed: {mapped_error}");
-                            break;
-                        }
-                    }
-                }
-                frame = audio_rx.recv() => {
-                    match frame {
-                        Ok(frame) => {
-                            if let Err(err) = writer.send_audio_frame(frame).await {
-                                let mapped_error = map_api_error(err);
-                                error!("failed to send input audio: {mapped_error}");
-                                let _ = events_tx
-                                    .send(RealtimeEvent::Error(mapped_error.to_string()))
-                                    .await;
-                                break;
-                            }
+                                .await;
+                            return Err(mapped_error.into());
                         }
-                        Err(_) => break,
+                        response_create_state.pending_response_create = false;
+                        response_create_state.response_in_progress = true;
                     }
                 }
             }
+            false
         }
-    })
+        RealtimeEvent::HandoffRequested(handoff) => {
+            *handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone());
+            *handoff_state.last_output_text.lock().await = None;
+            response_create_state.response_in_progress = false;
+            *output_audio_state = None;
+            false
+        }
+        RealtimeEvent::Error(message) => match session_kind {
+            RealtimeSessionKind::V1 => true,
+            RealtimeSessionKind::V2 => {
+                if message.starts_with(ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX) {
+                    warn!(
+                        "realtime rejected response.create because a response is already in progress; deferring follow-up response.create"
+                    );
+                    response_create_state.pending_response_create = true;
+                    response_create_state.response_in_progress = true;
+                    forward_event = false;
+                    false
+                } else {
+                    true
+                }
+            }
+        },
+        RealtimeEvent::SessionUpdated { .. }
+        | RealtimeEvent::InputTranscriptDelta(_)
+        | RealtimeEvent::OutputTranscriptDelta(_)
+        | RealtimeEvent::ConversationItemDone { .. } => false,
+    };
+
+    if forward_event && events_tx.send(event).await.is_err() {
+        anyhow::bail!("realtime output event channel closed");
+    }
+    if should_stop {
+        error!("realtime stream error event received");
+        anyhow::bail!("realtime stream error event received");
+    }
+    Ok(())
+}
+
+async fn handle_user_audio_input(
+    frame: Result<RealtimeAudioFrame, RecvError>,
+    writer: &RealtimeWebsocketWriter,
+    events_tx: &Sender<RealtimeEvent>,
+) -> anyhow::Result<()> {
+    let frame = frame.context("user audio input channel closed")?;
+
+    if let Err(err) = writer.send_audio_frame(frame).await {
+        let mapped_error = map_api_error(err);
+        error!("failed to send input audio: {mapped_error}");
+        let _ = events_tx
+            .send(RealtimeEvent::Error(mapped_error.to_string()))
+            .await;
+        return Err(mapped_error.into());
+    }
+    Ok(())
 }
 
 fn update_output_audio_state(
