diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index d2296c75c5..f3053a0daf 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -37,6 +37,7 @@ use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::ImageGenerationBeginEvent; use codex_protocol::protocol::ImageGenerationEndEvent; +use codex_protocol::protocol::InjectedMessageEvent; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::McpToolCallBeginEvent; @@ -207,6 +208,7 @@ impl ThreadHistoryBuilder { EventMsg::TurnAborted(payload) => self.handle_turn_aborted(payload), EventMsg::TurnStarted(payload) => self.handle_turn_started(payload), EventMsg::TurnComplete(payload) => self.handle_turn_complete(payload), + EventMsg::InjectedMessage(payload) => self.handle_injected_message(payload), _ => {} } } @@ -268,6 +270,21 @@ impl ThreadHistoryBuilder { self.current_turn = Some(turn); } + fn handle_injected_message(&mut self, payload: &InjectedMessageEvent) { + if payload.content.trim().is_empty() { + return; + } + + let id = self.next_item_id(); + self.ensure_turn().items.push(ThreadItem::UserMessage { + id, + content: vec![UserInput::Text { + text: payload.content.clone(), + text_elements: Vec::new(), + }], + }); + } + fn handle_agent_message( &mut self, text: String, diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 08a51108ae..4643766610 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -99,6 +99,7 @@ use codex_app_server_protocol::TurnPlanStep; use codex_app_server_protocol::TurnPlanUpdatedNotification; use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput as V2UserInput; use codex_app_server_protocol::build_command_execution_end_item; use codex_app_server_protocol::build_file_change_approval_request_item; use codex_app_server_protocol::build_file_change_begin_item; @@ -1470,6 +1471,24 @@ pub(crate) async fn apply_bespoke_event_handling( .send_server_notification(ServerNotification::ItemCompleted(completed)) .await; } + EventMsg::InjectedMessage(event) => { + if !event.content.trim().is_empty() { + let notification = ItemCompletedNotification { + thread_id: conversation_id.to_string(), + turn_id: event_turn_id.clone(), + item: ThreadItem::UserMessage { + id: uuid::Uuid::new_v4().to_string(), + content: vec![V2UserInput::Text { + text: event.content, + text_elements: Vec::new(), + }], + }, + }; + outgoing + .send_server_notification(ServerNotification::ItemCompleted(notification)) + .await; + } + } EventMsg::ItemStarted(item_started_event) => { let item: ThreadItem = item_started_event.item.clone().into(); let notification = ItemStartedNotification { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index c19c4b48fd..cee58f4527 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -113,6 +113,7 @@ use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::FileChange; use codex_protocol::protocol::HasLegacyEvent; +use codex_protocol::protocol::InjectedMessageEvent; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; @@ -283,6 +284,7 @@ use crate::mentions::collect_explicit_app_ids; use crate::mentions::collect_explicit_plugin_mentions; use crate::mentions::collect_tool_mentions_from_messages; use crate::network_policy_decision::execpolicy_network_rule_amendment; +use crate::pending_input::PendingInputItem; use crate::plugins::PluginsManager; use crate::plugins::build_plugin_injections; use crate::plugins::render_plugins_section; @@ -839,7 +841,7 @@ pub(crate) struct Session { pub(crate) active_turn: Mutex>, mailbox: Mailbox, mailbox_rx: Mutex, - idle_pending_input: Mutex>, // TODO (jif) merge with mailbox! + idle_pending_input: Mutex>, // TODO (jif) merge with mailbox! pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) services: SessionServices, js_repl: Arc, @@ -4086,6 +4088,19 @@ impl Session { self.ensure_rollout_materialized().await; } + pub(crate) async fn record_generated_message_and_emit_display( + &self, + turn_context: &TurnContext, + response_item: ResponseItem, + injected_event: InjectedMessageEvent, + ) { + self.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) + .await; + self.send_event(turn_context, EventMsg::InjectedMessage(injected_event)) + .await; + self.ensure_rollout_materialized().await; + } + pub(crate) async fn notify_background_event( &self, turn_context: &TurnContext, @@ -4199,7 +4214,7 @@ impl Session { } let mut turn_state = active_turn.turn_state.lock().await; - turn_state.push_pending_input(input.into()); + turn_state.push_pending_input(ResponseInputItem::from(input).into()); turn_state.accept_mailbox_delivery_for_current_turn(); Ok(active_turn_id.clone()) } @@ -4214,7 +4229,7 @@ impl Session { Some(at) => { let mut ts = at.turn_state.lock().await; for item in input { - ts.push_pending_input(item); + ts.push_pending_input(item.into()); } Ok(()) } @@ -4270,7 +4285,16 @@ impl Session { self.mailbox_rx.lock().await.has_pending_trigger_turn() } + #[cfg(test)] pub async fn prepend_pending_input(&self, input: Vec) -> Result<(), ()> { + self.prepend_pending_input_items(input.into_iter().map(Into::into).collect()) + .await + } + + pub(crate) async fn prepend_pending_input_items( + &self, + input: Vec, + ) -> Result<(), ()> { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { @@ -4282,7 +4306,16 @@ impl Session { } } + #[cfg(test)] pub async fn get_pending_input(&self) -> Vec { + self.take_pending_input_items() + .await + .into_iter() + .map(PendingInputItem::into_model_input) + .collect() + } + + pub(crate) async fn take_pending_input_items(&self) -> Vec { let (pending_input, accepts_mailbox_delivery) = { let mut active = self.active_turn.lock().await; match active.as_mut() { @@ -4305,6 +4338,7 @@ impl Session { .drain() .into_iter() .map(|mail| mail.to_response_input_item()) + .map(Into::into) .collect::>() }; if pending_input.is_empty() { @@ -4326,10 +4360,28 @@ impl Session { } let mut idle_pending_input = self.idle_pending_input.lock().await; - idle_pending_input.extend(items); + idle_pending_input.extend(items.into_iter().map(PendingInputItem::from)); } - pub(crate) async fn take_queued_response_items_for_next_turn(&self) -> Vec { + pub(crate) async fn queue_pending_input_for_next_turn(&self, items: Vec) { + if items.is_empty() { + return; + } + + let mut idle_pending_input = self.idle_pending_input.lock().await; + for item in items { + if let Some(timer_source) = item.timer_source() + && idle_pending_input + .iter() + .any(|queued| queued.timer_source() == Some(timer_source)) + { + continue; + } + idle_pending_input.push(item); + } + } + + pub(crate) async fn take_queued_pending_input_for_next_turn(&self) -> Vec { std::mem::take(&mut *self.idle_pending_input.lock().await) } @@ -6248,7 +6300,7 @@ pub(crate) async fn run_turn( // submitted through the UI while the model was running. Though the UI // may support this, the model might not. let pending_input = if can_drain_pending_input { - sess.get_pending_input().await + sess.take_pending_input_items().await } else { Vec::new() }; @@ -6269,7 +6321,9 @@ pub(crate) async fn run_turn( } => { let remaining_pending_input = pending_input_iter.collect::>(); if !remaining_pending_input.is_empty() { - let _ = sess.prepend_pending_input(remaining_pending_input).await; + let _ = sess + .prepend_pending_input_items(remaining_pending_input) + .await; requeued_pending_input = true; } blocked_pending_input_contexts = additional_contexts; @@ -7234,6 +7288,7 @@ fn realtime_text_for_event(msg: &EventMsg) -> Option { | EventMsg::TurnComplete(_) | EventMsg::TokenCount(_) | EventMsg::UserMessage(_) + | EventMsg::InjectedMessage(_) | EventMsg::AgentMessageDelta(_) | EventMsg::AgentReasoning(_) | EventMsg::AgentReasoningDelta(_) diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 0d620119f7..af1f5193b2 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -10,19 +10,20 @@ use codex_hooks::UserPromptSubmitOutcome; use codex_hooks::UserPromptSubmitRequest; use codex_protocol::items::TurnItem; use codex_protocol::models::DeveloperInstructions; -use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookRunSummary; use codex_protocol::protocol::HookStartedEvent; +use codex_protocol::protocol::InjectedMessageEvent; use codex_protocol::user_input::UserInput; use serde_json::Value; use crate::codex::Session; use crate::codex::TurnContext; use crate::event_mapping::parse_turn_item; +use crate::pending_input::PendingInputItem; pub(crate) struct HookRuntimeOutcome { pub should_stop: bool, @@ -38,6 +39,7 @@ pub(crate) enum PendingInputRecord { UserMessage { content: Vec, response_item: ResponseItem, + injected_event: Option, additional_contexts: Vec, }, ConversationItem { @@ -199,9 +201,10 @@ pub(crate) async fn run_user_prompt_submit_hooks( pub(crate) async fn inspect_pending_input( sess: &Arc, turn_context: &Arc, - pending_input_item: ResponseInputItem, + pending_input_item: PendingInputItem, ) -> PendingInputHookDisposition { - let response_item = ResponseItem::from(pending_input_item); + let (input_item, injected_event) = pending_input_item.into_parts(); + let response_item = ResponseItem::from(input_item); if let Some(TurnItem::UserMessage(user_message)) = parse_turn_item(&response_item) { let user_prompt_submit_outcome = run_user_prompt_submit_hooks(sess, turn_context, user_message.message()).await; @@ -213,6 +216,7 @@ pub(crate) async fn inspect_pending_input( PendingInputHookDisposition::Accepted(Box::new(PendingInputRecord::UserMessage { content: user_message.content, response_item, + injected_event, additional_contexts: user_prompt_submit_outcome.additional_contexts, })) } @@ -232,14 +236,24 @@ pub(crate) async fn record_pending_input( PendingInputRecord::UserMessage { content, response_item, + injected_event, additional_contexts, } => { - sess.record_user_prompt_and_emit_turn_item( - turn_context.as_ref(), - content.as_slice(), - response_item, - ) - .await; + if let Some(injected_event) = injected_event { + sess.record_generated_message_and_emit_display( + turn_context.as_ref(), + response_item, + injected_event, + ) + .await; + } else { + sess.record_user_prompt_and_emit_turn_item( + turn_context.as_ref(), + content.as_slice(), + response_item, + ) + .await; + } record_additional_contexts(sess, turn_context, additional_contexts).await; } PendingInputRecord::ConversationItem { response_item } => { diff --git a/codex-rs/core/src/injected_message.rs b/codex-rs/core/src/injected_message.rs new file mode 100644 index 0000000000..52f7d5f163 --- /dev/null +++ b/codex-rs/core/src/injected_message.rs @@ -0,0 +1,144 @@ +//! Generated user-message payloads and model-visible XML envelopes. +//! +//! This module owns the representation for messages that are delivered into a +//! thread by the harness rather than typed directly by the user. The model sees +//! the XML envelope recorded in history; transcript clients receive a separate +//! structured event with the human-facing text so they do not need to parse or +//! hide that XML. + +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseInputItem; +use codex_protocol::protocol::InjectedMessageEvent; + +const EXTERNAL_MESSAGE_OPEN_TAG: &str = ""; +const EXTERNAL_MESSAGE_CLOSE_TAG: &str = ""; +const TIMER_MESSAGE_OPEN_TAG: &str = ""; +const TIMER_MESSAGE_CLOSE_TAG: &str = ""; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum InjectedMessage { + External { + source: String, + content: String, + }, + Timer { + timer_id: String, + content: String, + instructions: Option, + }, +} + +impl InjectedMessage { + pub(crate) fn prompt_input_item(&self) -> ResponseInputItem { + ResponseInputItem::Message { + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: self.render_prompt(), + }], + } + } + + pub(crate) fn event(&self) -> InjectedMessageEvent { + match self { + Self::External { source, content } => InjectedMessageEvent { + content: content.clone(), + source: source.clone(), + }, + Self::Timer { + timer_id, content, .. + } => InjectedMessageEvent { + content: content.clone(), + source: format!("timer {timer_id}"), + }, + } + } + + fn render_prompt(&self) -> String { + let mut rendered = String::new(); + match self { + Self::External { content, .. } => { + rendered.push_str(EXTERNAL_MESSAGE_OPEN_TAG); + rendered.push('\n'); + push_block_tag(&mut rendered, "content", content); + rendered.push_str(EXTERNAL_MESSAGE_CLOSE_TAG); + } + Self::Timer { + timer_id, + content, + instructions, + } => { + rendered.push_str(TIMER_MESSAGE_OPEN_TAG); + rendered.push('\n'); + push_tag(&mut rendered, "timer_id", timer_id); + push_block_tag(&mut rendered, "content", content); + if let Some(instructions) = instructions.as_deref() { + push_block_tag(&mut rendered, "instructions", instructions); + } + rendered.push_str(TIMER_MESSAGE_CLOSE_TAG); + } + } + rendered + } +} + +fn push_tag(rendered: &mut String, tag: &str, value: &str) { + rendered.push('<'); + rendered.push_str(tag); + rendered.push('>'); + rendered.push_str(&xml_escape(value)); + rendered.push_str("\n"); +} + +fn push_block_tag(rendered: &mut String, tag: &str, value: &str) { + rendered.push('<'); + rendered.push_str(tag); + rendered.push_str(">\n"); + rendered.push_str(&xml_escape(value)); + rendered.push('\n'); + rendered.push_str("\n"); +} + +fn xml_escape(value: &str) -> String { + value + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + + #[test] + fn renders_external_message_prompt_with_only_content() { + let message = InjectedMessage::External { + source: "external".to_string(), + content: "run ".to_string(), + }; + + assert_eq!( + message.render_prompt(), + "\n\nrun <tests>\n\n" + ); + } + + #[test] + fn renders_timer_message_prompt_with_timer_id() { + let message = InjectedMessage::Timer { + timer_id: "timer-1".to_string(), + content: "run ".to_string(), + instructions: Some("stay \"brief\"".to_string()), + }; + + assert_eq!( + message.render_prompt(), + "\ntimer-1\n\nrun <tests>\n\n\nstay "brief"\n\n" + ); + } +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 5e62199d2d..63894ce588 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -39,6 +39,7 @@ mod flags; mod git_info_tests; mod guardian; mod hook_runtime; +pub mod injected_message; mod installation_id; pub(crate) mod instructions; pub(crate) mod landlock; @@ -62,6 +63,7 @@ mod mcp_tool_call; mod memories; pub(crate) mod mention_syntax; pub(crate) mod message_history; +pub(crate) mod pending_input; pub(crate) mod utils; pub use mention_syntax::PLUGIN_TEXT_MENTION_SIGIL; pub use mention_syntax::TOOL_MENTION_SIGIL; diff --git a/codex-rs/core/src/pending_input.rs b/codex-rs/core/src/pending_input.rs new file mode 100644 index 0000000000..a680ea5742 --- /dev/null +++ b/codex-rs/core/src/pending_input.rs @@ -0,0 +1,47 @@ +//! Pending input queued for delivery into a future or active regular turn. +//! +//! Most pending input is an ordinary model input item. Generated input carries +//! the model-visible item plus a separate display event so clients can render +//! the human-facing content without parsing the XML envelope recorded in model +//! history. + +use codex_protocol::models::ResponseInputItem; +use codex_protocol::protocol::InjectedMessageEvent; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct PendingInputItem { + item: ResponseInputItem, + injected_event: Option, +} + +impl PendingInputItem { + pub(crate) fn injected(item: ResponseInputItem, event: InjectedMessageEvent) -> Self { + Self { + item, + injected_event: Some(event), + } + } + + pub(crate) fn timer_source(&self) -> Option<&str> { + let source = self.injected_event.as_ref()?.source.as_str(); + source.starts_with("timer ").then_some(source) + } + + pub(crate) fn into_parts(self) -> (ResponseInputItem, Option) { + (self.item, self.injected_event) + } + + #[cfg(test)] + pub(crate) fn into_model_input(self) -> ResponseInputItem { + self.item + } +} + +impl From for PendingInputItem { + fn from(item: ResponseInputItem) -> Self { + Self { + item, + injected_event: None, + } + } +} diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 214fd8be15..d34a65286e 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -10,7 +10,6 @@ use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use codex_protocol::dynamic_tools::DynamicToolResponse; -use codex_protocol::models::ResponseInputItem; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_rmcp_client::ElicitationResponse; @@ -18,6 +17,7 @@ use rmcp::model::RequestId; use tokio::sync::oneshot; use crate::codex::TurnContext; +use crate::pending_input::PendingInputItem; use crate::tasks::AnySessionTask; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::ReviewDecision; @@ -101,7 +101,7 @@ pub(crate) struct TurnState { pending_user_input: HashMap>, pending_elicitations: HashMap<(String, RequestId), oneshot::Sender>, pending_dynamic_tools: HashMap>, - pending_input: Vec, + pending_input: Vec, mailbox_delivery_phase: MailboxDeliveryPhase, granted_permissions: Option, pub(crate) tool_calls: u64, @@ -197,11 +197,11 @@ impl TurnState { self.pending_dynamic_tools.remove(key) } - pub(crate) fn push_pending_input(&mut self, input: ResponseInputItem) { + pub(crate) fn push_pending_input(&mut self, input: PendingInputItem) { self.pending_input.push(input); } - pub(crate) fn prepend_pending_input(&mut self, mut input: Vec) { + pub(crate) fn prepend_pending_input(&mut self, mut input: Vec) { if input.is_empty() { return; } @@ -210,7 +210,7 @@ impl TurnState { self.pending_input = input; } - pub(crate) fn take_pending_input(&mut self) -> Vec { + pub(crate) fn take_pending_input(&mut self) -> Vec { if self.pending_input.is_empty() { Vec::with_capacity(0) } else { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index b46db8e98f..1dd1d73885 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -27,6 +27,7 @@ use crate::hook_runtime::PendingInputHookDisposition; use crate::hook_runtime::inspect_pending_input; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::record_pending_input; +use crate::pending_input::PendingInputItem; use crate::state::ActiveTurn; use crate::state::RunningTask; use crate::state::TaskKind; @@ -38,7 +39,6 @@ use codex_otel::TURN_NETWORK_PROXY_METRIC; use codex_otel::TURN_TOKEN_USAGE_METRIC; use codex_otel::TURN_TOOL_CALL_METRIC; use codex_protocol::models::ContentItem; -use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; @@ -255,8 +255,8 @@ impl Session { let cancellation_token = CancellationToken::new(); let done = Arc::new(Notify::new()); - let queued_response_items = self.take_queued_response_items_for_next_turn().await; - let mailbox_items = self.get_pending_input().await; + let queued_response_items = self.take_queued_pending_input_for_next_turn().await; + let mailbox_items = self.take_pending_input_items().await; let turn_state = { let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); @@ -405,7 +405,7 @@ impl Session { .turn_metadata_state .cancel_git_enrichment_task(); - let mut pending_input = Vec::::new(); + let mut pending_input = Vec::::new(); let mut should_clear_active_turn = false; let mut token_usage_at_turn_start = None; let mut turn_tool_calls = 0_u64; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 65d6dab04c..f8dac0fe02 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1412,6 +1412,9 @@ pub enum EventMsg { /// User/system input message (what was sent to the model) UserMessage(UserMessageEvent), + /// Transcript-safe display text for a timer-fired or external queued message. + InjectedMessage(InjectedMessageEvent), + /// Agent text output delta message AgentMessageDelta(AgentMessageDeltaEvent), @@ -2283,6 +2286,19 @@ pub struct UserMessageEvent { pub text_elements: Vec, } +/// Display-only event for a timer-fired or external queued message. +/// +/// The corresponding model-visible XML is recorded separately as a raw response +/// item. Clients should render this event's content instead of parsing or +/// hiding that XML themselves. +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS, PartialEq, Eq)] +pub struct InjectedMessageEvent { + /// Human-facing message text to show in transcript views. + pub content: String, + /// Origin label for the injected input, such as "external" or "timer ". + pub source: String, +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct AgentMessageDeltaEvent { pub delta: String, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index e7ef6a0606..d3626eed89 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -6735,6 +6735,17 @@ impl ChatWidget { self.on_user_message_event(ev); } } + EventMsg::InjectedMessage(ev) => { + if !ev.content.trim().is_empty() { + self.add_to_history(history_cell::new_user_prompt( + ev.content, + Vec::new(), + Vec::new(), + Vec::new(), + )); + self.needs_final_message_separator = false; + } + } EventMsg::EnteredReviewMode(review_request) => { self.on_entered_review_mode(review_request, from_replay) }