mirror of
https://github.com/openai/codex.git
synced 2026-10-14 16:38:43 +00:00
acrognale-oai
888e02db34
## Why Destination restrictions and policy revocation must remain effective during redirects, response body reads, and established WebSocket traffic. Policy denials must also survive error handling so callers do not retry them or report a revoked operation as successful. ## What changed - Route managed HTTP clients through a shared `RequestBuilder` and policy-aware execution, checking each redirect destination before route resolution and retaining a network permit while consuming response bodies. - Guard WebSocket connection setup, reads, and writes with revocable permits, including independent wakeups for split readers and writers. - Preserve `TransportError::Policy` through HTTP, SSE, and realtime error handling, and treat policy denials as non-retryable. - Keep the supplied network policy in plugin startup HTTP requests and propagate response body failures from backend and plugin requests. ## Testing Add regression coverage for rejection before connecting, revocation during redirect routing and streamed body reads, split WebSocket revocation, realtime writer error propagation, and revoked plugin upload responses. GitOrigin-RevId: fba36700444c98a8364c09b4dc5452c4d78a90fb
653 lines
20 KiB
Rust
653 lines
20 KiB
Rust
#![allow(clippy::expect_used)]
|
|
use std::collections::HashMap;
|
|
use std::future::Future;
|
|
use std::time::Duration;
|
|
|
|
use codex_api::Provider;
|
|
use codex_api::RealtimeAudioFrame;
|
|
use codex_api::RealtimeEvent;
|
|
use codex_api::RealtimeEventParser;
|
|
use codex_api::RealtimeOutputModality;
|
|
use codex_api::RealtimeSessionConfig;
|
|
use codex_api::RealtimeSessionMode;
|
|
use codex_api::RealtimeTranscriptState;
|
|
use codex_api::RealtimeWebsocketClient;
|
|
use codex_api::RetryConfig;
|
|
use codex_http_client::HttpClientFactory;
|
|
use codex_http_client::OutboundProxyPolicy::ReqwestDefault;
|
|
use codex_protocol::protocol::RealtimeHandoffRequested;
|
|
use codex_protocol::protocol::RealtimeTranscriptDelta;
|
|
use codex_protocol::protocol::RealtimeTranscriptDone;
|
|
use codex_protocol::protocol::RealtimeTranscriptEntry;
|
|
use codex_protocol::protocol::RealtimeVoice;
|
|
use futures::SinkExt;
|
|
use futures::StreamExt;
|
|
use http::HeaderMap;
|
|
use serde_json::Value;
|
|
use serde_json::json;
|
|
use tokio::net::TcpListener;
|
|
use tokio_tungstenite::accept_async;
|
|
use tokio_tungstenite::tungstenite::Message;
|
|
|
|
type RealtimeWsStream = tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>;
|
|
|
|
async fn spawn_realtime_ws_server<Handler, Fut>(
|
|
handler: Handler,
|
|
) -> (String, tokio::task::JoinHandle<()>)
|
|
where
|
|
Handler: FnOnce(RealtimeWsStream) -> Fut + Send + 'static,
|
|
Fut: Future<Output = ()> + Send + 'static,
|
|
{
|
|
let listener = TcpListener::bind("127.0.0.1:0")
|
|
.await
|
|
.expect("test websocket listener should bind");
|
|
let addr = listener
|
|
.local_addr()
|
|
.expect("test websocket listener should have a local address")
|
|
.to_string();
|
|
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener
|
|
.accept()
|
|
.await
|
|
.expect("test websocket connection should be accepted");
|
|
let ws = accept_async(stream)
|
|
.await
|
|
.expect("test websocket handshake should complete");
|
|
handler(ws).await;
|
|
});
|
|
|
|
(addr, server)
|
|
}
|
|
|
|
fn test_provider(base_url: String) -> Provider {
|
|
Provider {
|
|
name: "test".to_string(),
|
|
base_url,
|
|
query_params: Some(HashMap::new()),
|
|
headers: HeaderMap::new(),
|
|
retry: RetryConfig {
|
|
max_attempts: 1,
|
|
base_delay: Duration::from_millis(1),
|
|
retry_429: false,
|
|
retry_5xx: false,
|
|
retry_transport: false,
|
|
},
|
|
stream_idle_timeout: Duration::from_secs(5),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_e2e_session_create_and_event_flow() {
|
|
let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
assert_eq!(
|
|
first_json["session"]["type"],
|
|
Value::String("quicksilver".to_string())
|
|
);
|
|
assert_eq!(
|
|
first_json["session"]["instructions"],
|
|
Value::String("backend prompt".to_string())
|
|
);
|
|
assert_eq!(
|
|
first_json["session"]["audio"]["input"]["format"]["type"],
|
|
Value::String("audio/pcm".to_string())
|
|
);
|
|
assert_eq!(
|
|
first_json["session"]["audio"]["input"]["format"]["rate"],
|
|
Value::from(24_000)
|
|
);
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "session.updated",
|
|
"session": {"id": "sess_mock", "instructions": "backend prompt"}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send session.updated");
|
|
|
|
let second = ws
|
|
.next()
|
|
.await
|
|
.expect("second msg")
|
|
.expect("second msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let second_json: Value = serde_json::from_str(&second).expect("json");
|
|
assert_eq!(second_json["type"], "input_audio_buffer.append");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "conversation.output_audio.delta",
|
|
"delta": "AQID",
|
|
"sample_rate": 48000,
|
|
"channels": 1
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send audio out");
|
|
})
|
|
.await;
|
|
|
|
let client = test_client(test_provider(format!("http://{addr}")));
|
|
let connection = client
|
|
.connect(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::V1,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Cove,
|
|
},
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
)
|
|
.await
|
|
.expect("connect");
|
|
|
|
let created = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
created,
|
|
RealtimeEvent::SessionUpdated {
|
|
realtime_session_id: "sess_mock".to_string(),
|
|
instructions: Some("backend prompt".to_string()),
|
|
}
|
|
);
|
|
|
|
connection
|
|
.send_audio_frame(RealtimeAudioFrame {
|
|
data: "AQID".to_string(),
|
|
sample_rate: 48000,
|
|
num_channels: 1,
|
|
samples_per_channel: Some(960),
|
|
item_id: None,
|
|
})
|
|
.await
|
|
.expect("send audio");
|
|
|
|
let audio_event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
audio_event,
|
|
RealtimeEvent::AudioOut(RealtimeAudioFrame {
|
|
data: "AQID".to_string(),
|
|
sample_rate: 48000,
|
|
num_channels: 1,
|
|
samples_per_channel: None,
|
|
item_id: None,
|
|
})
|
|
);
|
|
|
|
connection.close().await.expect("close");
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_connect_webrtc_sideband_retries_join_until_server_is_available() {
|
|
let reserving_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
|
|
let addr = reserving_listener.local_addr().expect("local addr");
|
|
drop(reserving_listener);
|
|
|
|
let server = tokio::spawn(async move {
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
let listener = TcpListener::bind(addr).await.expect("bind delayed server");
|
|
let (stream, _) = listener.accept().await.expect("accept");
|
|
let mut ws = accept_async(stream).await.expect("accept ws");
|
|
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
assert_eq!(
|
|
first_json["session"]["instructions"],
|
|
Value::String("backend prompt".to_string())
|
|
);
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "session.updated",
|
|
"session": {"id": "sess_joined", "instructions": "backend prompt"}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send session.updated");
|
|
});
|
|
|
|
let mut provider = test_provider(format!("http://{addr}"));
|
|
provider.retry.max_attempts = 1;
|
|
provider.retry.base_delay = Duration::from_millis(100);
|
|
|
|
let client = test_client(provider).with_webrtc_sideband_base_url(format!("http://{addr}"));
|
|
let connection = client
|
|
.connect_webrtc_sideband(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::RealtimeV2,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Marin,
|
|
},
|
|
"rtc_test",
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
RealtimeTranscriptState::default(),
|
|
)
|
|
.await
|
|
.expect("connect on retry");
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
event,
|
|
RealtimeEvent::SessionUpdated {
|
|
realtime_session_id: "sess_joined".to_string(),
|
|
instructions: Some("backend prompt".to_string()),
|
|
}
|
|
);
|
|
|
|
connection.close().await.expect("close");
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_e2e_send_while_next_event_waits() {
|
|
let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
|
|
let second = ws
|
|
.next()
|
|
.await
|
|
.expect("second msg")
|
|
.expect("second msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let second_json: Value = serde_json::from_str(&second).expect("json");
|
|
assert_eq!(second_json["type"], "input_audio_buffer.append");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "session.updated",
|
|
"session": {"id": "sess_after_send", "instructions": "backend prompt"}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send session.updated");
|
|
})
|
|
.await;
|
|
|
|
let client = test_client(test_provider(format!("http://{addr}")));
|
|
let connection = client
|
|
.connect(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::V1,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Cove,
|
|
},
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
)
|
|
.await
|
|
.expect("connect");
|
|
|
|
let (send_result, next_result) = tokio::join!(
|
|
async {
|
|
tokio::time::timeout(
|
|
Duration::from_millis(200),
|
|
connection.send_audio_frame(RealtimeAudioFrame {
|
|
data: "AQID".to_string(),
|
|
sample_rate: 48000,
|
|
num_channels: 1,
|
|
samples_per_channel: Some(960),
|
|
item_id: None,
|
|
}),
|
|
)
|
|
.await
|
|
},
|
|
connection.next_event()
|
|
);
|
|
|
|
send_result
|
|
.expect("send should not block on next_event")
|
|
.expect("send audio");
|
|
let next_event = next_result.expect("next event").expect("event");
|
|
assert_eq!(
|
|
next_event,
|
|
RealtimeEvent::SessionUpdated {
|
|
realtime_session_id: "sess_after_send".to_string(),
|
|
instructions: Some("backend prompt".to_string()),
|
|
}
|
|
);
|
|
|
|
connection.close().await.expect("close");
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_e2e_disconnected_emitted_once() {
|
|
let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
|
|
ws.send(Message::Close(None)).await.expect("send close");
|
|
})
|
|
.await;
|
|
|
|
let client = test_client(test_provider(format!("http://{addr}")));
|
|
let connection = client
|
|
.connect(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::V1,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Cove,
|
|
},
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
)
|
|
.await
|
|
.expect("connect");
|
|
|
|
let first = connection.next_event().await.expect("next event");
|
|
assert_eq!(first, None);
|
|
|
|
let second = connection.next_event().await.expect("next event");
|
|
assert_eq!(second, None);
|
|
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_e2e_ignores_unknown_text_events() {
|
|
let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "response.created",
|
|
"response": {"id": "resp_unknown"}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send unknown event");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "session.updated",
|
|
"session": {"id": "sess_after_unknown", "instructions": "backend prompt"}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send session.updated");
|
|
})
|
|
.await;
|
|
|
|
let client = test_client(test_provider(format!("http://{addr}")));
|
|
let connection = client
|
|
.connect(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::V1,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Cove,
|
|
},
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
)
|
|
.await
|
|
.expect("connect");
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
event,
|
|
RealtimeEvent::SessionUpdated {
|
|
realtime_session_id: "sess_after_unknown".to_string(),
|
|
instructions: Some("backend prompt".to_string()),
|
|
}
|
|
);
|
|
|
|
connection.close().await.expect("close");
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() {
|
|
let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {
|
|
let first = ws
|
|
.next()
|
|
.await
|
|
.expect("first msg")
|
|
.expect("first msg ok")
|
|
.into_text()
|
|
.expect("text");
|
|
let first_json: Value = serde_json::from_str(&first).expect("json");
|
|
assert_eq!(first_json["type"], "session.update");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "conversation.item.input_audio_transcription.completed",
|
|
"transcript": "delegate now"
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send input transcript");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "response.output_audio_transcript.delta",
|
|
"delta": "secret context"
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send output transcript");
|
|
|
|
ws.send(Message::Text(
|
|
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>"
|
|
}]
|
|
}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send control item echo");
|
|
|
|
ws.send(Message::Text(
|
|
json!({
|
|
"type": "conversation.item.done",
|
|
"item": {
|
|
"id": "item_123",
|
|
"type": "function_call",
|
|
"name": "background_agent",
|
|
"call_id": "call_123",
|
|
"arguments": "{\"prompt\":\"delegate now\"}"
|
|
}
|
|
})
|
|
.to_string()
|
|
.into(),
|
|
))
|
|
.await
|
|
.expect("send function call");
|
|
})
|
|
.await;
|
|
|
|
let client = test_client(test_provider(format!("http://{addr}")));
|
|
let connection = client
|
|
.connect(
|
|
RealtimeSessionConfig {
|
|
instructions: "backend prompt".to_string(),
|
|
initial_items: Vec::new(),
|
|
delegation_ack_filler: None,
|
|
model: Some("realtime-test-model".to_string()),
|
|
session_id: Some("conv_123".to_string()),
|
|
event_parser: RealtimeEventParser::RealtimeV2,
|
|
session_mode: RealtimeSessionMode::Conversational,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
voice: RealtimeVoice::Marin,
|
|
},
|
|
HeaderMap::new(),
|
|
HeaderMap::new(),
|
|
)
|
|
.await
|
|
.expect("connect");
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
event,
|
|
RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone {
|
|
text: "delegate now".to_string()
|
|
})
|
|
);
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
event,
|
|
RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta {
|
|
delta: "secret context".to_string()
|
|
})
|
|
);
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert!(matches!(event, RealtimeEvent::ConversationItemAdded(_)));
|
|
|
|
let event = connection
|
|
.next_event()
|
|
.await
|
|
.expect("next event")
|
|
.expect("event");
|
|
assert_eq!(
|
|
event,
|
|
RealtimeEvent::HandoffRequested(RealtimeHandoffRequested {
|
|
handoff_id: "call_123".to_string(),
|
|
item_id: "item_123".to_string(),
|
|
input_transcript: "delegate now".to_string(),
|
|
active_transcript: vec![
|
|
RealtimeTranscriptEntry {
|
|
role: "user".to_string(),
|
|
text: "delegate now".to_string(),
|
|
},
|
|
RealtimeTranscriptEntry {
|
|
role: "assistant".to_string(),
|
|
text: "secret context".to_string(),
|
|
},
|
|
],
|
|
})
|
|
);
|
|
|
|
connection.close().await.expect("close");
|
|
server.await.expect("server task");
|
|
}
|
|
|
|
fn test_client(provider: Provider) -> RealtimeWebsocketClient {
|
|
RealtimeWebsocketClient::new(provider, HttpClientFactory::new(ReqwestDefault))
|
|
}
|