Record realtime conversation history in Core

codex-rs/app-server/README.md

@@ -1709,6 +1709,8 @@ Event notifications are the server-initiated event stream for thread lifecycles,
 
 Thread realtime publishes thread-scoped timeline item lifecycle notifications for paginated threads alongside its existing realtime notifications. Completed timeline items are durably interleaved with ordinary turn items by `thread/timeline/list`. Neither surface changes `ThreadItem`, `thread/read`, `thread/resume`, or `thread/fork`; clients ignore notification methods they do not recognize.
 
+Core records transcript segments, session boundaries, and backing-agent artifact promotions through its injected thread store, even without an app-server event listener. Presentation selection uses the same rules for every Core host. App-server translates Core's history events into the notifications below; it does not append those items again. Recording remains limited to paginated threads. A completed notification follows acceptance by the thread store, not an additional flush or power-loss durability barrier.
+
 Each realtime item has an `id`, a `realtimeSessionId`, and one of four types: `realtimeSessionStarted`, `transcriptSegment`, `bemItemPromoted`, or `realtimeSessionClosed`. A `bemItemPromoted` item references an existing backing-agent item by `turnId` and `itemId`; its `presentation` is `wholeItem`, `inlineMarkdown`, or `inlineVisualization` with an `index`.
 
 Recoverable configuration and initialization warnings use the existing `configWarning` notification: `{ summary, details?, path?, range? }`. App-server may emit it during initialization for config parsing and related setup diagnostics, or to the requesting connection during `thread/start` when that thread's exec-policy rules fail to parse.

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -61,6 +61,9 @@ use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadRealtimeClosedNotification;
 use codex_app_server_protocol::ThreadRealtimeErrorNotification;
 use codex_app_server_protocol::ThreadRealtimeItemAddedNotification;
+use codex_app_server_protocol::ThreadRealtimeItemCompletedNotification;
+use codex_app_server_protocol::ThreadRealtimeItemStartedNotification;
+use codex_app_server_protocol::ThreadRealtimeItemTranscriptDeltaNotification;
 use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification;
 use codex_app_server_protocol::ThreadRealtimeSdpNotification;
 use codex_app_server_protocol::ThreadRealtimeStartedNotification;
@@ -463,6 +466,39 @@ pub(crate) async fn apply_bespoke_event_handling(
                 .await;
         }
         EventMsg::RealtimeConversationRealtime(event) => match event.payload {
+            RealtimeEvent::HistoryItemStarted(item) => {
+                outgoing
+                    .send_server_notification(ServerNotification::ThreadRealtimeItemStarted(
+                        ThreadRealtimeItemStartedNotification {
+                            thread_id: conversation_id.to_string(),
+                            item: item.into(),
+                        },
+                    ))
+                    .await;
+            }
+            RealtimeEvent::HistoryTranscriptDelta { item_id, delta } => {
+                outgoing
+                    .send_server_notification(
+                        ServerNotification::ThreadRealtimeItemTranscriptDelta(
+                            ThreadRealtimeItemTranscriptDeltaNotification {
+                                thread_id: conversation_id.to_string(),
+                                item_id,
+                                delta,
+                            },
+                        ),
+                    )
+                    .await;
+            }
+            RealtimeEvent::HistoryItemCompleted(item) => {
+                outgoing
+                    .send_server_notification(ServerNotification::ThreadRealtimeItemCompleted(
+                        ThreadRealtimeItemCompletedNotification {
+                            thread_id: conversation_id.to_string(),
+                            item: item.into(),
+                        },
+                    ))
+                    .await;
+            }
             RealtimeEvent::SessionUpdated { .. } => {}
             RealtimeEvent::InputAudioSpeechStarted(event) => {
                 let notification = ThreadRealtimeItemAddedNotification {

codex-rs/app-server/src/lib.rs

@@ -122,8 +122,6 @@ mod models_refresh_worker;
 mod notification_media;
 mod otel_reloader;
 mod outgoing_message;
-mod realtime_event_handling;
-mod realtime_history;
 mod request_processors;
 mod request_serialization;
 mod server_request_error;

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -1,12 +1,8 @@
 use super::*;
 use crate::extensions::send_thread_warning;
-use crate::realtime_event_handling::apply_realtime_event_effects;
-use crate::realtime_event_handling::persist_realtime_items;
-use crate::realtime_history::RealtimeEventEffects;
 use codex_app_server_protocol::ThreadQueueChangedNotification;
 use codex_extension_api::ThreadIdleCause;
 use codex_protocol::config_types::MultiAgentMode;
-use codex_protocol::protocol::ThreadHistoryMode;
 
 pub(super) const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60);
 
@@ -247,8 +243,6 @@ pub(super) async fn ensure_listener_task_running(
         )
         .await;
     let config_snapshot = conversation.config_snapshot().await;
-    let realtime_history_enabled =
-        matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated);
     let thread_settings_baseline = thread_settings_from_config_snapshot(&config_snapshot);
     let (mut listener_command_rx, listener_generation) = {
         let mut thread_state = thread_state.lock().await;
@@ -331,20 +325,10 @@ pub(super) async fn ensure_listener_task_running(
                     // Track the event before emitting any typed translations
                     // so thread-local state such as raw event opt-in stays
                     // synchronized with the conversation.
-                    let (raw_events_enabled, realtime_effects) = {
+                    let raw_events_enabled = {
                         let mut thread_state = thread_state.lock().await;
                         thread_state.track_current_turn_event(&event.id, &event.msg);
-                        let realtime_effects = if realtime_history_enabled
-                            && thread_state.realtime_history.should_observe(&event.msg)
-                        {
-                            let active_turn_id = thread_state.active_turn_snapshot().map(|turn| turn.id);
-                            thread_state
-                                .realtime_history
-                                .observe(&event.msg, active_turn_id.as_deref())
-                        } else {
-                            RealtimeEventEffects::default()
-                        };
-                        (thread_state.experimental_raw_events, realtime_effects)
+                        thread_state.experimental_raw_events
                     };
                     if matches!(
                         &event.msg,
@@ -362,14 +346,6 @@ pub(super) async fn ensure_listener_task_running(
                         conversation_id,
                     );
 
-                    apply_realtime_event_effects(
-                        conversation.as_ref(),
-                        &thread_outgoing,
-                        conversation_id,
-                        realtime_effects,
-                    )
-                    .await;
-
                     apply_bespoke_event_handling(
                         event.clone(),
                         conversation_id,
@@ -582,32 +558,6 @@ pub(super) async fn handle_thread_listener_command(
             .await;
             let _ = completion_tx.send(());
         }
-        ThreadListenerCommand::SealRealtimeUserInput {
-            input,
-            completion_tx,
-        } => {
-            let items = thread_state
-                .lock()
-                .await
-                .realtime_history
-                .seal_user_input(&input);
-            let subscribed_connection_ids = thread_state_manager
-                .subscribed_connection_ids(conversation_id)
-                .await;
-            let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
-                outgoing.clone(),
-                subscribed_connection_ids,
-                conversation_id,
-            );
-            let result = persist_realtime_items(
-                conversation.as_ref(),
-                &thread_outgoing,
-                &conversation_id.to_string(),
-                items,
-            )
-            .await;
-            let _ = completion_tx.send(result);
-        }
     }
 }
 

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -619,10 +619,6 @@ impl TurnRequestProcessor {
                 },
             )
             .await?;
-        if let TurnInput::UserInput { content, .. } = &input {
-            self.seal_realtime_transcript_before_user_input(thread_id, content)
-                .await?;
-        }
 
         let submission = thread
             .start_or_steer_turn(
@@ -997,12 +993,12 @@ impl TurnRequestProcessor {
         request_id: &ConnectionRequestId,
         params: TurnSteerParams,
     ) -> Result<TurnSteerResponse, JSONRPCErrorError> {
-        let (thread_id, thread) =
-            self.load_thread(&params.thread_id)
-                .await
-                .inspect_err(|error| {
-                    self.track_error_response(request_id, error, /*error_type*/ None);
-                })?;
+        let (_, thread) = self
+            .load_thread(&params.thread_id)
+            .await
+            .inspect_err(|error| {
+                self.track_error_response(request_id, error, /*error_type*/ None);
+            })?;
         self.ensure_direct_input_allowed(request_id, thread.as_ref())
             .await?;
 
@@ -1028,9 +1024,6 @@ impl TurnRequestProcessor {
             .collect();
         let additional_context = map_additional_context(params.additional_context);
 
-        self.seal_realtime_transcript_before_user_input(thread_id, &mapped_items)
-            .await?;
-
         let submission = thread
             .steer_turn(
                 TurnInputRequest::new(TurnInput::UserInput {
@@ -1127,37 +1120,6 @@ impl TurnRequestProcessor {
         Ok(TurnSteerResponse { turn_id })
     }
 
-    async fn seal_realtime_transcript_before_user_input(
-        &self,
-        thread_id: ThreadId,
-        input: &[CoreInputItem],
-    ) -> Result<(), JSONRPCErrorError> {
-        let thread_state = self.thread_state_manager.thread_state(thread_id).await;
-        if !thread_state
-            .lock()
-            .await
-            .realtime_history
-            .should_seal_user_input(input)
-        {
-            return Ok(());
-        }
-        let listener = self
-            .thread_state_manager
-            .current_listener_command_tx(thread_id)
-            .ok_or_else(|| internal_error("thread listener is not running"))?;
-        let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
-        listener
-            .send(ThreadListenerCommand::SealRealtimeUserInput {
-                input: input.to_vec(),
-                completion_tx,
-            })
-            .map_err(|_| internal_error("thread listener is not running"))?;
-        completion_rx
-            .await
-            .map_err(|_| internal_error("thread listener stopped before sealing realtime input"))?
-            .map_err(internal_error)
-    }
-
     async fn prepare_realtime_conversation_thread(
         &self,
         request_id: &ConnectionRequestId,

codex-rs/app-server/src/thread_state.rs

@@ -1,6 +1,5 @@
 use crate::outgoing_message::ConnectionId;
 use crate::outgoing_message::ConnectionRequestId;
-use crate::realtime_history::RealtimeHistoryState;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadGoal;
 use codex_app_server_protocol::ThreadHistoryBuilder;
@@ -18,7 +17,6 @@ use codex_protocol::items::AgentMessageContent as CoreAgentMessageContent;
 use codex_protocol::items::TurnItem as CoreTurnItem;
 use codex_protocol::models::MessagePhase;
 use codex_protocol::protocol::EventMsg;
-use codex_protocol::user_input::UserInput;
 use codex_rollout::RolloutItem;
 use codex_rollout::state_db::StateDbHandle;
 use codex_utils_path_uri::LegacyAppPathString;
@@ -83,10 +81,6 @@ pub(crate) enum ThreadListenerCommand {
         request_id: RequestId,
         completion_tx: oneshot::Sender<()>,
     },
-    SealRealtimeUserInput {
-        input: Vec<UserInput>,
-        completion_tx: oneshot::Sender<Result<(), String>>,
-    },
 }
 
 /// Per-conversation accumulation of the latest states e.g. error message while a turn runs.
@@ -110,7 +104,6 @@ pub(crate) struct ThreadState {
     pub(crate) cancel_tx: Option<oneshot::Sender<()>>,
     pub(crate) experimental_raw_events: bool,
     pub(crate) listener_generation: u64,
-    pub(crate) realtime_history: RealtimeHistoryState,
     last_thread_settings: Option<ThreadSettings>,
     listener_command_tx: Option<mpsc::UnboundedSender<ThreadListenerCommand>>,
     current_turn_history: ThreadHistoryBuilder,