From a6d3e2a3340adcb7b6e259cdbdaffd5c29570906 Mon Sep 17 00:00:00 2001 From: jimmyfraiture Date: Wed, 1 Oct 2025 16:42:01 +0100 Subject: [PATCH] V1 --- codex-rs/app-server-protocol/src/protocol.rs | 2 + codex-rs/app-server/tests/suite/config.rs | 1 + codex-rs/core/src/client.rs | 33 +- codex-rs/core/src/client_common.rs | 40 +++ codex-rs/core/src/codex.rs | 195 ++++------- codex-rs/core/src/config.rs | 122 +++---- codex-rs/core/src/model_family.rs | 7 + codex-rs/core/src/model_provider_info.rs | 53 +-- codex-rs/core/src/tools/executor.rs | 314 ++++++++++++++++++ codex-rs/core/src/tools/mod.rs | 1 + codex-rs/core/src/tools/registry.rs | 71 +++- codex-rs/core/src/tools/router.rs | 36 +- codex-rs/core/src/tools/spec.rs | 29 +- .../core/tests/chat_completions_payload.rs | 6 +- codex-rs/core/tests/chat_completions_sse.rs | 7 +- codex-rs/core/tests/suite/client.rs | 73 +++- .../suite/stream_error_allows_next_turn.rs | 6 +- .../core/tests/suite/stream_no_completed.rs | 6 +- 18 files changed, 690 insertions(+), 312 deletions(-) create mode 100644 codex-rs/core/src/tools/executor.rs diff --git a/codex-rs/app-server-protocol/src/protocol.rs b/codex-rs/app-server-protocol/src/protocol.rs index 4bc10b2e65..721f4ecb46 100644 --- a/codex-rs/app-server-protocol/src/protocol.rs +++ b/codex-rs/app-server-protocol/src/protocol.rs @@ -515,6 +515,8 @@ pub struct Tools { pub web_search: Option, #[serde(skip_serializing_if = "Option::is_none")] pub view_image: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub parallel_read_only: Option, // todo check this } /// MCP representation of a [`codex_core::config_types::SandboxWorkspaceWrite`]. diff --git a/codex-rs/app-server/tests/suite/config.rs b/codex-rs/app-server/tests/suite/config.rs index 577eeb388d..85a215d44e 100644 --- a/codex-rs/app-server/tests/suite/config.rs +++ b/codex-rs/app-server/tests/suite/config.rs @@ -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([( diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 9e0c1c0430..7cf12fbba3 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -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(); diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 8e98bd6918..55ce16f1af 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -36,6 +36,9 @@ pub struct Prompt { /// Optional the output schema for the model's response. pub output_schema: Option, + + /// 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 = vec![]; + let tools: Vec = 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)) + ); + } } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 5884c6d4b6..343dafe1c4 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -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, + turn_context: Arc, turn_diff_tracker: &mut TurnDiffTracker, sub_id: String, input: Vec, ) -> CodexResult { 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, -} - #[derive(Debug)] struct TurnRunResult { processed_items: Vec, total_token_usage: Option, } - async fn try_run_turn( - router: &crate::tools::Router, - sess: &Session, - turn_context: &TurnContext, + mut tool_executor: ToolCallExecutor, + sess: Arc, + turn_context: Arc, turn_diff_tracker: &mut TurnDiffTracker, sub_id: &str, prompt: &Prompt, ) -> CodexResult { - // 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::>(); - // 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> { - 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 { 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, diff --git a/codex-rs/core/src/config.rs b/codex-rs/core/src/config.rs index af595b495d..5eeef60bc2 100644 --- a/codex-rs/core/src/config.rs +++ b/codex-rs/core/src/config.rs @@ -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, + + #[serde(default)] + pub parallel_read_only: Option, } impl From for Tools { @@ -774,6 +780,7 @@ impl From 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, 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(), diff --git a/codex-rs/core/src/model_family.rs b/codex-rs/core/src/model_family.rs index 54c18daec3..a8adc94b41 100644 --- a/codex-rs/core/src/model_family.rs +++ b/codex-rs/core/src/model_family.rs @@ -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 { 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, } } diff --git a/codex-rs/core/src/model_provider_info.rs b/codex-rs/core/src/model_provider_info.rs index a5707b34d1..ac51660f79 100644 --- a/codex-rs/core/src/model_provider_info.rs +++ b/codex-rs/core/src/model_provider_info.rs @@ -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 { 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()); diff --git a/codex-rs/core/src/tools/executor.rs b/codex-rs/core/src/tools/executor.rs new file mode 100644 index 0000000000..54985752c6 --- /dev/null +++ b/codex-rs/core/src/tools/executor.rs @@ -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, +} + +pub(crate) struct ToolCallExecutor { + router: Arc, + session: Arc, + turn_context: Arc, + allow_parallel_read_only: bool, + read_only_tasks: JoinSet<(usize, Result)>, + read_only_meta: HashMap, + processed_items: Vec, +} + +impl ToolCallExecutor { + pub(crate) fn new( + router: Arc, + session: Arc, // todo why + turn_context: Arc, // 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 { + 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) -> String { + if let Some(s) = payload.downcast_ref::<&str>() { + (*s).to_string() + } else if let Some(s) = payload.downcast_ref::() { + s.clone() + } else { + "panic without message".to_string() + } + } +} diff --git a/codex-rs/core/src/tools/mod.rs b/codex-rs/core/src/tools/mod.rs index 0033702f1a..9997e32e4d 100644 --- a/codex-rs/core/src/tools/mod.rs +++ b/codex-rs/core/src/tools/mod.rs @@ -1,4 +1,5 @@ pub mod context; +pub mod executor; pub(crate) mod handlers; pub mod registry; pub mod router; diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index 8ec4125495..e8016ba554 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -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, + capabilities: ToolCapabilities, +} + #[async_trait] pub trait ToolHandler: Send + Sync { fn kind(&self) -> ToolKind; @@ -36,17 +57,20 @@ pub trait ToolHandler: Send + Sync { -> Result; } +#[derive(Clone)] pub struct ToolRegistry { - handlers: HashMap>, + handlers: HashMap, } impl ToolRegistry { - pub fn new(handlers: HashMap>) -> Self { - Self { handlers } + pub fn capabilities(&self, name: &str) -> Option { + self.handlers.get(name).map(|entry| entry.capabilities) } - pub fn handler(&self, name: &str) -> Option> { - 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>, + handlers: HashMap, } impl ToolRegistryBuilder { @@ -148,10 +174,33 @@ impl ToolRegistryBuilder { } pub fn register_handler(&mut self, name: impl Into, handler: Arc) { + self.register_with_capabilities(name, handler, ToolCapabilities::mutating()); + } + + pub fn register_read_only_handler( + &mut self, + name: impl Into, + handler: Arc, + ) { + self.register_with_capabilities(name, handler, ToolCapabilities::read_only()); + } + + pub fn register_with_capabilities( + &mut self, + name: impl Into, + handler: Arc, + 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, + } } } diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index 8ec02ccd71..5de33ed152 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -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, @@ -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, 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 { diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 38a0df7bd2..8f5e66b436 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -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( diff --git a/codex-rs/core/tests/chat_completions_payload.rs b/codex-rs/core/tests/chat_completions_payload.rs index 9e10b37822..eeeb1ed3da 100644 --- a/codex-rs/core/tests/chat_completions_payload.rs +++ b/codex-rs/core/tests/chat_completions_payload.rs @@ -48,16 +48,12 @@ async fn run_request(input: Vec) -> 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() { diff --git a/codex-rs/core/tests/chat_completions_sse.rs b/codex-rs/core/tests/chat_completions_sse.rs index d8a8212932..682e451b6b 100644 --- a/codex-rs/core/tests/chat_completions_sse.rs +++ b/codex-rs/core/tests/chat_completions_sse.rs @@ -46,16 +46,11 @@ async fn run_stream_with_bytes(sse_body: &[u8]) -> Vec { 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() { diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index d4ff9da2d6..c6f0c1fa0b 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -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::().unwrap(); + assert_eq!( + request_body.get("parallel_tool_calls"), + Some(&serde_json::Value::Bool(true)) + ); +} diff --git a/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs b/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs index 9c32c351e0..905694ac18 100644 --- a/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs +++ b/codex-rs/core/tests/suite/stream_error_allows_next_turn.rs @@ -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() diff --git a/codex-rs/core/tests/suite/stream_no_completed.rs b/codex-rs/core/tests/suite/stream_no_completed.rs index ce33d8075f..c6356d2a80 100644 --- a/codex-rs/core/tests/suite/stream_no_completed.rs +++ b/codex-rs/core/tests/suite/stream_no_completed.rs @@ -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()