diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 2798dcb92e..9644b93263 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -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, diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 165160468f..ca33aae852 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -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(¬ify_file, Duration::from_secs(5)).await?; - let payload_raw = tokio::fs::read_to_string(¬ify_file).await?; + let notify_file = Path::new(¬ify_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"); diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 564de50d3a..469d856207 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -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 { + 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 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??; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs index 5808f0fe79..da0304664d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs @@ -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( diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 3155a4be9c..68add83dad 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -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 { + 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", )?, diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 78113b126d..8254814166 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -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" }, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 9157228df3..9b17791acd 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -18,8 +18,13 @@ use crate::thread_manager::ThreadManagerState; use codex_features::Feature; use codex_protocol::AgentPath; use codex_protocol::ThreadId; +use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputBody; use codex_protocol::models::FunctionCallOutputPayload; +use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::AGENT_INBOX_KIND; +use codex_protocol::protocol::AgentInboxPayload; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::Op; @@ -36,6 +41,7 @@ use std::sync::Arc; use std::sync::Weak; use tokio::sync::watch; use tracing::warn; +use uuid::Uuid; const AGENT_NAMES: &str = include_str!("agent_names.txt"); const FORKED_SPAWN_AGENT_OUTPUT_MESSAGE: &str = "You are the newly spawned agent. The prior conversation history was forked from your parent agent. Treat the next user message as your new task, and use the forked history only as background context."; @@ -527,6 +533,76 @@ impl AgentControl { result } + pub(crate) async fn send_prompt( + &self, + agent_id: ThreadId, + prompt: String, + ) -> CodexResult { + self.send_input( + agent_id, + vec![UserInput::Text { + text: prompt, + text_elements: Vec::new(), + }], + ) + .await + } + + pub(crate) async fn send_agent_message( + &self, + agent_id: ThreadId, + sender_thread_id: ThreadId, + message: String, + ) -> CodexResult { + let state = self.upgrade()?; + let thread = state.get_thread(agent_id).await?; + let snapshot = thread.config_snapshot().await; + if matches!(snapshot.session_source, SessionSource::SubAgent(_)) + || !snapshot.agent_use_function_call_inbox + { + return self.send_prompt(agent_id, message).await; + } + + let prepend_turn_start_user_message = + { !thread.codex.session.active_turn.lock().await.is_some() }; + let result = state + .send_op( + agent_id, + Op::InjectResponseItems { + items: build_agent_inbox_items( + sender_thread_id, + message, + prepend_turn_start_user_message, + )?, + }, + ) + .await; + if matches!(result, Err(CodexErr::InternalAgentDied)) { + let _ = state.remove_thread(&agent_id).await; + self.state.release_spawned_thread(agent_id); + } + result + } + + pub(crate) async fn send_agent_message_or_input( + &self, + agent_id: ThreadId, + sender_thread_id: ThreadId, + message: Option, + items: Option>, + ) -> CodexResult { + match (message, items) { + (Some(message), None) => { + self.send_agent_message(agent_id, sender_thread_id, message) + .await + } + (None, Some(items)) => self.send_input(agent_id, items).await, + _ => Err(CodexErr::UnsupportedOperation( + "invalid agent input".to_string(), + )), + } + } + /// Interrupt the current task for an existing agent thread. pub(crate) async fn interrupt_agent(&self, agent_id: ThreadId) -> CodexResult { let state = self.upgrade()?; @@ -1088,6 +1164,1332 @@ fn thread_spawn_depth(session_source: &SessionSource) -> Option { _ => None, } } + +fn build_agent_inbox_items( + sender_thread_id: ThreadId, + message: String, + prepend_turn_start_user_message: bool, +) -> CodexResult> { + let mut items = Vec::new(); + if prepend_turn_start_user_message { + items.push(ResponseInputItem::Message { + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: String::new(), + }], + }); + } + + let call_id = format!("agent_inbox_{}", Uuid::new_v4()); + let output = serde_json::to_string(&AgentInboxPayload::new(sender_thread_id, message)) + .map_err(|err| { + CodexErr::UnsupportedOperation(format!( + "failed to serialize agent inbox payload: {err}" + )) + })?; + + items.extend([ + ResponseInputItem::FunctionCall { + name: AGENT_INBOX_KIND.to_string(), + arguments: "{}".to_string(), + call_id: call_id.clone(), + }, + ResponseInputItem::FunctionCallOutput { + call_id, + output: FunctionCallOutputPayload { + body: FunctionCallOutputBody::Text(output), + ..Default::default() + }, + }, + ]); + + Ok(items) +} + #[cfg(test)] #[path = "control_tests.rs"] mod tests; +#[cfg(test)] +mod inbox_tests { + use super::*; + use crate::CodexAuth; + use crate::CodexThread; + use crate::ThreadManager; + use crate::agent::agent_status_from_event; + use crate::config::AgentRoleConfig; + use crate::config::Config; + use crate::config::ConfigBuilder; + use crate::config_loader::LoaderOverrides; + use crate::contextual_user_message::SUBAGENT_NOTIFICATION_OPEN_TAG; + use crate::features::Feature; + use assert_matches::assert_matches; + use codex_protocol::config_types::ModeKind; + use codex_protocol::models::ContentItem; + use codex_protocol::models::ResponseInputItem; + use codex_protocol::models::ResponseItem; + use codex_protocol::protocol::ErrorEvent; + use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::SubAgentSource; + use codex_protocol::protocol::TurnAbortReason; + use codex_protocol::protocol::TurnAbortedEvent; + use codex_protocol::protocol::TurnCompleteEvent; + use codex_protocol::protocol::TurnStartedEvent; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use tokio::time::Duration; + use tokio::time::sleep; + use tokio::time::timeout; + use toml::Value as TomlValue; + + async fn test_config_with_cli_overrides( + cli_overrides: Vec<(String, TomlValue)>, + ) -> (TempDir, Config) { + let home = TempDir::new().expect("create temp dir"); + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .cli_overrides(cli_overrides) + .loader_overrides(LoaderOverrides { + #[cfg(target_os = "macos")] + managed_preferences_base64: Some(String::new()), + macos_managed_config_requirements_base64: Some(String::new()), + ..LoaderOverrides::default() + }) + .build() + .await + .expect("load default test config"); + (home, config) + } + + async fn test_config() -> (TempDir, Config) { + test_config_with_cli_overrides(Vec::new()).await + } + + fn text_input(text: &str) -> Vec { + vec![UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }] + } + + struct AgentControlHarness { + _home: TempDir, + config: Config, + manager: ThreadManager, + control: AgentControl, + } + + impl AgentControlHarness { + async fn new() -> Self { + let (home, config) = test_config().await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + Self { + _home: home, + config, + manager, + control, + } + } + + async fn start_thread(&self) -> (ThreadId, Arc) { + let new_thread = self + .manager + .start_thread(self.config.clone()) + .await + .expect("start thread"); + (new_thread.thread_id, new_thread.thread) + } + } + + #[test] + fn build_agent_inbox_items_emits_function_call_and_output() { + let sender_thread_id = ThreadId::new(); + let items = build_agent_inbox_items(sender_thread_id, "watchdog update".to_string(), false) + .expect("tool role should build inbox items"); + + assert_eq!(items.len(), 2); + + let call_id = match &items[0] { + ResponseInputItem::FunctionCall { + name, + arguments, + call_id, + } => { + assert_eq!(name, AGENT_INBOX_KIND); + assert_eq!(arguments, "{}"); + call_id.clone() + } + other => panic!("expected function call item, got {other:?}"), + }; + + match &items[1] { + ResponseInputItem::FunctionCallOutput { + call_id: output_call_id, + output, + } => { + assert_eq!(output_call_id, &call_id); + let output_text = output + .body + .to_text() + .expect("payload should convert to text"); + let payload: AgentInboxPayload = + serde_json::from_str(&output_text).expect("payload should be valid json"); + assert!(payload.injected); + assert_eq!(payload.kind, AGENT_INBOX_KIND); + assert_eq!(payload.sender_thread_id, sender_thread_id); + assert_eq!(payload.message, "watchdog update"); + } + other => panic!("expected function call output item, got {other:?}"), + } + } + + #[test] + fn build_agent_inbox_items_prepends_empty_user_message_when_requested() { + let sender_thread_id = ThreadId::new(); + let items = build_agent_inbox_items(sender_thread_id, "watchdog update".to_string(), true) + .expect("tool role should build inbox items"); + + assert_eq!(items.len(), 3); + assert_eq!( + items[0], + ResponseInputItem::Message { + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: String::new(), + }], + } + ); + assert_matches!(&items[1], ResponseInputItem::FunctionCall { .. }); + assert_matches!(&items[2], ResponseInputItem::FunctionCallOutput { .. }); + } + + #[tokio::test] + async fn send_agent_message_to_root_thread_defaults_to_user_input() { + let harness = AgentControlHarness::new().await; + let (receiver_thread_id, _thread) = harness.start_thread().await; + let sender_thread_id = ThreadId::new(); + + let submission_id = harness + .control + .send_agent_message( + receiver_thread_id, + sender_thread_id, + "watchdog update".to_string(), + ) + .await + .expect("send_agent_message should succeed"); + assert!(!submission_id.is_empty()); + + let expected = ( + receiver_thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "watchdog update".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + + assert_eq!(captured, Some(expected)); + } + + #[tokio::test] + async fn send_agent_message_to_root_thread_injects_response_items_when_enabled() { + let mut harness = AgentControlHarness::new().await; + harness.config.agent_use_function_call_inbox = true; + let (receiver_thread_id, _thread) = harness.start_thread().await; + let sender_thread_id = ThreadId::new(); + + let submission_id = harness + .control + .send_agent_message( + receiver_thread_id, + sender_thread_id, + "watchdog update".to_string(), + ) + .await + .expect("send_agent_message should succeed"); + assert!(!submission_id.is_empty()); + + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|(thread_id, op)| { + *thread_id == receiver_thread_id && matches!(op, Op::InjectResponseItems { .. }) + }) + .expect("expected injected agent inbox op"); + + let Op::InjectResponseItems { items } = captured.1 else { + unreachable!("matched above"); + }; + assert_eq!(items.len(), 3); + match &items[0] { + ResponseInputItem::Message { role, content } => { + assert_eq!(role, "user"); + assert_eq!( + content, + &vec![ContentItem::InputText { + text: String::new(), + }] + ); + } + other => panic!("expected prepended user message, got {other:?}"), + } + match &items[1] { + ResponseInputItem::FunctionCall { + name, arguments, .. + } => { + assert_eq!(name, AGENT_INBOX_KIND); + assert_eq!(arguments, "{}"); + } + other => panic!("expected function call item, got {other:?}"), + } + match &items[2] { + ResponseInputItem::FunctionCallOutput { output, .. } => { + let output_text = output + .body + .to_text() + .expect("payload should convert to text"); + let payload: AgentInboxPayload = + serde_json::from_str(&output_text).expect("payload should be valid json"); + assert_eq!(payload.sender_thread_id, sender_thread_id); + assert_eq!(payload.message, "watchdog update"); + } + other => panic!("expected function call output item, got {other:?}"), + } + } + + fn has_subagent_notification(history_items: &[ResponseItem]) -> bool { + history_items.iter().any(|item| { + let ResponseItem::Message { role, content, .. } = item else { + return false; + }; + if role != "user" { + return false; + } + content.iter().any(|content_item| match content_item { + ContentItem::InputText { text } | ContentItem::OutputText { text } => { + text.contains(SUBAGENT_NOTIFICATION_OPEN_TAG) + } + ContentItem::InputImage { .. } => false, + }) + }) + } + + /// Returns true when any message item contains `needle` in a text span. + fn history_contains_text(history_items: &[ResponseItem], needle: &str) -> bool { + history_items.iter().any(|item| { + let ResponseItem::Message { content, .. } = item else { + return false; + }; + content.iter().any(|content_item| match content_item { + ContentItem::InputText { text } | ContentItem::OutputText { text } => { + text.contains(needle) + } + ContentItem::InputImage { .. } => false, + }) + }) + } + + async fn wait_for_subagent_notification(parent_thread: &Arc) -> bool { + let wait = async { + loop { + let history_items = parent_thread + .codex + .session + .clone_history() + .await + .raw_items() + .to_vec(); + if has_subagent_notification(&history_items) { + return true; + } + sleep(Duration::from_millis(25)).await; + } + }; + timeout(Duration::from_secs(5), wait).await.is_ok() + } + + #[tokio::test] + async fn send_input_errors_when_manager_dropped() { + let control = AgentControl::default(); + let err = control + .send_input( + ThreadId::new(), + vec![UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }], + ) + .await + .expect_err("send_input should fail without a manager"); + assert_eq!( + err.to_string(), + "unsupported operation: thread manager dropped" + ); + } + + #[tokio::test] + async fn get_status_returns_not_found_without_manager() { + let control = AgentControl::default(); + let got = control.get_status(ThreadId::new()).await; + assert_eq!(got, AgentStatus::NotFound); + } + + #[tokio::test] + async fn on_event_updates_status_from_task_started() { + let status = agent_status_from_event(&EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-1".to_string(), + model_context_window: None, + collaboration_mode_kind: ModeKind::Default, + })); + assert_eq!(status, Some(AgentStatus::Running)); + } + + #[tokio::test] + async fn on_event_updates_status_from_task_complete() { + let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: "turn-1".to_string(), + last_agent_message: Some("done".to_string()), + })); + let expected = AgentStatus::Completed(Some("done".to_string())); + assert_eq!(status, Some(expected)); + } + + #[tokio::test] + async fn on_event_updates_status_from_error() { + let status = agent_status_from_event(&EventMsg::Error(ErrorEvent { + message: "boom".to_string(), + codex_error_info: None, + })); + + let expected = AgentStatus::Errored("boom".to_string()); + assert_eq!(status, Some(expected)); + } + + #[tokio::test] + async fn on_event_updates_status_from_turn_aborted() { + let status = agent_status_from_event(&EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: Some("turn-1".to_string()), + reason: TurnAbortReason::Interrupted, + })); + + let expected = AgentStatus::Errored("Interrupted".to_string()); + assert_eq!(status, Some(expected)); + } + + #[tokio::test] + async fn on_event_updates_status_from_shutdown_complete() { + let status = agent_status_from_event(&EventMsg::ShutdownComplete); + assert_eq!(status, Some(AgentStatus::Shutdown)); + } + + #[tokio::test] + async fn spawn_agent_errors_when_manager_dropped() { + let control = AgentControl::default(); + let (_home, config) = test_config().await; + let err = control + .spawn_agent(config, text_input("hello"), None) + .await + .expect_err("spawn_agent should fail without a manager"); + assert_eq!( + err.to_string(), + "unsupported operation: thread manager dropped" + ); + } + + #[tokio::test] + async fn resume_agent_errors_when_manager_dropped() { + let control = AgentControl::default(); + let (_home, config) = test_config().await; + let err = control + .resume_agent_from_rollout(config, ThreadId::new(), SessionSource::Exec) + .await + .expect_err("resume_agent should fail without a manager"); + assert_eq!( + err.to_string(), + "unsupported operation: thread manager dropped" + ); + } + + #[tokio::test] + async fn send_input_errors_when_thread_missing() { + let harness = AgentControlHarness::new().await; + let thread_id = ThreadId::new(); + let err = harness + .control + .send_input( + thread_id, + vec![UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }], + ) + .await + .expect_err("send_input should fail for missing thread"); + assert_matches!(err, CodexErr::ThreadNotFound(id) if id == thread_id); + } + + #[tokio::test] + async fn get_status_returns_not_found_for_missing_thread() { + let harness = AgentControlHarness::new().await; + let status = harness.control.get_status(ThreadId::new()).await; + assert_eq!(status, AgentStatus::NotFound); + } + + #[tokio::test] + async fn get_status_returns_pending_init_for_new_thread() { + let harness = AgentControlHarness::new().await; + let (thread_id, _) = harness.start_thread().await; + let status = harness.control.get_status(thread_id).await; + assert_eq!(status, AgentStatus::PendingInit); + } + + #[tokio::test] + async fn subscribe_status_errors_for_missing_thread() { + let harness = AgentControlHarness::new().await; + let thread_id = ThreadId::new(); + let err = harness + .control + .subscribe_status(thread_id) + .await + .expect_err("subscribe_status should fail for missing thread"); + assert_matches!(err, CodexErr::ThreadNotFound(id) if id == thread_id); + } + + #[tokio::test] + async fn subscribe_status_updates_on_shutdown() { + let harness = AgentControlHarness::new().await; + let (thread_id, thread) = harness.start_thread().await; + let mut status_rx = harness + .control + .subscribe_status(thread_id) + .await + .expect("subscribe_status should succeed"); + assert_eq!(status_rx.borrow().clone(), AgentStatus::PendingInit); + + let _ = thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); + + let _ = status_rx.changed().await; + assert_eq!(status_rx.borrow().clone(), AgentStatus::Shutdown); + } + + #[tokio::test] + async fn send_input_submits_user_message() { + let harness = AgentControlHarness::new().await; + let (thread_id, _thread) = harness.start_thread().await; + + let submission_id = harness + .control + .send_input( + thread_id, + vec![UserInput::Text { + text: "hello from tests".to_string(), + text_elements: Vec::new(), + }], + ) + .await + .expect("send_input should succeed"); + assert!(!submission_id.is_empty()); + let expected = ( + thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "hello from tests".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + assert_eq!(captured, Some(expected)); + } + + #[tokio::test] + async fn spawn_agent_creates_thread_and_sends_prompt() { + let harness = AgentControlHarness::new().await; + let thread_id = harness + .control + .spawn_agent(harness.config.clone(), text_input("spawned"), None) + .await + .expect("spawn_agent should succeed"); + let _thread = harness + .manager + .get_thread(thread_id) + .await + .expect("thread should be registered"); + let expected = ( + thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "spawned".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + assert_eq!(captured, Some(expected)); + } + + #[tokio::test] + async fn spawn_agent_can_fork_parent_thread_history() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + parent_thread + .inject_user_message_without_turn("parent seed context".to_string()) + .await; + let turn_context = parent_thread.codex.session.new_default_turn().await; + let parent_spawn_call_id = "spawn-call-history".to_string(); + let parent_spawn_call = ResponseItem::FunctionCall { + id: None, + name: "spawn_agent".to_string(), + namespace: None, + arguments: "{}".to_string(), + call_id: parent_spawn_call_id.clone(), + }; + parent_thread + .codex + .session + .record_conversation_items(turn_context.as_ref(), &[parent_spawn_call]) + .await; + parent_thread + .codex + .session + .ensure_rollout_materialized() + .await; + parent_thread.codex.session.flush_rollout().await; + + let child_thread_id = harness + .control + .spawn_agent_with_options( + harness.config.clone(), + text_input("child task"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: None, + })), + SpawnAgentOptions { + fork_parent_spawn_call_id: Some(parent_spawn_call_id), + }, + ) + .await + .expect("forked spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + assert_ne!(child_thread_id, parent_thread_id); + let history = child_thread.codex.session.clone_history().await; + assert!(history_contains_text( + history.raw_items(), + "parent seed context" + )); + + let expected = ( + child_thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "child task".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + assert_eq!(captured, Some(expected)); + + let _ = harness + .control + .shutdown_agent(child_thread_id) + .await + .expect("child shutdown should submit"); + let _ = parent_thread + .submit(Op::Shutdown {}) + .await + .expect("parent shutdown should submit"); + } + + #[tokio::test] + async fn spawn_agent_fork_injects_output_for_parent_spawn_call() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + let turn_context = parent_thread.codex.session.new_default_turn().await; + let parent_spawn_call_id = "spawn-call-1".to_string(); + let parent_spawn_call = ResponseItem::FunctionCall { + id: None, + name: "spawn_agent".to_string(), + namespace: None, + arguments: "{}".to_string(), + call_id: parent_spawn_call_id.clone(), + }; + parent_thread + .codex + .session + .record_conversation_items(turn_context.as_ref(), &[parent_spawn_call]) + .await; + parent_thread + .codex + .session + .ensure_rollout_materialized() + .await; + parent_thread.codex.session.flush_rollout().await; + + let child_thread_id = harness + .control + .spawn_agent_with_options( + harness.config.clone(), + text_input("child task"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: None, + })), + SpawnAgentOptions { + fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), + }, + ) + .await + .expect("forked spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + let history = child_thread.codex.session.clone_history().await; + let injected_output = history.raw_items().iter().find_map(|item| match item { + ResponseItem::FunctionCallOutput { call_id, output } + if call_id == &parent_spawn_call_id => + { + Some(output) + } + _ => None, + }); + let injected_output = + injected_output.expect("forked child should contain synthetic tool output"); + assert_eq!( + injected_output.text_content(), + Some(FORKED_SPAWN_AGENT_OUTPUT_MESSAGE) + ); + assert_eq!(injected_output.success, Some(true)); + + let _ = harness + .control + .shutdown_agent(child_thread_id) + .await + .expect("child shutdown should submit"); + let _ = parent_thread + .submit(Op::Shutdown {}) + .await + .expect("parent shutdown should submit"); + } + + #[tokio::test] + async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + let turn_context = parent_thread.codex.session.new_default_turn().await; + let parent_spawn_call_id = "spawn-call-unflushed".to_string(); + let parent_spawn_call = ResponseItem::FunctionCall { + id: None, + name: "spawn_agent".to_string(), + namespace: None, + arguments: "{}".to_string(), + call_id: parent_spawn_call_id.clone(), + }; + parent_thread + .codex + .session + .record_conversation_items(turn_context.as_ref(), &[parent_spawn_call]) + .await; + + let child_thread_id = harness + .control + .spawn_agent_with_options( + harness.config.clone(), + text_input("child task"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: None, + })), + SpawnAgentOptions { + fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), + }, + ) + .await + .expect("forked spawn should flush parent rollout before loading history"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + let history = child_thread.codex.session.clone_history().await; + + let mut parent_call_index = None; + let mut injected_output_index = None; + for (idx, item) in history.raw_items().iter().enumerate() { + match item { + ResponseItem::FunctionCall { call_id, .. } if call_id == &parent_spawn_call_id => { + parent_call_index = Some(idx); + } + ResponseItem::FunctionCallOutput { call_id, .. } + if call_id == &parent_spawn_call_id => + { + injected_output_index = Some(idx); + } + _ => {} + } + } + + let parent_call_index = + parent_call_index.expect("forked child should include the parent spawn_agent call"); + let injected_output_index = injected_output_index + .expect("forked child should include synthetic output for the parent spawn_agent call"); + assert!(parent_call_index < injected_output_index); + + let _ = harness + .control + .shutdown_agent(child_thread_id) + .await + .expect("child shutdown should submit"); + let _ = parent_thread + .submit(Op::Shutdown {}) + .await + .expect("parent shutdown should submit"); + } + + #[tokio::test] + async fn spawn_agent_respects_max_threads_limit() { + let max_threads = 1usize; + let (_home, config) = test_config_with_cli_overrides(vec![( + "agents.max_threads".to_string(), + TomlValue::Integer(max_threads as i64), + )]) + .await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + + let _ = manager + .start_thread(config.clone()) + .await + .expect("start thread"); + + let first_agent_id = control + .spawn_agent(config.clone(), text_input("hello"), None) + .await + .expect("spawn_agent should succeed"); + + let err = control + .spawn_agent(config, text_input("hello again"), None) + .await + .expect_err("spawn_agent should respect max threads"); + let CodexErr::AgentLimitReached { + max_threads: seen_max_threads, + } = err + else { + panic!("expected CodexErr::AgentLimitReached"); + }; + assert_eq!(seen_max_threads, max_threads); + + let _ = control + .shutdown_agent(first_agent_id) + .await + .expect("shutdown agent"); + } + + #[tokio::test] + async fn spawn_agent_releases_slot_after_shutdown() { + let max_threads = 1usize; + let (_home, config) = test_config_with_cli_overrides(vec![( + "agents.max_threads".to_string(), + TomlValue::Integer(max_threads as i64), + )]) + .await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + + let first_agent_id = control + .spawn_agent(config.clone(), text_input("hello"), None) + .await + .expect("spawn_agent should succeed"); + let _ = control + .shutdown_agent(first_agent_id) + .await + .expect("shutdown agent"); + + let second_agent_id = control + .spawn_agent(config.clone(), text_input("hello again"), None) + .await + .expect("spawn_agent should succeed after shutdown"); + let _ = control + .shutdown_agent(second_agent_id) + .await + .expect("shutdown agent"); + } + + #[tokio::test] + async fn spawn_agent_limit_shared_across_clones() { + let max_threads = 1usize; + let (_home, config) = test_config_with_cli_overrides(vec![( + "agents.max_threads".to_string(), + TomlValue::Integer(max_threads as i64), + )]) + .await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + let cloned = control.clone(); + + let first_agent_id = cloned + .spawn_agent(config.clone(), text_input("hello"), None) + .await + .expect("spawn_agent should succeed"); + + let err = control + .spawn_agent(config, text_input("hello again"), None) + .await + .expect_err("spawn_agent should respect shared guard"); + let CodexErr::AgentLimitReached { max_threads } = err else { + panic!("expected CodexErr::AgentLimitReached"); + }; + assert_eq!(max_threads, 1); + + let _ = control + .shutdown_agent(first_agent_id) + .await + .expect("shutdown agent"); + } + + #[tokio::test] + async fn resume_agent_respects_max_threads_limit() { + let max_threads = 1usize; + let (_home, config) = test_config_with_cli_overrides(vec![( + "agents.max_threads".to_string(), + TomlValue::Integer(max_threads as i64), + )]) + .await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + + let resumable_id = control + .spawn_agent(config.clone(), text_input("hello"), None) + .await + .expect("spawn_agent should succeed"); + let _ = control + .shutdown_agent(resumable_id) + .await + .expect("shutdown resumable thread"); + + let active_id = control + .spawn_agent(config.clone(), text_input("occupy"), None) + .await + .expect("spawn_agent should succeed for active slot"); + + let err = control + .resume_agent_from_rollout(config, resumable_id, SessionSource::Exec) + .await + .expect_err("resume should respect max threads"); + let CodexErr::AgentLimitReached { + max_threads: seen_max_threads, + } = err + else { + panic!("expected CodexErr::AgentLimitReached"); + }; + assert_eq!(seen_max_threads, max_threads); + + let _ = control + .shutdown_agent(active_id) + .await + .expect("shutdown active thread"); + } + + #[tokio::test] + async fn resume_agent_releases_slot_after_resume_failure() { + let max_threads = 1usize; + let (_home, config) = test_config_with_cli_overrides(vec![( + "agents.max_threads".to_string(), + TomlValue::Integer(max_threads as i64), + )]) + .await; + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + + let _ = control + .resume_agent_from_rollout(config.clone(), ThreadId::new(), SessionSource::Exec) + .await + .expect_err("resume should fail for missing rollout path"); + + let resumed_id = control + .spawn_agent(config, text_input("hello"), None) + .await + .expect("spawn should succeed after failed resume"); + let _ = control + .shutdown_agent(resumed_id) + .await + .expect("shutdown resumed thread"); + } + + #[tokio::test] + async fn spawn_child_completion_notifies_parent_history() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + + let child_thread_id = harness + .control + .spawn_agent( + harness.config.clone(), + text_input("hello child"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: Some("explorer".to_string()), + })), + ) + .await + .expect("child spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should exist"); + let mut status_rx = harness + .control + .subscribe_status(child_thread_id) + .await + .expect("status subscription should succeed"); + if matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { + timeout(Duration::from_secs(5), async { + loop { + status_rx + .changed() + .await + .expect("child status should advance past pending init"); + if !matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { + break; + } + } + }) + .await + .expect("child should initialize before shutdown"); + } + let _ = child_thread + .submit(Op::Shutdown {}) + .await + .expect("child shutdown should submit"); + + assert_eq!(wait_for_subagent_notification(&parent_thread).await, true); + } + + #[tokio::test] + async fn completion_watcher_notifies_parent_when_child_is_missing() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + let child_thread_id = ThreadId::new(); + + harness.control.maybe_start_completion_watcher( + child_thread_id, + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: Some("explorer".to_string()), + })), + ); + + assert_eq!(wait_for_subagent_notification(&parent_thread).await, true); + + let history_items = parent_thread + .codex + .session + .clone_history() + .await + .raw_items() + .to_vec(); + assert_eq!( + history_contains_text( + &history_items, + &format!("\"agent_id\":\"{child_thread_id}\"") + ), + true + ); + assert_eq!( + history_contains_text(&history_items, "\"status\":\"not_found\""), + true + ); + } + + #[tokio::test] + async fn spawn_thread_subagent_gets_random_nickname_in_session_source() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, _parent_thread) = harness.start_thread().await; + + let child_thread_id = harness + .control + .spawn_agent( + harness.config.clone(), + text_input("hello child"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: Some("explorer".to_string()), + })), + ) + .await + .expect("child spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + let snapshot = child_thread.config_snapshot().await; + + let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: seen_parent_thread_id, + depth, + agent_nickname, + agent_role, + }) = snapshot.session_source + else { + panic!("expected thread-spawn sub-agent source"); + }; + assert_eq!(seen_parent_thread_id, parent_thread_id); + assert_eq!(depth, 1); + assert!(agent_nickname.is_some()); + assert_eq!(agent_role, Some("explorer".to_string())); + } + + #[tokio::test] + async fn spawn_thread_subagent_uses_role_specific_nickname_candidates() { + let mut harness = AgentControlHarness::new().await; + harness.config.agent_roles.insert( + "researcher".to_string(), + AgentRoleConfig { + description: Some("Research role".to_string()), + config_file: None, + nickname_candidates: Some(vec!["Atlas".to_string()]), + }, + ); + let (parent_thread_id, _parent_thread) = harness.start_thread().await; + + let child_thread_id = harness + .control + .spawn_agent( + harness.config.clone(), + text_input("hello child"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: Some("researcher".to_string()), + })), + ) + .await + .expect("child spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + let snapshot = child_thread.config_snapshot().await; + + let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_nickname, .. }) = + snapshot.session_source + else { + panic!("expected thread-spawn sub-agent source"); + }; + assert_eq!(agent_nickname, Some("Atlas".to_string())); + } + + #[tokio::test] + async fn resume_thread_subagent_restores_stored_nickname_and_role() { + let (home, mut config) = test_config().await; + config + .features + .enable(Feature::Sqlite) + .expect("test config should allow sqlite"); + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + let harness = AgentControlHarness { + _home: home, + config, + manager, + control, + }; + let (parent_thread_id, _parent_thread) = harness.start_thread().await; + + let child_thread_id = harness + .control + .spawn_agent( + harness.config.clone(), + text_input("hello child"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: Some("explorer".to_string()), + })), + ) + .await + .expect("child spawn should succeed"); + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should exist"); + let mut status_rx = harness + .control + .subscribe_status(child_thread_id) + .await + .expect("status subscription should succeed"); + if matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { + timeout(Duration::from_secs(5), async { + loop { + status_rx + .changed() + .await + .expect("child status should advance past pending init"); + if !matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { + break; + } + } + }) + .await + .expect("child should initialize before shutdown"); + } + let original_snapshot = child_thread.config_snapshot().await; + let original_nickname = original_snapshot + .session_source + .get_nickname() + .expect("spawned sub-agent should have a nickname"); + let state_db = child_thread + .state_db() + .expect("sqlite state db should be available for nickname resume test"); + timeout(Duration::from_secs(5), async { + loop { + if let Ok(Some(metadata)) = state_db.get_thread(child_thread_id).await + && metadata.agent_nickname.is_some() + && metadata.agent_role.as_deref() == Some("explorer") + { + break; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("child thread metadata should be persisted to sqlite before shutdown"); + + let _ = harness + .control + .shutdown_agent(child_thread_id) + .await + .expect("child shutdown should submit"); + + let resumed_thread_id = harness + .control + .resume_agent_from_rollout( + harness.config.clone(), + child_thread_id, + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_nickname: None, + agent_role: None, + }), + ) + .await + .expect("resume should succeed"); + assert_eq!(resumed_thread_id, child_thread_id); + + let resumed_snapshot = harness + .manager + .get_thread(resumed_thread_id) + .await + .expect("resumed child thread should exist") + .config_snapshot() + .await; + let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: resumed_parent_thread_id, + depth: resumed_depth, + agent_nickname: resumed_nickname, + agent_role: resumed_role, + }) = resumed_snapshot.session_source + else { + panic!("expected thread-spawn sub-agent source"); + }; + assert_eq!(resumed_parent_thread_id, parent_thread_id); + assert_eq!(resumed_depth, 1); + assert_eq!(resumed_nickname, Some(original_nickname)); + assert_eq!(resumed_role, Some("explorer".to_string())); + + let _ = harness + .control + .shutdown_agent(resumed_thread_id) + .await + .expect("resumed child shutdown should submit"); + } +} diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 9400af7219..e347ef0772 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -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, config: Arc, 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, + sub_id: String, + items: Vec, + ) { + 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(¤t_context) + .await; + let regular_task = sess.take_startup_regular_task().await.unwrap_or_default(); + sess.spawn_task(Arc::clone(¤t_context), turn_input, regular_task) + .await; + + if pending_items.is_empty() { + return; + } + } + } + + fn pop_leading_user_message_input( + items: &mut Vec, + ) -> Option> { + 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, sub_id: String, command: String) { if let Some((turn_context, cancellation_token)) = sess.active_turn_context_and_cancellation_token().await diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 0635718c9c..2b9bb940b6 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -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, pub personality: Option, pub session_source: SessionSource, diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 00227fe2b3..6a683e36e8 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -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"), diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 1a0722119b..6181a22df8 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -403,6 +403,10 @@ pub struct Config { pub agent_max_threads: Option, /// Maximum runtime in seconds for agent job workers before they are failed. pub agent_job_max_runtime_seconds: Option, + /// 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, diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index cd77f1d5a3..af0d109dc9 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -428,6 +428,17 @@ pub(crate) fn last_assistant_message_from_item( pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option { 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(), diff --git a/codex-rs/core/src/tools/context.rs b/codex-rs/core/src/tools/context.rs index 58d040cbe6..d7ac456a18 100644 --- a/codex-rs/core/src/tools/context.rs +++ b/codex-rs/core/src/tools/context.rs @@ -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, + }), } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index 3c8527712b..6d6b7c587f 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -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 diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 8006ee2204..f95832fa70 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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(); diff --git a/codex-rs/core/src/tools/js_repl/mod.rs b/codex-rs/core/src/tools/js_repl/mod.rs index beb2930b22..d3e66b0ee1 100644 --- a/codex-rs/core/src/tools/js_repl/mod.rs +++ b/codex-rs/core/src/tools/js_repl/mod.rs @@ -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 diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 48927f4915..f3c081c188 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -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) + ); + } +} diff --git a/codex-rs/protocol/src/models.rs b/codex-rs/protocol/src/models.rs index d4e568fea9..ab0e40d6f0 100644 --- a/codex-rs/protocol/src/models.rs +++ b/codex-rs/protocol/src/models.rs @@ -122,6 +122,11 @@ pub enum ResponseInputItem { role: String, content: Vec, }, + FunctionCall { + name: String, + arguments: String, + call_id: String, + }, FunctionCallOutput { call_id: String, #[ts(as = "FunctionCallOutputBody")] @@ -915,6 +920,17 @@ impl From 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 } } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 27e7b4e464..87c9ca085b 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -99,6 +99,26 @@ pub const COLLABORATION_MODE_CLOSE_TAG: &str = ""; pub const REALTIME_CONVERSATION_OPEN_TAG: &str = ""; pub const REALTIME_CONVERSATION_CLOSE_TAG: &str = ""; 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, }, + /// Inject non-user response items into an existing turn, or start a turn if needed. + InjectResponseItems { items: Vec }, + /// 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", diff --git a/docs/config.md b/docs/config.md index 71f3548deb..9036785451 100644 --- a/docs/config.md +++ b/docs/config.md @@ -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`.