diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 17d5c35e83..0dfafc4d13 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2224,6 +2224,7 @@ dependencies = [ "codex-keyring-store", "codex-protocol", "codex-terminal-detection", + "codex-utils-template", "core_test_support", "keyring", "once_cell", @@ -2385,6 +2386,7 @@ dependencies = [ "codex-utils-absolute-path", "codex-utils-image", "codex-utils-string", + "codex-utils-template", "icu_decimal", "icu_locale_core", "icu_provider", @@ -2612,6 +2614,7 @@ dependencies = [ name = "codex-tools" version = "0.0.0" dependencies = [ + "codex-protocol", "pretty_assertions", "rmcp", "serde", diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 1efd514f62..9157228df3 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -5,10 +5,8 @@ use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::role::resolve_role_config; use crate::agent::status::is_final; use crate::codex_thread::ThreadConfigSnapshot; -use crate::context_manager::is_user_turn_boundary; use crate::error::CodexErr; use crate::error::Result as CodexResult; -use crate::event_mapping::parse_turn_item; use crate::find_archived_thread_path_by_id_str; use crate::find_thread_path_by_id_str; use crate::rollout::RolloutRecorder; @@ -20,10 +18,7 @@ use crate::thread_manager::ThreadManagerState; use codex_features::Feature; use codex_protocol::AgentPath; use codex_protocol::ThreadId; -use codex_protocol::items::TurnItem; -use codex_protocol::models::ContentItem; use codex_protocol::models::FunctionCallOutputPayload; -use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; @@ -118,30 +113,36 @@ impl AgentControl { pub(crate) async fn spawn_agent( &self, config: crate::config::Config, - items: Vec, + initial_operation: Op, session_source: Option, ) -> CodexResult { Ok(self - .spawn_agent_internal(config, items, session_source, SpawnAgentOptions::default()) + .spawn_agent_internal( + config, + initial_operation, + session_source, + SpawnAgentOptions::default(), + ) .await? .thread_id) } + /// Spawn an agent thread with some metadata. pub(crate) async fn spawn_agent_with_metadata( &self, config: crate::config::Config, - items: Vec, + initial_operation: Op, session_source: Option, - options: SpawnAgentOptions, + options: SpawnAgentOptions, // TODO(jif) drop with new fork. ) -> CodexResult { - self.spawn_agent_internal(config, items, session_source, options) + self.spawn_agent_internal(config, initial_operation, session_source, options) .await } async fn spawn_agent_internal( &self, config: crate::config::Config, - items: Vec, + initial_operation: Op, session_source: Option, options: SpawnAgentOptions, ) -> CodexResult { @@ -270,7 +271,8 @@ impl AgentControl { ) .await; - self.send_input(new_thread.thread_id, items).await?; + self.send_input(new_thread.thread_id, initial_operation) + .await?; let child_reference = agent_metadata .agent_path .as_ref() @@ -468,23 +470,15 @@ impl AgentControl { pub(crate) async fn send_input( &self, agent_id: ThreadId, - items: Vec, + initial_operation: Op, ) -> CodexResult { - let last_task_message = render_input_preview(&items); + let last_task_message = render_input_preview(&initial_operation); let state = self.upgrade()?; let result = self .handle_thread_request_result( agent_id, &state, - state - .send_op( - agent_id, - Op::UserInput { - items, - final_output_json_schema: None, - }, - ) - .await, + state.send_op(agent_id, initial_operation).await, ) .await; if result.is_ok() { @@ -766,10 +760,7 @@ impl AgentControl { .as_ref() .map(ToString::to_string) .unwrap_or_else(|| thread_id.to_string()); - let last_task_message = match metadata.last_task_message.clone() { - Some(last_task_message) => Some(last_task_message), - None => last_task_message_for_thread(thread.as_ref()).await, - }; + let last_task_message = metadata.last_task_message.clone(); agents.push(ListedAgent { agent_name, agent_status: thread.agent_status().await, @@ -1072,79 +1063,23 @@ fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> b }) } -async fn last_task_message_for_thread(thread: &crate::CodexThread) -> Option { - let pending_input = thread.codex.session.pending_input_snapshot().await; - if let Some(message) = pending_input - .iter() - .rev() - .find_map(last_task_message_from_input_item) - { - return Some(message); +pub(crate) fn render_input_preview(initial_operation: &Op) -> String { + match initial_operation { + Op::UserInput { items, .. } => items + .iter() + .map(|item| match item { + UserInput::Text { text, .. } => text.clone(), + UserInput::Image { .. } => "[image]".to_string(), + UserInput::LocalImage { path } => format!("[local_image:{}]", path.display()), + UserInput::Skill { name, path } => format!("[skill:${name}]({})", path.display()), + UserInput::Mention { name, path } => format!("[mention:${name}]({path})"), + _ => "[input]".to_string(), + }) + .collect::>() + .join("\n"), + Op::InterAgentCommunication { communication } => communication.content.clone(), + _ => String::new(), } - - let queued_input = thread - .codex - .session - .queued_response_items_for_next_turn_snapshot() - .await; - if let Some(message) = queued_input - .iter() - .rev() - .find_map(last_task_message_from_input_item) - { - return Some(message); - } - - let history = thread.codex.session.clone_history().await; - history - .raw_items() - .iter() - .rev() - .find_map(last_task_message_from_item) -} - -fn last_task_message_from_input_item(item: &ResponseInputItem) -> Option { - let response_item: ResponseItem = item.clone().into(); - last_task_message_from_item(&response_item) -} - -fn last_task_message_from_item(item: &ResponseItem) -> Option { - if !is_user_turn_boundary(item) { - return None; - } - - match item { - ResponseItem::Message { role, .. } if role == "user" => { - let Some(TurnItem::UserMessage(message)) = parse_turn_item(item) else { - return None; - }; - Some(render_input_preview(&message.content)) - } - ResponseItem::Message { content, .. } => match content.as_slice() { - [ContentItem::InputText { text }] | [ContentItem::OutputText { text }] => { - serde_json::from_str::(text) - .ok() - .map(|communication| communication.content) - } - _ => None, - }, - _ => None, - } -} - -fn render_input_preview(items: &[UserInput]) -> String { - items - .iter() - .map(|item| match item { - UserInput::Text { text, .. } => text.clone(), - UserInput::Image { .. } => "[image]".to_string(), - UserInput::LocalImage { path } => format!("[local_image:{}]", path.display()), - UserInput::Skill { name, path } => format!("[skill:${name}]({})", path.display()), - UserInput::Mention { name, path } => format!("[mention:${name}]({path})"), - _ => "[input]".to_string(), - }) - .collect::>() - .join("\n") } fn thread_spawn_depth(session_source: &SessionSource) -> Option { diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 878d7d0991..41a01c5ff7 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -54,11 +54,12 @@ async fn test_config() -> (TempDir, Config) { test_config_with_cli_overrides(Vec::new()).await } -fn text_input(text: &str) -> Vec { +fn text_input(text: &str) -> Op { vec![UserInput::Text { text: text.to_string(), text_elements: Vec::new(), }] + .into() } struct AgentControlHarness { @@ -217,7 +218,8 @@ async fn send_input_errors_when_manager_dropped() { vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), - }], + }] + .into(), ) .await .expect_err("send_input should fail without a manager"); @@ -321,7 +323,8 @@ async fn send_input_errors_when_thread_missing() { vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), - }], + }] + .into(), ) .await .expect_err("send_input should fail for missing thread"); @@ -387,7 +390,8 @@ async fn send_input_submits_user_message() { vec![UserInput::Text { text: "hello from tests".to_string(), text_elements: Vec::new(), - }], + }] + .into(), ) .await .expect("send_input should succeed"); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 54f5c93f2d..9400af7219 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -3980,17 +3980,6 @@ impl Session { } } - pub(crate) async fn pending_input_snapshot(&self) -> Vec { - let active = self.active_turn.lock().await; - match active.as_ref() { - Some(at) => { - let ts = at.turn_state.lock().await; - ts.pending_input_snapshot() - } - None => Vec::with_capacity(0), - } - } - /// Queue response items to be injected into the next active turn created for this session. pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec) { if items.is_empty() { @@ -4005,12 +3994,6 @@ impl Session { std::mem::take(&mut *self.idle_pending_input.lock().await) } - pub(crate) async fn queued_response_items_for_next_turn_snapshot( - &self, - ) -> Vec { - self.idle_pending_input.lock().await.clone() - } - pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool { !self.idle_pending_input.lock().await.is_empty() } diff --git a/codex-rs/core/src/memories/phase2.rs b/codex-rs/core/src/memories/phase2.rs index 321b314abb..78f762ed0e 100644 --- a/codex-rs/core/src/memories/phase2.rs +++ b/codex-rs/core/src/memories/phase2.rs @@ -132,7 +132,7 @@ pub(super) async fn run(session: &Arc, config: Arc) { let thread_id = match session .services .agent_control - .spawn_agent(agent_config, prompt, Some(source)) + .spawn_agent(agent_config, prompt.into(), Some(source)) .await { Ok(thread_id) => thread_id, diff --git a/codex-rs/core/src/models_manager/collaboration_mode_presets.rs b/codex-rs/core/src/models_manager/collaboration_mode_presets.rs index dceab9f3bd..afb002bce8 100644 --- a/codex-rs/core/src/models_manager/collaboration_mode_presets.rs +++ b/codex-rs/core/src/models_manager/collaboration_mode_presets.rs @@ -2,13 +2,19 @@ use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::TUI_VISIBLE_COLLABORATION_MODES; use codex_protocol::openai_models::ReasoningEffort; +use codex_utils_template::Template; +use std::sync::LazyLock; const COLLABORATION_MODE_PLAN: &str = include_str!("../../templates/collaboration_mode/plan.md"); const COLLABORATION_MODE_DEFAULT: &str = include_str!("../../templates/collaboration_mode/default.md"); -const KNOWN_MODE_NAMES_PLACEHOLDER: &str = "{{KNOWN_MODE_NAMES}}"; -const REQUEST_USER_INPUT_AVAILABILITY_PLACEHOLDER: &str = "{{REQUEST_USER_INPUT_AVAILABILITY}}"; -const ASKING_QUESTIONS_GUIDANCE_PLACEHOLDER: &str = "{{ASKING_QUESTIONS_GUIDANCE}}"; +const KNOWN_MODE_NAMES_TEMPLATE_KEY: &str = "KNOWN_MODE_NAMES"; +const REQUEST_USER_INPUT_AVAILABILITY_TEMPLATE_KEY: &str = "REQUEST_USER_INPUT_AVAILABILITY"; +const ASKING_QUESTIONS_GUIDANCE_TEMPLATE_KEY: &str = "ASKING_QUESTIONS_GUIDANCE"; +static COLLABORATION_MODE_DEFAULT_TEMPLATE: LazyLock