From 947092283ac3680582fe2fee4b246a61f3c87129 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Wed, 25 Feb 2026 09:59:10 -0800 Subject: [PATCH] Add app-server v2 thread realtime API (#12715) Add experimental `thread/realtime/*` v2 requests and notifications, then route app-server realtime events through that thread-scoped surface with integration coverage. --------- Co-authored-by: Codex --- .../schema/json/ClientRequest.json | 32 ++ .../schema/json/ServerNotification.json | 214 ++++++++++ .../codex_app_server_protocol.schemas.json | 224 ++++++++++ .../v2/ThreadRealtimeClosedNotification.json | 20 + .../v2/ThreadRealtimeErrorNotification.json | 18 + .../ThreadRealtimeItemAddedNotification.json | 16 + ...dRealtimeOutputAudioDeltaNotification.json | 52 +++ .../v2/ThreadRealtimeStartedNotification.json | 20 + .../schema/typescript/ServerNotification.ts | 7 +- .../typescript/v2/ThreadRealtimeAudioChunk.ts | 8 + .../v2/ThreadRealtimeClosedNotification.ts | 8 + .../v2/ThreadRealtimeErrorNotification.ts | 8 + .../v2/ThreadRealtimeItemAddedNotification.ts | 9 + ...eadRealtimeOutputAudioDeltaNotification.ts | 9 + .../v2/ThreadRealtimeStartedNotification.ts | 8 + .../schema/typescript/v2/index.ts | 6 + .../src/protocol/common.rs | 138 +++++- .../app-server-protocol/src/protocol/v2.rs | 152 +++++++ codex-rs/app-server/README.md | 18 + .../app-server/src/bespoke_event_handling.rs | 73 ++++ .../app-server/src/codex_message_processor.rs | 198 +++++++++ .../app-server/tests/common/mcp_process.rs | 42 ++ .../tests/suite/v2/experimental_api.rs | 35 ++ codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/realtime_conversation.rs | 392 ++++++++++++++++++ 25 files changed, 1706 insertions(+), 2 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeClosedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeErrorNotification.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemAddedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeStartedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeClosedNotification.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeErrorNotification.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItemAddedNotification.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOutputAudioDeltaNotification.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartedNotification.ts create mode 100644 codex-rs/app-server/tests/suite/v2/realtime_conversation.rs diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index f4b94339e..b5b85d88a 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -2052,6 +2052,38 @@ ], "type": "object" }, + "ThreadRealtimeAudioChunk": { + "description": "EXPERIMENTAL - thread realtime audio chunk.", + "properties": { + "data": { + "type": "string" + }, + "numChannels": { + "format": "uint16", + "minimum": 0.0, + "type": "integer" + }, + "sampleRate": { + "format": "uint32", + "minimum": 0.0, + "type": "integer" + }, + "samplesPerChannel": { + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + } + }, + "required": [ + "data", + "numChannels", + "sampleRate" + ], + "type": "object" + }, "ThreadResumeParams": { "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nThe precedence is: history > path > thread_id. If using history or path, the thread_id param will be ignored.\n\nPrefer using thread_id whenever possible.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index b2dc4c3c3..47cdc6472 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2177,6 +2177,120 @@ ], "type": "object" }, + "ThreadRealtimeAudioChunk": { + "description": "EXPERIMENTAL - thread realtime audio chunk.", + "properties": { + "data": { + "type": "string" + }, + "numChannels": { + "format": "uint16", + "minimum": 0.0, + "type": "integer" + }, + "sampleRate": { + "format": "uint32", + "minimum": 0.0, + "type": "integer" + }, + "samplesPerChannel": { + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + } + }, + "required": [ + "data", + "numChannels", + "sampleRate" + ], + "type": "object" + }, + "ThreadRealtimeClosedNotification": { + "description": "EXPERIMENTAL - emitted when thread realtime transport closes.", + "properties": { + "reason": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, + "ThreadRealtimeErrorNotification": { + "description": "EXPERIMENTAL - emitted when thread realtime encounters an error.", + "properties": { + "message": { + "type": "string" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "type": "object" + }, + "ThreadRealtimeItemAddedNotification": { + "description": "EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.", + "properties": { + "item": true, + "threadId": { + "type": "string" + } + }, + "required": [ + "item", + "threadId" + ], + "type": "object" + }, + "ThreadRealtimeOutputAudioDeltaNotification": { + "description": "EXPERIMENTAL - streamed output audio emitted by thread realtime.", + "properties": { + "audio": { + "$ref": "#/definitions/ThreadRealtimeAudioChunk" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "audio", + "threadId" + ], + "type": "object" + }, + "ThreadRealtimeStartedNotification": { + "description": "EXPERIMENTAL - emitted when thread realtime startup is accepted.", + "properties": { + "sessionId": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, "ThreadStartedNotification": { "properties": { "thread": { @@ -3453,6 +3567,106 @@ "title": "FuzzyFileSearch/sessionCompletedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/started" + ], + "title": "Thread/realtime/startedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRealtimeStartedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/startedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/itemAdded" + ], + "title": "Thread/realtime/itemAddedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRealtimeItemAddedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/itemAddedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/outputAudio/delta" + ], + "title": "Thread/realtime/outputAudio/deltaNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRealtimeOutputAudioDeltaNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/outputAudio/deltaNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/error" + ], + "title": "Thread/realtime/errorNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRealtimeErrorNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/errorNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/closed" + ], + "title": "Thread/realtime/closedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRealtimeClosedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/closedNotification", + "type": "object" + }, { "description": "Notifies the user of world-writable directories on Windows, which cannot be protected by the sandbox.", "properties": { 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 4282f1c05..4bceb1d98 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 @@ -6425,6 +6425,106 @@ "title": "FuzzyFileSearch/sessionCompletedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/started" + ], + "title": "Thread/realtime/startedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRealtimeStartedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/startedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/itemAdded" + ], + "title": "Thread/realtime/itemAddedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRealtimeItemAddedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/itemAddedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/outputAudio/delta" + ], + "title": "Thread/realtime/outputAudio/deltaNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRealtimeOutputAudioDeltaNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/outputAudio/deltaNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/error" + ], + "title": "Thread/realtime/errorNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRealtimeErrorNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/errorNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/realtime/closed" + ], + "title": "Thread/realtime/closedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRealtimeClosedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/realtime/closedNotification", + "type": "object" + }, { "description": "Notifies the user of world-writable directories on Windows, which cannot be protected by the sandbox.", "properties": { @@ -13185,6 +13285,130 @@ "title": "ThreadReadResponse", "type": "object" }, + "ThreadRealtimeAudioChunk": { + "description": "EXPERIMENTAL - thread realtime audio chunk.", + "properties": { + "data": { + "type": "string" + }, + "numChannels": { + "format": "uint16", + "minimum": 0.0, + "type": "integer" + }, + "sampleRate": { + "format": "uint32", + "minimum": 0.0, + "type": "integer" + }, + "samplesPerChannel": { + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + } + }, + "required": [ + "data", + "numChannels", + "sampleRate" + ], + "type": "object" + }, + "ThreadRealtimeClosedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime transport closes.", + "properties": { + "reason": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRealtimeClosedNotification", + "type": "object" + }, + "ThreadRealtimeErrorNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime encounters an error.", + "properties": { + "message": { + "type": "string" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "title": "ThreadRealtimeErrorNotification", + "type": "object" + }, + "ThreadRealtimeItemAddedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.", + "properties": { + "item": true, + "threadId": { + "type": "string" + } + }, + "required": [ + "item", + "threadId" + ], + "title": "ThreadRealtimeItemAddedNotification", + "type": "object" + }, + "ThreadRealtimeOutputAudioDeltaNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - streamed output audio emitted by thread realtime.", + "properties": { + "audio": { + "$ref": "#/definitions/v2/ThreadRealtimeAudioChunk" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "audio", + "threadId" + ], + "title": "ThreadRealtimeOutputAudioDeltaNotification", + "type": "object" + }, + "ThreadRealtimeStartedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime startup is accepted.", + "properties": { + "sessionId": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRealtimeStartedNotification", + "type": "object" + }, "ThreadResumeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nThe precedence is: history > path > thread_id. If using history or path, the thread_id param will be ignored.\n\nPrefer using thread_id whenever possible.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeClosedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeClosedNotification.json new file mode 100644 index 000000000..edfba83b7 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeClosedNotification.json @@ -0,0 +1,20 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime transport closes.", + "properties": { + "reason": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRealtimeClosedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeErrorNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeErrorNotification.json new file mode 100644 index 000000000..e7ec76030 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeErrorNotification.json @@ -0,0 +1,18 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime encounters an error.", + "properties": { + "message": { + "type": "string" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "title": "ThreadRealtimeErrorNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemAddedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemAddedNotification.json new file mode 100644 index 000000000..06de7e00e --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemAddedNotification.json @@ -0,0 +1,16 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.", + "properties": { + "item": true, + "threadId": { + "type": "string" + } + }, + "required": [ + "item", + "threadId" + ], + "title": "ThreadRealtimeItemAddedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json new file mode 100644 index 000000000..d4df6194f --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json @@ -0,0 +1,52 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "ThreadRealtimeAudioChunk": { + "description": "EXPERIMENTAL - thread realtime audio chunk.", + "properties": { + "data": { + "type": "string" + }, + "numChannels": { + "format": "uint16", + "minimum": 0.0, + "type": "integer" + }, + "sampleRate": { + "format": "uint32", + "minimum": 0.0, + "type": "integer" + }, + "samplesPerChannel": { + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + } + }, + "required": [ + "data", + "numChannels", + "sampleRate" + ], + "type": "object" + } + }, + "description": "EXPERIMENTAL - streamed output audio emitted by thread realtime.", + "properties": { + "audio": { + "$ref": "#/definitions/ThreadRealtimeAudioChunk" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "audio", + "threadId" + ], + "title": "ThreadRealtimeOutputAudioDeltaNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeStartedNotification.json new file mode 100644 index 000000000..158411264 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeStartedNotification.json @@ -0,0 +1,20 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "EXPERIMENTAL - emitted when thread realtime startup is accepted.", + "properties": { + "sessionId": { + "type": [ + "string", + "null" + ] + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRealtimeStartedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 20dfa846f..92fecbfcc 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -30,6 +30,11 @@ import type { ReasoningTextDeltaNotification } from "./v2/ReasoningTextDeltaNoti import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification"; import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification"; 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 { ThreadRealtimeStartedNotification } from "./v2/ThreadRealtimeStartedNotification"; import type { ThreadStartedNotification } from "./v2/ThreadStartedNotification"; import type { ThreadStatusChangedNotification } from "./v2/ThreadStatusChangedNotification"; import type { ThreadTokenUsageUpdatedNotification } from "./v2/ThreadTokenUsageUpdatedNotification"; @@ -44,4 +49,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW /** * Notification sent from the server to the client. */ -export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "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": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification }; +export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "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": "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/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "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 } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts new file mode 100644 index 000000000..078f64224 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts @@ -0,0 +1,8 @@ +// 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. + +/** + * EXPERIMENTAL - thread realtime audio chunk. + */ +export type ThreadRealtimeAudioChunk = { data: string, sampleRate: number, numChannels: number, samplesPerChannel: number | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeClosedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeClosedNotification.ts new file mode 100644 index 000000000..a39cd71ed --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeClosedNotification.ts @@ -0,0 +1,8 @@ +// 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. + +/** + * EXPERIMENTAL - emitted when thread realtime transport closes. + */ +export type ThreadRealtimeClosedNotification = { threadId: string, reason: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeErrorNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeErrorNotification.ts new file mode 100644 index 000000000..0b24879ec --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeErrorNotification.ts @@ -0,0 +1,8 @@ +// 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. + +/** + * EXPERIMENTAL - emitted when thread realtime encounters an error. + */ +export type ThreadRealtimeErrorNotification = { threadId: string, message: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItemAddedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItemAddedNotification.ts new file mode 100644 index 000000000..f996e77cd --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItemAddedNotification.ts @@ -0,0 +1,9 @@ +// 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 { JsonValue } from "../serde_json/JsonValue"; + +/** + * EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend. + */ +export type ThreadRealtimeItemAddedNotification = { threadId: string, item: JsonValue, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOutputAudioDeltaNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOutputAudioDeltaNotification.ts new file mode 100644 index 000000000..1d03fd89c --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeOutputAudioDeltaNotification.ts @@ -0,0 +1,9 @@ +// 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 { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; + +/** + * EXPERIMENTAL - streamed output audio emitted by thread realtime. + */ +export type ThreadRealtimeOutputAudioDeltaNotification = { threadId: string, audio: ThreadRealtimeAudioChunk, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartedNotification.ts new file mode 100644 index 000000000..736ecde1f --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartedNotification.ts @@ -0,0 +1,8 @@ +// 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. + +/** + * EXPERIMENTAL - emitted when thread realtime startup is accepted. + */ +export type ThreadRealtimeStartedNotification = { threadId: string, sessionId: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 5d033d363..3f9041dcf 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -179,6 +179,12 @@ export type { ThreadLoadedListResponse } from "./ThreadLoadedListResponse"; export type { ThreadNameUpdatedNotification } from "./ThreadNameUpdatedNotification"; export type { ThreadReadParams } from "./ThreadReadParams"; export type { ThreadReadResponse } from "./ThreadReadResponse"; +export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; +export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification"; +export type { ThreadRealtimeErrorNotification } from "./ThreadRealtimeErrorNotification"; +export type { ThreadRealtimeItemAddedNotification } from "./ThreadRealtimeItemAddedNotification"; +export type { ThreadRealtimeOutputAudioDeltaNotification } from "./ThreadRealtimeOutputAudioDeltaNotification"; +export type { ThreadRealtimeStartedNotification } from "./ThreadRealtimeStartedNotification"; export type { ThreadResumeParams } from "./ThreadResumeParams"; export type { ThreadResumeResponse } from "./ThreadResumeResponse"; export type { ThreadRollbackParams } from "./ThreadRollbackParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index b7df2593b..7a918ada0 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -7,6 +7,7 @@ use crate::export::GeneratedSchema; use crate::export::write_json_schema; use crate::protocol::v1; use crate::protocol::v2; +use codex_experimental_api_macros::ExperimentalApi; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; @@ -268,6 +269,26 @@ client_request_definitions! { params: v2::TurnInterruptParams, response: v2::TurnInterruptResponse, }, + #[experimental("thread/realtime/start")] + ThreadRealtimeStart => "thread/realtime/start" { + params: v2::ThreadRealtimeStartParams, + response: v2::ThreadRealtimeStartResponse, + }, + #[experimental("thread/realtime/appendAudio")] + ThreadRealtimeAppendAudio => "thread/realtime/appendAudio" { + params: v2::ThreadRealtimeAppendAudioParams, + response: v2::ThreadRealtimeAppendAudioResponse, + }, + #[experimental("thread/realtime/appendText")] + ThreadRealtimeAppendText => "thread/realtime/appendText" { + params: v2::ThreadRealtimeAppendTextParams, + response: v2::ThreadRealtimeAppendTextResponse, + }, + #[experimental("thread/realtime/stop")] + ThreadRealtimeStop => "thread/realtime/stop" { + params: v2::ThreadRealtimeStopParams, + response: v2::ThreadRealtimeStopResponse, + }, ReviewStart => "review/start" { params: v2::ReviewStartParams, response: v2::ReviewStartResponse, @@ -586,7 +607,16 @@ macro_rules! server_notification_definitions { ),* $(,)? ) => { /// Notification sent from the server to the client. - #[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS, Display)] + #[derive( + Serialize, + Deserialize, + Debug, + Clone, + JsonSchema, + TS, + Display, + ExperimentalApi, + )] #[serde(tag = "method", content = "params", rename_all = "camelCase")] #[strum(serialize_all = "camelCase")] pub enum ServerNotification { @@ -823,6 +853,16 @@ server_notification_definitions! { ConfigWarning => "configWarning" (v2::ConfigWarningNotification), FuzzyFileSearchSessionUpdated => "fuzzyFileSearch/sessionUpdated" (FuzzyFileSearchSessionUpdatedNotification), FuzzyFileSearchSessionCompleted => "fuzzyFileSearch/sessionCompleted" (FuzzyFileSearchSessionCompletedNotification), + #[experimental("thread/realtime/started")] + ThreadRealtimeStarted => "thread/realtime/started" (v2::ThreadRealtimeStartedNotification), + #[experimental("thread/realtime/itemAdded")] + ThreadRealtimeItemAdded => "thread/realtime/itemAdded" (v2::ThreadRealtimeItemAddedNotification), + #[experimental("thread/realtime/outputAudio/delta")] + ThreadRealtimeOutputAudioDelta => "thread/realtime/outputAudio/delta" (v2::ThreadRealtimeOutputAudioDeltaNotification), + #[experimental("thread/realtime/error")] + ThreadRealtimeError => "thread/realtime/error" (v2::ThreadRealtimeErrorNotification), + #[experimental("thread/realtime/closed")] + ThreadRealtimeClosed => "thread/realtime/closed" (v2::ThreadRealtimeClosedNotification), /// Notifies the user of world-writable directories on Windows, which cannot be protected by the sandbox. WindowsWorldWritableWarning => "windows/worldWritableWarning" (v2::WindowsWorldWritableWarningNotification), @@ -1350,6 +1390,31 @@ mod tests { Ok(()) } + #[test] + fn serialize_thread_realtime_start() -> Result<()> { + let request = ClientRequest::ThreadRealtimeStart { + request_id: RequestId::Integer(9), + params: v2::ThreadRealtimeStartParams { + thread_id: "thr_123".to_string(), + prompt: "You are on a call".to_string(), + session_id: Some("sess_456".to_string()), + }, + }; + assert_eq!( + json!({ + "method": "thread/realtime/start", + "id": 9, + "params": { + "threadId": "thr_123", + "prompt": "You are on a call", + "sessionId": "sess_456" + } + }), + serde_json::to_value(&request)?, + ); + Ok(()) + } + #[test] fn serialize_thread_status_changed_notification() -> Result<()> { let notification = @@ -1372,6 +1437,37 @@ mod tests { Ok(()) } + #[test] + fn serialize_thread_realtime_output_audio_delta_notification() -> Result<()> { + let notification = ServerNotification::ThreadRealtimeOutputAudioDelta( + v2::ThreadRealtimeOutputAudioDeltaNotification { + thread_id: "thr_123".to_string(), + audio: v2::ThreadRealtimeAudioChunk { + data: "AQID".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: Some(512), + }, + }, + ); + assert_eq!( + json!({ + "method": "thread/realtime/outputAudio/delta", + "params": { + "threadId": "thr_123", + "audio": { + "data": "AQID", + "sampleRate": 24000, + "numChannels": 1, + "samplesPerChannel": 512 + } + } + }), + serde_json::to_value(¬ification)?, + ); + Ok(()) + } + #[test] fn mock_experimental_method_is_marked_experimental() { let request = ClientRequest::MockExperimentalMethod { @@ -1381,6 +1477,46 @@ mod tests { let reason = crate::experimental_api::ExperimentalApi::experimental_reason(&request); assert_eq!(reason, Some("mock/experimentalMethod")); } + #[test] + fn thread_realtime_start_is_marked_experimental() { + let request = ClientRequest::ThreadRealtimeStart { + request_id: RequestId::Integer(1), + params: v2::ThreadRealtimeStartParams { + thread_id: "thr_123".to_string(), + prompt: "You are on a call".to_string(), + session_id: None, + }, + }; + let reason = crate::experimental_api::ExperimentalApi::experimental_reason(&request); + assert_eq!(reason, Some("thread/realtime/start")); + } + #[test] + fn thread_realtime_started_notification_is_marked_experimental() { + let notification = + ServerNotification::ThreadRealtimeStarted(v2::ThreadRealtimeStartedNotification { + thread_id: "thr_123".to_string(), + session_id: Some("sess_456".to_string()), + }); + let reason = crate::experimental_api::ExperimentalApi::experimental_reason(¬ification); + assert_eq!(reason, Some("thread/realtime/started")); + } + + #[test] + fn thread_realtime_output_audio_delta_notification_is_marked_experimental() { + let notification = ServerNotification::ThreadRealtimeOutputAudioDelta( + v2::ThreadRealtimeOutputAudioDeltaNotification { + thread_id: "thr_123".to_string(), + audio: v2::ThreadRealtimeAudioChunk { + data: "AQID".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: Some(512), + }, + }, + ); + let reason = crate::experimental_api::ExperimentalApi::experimental_reason(¬ification); + assert_eq!(reason, Some("thread/realtime/outputAudio/delta")); + } #[test] fn command_execution_request_approval_additional_permissions_is_marked_experimental() { diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 22b7fe05c..1e4e5d0c3 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -46,6 +46,7 @@ use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus; use codex_protocol::protocol::RateLimitSnapshot as CoreRateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow as CoreRateLimitWindow; use codex_protocol::protocol::ReadOnlyAccess as CoreReadOnlyAccess; +use codex_protocol::protocol::RealtimeAudioFrame as CoreRealtimeAudioFrame; use codex_protocol::protocol::RejectConfig as CoreRejectConfig; use codex_protocol::protocol::SessionSource as CoreSessionSource; use codex_protocol::protocol::SkillDependencies as CoreSkillDependencies; @@ -2551,6 +2552,157 @@ pub struct ErrorNotification { pub turn_id: String, } +/// EXPERIMENTAL - thread realtime audio chunk. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeAudioChunk { + pub data: String, + pub sample_rate: u32, + pub num_channels: u16, + pub samples_per_channel: Option, +} + +impl From for ThreadRealtimeAudioChunk { + fn from(value: CoreRealtimeAudioFrame) -> Self { + let CoreRealtimeAudioFrame { + data, + sample_rate, + num_channels, + samples_per_channel, + } = value; + Self { + data, + sample_rate, + num_channels, + samples_per_channel, + } + } +} + +impl From for CoreRealtimeAudioFrame { + fn from(value: ThreadRealtimeAudioChunk) -> Self { + let ThreadRealtimeAudioChunk { + data, + sample_rate, + num_channels, + samples_per_channel, + } = value; + Self { + data, + sample_rate, + num_channels, + samples_per_channel, + } + } +} + +/// EXPERIMENTAL - start a thread-scoped realtime session. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeStartParams { + pub thread_id: String, + pub prompt: String, + #[ts(optional = nullable)] + pub session_id: Option, +} + +/// EXPERIMENTAL - response for starting thread realtime. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeStartResponse {} + +/// EXPERIMENTAL - append audio input to thread realtime. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeAppendAudioParams { + pub thread_id: String, + pub audio: ThreadRealtimeAudioChunk, +} + +/// EXPERIMENTAL - response for appending realtime audio input. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeAppendAudioResponse {} + +/// EXPERIMENTAL - append text input to thread realtime. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeAppendTextParams { + pub thread_id: String, + pub text: String, +} + +/// EXPERIMENTAL - response for appending realtime text input. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeAppendTextResponse {} + +/// EXPERIMENTAL - stop thread realtime. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeStopParams { + pub thread_id: String, +} + +/// EXPERIMENTAL - response for stopping thread realtime. +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeStopResponse {} + +/// EXPERIMENTAL - emitted when thread realtime startup is accepted. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeStartedNotification { + pub thread_id: String, + pub session_id: Option, +} + +/// EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeItemAddedNotification { + pub thread_id: String, + pub item: JsonValue, +} + +/// EXPERIMENTAL - streamed output audio emitted by thread realtime. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeOutputAudioDeltaNotification { + pub thread_id: String, + pub audio: ThreadRealtimeAudioChunk, +} + +/// EXPERIMENTAL - emitted when thread realtime encounters an error. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeErrorNotification { + pub thread_id: String, + pub message: String, +} + +/// EXPERIMENTAL - emitted when thread realtime transport closes. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeClosedNotification { + pub thread_id: String, + pub reason: Option, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0be424a40..08b7c8df0 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -135,6 +135,10 @@ Example with notification opt-out: - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". - `turn/steer` — add user input to an already in-flight turn without starting a new turn; returns the active `turnId` that accepted the input. - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`. +- `thread/realtime/start` — start a thread-scoped realtime session (experimental); returns `{}` and streams `thread/realtime/*` notifications. +- `thread/realtime/appendAudio` — append an input audio chunk to the active realtime session (experimental); returns `{}`. +- `thread/realtime/appendText` — append text input to the active realtime session (experimental); returns `{}`. +- `thread/realtime/stop` — stop the active realtime session for the thread (experimental); returns `{}`. - `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start` and emits `item/started`/`item/completed` notifications with `enteredReviewMode` and `exitedReviewMode` items, plus a final assistant `agentMessage` containing the review. - `command/exec` — run a single command under the server sandbox without starting a thread/turn (handy for utilities and validation). - `model/list` — list available models (set `includeHidden: true` to include entries with `hidden: true`), with reasoning effort options and optional `upgrade` model ids. @@ -553,6 +557,8 @@ Notes: Event notifications are the server-initiated event stream for thread lifecycles, turn lifecycles, and the items within them. After you start or resume a thread, keep reading stdout for `thread/started`, `thread/archived`, `thread/unarchived`, `turn/*`, and `item/*` notifications. +Thread realtime uses a separate thread-scoped notification surface. `thread/realtime/*` notifications are ephemeral transport events, not `ThreadItem`s, and are not returned by `thread/read`, `thread/resume`, or `thread/fork`. + ### Notification opt-out Clients can suppress specific notifications per connection by sending exact method names in `initialize.params.capabilities.optOutNotificationMethods`. @@ -574,6 +580,18 @@ The fuzzy file search session API emits per-query notifications: - `fuzzyFileSearch/sessionUpdated` — `{ sessionId, query, files }` with the current matching files for the active query. - `fuzzyFileSearch/sessionCompleted` — `{ sessionId, query }` once indexing/matching for that query has completed. +### Thread realtime events (experimental) + +The thread realtime API emits thread-scoped notifications for session lifecycle and streaming media: + +- `thread/realtime/started` — `{ threadId, sessionId }` once realtime starts for the thread (experimental). +- `thread/realtime/itemAdded` — `{ threadId, item }` for non-audio realtime items (experimental). `item` is forwarded as raw JSON while the upstream websocket item schema remains unstable. +- `thread/realtime/outputAudio/delta` — `{ threadId, audio }` for streamed output audio chunks (experimental). `audio` uses camelCase fields (`data`, `sampleRate`, `numChannels`, `samplesPerChannel`). +- `thread/realtime/error` — `{ threadId, message }` when realtime encounters a transport or backend error (experimental). +- `thread/realtime/closed` — `{ threadId, reason }` when the realtime transport closes (experimental). + +Because audio is intentionally separate from `ThreadItem`, clients can opt out of `thread/realtime/outputAudio/delta` independently with `optOutNotificationMethods`. + ### Windows sandbox setup events - `windowsSandbox/setupCompleted` — `{ mode, success, error }` after a `windowsSandbox/setupStart` request finishes. diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 0b81a4c89..56e55823c 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -62,6 +62,11 @@ use codex_app_server_protocol::SkillRequestApprovalResponse; use codex_app_server_protocol::TerminalInteractionNotification; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadNameUpdatedNotification; +use codex_app_server_protocol::ThreadRealtimeClosedNotification; +use codex_app_server_protocol::ThreadRealtimeErrorNotification; +use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; +use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification; +use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::ThreadRollbackResponse; use codex_app_server_protocol::ThreadTokenUsage; use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification; @@ -97,6 +102,7 @@ use codex_protocol::protocol::ExecCommandEndEvent; use codex_protocol::protocol::McpToolCallBeginEvent; use codex_protocol::protocol::McpToolCallEndEvent; use codex_protocol::protocol::Op; +use codex_protocol::protocol::RealtimeEvent; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::ReviewOutputEvent; use codex_protocol::protocol::TokenCountEvent; @@ -173,6 +179,73 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } } + EventMsg::RealtimeConversationStarted(event) => { + if let ApiVersion::V2 = api_version { + let notification = ThreadRealtimeStartedNotification { + thread_id: conversation_id.to_string(), + session_id: event.session_id, + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeStarted( + notification, + )) + .await; + } + } + EventMsg::RealtimeConversationRealtime(event) => { + if let ApiVersion::V2 = api_version { + match event.payload { + RealtimeEvent::SessionCreated { .. } => {} + RealtimeEvent::SessionUpdated { .. } => {} + RealtimeEvent::AudioOut(audio) => { + let notification = ThreadRealtimeOutputAudioDeltaNotification { + thread_id: conversation_id.to_string(), + audio: audio.into(), + }; + outgoing + .send_server_notification( + ServerNotification::ThreadRealtimeOutputAudioDelta(notification), + ) + .await; + } + RealtimeEvent::ConversationItemAdded(item) => { + let notification = ThreadRealtimeItemAddedNotification { + thread_id: conversation_id.to_string(), + item, + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( + notification, + )) + .await; + } + RealtimeEvent::Error(message) => { + let notification = ThreadRealtimeErrorNotification { + thread_id: conversation_id.to_string(), + message, + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeError( + notification, + )) + .await; + } + } + } + } + EventMsg::RealtimeConversationClosed(event) => { + if let ApiVersion::V2 = api_version { + let notification = ThreadRealtimeClosedNotification { + thread_id: conversation_id.to_string(), + reason: event.reason, + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeClosed( + notification, + )) + .await; + } + } EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { call_id, turn_id, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index ebe1b77ed..189789871 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -138,6 +138,14 @@ use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; +use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse; +use codex_app_server_protocol::ThreadRealtimeAppendTextParams; +use codex_app_server_protocol::ThreadRealtimeAppendTextResponse; +use codex_app_server_protocol::ThreadRealtimeStartParams; +use codex_app_server_protocol::ThreadRealtimeStartResponse; +use codex_app_server_protocol::ThreadRealtimeStopParams; +use codex_app_server_protocol::ThreadRealtimeStopResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadRollbackParams; @@ -235,6 +243,9 @@ use codex_protocol::dynamic_tools::DynamicToolSpec as CoreDynamicToolSpec; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::ConversationAudioParams; +use codex_protocol::protocol::ConversationStartParams; +use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GitInfo as CoreGitInfo; use codex_protocol::protocol::InitialHistory; @@ -625,6 +636,22 @@ impl CodexMessageProcessor { self.turn_interrupt(to_connection_request_id(request_id), params) .await; } + ClientRequest::ThreadRealtimeStart { request_id, params } => { + self.thread_realtime_start(to_connection_request_id(request_id), params) + .await; + } + ClientRequest::ThreadRealtimeAppendAudio { request_id, params } => { + self.thread_realtime_append_audio(to_connection_request_id(request_id), params) + .await; + } + ClientRequest::ThreadRealtimeAppendText { request_id, params } => { + self.thread_realtime_append_text(to_connection_request_id(request_id), params) + .await; + } + ClientRequest::ThreadRealtimeStop { request_id, params } => { + self.thread_realtime_stop(to_connection_request_id(request_id), params) + .await; + } ClientRequest::ReviewStart { request_id, params } => { self.review_start(to_connection_request_id(request_id), params) .await; @@ -5518,6 +5545,177 @@ impl CodexMessageProcessor { } } + async fn prepare_realtime_conversation_thread( + &mut self, + request_id: ConnectionRequestId, + thread_id: &str, + ) -> Option<(ThreadId, Arc)> { + let (thread_id, thread) = match self.load_thread(thread_id).await { + Ok(v) => v, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return None; + } + }; + + if let Err(error) = self + .ensure_conversation_listener( + thread_id, + request_id.connection_id, + false, + ApiVersion::V2, + ) + .await + { + self.outgoing.send_error(request_id, error).await; + return None; + } + + if !thread.enabled(Feature::RealtimeConversation) { + self.send_invalid_request_error( + request_id, + format!("thread {thread_id} does not support realtime conversation"), + ) + .await; + return None; + } + + Some((thread_id, thread)) + } + + async fn thread_realtime_start( + &mut self, + request_id: ConnectionRequestId, + params: ThreadRealtimeStartParams, + ) { + let Some((_, thread)) = self + .prepare_realtime_conversation_thread(request_id.clone(), ¶ms.thread_id) + .await + else { + return; + }; + + let submit = thread + .submit(Op::RealtimeConversationStart(ConversationStartParams { + prompt: params.prompt, + session_id: params.session_id, + })) + .await; + + match submit { + Ok(_) => { + self.outgoing + .send_response(request_id, ThreadRealtimeStartResponse::default()) + .await; + } + Err(err) => { + self.send_internal_error( + request_id, + format!("failed to start realtime conversation: {err}"), + ) + .await; + } + } + } + + async fn thread_realtime_append_audio( + &mut self, + request_id: ConnectionRequestId, + params: ThreadRealtimeAppendAudioParams, + ) { + let Some((_, thread)) = self + .prepare_realtime_conversation_thread(request_id.clone(), ¶ms.thread_id) + .await + else { + return; + }; + + let submit = thread + .submit(Op::RealtimeConversationAudio(ConversationAudioParams { + frame: params.audio.into(), + })) + .await; + + match submit { + Ok(_) => { + self.outgoing + .send_response(request_id, ThreadRealtimeAppendAudioResponse::default()) + .await; + } + Err(err) => { + self.send_internal_error( + request_id, + format!("failed to append realtime conversation audio: {err}"), + ) + .await; + } + } + } + + async fn thread_realtime_append_text( + &mut self, + request_id: ConnectionRequestId, + params: ThreadRealtimeAppendTextParams, + ) { + let Some((_, thread)) = self + .prepare_realtime_conversation_thread(request_id.clone(), ¶ms.thread_id) + .await + else { + return; + }; + + let submit = thread + .submit(Op::RealtimeConversationText(ConversationTextParams { + text: params.text, + })) + .await; + + match submit { + Ok(_) => { + self.outgoing + .send_response(request_id, ThreadRealtimeAppendTextResponse::default()) + .await; + } + Err(err) => { + self.send_internal_error( + request_id, + format!("failed to append realtime conversation text: {err}"), + ) + .await; + } + } + } + + async fn thread_realtime_stop( + &mut self, + request_id: ConnectionRequestId, + params: ThreadRealtimeStopParams, + ) { + let Some((_, thread)) = self + .prepare_realtime_conversation_thread(request_id.clone(), ¶ms.thread_id) + .await + else { + return; + }; + + let submit = thread.submit(Op::RealtimeConversationClose).await; + + match submit { + Ok(_) => { + self.outgoing + .send_response(request_id, ThreadRealtimeStopResponse::default()) + .await; + } + Err(err) => { + self.send_internal_error( + request_id, + format!("failed to stop realtime conversation: {err}"), + ) + .await; + } + } + } + fn build_review_turn(turn_id: String, display_text: &str) -> Turn { let items = if display_text.is_empty() { Vec::new() diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index bb35c2c0a..f1917add8 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -57,6 +57,10 @@ use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadReadParams; +use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; +use codex_app_server_protocol::ThreadRealtimeAppendTextParams; +use codex_app_server_protocol::ThreadRealtimeStartParams; +use codex_app_server_protocol::ThreadRealtimeStopParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadRollbackParams; use codex_app_server_protocol::ThreadSetNameParams; @@ -584,6 +588,44 @@ impl McpProcess { self.send_request("turn/interrupt", params).await } + /// Send a `thread/realtime/start` JSON-RPC request (v2). + pub async fn send_thread_realtime_start_request( + &mut self, + params: ThreadRealtimeStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/realtime/start", params).await + } + + /// Send a `thread/realtime/appendAudio` JSON-RPC request (v2). + pub async fn send_thread_realtime_append_audio_request( + &mut self, + params: ThreadRealtimeAppendAudioParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/realtime/appendAudio", params) + .await + } + + /// Send a `thread/realtime/appendText` JSON-RPC request (v2). + pub async fn send_thread_realtime_append_text_request( + &mut self, + params: ThreadRealtimeAppendTextParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/realtime/appendText", params) + .await + } + + /// Send a `thread/realtime/stop` JSON-RPC request (v2). + pub async fn send_thread_realtime_stop_request( + &mut self, + params: ThreadRealtimeStopParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/realtime/stop", params).await + } + /// Deterministically clean up an intentionally in-flight turn. /// /// Some tests assert behavior while a turn is still running. Returning from those tests diff --git a/codex-rs/app-server/tests/suite/v2/experimental_api.rs b/codex-rs/app-server/tests/suite/v2/experimental_api.rs index 798d52abf..9af3aa4c7 100644 --- a/codex-rs/app-server/tests/suite/v2/experimental_api.rs +++ b/codex-rs/app-server/tests/suite/v2/experimental_api.rs @@ -10,6 +10,7 @@ use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::MockExperimentalMethodParams; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadRealtimeStartParams; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use pretty_assertions::assert_eq; @@ -50,6 +51,40 @@ async fn mock_experimental_method_requires_experimental_api_capability() -> Resu Ok(()) } +#[tokio::test] +async fn realtime_conversation_start_requires_experimental_api_capability() -> Result<()> { + let codex_home = TempDir::new()?; + let mut mcp = McpProcess::new(codex_home.path()).await?; + + let init = mcp + .initialize_with_capabilities( + default_client_info(), + Some(InitializeCapabilities { + experimental_api: false, + opt_out_notification_methods: None, + }), + ) + .await?; + let JSONRPCMessage::Response(_) = init else { + anyhow::bail!("expected initialize response, got {init:?}"); + }; + + let request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: "thr_123".to_string(), + prompt: "hello".to_string(), + session_id: None, + }) + .await?; + let error = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_experimental_capability_error(error, "thread/realtime/start"); + Ok(()) +} + #[tokio::test] async fn thread_start_mock_field_requires_experimental_api_capability() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 67e7ae3ea..8afb1e3f4 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -15,6 +15,7 @@ mod model_list; mod output_schema; mod plan_item; mod rate_limits; +mod realtime_conversation; mod request_user_input; mod review; mod safety_check_downgrade; diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs new file mode 100644 index 000000000..f2664848c --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -0,0 +1,392 @@ +use anyhow::Context; +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_mock_responses_server_sequence_unchecked; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; +use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse; +use codex_app_server_protocol::ThreadRealtimeAppendTextParams; +use codex_app_server_protocol::ThreadRealtimeAppendTextResponse; +use codex_app_server_protocol::ThreadRealtimeAudioChunk; +use codex_app_server_protocol::ThreadRealtimeClosedNotification; +use codex_app_server_protocol::ThreadRealtimeErrorNotification; +use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; +use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification; +use codex_app_server_protocol::ThreadRealtimeStartParams; +use codex_app_server_protocol::ThreadRealtimeStartResponse; +use codex_app_server_protocol::ThreadRealtimeStartedNotification; +use codex_app_server_protocol::ThreadRealtimeStopParams; +use codex_app_server_protocol::ThreadRealtimeStopResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_core::features::FEATURES; +use codex_core::features::Feature; +use core_test_support::responses::start_websocket_server; +use core_test_support::skip_if_no_network; +use pretty_assertions::assert_eq; +use serde::de::DeserializeOwned; +use serde_json::json; +use std::path::Path; +use std::time::Duration; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); + +#[tokio::test] +async fn realtime_conversation_streams_v2_notifications() -> Result<()> { + skip_if_no_network!(Ok(())); + + let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + let realtime_server = start_websocket_server(vec![vec![ + vec![json!({ + "type": "session.created", + "session": { "id": "sess_backend" } + })], + vec![json!({ + "type": "session.updated", + "session": { "backend_prompt": "backend prompt" } + })], + vec![ + json!({ + "type": "response.output_audio.delta", + "delta": "AQID", + "sample_rate": 24_000, + "num_channels": 1, + "samples_per_channel": 512 + }), + json!({ + "type": "conversation.item.added", + "item": { + "type": "message", + "role": "assistant", + "content": [{ "type": "text", "text": "hi" }] + } + }), + json!({ + "type": "error", + "message": "upstream boom" + }), + ], + ]]) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &responses_server.uri(), + realtime_server.uri(), + true, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + mcp.initialize().await?; + + let thread_start_request_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)), + ) + .await??; + let thread_start: ThreadStartResponse = to_response(thread_start_response)?; + + let start_request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: thread_start.thread.id.clone(), + prompt: "backend prompt".to_string(), + session_id: None, + }) + .await?; + let start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)), + ) + .await??; + let _: ThreadRealtimeStartResponse = to_response(start_response)?; + + let started = + read_notification::(&mut mcp, "thread/realtime/started") + .await?; + assert_eq!(started.thread_id, thread_start.thread.id); + assert!(started.session_id.is_some()); + + let audio_append_request_id = mcp + .send_thread_realtime_append_audio_request(ThreadRealtimeAppendAudioParams { + thread_id: started.thread_id.clone(), + audio: ThreadRealtimeAudioChunk { + data: "BQYH".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: Some(480), + }, + }) + .await?; + let audio_append_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(audio_append_request_id)), + ) + .await??; + let _: ThreadRealtimeAppendAudioResponse = to_response(audio_append_response)?; + + let text_append_request_id = mcp + .send_thread_realtime_append_text_request(ThreadRealtimeAppendTextParams { + thread_id: started.thread_id.clone(), + text: "hello".to_string(), + }) + .await?; + let text_append_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(text_append_request_id)), + ) + .await??; + let _: ThreadRealtimeAppendTextResponse = to_response(text_append_response)?; + + let output_audio = read_notification::( + &mut mcp, + "thread/realtime/outputAudio/delta", + ) + .await?; + assert_eq!(output_audio.audio.data, "AQID"); + assert_eq!(output_audio.audio.sample_rate, 24_000); + assert_eq!(output_audio.audio.num_channels, 1); + assert_eq!(output_audio.audio.samples_per_channel, Some(512)); + + let item_added = read_notification::( + &mut mcp, + "thread/realtime/itemAdded", + ) + .await?; + assert_eq!(item_added.thread_id, output_audio.thread_id); + assert_eq!(item_added.item["type"], json!("message")); + + let realtime_error = + read_notification::(&mut mcp, "thread/realtime/error") + .await?; + assert_eq!(realtime_error.thread_id, output_audio.thread_id); + assert_eq!(realtime_error.message, "upstream boom"); + + let closed = + read_notification::(&mut mcp, "thread/realtime/closed") + .await?; + assert_eq!(closed.thread_id, output_audio.thread_id); + assert_eq!(closed.reason.as_deref(), Some("transport_closed")); + + let connections = realtime_server.connections(); + assert_eq!(connections.len(), 1); + let connection = &connections[0]; + assert_eq!(connection.len(), 3); + assert_eq!( + connection[0].body_json()["type"].as_str(), + Some("session.create") + ); + let mut request_types = [ + connection[1].body_json()["type"] + .as_str() + .context("expected websocket request type")? + .to_string(), + connection[2].body_json()["type"] + .as_str() + .context("expected websocket request type")? + .to_string(), + ]; + request_types.sort(); + assert_eq!( + request_types, + [ + "conversation.item.create".to_string(), + "response.input_audio.delta".to_string(), + ] + ); + + realtime_server.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> { + skip_if_no_network!(Ok(())); + + let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + let realtime_server = start_websocket_server(vec![vec![ + vec![json!({ + "type": "session.created", + "session": { "id": "sess_backend" } + })], + vec![], + ]]) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &responses_server.uri(), + realtime_server.uri(), + true, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + mcp.initialize().await?; + + let thread_start_request_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)), + ) + .await??; + let thread_start: ThreadStartResponse = to_response(thread_start_response)?; + + let start_request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: thread_start.thread.id.clone(), + prompt: "backend prompt".to_string(), + session_id: None, + }) + .await?; + let start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)), + ) + .await??; + let _: ThreadRealtimeStartResponse = to_response(start_response)?; + + let started = + read_notification::(&mut mcp, "thread/realtime/started") + .await?; + + let stop_request_id = mcp + .send_thread_realtime_stop_request(ThreadRealtimeStopParams { + thread_id: started.thread_id.clone(), + }) + .await?; + let stop_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(stop_request_id)), + ) + .await??; + let _: ThreadRealtimeStopResponse = to_response(stop_response)?; + + let closed = + read_notification::(&mut mcp, "thread/realtime/closed") + .await?; + assert_eq!(closed.thread_id, started.thread_id); + assert!(matches!( + closed.reason.as_deref(), + Some("requested" | "transport_closed") + )); + + realtime_server.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn realtime_conversation_requires_feature_flag() -> Result<()> { + skip_if_no_network!(Ok(())); + + let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + let realtime_server = start_websocket_server(vec![vec![]]).await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &responses_server.uri(), + realtime_server.uri(), + false, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + mcp.initialize().await?; + + let thread_start_request_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)), + ) + .await??; + let thread_start: ThreadStartResponse = to_response(thread_start_response)?; + + let start_request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: thread_start.thread.id.clone(), + prompt: "backend prompt".to_string(), + session_id: None, + }) + .await?; + let error = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(start_request_id)), + ) + .await??; + assert_invalid_request( + error, + format!( + "thread {} does not support realtime conversation", + thread_start.thread.id + ), + ); + + realtime_server.shutdown().await; + Ok(()) +} + +async fn read_notification(mcp: &mut McpProcess, method: &str) -> Result { + let notification = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_notification_message(method), + ) + .await??; + let params = notification + .params + .context("expected notification params to be present")?; + Ok(serde_json::from_value(params)?) +} + +fn create_config_toml( + codex_home: &Path, + responses_server_uri: &str, + realtime_server_uri: &str, + realtime_enabled: bool, +) -> std::io::Result<()> { + let realtime_feature_key = FEATURES + .iter() + .find(|spec| spec.id == Feature::RealtimeConversation) + .map(|spec| spec.key) + .unwrap_or("realtime_conversation"); + + std::fs::write( + codex_home.join("config.toml"), + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +model_provider = "mock_provider" +experimental_realtime_ws_base_url = "{realtime_server_uri}" + +[features] +{realtime_feature_key} = {realtime_enabled} + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{responses_server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + +fn assert_invalid_request(error: JSONRPCError, message: String) { + assert_eq!(error.error.code, -32600); + assert_eq!(error.error.message, message); + assert_eq!(error.error.data, None); +}