From 5dd992acd3f5242196ec690bb462e0d2687b485d Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Fri, 24 Jul 2026 12:22:39 +0000 Subject: [PATCH] Route extension warnings to app-server threads (#35168) ## Why Extension warnings need a stable thread target even when they are emitted outside an active turn. ## What changed - Add `ExtensionWarning` and a dedicated `ExtensionEventSink::emit_warning` path with an explicit thread ID and optional turn ID. - Deliver extension warnings as thread-scoped app-server `warning` notifications, preserving listener ordering and waiting briefly for a subscriber when no listener is active. - Move skills warnings to the new path, limiting provider catalog warnings to four messages of 256 bytes each. ## Testing - Cover warning routing, ordering, UTF-8-safe truncation, subscriber fallback, and invalid thread IDs. - Verify skills warning bounds and app-server delivery of executor catalog budget warnings. GitOrigin-RevId: 4dcb1ba7f17c0edb1a9d548f9568e0c734bffc15 --- codex-rs/app-server/src/extensions.rs | 335 +++++++++++++++++- .../request_processors/thread_lifecycle.rs | 4 + codex-rs/app-server/src/thread_state.rs | 4 + .../tests/suite/v2/executor_skills.rs | 39 +- codex-rs/core/tests/suite/skills_extension.rs | 33 +- .../extension-api/src/capabilities/events.rs | 19 + .../ext/extension-api/src/capabilities/mod.rs | 1 + codex-rs/ext/extension-api/src/lib.rs | 1 + codex-rs/ext/extension-api/tests/registry.rs | 14 + .../ext/goal/tests/goal_extension_backend.rs | 5 + codex-rs/ext/skills/src/extension.rs | 40 ++- codex-rs/ext/skills/src/lib.rs | 1 + codex-rs/ext/skills/src/tools/list.rs | 17 +- codex-rs/ext/skills/src/warnings.rs | 12 + codex-rs/ext/skills/tests/skills_extension.rs | 132 ++++++- 15 files changed, 592 insertions(+), 65 deletions(-) create mode 100644 codex-rs/ext/skills/src/warnings.rs diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index 10fa319f62..c394c8ad81 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -1,10 +1,12 @@ use std::sync::Arc; use std::sync::Weak; +use std::time::Duration; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadGoal; use codex_app_server_protocol::ThreadGoalUpdatedNotification; +use codex_app_server_protocol::WarningNotification; use codex_core::NewThread; use codex_core::StartThreadOptions; use codex_core::ThreadManager; @@ -15,6 +17,7 @@ use codex_extension_api::AgentSpawner; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistry; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_goal_extension::GoalService; use codex_http_client::HttpClientFactory; use codex_login::AuthManager; @@ -26,6 +29,7 @@ use codex_rollout::state_db::StateDbHandle; use codex_thread_store::ThreadStore; use crate::outgoing_message::OutgoingMessageSender; +use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::thread_state::ThreadListenerCommand; use crate::thread_state::ThreadStateManager; @@ -122,11 +126,36 @@ pub(crate) fn app_server_extension_event_sink( }) } +pub(crate) async fn send_thread_warning( + outgoing: &Arc, + thread_state_manager: &ThreadStateManager, + thread_id: ThreadId, + message: String, +) { + let subscribed_connection_ids = thread_state_manager + .subscribed_connection_ids(thread_id) + .await; + let thread_outgoing = ThreadScopedOutgoingMessageSender::new( + Arc::clone(outgoing), + subscribed_connection_ids, + thread_id, + ); + thread_outgoing + .send_server_notification(ServerNotification::Warning(WarningNotification { + thread_id: Some(thread_id.to_string()), + message, + })) + .await; +} + struct AppServerExtensionEventSink { outgoing: Arc, thread_state_manager: ThreadStateManager, } +const MAX_EXTENSION_WARNING_BYTES: usize = 256; +const EXTENSION_WARNING_SUBSCRIBER_TIMEOUT: Duration = Duration::from_secs(10); + impl ExtensionEventSink for AppServerExtensionEventSink { fn emit(&self, event: Event) { match event.msg { @@ -167,6 +196,62 @@ impl ExtensionEventSink for AppServerExtensionEventSink { } } } + + fn emit_warning(&self, warning: ExtensionWarning) { + let ExtensionWarning { + thread_id, + turn_id: _, + message, + } = warning; + let Ok(thread_id) = ThreadId::from_string(&thread_id) else { + tracing::warn!( + %thread_id, + "dropping extension warning with invalid thread id" + ); + return; + }; + let mut message = message; + if message.len() > MAX_EXTENSION_WARNING_BYTES { + let mut truncate_at = MAX_EXTENSION_WARNING_BYTES; + while !message.is_char_boundary(truncate_at) { + truncate_at -= 1; + } + message.truncate(truncate_at); + } + if let Some(listener_command_tx) = self + .thread_state_manager + .current_listener_command_tx(thread_id) + { + let command = ThreadListenerCommand::EmitWarning { + message: message.clone(), + }; + if listener_command_tx.send(command).is_ok() { + return; + } + tracing::warn!( + "failed to enqueue extension warning for {thread_id}: listener command channel is closed" + ); + } + let outgoing = Arc::clone(&self.outgoing); + let thread_state_manager = self.thread_state_manager.clone(); + tokio::spawn(async move { + if tokio::time::timeout( + EXTENSION_WARNING_SUBSCRIBER_TIMEOUT, + thread_state_manager.wait_for_thread_subscriber(thread_id), + ) + .await + .is_err() + { + tracing::warn!( + %thread_id, + timeout_secs = EXTENSION_WARNING_SUBSCRIBER_TIMEOUT.as_secs(), + "dropping extension warning after waiting for a thread subscriber" + ); + return; + } + send_thread_warning(&outgoing, &thread_state_manager, thread_id, message).await; + }); + } } pub(crate) fn guardian_agent_spawner( @@ -189,8 +274,6 @@ pub(crate) fn guardian_agent_spawner( #[cfg(test)] mod tests { - use std::time::Duration; - use codex_protocol::protocol::ThreadGoal as CoreThreadGoal; use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::ThreadGoalUpdatedEvent; @@ -198,10 +281,15 @@ mod tests { use tokio::sync::mpsc; use tokio::time::timeout; + use crate::outgoing_message::ConnectionId; + use crate::outgoing_message::OutgoingEnvelope; + use crate::outgoing_message::OutgoingMessage; + use crate::thread_state::ConnectionCapabilities; + use super::*; #[tokio::test] - async fn app_server_event_sink_uses_listener_fifo_for_goal_updates_and_clears() { + async fn app_server_event_sink_uses_listener_fifo_for_goal_updates_warnings_and_clears() { let (outgoing_tx, _outgoing_rx) = mpsc::channel(4); let outgoing = Arc::new(OutgoingMessageSender::new( outgoing_tx, @@ -213,15 +301,19 @@ mod tests { thread_state_manager.register_listener_command_tx(thread_id, listener_command_tx.clone()); let sink = app_server_extension_event_sink(outgoing, thread_state_manager); - for turn_id in ["turn-1", "turn-2"] { - sink.emit(thread_goal_updated_event(thread_id, turn_id)); - } + sink.emit(thread_goal_updated_event(thread_id, "turn-1")); + sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: Some("turn-warning".to_string()), + message: "catalog was shortened".to_string(), + }); + sink.emit(thread_goal_updated_event(thread_id, "turn-2")); listener_command_tx .send(ThreadListenerCommand::EmitThreadGoalCleared) .expect("listener command channel should be open"); let mut observed = Vec::new(); - for _ in 0..3 { + for _ in 0..4 { let command = timeout(Duration::from_secs(1), listener_command_rx.recv()) .await .expect("timed out waiting for listener command") @@ -230,6 +322,7 @@ mod tests { ThreadListenerCommand::EmitThreadGoalUpdated { turn_id, .. } => { observed.push(turn_id.expect("extension goal updates should include turn ids")); } + ThreadListenerCommand::EmitWarning { message } => observed.push(message), ThreadListenerCommand::EmitThreadGoalCleared => { observed.push("cleared".to_string()) } @@ -240,6 +333,7 @@ mod tests { assert_eq!( vec![ "turn-1".to_string(), + "catalog was shortened".to_string(), "turn-2".to_string(), "cleared".to_string() ], @@ -247,6 +341,233 @@ mod tests { ); } + #[tokio::test] + async fn app_server_event_sink_truncates_warning_before_listener_enqueue() { + let (outgoing_tx, _outgoing_rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let thread_state_manager = ThreadStateManager::new(); + let thread_id = ThreadId::default(); + let (listener_command_tx, mut listener_command_rx) = mpsc::unbounded_channel(); + thread_state_manager.register_listener_command_tx(thread_id, listener_command_tx); + let sink = app_server_extension_event_sink(outgoing, thread_state_manager); + + sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: Some("turn-warning".to_string()), + message: "🙂".repeat(65), + }); + + let command = timeout(Duration::from_secs(1), listener_command_rx.recv()) + .await + .expect("timed out waiting for listener command") + .expect("listener command channel closed unexpectedly"); + let ThreadListenerCommand::EmitWarning { message } = command else { + panic!("expected warning listener command"); + }; + assert_eq!(message, "🙂".repeat(64)); + } + + #[tokio::test] + async fn app_server_event_sink_targets_subscriber_without_listener() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let thread_id = ThreadId::new(); + let subscribed_connection = ConnectionId(1); + let unrelated_connection = ConnectionId(2); + let thread_state_manager = ThreadStateManager::new(); + for connection_id in [subscribed_connection, unrelated_connection] { + thread_state_manager + .connection_initialized(connection_id, ConnectionCapabilities::default()) + .await; + } + thread_state_manager + .try_ensure_connection_subscribed( + thread_id, + subscribed_connection, + /*experimental_raw_events*/ false, + ) + .await + .expect("connection should be subscribed"); + let sink = app_server_extension_event_sink(outgoing, thread_state_manager); + + sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".to_string()), + message: "catalog was shortened".to_string(), + }); + + let envelope = timeout(Duration::from_secs(1), outgoing_rx.recv()) + .await + .expect("timed out waiting for warning notification") + .expect("outgoing channel closed unexpectedly"); + let OutgoingEnvelope::ToConnection { + connection_id, + message, + write_complete_tx: _, + } = envelope + else { + panic!("expected connection-targeted warning notification"); + }; + assert_eq!(connection_id, subscribed_connection); + let OutgoingMessage::AppServerNotification(envelope) = message else { + panic!("expected app-server warning notification"); + }; + let ServerNotification::Warning(notification) = envelope.notification else { + panic!("expected warning notification"); + }; + assert_eq!( + notification, + WarningNotification { + thread_id: Some(thread_id.to_string()), + message: "catalog was shortened".to_string(), + } + ); + assert!(outgoing_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn app_server_event_sink_waits_for_subscriber_without_listener() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let thread_id = ThreadId::new(); + let subscribed_connection = ConnectionId(1); + let thread_state_manager = ThreadStateManager::new(); + thread_state_manager + .connection_initialized(subscribed_connection, ConnectionCapabilities::default()) + .await; + let sink = app_server_extension_event_sink(outgoing, thread_state_manager.clone()); + + sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".to_string()), + message: "catalog was shortened".to_string(), + }); + tokio::task::yield_now().await; + thread_state_manager + .try_ensure_connection_subscribed( + thread_id, + subscribed_connection, + /*experimental_raw_events*/ false, + ) + .await + .expect("connection should be subscribed"); + + let envelope = timeout(Duration::from_secs(1), outgoing_rx.recv()) + .await + .expect("timed out waiting for warning notification") + .expect("outgoing channel closed unexpectedly"); + let OutgoingEnvelope::ToConnection { + connection_id, + message, + write_complete_tx: _, + } = envelope + else { + panic!("expected connection-targeted warning notification"); + }; + assert_eq!(connection_id, subscribed_connection); + let OutgoingMessage::AppServerNotification(envelope) = message else { + panic!("expected app-server warning notification"); + }; + let ServerNotification::Warning(notification) = envelope.notification else { + panic!("expected warning notification"); + }; + assert_eq!( + notification, + WarningNotification { + thread_id: Some(thread_id.to_string()), + message: "catalog was shortened".to_string(), + } + ); + } + + #[tokio::test] + async fn app_server_event_sink_targets_subscriber_after_listener_closes() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let thread_id = ThreadId::new(); + let subscribed_connection = ConnectionId(1); + let thread_state_manager = ThreadStateManager::new(); + thread_state_manager + .connection_initialized(subscribed_connection, ConnectionCapabilities::default()) + .await; + thread_state_manager + .try_ensure_connection_subscribed( + thread_id, + subscribed_connection, + /*experimental_raw_events*/ false, + ) + .await + .expect("connection should be subscribed"); + let (listener_command_tx, listener_command_rx) = mpsc::unbounded_channel(); + drop(listener_command_rx); + thread_state_manager.register_listener_command_tx(thread_id, listener_command_tx); + let sink = app_server_extension_event_sink(outgoing, thread_state_manager); + + sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".to_string()), + message: "catalog was shortened".to_string(), + }); + + let envelope = timeout(Duration::from_secs(1), outgoing_rx.recv()) + .await + .expect("timed out waiting for warning notification") + .expect("outgoing channel closed unexpectedly"); + let OutgoingEnvelope::ToConnection { + connection_id, + message, + write_complete_tx: _, + } = envelope + else { + panic!("expected connection-targeted warning notification"); + }; + assert_eq!(connection_id, subscribed_connection); + let OutgoingMessage::AppServerNotification(envelope) = message else { + panic!("expected app-server warning notification"); + }; + let ServerNotification::Warning(notification) = envelope.notification else { + panic!("expected warning notification"); + }; + assert_eq!( + notification, + WarningNotification { + thread_id: Some(thread_id.to_string()), + message: "catalog was shortened".to_string(), + } + ); + assert!(outgoing_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn app_server_event_sink_drops_warning_with_invalid_thread_id() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let sink = app_server_extension_event_sink(outgoing, ThreadStateManager::new()); + + sink.emit_warning(ExtensionWarning { + thread_id: "not-a-thread-id".to_string(), + turn_id: Some("turn-1".to_string()), + message: "catalog was shortened".to_string(), + }); + + assert!(outgoing_rx.try_recv().is_err()); + } + fn thread_goal_updated_event(thread_id: ThreadId, turn_id: &str) -> Event { Event { id: turn_id.to_string(), diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index d83cf7edd4..84d66bb294 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -1,4 +1,5 @@ use super::*; +use crate::extensions::send_thread_warning; use codex_protocol::config_types::MultiAgentMode; pub(super) const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60); @@ -488,6 +489,9 @@ pub(super) async fn handle_thread_listener_command( )) .await; } + ThreadListenerCommand::EmitWarning { message } => { + send_thread_warning(outgoing, thread_state_manager, conversation_id, message).await; + } ThreadListenerCommand::EmitThreadGoalCleared => { outgoing .send_server_notification(ServerNotification::ThreadGoalCleared( diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index bb9b3ecabb..72dc7e15b9 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -61,6 +61,10 @@ pub(crate) enum ThreadListenerCommand { turn_id: Option, goal: ThreadGoal, }, + // EmitWarning is used to order extension warnings with other thread notifications. + EmitWarning { + message: String, + }, // EmitThreadGoalCleared is used to order app-server goal clears with running-thread resume responses. EmitThreadGoalCleared, // EmitThreadGoalSnapshot is used to read and emit the latest goal state in the listener order. diff --git a/codex-rs/app-server/tests/suite/v2/executor_skills.rs b/codex-rs/app-server/tests/suite/v2/executor_skills.rs index 3fe9e5b5d4..00039e5a46 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_skills.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_skills.rs @@ -11,9 +11,11 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::UserInput; +use codex_app_server_protocol::WarningNotification; use codex_utils_path_uri::PathUri; use core_test_support::responses; use core_test_support::skip_if_remote; +use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::time::timeout; @@ -23,7 +25,7 @@ const SKILL_MARKER: &str = "EXECUTOR_SKILL_BODY_MARKER"; const LOCAL_SKILL_MARKER: &str = "LOCAL_SKILL_BODY_MARKER"; #[tokio::test] -async fn selected_executor_root_exposes_plugin_skill() -> Result<()> { +async fn selected_executor_root_exposes_plugin_skill_and_forwards_budget_warning() -> Result<()> { // TODO(anp): Remove after selected capability-root fixtures can be materialized in remote exec. skip_if_remote!( Ok(()), @@ -87,6 +89,20 @@ stream_max_retries = 0 "---\nname: deploy\ndescription: Deploy through the executor.\n---\n\n# Deploy\n\n{SKILL_MARKER}\n" ), )?; + for index in 0..200 { + let skill_dir = plugin_dir + .path() + .join("skills") + .join(format!("skill-{index:03}")); + std::fs::create_dir_all(&skill_dir)?; + std::fs::write( + skill_dir.join("SKILL.md"), + format!( + "---\nname: skill-{index:03}\ndescription: {}\n---\n", + "x".repeat(1_025) + ), + )?; + } let mut app_server = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -113,10 +129,11 @@ stream_max_retries = 0 ) .await??; let ThreadStartResponse { thread, .. } = to_response(response)?; + let thread_id = thread.id; let request_id = app_server .send_turn_start_request(TurnStartParams { - thread_id: thread.id, + thread_id: thread_id.clone(), input: vec![UserInput::Text { text: format!("Use ${SKILL_NAME}"), text_elements: Vec::new(), @@ -129,6 +146,24 @@ stream_max_retries = 0 app_server.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; + let warning = timeout(READ_TIMEOUT, async { + loop { + let warning: WarningNotification = app_server.read_notification("warning").await?; + if warning + .message + .starts_with("Exceeded skills context budget.") + { + return Ok::(warning); + } + } + }) + .await??; + assert_eq!(warning.thread_id, Some(thread_id)); + assert!( + warning + .message + .starts_with("Exceeded skills context budget.") + ); timeout( READ_TIMEOUT, app_server.read_stream_until_notification_message("turn/completed"), diff --git a/codex-rs/core/tests/suite/skills_extension.rs b/codex-rs/core/tests/suite/skills_extension.rs index 2d37b2579a..9eb6583863 100644 --- a/codex-rs/core/tests/suite/skills_extension.rs +++ b/codex-rs/core/tests/suite/skills_extension.rs @@ -4,8 +4,8 @@ use anyhow::Result; use codex_core::config::Config; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; use codex_skills_extension::SkillProvider; use codex_skills_extension::SkillProviderSource; use codex_skills_extension::SkillProviders; @@ -37,11 +37,30 @@ struct StaticSkillProvider { catalog: SkillCatalog, } -struct ChannelEventSink(std::sync::mpsc::Sender); +#[derive(Debug)] +enum CapturedExtensionEvent { + Event(Box), + Warning(ExtensionWarning), +} + +impl CapturedExtensionEvent { + fn into_warning(self) -> ExtensionWarning { + match self { + Self::Warning(warning) => warning, + Self::Event(event) => panic!("expected extension warning, got {event:?}"), + } + } +} + +struct ChannelEventSink(std::sync::mpsc::Sender); impl ExtensionEventSink for ChannelEventSink { fn emit(&self, event: Event) { - let _ = self.0.send(event); + let _ = self.0.send(CapturedExtensionEvent::Event(Box::new(event))); + } + + fn emit_warning(&self, warning: ExtensionWarning) { + let _ = self.0.send(CapturedExtensionEvent::Warning(warning)); } } @@ -153,9 +172,7 @@ async fn production_turn_scales_extension_catalog_from_resolved_model_window() - .iter() .filter(|line| line.starts_with("- skill-")) .count(); - let EventMsg::Warning(warning) = event_rx.try_recv()?.msg else { - panic!("expected catalog budget warning"); - }; + let warning = event_rx.try_recv()?.into_warning(); let omitted_count = 400 - included_count; assert!(catalog_text.contains("additional skills omitted")); @@ -258,9 +275,7 @@ async fn production_turn_fairly_shortens_extension_catalog_descriptions() -> Res .all(|length| *length > 0 && *length < 1_024) ); assert!(!catalog_text.contains("additional skills omitted")); - let EventMsg::Warning(warning) = event_rx.try_recv()?.msg else { - panic!("expected catalog budget warning"); - }; + let warning = event_rx.try_recv()?.into_warning(); assert_eq!( warning.message, "Skill descriptions were shortened to fit the skills context budget. Codex can still see every skill, but some descriptions are shorter. Disable unused skills or plugins to leave more room for the rest." diff --git a/codex-rs/ext/extension-api/src/capabilities/events.rs b/codex-rs/ext/extension-api/src/capabilities/events.rs index d8fda1aea7..c19624a170 100644 --- a/codex-rs/ext/extension-api/src/capabilities/events.rs +++ b/codex-rs/ext/extension-api/src/capabilities/events.rs @@ -1,5 +1,16 @@ use codex_protocol::protocol::Event; +/// Extension warning with an explicit thread target and optional turn correlation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExtensionWarning { + /// Stable host-owned thread identifier used for delivery. + pub thread_id: String, + /// Stable host-owned turn identifier when the warning arose in a turn callback. + pub turn_id: Option, + /// Concise warning message for the user. + pub message: String, +} + /// Host-provided fire-and-forget sink for extension-generated events. /// /// Extensions construct protocol events with the correlation id appropriate for @@ -8,6 +19,12 @@ use codex_protocol::protocol::Event; pub trait ExtensionEventSink: Send + Sync { /// Queue one protocol event for host-owned delivery. fn emit(&self, event: Event); + + /// Queue one warning for host-owned delivery. + /// + /// Implementations must use [`ExtensionWarning::thread_id`] for routing. The optional + /// [`ExtensionWarning::turn_id`] is correlation metadata and does not identify a thread. + fn emit_warning(&self, warning: ExtensionWarning); } /// Event sink used when the host does not expose extension event emission. @@ -16,4 +33,6 @@ pub struct NoopExtensionEventSink; impl ExtensionEventSink for NoopExtensionEventSink { fn emit(&self, _event: Event) {} + + fn emit_warning(&self, _warning: ExtensionWarning) {} } diff --git a/codex-rs/ext/extension-api/src/capabilities/mod.rs b/codex-rs/ext/extension-api/src/capabilities/mod.rs index 37c36e573b..39340c255c 100644 --- a/codex-rs/ext/extension-api/src/capabilities/mod.rs +++ b/codex-rs/ext/extension-api/src/capabilities/mod.rs @@ -5,6 +5,7 @@ mod response_items; pub use agent::AgentSpawnFuture; pub use agent::AgentSpawner; pub use events::ExtensionEventSink; +pub use events::ExtensionWarning; pub use events::NoopExtensionEventSink; pub use response_items::NoopResponseItemInjector; pub use response_items::ResponseItemInjectionFuture; diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index 0c7ddce86d..cdfb0b4cbd 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -7,6 +7,7 @@ mod user_instructions; pub use capabilities::AgentSpawnFuture; pub use capabilities::AgentSpawner; pub use capabilities::ExtensionEventSink; +pub use capabilities::ExtensionWarning; pub use capabilities::NoopExtensionEventSink; pub use capabilities::NoopResponseItemInjector; pub use capabilities::ResponseItemInjectionFuture; diff --git a/codex-rs/ext/extension-api/tests/registry.rs b/codex-rs/ext/extension-api/tests/registry.rs index de9cdf1539..31a8805b78 100644 --- a/codex-rs/ext/extension-api/tests/registry.rs +++ b/codex-rs/ext/extension-api/tests/registry.rs @@ -11,6 +11,7 @@ use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_extension_api::PromptFragment; use codex_extension_api::PromptSlot; use codex_extension_api::SkillInvocationContributor; @@ -370,6 +371,13 @@ impl ExtensionEventSink for RecordingEventSink { .expect("recording event sink lock should not be poisoned") .push((event.id, warning.message)); } + + fn emit_warning(&self, warning: ExtensionWarning) { + self.events + .lock() + .expect("recording event sink lock should not be poisoned") + .push((warning.thread_id, warning.message)); + } } #[test] @@ -384,6 +392,11 @@ fn custom_event_sink_survives_registry_build() { registry .event_sink() .emit(warning_event("registry", "after")); + registry.event_sink().emit_warning(ExtensionWarning { + thread_id: "thread".to_string(), + turn_id: Some("turn".to_string()), + message: "warning".to_string(), + }); assert_eq!( sink.events @@ -393,6 +406,7 @@ fn custom_event_sink_survives_registry_build() { [ ("builder".to_string(), "before".to_string()), ("registry".to_string(), "after".to_string()), + ("thread".to_string(), "warning".to_string()), ] ); } diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index eb751321df..de1615bb64 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -12,6 +12,7 @@ use codex_analytics::AnalyticsEventsClient; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_extension_api::FunctionCallError; use codex_extension_api::NoopTurnItemEmitter; use codex_extension_api::ThreadResumeInput; @@ -1420,6 +1421,10 @@ impl ExtensionEventSink for RecordingEventSink { fn emit(&self, event: Event) { self.events().push(event); } + + fn emit_warning(&self, _warning: ExtensionWarning) { + panic!("goal extension tests do not emit warnings"); + } } #[derive(Debug, PartialEq, Eq)] diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index c4700fc66d..230e7563e0 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -11,6 +11,7 @@ use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_extension_api::PromptFragment; use codex_extension_api::SkillInvocationContributor; use codex_extension_api::SkillInvocationInput; @@ -27,9 +28,6 @@ use codex_extension_api::WorldStateSectionContribution; use codex_mcp::McpResourceClient; use codex_otel::MetricsClient; use codex_protocol::openai_models::ModelInfo; -use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::WarningEvent; use crate::SkillsExtensionConfig; use crate::catalog::SkillCatalog; @@ -58,6 +56,7 @@ use crate::state::SkillsSessionState; use crate::state::SkillsThreadState; use crate::state::SkillsTurnState; use crate::tools::skill_tools; +use crate::warnings::bounded_warnings; use crate::world_state::executor_skills_world_state_section; use crate::world_state::host_skills_world_state_section; @@ -170,8 +169,8 @@ where &thread_state, ) .await; - for warning in &catalog.warnings { - self.emit_warning(thread_store.level_id(), warning.clone()); + for warning in bounded_warnings(&catalog.warnings) { + self.emit_warning(thread_store.level_id(), /*turn_id*/ None, warning); } let include_usage = thread_store .get::() @@ -183,7 +182,7 @@ where capped_skill_metadata_budget(/*context_window*/ None), ); if let Some(message) = rendered.warning_message { - self.emit_warning(thread_store.level_id(), message); + self.emit_warning(thread_store.level_id(), /*turn_id*/ None, message); } rendered .fragment @@ -247,7 +246,7 @@ where .get_or_init(EmittedCatalogBudgetWarnings::default) .insert(&message) { - self.emit_warning(input.turn_id, message); + self.emit_warning(input.thread_store.level_id(), Some(input.turn_id), message); } let executor_body = rendered.fragment.map(|fragment| fragment.body()); let mut sections = vec![executor_skills_world_state_section( @@ -364,8 +363,8 @@ where .map(|executor_skills| executor_skills.0.clone()) .unwrap_or_default(); catalog.extend(self.list_skills(query, &thread_state).await); - for warning in &catalog.warnings { - self.emit_warning(&input.turn_id, warning.clone()); + for warning in bounded_warnings(&catalog.warnings) { + self.emit_warning(thread_store.level_id(), Some(&input.turn_id), warning); } let selected_entries = collect_explicit_skill_mentions(&input.user_input, &catalog); @@ -407,7 +406,7 @@ where metadata_budget, ); if let Some(message) = rendered.warning_message { - self.emit_warning(&input.turn_id, message); + self.emit_warning(thread_store.level_id(), Some(&input.turn_id), message); } if let Some(fragment) = rendered.fragment { fragments.push(Box::new(fragment)); @@ -435,7 +434,11 @@ where "Skill `{}` exceeded the main prompt context limit and was truncated.", entry.name ); - self.emit_warning(&input.turn_id, warning.clone()); + self.emit_warning( + thread_store.level_id(), + Some(&input.turn_id), + warning.clone(), + ); warnings.push(warning); } let fragment = SkillInstructions { @@ -455,7 +458,11 @@ where } Err(message) => { let warning = format!("Failed to load skill `{}`: {message}", entry.name); - self.emit_warning(&input.turn_id, warning.clone()); + self.emit_warning( + thread_store.level_id(), + Some(&input.turn_id), + warning.clone(), + ); warnings.push(warning); } } @@ -542,10 +549,11 @@ impl SkillsExtension { .map_err(|err| err.message) } - fn emit_warning(&self, turn_id: &str, message: String) { - self.event_sink.emit(Event { - id: turn_id.to_string(), - msg: EventMsg::Warning(WarningEvent { message }), + fn emit_warning(&self, thread_id: &str, turn_id: Option<&str>, message: String) { + self.event_sink.emit_warning(ExtensionWarning { + thread_id: thread_id.to_string(), + turn_id: turn_id.map(str::to_string), + message, }); } } diff --git a/codex-rs/ext/skills/src/lib.rs b/codex-rs/ext/skills/src/lib.rs index a01ff2bfdc..3e129db169 100644 --- a/codex-rs/ext/skills/src/lib.rs +++ b/codex-rs/ext/skills/src/lib.rs @@ -10,6 +10,7 @@ mod shadow_selection_experiment; mod sources; mod state; mod tools; +mod warnings; mod world_state; pub use config::SkillsExtensionConfig; diff --git a/codex-rs/ext/skills/src/tools/list.rs b/codex-rs/ext/skills/src/tools/list.rs index e556f9783f..b9bf6d90cd 100644 --- a/codex-rs/ext/skills/src/tools/list.rs +++ b/codex-rs/ext/skills/src/tools/list.rs @@ -9,7 +9,7 @@ use serde::Serialize; use crate::catalog::SkillCatalogEntry; use crate::render::truncate_catalog_skill_description; -use crate::render::truncate_utf8_to_bytes; +use crate::warnings::bounded_warnings; use super::MAX_HANDLE_BYTES; use super::SkillToolAuthority; @@ -21,8 +21,6 @@ use super::skill_function_tool; use super::skill_tool_name; const TOOL_NAME: &str = "list"; -const MAX_WARNINGS: usize = 4; -const MAX_WARNING_BYTES: usize = 256; #[derive(Deserialize, JsonSchema)] #[serde(deny_unknown_fields)] @@ -76,7 +74,7 @@ impl ToolExecutor for ListTool { .filter(|entry| entry.enabled && entry.authority == authority) .filter_map(listed_skill) .collect(), - warnings: bounded_warnings(catalog.warnings), + warnings: bounded_warnings(&catalog.warnings), }; external_json_output(&response) @@ -100,14 +98,3 @@ fn listed_skill(entry: SkillCatalogEntry) -> Option { main_resource: entry.main_prompt.as_str().to_string(), }) } - -fn bounded_warnings(warnings: Vec) -> Vec { - warnings - .into_iter() - .take(MAX_WARNINGS) - .map(|warning| { - let (warning, _) = truncate_utf8_to_bytes(&warning, MAX_WARNING_BYTES); - warning - }) - .collect() -} diff --git a/codex-rs/ext/skills/src/warnings.rs b/codex-rs/ext/skills/src/warnings.rs new file mode 100644 index 0000000000..68c3cbf1ff --- /dev/null +++ b/codex-rs/ext/skills/src/warnings.rs @@ -0,0 +1,12 @@ +use crate::render::truncate_utf8_to_bytes; + +const MAX_WARNINGS: usize = 4; +const MAX_WARNING_BYTES: usize = 256; + +pub(crate) fn bounded_warnings(warnings: &[String]) -> Vec { + warnings + .iter() + .take(MAX_WARNINGS) + .map(|warning| truncate_utf8_to_bytes(warning, MAX_WARNING_BYTES).0) + .collect() +} diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index 189e4bf046..d0a0afd61d 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -12,6 +12,7 @@ use codex_extension_api::ConversationHistory; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; use codex_extension_api::NoopTurnItemEmitter; use codex_extension_api::PreviousWorldStateSection; use codex_extension_api::ThreadStartInput; @@ -23,7 +24,6 @@ use codex_models_manager::model_info::model_info_from_slug; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::SKILLS_INSTRUCTIONS_CLOSE_TAG; use codex_protocol::protocol::SKILLS_INSTRUCTIONS_OPEN_TAG; use codex_protocol::protocol::SessionSource; @@ -540,11 +540,9 @@ async fn extreme_budget_pressure_removes_descriptions_before_omitting_entries() assert!(!rendered.contains("description-")); assert!(rendered.contains("additional skills omitted from this bounded skills list")); let omitted_count = 200 - included_count; - let event = event_rx.try_recv()?; - assert_eq!("thread", event.id); - let EventMsg::Warning(warning) = event.msg else { - panic!("expected catalog budget warning"); - }; + let warning = event_rx.try_recv()?.into_warning(); + assert_eq!(warning.thread_id, "thread"); + assert_eq!(warning.turn_id, None); assert_eq!( warning.message, format!( @@ -673,9 +671,9 @@ async fn orchestrator_catalog_snapshot_caches_failure() -> TestResult { .contribute_thread_context(&session_store, &thread_store) .await; assert!(initial_fragments.is_empty()); - let EventMsg::Warning(warning) = event_rx.try_recv()?.msg else { - panic!("expected warning event"); - }; + let warning = event_rx.try_recv()?.into_warning(); + assert_eq!(warning.thread_id, thread_store.level_id()); + assert_eq!(warning.turn_id, None); assert_eq!( warning.message, "orchestrator skills unavailable: temporary orchestrator failure" @@ -698,6 +696,13 @@ async fn orchestrator_catalog_snapshot_caches_failure() -> TestResult { ) .await; assert!(fragments.is_empty()); + let warning = event_rx.try_recv()?.into_warning(); + assert_eq!(warning.thread_id, thread_store.level_id()); + assert_eq!(warning.turn_id.as_deref(), Some(turn_id)); + assert_eq!( + warning.message, + "orchestrator skills unavailable: temporary orchestrator failure" + ); } assert_eq!(1, list_calls.load(Ordering::Relaxed)); @@ -916,11 +921,9 @@ async fn model_context_window_scales_executor_catalog_but_not_thread_catalog() - turn_store: &turn_store, }) .await; - let event = event_rx.try_recv()?; - assert_eq!("turn-1", event.id); - let EventMsg::Warning(warning) = event.msg else { - panic!("expected catalog budget warning"); - }; + let warning = event_rx.try_recv()?.into_warning(); + assert_eq!(warning.thread_id, thread_store.level_id()); + assert_eq!(warning.turn_id.as_deref(), Some("turn-1")); assert!( warning .message @@ -947,6 +950,84 @@ async fn model_context_window_scales_executor_catalog_but_not_thread_catalog() - Ok(()) } +#[tokio::test] +async fn executor_catalog_emits_at_most_four_warnings() -> TestResult { + let executor_provider = Arc::new(StaticSkillProvider { + catalog: SkillCatalog { + entries: Vec::new(), + warnings: (0..6).map(|index| format!("warning-{index}")).collect(), + }, + read_requests: Arc::new(Mutex::new(Vec::new())), + list_calls: None, + fail_first_list: false, + }); + let providers = SkillProviders::new().with_executor_provider(executor_provider); + let (event_tx, event_rx) = std::sync::mpsc::channel(); + let mut builder = + ExtensionRegistryBuilder::with_event_sink(Arc::new(ChannelEventSink(event_tx))); + install_with_providers(&mut builder, providers, skills_extension_config); + let registry = builder.build(); + let session_store = ExtensionData::new("session"); + let thread_store = ExtensionData::new("thread"); + let mut config = default_config(); + config.bundled_skills_enabled = false; + registry.thread_lifecycle_contributors()[0] + .on_thread_start(ThreadStartInput { + config: &config, + session_source: &SessionSource::Cli, + persistent_thread_state_available: true, + environments: &[], + mcp_resource_client: None, + session_store: &session_store, + thread_store: &thread_store, + }) + .await; + let selected_roots = vec![SelectedCapabilityRoot { + id: "skills".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: "env-1".to_string(), + path: PathUri::parse("file:///skills").expect("skill root URI"), + }, + }]; + let turn_store = ExtensionData::new("turn-1"); + + registry.context_contributors()[0] + .contribute_world_state(WorldStateContributionInput { + thread_id: codex_protocol::ThreadId::new(), + turn_id: "turn-1", + environments: &[], + ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, + session_store: &session_store, + thread_store: &thread_store, + turn_store: &turn_store, + }) + .await; + registry.turn_input_contributors()[0] + .contribute( + TurnInputContext { + turn_id: "turn-1".to_string(), + user_input: Vec::new(), + environments: Vec::new(), + }, + &session_store, + &thread_store, + &turn_store, + ) + .await; + + let messages = event_rx + .try_iter() + .map(|event| event.into_warning().message) + .collect::>(); + assert_eq!( + messages, + vec!["warning-0", "warning-1", "warning-2", "warning-3"] + ); + + Ok(()) +} + #[tokio::test] async fn prompt_hidden_skill_can_still_be_invoked() -> TestResult { let read_requests = Arc::new(Mutex::new(Vec::new())); @@ -1033,11 +1114,30 @@ struct StaticSkillProvider { fail_first_list: bool, } -struct ChannelEventSink(std::sync::mpsc::Sender); +#[derive(Debug)] +enum CapturedExtensionEvent { + Event(Box), + Warning(ExtensionWarning), +} + +impl CapturedExtensionEvent { + fn into_warning(self) -> ExtensionWarning { + match self { + Self::Warning(warning) => warning, + Self::Event(event) => panic!("expected extension warning, got {event:?}"), + } + } +} + +struct ChannelEventSink(std::sync::mpsc::Sender); impl ExtensionEventSink for ChannelEventSink { fn emit(&self, event: Event) { - let _ = self.0.send(event); + let _ = self.0.send(CapturedExtensionEvent::Event(Box::new(event))); + } + + fn emit_warning(&self, warning: ExtensionWarning) { + let _ = self.0.send(CapturedExtensionEvent::Warning(warning)); } }