openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-31 19:28:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 99ef3ed879 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-04-29 10:03:45 -07:00
committed by GitHub
parent e953514cb6 99ef3ed879
commit 2ec581eba5
51 changed files with 1933 additions and 1259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/Cargo.toml
+1
View File
@@ -47,6 +47,7 @@ codex-shell-command = { workspace = true }
codex-execpolicy = { workspace = true }
codex-git-utils = { workspace = true }
codex-hooks = { workspace = true }
codex-kernel = { workspace = true }
codex-network-proxy = { workspace = true }
codex-otel = { workspace = true }
codex-plugin = { workspace = true }
codex-rs/core/src/client_common.rs
+1 -159
View File
@@ -1,14 +1,7 @@
pub use codex_api::ResponseEvent;
use codex_config::types::Personality;
pub use codex_kernel::Prompt;
use codex_protocol::error::Result;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::FunctionCallOutputBody;
use codex_protocol::models::ResponseItem;
use codex_tools::ToolSpec;
use futures::Stream;
use serde::Deserialize;
use serde_json::Value;
use std::collections::HashSet;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
@@ -23,157 +16,6 @@ pub const REVIEW_EXIT_SUCCESS_TMPL: &str = include_str!("../templates/review/exi
pub const REVIEW_EXIT_INTERRUPTED_TMPL: &str =
include_str!("../templates/review/exit_interrupted.xml");
/// API request payload for a single model turn
#[derive(Debug, Clone)]
pub struct Prompt {
/// Conversation context input items.
pub input: Vec<ResponseItem>,
/// Tools available to the model, including additional tools sourced from
/// external MCP servers.
pub(crate) tools: Vec<ToolSpec>,
/// Whether parallel tool calls are permitted for this prompt.
pub(crate) parallel_tool_calls: bool,
pub base_instructions: BaseInstructions,
/// Optionally specify the personality of the model.
pub personality: Option<Personality>,
/// Optional the output schema for the model's response.
pub output_schema: Option<Value>,
/// Whether the Responses API should strictly validate `output_schema`.
pub output_schema_strict: bool,
}
impl Default for Prompt {
fn default() -> Self {
Self {
input: Vec::new(),
tools: Vec::new(),
parallel_tool_calls: false,
base_instructions: BaseInstructions::default(),
personality: None,
output_schema: None,
output_schema_strict: true,
}
}
}
impl Prompt {
pub(crate) fn get_formatted_input(&self) -> Vec<ResponseItem> {
let mut input = self.input.clone();
// when using the *Freeform* apply_patch tool specifically, tool outputs
// should be structured text, not json. Do NOT reserialize when using
// the Function tool - note that this differs from the check above for
// instructions. We declare the result as a named variable for clarity.
let is_freeform_apply_patch_tool_present = self.tools.iter().any(|tool| match tool {
ToolSpec::Freeform(f) => f.name == "apply_patch",
_ => false,
});
if is_freeform_apply_patch_tool_present {
reserialize_shell_outputs(&mut input);
}
input
}
}
fn reserialize_shell_outputs(items: &mut [ResponseItem]) {
let mut shell_call_ids: HashSet<String> = HashSet::new();
items.iter_mut().for_each(|item| match item {
ResponseItem::LocalShellCall { call_id, id, .. } => {
if let Some(identifier) = call_id.clone().or_else(|| id.clone()) {
shell_call_ids.insert(identifier);
}
}
ResponseItem::CustomToolCall {
id: _,
status: _,
call_id,
name,
input: _,
} => {
if name == "apply_patch" {
shell_call_ids.insert(call_id.clone());
}
}
ResponseItem::FunctionCall { name, call_id, .. }
if is_shell_tool_name(name) || name == "apply_patch" =>
{
shell_call_ids.insert(call_id.clone());
}
ResponseItem::FunctionCallOutput {
call_id, output, ..
}
| ResponseItem::CustomToolCallOutput {
call_id, output, ..
} => {
if shell_call_ids.remove(call_id)
&& let Some(structured) = output
.text_content()
.and_then(parse_structured_shell_output)
{
output.body = FunctionCallOutputBody::Text(structured);
}
}
_ => {}
})
}
fn is_shell_tool_name(name: &str) -> bool {
matches!(name, "shell" | "container.exec")
}
#[derive(Deserialize)]
struct ExecOutputJson {
output: String,
metadata: ExecOutputMetadataJson,
}
#[derive(Deserialize)]
struct ExecOutputMetadataJson {
exit_code: i32,
duration_seconds: f32,
}
fn parse_structured_shell_output(raw: &str) -> Option<String> {
let parsed: ExecOutputJson = serde_json::from_str(raw).ok()?;
Some(build_structured_output(&parsed))
}
fn build_structured_output(parsed: &ExecOutputJson) -> String {
let mut sections = Vec::new();
sections.push(format!("Exit code: {}", parsed.metadata.exit_code));
sections.push(format!(
"Wall time: {} seconds",
parsed.metadata.duration_seconds
));
let mut output = parsed.output.clone();
if let Some((stripped, total_lines)) = strip_total_output_header(&parsed.output) {
sections.push(format!("Total output lines: {total_lines}"));
output = stripped.to_string();
}
sections.push("Output:".to_string());
sections.push(output);
sections.join("\n")
}
fn strip_total_output_header(output: &str) -> Option<(&str, u32)> {
let after_prefix = output.strip_prefix("Total output lines: ")?;
let (total_segment, remainder) = after_prefix.split_once('\n')?;
let total_lines = total_segment.parse::<u32>().ok()?;
let remainder = remainder.strip_prefix('\n').unwrap_or(remainder);
Some((remainder, total_lines))
}
pub struct ResponseStream {
pub(crate) rx_event: mpsc::Receiver<Result<ResponseEvent>>,
/// Signals the mapper task that the consumer stopped polling before the
codex-rs/core/src/client_common_tests.rs
+1 -65
View File
@@ -3,11 +3,9 @@ use codex_api::ResponsesApiRequest;
use codex_api::TextControls;
use codex_api::create_text_param_for_request;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseItem;
use pretty_assertions::assert_eq;
use super::*;
#[test]
fn serializes_text_verbosity_when_set() {
let input: Vec<ResponseItem> = vec![];
@@ -166,65 +164,3 @@ fn serializes_flex_service_tier_when_set() {
Some("flex")
);
}
#[test]
fn reserializes_shell_outputs_for_function_and_custom_tool_calls() {
let raw_output = r#"{"output":"hello","metadata":{"exit_code":0,"duration_seconds":0.5}}"#;
let expected_output = "Exit code: 0\nWall time: 0.5 seconds\nOutput:\nhello";
let mut items = vec![
ResponseItem::FunctionCall {
id: None,
name: "shell".to_string(),
namespace: None,
arguments: "{}".to_string(),
call_id: "call-1".to_string(),
},
ResponseItem::FunctionCallOutput {
call_id: "call-1".to_string(),
output: FunctionCallOutputPayload::from_text(raw_output.to_string()),
},
ResponseItem::CustomToolCall {
id: None,
status: None,
call_id: "call-2".to_string(),
name: "apply_patch".to_string(),
input: "*** Begin Patch".to_string(),
},
ResponseItem::CustomToolCallOutput {
call_id: "call-2".to_string(),
name: None,
output: FunctionCallOutputPayload::from_text(raw_output.to_string()),
},
];
reserialize_shell_outputs(&mut items);
assert_eq!(
items,
vec![
ResponseItem::FunctionCall {
id: None,
name: "shell".to_string(),
namespace: None,
arguments: "{}".to_string(),
call_id: "call-1".to_string(),
},
ResponseItem::FunctionCallOutput {
call_id: "call-1".to_string(),
output: FunctionCallOutputPayload::from_text(expected_output.to_string()),
},
ResponseItem::CustomToolCall {
id: None,
status: None,
call_id: "call-2".to_string(),
name: "apply_patch".to_string(),
input: "*** Begin Patch".to_string(),
},
ResponseItem::CustomToolCallOutput {
call_id: "call-2".to_string(),
name: None,
output: FunctionCallOutputPayload::from_text(expected_output.to_string()),
},
]
);
}
codex-rs/core/src/config/mod.rs
+25 -1
View File
@@ -8,6 +8,7 @@ use crate::windows_sandbox::WindowsSandboxLevelExt;
use crate::windows_sandbox::resolve_windows_sandbox_mode;
use crate::windows_sandbox::resolve_windows_sandbox_private_desktop;
use codex_config::CloudRequirementsLoader;
use codex_config::ConfigLayerSource;
use codex_config::ConfigLayerStack;
use codex_config::ConfigLayerStackOrdering;
use codex_config::ConfigRequirements;
@@ -1558,7 +1559,30 @@ fn resolve_permission_config_syntax(
sandbox_mode_override: Option<SandboxMode>,
profile_sandbox_mode: Option<SandboxMode>,
) -> Option<PermissionConfigSyntax> {
if sandbox_mode_override.is_some() || profile_sandbox_mode.is_some() {
if sandbox_mode_override.is_some() {
return Some(PermissionConfigSyntax::Legacy);
}
let session_flags_select_profiles = config_layer_stack
.get_layers(
ConfigLayerStackOrdering::HighestPrecedenceFirst,
/*include_disabled*/ false,
)
.into_iter()
.find(|layer| matches!(layer.name, ConfigLayerSource::SessionFlags))
.and_then(|layer| {
layer
.config
.clone()
.try_into::<PermissionSelectionToml>()
.ok()
})
.is_some_and(|selection| selection.default_permissions.is_some());
if session_flags_select_profiles {
return Some(PermissionConfigSyntax::Profiles);
}
if profile_sandbox_mode.is_some() {
return Some(PermissionConfigSyntax::Legacy);
}
codex-rs/core/src/connectors.rs
+1 -1
View File
@@ -201,7 +201,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status(
config.codex_linux_sandbox_exe.clone(),
)?;
let environment_manager =
EnvironmentManager::new(EnvironmentManagerArgs::from_env(local_runtime_paths));
EnvironmentManager::new(EnvironmentManagerArgs::new(local_runtime_paths)).await;
list_accessible_connectors_from_mcp_tools_with_environment_manager(
config,
force_refetch,
codex-rs/core/src/environment_selection.rs
+6 -9
View File
@@ -61,7 +61,6 @@ pub(crate) fn selected_primary_environment(
#[cfg(test)]
mod tests {
use codex_exec_server::EnvironmentManagerArgs;
use codex_exec_server::ExecServerRuntimePaths;
use codex_exec_server::REMOTE_ENVIRONMENT_ID;
use codex_protocol::protocol::TurnEnvironmentSelection;
@@ -81,10 +80,11 @@ mod tests {
#[tokio::test]
async fn default_thread_environment_selections_use_manager_default_id() {
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
let manager = EnvironmentManager::new(EnvironmentManagerArgs {
exec_server_url: Some("ws://127.0.0.1:8765".to_string()),
local_runtime_paths: test_runtime_paths(),
});
let manager = EnvironmentManager::create_for_tests(
Some("ws://127.0.0.1:8765".to_string()),
test_runtime_paths(),
)
.await;
assert_eq!(
default_thread_environment_selections(&manager, &cwd),
@@ -98,10 +98,7 @@ mod tests {
#[tokio::test]
async fn default_thread_environment_selections_empty_when_default_disabled() {
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
let manager = EnvironmentManager::new(EnvironmentManagerArgs {
exec_server_url: Some("none".to_string()),
local_runtime_paths: test_runtime_paths(),
});
let manager = EnvironmentManager::disabled_for_tests(test_runtime_paths());
assert_eq!(
default_thread_environment_selections(&manager, &cwd),
codex-rs/core/src/prompt_debug.rs
+1 -3
View File
@@ -45,9 +45,7 @@ pub async fn build_prompt_input(
.features
.enabled(Feature::DefaultModeRequestUserInput),
},
Arc::new(EnvironmentManager::new(EnvironmentManagerArgs::from_env(
local_runtime_paths,
))),
Arc::new(EnvironmentManager::new(EnvironmentManagerArgs::new(local_runtime_paths)).await),
/*analytics_events_client*/ None,
);
let thread = thread_manager.start_thread(config).await?;
codex-rs/core/src/session/mod.rs
+1
View File
@@ -189,6 +189,7 @@ mod mcp;
mod multi_agents;
mod review;
mod rollout_reconstruction;
mod sampling_loop;
#[allow(clippy::module_inception)]
pub(crate) mod session;
pub(crate) mod turn;
codex-rs/core/src/session/sampling_loop.rs
+221
View File
@@ -0,0 +1,221 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use crate::SkillLoadOutcome;
use crate::client::ModelClientSession;
use crate::session::session::Session;
use crate::session::turn::build_prompt_config;
use crate::session::turn::build_tool_config;
use crate::session::turn::built_tools;
use crate::session::turn::try_run_sampling_request;
use crate::session::turn_context::TurnContext;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::parallel::ToolCallRuntime;
use codex_kernel::PreparedSamplingRequest;
use codex_kernel::Prompt;
use codex_kernel::SamplingLoopHost;
use codex_kernel::SamplingRequestResult;
use codex_kernel::run_sampling_request_loop;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::StreamErrorEvent;
use codex_protocol::protocol::WarningEvent;
use tokio_util::sync::CancellationToken;
use tracing::instrument;
use tracing::warn;
struct SamplingRequestRuntime {
_code_mode_worker: Option<codex_code_mode::CodeModeTurnWorker>,
}
#[allow(clippy::too_many_arguments)]
#[instrument(
level = "trace",
skip_all,
fields(
turn_id = %turn_context.sub_id,
model = %turn_context.model_info.slug,
cwd = %turn_context.cwd.display()
)
)]
pub(super) async fn run_sampling_request(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: SharedTurnDiffTracker,
client_session: &mut ModelClientSession,
turn_metadata_header: Option<&str>,
input: Vec<ResponseItem>,
explicitly_enabled_connectors: &HashSet<String>,
skills_outcome: Option<&SkillLoadOutcome>,
cancellation_token: CancellationToken,
) -> CodexResult<SamplingRequestResult> {
let host = CoreSamplingLoopHost {
sess,
turn_context,
turn_diff_tracker,
turn_metadata_header,
explicitly_enabled_connectors,
skills_outcome,
};
run_sampling_request_loop(&host, client_session, input, cancellation_token).await
}
struct CoreSamplingLoopHost<'a> {
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: SharedTurnDiffTracker,
turn_metadata_header: Option<&'a str>,
explicitly_enabled_connectors: &'a HashSet<String>,
skills_outcome: Option<&'a SkillLoadOutcome>,
}
impl SamplingLoopHost for CoreSamplingLoopHost<'_> {
type ClientSession = ModelClientSession;
type Runtime = SamplingRequestRuntime;
type Tools = ToolCallRuntime;
async fn prepare_sampling_request(
&self,
input: &[ResponseItem],
cancellation_token: &CancellationToken,
) -> CodexResult<PreparedSamplingRequest<Self::Runtime, Self::Tools>> {
let router = built_tools(
self.sess.as_ref(),
self.turn_context.as_ref(),
input,
self.explicitly_enabled_connectors,
self.skills_outcome,
cancellation_token,
)
.await?;
let base_instructions = self.sess.get_base_instructions().await;
let tool_runtime = ToolCallRuntime::new(
Arc::clone(&router),
Arc::clone(&self.sess),
Arc::clone(&self.turn_context),
Arc::clone(&self.turn_diff_tracker),
);
let code_mode_worker = self
.sess
.services
.code_mode_service
.start_turn_worker(
&self.sess,
&self.turn_context,
Arc::clone(&router),
Arc::clone(&self.turn_diff_tracker),
)
.await;
Ok(PreparedSamplingRequest {
prompt_config: build_prompt_config(self.turn_context.as_ref(), base_instructions),
tool_config: build_tool_config(router.as_ref(), self.turn_context.as_ref()),
runtime: SamplingRequestRuntime {
_code_mode_worker: code_mode_worker,
},
tools: tool_runtime,
})
}
async fn history_prompt_input(&self) -> Vec<ResponseItem> {
self.sess
.clone_history()
.await
.for_prompt(&self.turn_context.model_info.input_modalities)
}
async fn run_single_sampling_request(
&self,
_runtime: &Self::Runtime,
tools: &Self::Tools,
client_session: &mut Self::ClientSession,
prompt: &Prompt,
cancellation_token: CancellationToken,
) -> CodexResult<SamplingRequestResult> {
try_run_sampling_request(
tools.clone(),
Arc::clone(&self.sess),
Arc::clone(&self.turn_context),
client_session,
self.turn_metadata_header,
Arc::clone(&self.turn_diff_tracker),
prompt,
cancellation_token,
)
.await
}
fn stream_max_retries(&self) -> u64 {
self.turn_context.provider.info().stream_max_retries()
}
fn try_switch_fallback_transport(&self, client_session: &mut Self::ClientSession) -> bool {
client_session.try_switch_fallback_transport(
&self.turn_context.session_telemetry,
&self.turn_context.model_info,
)
}
fn should_notify_stream_retry(&self, retries: u64, _err: &CodexErr) -> bool {
retries > 1
|| cfg!(debug_assertions)
|| !self
.sess
.services
.model_client
.responses_websocket_enabled()
}
async fn handle_context_window_exceeded(&self) {
self.sess.set_total_tokens_full(&self.turn_context).await;
}
async fn handle_usage_limit_reached(&self, rate_limits: Option<RateLimitSnapshot>) {
if let Some(rate_limits) = rate_limits {
self.sess
.update_rate_limits(&self.turn_context, rate_limits)
.await;
}
}
async fn notify_fallback_to_http(&self, err: &CodexErr) {
self.sess
.send_event(
&self.turn_context,
EventMsg::Warning(WarningEvent {
message: format!("Falling back from WebSockets to HTTPS transport. {err:#}"),
}),
)
.await;
}
async fn notify_stream_retry(
&self,
retries: u64,
max_retries: u64,
delay: Duration,
err: &CodexErr,
) {
warn!(
"stream disconnected - retrying sampling request ({retries}/{max_retries} in {delay:?})..."
);
self.sess
.send_event(
&self.turn_context,
EventMsg::StreamError(StreamErrorEvent {
message: format!("Reconnecting... {retries}/{max_retries}"),
codex_error_info: Some(CodexErrorInfo::ResponseStreamDisconnected {
http_status_code: err.http_status_code_value(),
}),
additional_details: Some(err.to_string()),
}),
)
.await;
}
}
codex-rs/core/src/session/tests.rs
+3 -3
View File
@@ -5615,7 +5615,7 @@ fn emit_thread_start_skill_metrics_records_enabled_kept_and_truncated_values() {
assert_eq!(
rendered.warning_message,
Some(
"Warning: Exceeded skills context budget. All skill descriptions were removed and 1 additional skill was not included in the model-visible skills list."
"Exceeded skills context budget. All skill descriptions were removed and 1 additional skill was not included in the model-visible skills list."
.to_string()
)
);
@@ -5726,7 +5726,7 @@ async fn build_initial_context_emits_thread_start_skill_warning_on_repeated_buil
assert!(matches!(
warning_event.msg,
EventMsg::Warning(WarningEvent { message })
if message == "Warning: Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list."
if message == "Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list."
));
let _ = session.build_initial_context(&turn_context).await;
@@ -5737,7 +5737,7 @@ async fn build_initial_context_emits_thread_start_skill_warning_on_repeated_buil
assert!(matches!(
warning_event.msg,
EventMsg::Warning(WarningEvent { message })
if message == "Warning: Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list."
if message == "Exceeded skills context budget of 2%. All skill descriptions were removed and 2 additional skills were not included in the model-visible skills list."
));
}
codex-rs/core/src/session/turn.rs
+22 -163
View File
@@ -40,6 +40,7 @@ use crate::parse_turn_item;
use crate::plugins::build_plugin_injections;
use crate::resolve_skill_dependencies_for_turn;
use crate::session::PreviousTurnSettings;
use crate::session::sampling_loop::run_sampling_request;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::stream_events_utils::HandleOutputCtx;
@@ -57,7 +58,6 @@ use crate::tools::router::ToolRouterParams;
use crate::turn_diff_tracker::TurnDiffTracker;
use crate::turn_timing::record_turn_ttft_metric;
use crate::unavailable_tool::collect_unavailable_called_tools;
use crate::util::backoff;
use crate::util::error_or_panic;
use codex_analytics::AppInvocation;
use codex_analytics::CompactionPhase;
@@ -71,6 +71,9 @@ use codex_hooks::HookEvent;
use codex_hooks::HookEventAfterAgent;
use codex_hooks::HookPayload;
use codex_hooks::HookResult;
use codex_kernel::PromptConfig;
use codex_kernel::SamplingRequestResult;
use codex_kernel::ToolConfig;
use codex_protocol::config_types::ModeKind;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
@@ -932,16 +935,11 @@ fn connector_inserted_in_messages(
connector_count == 1 && skill_count == 0 && mention_names_lower.contains(&mention_slug)
}
pub(crate) fn build_prompt(
input: Vec<ResponseItem>,
router: &ToolRouter,
pub(super) fn build_prompt_config(
turn_context: &TurnContext,
base_instructions: BaseInstructions,
) -> Prompt {
Prompt {
input,
tools: router.model_visible_specs(),
parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
) -> PromptConfig {
PromptConfig {
base_instructions,
personality: turn_context.personality,
output_schema: turn_context.final_output_json_schema.clone(),
@@ -951,156 +949,23 @@ pub(crate) fn build_prompt(
}
}
#[allow(clippy::too_many_arguments)]
#[instrument(level = "trace",
skip_all,
fields(
turn_id = %turn_context.sub_id,
model = %turn_context.model_info.slug,
cwd = %turn_context.cwd.display()
)
)]
async fn run_sampling_request(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: SharedTurnDiffTracker,
client_session: &mut ModelClientSession,
turn_metadata_header: Option<&str>,
input: Vec<ResponseItem>,
explicitly_enabled_connectors: &HashSet<String>,
skills_outcome: Option<&SkillLoadOutcome>,
cancellation_token: CancellationToken,
) -> CodexResult<SamplingRequestResult> {
let router = built_tools(
sess.as_ref(),
turn_context.as_ref(),
&input,
explicitly_enabled_connectors,
skills_outcome,
&cancellation_token,
)
.await?;
let base_instructions = sess.get_base_instructions().await;
let tool_runtime = ToolCallRuntime::new(
Arc::clone(&router),
Arc::clone(&sess),
Arc::clone(&turn_context),
Arc::clone(&turn_diff_tracker),
);
let _code_mode_worker = sess
.services
.code_mode_service
.start_turn_worker(
&sess,
&turn_context,
Arc::clone(&router),
Arc::clone(&turn_diff_tracker),
)
.await;
let mut retries = 0;
let mut initial_input = Some(input);
loop {
let prompt_input = if let Some(input) = initial_input.take() {
input
} else {
sess.clone_history()
.await
.for_prompt(&turn_context.model_info.input_modalities)
};
let prompt = build_prompt(
prompt_input,
router.as_ref(),
turn_context.as_ref(),
base_instructions.clone(),
);
let err = match try_run_sampling_request(
tool_runtime.clone(),
Arc::clone(&sess),
Arc::clone(&turn_context),
client_session,
turn_metadata_header,
Arc::clone(&turn_diff_tracker),
&prompt,
cancellation_token.child_token(),
)
.await
{
Ok(output) => {
return Ok(output);
}
Err(CodexErr::ContextWindowExceeded) => {
sess.set_total_tokens_full(&turn_context).await;
return Err(CodexErr::ContextWindowExceeded);
}
Err(CodexErr::UsageLimitReached(e)) => {
let rate_limits = e.rate_limits.clone();
if let Some(rate_limits) = rate_limits {
sess.update_rate_limits(&turn_context, *rate_limits).await;
}
return Err(CodexErr::UsageLimitReached(e));
}
Err(err) => err,
};
if !err.is_retryable() {
return Err(err);
}
// Use the configured provider-specific stream retry budget.
let max_retries = turn_context.provider.info().stream_max_retries();
if retries >= max_retries
&& client_session.try_switch_fallback_transport(
&turn_context.session_telemetry,
&turn_context.model_info,
)
{
sess.send_event(
&turn_context,
EventMsg::Warning(WarningEvent {
message: format!("Falling back from WebSockets to HTTPS transport. {err:#}"),
}),
)
.await;
retries = 0;
continue;
}
if retries < max_retries {
retries += 1;
let delay = match &err {
CodexErr::Stream(_, requested_delay) => {
requested_delay.unwrap_or_else(|| backoff(retries))
}
_ => backoff(retries),
};
warn!(
"stream disconnected - retrying sampling request ({retries}/{max_retries} in {delay:?})...",
);
// In release builds, hide the first websocket retry notification to reduce noisy
// transient reconnect messages. In debug builds, keep full visibility for diagnosis.
let report_error = retries > 1
|| cfg!(debug_assertions)
|| !sess.services.model_client.responses_websocket_enabled();
if report_error {
// Surface retry information to any UI/front‑end so the
// user understands what is happening instead of staring
// at a seemingly frozen screen.
sess.notify_stream_error(
&turn_context,
format!("Reconnecting... {retries}/{max_retries}"),
err,
)
.await;
}
tokio::time::sleep(delay).await;
} else {
return Err(err);
}
pub(super) fn build_tool_config(router: &ToolRouter, turn_context: &TurnContext) -> ToolConfig {
ToolConfig {
tools: router.model_visible_specs(),
parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
}
}
pub(crate) fn build_prompt(
input: Vec<ResponseItem>,
router: &ToolRouter,
turn_context: &TurnContext,
base_instructions: BaseInstructions,
) -> Prompt {
build_prompt_config(turn_context, base_instructions)
.build_prompt(input, &build_tool_config(router, turn_context))
}
#[expect(
clippy::await_holding_invalid_type,
reason = "tool router construction reads through the session-owned manager guard"
@@ -1243,12 +1108,6 @@ pub(crate) async fn built_tools(
)))
}
#[derive(Debug)]
struct SamplingRequestResult {
needs_follow_up: bool,
last_agent_message: Option<String>,
}
/// Ephemeral per-response state for streaming a single proposed plan.
/// This is intentionally not persisted or stored in session/state since it
/// only exists while a response is actively streaming. The final plan text
@@ -1806,7 +1665,7 @@ async fn drain_in_flight(
model = %turn_context.model_info.slug
)
)]
async fn try_run_sampling_request(
pub(super) async fn try_run_sampling_request(
tool_runtime: ToolCallRuntime,
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
codex-rs/core/src/stream_events_utils.rs
+12 -94
View File
@@ -10,18 +10,18 @@ use tokio_util::sync::CancellationToken;
use crate::context::ContextualUserFragment;
use crate::context::ImageGenerationInstructions;
use crate::function_tool::FunctionCallError;
use crate::parse_turn_item;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::tools::parallel::ToolCallRuntime;
use crate::tools::router::ToolRouter;
use codex_kernel::CompletedResponseItem;
use codex_kernel::KernelToolExecutor;
use codex_kernel::execute_tool_call_with_default_output;
use codex_kernel::response_input_to_response_item;
use codex_memories_read::citations::parse_memory_citation;
use codex_memories_read::citations::thread_ids_from_memory_citation;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result;
use codex_protocol::models::FunctionCallOutputBody;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
@@ -225,9 +225,8 @@ pub(crate) async fn handle_output_item_done(
let mut output = OutputItemResult::default();
let plan_mode = ctx.turn_context.collaboration_mode.mode == ModeKind::Plan;
match ToolRouter::build_tool_call(ctx.sess.as_ref(), item.clone()).await {
// The model emitted a tool call; log it, persist the item immediately, and queue the tool execution.
Ok(Some(call)) => {
match ctx.tool_runtime.classify_response_item(item).await? {
CompletedResponseItem::ToolCall { item, call } => {
ctx.sess
.accept_mailbox_delivery_for_current_turn(&ctx.turn_context.sub_id)
.await;
@@ -244,17 +243,15 @@ pub(crate) async fn handle_output_item_done(
.await;
let cancellation_token = ctx.cancellation_token.child_token();
let tool_future: InFlightFuture<'static> = Box::pin(
ctx.tool_runtime
.clone()
.handle_tool_call(call, cancellation_token),
);
let tool_runtime = ctx.tool_runtime.clone();
let tool_future: InFlightFuture<'static> = Box::pin(async move {
execute_tool_call_with_default_output(&tool_runtime, call, cancellation_token).await
});
output.needs_follow_up = true;
output.tool_future = Some(tool_future);
}
// No tool call: convert messages/reasoning into turn items and mark them as complete.
Ok(None) => {
CompletedResponseItem::NonTool(item) => {
if let Some(turn_item) = handle_non_tool_response_item(
ctx.sess.as_ref(),
ctx.turn_context.as_ref(),
@@ -286,21 +283,7 @@ pub(crate) async fn handle_output_item_done(
output.last_agent_message = last_agent_message;
}
// Guardrail: the model issued a LocalShellCall without an id; surface the error back into history.
Err(FunctionCallError::MissingLocalShellCallId) => {
let msg = "LocalShellCall without call_id or id";
ctx.turn_context
.session_telemetry
.log_tool_failed("local_shell", msg);
tracing::error!(msg);
let response = ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: FunctionCallOutputPayload {
body: FunctionCallOutputBody::Text(msg.to_string()),
..Default::default()
},
};
CompletedResponseItem::ImmediateResponse { item, response } => {
record_completed_response_item(ctx.sess.as_ref(), ctx.turn_context.as_ref(), &item)
.await;
if let Some(response_item) = response_input_to_response_item(&response) {
@@ -314,32 +297,6 @@ pub(crate) async fn handle_output_item_done(
output.needs_follow_up = true;
}
// The tool request should be answered directly (or was denied); push that response into the transcript.
Err(FunctionCallError::RespondToModel(message)) => {
let response = ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: FunctionCallOutputPayload {
body: FunctionCallOutputBody::Text(message),
..Default::default()
},
};
record_completed_response_item(ctx.sess.as_ref(), ctx.turn_context.as_ref(), &item)
.await;
if let Some(response_item) = response_input_to_response_item(&response) {
ctx.sess
.record_conversation_items(
&ctx.turn_context,
std::slice::from_ref(&response_item),
)
.await;
}
output.needs_follow_up = true;
}
// A fatal error occurred; surface it back into history.
Err(FunctionCallError::Fatal(message)) => {
return Err(CodexErr::Fatal(message));
}
}
Ok(output)
@@ -465,45 +422,6 @@ fn completed_item_defers_mailbox_delivery_to_next_turn(
}
}
pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option<ResponseItem> {
match input {
ResponseInputItem::FunctionCallOutput { call_id, output } => {
Some(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output: output.clone(),
})
}
ResponseInputItem::CustomToolCallOutput {
call_id,
name,
output,
} => Some(ResponseItem::CustomToolCallOutput {
call_id: call_id.clone(),
name: name.clone(),
output: output.clone(),
}),
ResponseInputItem::McpToolCallOutput { call_id, output } => {
let output = output.as_function_call_output_payload();
Some(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output,
})
}
ResponseInputItem::ToolSearchOutput {
call_id,
status,
execution,
tools,
} => Some(ResponseItem::ToolSearchOutput {
call_id: Some(call_id.clone()),
status: status.clone(),
execution: execution.clone(),
tools: tools.clone(),
}),
_ => None,
}
}
#[cfg(test)]
#[path = "stream_events_utils_tests.rs"]
mod tests;
codex-rs/core/src/thread_manager_tests.rs
+13 -15
View File
@@ -47,20 +47,6 @@ fn assistant_msg(text: &str) -> ResponseItem {
}
}
fn disabled_environment_manager_for_tests() -> Arc<codex_exec_server::EnvironmentManager> {
let runtime_paths = codex_exec_server::ExecServerRuntimePaths::new(
std::env::current_exe().expect("current exe path"),
/*codex_linux_sandbox_exe*/ None,
)
.expect("runtime paths");
Arc::new(codex_exec_server::EnvironmentManager::new(
codex_exec_server::EnvironmentManagerArgs {
exec_server_url: Some("none".to_string()),
local_runtime_paths: runtime_paths,
},
))
}
fn contextual_user_interrupted_marker() -> ResponseItem {
interrupted_turn_history_marker(InterruptedTurnHistoryMarker::ContextualUser)
.expect("contextual-user interrupted marker should be enabled")
@@ -307,11 +293,23 @@ async fn start_thread_accepts_explicit_environment_when_default_environment_is_d
config.cwd = config.codex_home.abs();
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
let runtime_paths = codex_exec_server::ExecServerRuntimePaths::new(
std::env::current_exe().expect("current exe path"),
/*codex_linux_sandbox_exe*/ None,
)
.expect("runtime paths");
let environment_manager = Arc::new(
codex_exec_server::EnvironmentManager::create_for_tests(
Some("none".to_string()),
runtime_paths,
)
.await,
);
let manager = ThreadManager::with_models_provider_and_home_for_tests(
CodexAuth::from_api_key("dummy"),
config.model_provider.clone(),
config.codex_home.to_path_buf(),
disabled_environment_manager_for_tests(),
environment_manager,
);
let thread = manager
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
-370
View File
@@ -1,15 +1,10 @@
use super::*;
use crate::CodexThread;
use crate::ThreadManager;
use crate::config::AgentRoleConfig;
use crate::config::DEFAULT_AGENT_MAX_DEPTH;
use crate::context::TurnAborted;
use crate::function_tool::FunctionCallError;
use crate::session::tests::make_session_and_context;
use crate::session_prefix::format_subagent_notification_message;
use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
use crate::tools::context::ToolOutput;
use crate::tools::handlers::multi_agents_v2::CloseAgentHandler as CloseAgentHandlerV2;
use crate::tools::handlers::multi_agents_v2::FollowupTaskHandler as FollowupTaskHandlerV2;
@@ -134,121 +129,6 @@ model_reasoning_effort = "minimal"
role_name
}
fn history_contains_inter_agent_communication(
history_items: &[ResponseItem],
expected: &InterAgentCommunication,
) -> bool {
history_items.iter().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
if role != "assistant" {
return false;
}
content.iter().any(|content_item| match content_item {
ContentItem::OutputText { text } => {
serde_json::from_str::<InterAgentCommunication>(text)
.ok()
.as_ref()
== Some(expected)
}
ContentItem::InputText { .. } | ContentItem::InputImage { .. } => false,
})
})
}
async fn wait_for_turn_aborted(
thread: &Arc<CodexThread>,
expected_turn_id: &str,
expected_reason: TurnAbortReason,
) {
timeout(Duration::from_secs(5), async {
loop {
let event = thread
.next_event()
.await
.expect("child thread should emit events");
if matches!(
event.msg,
EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some(ref turn_id),
ref reason,
..
}) if turn_id == expected_turn_id && *reason == expected_reason
) {
break;
}
}
})
.await
.expect("expected child turn to be interrupted");
}
async fn wait_for_redirected_envelope_in_history(
thread: &Arc<CodexThread>,
expected: &InterAgentCommunication,
) {
timeout(Duration::from_secs(5), async {
loop {
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
let saw_envelope =
history_contains_inter_agent_communication(&history_items, expected);
let saw_user_message = history_items.iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text }
if text == &expected.content
))
)
});
if saw_envelope {
assert!(
!saw_user_message,
"redirected followup should be stored as an assistant envelope, not a plain user message"
);
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("redirected followup envelope should appear in history");
}
#[derive(Clone, Copy)]
struct NeverEndingTask;
impl SessionTask for NeverEndingTask {
fn kind(&self) -> TaskKind {
TaskKind::Regular
}
fn span_name(&self) -> &'static str {
"session_task.multi_agent_never_ending"
}
async fn run(
self: Arc<Self>,
_session: Arc<SessionTaskContext>,
_ctx: Arc<TurnContext>,
_input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> Option<String> {
cancellation_token.cancelled().await;
None
}
}
fn expect_text_output<T>(output: T) -> (String, Option<bool>)
where
T: ToolOutput,
@@ -1086,7 +966,6 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
function_payload(json!({
"target": "/root",
"message": "run this",
"interrupt": true
})),
))
.await
@@ -1485,255 +1364,6 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
)));
}
#[tokio::test]
async fn multi_agent_v2_followup_task_interrupts_busy_child_without_losing_message() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
session.conversation_id = root.thread_id;
let mut config = turn.config.as_ref().clone();
let _ = config.features.enable(Feature::MultiAgentV2);
turn.config = Arc::new(config);
let session = Arc::new(session);
let turn = Arc::new(turn);
let worker_path = AgentPath::try_from("/root/worker").expect("worker path");
let agent_id = session
.services
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
Op::CleanBackgroundTerminals,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
agent_path: Some(worker_path.clone()),
agent_nickname: None,
agent_role: None,
})),
crate::agent::control::SpawnAgentOptions::default(),
)
.await
.expect("worker spawn should succeed")
.thread_id;
let thread = manager
.get_thread(agent_id)
.await
.expect("worker thread should exist");
let active_turn = thread.codex.session.new_default_turn().await;
let interrupted_turn_id = active_turn.sub_id.clone();
thread
.codex
.session
.spawn_task(
Arc::clone(&active_turn),
vec![UserInput::Text {
text: "working".to_string(),
text_elements: Vec::new(),
}],
NeverEndingTask,
)
.await;
FollowupTaskHandlerV2
.handle(invocation(
session,
turn,
"followup_task",
function_payload(json!({
"target": agent_id.to_string(),
"message": "continue",
"interrupt": true
})),
))
.await
.expect("interrupting v2 followup_task should succeed");
let ops = manager.captured_ops();
let ops_for_agent: Vec<&Op> = ops
.iter()
.filter_map(|(id, op)| (*id == agent_id).then_some(op))
.collect();
assert!(ops_for_agent.iter().any(|op| matches!(op, Op::Interrupt)));
assert!(ops_for_agent.iter().any(|op| {
matches!(
op,
Op::InterAgentCommunication { communication }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/worker"
&& communication.other_recipients.is_empty()
&& communication.content == "continue"
)
}));
wait_for_turn_aborted(&thread, &interrupted_turn_id, TurnAbortReason::Interrupted).await;
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(
history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "developer"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text }
if text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"v2 interrupted-turn marker should be recorded as a developer input message"
);
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_GUIDANCE)
))
)),
"v2 interrupted-turn marker should not be recorded as a user message"
);
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "assistant"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"v2 interrupted-turn marker should not be recorded as an assistant message"
);
wait_for_redirected_envelope_in_history(
&thread,
&InterAgentCommunication::new(
AgentPath::root(),
worker_path,
Vec::new(),
"continue".to_string(),
/*trigger_turn*/ true,
),
)
.await;
let _ = thread
.submit(Op::Shutdown {})
.await
.expect("shutdown should submit");
}
#[tokio::test]
async fn multi_agent_v2_followup_task_can_disable_interrupted_marker() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
session.conversation_id = root.thread_id;
let mut config = turn.config.as_ref().clone();
let _ = config.features.enable(Feature::MultiAgentV2);
config.agent_interrupt_message_enabled = false;
turn.config = Arc::new(config);
let session = Arc::new(session);
let turn = Arc::new(turn);
let worker_path = AgentPath::try_from("/root/worker").expect("worker path");
let agent_id = session
.services
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
Op::CleanBackgroundTerminals,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
agent_path: Some(worker_path),
agent_nickname: None,
agent_role: None,
})),
crate::agent::control::SpawnAgentOptions::default(),
)
.await
.expect("worker spawn should succeed")
.thread_id;
let thread = manager
.get_thread(agent_id)
.await
.expect("worker thread should exist");
let active_turn = thread.codex.session.new_default_turn().await;
let interrupted_turn_id = active_turn.sub_id.clone();
thread
.codex
.session
.spawn_task(
Arc::clone(&active_turn),
vec![UserInput::Text {
text: "working".to_string(),
text_elements: Vec::new(),
}],
NeverEndingTask,
)
.await;
FollowupTaskHandlerV2
.handle(invocation(
session,
turn,
"followup_task",
function_payload(json!({
"target": agent_id.to_string(),
"message": "continue",
"interrupt": true
})),
))
.await
.expect("interrupting v2 followup_task should succeed");
wait_for_turn_aborted(&thread, &interrupted_turn_id, TurnAbortReason::Interrupted).await;
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { content, .. }
if content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_GUIDANCE)
|| text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"disabled interrupted-turn marker should not be recorded in history"
);
let _ = thread
.submit(Op::Shutdown {})
.await
.expect("shutdown should submit");
}
#[tokio::test]
async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() {
let (mut session, mut turn) = make_session_and_context().await;
codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs
-1
View File
@@ -25,7 +25,6 @@ impl ToolHandler for Handler {
MessageDeliveryMode::TriggerTurn,
args.target,
args.message,
args.interrupt,
)
.await
}
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+1 -20
View File
@@ -43,8 +43,6 @@ pub(crate) struct SendMessageArgs {
pub(crate) struct FollowupTaskArgs {
pub(crate) target: String,
pub(crate) message: String,
#[serde(default)]
pub(crate) interrupt: bool,
}
fn message_content(message: String) -> Result<String, FunctionCallError> {
@@ -62,16 +60,8 @@ pub(crate) async fn handle_message_string_tool(
mode: MessageDeliveryMode,
target: String,
message: String,
interrupt: bool,
) -> Result<FunctionToolOutput, FunctionCallError> {
handle_message_submission(
invocation,
mode,
target,
message_content(message)?,
interrupt,
)
.await
handle_message_submission(invocation, mode, target, message_content(message)?).await
}
async fn handle_message_submission(
@@ -79,7 +69,6 @@ async fn handle_message_submission(
mode: MessageDeliveryMode,
target: String,
prompt: String,
interrupt: bool,
) -> Result<FunctionToolOutput, FunctionCallError> {
let ToolInvocation {
session,
@@ -103,14 +92,6 @@ async fn handle_message_submission(
"Tasks can't be assigned to the root agent".to_string(),
));
}
if interrupt {
session
.services
.agent_control
.interrupt_agent(receiver_thread_id)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err))?;
}
session
.send_event(
&turn,
codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs
-1
View File
@@ -25,7 +25,6 @@ impl ToolHandler for Handler {
MessageDeliveryMode::QueueOnly,
args.target,
args.message,
/*interrupt*/ false,
)
.await
}
codex-rs/core/src/tools/parallel.rs
+90 -46
View File
@@ -20,8 +20,14 @@ use crate::tools::registry::ToolArgumentDiffConsumer;
use crate::tools::router::ToolCall;
use crate::tools::router::ToolCallSource;
use crate::tools::router::ToolRouter;
use codex_kernel::CompletedResponseItem;
use codex_kernel::KernelToolCall;
use codex_kernel::KernelToolExecutor;
use codex_kernel::ToolCallError;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_tools::ToolSpec;
#[derive(Clone)]
@@ -60,25 +66,6 @@ impl ToolCallRuntime {
self.router.create_diff_consumer(tool_name)
}
#[instrument(level = "trace", skip_all)]
pub(crate) fn handle_tool_call(
self,
call: ToolCall,
cancellation_token: CancellationToken,
) -> impl std::future::Future<Output = Result<ResponseInputItem, CodexErr>> {
let error_call = call.clone();
let future =
self.handle_tool_call_with_source(call, ToolCallSource::Direct, cancellation_token);
async move {
match future.await {
Ok(response) => Ok(response.into_response()),
Err(FunctionCallError::Fatal(message)) => Err(CodexErr::Fatal(message)),
Err(other) => Ok(Self::failure_response(error_call, other)),
}
}
.in_current_span()
}
#[instrument(level = "trace", skip_all)]
pub(crate) fn handle_tool_call_with_source(
self,
@@ -144,33 +131,6 @@ impl ToolCallRuntime {
}
impl ToolCallRuntime {
fn failure_response(call: ToolCall, err: FunctionCallError) -> ResponseInputItem {
let message = err.to_string();
match call.payload {
ToolPayload::ToolSearch { .. } => ResponseInputItem::ToolSearchOutput {
call_id: call.call_id,
status: "completed".to_string(),
execution: "client".to_string(),
tools: Vec::new(),
},
ToolPayload::Custom { .. } => ResponseInputItem::CustomToolCallOutput {
call_id: call.call_id,
name: None,
output: codex_protocol::models::FunctionCallOutputPayload {
body: codex_protocol::models::FunctionCallOutputBody::Text(message),
success: Some(false),
},
},
_ => ResponseInputItem::FunctionCallOutput {
call_id: call.call_id,
output: codex_protocol::models::FunctionCallOutputPayload {
body: codex_protocol::models::FunctionCallOutputBody::Text(message),
success: Some(false),
},
},
}
}
fn aborted_response(call: &ToolCall, secs: f32) -> AnyToolResult {
AnyToolResult {
call_id: call.call_id.clone(),
@@ -195,3 +155,87 @@ impl ToolCallRuntime {
}
}
}
impl KernelToolCall for ToolCall {
fn error_response(&self, message: String) -> ResponseInputItem {
failure_response(self, message)
}
}
impl KernelToolExecutor for ToolCallRuntime {
type Call = ToolCall;
async fn classify_response_item(
&self,
item: ResponseItem,
) -> CodexResult<CompletedResponseItem<Self::Call>> {
match ToolRouter::build_tool_call(self.session.as_ref(), item.clone()).await {
Ok(Some(call)) => Ok(CompletedResponseItem::ToolCall { item, call }),
Ok(None) => Ok(CompletedResponseItem::NonTool(item)),
Err(FunctionCallError::MissingLocalShellCallId) => {
let response = ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: codex_protocol::models::FunctionCallOutputPayload::from_text(
"LocalShellCall without call_id or id".to_string(),
),
};
Ok(CompletedResponseItem::ImmediateResponse { item, response })
}
Err(FunctionCallError::RespondToModel(message)) => {
let response = ResponseInputItem::FunctionCallOutput {
call_id: String::new(),
output: codex_protocol::models::FunctionCallOutputPayload::from_text(message),
};
Ok(CompletedResponseItem::ImmediateResponse { item, response })
}
Err(FunctionCallError::Fatal(message)) => Err(CodexErr::Fatal(message)),
}
}
async fn execute_tool_call(
&self,
call: Self::Call,
cancellation_token: CancellationToken,
) -> Result<ResponseInputItem, ToolCallError> {
match self
.clone()
.handle_tool_call_with_source(call, ToolCallSource::Direct, cancellation_token)
.await
{
Ok(result) => Ok(result.into_response()),
Err(FunctionCallError::RespondToModel(message)) => {
Err(ToolCallError::RespondToModel(message))
}
Err(FunctionCallError::MissingLocalShellCallId) => Err(ToolCallError::RespondToModel(
"LocalShellCall without call_id or id".to_string(),
)),
Err(FunctionCallError::Fatal(message)) => Err(ToolCallError::Fatal(message)),
}
}
}
fn failure_response(call: &ToolCall, message: String) -> ResponseInputItem {
match &call.payload {
ToolPayload::ToolSearch { .. } => ResponseInputItem::ToolSearchOutput {
call_id: call.call_id.clone(),
status: "completed".to_string(),
execution: "client".to_string(),
tools: Vec::new(),
},
ToolPayload::Custom { .. } => ResponseInputItem::CustomToolCallOutput {
call_id: call.call_id.clone(),
name: None,
output: codex_protocol::models::FunctionCallOutputPayload {
body: codex_protocol::models::FunctionCallOutputBody::Text(message),
success: Some(false),
},
},
_ => ResponseInputItem::FunctionCallOutput {
call_id: call.call_id.clone(),
output: codex_protocol::models::FunctionCallOutputPayload {
body: codex_protocol::models::FunctionCallOutputBody::Text(message),
success: Some(false),
},
},
}
}
codex-rs/core/src/tools/runtimes/mod_tests.rs
+7 -4
View File
@@ -675,10 +675,13 @@ fn maybe_wrap_shell_lc_with_snapshot_keeps_user_proxy_env_when_proxy_inactive()
&HashMap::new(),
&HashMap::new(),
);
let output = Command::new(&rewritten[0])
.args(&rewritten[1..])
.output()
.expect("run rewritten command");
let mut process = Command::new(&rewritten[0]);
process.args(&rewritten[1..]);
for key in PROXY_ENV_KEYS {
process.env_remove(key);
}
process.env_remove(PROXY_ACTIVE_ENV_KEY);
let output = process.output().expect("run rewritten command");
assert!(output.status.success(), "command failed: {output:?}");
assert_eq!(
codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs
+16 -14
View File
@@ -338,23 +338,25 @@ async fn execve_permission_request_hook_short_circuits_prompt() -> anyhow::Resul
std::fs::write(
&script_path,
format!(
"#!/bin/sh\ncat > {log_path}\nprintf '%s\\n' '{response}'\n",
log_path = shlex::try_quote(log_path.to_string_lossy().as_ref())?,
r#"import json
from pathlib import Path
import sys
payload = json.load(sys.stdin)
with Path(r"{log_path}").open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload) + "\n")
print({response:?})
"#,
log_path = log_path.display(),
response = "{\"hookSpecificOutput\":{\"hookEventName\":\"PermissionRequest\",\"decision\":{\"behavior\":\"allow\"}}}",
),
)
.with_context(|| format!("write hook script to {}", script_path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut permissions = std::fs::metadata(&script_path)
.with_context(|| format!("read hook script metadata from {}", script_path.display()))?
.permissions();
permissions.set_mode(0o755);
std::fs::set_permissions(&script_path, permissions)
.with_context(|| format!("set hook script permissions on {}", script_path.display()))?;
}
let script_path_arg = format!(
"'{}'",
script_path.display().to_string().replace('\'', "'\\''")
);
std::fs::write(
turn_context.config.codex_home.join("hooks.json"),
serde_json::json!({
@@ -362,7 +364,7 @@ async fn execve_permission_request_hook_short_circuits_prompt() -> anyhow::Resul
"PermissionRequest": [{
"hooks": [{
"type": "command",
"command": script_path.display().to_string(),
"command": format!("python3 {script_path_arg}"),
}]
}]
}
codex-rs/core/tests/common/test_codex.rs
+10 -8
View File
@@ -384,15 +384,17 @@ impl TestCodexBuilder {
.exec_server_url
.clone()
.or_else(|| test_env.exec_server_url().map(str::to_owned));
let environment_manager = Arc::new(codex_exec_server::EnvironmentManager::new(
codex_exec_server::EnvironmentManagerArgs {
let local_runtime_paths = codex_exec_server::ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?;
let environment_manager = Arc::new(
codex_exec_server::EnvironmentManager::create_for_tests(
exec_server_url,
local_runtime_paths: codex_exec_server::ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?,
},
));
local_runtime_paths,
)
.await,
);
let file_system = test_env.environment().get_filesystem();
let mut workspace_setups = vec![];
swap(&mut self.workspace_setups, &mut workspace_setups);
codex-rs/core/tests/suite/skills.rs
+5 -8
View File
@@ -240,14 +240,11 @@ async fn list_skills_skips_cwd_roots_when_environment_disabled() -> Result<()> {
codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
SessionSource::Exec,
CollaborationModesConfig::default(),
Arc::new(EnvironmentManager::new(
codex_exec_server::EnvironmentManagerArgs {
exec_server_url: Some("none".to_string()),
local_runtime_paths: ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?,
},
Arc::new(EnvironmentManager::disabled_for_tests(
ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?,
)),
/*analytics_events_client*/ None,
);