diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index dbd61d3875..b0ce3d4a0f 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -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" } \ No newline at end of file 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 3db8b10e9f..bfc6561069 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 @@ -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": { 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 d3a15f2a49..c45d0f40a4 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 @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts new file mode 100644 index 0000000000..83891c6a95 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts @@ -0,0 +1,87 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { FuzzyFileSearchSessionCompletedNotification } from "./FuzzyFileSearchSessionCompletedNotification"; +import type { FuzzyFileSearchSessionUpdatedNotification } from "./FuzzyFileSearchSessionUpdatedNotification"; +import type { AccountLoginCompletedNotification } from "./v2/AccountLoginCompletedNotification"; +import type { AccountRateLimitsUpdatedNotification } from "./v2/AccountRateLimitsUpdatedNotification"; +import type { AccountUpdatedNotification } from "./v2/AccountUpdatedNotification"; +import type { AgentMessageDeltaNotification } from "./v2/AgentMessageDeltaNotification"; +import type { AppListUpdatedNotification } from "./v2/AppListUpdatedNotification"; +import type { CommandExecOutputDeltaNotification } from "./v2/CommandExecOutputDeltaNotification"; +import type { CommandExecutionOutputDeltaNotification } from "./v2/CommandExecutionOutputDeltaNotification"; +import type { ConfigWarningNotification } from "./v2/ConfigWarningNotification"; +import type { ContextCompactedNotification } from "./v2/ContextCompactedNotification"; +import type { DeprecationNoticeNotification } from "./v2/DeprecationNoticeNotification"; +import type { ErrorNotification } from "./v2/ErrorNotification"; +import type { ExternalAgentConfigImportCompletedNotification } from "./v2/ExternalAgentConfigImportCompletedNotification"; +import type { ExternalAgentConfigImportProgressNotification } from "./v2/ExternalAgentConfigImportProgressNotification"; +import type { FileChangeOutputDeltaNotification } from "./v2/FileChangeOutputDeltaNotification"; +import type { FileChangePatchUpdatedNotification } from "./v2/FileChangePatchUpdatedNotification"; +import type { FsChangedNotification } from "./v2/FsChangedNotification"; +import type { GuardianWarningNotification } from "./v2/GuardianWarningNotification"; +import type { HookCompletedNotification } from "./v2/HookCompletedNotification"; +import type { HookStartedNotification } from "./v2/HookStartedNotification"; +import type { ItemCompletedNotification } from "./v2/ItemCompletedNotification"; +import type { ItemGuardianApprovalReviewCompletedNotification } from "./v2/ItemGuardianApprovalReviewCompletedNotification"; +import type { ItemGuardianApprovalReviewStartedNotification } from "./v2/ItemGuardianApprovalReviewStartedNotification"; +import type { ItemStartedNotification } from "./v2/ItemStartedNotification"; +import type { McpServerOauthLoginCompletedNotification } from "./v2/McpServerOauthLoginCompletedNotification"; +import type { McpServerStatusUpdatedNotification } from "./v2/McpServerStatusUpdatedNotification"; +import type { McpToolCallProgressNotification } from "./v2/McpToolCallProgressNotification"; +import type { ModelReroutedNotification } from "./v2/ModelReroutedNotification"; +import type { ModelSafetyBufferingUpdatedNotification } from "./v2/ModelSafetyBufferingUpdatedNotification"; +import type { ModelVerificationNotification } from "./v2/ModelVerificationNotification"; +import type { PlanDeltaNotification } from "./v2/PlanDeltaNotification"; +import type { ProcessExitedNotification } from "./v2/ProcessExitedNotification"; +import type { ProcessOutputDeltaNotification } from "./v2/ProcessOutputDeltaNotification"; +import type { RawResponseItemCompletedNotification } from "./v2/RawResponseItemCompletedNotification"; +import type { ReasoningSummaryPartAddedNotification } from "./v2/ReasoningSummaryPartAddedNotification"; +import type { ReasoningSummaryTextDeltaNotification } from "./v2/ReasoningSummaryTextDeltaNotification"; +import type { ReasoningTextDeltaNotification } from "./v2/ReasoningTextDeltaNotification"; +import type { RemoteControlStatusChangedNotification } from "./v2/RemoteControlStatusChangedNotification"; +import type { ServerRequestResolvedNotification } from "./v2/ServerRequestResolvedNotification"; +import type { SkillsChangedNotification } from "./v2/SkillsChangedNotification"; +import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification"; +import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification"; +import type { ThreadClosedNotification } from "./v2/ThreadClosedNotification"; +import type { ThreadDeletedNotification } from "./v2/ThreadDeletedNotification"; +import type { ThreadGoalClearedNotification } from "./v2/ThreadGoalClearedNotification"; +import type { ThreadGoalUpdatedNotification } from "./v2/ThreadGoalUpdatedNotification"; +import type { ThreadNameUpdatedNotification } from "./v2/ThreadNameUpdatedNotification"; +import type { ThreadRealtimeClosedNotification } from "./v2/ThreadRealtimeClosedNotification"; +import type { ThreadRealtimeErrorNotification } from "./v2/ThreadRealtimeErrorNotification"; +import type { ThreadRealtimeItemAddedNotification } from "./v2/ThreadRealtimeItemAddedNotification"; +import type { ThreadRealtimeOutputAudioDeltaNotification } from "./v2/ThreadRealtimeOutputAudioDeltaNotification"; +import type { ThreadRealtimeSdpNotification } from "./v2/ThreadRealtimeSdpNotification"; +import type { ThreadRealtimeStartedNotification } from "./v2/ThreadRealtimeStartedNotification"; +import type { ThreadRealtimeTranscriptDeltaNotification } from "./v2/ThreadRealtimeTranscriptDeltaNotification"; +import type { ThreadRealtimeTranscriptDoneNotification } from "./v2/ThreadRealtimeTranscriptDoneNotification"; +import type { ThreadSettingsUpdatedNotification } from "./v2/ThreadSettingsUpdatedNotification"; +import type { ThreadStartedNotification } from "./v2/ThreadStartedNotification"; +import type { ThreadStatusChangedNotification } from "./v2/ThreadStatusChangedNotification"; +import type { ThreadTokenUsageUpdatedNotification } from "./v2/ThreadTokenUsageUpdatedNotification"; +import type { ThreadUnarchivedNotification } from "./v2/ThreadUnarchivedNotification"; +import type { TurnCompletedNotification } from "./v2/TurnCompletedNotification"; +import type { TurnDiffUpdatedNotification } from "./v2/TurnDiffUpdatedNotification"; +import type { TurnModerationMetadataNotification } from "./v2/TurnModerationMetadataNotification"; +import type { TurnPlanUpdatedNotification } from "./v2/TurnPlanUpdatedNotification"; +import type { TurnStartedNotification } from "./v2/TurnStartedNotification"; +import type { WarningNotification } from "./v2/WarningNotification"; +import type { WindowsSandboxSetupCompletedNotification } from "./v2/WindowsSandboxSetupCompletedNotification"; +import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldWritableWarningNotification"; + +/** + * 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. + */ +export type ServerNotificationEnvelope = { +/** + * 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. + */ +emittedAtMs?: number, } & ({ "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }); diff --git a/codex-rs/app-server-protocol/schema/typescript/index.ts b/codex-rs/app-server-protocol/schema/typescript/index.ts index a30ee85b10..8afbbcc981 100644 --- a/codex-rs/app-server-protocol/schema/typescript/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/index.ts @@ -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"; diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 9f27dff7fa..f6dddb936e 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -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 { diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index bf56f7f28a..7bfc33a2c6 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -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, +} + client_notification_definitions! { Initialized, } diff --git a/codex-rs/app-server-protocol/src/schema_fixtures.rs b/codex-rs/app-server-protocol/src/schema_fixtures.rs index 18ed557ddf..6bf60e4a55 100644 --- a/codex-rs/app-server-protocol/src/schema_fixtures.rs +++ b/codex-rs/app-server-protocol/src/schema_fixtures.rs @@ -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(&mut files, &mut seen)?; + collect_typescript_fixture_file::(&mut files, &mut seen)?; filter_experimental_ts_tree(&mut files)?; generate_index_ts_tree(&mut files); diff --git a/codex-rs/app-server-transport/src/outgoing_message.rs b/codex-rs/app-server-transport/src/outgoing_message.rs index ff56b9fef9..f080d1fb13 100644 --- a/codex-rs/app-server-transport/src/outgoing_message.rs +++ b/codex-rs/app-server-transport/src/outgoing_message.rs @@ -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), } diff --git a/codex-rs/app-server-transport/src/transport/mod.rs b/codex-rs/app-server-transport/src/transport/mod.rs index 040de5c442..1c45a725ed 100644 --- a/codex-rs/app-server-transport/src/transport/mod.rs +++ b/codex-rs/app-server-transport/src/transport/mod.rs @@ -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, }) ); } diff --git a/codex-rs/app-server-transport/src/transport/remote_control/client_tracker.rs b/codex-rs/app-server-transport/src/transport/remote_control/client_tracker.rs index aef2bfd25d..891a7f722d 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/client_tracker.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/client_tracker.rs @@ -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"); diff --git a/codex-rs/app-server-transport/src/transport/remote_control/segment_tests.rs b/codex-rs/app-server-transport/src/transport/remote_control/segment_tests.rs index dc15bdf8ba..37c5df6772 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/segment_tests.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/segment_tests.rs @@ -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()), diff --git a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs index 465c6b13ec..92ad438770 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs @@ -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, } }) ); diff --git a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs index 7434454499..3cd41ff8b4 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs @@ -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(), diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 8fd4ad0a14..d9fb208b02 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -2115,6 +2115,16 @@ mod tests { } } + async fn recv_broadcast_notification( + rx: &mut mpsc::Receiver, + ) -> Result { + 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); diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index d8a2bc2276..84fec13b22 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -667,7 +667,8 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult { + OutgoingMessage::AppServerNotification(envelope) => { + let notification = envelope.notification; if server_notification_requires_delivery(¬ification) { if event_tx .send(InProcessServerEvent::ServerNotification(notification)) diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index 7c4ae8c985..e7451cb3fa 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -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; diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 16d5c2ff9d..7f93f27cb8 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -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::(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::(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(()) diff --git a/codex-rs/app-server/src/thread_status.rs b/codex-rs/app-server/src/thread_status.rs index a29b649788..b7e450e0ab 100644 --- a/codex-rs/app-server/src/thread_status.rs +++ b/codex-rs/app-server/src/thread_status.rs @@ -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 diff --git a/codex-rs/app-server/src/transport.rs b/codex-rs/app-server/src/transport.rs index 9c40653764..89b2d1e814 100644 --- a/codex-rs/app-server/src/transport.rs +++ b/codex-rs/app-server/src/transport.rs @@ -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, diff --git a/codex-rs/app-server/src/transport_tests.rs b/codex-rs/app-server/src/transport_tests.rs index 439284a176..968b5a0181 100644 --- a/codex-rs/app-server/src/transport_tests.rs +++ b/codex-rs/app-server/src/transport_tests.rs @@ -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" )); } diff --git a/codex-rs/app-server/tests/suite/v2/thread_status.rs b/codex-rs/app-server/tests/suite/v2/thread_status.rs index dad1db88af..6c54740443 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_status.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_status.rs @@ -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 {