openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-14 11:57:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

V1

Browse Source
This commit is contained in:
jimmyfraiture
2025-10-01 16:42:01 +01:00
parent 1a58229438
commit a6d3e2a334
18 changed files with 690 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/app-server-protocol/src/protocol.rs
View File

@@ -515,6 +515,8 @@ pub struct Tools {
pub web_search: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub view_image: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub parallel_read_only: Option<bool>, // todo check this
}
/// MCP representation of a [`codex_core::config_types::SandboxWorkspaceWrite`].

1
codex-rs/app-server/tests/suite/config.rs
View File

@@ -99,6 +99,7 @@ async fn get_config_toml_parses_all_fields() {
tools: Some(Tools {
web_search: Some(false),
view_image: Some(true),
parallel_read_only: Some(false),
}),
profile: Some("test".to_string()),
profiles: HashMap::from([(

33
codex-rs/core/src/client.rs
View File

@@ -118,6 +118,12 @@ impl ModelClient {
})
}
pub fn supports_parallel_read_only_tools(&self) -> bool {
self.config.enable_parallel_read_only_tools
&& self.config.model_family.supports_parallel_read_only_tools
&& self.provider.supports_parallel_tool_calls
}
/// Dispatches to either the Responses or Chat implementation depending on
/// the provider config. Public callers always invoke `stream()` – the
/// specialised helpers are private to avoid accidental misuse.
@@ -226,7 +232,8 @@ impl ModelClient {
input: &input_with_instructions,
tools: &tools_json,
tool_choice: "auto",
parallel_tool_calls: false,
parallel_tool_calls: prompt.allow_parallel_read_only_tools
&& self.supports_parallel_read_only_tools(),
reasoning,
store: azure_workaround,
stream: true,
@@ -1038,15 +1045,11 @@ mod tests {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
..Default::default()
};
let otel_event_manager = otel_event_manager();
@@ -1101,15 +1104,11 @@ mod tests {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
..Default::default()
};
let otel_event_manager = otel_event_manager();
@@ -1137,15 +1136,11 @@ mod tests {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
..Default::default()
};
let otel_event_manager = otel_event_manager();
@@ -1244,15 +1239,11 @@ mod tests {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
..Default::default()
};
let otel_event_manager = otel_event_manager();

40
codex-rs/core/src/client_common.rs
View File

@@ -36,6 +36,9 @@ pub struct Prompt {
/// Optional the output schema for the model's response.
pub output_schema: Option<Value>,
/// Allow parallel tool calls for read-only tools.
pub allow_parallel_read_only_tools: bool,
}
impl Prompt {
@@ -368,4 +371,41 @@ mod tests {
let v = serde_json::to_value(&req).expect("json");
assert!(v.get("text").is_none());
}
#[test]
fn serializes_parallel_tool_calls_flag() {
let input: Vec<ResponseItem> = vec![];
let tools: Vec<serde_json::Value> = vec![];
let req_true = ResponsesApiRequest {
model: "gpt-5",
instructions: "i",
input: &input,
tools: &tools,
tool_choice: "auto",
parallel_tool_calls: true,
reasoning: None,
store: false,
stream: true,
include: vec![],
prompt_cache_key: None,
text: None,
};
let v_true = serde_json::to_value(&req_true).expect("json");
assert_eq!(
v_true.get("parallel_tool_calls"),
Some(&serde_json::Value::Bool(true))
);
let req_false = ResponsesApiRequest {
parallel_tool_calls: false,
..req_true
};
let v_false = serde_json::to_value(&req_false).expect("json");
assert_eq!(
v_false.get("parallel_tool_calls"),
Some(&serde_json::Value::Bool(false))
);
}
}

195
codex-rs/core/src/codex.rs
View File

@@ -99,6 +99,8 @@ use crate::tasks::CompactTask;
use crate::tasks::RegularTask;
use crate::tasks::ReviewTask;
use crate::tools::Router;
use crate::tools::executor::ProcessedResponseItem;
use crate::tools::executor::ToolCallExecutor;
use crate::tools::format_exec_output_str;
use crate::turn_diff_tracker::TurnDiffTracker;
use crate::unified_exec::UnifiedExecSessionManager;
@@ -438,6 +440,7 @@ impl Session {
use_streamable_shell_tool: config.use_experimental_streamable_shell_tool,
include_view_image_tool: config.include_view_image_tool,
experimental_unified_exec_tool: config.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
}),
user_instructions,
base_instructions,
@@ -1076,7 +1079,7 @@ impl Session {
&self.services.user_shell
}
fn show_raw_agent_reasoning(&self) -> bool {
pub(crate) fn show_raw_agent_reasoning(&self) -> bool {
self.services.show_raw_agent_reasoning
}
}
@@ -1166,6 +1169,7 @@ async fn submission_loop(
use_streamable_shell_tool: config.use_experimental_streamable_shell_tool,
include_view_image_tool: config.include_view_image_tool,
experimental_unified_exec_tool: config.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
});
let new_turn_context = TurnContext {
@@ -1270,6 +1274,7 @@ async fn submission_loop(
include_view_image_tool: config.include_view_image_tool,
experimental_unified_exec_tool: config
.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
}),
user_instructions: turn_context.user_instructions.clone(),
base_instructions: turn_context.base_instructions.clone(),
@@ -1501,6 +1506,7 @@ async fn spawn_review_thread(
use_streamable_shell_tool: false,
include_view_image_tool: false,
experimental_unified_exec_tool: config.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
});
let base_instructions = REVIEW_PROMPT.to_string();
@@ -1669,8 +1675,8 @@ pub(crate) async fn run_task(
})
.collect();
match run_turn(
&sess,
turn_context.as_ref(),
sess.clone(),
turn_context.clone(),
&mut turn_diff_tracker,
sub_id.clone(),
turn_input,
@@ -1894,28 +1900,40 @@ fn parse_review_output_event(text: &str) -> ReviewOutputEvent {
}
async fn run_turn(
sess: &Session,
turn_context: &TurnContext,
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: &mut TurnDiffTracker,
sub_id: String,
input: Vec<ResponseItem>,
) -> CodexResult<TurnRunResult> {
let mcp_tools = sess.services.mcp_connection_manager.list_all_tools();
let router = Router::from_config(&turn_context.tools_config, Some(mcp_tools));
let router = Arc::new(Router::from_config(
&turn_context.tools_config,
Some(mcp_tools),
));
let template_executor =
ToolCallExecutor::new(router.clone(), sess.clone(), turn_context.clone());
let allow_parallel_read_only = template_executor.allow_parallel_read_only();
let tool_specs = template_executor.specs().to_vec();
drop(template_executor);
let prompt = Prompt {
input,
tools: router.specs().to_vec(),
tools: tool_specs,
base_instructions_override: turn_context.base_instructions.clone(),
output_schema: turn_context.final_output_json_schema.clone(),
allow_parallel_read_only_tools: allow_parallel_read_only,
};
let mut retries = 0;
loop {
let tool_executor =
ToolCallExecutor::new(router.clone(), sess.clone(), turn_context.clone());
match try_run_turn(
&router,
sess,
turn_context,
tool_executor,
sess.clone(),
turn_context.clone(),
turn_diff_tracker,
&sub_id,
&prompt,
@@ -1966,31 +1984,22 @@ async fn run_turn(
}
}
/// When the model is prompted, it returns a stream of events. Some of these
/// events map to a `ResponseItem`. A `ResponseItem` may need to be
/// "handled" such that it produces a `ResponseInputItem` that needs to be
/// sent back to the model on the next turn.
#[derive(Debug)]
struct ProcessedResponseItem {
item: ResponseItem,
response: Option<ResponseInputItem>,
}
#[derive(Debug)]
struct TurnRunResult {
processed_items: Vec<ProcessedResponseItem>,
total_token_usage: Option<TokenUsage>,
}
async fn try_run_turn(
router: &crate::tools::Router,
sess: &Session,
turn_context: &TurnContext,
mut tool_executor: ToolCallExecutor,
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: &mut TurnDiffTracker,
sub_id: &str,
prompt: &Prompt,
) -> CodexResult<TurnRunResult> {
// call_ids that are part of this response.
let sess_ref = sess.as_ref();
let turn_context_ref = turn_context.as_ref();
let completed_call_ids = prompt
.input
.iter()
@@ -2005,9 +2014,6 @@ async fn try_run_turn(
})
.collect::<Vec<_>>();
// call_ids that were pending but are not part of this response.
// This usually happens because the user interrupted the model before we responded to one of its tool calls
// and then the user sent a follow-up message.
let missing_calls = {
prompt
.input
@@ -2046,22 +2052,20 @@ async fn try_run_turn(
};
let rollout_item = RolloutItem::TurnContext(TurnContextItem {
cwd: turn_context.cwd.clone(),
approval_policy: turn_context.approval_policy,
sandbox_policy: turn_context.sandbox_policy.clone(),
model: turn_context.client.get_model(),
effort: turn_context.client.get_reasoning_effort(),
summary: turn_context.client.get_reasoning_summary(),
cwd: turn_context_ref.cwd.clone(),
approval_policy: turn_context_ref.approval_policy,
sandbox_policy: turn_context_ref.sandbox_policy.clone(),
model: turn_context_ref.client.get_model(),
effort: turn_context_ref.client.get_reasoning_effort(),
summary: turn_context_ref.client.get_reasoning_summary(),
});
sess.persist_rollout_items(&[rollout_item]).await;
let mut stream = turn_context.client.clone().stream(&prompt).await?;
sess_ref.persist_rollout_items(&[rollout_item]).await;
let mut output = Vec::new();
let mut stream = turn_context_ref.client.clone().stream(&prompt).await?;
loop {
// Poll the next item from the model stream. We must inspect *both* Ok and Err
// cases so that transient stream failures (e.g., dropped SSE connection before
// `response.completed`) bubble up and trigger the caller's retry logic.
tool_executor.drain_ready();
let event = stream.next().await;
let Some(event) = event else {
// Channel closed without yielding a final Completed event or explicit error.
@@ -2084,19 +2088,12 @@ async fn try_run_turn(
match event {
ResponseEvent::Created => {}
ResponseEvent::OutputItemDone(item) => {
let response = handle_response_item(
router,
sess,
turn_context,
turn_diff_tracker,
sub_id,
item.clone(),
)
.await?;
output.push(ProcessedResponseItem { item, response });
tool_executor
.handle_output_item(item, turn_diff_tracker, sub_id)
.await?;
}
ResponseEvent::WebSearchCallBegin { call_id } => {
let _ = sess
let _ = sess_ref
.tx_event
.send(Event {
id: sub_id.to_string(),
@@ -2105,15 +2102,16 @@ async fn try_run_turn(
.await;
}
ResponseEvent::RateLimits(snapshot) => {
// Update internal state with latest rate limits, but defer sending until
// token usage is available to avoid duplicate TokenCount events.
sess.update_rate_limits(sub_id, snapshot).await;
sess_ref.update_rate_limits(sub_id, snapshot).await;
}
ResponseEvent::Completed {
response_id: _,
token_usage,
} => {
sess.update_token_usage_info(sub_id, turn_context, token_usage.as_ref())
tool_executor.flush().await;
sess_ref
.update_token_usage_info(sub_id, turn_context_ref, token_usage.as_ref())
.await;
let unified_diff = turn_diff_tracker.get_unified_diff();
@@ -2123,11 +2121,11 @@ async fn try_run_turn(
id: sub_id.to_string(),
msg,
};
sess.send_event(event).await;
sess_ref.send_event(event).await;
}
let result = TurnRunResult {
processed_items: output,
processed_items: tool_executor.take_processed_items(),
total_token_usage: token_usage.clone(),
};
@@ -2141,7 +2139,7 @@ async fn try_run_turn(
id: sub_id.to_string(),
msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }),
};
sess.send_event(event).await;
sess_ref.send_event(event).await;
} else {
trace!("suppressing OutputTextDelta in review mode");
}
@@ -2151,101 +2149,30 @@ async fn try_run_turn(
id: sub_id.to_string(),
msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }),
};
sess.send_event(event).await;
sess_ref.send_event(event).await;
}
ResponseEvent::ReasoningSummaryPartAdded => {
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent {}),
};
sess.send_event(event).await;
sess_ref.send_event(event).await;
}
ResponseEvent::ReasoningContentDelta(delta) => {
if sess.show_raw_agent_reasoning() {
if sess_ref.show_raw_agent_reasoning() {
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentReasoningRawContentDelta(
AgentReasoningRawContentDeltaEvent { delta },
),
};
sess.send_event(event).await;
sess_ref.send_event(event).await;
}
}
}
}
}
async fn handle_response_item(
router: &crate::tools::Router,
sess: &Session,
turn_context: &TurnContext,
turn_diff_tracker: &mut TurnDiffTracker,
sub_id: &str,
item: ResponseItem,
) -> CodexResult<Option<ResponseInputItem>> {
debug!(?item, "Output item");
match Router::build_tool_call(sess, item.clone()) {
Ok(Some(call)) => {
let payload_preview = call.payload.log_payload().into_owned();
tracing::info!("ToolCall: {} {}", call.tool_name, payload_preview);
let response = router
.dispatch_tool_call(sess, turn_context, turn_diff_tracker, sub_id, call)
.await;
Ok(Some(response))
}
Ok(None) => {
match &item {
ResponseItem::Message { .. }
| ResponseItem::Reasoning { .. }
| ResponseItem::WebSearchCall { .. } => {
let msgs = match &item {
ResponseItem::Message { .. } if turn_context.is_review_mode => {
trace!("suppressing assistant Message in review mode");
Vec::new()
}
_ => map_response_item_to_event_messages(
&item,
sess.show_raw_agent_reasoning(),
),
};
for msg in msgs {
let event = Event {
id: sub_id.to_string(),
msg,
};
sess.send_event(event).await;
}
}
ResponseItem::FunctionCallOutput { .. }
| ResponseItem::CustomToolCallOutput { .. } => {
debug!("unexpected tool output from stream");
}
_ => {}
}
Ok(None)
}
Err(FunctionCallError::RespondToModel(msg)) => {
if msg == "LocalShellCall without call_id or id" {
turn_context
.client
.get_otel_event_manager()
.log_tool_failed("local_shell", &msg);
error!(msg);
}
Ok(Some(ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: FunctionCallOutputPayload {
content: msg,
success: None,
},
}))
}
}
}
pub(super) fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -> Option<String> {
responses.iter().rev().find_map(|item| {
if let ResponseItem::Message { role, content, .. } = item {
@@ -2658,6 +2585,7 @@ mod tests {
use_streamable_shell_tool: config.use_experimental_streamable_shell_tool,
include_view_image_tool: config.include_view_image_tool,
experimental_unified_exec_tool: config.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
});
let turn_context = TurnContext {
client,
@@ -2731,6 +2659,7 @@ mod tests {
use_streamable_shell_tool: config.use_experimental_streamable_shell_tool,
include_view_image_tool: config.include_view_image_tool,
experimental_unified_exec_tool: config.use_experimental_unified_exec_tool,
enable_parallel_read_only: config.enable_parallel_read_only_tools,
});
let turn_context = Arc::new(TurnContext {
client,

122
codex-rs/core/src/config.rs
View File

@@ -206,6 +206,9 @@ pub struct Config {
/// OTEL configuration (exporter type, endpoint, headers, etc.).
pub otel: crate::config_types::OtelConfig,
/// Enable read-only tools to run in parallel.
pub enable_parallel_read_only_tools: bool,
}
impl Config {
@@ -767,6 +770,9 @@ pub struct ToolsToml {
/// Enable the `view_image` tool that lets the agent attach local images.
#[serde(default)]
pub view_image: Option<bool>,
#[serde(default)]
pub parallel_read_only: Option<bool>,
}
impl From<ToolsToml> for Tools {
@@ -774,6 +780,7 @@ impl From<ToolsToml> for Tools {
Self {
web_search: tools_toml.web_search,
view_image: tools_toml.view_image,
parallel_read_only: tools_toml.parallel_read_only,
}
}
}
@@ -970,6 +977,12 @@ impl Config {
.or(cfg.tools.as_ref().and_then(|t| t.view_image))
.unwrap_or(true);
let enable_parallel_read_only_tools = cfg
.tools
.as_ref()
.and_then(|t| t.parallel_read_only)
.unwrap_or(false);
let model = model
.or(config_profile.model)
.or(cfg.model)
@@ -1071,6 +1084,7 @@ impl Config {
.unwrap_or(false),
use_experimental_use_rmcp_client: cfg.experimental_use_rmcp_client.unwrap_or(false),
include_view_image_tool,
enable_parallel_read_only_tools,
active_profile: active_profile_name,
disable_paste_burst: cfg.disable_paste_burst.unwrap_or(false),
tui_notifications: cfg
@@ -1658,9 +1672,7 @@ model = "gpt-5-codex"
cwd: TempDir,
codex_home: TempDir,
cfg: ConfigToml,
model_provider_map: HashMap<String, ModelProviderInfo>,
openai_provider: ModelProviderInfo,
openai_chat_completions_provider: ModelProviderInfo,
}
impl PrecedenceTestFixture {
@@ -1733,14 +1745,10 @@ model_verbosity = "high"
base_url: Some("https://api.openai.com/v1".to_string()),
env_key: Some("OPENAI_API_KEY".to_string()),
wire_api: crate::WireApi::Chat,
env_key_instructions: None,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(4),
stream_max_retries: Some(10),
stream_idle_timeout_ms: Some(300_000),
requires_openai_auth: false,
..Default::default()
};
let model_provider_map = {
let mut model_provider_map = built_in_model_providers();
@@ -1760,9 +1768,7 @@ model_verbosity = "high"
cwd: cwd_temp_dir,
codex_home: codex_home_temp_dir,
cfg,
model_provider_map,
openai_provider,
openai_chat_completions_provider,
})
}
@@ -1792,50 +1798,49 @@ model_verbosity = "high"
o3_profile_overrides,
fixture.codex_home(),
)?;
assert_eq!(
Config {
model: "o3".to_string(),
review_model: OPENAI_DEFAULT_REVIEW_MODEL.to_string(),
model_family: find_family_for_model("o3").expect("known model slug"),
model_context_window: Some(200_000),
model_max_output_tokens: Some(100_000),
model_auto_compact_token_limit: None,
model_provider_id: "openai".to_string(),
model_provider: fixture.openai_provider.clone(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
shell_environment_policy: ShellEnvironmentPolicy::default(),
user_instructions: None,
notify: None,
cwd: fixture.cwd(),
mcp_servers: HashMap::new(),
model_providers: fixture.model_provider_map.clone(),
project_doc_max_bytes: PROJECT_DOC_MAX_BYTES,
codex_home: fixture.codex_home(),
history: History::default(),
file_opener: UriBasedFileOpener::VsCode,
codex_linux_sandbox_exe: None,
hide_agent_reasoning: false,
show_raw_agent_reasoning: false,
model_reasoning_effort: Some(ReasoningEffort::High),
model_reasoning_summary: ReasoningSummary::Detailed,
model_verbosity: None,
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
base_instructions: None,
include_plan_tool: false,
include_apply_patch_tool: false,
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
active_profile: Some("o3".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
},
o3_profile_config
);
let expected_o3_profile_config = Config {
model: "o3".to_string(),
review_model: OPENAI_DEFAULT_REVIEW_MODEL.to_string(),
model_family: find_family_for_model("o3").expect("known model slug"),
model_context_window: Some(200_000),
model_max_output_tokens: Some(100_000),
model_auto_compact_token_limit: None,
model_provider_id: "openai".to_string(),
model_provider: fixture.openai_provider.clone(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
shell_environment_policy: ShellEnvironmentPolicy::default(),
user_instructions: None,
notify: None,
cwd: fixture.cwd(),
mcp_servers: HashMap::new(),
model_providers: o3_profile_config.model_providers.clone(),
project_doc_max_bytes: PROJECT_DOC_MAX_BYTES,
codex_home: fixture.codex_home(),
history: History::default(),
file_opener: UriBasedFileOpener::VsCode,
codex_linux_sandbox_exe: None,
hide_agent_reasoning: false,
show_raw_agent_reasoning: false,
model_reasoning_effort: Some(ReasoningEffort::High),
model_reasoning_summary: ReasoningSummary::Detailed,
model_verbosity: None,
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
base_instructions: None,
include_plan_tool: false,
include_apply_patch_tool: false,
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
enable_parallel_read_only_tools: false,
active_profile: Some("o3".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
};
assert_eq!(expected_o3_profile_config, o3_profile_config);
Ok(())
}
@@ -1861,7 +1866,7 @@ model_verbosity = "high"
model_max_output_tokens: Some(4_096),
model_auto_compact_token_limit: None,
model_provider_id: "openai-chat-completions".to_string(),
model_provider: fixture.openai_chat_completions_provider.clone(),
model_provider: gpt3_profile_config.model_provider.clone(),
approval_policy: AskForApproval::UnlessTrusted,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
shell_environment_policy: ShellEnvironmentPolicy::default(),
@@ -1869,7 +1874,7 @@ model_verbosity = "high"
notify: None,
cwd: fixture.cwd(),
mcp_servers: HashMap::new(),
model_providers: fixture.model_provider_map.clone(),
model_providers: gpt3_profile_config.model_providers.clone(),
project_doc_max_bytes: PROJECT_DOC_MAX_BYTES,
codex_home: fixture.codex_home(),
history: History::default(),
@@ -1889,6 +1894,7 @@ model_verbosity = "high"
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
enable_parallel_read_only_tools: false,
active_profile: Some("gpt3".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
@@ -1944,7 +1950,7 @@ model_verbosity = "high"
notify: None,
cwd: fixture.cwd(),
mcp_servers: HashMap::new(),
model_providers: fixture.model_provider_map.clone(),
model_providers: zdr_profile_config.model_providers.clone(),
project_doc_max_bytes: PROJECT_DOC_MAX_BYTES,
codex_home: fixture.codex_home(),
history: History::default(),
@@ -1964,6 +1970,7 @@ model_verbosity = "high"
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
enable_parallel_read_only_tools: false,
active_profile: Some("zdr".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
@@ -2005,7 +2012,7 @@ model_verbosity = "high"
notify: None,
cwd: fixture.cwd(),
mcp_servers: HashMap::new(),
model_providers: fixture.model_provider_map.clone(),
model_providers: gpt5_profile_config.model_providers.clone(),
project_doc_max_bytes: PROJECT_DOC_MAX_BYTES,
codex_home: fixture.codex_home(),
history: History::default(),
@@ -2025,6 +2032,7 @@ model_verbosity = "high"
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
enable_parallel_read_only_tools: false,
active_profile: Some("gpt5".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),

7
codex-rs/core/src/model_family.rs
View File

@@ -41,6 +41,9 @@ pub struct ModelFamily {
// Instructions to use for querying the model
pub base_instructions: String,
// todo check if all those configs are necessary...
pub supports_parallel_read_only_tools: bool,
}
macro_rules! model_family {
@@ -57,6 +60,7 @@ macro_rules! model_family {
uses_local_shell_tool: false,
apply_patch_tool_type: None,
base_instructions: BASE_INSTRUCTIONS.to_string(),
supports_parallel_read_only_tools: false,
};
// apply overrides
$(
@@ -105,12 +109,14 @@ pub fn find_family_for_model(slug: &str) -> Option<ModelFamily> {
supports_reasoning_summaries: true,
reasoning_summary_format: ReasoningSummaryFormat::Experimental,
base_instructions: GPT_5_CODEX_INSTRUCTIONS.to_string(),
supports_parallel_read_only_tools: true,
)
} else if slug.starts_with("gpt-5") {
model_family!(
slug, "gpt-5",
supports_reasoning_summaries: true,
needs_special_apply_patch_instructions: true,
supports_parallel_read_only_tools: true,
)
} else {
None
@@ -127,5 +133,6 @@ pub fn derive_default_model_family(model: &str) -> ModelFamily {
uses_local_shell_tool: false,
apply_patch_tool_type: None,
base_instructions: BASE_INSTRUCTIONS.to_string(),
supports_parallel_read_only_tools: false,
}
}

53
codex-rs/core/src/model_provider_info.rs
View File

@@ -40,7 +40,7 @@ pub enum WireApi {
}
/// Serializable representation of a provider definition.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Default)]
pub struct ModelProviderInfo {
/// Friendly display name.
pub name: String,
@@ -86,6 +86,10 @@ pub struct ModelProviderInfo {
/// and API key (if needed) comes from the "env_key" environment variable.
#[serde(default)]
pub requires_openai_auth: bool,
/// Does the model support parallel tool calls.
#[serde(default)]
pub supports_parallel_tool_calls: bool,
}
impl ModelProviderInfo {
@@ -297,6 +301,7 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: true,
supports_parallel_tool_calls: true,
},
),
(BUILT_IN_OSS_MODEL_PROVIDER_ID, create_oss_provider()),
@@ -341,6 +346,7 @@ pub fn create_oss_provider_with_base_url(base_url: &str) -> ModelProviderInfo {
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
supports_parallel_tool_calls: false,
}
}
@@ -370,16 +376,8 @@ base_url = "http://localhost:11434/v1"
let expected_provider = ModelProviderInfo {
name: "Ollama".into(),
base_url: Some("http://localhost:11434/v1".into()),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Chat,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();
@@ -398,17 +396,11 @@ query_params = { api-version = "2025-04-01-preview" }
name: "Azure".into(),
base_url: Some("https://xxxxx.openai.azure.com/openai".into()),
env_key: Some("AZURE_OPENAI_API_KEY".into()),
env_key_instructions: None,
wire_api: WireApi::Chat,
query_params: Some(maplit::hashmap! {
"api-version".to_string() => "2025-04-01-preview".to_string(),
}),
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();
@@ -428,19 +420,14 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" }
name: "Example".into(),
base_url: Some("https://example.com".into()),
env_key: Some("API_KEY".into()),
env_key_instructions: None,
wire_api: WireApi::Chat,
query_params: None,
http_headers: Some(maplit::hashmap! {
"X-Example-Header".to_string() => "example-value".to_string(),
}),
env_http_headers: Some(maplit::hashmap! {
"X-Example-Env-Header".to_string() => "EXAMPLE_ENV_VAR".to_string(),
}),
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();
@@ -453,16 +440,8 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" }
ModelProviderInfo {
name: "test".into(),
base_url: Some(base_url.into()),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
}
}
@@ -485,16 +464,8 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" }
let named_provider = ModelProviderInfo {
name: "Azure".into(),
base_url: Some("https://example.com".into()),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
assert!(named_provider.is_azure_responses_endpoint());

314
codex-rs/core/src/tools/executor.rs Normal file
View File

@@ -0,0 +1,314 @@
use std::collections::HashMap;
use std::panic::AssertUnwindSafe;
use std::sync::Arc;
use futures::FutureExt;
use tokio::task::JoinSet;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::trace;
use tracing::warn;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::error::Result as CodexResult;
use crate::event_mapping::map_response_item_to_event_messages;
use crate::function_tool::FunctionCallError;
use crate::protocol::Event;
use crate::tools::context::ToolPayload;
use crate::tools::router::Router;
use crate::tools::router::ToolCall;
use crate::tools::spec::ToolSpec;
use crate::turn_diff_tracker::TurnDiffTracker;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
#[derive(Debug)]
pub(crate) struct ProcessedResponseItem {
pub item: ResponseItem,
pub response: Option<ResponseInputItem>,
}
pub(crate) struct ToolCallExecutor {
router: Arc<Router>,
session: Arc<Session>,
turn_context: Arc<TurnContext>,
allow_parallel_read_only: bool,
read_only_tasks: JoinSet<(usize, Result<ResponseInputItem, String>)>,
read_only_meta: HashMap<usize, (String, ToolPayload, String)>,
processed_items: Vec<ProcessedResponseItem>,
}
impl ToolCallExecutor {
pub(crate) fn new(
router: Arc<Router>,
session: Arc<Session>, // todo why
turn_context: Arc<TurnContext>, // todo why
) -> Self {
let allow_parallel_read_only = router.has_read_only_tools()
&& turn_context.tools_config.enable_parallel_read_only
&& turn_context.client.supports_parallel_read_only_tools();
Self {
router,
session,
turn_context,
allow_parallel_read_only,
read_only_tasks: JoinSet::new(),
read_only_meta: HashMap::new(),
processed_items: Vec::new(),
}
}
pub(crate) fn specs(&self) -> &[ToolSpec] {
self.router.specs()
}
pub(crate) fn allow_parallel_read_only(&self) -> bool {
self.allow_parallel_read_only
}
pub(crate) fn drain_ready(&mut self) {
while let Some(res) = self.read_only_tasks.try_join_next() {
match res {
Ok((idx, Ok(response))) => self.assign_parallel_success(idx, response),
Ok((idx, Err(err))) => self.assign_parallel_failure(idx, err),
Err(join_err) => {
warn!(
?join_err,
"parallel read-only task aborted before completion"
);
}
}
}
}
pub(crate) async fn flush(&mut self) {
while let Some(res) = self.read_only_tasks.join_next().await {
match res {
Ok((idx, Ok(response))) => self.assign_parallel_success(idx, response),
Ok((idx, Err(err))) => self.assign_parallel_failure(idx, err),
Err(join_err) => {
warn!(
?join_err,
"parallel read-only task aborted before completion"
);
}
}
}
}
pub(crate) async fn handle_output_item(
&mut self,
item: ResponseItem,
turn_diff_tracker: &mut TurnDiffTracker,
sub_id: &str,
) -> CodexResult<()> {
match self
.router
.build_tool_call(self.session.as_ref(), item.clone())
{
Ok(Some(call)) => {
let payload_preview = call.payload.log_payload().into_owned();
info!("ToolCall: {} {}", call.tool_name, payload_preview);
let idx = self.processed_items.len();
self.processed_items.push(ProcessedResponseItem {
item,
response: None,
});
if self.allow_parallel_read_only && call.capabilities.read_only {
self.schedule_parallel_task(idx, call, sub_id);
} else {
self.flush().await;
let response = self
.router
.dispatch_tool_call(
self.session.as_ref(),
self.turn_context.as_ref(),
turn_diff_tracker,
sub_id,
call,
)
.await;
if let Some(slot) = self.processed_items.get_mut(idx) {
slot.response = Some(response);
}
}
}
Ok(None) => {
self.emit_response_item_events(sub_id, &item).await?;
self.processed_items.push(ProcessedResponseItem {
item,
response: None,
});
}
Err(FunctionCallError::RespondToModel(msg)) => {
if msg == "LocalShellCall without call_id or id" {
self.turn_context
.client
.get_otel_event_manager()
.log_tool_failed("local_shell", &msg);
error!(msg);
}
self.flush().await;
self.processed_items.push(ProcessedResponseItem {
item,
response: Some(ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: FunctionCallOutputPayload {
content: msg,
success: None,
},
}),
});
}
}
Ok(())
}
pub(crate) fn take_processed_items(mut self) -> Vec<ProcessedResponseItem> {
self.drain_ready();
self.processed_items
}
fn schedule_parallel_task(&mut self, idx: usize, call: ToolCall, sub_id: &str) {
let payload_clone = call.payload.clone();
self.read_only_meta.insert(
idx,
(call.call_id.clone(), payload_clone, call.tool_name.clone()),
);
let router_for_task = self.router.clone();
let session_for_task = self.session.clone();
let turn_context_for_task = self.turn_context.clone();
let sub_id_for_task = sub_id.to_string();
self.read_only_tasks.spawn(async move {
let mut tracker = TurnDiffTracker::new();
let fut = async {
router_for_task
.dispatch_tool_call(
session_for_task.as_ref(),
turn_context_for_task.as_ref(),
&mut tracker,
&sub_id_for_task,
call,
)
.await
};
let result = AssertUnwindSafe(fut)
.catch_unwind()
.await
.map_err(Self::panic_to_message);
(idx, result)
});
}
async fn emit_response_item_events(
&self,
sub_id: &str,
item: &ResponseItem,
) -> CodexResult<()> {
match item {
ResponseItem::Message { .. }
| ResponseItem::Reasoning { .. }
| ResponseItem::WebSearchCall { .. } => {
let msgs = match item {
ResponseItem::Message { .. } if self.turn_context.is_review_mode => {
trace!("suppressing assistant Message in review mode");
Vec::new()
}
_ => map_response_item_to_event_messages(
item,
self.session.show_raw_agent_reasoning(),
),
};
for msg in msgs {
let event = Event {
id: sub_id.to_string(),
msg,
};
self.session.send_event(event).await;
}
}
ResponseItem::FunctionCallOutput { .. } | ResponseItem::CustomToolCallOutput { .. } => {
debug!("unexpected tool output from stream");
}
_ => {}
}
Ok(())
}
fn assign_parallel_success(&mut self, idx: usize, response: ResponseInputItem) {
self.read_only_meta.remove(&idx);
if let Some(slot) = self.processed_items.get_mut(idx) {
slot.response = Some(response);
} else {
warn!(idx, "parallel tool completion missing output slot");
}
}
fn assign_parallel_failure(&mut self, idx: usize, reason: String) {
let (call_id, payload, tool_name) = self.read_only_meta.remove(&idx).unwrap_or_else(|| {
(
String::new(),
ToolPayload::Function {
arguments: String::new(),
},
String::from("unknown"),
)
});
let message = if tool_name == "unknown" {
reason
} else {
format!("{tool_name} failed: {reason}")
};
let response = Self::fallback_response(call_id, payload, message);
if let Some(slot) = self.processed_items.get_mut(idx) {
slot.response = Some(response);
} else {
warn!(idx, "parallel tool failure missing output slot");
}
}
fn fallback_response(
call_id: String,
payload: ToolPayload,
message: String,
) -> ResponseInputItem {
match payload {
ToolPayload::Custom { .. } => ResponseInputItem::CustomToolCallOutput {
call_id,
output: message,
},
_ => ResponseInputItem::FunctionCallOutput {
call_id,
output: FunctionCallOutputPayload {
content: message,
success: Some(false),
},
},
}
}
fn panic_to_message(payload: Box<dyn std::any::Any + Send>) -> String {
if let Some(s) = payload.downcast_ref::<&str>() {
(*s).to_string()
} else if let Some(s) = payload.downcast_ref::<String>() {
s.clone()
} else {
"panic without message".to_string()
}
}
}

1
codex-rs/core/src/tools/mod.rs
View File

@@ -1,4 +1,5 @@
pub mod context;
pub mod executor;
pub(crate) mod handlers;
pub mod registry;
pub mod router;

71
codex-rs/core/src/tools/registry.rs
View File

@@ -18,6 +18,27 @@ pub enum ToolKind {
Mcp,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ToolCapabilities {
pub read_only: bool,
}
impl ToolCapabilities {
pub const fn mutating() -> Self {
Self { read_only: false }
}
pub const fn read_only() -> Self {
Self { read_only: true }
}
}
#[derive(Clone)]
struct ToolEntry {
handler: Arc<dyn ToolHandler>,
capabilities: ToolCapabilities,
}
#[async_trait]
pub trait ToolHandler: Send + Sync {
fn kind(&self) -> ToolKind;
@@ -36,17 +57,20 @@ pub trait ToolHandler: Send + Sync {
-> Result<ToolOutput, FunctionCallError>;
}
#[derive(Clone)]
pub struct ToolRegistry {
handlers: HashMap<String, Arc<dyn ToolHandler>>,
handlers: HashMap<String, ToolEntry>,
}
impl ToolRegistry {
pub fn new(handlers: HashMap<String, Arc<dyn ToolHandler>>) -> Self {
Self { handlers }
pub fn capabilities(&self, name: &str) -> Option<ToolCapabilities> {
self.handlers.get(name).map(|entry| entry.capabilities)
}
pub fn handler(&self, name: &str) -> Option<Arc<dyn ToolHandler>> {
self.handlers.get(name).map(Arc::clone)
pub fn has_read_only_tools(&self) -> bool {
self.handlers
.values()
.any(|entry| entry.capabilities.read_only)
}
// TODO(jif) for dynamic tools.
@@ -67,8 +91,8 @@ impl ToolRegistry {
let payload_for_response = invocation.payload.clone();
let log_payload = payload_for_response.log_payload().into_owned();
let handler = match self.handler(tool_name.as_ref()) {
Some(handler) => handler,
let entry = match self.handlers.get(tool_name.as_str()) {
Some(entry) => entry,
None => {
let message =
unsupported_tool_call_message(&invocation.payload, tool_name.as_ref());
@@ -84,6 +108,8 @@ impl ToolRegistry {
}
};
let handler = Arc::clone(&entry.handler);
if !handler.matches_kind(&invocation.payload) {
let message = format!("tool {tool_name} invoked with incompatible payload");
otel.tool_result(
@@ -137,7 +163,7 @@ impl ToolRegistry {
}
pub struct ToolRegistryBuilder {
handlers: HashMap<String, Arc<dyn ToolHandler>>,
handlers: HashMap<String, ToolEntry>,
}
impl ToolRegistryBuilder {
@@ -148,10 +174,33 @@ impl ToolRegistryBuilder {
}
pub fn register_handler(&mut self, name: impl Into<String>, handler: Arc<dyn ToolHandler>) {
self.register_with_capabilities(name, handler, ToolCapabilities::mutating());
}
pub fn register_read_only_handler(
&mut self,
name: impl Into<String>,
handler: Arc<dyn ToolHandler>,
) {
self.register_with_capabilities(name, handler, ToolCapabilities::read_only());
}
pub fn register_with_capabilities(
&mut self,
name: impl Into<String>,
handler: Arc<dyn ToolHandler>,
capabilities: ToolCapabilities,
) {
let name = name.into();
if self
.handlers
.insert(name.clone(), handler.clone())
.insert(
name.clone(),
ToolEntry {
handler: handler.clone(),
capabilities,
},
)
.is_some()
{
warn!("overwriting handler for tool {name}");
@@ -177,7 +226,9 @@ impl ToolRegistryBuilder {
// }
pub fn build(self) -> ToolRegistry {
ToolRegistry::new(self.handlers)
ToolRegistry {
handlers: self.handlers,
}
}
}

36
codex-rs/core/src/tools/router.rs
View File

@@ -5,6 +5,7 @@ use crate::codex::TurnContext;
use crate::function_tool::FunctionCallError;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
use crate::tools::registry::ToolCapabilities;
use crate::tools::registry::ToolRegistry;
use crate::tools::spec::ToolSpec;
use crate::tools::spec::ToolsConfig;
@@ -20,8 +21,10 @@ pub struct ToolCall {
pub tool_name: String,
pub call_id: String,
pub payload: ToolPayload,
pub capabilities: ToolCapabilities,
}
#[derive(Clone)]
pub struct Router {
registry: ToolRegistry,
specs: Vec<ToolSpec>,
@@ -41,7 +44,12 @@ impl Router {
&self.specs
}
pub fn has_read_only_tools(&self) -> bool {
self.registry.has_read_only_tools()
}
pub fn build_tool_call(
&self,
session: &Session,
item: ResponseItem,
) -> Result<Option<ToolCall>, FunctionCallError> {
@@ -53,7 +61,7 @@ impl Router {
..
} => {
if let Some((server, tool)) = session.parse_mcp_tool_name(&name) {
Ok(Some(ToolCall {
Ok(Some(self.attach_capabilities(ToolCall {
tool_name: name,
call_id,
payload: ToolPayload::Mcp {
@@ -61,18 +69,20 @@ impl Router {
tool,
raw_arguments: arguments,
},
}))
capabilities: ToolCapabilities::mutating(),
})))
} else {
let payload = if name == "unified_exec" {
ToolPayload::UnifiedExec { arguments }
} else {
ToolPayload::Function { arguments }
};
Ok(Some(ToolCall {
Ok(Some(self.attach_capabilities(ToolCall {
tool_name: name,
call_id,
payload,
}))
capabilities: ToolCapabilities::mutating(),
})))
}
}
ResponseItem::CustomToolCall {
@@ -80,11 +90,12 @@ impl Router {
input,
call_id,
..
} => Ok(Some(ToolCall {
} => Ok(Some(self.attach_capabilities(ToolCall {
tool_name: name,
call_id,
payload: ToolPayload::Custom { input },
})),
capabilities: ToolCapabilities::mutating(),
}))),
ResponseItem::LocalShellCall {
id,
call_id,
@@ -106,11 +117,12 @@ impl Router {
with_escalated_permissions: None,
justification: None,
};
Ok(Some(ToolCall {
Ok(Some(self.attach_capabilities(ToolCall {
tool_name: "local_shell".to_string(),
call_id,
payload: ToolPayload::LocalShell { params },
}))
capabilities: ToolCapabilities::mutating(),
})))
}
}
}
@@ -118,6 +130,13 @@ impl Router {
}
}
fn attach_capabilities(&self, mut call: ToolCall) -> ToolCall {
if let Some(capabilities) = self.registry.capabilities(call.tool_name.as_str()) {
call.capabilities = capabilities;
}
call
}
pub async fn dispatch_tool_call(
&self,
session: &Session,
@@ -131,6 +150,7 @@ impl Router {
tool_name,
call_id,
payload,
..
} = call;
let invocation = ToolInvocation {

29
codex-rs/core/src/tools/spec.rs
View File

@@ -1,3 +1,4 @@
use crate::tools::registry::ToolCapabilities;
use crate::tools::registry::ToolRegistryBuilder;
use serde::Deserialize;
use serde::Serialize;
@@ -69,6 +70,7 @@ pub(crate) struct ToolsConfig {
pub web_search_request: bool,
pub include_view_image_tool: bool,
pub experimental_unified_exec_tool: bool,
pub enable_parallel_read_only: bool,
}
pub(crate) struct ToolsConfigParams<'a> {
@@ -79,6 +81,7 @@ pub(crate) struct ToolsConfigParams<'a> {
pub(crate) use_streamable_shell_tool: bool,
pub(crate) include_view_image_tool: bool,
pub(crate) experimental_unified_exec_tool: bool,
pub(crate) enable_parallel_read_only: bool,
}
impl ToolsConfig {
@@ -91,6 +94,7 @@ impl ToolsConfig {
use_streamable_shell_tool,
include_view_image_tool,
experimental_unified_exec_tool,
enable_parallel_read_only,
} = params;
let shell_type = if *use_streamable_shell_tool {
ConfigShellToolType::Streamable
@@ -119,6 +123,7 @@ impl ToolsConfig {
web_search_request: *include_web_search_request,
include_view_image_tool: *include_view_image_tool,
experimental_unified_exec_tool: *experimental_unified_exec_tool,
enable_parallel_read_only: *enable_parallel_read_only,
}
}
}
@@ -589,7 +594,7 @@ pub(crate) fn build_specs(
}
specs.push(create_read_file_tool());
builder.register_handler("read_file", read_file_handler);
builder.register_read_only_handler("read_file", read_file_handler);
if config.web_search_request {
specs.push(ToolSpec::WebSearch {});
@@ -597,7 +602,7 @@ pub(crate) fn build_specs(
if config.include_view_image_tool {
specs.push(create_view_image_tool());
builder.register_handler("view_image", view_image_handler);
builder.register_read_only_handler("view_image", view_image_handler);
}
if let Some(mcp_tools) = mcp_tools {
@@ -605,10 +610,20 @@ pub(crate) fn build_specs(
entries.sort_by(|a, b| a.0.cmp(&b.0));
for (name, tool) in entries.into_iter() {
let capabilities = if tool
.annotations
.as_ref()
.and_then(|ann| ann.read_only_hint)
.unwrap_or(false)
{
ToolCapabilities::read_only()
} else {
ToolCapabilities::mutating()
};
match mcp_tool_to_openai_tool(name.clone(), tool.clone()) {
Ok(converted_tool) => {
specs.push(ToolSpec::Function(converted_tool));
builder.register_handler(name, mcp_handler.clone());
builder.register_with_capabilities(name, mcp_handler.clone(), capabilities);
}
Err(e) => {
tracing::error!("Failed to convert {name:?} MCP tool to OpenAI tool: {e:?}");
@@ -664,6 +679,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(&config, Some(HashMap::new())).0;
@@ -690,6 +706,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(&config, Some(HashMap::new())).0;
@@ -716,6 +733,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(
&config,
@@ -822,6 +840,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
// Intentionally construct a map with keys that would sort alphabetically.
@@ -899,6 +918,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(
@@ -967,6 +987,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(
@@ -1030,6 +1051,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(
@@ -1096,6 +1118,7 @@ mod tests {
use_streamable_shell_tool: false,
include_view_image_tool: true,
experimental_unified_exec_tool: true,
enable_parallel_read_only: false,
});
let tools = build_specs(

6
codex-rs/core/tests/chat_completions_payload.rs
View File

@@ -48,16 +48,12 @@ async fn run_request(input: Vec<ResponseItem>) -> Value {
let provider = ModelProviderInfo {
name: "mock".into(),
base_url: Some(format!("{}/v1", server.uri())),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Chat,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
requires_openai_auth: false,
..Default::default()
};
let codex_home = match TempDir::new() {

7
codex-rs/core/tests/chat_completions_sse.rs
View File

@@ -46,16 +46,11 @@ async fn run_stream_with_bytes(sse_body: &[u8]) -> Vec<ResponseEvent> {
let provider = ModelProviderInfo {
name: "mock".into(),
base_url: Some(format!("{}/v1", server.uri())),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Chat,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
requires_openai_auth: false,
..Default::default()
};
let codex_home = match TempDir::new() {

73
codex-rs/core/tests/suite/client.rs
View File

@@ -14,6 +14,7 @@ use codex_core::ResponseEvent;
use codex_core::ResponseItem;
use codex_core::WireApi;
use codex_core::built_in_model_providers;
use codex_core::model_family::find_family_for_model;
use codex_core::protocol::EventMsg;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
@@ -25,6 +26,7 @@ use core_test_support::load_default_config_for_test;
use core_test_support::load_sse_fixture_with_id;
use core_test_support::responses;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use futures::StreamExt;
@@ -646,16 +648,11 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
let provider = ModelProviderInfo {
name: "azure".into(),
base_url: Some(format!("{}/openai", server.uri())),
env_key: None,
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
requires_openai_auth: false,
..Default::default()
};
let codex_home = TempDir::new().unwrap();
@@ -1035,17 +1032,12 @@ async fn azure_overrides_assign_properties_used_for_responses_url() {
"api-version".to_string(),
"2025-04-01-preview".to_string(),
)])),
env_key_instructions: None,
wire_api: WireApi::Responses,
http_headers: Some(std::collections::HashMap::from([(
"Custom-Header".to_string(),
"Value".to_string(),
)])),
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
// Init session
@@ -1112,17 +1104,12 @@ async fn env_var_overrides_loaded_auth() {
"api-version".to_string(),
"2025-04-01-preview".to_string(),
)])),
env_key_instructions: None,
wire_api: WireApi::Responses,
http_headers: Some(std::collections::HashMap::from([(
"Custom-Header".to_string(),
"Value".to_string(),
)])),
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
requires_openai_auth: false,
..Default::default()
};
// Init session
@@ -1290,3 +1277,53 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() {
"request 3 tail mismatch",
);
}
#[tokio::test]
async fn parallel_tool_calls_enabled_when_supported() {
let server = MockServer::start().await;
let template = ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse_completed("resp_parallel"), "text/event-stream");
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(template)
.expect(1)
.mount(&server)
.await;
let mut provider = built_in_model_providers()["openai"].clone();
provider.base_url = Some(format!("{}/v1", server.uri()));
provider.supports_parallel_tool_calls = true;
let provider_clone = provider.clone();
let TestCodex { codex, .. } = test_codex()
.with_config(move |config| {
config.model = "gpt-5".to_string();
config.model_family = find_family_for_model("gpt-5").expect("model family");
config.enable_parallel_read_only_tools = true;
config.model_provider = provider_clone.clone();
config.model_provider_id = "openai".to_string();
})
.build(&server)
.await
.expect("build codex");
codex
.submit(Op::UserInput {
items: vec![InputItem::Text {
text: "hello".into(),
}],
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
let request = &server.received_requests().await.expect("requests")[0];
let request_body = request.body_json::<serde_json::Value>().unwrap();
assert_eq!(
request_body.get("parallel_tool_calls"),
Some(&serde_json::Value::Bool(true))
);
}

6
codex-rs/core/tests/suite/stream_error_allows_next_turn.rs
View File

@@ -65,15 +65,11 @@ async fn continue_after_stream_error() {
name: "mock-openai".into(),
base_url: Some(format!("{}/v1", server.uri())),
env_key: Some("PATH".into()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(1),
stream_max_retries: Some(1),
stream_idle_timeout_ms: Some(2_000),
requires_openai_auth: false,
..Default::default()
};
let TestCodex { codex, .. } = test_codex()

6
codex-rs/core/tests/suite/stream_no_completed.rs
View File

@@ -72,16 +72,12 @@ async fn retries_on_early_close() {
// ModelClient will return an error if the environment variable for the
// provider is not set.
env_key: Some("PATH".into()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
// exercise retry path: first attempt yields incomplete stream, so allow 1 retry
request_max_retries: Some(0),
stream_max_retries: Some(1),
stream_idle_timeout_ms: Some(2000),
requires_openai_auth: false,
..Default::default()
};
let TestCodex { codex, .. } = test_codex()