github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-11 22:33:20 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
50fffd5ed367aa99491d9ec58575626fce4e9dd4
codex/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
T
Charlie Marsh 8436b749a4 Bound Git root discovery for metadata enrichment (#42132) 
## Why

Git root discovery is optional metadata work, but filesystem probes can block. They should not exhaust Tokio's blocking pool, delay runtime shutdown, or prevent later turns from observing repository changes.

## What changed

- Add a shared `GitRootDiscovery` service that coalesces concurrent lookups for the same working directory and limits probes across directories.
- Run probes on detached threads, retain in-flight work across caller cancellation, and discard completed results instead of caching them.
- Use the service for turn and memory metadata enrichment, abort unused turn enrichment when its state is dropped, and limit memory metadata waits to one second.

## Testing

Add coverage for probe sharing, capacity limits, cancellation, fresh discovery, runtime shutdown, memory timeouts, and repositories restored after startup prewarming.

GitOrigin-RevId: bca46fc263e7a12a2f69146d8a0b3e7c7e0846cb
2026-09-01 20:41:32 +00:00

4195 lines
147 KiB
Rust
Raw Blame History

use anyhow::Context;
use anyhow::Result;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use app_test_support::create_command_execution_sse_response;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::ItemCompletedNotification;
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;
use codex_app_server_protocol::ThreadRealtimeAppendSpeechParams;
use codex_app_server_protocol::ThreadRealtimeAppendSpeechResponse;
use codex_app_server_protocol::ThreadRealtimeAppendTextParams;
use codex_app_server_protocol::ThreadRealtimeAppendTextResponse;
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;
use codex_app_server_protocol::ThreadRealtimeSdpNotification;
use codex_app_server_protocol::ThreadRealtimeStartParams;
use codex_app_server_protocol::ThreadRealtimeStartResponse;
use codex_app_server_protocol::ThreadRealtimeStartTransport;
use codex_app_server_protocol::ThreadRealtimeStartedNotification;
use codex_app_server_protocol::ThreadRealtimeStopParams;
use codex_app_server_protocol::ThreadRealtimeStopResponse;
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;
use codex_protocol::protocol::ConversationTextRole;
use codex_protocol::protocol::RealtimeConversationVersion;
use codex_protocol::protocol::RealtimeOutputModality;
use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RealtimeVoicesList;
use core_test_support::responses;
use core_test_support::responses::WebSocketConnectionConfig;
use core_test_support::responses::WebSocketRequest;
use core_test_support::responses::WebSocketTestServer;
use core_test_support::responses::start_websocket_server;
use core_test_support::responses::start_websocket_server_with_headers;
use core_test_support::skip_if_no_network;
use core_test_support::skip_if_remote;
use pretty_assertions::assert_eq;
use serde::de::DeserializeOwned;
use serde_json::Value;
use serde_json::json;
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::mpsc;
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;
use wiremock::Request as WiremockRequest;
use wiremock::Respond;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::path_regex;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
const DELEGATED_SHELL_TURN_TIMEOUT: Duration = Duration::from_secs(30);
const DELEGATED_SHELL_TOOL_TIMEOUT_MS: u64 = 30_000;
const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.";
const V2_STEERING_ACKNOWLEDGEMENT: &str =
"This was sent to steer the previous background agent task.";
const V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str =
"Background agent finished. Use the preceding [BACKEND] messages as the result.";
const RESPONSE_ITEM_PREFIX: &str =
"Use the following context to inform future responses, but do not speak it to the user.";
#[derive(Debug, Clone, Copy)]
enum StartupContextConfig<'a> {
Generated,
Override(&'a str),
}
#[derive(Debug, Clone)]
struct RealtimeCallRequestCapture {
requests: Arc<Mutex<Vec<WiremockRequest>>>,
}
impl RealtimeCallRequestCapture {
fn new() -> Self {
Self {
requests: Arc::new(Mutex::new(Vec::new())),
}
}
fn single_request(&self) -> WiremockRequest {
let requests = self
.requests
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
assert_eq!(requests.len(), 1, "expected one realtime call request");
requests[0].clone()
}
}
impl Match for RealtimeCallRequestCapture {
fn matches(&self, request: &WiremockRequest) -> bool {
self.requests
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(request.clone());
true
}
}
fn normalized_json_string(raw: &str) -> Result<String> {
let value: Value = serde_json::from_str(raw).context("expected JSON fixture to parse")?;
serde_json::to_string(&value).context("expected JSON fixture to serialize")
}
struct GatedSseResponse {
gate_rx: Mutex<Option<mpsc::Receiver<()>>>,
response: String,
}
impl Respond for GatedSseResponse {
fn respond(&self, _: &WiremockRequest) -> ResponseTemplate {
let gate_rx = self
.gate_rx
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
if let Some(gate_rx) = gate_rx {
let _ = gate_rx.recv();
}
responses::sse_response(self.response.clone())
}
}
#[derive(Debug, Clone, Copy)]
enum RealtimeTestVersion {
V1,
V2,
}
impl RealtimeTestVersion {
fn config_value(self) -> &'static str {
match self {
RealtimeTestVersion::V1 => "v1",
RealtimeTestVersion::V2 => "v2",
}
}
}
#[derive(Debug, Clone, Copy)]
enum RealtimeTestSandbox {
ReadOnly,
DangerFullAccess,
}
impl RealtimeTestSandbox {
fn config_value(self) -> &'static str {
match self {
RealtimeTestSandbox::ReadOnly => "read-only",
RealtimeTestSandbox::DangerFullAccess => "danger-full-access",
}
}
}
#[derive(Debug, PartialEq)]
struct StartedWebrtcRealtime {
started: ThreadRealtimeStartedNotification,
sdp: ThreadRealtimeSdpNotification,
}
// Scripted SSE responses for the normal background agent loop. Realtime can ask for a delegated
// background agent turn; that turn talks to this mock `/responses` endpoint and may request
// ordinary tools.
struct MainLoopResponsesScript {
responses: Vec<String>,
}
// Scripted server events for the direct realtime sideband WebSocket. This mock is the realtime
// session app-server joins after call creation; it is not the background agent Responses stream.
struct RealtimeSidebandScript {
connections: Vec<WebSocketConnectionConfig>,
}
struct RealtimeE2eHarness {
mcp: TestAppServer,
_codex_home: TempDir,
main_loop_responses_server: MockServer,
realtime_server: WebSocketTestServer,
call_capture: RealtimeCallRequestCapture,
thread_id: String,
}
impl RealtimeE2eHarness {
// Owns the full mocked app-server realtime route: MCP client, Responses mocks, WebRTC call
// creation capture, sideband WebSocket server, login, config, and a started thread.
async fn new(
realtime_version: RealtimeTestVersion,
main_loop: MainLoopResponsesScript,
realtime_sideband: RealtimeSidebandScript,
) -> Result<Self> {
let main_loop_responses_server =
create_mock_responses_server_sequence_unchecked(main_loop.responses).await;
Self::new_with_main_loop_responses_server_and_sandbox(
realtime_version,
main_loop_responses_server,
realtime_sideband,
RealtimeTestSandbox::ReadOnly,
)
.await
}
async fn new_with_sandbox(
realtime_version: RealtimeTestVersion,
main_loop: MainLoopResponsesScript,
realtime_sideband: RealtimeSidebandScript,
sandbox: RealtimeTestSandbox,
) -> Result<Self> {
let main_loop_responses_server =
create_mock_responses_server_sequence_unchecked(main_loop.responses).await;
Self::new_with_main_loop_responses_server_and_sandbox(
realtime_version,
main_loop_responses_server,
realtime_sideband,
sandbox,
)
.await
}
async fn new_with_main_loop_responses_server(
realtime_version: RealtimeTestVersion,
main_loop_responses_server: MockServer,
realtime_sideband: RealtimeSidebandScript,
) -> Result<Self> {
Self::new_with_main_loop_responses_server_and_sandbox(
realtime_version,
main_loop_responses_server,
realtime_sideband,
RealtimeTestSandbox::ReadOnly,
)
.await
}
async fn new_with_main_loop_responses_server_and_sandbox(
realtime_version: RealtimeTestVersion,
main_loop_responses_server: MockServer,
realtime_sideband: RealtimeSidebandScript,
sandbox: RealtimeTestSandbox,
) -> Result<Self> {
let call_capture = RealtimeCallRequestCapture::new();
Mock::given(method("POST"))
.and(path("/v1/realtime/calls"))
.and(call_capture.clone())
.respond_with(
ResponseTemplate::new(200)
.insert_header("Location", "/v1/realtime/calls/rtc_e2e")
.set_body_string("v=answer\r\n"),
)
.mount(&main_loop_responses_server)
.await;
Mock::given(method("POST"))
.and(path("/v1/live"))
.and(call_capture.clone())
.respond_with(
ResponseTemplate::new(200)
.insert_header("Location", "/v1/live/rtc_e2e")
.set_body_string("v=answer\r\n"),
)
.mount(&main_loop_responses_server)
.await;
let realtime_server =
start_websocket_server_with_headers(realtime_sideband.connections).await;
let codex_home = TempDir::new()?;
create_config_toml_with_realtime_version(
codex_home.path(),
&main_loop_responses_server.uri(),
realtime_server.uri(),
/*realtime_enabled*/ true,
StartupContextConfig::Override("startup context"),
realtime_version,
sandbox,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
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())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
Ok(Self {
mcp,
_codex_home: codex_home,
main_loop_responses_server,
realtime_server,
call_capture,
thread_id: thread_start.thread.id,
})
}
async fn start_webrtc_realtime(&mut self, offer_sdp: &str) -> Result<StartedWebrtcRealtime> {
self.start_webrtc_realtime_with_codex_response_routing(
offer_sdp,
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
/*codex_response_handoff_mode*/ None,
/*delegation_ack_filler*/ None,
RealtimeConversationVersion::V1,
)
.await
}
async fn start_webrtc_realtime_with_codex_response_items(
&mut self,
offer_sdp: &str,
) -> Result<StartedWebrtcRealtime> {
self.start_webrtc_realtime_with_codex_response_routing(
offer_sdp,
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ Some(true),
/*codex_response_handoff_mode*/ None,
/*delegation_ack_filler*/ None,
RealtimeConversationVersion::V1,
)
.await
}
async fn start_webrtc_realtime_with_codex_response_routing(
&mut self,
offer_sdp: &str,
client_managed_handoffs: Option<bool>,
codex_responses_as_items: Option<bool>,
codex_response_handoff_mode: Option<CodexResponseHandoffMode>,
delegation_ack_filler: Option<bool>,
version: RealtimeConversationVersion,
) -> Result<StartedWebrtcRealtime> {
// Starts realtime through the public JSON-RPC method, then waits for the same client-visible
// notifications a desktop app needs: started first, SDP answer second.
let start_request_id = self
.mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
client_managed_handoffs,
delegation_ack_filler,
flush_transcript_tail_on_session_end: None,
thread_id: self.thread_id.clone(),
codex_response_item_prefix: codex_responses_as_items
.unwrap_or(false)
.then(|| RESPONSE_ITEM_PREFIX.to_string()),
codex_response_handoff_mode,
codex_response_handoff_channel_prefixes: None,
codex_responses_as_items,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: Some(ThreadRealtimeStartTransport::Webrtc {
sdp: offer_sdp.to_string(),
}),
version: Some(version),
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(start_request_id)).await??;
let started = self
.read_notification::<ThreadRealtimeStartedNotification>("thread/realtime/started")
.await?;
let sdp = self
.read_notification::<ThreadRealtimeSdpNotification>("thread/realtime/sdp")
.await?;
Ok(StartedWebrtcRealtime { started, sdp })
}
async fn start_websocket_realtime(&mut self) -> Result<ThreadRealtimeStartedNotification> {
self.start_websocket_realtime_with_codex_responses_as_items(
/*codex_responses_as_items*/ None,
)
.await
}
async fn start_websocket_realtime_with_codex_response_items(
&mut self,
) -> Result<ThreadRealtimeStartedNotification> {
self.start_websocket_realtime_with_codex_responses_as_items(
/*codex_responses_as_items*/ Some(true),
)
.await
}
async fn start_websocket_realtime_with_codex_responses_as_items(
&mut self,
codex_responses_as_items: Option<bool>,
) -> Result<ThreadRealtimeStartedNotification> {
let start_request_id = self
.mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
thread_id: self.thread_id.clone(),
client_managed_handoffs: None,
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: None,
codex_response_item_prefix: codex_responses_as_items
.unwrap_or(false)
.then(|| RESPONSE_ITEM_PREFIX.to_string()),
codex_response_handoff_mode: None,
codex_response_handoff_channel_prefixes: None,
codex_responses_as_items,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(start_request_id)).await??;
self.read_notification::<ThreadRealtimeStartedNotification>("thread/realtime/started")
.await
}
async fn start_frameless_bidi_realtime(
&mut self,
codex_response_handoff_mode: Option<CodexResponseHandoffMode>,
codex_response_handoff_channel_prefixes: Option<BTreeMap<String, Vec<String>>>,
initial_items: Option<Vec<ThreadRealtimeInitialItem>>,
) -> Result<ThreadRealtimeStartedNotification> {
let start_request_id = self
.mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
thread_id: self.thread_id.clone(),
client_managed_handoffs: None,
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: None,
codex_response_item_prefix: None,
codex_response_handoff_mode,
codex_response_handoff_channel_prefixes,
codex_responses_as_items: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: Some(RealtimeConversationVersion::V3),
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(start_request_id)).await??;
self.read_notification::<ThreadRealtimeStartedNotification>("thread/realtime/started")
.await
}
async fn read_notification<T: DeserializeOwned>(&mut self, method: &str) -> Result<T> {
read_notification(&mut self.mcp, method).await
}
async fn complete_turn(&mut self, text: &str) -> Result<()> {
let request_id = self
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: self.thread_id.clone(),
input: vec![V2UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(request_id)).await??;
self.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
Ok(())
}
/// Returns the nth JSON message app-server wrote to the fake Realtime API
/// sideband websocket.
async fn sideband_outbound_request(&self, request_index: usize) -> Value {
timeout(
DEFAULT_TIMEOUT,
self.realtime_server
.wait_for_request(/*connection_index*/ 0, request_index),
)
.await
.expect("realtime sideband request should arrive before timeout")
.body_json()
}
async fn append_audio(&mut self, thread_id: String) -> Result<()> {
let request_id = self
.mcp
.send_thread_realtime_append_audio_request(ThreadRealtimeAppendAudioParams {
thread_id,
audio: ThreadRealtimeAudioChunk {
data: "BQYH".to_string(),
sample_rate: 24_000,
num_channels: 1,
samples_per_channel: Some(480),
item_id: None,
},
})
.await?;
let _: ThreadRealtimeAppendAudioResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(request_id)).await??;
Ok(())
}
async fn append_text(&mut self, thread_id: String, text: &str) -> Result<()> {
let request_id = self
.mcp
.send_thread_realtime_append_text_request(ThreadRealtimeAppendTextParams {
thread_id,
text: text.to_string(),
role: ConversationTextRole::User,
})
.await?;
let _: ThreadRealtimeAppendTextResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(request_id)).await??;
Ok(())
}
async fn append_speech(&mut self, thread_id: String, text: &str) -> Result<()> {
let request_id = self
.mcp
.send_thread_realtime_append_speech_request(ThreadRealtimeAppendSpeechParams {
thread_id,
text: text.to_string(),
})
.await?;
let _: ThreadRealtimeAppendSpeechResponse =
timeout(DEFAULT_TIMEOUT, self.mcp.read_response(request_id)).await??;
Ok(())
}
async fn main_loop_responses_requests(&self) -> Result<Vec<Value>> {
responses_requests(&self.main_loop_responses_server).await
}
async fn shutdown(self) {
self.realtime_server.shutdown().await;
}
}
fn main_loop_responses(responses: Vec<String>) -> MainLoopResponsesScript {
MainLoopResponsesScript { responses }
}
fn no_main_loop_responses() -> MainLoopResponsesScript {
main_loop_responses(Vec::new())
}
fn realtime_sideband(connections: Vec<WebSocketConnectionConfig>) -> RealtimeSidebandScript {
RealtimeSidebandScript { connections }
}
fn realtime_sideband_connection(
realtime_server_events: Vec<Vec<Value>>,
) -> WebSocketConnectionConfig {
WebSocketConnectionConfig {
requests: realtime_server_events,
response_headers: Vec::new(),
accept_delay: None,
close_after_requests: true,
}
}
fn open_realtime_sideband_connection(
realtime_server_events: Vec<Vec<Value>>,
) -> WebSocketConnectionConfig {
WebSocketConnectionConfig {
close_after_requests: false,
..realtime_sideband_connection(realtime_server_events)
}
}
fn session_updated(realtime_session_id: &str) -> Value {
json!({
"type": "session.updated",
"session": { "id": realtime_session_id, "instructions": "backend prompt" }
})
}
fn session_started(realtime_session_id: &str) -> Value {
json!({
"type": "session.started",
"session": { "id": realtime_session_id, "instructions": "backend prompt" }
})
}
fn v2_background_agent_tool_call(call_id: &str, prompt: &str) -> Value {
json!({
"type": "conversation.item.done",
"item": {
"id": format!("item_{call_id}"),
"type": "function_call",
"name": "background_agent",
"call_id": call_id,
"arguments": json!({ "prompt": prompt }).to_string()
}
})
}
#[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"
));
let closed = read_notification::<ThreadRealtimeItemCompletedNotification>(
&mut mcp,
"thread/realtime/item/completed",
)
.await?;
let _: ThreadRealtimeClosedNotification =
read_notification(&mut mcp, "thread/realtime/closed").await?;
let request = mcp
.send_thread_timeline_list_request(ThreadTimelineListParams {
thread_id: thread.thread.id,
cursor: None,
limit: Some(100),
})
.await?;
let page: ThreadTimelineListResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(request)).await??;
let persisted = page
.data
.into_iter()
.filter_map(|entry| match entry {
ThreadTimelineEntry::Realtime { item, .. } => Some(item),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(
persisted,
vec![session_completed.item, completed.item, closed.item]
);
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
skip_if_no_network!(Ok(()));
let responses_server = create_mock_responses_server_sequence_unchecked(vec![
create_final_assistant_message_sse_response("delegated")?,
])
.await;
let realtime_server = start_websocket_server(vec![vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_backend", "instructions": "backend prompt" }
})],
vec![],
vec![],
vec![
json!({
"type": "response.output_audio.delta",
"delta": "AQID",
"sample_rate": 24_000,
"channels": 1,
"samples_per_channel": 512
}),
json!({
"type": "conversation.item.added",
"item": {
"type": "message",
"role": "assistant",
"content": [{ "type": "text", "text": "hi" }]
}
}),
json!({
"type": "conversation.item.input_audio_transcription.delta",
"delta": "delegate now"
}),
json!({
"type": "response.output_text.delta",
"delta": "working"
}),
json!({
"type": "response.output_text.done",
"text": "working on it"
}),
json!({
"type": "conversation.item.done",
"item": {
"id": "item_assistant_1",
"type": "message",
"role": "assistant",
"content": [{ "type": "output_text", "text": "working on it" }]
}
}),
json!({
"type": "conversation.item.done",
"item": {
"id": "item_2",
"type": "function_call",
"name": "background_agent",
"call_id": "handoff_1",
"arguments": "{\"input_transcript\":\"delegate now\"}"
}
}),
json!({
"type": "error",
"message": "upstream boom"
}),
],
]])
.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_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
login_with_api_key(&mut mcp, "sk-test-key").await?;
let thread_start_request_id = mcp
.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??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id.clone(),
model: Some("realtime-treatment-model".to_string()),
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: Some(RealtimeVoice::Cedar),
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(start_request_id)).await??;
let started =
read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
.await?;
assert_eq!(started.thread_id, thread_start.thread.id);
assert!(started.realtime_session_id.is_some());
assert_eq!(started.version, RealtimeConversationVersion::V2);
let startup_context_request = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0)
.await;
assert_eq!(
startup_context_request.body_json()["type"].as_str(),
Some("session.update")
);
assert_eq!(
startup_context_request.body_json()["session"]["audio"]["output"]["voice"],
"cedar"
);
assert_eq!(
realtime_server.single_handshake().uri(),
"/v1/realtime?model=realtime-treatment-model"
);
assert_eq!(
startup_context_request.body_json()["session"]["output_modalities"],
json!(["audio"])
);
let startup_context_instructions =
startup_context_request.body_json()["session"]["instructions"]
.as_str()
.context("expected startup context instructions")?
.to_string();
assert!(startup_context_instructions.starts_with("backend prompt"));
assert!(startup_context_instructions.contains(STARTUP_CONTEXT_HEADER));
let audio_append_request_id = mcp
.send_thread_realtime_append_audio_request(ThreadRealtimeAppendAudioParams {
thread_id: started.thread_id.clone(),
audio: ThreadRealtimeAudioChunk {
data: "BQYH".to_string(),
sample_rate: 24_000,
num_channels: 1,
samples_per_channel: Some(480),
item_id: None,
},
})
.await?;
let _: ThreadRealtimeAppendAudioResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(audio_append_request_id)).await??;
let text_append_request_id = mcp
.send_thread_realtime_append_text_request(ThreadRealtimeAppendTextParams {
thread_id: started.thread_id.clone(),
text: "hello".to_string(),
role: ConversationTextRole::Developer,
})
.await?;
let _: ThreadRealtimeAppendTextResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(text_append_request_id)).await??;
let assistant_append_request_id = mcp
.send_thread_realtime_append_text_request(ThreadRealtimeAppendTextParams {
thread_id: started.thread_id.clone(),
text: "welcome back".to_string(),
role: ConversationTextRole::Assistant,
})
.await?;
let _: ThreadRealtimeAppendTextResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_response(assistant_append_request_id),
)
.await??;
let output_audio = read_notification::<ThreadRealtimeOutputAudioDeltaNotification>(
&mut mcp,
"thread/realtime/outputAudio/delta",
)
.await?;
assert_eq!(output_audio.audio.data, "AQID");
assert_eq!(output_audio.audio.sample_rate, 24_000);
assert_eq!(output_audio.audio.num_channels, 1);
assert_eq!(output_audio.audio.samples_per_channel, Some(512));
let item_added = read_notification::<ThreadRealtimeItemAddedNotification>(
&mut mcp,
"thread/realtime/itemAdded",
)
.await?;
assert_eq!(item_added.thread_id, output_audio.thread_id);
assert_eq!(item_added.item["type"], json!("message"));
let first_transcript_delta = read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
&mut mcp,
"thread/realtime/transcript/delta",
)
.await?;
assert_eq!(first_transcript_delta.thread_id, output_audio.thread_id);
assert_eq!(first_transcript_delta.role, "user");
assert_eq!(first_transcript_delta.delta, "delegate now");
let second_transcript_delta = read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
&mut mcp,
"thread/realtime/transcript/delta",
)
.await?;
assert_eq!(second_transcript_delta.thread_id, output_audio.thread_id);
assert_eq!(second_transcript_delta.role, "assistant");
assert_eq!(second_transcript_delta.delta, "working");
let final_transcript_done = read_notification::<ThreadRealtimeTranscriptDoneNotification>(
&mut mcp,
"thread/realtime/transcript/done",
)
.await?;
assert_eq!(final_transcript_done.thread_id, output_audio.thread_id);
assert_eq!(final_transcript_done.role, "assistant");
assert_eq!(final_transcript_done.text, "working on it");
let handoff_item_added = read_notification::<ThreadRealtimeItemAddedNotification>(
&mut mcp,
"thread/realtime/itemAdded",
)
.await?;
assert_eq!(handoff_item_added.thread_id, output_audio.thread_id);
assert_eq!(handoff_item_added.item["type"], json!("handoff_request"));
assert_eq!(handoff_item_added.item["handoff_id"], json!("handoff_1"));
assert_eq!(handoff_item_added.item["item_id"], json!("item_2"));
assert_eq!(
handoff_item_added.item["input_transcript"],
json!("delegate now")
);
assert_eq!(
handoff_item_added.item["active_transcript"],
json!([
{"role": "user", "text": "delegate now"},
{"role": "assistant", "text": "working on it"}
])
);
let realtime_error =
read_notification::<ThreadRealtimeErrorNotification>(&mut mcp, "thread/realtime/error")
.await?;
assert_eq!(realtime_error.thread_id, output_audio.thread_id);
assert_eq!(realtime_error.message, "upstream boom");
let closed =
read_notification::<ThreadRealtimeClosedNotification>(&mut mcp, "thread/realtime/closed")
.await?;
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 { .. },
..
},
..
}
)));
// A background handoff can append timeline entries after realtime closes.
assert!(history.data.iter().any(|entry| matches!(
entry,
ThreadTimelineEntry::Realtime {
item: ThreadRealtimeItem {
content: ThreadRealtimeItemContent::RealtimeSessionClosed { .. },
..
},
..
}
)));
let connections = realtime_server.connections();
assert_eq!(connections.len(), 1);
let connection = &connections[0];
assert_eq!(connection.len(), 4);
assert_eq!(
connection[0].body_json()["type"].as_str(),
Some("session.update")
);
assert_eq!(
connection[0].body_json()["session"]["instructions"].as_str(),
Some(startup_context_instructions.as_str()),
);
let text_requests = connection
.iter()
.map(WebSocketRequest::body_json)
.filter(|request| request["type"] == "conversation.item.create")
.collect::<Vec<_>>();
assert_eq!(text_requests.len(), 2);
assert_eq!(
text_requests[0],
json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "developer",
"content": [{
"type": "input_text",
"text": "hello",
}],
},
})
);
assert_eq!(
text_requests[1],
json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "assistant",
"content": [{
"type": "output_text",
"text": "welcome back",
}],
},
})
);
let mut request_types = [
connection[1].body_json()["type"]
.as_str()
.context("expected websocket request type")?
.to_string(),
connection[2].body_json()["type"]
.as_str()
.context("expected websocket request type")?
.to_string(),
connection[3].body_json()["type"]
.as_str()
.context("expected websocket request type")?
.to_string(),
];
request_types.sort();
assert_eq!(
request_types,
[
"conversation.item.create".to_string(),
"conversation.item.create".to_string(),
"input_audio_buffer.append".to_string(),
]
);
realtime_server.shutdown().await;
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"
})],
vec![json!({
"type": "response.output_text.delta",
"delta": " and after rejected 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?;
for (input, expected_turn_id) in [
(Vec::new(), turn.turn.id.clone()),
(
vec![V2UserInput::Text {
text: "Rejected steering".to_string(),
text_elements: Vec::new(),
}],
"stale-turn".to_string(),
),
] {
let request = harness
.mcp
.send_turn_steer_request(TurnSteerParams {
thread_id: harness.thread_id.clone(),
input,
expected_turn_id,
additional_context: None,
client_user_message_id: None,
responsesapi_client_metadata: None,
})
.await?;
let rejected = timeout(
DEFAULT_TIMEOUT,
harness
.mcp
.read_stream_until_error_message(RequestId::Integer(request)),
)
.await??;
assert_eq!(rejected.error.code, -32600);
}
harness
.append_text(harness.thread_id.clone(), "Continue speech after rejection")
.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 and after rejected 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);
// Inline artifacts are promoted while streaming, before their final item.
assert_eq!(
page.data
.iter()
.filter_map(|entry| match entry {
ThreadTimelineEntry::Item { item, .. }
if matches!(item.as_ref(), ThreadItem::AgentMessage { id, .. } if id == "promoted-message") => Some("artifact"),
ThreadTimelineEntry::Realtime {
item: ThreadRealtimeItem {
content: ThreadRealtimeItemContent::BemItemPromoted { item_id, .. },
..
},
..
} if item_id == "promoted-message" => Some("promotion"),
_ => None,
})
.collect::<Vec<_>>(),
vec!["promotion", "artifact"]
);
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(()));
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let realtime_server = start_websocket_server(vec![vec![vec![json!({
"type": "session.updated",
"session": { "id": "sess_backend", "instructions": "backend prompt" }
})]]])
.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_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
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())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: Some(false),
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(start_request_id)).await??;
read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
.await?;
let startup_context_request = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0)
.await;
let startup_context_body = startup_context_request.body_json();
let instructions = startup_context_body["session"]["instructions"]
.as_str()
.context("expected realtime instructions")?;
assert_eq!(instructions, "backend prompt");
assert!(!instructions.contains(STARTUP_CONTEXT_HEADER));
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_text_output_modality_requests_text_output_and_final_transcript() -> 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![
json!({
"type": "session.updated",
"session": { "id": "sess_text", "instructions": "backend prompt" }
}),
json!({
"type": "response.output_text.delta",
"delta": "hello "
}),
json!({
"type": "response.output_text.delta",
"delta": "world"
}),
json!({
"type": "response.output_audio_transcript.done",
"transcript": "hello world"
}),
json!({
"type": "conversation.item.done",
"item": {
"id": "item_output_1",
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "hello world"}]
}
}),
]]])
.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_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
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())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Text,
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(start_request_id)).await??;
let session_update = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0)
.await;
assert_eq!(
session_update.body_json()["session"]["output_modalities"],
json!(["text"])
);
let first_delta = read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
&mut mcp,
"thread/realtime/transcript/delta",
)
.await?;
let second_delta = read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
&mut mcp,
"thread/realtime/transcript/delta",
)
.await?;
let done = read_notification::<ThreadRealtimeTranscriptDoneNotification>(
&mut mcp,
"thread/realtime/transcript/done",
)
.await?;
assert_eq!(
vec![first_delta, second_delta],
vec![
ThreadRealtimeTranscriptDeltaNotification {
thread_id: thread_start.thread.id.clone(),
role: "assistant".to_string(),
delta: "hello ".to_string(),
},
ThreadRealtimeTranscriptDeltaNotification {
thread_id: thread_start.thread.id.clone(),
role: "assistant".to_string(),
delta: "world".to_string(),
},
]
);
assert_eq!(
done,
ThreadRealtimeTranscriptDoneNotification {
thread_id: thread_start.thread.id,
role: "assistant".to_string(),
text: "hello world".to_string(),
}
);
assert!(
timeout(
Duration::from_millis(200),
mcp.read_stream_until_notification_message("thread/realtime/transcript/done"),
)
.await
.is_err(),
"should not emit duplicate transcript done from audio transcript done"
);
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_list_voices_returns_supported_names() -> Result<()> {
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
"http://127.0.0.1:1",
"ws://127.0.0.1:1",
/*realtime_enabled*/ true,
StartupContextConfig::Generated,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
let request_id = mcp
.send_thread_realtime_list_voices_request(ThreadRealtimeListVoicesParams {})
.await?;
let response: ThreadRealtimeListVoicesResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
assert_eq!(
response,
ThreadRealtimeListVoicesResponse {
voices: RealtimeVoicesList {
v1: vec![
RealtimeVoice::Juniper,
RealtimeVoice::Maple,
RealtimeVoice::Spruce,
RealtimeVoice::Ember,
RealtimeVoice::Vale,
RealtimeVoice::Breeze,
RealtimeVoice::Arbor,
RealtimeVoice::Sol,
RealtimeVoice::Cove,
],
v2: vec![
RealtimeVoice::Alloy,
RealtimeVoice::Ash,
RealtimeVoice::Ballad,
RealtimeVoice::Coral,
RealtimeVoice::Echo,
RealtimeVoice::Sage,
RealtimeVoice::Shimmer,
RealtimeVoice::Verse,
RealtimeVoice::Marin,
RealtimeVoice::Cedar,
],
default_v1: RealtimeVoice::Cove,
default_v2: RealtimeVoice::Marin,
},
}
);
Ok(())
}
#[tokio::test]
async fn realtime_conversation_stop_emits_closed_notification() -> 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![json!({
"type": "session.updated",
"session": { "id": "sess_backend", "instructions": "backend prompt" }
})],
vec![],
]])
.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_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
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())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(start_request_id)).await??;
let started =
read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
.await?;
let stop_request_id = mcp
.send_thread_realtime_stop_request(ThreadRealtimeStopParams {
thread_id: started.thread_id.clone(),
})
.await?;
let _: ThreadRealtimeStopResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(stop_request_id)).await??;
let closed =
read_notification::<ThreadRealtimeClosedNotification>(&mut mcp, "thread/realtime/closed")
.await?;
assert_eq!(closed.thread_id, started.thread_id);
assert!(matches!(
closed.reason.as_deref(),
Some("requested" | "transport_closed")
));
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_mode_uses_client_instructions_on_entry_and_exit() -> Result<()> {
skip_if_no_network!(Ok(()));
let start_instructions = "Use [analysis], [final], and ::realtime-inline for voice output.";
let end_instructions = "Voice has ended. Resume the normal text output protocol.";
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![
create_final_assistant_message_sse_response("first voice response")?,
create_final_assistant_message_sse_response("second voice response")?,
create_final_assistant_message_sse_response("text response after voice")?,
]),
realtime_sideband(vec![open_realtime_sideband_connection(vec![
vec![session_updated("sess_client_controlled_mode")],
vec![],
vec![],
])]),
)
.await?;
let start_request_id = harness
.mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
thread_id: harness.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: Some(start_instructions.to_string()),
realtime_end_instructions: Some(end_instructions.to_string()),
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(start_request_id)).await??;
harness
.read_notification::<ThreadRealtimeStartedNotification>("thread/realtime/started")
.await?;
for input in ["first voice turn", "second voice turn"] {
harness.complete_turn(input).await?;
}
let stop_request_id = harness
.mcp
.send_thread_realtime_stop_request(ThreadRealtimeStopParams {
thread_id: harness.thread_id.clone(),
})
.await?;
let _: ThreadRealtimeStopResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(stop_request_id)).await??;
harness
.read_notification::<ThreadRealtimeClosedNotification>("thread/realtime/closed")
.await?;
harness.complete_turn("continue in text").await?;
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 3);
assert!(response_request_contains_text(
&requests[0],
start_instructions
));
assert!(!response_request_contains_text(
&requests[0],
end_instructions
));
assert!(!response_request_contains_text(
&requests[1],
end_instructions
));
assert!(response_request_contains_text(
&requests[2],
end_instructions
));
assert!(response_request_contains_text(
&requests[2],
&format!("<realtime_conversation>\n{end_instructions}\n</realtime_conversation>"),
));
let start_message_count = requests[1]["input"]
.as_array()
.context("second voice Responses request should contain input")?
.iter()
.filter(|item| {
item["role"] == "developer" && response_request_contains_text(item, start_instructions)
})
.count();
assert!(
start_message_count <= 1,
"realtime entry instructions should not be injected again on subsequent turns"
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
skip_if_no_network!(Ok(()));
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let call_capture = RealtimeCallRequestCapture::new();
Mock::given(method("POST"))
.and(path("/v1/realtime/calls"))
.and(call_capture.clone())
.respond_with(
ResponseTemplate::new(200)
.insert_header("Location", "/v1/realtime/calls/rtc_app_test")
.set_body_string("v=answer\r\n"),
)
.mount(&responses_server)
.await;
let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
requests: vec![vec![json!({
"type": "session.updated",
"session": { "id": "sess_webrtc", "instructions": "backend prompt" }
})]],
response_headers: Vec::new(),
accept_delay: None,
close_after_requests: false,
}])
.await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
/*realtime_enabled*/ true,
StartupContextConfig::Override("startup context"),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
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())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let thread_id = thread_start.thread.id;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: Some(ThreadRealtimeStartTransport::Webrtc {
sdp: "v=offer\r\n".to_string(),
}),
version: Some(RealtimeConversationVersion::V1),
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(start_request_id)).await??;
let started =
read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
.await?;
assert_eq!(started.thread_id, thread_id);
assert_eq!(started.version, RealtimeConversationVersion::V1);
let sdp_notification =
read_notification::<ThreadRealtimeSdpNotification>(&mut mcp, "thread/realtime/sdp").await?;
assert_eq!(
sdp_notification,
ThreadRealtimeSdpNotification {
thread_id: thread_id.clone(),
sdp: "v=answer\r\n".to_string()
}
);
let session_update = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0)
.await;
assert_eq!(
session_update.body_json()["type"].as_str(),
Some("session.update")
);
assert!(
session_update.body_json()["session"]["instructions"]
.as_str()
.context("expected session.update instructions")?
.contains("startup context")
);
assert_eq!(
realtime_server.single_handshake().uri(),
"/v1/realtime?intent=quicksilver&call_id=rtc_app_test"
);
let stop_request_id = mcp
.send_thread_realtime_stop_request(ThreadRealtimeStopParams {
thread_id: thread_id.clone(),
})
.await?;
let _: ThreadRealtimeStopResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(stop_request_id)).await??;
let closed_notification =
read_notification::<ThreadRealtimeClosedNotification>(&mut mcp, "thread/realtime/closed")
.await?;
assert_eq!(closed_notification.thread_id, thread_id);
assert!(
matches!(
closed_notification.reason.as_deref(),
Some("requested" | "transport_closed")
),
"unexpected close reason: {closed_notification:?}"
);
let request = call_capture.single_request();
assert_eq!(request.url.path(), "/v1/realtime/calls");
assert_eq!(
request.url.query(),
Some("intent=quicksilver&architecture=avas")
);
assert_eq!(
request
.headers
.get("content-type")
.and_then(|value| value.to_str().ok()),
Some("multipart/form-data; boundary=codex-realtime-call-boundary")
);
let body = String::from_utf8(request.body).context("multipart body should be utf-8")?;
let session = normalized_json_string(v1_session_create_json())?;
assert_eq!(
body,
format!(
"--codex-realtime-call-boundary\r\n\
Content-Disposition: form-data; name=\"sdp\"\r\n\
Content-Type: application/sdp\r\n\
\r\n\
v=offer\r\n\
\r\n\
--codex-realtime-call-boundary\r\n\
Content-Disposition: form-data; name=\"session\"\r\n\
Content-Type: application/json\r\n\
\r\n\
{session}\r\n\
--codex-realtime-call-boundary--\r\n"
)
);
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: build a v1 realtime thread with a mocked call-create response and a sideband socket
// that immediately proves the joined connection can receive server events.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
no_main_loop_responses(),
realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![
session_updated("sess_v1_webrtc"),
]])]),
)
.await?;
// Phase 2: start through app-server and assert the app receives both the started notification
// and the answer SDP.
let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
assert_eq!(
started,
StartedWebrtcRealtime {
started: ThreadRealtimeStartedNotification {
thread_id: harness.thread_id.clone(),
realtime_session_id: Some(harness.thread_id.clone()),
version: RealtimeConversationVersion::V1,
},
sdp: ThreadRealtimeSdpNotification {
thread_id: harness.thread_id.clone(),
sdp: "v=answer\r\n".to_string(),
},
}
);
// Phase 3: verify the HTTP call-create leg, the direct sideband join, and the normal v1
// session.update; the WebRTC transport should remain alive instead of closing after SDP.
assert_call_create_multipart(
harness.call_capture.single_request(),
"v=offer\r\n",
v1_session_create_json(),
"/v1/realtime/calls?intent=quicksilver&architecture=avas",
)?;
let session_update = harness.sideband_outbound_request(/*request_index*/ 0).await;
assert_v1_session_update(&session_update)?;
assert_eq!(
harness.realtime_server.single_handshake().uri(),
"/v1/realtime?intent=quicksilver&call_id=rtc_e2e"
);
let closed = timeout(
Duration::from_millis(100),
harness
.mcp
.read_stream_until_notification_message("thread/realtime/closed"),
)
.await;
assert!(closed.is_err(), "WebRTC start should not close immediately");
harness.shutdown().await;
Ok(())
}
#[test_case(
None,
None,
RealtimeConversationVersion::V1,
"/v1/realtime?intent=quicksilver&call_id=rtc_existing";
"defaults to v1"
)]
#[test_case(
Some(RealtimeConversationVersion::V3),
Some("sess_client_owned"),
RealtimeConversationVersion::V3,
"/v1/live/rtc_existing";
"supports v3"
)]
#[test_case(
Some(RealtimeConversationVersion::V2),
None,
RealtimeConversationVersion::V2,
"";
"rejects v2"
)]
#[tokio::test]
async fn existing_call_attaches_without_reinitializing_the_client_session(
version: Option<RealtimeConversationVersion>,
realtime_session_id: Option<&str>,
expected_version: RealtimeConversationVersion,
expected_handshake_uri: &str,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
no_main_loop_responses(),
realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![]])]),
)
.await?;
let request_id = harness
.mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
thread_id: harness.thread_id.clone(),
client_managed_handoffs: None,
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: None,
codex_response_item_prefix: None,
codex_response_handoff_mode: None,
codex_response_handoff_channel_prefixes: None,
codex_responses_as_items: 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: realtime_session_id.map(str::to_string),
transport: Some(ThreadRealtimeStartTransport::ExistingCall {
call_id: "rtc_existing".to_string(),
}),
version,
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(request_id)).await??;
if expected_version == RealtimeConversationVersion::V2 {
let error = harness
.read_notification::<ThreadRealtimeErrorNotification>("thread/realtime/error")
.await?;
assert_eq!(
error.message,
"AVAS realtime calls require realtime v1 or v3"
);
assert!(harness.realtime_server.handshakes().is_empty());
harness.shutdown().await;
return Ok(());
}
let started = harness
.read_notification::<ThreadRealtimeStartedNotification>("thread/realtime/started")
.await?;
assert_eq!(
started,
ThreadRealtimeStartedNotification {
thread_id: harness.thread_id.clone(),
realtime_session_id: realtime_session_id.map(str::to_string),
version: expected_version,
}
);
assert_eq!(
harness.realtime_server.single_handshake().uri(),
expected_handshake_uri
);
assert!(
harness.realtime_server.single_connection().is_empty(),
"attaching to an existing call must not overwrite the client session"
);
assert!(
harness
.call_capture
.requests
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.is_empty(),
"an existing call must not issue a second call-create request"
);
harness.shutdown().await;
Ok(())
}
#[test_case("includeStartupContext"; "rejects startup context")]
#[test_case("prompt"; "rejects prompt")]
#[test_case("initialItems"; "rejects initial items")]
#[test_case("model"; "rejects model")]
#[test_case("voice"; "rejects voice")]
#[test_case("delegationAckFiller"; "rejects delegation acknowledgement filler")]
#[tokio::test]
async fn existing_call_rejects_client_owned_session_configuration(option: &str) -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
no_main_loop_responses(),
realtime_sideband(Vec::new()),
)
.await?;
let mut params = ThreadRealtimeStartParams {
thread_id: harness.thread_id.clone(),
client_managed_handoffs: None,
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: None,
codex_response_item_prefix: None,
codex_response_handoff_mode: None,
codex_response_handoff_channel_prefixes: None,
codex_responses_as_items: 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: Some(ThreadRealtimeStartTransport::ExistingCall {
call_id: "rtc_existing".to_string(),
}),
version: Some(RealtimeConversationVersion::V3),
voice: None,
};
match option {
"includeStartupContext" => params.include_startup_context = Some(true),
"prompt" => params.prompt = Some(Some("backend prompt".to_string())),
"initialItems" => {
params.initial_items = Some(vec![ThreadRealtimeInitialItem {
role: ConversationTextRole::User,
text: "client-owned history".to_string(),
}]);
}
"model" => params.model = Some("another-model".to_string()),
"voice" => params.voice = Some(RealtimeVoice::Cove),
"delegationAckFiller" => params.delegation_ack_filler = Some(true),
option => anyhow::bail!("unsupported test option: {option}"),
}
let request_id = harness
.mcp
.send_thread_realtime_start_request(params)
.await?;
let error = timeout(
DEFAULT_TIMEOUT,
harness
.mcp
.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert_invalid_request(
error,
format!("existingCall transport does not support {option}"),
);
assert!(harness.realtime_server.handshakes().is_empty());
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v3_start_posts_live_session_and_joins_without_session_update() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
no_main_loop_responses(),
realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![]])]),
)
.await?;
let started = harness
.start_webrtc_realtime_with_codex_response_routing(
"v=offer\r\n",
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
/*codex_response_handoff_mode*/ None,
/*delegation_ack_filler*/ Some(false),
RealtimeConversationVersion::V3,
)
.await?;
assert_eq!(
started,
StartedWebrtcRealtime {
started: ThreadRealtimeStartedNotification {
thread_id: harness.thread_id.clone(),
realtime_session_id: Some(harness.thread_id.clone()),
version: RealtimeConversationVersion::V3,
},
sdp: ThreadRealtimeSdpNotification {
thread_id: harness.thread_id.clone(),
sdp: "v=answer\r\n".to_string(),
},
}
);
assert_call_create_multipart(
harness.call_capture.single_request(),
"v=offer\r\n",
r#"{"audio":{"output":{"voice":"cove"}},"delegation":{"ack_filler":false,"type":"client"},"instructions":"backend prompt\n\nstartup context","model":"gpt-live-1-codex"}"#,
"/v1/live",
)?;
assert!(
harness
.realtime_server
.wait_for_handshakes(/*expected*/ 1, DEFAULT_TIMEOUT)
.await,
"Frameless sideband should connect"
);
assert_eq!(
harness.realtime_server.single_handshake().uri(),
"/v1/live/rtc_e2e"
);
assert_eq!(
harness
.realtime_server
.single_handshake()
.header("openai-alpha")
.as_deref(),
Some("quicksilver=v2")
);
assert!(
harness.realtime_server.single_connection().is_empty(),
"Frameless WebRTC sideband must not send a second session.update"
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_default_automatic_output_uses_handoff_append() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"legacy automatic speech",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_v1_default_handoff")],
vec![],
vec![],
])]),
)
.await?;
let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
assert_v1_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
let turn_request_id = harness
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: harness.thread_id.clone(),
input: vec![V2UserInput::Text {
text: "say the default output".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(turn_request_id)).await??;
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 1).await,
json!({
"type": "conversation.handoff.append",
"handoff_id": "codex",
"output_text": "\"Agent Final Message\":\n\nlegacy automatic speech",
})
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_client_managed_handoffs_disable_automatic_output() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"client-managed output",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_v1_client_managed_handoffs")],
vec![],
])]),
)
.await?;
let started = harness
.start_webrtc_realtime_with_codex_response_routing(
"v=offer\r\n",
/*client_managed_handoffs*/ Some(true),
/*codex_responses_as_items*/ None,
/*codex_response_handoff_mode*/ None,
/*delegation_ack_filler*/ None,
RealtimeConversationVersion::V1,
)
.await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
assert_v1_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
let turn_request_id = harness
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: harness.thread_id.clone(),
input: vec![V2UserInput::Text {
text: "leave realtime delivery to the client".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(turn_request_id)).await??;
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
let automatic_handoff = timeout(
Duration::from_millis(200),
harness
.realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1),
)
.await;
assert!(
automatic_handoff.is_err(),
"automatic Codex output should not reach realtime in client-managed handoff mode"
);
harness
.append_speech(harness.thread_id.clone(), "client-selected speech")
.await?;
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 1).await,
json!({
"type": "conversation.handoff.append",
"handoff_id": "codex",
"output_text": "client-selected speech",
})
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_ignores_codex_response_handoff_mode() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut commentary = responses::ev_assistant_message("msg-commentary", "background progress");
commentary["item"]["phase"] = json!("commentary");
let mut final_answer = responses::ev_assistant_message("msg-final", "background complete");
final_answer["item"]["phase"] = json!("final_answer");
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![responses::sse(vec![
responses::ev_response_created("resp-1"),
commentary,
final_answer,
responses::ev_completed("resp-1"),
])]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v1_channel_handoff"),
json!({
"type": "conversation.handoff.requested",
"handoff_id": "handoff_channel",
"item_id": "item_channel",
"input_transcript": "run the background task"
}),
],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_webrtc_realtime_with_codex_response_routing(
"v=offer\r\n",
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
/*codex_response_handoff_mode*/ Some(CodexResponseHandoffMode::BemTags),
/*delegation_ack_filler*/ None,
RealtimeConversationVersion::V1,
)
.await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 1).await,
json!({
"type": "conversation.handoff.append",
"handoff_id": "handoff_channel",
"output_text": "background progress",
})
);
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 2).await,
json!({
"type": "conversation.handoff.append",
"handoff_id": "handoff_channel",
"output_text": "\"Agent Final Message\":\n\nbackground complete",
})
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_handoff_request_delegates_context_and_manual_append_speaks() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: script one v1 handoff request on the sideband and one delegated Responses turn.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"delegated from v1",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v1_handoff"),
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "delegate from v1"
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "the secret word is "
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "kumquat"
}),
json!({
"type": "conversation.handoff.requested",
"handoff_id": "handoff_v1",
"item_id": "item_v1",
"input_transcript": "delegate from v1"
}),
],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_webrtc_realtime_with_codex_response_items("v=offer\r\n")
.await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
assert_call_create_multipart(
harness.call_capture.single_request(),
"v=offer\r\n",
v1_session_create_json(),
"/v1/realtime/calls?intent=quicksilver&architecture=avas",
)?;
assert_v1_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
// Phase 2: wait for the delegated background agent turn that is launched by the handoff request.
let turn_started = harness
.read_notification::<TurnStartedNotification>("turn/started")
.await?;
assert_eq!(turn_started.thread_id, harness.thread_id);
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
// Phase 3: assert the delegated prompt went to Responses, then the automatic v1 output went
// back over the existing sideband connection as a conversation item.
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 1);
assert!(
response_request_contains_text(
&requests[0],
"<realtime_delegation>\n <input>delegate from v1</input>\n <transcript_delta>user: delegate from v1\nassistant: the secret word is kumquat</transcript_delta>\n</realtime_delegation>",
),
"delegated Responses request should contain realtime delegation envelope: {}",
requests[0]
);
let context_update = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_eq!(
context_update,
json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "developer",
"content": [{
"type": "input_text",
"text": format!("{RESPONSE_ITEM_PREFIX}\n\ndelegated from v1")
}]
}
})
);
harness
.append_speech(harness.thread_id.clone(), "manual spoken v1 update")
.await?;
let spoken_append = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_eq!(
spoken_append,
json!({
"type": "conversation.handoff.append",
"handoff_id": "codex",
"output_text": "manual spoken v1 update",
})
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_automatic_standalone_output_is_item_and_append_speaks() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"automatic output",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_manual_handoff")],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_websocket_realtime_with_codex_response_items()
.await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 0).await["type"].as_str(),
Some("session.update")
);
let turn_request_id = harness
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: harness.thread_id.clone(),
input: vec![V2UserInput::Text {
text: "do something quietly".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(turn_request_id)).await??;
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_v2_backend_item_update(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
"automatic output",
);
let automatic_response_create = timeout(
Duration::from_millis(200),
harness
.realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2),
)
.await;
assert!(
automatic_response_create.is_err(),
"automatic item should not request a realtime response"
);
harness
.append_speech(harness.thread_id.clone(), "manual voice update")
.await?;
assert_v2_progress_update(
&harness.sideband_outbound_request(/*request_index*/ 2).await,
"manual voice update",
);
assert_v2_response_create(&harness.sideband_outbound_request(/*request_index*/ 3).await);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_automatic_handoff_output_is_item_and_append_speaks() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"automatic final response",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_manual_update"),
v2_background_agent_tool_call("call_quiet", "delegate quietly"),
],
vec![],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_websocket_realtime_with_codex_response_items()
.await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 0).await["type"].as_str(),
Some("session.update")
);
let turn_started = harness
.read_notification::<TurnStartedNotification>("turn/started")
.await?;
assert_eq!(turn_started.thread_id, harness.thread_id);
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
assert_v2_backend_item_update(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
"automatic final response",
);
assert_v2_function_call_output(
&harness.sideband_outbound_request(/*request_index*/ 2).await,
"call_quiet",
"",
);
let automatic_response_create = timeout(
Duration::from_millis(200),
harness
.realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 3),
)
.await;
assert!(
automatic_response_create.is_err(),
"automatic handoff item should not request a realtime response"
);
harness
.append_speech(harness.thread_id.clone(), "manual spoken update")
.await?;
assert_v2_progress_update(
&harness.sideband_outbound_request(/*request_index*/ 3).await,
"manual spoken update",
);
assert_v2_response_create(&harness.sideband_outbound_request(/*request_index*/ 4).await);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn websocket_v2_assistant_output_without_handoff_reaches_realtime_context() -> Result<()> {
skip_if_no_network!(Ok(()));
let final_answer = "long output ".repeat(1_000);
let preamble = "direct preamble from v2";
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
main_loop_responses(vec![responses::sse(vec![
responses::ev_response_created("resp-1"),
json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": "msg-preamble",
"phase": "commentary",
"content": [{"type": "output_text", "text": preamble}]
}
}),
responses::ev_assistant_message("msg-final", &final_answer),
responses::ev_completed("resp-1"),
])]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_standalone_output")],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_websocket_realtime_with_codex_response_items()
.await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
let request_id = harness
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: harness.thread_id.clone(),
input: vec![V2UserInput::Text {
text: "direct text turn".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(request_id)).await??;
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_v2_backend_item_update(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
preamble,
);
let final_request = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_eq!(final_request["type"], "conversation.item.create");
assert_eq!(final_request["item"]["type"], "message");
assert_eq!(final_request["item"]["role"], "developer");
assert_eq!(final_request["item"]["content"][0]["type"], "input_text");
let output_text = final_request["item"]["content"][0]["text"]
.as_str()
.expect("output text");
assert!(output_text.starts_with(&format!("{RESPONSE_ITEM_PREFIX}\n\n[BACKEND] ")));
assert!(output_text.contains("tokens truncated"));
assert!(output_text.len() <= 4_000);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn websocket_v3_passes_initial_items_through_session_start() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(Vec::new()),
realtime_sideband(vec![realtime_sideband_connection(vec![vec![
session_started("sess_initial_items"),
]])]),
)
.await?;
let started = harness
.start_frameless_bidi_realtime(
/*codex_response_handoff_mode*/ None,
/*codex_response_handoff_channel_prefixes*/ None,
Some(vec![
ThreadRealtimeInitialItem {
role: ConversationTextRole::Developer,
text: "Remember this.".to_string(),
},
ThreadRealtimeInitialItem {
role: ConversationTextRole::Assistant,
text: "Understood.".to_string(),
},
]),
)
.await?;
assert_eq!(started.version, RealtimeConversationVersion::V3);
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 0).await["session"]["initial_items"],
json!([
{
"type": "message",
"role": "developer",
"content": [{"type": "input_text", "text": "Remember this."}],
},
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Understood."}],
},
])
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn websocket_v3_routes_handoffs_by_session_mode() -> Result<()> {
skip_if_no_network!(Ok(()));
for (mode, channel_prefixes, texts, expected_channels) in [
(
None,
None,
[
"[ANALYSIS]silent context",
"[COMMENTARY]still working",
"[FINAL]finished",
"unparsable BEM output",
],
[None, None, None, None],
),
(
Some(CodexResponseHandoffMode::Commentary),
None,
[
"[ANALYSIS]silent context",
"[COMMENTARY]still working",
"[FINAL]finished",
"unparsable BEM output",
],
[
Some("commentary"),
Some("commentary"),
Some("commentary"),
Some("commentary"),
],
),
(
Some(CodexResponseHandoffMode::BemTags),
None,
[
"[ANALYSIS]silent context",
"[COMMENTARY]still working",
"[FINAL]finished",
"unparsable BEM output",
],
[
Some("commentary"),
Some("commentary"),
Some("speakable"),
Some("speakable"),
],
),
(
Some(CodexResponseHandoffMode::BemTags),
Some(BTreeMap::from([
("analysis".to_string(), vec!["[THOUGHT]".to_string()]),
(
"commentary".to_string(),
vec!["[PROGRESS]".to_string(), "[UPDATE]".to_string()],
),
("final".to_string(), vec!["[DONE]".to_string()]),
])),
[
"[THOUGHT]silent context",
"[UPDATE]still working",
"[DONE]finished",
"unparsable BEM output",
],
[
Some("commentary"),
Some("commentary"),
Some("speakable"),
Some("speakable"),
],
),
] {
let [analysis_text, commentary_text, final_text, fallback_text] = texts;
let analysis = responses::ev_assistant_message("msg-analysis", analysis_text);
let commentary = responses::ev_assistant_message("msg-commentary", commentary_text);
let final_answer = responses::ev_assistant_message("msg-final", final_text);
let fallback = responses::ev_assistant_message("msg-fallback", fallback_text);
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![responses::sse(vec![
responses::ev_response_created("resp-1"),
analysis,
commentary,
final_answer,
fallback,
responses::ev_completed("resp-1"),
])]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_started("sess_frameless"),
json!({
"type": "delegation.created",
"offset_ms": 100,
"item": {
"id": "delegation_frameless",
"type": "delegation",
"target": "client",
"content": [{
"type": "input_text",
"text": "delegate from frameless"
}]
}
}),
],
vec![],
vec![],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_frameless_bidi_realtime(mode, channel_prefixes, /*initial_items*/ None)
.await?;
assert_eq!(started.version, RealtimeConversationVersion::V3);
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
for (request_index, (text, channel)) in
[analysis_text, commentary_text, final_text, fallback_text]
.into_iter()
.zip(expected_channels)
.enumerate()
{
let mut expected = json!({
"type": "delegation.context.append",
"delegation_item_id": "delegation_frameless",
"content": [{
"type": "input_text",
"text": text
}]
});
if let Some(channel) = channel {
expected["channel"] = json!(channel);
}
assert_eq!(
harness
.sideband_outbound_request(/*request_index*/ request_index + 1)
.await,
expected
);
}
harness
.append_speech(harness.thread_id.clone(), "manual spoken update")
.await?;
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 5).await,
json!({
"type": "session.context.append",
"content": [{
"type": "input_text",
"text": "manual spoken update"
}],
"channel": "speakable"
})
);
harness.shutdown().await;
}
Ok(())
}
#[tokio::test]
async fn websocket_v2_forwards_audio_and_text_between_client_and_sideband() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: create a v2 websocket conversation whose sideband sends transcript + output audio
// after the client has had a chance to append input.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
no_main_loop_responses(),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_v2_stream")],
vec![],
vec![
json!({
"type": "conversation.item.input_audio_transcription.delta",
"delta": "transcribed audio"
}),
json!({
"type": "response.output_audio.delta",
"delta": "AQID",
"sample_rate": 24_000,
"channels": 1,
"samples_per_channel": 512
}),
],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
assert_v2_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
// Phase 2: drive app-server as the client would: append audio, append text, then receive
// transcript/audio notifications that came from the sideband socket.
let thread_id = started.thread_id.clone();
harness.append_audio(thread_id.clone()).await?;
harness.append_text(thread_id, "hello").await?;
let transcript = harness
.read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
"thread/realtime/transcript/delta",
)
.await?;
assert_eq!(transcript.delta, "transcribed audio");
let output_audio = harness
.read_notification::<ThreadRealtimeOutputAudioDeltaNotification>(
"thread/realtime/outputAudio/delta",
)
.await?;
assert_eq!(output_audio.audio.data, "AQID");
// Phase 3: prove the client inputs were translated into the v2 realtime sideband events.
let requests = [
harness.sideband_outbound_request(/*request_index*/ 1).await,
harness.sideband_outbound_request(/*request_index*/ 2).await,
];
assert!(
requests
.iter()
.any(|request| request["type"] == "input_audio_buffer.append"
&& request["audio"] == "BQYH"),
"sideband requests should include audio append: {requests:?}"
);
assert!(
requests.iter().any(|request| {
request["type"] == "conversation.item.create"
&& request["item"]["type"] == "message"
&& request["item"]["role"] == "user"
&& request["item"]["content"][0]["type"] == "input_text"
&& request["item"]["content"][0]["text"] == "[USER] hello"
}),
"sideband requests should include user text item: {requests:?}"
);
harness.shutdown().await;
Ok(())
}
/// Regression coverage for Realtime V2 text input while a response is active.
///
/// Text input is append-only, so app-server should send the user message without
/// requesting a new realtime response.
#[tokio::test]
async fn websocket_v2_text_input_is_append_only_while_response_is_active() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: script a server-side response that becomes active after the first
// user text turn, then finishes only after a later audio input.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
no_main_loop_responses(),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_v2_response_queue")],
vec![
json!({
"type": "response.created",
"response": { "id": "resp_active" }
}),
json!({
"type": "response.output_text.delta",
"delta": "active response started"
}),
],
vec![],
vec![json!({
"type": "response.done",
"response": { "id": "resp_active" }
})],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
// From here on, `sideband_outbound_request(n)` reads outbound messages to
// the fake Realtime API sideband websocket. These are not client-facing
// notifications; they are the protocol frames app-server sends upstream.
assert_v2_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
// Phase 2: send the first text turn. Text input is append-only, so this
// sends only the user text item.
let thread_id = started.thread_id.clone();
harness.append_text(thread_id.clone(), "first").await?;
assert_v2_user_text_item(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
"first",
);
let transcript = harness
.read_notification::<ThreadRealtimeTranscriptDeltaNotification>(
"thread/realtime/transcript/delta",
)
.await?;
assert_eq!(transcript.delta, "active response started");
// Phase 3: send a second text turn while `resp_active` is still open. The
// user message must reach realtime without requesting another response.
harness.append_text(thread_id.clone(), "second").await?;
assert_v2_user_text_item(
&harness.sideband_outbound_request(/*request_index*/ 2).await,
"second",
);
// Phase 4: audio still forwards normally after text input.
harness.append_audio(thread_id).await?;
let audio = harness.sideband_outbound_request(/*request_index*/ 3).await;
assert_eq!(audio["type"], "input_audio_buffer.append");
assert_eq!(audio["audio"], "BQYH");
harness.shutdown().await;
Ok(())
}
/// Regression coverage for append-only Realtime V2 text input when the active
/// response is cancelled instead of completed.
#[tokio::test]
async fn websocket_v2_text_input_is_append_only_when_response_is_cancelled() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: script a server-side response that becomes active after the first
// text turn, then is cancelled only after a later audio input.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
no_main_loop_responses(),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![session_updated("sess_v2_response_cancel_queue")],
vec![json!({
"type": "response.created",
"response": { "id": "resp_cancelled" }
})],
vec![],
vec![json!({
"type": "response.cancelled",
"response": { "id": "resp_cancelled" }
})],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
assert_v2_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
// Phase 2: send the first text turn. Text input is append-only, so this
// sends only the user text item.
let thread_id = started.thread_id.clone();
harness.append_text(thread_id.clone(), "first").await?;
assert_v2_user_text_item(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
"first",
);
// Phase 3: send a second text turn while `resp_cancelled` is still open.
// The user message must reach realtime without requesting another response.
harness.append_text(thread_id.clone(), "second").await?;
assert_v2_user_text_item(
&harness.sideband_outbound_request(/*request_index*/ 2).await,
"second",
);
// Phase 4: audio still forwards normally after text input.
harness.append_audio(thread_id).await?;
let audio = harness.sideband_outbound_request(/*request_index*/ 3).await;
assert_eq!(audio["type"], "input_audio_buffer.append");
assert_eq!(audio["audio"], "BQYH");
harness.shutdown().await;
Ok(())
}
/// Regression coverage for the Realtime V2 background-agent final-output path.
///
/// Once the background agent finishes, app-server sends the final function-call
/// output to realtime and then requests a new `response.create` so realtime can
/// react to that final output.
#[tokio::test]
async fn websocket_v2_background_agent_returns_function_output() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: script a v2 background agent function call and a delegated Responses turn that
// returns final assistant text.
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"delegated from v2",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_tool"),
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "Hi how are you"
}),
json!({
"type": "response.output_audio_transcript.done",
"transcript": "Doing well, what can I help you with?"
}),
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "The secret word is strawberry"
}),
json!({
"type": "conversation.item.created",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": "<realtime_collaboration_update><voice_policy>silent_delegate</voice_policy></realtime_collaboration_update>"
}]
}
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "Got it-strawberry. What's next on the menu?"
}),
v2_background_agent_tool_call("call_v2", "run ls"),
],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
// Phase 2: wait for the delegated turn lifecycle kicked off by the v2 function-call item.
let turn_started = harness
.read_notification::<TurnStartedNotification>("turn/started")
.await?;
assert_eq!(turn_started.thread_id, harness.thread_id);
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
// Phase 3: assert the delegated prompt went to Responses and the result
// returned as exactly one v2 function-call output event on the sideband.
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 1);
assert!(
response_request_contains_text(
&requests[0],
"<realtime_delegation>\n <input>run ls</input>\n <transcript_delta>user: Hi how are you\nassistant: Doing well, what can I help you with?\nuser: The secret word is strawberry\nassistant: Got it-strawberry. What's next on the menu?\nuser: run ls</transcript_delta>\n</realtime_delegation>",
),
"delegated Responses request should contain realtime delegation envelope: {}",
requests[0]
);
assert!(
!response_request_contains_text(&requests[0], "<realtime_collaboration_update>"),
"delegated Responses request should not include realtime control injects: {}",
requests[0]
);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "delegated from v2");
let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_v2_function_call_output(&tool_output, "call_v2", V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT);
harness.shutdown().await;
Ok(())
}
/// Regression coverage for Realtime V2 steering while a background-agent task is
/// already active.
///
/// The second background-agent tool call is treated as guidance for the active
/// task. App-server acknowledges that steering message to realtime and then
/// emits `response.create` so realtime can speak that acknowledgement.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn websocket_v2_background_agent_steering_ack_requests_response_create() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: gate the delegated Responses turn from the first tool call so
// the background-agent handoff stays active while realtime sends a second
// tool call that should steer the active task.
let main_loop_responses_server = responses::start_mock_server().await;
let (gate_completed_tx, gate_completed_rx) = mpsc::channel();
let gated_response = responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "first task finished"),
responses::ev_completed("resp-1"),
]);
Mock::given(method("POST"))
.and(path_regex(".*/responses$"))
.respond_with(GatedSseResponse {
gate_rx: Mutex::new(Some(gate_completed_rx)),
response: gated_response,
})
.expect(2)
.mount(&main_loop_responses_server)
.await;
let mut harness = RealtimeE2eHarness::new_with_main_loop_responses_server(
RealtimeTestVersion::V2,
main_loop_responses_server,
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_steering_ack"),
v2_background_agent_tool_call("call_active", "start a task"),
v2_background_agent_tool_call("call_steer", "steer the active task"),
],
vec![],
vec![],
vec![],
vec![],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
assert_v2_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
let turn_started = harness
.read_notification::<TurnStartedNotification>("turn/started")
.await?;
assert_eq!(turn_started.thread_id, harness.thread_id);
// Phase 2: the second tool call happens while `call_active` is still
// running, so app-server sends a steering acknowledgement as a function-call
// output for the second call.
assert_v2_function_call_output(
&harness.sideband_outbound_request(/*request_index*/ 1).await,
"call_steer",
V2_STEERING_ACKNOWLEDGEMENT,
);
// Phase 3: realtime needs a `response.create` after the steering
// acknowledgement so it can surface that acknowledgement to the user.
assert_v2_response_create(&harness.sideband_outbound_request(/*request_index*/ 2).await);
// Phase 4: release the gated delegated turn. Codex should then continue
// the same run with the steering text included in the follow-up Responses
// request, proving realtime did not merely acknowledge and drop it.
let _ = gate_completed_tx.send(());
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 2);
assert!(
response_request_contains_text(&requests[1], "steer the active task"),
"follow-up Responses request should contain steering prompt: {}",
requests[1]
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn websocket_v2_background_agent_progress_is_sent_before_function_output() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V2,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"progress before final",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_progress_before_final"),
v2_background_agent_tool_call("call_progress_order", "stream progress"),
],
vec![],
vec![],
])]),
)
.await?;
let started = harness.start_websocket_realtime().await?;
assert_eq!(started.version, RealtimeConversationVersion::V2);
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "progress before final");
let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_v2_function_call_output(
&tool_output,
"call_progress_order",
V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT,
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn websocket_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<()> {
// TODO(anp): Remove after delegated shell commands resolve target-native cwd in remote environments.
skip_if_remote!(
Ok(()),
"delegated shell command cwd is only materialized on the host"
);
skip_if_no_network!(Ok(()));
// Phase 1: keep the two mocked OpenAI conversations explicit. The realtime sideband only
// calls the `background_agent` function; the shell command is requested by the delegated
// background agent Responses turn that app-server starts after receiving that function call.
let main_loop = main_loop_responses(vec![
create_command_execution_sse_response(
realtime_tool_ok_command(),
/*workdir*/ None,
// Windows CI can spend several seconds starting the nested PowerShell command. This
// test verifies delegated shell-tool plumbing, not timeout enforcement.
Some(DELEGATED_SHELL_TOOL_TIMEOUT_MS),
"shell_call",
)?,
create_final_assistant_message_sse_response("shell tool finished")?,
]);
let realtime = realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_shell"),
v2_background_agent_tool_call("call_shell", "run shell through delegated turn"),
],
vec![],
vec![],
])]);
let mut harness = RealtimeE2eHarness::new_with_sandbox(
RealtimeTestVersion::V2,
main_loop,
realtime,
RealtimeTestSandbox::DangerFullAccess,
)
.await?;
let _ = harness.start_websocket_realtime().await?;
// Phase 2: observe the delegated background agent turn executing the requested shell command.
let started_command = wait_for_started_command_execution(&mut harness.mcp).await?;
let ThreadItem::CommandExecution { id, status, .. } = started_command.item else {
unreachable!("helper returns command execution items");
};
assert_eq!(
(id.as_str(), status),
("shell_call", CommandExecutionStatus::InProgress)
);
let completed_command = wait_for_completed_command_execution(&mut harness.mcp).await?;
let ThreadItem::CommandExecution {
id,
status,
aggregated_output,
..
} = completed_command.item
else {
unreachable!("helper returns command execution items");
};
assert_eq!(id.as_str(), "shell_call");
assert_eq!(status, CommandExecutionStatus::Completed);
assert_eq!(aggregated_output.as_deref(), Some("realtime-tool-ok"));
// Phase 3: verify the shell output reached Responses and the final delegated answer returned
// to realtime as a single function-call-output item.
let turn_completed = read_notification_with_timeout::<TurnCompletedNotification>(
&mut harness.mcp,
"turn/completed",
DELEGATED_SHELL_TURN_TIMEOUT,
)
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 2);
assert!(
response_request_contains_text(&requests[1], "realtime-tool-ok"),
"follow-up Responses request should contain shell output: {}",
requests[1]
);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "shell tool finished");
let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_v2_function_call_output(
&tool_output,
"call_shell",
V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT,
);
harness.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn websocket_v2_tool_call_does_not_block_sideband_audio() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: gate the delegated Responses stream so the sideband can send audio while the tool
// call is still waiting on its delegated turn.
let main_loop_responses_server = responses::start_mock_server().await;
let (gate_completed_tx, gate_completed_rx) = mpsc::channel();
let gated_response = responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "late delegated result"),
responses::ev_completed("resp-1"),
]);
Mock::given(method("POST"))
.and(path_regex(".*/responses$"))
.respond_with(GatedSseResponse {
gate_rx: Mutex::new(Some(gate_completed_rx)),
response: gated_response,
})
.expect(1)
.mount(&main_loop_responses_server)
.await;
let mut harness = RealtimeE2eHarness::new_with_main_loop_responses_server(
RealtimeTestVersion::V2,
main_loop_responses_server,
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_nonblocking"),
v2_background_agent_tool_call("call_audio", "delegate while audio continues"),
json!({
"type": "response.output_audio.delta",
"delta": "CQoL",
"sample_rate": 24_000,
"channels": 1,
"samples_per_channel": 256
}),
],
vec![],
vec![],
])]),
)
.await?;
let _ = harness.start_websocket_realtime().await?;
let _ = harness
.read_notification::<TurnStartedNotification>("turn/started")
.await?;
// Phase 2: require app-server to fan out sideband audio before the delegated tool call is
// allowed to finish.
let audio = harness
.read_notification::<ThreadRealtimeOutputAudioDeltaNotification>(
"thread/realtime/outputAudio/delta",
)
.await?;
assert_eq!(audio.audio.data, "CQoL");
// Phase 3: release the delegated turn and assert the sideband function-call output is delivered
// after the nonblocking audio.
let _ = gate_completed_tx.send(());
let turn_completed = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "late delegated result");
let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await;
assert_v2_function_call_output(
&tool_output,
"call_audio",
V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT,
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> {
skip_if_no_network!(Ok(()));
// Phase 1: make call creation fail before any sideband connection can matter.
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
Mock::given(method("POST"))
.and(path("/v1/realtime/calls"))
.respond_with(ResponseTemplate::new(500).set_body_string("boom"))
.mount(&responses_server)
.await;
let realtime_server = start_websocket_server(vec![vec![]]).await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
/*realtime_enabled*/ true,
StartupContextConfig::Override("startup context"),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
login_with_api_key(&mut mcp, "sk-test-key").await?;
// Phase 2: start a normal app-server thread and request realtime over WebRTC.
let thread_start_request_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams::default())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: Some(ThreadRealtimeStartTransport::Webrtc {
sdp: "v=offer\r\n".to_string(),
}),
version: Some(RealtimeConversationVersion::V1),
voice: None,
})
.await?;
let _: ThreadRealtimeStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(start_request_id)).await??;
// Phase 3: the JSON-RPC start request returns, and the realtime failure is delivered as the
// typed realtime error notification.
let error =
read_notification::<ThreadRealtimeErrorNotification>(&mut mcp, "thread/realtime/error")
.await?;
assert!(error.message.contains("currently experiencing high demand"));
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test]
async fn realtime_conversation_requires_feature_flag() -> 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![]]).await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
/*realtime_enabled*/ false,
StartupContextConfig::Generated,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_TIMEOUT)
.await?;
let thread_start_request_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams::default())
.await?;
let thread_start: ThreadStartResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
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,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: None,
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
})
.await?;
let error = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(start_request_id)),
)
.await??;
assert_invalid_request(
error,
format!(
"thread {} does not support realtime conversation",
thread_start.thread.id
),
);
realtime_server.shutdown().await;
Ok(())
}
async fn read_notification<T: DeserializeOwned>(
mcp: &mut TestAppServer,
method: &str,
) -> Result<T> {
read_notification_with_timeout(mcp, method, DEFAULT_TIMEOUT).await
}
async fn read_notification_with_timeout<T: DeserializeOwned>(
mcp: &mut TestAppServer,
method: &str,
timeout_duration: Duration,
) -> Result<T> {
timeout(timeout_duration, mcp.read_notification(method)).await?
}
async fn login_with_api_key(mcp: &mut TestAppServer, api_key: &str) -> Result<()> {
let request_id = mcp.send_login_account_api_key_request(api_key).await?;
let login: LoginAccountResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??;
assert_eq!(login, LoginAccountResponse::ApiKey {});
Ok(())
}
async fn wait_for_started_command_execution(
mcp: &mut TestAppServer,
) -> Result<ItemStartedNotification> {
loop {
let started = read_notification::<ItemStartedNotification>(mcp, "item/started").await?;
if let ThreadItem::CommandExecution { .. } = &started.item {
return Ok(started);
}
}
}
async fn wait_for_completed_command_execution(
mcp: &mut TestAppServer,
) -> Result<ItemCompletedNotification> {
loop {
let completed =
read_notification::<ItemCompletedNotification>(mcp, "item/completed").await?;
if let ThreadItem::CommandExecution { .. } = &completed.item {
return Ok(completed);
}
}
}
async fn responses_requests(server: &MockServer) -> Result<Vec<Value>> {
server
.received_requests()
.await
.context("failed to fetch received requests")?
.into_iter()
.filter(|request| request.url.path().ends_with("/responses"))
.map(|request| {
request
.body_json::<Value>()
.context("Responses request body should be JSON")
})
.collect()
}
fn response_request_contains_text(request: &Value, text: &str) -> bool {
match request {
Value::String(value) => value.contains(text),
Value::Array(values) => values
.iter()
.any(|value| response_request_contains_text(value, text)),
Value::Object(map) => map
.values()
.any(|value| response_request_contains_text(value, text)),
Value::Null | Value::Bool(_) | Value::Number(_) => false,
}
}
fn realtime_tool_ok_command() -> Vec<String> {
#[cfg(windows)]
{
vec![
"powershell.exe".to_string(),
"-NoProfile".to_string(),
"-Command".to_string(),
"[Console]::Write('realtime-tool-ok')".to_string(),
]
}
#[cfg(not(windows))]
{
vec!["printf".to_string(), "realtime-tool-ok".to_string()]
}
}
fn assert_v2_function_call_output(request: &Value, call_id: &str, expected_output: &str) {
assert_eq!(
request,
&json!({
"type": "conversation.item.create",
"item": {
"type": "function_call_output",
"call_id": call_id,
"output": expected_output,
}
})
);
}
fn assert_v2_progress_update(request: &Value, expected_text: &str) {
assert_eq!(
request,
&json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": format!("[BACKEND] {expected_text}")
}]
}
})
);
}
fn assert_v2_backend_item_update(request: &Value, expected_text: &str) {
assert_v2_items_update(request, &format!("[BACKEND] {expected_text}"));
}
fn assert_v2_items_update(request: &Value, expected_text: &str) {
assert_eq!(
request,
&json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "developer",
"content": [{
"type": "input_text",
"text": format!("{RESPONSE_ITEM_PREFIX}\n\n{expected_text}")
}]
}
})
);
}
fn assert_v2_user_text_item(request: &Value, expected_text: &str) {
assert_eq!(
request,
&json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": format!("[USER] {expected_text}")
}]
}
})
);
}
fn assert_v2_response_create(request: &Value) {
assert_eq!(
request,
&json!({
"type": "response.create"
})
);
}
fn assert_v1_session_update(request: &Value) -> Result<()> {
assert_eq!(request["type"].as_str(), Some("session.update"));
assert_eq!(request["session"]["type"].as_str(), Some("quicksilver"));
assert!(
request["session"]["instructions"]
.as_str()
.context("v1 session.update instructions")?
.contains("startup context")
);
assert_eq!(
request["session"]["audio"]["output"]["voice"].as_str(),
Some("cove")
);
assert_eq!(request["session"]["tools"], Value::Null);
Ok(())
}
fn assert_v2_session_update(request: &Value) -> Result<()> {
assert_eq!(request["type"].as_str(), Some("session.update"));
assert_eq!(request["session"]["type"].as_str(), Some("realtime"));
assert!(
request["session"]["instructions"]
.as_str()
.context("v2 session.update instructions")?
.contains("startup context")
);
assert_eq!(
request["session"]["tools"][0]["name"].as_str(),
Some("background_agent")
);
assert_eq!(
request["session"]["tools"][1]["name"].as_str(),
Some("remain_silent")
);
assert_eq!(
request["session"]["audio"]["input"]["transcription"]["model"].as_str(),
Some("gpt-4o-mini-transcribe")
);
Ok(())
}
fn assert_call_create_multipart(
request: WiremockRequest,
offer_sdp: &str,
expected_session: &str,
expected_path_and_query: &str,
) -> Result<()> {
let path_and_query = match request.url.query() {
Some(query) => format!("{}?{query}", request.url.path()),
None => request.url.path().to_string(),
};
assert_eq!(path_and_query, expected_path_and_query);
assert_eq!(
request
.headers
.get("content-type")
.and_then(|value| value.to_str().ok()),
Some("multipart/form-data; boundary=codex-realtime-call-boundary")
);
let body = String::from_utf8(request.body).context("multipart body should be utf-8")?;
let session_prefix = format!(
"--codex-realtime-call-boundary\r\n\
Content-Disposition: form-data; name=\"sdp\"\r\n\
Content-Type: application/sdp\r\n\
\r\n\
{offer_sdp}\r\n\
--codex-realtime-call-boundary\r\n\
Content-Disposition: form-data; name=\"session\"\r\n\
Content-Type: application/json\r\n\
\r\n"
);
let actual_session = body
.strip_prefix(&session_prefix)
.and_then(|body| body.strip_suffix("\r\n--codex-realtime-call-boundary--\r\n"))
.context("multipart body should contain one JSON session part")?;
let actual_session: Value =
serde_json::from_str(actual_session).context("session part should be valid JSON")?;
let expected_session: Value = serde_json::from_str(expected_session)
.context("expected session fixture should be valid JSON")?;
assert_eq!(actual_session, expected_session);
Ok(())
}
fn v1_session_create_json() -> &'static str {
r#"{"audio":{"input":{"format":{"type":"audio/pcm","rate":24000}},"output":{"voice":"cove"}},"type":"quicksilver","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context"}"#
}
fn create_config_toml(
codex_home: &Path,
responses_server_uri: &str,
realtime_server_uri: &str,
realtime_enabled: bool,
startup_context: StartupContextConfig<'_>,
) -> std::io::Result<()> {
create_config_toml_with_realtime_version(
codex_home,
responses_server_uri,
realtime_server_uri,
realtime_enabled,
startup_context,
RealtimeTestVersion::V2,
RealtimeTestSandbox::ReadOnly,
)
}
fn create_config_toml_with_realtime_version(
codex_home: &Path,
responses_server_uri: &str,
realtime_server_uri: &str,
realtime_enabled: bool,
startup_context: StartupContextConfig<'_>,
realtime_version: RealtimeTestVersion,
sandbox: RealtimeTestSandbox,
) -> std::io::Result<()> {
let mut config = MockResponsesConfig::new(responses_server_uri)
.with_sandbox_mode(sandbox.config_value())
.with_root_config(&format!(
"experimental_realtime_ws_base_url = \"{realtime_server_uri}\"\n\
experimental_realtime_ws_backend_prompt = \"backend prompt\""
))
.with_extra_config(&format!(
"[realtime]\nversion = \"{}\"\ntype = \"conversational\"",
realtime_version.config_value()
));
if let StartupContextConfig::Override(context) = startup_context {
config = config.with_root_config(&format!(
"experimental_realtime_ws_startup_context = {context:?}"
));
}
config = if realtime_enabled {
config.enable_feature(Feature::RealtimeConversation)
} else {
config.disable_feature(Feature::RealtimeConversation)
};
config.write(codex_home)
}
fn assert_invalid_request(error: JSONRPCError, message: String) {
assert_eq!(error.error.code, -32600);
assert_eq!(error.error.message, message);
assert_eq!(error.error.data, None);
}
Reference in new issue View Git Blame Copy Permalink