openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into dev/realtime-connection-text-done

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-04-13 16:30:48 -07:00
parent a4320be955 d4be06adea
commit 3cd3966282
53 changed files with 1445 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
+42
View File
@@ -2722,6 +2722,23 @@
],
"type": "object"
},
"ThreadInjectItemsParams": {
"properties": {
"items": {
"description": "Raw Responses API items to append to the thread's model-visible history.",
"items": true,
"type": "array"
},
"threadId": {
"type": "string"
}
},
"required": [
"items",
"threadId"
],
"type": "object"
},
"ThreadListParams": {
"properties": {
"archived": {
@@ -3966,6 +3983,31 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/inject_items"
],
"title": "Thread/injectItemsRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadInjectItemsParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/injectItemsRequest",
"type": "object"
},
{
"properties": {
"id": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
+49
View File
@@ -578,6 +578,31 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
"id": {
"$ref": "#/definitions/v2/RequestId"
},
"method": {
"enum": [
"thread/inject_items"
],
"title": "Thread/injectItemsRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/ThreadInjectItemsParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/injectItemsRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -12889,6 +12914,30 @@
"ThreadId": {
"type": "string"
},
"ThreadInjectItemsParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"items": {
"description": "Raw Responses API items to append to the thread's model-visible history.",
"items": true,
"type": "array"
},
"threadId": {
"type": "string"
}
},
"required": [
"items",
"threadId"
],
"title": "ThreadInjectItemsParams",
"type": "object"
},
"ThreadInjectItemsResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "ThreadInjectItemsResponse",
"type": "object"
},
"ThreadItem": {
"oneOf": [
{
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
+49
View File
@@ -1160,6 +1160,31 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/inject_items"
],
"title": "Thread/injectItemsRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadInjectItemsParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/injectItemsRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -10737,6 +10762,30 @@
"ThreadId": {
"type": "string"
},
"ThreadInjectItemsParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"items": {
"description": "Raw Responses API items to append to the thread's model-visible history.",
"items": true,
"type": "array"
},
"threadId": {
"type": "string"
}
},
"required": [
"items",
"threadId"
],
"title": "ThreadInjectItemsParams",
"type": "object"
},
"ThreadInjectItemsResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "ThreadInjectItemsResponse",
"type": "object"
},
"ThreadItem": {
"oneOf": [
{
codex-rs/app-server-protocol/schema/json/v2/ThreadInjectItemsParams.json
+19
View File
@@ -0,0 +1,19 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"items": {
"description": "Raw Responses API items to append to the thread's model-visible history.",
"items": true,
"type": "array"
},
"threadId": {
"type": "string"
}
},
"required": [
"items",
"threadId"
],
"title": "ThreadInjectItemsParams",
"type": "object"
}
codex-rs/app-server-protocol/schema/json/v2/ThreadInjectItemsResponse.json
+5
View File
@@ -0,0 +1,5 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "ThreadInjectItemsResponse",
"type": "object"
}
codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/ThreadInjectItemsParams.ts
+10
View File
@@ -0,0 +1,10 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { JsonValue } from "../serde_json/JsonValue";
export type ThreadInjectItemsParams = { threadId: string,
/**
* Raw Responses API items to append to the thread's model-visible history.
*/
items: Array<JsonValue>, };
codex-rs/app-server-protocol/schema/typescript/v2/ThreadInjectItemsResponse.ts
+5
View File
@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type ThreadInjectItemsResponse = Record<string, never>;
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
+2
View File
@@ -282,6 +282,8 @@ export type { ThreadCompactStartParams } from "./ThreadCompactStartParams";
export type { ThreadCompactStartResponse } from "./ThreadCompactStartResponse";
export type { ThreadForkParams } from "./ThreadForkParams";
export type { ThreadForkResponse } from "./ThreadForkResponse";
export type { ThreadInjectItemsParams } from "./ThreadInjectItemsParams";
export type { ThreadInjectItemsResponse } from "./ThreadInjectItemsResponse";
export type { ThreadItem } from "./ThreadItem";
export type { ThreadListParams } from "./ThreadListParams";
export type { ThreadListResponse } from "./ThreadListResponse";
codex-rs/app-server-protocol/src/protocol/common.rs
+5
View File
@@ -322,6 +322,11 @@ client_request_definitions! {
params: v2::ThreadReadParams,
response: v2::ThreadReadResponse,
},
/// Append raw Responses API items to the thread history without starting a user turn.
ThreadInjectItems => "thread/inject_items" {
params: v2::ThreadInjectItemsParams,
response: v2::ThreadInjectItemsResponse,
},
SkillsList => "skills/list" {
params: v2::SkillsListParams,
response: v2::SkillsListResponse,
codex-rs/app-server-protocol/src/protocol/v2.rs
+14
View File
@@ -4268,6 +4268,20 @@ pub struct TurnStartResponse {
pub turn: Turn,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadInjectItemsParams {
pub thread_id: String,
/// Raw Responses API items to append to the thread's model-visible history.
pub items: Vec<JsonValue>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadInjectItemsResponse {}
#[derive(
Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi,
)]
codex-rs/app-server/README.md
+19
View File
@@ -151,6 +151,7 @@ Example with notification opt-out:
- `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
- `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
- `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success.
- `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. Review and manual compaction turns reject `turn/steer`.
- `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`.
- `thread/realtime/start` — start a thread-scoped realtime session (experimental); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`.
@@ -581,6 +582,24 @@ Invoke a plugin by including a UI mention token such as `@sample` in the text in
} } }
```
### Example: Inject raw history items
Use `thread/inject_items` to append prebuilt Responses API items to a loaded thread’s prompt history without starting a user turn. These items are persisted to the rollout and included in subsequent model requests.
```json
{ "method": "thread/inject_items", "id": 36, "params": {
"threadId": "thr_123",
"items": [
{
"type": "message",
"role": "assistant",
"content": [{ "type": "output_text", "text": "Previously computed context." }]
}
]
} }
{ "id": 36, "result": {} }
```
### Example: Start realtime with WebRTC
Use `thread/realtime/start` with `transport.type: "webrtc"` when a browser or webview owns the `RTCPeerConnection` and app-server should create the server-side realtime session. The transport `sdp` must be the offer SDP produced by `RTCPeerConnection.createOffer()`, not a hand-written or minimal SDP string.
codex-rs/app-server/src/codex_message_processor.rs
+55
View File
@@ -132,6 +132,8 @@ use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadIncrementElicitationParams;
use codex_app_server_protocol::ThreadIncrementElicitationResponse;
use codex_app_server_protocol::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadInjectItemsResponse;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
@@ -958,6 +960,10 @@ impl CodexMessageProcessor {
)
.await;
}
ClientRequest::ThreadInjectItems { request_id, params } => {
self.thread_inject_items(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::TurnSteer { request_id, params } => {
self.turn_steer(to_connection_request_id(request_id), params)
.await;
@@ -6986,6 +6992,55 @@ impl CodexMessageProcessor {
}
}
async fn thread_inject_items(
&self,
request_id: ConnectionRequestId,
params: ThreadInjectItemsParams,
) {
let (_, thread) = match self.load_thread(&params.thread_id).await {
Ok(value) => value,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let items = match params
.items
.into_iter()
.enumerate()
.map(|(index, value)| {
serde_json::from_value::<ResponseItem>(value)
.map_err(|err| format!("items[{index}] is not a valid response item: {err}"))
})
.collect::<std::result::Result<Vec<_>, _>>()
{
Ok(items) => items,
Err(message) => {
self.send_invalid_request_error(request_id, message).await;
return;
}
};
match thread.inject_response_items(items).await {
Ok(()) => {
self.outgoing
.send_response(request_id, ThreadInjectItemsResponse {})
.await;
}
Err(CodexErr::InvalidRequest(message)) => {
self.send_invalid_request_error(request_id, message).await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to inject response items: {err}"),
)
.await;
}
}
}
async fn set_app_server_client_info(
thread: &CodexThread,
app_server_client_name: Option<String>,
codex-rs/app-server/tests/common/mcp_process.rs
+10
View File
@@ -62,6 +62,7 @@ use codex_app_server_protocol::SkillsListParams;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadCompactStartParams;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadLoadedListParams;
use codex_app_server_protocol::ThreadMemoryModeSetParams;
@@ -602,6 +603,15 @@ impl McpProcess {
self.send_request("turn/start", params).await
}
/// Send a `thread/inject_items` JSON-RPC request (v2).
pub async fn send_thread_inject_items_request(
&mut self,
params: ThreadInjectItemsParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/inject_items", params).await
}
/// Send a `command/exec` JSON-RPC request (v2).
pub async fn send_command_exec_request(
&mut self,
codex-rs/app-server/tests/suite/v2/mod.rs
+1
View File
@@ -35,6 +35,7 @@ mod safety_check_downgrade;
mod skills_list;
mod thread_archive;
mod thread_fork;
mod thread_inject_items;
mod thread_list;
mod thread_loaded_list;
mod thread_memory_mode_set;
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+1 -1
View File
@@ -1111,7 +1111,7 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
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 = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"background_agent","description":"Send a user request to the background agent. Use this as the default action. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to the background agent."}},"required":["prompt"],"additionalProperties":false}}]}"#;
let session = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"background_agent","description":"Send a user request to the background agent. Use this as the default action. Do not rephrase the user's ask or rewrite it in your own words; pass along the user's own words. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to the background agent."}},"required":["prompt"],"additionalProperties":false}}]}"#;
assert_eq!(
body,
format!(
codex-rs/app-server/tests/suite/v2/thread_inject_items.rs
+288
View File
@@ -0,0 +1,288 @@
use anyhow::Context;
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadInjectItemsResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::RolloutRecorder;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::RolloutItem;
use core_test_support::responses;
use serde_json::Value;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Result<()> {
let server = responses::start_mock_server().await;
let body = responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-1"),
]);
let response_mock = responses::mount_sse_once(&server, body).await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_req = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let thread_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
let injected_text = "Injected assistant context";
let injected_item = ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: injected_text.to_string(),
}],
end_turn: None,
phase: None,
};
let inject_req = mcp
.send_thread_inject_items_request(ThreadInjectItemsParams {
thread_id: thread.id.clone(),
items: vec![serde_json::to_value(&injected_item)?],
})
.await?;
let inject_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(inject_req)),
)
.await??;
let _response: ThreadInjectItemsResponse =
to_response::<ThreadInjectItemsResponse>(inject_resp)?;
let rollout_path = thread.path.as_ref().context("thread path missing")?;
let history = RolloutRecorder::get_rollout_history(rollout_path).await?;
let InitialHistory::Resumed(resumed_history) = history else {
panic!("expected resumed rollout history");
};
assert!(
resumed_history
.history
.iter()
.any(|item| matches!(item, RolloutItem::ResponseItem(response_item) if response_item == &injected_item)),
"injected item should be persisted in rollout history"
);
let turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let injected_value = serde_json::to_value(&injected_item)?;
let model_input = response_mock.single_request().input();
let environment_context_index =
response_item_text_position(&model_input, "<environment_context>")
.expect("environment context should be injected before the first user turn");
let injected_index = model_input
.iter()
.position(|item| item == &injected_value)
.expect("injected item should be sent in the next model request");
let user_prompt_index = response_item_text_position(&model_input, "Hello")
.expect("user prompt should be sent in the next model request");
assert!(
environment_context_index < injected_index,
"standard initial context should be sent before injected items"
);
assert!(
injected_index < user_prompt_index,
"injected items should be sent before the user prompt"
);
Ok(())
}
#[tokio::test]
async fn thread_inject_items_adds_raw_response_items_after_a_turn() -> Result<()> {
let server = responses::start_mock_server().await;
let first_body = responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "First done"),
responses::ev_completed("resp-1"),
]);
let second_body = responses::sse(vec![
responses::ev_response_created("resp-2"),
responses::ev_assistant_message("msg-2", "Second done"),
responses::ev_completed("resp-2"),
]);
let response_mock = responses::mount_sse_sequence(&server, vec![first_body, second_body]).await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_req = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let thread_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
let first_turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "First turn".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(first_turn_req)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let injected_item = ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: "Injected after first turn".to_string(),
}],
end_turn: None,
phase: None,
};
let injected_value = serde_json::to_value(&injected_item)?;
let inject_req = mcp
.send_thread_inject_items_request(ThreadInjectItemsParams {
thread_id: thread.id.clone(),
items: vec![injected_value.clone()],
})
.await?;
let inject_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(inject_req)),
)
.await??;
let _response: ThreadInjectItemsResponse =
to_response::<ThreadInjectItemsResponse>(inject_resp)?;
let second_turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "Second turn".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(second_turn_req)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
assert!(
!requests[0].input().contains(&injected_value),
"injected item should not be sent before it is injected"
);
assert!(
requests[1].input().contains(&injected_value),
"injected item should be sent after being injected into existing history"
);
Ok(())
}
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
"#
),
)
}
fn response_item_text_position(items: &[Value], needle: &str) -> Option<usize> {
items.iter().position(|item| {
item.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
.any(|content| {
content
.get("text")
.and_then(Value::as_str)
.is_some_and(|text| text.contains(needle))
})
})
}
codex-rs/cli/src/mcp_cmd.rs
+1
View File
@@ -299,6 +299,7 @@ async fn run_add(config_overrides: &CliConfigOverrides, add_args: AddArgs) -> Re
transport: transport.clone(),
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
+1 -1
View File
@@ -30,7 +30,7 @@ const REALTIME_V2_OUTPUT_MODALITY_AUDIO: &str = "audio";
const REALTIME_V2_OUTPUT_MODALITY_TEXT: &str = "text";
const REALTIME_V2_TOOL_CHOICE: &str = "auto";
const REALTIME_V2_BACKGROUND_AGENT_TOOL_NAME: &str = "background_agent";
const REALTIME_V2_BACKGROUND_AGENT_TOOL_DESCRIPTION: &str = "Send a user request to the background agent. Use this as the default action. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
const REALTIME_V2_BACKGROUND_AGENT_TOOL_DESCRIPTION: &str = "Send a user request to the background agent. Use this as the default action. Do not rephrase the user's ask or rewrite it in your own words; pass along the user's own words. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage {
RealtimeOutboundMessage::ConversationItemCreate {
codex-rs/codex-mcp/src/mcp/mod.rs
+1
View File
@@ -272,6 +272,7 @@ fn codex_apps_mcp_server_config(config: &McpConfig, auth: Option<&CodexAuth>) ->
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(30)),
tool_timeout_sec: None,
codex-rs/codex-mcp/src/mcp/mod_tests.rs
+2
View File
@@ -195,6 +195,7 @@ async fn effective_mcp_servers_preserve_user_servers_and_add_codex_apps() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -216,6 +217,7 @@ async fn effective_mcp_servers_preserve_user_servers_and_add_codex_apps() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/codex-mcp/src/mcp/skill_dependencies.rs
+2
View File
@@ -121,6 +121,7 @@ fn mcp_dependency_to_server_config(
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -147,6 +148,7 @@ fn mcp_dependency_to_server_config(
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/codex-mcp/src/mcp/skill_dependencies_tests.rs
+2
View File
@@ -41,6 +41,7 @@ fn collect_missing_respects_canonical_installed_key() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -91,6 +92,7 @@ fn collect_missing_dedupes_by_canonical_key_but_preserves_original_name() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/codex-mcp/src/mcp_connection_manager_tests.rs
+2
View File
@@ -757,6 +757,7 @@ fn mcp_init_error_display_prompts_for_github_pat() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -806,6 +807,7 @@ fn mcp_init_error_display_reports_generic_errors() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/config/src/mcp_edit.rs
+3
View File
@@ -177,6 +177,9 @@ fn serialize_mcp_server(config: &McpServerConfig) -> TomlItem {
if config.required {
entry["required"] = value(true);
}
if config.supports_parallel_tool_calls {
entry["supports_parallel_tool_calls"] = value(true);
}
if let Some(timeout) = config.startup_timeout_sec {
entry["startup_timeout_sec"] = value(timeout.as_secs_f64());
}
codex-rs/config/src/mcp_edit_tests.rs
+2
View File
@@ -24,6 +24,7 @@ async fn replace_mcp_servers_serializes_per_tool_approval_overrides() -> anyhow:
},
enabled: true,
required: false,
supports_parallel_tool_calls: true,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -59,6 +60,7 @@ async fn replace_mcp_servers_serializes_per_tool_approval_overrides() -> anyhow:
serialized,
r#"[mcp_servers.docs]
command = "docs-server"
supports_parallel_tool_calls = true
[mcp_servers.docs.tools]
codex-rs/config/src/mcp_types.rs
+8
View File
@@ -69,6 +69,10 @@ pub struct McpServerConfig {
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub required: bool,
/// When `true`, every tool from this server is advertised as safe for parallel tool calls.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub supports_parallel_tool_calls: bool,
/// Reason this server was disabled after applying requirements.
#[serde(skip)]
pub disabled_reason: Option<McpServerDisabledReason>,
@@ -146,6 +150,8 @@ pub struct RawMcpServerConfig {
#[serde(default)]
pub required: Option<bool>,
#[serde(default)]
pub supports_parallel_tool_calls: Option<bool>,
#[serde(default)]
pub enabled_tools: Option<Vec<String>>,
#[serde(default)]
pub disabled_tools: Option<Vec<String>>,
@@ -180,6 +186,7 @@ impl TryFrom<RawMcpServerConfig> for McpServerConfig {
tool_timeout_sec,
enabled,
required,
supports_parallel_tool_calls,
enabled_tools,
disabled_tools,
scopes,
@@ -243,6 +250,7 @@ impl TryFrom<RawMcpServerConfig> for McpServerConfig {
tool_timeout_sec,
enabled: enabled.unwrap_or_else(default_enabled),
required: required.unwrap_or_default(),
supports_parallel_tool_calls: supports_parallel_tool_calls.unwrap_or_default(),
disabled_reason: None,
enabled_tools,
disabled_tools,
codex-rs/config/src/mcp_types_tests.rs
+33
View File
@@ -245,6 +245,38 @@ fn deserialize_server_config_with_tool_filters() {
assert_eq!(cfg.disabled_tools, Some(vec!["blocked".to_string()]));
}
#[test]
fn deserialize_server_config_with_parallel_tool_calls() {
let cfg: McpServerConfig = toml::from_str(
r#"
command = "echo"
supports_parallel_tool_calls = true
"#,
)
.expect("should deserialize supports_parallel_tool_calls");
assert!(cfg.supports_parallel_tool_calls);
}
#[test]
fn serialize_round_trips_server_config_with_parallel_tool_calls() {
let cfg: McpServerConfig = toml::from_str(
r#"
command = "echo"
supports_parallel_tool_calls = true
tool_timeout_sec = 2.0
"#,
)
.expect("should deserialize supports_parallel_tool_calls");
let serialized = toml::to_string(&cfg).expect("should serialize MCP config");
assert!(serialized.contains("supports_parallel_tool_calls = true"));
let round_tripped: McpServerConfig =
toml::from_str(&serialized).expect("should deserialize serialized MCP config");
assert_eq!(round_tripped, cfg);
}
#[test]
fn deserialize_ignores_unknown_server_fields() {
let cfg: McpServerConfig = toml::from_str(
@@ -267,6 +299,7 @@ fn deserialize_ignores_unknown_server_fields() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/core/config.schema.json
+4
View File
@@ -1522,6 +1522,10 @@
"format": "double",
"type": "number"
},
"supports_parallel_tool_calls": {
"default": null,
"type": "boolean"
},
"tool_timeout_sec": {
"default": null,
"format": "double",
codex-rs/core/src/codex.rs
+20 -1
View File
@@ -4904,6 +4904,8 @@ mod handlers {
use crate::config_loader::CloudRequirementsLoader;
use crate::config_loader::LoaderOverrides;
use crate::config_loader::load_config_layers_state;
use crate::realtime_context::REALTIME_TURN_TOKEN_BUDGET;
use crate::realtime_context::truncate_realtime_text_to_token_budget;
use codex_features::Feature;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -5124,6 +5126,10 @@ mod handlers {
if text.is_empty() {
return;
}
let text = truncate_realtime_text_to_token_budget(&text, REALTIME_TURN_TOKEN_BUDGET);
if text.is_empty() {
return;
}
if sess.conversation.running_state().await.is_none() {
return;
}
@@ -7127,11 +7133,24 @@ pub(crate) async fn built_tools(
);
let direct_mcp_tools = has_mcp_servers.then_some(mcp_tool_exposure.direct_tools);
let parallel_mcp_server_names = turn_context
.config
.mcp_servers
.get()
.iter()
.filter_map(|(server_name, server_config)| {
server_config
.supports_parallel_tool_calls
.then_some(server_name.clone())
})
.collect::<HashSet<_>>();
Ok(Arc::new(ToolRouter::from_config(
&turn_context.tools_config,
ToolRouterParams {
deferred_mcp_tools: mcp_tool_exposure.deferred_tools,
mcp_tools: direct_mcp_tools,
deferred_mcp_tools: mcp_tool_exposure.deferred_tools,
parallel_mcp_server_names,
discoverable_tools,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
},
codex-rs/core/src/codex_tests.rs
+2
View File
@@ -312,6 +312,7 @@ fn test_tool_runtime(session: Arc<Session>, turn_context: Arc<TurnContext>) -> T
crate::tools::router::ToolRouterParams {
mcp_tools: None,
deferred_mcp_tools: None,
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
},
@@ -5353,6 +5354,7 @@ async fn fatal_tool_error_stops_turn_and_reports_error() {
crate::tools::router::ToolRouterParams {
deferred_mcp_tools,
mcp_tools: Some(tools),
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
},
codex-rs/core/src/codex_thread.rs
+23
View File
@@ -199,6 +199,29 @@ impl CodexThread {
Ok(submission_id)
}
/// Append raw Responses API items to the thread's model-visible history.
pub async fn inject_response_items(&self, items: Vec<ResponseItem>) -> CodexResult<()> {
if items.is_empty() {
return Err(CodexErr::InvalidRequest(
"items must not be empty".to_string(),
));
}
let turn_context = self.codex.session.new_default_turn().await;
if self.codex.session.reference_context_item().await.is_none() {
self.codex
.session
.record_context_updates_and_set_reference_context_item(turn_context.as_ref())
.await;
}
self.codex
.session
.record_conversation_items(turn_context.as_ref(), &items)
.await;
self.codex.session.flush_rollout().await?;
Ok(())
}
pub fn rollout_path(&self) -> Option<PathBuf> {
self.rollout_path.clone()
}
codex-rs/core/src/config/config_tests.rs
+16
View File
@@ -83,6 +83,7 @@ fn stdio_mcp(command: &str) -> McpServerConfig {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -104,6 +105,7 @@ fn http_mcp(url: &str) -> McpServerConfig {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2011,6 +2013,7 @@ async fn replace_mcp_servers_round_trips_entries() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(3)),
tool_timeout_sec: Some(Duration::from_secs(5)),
@@ -2257,6 +2260,7 @@ async fn replace_mcp_servers_serializes_env_sorted() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2330,6 +2334,7 @@ async fn replace_mcp_servers_serializes_env_vars() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2383,6 +2388,7 @@ async fn replace_mcp_servers_serializes_cwd() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2434,6 +2440,7 @@ async fn replace_mcp_servers_streamable_http_serializes_bearer_token() -> anyhow
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(2)),
tool_timeout_sec: None,
@@ -2501,6 +2508,7 @@ async fn replace_mcp_servers_streamable_http_serializes_custom_headers() -> anyh
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(2)),
tool_timeout_sec: None,
@@ -2580,6 +2588,7 @@ async fn replace_mcp_servers_streamable_http_removes_optional_sections() -> anyh
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(2)),
tool_timeout_sec: None,
@@ -2612,6 +2621,7 @@ async fn replace_mcp_servers_streamable_http_removes_optional_sections() -> anyh
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2679,6 +2689,7 @@ async fn replace_mcp_servers_streamable_http_isolates_headers_between_servers()
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(2)),
tool_timeout_sec: None,
@@ -2701,6 +2712,7 @@ async fn replace_mcp_servers_streamable_http_isolates_headers_between_servers()
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2786,6 +2798,7 @@ async fn replace_mcp_servers_serializes_disabled_flag() -> anyhow::Result<()> {
},
enabled: false,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2833,6 +2846,7 @@ async fn replace_mcp_servers_serializes_required_flag() -> anyhow::Result<()> {
},
enabled: true,
required: true,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2880,6 +2894,7 @@ async fn replace_mcp_servers_serializes_tool_filters() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -2931,6 +2946,7 @@ async fn replace_mcp_servers_streamable_http_serializes_oauth_resource() -> anyh
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/core/src/config/edit.rs
+3
View File
@@ -225,6 +225,9 @@ mod document_helpers {
if config.required {
entry["required"] = value(true);
}
if config.supports_parallel_tool_calls {
entry["supports_parallel_tool_calls"] = value(true);
}
if let Some(timeout) = config.startup_timeout_sec {
entry["startup_timeout_sec"] = value(timeout.as_secs_f64());
}
codex-rs/core/src/config/edit_tests.rs
+8
View File
@@ -577,6 +577,7 @@ fn blocking_replace_mcp_servers_round_trips() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: true,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -603,6 +604,7 @@ fn blocking_replace_mcp_servers_round_trips() {
},
enabled: false,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(std::time::Duration::from_secs(5)),
tool_timeout_sec: None,
@@ -638,6 +640,7 @@ Z-Header = \"z\"
command = \"cmd\"
args = [\"--flag\"]
env_vars = [\"FOO\"]
supports_parallel_tool_calls = true
enabled_tools = [\"one\", \"two\"]
[mcp_servers.stdio.env]
@@ -665,6 +668,7 @@ fn blocking_replace_mcp_servers_serializes_tool_approval_overrides() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -725,6 +729,7 @@ foo = { command = "cmd" }
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -776,6 +781,7 @@ foo = { command = "cmd" } # keep me
},
enabled: false,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -826,6 +832,7 @@ foo = { command = "cmd", args = ["--flag"] } # keep me
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -877,6 +884,7 @@ foo = { command = "cmd" }
},
enabled: false,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/core/src/mcp_skill_dependencies.rs
+2
View File
@@ -364,6 +364,7 @@ fn mcp_dependency_to_server_config(
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -390,6 +391,7 @@ fn mcp_dependency_to_server_config(
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/core/src/plugins/manager_tests.rs
+4
View File
@@ -179,6 +179,7 @@ fn load_plugins_loads_default_skills_and_mcp_servers() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -510,6 +511,7 @@ fn load_plugins_uses_manifest_configured_component_paths() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -617,6 +619,7 @@ fn load_plugins_ignores_manifest_component_paths_without_dot_slash() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
@@ -772,6 +775,7 @@ fn capability_index_filters_inactive_and_zero_capability_plugins() {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
codex-rs/core/src/realtime_context.rs
+28 -24
View File
@@ -27,7 +27,7 @@ const CURRENT_THREAD_SECTION_TOKEN_BUDGET: usize = 1_200;
const RECENT_WORK_SECTION_TOKEN_BUDGET: usize = 2_200;
const WORKSPACE_SECTION_TOKEN_BUDGET: usize = 1_600;
const NOTES_SECTION_TOKEN_BUDGET: usize = 300;
const CURRENT_THREAD_TURN_TOKEN_BUDGET: usize = 300;
pub(crate) const REALTIME_TURN_TOKEN_BUDGET: usize = 300;
const MAX_RECENT_THREADS: usize = 40;
const MAX_RECENT_WORK_GROUPS: usize = 8;
const MAX_CURRENT_CWD_ASKS: usize = 8;
@@ -262,30 +262,9 @@ fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
turn_lines.push(assistant_messages.join("\n\n"));
}
let turn_budget = CURRENT_THREAD_TURN_TOKEN_BUDGET.min(remaining_budget);
let turn_budget = REALTIME_TURN_TOKEN_BUDGET.min(remaining_budget);
let turn_text = turn_lines.join("\n");
let mut truncation_budget = turn_budget;
let turn_text = loop {
let candidate = truncate_text(&turn_text, TruncationPolicy::Tokens(truncation_budget));
let candidate_tokens = approx_token_count(&candidate);
if candidate_tokens <= turn_budget {
break candidate;
}
// The shared truncator adds its marker after choosing preserved
// content, so tighten the content budget until the rendered turn
// itself fits the per-turn cap.
let excess_tokens = candidate_tokens.saturating_sub(turn_budget);
let next_budget = truncation_budget.saturating_sub(excess_tokens.max(1));
if next_budget == 0 {
let candidate = truncate_text(&turn_text, TruncationPolicy::Tokens(0));
if approx_token_count(&candidate) <= turn_budget {
break candidate;
}
break String::new();
}
truncation_budget = next_budget;
};
let turn_text = truncate_realtime_text_to_token_budget(&turn_text, turn_budget);
let turn_tokens = approx_token_count(&turn_text);
if turn_tokens == 0 {
continue;
@@ -300,6 +279,31 @@ fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
(retained_turn_count > 0).then(|| lines.join("\n"))
}
pub(crate) fn truncate_realtime_text_to_token_budget(text: &str, budget_tokens: usize) -> String {
let mut truncation_budget = budget_tokens;
loop {
let candidate = truncate_text(text, TruncationPolicy::Tokens(truncation_budget));
let candidate_tokens = approx_token_count(&candidate);
if candidate_tokens <= budget_tokens {
break candidate;
}
// The shared truncator adds its marker after choosing preserved
// content, so tighten the content budget until the rendered turn
// itself fits the per-turn cap.
let excess_tokens = candidate_tokens.saturating_sub(budget_tokens);
let next_budget = truncation_budget.saturating_sub(excess_tokens.max(1));
if next_budget == 0 {
let candidate = truncate_text(text, TruncationPolicy::Tokens(0));
if approx_token_count(&candidate) <= budget_tokens {
break candidate;
}
break String::new();
}
truncation_budget = next_budget;
}
}
fn build_workspace_section_with_user_root(
cwd: &Path,
user_root: Option<PathBuf>,
codex-rs/core/src/tools/code_mode/mod.rs
+16 -2
View File
@@ -2,6 +2,7 @@ mod execute_handler;
mod response_adapter;
mod wait_handler;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
@@ -251,7 +252,7 @@ pub(super) async fn build_enabled_tools(
async fn build_nested_router(exec: &ExecContext) -> ToolRouter {
let nested_tools_config = exec.turn.tools_config.for_code_mode_nested_tools();
let mcp_tools = exec
let listed_mcp_tools = exec
.session
.services
.mcp_connection_manager
@@ -259,12 +260,25 @@ async fn build_nested_router(exec: &ExecContext) -> ToolRouter {
.await
.list_all_tools()
.await;
let parallel_mcp_server_names = exec
.turn
.config
.mcp_servers
.get()
.iter()
.filter_map(|(server_name, server_config)| {
server_config
.supports_parallel_tool_calls
.then_some(server_name.clone())
})
.collect::<HashSet<_>>();
ToolRouter::from_config(
&nested_tools_config,
ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: Some(mcp_tools),
mcp_tools: Some(listed_mcp_tools),
parallel_mcp_server_names,
discoverable_tools: None,
dynamic_tools: exec.turn.dynamic_tools.as_slice(),
},
codex-rs/core/src/tools/js_repl/mod.rs
+3 -1
View File
@@ -1561,12 +1561,14 @@ impl JsReplManager {
.await
.list_all_tools()
.await;
let router = ToolRouter::from_config(
&exec.turn.tools_config,
crate::tools::router::ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: Some(mcp_tools),
// JS REPL dispatches nested tool calls directly, not through
// `ToolCallRuntime`'s parallel scheduling lock.
parallel_mcp_server_names: std::collections::HashSet::new(),
discoverable_tools: None,
dynamic_tools: exec.turn.dynamic_tools.as_slice(),
},
codex-rs/core/src/tools/parallel.rs
+1 -1
View File
@@ -78,7 +78,7 @@ impl ToolCallRuntime {
source: ToolCallSource,
cancellation_token: CancellationToken,
) -> impl std::future::Future<Output = Result<AnyToolResult, FunctionCallError>> {
let supports_parallel = self.router.tool_supports_parallel(&call.tool_name);
let supports_parallel = self.router.tool_supports_parallel(&call);
let router = Arc::clone(&self.router);
let session = Arc::clone(&self.session);
let turn = Arc::clone(&self.turn_context);
codex-rs/core/src/tools/router.rs
+16 -1
View File
@@ -20,6 +20,7 @@ use codex_tools::ToolName;
use codex_tools::ToolSpec;
use codex_tools::ToolsConfig;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use tracing::instrument;
@@ -36,11 +37,13 @@ pub struct ToolRouter {
registry: ToolRegistry,
specs: Vec<ConfiguredToolSpec>,
model_visible_specs: Vec<ToolSpec>,
parallel_mcp_server_names: HashSet<String>,
}
pub(crate) struct ToolRouterParams<'a> {
pub(crate) mcp_tools: Option<HashMap<String, ToolInfo>>,
pub(crate) deferred_mcp_tools: Option<HashMap<String, ToolInfo>>,
pub(crate) parallel_mcp_server_names: HashSet<String>,
pub(crate) discoverable_tools: Option<Vec<DiscoverableTool>>,
pub(crate) dynamic_tools: &'a [DynamicToolSpec],
}
@@ -50,6 +53,7 @@ impl ToolRouter {
let ToolRouterParams {
mcp_tools,
deferred_mcp_tools,
parallel_mcp_server_names,
discoverable_tools,
dynamic_tools,
} = params;
@@ -83,6 +87,7 @@ impl ToolRouter {
registry,
specs,
model_visible_specs,
parallel_mcp_server_names,
}
}
@@ -104,7 +109,7 @@ impl ToolRouter {
.map(|config| config.spec.clone())
}
pub fn tool_supports_parallel(&self, tool_name: &ToolName) -> bool {
fn configured_tool_supports_parallel(&self, tool_name: &ToolName) -> bool {
tool_name.namespace.is_none()
&& self
.specs
@@ -113,6 +118,16 @@ impl ToolRouter {
.any(|config| config.name() == tool_name.name.as_str())
}
pub fn tool_supports_parallel(&self, call: &ToolCall) -> bool {
match &call.payload {
// MCP parallel support is configured per server, including for deferred
// tools that may not have a matching spec entry. Use the parsed payload
// server so similarly named servers/tools cannot collide.
ToolPayload::Mcp { server, .. } => self.parallel_mcp_server_names.contains(server),
_ => self.configured_tool_supports_parallel(&call.tool_name),
}
}
#[instrument(level = "trace", skip_all, err)]
pub async fn build_tool_call(
session: &Session,
codex-rs/core/src/tools/router_tests.rs
+60 -4
View File
@@ -1,3 +1,4 @@
use std::collections::HashSet;
use std::sync::Arc;
use crate::codex::make_session_and_context;
@@ -32,6 +33,7 @@ async fn js_repl_tools_only_blocks_direct_tool_calls() -> anyhow::Result<()> {
ToolRouterParams {
deferred_mcp_tools,
mcp_tools: Some(mcp_tools),
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
},
@@ -84,6 +86,7 @@ async fn js_repl_tools_only_allows_js_repl_source_calls() -> anyhow::Result<()>
ToolRouterParams {
deferred_mcp_tools,
mcp_tools: Some(mcp_tools),
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
},
@@ -129,6 +132,7 @@ async fn js_repl_tools_only_blocks_namespaced_js_repl_tool() -> anyhow::Result<(
ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: None,
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
},
@@ -178,6 +182,7 @@ async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow
ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: Some(mcp_tools),
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
},
@@ -185,12 +190,24 @@ async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow
let parallel_tool_name = ["shell", "local_shell", "exec_command", "shell_command"]
.into_iter()
.find(|name| router.tool_supports_parallel(&ToolName::plain(*name)))
.find(|name| {
router.tool_supports_parallel(&ToolCall {
tool_name: ToolName::plain(*name),
call_id: "call-parallel-tool".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
})
})
.expect("test session should expose a parallel shell-like tool");
assert!(
!router.tool_supports_parallel(&ToolName::namespaced("mcp__server__", parallel_tool_name))
);
assert!(!router.tool_supports_parallel(&ToolCall {
tool_name: ToolName::namespaced("mcp__server__", parallel_tool_name),
call_id: "call-namespaced-tool".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
}));
Ok(())
}
@@ -228,3 +245,42 @@ async fn build_tool_call_uses_namespace_for_registry_name() -> anyhow::Result<()
Ok(())
}
#[tokio::test]
async fn mcp_parallel_support_uses_exact_payload_server() -> anyhow::Result<()> {
let (_, turn) = make_session_and_context().await;
let router = ToolRouter::from_config(
&turn.tools_config,
ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: None,
parallel_mcp_server_names: HashSet::from(["echo".to_string()]),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
},
);
let deferred_call = ToolCall {
tool_name: ToolName::namespaced("mcp__echo__", "query_with_delay"),
call_id: "call-deferred".to_string(),
payload: ToolPayload::Mcp {
server: "echo".to_string(),
tool: "query_with_delay".to_string(),
raw_arguments: "{}".to_string(),
},
};
assert!(router.tool_supports_parallel(&deferred_call));
let different_server_call = ToolCall {
tool_name: ToolName::namespaced("mcp__hello_echo__", "query_with_delay"),
call_id: "call-other-server".to_string(),
payload: ToolPayload::Mcp {
server: "hello_echo".to_string(),
tool: "query_with_delay".to_string(),
raw_arguments: "{}".to_string(),
},
};
assert!(!router.tool_supports_parallel(&different_server_call));
Ok(())
}
codex-rs/core/src/tools/spec_tests.rs
+1
View File
@@ -314,6 +314,7 @@ fn assert_model_tools(
ToolRouterParams {
mcp_tools: None,
deferred_mcp_tools: None,
parallel_mcp_server_names: std::collections::HashSet::new(),
discoverable_tools: None,
dynamic_tools: &[],
},
codex-rs/core/tests/suite/code_mode.rs
+1
View File
@@ -200,6 +200,7 @@ async fn run_code_mode_turn_with_rmcp(
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
codex-rs/core/tests/suite/realtime_conversation.rs
+118
View File
@@ -26,6 +26,7 @@ use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::user_input::UserInput;
use codex_utils_output_truncation::approx_token_count;
use core_test_support::responses;
use core_test_support::responses::WebSocketConnectionConfig;
use core_test_support::responses::start_mock_server;
@@ -1934,6 +1935,123 @@ async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_user_text_turn_is_capped_when_mirrored_to_realtime() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = start_mock_server().await;
let response_mock = responses::mount_sse_once(
&api_server,
responses::sse(vec![
responses::ev_response_created("resp_long_user_text"),
responses::ev_assistant_message("msg_long_user_text", "ack"),
responses::ev_completed("resp_long_user_text"),
]),
)
.await;
let realtime_server = start_websocket_server(vec![vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_long_user_text", "instructions": "backend prompt" }
})],
vec![],
]])
.await;
let mut builder = test_codex().with_config({
let realtime_base_url = realtime_server.uri().to_string();
move |config| {
config.experimental_realtime_ws_base_url = Some(realtime_base_url);
config.experimental_realtime_ws_startup_context = Some(String::new());
}
});
let test = builder.build(&api_server).await?;
// Phase 1: start realtime so the next normal user turn mirrors over the
// active WebSocket session.
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
prompt: Some(Some("backend prompt".to_string())),
session_id: None,
transport: None,
voice: None,
}))
.await?;
let session_updated = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::SessionUpdated { session_id, .. },
}) => Some(session_id.clone()),
_ => None,
})
.await;
assert_eq!(session_updated, "sess_long_user_text");
// Phase 2: submit one oversized text turn. The model request should keep
// the exact user text, while the realtime mirror should get the capped copy.
let user_text = format!(
"mirror-head {} mirror-middle {} mirror-tail",
"alpha ".repeat(900),
"omega ".repeat(900),
);
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: user_text.clone(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
// Phase 3: capture the mirrored WebSocket item; the snapshot below records
// the capped payload shape.
let realtime_text_request = wait_for_matching_websocket_request(
&realtime_server,
"capped normal user turn text mirrored to realtime",
|request| websocket_request_text(request).is_some_and(|text| text.contains("mirror-head")),
)
.await;
let realtime_text =
websocket_request_text(&realtime_text_request).expect("realtime request text");
let model_user_texts = response_mock.single_request().message_input_texts("user");
let realtime_request_body = realtime_text_request.body_json();
let content = &realtime_request_body["item"]["content"][0];
// Snapshot the request envelope and capped text together so reviewers can
// see the preserved head/tail and truncation marker in one place.
let snapshot = format!(
"type: {}\nitem.type: {}\nitem.role: {}\ncontent[0].type: {}\nmodel_has_full_user_text: {}\nrealtime_text_equal_full_user_text: {}\nrealtime_text_approx_tokens: {}\ncontent[0].text: {}",
realtime_request_body["type"].as_str().unwrap_or_default(),
realtime_request_body["item"]["type"]
.as_str()
.unwrap_or_default(),
realtime_request_body["item"]["role"]
.as_str()
.unwrap_or_default(),
content["type"].as_str().unwrap_or_default(),
model_user_texts.iter().any(|text| text == &user_text),
realtime_text == user_text,
approx_token_count(&realtime_text),
realtime_text,
);
insta::assert_snapshot!(
"conversation_user_text_turn_is_capped_when_mirrored_to_realtime",
snapshot
);
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/rmcp_client.rs
+269
View File
@@ -75,6 +75,12 @@ fn assert_wall_time_header(output: &str) {
assert_eq!(marker, "Output:");
}
#[derive(Debug, PartialEq, Eq)]
enum McpCallEvent {
Begin(String),
End(String),
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial(mcp_test_value)]
async fn stdio_server_round_trip() -> anyhow::Result<()> {
@@ -125,6 +131,7 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -230,6 +237,263 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stdio_mcp_parallel_tool_calls_default_false_runs_serially() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let first_call_id = "sync-serial-1";
let second_call_id = "sync-serial-2";
let server_name = "rmcp";
let tool_name = format!("mcp__{server_name}__sync");
let args = json!({ "sleep_after_ms": 100 }).to_string();
mount_sse_once(
&server,
responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_function_call(first_call_id, &tool_name, &args),
responses::ev_function_call(second_call_id, &tool_name, &args),
responses::ev_completed("resp-1"),
]),
)
.await;
let final_mock = mount_sse_once(
&server,
responses::sse(vec![
responses::ev_assistant_message("msg-1", "rmcp sync tools completed successfully."),
responses::ev_completed("resp-2"),
]),
)
.await;
let rmcp_test_server_bin = stdio_server_bin()?;
let fixture = test_codex()
.with_config(move |config| {
let mut servers = config.mcp_servers.get().clone();
servers.insert(
server_name.to_string(),
McpServerConfig {
transport: McpServerTransportConfig::Stdio {
command: rmcp_test_server_bin,
args: Vec::new(),
env: None,
env_vars: Vec::new(),
cwd: None,
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: Some(Duration::from_secs(2)),
enabled_tools: None,
disabled_tools: None,
scopes: None,
oauth_resource: None,
tools: HashMap::new(),
},
);
config
.mcp_servers
.set(servers)
.expect("test mcp servers should accept any configuration");
})
.build(&server)
.await?;
let session_model = fixture.session_configured.model.clone();
fixture
.codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "call the rmcp sync tool twice".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
cwd: fixture.cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
approvals_reviewer: None,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
model: session_model,
effort: None,
summary: None,
service_tier: None,
collaboration_mode: None,
personality: None,
})
.await?;
let mut call_events = Vec::new();
while call_events.len() < 4 {
let event = wait_for_event(&fixture.codex, |ev| {
matches!(
ev,
EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_)
)
})
.await;
match event {
EventMsg::McpToolCallBegin(begin) => {
call_events.push(McpCallEvent::Begin(begin.call_id));
}
EventMsg::McpToolCallEnd(end) => {
call_events.push(McpCallEvent::End(end.call_id));
}
_ => unreachable!("event guard guarantees MCP call events"),
}
}
let event_index = |needle: McpCallEvent| {
call_events
.iter()
.position(|event| event == &needle)
.expect("expected MCP call event")
};
let first_begin = event_index(McpCallEvent::Begin(first_call_id.to_string()));
let first_end = event_index(McpCallEvent::End(first_call_id.to_string()));
let second_begin = event_index(McpCallEvent::Begin(second_call_id.to_string()));
let second_end = event_index(McpCallEvent::End(second_call_id.to_string()));
assert!(
first_end < second_begin || second_end < first_begin,
"default MCP tool calls should run serially; saw events: {call_events:?}"
);
wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = final_mock.single_request();
for call_id in [first_call_id, second_call_id] {
let output_text = request
.function_call_output_text(call_id)
.expect("function_call_output present for rmcp sync call");
let wrapped_payload = split_wall_time_wrapped_output(&output_text);
let output_json: Value = serde_json::from_str(wrapped_payload)
.expect("wrapped MCP output should preserve structured JSON");
assert_eq!(output_json, json!({ "result": "ok" }));
}
server.verify().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stdio_mcp_parallel_tool_calls_opt_in_runs_concurrently() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let first_call_id = "sync-1";
let second_call_id = "sync-2";
let server_name = "rmcp";
let tool_name = format!("mcp__{server_name}__sync");
let args = json!({
"sleep_after_ms": 100,
"barrier": {
"id": "stdio-mcp-parallel-tool-calls",
"participants": 2,
"timeout_ms": 1_000
}
})
.to_string();
mount_sse_once(
&server,
responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_function_call(first_call_id, &tool_name, &args),
responses::ev_function_call(second_call_id, &tool_name, &args),
responses::ev_completed("resp-1"),
]),
)
.await;
let final_mock = mount_sse_once(
&server,
responses::sse(vec![
responses::ev_assistant_message("msg-1", "rmcp sync tools completed successfully."),
responses::ev_completed("resp-2"),
]),
)
.await;
let rmcp_test_server_bin = stdio_server_bin()?;
let fixture = test_codex()
.with_config(move |config| {
let mut servers = config.mcp_servers.get().clone();
servers.insert(
server_name.to_string(),
McpServerConfig {
transport: McpServerTransportConfig::Stdio {
command: rmcp_test_server_bin,
args: Vec::new(),
env: None,
env_vars: Vec::new(),
cwd: None,
},
enabled: true,
required: false,
supports_parallel_tool_calls: true,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: Some(Duration::from_secs(2)),
enabled_tools: None,
disabled_tools: None,
scopes: None,
oauth_resource: None,
tools: HashMap::new(),
},
);
config
.mcp_servers
.set(servers)
.expect("test mcp servers should accept any configuration");
})
.build(&server)
.await?;
let session_model = fixture.session_configured.model.clone();
fixture
.codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "call the rmcp sync tool twice".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
cwd: fixture.cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
approvals_reviewer: None,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
model: session_model,
effort: None,
summary: None,
service_tier: None,
collaboration_mode: None,
personality: None,
})
.await?;
wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = final_mock.single_request();
for call_id in [first_call_id, second_call_id] {
let output_text = request
.function_call_output_text(call_id)
.expect("function_call_output present for rmcp sync call");
let wrapped_payload = split_wall_time_wrapped_output(&output_text);
let output_json: Value = serde_json::from_str(wrapped_payload)
.expect("wrapped MCP output should preserve structured JSON");
assert_eq!(output_json, json!({ "result": "ok" }));
}
server.verify().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial(mcp_test_value)]
async fn stdio_image_responses_round_trip() -> anyhow::Result<()> {
@@ -282,6 +546,7 @@ async fn stdio_image_responses_round_trip() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -514,6 +779,7 @@ async fn stdio_image_responses_are_sanitized_for_text_only_model() -> anyhow::Re
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -637,6 +903,7 @@ async fn stdio_server_propagates_whitelisted_env_vars() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -800,6 +1067,7 @@ async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -1023,6 +1291,7 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
codex-rs/core/tests/suite/search_tool.rs
+1
View File
@@ -649,6 +649,7 @@ async fn tool_search_indexes_only_enabled_non_app_mcp_tools() -> Result<()> {
disabled_tools: Some(vec!["image".to_string()]),
scopes: None,
oauth_resource: None,
supports_parallel_tool_calls: false,
tools: HashMap::new(),
},
);
codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_capped_when_mirrored_to_realtime.snap
+12
View File
@@ -0,0 +1,12 @@
---
source: core/tests/suite/realtime_conversation.rs
expression: snapshot
---
type: conversation.item.create
item.type: message
item.role: user
content[0].type: input_text
model_has_full_user_text: true
realtime_text_equal_full_user_text: false
realtime_text_approx_tokens: 300
content[0].text: mirror-head alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alph…2417 tokens truncated…ega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega mirror-tail
codex-rs/core/tests/suite/sqlite_state.rs
+1
View File
@@ -368,6 +368,7 @@ async fn mcp_call_marks_thread_memory_mode_polluted_when_configured() -> Result<
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
codex-rs/core/tests/suite/truncation.rs
+3
View File
@@ -372,6 +372,7 @@ async fn mcp_tool_call_output_exceeds_limit_truncated_for_model() -> Result<()>
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(std::time::Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -468,6 +469,7 @@ async fn mcp_image_output_preserves_image_and_no_text_summary() -> Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
@@ -742,6 +744,7 @@ async fn mcp_tool_call_output_not_truncated_with_custom_limit() -> Result<()> {
},
enabled: true,
required: false,
supports_parallel_tool_calls: false,
disabled_reason: None,
startup_timeout_sec: Some(std::time::Duration::from_secs(10)),
tool_timeout_sec: None,
codex-rs/rmcp-client/src/bin/test_stdio_server.rs
+186
View File
@@ -1,6 +1,9 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration;
use rmcp::ErrorData as McpError;
use rmcp::ServiceExt;
@@ -25,7 +28,9 @@ use rmcp::model::Tool;
use rmcp::model::ToolAnnotations;
use serde::Deserialize;
use serde_json::json;
use tokio::sync::Barrier;
use tokio::task;
use tokio::time::sleep;
#[derive(Clone)]
struct TestToolServer {
@@ -47,6 +52,7 @@ impl TestToolServer {
let tools = vec![
Self::echo_tool(),
Self::echo_dash_tool(),
Self::sync_tool(),
Self::image_tool(),
Self::image_scenario_tool(),
];
@@ -112,6 +118,50 @@ impl TestToolServer {
tool
}
fn sync_tool() -> Tool {
#[expect(clippy::expect_used)]
let schema: JsonObject = serde_json::from_value(json!({
"type": "object",
"properties": {
"sleep_before_ms": { "type": "number" },
"sleep_after_ms": { "type": "number" },
"barrier": {
"type": "object",
"properties": {
"id": { "type": "string" },
"participants": { "type": "number" },
"timeout_ms": { "type": "number" }
},
"required": ["id", "participants"],
"additionalProperties": false
}
},
"additionalProperties": false
}))
.expect("sync tool schema should deserialize");
let mut tool = Tool::new(
Cow::Borrowed("sync"),
Cow::Borrowed(
"Synchronize concurrent test calls and optionally delay before or after the barrier.",
),
Arc::new(schema),
);
#[expect(clippy::expect_used)]
let output_schema: JsonObject = serde_json::from_value(json!({
"type": "object",
"properties": {
"result": { "type": "string" }
},
"required": ["result"],
"additionalProperties": false
}))
.expect("sync tool output schema should deserialize");
tool.output_schema = Some(Arc::new(output_schema));
tool.annotations = Some(ToolAnnotations::new().read_only(true));
tool
}
fn image_tool() -> Tool {
#[expect(clippy::expect_used)]
let schema: JsonObject = serde_json::from_value(serde_json::json!({
@@ -227,6 +277,42 @@ struct EchoArgs {
env_var: Option<String>,
}
const DEFAULT_SYNC_TIMEOUT_MS: u64 = 1_000;
static SYNC_BARRIERS: OnceLock<tokio::sync::Mutex<HashMap<String, SyncBarrierState>>> =
OnceLock::new();
struct SyncBarrierState {
barrier: Arc<Barrier>,
participants: usize,
}
#[derive(Debug, Deserialize)]
struct SyncBarrierArgs {
id: String,
participants: usize,
#[serde(default = "default_sync_timeout_ms")]
timeout_ms: u64,
}
#[derive(Debug, Deserialize)]
struct SyncArgs {
#[serde(default)]
sleep_before_ms: Option<u64>,
#[serde(default)]
sleep_after_ms: Option<u64>,
#[serde(default)]
barrier: Option<SyncBarrierArgs>,
}
fn default_sync_timeout_ms() -> u64 {
DEFAULT_SYNC_TIMEOUT_MS
}
fn sync_barrier_map() -> &'static tokio::sync::Mutex<HashMap<String, SyncBarrierState>> {
SYNC_BARRIERS.get_or_init(|| tokio::sync::Mutex::new(HashMap::new()))
}
#[derive(Deserialize, Debug)]
#[serde(rename_all = "snake_case")]
/// Scenarios for `image_scenario`, intended to exercise Codex TUI handling of MCP image outputs.
@@ -387,6 +473,10 @@ impl ServerHandler for TestToolServer {
let args = Self::parse_call_args::<ImageScenarioArgs>(&request, "image_scenario")?;
Self::image_scenario_result(args)
}
"sync" => {
let args = Self::parse_call_args::<SyncArgs>(&request, "sync")?;
Self::sync_result(args).await
}
other => Err(McpError::invalid_params(
format!("unknown tool: {other}"),
None,
@@ -469,6 +559,102 @@ impl TestToolServer {
Ok(CallToolResult::success(content))
}
async fn sync_result(args: SyncArgs) -> Result<CallToolResult, McpError> {
if let Some(delay) = args.sleep_before_ms
&& delay > 0
{
sleep(Duration::from_millis(delay)).await;
}
if let Some(barrier) = args.barrier {
wait_on_sync_barrier(barrier).await?;
}
if let Some(delay) = args.sleep_after_ms
&& delay > 0
{
sleep(Duration::from_millis(delay)).await;
}
Ok(CallToolResult {
content: Vec::new(),
structured_content: Some(json!({ "result": "ok" })),
is_error: Some(false),
meta: None,
})
}
}
async fn wait_on_sync_barrier(args: SyncBarrierArgs) -> Result<(), McpError> {
if args.participants == 0 {
return Err(McpError::invalid_params(
"barrier participants must be greater than zero",
None,
));
}
if args.timeout_ms == 0 {
return Err(McpError::invalid_params(
"barrier timeout must be greater than zero",
None,
));
}
let barrier_id = args.id.clone();
let barrier = {
let mut map = sync_barrier_map().lock().await;
match map.entry(barrier_id.clone()) {
Entry::Occupied(entry) => {
let state = entry.get();
if state.participants != args.participants {
let existing = state.participants;
return Err(McpError::invalid_params(
format!(
"barrier {barrier_id} already registered with {existing} participants"
),
None,
));
}
state.barrier.clone()
}
Entry::Vacant(entry) => {
let barrier = Arc::new(Barrier::new(args.participants));
entry.insert(SyncBarrierState {
barrier: barrier.clone(),
participants: args.participants,
});
barrier
}
}
};
let wait_result =
match tokio::time::timeout(Duration::from_millis(args.timeout_ms), barrier.wait()).await {
Ok(wait_result) => wait_result,
Err(_) => {
remove_sync_barrier_if_current(&barrier_id, &barrier).await;
return Err(McpError::invalid_params(
"sync barrier wait timed out",
None,
));
}
};
if wait_result.is_leader() {
remove_sync_barrier_if_current(&barrier_id, &barrier).await;
}
Ok(())
}
async fn remove_sync_barrier_if_current(barrier_id: &str, barrier: &Arc<Barrier>) {
let mut map = sync_barrier_map().lock().await;
if let Some(state) = map.get(barrier_id)
&& Arc::ptr_eq(&state.barrier, barrier)
{
map.remove(barrier_id);
}
}
fn parse_data_url(url: &str) -> Option<(String, String)> {
docs/config.md
+14
View File
@@ -12,6 +12,20 @@ Codex can connect to MCP servers configured in `~/.codex/config.toml`. See the c
- https://developers.openai.com/codex/config-reference
MCP tools default to serialized calls. To mark every tool exposed by one server
as eligible for parallel tool calls, set `supports_parallel_tool_calls` on that
server:
```toml
[mcp_servers.docs]
command = "docs-server"
supports_parallel_tool_calls = true
```
Only enable parallel calls for MCP servers whose tools are safe to run at the
same time. If tools read and write shared state, files, databases, or external
resources, review those read/write race conditions before enabling this setting.
## MCP tool approvals
Codex stores per-tool approval overrides for custom MCP servers under