diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 84fd8014b..4e7cc8827 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -813,7 +813,7 @@ "type": "boolean" }, "optOutNotificationMethods": { - "description": "Exact notification method names that should be suppressed for this connection (for example `codex/event/session_configured`).", + "description": "Exact notification method names that should be suppressed for this connection (for example `thread/started`).", "items": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 421423bbc..6b4729ace 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -5423,7 +5423,7 @@ "type": "boolean" }, "optOutNotificationMethods": { - "description": "Exact notification method names that should be suppressed for this connection (for example `codex/event/session_configured`).", + "description": "Exact notification method names that should be suppressed for this connection (for example `thread/started`).", "items": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index f7a0dbb47..793ed16a8 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -7705,7 +7705,7 @@ "type": "boolean" }, "optOutNotificationMethods": { - "description": "Exact notification method names that should be suppressed for this connection (for example `codex/event/session_configured`).", + "description": "Exact notification method names that should be suppressed for this connection (for example `thread/started`).", "items": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v1/InitializeParams.json b/codex-rs/app-server-protocol/schema/json/v1/InitializeParams.json index 050bcb9c5..6048b8224 100644 --- a/codex-rs/app-server-protocol/schema/json/v1/InitializeParams.json +++ b/codex-rs/app-server-protocol/schema/json/v1/InitializeParams.json @@ -31,7 +31,7 @@ "type": "boolean" }, "optOutNotificationMethods": { - "description": "Exact notification method names that should be suppressed for this connection (for example `codex/event/session_configured`).", + "description": "Exact notification method names that should be suppressed for this connection (for example `thread/started`).", "items": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/typescript/InitializeCapabilities.ts b/codex-rs/app-server-protocol/schema/typescript/InitializeCapabilities.ts index a6ac24efc..125b4b1f1 100644 --- a/codex-rs/app-server-protocol/schema/typescript/InitializeCapabilities.ts +++ b/codex-rs/app-server-protocol/schema/typescript/InitializeCapabilities.ts @@ -12,6 +12,6 @@ export type InitializeCapabilities = { experimentalApi: boolean, /** * Exact notification method names that should be suppressed for this - * connection (for example `codex/event/session_configured`). + * connection (for example `thread/started`). */ optOutNotificationMethods?: Array | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index bb3c486ee..c7fa75c10 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -955,7 +955,7 @@ mod tests { capabilities: Some(v1::InitializeCapabilities { experimental_api: true, opt_out_notification_methods: Some(vec![ - "codex/event/session_configured".to_string(), + "thread/started".to_string(), "item/agentMessage/delta".to_string(), ]), }), @@ -975,7 +975,7 @@ mod tests { "capabilities": { "experimentalApi": true, "optOutNotificationMethods": [ - "codex/event/session_configured", + "thread/started", "item/agentMessage/delta" ] } @@ -1000,7 +1000,7 @@ mod tests { "capabilities": { "experimentalApi": true, "optOutNotificationMethods": [ - "codex/event/session_configured", + "thread/started", "item/agentMessage/delta" ] } @@ -1020,7 +1020,7 @@ mod tests { capabilities: Some(v1::InitializeCapabilities { experimental_api: true, opt_out_notification_methods: Some(vec![ - "codex/event/session_configured".to_string(), + "thread/started".to_string(), "item/agentMessage/delta".to_string(), ]), }), diff --git a/codex-rs/app-server-protocol/src/protocol/v1.rs b/codex-rs/app-server-protocol/src/protocol/v1.rs index b46da59ad..903845824 100644 --- a/codex-rs/app-server-protocol/src/protocol/v1.rs +++ b/codex-rs/app-server-protocol/src/protocol/v1.rs @@ -54,7 +54,7 @@ pub struct InitializeCapabilities { #[serde(default)] pub experimental_api: bool, /// Exact notification method names that should be suppressed for this - /// connection (for example `codex/event/session_configured`). + /// connection (for example `thread/started`). #[ts(optional = nullable)] pub opt_out_notification_methods: Option>, } diff --git a/codex-rs/app-server-test-client/src/lib.rs b/codex-rs/app-server-test-client/src/lib.rs index 14ca0cff5..3547bc5f1 100644 --- a/codex-rs/app-server-test-client/src/lib.rs +++ b/codex-rs/app-server-test-client/src/lib.rs @@ -88,20 +88,6 @@ use url::Url; use uuid::Uuid; const NOTIFICATIONS_TO_OPT_OUT: &[&str] = &[ - // Legacy codex/event (v1-style) deltas. - "codex/event/agent_message_content_delta", - "codex/event/agent_message_delta", - "codex/event/agent_reasoning_delta", - "codex/event/reasoning_content_delta", - "codex/event/reasoning_raw_content_delta", - "codex/event/exec_command_output_delta", - // Other legacy events. - "codex/event/exec_approval_request", - "codex/event/exec_command_begin", - "codex/event/exec_command_end", - "codex/event/exec_output", - "codex/event/item_started", - "codex/event/item_completed", // v2 item deltas. "command/exec/outputDelta", "item/agentMessage/delta", diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 1fec3a35d..8d545ddda 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -115,7 +115,7 @@ Example with notification opt-out: "capabilities": { "experimentalApi": true, "optOutNotificationMethods": [ - "codex/event/session_configured", + "thread/started", "item/agentMessage/delta" ] } @@ -722,12 +722,12 @@ Clients can suppress specific notifications per connection by sending exact meth - Exact-match only: `item/agentMessage/delta` suppresses only that method. - Unknown method names are ignored. -- Applies to both legacy (`codex/event/*`) and v2 (`thread/*`, `turn/*`, `item/*`, etc.) notifications. +- Applies to app-server typed notifications such as `thread/*`, `turn/*`, `item/*`, and `rawResponseItem/*`. - Does not apply to requests/responses/errors. Examples: -- Opt out of legacy session setup event: `codex/event/session_configured` +- Opt out of thread lifecycle notifications: `thread/started` - Opt out of streamed agent text deltas: `item/agentMessage/delta` ### Fuzzy file search events (experimental) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index a4909c774..0ab906620 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6506,9 +6506,17 @@ impl CodexMessageProcessor { }; // For now, we send a notification for every event, - // JSON-serializing the `Event` as-is, but these should - // be migrated to be variants of `ServerNotification` - // instead. + // Legacy `codex/event/*` notifications are still + // produced here because the in-process app-server lane + // (`codex exec` and other in-process consumers) still + // depends on them. External transports now drop + // `OutgoingMessage::Notification` in `transport.rs`, + // so stdio/websocket clients only observe the typed + // `ServerNotification` translations emitted below. + // + // TODO: remove this raw legacy-notification emission + // entirely once the remaining in-process consumers are + // migrated off `codex/event/*`. let event_formatted = match &event.msg { EventMsg::TurnStarted(_) => "task_started", EventMsg::TurnComplete(_) => "task_complete", diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 4e3572ee5..c31236f1a 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -384,6 +384,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { Arc::clone(&outbound_initialized), Arc::clone(&outbound_experimental_api_enabled), Arc::clone(&outbound_opted_out_notification_methods), + true, None, ), ); diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 887c2c650..5616dcef3 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -103,6 +103,8 @@ enum OutboundControlEvent { Opened { connection_id: ConnectionId, writer: mpsc::Sender, + // Allow codex/event/* notifications to be emitted. + allow_legacy_notifications: bool, disconnect_sender: Option, initialized: Arc, experimental_api_enabled: Arc, @@ -541,6 +543,7 @@ pub async fn run_main_with_transport( OutboundControlEvent::Opened { connection_id, writer, + allow_legacy_notifications, disconnect_sender, initialized, experimental_api_enabled, @@ -553,6 +556,7 @@ pub async fn run_main_with_transport( initialized, experimental_api_enabled, opted_out_notification_methods, + allow_legacy_notifications, disconnect_sender, ), ); @@ -650,6 +654,7 @@ pub async fn run_main_with_transport( TransportEvent::ConnectionOpened { connection_id, writer, + allow_legacy_notifications, disconnect_sender, } => { let outbound_initialized = Arc::new(AtomicBool::new(false)); @@ -661,6 +666,7 @@ pub async fn run_main_with_transport( .send(OutboundControlEvent::Opened { connection_id, writer, + allow_legacy_notifications, disconnect_sender, initialized: Arc::clone(&outbound_initialized), experimental_api_enabled: Arc::clone( diff --git a/codex-rs/app-server/src/transport.rs b/codex-rs/app-server/src/transport.rs index f537d8ba5..d0aa75335 100644 --- a/codex-rs/app-server/src/transport.rs +++ b/codex-rs/app-server/src/transport.rs @@ -166,6 +166,7 @@ pub(crate) enum TransportEvent { ConnectionOpened { connection_id: ConnectionId, writer: mpsc::Sender, + allow_legacy_notifications: bool, disconnect_sender: Option, }, ConnectionClosed { @@ -203,6 +204,7 @@ pub(crate) struct OutboundConnectionState { pub(crate) initialized: Arc, pub(crate) experimental_api_enabled: Arc, pub(crate) opted_out_notification_methods: Arc>>, + pub(crate) allow_legacy_notifications: bool, pub(crate) writer: mpsc::Sender, disconnect_sender: Option, } @@ -213,12 +215,14 @@ impl OutboundConnectionState { initialized: Arc, experimental_api_enabled: Arc, opted_out_notification_methods: Arc>>, + allow_legacy_notifications: bool, disconnect_sender: Option, ) -> Self { Self { initialized, experimental_api_enabled, opted_out_notification_methods, + allow_legacy_notifications, writer, disconnect_sender, } @@ -246,6 +250,7 @@ pub(crate) async fn start_stdio_connection( .send(TransportEvent::ConnectionOpened { connection_id, writer: writer_tx, + allow_legacy_notifications: false, disconnect_sender: None, }) .await @@ -348,6 +353,7 @@ async fn run_websocket_connection( .send(TransportEvent::ConnectionOpened { connection_id, writer: writer_tx, + allow_legacy_notifications: false, disconnect_sender: Some(disconnect_token.clone()), }) .await @@ -555,6 +561,16 @@ fn should_skip_notification_for_connection( connection_state: &OutboundConnectionState, message: &OutgoingMessage, ) -> bool { + if !connection_state.allow_legacy_notifications + && matches!(message, OutgoingMessage::Notification(_)) + { + // Raw legacy `codex/event/*` notifications are still emitted upstream + // for in-process compatibility, but they are no longer part of the + // external app-server contract. Keep dropping them here until the + // producer path can be deleted entirely. + return true; + } + let Ok(opted_out_notification_methods) = connection_state.opted_out_notification_methods.read() else { warn!("failed to read outbound opted-out notifications"); @@ -931,6 +947,7 @@ mod tests { initialized, Arc::new(AtomicBool::new(true)), opted_out_notification_methods, + false, None, ), ); @@ -955,6 +972,89 @@ mod tests { ); } + #[tokio::test] + async fn to_connection_legacy_notifications_are_dropped_for_external_clients() { + let connection_id = ConnectionId(10); + let (writer_tx, mut writer_rx) = mpsc::channel(1); + + let mut connections = HashMap::new(); + connections.insert( + connection_id, + OutboundConnectionState::new( + writer_tx, + Arc::new(AtomicBool::new(true)), + Arc::new(AtomicBool::new(true)), + Arc::new(RwLock::new(HashSet::new())), + false, + None, + ), + ); + + route_outgoing_envelope( + &mut connections, + OutgoingEnvelope::ToConnection { + connection_id, + message: OutgoingMessage::Notification( + crate::outgoing_message::OutgoingNotification { + method: "codex/event/task_started".to_string(), + params: None, + }, + ), + }, + ) + .await; + + assert!( + writer_rx.try_recv().is_err(), + "legacy notifications should not reach external clients" + ); + } + + #[tokio::test] + async fn to_connection_legacy_notifications_are_preserved_for_in_process_clients() { + let connection_id = ConnectionId(11); + let (writer_tx, mut writer_rx) = mpsc::channel(1); + + let mut connections = HashMap::new(); + connections.insert( + connection_id, + OutboundConnectionState::new( + writer_tx, + Arc::new(AtomicBool::new(true)), + Arc::new(AtomicBool::new(true)), + Arc::new(RwLock::new(HashSet::new())), + true, + None, + ), + ); + + route_outgoing_envelope( + &mut connections, + OutgoingEnvelope::ToConnection { + connection_id, + message: OutgoingMessage::Notification( + crate::outgoing_message::OutgoingNotification { + method: "codex/event/task_started".to_string(), + params: None, + }, + ), + }, + ) + .await; + + let message = writer_rx + .recv() + .await + .expect("legacy notification should reach in-process clients"); + assert!(matches!( + message, + OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { + method, + params: None, + }) if method == "codex/event/task_started" + )); + } + #[tokio::test] async fn command_execution_request_approval_strips_experimental_fields_without_capability() { let connection_id = ConnectionId(8); @@ -968,6 +1068,7 @@ mod tests { Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(false)), Arc::new(RwLock::new(HashSet::new())), + false, None, ), ); @@ -1034,6 +1135,7 @@ mod tests { Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)), Arc::new(RwLock::new(HashSet::new())), + false, None, ), ); @@ -1121,6 +1223,7 @@ mod tests { Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)), Arc::new(RwLock::new(HashSet::new())), + false, Some(fast_disconnect_token.clone()), ), ); @@ -1131,6 +1234,7 @@ mod tests { Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)), Arc::new(RwLock::new(HashSet::new())), + false, Some(slow_disconnect_token.clone()), ), ); @@ -1159,20 +1263,14 @@ mod tests { ), ) .await - .expect("broadcast should not block on a full writer"); - assert!(!connections.contains_key(&slow_connection_id)); - assert!(slow_disconnect_token.is_cancelled()); + .expect("broadcast should return even when legacy notifications are dropped"); + assert!(connections.contains_key(&slow_connection_id)); + assert!(!slow_disconnect_token.is_cancelled()); assert!(!fast_disconnect_token.is_cancelled()); - let fast_message = fast_writer_rx - .try_recv() - .expect("fast connection should receive broadcast"); - assert!(matches!( - fast_message, - OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { - method, - params: None, - }) if method == "codex/event/test" - )); + assert!( + fast_writer_rx.try_recv().is_err(), + "broadcast legacy notification should be dropped for fast connections" + ); let slow_message = slow_writer_rx .try_recv() @@ -1208,6 +1306,7 @@ mod tests { Arc::new(AtomicBool::new(true)), Arc::new(AtomicBool::new(true)), Arc::new(RwLock::new(HashSet::new())), + false, None, ), ); @@ -1232,14 +1331,9 @@ mod tests { .await .expect("first queued message should be readable") .expect("first queued message should exist"); - let second = timeout(Duration::from_millis(100), writer_rx.recv()) - .await - .expect("second message should eventually be delivered") - .expect("second message should exist"); - timeout(Duration::from_millis(100), route_task) .await - .expect("routing should finish after writer drains") + .expect("routing should finish immediately when legacy notifications are dropped") .expect("routing task should succeed"); assert!(matches!( @@ -1250,11 +1344,9 @@ mod tests { }) if method == "queued" )); assert!(matches!( - second, - OutgoingMessage::Notification(crate::outgoing_message::OutgoingNotification { - method, - params: None, - }) if method == "second" + writer_rx.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + | Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) )); } } diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index 2398cd8fc..f84b70ba0 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -594,7 +594,7 @@ impl McpProcess { /// Deterministically clean up an intentionally in-flight turn. /// /// Some tests assert behavior while a turn is still running. Returning from those tests - /// without an explicit interrupt + `codex/event/turn_aborted` wait can leave in-flight work + /// without an explicit interrupt + terminal turn notification wait can leave in-flight work /// racing teardown and intermittently show up as `LEAK` in nextest. /// /// In rare races, the turn can also fail or complete on its own after we send @@ -631,18 +631,19 @@ impl McpProcess { } match tokio::time::timeout( read_timeout, - self.read_stream_until_notification_message("codex/event/turn_aborted"), + self.read_stream_until_notification_message("turn/completed"), ) .await { Ok(result) => { - result.with_context(|| "failed while waiting for turn aborted notification")?; + result.with_context(|| "failed while waiting for terminal turn notification")?; } Err(err) => { if self.pending_turn_completed_notification(&thread_id, &turn_id) { return Ok(()); } - return Err(err).with_context(|| "timed out waiting for turn aborted notification"); + return Err(err) + .with_context(|| "timed out waiting for terminal turn notification"); } } Ok(()) diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 8887e87f5..6b5bf1156 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -139,10 +139,7 @@ async fn initialize_opt_out_notification_methods_filters_notifications() -> Resu }, Some(InitializeCapabilities { experimental_api: true, - opt_out_notification_methods: Some(vec![ - "thread/started".to_string(), - "codex/event/session_configured".to_string(), - ]), + opt_out_notification_methods: Some(vec!["thread/started".to_string()]), }), ), ) diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 15b309ddf..8f0847527 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -1152,11 +1152,6 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> { .await??; // Ensure we do NOT receive a CommandExecutionRequestApproval request before task completes - timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), - ) - .await??; timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), @@ -1462,7 +1457,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; @@ -1651,7 +1646,7 @@ async fn turn_start_file_change_approval_v2() -> Result<()> { timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; @@ -1782,7 +1777,7 @@ async fn turn_start_file_change_approval_accept_for_session_persists_v2() -> Res .await??; timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; @@ -1840,7 +1835,7 @@ async fn turn_start_file_change_approval_accept_for_session_persists_v2() -> Res .await??; timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; @@ -1991,7 +1986,7 @@ async fn turn_start_file_change_approval_decline_v2() -> Result<()> { timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; diff --git a/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs b/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs index 8fd1bb6e5..559be8b18 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs @@ -303,7 +303,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> { timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_notification_message("turn/completed"), ) .await??; diff --git a/codex-rs/app-server/tests/suite/v2/turn_steer.rs b/codex-rs/app-server/tests/suite/v2/turn_steer.rs index 61a8a9794..5d1b3cc22 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_steer.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_steer.rs @@ -133,7 +133,7 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> { let _task_started: JSONRPCNotification = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_started"), + mcp.read_stream_until_notification_message("turn/started"), ) .await??; @@ -236,7 +236,7 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> { let _task_started: JSONRPCNotification = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_started"), + mcp.read_stream_until_notification_message("turn/started"), ) .await??;