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

Merge current main into PSEC-4922 stack base

Browse Source
This commit is contained in:
Chris Bookholt
2026-07-02 19:07:50 -07:00
parent 3fae2503b3 da4c8ca57d
commit 9f9db61ddf
64 changed files with 2459 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/config.schema.json
+3
View File
@@ -1750,6 +1750,9 @@
"minimum": 0.0,
"type": "integer"
},
"multi_agent_mode_hint_text": {
"type": "string"
},
"non_code_mode_only": {
"type": "boolean"
},
codex-rs/core/src/agent/control.rs
+78 -37
View File
@@ -4,6 +4,8 @@ use crate::agent::registry::AgentRegistry;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::resolve_role_config;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::codex_thread::ThreadConfigSnapshot;
use crate::config::Config;
use crate::config::RolloutBudgetConfig;
@@ -139,12 +141,12 @@ impl AgentControl {
pub(crate) async fn send_input(
&self,
agent_id: ThreadId,
initial_operation: Op,
input: Vec<UserInput>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_op(agent_id, &initial_operation)
self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true)
.await?;
self.send_input_after_capacity_check(agent_id, &state, initial_operation)
self.send_input_after_capacity_check(agent_id, &state, input)
.await
}
@@ -152,19 +154,14 @@ impl AgentControl {
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
initial_operation: Op,
input: Vec<UserInput>,
) -> CodexResult<String> {
let last_task_message = match &initial_operation {
Op::InterAgentCommunication { communication } => {
last_task_message_from_communication(communication)
}
_ => non_empty_task_message(render_input_preview(&initial_operation)),
};
let last_task_message = non_empty_task_message(render_input_preview(&input));
let result = self
.handle_thread_request_result(
agent_id,
state,
state.send_op(agent_id, initial_operation).await,
state.send_op(agent_id, input.into()).await,
)
.await;
if result.is_ok() {
@@ -182,14 +179,60 @@ impl AgentControl {
&self,
agent_id: ThreadId,
communication: InterAgentCommunication,
agent_communication_context: AgentCommunicationContext,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn)
.await?;
self.send_inter_agent_communication_after_capacity_check(
agent_id,
&state,
communication,
agent_communication_context,
)
.await
}
async fn send_inter_agent_communication_after_capacity_check(
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
self.submit_inter_agent_communication(agent_id, state, communication, context)
.await
}
async fn submit_inter_agent_communication(
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
let last_task_message = last_task_message_from_communication(&communication);
let state = self.upgrade()?;
let op = Op::InterAgentCommunication { communication };
self.ensure_execution_capacity_for_op(agent_id, &op).await?;
let communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let result = self
.handle_thread_request_result(agent_id, &state, state.send_op(agent_id, op).await)
.handle_thread_request_result(
agent_id,
state,
state
.send_op(agent_id, Op::InterAgentCommunication { communication })
.await,
)
.await;
if let (Some(communication), Ok(communication_id)) =
(communication_for_log, result.as_ref())
{
crate::agent_communication::emit_agent_communication_send(
communication_id,
&context,
&communication,
agent_id,
);
}
if result.is_ok() {
match last_task_message {
Some(last_task_message) => self
@@ -480,8 +523,10 @@ impl AgentControl {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id);
let _ = control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await;
return;
}
@@ -706,27 +751,23 @@ fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> b
})
}
pub(crate) fn render_input_preview(initial_operation: &Op) -> String {
match initial_operation {
Op::UserInput { items, .. } => items
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n"),
Op::InterAgentCommunication { communication } => communication.content.clone(),
_ => String::new(),
}
pub(crate) fn render_input_preview(input: &[UserInput]) -> String {
input
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n")
}
fn last_task_message_from_communication(communication: &InterAgentCommunication) -> Option<String> {
codex-rs/core/src/agent/control/execution.rs
+10 -1
View File
@@ -32,7 +32,16 @@ impl AgentControl {
thread_id: ThreadId,
op: &Op,
) -> CodexResult<()> {
if !op_starts_turn(op) {
self.ensure_execution_capacity_for_turn_start(thread_id, op_starts_turn(op))
.await
}
pub(super) async fn ensure_execution_capacity_for_turn_start(
&self,
thread_id: ThreadId,
starts_turn: bool,
) -> CodexResult<()> {
if !starts_turn {
return Ok(());
}
let state = self.upgrade()?;
codex-rs/core/src/agent/control/spawn.rs
+54 -8
View File
@@ -9,6 +9,17 @@ struct SpawnAgentThreadInheritance {
exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
}
/// Initial input delivered after a spawned agent acquires execution capacity.
///
/// V2 communication spawns keep the communication and its context paired so centralized
/// submission and lifecycle logging cannot receive one without the other. Other spawn sources
/// provide user input directly, making an uncontextualized inter-agent communication
/// unrepresentable.
enum SpawnInitialInput {
UserInput(Vec<UserInput>),
InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext),
}
fn default_agent_nickname_list() -> Vec<&'static str> {
AGENT_NAMES
.lines()
@@ -89,12 +100,12 @@ impl AgentControl {
pub(crate) async fn spawn_agent(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
session_source: Option<SessionSource>,
) -> CodexResult<ThreadId> {
let spawned_agent = Box::pin(self.spawn_agent_internal(
config,
initial_operation,
SpawnInitialInput::UserInput(initial_input),
session_source,
SpawnAgentOptions::default(),
))
@@ -106,12 +117,34 @@ impl AgentControl {
pub(crate) async fn spawn_agent_with_metadata(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
session_source: Option<SessionSource>,
options: SpawnAgentOptions, // TODO(jif) drop with new fork.
) -> CodexResult<LiveAgent> {
Box::pin(self.spawn_agent_internal(config, initial_operation, session_source, options))
.await
Box::pin(self.spawn_agent_internal(
config,
SpawnInitialInput::UserInput(initial_input),
session_source,
options,
))
.await
}
pub(crate) async fn spawn_agent_with_communication(
&self,
config: Config,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
session_source: Option<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
Box::pin(self.spawn_agent_internal(
config,
SpawnInitialInput::InterAgentCommunication(communication, context),
session_source,
options,
))
.await
}
pub(crate) async fn ensure_v2_agent_loaded(
@@ -197,7 +230,7 @@ impl AgentControl {
async fn spawn_agent_internal(
&self,
config: Config,
initial_operation: Op,
initial_input: SpawnInitialInput,
session_source: Option<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
@@ -356,8 +389,21 @@ impl AgentControl {
)
.await;
self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation)
.await?;
match initial_input {
SpawnInitialInput::UserInput(input) => {
self.send_input_after_capacity_check(new_thread.thread_id, &state, input)
.await?;
}
SpawnInitialInput::InterAgentCommunication(communication, context) => {
self.send_inter_agent_communication_after_capacity_check(
new_thread.thread_id,
&state,
communication,
context,
)
.await?;
}
}
if multi_agent_version != MultiAgentVersion::V2 {
let child_reference = agent_metadata
.agent_path
codex-rs/core/src/agent/control_tests.rs
+21 -11
View File
@@ -3,6 +3,8 @@ use crate::CodexThread;
use crate::StateDbHandle;
use crate::ThreadManager;
use crate::agent::agent_status_from_event;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::config::AgentRoleConfig;
use crate::config::Config;
use crate::config::ConfigBuilder;
@@ -65,12 +67,11 @@ async fn test_config() -> (TempDir, Config) {
test_config_with_cli_overrides(Vec::new()).await
}
fn text_input(text: &str) -> Op {
fn text_input(text: &str) -> Vec<UserInput> {
vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}]
.into()
}
fn assistant_message(text: &str, phase: Option<MessagePhase>) -> ResponseItem {
@@ -311,8 +312,7 @@ async fn send_input_errors_when_manager_dropped() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail without a manager");
@@ -423,8 +423,7 @@ async fn send_input_errors_when_thread_missing() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail for missing thread");
@@ -490,8 +489,7 @@ async fn send_input_submits_user_message() {
vec![UserInput::Text {
text: "hello from tests".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect("send_input should succeed");
@@ -531,7 +529,11 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
let submission_id = harness
.control
.send_inter_agent_communication(thread_id, communication.clone())
.send_inter_agent_communication(
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
assert!(!submission_id.is_empty());
@@ -656,7 +658,11 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication.clone())
.send_inter_agent_communication(
spawned_agent.thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed after reload");
let expected = (
@@ -803,7 +809,11 @@ async fn encrypted_inter_agent_communication_clears_existing_last_task_message()
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication)
.send_inter_agent_communication(
spawned_agent.thread_id,
communication,
AgentCommunicationContext::new(AgentCommunicationKind::Followup, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
codex-rs/core/src/agent_communication.rs
+79
View File
@@ -0,0 +1,79 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::InterAgentCommunication;
const AGENT_COMMUNICATION_TARGET: &str = "codex_otel.agent_communication";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum AgentCommunicationKind {
Spawn,
Message,
Followup,
Result,
}
impl AgentCommunicationKind {
fn as_str(self) -> &'static str {
match self {
Self::Spawn => "spawn",
Self::Message => "message",
Self::Followup => "followup",
Self::Result => "result",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct AgentCommunicationContext {
kind: AgentCommunicationKind,
sender_thread_id: ThreadId,
}
impl AgentCommunicationContext {
pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self {
Self {
kind,
sender_thread_id,
}
}
}
pub(crate) fn logging_enabled() -> bool {
tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO)
}
pub(crate) fn emit_agent_communication_send(
communication_id: &str,
context: &AgentCommunicationContext,
communication: &InterAgentCommunication,
receiver_thread_id: ThreadId,
) {
tracing::info!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
kind = context.kind.as_str(),
state = "send",
sender_thread_id = %context.sender_thread_id,
receiver_thread_id = %receiver_thread_id,
content = if communication.content.is_empty() {
communication.encrypted_content.as_deref().unwrap_or_default()
} else {
communication.content.as_str()
},
},
"agent communication"
);
}
pub(crate) fn emit_agent_communication_receive(communication_id: &str) {
tracing::info!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
state = "receive",
},
"agent communication"
);
}
codex-rs/core/src/client.rs
+33 -20
View File
@@ -1123,42 +1123,48 @@ impl ModelClientSession {
}
}
/// Checks whether the current request is an incremental extension of the previous request.
/// We only reuse an incremental input delta when non-input request fields are unchanged and
/// `input` is a strict extension of the previous known input. Server-returned output items
/// are treated as part of the baseline so we do not resend them.
fn get_incremental_items(
&self,
request: &ResponsesApiRequest,
last_response: Option<&LastResponse>,
allow_empty_delta: bool,
) -> Option<Vec<ResponseItem>> {
// Checks whether the current request is an incremental extension of the previous request.
// We only reuse an incremental input delta when non-input request fields are unchanged and
// `input` is a strict
// extension of the previous known input. Server-returned output items are treated as part
// of the baseline so we do not resend them.
let previous_request = self.websocket_session.last_request.as_ref()?;
if !responses_request_properties_match(previous_request, request) {
trace!("incremental request failed, websocket reuse properties didn't match");
return None;
}
let Some(after_previous_input) = request
.input
.strip_prefix(previous_request.input.as_slice())
else {
trace!("incremental request failed, items didn't match");
return None;
};
let mut response_items =
last_response.map_or_else(Vec::new, |response| response.items_added.clone());
if !self.client.state.provider.info().is_openai() {
response_items
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
// To compare the inputs, we concatenate the previous request items with the response items,
// then compare that against the equivalent slice of request items, ignoring metadata. If
// they match, we can consider the remaining items the incremental request.
let mut previous_items = previous_request.input.clone();
if let Some(response) = last_response {
previous_items.extend_from_slice(&response.items_added);
}
let Some(incremental_items) = after_previous_input.strip_prefix(response_items.as_slice())
previous_items
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
let Some((request_items_to_compare, incremental_items)) =
request.input.split_at_checked(previous_items.len())
else {
trace!("incremental request failed, items didn't match");
trace!("incremental request failed, incompatible request length");
return None;
};
let mut request_prefix = request_items_to_compare.to_vec();
request_prefix
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
if previous_items != request_prefix {
trace!("incremental request failed, items didn't match");
return None;
}
if !allow_empty_delta && incremental_items.is_empty() {
return None;
}
@@ -1912,6 +1918,7 @@ where
let mut logged_error = false;
let mut tx_last_response = Some(tx_last_response);
let mut items_added: Vec<ResponseItem> = Vec::new();
let (request_start, mut ttft_ms) = (Instant::now(), None);
let mut api_stream = api_stream;
let upstream_request_id = upstream_request_id.as_deref();
if let Some(upstream_request_id) = upstream_request_id {
@@ -1961,6 +1968,7 @@ where
Some(usage.cached_input_tokens),
Some(usage.reasoning_output_tokens),
usage.total_tokens,
ttft_ms,
);
}
inference_trace_attempt.record_completed(
@@ -1988,6 +1996,11 @@ where
}
}
Ok(event) => {
if matches!(&event, ResponseEvent::OutputItemAdded(_)) && ttft_ms.is_none() {
ttft_ms = Some(
i64::try_from(request_start.elapsed().as_millis()).unwrap_or(i64::MAX),
);
}
if tx_event.send(Ok(event)).await.is_err() {
inference_trace_attempt.record_cancelled(
STREAM_DROPPED_REASON,
codex-rs/core/src/config/config_tests.rs
+21
View File
@@ -10307,6 +10307,7 @@ default_wait_timeout_ms = 30000
usage_hint_text = "Custom delegation guidance."
root_agent_usage_hint_text = "Root guidance."
subagent_usage_hint_text = "Subagent guidance."
multi_agent_mode_hint_text = "Custom mode guidance."
tool_namespace = "agents"
hide_spawn_agent_metadata = true
non_code_mode_only = true
@@ -10343,6 +10344,10 @@ non_code_mode_only = true
config.multi_agent_v2.subagent_usage_hint_text.as_deref(),
Some("Subagent guidance.")
);
assert_eq!(
config.multi_agent_v2.multi_agent_mode_hint_text.as_deref(),
Some("Custom mode guidance.")
);
assert_eq!(
config.multi_agent_v2.tool_namespace.as_deref(),
Some("agents")
@@ -10405,6 +10410,22 @@ max_concurrent_threads_per_session = 17
);
}
#[test]
fn multi_agent_v2_preserves_empty_mode_hint_override() {
let config_toml = toml::from_str(
r#"[features.multi_agent_v2]
multi_agent_mode_hint_text = ""
"#,
)
.expect("multi-agent v2 config should parse");
let expected = MultiAgentV2Config {
multi_agent_mode_hint_text: Some(String::new()),
..Default::default()
};
assert_eq!(resolve_multi_agent_v2_config(&config_toml), expected);
}
#[tokio::test]
async fn multi_agent_v2_empty_usage_hint_overrides_clear_default_hints() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
codex-rs/core/src/config/mod.rs
+7
View File
@@ -1144,6 +1144,7 @@ pub struct MultiAgentV2Config {
pub usage_hint_text: Option<String>,
pub root_agent_usage_hint_text: Option<String>,
pub subagent_usage_hint_text: Option<String>,
pub multi_agent_mode_hint_text: Option<String>,
pub tool_namespace: Option<String>,
pub hide_spawn_agent_metadata: bool,
pub non_code_mode_only: bool,
@@ -1165,6 +1166,7 @@ impl MultiAgentV2Config {
DEFAULT_MULTI_AGENT_V2_SUBAGENT_USAGE_HINT_TEXT,
max_concurrent_threads_per_session,
)),
multi_agent_mode_hint_text: None,
tool_namespace: Some(DEFAULT_MULTI_AGENT_V2_TOOL_NAMESPACE.to_string()),
hide_spawn_agent_metadata: true,
non_code_mode_only: true,
@@ -2501,6 +2503,10 @@ fn resolve_multi_agent_v2_config(config_toml: &ConfigToml) -> MultiAgentV2Config
base.map(|config| &config.subagent_usage_hint_text),
default.subagent_usage_hint_text,
);
let multi_agent_mode_hint_text = base
.and_then(|config| config.multi_agent_mode_hint_text.as_ref())
.cloned()
.or(default.multi_agent_mode_hint_text);
let tool_namespace = base
.and_then(|config| config.tool_namespace.as_ref())
.cloned()
@@ -2520,6 +2526,7 @@ fn resolve_multi_agent_v2_config(config_toml: &ConfigToml) -> MultiAgentV2Config
usage_hint_text,
root_agent_usage_hint_text,
subagent_usage_hint_text,
multi_agent_mode_hint_text,
tool_namespace,
hide_spawn_agent_metadata,
non_code_mode_only,
codex-rs/core/src/context/multi_agent_mode_instructions.rs
+2 -3
View File
@@ -4,7 +4,6 @@ use codex_protocol::protocol::MULTI_AGENT_MODE_CLOSE_TAG;
use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG;
const EXPLICIT_REQUEST_ONLY_MULTI_AGENT_MODE_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const NO_MULTI_AGENT_MODE_TEXT: &str = "Multi-agent delegation mode instructions are inactive. Any earlier multi-agent mode developer message no longer applies.";
const PROACTIVE_MULTI_AGENT_MODE_TEXT: &str = "Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.";
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -32,8 +31,8 @@ impl ContextualUserFragment for MultiAgentModeInstructions {
}
fn body(&self) -> String {
match self.multi_agent_mode {
MultiAgentMode::None => NO_MULTI_AGENT_MODE_TEXT.to_string(),
match &self.multi_agent_mode {
MultiAgentMode::Custom(hint_text) => hint_text.clone(),
MultiAgentMode::ExplicitRequestOnly => {
EXPLICIT_REQUEST_ONLY_MULTI_AGENT_MODE_TEXT.to_string()
}
codex-rs/core/src/context_manager/updates.rs
-3
View File
@@ -85,9 +85,6 @@ fn build_multi_agent_mode_update_item(
}
match effective_multi_agent_mode {
Some(MultiAgentMode::None) => {
Some(MultiAgentModeInstructions::new(MultiAgentMode::None).render())
}
Some(multi_agent_mode) => Some(MultiAgentModeInstructions::new(multi_agent_mode).render()),
None if previous.multi_agent_mode == Some(MultiAgentMode::Proactive) => {
Some(MultiAgentModeInstructions::new(MultiAgentMode::ExplicitRequestOnly).render())
codex-rs/core/src/lib.rs
+1
View File
@@ -31,6 +31,7 @@ pub use codex_thread::TryStartTurnIfIdleError;
pub use codex_thread::TryStartTurnIfIdleRejectionReason;
pub use session::turn_context::TurnContext;
mod agent;
mod agent_communication;
mod attestation;
mod codex_delegate;
mod command_canonicalization;
codex-rs/core/src/session/handlers.rs
+1
View File
@@ -285,6 +285,7 @@ pub async fn inter_agent_communication(
sess.input_queue
.enqueue_mailbox_communication(communication)
.await;
crate::agent_communication::emit_agent_communication_receive(&sub_id);
if trigger_turn {
sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
.await;
codex-rs/core/src/session/mod.rs
+9 -12
View File
@@ -14,6 +14,8 @@ use crate::agent::AgentControl;
use crate::agent::AgentStatus;
use crate::agent::agent_status_from_event;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::attestation::AttestationProvider;
use crate::build_available_skills;
use crate::compact;
@@ -92,7 +94,6 @@ use codex_protocol::approvals::NetworkPolicyRuleAction;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::AutoCompactTokenLimitScope;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::MultiAgentMode;
use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::WebSearchMode;
@@ -1867,10 +1868,12 @@ impl Session {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id);
if let Err(err) = self
.services
.agent_control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
@@ -3426,16 +3429,10 @@ impl Session {
{
items.push(usage_hint_message);
}
match multi_agents::effective_multi_agent_mode(turn_context) {
Some(
multi_agent_mode
@ (MultiAgentMode::ExplicitRequestOnly | MultiAgentMode::Proactive),
) => {
items.push(ContextualUserFragment::into(
MultiAgentModeInstructions::new(multi_agent_mode),
));
}
Some(MultiAgentMode::None) | None => {}
if let Some(multi_agent_mode) = multi_agents::effective_multi_agent_mode(turn_context) {
items.push(ContextualUserFragment::into(
MultiAgentModeInstructions::new(multi_agent_mode),
));
}
if let Some(contextual_user_message) =
crate::context_manager::updates::build_contextual_user_message(contextual_user_sections)
codex-rs/core/src/session/multi_agents.rs
+12 -3
View File
@@ -41,9 +41,18 @@ pub(crate) fn effective_multi_agent_mode(turn_context: &TurnContext) -> Option<M
return None;
}
let multi_agent_mode = match turn_context.effective_reasoning_effort() {
Some(ReasoningEffort::Ultra) => MultiAgentMode::Proactive,
_ => MultiAgentMode::ExplicitRequestOnly,
// A configured hint, including an empty string, defines a custom policy instead of an
// effort-derived built-in policy.
let multi_agent_mode = match &turn_context
.config
.multi_agent_v2
.multi_agent_mode_hint_text
{
Some(hint_text) => MultiAgentMode::Custom(hint_text.clone()),
None => match turn_context.effective_reasoning_effort() {
Some(ReasoningEffort::Ultra) => MultiAgentMode::Proactive,
_ => MultiAgentMode::ExplicitRequestOnly,
},
};
match &turn_context.session_source {
codex-rs/core/src/tools/handlers/agent_jobs.rs
+1 -1
View File
@@ -203,7 +203,7 @@ async fn run_agent_job_loop(
.agent_control
.spawn_agent_with_metadata(
options.spawn_config.clone(),
items.into(),
items,
Some(SessionSource::SubAgent(SubAgentSource::Other(format!(
"agent_job:{job_id}"
)))),
codex-rs/core/src/tools/handlers/multi_agents_common.rs
+3 -5
View File
@@ -18,7 +18,6 @@ use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::openai_models::ReasoningEffortPreset;
use codex_protocol::protocol::CollabAgentRef;
use codex_protocol::protocol::CollabAgentStatusEntry;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
@@ -163,7 +162,7 @@ pub(crate) fn thread_spawn_source(
pub(crate) fn parse_collab_input(
message: Option<String>,
items: Option<Vec<UserInput>>,
) -> Result<Op, FunctionCallError> {
) -> Result<Vec<UserInput>, FunctionCallError> {
match (message, items) {
(Some(_), Some(_)) => Err(FunctionCallError::RespondToModel(
"Provide either message or items, but not both".to_string(),
@@ -180,8 +179,7 @@ pub(crate) fn parse_collab_input(
Ok(vec![UserInput::Text {
text: message,
text_elements: Vec::new(),
}]
.into())
}])
}
(None, Some(items)) => {
if items.is_empty() {
@@ -189,7 +187,7 @@ pub(crate) fn parse_collab_input(
"Items can't be empty".to_string(),
));
}
Ok(items.into())
Ok(items)
}
}
}
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+6 -12
View File
@@ -1377,8 +1377,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1454,8 +1453,7 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1626,8 +1624,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
vec![UserInput::Text {
text: "research".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1647,8 +1644,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
vec![UserInput::Text {
text: "build".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 2,
@@ -4097,8 +4093,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_id() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -4165,8 +4160,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_task_name() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
codex-rs/core/src/tools/handlers/multi_agents_v2.rs
-1
View File
@@ -19,7 +19,6 @@ use codex_protocol::protocol::CollabWaitingEndEvent;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::SubAgentActivityEvent;
use codex_protocol::protocol::SubAgentActivityKind;
use codex_protocol::user_input::UserInput;
use codex_tools::ToolName;
use serde::Deserialize;
use serde::Serialize;
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+9 -2
View File
@@ -4,6 +4,8 @@
//! resulting `InterAgentCommunication` should wake the target immediately.
use super::*;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::context::FunctionToolOutput;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::protocol::InterAgentCommunication;
@@ -46,7 +48,7 @@ pub(crate) struct FollowupTaskArgs {
pub(crate) message: String,
}
fn message_content(message: String) -> Result<String, FunctionCallError> {
pub(super) fn message_content(message: String) -> Result<String, FunctionCallError> {
if message.trim().is_empty() {
return Err(FunctionCallError::RespondToModel(
"Empty message can't be sent to an agent".to_string(),
@@ -101,10 +103,15 @@ pub(crate) async fn handle_message_string_tool(
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, receiver_agent_path.clone(), message);
let kind = match mode {
MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message,
MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup,
};
let context = AgentCommunicationContext::new(kind, session.thread_id);
let result = session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication))
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication), context)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
result?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+25 -29
View File
@@ -4,11 +4,13 @@ use crate::agent::control::SpawnAgentOptions;
use crate::agent::next_thread_spawn_depth;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::apply_role_to_config;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions;
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2;
use crate::tools::handlers::multi_agents_v2::message_tool::message_content;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::AgentPath;
use codex_protocol::protocol::Op;
use codex_tools::ToolSpec;
#[derive(Default)]
@@ -55,8 +57,7 @@ async fn handle_spawn_agent(
.map(str::trim)
.filter(|role| !role.is_empty());
let message = args.message.clone();
let initial_operation = parse_collab_input(Some(args.message), /*items*/ None)?;
let message = message_content(args.message)?;
let session_source = turn.session_source.clone();
let child_depth = next_thread_spawn_depth(&session_source);
let mut config =
@@ -104,33 +105,28 @@ async fn handle_spawn_agent(
"spawned agent is missing a canonical task name".to_string(),
)
})?;
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication = communication_from_tool_message(author, new_agent_path.clone(), message);
let context = AgentCommunicationContext::new(AgentCommunicationKind::Spawn, session.thread_id);
let spawned_agent = Box::pin(
session.services.agent_control.spawn_agent_with_metadata(
config,
match initial_operation {
Op::UserInput { items, .. }
if items
.iter()
.all(|item| matches!(item, UserInput::Text { .. })) =>
{
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, new_agent_path.clone(), message);
Op::InterAgentCommunication { communication }
}
initial_operation => initial_operation,
},
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
),
session
.services
.agent_control
.spawn_agent_with_communication(
config,
communication,
context,
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
),
)
.await
.map_err(collab_spawn_error)?;
codex-rs/core/src/tools/parallel.rs
+260 -8
View File
@@ -1,4 +1,5 @@
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Instant;
@@ -9,6 +10,7 @@ use tokio_util::either::Either;
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;
use tracing::Instrument;
use tracing::info;
use tracing::instrument;
use tracing::trace_span;
@@ -27,6 +29,15 @@ use crate::tools::router::ToolRouter;
use codex_protocol::error::CodexErr;
use codex_protocol::models::ResponseInputItem;
struct ToolCallTimingGuard {
started_at: Instant,
execution_started_at: Arc<OnceLock<Instant>>,
conversation_id: String,
turn_id: String,
call_id: String,
tool_name: codex_tools::ToolName,
}
#[derive(Clone)]
pub(crate) struct ToolCallRuntime {
router: Arc<ToolRouter>,
@@ -96,6 +107,11 @@ impl ToolCallRuntime {
let invocation_cancellation_token = cancellation_token.clone();
let wait_for_runtime_cancellation = self.router.tool_waits_for_runtime_cancellation(&call);
let started = Instant::now();
let tool_call_timing_guard =
ToolCallTimingGuard::capture(started, &session.thread_id, &turn.sub_id, &call, &source);
let execution_started_at = tool_call_timing_guard
.as_ref()
.map(|timing| Arc::clone(&timing.execution_started_at));
let abort_session = Arc::clone(&session);
let abort_source = source.clone();
let abort_turn = Arc::clone(&turn);
@@ -112,13 +128,18 @@ impl ToolCallRuntime {
);
let abort_dispatch_span = dispatch_span.clone();
let mut handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
let mut dispatch_handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
AbortOnDropHandle::new(tokio::spawn(async move {
let _guard = if supports_parallel {
Either::Left(lock.read().await)
} else {
Either::Right(lock.write().await)
};
// Admission through the parallel-execution gate marks the end
// of dispatch waiting and the start of handler execution.
if let Some(execution_started_at) = execution_started_at {
let _ = execution_started_at.set(Instant::now());
}
router
.dispatch_tool_call_with_terminal_outcome(
@@ -135,28 +156,29 @@ impl ToolCallRuntime {
}));
async move {
let _tool_call_timing_guard = tool_call_timing_guard;
tokio::select! {
res = &mut handle => res.map_err(Self::tool_task_join_error)?,
res = &mut dispatch_handle => res.map_err(Self::tool_task_join_error)?,
_ = cancellation_token.cancelled() => {
if terminal_outcome_reached.load(Ordering::Acquire) || handle.is_finished() {
handle.await.map_err(Self::tool_task_join_error)?
if terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() {
dispatch_handle.await.map_err(Self::tool_task_join_error)?
} else {
let secs = started.elapsed().as_secs_f32().max(0.1);
abort_dispatch_span.record("aborted", true);
if wait_for_runtime_cancellation {
if terminal_outcome_reached.swap(true, Ordering::AcqRel) {
return handle.await.map_err(Self::tool_task_join_error)?;
return dispatch_handle.await.map_err(Self::tool_task_join_error)?;
}
// The abort owns the terminal outcome; await only so
// the runtime can finish process teardown.
match handle.await {
match dispatch_handle.await {
Ok(_) => {}
Err(err) if err.is_cancelled() => {}
Err(err) => return Err(Self::tool_task_join_error(err)),
}
} else {
handle.abort();
match handle.await {
dispatch_handle.abort();
match dispatch_handle.await {
Ok(result) => return result,
Err(err) if err.is_cancelled() => {}
Err(err) => return Err(Self::tool_task_join_error(err)),
@@ -237,6 +259,88 @@ impl ToolCallRuntime {
}
}
impl ToolCallTimingGuard {
fn capture(
started_at: Instant,
conversation_id: &impl std::fmt::Display,
turn_id: &str,
call: &ToolCall,
source: &ToolCallSource,
) -> Option<Self> {
// Code-mode calls are nested within a direct code-mode tool call whose
// timing already includes them. Suppress nested guards so consumers do
// not mistake overlapping events for independent tool-call latency.
if !matches!(source, ToolCallSource::Direct) || !tracing::enabled!(tracing::Level::INFO) {
return None;
}
Some(Self {
started_at,
execution_started_at: Arc::new(OnceLock::new()),
conversation_id: conversation_id.to_string(),
turn_id: turn_id.to_string(),
call_id: call.call_id.clone(),
tool_name: call.tool_name.clone(),
})
}
}
impl Drop for ToolCallTimingGuard {
fn drop(&mut self) {
let completed_at = Instant::now();
// Snapshot once so a concurrently-starting dispatch cannot make one
// event internally inconsistent.
let execution_started_at = self
.execution_started_at
.get()
.copied()
.filter(|execution_started_at| *execution_started_at <= completed_at);
let duration_ms = |duration: std::time::Duration| u64::try_from(duration.as_millis()).ok();
let total_duration_ms = duration_ms(completed_at.duration_since(self.started_at));
let dispatch_duration_ms = execution_started_at.map_or_else(
|| total_duration_ms,
|execution_started_at| {
duration_ms(execution_started_at.duration_since(self.started_at))
},
);
let handler_duration_ms = execution_started_at.map_or(Some(0), |execution_started_at| {
duration_ms(completed_at.duration_since(execution_started_at))
});
macro_rules! log_tool_call {
($dispatch_duration_ms:expr, $handler_duration_ms:expr, $total_duration_ms:expr) => {
info!(
event.name = "codex.tool_call",
trace_id = %codex_otel::current_span_trace_id().unwrap_or_default(),
conversation.id = %self.conversation_id,
turn_id = %self.turn_id,
tool_name = %self.tool_name,
call_id = %self.call_id,
tool_source = "direct",
execution_started = execution_started_at.is_some(),
dispatch_duration_ms = $dispatch_duration_ms,
handler_duration_ms = $handler_duration_ms,
total_duration_ms = $total_duration_ms,
"tool call completed"
);
};
}
match (dispatch_duration_ms, handler_duration_ms, total_duration_ms) {
(Some(dispatch_duration_ms), Some(handler_duration_ms), Some(total_duration_ms)) => {
log_tool_call!(dispatch_duration_ms, handler_duration_ms, total_duration_ms);
}
_ => {
log_tool_call!(
tracing::field::Empty,
tracing::field::Empty,
tracing::field::Empty
);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -255,6 +359,154 @@ mod tests {
use pretty_assertions::assert_eq;
use tokio::sync::Notify;
use tokio::sync::oneshot;
use tracing_test::internal::MockWriter;
#[test]
fn tool_call_timing_guard_ignores_code_mode_source() {
let subscriber = tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.finish();
tracing::subscriber::with_default(subscriber, || {
let call = ToolCall {
tool_name: codex_tools::ToolName::plain("test_tool"),
call_id: "call-1".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
};
let direct_guard = ToolCallTimingGuard::capture(
Instant::now(),
&"conversation-id",
"turn-id",
&call,
&ToolCallSource::Direct,
);
assert!(
direct_guard.is_some(),
"direct tool calls should create a timing guard"
);
drop(direct_guard);
let code_mode_guard = ToolCallTimingGuard::capture(
Instant::now(),
&"conversation-id",
"turn-id",
&call,
&ToolCallSource::CodeMode {
cell_id: "cell-1".to_string(),
runtime_tool_call_id: "runtime-call-1".to_string(),
},
);
assert!(
code_mode_guard.is_none(),
"nested code-mode calls should not create overlapping timing events"
);
});
}
#[tokio::test]
async fn cancellation_before_dispatch_admission_logs_dispatch_only_timing() -> anyhow::Result<()>
{
let (session, turn_context) = crate::session::tests::make_session_and_context().await;
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let tool_name = codex_tools::ToolName::plain("test_tool");
let handler = Arc::new(ImmediateHandler {
tool_name: tool_name.clone(),
}) as Arc<dyn CoreToolRuntime>;
let step_context = StepContext::for_test(Arc::clone(&turn_context));
let router = Arc::new(ToolRouter::from_parts(
ToolRegistry::from_tools([handler]),
Vec::new(),
));
let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new()));
let runtime = ToolCallRuntime::new(router, session, step_context, tracker);
let execution_gate = Arc::clone(&runtime.parallel_execution);
let execution_gate_guard = execution_gate
.try_write_owned()
.expect("execution gate should be available before dispatch starts");
let (release_execution_gate_tx, release_execution_gate_rx) = std::sync::mpsc::channel();
let execution_gate_task = tokio::task::spawn_blocking(move || {
let _execution_gate_guard = execution_gate_guard;
release_execution_gate_rx
.recv()
.expect("test should release the execution gate");
});
let buffer: &'static std::sync::Mutex<Vec<u8>> =
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(tracing::Level::INFO)
.with_writer(MockWriter::new(buffer))
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
let cancellation_token = CancellationToken::new();
let call = ToolCall {
tool_name,
call_id: "call-1".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
};
let response_task =
tokio::spawn(runtime.handle_tool_call(call, cancellation_token.clone()));
cancellation_token.cancel();
tokio::time::timeout(Duration::from_secs(1), response_task)
.await
.expect("timed out waiting for cancelled tool response")
.expect("cancelled tool response task should join")
.expect("cancelled tool call should produce a response");
let logs = String::from_utf8(
buffer
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone(),
)?;
let timing_events = logs
.lines()
.filter(|line| line.contains("event.name=\"codex.tool_call\""))
.collect::<Vec<_>>();
assert_eq!(
timing_events.len(),
1,
"cancelled tool call should emit exactly one timing event; logs:\n{logs}"
);
let timing_event = timing_events[0];
assert!(
timing_event.contains("execution_started=false"),
"tool cancelled before admission should not report execution started: {timing_event}"
);
assert!(
timing_event.contains("handler_duration_ms=0"),
"tool cancelled before admission should report zero handler duration: {timing_event}"
);
let duration_field = |name: &str| {
timing_event.split_whitespace().find_map(|field| {
field
.strip_prefix(&format!("{name}="))
.and_then(|value| value.parse::<u64>().ok())
})
};
let dispatch_duration_ms = duration_field("dispatch_duration_ms")
.expect("timing event should include dispatch_duration_ms");
let total_duration_ms = duration_field("total_duration_ms")
.expect("timing event should include total_duration_ms");
assert_eq!(
dispatch_duration_ms, total_duration_ms,
"tool cancelled before admission should attribute all elapsed time to dispatch: {timing_event}"
);
release_execution_gate_tx
.send(())
.expect("execution gate task should remain available");
execution_gate_task
.await
.expect("execution gate task should join");
Ok(())
}
struct ImmediateHandler {
tool_name: codex_tools::ToolName,
codex-rs/core/tests/suite/client_websockets.rs
+63 -12
View File
@@ -966,35 +966,77 @@ async fn responses_websocket_v2_requests_use_v2_when_provider_supports_websocket
async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
skip_if_no_network!();
let mut assistant_output_with_turn_id = ev_assistant_message("msg-1", "assistant output");
assistant_output_with_turn_id["item"]["internal_chat_message_metadata_passthrough"] =
json!({ "turn_id": "turn-1" });
let assistant_output_without_turn_id = ev_assistant_message("msg-2", "second assistant output");
let server = start_websocket_server(vec![vec![
vec![
ev_response_created("resp-1"),
ev_assistant_message("msg-1", "assistant output"),
assistant_output_with_turn_id,
ev_completed("resp-1"),
],
vec![ev_response_created("resp-2"), ev_completed("resp-2")],
vec![
ev_response_created("resp-2"),
assistant_output_without_turn_id,
ev_completed("resp-2"),
],
vec![ev_response_created("resp-3"), ev_completed("resp-3")],
]])
.await;
let harness = websocket_harness_with_options(&server, /*runtime_metrics_enabled*/ false).await;
let prompt_one = prompt_with_input(vec![message_item("hello")]);
let prompt_two = prompt_with_input(vec![
message_item("hello"),
assistant_message_item("msg-1", "assistant output"),
message_item("second"),
]);
// use OpenAI provider to check metadata logic
let mut provider = websocket_provider(&server);
provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name;
let harness =
websocket_harness_with_provider_options(provider, /*runtime_metrics_enabled*/ false).await;
// Turn one: initiate
let prompt_one = prompt_with_input(vec![message_item("hello")]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_one).await;
}
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_two).await;
// Turn two: the response and reconstructed history have matching metadata.
let mut first_assistant_output = assistant_message_item("msg-1", "assistant output");
first_assistant_output.set_turn_id_if_missing("turn-1");
let prompt_two = prompt_with_input(vec![
message_item("hello"),
first_assistant_output.clone(),
message_item("second"),
]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_two).await;
}
// Turn three: the reconstructed history has metadata that the response omitted.
let mut second_assistant_output = assistant_message_item("msg-2", "second assistant output");
second_assistant_output.set_turn_id_if_missing("turn-2");
let prompt_three = prompt_with_input(vec![
message_item("hello"),
first_assistant_output,
message_item("second"),
second_assistant_output,
message_item("third"),
]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_three).await;
}
assert_eq!(server.handshakes().len(), 1);
let connection = server.single_connection();
assert_eq!(connection.len(), 2);
assert_eq!(connection.len(), 3);
// validate second turn
let second = connection.get(1).expect("missing request").body_json();
assert_eq!(second["type"].as_str(), Some("response.create"));
assert_eq!(second["previous_response_id"].as_str(), Some("resp-1"));
@@ -1003,6 +1045,15 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
serde_json::to_value(&prompt_two.input[2..]).unwrap()
);
// validate third turn
let third = connection.get(2).expect("missing request").body_json();
assert_eq!(third["type"].as_str(), Some("response.create"));
assert_eq!(third["previous_response_id"].as_str(), Some("resp-2"));
assert_eq!(
third["input"],
serde_json::to_value(&prompt_three.input[4..]).unwrap()
);
server.shutdown().await;
}
codex-rs/core/tests/suite/multi_agent_mode.rs
+107
View File
@@ -20,9 +20,11 @@ use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
const NO_SPAWN_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const PROACTIVE_TEXT: &str = "Proactive multi-agent delegation is active.";
const CUSTOM_MODE_HINT_TEXT: &str = "Use the configured delegation policy.";
fn add_ultra_reasoning(model_info: &mut ModelInfo) {
model_info.supports_reasoning_summaries = true;
@@ -41,6 +43,12 @@ fn configure_multi_agent_v2(config: &mut Config) {
.expect("test config should allow feature update");
}
// Configuring a custom mode hint also enables multi-agent V2 for the test.
fn configure_custom_mode_hint(config: &mut Config) {
configure_multi_agent_v2(config);
config.multi_agent_v2.multi_agent_mode_hint_text = Some(CUSTOM_MODE_HINT_TEXT.to_string());
}
fn configure_ultra(config: &mut Config) {
configure_multi_agent_v2(config);
config.model_reasoning_effort = Some(ReasoningEffort::Ultra);
@@ -120,6 +128,105 @@ async fn ultra_reasoning_uses_max_and_proactive_mode() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn configured_mode_hint_uses_custom_mode_across_reasoning_efforts() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let responses = mount_sse_sequence(
&server,
(1..=2)
.map(|index| {
sse(vec![
ev_response_created(&format!("resp-{index}")),
ev_completed(&format!("resp-{index}")),
])
})
.collect(),
)
.await;
let test = test_codex()
.with_model_info_override("gpt-5.4", add_ultra_reasoning)
.with_config(configure_custom_mode_hint)
.build(&server)
.await?;
let rollout_path = test
.session_configured
.rollout_path
.clone()
.expect("rollout path");
submit_turn(&test.codex, "explicit", Some(ReasoningEffort::High)).await?;
submit_turn(&test.codex, "proactive", Some(ReasoningEffort::Ultra)).await?;
let requests = responses.requests();
let first_input = requests[0].input();
let first_texts = developer_texts(&first_input);
let second_input = requests[1].input();
let second_texts = developer_texts(&second_input);
let instruction_counts = |texts: &[&str]| {
(
count_containing(texts, CUSTOM_MODE_HINT_TEXT),
count_containing(texts, NO_SPAWN_TEXT),
count_containing(texts, PROACTIVE_TEXT),
)
};
assert_eq!(instruction_counts(&first_texts), (1, 0, 0));
assert_eq!(instruction_counts(&second_texts), (1, 0, 0));
let rollout_values = std::fs::read_to_string(rollout_path)?
.lines()
.map(serde_json::from_str::<Value>)
.collect::<serde_json::Result<Vec<_>>>()?;
let recorded_modes = rollout_values
.iter()
.filter(|value| value.get("type").and_then(Value::as_str) == Some("turn_context"))
.filter_map(|value| value.pointer("/payload/multi_agent_mode").cloned())
.collect::<Vec<_>>();
assert_eq!(
recorded_modes,
[
json!({"custom": CUSTOM_MODE_HINT_TEXT}),
json!({"custom": CUSTOM_MODE_HINT_TEXT}),
]
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_configured_mode_hint_suppresses_builtin_text() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response = mount_sse_once(
&server,
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
)
.await;
let test = test_codex()
.with_config(|config| {
configure_multi_agent_v2(config);
config.multi_agent_v2.multi_agent_mode_hint_text = Some(String::new());
})
.build(&server)
.await?;
submit_turn(&test.codex, "hello", Some(ReasoningEffort::High)).await?;
let input = response.single_request().input();
let texts = developer_texts(&input);
assert_eq!(
(
count_containing(&texts, MULTI_AGENT_MODE_OPEN_TAG),
count_containing(&texts, NO_SPAWN_TEXT),
count_containing(&texts, PROACTIVE_TEXT),
),
(1, 0, 0)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn leaving_ultra_after_cold_resume_emits_explicit_mode() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/otel.rs
+18 -12
View File
@@ -567,19 +567,22 @@ async fn process_sse_emits_completed_telemetry() {
mount_sse_once(
&server,
sse(vec![serde_json::json!({
"type": "response.completed",
"response": {
"id": "resp1",
"usage": {
"input_tokens": 3,
"input_tokens_details": { "cached_tokens": 1 },
"output_tokens": 5,
"output_tokens_details": { "reasoning_tokens": 2 },
"total_tokens": 9
sse(vec![
ev_reasoning_item_added("reasoning-1", &[]),
serde_json::json!({
"type": "response.completed",
"response": {
"id": "resp1",
"usage": {
"input_tokens": 3,
"input_tokens_details": { "cached_tokens": 1 },
"output_tokens": 5,
"output_tokens_details": { "reasoning_tokens": 2 },
"total_tokens": 9
}
}
}
})]),
}),
]),
)
.await;
@@ -612,6 +615,9 @@ async fn process_sse_emits_completed_telemetry() {
&& line.contains("cached_token_count=1")
&& line.contains("reasoning_token_count=2")
&& line.contains("tool_token_count=9")
&& extract_log_field(line, "ttft_ms")
.and_then(|ttft_ms| ttft_ms.parse::<i64>().ok())
.is_some()
})
.map(|_| Ok(()))
.unwrap_or(Err("missing response.completed telemetry".to_string()))
codex-rs/core/tests/suite/subagent_notifications.rs
+55 -1
View File
@@ -39,10 +39,13 @@ use serde_json::Value;
use serde_json::json;
use std::fs;
use std::path::Path;
use std::sync::Mutex;
use std::time::Duration;
use test_case::test_case;
use tokio::time::Instant;
use tokio::time::sleep;
use tracing::Level;
use tracing_test::internal::MockWriter;
use wiremock::MockServer;
const SPAWN_CALL_ID: &str = "spawn-call-1";
@@ -96,6 +99,13 @@ fn decoded_body(req: &wiremock::Request) -> Option<Vec<u8>> {
}
}
fn log_field<'a>(line: &'a str, name: &str) -> Option<&'a str> {
let prefix = format!("{name}=");
line.split_ascii_whitespace()
.find_map(|field| field.strip_prefix(&prefix))
.map(|value| value.trim_matches('"'))
}
fn has_subagent_notification(req: &ResponsesRequest) -> bool {
req.message_input_texts("user")
.iter()
@@ -1042,8 +1052,16 @@ async fn spawned_multi_agent_v2_child_inherits_parent_developer_context() -> Res
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[tokio::test]
async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result<()> {
let output: &'static Mutex<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(Level::INFO)
.with_writer(MockWriter::new(output))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = start_mock_server().await;
let encrypted_message = "opaque-encrypted-message";
let spawn_args = serde_json::to_string(&json!({
@@ -1098,6 +1116,7 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result
.expect("test config should allow feature update");
});
let test = builder.build(&server).await?;
let root_thread_id = test.session_configured.thread_id;
test.submit_turn(TURN_1_PROMPT).await?;
@@ -1124,6 +1143,41 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result
})])
);
let child_thread_id = test
.thread_manager
.list_thread_ids()
.await
.into_iter()
.find(|thread_id| *thread_id != root_thread_id)
.expect("child thread ID");
let logs = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let logs = String::from_utf8(output.lock().expect("buffer lock").clone())
.expect("logs should be UTF-8");
if logs.contains("kind=\"spawn\"") && logs.contains("state=\"receive\"") {
break logs;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("spawn communication logs should be emitted");
let send = logs
.lines()
.find(|line| line.contains("kind=\"spawn\"") && line.contains("state=\"send\""))
.expect("spawn send event");
assert!(send.contains(&format!("sender_thread_id={root_thread_id}")));
assert!(send.contains(&format!("receiver_thread_id={child_thread_id}")));
assert!(send.contains(&format!("content=\"{encrypted_message}\"")));
let communication_id = log_field(send, "communication_id").expect("communication ID");
logs.lines()
.find(|line| {
line.contains("state=\"receive\"")
&& log_field(line, "communication_id") == Some(communication_id)
})
.expect("correlated receive event");
Ok(())
}