mirror of
https://github.com/openai/codex.git
synced 2026-11-09 08:40:33 +00:00
Persist realtime events in the thread timeline (#40508)
## Why Clients need a bounded view of realtime conversations that preserves the order of speech, agent work, and turn lifecycle events without loading the full thread history. ## What changed - Persist realtime session boundaries, transcript segments, and promoted agent items in paginated thread rollouts. - Add the experimental `thread/timeline/list` API to page ordinary items, realtime items, and turn boundaries in canonical rollout order. Include `activeRealtimeSessionAtPageStart` so each page can be rendered independently. - Emit typed `thread/realtime/item/started`, `thread/realtime/item/transcript/delta`, and `thread/realtime/item/completed` notifications around durable realtime items. - Keep the existing `thread/items/list` response and legacy thread history behavior unchanged. ## Testing - Cover mixed timeline pagination, realtime-session state at page boundaries, rollout replay, subagent history prefixes, transcript streaming, steering, and promoted agent artifacts. GitOrigin-RevId: faa2c420192d63e060e8fb32446e3c797422656e
This commit is contained in:
guinness-oai authored and copyberry committed
1 parent
80e871309a
commit
41ca88e95c
80 files changed
+5140
-47
No files matched your search
@@ -110,6 +110,7 @@ mod thread_settings_update;
|
||||
mod thread_shell_command;
|
||||
mod thread_start;
|
||||
mod thread_status;
|
||||
mod thread_timeline;
|
||||
mod thread_unarchive;
|
||||
mod thread_unsubscribe;
|
||||
mod turn_interrupt;
|
||||
|
||||
@@ -11,6 +11,7 @@ use codex_app_server_protocol::ItemStartedNotification;
|
||||
use codex_app_server_protocol::JSONRPCError;
|
||||
use codex_app_server_protocol::LoginAccountResponse;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ThreadHistoryMode;
|
||||
use codex_app_server_protocol::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
|
||||
use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse;
|
||||
@@ -22,7 +23,12 @@ use codex_app_server_protocol::ThreadRealtimeAudioChunk;
|
||||
use codex_app_server_protocol::ThreadRealtimeClosedNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeErrorNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeInitialItem;
|
||||
use codex_app_server_protocol::ThreadRealtimeItem;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemAddedNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemCompletedNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemContent;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemStartedNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemTranscriptDeltaNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeListVoicesParams;
|
||||
use codex_app_server_protocol::ThreadRealtimeListVoicesResponse;
|
||||
use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification;
|
||||
@@ -37,10 +43,15 @@ use codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification;
|
||||
use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::ThreadTimelineEntry;
|
||||
use codex_app_server_protocol::ThreadTimelineListParams;
|
||||
use codex_app_server_protocol::ThreadTimelineListResponse;
|
||||
use codex_app_server_protocol::TurnCompletedNotification;
|
||||
use codex_app_server_protocol::TurnStartParams;
|
||||
use codex_app_server_protocol::TurnStartResponse;
|
||||
use codex_app_server_protocol::TurnStartedNotification;
|
||||
use codex_app_server_protocol::TurnSteerParams;
|
||||
use codex_app_server_protocol::TurnSteerResponse;
|
||||
use codex_app_server_protocol::UserInput as V2UserInput;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::protocol::CodexResponseHandoffMode;
|
||||
@@ -70,6 +81,7 @@ use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use test_case::test_case;
|
||||
use tokio::time::timeout;
|
||||
use uuid::Uuid;
|
||||
use wiremock::Match;
|
||||
use wiremock::Mock;
|
||||
use wiremock::MockServer;
|
||||
@@ -643,6 +655,119 @@ fn v2_background_agent_tool_call(call_id: &str, prompt: &str) -> Value {
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn realtime_conversation_streams_timeline_items() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
|
||||
let realtime_server = start_websocket_server(vec![vec![vec![
|
||||
session_updated("voice-1"),
|
||||
json!({ "type": "response.output_text.delta", "delta": "hello" }),
|
||||
json!({
|
||||
"type": "response.output_text.done",
|
||||
"text": "a substantially different final revision"
|
||||
}),
|
||||
]]])
|
||||
.await;
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml(
|
||||
codex_home.path(),
|
||||
&responses_server.uri(),
|
||||
realtime_server.uri(),
|
||||
/*realtime_enabled*/ true,
|
||||
StartupContextConfig::Generated,
|
||||
)?;
|
||||
|
||||
let mut mcp = TestAppServer::builder()
|
||||
.with_codex_home(codex_home.path())
|
||||
.build()
|
||||
.await?;
|
||||
mcp.initialize().await?;
|
||||
login_with_api_key(&mut mcp, "sk-test-key").await?;
|
||||
|
||||
let thread_request = mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
history_mode: Some(ThreadHistoryMode::Paginated),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread: ThreadStartResponse =
|
||||
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_request)).await??;
|
||||
let request = mcp
|
||||
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
|
||||
thread_id: thread.thread.id.clone(),
|
||||
client_managed_handoffs: None,
|
||||
delegation_ack_filler: None,
|
||||
flush_transcript_tail_on_session_end: None,
|
||||
codex_responses_as_items: None,
|
||||
codex_response_item_prefix: None,
|
||||
codex_response_handoff_mode: None,
|
||||
codex_response_handoff_channel_prefixes: None,
|
||||
model: None,
|
||||
output_modality: RealtimeOutputModality::Audio,
|
||||
include_startup_context: None,
|
||||
initial_items: None,
|
||||
realtime_start_instructions: None,
|
||||
realtime_end_instructions: None,
|
||||
prompt: None,
|
||||
realtime_session_id: None,
|
||||
transport: None,
|
||||
version: None,
|
||||
voice: None,
|
||||
})
|
||||
.await?;
|
||||
let _: ThreadRealtimeStartResponse =
|
||||
timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
|
||||
|
||||
let session_started = read_notification::<ThreadRealtimeItemStartedNotification>(
|
||||
&mut mcp,
|
||||
"thread/realtime/item/started",
|
||||
)
|
||||
.await?;
|
||||
let session_completed = read_notification::<ThreadRealtimeItemCompletedNotification>(
|
||||
&mut mcp,
|
||||
"thread/realtime/item/completed",
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(session_started.item, session_completed.item);
|
||||
assert_eq!(
|
||||
Uuid::parse_str(&session_started.item.id)?.get_version_num(),
|
||||
7
|
||||
);
|
||||
assert!(matches!(
|
||||
session_completed.item.content,
|
||||
ThreadRealtimeItemContent::RealtimeSessionStarted
|
||||
));
|
||||
|
||||
let started = read_notification::<ThreadRealtimeItemStartedNotification>(
|
||||
&mut mcp,
|
||||
"thread/realtime/item/started",
|
||||
)
|
||||
.await?;
|
||||
let delta = read_notification::<ThreadRealtimeItemTranscriptDeltaNotification>(
|
||||
&mut mcp,
|
||||
"thread/realtime/item/transcript/delta",
|
||||
)
|
||||
.await?;
|
||||
let completed = read_notification::<ThreadRealtimeItemCompletedNotification>(
|
||||
&mut mcp,
|
||||
"thread/realtime/item/completed",
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(Uuid::parse_str(&started.item.id)?.get_version_num(), 7);
|
||||
assert_ne!(started.item.id, session_started.item.id);
|
||||
assert_eq!(delta.item_id, started.item.id);
|
||||
assert_eq!(completed.item.id, started.item.id);
|
||||
assert_eq!(delta.delta, "hello");
|
||||
assert!(matches!(
|
||||
completed.item.content,
|
||||
ThreadRealtimeItemContent::TranscriptSegment { text, .. } if text == "hello"
|
||||
));
|
||||
|
||||
realtime_server.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
@@ -729,7 +854,10 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
|
||||
login_with_api_key(&mut mcp, "sk-test-key").await?;
|
||||
|
||||
let thread_start_request_id = mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams::default())
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
history_mode: Some(ThreadHistoryMode::Paginated),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread_start: ThreadStartResponse =
|
||||
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
|
||||
@@ -910,6 +1038,46 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
|
||||
assert_eq!(closed.thread_id, output_audio.thread_id);
|
||||
assert_eq!(closed.reason.as_deref(), Some("error"));
|
||||
|
||||
let history_request = mcp
|
||||
.send_thread_timeline_list_request(ThreadTimelineListParams {
|
||||
thread_id: thread_start.thread.id.clone(),
|
||||
cursor: None,
|
||||
limit: Some(100),
|
||||
})
|
||||
.await?;
|
||||
let history: ThreadTimelineListResponse =
|
||||
timeout(DEFAULT_TIMEOUT, mcp.read_response(history_request)).await??;
|
||||
assert!(matches!(
|
||||
history.data.first(),
|
||||
Some(ThreadTimelineEntry::Realtime {
|
||||
item: ThreadRealtimeItem {
|
||||
content: ThreadRealtimeItemContent::RealtimeSessionStarted,
|
||||
..
|
||||
},
|
||||
..
|
||||
})
|
||||
));
|
||||
assert!(history.data.iter().any(|entry| matches!(
|
||||
entry,
|
||||
ThreadTimelineEntry::Realtime {
|
||||
item: ThreadRealtimeItem {
|
||||
content: ThreadRealtimeItemContent::TranscriptSegment { .. },
|
||||
..
|
||||
},
|
||||
..
|
||||
}
|
||||
)));
|
||||
assert!(matches!(
|
||||
history.data.last(),
|
||||
Some(ThreadTimelineEntry::Realtime {
|
||||
item: ThreadRealtimeItem {
|
||||
content: ThreadRealtimeItemContent::RealtimeSessionClosed { .. },
|
||||
..
|
||||
},
|
||||
..
|
||||
})
|
||||
));
|
||||
|
||||
let connections = realtime_server.connections();
|
||||
assert_eq!(connections.len(), 1);
|
||||
let connection = &connections[0];
|
||||
@@ -984,6 +1152,201 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn realtime_timeline_splits_accepted_steering_and_persists_promoted_artifacts() -> Result<()>
|
||||
{
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let responses_server = responses::start_mock_server().await;
|
||||
let (gate_tx, gate_rx) = mpsc::channel();
|
||||
Mock::given(method("POST"))
|
||||
.and(path_regex(".*/responses$"))
|
||||
.respond_with(GatedSseResponse {
|
||||
gate_rx: Mutex::new(Some(gate_rx)),
|
||||
response: responses::sse(vec![
|
||||
responses::ev_response_created("response-1"),
|
||||
responses::ev_assistant_message(
|
||||
"promoted-message",
|
||||
"::codex-realtime-inline{}\nVisible artifact",
|
||||
),
|
||||
responses::ev_completed("response-1"),
|
||||
]),
|
||||
})
|
||||
.expect(2)
|
||||
.mount(&responses_server)
|
||||
.await;
|
||||
|
||||
let mut harness = RealtimeE2eHarness::new_with_main_loop_responses_server(
|
||||
RealtimeTestVersion::V2,
|
||||
responses_server,
|
||||
realtime_sideband(vec![open_realtime_sideband_connection(vec![
|
||||
vec![session_updated("voice-steering")],
|
||||
vec![json!({
|
||||
"type": "response.output_text.delta",
|
||||
"delta": "Spoken before steering"
|
||||
})],
|
||||
])]),
|
||||
)
|
||||
.await?;
|
||||
let thread_request = harness
|
||||
.mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
history_mode: Some(ThreadHistoryMode::Paginated),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread: ThreadStartResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(thread_request)).await??;
|
||||
harness.thread_id = thread.thread.id;
|
||||
let realtime_session_id = harness
|
||||
.start_websocket_realtime()
|
||||
.await?
|
||||
.realtime_session_id
|
||||
.context("realtime started notification should include a session ID")?;
|
||||
|
||||
let turn_request = harness
|
||||
.mcp
|
||||
.send_turn_start_request(TurnStartParams {
|
||||
thread_id: harness.thread_id.clone(),
|
||||
input: vec![V2UserInput::Text {
|
||||
text: "Start work".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let turn: TurnStartResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(turn_request)).await??;
|
||||
harness
|
||||
.read_notification::<TurnStartedNotification>("turn/started")
|
||||
.await?;
|
||||
|
||||
harness
|
||||
.append_text(harness.thread_id.clone(), "Trigger speech")
|
||||
.await?;
|
||||
harness
|
||||
.read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
|
||||
"thread/realtime/transcript/delta",
|
||||
)
|
||||
.await?;
|
||||
|
||||
let steering_request = harness
|
||||
.mcp
|
||||
.send_turn_steer_request(TurnSteerParams {
|
||||
thread_id: harness.thread_id.clone(),
|
||||
input: vec![V2UserInput::Text {
|
||||
text: "Accepted steering".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
expected_turn_id: turn.turn.id.clone(),
|
||||
additional_context: None,
|
||||
client_user_message_id: Some("accepted-steer".to_string()),
|
||||
responsesapi_client_metadata: None,
|
||||
})
|
||||
.await?;
|
||||
let accepted: TurnSteerResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(steering_request)).await??;
|
||||
assert_eq!(accepted.turn_id, turn.turn.id);
|
||||
|
||||
let _ = gate_tx.send(());
|
||||
harness
|
||||
.read_notification::<TurnCompletedNotification>("turn/completed")
|
||||
.await?;
|
||||
let page_request = harness
|
||||
.mcp
|
||||
.send_thread_timeline_list_request(ThreadTimelineListParams {
|
||||
thread_id: harness.thread_id.clone(),
|
||||
cursor: None,
|
||||
limit: Some(100),
|
||||
})
|
||||
.await?;
|
||||
let page: ThreadTimelineListResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(page_request)).await??;
|
||||
let transcript_index = page
|
||||
.data
|
||||
.iter()
|
||||
.position(|entry| {
|
||||
matches!(
|
||||
entry,
|
||||
ThreadTimelineEntry::Realtime {
|
||||
item: ThreadRealtimeItem {
|
||||
content: ThreadRealtimeItemContent::TranscriptSegment { text, .. },
|
||||
..
|
||||
},
|
||||
..
|
||||
} if text == "Spoken before steering"
|
||||
)
|
||||
})
|
||||
.context("accepted steering should seal the active transcript")?;
|
||||
let steering_index = page
|
||||
.data
|
||||
.iter()
|
||||
.position(|entry| {
|
||||
matches!(
|
||||
entry,
|
||||
ThreadTimelineEntry::Item {
|
||||
item,
|
||||
..
|
||||
} if matches!(
|
||||
item.as_ref(),
|
||||
ThreadItem::UserMessage { client_id, .. }
|
||||
if client_id.as_deref() == Some("accepted-steer")
|
||||
)
|
||||
)
|
||||
})
|
||||
.context("accepted steering should be included in the timeline")?;
|
||||
assert!(transcript_index < steering_index);
|
||||
assert!(page.data.iter().any(|entry| matches!(
|
||||
entry,
|
||||
ThreadTimelineEntry::Realtime {
|
||||
item: ThreadRealtimeItem {
|
||||
content: ThreadRealtimeItemContent::BemItemPromoted { item_id, .. },
|
||||
..
|
||||
},
|
||||
..
|
||||
} if item_id == "promoted-message"
|
||||
)));
|
||||
for entry in &page.data {
|
||||
if let ThreadTimelineEntry::Realtime { item, .. } = entry {
|
||||
assert_eq!(Uuid::parse_str(&item.id)?.get_version_num(), 7);
|
||||
}
|
||||
}
|
||||
|
||||
let bounded_request = harness
|
||||
.mcp
|
||||
.send_thread_timeline_list_request(ThreadTimelineListParams {
|
||||
thread_id: harness.thread_id.clone(),
|
||||
cursor: None,
|
||||
limit: Some(1),
|
||||
})
|
||||
.await?;
|
||||
let bounded: ThreadTimelineListResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(bounded_request)).await??;
|
||||
assert_eq!(bounded.data.len(), 1);
|
||||
assert_eq!(
|
||||
bounded.active_realtime_session_at_page_start.as_deref(),
|
||||
Some(realtime_session_id.as_str())
|
||||
);
|
||||
let older_request = harness
|
||||
.mcp
|
||||
.send_thread_timeline_list_request(ThreadTimelineListParams {
|
||||
thread_id: harness.thread_id.clone(),
|
||||
cursor: Some(bounded.next_cursor.context("bounded timeline cursor")?),
|
||||
limit: Some(1),
|
||||
})
|
||||
.await?;
|
||||
let older: ThreadTimelineListResponse =
|
||||
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(older_request)).await??;
|
||||
assert_eq!(older.data.len(), 1);
|
||||
assert_eq!(
|
||||
older.active_realtime_session_at_page_start.as_deref(),
|
||||
Some(realtime_session_id.as_str())
|
||||
);
|
||||
|
||||
harness.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn realtime_start_can_skip_startup_context() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
use anyhow::Result;
|
||||
use app_test_support::MockResponsesConfig;
|
||||
use app_test_support::TestAppServer;
|
||||
use app_test_support::create_mock_responses_server_repeating_assistant;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::SortDirection;
|
||||
use codex_app_server_protocol::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadItemsListParams;
|
||||
use codex_app_server_protocol::ThreadItemsListResponse;
|
||||
use codex_app_server_protocol::ThreadRealtimeItemContent;
|
||||
use codex_app_server_protocol::ThreadTimelineEntry;
|
||||
use codex_app_server_protocol::ThreadTimelineListParams;
|
||||
use codex_app_server_protocol::ThreadTimelineListResponse;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::items::TurnItem;
|
||||
use codex_protocol::items::UserMessageItem;
|
||||
use codex_protocol::models::BaseInstructions;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::ItemCompletedEvent;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::ThreadHistoryMode;
|
||||
use codex_protocol::protocol::ThreadMemoryMode;
|
||||
use codex_protocol::protocol::TurnCompleteEvent;
|
||||
use codex_protocol::protocol::TurnStartedEvent;
|
||||
use codex_protocol::realtime::RealtimeItem;
|
||||
use codex_protocol::realtime::RealtimeItemContent;
|
||||
use codex_protocol::realtime::RealtimeTranscriptRole;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use codex_rollout::RolloutItem;
|
||||
use codex_thread_store::AppendThreadItemsParams;
|
||||
use codex_thread_store::CreateThreadParams;
|
||||
use codex_thread_store::LocalThreadStore;
|
||||
use codex_thread_store::LocalThreadStoreConfig;
|
||||
use codex_thread_store::PersistContext;
|
||||
use codex_thread_store::ThreadPersistenceMetadata;
|
||||
use codex_thread_store::ThreadStore;
|
||||
use codex_utils_absolute_path::test_support::PathExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
use tempfile::TempDir;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[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;
|
||||
let codex_home = TempDir::new()?;
|
||||
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),
|
||||
);
|
||||
store
|
||||
.create_thread(CreateThreadParams {
|
||||
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: Uuid::now_v7().to_string(),
|
||||
metadata: ThreadPersistenceMetadata {
|
||||
cwd: Some(codex_home.path().to_path_buf()),
|
||||
model_provider: "mock_provider".to_string(),
|
||||
memory_mode: ThreadMemoryMode::Enabled,
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
store
|
||||
.persist_thread(thread_id, PersistContext::Standard)
|
||||
.await?;
|
||||
store
|
||||
.append_items(AppendThreadItemsParams {
|
||||
thread_id,
|
||||
items: vec![
|
||||
RolloutItem::RealtimeItem(RealtimeItem {
|
||||
id: "voice-1:started".to_string(),
|
||||
realtime_session_id: "voice-1".to_string(),
|
||||
content: RealtimeItemContent::RealtimeSessionStarted,
|
||||
}),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-1".to_string(),
|
||||
trace_id: None,
|
||||
started_at: Some(10),
|
||||
model_context_window: None,
|
||||
collaboration_mode_kind: Default::default(),
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
|
||||
thread_id,
|
||||
turn_id: "turn-1".to_string(),
|
||||
item: TurnItem::UserMessage(UserMessageItem {
|
||||
id: "user-1".to_string(),
|
||||
client_id: None,
|
||||
content: vec![UserInput::Text {
|
||||
text: "check staging".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
}),
|
||||
started_at_ms: Some(0),
|
||||
completed_at_ms: 1,
|
||||
})),
|
||||
RolloutItem::RealtimeItem(RealtimeItem {
|
||||
id: "voice-1:transcript-1".to_string(),
|
||||
realtime_session_id: "voice-1".to_string(),
|
||||
content: RealtimeItemContent::TranscriptSegment {
|
||||
role: RealtimeTranscriptRole::Assistant,
|
||||
text: "Checking staging".to_string(),
|
||||
},
|
||||
}),
|
||||
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
|
||||
turn_id: "turn-1".to_string(),
|
||||
last_agent_message: None,
|
||||
error: None,
|
||||
started_at: Some(10),
|
||||
completed_at: Some(12),
|
||||
duration_ms: Some(2000),
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
],
|
||||
})
|
||||
.await?;
|
||||
store.shutdown_thread(thread_id).await?;
|
||||
|
||||
let mut app_server = TestAppServer::builder()
|
||||
.with_codex_home(codex_home.path())
|
||||
.without_auto_env()
|
||||
.build_initialized()
|
||||
.await?;
|
||||
let page: ThreadTimelineListResponse = app_server
|
||||
.request(|request_id| ClientRequest::ThreadTimelineList {
|
||||
request_id,
|
||||
params: ThreadTimelineListParams {
|
||||
thread_id: thread_id.to_string(),
|
||||
cursor: None,
|
||||
limit: Some(3),
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(page.data.len(), 3);
|
||||
assert!(matches!(
|
||||
&page.data[0],
|
||||
ThreadTimelineEntry::Item { turn_id, item, .. }
|
||||
if turn_id == "turn-1"
|
||||
&& matches!(item.as_ref(), ThreadItem::UserMessage { .. })
|
||||
));
|
||||
assert!(matches!(
|
||||
&page.data[1],
|
||||
ThreadTimelineEntry::Realtime { item, .. }
|
||||
if matches!(
|
||||
&item.content,
|
||||
ThreadRealtimeItemContent::TranscriptSegment { text, .. }
|
||||
if text == "Checking staging"
|
||||
)
|
||||
));
|
||||
assert_eq!(
|
||||
page.active_realtime_session_at_page_start.as_deref(),
|
||||
Some("voice-1")
|
||||
);
|
||||
assert!(matches!(
|
||||
&page.data[2],
|
||||
ThreadTimelineEntry::TurnCompleted { turn_id, duration_ms: Some(2000), .. }
|
||||
if turn_id == "turn-1"
|
||||
));
|
||||
|
||||
let older: ThreadTimelineListResponse = app_server
|
||||
.request(|request_id| ClientRequest::ThreadTimelineList {
|
||||
request_id,
|
||||
params: ThreadTimelineListParams {
|
||||
thread_id: thread_id.to_string(),
|
||||
cursor: page.next_cursor,
|
||||
limit: Some(2),
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(older.data.len(), 2);
|
||||
assert!(matches!(
|
||||
&older.data[0],
|
||||
ThreadTimelineEntry::Realtime { item, .. }
|
||||
if matches!(
|
||||
item.content,
|
||||
ThreadRealtimeItemContent::RealtimeSessionStarted
|
||||
)
|
||||
));
|
||||
assert!(matches!(
|
||||
&older.data[1],
|
||||
ThreadTimelineEntry::TurnStarted { turn_id, started_at: Some(10), .. }
|
||||
if turn_id == "turn-1"
|
||||
));
|
||||
assert_eq!(older.active_realtime_session_at_page_start, None);
|
||||
assert_eq!(older.next_cursor, None);
|
||||
|
||||
let ordinary: ThreadItemsListResponse = app_server
|
||||
.request(|request_id| ClientRequest::ThreadItemsList {
|
||||
request_id,
|
||||
params: ThreadItemsListParams {
|
||||
thread_id: thread_id.to_string(),
|
||||
turn_id: None,
|
||||
cursor: None,
|
||||
limit: Some(10),
|
||||
sort_direction: Some(SortDirection::Asc),
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(ordinary.data.len(), 1);
|
||||
assert_eq!(ordinary.data[0].turn_id, "turn-1");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Reference in new issue
Block a user