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

Route extension warnings to app-server threads (#35168)

Browse Source 
## 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
This commit is contained in:
felixxia-oai
2026-07-24 12:28:29 +00:00
committed by copyberry
parent c8957bbf0f
commit 5dd992acd3
15 changed files with 592 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/extensions.rs
+328 -7
View File
@@ -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<OutgoingMessageSender>,
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<OutgoingMessageSender>,
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(),
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+4
View File
@@ -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(
codex-rs/app-server/src/thread_state.rs
+4
View File
@@ -61,6 +61,10 @@ pub(crate) enum ThreadListenerCommand {
turn_id: Option<String>,
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.
codex-rs/app-server/tests/suite/v2/executor_skills.rs
+37 -2
View File
@@ -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::<WarningNotification, anyhow::Error>(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"),
codex-rs/core/tests/suite/skills_extension.rs
+24 -9
View File
@@ -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<Event>);
#[derive(Debug)]
enum CapturedExtensionEvent {
Event(Box<Event>),
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<CapturedExtensionEvent>);
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."
codex-rs/ext/extension-api/src/capabilities/events.rs
+19
View File
@@ -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<String>,
/// 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) {}
}
codex-rs/ext/extension-api/src/capabilities/mod.rs
+1
View File
@@ -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;
codex-rs/ext/extension-api/src/lib.rs
+1
View File
@@ -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;
codex-rs/ext/extension-api/tests/registry.rs
+14
View File
@@ -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()),
]
);
}
codex-rs/ext/goal/tests/goal_extension_backend.rs
+5
View File
@@ -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)]
codex-rs/ext/skills/src/extension.rs
+24 -16
View File
@@ -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::<ModelInfo>()
@@ -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<C> SkillsExtension<C> {
.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,
});
}
}
codex-rs/ext/skills/src/lib.rs
+1
View File
@@ -10,6 +10,7 @@ mod shadow_selection_experiment;
mod sources;
mod state;
mod tools;
mod warnings;
mod world_state;
pub use config::SkillsExtensionConfig;
codex-rs/ext/skills/src/tools/list.rs
+2 -15
View File
@@ -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<ToolCall> 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<ListedSkill> {
main_resource: entry.main_prompt.as_str().to_string(),
})
}
fn bounded_warnings(warnings: Vec<String>) -> Vec<String> {
warnings
.into_iter()
.take(MAX_WARNINGS)
.map(|warning| {
let (warning, _) = truncate_utf8_to_bytes(&warning, MAX_WARNING_BYTES);
warning
})
.collect()
}
codex-rs/ext/skills/src/warnings.rs
+12
View File
@@ -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<String> {
warnings
.iter()
.take(MAX_WARNINGS)
.map(|warning| truncate_utf8_to_bytes(warning, MAX_WARNING_BYTES).0)
.collect()
}
codex-rs/ext/skills/tests/skills_extension.rs
+116 -16
View File
@@ -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::<Vec<_>>();
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<Event>);
#[derive(Debug)]
enum CapturedExtensionEvent {
Event(Box<Event>),
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<CapturedExtensionEvent>);
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));
}
}