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

feat(agents): enable subagent inbox delivery

Browse Source 
Preserve subagent inbox delivery on the current origin/main base and collapse the branch back to a single commit for easier future restacks.
This commit is contained in:
Friel
2026-03-28 11:35:15 -07:00
parent 65f631c3d6
commit 0e91619094
21 changed files with 1840 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+1
View File
@@ -8791,6 +8791,7 @@ mod tests {
sandbox_policy: codex_protocol::protocol::SandboxPolicy::DangerFullAccess,
cwd: PathBuf::from("/tmp"),
ephemeral: false,
agent_use_function_call_inbox: false,
reasoning_effort: None,
personality: None,
session_source: SessionSource::Cli,
codex-rs/app-server/tests/suite/v2/initialize.rs
+10 -4
View File
@@ -20,11 +20,17 @@ use core_test_support::fs_wait;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[cfg(windows)]
const DEFAULT_NOTIFY_FILE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_NOTIFY_FILE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
#[tokio::test]
async fn initialize_uses_client_info_name_as_originator() -> Result<()> {
@@ -270,9 +276,9 @@ async fn turn_start_notify_payload_includes_initialize_client_name() -> Result<(
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
fs_wait::wait_for_path_exists(&notify_file, Duration::from_secs(5)).await?;
let payload_raw = tokio::fs::read_to_string(&notify_file).await?;
let notify_file = Path::new(&notify_file);
fs_wait::wait_for_path_exists(notify_file, DEFAULT_NOTIFY_FILE_TIMEOUT).await?;
let payload_raw = tokio::fs::read_to_string(notify_file).await?;
let payload: Value = serde_json::from_str(&payload_raw)?;
assert_eq!(payload["client"], "xcode");
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+51 -49
View File
@@ -62,6 +62,7 @@ use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use tempfile::TempDir;
use tokio::time::sleep;
use tokio::time::timeout;
use uuid::Uuid;
use wiremock::Mock;
@@ -70,39 +71,12 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
async fn wait_for_responses_request_count(
server: &wiremock::MockServer,
expected_count: usize,
) -> Result<()> {
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let Some(requests) = server.received_requests().await else {
anyhow::bail!("wiremock did not record requests");
};
let responses_request_count = requests
.iter()
.filter(|request| {
request.method == "POST" && request.url.path().ends_with("/responses")
})
.count();
if responses_request_count == expected_count {
return Ok::<(), anyhow::Error>(());
}
if responses_request_count > expected_count {
anyhow::bail!(
"expected exactly {expected_count} /responses requests, got {responses_request_count}"
);
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
#[tokio::test]
async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -171,7 +145,7 @@ async fn thread_resume_returns_rollout_history() -> Result<()> {
.map(|elem| serde_json::to_value(elem).expect("serialize text element"))
.collect(),
Some("mock_provider"),
/*git_info*/ None,
None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -368,9 +342,7 @@ stream_max_retries = 0
)?;
let state_db =
StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?;
state_db
.mark_backfill_complete(/*last_watermark*/ None)
.await?;
state_db.mark_backfill_complete(None).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -431,7 +403,7 @@ async fn thread_resume_and_read_interrupt_incomplete_rollout_turn_when_thread_is
"Saved user message",
Vec::new(),
Some("mock_provider"),
/*git_info*/ None,
None,
)?;
let rollout_file_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
let persisted_rollout = std::fs::read_to_string(&rollout_file_path)?;
@@ -1018,16 +990,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R
async fn thread_resume_replays_pending_command_execution_request_approval() -> Result<()> {
let responses = vec![
create_final_assistant_message_sse_response("seeded")?,
create_shell_command_sse_response(
vec![
"python3".to_string(),
"-c".to_string(),
"print(42)".to_string(),
],
/*workdir*/ None,
Some(5000),
"call-1",
)?,
create_shell_command_sse_response(fast_shell_command(), None, Some(1000), "call-1")?,
create_final_assistant_message_sse_response("done")?,
];
let server = create_mock_responses_server_sequence_unchecked(responses).await;
@@ -1145,7 +1108,7 @@ async fn thread_resume_replays_pending_command_execution_request_approval() -> R
primary.read_stream_until_notification_message("turn/completed"),
)
.await??;
wait_for_responses_request_count(&server, /*expected_count*/ 3).await?;
wait_for_mock_request_count(&server, 3).await?;
Ok(())
}
@@ -1311,11 +1274,50 @@ async fn thread_resume_replays_pending_file_change_request_approval() -> Result<
primary.read_stream_until_notification_message("turn/completed"),
)
.await??;
wait_for_responses_request_count(&server, /*expected_count*/ 3).await?;
wait_for_mock_request_count(&server, 3).await?;
Ok(())
}
fn fast_shell_command() -> Vec<String> {
if cfg!(windows) {
vec![
"cmd".to_string(),
"/d".to_string(),
"/c".to_string(),
"echo 42".to_string(),
]
} else {
vec![
"python3".to_string(),
"-c".to_string(),
"print(42)".to_string(),
]
}
}
async fn wait_for_mock_request_count(server: &MockServer, expected: usize) -> Result<()> {
let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
loop {
let requests = server
.received_requests()
.await
.ok_or_else(|| anyhow::anyhow!("failed to fetch received requests"))?;
if requests.len() >= expected {
return Ok(());
}
if tokio::time::Instant::now() >= deadline {
anyhow::bail!(
"expected at least {expected} mock requests, observed {}",
requests.len()
);
}
sleep(std::time::Duration::from_millis(50)).await;
}
}
#[tokio::test]
async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -1467,7 +1469,7 @@ async fn thread_resume_surfaces_cloud_requirements_load_errors() -> Result<()> {
"Saved user message",
Vec::new(),
Some("mock_provider"),
/*git_info*/ None,
None,
)?;
let refresh_token_url = format!("{}/oauth/token", server.uri());
let mut mcp = McpProcess::new_with_env(
@@ -1935,7 +1937,7 @@ fn setup_rollout_fixture(codex_home: &Path, server_uri: &str) -> Result<RolloutF
preview,
Vec::new(),
Some("mock_provider"),
/*git_info*/ None,
None,
)?;
let rollout_file_path = rollout_path(codex_home, filename_ts, &conversation_id);
set_rollout_mtime(rollout_file_path.as_path(), expected_updated_at_rfc3339)?;
codex-rs/app-server/tests/suite/v2/thread_start.rs
+7 -2
View File
@@ -200,9 +200,14 @@ async fn thread_start_accepts_flex_service_tier() -> Result<()> {
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let read_timeout = if cfg!(windows) {
std::time::Duration::from_secs(15)
} else {
DEFAULT_READ_TIMEOUT
};
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
timeout(read_timeout, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams {
@@ -212,7 +217,7 @@ async fn thread_start_accepts_flex_service_tier() -> Result<()> {
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
read_timeout,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs
+3
View File
@@ -33,6 +33,9 @@ use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::time::timeout;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
async fn wait_for_responses_request_count_to_stabilize(
codex-rs/app-server/tests/suite/v2/turn_start.rs
+23 -24
View File
@@ -888,7 +888,7 @@ async fn turn_start_uses_migrated_pragmatic_personality_without_override_v2() ->
"2025-01-01T00:00:00Z",
"history user message",
Some("mock_provider"),
/*git_info*/ None,
None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -1025,27 +1025,9 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> {
// Mock server: first turn requests a shell call (elicitation), then completes.
// Second turn same, but we'll set approval_policy=never to avoid elicitation.
let responses = vec![
create_shell_command_sse_response(
vec![
"python3".to_string(),
"-c".to_string(),
"print(42)".to_string(),
],
/*workdir*/ None,
Some(5000),
"call1",
)?,
create_shell_command_sse_response(fast_shell_command(), None, Some(1000), "call1")?,
create_final_assistant_message_sse_response("done 1")?,
create_shell_command_sse_response(
vec![
"python3".to_string(),
"-c".to_string(),
"print(42)".to_string(),
],
/*workdir*/ None,
Some(5000),
"call2",
)?,
create_shell_command_sse_response(fast_shell_command(), None, Some(1000), "call2")?,
create_final_assistant_message_sse_response("done 2")?,
];
let server = create_mock_responses_server_sequence(responses).await;
@@ -1170,6 +1152,23 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> {
Ok(())
}
fn fast_shell_command() -> Vec<String> {
if cfg!(windows) {
vec![
"cmd".to_string(),
"/d".to_string(),
"/c".to_string(),
"echo 42".to_string(),
]
} else {
vec![
"python3".to_string(),
"-c".to_string(),
"print(42)".to_string(),
]
}
}
#[tokio::test]
async fn turn_start_exec_approval_decline_v2() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -1186,7 +1185,7 @@ async fn turn_start_exec_approval_decline_v2() -> Result<()> {
"-c".to_string(),
"print(42)".to_string(),
],
/*workdir*/ None,
None,
Some(5000),
"call-decline",
)?,
@@ -1331,14 +1330,14 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
let responses = vec![
create_shell_command_sse_response(
vec!["echo".to_string(), "first".to_string(), "turn".to_string()],
/*workdir*/ None,
None,
Some(5000),
"call-first",
)?,
create_final_assistant_message_sse_response("done first")?,
create_shell_command_sse_response(
vec!["echo".to_string(), "second".to_string(), "turn".to_string()],
/*workdir*/ None,
None,
Some(5000),
"call-second",
)?,
codex-rs/core/config.schema.json
+6
View File
@@ -332,6 +332,9 @@
"default": null,
"description": "Optional feature toggles scoped to this profile.",
"properties": {
"agent_function_call_inbox": {
"type": "boolean"
},
"apply_patch_freeform": {
"type": "boolean"
},
@@ -1977,6 +1980,9 @@
"default": null,
"description": "Centralized feature flags (new). Prefer this over individual toggles.",
"properties": {
"agent_function_call_inbox": {
"type": "boolean"
},
"apply_patch_freeform": {
"type": "boolean"
},
codex-rs/core/src/agent/control.rs
+1402
View File
File diff suppressed because it is too large Load Diff
codex-rs/core/src/codex.rs
+77
View File
@@ -1124,6 +1124,9 @@ impl SessionConfiguration {
sandbox_policy: self.sandbox_policy.get().clone(),
cwd: self.cwd.to_path_buf(),
ephemeral: self.original_config_do_not_use.ephemeral,
agent_use_function_call_inbox: self
.original_config_do_not_use
.agent_use_function_call_inbox,
reasoning_effort: self.collaboration_mode.reasoning_effort(),
personality: self.personality,
session_source: self.session_source.clone(),
@@ -4334,6 +4337,10 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
handlers::inter_agent_communication(&sess, sub.id.clone(), communication).await;
false
}
Op::InjectResponseItems { items } => {
handlers::inject_response_items(&sess, sub.id.clone(), items).await;
false
}
Op::ExecApproval {
id: approval_id,
turn_id,
@@ -4540,7 +4547,10 @@ mod handlers {
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::items::TurnItem;
use codex_protocol::mcp::RequestId as ProtocolRequestId;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::user_input::UserInput;
use codex_rmcp_client::ElicitationAction;
use codex_rmcp_client::ElicitationResponse;
@@ -4686,6 +4696,73 @@ mod handlers {
}
}
pub async fn inject_response_items(
sess: &Arc<Session>,
sub_id: String,
items: Vec<ResponseInputItem>,
) {
const MAX_TURN_RESTART_ATTEMPTS: usize = 3;
let mut pending_items = items;
let mut attempts = 0usize;
loop {
match sess.inject_response_items(pending_items).await {
Ok(()) => return,
Err(items_without_active_turn) => {
pending_items = items_without_active_turn;
}
}
if attempts >= MAX_TURN_RESTART_ATTEMPTS {
warn!(
attempts,
remaining_items = pending_items.len(),
"dropping response items after repeated turn restart failures"
);
return;
}
attempts += 1;
let turn_input =
pop_leading_user_message_input(&mut pending_items).unwrap_or_else(|| {
vec![UserInput::Text {
text: String::new(),
text_elements: Vec::new(),
}]
});
let turn_sub_id = if attempts == 1 {
sub_id.clone()
} else {
format!("{sub_id}-retry-{attempts}")
};
let current_context = sess.new_default_turn_with_sub_id(turn_sub_id).await;
// Keep injected inbox wakeups visible to telemetry after the TurnContext field rename.
current_context.session_telemetry.user_prompt(&turn_input);
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
sess.spawn_task(Arc::clone(&current_context), turn_input, regular_task)
.await;
if pending_items.is_empty() {
return;
}
}
}
fn pop_leading_user_message_input(
items: &mut Vec<ResponseInputItem>,
) -> Option<Vec<UserInput>> {
let first_item = items.first().cloned()?;
let response_item = ResponseItem::from(first_item);
let TurnItem::UserMessage(user_message) = crate::parse_turn_item(&response_item)? else {
return None;
};
let _ = items.remove(0);
Some(user_message.content)
}
pub async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command: String) {
if let Some((turn_context, cancellation_token)) =
sess.active_turn_context_and_cancellation_token().await
codex-rs/core/src/codex_thread.rs
+1
View File
@@ -38,6 +38,7 @@ pub struct ThreadConfigSnapshot {
pub sandbox_policy: SandboxPolicy,
pub cwd: PathBuf,
pub ephemeral: bool,
pub agent_use_function_call_inbox: bool,
pub reasoning_effort: Option<ReasoningEffort>,
pub personality: Option<Personality>,
pub session_source: SessionSource,
codex-rs/core/src/config/config_tests.rs
+26
View File
@@ -1715,6 +1715,28 @@ fn feature_table_overrides_legacy_flags() -> std::io::Result<()> {
Ok(())
}
#[test]
fn feature_table_enables_agent_function_call_inbox() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
let mut entries = BTreeMap::new();
entries.insert("agent_function_call_inbox".to_string(), true);
let cfg = ConfigToml {
features: Some(crate::features::FeaturesToml { entries }),
..Default::default()
};
let config = Config::load_from_base_config_with_overrides(
cfg,
ConfigOverrides::default(),
codex_home.path().to_path_buf(),
)?;
assert!(config.features.enabled(Feature::AgentFunctionCallInbox));
assert!(config.agent_use_function_call_inbox);
Ok(())
}
#[test]
fn legacy_toggles_map_to_features() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
@@ -4414,6 +4436,7 @@ fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> {
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
agent_use_function_call_inbox: false,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -4556,6 +4579,7 @@ fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> {
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
agent_use_function_call_inbox: false,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -4696,6 +4720,7 @@ fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> {
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
agent_use_function_call_inbox: false,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -4822,6 +4847,7 @@ fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> {
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
agent_use_function_call_inbox: false,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
codex-rs/core/src/config/mod.rs
+6
View File
@@ -403,6 +403,10 @@ pub struct Config {
pub agent_max_threads: Option<usize>,
/// Maximum runtime in seconds for agent job workers before they are failed.
pub agent_job_max_runtime_seconds: Option<u64>,
/// When true, inbound agent messages to non-subagent threads are delivered
/// as a synthetic function_call/function_call_output pair instead of plain
/// user input.
pub agent_use_function_call_inbox: bool,
/// Maximum nesting depth allowed for spawned agent threads.
pub agent_max_depth: i32,
@@ -2309,6 +2313,7 @@ impl Config {
.as_ref()
.and_then(|agents| agents.job_max_runtime_seconds)
.or(DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS);
let agent_use_function_call_inbox = features.enabled(Feature::AgentFunctionCallInbox);
if agent_job_max_runtime_seconds == Some(0) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
@@ -2597,6 +2602,7 @@ impl Config {
agent_roles,
memories: cfg.memories.unwrap_or_default().into(),
agent_job_max_runtime_seconds,
agent_use_function_call_inbox,
codex_home,
sqlite_home,
log_dir,
codex-rs/core/src/stream_events_utils.rs
+11
View File
@@ -428,6 +428,17 @@ pub(crate) fn last_assistant_message_from_item(
pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option<ResponseItem> {
match input {
ResponseInputItem::FunctionCall {
name,
arguments,
call_id,
} => Some(ResponseItem::FunctionCall {
id: None,
name: name.clone(),
namespace: None,
arguments: arguments.clone(),
call_id: call_id.clone(),
}),
ResponseInputItem::FunctionCallOutput { call_id, output } => {
Some(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
codex-rs/core/src/tools/context.rs
+10
View File
@@ -431,6 +431,16 @@ pub(crate) fn response_input_to_code_mode_result(response: ResponseInputItem) ->
raw_arguments: String::new(),
})
}
ResponseInputItem::FunctionCall {
name,
arguments,
call_id,
} => serde_json::json!({
"type": "function_call",
"name": name,
"arguments": arguments,
"call_id": call_id,
}),
}
}
codex-rs/core/src/tools/handlers/multi_agents/send_input.rs
+12 -6
View File
@@ -25,9 +25,9 @@ impl ToolHandler for Handler {
} = invocation;
let arguments = function_arguments(payload)?;
let args: SendInputArgs = parse_arguments(&arguments)?;
let receiver_thread_id = parse_agent_id_target(&args.target)?;
let input_items = parse_collab_input(args.message, args.items)?;
let prompt = render_input_preview(&input_items);
let receiver_thread_id = resolve_agent_target(&session, &turn, &args.target).await?;
let input_items = parse_collab_input(args.message.clone(), args.items.clone())?;
let prompt = input_preview(&input_items);
let receiver_agent = session
.services
.agent_control
@@ -53,9 +53,15 @@ impl ToolHandler for Handler {
.into(),
)
.await;
let agent_control = session.services.agent_control.clone();
let result = agent_control
.send_input(receiver_thread_id, input_items)
let result = session
.services
.agent_control
.send_agent_message_or_input(
receiver_thread_id,
session.conversation_id,
args.message,
args.items,
)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
let status = session
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+1
View File
@@ -2507,6 +2507,7 @@ async fn build_agent_spawn_config_uses_turn_context_values() {
let config = build_agent_spawn_config(&base_instructions, &turn).expect("spawn config");
let mut expected = (*turn.config).clone();
expected.features = config.features.clone();
expected.base_instructions = Some(base_instructions.text);
expected.model = Some(turn.model_info.slug.clone());
expected.model_provider = turn.provider.clone();
codex-rs/core/src/tools/js_repl/mod.rs
+5
View File
@@ -739,6 +739,11 @@ impl JsReplManager {
fn summarize_tool_call_response(response: &ResponseInputItem) -> JsReplToolCallResponseSummary {
match response {
ResponseInputItem::Message { content, .. } => Self::summarize_message_payload(content),
ResponseInputItem::FunctionCall { arguments, .. } => Self::summarize_text_payload(
Some("function_call"),
JsReplToolCallPayloadKind::FunctionText,
arguments,
),
ResponseInputItem::FunctionCallOutput { output, .. } => {
let payload_kind = if output.content_items().is_some() {
JsReplToolCallPayloadKind::FunctionContentItems
codex-rs/features/src/lib.rs
+130
View File
@@ -140,6 +140,8 @@ pub enum Feature {
MultiAgentV2,
/// Enable CSV-backed agent job tools.
SpawnCsv,
/// Deliver inbound agent messages via a synthetic function-call inbox envelope.
AgentFunctionCallInbox,
/// Enable apps.
Apps,
/// Enable the tool_search tool for apps.
@@ -713,6 +715,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::AgentFunctionCallInbox,
key: "agent_function_call_inbox",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::SpawnCsv,
key: "enable_fanout",
@@ -907,3 +915,125 @@ pub fn unstable_features_warning_event(
#[cfg(test)]
mod tests;
#[cfg(test)]
mod inbox_feature_tests {
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn under_development_features_are_disabled_by_default() {
for spec in FEATURES {
if matches!(spec.stage, Stage::UnderDevelopment) {
assert_eq!(
spec.default_enabled, false,
"feature `{}` is under development and must be disabled by default",
spec.key
);
}
}
}
#[test]
fn default_enabled_features_are_stable() {
for spec in FEATURES {
if spec.default_enabled {
assert!(
matches!(spec.stage, Stage::Stable | Stage::Removed),
"feature `{}` is enabled by default but is not stable/removed ({:?})",
spec.key,
spec.stage
);
}
}
}
#[test]
fn use_linux_sandbox_bwrap_is_removed() {
assert_eq!(Feature::UseLinuxSandboxBwrap.stage(), Stage::Removed);
assert_eq!(Feature::UseLinuxSandboxBwrap.default_enabled(), false);
}
#[test]
fn js_repl_is_experimental_and_user_toggleable() {
let spec = Feature::JsRepl.info();
let stage = spec.stage;
let expected_node_version = include_str!("../../node-version.txt").trim_end();
assert!(matches!(stage, Stage::Experimental { .. }));
assert_eq!(stage.experimental_menu_name(), Some("JavaScript REPL"));
assert_eq!(
stage.experimental_menu_description().map(str::to_owned),
Some(format!(
"Enable a persistent Node-backed JavaScript REPL for interactive website debugging and other inline JavaScript execution capabilities. Requires Node >= v{expected_node_version} installed."
))
);
assert_eq!(Feature::JsRepl.default_enabled(), false);
}
#[test]
fn guardian_approval_is_experimental_and_user_toggleable() {
let spec = Feature::GuardianApproval.info();
let stage = spec.stage;
assert!(matches!(stage, Stage::Experimental { .. }));
assert_eq!(stage.experimental_menu_name(), Some("Smart Approvals"));
assert_eq!(
stage.experimental_menu_description().map(str::to_owned),
Some(
"When Codex needs approval for higher-risk actions (e.g. sandbox escapes or blocked network access), route eligible approval requests to a carefully-prompted security reviewer subagent rather than blocking the agent on your input. This can consume significantly more tokens because it runs a subagent on every approval request.".to_string()
)
);
assert_eq!(stage.experimental_announcement(), None);
assert_eq!(Feature::GuardianApproval.default_enabled(), false);
}
#[test]
fn request_permissions_tool_is_under_development() {
assert_eq!(
Feature::RequestPermissionsTool.stage(),
Stage::UnderDevelopment
);
assert_eq!(Feature::RequestPermissionsTool.default_enabled(), false);
}
#[test]
fn image_generation_is_under_development() {
assert_eq!(Feature::ImageGeneration.stage(), Stage::UnderDevelopment);
assert_eq!(Feature::ImageGeneration.default_enabled(), false);
}
#[test]
fn collab_is_legacy_alias_for_multi_agent() {
assert_eq!(feature_for_key("multi_agent"), Some(Feature::Collab));
assert_eq!(feature_for_key("collab"), Some(Feature::Collab));
}
#[test]
fn apps_require_feature_flag_and_chatgpt_auth() {
let mut features = Features::with_defaults();
assert!(!features.apps_enabled_for_auth(None));
features.enable(Feature::Apps);
assert!(!features.apps_enabled_for_auth(None));
let api_key_auth = CodexAuth::from_api_key("test-api-key");
assert!(!features.apps_enabled_for_auth(Some(&api_key_auth)));
let chatgpt_auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
assert!(features.apps_enabled_for_auth(Some(&chatgpt_auth)));
}
#[test]
fn agent_function_call_inbox_is_under_development() {
assert_eq!(
Feature::AgentFunctionCallInbox.stage(),
Stage::UnderDevelopment
);
assert_eq!(Feature::AgentFunctionCallInbox.default_enabled(), false);
assert_eq!(
feature_for_key("agent_function_call_inbox"),
Some(Feature::AgentFunctionCallInbox)
);
}
}
codex-rs/protocol/src/models.rs
+16
View File
@@ -122,6 +122,11 @@ pub enum ResponseInputItem {
role: String,
content: Vec<ContentItem>,
},
FunctionCall {
name: String,
arguments: String,
call_id: String,
},
FunctionCallOutput {
call_id: String,
#[ts(as = "FunctionCallOutputBody")]
@@ -915,6 +920,17 @@ impl From<ResponseInputItem> for ResponseItem {
end_turn: None,
phase: None,
},
ResponseInputItem::FunctionCall {
name,
arguments,
call_id,
} => Self::FunctionCall {
id: None,
name,
namespace: None,
arguments,
call_id,
},
ResponseInputItem::FunctionCallOutput { call_id, output } => {
Self::FunctionCallOutput { call_id, output }
}
codex-rs/protocol/src/protocol.rs
+24
View File
@@ -99,6 +99,26 @@ pub const COLLABORATION_MODE_CLOSE_TAG: &str = "</collaboration_mode>";
pub const REALTIME_CONVERSATION_OPEN_TAG: &str = "<realtime_conversation>";
pub const REALTIME_CONVERSATION_CLOSE_TAG: &str = "</realtime_conversation>";
pub const USER_MESSAGE_BEGIN: &str = "## My request for Codex:";
pub const AGENT_INBOX_KIND: &str = "agent_inbox";
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema)]
pub struct AgentInboxPayload {
pub injected: bool,
pub kind: String,
pub sender_thread_id: ThreadId,
pub message: String,
}
impl AgentInboxPayload {
pub fn new(sender_thread_id: ThreadId, message: String) -> Self {
Self {
injected: true,
kind: AGENT_INBOX_KIND.to_string(),
sender_thread_id,
message,
}
}
}
/// Submission Queue Entry - requests from user
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
@@ -243,6 +263,9 @@ pub enum Op {
final_output_json_schema: Option<Value>,
},
/// Inject non-user response items into an existing turn, or start a turn if needed.
InjectResponseItems { items: Vec<ResponseInputItem> },
/// Similar to [`Op::UserInput`], but contains additional context required
/// for a turn of a [`crate::codex_thread::CodexThread`].
UserTurn {
@@ -590,6 +613,7 @@ impl Op {
Self::UserInputAnswer { .. } => "user_input_answer",
Self::RequestPermissionsResponse { .. } => "request_permissions_response",
Self::DynamicToolResponse { .. } => "dynamic_tool_response",
Self::InjectResponseItems { .. } => "inject_response_items",
Self::AddToHistory { .. } => "add_to_history",
Self::GetHistoryEntryRequest { .. } => "get_history_entry_request",
Self::ListMcpTools => "list_mcp_tools",
docs/config.md
+18
View File
@@ -36,6 +36,24 @@ Codex can run a notification hook when the agent finishes a turn. See the config
When Codex knows which client started the turn, the legacy notify JSON payload also includes a top-level `client` field. The TUI reports `codex-tui`, and the app server reports the `clientInfo.name` value from `initialize`.
## Agent Inbox Delivery
By default, inbound messages from other agents are delivered to non-subagent threads as normal
user input. If you want those handoffs to appear as explicit non-user transcript activity, you can
opt into a synthetic function-call/function-call-output envelope:
```toml
[features]
agent_function_call_inbox = true
```
When enabled, Codex injects inbound agent messages into non-subagent threads as an `agent_inbox`
function-call/function-call-output pair. This is primarily a model-behavior workaround for cases
where you want a subagent handoff to start a valid turn while still being clearly marked as
non-user activity in the transcript.
Messages sent to subagents continue to arrive as normal user input.
## JSON Schema
The generated JSON Schema for `config.toml` lives at `codex-rs/core/config.schema.json`.