diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 934de59603..a4a3231dc7 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/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": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index ecef4575ce..00d679c4c6 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/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.", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index e9e4e692f4..8970872b6e 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/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.", diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 7c3428d238..c4255e2947 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 0b1c8e6fcb..d53acc9c8e 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeCoverage.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeCoverage.ts new file mode 100644 index 0000000000..bf7b792b5a --- /dev/null +++ b/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"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeEntry.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeEntry.ts new file mode 100644 index 0000000000..731958bad4 --- /dev/null +++ b/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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOrder.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOrder.ts new file mode 100644 index 0000000000..147bed171d --- /dev/null +++ b/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"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index c85d38ce01..c9a7b4ea1f 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/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"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 358a8c17fd..266ecc98d9 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/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, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs b/codex-rs/app-server-protocol/src/protocol/v2/mod.rs index 77e2c34ffb..d9fd755335 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs +++ b/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::*; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/realtime_history.rs b/codex-rs/app-server-protocol/src/protocol/v2/realtime_history.rs new file mode 100644 index 0000000000..ad6a053388 --- /dev/null +++ b/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, + /// Inclusive canonical observation time lower bound. + #[ts(optional = nullable, type = "number | null")] + pub from_observed_at_ms: Option, + /// Exclusive canonical observation time upper bound. + #[ts(optional = nullable, type = "number | null")] + pub to_observed_at_ms: Option, + /// Defaults to canonical observation time. + #[ts(optional = nullable)] + pub order: Option, + #[ts(optional = nullable, type = "number | null")] + pub after_position: Option, + /// 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, + #[ts(optional = nullable)] + pub cursor: Option, + #[ts(optional = nullable)] + pub limit: Option, + /// Exclusive committed position delimiting an immutable snapshot. + #[ts(optional = nullable, type = "number | null")] + pub snapshot_end: Option, + /// Return pending until the projection includes this committed position. + #[ts(optional = nullable, type = "number | null")] + pub required_through_position: Option, +} + +#[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, + pub next_cursor: Option, + #[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, +} diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index fd89626e02..f04c821c05 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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 } diff --git a/codex-rs/app-server/src/request_processors/realtime_history.rs b/codex-rs/app-server/src/request_processors/realtime_history.rs new file mode 100644 index 0000000000..422d2de20e --- /dev/null +++ b/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, JSONRPCErrorError> { + let thread_id = ThreadId::from_string(¶ms.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())) + } +} diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 7b81d75021..5ce6bcfb15 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/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"] diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index 2f09a238e2..aa8a1cd99c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/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 { + 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::>(), + ["child-frame"], + ); + } + Ok(()) +} diff --git a/codex-rs/state/thread_history_migrations/0009_realtime_observation_indexes.sql b/codex-rs/state/thread_history_migrations/0009_realtime_observation_indexes.sql new file mode 100644 index 0000000000..882fe92322 --- /dev/null +++ b/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; diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 495e9b0cd5..1270d198ae 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 2e1f515ff7..1d3e99fe02 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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 { + 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, diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index aa8ff7a801..6923a7333c 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/thread_history/realtime_stream.rs b/codex-rs/thread-store/src/local/thread_history/realtime_stream.rs new file mode 100644 index 0000000000..76eaf76268 --- /dev/null +++ b/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, + position: u64, +} + +pub(in crate::local) async fn list_realtime( + store: &LocalThreadStore, + params: ListRealtimeParams, +) -> ThreadStoreResult { + 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 { + 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::) + .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::::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) = ¶ms.session_id { + sql.push(" AND json_extract(item_json, '$.realtime_session_id') = ") + .push_bind(session); + } + if let Some(event_id) = ¶ms.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::(&json).map_err(thread_history_error)?, + ), + }) + }) + .collect::>>()?; + 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) = ¶ms.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; diff --git a/codex-rs/thread-store/src/local/thread_history/realtime_stream_tests.rs b/codex-rs/thread-store/src/local/thread_history/realtime_stream_tests.rs new file mode 100644 index 0000000000..193713454b --- /dev/null +++ b/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) { + 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::>(), + [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::>(), + [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); +} diff --git a/codex-rs/thread-store/src/realtime.rs b/codex-rs/thread-store/src/realtime.rs new file mode 100644 index 0000000000..8ac8baf284 --- /dev/null +++ b/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, + pub from_observed_at_ms: Option, + pub to_observed_at_ms: Option, + pub order: ThreadRealtimeOrder, + pub after_position: Option, + pub event_id: Option, + pub cursor: Option, + pub page_size: usize, + pub snapshot_end: Option, + pub required_through_position: Option, +} diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index b1d0300901..2faa9d2332 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/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, diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 20219f7b98..5f999438c3 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/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,