mirror of
https://github.com/openai/codex.git
synced 2026-10-24 18:18:41 +00:00
## Why A realtime session can end after more transcript has accumulated than was included in its last handoff. That tail already lives in core's active transcript state, but the stop path aborted the realtime input/fanout tasks before routing it, so the final bit of the conversation could disappear before `thread/realtime/closed`. This behavior is still being evaluated, so clients must opt in per realtime session. Omitted or false leaves shutdown behavior unchanged. ## What changed - Add optional `flushTranscriptTailOnSessionEnd` to `thread/realtime/start`, defaulting to false in app-server. - Expose an idempotent `take_transcript_tail()` from the existing active transcript state using `last_handoff_entry_count`. - When enabled, let shutdown cancel the input owner cleanly and publish at most one final existing `<realtime_delegation>` with the remaining text in `<transcript_delta>`. - Have the existing fanout drain already-parsed events before routing that final delegation, so a queued handoff wins first and is not duplicated in the tail. - Flush realtime shutdown before ordinary session task abort during core cleanup. ## Validation - `just test -p codex-app-server-protocol` - `just test -p codex-app-server realtime_conversation_stop_emits_closed_notification` - `just test -p codex-core conversation_transport_close_` - `just test -p codex-core conversation_close_routes_only_remaining_transcript_tail_once` - scoped `just fix` for `codex-app-server-protocol`, `codex-app-server`, `codex-protocol`, and `codex-core` - `just fmt`
388 lines
13 KiB
Rust
388 lines
13 KiB
Rust
use anyhow::Result;
|
|
use app_test_support::DEFAULT_CLIENT_NAME;
|
|
use app_test_support::TestAppServer;
|
|
use app_test_support::create_mock_responses_server_sequence_unchecked;
|
|
use app_test_support::to_response;
|
|
use codex_app_server_protocol::AskForApproval;
|
|
use codex_app_server_protocol::ClientInfo;
|
|
use codex_app_server_protocol::InitializeCapabilities;
|
|
use codex_app_server_protocol::JSONRPCError;
|
|
use codex_app_server_protocol::JSONRPCMessage;
|
|
use codex_app_server_protocol::JSONRPCResponse;
|
|
use codex_app_server_protocol::MockExperimentalMethodParams;
|
|
use codex_app_server_protocol::RequestId;
|
|
use codex_app_server_protocol::ThreadMemoryMode;
|
|
use codex_app_server_protocol::ThreadMemoryModeSetParams;
|
|
use codex_app_server_protocol::ThreadRealtimeStartParams;
|
|
use codex_app_server_protocol::ThreadRealtimeStartTransport;
|
|
use codex_app_server_protocol::ThreadSettingsUpdateParams;
|
|
use codex_app_server_protocol::ThreadStartParams;
|
|
use codex_app_server_protocol::ThreadStartResponse;
|
|
use codex_protocol::protocol::RealtimeOutputModality;
|
|
use pretty_assertions::assert_eq;
|
|
use std::path::Path;
|
|
use std::time::Duration;
|
|
use tempfile::TempDir;
|
|
use tokio::time::timeout;
|
|
|
|
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
#[tokio::test]
|
|
async fn mock_experimental_method_requires_experimental_api_capability() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_mock_experimental_method_request(MockExperimentalMethodParams::default())
|
|
.await?;
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "mock/experimentalMethod");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_conversation_start_requires_experimental_api_capability() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
|
|
client_managed_handoffs: None,
|
|
flush_transcript_tail_on_session_end: None,
|
|
codex_responses_as_items: None,
|
|
codex_response_item_prefix: None,
|
|
codex_response_handoff_prefix: None,
|
|
thread_id: "thr_123".to_string(),
|
|
model: None,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
include_startup_context: None,
|
|
prompt: Some(Some("hello".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(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "thread/realtime/start");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_memory_mode_set_requires_experimental_api_capability() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_memory_mode_set_request(ThreadMemoryModeSetParams {
|
|
thread_id: "thr_123".to_string(),
|
|
mode: ThreadMemoryMode::Disabled,
|
|
})
|
|
.await?;
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "thread/memoryMode/set");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_settings_update_requires_experimental_api_capability() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_settings_update_request(ThreadSettingsUpdateParams {
|
|
thread_id: "thr_123".to_string(),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "thread/settings/update");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn realtime_webrtc_start_requires_experimental_api_capability() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
|
|
client_managed_handoffs: None,
|
|
flush_transcript_tail_on_session_end: None,
|
|
codex_responses_as_items: None,
|
|
codex_response_item_prefix: None,
|
|
codex_response_handoff_prefix: None,
|
|
thread_id: "thr_123".to_string(),
|
|
model: None,
|
|
output_modality: RealtimeOutputModality::Audio,
|
|
include_startup_context: None,
|
|
prompt: Some(Some("hello".to_string())),
|
|
realtime_session_id: None,
|
|
transport: Some(ThreadRealtimeStartTransport::Webrtc {
|
|
sdp: "v=offer\r\n".to_string(),
|
|
}),
|
|
version: None,
|
|
voice: None,
|
|
})
|
|
.await?;
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "thread/realtime/start");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_start_mock_field_requires_experimental_api_capability() -> Result<()> {
|
|
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_start_request(ThreadStartParams {
|
|
mock_experimental_field: Some("mock".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "thread/start.mockExperimentalField");
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_start_without_dynamic_tools_allows_without_experimental_api_capability()
|
|
-> Result<()> {
|
|
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_start_request(ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let response: JSONRPCResponse = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
let _: ThreadStartResponse = to_response(response)?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_start_granular_approval_policy_requires_experimental_api_capability() -> Result<()>
|
|
{
|
|
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
let init = mcp
|
|
.initialize_with_capabilities(
|
|
default_client_info(),
|
|
Some(InitializeCapabilities {
|
|
experimental_api: false,
|
|
request_attestation: false,
|
|
opt_out_notification_methods: None,
|
|
mcp_server_openai_form_elicitation: false,
|
|
}),
|
|
)
|
|
.await?;
|
|
let JSONRPCMessage::Response(_) = init else {
|
|
anyhow::bail!("expected initialize response, got {init:?}");
|
|
};
|
|
|
|
let request_id = mcp
|
|
.send_thread_start_request(ThreadStartParams {
|
|
approval_policy: Some(AskForApproval::Granular {
|
|
sandbox_approval: true,
|
|
rules: false,
|
|
skill_approval: false,
|
|
request_permissions: true,
|
|
mcp_elicitations: false,
|
|
}),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
|
|
let error = timeout(
|
|
DEFAULT_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
|
|
)
|
|
.await??;
|
|
assert_experimental_capability_error(error, "askForApproval.granular");
|
|
Ok(())
|
|
}
|
|
|
|
fn default_client_info() -> ClientInfo {
|
|
ClientInfo {
|
|
name: DEFAULT_CLIENT_NAME.to_string(),
|
|
title: None,
|
|
version: "0.1.0".to_string(),
|
|
}
|
|
}
|
|
|
|
fn assert_experimental_capability_error(error: JSONRPCError, reason: &str) {
|
|
assert_eq!(error.error.code, -32600);
|
|
assert_eq!(
|
|
error.error.message,
|
|
format!("{reason} requires experimentalApi capability")
|
|
);
|
|
assert_eq!(error.error.data, None);
|
|
}
|
|
|
|
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
|
|
let config_toml = codex_home.join("config.toml");
|
|
std::fs::write(
|
|
config_toml,
|
|
format!(
|
|
r#"
|
|
model = "mock-model"
|
|
approval_policy = "never"
|
|
sandbox_mode = "read-only"
|
|
|
|
model_provider = "mock_provider"
|
|
|
|
[model_providers.mock_provider]
|
|
name = "Mock provider for test"
|
|
base_url = "{server_uri}/v1"
|
|
wire_api = "responses"
|
|
request_max_retries = 0
|
|
stream_max_retries = 0
|
|
"#
|
|
),
|
|
)
|
|
}
|