Add snapshot pagination for persisted realtime history (#61480)

## What changed

Add experimental `thread/realtime/list` reads independent of turn pagination or a live runtime. Support session, observation-time, and event-ID filters, with ordering by observation time or committed position.

Keep cursor pagination within a fixed snapshot and expose positions for catching up on events committed later. Return `unsupported`, `pending`, or `ready` coverage so clients can distinguish unavailable history from events awaiting indexing. Limit fork reads to the child's own realtime history.

## Testing

Add SQLite tests for snapshot stability, late-event catch-up, cursor validation, and coverage tracking. Extend app-server tests to cover filtering, exact event lookup, and fork isolation.

GitOrigin-RevId: 8158df53f10d948ede4911654aa8adbc7297b258

codex-rs/app-server-protocol/schema/json/ClientRequest.json

@@ -5476,6 +5476,24 @@
       ],
       "type": "object"
     },
+    "ThreadRealtimeOrder": {
+      "oneOf": [
+        {
+          "description": "Sort by canonical observation time, then committed position.",
+          "enum": [
+            "observedAt"
+          ],
+          "type": "string"
+        },
+        {
+          "description": "Sort by committed position, including events sealed after older snapshots.",
+          "enum": [
+            "position"
+          ],
+          "type": "string"
+        }
+      ]
+    },
     "ThreadRealtimeStartTransport": {
       "description": "EXPERIMENTAL - transport used by thread realtime.",
       "oneOf": [

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -25561,6 +25561,31 @@
         "title": "ThreadRealtimeClosedNotification",
         "type": "object"
       },
+      "ThreadRealtimeCoverage": {
+        "enum": [
+          "unsupported",
+          "pending",
+          "ready"
+        ],
+        "type": "string"
+      },
+      "ThreadRealtimeEntry": {
+        "properties": {
+          "item": {
+            "$ref": "#/definitions/v2/ThreadRealtimeItem"
+          },
+          "position": {
+            "format": "uint64",
+            "minimum": 0.0,
+            "type": "integer"
+          }
+        },
+        "required": [
+          "item",
+          "position"
+        ],
+        "type": "object"
+      },
       "ThreadRealtimeErrorNotification": {
         "$schema": "http://json-schema.org/draft-07/schema#",
         "description": "EXPERIMENTAL - emitted when thread realtime encounters an error.",
@@ -25896,6 +25921,24 @@
         "title": "ThreadRealtimeItemTranscriptDeltaNotification",
         "type": "object"
       },
+      "ThreadRealtimeOrder": {
+        "oneOf": [
+          {
+            "description": "Sort by canonical observation time, then committed position.",
+            "enum": [
+              "observedAt"
+            ],
+            "type": "string"
+          },
+          {
+            "description": "Sort by committed position, including events sealed after older snapshots.",
+            "enum": [
+              "position"
+            ],
+            "type": "string"
+          }
+        ]
+      },
       "ThreadRealtimeOutputAudioDeltaNotification": {
         "$schema": "http://json-schema.org/draft-07/schema#",
         "description": "EXPERIMENTAL - streamed output audio emitted by thread realtime.",

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json

@@ -23257,6 +23257,31 @@
       "title": "ThreadRealtimeClosedNotification",
       "type": "object"
     },
+    "ThreadRealtimeCoverage": {
+      "enum": [
+        "unsupported",
+        "pending",
+        "ready"
+      ],
+      "type": "string"
+    },
+    "ThreadRealtimeEntry": {
+      "properties": {
+        "item": {
+          "$ref": "#/definitions/ThreadRealtimeItem"
+        },
+        "position": {
+          "format": "uint64",
+          "minimum": 0.0,
+          "type": "integer"
+        }
+      },
+      "required": [
+        "item",
+        "position"
+      ],
+      "type": "object"
+    },
     "ThreadRealtimeErrorNotification": {
       "$schema": "http://json-schema.org/draft-07/schema#",
       "description": "EXPERIMENTAL - emitted when thread realtime encounters an error.",
@@ -23592,6 +23617,24 @@
       "title": "ThreadRealtimeItemTranscriptDeltaNotification",
       "type": "object"
     },
+    "ThreadRealtimeOrder": {
+      "oneOf": [
+        {
+          "description": "Sort by canonical observation time, then committed position.",
+          "enum": [
+            "observedAt"
+          ],
+          "type": "string"
+        },
+        {
+          "description": "Sort by committed position, including events sealed after older snapshots.",
+          "enum": [
+            "position"
+          ],
+          "type": "string"
+        }
+      ]
+    },
     "ThreadRealtimeOutputAudioDeltaNotification": {
       "$schema": "http://json-schema.org/draft-07/schema#",
       "description": "EXPERIMENTAL - streamed output audio emitted by thread realtime.",

codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeCoverage.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ThreadRealtimeCoverage = "unsupported" | "pending" | "ready";

codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeEntry.ts

@@ -0,0 +1,6 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { ThreadRealtimeItem } from "./ThreadRealtimeItem";
+
+export type ThreadRealtimeEntry = { position: number, item: ThreadRealtimeItem, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOrder.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ThreadRealtimeOrder = "observedAt" | "position";

codex-rs/app-server-protocol/schema/typescript/v2/index.ts

@@ -584,6 +584,8 @@ export type { ThreadReadStateOperation } from "./ThreadReadStateOperation";
 export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk";
 export type { ThreadRealtimeBemItemPresentation } from "./ThreadRealtimeBemItemPresentation";
 export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification";
+export type { ThreadRealtimeCoverage } from "./ThreadRealtimeCoverage";
+export type { ThreadRealtimeEntry } from "./ThreadRealtimeEntry";
 export type { ThreadRealtimeErrorNotification } from "./ThreadRealtimeErrorNotification";
 export type { ThreadRealtimeFrameChannels } from "./ThreadRealtimeFrameChannels";
 export type { ThreadRealtimeHandoffMode } from "./ThreadRealtimeHandoffMode";
@@ -593,6 +595,7 @@ export type { ThreadRealtimeItemAddedNotification } from "./ThreadRealtimeItemAd
 export type { ThreadRealtimeItemCompletedNotification } from "./ThreadRealtimeItemCompletedNotification";
 export type { ThreadRealtimeItemStartedNotification } from "./ThreadRealtimeItemStartedNotification";
 export type { ThreadRealtimeItemTranscriptDeltaNotification } from "./ThreadRealtimeItemTranscriptDeltaNotification";
+export type { ThreadRealtimeOrder } from "./ThreadRealtimeOrder";
 export type { ThreadRealtimeOutputAudioDeltaNotification } from "./ThreadRealtimeOutputAudioDeltaNotification";
 export type { ThreadRealtimeSdpNotification } from "./ThreadRealtimeSdpNotification";
 export type { ThreadRealtimeSessionOutcome } from "./ThreadRealtimeSessionOutcome";

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -1170,6 +1170,12 @@ client_request_definitions! {
         serialization: thread_id(params.thread_id),
         response: v2::ThreadTimelineListResponse,
     },
+    #[experimental("thread/realtime/list")]
+    ThreadRealtimeList => "thread/realtime/list" {
+        params: v2::ThreadRealtimeListParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadRealtimeListResponse,
+    },
     #[experimental("thread/realtime/listVoices")]
     ThreadRealtimeListVoices => "thread/realtime/listVoices" {
         params: v2::ThreadRealtimeListVoicesParams,

codex-rs/app-server-protocol/src/protocol/v2/mod.rs

@@ -30,6 +30,7 @@ mod plugin_search;
 mod process;
 mod project;
 mod realtime;
+mod realtime_history;
 mod remote_control;
 mod review;
 mod rollout;
@@ -73,6 +74,7 @@ pub use plugin_search::*;
 pub use process::*;
 pub use project::*;
 pub use realtime::*;
+pub use realtime_history::*;
 pub use remote_control::*;
 pub use review::*;
 pub use rollout::*;

codex-rs/app-server-protocol/src/protocol/v2/realtime_history.rs

@@ -0,0 +1,84 @@
+//! Persisted realtime history reads, independent of turn pagination or a live runtime.
+
+use super::ThreadRealtimeItem;
+use crate::JsonSchema;
+use crate::TS;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadRealtimeListParams {
+    pub thread_id: String,
+    #[ts(optional = nullable)]
+    pub session_id: Option<String>,
+    /// Inclusive canonical observation time lower bound.
+    #[ts(optional = nullable, type = "number | null")]
+    pub from_observed_at_ms: Option<i64>,
+    /// Exclusive canonical observation time upper bound.
+    #[ts(optional = nullable, type = "number | null")]
+    pub to_observed_at_ms: Option<i64>,
+    /// Defaults to canonical observation time.
+    #[ts(optional = nullable)]
+    pub order: Option<ThreadRealtimeOrder>,
+    #[ts(optional = nullable, type = "number | null")]
+    pub after_position: Option<u64>,
+    /// A required exact event ID, such as a completed item's acknowledgement.
+    /// Returns pending until projected. Explicit filters can exclude a known event.
+    #[ts(optional = nullable)]
+    pub event_id: Option<String>,
+    #[ts(optional = nullable)]
+    pub cursor: Option<String>,
+    #[ts(optional = nullable)]
+    pub limit: Option<u32>,
+    /// Exclusive committed position delimiting an immutable snapshot.
+    #[ts(optional = nullable, type = "number | null")]
+    pub snapshot_end: Option<u64>,
+    /// Return pending until the projection includes this committed position.
+    #[ts(optional = nullable, type = "number | null")]
+    pub required_through_position: Option<u64>,
+}
+
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(rename_all = "camelCase", export_to = "v2/")]
+pub enum ThreadRealtimeOrder {
+    /// Sort by canonical observation time, then committed position.
+    #[default]
+    ObservedAt,
+    /// Sort by committed position, including events sealed after older snapshots.
+    Position,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(rename_all = "camelCase", export_to = "v2/")]
+pub enum ThreadRealtimeCoverage {
+    Unsupported,
+    Pending,
+    Ready,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadRealtimeEntry {
+    #[ts(type = "number")]
+    pub position: u64,
+    pub item: ThreadRealtimeItem,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadRealtimeListResponse {
+    pub data: Vec<ThreadRealtimeEntry>,
+    pub next_cursor: Option<String>,
+    #[ts(type = "number")]
+    pub snapshot_end: u64,
+    /// Exclusive position through which the projection has processed canonical history.
+    #[ts(type = "number")]
+    pub projected_through_item_index: u64,
+    pub coverage: ThreadRealtimeCoverage,
+}

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

@@ -1820,6 +1820,9 @@ impl MessageProcessor {
             ClientRequest::ThreadTimelineList { params, .. } => {
                 self.thread_processor.thread_timeline_list(params).await
             }
+            ClientRequest::ThreadRealtimeList { params, .. } => {
+                self.thread_processor.thread_realtime_list(params).await
+            }
             ClientRequest::ThreadRealtimeListVoices { .. } => {
                 self.turn_processor.thread_realtime_list_voices().await
             }

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

@@ -0,0 +1,38 @@
+//! Serves persisted realtime history through the store without acquiring a runtime.
+
+use super::ClientResponsePayload;
+use super::JSONRPCErrorError;
+use super::ThreadRequestProcessor;
+use super::invalid_request;
+use super::paginated_history_list_error;
+use codex_app_server_protocol::ThreadRealtimeListParams;
+use codex_protocol::ThreadId;
+use codex_thread_store::ListRealtimeParams;
+
+impl ThreadRequestProcessor {
+    pub(crate) async fn thread_realtime_list(
+        &self,
+        params: ThreadRealtimeListParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        let thread_id = ThreadId::from_string(&params.thread_id)
+            .map_err(|error| invalid_request(format!("invalid thread id: {error}")))?;
+        let page = self
+            .thread_store
+            .list_realtime(ListRealtimeParams {
+                thread_id,
+                session_id: params.session_id,
+                from_observed_at_ms: params.from_observed_at_ms,
+                to_observed_at_ms: params.to_observed_at_ms,
+                order: params.order.unwrap_or_default(),
+                after_position: params.after_position,
+                event_id: params.event_id,
+                cursor: params.cursor,
+                page_size: params.limit.unwrap_or(100) as usize,
+                snapshot_end: params.snapshot_end,
+                required_through_position: params.required_through_position,
+            })
+            .await
+            .map_err(paginated_history_list_error)?;
+        Ok(Some(page.into()))
+    }
+}

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

@@ -4,6 +4,9 @@ mod daemon_continuation;
 #[path = "daemon_snapshot.rs"]
 mod daemon_snapshot;
 
+#[path = "realtime_history.rs"]
+mod realtime_history;
+
 mod workspace_switching;
 
 #[path = "thread_resume_account.rs"]

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

@@ -29,15 +29,64 @@ use codex_protocol::user_input::UserInput;
 use codex_rollout::RolloutItem;
 use codex_thread_store::AppendThreadItemsParams;
 use codex_thread_store::CreateThreadParams;
+use codex_thread_store::ForkBoundary;
+use codex_thread_store::LiveThread;
 use codex_thread_store::LocalThreadStore;
 use codex_thread_store::LocalThreadStoreConfig;
 use codex_thread_store::PersistContext;
+use codex_thread_store::PrepareForkParams;
 use codex_thread_store::ThreadPersistenceMetadata;
 use codex_thread_store::ThreadStore;
 use codex_utils_absolute_path::test_support::PathExt;
 use pretty_assertions::assert_eq;
+use std::path::Path;
+use std::sync::Arc;
 use tempfile::TempDir;
 
+async fn local_store(codex_home: &Path) -> Result<LocalThreadStore> {
+    let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.abs());
+    let state_db =
+        codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?;
+    Ok(LocalThreadStore::new(
+        LocalThreadStoreConfig {
+            codex_home: codex_home.to_path_buf(),
+            sqlite,
+            default_model_provider_id: "mock_provider".to_string(),
+        },
+        Some(state_db),
+    ))
+}
+
+fn create_params(codex_home: &Path, thread_id: ThreadId) -> CreateThreadParams {
+    CreateThreadParams {
+        attached_thread: false,
+        creator_user_id: None,
+        creator_account_id: None,
+        session_id: thread_id.into(),
+        thread_id,
+        extra_config: None,
+        forked_from_id: None,
+        parent_thread_id: None,
+        source: SessionSource::Cli,
+        thread_source: None,
+        originator: "test_originator".to_string(),
+        base_instructions: BaseInstructions::default(),
+        dynamic_tools: Vec::new(),
+        selected_capability_roots: Vec::new(),
+        multi_agent_version: None,
+        history_mode: ThreadHistoryMode::Paginated,
+        history_base: None,
+        subagent_history_start_ordinal: None,
+        initial_window_id: codex_protocol::ContextWindowId::new_uuid_v7(),
+        runtime_workspace_roots: None,
+        metadata: ThreadPersistenceMetadata {
+            cwd: Some(codex_home.to_path_buf()),
+            model_provider: "mock_provider".to_string(),
+            memory_mode: ThreadMemoryMode::Enabled,
+        },
+    }
+}
+
 #[tokio::test]
 async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -45,45 +94,9 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() ->
     MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
 
     let thread_id = ThreadId::default();
-    let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
-    let state_db =
-        codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?;
-    let store = LocalThreadStore::new(
-        LocalThreadStoreConfig {
-            codex_home: codex_home.path().to_path_buf(),
-            sqlite,
-            default_model_provider_id: "mock_provider".to_string(),
-        },
-        Some(state_db),
-    );
+    let store = local_store(codex_home.path()).await?;
     store
-        .create_thread(CreateThreadParams {
-            attached_thread: false,
-            creator_user_id: None,
-            creator_account_id: None,
-            session_id: thread_id.into(),
-            thread_id,
-            extra_config: None,
-            forked_from_id: None,
-            parent_thread_id: None,
-            source: SessionSource::Cli,
-            thread_source: None,
-            originator: "test_originator".to_string(),
-            base_instructions: BaseInstructions::default(),
-            dynamic_tools: Vec::new(),
-            selected_capability_roots: Vec::new(),
-            multi_agent_version: None,
-            history_mode: ThreadHistoryMode::Paginated,
-            history_base: None,
-            subagent_history_start_ordinal: None,
-            initial_window_id: codex_protocol::ContextWindowId::new_uuid_v7(),
-            runtime_workspace_roots: None,
-            metadata: ThreadPersistenceMetadata {
-                cwd: Some(codex_home.path().to_path_buf()),
-                model_provider: "mock_provider".to_string(),
-                memory_mode: ThreadMemoryMode::Enabled,
-            },
-        })
+        .create_thread(create_params(codex_home.path(), thread_id))
         .await?;
     store
         .persist_thread(thread_id, PersistContext::Standard)
@@ -125,10 +138,14 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() ->
                 RolloutItem::RealtimeItem(RealtimeItem {
                     id: "voice-1:transcript-1".to_string(),
                     realtime_session_id: "voice-1".to_string(),
-                    observed_at_ms: None,
-                    content: RealtimeItemContent::TranscriptSegment {
-                        role: RealtimeTranscriptRole::Assistant,
-                        text: "Checking staging".to_string(),
+                    observed_at_ms: Some(2_100),
+                    content: RealtimeItemContent::Frame {
+                        channels: codex_protocol::realtime::RealtimeFrameChannels {
+                            voice_transcript: codex_protocol::realtime::RealtimeVoiceTranscript {
+                                role: RealtimeTranscriptRole::Assistant,
+                                text: "Checking staging".to_string(),
+                            },
+                        },
                     },
                 }),
                 RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
@@ -151,6 +168,52 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() ->
         .without_auto_env()
         .build_initialized()
         .await?;
+    let realtime: codex_app_server_protocol::ThreadRealtimeListResponse = app_server
+        .request(|request_id| ClientRequest::ThreadRealtimeList {
+            request_id,
+            params: codex_app_server_protocol::ThreadRealtimeListParams {
+                thread_id: thread_id.to_string(),
+                session_id: Some("voice-1".into()),
+                from_observed_at_ms: Some(1_000),
+                to_observed_at_ms: Some(2_000),
+                order: None,
+                after_position: None,
+                event_id: None,
+                cursor: None,
+                limit: Some(10),
+                snapshot_end: None,
+                required_through_position: None,
+            },
+        })
+        .await?;
+    assert_eq!(
+        realtime.coverage,
+        codex_app_server_protocol::ThreadRealtimeCoverage::Ready
+    );
+    assert_eq!(realtime.data.len(), 1);
+    assert_eq!(realtime.data[0].item.id, "voice-1:started");
+    assert_eq!(realtime.data[0].item.observed_at_ms, Some(1_000));
+
+    let by_id: codex_app_server_protocol::ThreadRealtimeListResponse = app_server
+        .request(|request_id| ClientRequest::ThreadRealtimeList {
+            request_id,
+            params: codex_app_server_protocol::ThreadRealtimeListParams {
+                thread_id: thread_id.to_string(),
+                event_id: Some("voice-1:transcript-1".into()),
+                ..Default::default()
+            },
+        })
+        .await?;
+    let [entry] = by_id.data.as_slice() else {
+        panic!("expected the persisted voice frame");
+    };
+    assert_eq!(entry.item.observed_at_ms, Some(2_100));
+    assert!(matches!(
+        &entry.item.content,
+        ThreadRealtimeItemContent::Frame { channels }
+            if channels.voice_transcript.text == "Checking staging"
+    ));
+
     let page: ThreadTimelineListResponse = app_server
         .request(|request_id| ClientRequest::ThreadTimelineList {
             request_id,
@@ -234,3 +297,93 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() ->
 
     Ok(())
 }
+
+#[tokio::test]
+async fn realtime_list_reads_only_frames_sealed_on_the_paginated_fork() -> Result<()> {
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
+    let codex_home = TempDir::new()?;
+    MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
+    let store = Arc::new(local_store(codex_home.path()).await?);
+    let parent_id = ThreadId::default();
+    let child_id = ThreadId::default();
+    let frame = |id: &str, session: &str| RealtimeItem {
+        id: id.to_owned(),
+        realtime_session_id: session.to_owned(),
+        observed_at_ms: Some(1_000),
+        content: RealtimeItemContent::Frame {
+            channels: codex_protocol::realtime::RealtimeFrameChannels {
+                voice_transcript: codex_protocol::realtime::RealtimeVoiceTranscript {
+                    role: RealtimeTranscriptRole::User,
+                    text: session.to_owned(),
+                },
+            },
+        },
+    };
+    let parent =
+        LiveThread::create(store.clone(), create_params(codex_home.path(), parent_id)).await?;
+    parent.persist(PersistContext::Standard).await?;
+    parent
+        .append_items(&[RolloutItem::RealtimeItem(frame(
+            "parent-frame",
+            "parent-call",
+        ))])
+        .await?;
+    parent.flush().await?;
+    parent.shutdown().await?;
+
+    let fork = store
+        .prepare_fork(PrepareForkParams {
+            thread_id: parent_id,
+            boundary: ForkBoundary::Latest,
+        })
+        .await?;
+    let child = LiveThread::create(
+        store,
+        CreateThreadParams {
+            forked_from_id: Some(fork.source_thread_id),
+            history_base: fork.history_base,
+            ..create_params(codex_home.path(), child_id)
+        },
+    )
+    .await?;
+    child.persist(PersistContext::Standard).await?;
+    child
+        .append_items(&[RolloutItem::RealtimeItem(frame(
+            "child-frame",
+            "child-call",
+        ))])
+        .await?;
+    child.flush().await?;
+    child.shutdown().await?;
+    drop(fork);
+
+    let mut app_server = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_auto_env()
+        .build_initialized()
+        .await?;
+    for event_id in [None, Some("child-frame".to_owned())] {
+        let page: codex_app_server_protocol::ThreadRealtimeListResponse = app_server
+            .request(|request_id| ClientRequest::ThreadRealtimeList {
+                request_id,
+                params: codex_app_server_protocol::ThreadRealtimeListParams {
+                    thread_id: child_id.to_string(),
+                    event_id,
+                    ..Default::default()
+                },
+            })
+            .await?;
+        assert_eq!(
+            page.coverage,
+            codex_app_server_protocol::ThreadRealtimeCoverage::Ready
+        );
+        assert_eq!(
+            page.data
+                .iter()
+                .map(|entry| entry.item.id.as_str())
+                .collect::<Vec<_>>(),
+            ["child-frame"],
+        );
+    }
+    Ok(())
+}

codex-rs/state/thread_history_migrations/0009_realtime_observation_indexes.sql

@@ -0,0 +1,12 @@
+-- Support time ranges and position catch-up directly on the realtime projection.
+CREATE INDEX idx_thread_realtime_observed
+    ON thread_realtime_items(thread_id, json_extract(item_json, '$.observed_at_ms'), rollout_ordinal, item_id)
+    WHERE json_extract(item_json, '$.observed_at_ms') IS NOT NULL;
+
+CREATE INDEX idx_thread_realtime_session_observed
+    ON thread_realtime_items(thread_id, json_extract(item_json, '$.realtime_session_id'), json_extract(item_json, '$.observed_at_ms'), rollout_ordinal, item_id)
+    WHERE json_extract(item_json, '$.observed_at_ms') IS NOT NULL;
+
+CREATE INDEX idx_thread_realtime_session_position
+    ON thread_realtime_items(thread_id, json_extract(item_json, '$.realtime_session_id'), rollout_ordinal, item_id)
+    WHERE json_extract(item_json, '$.observed_at_ms') IS NOT NULL;

codex-rs/thread-store/src/lib.rs

@@ -12,6 +12,7 @@ mod local_history_account;
 mod original_history;
 mod projects;
 mod queue_store;
+mod realtime;
 mod store;
 mod thread_attachments;
 mod thread_metadata_sync;
@@ -61,6 +62,8 @@ pub use projects::UpdateProjectParams;
 pub use projects::UpdatedProject;
 pub use queue_store::LocalQueueStore;
 pub use queue_store::QueueStore;
+pub use realtime::ListRealtimeParams;
+pub use realtime::RealtimePage;
 pub use store::PendingDaybreakPreference;
 pub use store::PersistContext;
 pub use store::ThreadStore;

codex-rs/thread-store/src/local/mod.rs

@@ -498,6 +498,14 @@ impl LocalThreadStore {
         thread_history::list_items(self, params).await
     }
 
+    /// Reads the independent realtime projection without starting a runtime.
+    pub async fn list_realtime(
+        &self,
+        params: crate::ListRealtimeParams,
+    ) -> ThreadStoreResult<crate::RealtimePage> {
+        thread_history::list_realtime(self, params).await
+    }
+
     /// Lists bounded ordinary and realtime items from the rollout projection.
     pub async fn list_timeline(
         &self,
@@ -818,6 +826,13 @@ impl ThreadStore for LocalThreadStore {
         Box::pin(LocalThreadStore::list_timeline(self, params))
     }
 
+    fn list_realtime(
+        &self,
+        params: crate::ListRealtimeParams,
+    ) -> ThreadStoreFuture<'_, crate::RealtimePage> {
+        Box::pin(LocalThreadStore::list_realtime(self, params))
+    }
+
     fn search_threads(
         &self,
         params: SearchThreadsParams,

codex-rs/thread-store/src/local/thread_history.rs

@@ -12,6 +12,7 @@ use crate::ThreadStoreResult;
 
 mod read;
 mod realtime;
+mod realtime_stream;
 mod search;
 mod segment_paging;
 mod turn_lookup;
@@ -19,6 +20,7 @@ mod turn_lookup;
 pub(super) use read::list_items;
 pub(super) use read::list_turns;
 pub(super) use realtime::list_timeline;
+pub(super) use realtime_stream::list_realtime;
 pub(super) use search::search_thread_occurrences;
 pub(super) use turn_lookup::find_source_turn;
 pub(super) use turn_lookup::find_visible_turn;

codex-rs/thread-store/src/local/thread_history/realtime_stream.rs

@@ -0,0 +1,330 @@
+//! Reads immutable realtime snapshots by observation time or committed position.
+//! Cursors retain the query and snapshot bound so late seals appear in catch-up reads.
+
+use codex_app_server_protocol::ThreadRealtimeCoverage;
+use codex_app_server_protocol::ThreadRealtimeEntry;
+use codex_app_server_protocol::ThreadRealtimeItem;
+use codex_app_server_protocol::ThreadRealtimeOrder;
+use codex_protocol::ThreadId;
+use codex_protocol::realtime::RealtimeItem;
+use serde::Deserialize;
+use serde::Serialize;
+use sqlx::QueryBuilder;
+use sqlx::Sqlite;
+
+use super::super::LocalThreadStore;
+use super::read::validate_thread_for_paginated_reads;
+use super::sqlite_integer;
+use super::thread_history_error;
+use crate::ListRealtimeParams;
+use crate::RealtimePage;
+use crate::ThreadStoreError;
+use crate::ThreadStoreResult;
+
+const OBSERVED_AT: &str = "json_extract(item_json, '$.observed_at_ms')";
+
+#[derive(Deserialize, Serialize)]
+struct Cursor {
+    version: u8,
+    rollout_id: ThreadId,
+    query: ListRealtimeParams,
+    snapshot_end: u64,
+    observed_at_ms: Option<i64>,
+    position: u64,
+}
+
+pub(in crate::local) async fn list_realtime(
+    store: &LocalThreadStore,
+    params: ListRealtimeParams,
+) -> ThreadStoreResult<RealtimePage> {
+    if params.page_size == 0 || params.page_size > 500 {
+        return Err(invalid("realtime page size must be between 1 and 500"));
+    }
+    if params
+        .from_observed_at_ms
+        .zip(params.to_observed_at_ms)
+        .is_some_and(|(from, to)| from >= to)
+    {
+        return Err(invalid("realtime observation time range must be nonempty"));
+    }
+    match validate_thread_for_paginated_reads(
+        store,
+        params.thread_id,
+        /*include_archived*/ true,
+        "thread/realtime/list",
+    )
+    .await
+    {
+        Ok(()) => {}
+        Err(ThreadStoreError::Unsupported { .. }) => {
+            return Ok(empty_page(
+                ThreadRealtimeCoverage::Unsupported,
+                /*snapshot_end*/ 0,
+                /*projected_through_item_index*/ 0,
+            ));
+        }
+        Err(error) => return Err(error),
+    }
+    let _writer_guard = store.live_writer_locks.lock(params.thread_id).await;
+    let resolved = super::super::thread_rollout_resolver::resolve_current_including_archived(
+        store,
+        params.thread_id,
+    )
+    .await?;
+    let Some(resolved) = resolved else {
+        return read_page(
+            store.thread_history_db().await?,
+            params.clone(),
+            params.thread_id,
+            /*first_position*/ 0,
+            /*projected_end*/ 0,
+        )
+        .await;
+    };
+    let meta = codex_rollout::read_session_meta_line(&resolved.path)
+        .await
+        .map_err(thread_history_error)?
+        .meta;
+    let child_start = meta
+        .subagent_history_start_ordinal
+        .or_else(|| meta.history_base.map(|base| base.end_ordinal_exclusive));
+    // A copied fork without a boundary cannot distinguish its own frames from the parent's.
+    if meta.forked_from_id.is_some() && child_start.is_none() {
+        return Ok(empty_page(
+            ThreadRealtimeCoverage::Unsupported,
+            /*snapshot_end*/ 0,
+            /*projected_through_item_index*/ 0,
+        ));
+    }
+    super::super::thread_history_materialization::materialize_to_sqlite(
+        store,
+        resolved.rollout_id,
+        &resolved.path,
+    )
+    .await?;
+    let pool = store.thread_history_db().await?;
+    let projected_end = super::projection_state(store, resolved.rollout_id)
+        .await?
+        .map_or(0, |state| state.next_ordinal);
+    read_page(
+        pool,
+        params,
+        resolved.rollout_id,
+        child_start.unwrap_or(0),
+        projected_end,
+    )
+    .await
+}
+
+async fn read_page(
+    pool: &sqlx::SqlitePool,
+    params: ListRealtimeParams,
+    rollout_id: ThreadId,
+    first_position: u64,
+    projected_end: u64,
+) -> ThreadStoreResult<RealtimePage> {
+    let mut query = params.clone();
+    query.cursor = None;
+    query.page_size = 0;
+    // The effective snapshot is bound independently, so callers can omit it on later pages.
+    query.snapshot_end = None;
+    let cursor = params
+        .cursor
+        .as_deref()
+        .map(serde_json::from_str::<Cursor>)
+        .transpose()
+        .map_err(|_| invalid("invalid realtime cursor"))?;
+    if let Some(cursor) = &cursor
+        && (cursor.version != 1
+            || cursor.rollout_id != rollout_id
+            || cursor.query != query
+            || params
+                .snapshot_end
+                .is_some_and(|end| end != cursor.snapshot_end)
+            || cursor.position < first_position
+            || cursor.position >= cursor.snapshot_end)
+    {
+        return Err(invalid("realtime cursor does not match this query"));
+    }
+    let snapshot_end = cursor
+        .as_ref()
+        .map(|cursor| cursor.snapshot_end)
+        .or(params.snapshot_end)
+        .unwrap_or(projected_end);
+    let exact_id_uses_current_snapshot =
+        params.event_id.is_some() && cursor.is_none() && params.snapshot_end.is_none();
+    if projected_end < snapshot_end
+        || params
+            .required_through_position
+            .is_some_and(|position| projected_end <= position)
+    {
+        return Ok(empty_page(
+            ThreadRealtimeCoverage::Pending,
+            snapshot_end,
+            projected_end,
+        ));
+    }
+    let snapshot_end_sql = sqlite_integer(snapshot_end, "snapshot end")?;
+    let mut sql = QueryBuilder::<Sqlite>::new(
+        "SELECT rollout_ordinal, item_json FROM thread_realtime_items WHERE thread_id = ",
+    );
+    sql.push_bind(rollout_id.to_string());
+    sql.push(" AND rollout_ordinal >= ")
+        .push_bind(sqlite_integer(first_position, "first realtime position")?);
+    // Only new, enrolled calls write a canonical timestamp.
+    sql.push(" AND ")
+        .push(OBSERVED_AT)
+        .push(" IS NOT NULL AND item_type != 'transcript_segment'");
+    if !exact_id_uses_current_snapshot {
+        sql.push(" AND rollout_ordinal < ")
+            .push_bind(snapshot_end_sql);
+    }
+    if let Some(after) = params.after_position {
+        sql.push(" AND rollout_ordinal > ")
+            .push_bind(sqlite_integer(after, "after position")?);
+    }
+    if let Some(session) = &params.session_id {
+        sql.push(" AND json_extract(item_json, '$.realtime_session_id') = ")
+            .push_bind(session);
+    }
+    if let Some(event_id) = &params.event_id {
+        sql.push(" AND item_id = ").push_bind(event_id);
+    }
+    if let Some(from) = params.from_observed_at_ms {
+        sql.push(" AND ")
+            .push(OBSERVED_AT)
+            .push(" >= ")
+            .push_bind(from);
+    }
+    if let Some(to) = params.to_observed_at_ms {
+        sql.push(" AND ")
+            .push(OBSERVED_AT)
+            .push(" < ")
+            .push_bind(to);
+    }
+    if let Some(cursor) = &cursor {
+        match params.order {
+            ThreadRealtimeOrder::ObservedAt => {
+                sql.push(" AND (")
+                    .push(OBSERVED_AT)
+                    .push(", rollout_ordinal) > (")
+                    .push_bind(cursor.observed_at_ms)
+                    .push(", ")
+                    .push_bind(sqlite_integer(cursor.position, "cursor position")?)
+                    .push(")");
+            }
+            ThreadRealtimeOrder::Position => {
+                sql.push(" AND rollout_ordinal > ")
+                    .push_bind(sqlite_integer(cursor.position, "cursor position")?);
+            }
+        }
+    }
+    sql.push(" ORDER BY ");
+    if params.order == ThreadRealtimeOrder::ObservedAt {
+        sql.push(OBSERVED_AT).push(", ");
+    }
+    sql.push("rollout_ordinal LIMIT ")
+        .push_bind(i64::try_from(params.page_size + 1).map_err(thread_history_error)?);
+    // Classify an exact-ID miss against the same projection snapshot as the page query.
+    let mut transaction = pool.begin().await.map_err(thread_history_error)?;
+    let rows = sql
+        .build_query_as::<(i64, String)>()
+        .fetch_all(&mut *transaction)
+        .await
+        .map_err(thread_history_error)?;
+    // A second process can advance the projection after we capture its watermark.
+    // An acknowledged ID outside that implicit snapshot is still waiting for coverage.
+    if exact_id_uses_current_snapshot
+        && rows
+            .iter()
+            .any(|(position, _)| *position >= snapshot_end_sql)
+    {
+        return Ok(empty_page(
+            ThreadRealtimeCoverage::Pending,
+            snapshot_end,
+            projected_end,
+        ));
+    }
+    let has_more = rows.len() > params.page_size;
+    let data = rows
+        .into_iter()
+        .take(params.page_size)
+        .map(|(position, json)| {
+            Ok(ThreadRealtimeEntry {
+                position: u64::try_from(position).map_err(thread_history_error)?,
+                item: ThreadRealtimeItem::from_persisted(
+                    serde_json::from_str::<RealtimeItem>(&json).map_err(thread_history_error)?,
+                ),
+            })
+        })
+        .collect::<ThreadStoreResult<Vec<_>>>()?;
+    let next_cursor = if has_more {
+        data.last()
+            .map(|entry| {
+                serde_json::to_string(&Cursor {
+                    version: 1,
+                    rollout_id,
+                    query,
+                    snapshot_end,
+                    observed_at_ms: entry.item.observed_at_ms,
+                    position: entry.position,
+                })
+                .map_err(thread_history_error)
+            })
+            .transpose()?
+    } else {
+        None
+    };
+    let coverage = if let Some(event_id) = &params.event_id
+        && data.is_empty()
+    {
+        let exists: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM thread_realtime_items WHERE thread_id = ? AND item_id = ? AND rollout_ordinal >= ?)",
+        )
+        .bind(rollout_id.to_string())
+        .bind(event_id)
+        .bind(sqlite_integer(first_position, "first realtime position")?)
+        .fetch_one(&mut *transaction)
+        .await
+        .map_err(thread_history_error)?;
+        if exists {
+            ThreadRealtimeCoverage::Ready
+        } else {
+            ThreadRealtimeCoverage::Pending
+        }
+    } else {
+        ThreadRealtimeCoverage::Ready
+    };
+    transaction.commit().await.map_err(thread_history_error)?;
+    Ok(RealtimePage {
+        data,
+        next_cursor,
+        snapshot_end,
+        projected_through_item_index: projected_end,
+        coverage,
+    })
+}
+
+fn empty_page(
+    coverage: ThreadRealtimeCoverage,
+    snapshot_end: u64,
+    projected_through_item_index: u64,
+) -> RealtimePage {
+    RealtimePage {
+        data: Vec::new(),
+        next_cursor: None,
+        snapshot_end,
+        projected_through_item_index,
+        coverage,
+    }
+}
+
+fn invalid(message: &str) -> ThreadStoreError {
+    ThreadStoreError::InvalidRequest {
+        message: message.to_owned(),
+    }
+}
+
+#[cfg(test)]
+#[path = "realtime_stream_tests.rs"]
+mod tests;

codex-rs/thread-store/src/local/thread_history/realtime_stream_tests.rs

@@ -0,0 +1,250 @@
+#![cfg(test)]
+
+//! Verifies snapshot reads, late seals, and projection coverage against SQLite.
+
+use super::*;
+use codex_protocol::realtime::RealtimeItemContent;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+
+async fn database() -> (TempDir, sqlx::SqlitePool) {
+    let home = TempDir::new().unwrap();
+    let config = crate::local::test_support::test_config(home.path());
+    let pool = codex_state::open_thread_history_db(&config.sqlite)
+        .await
+        .unwrap();
+    (home, pool)
+}
+
+fn params(thread_id: ThreadId) -> ListRealtimeParams {
+    ListRealtimeParams {
+        thread_id,
+        session_id: None,
+        from_observed_at_ms: None,
+        to_observed_at_ms: None,
+        order: ThreadRealtimeOrder::ObservedAt,
+        after_position: None,
+        event_id: None,
+        cursor: None,
+        page_size: 1,
+        snapshot_end: None,
+        required_through_position: None,
+    }
+}
+
+async fn insert(pool: &sqlx::SqlitePool, thread_id: ThreadId, position: i64, time: Option<i64>) {
+    let item = RealtimeItem {
+        id: format!("event-{position}"),
+        realtime_session_id: "voice".to_owned(),
+        observed_at_ms: time,
+        content: RealtimeItemContent::RealtimeSessionStarted,
+    };
+    sqlx::query("INSERT INTO thread_realtime_items (thread_id, item_id, rollout_ordinal, created_at_ms, item_type, item_json) VALUES (?, ?, ?, 0, 'realtime_session_started', ?)")
+        .bind(thread_id.to_string()).bind(&item.id).bind(position)
+        .bind(serde_json::to_string(&item).unwrap()).execute(pool).await.unwrap();
+}
+
+#[tokio::test]
+async fn snapshot_stays_fixed_while_position_catchup_finds_a_late_seal() {
+    let (_home, pool) = database().await;
+    let thread = ThreadId::default();
+    insert(&pool, thread, /*position*/ 0, Some(200)).await;
+    insert(&pool, thread, /*position*/ 1, Some(200)).await;
+    insert(&pool, thread, /*position*/ 2, Some(100)).await;
+    let first = read_page(
+        &pool,
+        params(thread),
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 3,
+    )
+    .await
+    .unwrap();
+    assert_eq!(first.data[0].position, 2);
+    assert_eq!(first.snapshot_end, 3);
+
+    insert(&pool, thread, /*position*/ 3, Some(50)).await;
+    let next = read_page(
+        &pool,
+        ListRealtimeParams {
+            cursor: first.next_cursor.clone(),
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 4,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        next.data
+            .iter()
+            .map(|entry| entry.position)
+            .collect::<Vec<_>>(),
+        [0]
+    );
+    assert_eq!(next.snapshot_end, 3);
+    // Equal observation times on opposite sides of a page boundary use rollout position.
+    let last = read_page(
+        &pool,
+        ListRealtimeParams {
+            cursor: next.next_cursor,
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 4,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        last.data
+            .iter()
+            .map(|entry| entry.position)
+            .collect::<Vec<_>>(),
+        [1]
+    );
+    assert!(last.next_cursor.is_none());
+
+    let catchup = read_page(
+        &pool,
+        ListRealtimeParams {
+            order: ThreadRealtimeOrder::Position,
+            after_position: Some(first.snapshot_end - 1),
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 4,
+    )
+    .await
+    .unwrap();
+    assert_eq!(catchup.data[0].position, 3);
+
+    let error = read_page(
+        &pool,
+        ListRealtimeParams {
+            cursor: first.next_cursor,
+            session_id: Some("another-voice".into()),
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 4,
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(error, ThreadStoreError::InvalidRequest { .. }));
+}
+
+#[tokio::test]
+async fn projection_lag_is_explicit_and_descendants_hide_parent_positions() {
+    let (_home, pool) = database().await;
+    let thread = ThreadId::default();
+    insert(&pool, thread, /*position*/ 1, Some(100)).await;
+    let pending = read_page(
+        &pool,
+        ListRealtimeParams {
+            required_through_position: Some(2),
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 2,
+    )
+    .await
+    .unwrap();
+    assert_eq!(pending.coverage, ThreadRealtimeCoverage::Pending);
+    assert!(pending.data.is_empty());
+    let hidden = read_page(
+        &pool,
+        ListRealtimeParams {
+            order: ThreadRealtimeOrder::Position,
+            ..params(thread)
+        },
+        thread,
+        /*first_position*/ 2,
+        /*projected_end*/ 2,
+    )
+    .await
+    .unwrap();
+    assert!(hidden.data.is_empty());
+}
+
+#[tokio::test]
+async fn exact_event_id_waits_for_the_acknowledged_seal_to_be_projected() {
+    let (_home, pool) = database().await;
+    let thread = ThreadId::default();
+    let query = ListRealtimeParams {
+        event_id: Some("event-1".into()),
+        ..params(thread)
+    };
+    let pending = read_page(
+        &pool,
+        query.clone(),
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 1,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        pending,
+        empty_page(
+            ThreadRealtimeCoverage::Pending,
+            /*snapshot_end*/ 1,
+            /*projected_through_item_index*/ 1
+        )
+    );
+
+    insert(&pool, thread, /*position*/ 1, Some(100)).await;
+    let stale = read_page(
+        &pool,
+        query.clone(),
+        thread,
+        /*first_position*/ 0,
+        /*projected_end*/ 1,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        stale,
+        empty_page(
+            ThreadRealtimeCoverage::Pending,
+            /*snapshot_end*/ 1,
+            /*projected_through_item_index*/ 1
+        )
+    );
+    for excluded in [
+        ListRealtimeParams {
+            snapshot_end: Some(1),
+            ..query.clone()
+        },
+        ListRealtimeParams {
+            from_observed_at_ms: Some(200),
+            ..query.clone()
+        },
+    ] {
+        let page = read_page(
+            &pool, excluded, thread, /*first_position*/ 0, /*projected_end*/ 1,
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            page,
+            empty_page(
+                ThreadRealtimeCoverage::Ready,
+                /*snapshot_end*/ 1,
+                /*projected_through_item_index*/ 1
+            )
+        );
+    }
+    let ready = read_page(
+        &pool, query, thread, /*first_position*/ 0, /*projected_end*/ 2,
+    )
+    .await
+    .unwrap();
+    assert_eq!(ready.coverage, ThreadRealtimeCoverage::Ready);
+    assert_eq!(ready.data[0].position, 1);
+    assert_eq!(ready.snapshot_end, 2);
+}

codex-rs/thread-store/src/realtime.rs

@@ -0,0 +1,23 @@
+//! Shared query contract for the independent, immutable realtime history stream.
+
+use codex_app_server_protocol::ThreadRealtimeOrder;
+use codex_protocol::ThreadId;
+use serde::Deserialize;
+use serde::Serialize;
+
+pub use codex_app_server_protocol::ThreadRealtimeListResponse as RealtimePage;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ListRealtimeParams {
+    pub thread_id: ThreadId,
+    pub session_id: Option<String>,
+    pub from_observed_at_ms: Option<i64>,
+    pub to_observed_at_ms: Option<i64>,
+    pub order: ThreadRealtimeOrder,
+    pub after_position: Option<u64>,
+    pub event_id: Option<String>,
+    pub cursor: Option<String>,
+    pub page_size: usize,
+    pub snapshot_end: Option<u64>,
+    pub required_through_position: Option<u64>,
+}

codex-rs/thread-store/src/store.rs

@@ -599,6 +599,22 @@ pub trait ThreadStore: Any + Send + Sync {
         })
     }
 
+    /// Reads a bounded immutable snapshot of the independent realtime stream.
+    fn list_realtime(
+        &self,
+        _params: crate::ListRealtimeParams,
+    ) -> ThreadStoreFuture<'_, crate::RealtimePage> {
+        Box::pin(async {
+            Ok(crate::RealtimePage {
+                data: Vec::new(),
+                next_cursor: None,
+                snapshot_end: 0,
+                projected_through_item_index: 0,
+                coverage: codex_app_server_protocol::ThreadRealtimeCoverage::Unsupported,
+            })
+        })
+    }
+
     /// Lists bounded ordinary and realtime thread history in rollout order.
     fn list_timeline(
         &self,

sdk/python/src/openai_codex/generated/v2_all.py

@@ -6415,6 +6415,12 @@ class ThreadRealtimeClosedNotification(BaseModel):
     thread_id: Annotated[str, Field(alias="threadId")]
 
 
+class ThreadRealtimeCoverage(Enum):
+    unsupported = "unsupported"
+    pending = "pending"
+    ready = "ready"
+
+
 class ThreadRealtimeErrorNotification(BaseModel):
     model_config = ConfigDict(
         populate_by_name=True,
@@ -6538,6 +6544,11 @@ class ThreadRealtimeItemTranscriptDeltaNotification(BaseModel):
     thread_id: Annotated[str, Field(alias="threadId")]
 
 
+class ThreadRealtimeOrder(Enum):
+    observed_at = "observedAt"
+    position = "position"
+
+
 class ThreadRealtimeOutputAudioDeltaNotification(BaseModel):
     model_config = ConfigDict(
         populate_by_name=True,
@@ -12927,6 +12938,14 @@ class ThreadItemsListResponse(BaseModel):
     ] = None
 
 
+class ThreadRealtimeEntry(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+    )
+    item: ThreadRealtimeItem
+    position: Annotated[int, Field(ge=0)]
+
+
 class ItemThreadTimelineEntry(BaseModel):
     model_config = ConfigDict(
         populate_by_name=True,