Persist realtime transcript tails before closure without inference (#50531)

## Why

When `flush_transcript_tail_on_session_end` is enabled, remaining speech must be available in thread history as soon as realtime closure is reported, even while a model response is still running. Saving that transcript should not trigger another model request.

## What changed

Record the transcript tail directly in model and client history, attaching it to the active turn or creating a completed history turn when idle. Flush persistence before emitting `RealtimeConversationClosed`, and keep realtime context active until recording finishes so the return to text instructions follow the transcript.

## Testing

Extend core and app-server tests to cover immediate history visibility in paginated and legacy modes, idle and active turns, closure without extra inference, transcript reuse in subsequent text turns, and interruption after closure.

GitOrigin-RevId: ae323b711f22aa88db1432e24ad4bb3eb3f08f6b

codex-rs/app-server/tests/suite/v2/realtime_conversation.rs

@@ -15,6 +15,10 @@ use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadForkResponse;
 use codex_app_server_protocol::ThreadHistoryMode;
 use codex_app_server_protocol::ThreadItem;
+use codex_app_server_protocol::ThreadItemsListParams;
+use codex_app_server_protocol::ThreadItemsListResponse;
+use codex_app_server_protocol::ThreadReadParams;
+use codex_app_server_protocol::ThreadReadResponse;
 use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
 use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse;
 use codex_app_server_protocol::ThreadRealtimeAppendSpeechParams;
@@ -50,10 +54,13 @@ use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadTimelineEntry;
 use codex_app_server_protocol::ThreadTimelineListParams;
 use codex_app_server_protocol::ThreadTimelineListResponse;
+use codex_app_server_protocol::ThreadTurnsListParams;
+use codex_app_server_protocol::ThreadTurnsListResponse;
 use codex_app_server_protocol::TurnCompletedNotification;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnStartedNotification;
+use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::TurnSteerParams;
 use codex_app_server_protocol::TurnSteerResponse;
 use codex_app_server_protocol::UserInput as V2UserInput;
@@ -1713,16 +1720,32 @@ async fn realtime_list_voices_returns_supported_names() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
+#[test_matrix([ThreadHistoryMode::Paginated, ThreadHistoryMode::Legacy], [false, true])]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn realtime_conversation_stop_emits_closed_notification(
+    history_mode: ThreadHistoryMode,
+    running: bool,
+) -> Result<()> {
     skip_if_no_network!(Ok(()));
 
-    let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
+    let responses_server = MockServer::start().await;
+    let (response_tx, response_rx) = mpsc::channel();
+    Mock::given(method("POST"))
+        .and(path_regex(".*/responses$"))
+        .respond_with(GatedSseResponse {
+            gate_rx: Mutex::new(Some(response_rx)),
+            response: create_final_assistant_message_sse_response("Done.")?,
+        })
+        .mount(&responses_server)
+        .await;
     let realtime_server = start_websocket_server(vec![vec![
-        vec![json!({
-            "type": "session.updated",
-            "session": { "id": "sess_backend", "instructions": "backend prompt" }
-        })],
+        vec![
+            json!({
+                "type": "session.updated",
+                "session": { "id": "sess_backend", "instructions": "backend prompt" }
+            }),
+            json!({"type": "conversation.input_transcript.delta", "delta": "final words"}),
+        ],
         vec![],
     ]])
     .await;
@@ -1742,16 +1765,38 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
     login_with_api_key(&mut mcp, "sk-test-key").await?;
 
     let thread_start_request_id = mcp
-        .send_thread_start_request_with_auto_env(ThreadStartParams::default())
+        .send_thread_start_request_with_auto_env(ThreadStartParams {
+            history_mode: Some(history_mode),
+            ..Default::default()
+        })
         .await?;
     let thread_start: ThreadStartResponse =
         timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
 
+    let running_turn_id = if running {
+        let request = mcp
+            .send_turn_start_request(TurnStartParams {
+                thread_id: thread_start.thread.id.clone(),
+                input: vec![V2UserInput::Text {
+                    text: "keep working".into(),
+                    text_elements: Vec::new(),
+                }],
+                ..Default::default()
+            })
+            .await?;
+        let response: TurnStartResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
+        read_notification::<ItemCompletedNotification>(&mut mcp, "item/completed").await?;
+        Some(response.turn.id)
+    } else {
+        None
+    };
+
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             client_managed_handoffs: None,
             delegation_ack_filler: None,
-            flush_transcript_tail_on_session_end: None,
+            flush_transcript_tail_on_session_end: Some(true),
             codex_responses_as_items: None,
             codex_response_item_prefix: None,
             codex_response_handoff_mode: None,
@@ -1767,7 +1812,7 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: None,
-            version: None,
+            version: Some(RealtimeConversationVersion::V1),
             voice: None,
         })
         .await?;
@@ -1777,6 +1822,12 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
     let started =
         read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
             .await?;
+    let transcript = read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
+        &mut mcp,
+        "thread/realtime/transcript/delta",
+    )
+    .await?;
+    assert_eq!(transcript.delta, "final words");
 
     let stop_request_id = mcp
         .send_thread_realtime_stop_request(ThreadRealtimeStopParams {
@@ -1795,6 +1846,84 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
         Some("requested" | "transport_closed")
     ));
 
+    // The parent's handback uses this API immediately after Closed, even when
+    // the child's model response is still blocked on response_tx above.
+    let turns = if history_mode == ThreadHistoryMode::Paginated {
+        let request = mcp
+            .send_thread_items_list_request(ThreadItemsListParams {
+                thread_id: started.thread_id.clone(),
+                turn_id: None,
+                cursor: None,
+                limit: None,
+                sort_direction: None,
+            })
+            .await?;
+        let items: ThreadItemsListResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
+        let tails = items.data.iter().filter(|entry| matches!(
+            &entry.item, ThreadItem::UserMessage { content, .. }
+            if content.iter().any(|part| matches!(part, V2UserInput::Text { text, .. }
+                if text.contains("<source>transcript_tail_flush</source>") && text.contains("user: final words")))
+        )).collect::<Vec<_>>();
+        assert_eq!(tails.len(), 1);
+        if let Some(id) = &running_turn_id {
+            assert_eq!(&tails[0].turn_id, id);
+        }
+        let request = mcp
+            .send_thread_turns_list_request(ThreadTurnsListParams {
+                thread_id: started.thread_id.clone(),
+                cursor: None,
+                limit: None,
+                sort_direction: None,
+                items_view: None,
+            })
+            .await?;
+        let turns: ThreadTurnsListResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
+        assert_eq!(tails[0].turn_id, turns.data[0].id);
+        turns.data
+    } else {
+        let request = mcp
+            .send_thread_read_request(ThreadReadParams {
+                thread_id: started.thread_id.clone(),
+                include_turns: true,
+            })
+            .await?;
+        let history: ThreadReadResponse =
+            timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
+        assert_eq!(history.thread.turns.iter().flat_map(|turn| &turn.items).filter(|item| matches!(
+            item, ThreadItem::UserMessage { content, .. }
+            if content.iter().any(|part| matches!(part, V2UserInput::Text { text, .. }
+                if text.contains("<source>transcript_tail_flush</source>") && text.contains("user: final words")))
+        )).count(), 1);
+        history.thread.turns
+    };
+    assert_eq!(turns.len(), 1);
+    assert_eq!(
+        turns[0].status,
+        if running {
+            TurnStatus::InProgress
+        } else {
+            TurnStatus::Completed
+        }
+    );
+    assert_eq!(Uuid::parse_str(&turns[0].id)?.get_version_num(), 7);
+
+    response_tx.send(())?;
+    if running {
+        let completed =
+            read_notification::<TurnCompletedNotification>(&mut mcp, "turn/completed").await?;
+        assert_eq!(Some(completed.turn.id), running_turn_id);
+    }
+    assert_eq!(
+        responses_server
+            .received_requests()
+            .await
+            .expect("request recording is enabled")
+            .len(),
+        usize::from(running)
+    );
+
     realtime_server.shutdown().await;
     Ok(())
 }

codex-rs/core/src/realtime_conversation.rs

@@ -88,6 +88,7 @@ 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;
@@ -171,6 +172,8 @@ pub(crate) struct RealtimeConversationManager {
 struct RealtimeConversationManagerState {
     conversation: Option<ConversationState>,
     mode_instructions: Option<RealtimeModeInstructions>,
+    // Keep context active until the terminal transcript is recorded.
+    context_active: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -556,6 +559,7 @@ impl RealtimeConversationManager {
             state: Mutex::new(RealtimeConversationManagerState {
                 conversation: None,
                 mode_instructions: None,
+                context_active: false,
             }),
         }
     }
@@ -563,10 +567,7 @@ impl RealtimeConversationManager {
     pub(crate) async fn snapshot(&self) -> RealtimeConversationSnapshot {
         let state = self.state.lock().await;
         RealtimeConversationSnapshot {
-            active: state
-                .conversation
-                .as_ref()
-                .is_some_and(|conversation| conversation.realtime_active.load(Ordering::Relaxed)),
+            active: state.context_active,
             mode_instructions: state.mode_instructions.clone(),
         }
     }
@@ -780,6 +781,7 @@ impl RealtimeConversationManager {
             stop_token,
         });
         state.mode_instructions = Some(mode_instructions);
+        state.context_active = true;
         Ok(RealtimeStartOutput {
             realtime_active,
             route_handoffs,
@@ -1785,8 +1787,16 @@ async fn handle_start_inner(
         if handoff_error.is_none()
             && let Ok(text) = transcript_tail_rx.recv().await
         {
-            handoff_error = route_handoffs.route(&sess_clone, text).await.err();
+            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.
+        sess_clone.conversation.state.lock().await.context_active = false;
         if let Some(error) = handoff_error {
             end = RealtimeConversationEnd::Error;
             sess_clone

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

@@ -0,0 +1,96 @@
+//! Saves terminal speech in model and client history before realtime closure, without inference.
+
+use crate::session::new_submission_id;
+use crate::session::session::Session;
+use codex_protocol::items::TurnItem;
+use codex_protocol::items::UserMessageItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::HasLegacyEvent;
+use codex_protocol::protocol::ItemCompletedEvent;
+use codex_protocol::protocol::TurnCompleteEvent;
+use codex_protocol::protocol::TurnStartedEvent;
+use codex_protocol::user_input::UserInput;
+use codex_rollout::RolloutItem;
+use codex_thread_store::PersistContext;
+use std::sync::Arc;
+
+#[expect(
+    clippy::await_holding_invalid_type,
+    reason = "record history before the owning task can finish or a new task can start"
+)]
+pub(super) async fn record(session: &Session, text: String) -> std::io::Result<()> {
+    let mut recording_context = session.new_inject_items_context().await;
+    Arc::get_mut(&mut recording_context)
+        .ok_or_else(|| std::io::Error::other("realtime transcript context is shared"))?
+        .sub_id = new_submission_id();
+    let active = session.active_turn.lock().await;
+    let running_turn = active
+        .as_ref()
+        .and_then(|turn| turn.task.as_ref())
+        .map(|task| Arc::clone(&task.turn_context));
+    let recording_turn = running_turn.is_none();
+    let turn = running_turn.unwrap_or(recording_context);
+    let content = vec![UserInput::Text {
+        text,
+        text_elements: Vec::new(),
+    }];
+    let now = chrono::Utc::now();
+    if recording_turn {
+        let started = EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: turn.sub_id.clone(),
+            root_turn_id: Some(turn.sub_id.clone()),
+            trace_id: turn.trace_id.clone(),
+            started_at: Some(now.timestamp()),
+            model_context_window: turn.model_context_window(),
+            collaboration_mode_kind: turn.mode(),
+        });
+        session
+            .persist_rollout_items(&[RolloutItem::EventMsg(started)])
+            .await;
+        session.record_started_turn(&turn.sub_id).await;
+    }
+    let item = session.response_item_from_user_input(content.clone());
+    session
+        .record_conversation_items(&turn, turn.model_info(), &[item])
+        .await;
+
+    // ItemCompleted is the existing source for thread/items/list. Keep the legacy
+    // equivalent too; the store selects the appropriate format for this thread.
+    let completed = EventMsg::ItemCompleted(ItemCompletedEvent {
+        thread_id: session.thread_id,
+        turn_id: turn.sub_id.clone(),
+        item: TurnItem::UserMessage(UserMessageItem::new(&content)),
+        started_at_ms: Some(now.timestamp_millis()),
+        completed_at_ms: now.timestamp_millis(),
+    });
+    let mut events = Vec::new();
+    let legacy = completed.as_legacy_events(/*show_raw_agent_reasoning*/ false);
+    events.push(completed);
+    events.extend(legacy);
+    if recording_turn {
+        events.push(EventMsg::TurnComplete(TurnCompleteEvent {
+            turn_id: turn.sub_id.clone(),
+            last_agent_message: None,
+            error: None,
+            started_at: Some(now.timestamp()),
+            completed_at: Some(now.timestamp()),
+            duration_ms: Some(0),
+            time_to_first_token_ms: None,
+        }));
+    }
+    // Persist the complete history record without advertising a running UI turn
+    // or completing the model task that may still be streaming.
+    session
+        .persist_rollout_items(
+            &events
+                .into_iter()
+                .map(RolloutItem::EventMsg)
+                .collect::<Vec<_>>(),
+        )
+        .await;
+    drop(active);
+    session
+        .try_ensure_rollout_materialized(PersistContext::Standard)
+        .await?;
+    session.flush_rollout().await
+}

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -12,6 +12,7 @@ use codex_core::TurnInputRequest;
 use codex_core::TurnInputSubmission;
 use codex_core::test_support::auth_manager_from_auth;
 use codex_history::InitialHistory;
+use codex_history::ResponseItemEnvelope;
 use codex_history::RolloutItem;
 use codex_login::CodexAuth;
 use codex_login::OPENAI_API_KEY_ENV_VAR;
@@ -2238,6 +2239,7 @@ async fn assert_transport_close_tail_flush(
         }
     });
     let test = builder.build(&api_server).await?;
+    test.codex.ensure_rollout_materialized().await;
 
     test.codex
         .submit(Op::RealtimeConversationStart(ConversationStartParams {
@@ -2288,22 +2290,32 @@ async fn assert_transport_close_tail_flush(
 
     let closed = wait_for_event_match(&test.codex, |msg| match msg {
         EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
+        EventMsg::TurnStarted(_) | EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) => {
+            panic!("idle transcript flush must not create an unfinished client turn")
+        }
         _ => None,
     })
     .await;
     assert_eq!(closed.reason.as_deref(), Some("transport_closed"));
-    if flush_transcript_tail_on_session_end {
-        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
-        while response_mock.requests().is_empty() {
-            assert!(tokio::time::Instant::now() < deadline);
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-        assert!(response_mock.single_request().message_input_texts("user").iter().any(|text| text
-            == "<realtime_delegation>\n  <source>transcript_tail_flush</source>\n  <input>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.</input>\n  <transcript_delta>user: transport tail</transcript_delta>\n</realtime_delegation>"));
-    } else {
-        tokio::time::sleep(Duration::from_millis(200)).await;
-        assert!(response_mock.requests().is_empty());
-    }
+    // Closed must be a history barrier, including when there is no active turn.
+    let history = test.codex.load_history(/*include_archived*/ false).await?;
+    let tails = history.items.iter().filter(|item| {
+        matches!(item, RolloutItem::ResponseItem(ResponseItemEnvelope { item: ResponseItem::Message { role, content, .. }, .. })
+            if role == "user" && content.iter().any(|item| matches!(item,
+                ContentItem::InputText { text } if text.contains("<source>transcript_tail_flush</source>")
+                    && text.contains("user: transport tail"))))
+    }).count();
+    assert_eq!(tails, usize::from(flush_transcript_tail_on_session_end));
+    assert!(response_mock.requests().is_empty());
+
+    test.submit_text_turn("continue in text").await?;
+    let user_texts = response_mock.single_request().message_input_texts("user");
+    assert_eq!(
+        user_texts
+            .iter()
+            .any(|text| text.contains("user: transport tail")),
+        flush_transcript_tail_on_session_end
+    );
 
     realtime_server.shutdown().await;
     Ok(())
@@ -5073,33 +5085,38 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu
         ],
     )
     .await;
-    let realtime_server = start_websocket_server(vec![vec![
-        vec![
-            json!({
-                "type": "session.updated",
-                "session": { "id": "sess_tail", "instructions": "backend prompt" }
-            }),
-            json!({
-                "type": "conversation.input_transcript.delta",
-                "delta": "already handed off"
-            }),
-            json!({
-                "type": "conversation.handoff.requested",
-                "handoff_id": "handoff_tail",
-                "item_id": "item_tail",
-                "input_transcript": "already handed off"
-            }),
-            json!({
-                "type": "conversation.output_transcript.delta",
-                "delta": "remaining answer"
-            }),
-            json!({
-                "type": "conversation.input_transcript.delta",
-                "delta": "remaining question"
-            }),
+    let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
+        requests: vec![
+            vec![
+                json!({
+                    "type": "session.updated",
+                    "session": { "id": "sess_tail", "instructions": "backend prompt" }
+                }),
+                json!({
+                    "type": "conversation.input_transcript.delta",
+                    "delta": "already handed off"
+                }),
+                json!({
+                    "type": "conversation.handoff.requested",
+                    "handoff_id": "handoff_tail",
+                    "item_id": "item_tail",
+                    "input_transcript": "already handed off"
+                }),
+                json!({
+                    "type": "conversation.output_transcript.delta",
+                    "delta": "remaining answer"
+                }),
+                json!({
+                    "type": "conversation.input_transcript.delta",
+                    "delta": "remaining question"
+                }),
+            ],
+            vec![],
         ],
-        vec![],
-    ]])
+        response_headers: Vec::new(),
+        accept_delay: None,
+        close_after_requests: false,
+    }])
     .await;
     let mut builder = test_codex().with_config({
         let realtime_base_url = realtime_server.uri().to_string();
@@ -5143,19 +5160,30 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu
 
     let closed = wait_for_event_match(&test.codex, |msg| match msg {
         EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
+        EventMsg::TurnStarted(_) | EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) => {
+            panic!("idle transcript flush must not create an unfinished client turn")
+        }
         _ => None,
     })
     .await;
     assert_eq!(closed.reason.as_deref(), Some("requested"));
 
-    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
-    while response_mock.requests().len() < 2 {
-        assert!(tokio::time::Instant::now() < deadline);
-        tokio::time::sleep(Duration::from_millis(10)).await;
-    }
+    let history = test.codex.load_history(/*include_archived*/ false).await?;
+    let tails = history.items.iter().filter(|item| {
+        matches!(item, RolloutItem::ResponseItem(ResponseItemEnvelope { item: ResponseItem::Message { role, content, .. }, .. })
+            if role == "user" && content.iter().any(|item| matches!(item,
+                ContentItem::InputText { text } if text.contains("<source>transcript_tail_flush</source>")
+                    && text.contains("assistant: remaining answer\nuser: remaining question"))))
+    }).count();
+    assert_eq!(tails, 1);
+    assert_eq!(response_mock.requests().len(), 1);
 
     test.codex.submit(Op::RealtimeConversationClose).await?;
-    tokio::time::sleep(Duration::from_millis(200)).await;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::RealtimeConversationClosed(_))
+    })
+    .await;
+    test.submit_text_turn("continue in text").await?;
 
     let requests = response_mock.requests();
     assert_eq!(requests.len(), 2);
@@ -5581,8 +5609,14 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R
     Ok(())
 }
 
+#[test_case(false, false; "without_tail")]
+#[test_case(true, false; "with_tail")]
+#[test_case(true, true; "interrupt_after_close")]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn inbound_handoff_request_updates_realtime_state_during_active_turn() -> Result<()> {
+async fn inbound_handoff_request_updates_realtime_state_during_active_turn(
+    flush_transcript_tail_on_session_end: bool,
+    interrupt: bool,
+) -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
@@ -5647,6 +5681,10 @@ async fn inbound_handoff_request_updates_realtime_state_during_active_turn() ->
                     "item_id": "item_steer",
                     "input_transcript": "steer via realtime"
                 }),
+                json!({
+                    "type": "conversation.input_transcript.delta",
+                    "delta": "final transcript tail"
+                }),
             ],
         ],
         response_headers: Vec::new(),
@@ -5684,7 +5722,7 @@ async fn inbound_handoff_request_updates_realtime_state_during_active_turn() ->
         .submit(Op::RealtimeConversationStart(ConversationStartParams {
             client_managed_handoffs: false,
             delegation_ack_filler: None,
-            flush_transcript_tail_on_session_end: false,
+            flush_transcript_tail_on_session_end,
             codex_responses_as_items: false,
             codex_response_item_prefix: None,
             codex_response_handoff_mode:
@@ -5752,12 +5790,34 @@ async fn inbound_handoff_request_updates_realtime_state_during_active_turn() ->
     .await
     .context("steered request did not start")?;
 
-    // End the call while the same text turn is still in flight.
+    // Closed and the persisted tail must not wait for the in-flight model response.
     test.codex.submit(Op::RealtimeConversationClose).await?;
-    wait_for_event(&test.codex, |event| {
-        matches!(event, EventMsg::RealtimeConversationClosed(_))
+    let closed = wait_for_event_match(&test.codex, |event| match event {
+        EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
+        _ => None,
     })
     .await;
+    assert_eq!(closed.reason.as_deref(), Some("requested"));
+    if flush_transcript_tail_on_session_end {
+        let history = test.codex.load_history(/*include_archived*/ false).await?;
+        assert!(history.items.iter().any(|item| {
+            matches!(item, RolloutItem::ResponseItem(ResponseItemEnvelope {
+                item: ResponseItem::Message { role, content, .. }, ..
+            }) if role == "user" && content.iter().any(|item| matches!(item,
+                ContentItem::InputText { text } if text.contains("user: final transcript tail"))))
+        }));
+    }
+    if interrupt {
+        test.codex.submit(Op::Interrupt).await?;
+        wait_for_event(&test.codex, |event| {
+            matches!(event, EventMsg::TurnAborted(_))
+        })
+        .await;
+        let _ = second_completed_tx.send(());
+        realtime_server.shutdown().await;
+        api_server.shutdown().await;
+        return Ok(());
+    }
     let steered = test
         .codex
         .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
@@ -5811,14 +5871,31 @@ async fn inbound_handoff_request_updates_realtime_state_during_active_turn() ->
         "<realtime_conversation>\nGive frequent spoken progress updates.\n</realtime_conversation>";
     let end_instructions =
         "<realtime_conversation>\nReturn to normal text updates.\n</realtime_conversation>";
+    assert!(realtime_instructions[0].is_empty());
     assert_eq!(
-        realtime_instructions,
-        vec![
-            vec![],
-            vec![start_instructions.to_string()],
-            vec![start_instructions.to_string(), end_instructions.to_string()],
-        ]
+        realtime_instructions[1],
+        vec![start_instructions.to_string()]
     );
+    assert_eq!(
+        realtime_instructions.last().expect("missing final request"),
+        &vec![start_instructions.to_string(), end_instructions.to_string()]
+    );
+    if flush_transcript_tail_on_session_end {
+        for request in &requests {
+            let body: Value = serde_json::from_slice(request)?;
+            let input = body["input"].as_array().expect("request input array");
+            if let Some(end) = input
+                .iter()
+                .position(|item| item.to_string().contains("Return to normal text updates."))
+            {
+                let tail = input
+                    .iter()
+                    .position(|item| item.to_string().contains("user: final transcript tail"))
+                    .expect("realtime_end must not overtake the tail");
+                assert!(tail < end);
+            }
+        }
+    }
 
     realtime_server.shutdown().await;
     api_server.shutdown().await;