diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 5253ed41b7..6ef9c31d7f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -260,20 +260,13 @@ impl LocalAgentControl { &self, agent_id: ThreadId, state: &Arc, - mut communication: InterAgentCommunication, + communication: InterAgentCommunication, context: AgentCommunicationContext, start_options: TurnStartOptions, ) -> CodexResult { let observers = state.extensions.message_lifecycle_contributors(); - let delivery = (!observers.is_empty()).then(|| { - let message_id = communication - .id - .take() - .filter(|id| !id.is_empty()) - .unwrap_or_else(|| (self.runtime.item_id_generator)("amsg")); - communication.id = Some(message_id.clone()); - MessageDelivery::publish(context.sender_thread_id, message_id, observers) - }); + let delivery = (!observers.is_empty()) + .then(|| MessageDelivery::publish(context.sender_thread_id, observers)); let log_content = crate::agent_communication::send_log_content(&communication); { // Keep unloaded delivery atomic with publication of a reloaded session. diff --git a/codex-rs/core/src/agent_communication.rs b/codex-rs/core/src/agent_communication.rs index 0b5c8dbffa..141737c50d 100644 --- a/codex-rs/core/src/agent_communication.rs +++ b/codex-rs/core/src/agent_communication.rs @@ -4,7 +4,6 @@ use codex_extension_api::ExtensionData; use codex_extension_api::MessageLifecycleContributor; use codex_extension_api::MessageLifecycleInput; -use codex_protocol::ResponseItemId; use codex_protocol::ThreadId; use codex_protocol::protocol::InterAgentCommunication; use serde::Deserialize; @@ -156,13 +155,11 @@ pub(crate) struct MessageDelivery(Arc>>); impl MessageDelivery { pub(crate) fn publish( sender: ThreadId, - message_id: ResponseItemId, observers: &[Arc], ) -> Self { let input = MessageLifecycleInput { sender, - data: ExtensionData::new(message_id.to_string()), - message_id, + data: ExtensionData::new(uuid::Uuid::now_v7().to_string()), }; for observer in observers { observer.on_publish(&input); diff --git a/codex-rs/core/tests/suite/agent_mailbox.rs b/codex-rs/core/tests/suite/agent_mailbox.rs index a06ff09414..7fa6a3d646 100644 --- a/codex-rs/core/tests/suite/agent_mailbox.rs +++ b/codex-rs/core/tests/suite/agent_mailbox.rs @@ -7,7 +7,6 @@ use codex_core::config::Config; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::MessageLifecycleContributor; use codex_extension_api::MessageLifecycleInput; -use codex_protocol::ResponseItemId; use codex_protocol::ThreadId; use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::Op; @@ -20,15 +19,18 @@ use std::sync::mpsc; use tokio::sync::oneshot; #[derive(Default)] -struct Messages(Mutex>); +struct Messages(Mutex>); impl MessageLifecycleContributor for Messages { fn on_publish(&self, input: &MessageLifecycleInput) { let mut pending = self.0.lock().expect("messages lock"); - assert!(pending.insert(input.message_id.clone())); + let token = uuid::Uuid::new_v4(); + input.data.insert(token); + assert!(pending.insert(token)); } fn on_consumed(&self, input: &MessageLifecycleInput, _recipient: ThreadId) { let mut pending = self.0.lock().expect("messages lock"); - assert!(pending.remove(&input.message_id)); + let token = input.data.remove::().expect("published token"); + assert!(pending.remove(token.as_ref())); } } diff --git a/codex-rs/ext/extension-api/src/message_lifecycle.rs b/codex-rs/ext/extension-api/src/message_lifecycle.rs index 85d3e169e9..dbe844a4a6 100644 --- a/codex-rs/ext/extension-api/src/message_lifecycle.rs +++ b/codex-rs/ext/extension-api/src/message_lifecycle.rs @@ -1,12 +1,11 @@ //! Capture publication state once; consume it independently for each delivery. use crate::ExtensionData; -use codex_protocol::ResponseItemId; use codex_protocol::ThreadId; /// State shared by all deliveries of a publication, including recipients resolved later. +/// The model-visible item ID may still be unknown until the recipient records the message. pub struct MessageLifecycleInput { - pub message_id: ResponseItemId, pub sender: ThreadId, /// Capture contributor-specific state at publication; read it at consumption. pub data: ExtensionData,