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

codex: add injected message transcript events

Browse Source
This commit is contained in:
Eric Traut
2026-04-12 19:10:04 -07:00
parent 70b7a3700c
commit 2a40e791e8
11 changed files with 350 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+17
View File
@@ -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,
codex-rs/app-server/src/bespoke_event_handling.rs
+19
View File
@@ -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 {
codex-rs/core/src/codex.rs
+62 -7
View File
@@ -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<Option<ActiveTurn>>,
mailbox: Mailbox,
mailbox_rx: Mutex<MailboxReceiver>,
idle_pending_input: Mutex<Vec<ResponseInputItem>>, // TODO (jif) merge with mailbox!
idle_pending_input: Mutex<Vec<PendingInputItem>>, // TODO (jif) merge with mailbox!
pub(crate) guardian_review_session: GuardianReviewSessionManager,
pub(crate) services: SessionServices,
js_repl: Arc<JsReplHandle>,
@@ -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<ResponseInputItem>) -> 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<PendingInputItem>,
) -> 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<ResponseInputItem> {
self.take_pending_input_items()
.await
.into_iter()
.map(PendingInputItem::into_model_input)
.collect()
}
pub(crate) async fn take_pending_input_items(&self) -> Vec<PendingInputItem> {
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::<Vec<_>>()
};
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<ResponseInputItem> {
pub(crate) async fn queue_pending_input_for_next_turn(&self, items: Vec<PendingInputItem>) {
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<PendingInputItem> {
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::<Vec<_>>();
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<String> {
| EventMsg::TurnComplete(_)
| EventMsg::TokenCount(_)
| EventMsg::UserMessage(_)
| EventMsg::InjectedMessage(_)
| EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningDelta(_)
codex-rs/core/src/hook_runtime.rs
+23 -9
View File
@@ -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<UserInput>,
response_item: ResponseItem,
injected_event: Option<InjectedMessageEvent>,
additional_contexts: Vec<String>,
},
ConversationItem {
@@ -199,9 +201,10 @@ pub(crate) async fn run_user_prompt_submit_hooks(
pub(crate) async fn inspect_pending_input(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
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 } => {
codex-rs/core/src/injected_message.rs
+144
View File
@@ -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 = "<external_message>";
const EXTERNAL_MESSAGE_CLOSE_TAG: &str = "</external_message>";
const TIMER_MESSAGE_OPEN_TAG: &str = "<timer_message>";
const TIMER_MESSAGE_CLOSE_TAG: &str = "</timer_message>";
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum InjectedMessage {
External {
source: String,
content: String,
},
Timer {
timer_id: String,
content: String,
instructions: Option<String>,
},
}
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("</");
rendered.push_str(tag);
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("</");
rendered.push_str(tag);
rendered.push_str(">\n");
}
fn xml_escape(value: &str) -> String {
value
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
}
#[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 <tests>".to_string(),
};
assert_eq!(
message.render_prompt(),
"<external_message>\n<content>\nrun &lt;tests&gt;\n</content>\n</external_message>"
);
}
#[test]
fn renders_timer_message_prompt_with_timer_id() {
let message = InjectedMessage::Timer {
timer_id: "timer-1".to_string(),
content: "run <tests>".to_string(),
instructions: Some("stay \"brief\"".to_string()),
};
assert_eq!(
message.render_prompt(),
"<timer_message>\n<timer_id>timer-1</timer_id>\n<content>\nrun &lt;tests&gt;\n</content>\n<instructions>\nstay &quot;brief&quot;\n</instructions>\n</timer_message>"
);
}
}
codex-rs/core/src/lib.rs
+2
View File
@@ -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;
codex-rs/core/src/pending_input.rs
+47
View File
@@ -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<InjectedMessageEvent>,
}
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<InjectedMessageEvent>) {
(self.item, self.injected_event)
}
#[cfg(test)]
pub(crate) fn into_model_input(self) -> ResponseInputItem {
self.item
}
}
impl From<ResponseInputItem> for PendingInputItem {
fn from(item: ResponseInputItem) -> Self {
Self {
item,
injected_event: None,
}
}
}
codex-rs/core/src/state/turn.rs
+5 -5
View File
@@ -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<String, oneshot::Sender<RequestUserInputResponse>>,
pending_elicitations: HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>,
pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
pending_input: Vec<ResponseInputItem>,
pending_input: Vec<PendingInputItem>,
mailbox_delivery_phase: MailboxDeliveryPhase,
granted_permissions: Option<PermissionProfile>,
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<ResponseInputItem>) {
pub(crate) fn prepend_pending_input(&mut self, mut input: Vec<PendingInputItem>) {
if input.is_empty() {
return;
}
@@ -210,7 +210,7 @@ impl TurnState {
self.pending_input = input;
}
pub(crate) fn take_pending_input(&mut self) -> Vec<ResponseInputItem> {
pub(crate) fn take_pending_input(&mut self) -> Vec<PendingInputItem> {
if self.pending_input.is_empty() {
Vec::with_capacity(0)
} else {
codex-rs/core/src/tasks/mod.rs
+4 -4
View File
@@ -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::<ResponseInputItem>::new();
let mut pending_input = Vec::<PendingInputItem>::new();
let mut should_clear_active_turn = false;
let mut token_usage_at_turn_start = None;
let mut turn_tool_calls = 0_u64;
codex-rs/protocol/src/protocol.rs
+16
View File
@@ -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<crate::user_input::TextElement>,
}
/// 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 <id>".
pub source: String,
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct AgentMessageDeltaEvent {
pub delta: String,
codex-rs/tui/src/chatwidget.rs
+11
View File
@@ -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)
}