codex/codex-rs/core/src/realtime_conversation.rs

use crate::client::ModelClient;
use crate::client::X_CODEX_TURN_METADATA_HEADER;
use crate::context::ContextualUserFragment;
use crate::context::RealtimeDelegation;
use crate::context::RealtimeDelegationSource;
use crate::realtime_context::build_realtime_startup_context;
use crate::realtime_context::truncate_realtime_text_to_token_budget;
use crate::realtime_prompt::prepare_realtime_backend_prompt;
use crate::responses_metadata::THREAD_SOURCE_KEY;
use crate::session::session::Session;
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::RealtimeContextAppendChannel;
use codex_api::RealtimeEvent;
use codex_api::RealtimeEventParser;
use codex_api::RealtimeSessionConfig;
use codex_api::RealtimeSessionMode;
use codex_api::RealtimeWebsocketClient;
use codex_api::RealtimeWebsocketConnection;
use codex_api::RealtimeWebsocketEvents;
use codex_api::RealtimeWebsocketWriter;
use codex_api::build_session_headers;
use codex_api::map_api_error;
use codex_config::config_toml::RealtimeWsMode;
use codex_config::config_toml::RealtimeWsVersion;
use codex_http_client::HttpClientFactory;
use codex_login::CodexAuth;
use codex_login::default_client::add_originator_header;
use codex_login::default_client::default_headers;
use codex_login::read_openai_api_key_from_env;
use codex_model_provider_info::ModelProviderInfo;
use codex_protocol::auth::AuthMode;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::MessagePhase;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::CodexResponseHandoffMode;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationSpeechParams;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::ConversationStartTransport;
use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::ConversationTextRole;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RealtimeConversationClosedEvent;
use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
use codex_protocol::protocol::RealtimeConversationSdpEvent;
use codex_protocol::protocol::RealtimeConversationStartedEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeOutputModality;
use codex_protocol::protocol::RealtimeTranscriptEntry;
use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RealtimeVoicesList;
use codex_utils_output_truncation::approx_bytes_for_tokens;
use codex_utils_string::approx_token_count;
use codex_utils_string::take_bytes_at_char_boundary;
use codex_utils_string::to_ascii_json_string;
use http::HeaderMap;
use http::HeaderValue;
use http::header::AUTHORIZATION;
use serde_json::json;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::warn;

mod bem;
mod existing_call;
mod sideband;
mod transcript_tail;

use self::bem::ChannelParser as BemChannelParser;
use self::bem::message_phase as bem_message_phase;
use self::sideband::spawn_webrtc_sideband_input_task;

const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
const TEXT_IN_QUEUE_CAPACITY: usize = 64;
const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256;
const REALTIME_THREAD_SOURCE_MAX_BYTES: usize = 256;
const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300;
const REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET: usize = 1_000;
const REALTIME_REASONING_STATUS_MAX_BYTES: usize = 256;
const REALTIME_INITIAL_ITEMS_MAX_COUNT: usize = 128;
const REALTIME_INITIAL_ITEMS_MAX_TOKENS: usize = 8_192;
const REALTIME_MODE_INSTRUCTIONS_MAX_TOKENS: usize = 8_192;
const HANDOFF_STREAM_FLUSH_INTERVAL: Duration = Duration::from_millis(200);
const HANDOFF_STREAM_TRUNCATION_MARKER: &str = "\n…output truncated…\n";
const AGENT_FINAL_MESSAGE_PREFIX: &str = "\"Agent Final Message\":\n\n";
const STANDALONE_HANDOFF_ID: &str = "codex";
const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5";
const DEFAULT_FRAMELESS_REALTIME_MODEL: &str = "gpt-live-1-codex";
pub(crate) const REALTIME_USER_TEXT_PREFIX: &str = "[USER] ";
pub(crate) const REALTIME_BACKEND_TEXT_PREFIX: &str = "[BACKEND] ";
const REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str =
    "Background agent finished. Use the preceding [BACKEND] messages as the result.";
const REALTIME_V2_STEER_ACKNOWLEDGEMENT: &str =
    "This was sent to steer the previous background agent task.";
const REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX: &str =
    "Conversation already has an active response in progress:";
const REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION: &str = "The user just ended their realtime session. Here is the remaining handoff/transcript tail. You probably do not have to do anything; acknowledge the handoff unless the transcript itself asks for something.";

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RealtimeConversationEnd {
    Requested,
    TransportClosed,
    Error,
}

enum RealtimeFanoutTaskStop {
    Await,
    Detach,
}

/// Serializes turn admission with the safety cutoff for one voice session.
struct RealtimeHandoffAdmission {
    gate: Semaphore,
    retired: AtomicBool,
}

impl RealtimeHandoffAdmission {
    fn new() -> Self {
        Self {
            gate: Semaphore::new(1),
            retired: AtomicBool::new(false),
        }
    }

    async fn route(&self, session: &Arc<Session>, text: String) -> Result<(), &'static str> {
        let Ok(_permit) = self.gate.acquire().await else {
            return Ok(());
        };
        if !self.retired.load(Ordering::Acquire) {
            session.route_realtime_text_input(text).await?;
        }
        Ok(())
    }

    async fn retire(&self) {
        let Ok(_permit) = self.gate.acquire().await else {
            return;
        };
        self.retired.store(true, Ordering::Release);
    }
}

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

struct RealtimeConversationManagerState {
    conversation: Option<ConversationState>,
    mode_instructions: Option<RealtimeModeInstructions>,
    // Keep context active until the terminal transcript is recorded.
    context_active: bool,
}

#[derive(Clone, Debug)]
pub(crate) struct RealtimeModeInstructions {
    pub(crate) start: Option<String>,
    pub(crate) end: Option<String>,
}

#[derive(Clone, Debug)]
pub(crate) struct RealtimeConversationSnapshot {
    pub(crate) active: bool,
    pub(crate) mode_instructions: Option<RealtimeModeInstructions>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RealtimeSessionKind {
    V1,
    V2,
}

#[derive(Clone, Debug)]
struct RealtimeHandoffState {
    output_tx: Sender<RealtimeOutbound>,
    last_output: Arc<Mutex<Option<RealtimeHandoffOutput>>>,
    stream: Arc<Mutex<RealtimeHandoffStreamState>>,
    client_managed_handoffs: bool,
    codex_responses_as_items: bool,
    codex_response_item_prefix: Option<String>,
    codex_response_handoff_mode: CodexResponseHandoffMode,
    backend_reasoning_status: bool,
    codex_response_handoff_channel_prefixes: Arc<BTreeMap<String, Vec<String>>>,
    session_kind: RealtimeSessionKind,
    event_parser: RealtimeEventParser,
}

#[derive(Clone, Debug)]
struct RealtimeHandoffOutput {
    text: String,
    phase: Option<MessagePhase>,
}

#[derive(Debug, Default)]
struct RealtimeHandoffStreamState {
    active_handoff: Option<String>,
    items: HashMap<String, RealtimeStreamedItem>,
}

#[derive(Debug)]
struct RealtimeStreamedItem {
    handoff_id: String,
    phase: Option<MessagePhase>,
    bem_channel_parser: Option<BemChannelParser>,
    prefix_final_message: bool,
    sent_bytes: usize,
    buffered_text: String,
    tail_text: String,
    truncated: bool,
    last_flush_at: Instant,
    flush_scheduled: bool,
}

impl RealtimeStreamedItem {
    fn next_flush_delay(&self) -> Duration {
        HANDOFF_STREAM_FLUSH_INTERVAL.saturating_sub(self.last_flush_at.elapsed())
    }

    fn output_prefix(&self) -> &'static str {
        if self.prefix_final_message
            && self.sent_bytes == 0
            && matches!(self.phase, Some(MessagePhase::FinalAnswer) | None)
        {
            AGENT_FINAL_MESSAGE_PREFIX
        } else {
            ""
        }
    }

    fn stream_head_byte_limit(&self) -> usize {
        let output_byte_limit = approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET);
        output_byte_limit.saturating_sub(HANDOFF_STREAM_TRUNCATION_MARKER.len()) / 2
    }

    fn tail_byte_limit(&self) -> usize {
        approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
            .saturating_sub(self.stream_head_byte_limit())
            .saturating_sub(HANDOFF_STREAM_TRUNCATION_MARKER.len())
    }

    fn streamable_text_bytes(&self) -> usize {
        self.stream_head_byte_limit()
            .saturating_sub(self.sent_bytes)
            .saturating_sub(self.output_prefix().len())
    }

    fn push_text(&mut self, text: &str) {
        if text.is_empty() {
            return;
        }

        let text = if let Some(parser) = self.bem_channel_parser.as_mut() {
            let Some(text) = parser.push(text) else {
                return;
            };
            self.phase = parser.phase();
            text
        } else {
            text.to_string()
        };
        self.push_output_text(&text);
    }

    fn finish_input(&mut self) {
        let Some(parser) = self.bem_channel_parser.as_mut() else {
            return;
        };
        self.phase = parser.phase();
        let text = parser.finish();
        if self.phase.is_none() && !text.is_empty() {
            warn!("BEM output ended before a recognized channel header was received");
            self.phase = Some(MessagePhase::FinalAnswer);
        }
        self.push_output_text(&text);
    }

    fn push_output_text(&mut self, text: &str) {
        if text.is_empty() {
            return;
        }
        if self.truncated {
            self.tail_text.push_str(text);
            self.tail_text =
                take_last_bytes_at_char_boundary(&self.tail_text, self.tail_byte_limit())
                    .to_string();
            return;
        }

        self.buffered_text.push_str(text);
        let output_byte_limit = approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET);
        let remaining_text_bytes = output_byte_limit
            .saturating_sub(self.sent_bytes)
            .saturating_sub(self.output_prefix().len());
        if self.buffered_text.len() <= remaining_text_bytes {
            return;
        }

        let head_bytes =
            take_bytes_at_char_boundary(&self.buffered_text, self.streamable_text_bytes()).len();
        self.tail_text =
            take_last_bytes_at_char_boundary(&self.buffered_text, self.tail_byte_limit())
                .to_string();
        self.buffered_text.truncate(head_bytes);
        self.truncated = true;
    }

    fn drain_stream_chunk(&mut self) -> Option<String> {
        let prefix = self.output_prefix();
        let available_text_bytes = self.streamable_text_bytes();
        if self.buffered_text.is_empty() || available_text_bytes == 0 {
            return None;
        }

        let requested_bytes = available_text_bytes.min(self.buffered_text.len());
        let split_at = take_bytes_at_char_boundary(&self.buffered_text, requested_bytes).len();
        if split_at == 0 {
            return None;
        }
        let text = self.buffered_text.drain(..split_at).collect::<String>();
        let text = format!("{prefix}{text}");
        self.sent_bytes += text.len();
        Some(text)
    }

    fn drain_final_chunk(&mut self) -> Option<String> {
        let prefix = self.output_prefix();
        if !self.truncated {
            if self.buffered_text.is_empty() {
                return None;
            }
            let text = self.buffered_text.drain(..).collect::<String>();
            let text = format!("{prefix}{text}");
            self.sent_bytes += text.len();
            return Some(text);
        }

        let head = self.buffered_text.drain(..).collect::<String>();
        let tail = self.tail_text.drain(..).collect::<String>();
        let text = format!("{prefix}{head}{HANDOFF_STREAM_TRUNCATION_MARKER}{tail}");
        self.sent_bytes += text.len();
        Some(text)
    }
}

fn take_last_bytes_at_char_boundary(text: &str, max_bytes: usize) -> &str {
    let mut start = text.len().saturating_sub(max_bytes);
    while start < text.len() && !text.is_char_boundary(start) {
        start += 1;
    }
    &text[start..]
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum RealtimeOutbound {
    QuietReasoningStatus {
        handoff_id: String,
        text: String,
    },
    StandaloneHandoff {
        text: String,
        phase: Option<MessagePhase>,
    },
    StandaloneSpeech {
        text: String,
    },
    HandoffUpdate {
        handoff_id: String,
        text: String,
        phase: Option<MessagePhase>,
    },
    HandoffAppend {
        handoff_id: String,
        text: String,
        phase: Option<MessagePhase>,
    },
    CompletedHandoff {
        handoff_id: String,
        text: String,
        phase: Option<MessagePhase>,
    },
    ConversationItem {
        text: String,
        phase: Option<MessagePhase>,
    },
    HandoffCompleteAck {
        handoff_id: String,
    },
}

#[derive(Debug, PartialEq, Eq)]
struct OutputAudioState {
    item_id: String,
    audio_end_ms: u32,
}

#[derive(Default)]
struct RealtimeResponseCreateQueue {
    active_default_response: bool,
    pending_create: bool,
}

impl RealtimeResponseCreateQueue {
    async fn request_create(
        &mut self,
        writer: &RealtimeWebsocketWriter,
        reason: &str,
    ) -> anyhow::Result<()> {
        if self.active_default_response {
            self.pending_create = true;
            return Ok(());
        }
        self.send_create_now(writer, reason).await
    }

    fn mark_started(&mut self) {
        self.active_default_response = true;
    }

    async fn mark_finished(
        &mut self,
        writer: &RealtimeWebsocketWriter,
        reason: &str,
    ) -> anyhow::Result<()> {
        self.active_default_response = false;
        if !self.pending_create {
            return Ok(());
        }
        self.pending_create = false;
        self.send_create_now(writer, reason).await
    }

    async fn send_create_now(
        &mut self,
        writer: &RealtimeWebsocketWriter,
        reason: &str,
    ) -> anyhow::Result<()> {
        if let Err(err) = writer.send_response_create().await {
            if matches!(&err, ApiError::Stream(message) if message.starts_with(REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX))
            {
                warn!("realtime response.create raced an active response; deferring");
                self.active_default_response = true;
                self.pending_create = true;
                return Ok(());
            }
            warn!("failed to send {reason} response.create: {err}");
            return Err(err.into());
        }
        self.active_default_response = true;
        Ok(())
    }
}

struct RealtimeInputTask {
    writer: RealtimeWebsocketWriter,
    events: RealtimeWebsocketEvents,
    text_rx: Receiver<ConversationTextParams>,
    handoff_output_rx: Receiver<RealtimeOutbound>,
    audio_rx: Receiver<RealtimeAudioFrame>,
    events_tx: Sender<RealtimeEvent>,
    handoff_state: RealtimeHandoffState,
    session_kind: RealtimeSessionKind,
    event_parser: RealtimeEventParser,
    stop_token: CancellationToken,
}

struct RealtimeTranscriptTailFlush {
    enabled: bool,
    tx: Sender<String>,
}

struct RealtimeInputChannels {
    text_rx: Receiver<ConversationTextParams>,
    handoff_output_rx: Receiver<RealtimeOutbound>,
    audio_rx: Receiver<RealtimeAudioFrame>,
}

impl RealtimeHandoffState {
    fn streams_handoff_append(&self) -> bool {
        self.event_parser == RealtimeEventParser::FramelessBidi
            && !self.client_managed_handoffs
            && !self.codex_responses_as_items
    }

    fn routes_handoff_by_bem(&self) -> bool {
        self.event_parser == RealtimeEventParser::FramelessBidi
            && self.codex_response_handoff_mode == CodexResponseHandoffMode::BemTags
    }
}

#[allow(dead_code)]
struct ConversationState {
    session_id: Option<String>,
    sub_id: String,
    audio_tx: Sender<RealtimeAudioFrame>,
    text_tx: Sender<ConversationTextParams>,
    session_kind: RealtimeSessionKind,
    handoff: RealtimeHandoffState,
    input_task: JoinHandle<()>,
    fanout_task: Option<JoinHandle<()>>,
    realtime_active: Arc<AtomicBool>,
    // A misalignment failure retires only this session's queued handoffs.
    route_handoffs: Arc<RealtimeHandoffAdmission>,
    stop_token: CancellationToken,
}

struct RealtimeStart {
    api_provider: ApiProvider,
    http_client_factory: HttpClientFactory,
    realtime_sideband_base_url: Option<String>,
    extra_headers: Option<HeaderMap>,
    client_managed_handoffs: bool,
    flush_transcript_tail_on_session_end: bool,
    codex_responses_as_items: bool,
    codex_response_item_prefix: Option<String>,
    codex_response_handoff_mode: CodexResponseHandoffMode,
    backend_reasoning_status: bool,
    codex_response_handoff_channel_prefixes: Option<BTreeMap<String, Vec<String>>>,
    realtime_call_api_provider: Option<ApiProvider>,
    session_config: RealtimeSessionConfig,
    model_client: ModelClient,
    sdp: Option<String>,
    existing_call_id: Option<String>,
}

struct RealtimeStartOutput {
    realtime_active: Arc<AtomicBool>,
    route_handoffs: Arc<RealtimeHandoffAdmission>,
    events_rx: Receiver<RealtimeEvent>,
    transcript_tail_rx: Receiver<String>,
    sdp: Option<String>,
}

#[allow(dead_code)]
impl RealtimeConversationManager {
    pub(crate) fn new() -> Self {
        Self {
            state: Mutex::new(RealtimeConversationManagerState {
                conversation: None,
                mode_instructions: None,
                context_active: false,
            }),
        }
    }

    pub(crate) async fn snapshot(&self) -> RealtimeConversationSnapshot {
        let state = self.state.lock().await;
        RealtimeConversationSnapshot {
            active: state.context_active,
            mode_instructions: state.mode_instructions.clone(),
        }
    }

    #[tracing::instrument(
        name = "realtime_conversation.running_state",
        level = "trace",
        skip_all
    )]
    pub(crate) async fn running_state(&self) -> Option<()> {
        let state = self.state.lock().await;
        state
            .conversation
            .as_ref()
            .and_then(|state| state.realtime_active.load(Ordering::Relaxed).then_some(()))
    }

    pub(crate) async fn is_running_v2(&self) -> bool {
        let state = self.state.lock().await;
        matches!(
            state.conversation.as_ref(),
            Some(state)
                if state.realtime_active.load(Ordering::Relaxed)
                    && state.session_kind == RealtimeSessionKind::V2
        )
    }

    /// Prevent already parsed handoffs from starting turns after a safety failure.
    /// Keep the gate on the old session so a later voice session starts fresh.
    pub(crate) async fn retire_handoffs_for_misalignment(&self) {
        let route_handoffs = self
            .state
            .lock()
            .await
            .conversation
            .as_ref()
            .map(|state| Arc::clone(&state.route_handoffs));
        if let Some(route_handoffs) = route_handoffs {
            route_handoffs.retire().await;
        }
    }

    async fn start(
        &self,
        start: RealtimeStart,
        mode_instructions: RealtimeModeInstructions,
        sess: &Arc<Session>,
        sub_id: &str,
    ) -> CodexResult<RealtimeStartOutput> {
        let previous_state = {
            let mut guard = self.state.lock().await;
            guard.conversation.take()
        };
        if let Some(state) = previous_state {
            let previous_sub_id = state.sub_id.clone();
            stop_conversation_state(state, RealtimeFanoutTaskStop::Await).await;
            send_realtime_conversation_closed(
                sess,
                previous_sub_id,
                RealtimeConversationEnd::Requested,
            )
            .await;
        }

        self.start_inner(start, mode_instructions, sub_id).await
    }

    async fn start_inner(
        &self,
        start: RealtimeStart,
        mode_instructions: RealtimeModeInstructions,
        sub_id: &str,
    ) -> CodexResult<RealtimeStartOutput> {
        let session_id = start.session_config.session_id.clone();
        let RealtimeStart {
            api_provider,
            http_client_factory,
            realtime_sideband_base_url,
            extra_headers,
            client_managed_handoffs,
            flush_transcript_tail_on_session_end,
            codex_responses_as_items,
            codex_response_item_prefix,
            codex_response_handoff_mode,
            backend_reasoning_status,
            codex_response_handoff_channel_prefixes,
            realtime_call_api_provider,
            session_config,
            model_client,
            sdp,
            existing_call_id,
        } = start;
        let event_parser = session_config.event_parser;
        let session_kind = match event_parser {
            RealtimeEventParser::V1 | RealtimeEventParser::FramelessBidi => RealtimeSessionKind::V1,
            RealtimeEventParser::RealtimeV2 => RealtimeSessionKind::V2,
        };

        let (audio_tx, audio_rx) =
            async_channel::bounded::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
        let (text_tx, text_rx) =
            async_channel::bounded::<ConversationTextParams>(TEXT_IN_QUEUE_CAPACITY);
        let (handoff_output_tx, handoff_output_rx) =
            async_channel::bounded::<RealtimeOutbound>(HANDOFF_OUT_QUEUE_CAPACITY);
        let (events_tx, events_rx) =
            async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
        let (transcript_tail_tx, transcript_tail_rx) = async_channel::bounded::<String>(1);

        let realtime_active = Arc::new(AtomicBool::new(true));
        let route_handoffs = Arc::new(RealtimeHandoffAdmission::new());
        let stop_token = CancellationToken::new();
        let handoff = RealtimeHandoffState {
            output_tx: handoff_output_tx,
            last_output: Arc::new(Mutex::new(None)),
            stream: Arc::new(Mutex::new(RealtimeHandoffStreamState::default())),
            client_managed_handoffs,
            codex_responses_as_items,
            codex_response_item_prefix,
            codex_response_handoff_mode,
            backend_reasoning_status,
            codex_response_handoff_channel_prefixes: Arc::new(
                codex_response_handoff_channel_prefixes.unwrap_or_default(),
            ),
            session_kind,
            event_parser,
        };
        let input_channels = RealtimeInputChannels {
            text_rx,
            handoff_output_rx,
            audio_rx,
        };

        let client = RealtimeWebsocketClient::new(api_provider, http_client_factory);
        let client = match realtime_sideband_base_url {
            Some(base_url) => client.with_webrtc_sideband_base_url(base_url),
            None => client,
        };
        let transcript_tail_flush = RealtimeTranscriptTailFlush {
            enabled: flush_transcript_tail_on_session_end,
            tx: transcript_tail_tx,
        };
        let (task, sdp) = if let Some(sdp) = sdp {
            let call = model_client
                .create_realtime_call_with_headers(
                    sdp,
                    session_config.clone(),
                    extra_headers.unwrap_or_default(),
                    realtime_call_api_provider,
                )
                .await?;
            let task = spawn_webrtc_sideband_input_task(RealtimeWebrtcSidebandInputTask {
                client,
                session_config,
                call_id: call.call_id,
                sideband_headers: call.sideband_headers,
                session_initialization: RealtimeSidebandSessionInitialization::CoreCreated,
                input_channels,
                events_tx,
                handoff_state: handoff.clone(),
                session_kind,
                event_parser,
                realtime_active: Arc::clone(&realtime_active),
                transcript_tail_flush,
                stop_token: stop_token.clone(),
            });
            (task, Some(call.sdp))
        } else if let Some(call_id) = existing_call_id {
            let task = existing_call::attach(existing_call::ExistingCallAttachment {
                client,
                model_client,
                session_config,
                call_id,
                extra_headers: extra_headers.unwrap_or_default(),
                input_channels,
                events_tx,
                handoff_state: handoff.clone(),
                session_kind,
                event_parser,
                realtime_active: Arc::clone(&realtime_active),
                transcript_tail_flush,
                stop_token: stop_token.clone(),
            })
            .await?;
            (task, None)
        } else {
            let connection = client
                .connect(
                    session_config,
                    extra_headers.unwrap_or_default(),
                    default_headers(),
                )
                .await
                .map_err(map_api_error)?;
            let task = spawn_realtime_input_task(
                RealtimeInputTask {
                    writer: connection.writer(),
                    events: connection.events(),
                    text_rx: input_channels.text_rx,
                    handoff_output_rx: input_channels.handoff_output_rx,
                    audio_rx: input_channels.audio_rx,
                    events_tx,
                    handoff_state: handoff.clone(),
                    session_kind,
                    event_parser,
                    stop_token: stop_token.clone(),
                },
                transcript_tail_flush,
            );
            (task, None)
        };

        let mut state = self.state.lock().await;
        state.conversation = Some(ConversationState {
            session_id,
            sub_id: sub_id.to_owned(),
            audio_tx,
            text_tx,
            session_kind,
            handoff,
            input_task: task,
            fanout_task: None,
            realtime_active: Arc::clone(&realtime_active),
            route_handoffs: Arc::clone(&route_handoffs),
            stop_token,
        });
        state.mode_instructions = Some(mode_instructions);
        state.context_active = true;
        Ok(RealtimeStartOutput {
            realtime_active,
            route_handoffs,
            events_rx,
            transcript_tail_rx,
            sdp,
        })
    }

    pub(crate) async fn register_fanout_task(
        &self,
        realtime_active: &Arc<AtomicBool>,
        fanout_task: JoinHandle<()>,
    ) {
        let mut fanout_task = Some(fanout_task);
        {
            let mut guard = self.state.lock().await;
            if let Some(state) = guard.conversation.as_mut()
                && Arc::ptr_eq(&state.realtime_active, realtime_active)
            {
                state.fanout_task = fanout_task.take();
            }
        }

        if let Some(fanout_task) = fanout_task {
            fanout_task.abort();
            let _ = fanout_task.await;
        }
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "Publish the matching close before a replacement can start its history segment."
    )]
    async fn finish_if_active(
        &self,
        sess: &Arc<Session>,
        realtime_active: &Arc<AtomicBool>,
        sub_id: String,
        end: RealtimeConversationEnd,
    ) {
        // Keep selection and its close event together. An old fanout must not
        // close the history segment that belongs to a replacement.
        let mut guard = self.state.lock().await;
        if !guard
            .conversation
            .as_ref()
            .is_some_and(|state| Arc::ptr_eq(&state.realtime_active, realtime_active))
        {
            return;
        }
        let Some(state) = guard.conversation.take() else {
            return;
        };
        stop_conversation_state(state, RealtimeFanoutTaskStop::Detach).await;
        send_realtime_conversation_closed(sess, sub_id, end).await;
    }

    pub(crate) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> {
        let sender = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| state.audio_tx.clone())
        };

        let Some(sender) = sender else {
            return Err(CodexErr::InvalidRequest(
                "conversation is not running".to_string(),
            ));
        };

        match sender.try_send(frame) {
            Ok(()) => Ok(()),
            Err(TrySendError::Full(_)) => {
                warn!("dropping input audio frame due to full queue");
                Ok(())
            }
            Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest(
                "conversation is not running".to_string(),
            )),
        }
    }

    pub(crate) async fn text_in(&self, mut params: ConversationTextParams) -> CodexResult<()> {
        let sender = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| (state.text_tx.clone(), state.session_kind))
        };

        let Some((sender, session_kind)) = sender else {
            return Err(CodexErr::InvalidRequest(
                "conversation is not running".to_string(),
            ));
        };

        if params.role == ConversationTextRole::User {
            params.text =
                prefix_realtime_text(params.text, REALTIME_USER_TEXT_PREFIX, session_kind);
        }
        sender
            .send(params)
            .await
            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
        Ok(())
    }

    pub(crate) async fn send_reasoning_status(&self, text: &str) -> CodexResult<()> {
        let handoff = self
            .state
            .lock()
            .await
            .conversation
            .as_ref()
            .map(|state| state.handoff.clone());
        let Some(handoff) = handoff.filter(|handoff| {
            handoff.backend_reasoning_status
                && !handoff.client_managed_handoffs
                && handoff.event_parser == RealtimeEventParser::FramelessBidi
        }) else {
            return Ok(());
        };
        let Some(handoff_id) = handoff.stream.lock().await.active_handoff.clone() else {
            return Ok(());
        };
        let text = take_bytes_at_char_boundary(text, REALTIME_REASONING_STATUS_MAX_BYTES);
        handoff
            .output_tx
            .send(RealtimeOutbound::QuietReasoningStatus {
                handoff_id,
                text: format!("[STATUS] {text}"),
            })
            .await
            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))
    }

    pub(crate) async fn handoff_out(
        &self,
        output_text: String,
        phase: Option<MessagePhase>,
    ) -> CodexResult<()> {
        let handoff = {
            let guard = self.state.lock().await;
            let Some(state) = guard.conversation.as_ref() else {
                return Err(CodexErr::InvalidRequest(
                    "conversation is not running".to_string(),
                ));
            };
            state.handoff.clone()
        };

        if handoff.client_managed_handoffs {
            return Ok(());
        }
        let phase = if handoff.routes_handoff_by_bem()
            && !matches!(phase, Some(MessagePhase::PartialAnswer))
        {
            match bem_message_phase(
                &output_text,
                &handoff.codex_response_handoff_channel_prefixes,
            ) {
                Some(phase) => Some(phase),
                None => {
                    warn!("BEM output did not contain a recognized channel header");
                    Some(MessagePhase::FinalAnswer)
                }
            }
        } else {
            phase
        };
        let is_nonterminal = matches!(
            phase,
            Some(MessagePhase::Commentary | MessagePhase::PartialAnswer)
        );
        let active_handoff = handoff.stream.lock().await.active_handoff.clone();
        let output = match active_handoff {
            Some(handoff_id) => {
                let output_text = realtime_backend_output(output_text, handoff.session_kind);
                *handoff.last_output.lock().await = Some(RealtimeHandoffOutput {
                    text: output_text.clone(),
                    phase: phase.clone(),
                });
                if handoff.codex_responses_as_items {
                    RealtimeOutbound::ConversationItem {
                        text: realtime_backend_item(
                            output_text,
                            handoff.codex_response_item_prefix.as_deref(),
                        ),
                        phase,
                    }
                } else if handoff.event_parser == RealtimeEventParser::V1 && is_nonterminal {
                    RealtimeOutbound::HandoffAppend {
                        handoff_id,
                        text: output_text,
                        phase,
                    }
                } else {
                    RealtimeOutbound::HandoffUpdate {
                        handoff_id,
                        text: output_text,
                        phase,
                    }
                }
            }
            None if output_text.trim().is_empty() => return Ok(()),
            None => {
                let output_text = realtime_backend_output(output_text, handoff.session_kind);
                if handoff.codex_responses_as_items {
                    RealtimeOutbound::ConversationItem {
                        text: realtime_backend_item(
                            output_text,
                            handoff.codex_response_item_prefix.as_deref(),
                        ),
                        phase,
                    }
                } else {
                    RealtimeOutbound::StandaloneHandoff {
                        text: if handoff.event_parser == RealtimeEventParser::V1 && !is_nonterminal
                        {
                            format!("{AGENT_FINAL_MESSAGE_PREFIX}{output_text}")
                        } else {
                            output_text
                        },
                        phase,
                    }
                }
            }
        };
        handoff
            .output_tx
            .send(output)
            .await
            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
        Ok(())
    }

    pub(crate) async fn register_handoff_stream_item(
        &self,
        item_id: String,
        phase: Option<MessagePhase>,
        initial_text: String,
    ) {
        let handoff = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| state.handoff.clone())
        };
        let Some(handoff) = handoff else {
            return;
        };
        if !handoff.streams_handoff_append() {
            return;
        }
        let flush_delay = {
            let mut stream = handoff.stream.lock().await;
            let Some(handoff_id) = stream.active_handoff.clone() else {
                return;
            };
            // An explicit stable fragment is answer text, even when it starts
            // with characters that legacy BEM treats as a private channel tag.
            let parse_bem = handoff.routes_handoff_by_bem()
                && !matches!(phase, Some(MessagePhase::PartialAnswer));
            let mut streamed_item = RealtimeStreamedItem {
                handoff_id,
                phase: if parse_bem { None } else { phase },
                bem_channel_parser: parse_bem.then(|| {
                    BemChannelParser::new(Arc::clone(
                        &handoff.codex_response_handoff_channel_prefixes,
                    ))
                }),
                prefix_final_message: handoff.event_parser == RealtimeEventParser::V1,
                sent_bytes: 0,
                buffered_text: String::new(),
                tail_text: String::new(),
                truncated: false,
                last_flush_at: Instant::now(),
                flush_scheduled: false,
            };
            streamed_item.push_text(&initial_text);
            let flush_delay = if streamed_item.buffered_text.is_empty() {
                None
            } else {
                streamed_item.flush_scheduled = true;
                Some(streamed_item.next_flush_delay())
            };
            stream.items.insert(item_id.clone(), streamed_item);
            flush_delay
        };
        if let Some(flush_delay) = flush_delay {
            schedule_streamed_handoff_flush(&handoff, item_id, flush_delay);
        }
    }

    pub(crate) async fn stream_handoff_delta(
        &self,
        item_id: &str,
        delta: String,
    ) -> CodexResult<()> {
        if delta.is_empty() {
            return Ok(());
        }
        let handoff = {
            let guard = self.state.lock().await;
            let Some(state) = guard.conversation.as_ref() else {
                return Err(CodexErr::InvalidRequest(
                    "conversation is not running".to_string(),
                ));
            };
            state.handoff.clone()
        };
        if !handoff.streams_handoff_append() {
            return Ok(());
        }
        let flush_delay = {
            let mut stream = handoff.stream.lock().await;
            let Some(streamed_item) = stream.items.get_mut(item_id) else {
                return Ok(());
            };
            streamed_item.push_text(&delta);
            if streamed_item.flush_scheduled || streamed_item.streamable_text_bytes() == 0 {
                None
            } else {
                streamed_item.flush_scheduled = true;
                Some(streamed_item.next_flush_delay())
            }
        };
        if let Some(flush_delay) = flush_delay {
            schedule_streamed_handoff_flush(&handoff, item_id.to_string(), flush_delay);
        }
        Ok(())
    }

    pub(crate) async fn finish_handoff_stream_item(&self, item_id: &str) -> bool {
        let handoff = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| state.handoff.clone())
        };
        let Some(handoff) = handoff else {
            return false;
        };
        if !handoff.streams_handoff_append() {
            return false;
        }
        let Some(mut streamed_item) = handoff.stream.lock().await.items.remove(item_id) else {
            return false;
        };
        streamed_item.finish_input();
        let chunk = streamed_item.drain_final_chunk();
        let sent_output = streamed_item.sent_bytes > 0;
        if let Some(text) = chunk {
            let _ = handoff
                .output_tx
                .send(RealtimeOutbound::HandoffAppend {
                    handoff_id: streamed_item.handoff_id,
                    text,
                    phase: streamed_item.phase,
                })
                .await;
        }
        sent_output
    }

    pub(crate) async fn append_speech(&self, text: String) -> CodexResult<()> {
        if text.trim().is_empty() {
            return Ok(());
        }

        let handoff = {
            let guard = self.state.lock().await;
            let Some(state) = guard.conversation.as_ref() else {
                return Err(CodexErr::InvalidRequest(
                    "conversation is not running".to_string(),
                ));
            };
            state.handoff.clone()
        };

        handoff
            .output_tx
            .send(RealtimeOutbound::StandaloneSpeech {
                text: realtime_backend_output(text, handoff.session_kind),
            })
            .await
            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
        Ok(())
    }

    pub(crate) async fn handoff_complete(&self) -> CodexResult<()> {
        let handoff = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| state.handoff.clone())
        };
        let Some(handoff) = handoff else {
            return Ok(());
        };
        if handoff.client_managed_handoffs {
            return Ok(());
        }
        match handoff.session_kind {
            RealtimeSessionKind::V1 => return Ok(()),
            RealtimeSessionKind::V2 => {}
        }

        let Some(handoff_id) = handoff.stream.lock().await.active_handoff.clone() else {
            return Ok(());
        };
        let Some(last_output) = handoff.last_output.lock().await.clone() else {
            return Ok(());
        };

        let output = if handoff.codex_responses_as_items {
            RealtimeOutbound::HandoffCompleteAck { handoff_id }
        } else {
            RealtimeOutbound::CompletedHandoff {
                handoff_id,
                text: last_output.text,
                phase: last_output.phase,
            }
        };

        handoff
            .output_tx
            .send(output)
            .await
            .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))
    }

    pub(crate) async fn clear_active_handoff(&self) {
        let handoff = {
            let guard = self.state.lock().await;
            guard
                .conversation
                .as_ref()
                .map(|state| state.handoff.clone())
        };
        if let Some(handoff) = handoff {
            {
                let mut stream = handoff.stream.lock().await;
                stream.active_handoff = None;
                stream.items.clear();
            }
            *handoff.last_output.lock().await = None;
        }
    }

    pub(crate) async fn shutdown(&self) -> CodexResult<()> {
        let state = {
            let mut guard = self.state.lock().await;
            guard.conversation.take()
        };

        if let Some(state) = state {
            stop_conversation_state(state, RealtimeFanoutTaskStop::Await).await;
        }
        Ok(())
    }
}

async fn stop_conversation_state(
    mut state: ConversationState,
    fanout_task_stop: RealtimeFanoutTaskStop,
) {
    state.realtime_active.store(false, Ordering::Relaxed);
    state.stop_token.cancel();
    let _ = state.input_task.await;

    if let Some(fanout_task) = state.fanout_task.take() {
        match fanout_task_stop {
            RealtimeFanoutTaskStop::Await => {
                let _ = fanout_task.await;
            }
            RealtimeFanoutTaskStop::Detach => {}
        }
    }
}

pub(crate) async fn handle_start(
    sess: &Arc<Session>,
    sub_id: String,
    params: ConversationStartParams,
) -> CodexResult<()> {
    let prepared_start = match prepare_realtime_start(sess, params).await {
        Ok(prepared_start) => prepared_start,
        Err(err) => {
            error!("failed to prepare realtime conversation: {err}");
            let message = err.to_string();
            sess.send_event_raw(Event {
                id: sub_id,
                msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
                    payload: RealtimeEvent::Error(message),
                }),
            })
            .await;
            return Ok(());
        }
    };

    if let Err(err) = handle_start_inner(sess, &sub_id, prepared_start).await {
        error!("failed to start realtime conversation: {err}");
        let message = err.to_string();
        sess.send_event_raw(Event {
            id: sub_id.clone(),
            msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
                payload: RealtimeEvent::Error(message),
            }),
        })
        .await;
    }
    Ok(())
}

struct PreparedRealtimeConversationStart {
    api_provider: ApiProvider,
    http_client_factory: HttpClientFactory,
    realtime_sideband_base_url: Option<String>,
    extra_headers: Option<HeaderMap>,
    client_managed_handoffs: bool,
    flush_transcript_tail_on_session_end: bool,
    codex_responses_as_items: bool,
    codex_response_item_prefix: Option<String>,
    codex_response_handoff_mode: CodexResponseHandoffMode,
    backend_reasoning_status: bool,
    codex_response_handoff_channel_prefixes: Option<BTreeMap<String, Vec<String>>>,
    realtime_start_instructions: Option<String>,
    realtime_end_instructions: Option<String>,
    realtime_call_api_provider: Option<ApiProvider>,
    requested_realtime_session_id: Option<String>,
    version: RealtimeWsVersion,
    session_config: RealtimeSessionConfig,
    transport: ConversationStartTransport,
}

#[derive(Clone, Copy)]
pub(crate) enum ConfiguredRealtimeVoice {
    Use,
    Ignore,
}

async fn prepare_realtime_start(
    sess: &Arc<Session>,
    params: ConversationStartParams,
) -> CodexResult<PreparedRealtimeConversationStart> {
    let provider = sess.provider().await;
    let auth_manager = sess
        .services
        .model_client
        .auth_manager()
        .unwrap_or_else(|| Arc::clone(&sess.services.auth_manager));
    let config = sess.get_config().await;
    let (auth, http_client_factory) = match auth_manager.auth_with_http_client_factory().await {
        Some((auth, captured_factory)) if auth.is_chatgpt_auth() || auth.is_api_key_auth() => (
            Some(auth),
            config
                .http_client_factory()
                .with_network_policy(captured_factory.network_policy().clone()),
        ),
        Some((auth, _)) => (Some(auth), config.http_client_factory()),
        None => (None, config.http_client_factory()),
    };
    let transport = params
        .transport
        .clone()
        .unwrap_or(ConversationStartTransport::Websocket);
    let mut api_provider = provider.to_api_provider(Some(AuthMode::ApiKey))?;
    let realtime_sideband_base_url = match &transport {
        ConversationStartTransport::ExistingCall {
            sideband_base_url, ..
        } => sideband_base_url
            .clone()
            .or_else(|| config.experimental_realtime_ws_base_url.clone()),
        ConversationStartTransport::Websocket | ConversationStartTransport::Webrtc { .. } => {
            config.experimental_realtime_ws_base_url.clone()
        }
    };
    if let Some(realtime_ws_base_url) = &realtime_sideband_base_url {
        api_provider.base_url = realtime_ws_base_url.clone();
    }
    let realtime_call_api_provider =
        if let Some(realtime_call_base_url) = &config.experimental_realtime_webrtc_call_base_url {
            let mut api_provider = provider.to_api_provider(Some(AuthMode::ApiKey))?;
            api_provider.base_url = realtime_call_base_url.clone();
            Some(api_provider)
        } else {
            None
        };
    let version = params.version.unwrap_or(match &transport {
        ConversationStartTransport::Websocket => config.realtime.version,
        ConversationStartTransport::Webrtc { .. }
        | ConversationStartTransport::ExistingCall { .. } => RealtimeWsVersion::V1,
    });
    match &transport {
        ConversationStartTransport::Webrtc { .. } => {
            validate_avas_webrtc_start(version, config.realtime.session_type)?;
        }
        ConversationStartTransport::ExistingCall { .. } => {
            if version == RealtimeWsVersion::V2 {
                return Err(CodexErr::InvalidRequest(
                    "AVAS realtime calls require realtime v1 or v3".to_string(),
                ));
            }
            if params.include_startup_context
                || params.prompt.is_some()
                || !params.initial_items.is_empty()
                || params.model.is_some()
                || params.voice.is_some()
                || params.delegation_ack_filler.is_some()
            {
                return Err(CodexErr::InvalidRequest(
                    "existing realtime calls do not support session configuration options"
                        .to_string(),
                ));
            }
        }
        ConversationStartTransport::Websocket => {}
    }
    let configured_voice = match (&transport, params.version) {
        (ConversationStartTransport::ExistingCall { .. }, _)
        | (ConversationStartTransport::Webrtc { .. }, None) => ConfiguredRealtimeVoice::Ignore,
        (ConversationStartTransport::Webrtc { .. } | ConversationStartTransport::Websocket, _) => {
            ConfiguredRealtimeVoice::Use
        }
    };
    let mut session_config =
        build_realtime_session_config(sess, &params, version, configured_voice).await?;
    if matches!(&transport, ConversationStartTransport::ExistingCall { .. }) {
        session_config.session_id = params.realtime_session_id.clone();
    }
    let requested_realtime_session_id = session_config.session_id.clone();
    let event_parser = session_config.event_parser;
    let originator = sess.originator().await;
    let mut extra_headers = match transport {
        ConversationStartTransport::Websocket => {
            let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?;
            realtime_request_headers(
                requested_realtime_session_id.as_deref(),
                Some(realtime_api_key.as_str()),
                event_parser,
                originator.as_str(),
            )?
        }
        ConversationStartTransport::Webrtc { .. }
        | ConversationStartTransport::ExistingCall { .. } => {
            realtime_request_headers(
                requested_realtime_session_id.as_deref(),
                /*api_key*/ None,
                event_parser,
                originator.as_str(),
            )?
        }
    }
    .unwrap_or_default();
    extra_headers.extend(build_session_headers(
        Some(sess.session_id().to_string()),
        Some(sess.thread_id().to_string()),
    ));
    // Voice calls can span zero or many backing turns; send only the saved thread source.
    if let Some(thread_source) = sess.thread_config_snapshot().await.thread_source
        && thread_source.as_str().len() <= REALTIME_THREAD_SOURCE_MAX_BYTES
        && let Ok(metadata) = to_ascii_json_string(&json!({ THREAD_SOURCE_KEY: thread_source }))
        && let Ok(metadata) = HeaderValue::from_str(&metadata)
    {
        extra_headers.insert(X_CODEX_TURN_METADATA_HEADER, metadata);
    }
    Ok(PreparedRealtimeConversationStart {
        api_provider,
        http_client_factory,
        realtime_sideband_base_url,
        extra_headers: Some(extra_headers),
        client_managed_handoffs: params.client_managed_handoffs,
        flush_transcript_tail_on_session_end: params.flush_transcript_tail_on_session_end,
        codex_responses_as_items: params.codex_responses_as_items,
        codex_response_item_prefix: params.codex_response_item_prefix,
        codex_response_handoff_mode: params.codex_response_handoff_mode,
        backend_reasoning_status: params.backend_reasoning_status,
        codex_response_handoff_channel_prefixes: params.codex_response_handoff_channel_prefixes,
        realtime_start_instructions: params.realtime_start_instructions,
        realtime_end_instructions: params.realtime_end_instructions,
        realtime_call_api_provider,
        requested_realtime_session_id,
        version,
        session_config,
        transport,
    })
}

fn validate_avas_webrtc_start(
    version: RealtimeWsVersion,
    session_type: RealtimeWsMode,
) -> CodexResult<()> {
    if version == RealtimeWsVersion::V2 {
        return Err(CodexErr::InvalidRequest(
            "AVAS realtime calls require realtime v1 or v3".to_string(),
        ));
    }
    if session_type != RealtimeWsMode::Conversational {
        return Err(CodexErr::InvalidRequest(
            "AVAS realtime calls require conversational realtime".to_string(),
        ));
    }
    Ok(())
}

pub(crate) async fn build_realtime_session_config(
    sess: &Arc<Session>,
    params: &ConversationStartParams,
    version: RealtimeWsVersion,
    configured_voice: ConfiguredRealtimeVoice,
) -> CodexResult<RealtimeSessionConfig> {
    for (name, instructions) in [
        (
            "realtime start instructions",
            params.realtime_start_instructions.as_deref(),
        ),
        (
            "realtime end instructions",
            params.realtime_end_instructions.as_deref(),
        ),
    ] {
        if instructions.is_some_and(|instructions| {
            approx_token_count(instructions) > REALTIME_MODE_INSTRUCTIONS_MAX_TOKENS
        }) {
            return Err(CodexErr::InvalidRequest(format!(
                "{name} must not exceed {REALTIME_MODE_INSTRUCTIONS_MAX_TOKENS} estimated tokens"
            )));
        }
    }

    let config = sess.get_config().await;
    let prompt = prepare_realtime_backend_prompt(
        params.prompt.clone(),
        config.experimental_realtime_ws_backend_prompt.clone(),
    );
    let startup_context = if params.include_startup_context {
        match config.experimental_realtime_ws_startup_context.clone() {
            Some(startup_context) => startup_context,
            None => {
                build_realtime_startup_context(sess.as_ref(), REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET)
                    .await
                    .unwrap_or_default()
            }
        }
    } else {
        String::new()
    };
    let prompt = match (prompt.is_empty(), startup_context.is_empty()) {
        (true, true) => String::new(),
        (true, false) => startup_context,
        (false, true) => prompt,
        (false, false) => format!("{prompt}\n\n{startup_context}"),
    };
    if version != RealtimeWsVersion::V3 && !params.initial_items.is_empty() {
        return Err(CodexErr::InvalidRequest(
            "initial realtime items require realtime v3".to_string(),
        ));
    }
    if params.initial_items.len() > REALTIME_INITIAL_ITEMS_MAX_COUNT {
        return Err(CodexErr::InvalidRequest(format!(
            "initial realtime items must contain no more than {REALTIME_INITIAL_ITEMS_MAX_COUNT} items"
        )));
    }
    let mut total_initial_item_tokens: usize = 0;
    for item in &params.initial_items {
        let item_tokens = approx_token_count(&item.text);
        if item_tokens > REALTIME_INITIAL_ITEMS_MAX_TOKENS {
            return Err(CodexErr::InvalidRequest(format!(
                "each initial realtime item must not exceed {REALTIME_INITIAL_ITEMS_MAX_TOKENS} estimated tokens"
            )));
        }
        total_initial_item_tokens = total_initial_item_tokens.saturating_add(item_tokens);
    }
    if total_initial_item_tokens > REALTIME_INITIAL_ITEMS_MAX_TOKENS {
        return Err(CodexErr::InvalidRequest(format!(
            "initial realtime items must not exceed {REALTIME_INITIAL_ITEMS_MAX_TOKENS} estimated tokens in total"
        )));
    }
    let model = Some(
        params
            .model
            .clone()
            .or_else(|| config.experimental_realtime_ws_model.clone())
            .unwrap_or_else(|| match version {
                RealtimeWsVersion::V1 | RealtimeWsVersion::V2 => DEFAULT_REALTIME_MODEL.to_string(),
                RealtimeWsVersion::V3 => DEFAULT_FRAMELESS_REALTIME_MODEL.to_string(),
            }),
    );
    let event_parser = match version {
        RealtimeWsVersion::V1 => RealtimeEventParser::V1,
        RealtimeWsVersion::V2 => RealtimeEventParser::RealtimeV2,
        RealtimeWsVersion::V3 => RealtimeEventParser::FramelessBidi,
    };
    if version != RealtimeWsVersion::V2
        && matches!(params.output_modality, RealtimeOutputModality::Text)
    {
        return Err(CodexErr::InvalidRequest(
            "text realtime output modality requires realtime v2".to_string(),
        ));
    }
    let session_mode = match config.realtime.session_type {
        RealtimeWsMode::Conversational => RealtimeSessionMode::Conversational,
        RealtimeWsMode::Transcription => RealtimeSessionMode::Transcription,
    };
    let config_voice = match configured_voice {
        ConfiguredRealtimeVoice::Use => config.realtime.voice,
        ConfiguredRealtimeVoice::Ignore => None,
    };
    let voice = params
        .voice
        .or(config_voice)
        .unwrap_or_else(|| default_realtime_voice(version));
    validate_realtime_voice(version, voice)?;
    Ok(RealtimeSessionConfig {
        instructions: prompt,
        initial_items: params.initial_items.clone(),
        delegation_ack_filler: params.delegation_ack_filler,
        model,
        session_id: Some(
            params
                .realtime_session_id
                .clone()
                .unwrap_or_else(|| sess.thread_id.to_string()),
        ),
        event_parser,
        session_mode,
        output_modality: params.output_modality,
        voice,
    })
}

fn default_realtime_voice(version: RealtimeWsVersion) -> RealtimeVoice {
    let voices = RealtimeVoicesList::builtin();
    match version {
        RealtimeWsVersion::V1 | RealtimeWsVersion::V3 => voices.default_v1,
        RealtimeWsVersion::V2 => voices.default_v2,
    }
}

fn prefix_realtime_text(text: String, prefix: &str, session_kind: RealtimeSessionKind) -> String {
    if session_kind != RealtimeSessionKind::V2 || text.is_empty() || text.starts_with(prefix) {
        return text;
    }
    format!("{prefix}{text}")
}

fn realtime_backend_output(output_text: String, session_kind: RealtimeSessionKind) -> String {
    let output_text = prefix_realtime_text(output_text, REALTIME_BACKEND_TEXT_PREFIX, session_kind);
    truncate_realtime_text_to_token_budget(&output_text, REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
}

fn realtime_backend_item(text: String, prefix: Option<&str>) -> String {
    let text = match prefix.filter(|prefix| !prefix.is_empty()) {
        Some(prefix) => format!("{prefix}\n\n{text}"),
        None => text,
    };
    truncate_realtime_text_to_token_budget(&text, REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
}

fn validate_realtime_voice(version: RealtimeWsVersion, voice: RealtimeVoice) -> CodexResult<()> {
    let voices = RealtimeVoicesList::builtin();
    let allowed = match version {
        RealtimeWsVersion::V1 | RealtimeWsVersion::V3 => &voices.v1,
        RealtimeWsVersion::V2 => &voices.v2,
    };
    if allowed.contains(&voice) {
        return Ok(());
    }

    let version = match version {
        RealtimeWsVersion::V1 => "v1",
        RealtimeWsVersion::V2 => "v2",
        RealtimeWsVersion::V3 => "v3",
    };
    let allowed = allowed
        .iter()
        .map(|voice| voice.wire_name())
        .collect::<Vec<_>>()
        .join(", ");
    Err(CodexErr::InvalidRequest(format!(
        "realtime voice `{}` is not supported for {version}; supported voices: {allowed}",
        voice.wire_name()
    )))
}

async fn handle_start_inner(
    sess: &Arc<Session>,
    sub_id: &str,
    prepared_start: PreparedRealtimeConversationStart,
) -> CodexResult<()> {
    let PreparedRealtimeConversationStart {
        api_provider,
        http_client_factory,
        realtime_sideband_base_url,
        extra_headers,
        client_managed_handoffs,
        flush_transcript_tail_on_session_end,
        codex_responses_as_items,
        codex_response_item_prefix,
        codex_response_handoff_mode,
        backend_reasoning_status,
        codex_response_handoff_channel_prefixes,
        realtime_start_instructions,
        realtime_end_instructions,
        realtime_call_api_provider,
        requested_realtime_session_id,
        version,
        session_config,
        transport,
    } = prepared_start;
    info!("starting realtime conversation");
    let (sdp, existing_call_id) = match transport {
        ConversationStartTransport::Websocket => (None, None),
        ConversationStartTransport::Webrtc { sdp } => (Some(sdp), None),
        ConversationStartTransport::ExistingCall { call_id, .. } => (None, Some(call_id)),
    };
    let mode_instructions = RealtimeModeInstructions {
        start: realtime_start_instructions,
        end: realtime_end_instructions,
    };
    let start = RealtimeStart {
        api_provider,
        http_client_factory,
        realtime_sideband_base_url,
        extra_headers,
        client_managed_handoffs,
        flush_transcript_tail_on_session_end,
        codex_responses_as_items,
        codex_response_item_prefix,
        codex_response_handoff_mode,
        backend_reasoning_status,
        codex_response_handoff_channel_prefixes,
        realtime_call_api_provider,
        session_config,
        model_client: sess.services.model_client.clone(),
        sdp,
        existing_call_id,
    };
    let start_output = sess
        .conversation
        .start(start, mode_instructions, sess, sub_id)
        .await?;

    info!("realtime conversation started");

    sess.send_event_raw(Event {
        id: sub_id.to_string(),
        msg: EventMsg::RealtimeConversationStarted(RealtimeConversationStartedEvent {
            realtime_session_id: requested_realtime_session_id,
            version,
        }),
    })
    .await;

    let RealtimeStartOutput {
        realtime_active,
        route_handoffs,
        events_rx,
        transcript_tail_rx,
        sdp,
    } = start_output;
    if let Some(sdp) = sdp {
        sess.send_event_raw(Event {
            id: sub_id.to_string(),
            msg: EventMsg::RealtimeConversationSdp(RealtimeConversationSdpEvent { sdp }),
        })
        .await;
    }

    let sess_clone = Arc::clone(sess);
    let sub_id = sub_id.to_string();
    let fanout_realtime_active = Arc::clone(&realtime_active);
    let fanout_task = tokio::spawn(async move {
        let ev = |msg| Event {
            id: sub_id.clone(),
            msg,
        };
        let mut end = RealtimeConversationEnd::TransportClosed;
        let mut handoff_error = None;
        // Drain already-parsed events so a queued handoff is routed before the final tail.
        while let Ok(event) = events_rx.recv().await {
            match &event {
                RealtimeEvent::AudioOut(_) => {}
                _ => {
                    // Transcripts, handoffs, and conversation items may contain
                    // credentials. Keep this receipt free of event payloads.
                    info!("received realtime conversation event");
                }
            }
            if let RealtimeEvent::Error(_) = &event {
                end = RealtimeConversationEnd::Error;
            }
            let maybe_routed_text = match &event {
                RealtimeEvent::HandoffRequested(handoff) => {
                    realtime_delegation_from_handoff(handoff)
                }
                _ => None,
            };
            if let Some(text) = maybe_routed_text {
                // The routed text can contain spoken prompts or workspace secrets.
                debug!("[realtime-text] realtime conversation text output");
                handoff_error = route_handoffs.route(&sess_clone, text).await.err();
            }
            sess_clone
                .send_event_raw(ev(EventMsg::RealtimeConversationRealtime(
                    RealtimeConversationRealtimeEvent {
                        payload: event.clone(),
                    },
                )))
                .await;
            if handoff_error.is_some() {
                break;
            }
        }
        if handoff_error.is_none()
            && let Ok(text) = transcript_tail_rx.recv().await
        {
            let _permit = route_handoffs.gate.acquire().await;
            if !route_handoffs.retired.load(Ordering::Acquire)
                && let Err(err) = transcript_tail::record(&sess_clone, text).await
            {
                warn!("failed to flush realtime transcript before closure: {err}");
                handoff_error = Some("failed to save the realtime transcript before closure");
            }
        }
        // Make realtime_end eligible only after the tail has reached history.
        {
            let mut state = sess_clone.conversation.state.lock().await;
            if state.conversation.as_ref().is_none_or(|current| {
                Arc::ptr_eq(&current.realtime_active, &fanout_realtime_active)
            }) {
                state.context_active = false;
            }
        }
        if let Some(error) = handoff_error {
            end = RealtimeConversationEnd::Error;
            sess_clone
                .send_event_raw(ev(EventMsg::RealtimeConversationRealtime(
                    RealtimeConversationRealtimeEvent {
                        payload: RealtimeEvent::Error(error.to_string()),
                    },
                )))
                .await;
        }
        if fanout_realtime_active.swap(false, Ordering::Relaxed) {
            match end {
                RealtimeConversationEnd::TransportClosed => {
                    info!("realtime conversation transport closed");
                }
                RealtimeConversationEnd::Requested | RealtimeConversationEnd::Error => {}
            }
            sess_clone
                .conversation
                .finish_if_active(&sess_clone, &fanout_realtime_active, sub_id, end)
                .await;
        }
    });
    sess.conversation
        .register_fanout_task(&realtime_active, fanout_task)
        .await;

    Ok(())
}

pub(crate) async fn handle_audio(
    sess: &Arc<Session>,
    sub_id: String,
    params: ConversationAudioParams,
) {
    if let Err(err) = sess.conversation.audio_in(params.frame).await {
        error!("failed to append realtime audio: {err}");
        if sess.conversation.running_state().await.is_some() {
            warn!("realtime audio input failed while the session was already ending");
        } else {
            send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest)
                .await;
        }
    }
}

fn realtime_transcript_delta_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
    realtime_transcript_delta(&handoff.active_transcript)
}

fn realtime_transcript_delta(active_transcript: &[RealtimeTranscriptEntry]) -> Option<String> {
    let active_transcript = active_transcript
        .iter()
        .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
        .collect::<Vec<_>>()
        .join("\n");
    (!active_transcript.is_empty()).then_some(active_transcript)
}

fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
    (!handoff.input_transcript.is_empty())
        .then_some(handoff.input_transcript.clone())
        .or_else(|| realtime_transcript_delta_from_handoff(handoff))
}

fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
    let input = realtime_text_from_handoff_request(handoff)?;
    Some(wrap_realtime_delegation_input(
        &input,
        realtime_transcript_delta_from_handoff(handoff).as_deref(),
        RealtimeDelegationSource::Handoff,
    ))
}

fn wrap_realtime_delegation_input(
    input: &str,
    transcript_delta: Option<&str>,
    source: RealtimeDelegationSource,
) -> String {
    RealtimeDelegation::new(input, transcript_delta, source).render()
}

fn realtime_api_key(auth: Option<&CodexAuth>, provider: &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.into_inner());
    }

    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(
    realtime_session_id: Option<&str>,
    api_key: Option<&str>,
    event_parser: RealtimeEventParser,
    originator: &str,
) -> CodexResult<Option<HeaderMap>> {
    let mut headers = HeaderMap::new();

    match event_parser {
        RealtimeEventParser::V1 => {
            headers.insert("openai-alpha", HeaderValue::from_static("quicksilver=v1"));
        }
        RealtimeEventParser::FramelessBidi => {
            headers.insert("openai-alpha", HeaderValue::from_static("quicksilver=v2"));
        }
        RealtimeEventParser::RealtimeV2 => {}
    }

    if let Some(realtime_session_id) = realtime_session_id
        && let Ok(realtime_session_id) = HeaderValue::from_str(realtime_session_id)
    {
        headers.insert("x-session-id", realtime_session_id);
    }

    if let Some(api_key) = api_key {
        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);
    }

    add_originator_header(&mut headers, originator);

    Ok(Some(headers))
}

pub(crate) async fn handle_text(
    sess: &Arc<Session>,
    sub_id: String,
    params: ConversationTextParams,
) {
    debug!(text = %params.text, "[realtime-text] appending realtime conversation text input");
    if let Err(err) = sess.conversation.text_in(params).await {
        error!("failed to append realtime text: {err}");
        if sess.conversation.running_state().await.is_some() {
            warn!("realtime text input failed while the session was already ending");
        } else {
            send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest)
                .await;
        }
    }
}

pub(crate) async fn handle_speech(
    sess: &Arc<Session>,
    sub_id: String,
    params: ConversationSpeechParams,
) {
    debug!("[realtime-text] appending realtime speech");
    if let Err(err) = sess.conversation.append_speech(params.text).await {
        error!("failed to append realtime speech: {err}");
        if sess.conversation.running_state().await.is_some() {
            warn!("realtime speech append failed while the session was already ending");
        } else {
            send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest)
                .await;
        }
    }
}

pub(crate) async fn handle_close(sess: &Arc<Session>, sub_id: String) {
    handle_detach(sess, /*expected*/ None, &sub_id).await;
}

/// Uses the ordinary sequential start path and reports its connection result.
pub(crate) async fn handle_attach(
    sess: &Arc<Session>,
    sub_id: String,
    params: ConversationStartParams,
) -> CodexResult<()> {
    let prepared = prepare_realtime_start(sess, params).await?;
    handle_start_inner(sess, &sub_id, prepared).await
}

pub(crate) async fn handle_detach(sess: &Arc<Session>, expected: Option<&str>, stop_sub_id: &str) {
    let current = {
        let mut state = sess.conversation.state.lock().await;
        if expected.is_some_and(|expected| {
            state
                .conversation
                .as_ref()
                .and_then(|current| current.session_id.as_deref())
                != Some(expected)
        }) {
            return;
        }
        state.conversation.take()
    };
    let sub_id = if let Some(current) = current {
        let sub_id = current.sub_id.clone();
        stop_conversation_state(current, RealtimeFanoutTaskStop::Await).await;
        sub_id
    } else {
        stop_sub_id.to_owned()
    };
    send_realtime_conversation_closed(sess, sub_id, RealtimeConversationEnd::Requested).await;
}

fn spawn_realtime_input_task(
    input: RealtimeInputTask,
    transcript_tail_flush: RealtimeTranscriptTailFlush,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let events_tx = input.events_tx.clone();
        let transcript_state = input.events.transcript_state();
        let exit = run_realtime_input_task(input, /*pending_outbound*/ None).await;
        if let RealtimeInputTaskExit::TransportLost { err, .. } = exit {
            report_realtime_transport_loss(&events_tx, err).await;
        }
        let transcript_tail = transcript_state.take_transcript_tail().await;
        flush_realtime_transcript_tail(&transcript_tail_flush, &transcript_tail).await;
    })
}

enum RealtimeInputTaskExit {
    Terminal,
    TransportLost {
        err: ApiError,
        pending_outbound: Option<Box<RealtimePendingOutbound>>,
    },
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum RealtimePendingOutbound {
    Text(ConversationTextParams),
    Handoff(RealtimeOutbound),
}

struct RealtimeWebrtcSidebandInputTask {
    client: RealtimeWebsocketClient,
    session_config: RealtimeSessionConfig,
    call_id: String,
    sideband_headers: HeaderMap,
    session_initialization: RealtimeSidebandSessionInitialization,
    input_channels: RealtimeInputChannels,
    events_tx: Sender<RealtimeEvent>,
    handoff_state: RealtimeHandoffState,
    session_kind: RealtimeSessionKind,
    event_parser: RealtimeEventParser,
    realtime_active: Arc<AtomicBool>,
    transcript_tail_flush: RealtimeTranscriptTailFlush,
    stop_token: CancellationToken,
}

enum RealtimeSidebandSessionInitialization {
    CoreCreated,
    ExistingCall {
        initial_connection: Option<RealtimeWebsocketConnection>,
    },
}

async fn run_realtime_input_task(
    input: RealtimeInputTask,
    pending_outbound: Option<Box<RealtimePendingOutbound>>,
) -> RealtimeInputTaskExit {
    let RealtimeInputTask {
        writer,
        events,
        text_rx,
        handoff_output_rx,
        audio_rx,
        events_tx,
        handoff_state,
        session_kind,
        event_parser,
        stop_token,
    } = input;

    let mut output_audio_state: Option<OutputAudioState> = None;
    let mut response_create_queue = RealtimeResponseCreateQueue::default();

    if let Some(pending_outbound) = pending_outbound
        && let Err(err) = send_realtime_pending_outbound(
            &pending_outbound,
            &writer,
            &handoff_state,
            event_parser,
            &mut response_create_queue,
        )
        .await
    {
        return classify_realtime_input_error_with_pending(err, Some(pending_outbound));
    }

    loop {
        let result = tokio::select! {
            _ = stop_token.cancelled() => Err(RealtimeInputTaskExit::Terminal),
            realtime_event = events.next_event() => {
                match realtime_event {
                    Ok(Some(event)) => {
                        handle_realtime_server_event(
                            event,
                            &writer,
                            &events_tx,
                            &handoff_state,
                            session_kind,
                            &mut output_audio_state,
                            &mut response_create_queue,
                        )
                        .await
                        .map_err(classify_realtime_input_error)
                    }
                    Ok(None) => Err(RealtimeInputTaskExit::Terminal),
                    Err(err) => Err(RealtimeInputTaskExit::TransportLost {
                        err,
                        pending_outbound: None,
                    }),
                }
            }
            // Text input that should be sent into realtime.
            text = text_rx.recv() => {
                let pending_outbound = text
                    .as_ref()
                    .ok()
                    .cloned()
                    .map(RealtimePendingOutbound::Text)
                    .map(Box::new);
                handle_text_input(
                    text,
                    &writer,
                )
                .await
                .map_err(|err| {
                    classify_realtime_input_error_with_pending(err, pending_outbound)
                })
            }
            // Background agent progress or final output that should be sent back to realtime.
            background_agent_output = handoff_output_rx.recv() => {
                let pending_outbound = background_agent_output
                    .as_ref()
                    .ok()
                    .cloned()
                    .map(RealtimePendingOutbound::Handoff)
                    .map(Box::new);
                handle_handoff_output(
                    background_agent_output,
                    &writer,
                    &handoff_state,
                    event_parser,
                    &mut response_create_queue,
                )
                .await
                .map_err(|err| {
                    classify_realtime_input_error_with_pending(err, pending_outbound)
                })
            }
            // Audio frames captured from the user microphone.
            user_audio_frame = audio_rx.recv() => {
                handle_user_audio_input(user_audio_frame, &writer)
                    .await
                    .map_err(classify_realtime_input_error)
            }
        };
        if let Err(exit) = result {
            break exit;
        }
    }
}

fn classify_realtime_input_error(err: anyhow::Error) -> RealtimeInputTaskExit {
    classify_realtime_input_error_with_pending(err, /*pending_outbound*/ None)
}

fn classify_realtime_input_error_with_pending(
    err: anyhow::Error,
    pending_outbound: Option<Box<RealtimePendingOutbound>>,
) -> RealtimeInputTaskExit {
    match err.downcast::<ApiError>() {
        Ok(err) => RealtimeInputTaskExit::TransportLost {
            err,
            pending_outbound,
        },
        Err(err) => {
            warn!("realtime input task stopped: {err}");
            RealtimeInputTaskExit::Terminal
        }
    }
}

async fn send_realtime_pending_outbound(
    pending_outbound: &RealtimePendingOutbound,
    writer: &RealtimeWebsocketWriter,
    handoff_state: &RealtimeHandoffState,
    event_parser: RealtimeEventParser,
    response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
    match pending_outbound {
        RealtimePendingOutbound::Text(params) => writer
            .send_conversation_item_create(params.text.clone(), params.role)
            .await
            .map_err(anyhow::Error::from),
        RealtimePendingOutbound::Handoff(output) => {
            handle_handoff_output(
                Ok(output.clone()),
                writer,
                handoff_state,
                event_parser,
                response_create_queue,
            )
            .await
        }
    }
}

async fn report_realtime_transport_loss(events_tx: &Sender<RealtimeEvent>, err: ApiError) {
    let mapped_error = map_api_error(err);
    error!("realtime stream closed: {mapped_error}");
    let _ = events_tx
        .send(RealtimeEvent::Error(mapped_error.to_string()))
        .await;
}

async fn flush_realtime_transcript_tail(
    transcript_tail_flush: &RealtimeTranscriptTailFlush,
    transcript_tail: &[RealtimeTranscriptEntry],
) {
    if transcript_tail_flush.enabled
        && let Some(transcript_delta) = realtime_transcript_delta(transcript_tail)
    {
        let _ = transcript_tail_flush
            .tx
            .send(wrap_realtime_delegation_input(
                REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION,
                Some(&transcript_delta),
                RealtimeDelegationSource::TranscriptTailFlush,
            ))
            .await;
    }
}

async fn handle_text_input(
    params: Result<ConversationTextParams, RecvError>,
    writer: &RealtimeWebsocketWriter,
) -> anyhow::Result<()> {
    let params = params.context("text input channel closed")?;
    writer
        .send_conversation_item_create(params.text, params.role)
        .await
        .map_err(anyhow::Error::from)?;
    Ok(())
}

async fn flush_streamed_handoff_item(handoff: &RealtimeHandoffState, item_id: &str) {
    let (handoff_id, text, phase) = {
        let mut stream = handoff.stream.lock().await;
        let Some(streamed_item) = stream.items.get_mut(item_id) else {
            return;
        };
        streamed_item.flush_scheduled = false;
        let Some(text) = streamed_item.drain_stream_chunk() else {
            return;
        };
        streamed_item.last_flush_at = Instant::now();
        (
            streamed_item.handoff_id.clone(),
            text,
            streamed_item.phase.clone(),
        )
    };
    let _ = handoff
        .output_tx
        .send(RealtimeOutbound::HandoffAppend {
            handoff_id,
            text,
            phase,
        })
        .await;
}

fn schedule_streamed_handoff_flush(
    handoff: &RealtimeHandoffState,
    item_id: String,
    flush_delay: Duration,
) {
    let handoff = handoff.clone();
    let _flush_task = tokio::spawn(async move {
        tokio::time::sleep(flush_delay).await;
        flush_streamed_handoff_item(&handoff, &item_id).await;
    });
}

fn v3_output_writer(
    writer: &RealtimeWebsocketWriter,
    phase: Option<&MessagePhase>,
    handoff_mode: CodexResponseHandoffMode,
) -> RealtimeWebsocketWriter {
    let channel = match handoff_mode {
        CodexResponseHandoffMode::Thinking => None,
        CodexResponseHandoffMode::Commentary => Some(RealtimeContextAppendChannel::Commentary),
        CodexResponseHandoffMode::BemTags => match phase {
            Some(MessagePhase::PartialAnswer | MessagePhase::FinalAnswer) => {
                Some(RealtimeContextAppendChannel::Speakable)
            }
            Some(MessagePhase::Commentary) => Some(RealtimeContextAppendChannel::Commentary),
            None => Some(RealtimeContextAppendChannel::Speakable),
        },
    };
    match channel {
        Some(channel) => writer.clone().with_context_append_channel(channel),
        None => writer.clone(),
    }
}

async fn handle_handoff_output(
    handoff_output: Result<RealtimeOutbound, RecvError>,
    writer: &RealtimeWebsocketWriter,
    handoff_state: &RealtimeHandoffState,
    event_parser: RealtimeEventParser,
    response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
    let handoff_output = handoff_output.context("handoff output channel closed")?;
    let result = match event_parser {
        RealtimeEventParser::V1 => match handoff_output {
            RealtimeOutbound::QuietReasoningStatus { .. } => Ok(()),
            RealtimeOutbound::StandaloneHandoff { text, phase: _ } => {
                writer
                    .send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
                    .await
            }
            RealtimeOutbound::StandaloneSpeech { text } => {
                writer
                    .send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
                    .await
            }
            RealtimeOutbound::HandoffUpdate {
                handoff_id,
                text,
                phase: _,
            }
            | RealtimeOutbound::CompletedHandoff {
                handoff_id,
                text,
                phase: _,
            } => {
                writer
                    .send_conversation_function_call_output(handoff_id, text)
                    .await
            }
            RealtimeOutbound::HandoffAppend {
                handoff_id,
                text,
                phase: _,
            } => {
                writer
                    .send_conversation_handoff_append(handoff_id, text)
                    .await
            }
            RealtimeOutbound::ConversationItem { text, phase: _ } => {
                writer
                    .send_conversation_item_create(text, ConversationTextRole::Developer)
                    .await
            }
            RealtimeOutbound::HandoffCompleteAck { .. } => Ok(()),
        },
        RealtimeEventParser::FramelessBidi => match handoff_output {
            RealtimeOutbound::QuietReasoningStatus { handoff_id, text } => {
                if handoff_state.stream.lock().await.active_handoff.as_deref() != Some(&handoff_id)
                {
                    return Ok(());
                }
                // An omitted channel defaults to speakable on this wire protocol.
                writer
                    .clone()
                    .with_context_append_channel(RealtimeContextAppendChannel::Commentary)
                    .send_conversation_handoff_append(handoff_id, text)
                    .await
            }
            RealtimeOutbound::StandaloneHandoff { text, phase } => {
                v3_output_writer(
                    writer,
                    phase.as_ref(),
                    handoff_state.codex_response_handoff_mode,
                )
                .send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
                .await
            }
            RealtimeOutbound::StandaloneSpeech { text } => {
                writer
                    .clone()
                    .with_context_append_channel(RealtimeContextAppendChannel::Speakable)
                    .send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
                    .await
            }
            RealtimeOutbound::HandoffUpdate {
                handoff_id,
                text,
                phase,
            } => {
                v3_output_writer(
                    writer,
                    phase.as_ref(),
                    handoff_state.codex_response_handoff_mode,
                )
                .send_conversation_function_call_output(handoff_id, text)
                .await
            }
            RealtimeOutbound::HandoffAppend {
                handoff_id,
                text,
                phase,
            } => {
                v3_output_writer(
                    writer,
                    phase.as_ref(),
                    handoff_state.codex_response_handoff_mode,
                )
                .send_conversation_handoff_append(handoff_id, text)
                .await
            }
            RealtimeOutbound::CompletedHandoff {
                handoff_id,
                text,
                phase,
            } => {
                v3_output_writer(
                    writer,
                    phase.as_ref(),
                    handoff_state.codex_response_handoff_mode,
                )
                .send_conversation_function_call_output(handoff_id, text)
                .await
            }
            RealtimeOutbound::ConversationItem { text, phase } => {
                v3_output_writer(
                    writer,
                    phase.as_ref(),
                    handoff_state.codex_response_handoff_mode,
                )
                .send_conversation_item_create(text, ConversationTextRole::Developer)
                .await
            }
            RealtimeOutbound::HandoffCompleteAck { .. } => Ok(()),
        },
        RealtimeEventParser::RealtimeV2 => match handoff_output {
            RealtimeOutbound::QuietReasoningStatus { .. } => Ok(()),
            RealtimeOutbound::StandaloneHandoff { text, phase: _ } => {
                if let Err(err) = writer
                    .send_conversation_item_create(text, ConversationTextRole::User)
                    .await
                {
                    Err(err)
                } else {
                    return response_create_queue
                        .request_create(writer, "standalone handoff")
                        .await;
                }
            }
            RealtimeOutbound::StandaloneSpeech { text } => {
                if let Err(err) = writer
                    .send_conversation_item_create(text, ConversationTextRole::User)
                    .await
                {
                    Err(err)
                } else {
                    return response_create_queue
                        .request_create(writer, "standalone handoff")
                        .await;
                }
            }
            RealtimeOutbound::HandoffUpdate {
                handoff_id,
                text,
                phase: _,
            }
            | RealtimeOutbound::HandoffAppend {
                handoff_id,
                text,
                phase: _,
            } => {
                let active_handoff = handoff_state.stream.lock().await.active_handoff.clone();
                match active_handoff {
                    Some(active_handoff) if active_handoff == handoff_id => {}
                    Some(_) | None => {
                        debug!("dropping stale realtime handoff progress update");
                        return Ok(());
                    }
                }
                writer
                    .send_conversation_item_create(text, ConversationTextRole::User)
                    .await
            }
            RealtimeOutbound::CompletedHandoff {
                handoff_id,
                text: _,
                phase: _,
            } => {
                if let Err(err) = writer
                    .send_conversation_function_call_output(
                        handoff_id,
                        REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT.to_string(),
                    )
                    .await
                {
                    Err(err)
                } else {
                    return response_create_queue
                        .request_create(writer, "handoff")
                        .await;
                }
            }
            RealtimeOutbound::ConversationItem { text, phase: _ } => {
                writer
                    .send_conversation_item_create(text, ConversationTextRole::Developer)
                    .await
            }
            RealtimeOutbound::HandoffCompleteAck { handoff_id } => {
                writer
                    .send_conversation_function_call_output(handoff_id, String::new())
                    .await
            }
        },
    };
    if let Err(err) = result {
        warn!("failed to send handoff output: {err}");
        return Err(err.into());
    }
    Ok(())
}

async fn handle_realtime_server_event(
    event: RealtimeEvent,
    writer: &RealtimeWebsocketWriter,
    events_tx: &Sender<RealtimeEvent>,
    handoff_state: &RealtimeHandoffState,
    session_kind: RealtimeSessionKind,
    output_audio_state: &mut Option<OutputAudioState>,
    response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
    let should_stop = match &event {
        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)
                    {
                        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
                            .map_err(anyhow::Error::from)?;
                    }
                }
            }
            false
        }
        RealtimeEvent::ResponseCreated(_) => {
            match session_kind {
                RealtimeSessionKind::V1 => {}
                RealtimeSessionKind::V2 => response_create_queue.mark_started(),
            }
            false
        }
        RealtimeEvent::ResponseCancelled(_) => {
            *output_audio_state = None;
            match session_kind {
                RealtimeSessionKind::V1 => {}
                RealtimeSessionKind::V2 => {
                    response_create_queue
                        .mark_finished(writer, "deferred")
                        .await?;
                }
            }
            false
        }
        RealtimeEvent::ResponseDone(_) => {
            *output_audio_state = None;
            match session_kind {
                RealtimeSessionKind::V1 => {}
                RealtimeSessionKind::V2 => {
                    response_create_queue
                        .mark_finished(writer, "deferred")
                        .await?;
                }
            }
            false
        }
        RealtimeEvent::HandoffRequested(handoff) => {
            *output_audio_state = None;

            match session_kind {
                RealtimeSessionKind::V1 => {
                    let mut stream = handoff_state.stream.lock().await;
                    stream.items.clear();
                    stream.active_handoff = Some(handoff.handoff_id.clone());
                }
                RealtimeSessionKind::V2 => {
                    let active_handoff = handoff_state.stream.lock().await.active_handoff.clone();
                    match active_handoff {
                        Some(_) => {
                            if let Err(err) = writer
                                .send_conversation_function_call_output(
                                    handoff.handoff_id.clone(),
                                    REALTIME_V2_STEER_ACKNOWLEDGEMENT.to_string(),
                                )
                                .await
                            {
                                warn!("failed to send handoff steering acknowledgement: {err}");
                                return Err(err.into());
                            }
                            response_create_queue
                                .request_create(writer, "handoff steering")
                                .await?;
                        }
                        None => {
                            handoff_state.stream.lock().await.active_handoff =
                                Some(handoff.handoff_id.clone());
                        }
                    }
                }
            }
            false
        }
        RealtimeEvent::NoopRequested(noop) => {
            *output_audio_state = None;

            match session_kind {
                RealtimeSessionKind::V1 => {}
                RealtimeSessionKind::V2 => {
                    if let Err(err) = writer
                        .send_conversation_function_call_output(noop.call_id.clone(), String::new())
                        .await
                    {
                        warn!("failed to send realtime noop function output: {err}");
                        return Err(err.into());
                    }
                }
            }
            false
        }
        RealtimeEvent::Error(_) => true,
        RealtimeEvent::SessionUpdated {
            realtime_session_id,
            ..
        } => {
            info!(realtime_session_id = %realtime_session_id, "realtime session updated");
            false
        }
        RealtimeEvent::InputTranscriptDelta(_)
        | RealtimeEvent::InputTranscriptDone(_)
        | RealtimeEvent::OutputTranscriptDelta(_)
        | RealtimeEvent::OutputTranscriptDone(_)
        | RealtimeEvent::ConversationItemAdded(_)
        | RealtimeEvent::ConversationItemDone { .. }
        | RealtimeEvent::HistoryItemStarted(_)
        | RealtimeEvent::HistoryTranscriptDelta { .. }
        | RealtimeEvent::HistoryItemCompleted(_) => false,
    };

    if 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,
) -> anyhow::Result<()> {
    let frame = frame.context("user audio input channel closed")?;
    writer
        .send_audio_frame(frame)
        .await
        .map_err(anyhow::Error::from)?;
    Ok(())
}

fn update_output_audio_state(
    output_audio_state: &mut Option<OutputAudioState>,
    frame: &RealtimeAudioFrame,
) {
    let Some(item_id) = frame.item_id.clone() else {
        return;
    };
    let audio_end_ms = audio_duration_ms(frame);
    if audio_end_ms == 0 {
        return;
    }

    if let Some(current) = output_audio_state.as_mut()
        && current.item_id == item_id
    {
        current.audio_end_ms = current.audio_end_ms.saturating_add(audio_end_ms);
        return;
    }

    *output_audio_state = Some(OutputAudioState {
        item_id,
        audio_end_ms,
    });
}

fn audio_duration_ms(frame: &RealtimeAudioFrame) -> u32 {
    let Some(samples_per_channel) = frame
        .samples_per_channel
        .or(decoded_samples_per_channel(frame))
    else {
        return 0;
    };
    let sample_rate = u64::from(frame.sample_rate.max(1));
    ((u64::from(samples_per_channel) * 1_000) / sample_rate) as u32
}

fn decoded_samples_per_channel(frame: &RealtimeAudioFrame) -> Option<u32> {
    let bytes = BASE64_STANDARD.decode(&frame.data).ok()?;
    let channels = usize::from(frame.num_channels.max(1));
    let samples = bytes.len().checked_div(2)?.checked_div(channels)?;
    u32::try_from(samples).ok()
}

async fn send_conversation_error(
    sess: &Arc<Session>,
    sub_id: String,
    message: String,
    codex_error_info: CodexErrorInfo,
) {
    sess.send_event_raw(Event {
        id: sub_id,
        msg: EventMsg::Error(ErrorEvent {
            misalignment: None,
            message,
            codex_error_info: Some(codex_error_info),
        }),
    })
    .await;
}

async fn send_realtime_conversation_closed(
    sess: &Arc<Session>,
    sub_id: String,
    end: RealtimeConversationEnd,
) {
    let reason = match end {
        RealtimeConversationEnd::Requested => Some("requested".to_string()),
        RealtimeConversationEnd::TransportClosed => Some("transport_closed".to_string()),
        RealtimeConversationEnd::Error => Some("error".to_string()),
    };

    sess.send_event_raw(Event {
        id: sub_id,
        msg: EventMsg::RealtimeConversationClosed(RealtimeConversationClosedEvent { reason }),
    })
    .await;
}

#[cfg(test)]
#[path = "realtime_conversation_tests.rs"]
mod tests;