Record realtime conversation history in Core (#41924)

## Why

Realtime history should be recorded consistently for every Core host, including
when no app-server event listener is attached.

## What changed

- Move transcript segmentation, session boundaries, and backing-agent artifact
  promotion into Core for paginated threads.
- Persist canonical realtime items through the thread store in event order and
  emit dedicated history lifecycle events for hosts to present.
- Translate those Core events into the existing app-server realtime item
  notifications without app-server persisting the items a second time.

## Testing

- Cover Core-only persistence across repeated sessions, ephemeral sessions,
  accepted and rejected steering, typed input ordering, and artifact promotion.
- Verify app-server notifications correspond to the persisted timeline.

GitOrigin-RevId: 7cbef14129d77f6d6d7099b733be91f5279c55f5

codex-rs/core/src/session/mod.rs

@@ -40,6 +40,7 @@ use crate::image_preparation::prepare_response_items as prepare_image_response_i
 use crate::image_preparation::unified_image_budget_enabled;
 use crate::parse_turn_item;
 use crate::realtime_conversation::RealtimeConversationManager;
+use crate::realtime_history::RealtimeEventOrder;
 use crate::session::step_context::StepContext;
 use crate::session::step_settings::ResolvedStepSettings;
 use crate::session::step_settings::StepSettings;
@@ -227,6 +228,7 @@ mod mcp_prewarm;
 mod mcp_refresh;
 mod mcp_runtime;
 pub(crate) mod multi_agents;
+mod realtime_history;
 mod review;
 mod rollout_budget;
 mod rollout_reconstruction;
@@ -2339,7 +2341,32 @@ impl Session {
     }
 
     async fn send_event_raw_with_persistence(&self, event: Event, persist: bool) {
+        // Keep realtime reduction, canonical append, and delivery in the same order.
+        // This lock must not acquire SessionState or ActiveTurn: event producers can
+        // already hold those locks. Host presentation policies are synchronous.
+        let mut realtime_history = match &self.realtime_history {
+            Some(history) => {
+                let history = history.lock().await;
+                history.should_observe(&event.msg).then_some(history)
+            }
+            None => None,
+        };
         self.services.mcp_runtime.observe_event(&event.msg);
+        let (before_event, after_event) = match realtime_history.as_mut() {
+            Some(history) => {
+                let effects = history.observe(&event.msg);
+                match effects.order {
+                    RealtimeEventOrder::BeforeEvent => (Some(effects), None),
+                    RealtimeEventOrder::AfterEvent => (None, Some(effects)),
+                }
+            }
+            None => (None, None),
+        };
+        if let Some(effects) = before_event
+            && let Err(error) = self.send_realtime_history_effects(&event.id, effects).await
+        {
+            warn!("failed to persist realtime history: {error}");
+        }
         // Persist the event into rollout storage; the store applies its persistence policy.
         if persist {
             let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())];
@@ -2348,6 +2375,11 @@ impl Session {
         self.services
             .rollout_thread_trace
             .record_protocol_event(&event.msg);
+        if let Some(effects) = after_event
+            && let Err(error) = self.send_realtime_history_effects(&event.id, effects).await
+        {
+            warn!("failed to persist realtime history: {error}");
+        }
         self.deliver_event_raw(event).await;
     }
 

codex-rs/core/src/session/realtime_history.rs

@@ -0,0 +1,56 @@
+use super::session::Session;
+use crate::realtime_history::RealtimeEventEffects;
+use codex_history::RolloutItem;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
+use codex_protocol::protocol::RealtimeEvent;
+use codex_protocol::realtime::RealtimeItemContent;
+
+impl Session {
+    pub(super) async fn send_realtime_history_effects(
+        &self,
+        submission_id: &str,
+        effects: RealtimeEventEffects,
+    ) -> anyhow::Result<()> {
+        let event = |payload| Event {
+            id: submission_id.to_string(),
+            msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
+                payload,
+            }),
+        };
+        if let Some(stream) = effects.transcript_stream {
+            if let Some(item) = stream.started_item {
+                self.deliver_event_raw(event(RealtimeEvent::HistoryItemStarted(item)))
+                    .await;
+            }
+            self.deliver_event_raw(event(RealtimeEvent::HistoryTranscriptDelta {
+                item_id: stream.item_id,
+                delta: stream.delta,
+            }))
+            .await;
+        }
+        if effects.items.is_empty() {
+            return Ok(());
+        }
+        self.live_thread_for_persistence("append realtime history")?
+            .append_items(
+                &effects
+                    .items
+                    .iter()
+                    .cloned()
+                    .map(RolloutItem::RealtimeItem)
+                    .collect::<Vec<_>>(),
+            )
+            .await?;
+        for item in effects.items {
+            if !matches!(&item.content, RealtimeItemContent::TranscriptSegment { .. }) {
+                self.deliver_event_raw(event(RealtimeEvent::HistoryItemStarted(item.clone())))
+                    .await;
+            }
+            self.deliver_event_raw(event(RealtimeEvent::HistoryItemCompleted(item)))
+                .await;
+        }
+        Ok(())
+    }
+}

codex-rs/core/src/session/session.rs

@@ -64,6 +64,7 @@ pub(crate) struct Session {
     pub(super) mcp_prewarm_shutdown: CancellationToken,
     pub(super) mcp_prewarm_task: std::sync::Mutex<Option<JoinHandle<()>>>,
     pub(crate) conversation: Arc<RealtimeConversationManager>,
+    pub(crate) realtime_history: Option<Mutex<crate::realtime_history::RealtimeHistoryState>>,
     pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
     pub(crate) async_hook_results: async_channel::Receiver<HookCompletedEvent>,
     pub(crate) input_queue: InputQueue,
@@ -1492,6 +1493,9 @@ impl Session {
                 mcp_prewarm_shutdown: CancellationToken::new(),
                 mcp_prewarm_task: std::sync::Mutex::new(None),
                 conversation: Arc::new(RealtimeConversationManager::new()),
+                realtime_history: (session_configuration.history_mode == ThreadHistoryMode::Paginated
+                    && services.live_thread.is_some())
+                .then(|| Mutex::new(Default::default())),
                 active_turn: Mutex::new(None),
                 async_hook_results,
                 input_queue: InputQueue::new(),

codex-rs/core/src/session/tests.rs

@@ -6612,6 +6612,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         mcp_prewarm_shutdown: CancellationToken::new(),
         mcp_prewarm_task: std::sync::Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),
+        realtime_history: None,
         active_turn: Mutex::new(None),
         async_hook_results,
         input_queue: super::input_queue::InputQueue::new(),
@@ -8892,6 +8893,7 @@ where
         mcp_prewarm_shutdown: CancellationToken::new(),
         mcp_prewarm_task: std::sync::Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),
+        realtime_history: None,
         active_turn: Mutex::new(None),
         async_hook_results,
         input_queue: super::input_queue::InputQueue::new(),

codex-rs/core/src/session/turn_input_tests.rs

@@ -1,6 +1,7 @@
 use super::*;
 use crate::config::Constrained;
 use crate::session::step_settings::StepSettingsUpdate;
+use crate::session::tests::make_session_and_context;
 use crate::session::tests::make_session_and_context_with_rx;
 use crate::session::turn_context::TurnContext;
 use crate::state::TaskKind;
@@ -107,6 +108,65 @@ async fn submit_steer_only(
     .expect("steer-only submission should be valid")
 }
 
+#[tokio::test]
+#[expect(
+    clippy::await_holding_invalid_type,
+    reason = "simulate an in-flight realtime append while checking input admission"
+)]
+async fn steering_does_not_wait_for_realtime_history() {
+    let (mut session, turn_context) = make_session_and_context().await;
+    session.realtime_history = Some(tokio::sync::Mutex::new(Default::default()));
+    let session = Arc::new(session);
+    let turn_context = Arc::new(turn_context);
+    session
+        .spawn_task(
+            Arc::clone(&turn_context),
+            Vec::new(),
+            NeverEndingTask {
+                kind: TaskKind::Regular,
+                listen_to_cancellation_token: true,
+            },
+        )
+        .await;
+
+    let history = session
+        .realtime_history
+        .as_ref()
+        .expect("realtime history")
+        .lock()
+        .await;
+    for mode in [
+        TurnInputMode::StartOrSteer,
+        TurnInputMode::Steer {
+            expected_turn_id: turn_context.sub_id.clone(),
+        },
+    ] {
+        let submission = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            handle(
+                &session,
+                TurnInputRequest::user_input(vec![UserInput::Text {
+                    text: "steer without waiting for persistence".to_string(),
+                    text_elements: Vec::new(),
+                }]),
+                mode,
+                "steer-submission".to_string(),
+            ),
+        )
+        .await
+        .expect("steering must not wait for the realtime recorder")
+        .expect("steering should succeed");
+        assert_eq!(
+            submission,
+            TurnInputSubmission::Steered {
+                turn_id: turn_context.sub_id.clone()
+            }
+        );
+    }
+    drop(history);
+    session.abort_all_tasks(TurnAbortReason::Interrupted).await;
+}
+
 #[tokio::test]
 async fn accepted_input_applies_thread_settings() {
     let (session, turn_context, _rx) = make_session_and_context_with_rx().await;