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

Timestamp app-server notifications at emission (#32905)

Browse Source 
## What changed

- Add an optional `emittedAtMs` field to server notification envelopes and expose the envelope in the generated JSON and TypeScript schemas.
- Populate the Unix timestamp when app-server emits a notification, before transport routing, so every connection in a fan-out receives the same value.
- Keep the field optional for clients decoding notifications from older app-server versions.

## Testing

- Verify notification serialization includes `emittedAtMs`.
- Verify targeted fan-out reuses one timestamp across connections.

GitOrigin-RevId: b379cb11e8e91c558bcd3d46ce459311a6e81e54
This commit is contained in:
Owen Lin
2026-07-14 00:38:04 +00:00
committed by copyberry
parent 7f3eb8a223
commit c39520f3d1
22 changed files with 400 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+7
View File
@@ -7117,5 +7117,12 @@
"type": "object"
}
],
"properties": {
"emittedAtMs": {
"description": "Unix timestamp (in milliseconds) when app-server emitted this notification.",
"format": "int64",
"type": "integer"
}
},
"title": "ServerNotification"
}
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+7
View File
@@ -5476,6 +5476,13 @@
"type": "object"
}
],
"properties": {
"emittedAtMs": {
"description": "Unix timestamp (in milliseconds) when app-server emitted this notification.",
"format": "int64",
"type": "integer"
}
},
"title": "ServerNotification"
},
"ServerRequest": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+7
View File
@@ -14385,6 +14385,13 @@
"type": "object"
}
],
"properties": {
"emittedAtMs": {
"description": "Unix timestamp (in milliseconds) when app-server emitted this notification.",
"format": "int64",
"type": "integer"
}
},
"title": "ServerNotification"
},
"ServerRequestResolvedNotification": {
codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts
Generated
+87
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/index.ts
Generated
+1
View File
@@ -73,6 +73,7 @@ export type { ResponseItem } from "./ResponseItem";
export type { ResponseItemId } from "./ResponseItemId";
export type { ReviewDecision } from "./ReviewDecision";
export type { ServerNotification } from "./ServerNotification";
export type { ServerNotificationEnvelope } from "./ServerNotificationEnvelope";
export type { ServerRequest } from "./ServerRequest";
export type { SessionSource } from "./SessionSource";
export type { Settings } from "./Settings";
codex-rs/app-server-protocol/src/export.rs
+22
View File
@@ -1,6 +1,7 @@
use crate::ClientNotification;
use crate::ClientRequest;
use crate::ServerNotification;
use crate::ServerNotificationEnvelope;
use crate::ServerRequest;
use crate::experimental_api::experimental_fields;
use crate::export_client_notification_schemas;
@@ -128,6 +129,7 @@ pub fn generate_ts_with_options(
ServerRequest::export_all_to(out_dir)?;
export_server_responses(out_dir)?;
ServerNotification::export_all_to(out_dir)?;
ServerNotificationEnvelope::export_all_to(out_dir)?;
if !options.experimental_api {
filter_experimental_ts(out_dir)?;
@@ -1339,6 +1341,7 @@ where
strip_v1_client_request_variants_from_json_schema(&mut schema_value);
} else if file_stem == "ServerNotification" {
strip_v1_server_notification_variants_from_json_schema(&mut schema_value);
add_server_notification_emitted_at_to_json_schema(&mut schema_value)?;
}
enforce_numbered_definition_collision_overrides(file_stem, &mut schema_value);
annotate_schema(&mut schema_value, Some(file_stem));
@@ -1371,6 +1374,25 @@ where
})
}
fn add_server_notification_emitted_at_to_json_schema(schema: &mut Value) -> Result<()> {
let schema = schema
.as_object_mut()
.ok_or_else(|| anyhow!("expected ServerNotification schema to be an object"))?;
schema.insert(
"properties".to_string(),
serde_json::json!({
"emittedAtMs": {
"description": "Unix timestamp (in milliseconds) when app-server emitted this notification.",
"format": "int64",
"type": "integer"
}
}),
);
// Keep this optional in generated client schemas for compatibility with
// older app-server versions. New servers still always emit it.
Ok(())
}
fn enforce_numbered_definition_collision_overrides(schema_name: &str, schema: &mut Value) {
for defs_key in ["definitions", "$defs"] {
let Some(defs) = schema.get(defs_key).and_then(Value::as_object) else {
codex-rs/app-server-protocol/src/protocol/common.rs
+19
View File
@@ -1709,6 +1709,25 @@ server_notification_definitions! {
}
/// Server notification envelope sent over app-server transports.
///
/// `emitted_at_ms` records when app-server emitted the notification, before it
/// is fanned out to individual connections.
#[derive(Serialize, Deserialize, Debug, Clone, TS)]
#[serde(rename_all = "camelCase")]
pub struct ServerNotificationEnvelope {
#[serde(flatten)]
pub notification: ServerNotification,
/// Unix timestamp (in milliseconds) when app-server emitted this notification.
///
/// Optional so clients can decode notifications from older app-server
/// versions. Current app-server versions always populate it.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
#[ts(type = "number")]
pub emitted_at_ms: Option<i64>,
}
client_notification_definitions! {
Initialized,
}
codex-rs/app-server-protocol/src/schema_fixtures.rs
+2
View File
@@ -1,6 +1,7 @@
use crate::ClientNotification;
use crate::ClientRequest;
use crate::ServerNotification;
use crate::ServerNotificationEnvelope;
use crate::ServerRequest;
use crate::export::GENERATED_TS_HEADER;
use crate::export::filter_experimental_ts_tree;
@@ -66,6 +67,7 @@ pub fn generate_typescript_schema_fixture_subtree_for_tests() -> Result<BTreeMap
visit_server_response_types(visitor);
})?;
collect_typescript_fixture_file::<ServerNotification>(&mut files, &mut seen)?;
collect_typescript_fixture_file::<ServerNotificationEnvelope>(&mut files, &mut seen)?;
filter_experimental_ts_tree(&mut files)?;
generate_index_ts_tree(&mut files);
codex-rs/app-server-transport/src/outgoing_message.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use std::fmt;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use codex_app_server_protocol::ServerRequest;
use serde::Serialize;
use tokio::sync::oneshot;
@@ -25,7 +25,7 @@ pub enum OutgoingMessage {
Request(ServerRequest),
/// AppServerNotification is specific to the case where this is run as an
/// "app server" as opposed to an MCP server.
AppServerNotification(ServerNotification),
AppServerNotification(ServerNotificationEnvelope),
Response(OutgoingResponse),
Error(OutgoingError),
}
codex-rs/app-server-transport/src/transport/mod.rs
+7 -4
View File
@@ -281,6 +281,7 @@ mod tests {
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::time::Duration;
@@ -443,14 +444,15 @@ mod tests {
writer_tx
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "queued".to_string(),
details: None,
path: None,
range: None,
},
)),
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("writer queue should accept first message");
@@ -484,6 +486,7 @@ mod tests {
"summary": "queued",
"details": null,
},
"emittedAtMs": 1_234,
})
);
}
codex-rs/app-server-transport/src/transport/remote_control/client_tracker.rs
+6 -4
View File
@@ -448,6 +448,7 @@ mod tests {
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::time::timeout;
@@ -588,14 +589,15 @@ mod tests {
writer
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "test".to_string(),
details: None,
path: None,
range: None,
},
)),
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("writer should accept queued message");
codex-rs/app-server-transport/src/transport/remote_control/segment_tests.rs
+10 -6
View File
@@ -14,6 +14,7 @@ use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use pretty_assertions::assert_eq;
#[test]
@@ -74,12 +75,15 @@ fn splits_large_server_messages_into_wire_chunks() {
let envelope = ServerEnvelope {
event: ServerEvent::ServerMessage {
message: Box::new(OutgoingMessage::AppServerNotification(
ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "x".repeat(REMOTE_CONTROL_SEGMENT_MAX_BYTES),
details: None,
path: None,
range: None,
}),
ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "x".repeat(REMOTE_CONTROL_SEGMENT_MAX_BYTES),
details: None,
path: None,
range: None,
}),
emitted_at_ms: Some(1_234),
},
)),
},
client_id: ClientId("client-1".to_string()),
codex-rs/app-server-transport/src/transport/remote_control/tests.rs
+22 -15
View File
@@ -25,6 +25,7 @@ use codex_app_server_protocol::RemoteControlPairingStartParams;
use codex_app_server_protocol::RemoteControlPairingStatusParams;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::test_support::auth_manager_from_auth;
use codex_core::test_support::auth_manager_from_auth_with_home;
@@ -736,14 +737,15 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
writer
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "test".to_string(),
details: None,
path: None,
range: None,
},
)),
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("remote writer should accept outgoing message");
@@ -758,7 +760,8 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages()
"params": {
"summary": "test",
"details": null,
}
},
"emittedAtMs": 1_234,
}
})
);
@@ -1344,14 +1347,15 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
writer
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "stale".to_string(),
details: None,
path: None,
range: None,
},
)),
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("remote writer should accept outgoing message");
@@ -1367,7 +1371,8 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
"params": {
"summary": "stale",
"details": null,
}
},
"emittedAtMs": 1_234,
}
})
);
@@ -1634,14 +1639,15 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
writer
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "backend".to_string(),
details: None,
path: None,
range: None,
},
)),
}),
emitted_at_ms: Some(1_234),
}),
))
.await
.expect("remote writer should accept outgoing message");
@@ -1656,7 +1662,8 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
"params": {
"summary": "backend",
"details": null,
}
},
"emittedAtMs": 1_234,
}
})
);
codex-rs/app-server-transport/src/transport/remote_control/websocket.rs
+12 -6
View File
@@ -1825,6 +1825,7 @@ mod tests {
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::test_support::auth_manager_from_auth;
use codex_login::AuthDotJson;
@@ -3362,12 +3363,17 @@ mod tests {
ServerEnvelope {
event: ServerEvent::ServerMessage {
message: Box::new(OutgoingMessage::AppServerNotification(
ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: summary.to_string(),
details: None,
path: None,
range: None,
}),
ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: summary.to_string(),
details: None,
path: None,
range: None,
},
),
emitted_at_ms: Some(1_234),
},
)),
},
client_id: client_id.clone(),
codex-rs/app-server/src/bespoke_event_handling.rs
+57 -67
View File
@@ -2115,6 +2115,16 @@ mod tests {
}
}
async fn recv_broadcast_notification(
rx: &mut mpsc::Receiver<OutgoingEnvelope>,
) -> Result<ServerNotification> {
let message = recv_broadcast_message(rx).await?;
let OutgoingMessage::AppServerNotification(envelope) = message else {
bail!("unexpected message: {message:?}");
};
Ok(envelope.notification)
}
#[test]
fn rollback_response_rebuilds_pathless_thread_from_stored_history() -> Result<()> {
let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000789")?;
@@ -2473,9 +2483,9 @@ mod tests {
.await;
assert!(first_start);
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => {
ServerNotification::ItemStarted(payload) => {
assert_eq!(payload.thread_id, conversation_id.to_string());
assert_eq!(payload.turn_id, "turn-1");
assert_eq!(
@@ -2543,7 +2553,7 @@ mod tests {
&thread_state,
)
.await;
let _started = recv_broadcast_message(&mut rx).await?;
let _started = recv_broadcast_notification(&mut rx).await?;
complete_command_execution_item(
&conversation_id,
@@ -2560,9 +2570,9 @@ mod tests {
)
.await;
let completed = recv_broadcast_message(&mut rx).await?;
let completed = recv_broadcast_notification(&mut rx).await?;
match completed {
OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => {
ServerNotification::ItemCompleted(payload) => {
let ThreadItem::CommandExecution { id, status, .. } = payload.item else {
bail!("expected command execution completion");
};
@@ -2638,9 +2648,9 @@ mod tests {
GuardianAssessmentStatus::InProgress,
))
.await;
let first = recv_broadcast_message(&mut rx).await?;
let first = recv_broadcast_notification(&mut rx).await?;
match first {
OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => {
ServerNotification::ItemStarted(payload) => {
assert_eq!(payload.turn_id, "turn-guardian-approved");
let ThreadItem::CommandExecution { id, status, .. } = payload.item else {
bail!("expected command execution item");
@@ -2650,11 +2660,9 @@ mod tests {
}
other => bail!("unexpected message: {other:?}"),
}
let second = recv_broadcast_message(&mut rx).await?;
let second = recv_broadcast_notification(&mut rx).await?;
match second {
OutgoingMessage::AppServerNotification(
ServerNotification::ItemGuardianApprovalReviewStarted(payload),
) => {
ServerNotification::ItemGuardianApprovalReviewStarted(payload) => {
assert_eq!(payload.review_id, "review-cmd-guardian-approved");
assert_eq!(
payload.target_item_id.as_deref(),
@@ -2675,11 +2683,9 @@ mod tests {
GuardianAssessmentStatus::Approved,
))
.await;
let third = recv_broadcast_message(&mut rx).await?;
let third = recv_broadcast_notification(&mut rx).await?;
match third {
OutgoingMessage::AppServerNotification(
ServerNotification::ItemGuardianApprovalReviewCompleted(payload),
) => {
ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => {
assert_eq!(payload.review_id, "review-cmd-guardian-approved");
assert_eq!(
payload.target_item_id.as_deref(),
@@ -2705,9 +2711,9 @@ mod tests {
GuardianAssessmentStatus::InProgress,
))
.await;
let fourth = recv_broadcast_message(&mut rx).await?;
let fourth = recv_broadcast_notification(&mut rx).await?;
match fourth {
OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => {
ServerNotification::ItemStarted(payload) => {
assert_eq!(payload.turn_id, "turn-guardian-denied");
let ThreadItem::CommandExecution { id, status, .. } = payload.item else {
bail!("expected command execution item");
@@ -2717,11 +2723,9 @@ mod tests {
}
other => bail!("unexpected message: {other:?}"),
}
let fifth = recv_broadcast_message(&mut rx).await?;
let fifth = recv_broadcast_notification(&mut rx).await?;
match fifth {
OutgoingMessage::AppServerNotification(
ServerNotification::ItemGuardianApprovalReviewStarted(payload),
) => {
ServerNotification::ItemGuardianApprovalReviewStarted(payload) => {
assert_eq!(payload.review_id, "review-cmd-guardian-denied");
assert_eq!(
payload.target_item_id.as_deref(),
@@ -2742,11 +2746,9 @@ mod tests {
GuardianAssessmentStatus::Denied,
))
.await;
let sixth = recv_broadcast_message(&mut rx).await?;
let sixth = recv_broadcast_notification(&mut rx).await?;
match sixth {
OutgoingMessage::AppServerNotification(
ServerNotification::ItemGuardianApprovalReviewCompleted(payload),
) => {
ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => {
assert_eq!(payload.review_id, "review-cmd-guardian-denied");
assert_eq!(
payload.target_item_id.as_deref(),
@@ -2757,9 +2759,9 @@ mod tests {
}
other => bail!("unexpected message: {other:?}"),
}
let seventh = recv_broadcast_message(&mut rx).await?;
let seventh = recv_broadcast_notification(&mut rx).await?;
match seventh {
OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => {
ServerNotification::ItemCompleted(payload) => {
let ThreadItem::CommandExecution { id, status, .. } = payload.item else {
bail!("expected command execution completion");
};
@@ -2778,11 +2780,9 @@ mod tests {
guardian_context
.apply_guardian_assessment_event(missing_target)
.await;
let eighth = recv_broadcast_message(&mut rx).await?;
let eighth = recv_broadcast_notification(&mut rx).await?;
match eighth {
OutgoingMessage::AppServerNotification(
ServerNotification::ItemGuardianApprovalReviewStarted(payload),
) => {
ServerNotification::ItemGuardianApprovalReviewStarted(payload) => {
assert_eq!(payload.review_id, "review-cmd-guardian-missing-target");
assert_eq!(payload.target_item_id, None);
assert_eq!(
@@ -3255,9 +3255,9 @@ mod tests {
)
.await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnStarted(n)) => {
ServerNotification::TurnStarted(n) => {
assert_eq!(n.turn.id, "turn-1");
assert_eq!(n.turn.items_view, TurnItemsView::NotLoaded);
assert!(n.turn.items.is_empty());
@@ -3336,10 +3336,8 @@ mod tests {
ThreadStatus::NotLoaded
);
assert_eq!(thread_watch_manager.running_turn_count().await, 0);
let message = recv_broadcast_message(&mut rx).await?;
let OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) =
message
else {
let message = recv_broadcast_notification(&mut rx).await?;
let ServerNotification::ItemCompleted(payload) = message else {
bail!("unexpected message: {message:?}");
};
assert_eq!(
@@ -3418,10 +3416,8 @@ mod tests {
)
.await;
let item_started = recv_broadcast_message(&mut rx).await?;
let OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) =
item_started
else {
let item_started = recv_broadcast_notification(&mut rx).await?;
let ServerNotification::ItemStarted(payload) = item_started else {
bail!("unexpected message: {item_started:?}");
};
assert_eq!(payload.item.id(), "dynamic-1");
@@ -3487,9 +3483,9 @@ mod tests {
)
.await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, event_turn_id);
assert_eq!(n.turn.status, TurnStatus::Completed);
assert_eq!(n.turn.items_view, TurnItemsView::NotLoaded);
@@ -3540,9 +3536,9 @@ mod tests {
)
.await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, event_turn_id);
assert_eq!(n.turn.status, TurnStatus::Interrupted);
assert_eq!(n.turn.error, None);
@@ -3590,9 +3586,9 @@ mod tests {
)
.await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, event_turn_id);
assert_eq!(n.turn.status, TurnStatus::Failed);
assert_eq!(
@@ -3642,9 +3638,9 @@ mod tests {
handle_turn_plan_update(conversation_id, "turn-123", update, &outgoing).await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(n)) => {
ServerNotification::TurnPlanUpdated(n) => {
assert_eq!(n.thread_id, conversation_id.to_string());
assert_eq!(n.turn_id, "turn-123");
assert_eq!(n.explanation.as_deref(), Some("need plan"));
@@ -3722,11 +3718,9 @@ mod tests {
)
.await;
let first = recv_broadcast_message(&mut rx).await?;
let first = recv_broadcast_notification(&mut rx).await?;
match first {
OutgoingMessage::AppServerNotification(
ServerNotification::ThreadTokenUsageUpdated(payload),
) => {
ServerNotification::ThreadTokenUsageUpdated(payload) => {
assert_eq!(payload.thread_id, conversation_id.to_string());
assert_eq!(payload.turn_id, turn_id);
let usage = payload.token_usage;
@@ -3738,11 +3732,9 @@ mod tests {
other => bail!("unexpected notification: {other:?}"),
}
let second = recv_broadcast_message(&mut rx).await?;
let second = recv_broadcast_notification(&mut rx).await?;
match second {
OutgoingMessage::AppServerNotification(
ServerNotification::AccountRateLimitsUpdated(payload),
) => {
ServerNotification::AccountRateLimitsUpdated(payload) => {
assert_eq!(payload.rate_limits.limit_id.as_deref(), Some("codex"));
assert_eq!(payload.rate_limits.limit_name, None);
assert!(payload.rate_limits.primary.is_some());
@@ -3858,9 +3850,9 @@ mod tests {
.await;
// Verify: A turn 1
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, a_turn1);
assert_eq!(n.turn.status, TurnStatus::Failed);
assert_eq!(
@@ -3876,9 +3868,9 @@ mod tests {
}
// Verify: B turn 1
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, b_turn1);
assert_eq!(n.turn.status, TurnStatus::Failed);
assert_eq!(
@@ -3894,9 +3886,9 @@ mod tests {
}
// Verify: A turn 2
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => {
ServerNotification::TurnCompleted(n) => {
assert_eq!(n.turn.id, a_turn2);
assert_eq!(n.turn.status, TurnStatus::Completed);
assert_eq!(n.turn.error, None);
@@ -3933,11 +3925,9 @@ mod tests {
)
.await;
let msg = recv_broadcast_message(&mut rx).await?;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
OutgoingMessage::AppServerNotification(ServerNotification::TurnDiffUpdated(
notification,
)) => {
ServerNotification::TurnDiffUpdated(notification) => {
assert_eq!(notification.thread_id, conversation_id.to_string());
assert_eq!(notification.turn_id, "turn-1");
assert_eq!(notification.diff, unified_diff);
codex-rs/app-server/src/in_process.rs
+2 -1
View File
@@ -667,7 +667,8 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
.await;
}
}
OutgoingMessage::AppServerNotification(notification) => {
OutgoingMessage::AppServerNotification(envelope) => {
let notification = envelope.notification;
if server_notification_requires_delivery(&notification) {
if event_tx
.send(InProcessServerEvent::ServerNotification(notification))
codex-rs/app-server/src/message_processor_tracing_tests.rs
+2 -2
View File
@@ -477,7 +477,7 @@ async fn read_thread_started_notification(
continue;
};
if matches!(
notification,
notification.notification,
codex_app_server_protocol::ServerNotification::ThreadStarted(_)
) {
return;
@@ -490,7 +490,7 @@ async fn read_thread_started_notification(
continue;
};
if matches!(
notification,
notification.notification,
codex_app_server_protocol::ServerNotification::ThreadStarted(_)
) {
return;
codex-rs/app-server/src/outgoing_message.rs
+69 -20
View File
@@ -11,6 +11,7 @@ use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ServerResponse;
@@ -600,7 +601,7 @@ impl OutgoingMessageSender {
targeted_connections = connection_ids.len(),
"app-server event: {notification}"
);
let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone());
let outgoing_message = timestamped_server_notification(notification);
if connection_ids.is_empty() {
if let Err(err) = self
.sender
@@ -634,7 +635,7 @@ impl OutgoingMessageSender {
notification: ServerNotification,
) {
tracing::trace!("app-server event: {notification}");
let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone());
let outgoing_message = timestamped_server_notification(notification);
let (write_complete_tx, write_complete_rx) = oneshot::channel();
if let Err(err) = self
.sender
@@ -728,6 +729,13 @@ fn now_unix_timestamp_ms() -> u64 {
.unwrap_or_default()
}
fn timestamped_server_notification(notification: ServerNotification) -> OutgoingMessage {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification,
emitted_at_ms: Some(now_unix_timestamp_ms().try_into().unwrap_or_default()),
})
}
#[cfg(test)]
mod tests {
use std::time::Duration;
@@ -770,7 +778,11 @@ mod tests {
error: None,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
let jsonrpc_notification =
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification,
emitted_at_ms: Some(1_234),
});
assert_eq!(
json!({
"method": "account/login/completed",
@@ -779,6 +791,7 @@ mod tests {
"success": true,
"error": null,
},
"emittedAtMs": 1_234,
}),
serde_json::to_value(jsonrpc_notification)
.expect("ensure the strum macros serialize the method field correctly"),
@@ -795,7 +808,6 @@ mod tests {
error: None,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "account/login/completed",
@@ -805,7 +817,7 @@ mod tests {
"error": null,
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -831,7 +843,6 @@ mod tests {
},
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "account/rateLimits/updated",
@@ -852,7 +863,7 @@ mod tests {
}
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -865,7 +876,6 @@ mod tests {
plan_type: None,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "account/updated",
@@ -874,7 +884,7 @@ mod tests {
"planType": null
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -889,7 +899,6 @@ mod tests {
range: None,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!( {
"method": "configWarning",
@@ -898,7 +907,7 @@ mod tests {
"details": "error loading config: bad config",
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -911,7 +920,6 @@ mod tests {
message: "Automatic approval review denied the requested action.".to_string(),
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "guardianWarning",
@@ -920,7 +928,7 @@ mod tests {
"message": "Automatic approval review denied the requested action.",
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -936,7 +944,6 @@ mod tests {
reason: ModelRerouteReason::HighRiskCyberActivity,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "model/rerouted",
@@ -948,7 +955,7 @@ mod tests {
"reason": "highRiskCyberActivity",
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -962,7 +969,6 @@ mod tests {
verifications: vec![ModelVerification::TrustedAccessForCyber],
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "model/verification",
@@ -972,7 +978,7 @@ mod tests {
"verifications": ["trustedAccessForCyber"],
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -987,7 +993,6 @@ mod tests {
metadata: json!({"presentation": "inline"}),
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
assert_eq!(
json!({
"method": "turn/moderationMetadata",
@@ -997,7 +1002,7 @@ mod tests {
"metadata": {"presentation": "inline"},
},
}),
serde_json::to_value(jsonrpc_notification)
serde_json::to_value(notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
@@ -1152,6 +1157,43 @@ mod tests {
}
}
#[tokio::test]
async fn send_server_notification_to_connections_reuses_timestamp() {
let (tx, mut rx) = mpsc::channel::<OutgoingEnvelope>(2);
let outgoing =
OutgoingMessageSender::new(tx, codex_analytics::AnalyticsEventsClient::disabled());
outgoing
.send_server_notification_to_connections(
&[ConnectionId(1), ConnectionId(2)],
ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "test".to_string(),
details: None,
path: None,
range: None,
}),
)
.await;
let timestamps = [
rx.recv()
.await
.expect("first connection should receive notification"),
rx.recv()
.await
.expect("second connection should receive notification"),
]
.map(|envelope| match envelope {
OutgoingEnvelope::ToConnection {
message: OutgoingMessage::AppServerNotification(envelope),
..
} => envelope.emitted_at_ms,
_ => panic!("expected targeted server notification"),
});
assert_eq!(timestamps[0], timestamps[1]);
}
#[tokio::test]
async fn send_server_notification_to_connection_and_wait_tracks_write_completion() {
let (tx, mut rx) = mpsc::channel::<OutgoingEnvelope>(4);
@@ -1185,7 +1227,14 @@ mod tests {
panic!("expected targeted server notification envelope");
};
assert_eq!(connection_id, ConnectionId(42));
assert!(matches!(message, OutgoingMessage::AppServerNotification(_)));
let OutgoingMessage::AppServerNotification(envelope) = message else {
panic!("expected app-server notification");
};
assert!(
envelope
.emitted_at_ms
.is_some_and(|emitted_at_ms| emitted_at_ms > 0)
);
write_complete_tx
.expect("write completion sender should be attached")
.send(())
codex-rs/app-server/src/thread_status.rs
+4 -4
View File
@@ -877,10 +877,10 @@ mod tests {
let OutgoingEnvelope::Broadcast { message } = envelope else {
panic!("expected broadcast notification");
};
let OutgoingMessage::AppServerNotification(ServerNotification::ThreadStatusChanged(
notification,
)) = message
else {
let OutgoingMessage::AppServerNotification(envelope) = message else {
panic!("expected thread/status/changed notification");
};
let ServerNotification::ThreadStatusChanged(notification) = envelope.notification else {
panic!("expected thread/status/changed notification");
};
notification
codex-rs/app-server/src/transport.rs
+3 -3
View File
@@ -105,15 +105,15 @@ fn should_skip_notification_for_connection(
return false;
};
match message {
OutgoingMessage::AppServerNotification(notification) => {
if notification.experimental_reason().is_some()
OutgoingMessage::AppServerNotification(envelope) => {
if envelope.notification.experimental_reason().is_some()
&& !connection_state
.experimental_api_enabled
.load(Ordering::Acquire)
{
return true;
}
let method = notification.to_string();
let method = envelope.notification.to_string();
opted_out_notification_methods.contains(method.as_str())
}
_ => false,
codex-rs/app-server/src/transport_tests.rs
+51 -37
View File
@@ -2,6 +2,7 @@ use super::*;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerNotificationEnvelope;
use codex_app_server_protocol::ThreadRealtimeStartedNotification;
use codex_protocol::protocol::RealtimeConversationVersion;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -22,6 +23,13 @@ fn thread_realtime_started_notification() -> ServerNotification {
})
}
fn app_server_notification(notification: ServerNotification) -> OutgoingMessage {
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification,
emitted_at_ms: Some(1_234),
})
}
#[tokio::test]
async fn to_connection_notification_respects_opt_out_filters() {
let connection_id = ConnectionId(7);
@@ -46,7 +54,7 @@ async fn to_connection_notification_respects_opt_out_filters() {
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
message: app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "task_started".to_string(),
details: None,
@@ -86,7 +94,7 @@ async fn to_connection_notifications_are_dropped_for_opted_out_clients() {
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
message: app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "task_started".to_string(),
details: None,
@@ -126,7 +134,7 @@ async fn to_connection_notifications_are_preserved_for_non_opted_out_clients() {
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
message: app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "task_started".to_string(),
details: None,
@@ -145,9 +153,10 @@ async fn to_connection_notifications_are_preserved_for_non_opted_out_clients() {
.expect("notification should reach non-opted-out clients");
assert!(matches!(
message.message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "task_started"
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification { summary, .. }),
..
}) if summary == "task_started"
));
}
@@ -172,7 +181,7 @@ async fn experimental_notifications_are_dropped_without_capability() {
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(thread_realtime_started_notification()),
message: app_server_notification(thread_realtime_started_notification()),
write_complete_tx: None,
},
)
@@ -205,7 +214,7 @@ async fn experimental_notifications_are_preserved_with_capability() {
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(thread_realtime_started_notification()),
message: app_server_notification(thread_realtime_started_notification()),
write_complete_tx: None,
},
)
@@ -217,7 +226,10 @@ async fn experimental_notifications_are_preserved_with_capability() {
.expect("experimental notification should reach opted-in client");
assert!(matches!(
message.message,
OutgoingMessage::AppServerNotification(ServerNotification::ThreadRealtimeStarted(_))
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ThreadRealtimeStarted(_),
..
})
));
}
@@ -395,7 +407,7 @@ async fn broadcast_does_not_block_on_slow_connection() {
),
);
let queued_message = OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
let queued_message = app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "already-buffered".to_string(),
details: None,
@@ -407,14 +419,14 @@ async fn broadcast_does_not_block_on_slow_connection() {
.try_send(QueuedOutgoingMessage::new(queued_message))
.expect("channel should have room");
let broadcast_message = OutgoingMessage::AppServerNotification(
ServerNotification::ConfigWarning(ConfigWarningNotification {
let broadcast_message = app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "test".to_string(),
details: None,
path: None,
range: None,
}),
);
},
));
timeout(
Duration::from_millis(100),
route_outgoing_envelope(
@@ -434,9 +446,10 @@ async fn broadcast_does_not_block_on_slow_connection() {
.expect("fast connection should receive the broadcast notification");
assert!(matches!(
fast_message.message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "test"
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification { summary, .. }),
..
}) if summary == "test"
));
let slow_message = slow_writer_rx
@@ -444,9 +457,10 @@ async fn broadcast_does_not_block_on_slow_connection() {
.expect("slow connection should retain its original buffered message");
assert!(matches!(
slow_message.message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "already-buffered"
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification { summary, .. }),
..
}) if summary == "already-buffered"
));
}
@@ -455,16 +469,14 @@ async fn to_connection_stdio_waits_instead_of_disconnecting_when_writer_queue_is
let connection_id = ConnectionId(3);
let (writer_tx, mut writer_rx) = mpsc::channel(1);
writer_tx
.send(QueuedOutgoingMessage::new(
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "queued".to_string(),
details: None,
path: None,
range: None,
},
)),
))
.send(QueuedOutgoingMessage::new(app_server_notification(
ServerNotification::ConfigWarning(ConfigWarningNotification {
summary: "queued".to_string(),
details: None,
path: None,
range: None,
}),
)))
.await
.expect("channel should accept the first queued message");
@@ -485,7 +497,7 @@ async fn to_connection_stdio_waits_instead_of_disconnecting_when_writer_queue_is
&mut connections,
OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
message: app_server_notification(ServerNotification::ConfigWarning(
ConfigWarningNotification {
summary: "second".to_string(),
details: None,
@@ -510,17 +522,19 @@ async fn to_connection_stdio_waits_instead_of_disconnecting_when_writer_queue_is
assert!(matches!(
first.message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "queued"
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification { summary, .. }),
..
}) if summary == "queued"
));
let second = writer_rx
.try_recv()
.expect("second notification should be delivered once the queue has room");
assert!(matches!(
second.message,
OutgoingMessage::AppServerNotification(ServerNotification::ConfigWarning(
ConfigWarningNotification { summary, .. }
)) if summary == "second"
OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
notification: ServerNotification::ConfigWarning(ConfigWarningNotification { summary, .. }),
..
}) if summary == "second"
));
}
codex-rs/app-server/tests/suite/v2/thread_status.rs
+1
View File
@@ -80,6 +80,7 @@ async fn thread_status_changed_emits_runtime_updates() -> Result<()> {
JSONRPCMessage::Notification(JSONRPCNotification {
method,
params: Some(params),
..
}) if method == "thread/status/changed" => {
let notification: ThreadStatusChangedNotification = serde_json::from_value(params)?;
if notification.thread_id != thread.id {