From 1c9f890c051dbd091f1827ecd0149d0f43001b5f Mon Sep 17 00:00:00 2001 From: Dylan Hurd Date: Tue, 14 Jul 2026 06:34:16 +0000 Subject: [PATCH] Expose exact per-response usage in raw app-server events (#32985) ## Why Clients that account for individual upstream Responses API calls need the exact usage for each completion rather than accumulated or estimated thread usage. ## What changed - Add the opt-in `rawResponse/completed` app-server notification with the `threadId`, `turnId`, `responseId`, and upstream `usage` payload. - Emit the transient event for normal turns and compaction requests when `experimentalRawEvents` is enabled. The event is not persisted or replayed, and `usage` is `null` when the upstream response omits it. - Export the notification in the app-server schemas and document its behavior. ## Testing - Cover exact token breakdowns for turn and compaction completions. GitOrigin-RevId: 49040a060bf1784c1ff1dd62112efb4a778428c4 --- .../codex_app_server_protocol.schemas.json | 32 ++++++ .../codex_app_server_protocol.v2.schemas.json | 32 ++++++ .../v2/RawResponseCompletedNotification.json | 66 ++++++++++++ .../schema/typescript/ServerNotification.ts | 3 +- .../v2/RawResponseCompletedNotification.ts | 10 ++ .../schema/typescript/v2/index.ts | 1 + codex-rs/app-server-protocol/src/export.rs | 3 +- .../src/protocol/common.rs | 2 + .../src/protocol/v2/thread.rs | 12 +++ codex-rs/app-server/README.md | 1 + .../app-server/src/bespoke_event_handling.rs | 12 +++ .../request_processors/thread_lifecycle.rs | 6 +- .../app-server/tests/suite/v2/compaction.rs | 42 +++++++- .../app-server/tests/suite/v2/turn_start.rs | 100 ++++++++++++++++++ codex-rs/core/src/compact.rs | 15 ++- codex-rs/core/src/compact_remote_v2.rs | 14 ++- .../core/src/compact_remote_v2_attempt.rs | 13 +++ codex-rs/core/src/session/turn.rs | 12 ++- codex-rs/mcp-server/src/codex_tool_runner.rs | 1 + codex-rs/protocol/src/protocol.rs | 10 ++ codex-rs/rollout-trace/src/protocol_event.rs | 2 + codex-rs/rollout/src/policy.rs | 1 + .../tui/src/app/app_server_event_targets.rs | 3 + codex-rs/tui/src/chatwidget/protocol.rs | 1 + 24 files changed, 387 insertions(+), 7 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/RawResponseCompletedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/RawResponseCompletedNotification.ts 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 3f8bcc9429..e1100206d7 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 @@ -14839,6 +14839,38 @@ ], "type": "object" }, + "RawResponseCompletedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "Internal-only notification containing the exact usage from one upstream Responses API completion.", + "properties": { + "responseId": { + "type": "string" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "usage": { + "anyOf": [ + { + "$ref": "#/definitions/v2/TokenUsageBreakdown" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "responseId", + "threadId", + "turnId" + ], + "title": "RawResponseCompletedNotification", + "type": "object" + }, "RawResponseItemCompletedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { 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 5a69ec14aa..5464a392db 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 @@ -11236,6 +11236,38 @@ ], "type": "object" }, + "RawResponseCompletedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "Internal-only notification containing the exact usage from one upstream Responses API completion.", + "properties": { + "responseId": { + "type": "string" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "usage": { + "anyOf": [ + { + "$ref": "#/definitions/TokenUsageBreakdown" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "responseId", + "threadId", + "turnId" + ], + "title": "RawResponseCompletedNotification", + "type": "object" + }, "RawResponseItemCompletedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/RawResponseCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/RawResponseCompletedNotification.json new file mode 100644 index 0000000000..d0caf5f26b --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/RawResponseCompletedNotification.json @@ -0,0 +1,66 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "TokenUsageBreakdown": { + "properties": { + "cachedInputTokens": { + "format": "int64", + "type": "integer" + }, + "inputTokens": { + "format": "int64", + "type": "integer" + }, + "outputTokens": { + "format": "int64", + "type": "integer" + }, + "reasoningOutputTokens": { + "format": "int64", + "type": "integer" + }, + "totalTokens": { + "format": "int64", + "type": "integer" + } + }, + "required": [ + "cachedInputTokens", + "inputTokens", + "outputTokens", + "reasoningOutputTokens", + "totalTokens" + ], + "type": "object" + } + }, + "description": "Internal-only notification containing the exact usage from one upstream Responses API completion.", + "properties": { + "responseId": { + "type": "string" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "usage": { + "anyOf": [ + { + "$ref": "#/definitions/TokenUsageBreakdown" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "responseId", + "threadId", + "turnId" + ], + "title": "RawResponseCompletedNotification", + "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 f3f828b8ac..17e9150971 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -35,6 +35,7 @@ import type { ModelVerificationNotification } from "./v2/ModelVerificationNotifi import type { PlanDeltaNotification } from "./v2/PlanDeltaNotification"; import type { ProcessExitedNotification } from "./v2/ProcessExitedNotification"; import type { ProcessOutputDeltaNotification } from "./v2/ProcessOutputDeltaNotification"; +import type { RawResponseCompletedNotification } from "./v2/RawResponseCompletedNotification"; import type { RawResponseItemCompletedNotification } from "./v2/RawResponseItemCompletedNotification"; import type { ReasoningSummaryPartAddedNotification } from "./v2/ReasoningSummaryPartAddedNotification"; import type { ReasoningSummaryTextDeltaNotification } from "./v2/ReasoningSummaryTextDeltaNotification"; @@ -74,4 +75,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/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 }; +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/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": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "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/v2/RawResponseCompletedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/RawResponseCompletedNotification.ts new file mode 100644 index 0000000000..b06e74b4c7 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/RawResponseCompletedNotification.ts @@ -0,0 +1,10 @@ +// 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 { TokenUsageBreakdown } from "./TokenUsageBreakdown"; + +/** + * Internal-only notification containing the exact usage from one upstream + * Responses API completion. + */ +export type RawResponseCompletedNotification = { threadId: string, turnId: string, responseId: string, usage: TokenUsageBreakdown | 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 10fb369ef9..2cfb40140c 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -343,6 +343,7 @@ export type { RateLimitResetCreditsSummary } from "./RateLimitResetCreditsSummar export type { RateLimitResetType } from "./RateLimitResetType"; export type { RateLimitSnapshot } from "./RateLimitSnapshot"; export type { RateLimitWindow } from "./RateLimitWindow"; +export type { RawResponseCompletedNotification } from "./RawResponseCompletedNotification"; export type { RawResponseItemCompletedNotification } from "./RawResponseItemCompletedNotification"; export type { ReasoningEffortOption } from "./ReasoningEffortOption"; export type { ReasoningSummaryPartAddedNotification } from "./ReasoningSummaryPartAddedNotification"; diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 4c2557989b..5c4b83a8a9 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -60,7 +60,8 @@ const SPECIAL_DEFINITIONS: &[&str] = &[ const FLAT_V2_SHARED_DEFINITIONS: &[&str] = &["ClientRequest", "ServerNotification"]; const V1_CLIENT_REQUEST_METHODS: &[&str] = &["getConversationSummary", "gitDiffToRemote", "getAuthStatus"]; -const EXCLUDED_SERVER_NOTIFICATION_METHODS_FOR_JSON: &[&str] = &["rawResponseItem/completed"]; +const EXCLUDED_SERVER_NOTIFICATION_METHODS_FOR_JSON: &[&str] = + &["rawResponseItem/completed", "rawResponse/completed"]; #[derive(Clone)] pub struct GeneratedSchema { diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 6be48b5de1..584c66fed9 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1645,6 +1645,8 @@ server_notification_definitions! { ItemCompleted => "item/completed" (v2::ItemCompletedNotification), /// This event is internal-only. Used by Codex Cloud. RawResponseItemCompleted => "rawResponseItem/completed" (v2::RawResponseItemCompletedNotification), + /// This event is internal-only. Used by clients that need exact upstream usage. + RawResponseCompleted => "rawResponse/completed" (v2::RawResponseCompletedNotification), AgentMessageDelta => "item/agentMessage/delta" (v2::AgentMessageDeltaNotification), /// EXPERIMENTAL - proposed plan streaming deltas for plan items. PlanDelta => "item/plan/delta" (v2::PlanDeltaNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 367a4a64c8..2a22b76993 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1375,6 +1375,18 @@ pub struct ThreadTokenUsageUpdatedNotification { pub token_usage: ThreadTokenUsage, } +/// Internal-only notification containing the exact usage from one upstream +/// Responses API completion. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RawResponseCompletedNotification { + pub thread_id: String, + pub turn_id: String, + pub response_id: String, + pub usage: 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 b23003260b..489d0864a9 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1357,6 +1357,7 @@ The app-server streams JSON-RPC notifications while a turn is running. Each turn - `turn/completed` — `{ turn }` where `turn.status` is `completed`, `interrupted`, or `failed`; failures carry `{ error: { message, codexErrorInfo?, additionalDetails? } }`. - `turn/diff/updated` — `{ threadId, turnId, diff }` represents the up-to-date snapshot of the turn-level unified diff, emitted after every FileChange item. `diff` is the latest aggregated unified diff across every file change in the turn. UIs can render this to show the full "what changed" view without stitching individual `fileChange` items. - `turn/plan/updated` — `{ turnId, explanation?, plan }` whenever the agent shares or changes its plan; each `plan` entry is `{ step, status }` with `status` in `pending`, `inProgress`, or `completed`. +- `rawResponse/completed` — internal-only; when `thread/start.experimentalRawEvents` is enabled, emits `{ threadId, turnId, responseId, usage }` once for each upstream Responses API completion. `usage` is the exact upstream usage payload mapped to the app-server token breakdown shape and is `null` when the upstream completion omitted usage. Unlike `thread/tokenUsage/updated`, this notification is not accumulated, estimated, persisted, or replayed. - `model/safetyBuffering/updated` — `{ threadId, turnId, model, useCases, reasons, showBufferingUi, fasterModel }` when a response enters safety buffering. `fasterModel` is nullable. This notification is transient and is not persisted in rollout history. - `model/rerouted` — `{ threadId, turnId, fromModel, toModel, reason }` when the backend reroutes a request to a different model (for example, due to high-risk cyber safety checks). - `model/verification` — `{ threadId, turnId, verifications }` when the backend flags additional account verification, such as `trustedAccessForCyber`. diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index d9fb208b02..b3491afedd 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -46,6 +46,7 @@ use codex_app_server_protocol::NetworkPolicyAmendment as V2NetworkPolicyAmendmen use codex_app_server_protocol::NetworkPolicyRuleAction as V2NetworkPolicyRuleAction; use codex_app_server_protocol::PermissionsRequestApprovalParams; use codex_app_server_protocol::PermissionsRequestApprovalResponse; +use codex_app_server_protocol::RawResponseCompletedNotification; use codex_app_server_protocol::RawResponseItemCompletedNotification; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; @@ -1029,6 +1030,17 @@ pub(crate) async fn apply_bespoke_event_handling( ) .await; } + EventMsg::RawResponseCompleted(raw_response_completed_event) => { + let notification = RawResponseCompletedNotification { + thread_id: conversation_id.to_string(), + turn_id: event_turn_id, + response_id: raw_response_completed_event.response_id, + usage: raw_response_completed_event.token_usage.map(Into::into), + }; + outgoing + .send_server_notification(ServerNotification::RawResponseCompleted(notification)) + .await; + } EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) => { // Core still fans out these deprecated events for raw-event and rollout compatibility // consumers; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 25346de7ee..da5db51e48 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -316,7 +316,11 @@ pub(super) async fn ensure_listener_task_running( thread_state.track_current_turn_event(&event.id, &event.msg); thread_state.experimental_raw_events }; - if matches!(&event.msg, EventMsg::RawResponseItem(_)) && !raw_events_enabled { + if matches!( + &event.msg, + EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_) + ) && !raw_events_enabled + { continue; } let subscribed_connection_ids = thread_state_manager diff --git a/codex-rs/app-server/tests/suite/v2/compaction.rs b/codex-rs/app-server/tests/suite/v2/compaction.rs index a534912a89..d4e5891538 100644 --- a/codex-rs/app-server/tests/suite/v2/compaction.rs +++ b/codex-rs/app-server/tests/suite/v2/compaction.rs @@ -16,12 +16,15 @@ use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RawResponseCompletedNotification; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TokenUsageBreakdown; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; @@ -283,7 +286,20 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() - .await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; - let thread_id = start_thread(&mut mcp).await?; + let thread_req = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("mock-model".to_string()), + experimental_raw_events: true, + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + let thread_id = thread.id; let compact_id = mcp .send_thread_compact_start_request(ThreadCompactStartParams { thread_id: thread_id.clone(), @@ -298,6 +314,15 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() - to_response::(compact_resp)?; let started = wait_for_context_compaction_started(&mut mcp).await?; + let raw_completed = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("rawResponse/completed"), + ) + .await??; + let raw_completed: ServerNotification = raw_completed.try_into()?; + let ServerNotification::RawResponseCompleted(raw_completed) = raw_completed else { + anyhow::bail!("expected rawResponse/completed notification"); + }; let completed = wait_for_context_compaction_completed(&mut mcp).await?; let ThreadItem::ContextCompaction { id: started_id } = started.item else { @@ -310,6 +335,21 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() - assert_eq!(started.thread_id, thread_id); assert_eq!(completed.thread_id, thread_id); assert_eq!(started_id, completed_id); + assert_eq!( + raw_completed, + RawResponseCompletedNotification { + thread_id, + turn_id: started.turn_id, + response_id: "r1".to_string(), + usage: Some(TokenUsageBreakdown { + total_tokens: 200, + input_tokens: 200, + cached_input_tokens: 0, + output_tokens: 0, + reasoning_output_tokens: 0, + }), + } + ); Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 15c557641f..1938db2adf 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -36,6 +36,7 @@ use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PatchApplyStatus; use codex_app_server_protocol::PatchChangeKind; +use codex_app_server_protocol::RawResponseCompletedNotification; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequestResolvedNotification; @@ -51,6 +52,7 @@ use codex_app_server_protocol::ThreadSettingsUpdatedNotification; use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TokenUsageBreakdown; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnEnvironmentParams; use codex_app_server_protocol::TurnItemsView; @@ -765,6 +767,104 @@ async fn turn_start_sends_service_tier_id_to_model_request() -> Result<()> { Ok(()) } +#[tokio::test] +async fn turn_start_emits_raw_response_completed_with_upstream_usage() -> Result<()> { + let server = responses::start_mock_server().await; + let body = responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_assistant_message("msg-1", "Done"), + json!({ + "type": "response.completed", + "response": { + "id": "resp-1", + "usage": { + "input_tokens": 30, + "input_tokens_details": { "cached_tokens": 11 }, + "output_tokens": 7, + "output_tokens_details": { "reasoning_tokens": 3 }, + "total_tokens": 37 + } + } + }), + ]); + responses::mount_sse_once(&server, body).await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + "never", + &BTreeMap::default(), + )?; + write_models_cache(codex_home.path())?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + experimental_raw_events: true, + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_resp = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + + let notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("rawResponse/completed"), + ) + .await??; + let notification: codex_app_server_protocol::ServerNotification = notification.try_into()?; + let codex_app_server_protocol::ServerNotification::RawResponseCompleted(notification) = + notification + else { + anyhow::bail!("expected rawResponse/completed notification"); + }; + + assert_eq!( + notification, + RawResponseCompletedNotification { + thread_id: thread.id, + turn_id: turn.id, + response_id: "resp-1".to_string(), + usage: Some(TokenUsageBreakdown { + total_tokens: 37, + input_tokens: 30, + cached_input_tokens: 11, + output_tokens: 7, + reasoning_output_tokens: 3, + }), + } + ); + + Ok(()) +} + #[tokio::test] async fn thread_start_omits_empty_instruction_overrides_from_model_request() -> Result<()> { let server = responses::start_mock_server().await; diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 3a32441518..e7bf9d51a5 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -36,6 +36,7 @@ use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RawResponseCompletedEvent; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::WarningEvent; use codex_protocol::user_input::UserInput; @@ -698,7 +699,19 @@ async fn drain_to_completed( Ok(ResponseEvent::RateLimits(snapshot)) => { sess.update_rate_limits(turn_context, snapshot).await; } - Ok(ResponseEvent::Completed { token_usage, .. }) => { + Ok(ResponseEvent::Completed { + response_id, + token_usage, + .. + }) => { + sess.send_event( + turn_context, + EventMsg::RawResponseCompleted(RawResponseCompletedEvent { + response_id, + token_usage: token_usage.clone(), + }), + ) + .await; sess.update_token_usage_info(turn_context, token_usage.as_ref()) .await?; return Ok(()); diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index c2a8a6191a..a4e7918860 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -325,6 +325,7 @@ async fn run_remote_compact_task_inner_impl( struct RemoteCompactionV2Output { compaction_output: ResponseItem, + response_id: String, token_usage: Option, } @@ -385,6 +386,7 @@ async fn collect_compaction_output( let mut compaction_count = 0usize; let mut compaction_output = None; let mut saw_completed = false; + let mut completed_response_id = None; let mut completed_token_usage = None; while let Some(event) = stream.next().await { match event? { @@ -397,8 +399,13 @@ async fn collect_compaction_output( } } } - ResponseEvent::Completed { token_usage, .. } => { + ResponseEvent::Completed { + response_id, + token_usage, + .. + } => { saw_completed = true; + completed_response_id = Some(response_id); completed_token_usage = token_usage; break; } @@ -422,8 +429,12 @@ async fn collect_compaction_output( let Some(compaction_output) = compaction_output else { unreachable!("compaction output must exist when count is exactly one"); }; + let Some(response_id) = completed_response_id else { + unreachable!("response id must exist after response.completed"); + }; Ok(RemoteCompactionV2Output { compaction_output, + response_id, token_usage: completed_token_usage, }) } @@ -837,6 +848,7 @@ mod tests { .expect("compaction should be collected"); assert_eq!(output.compaction_output, compaction); + assert_eq!(output.response_id, "resp-compact"); assert_eq!( output.token_usage, Some(TokenUsage { diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index a145d00ae6..5ebe128e3b 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -13,6 +13,8 @@ use crate::session::step_context::StepContext; use crate::session::turn::built_tools; use codex_protocol::error::Result as CodexResult; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RawResponseCompletedEvent; use codex_protocol::protocol::TokenUsage; use codex_rollout_trace::CompactionTraceContext; use tokio_util::sync::CancellationToken; @@ -114,8 +116,19 @@ pub(super) async fn run_remote_compact_v2_attempt( ); let RemoteCompactionV2Output { compaction_output, + response_id, token_usage, } = compaction_output_result?; + // TODO: Emit this before compaction output validation so malformed completed + // responses still surface their raw upstream usage. + sess.send_event( + turn_context, + EventMsg::RawResponseCompleted(RawResponseCompletedEvent { + response_id, + token_usage: token_usage.clone(), + }), + ) + .await; Ok(RemoteCompactV2Attempt { trace_input_history, prompt_input, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index fd68248076..2ecdaacc55 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -100,6 +100,7 @@ use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::PlanDeltaEvent; +use codex_protocol::protocol::RawResponseCompletedEvent; use codex_protocol::protocol::ReasoningContentDeltaEvent; use codex_protocol::protocol::ReasoningRawContentDeltaEvent; use codex_protocol::protocol::SafetyBufferingEvent; @@ -1591,6 +1592,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<(String, Option< | EventMsg::EnteredReviewMode(_) | EventMsg::ExitedReviewMode(_) | EventMsg::RawResponseItem(_) + | EventMsg::RawResponseCompleted(_) | EventMsg::ItemStarted(_) | EventMsg::HookStarted(_) | EventMsg::HookCompleted(_) @@ -2276,9 +2278,9 @@ async fn try_run_sampling_request( .await; } ResponseEvent::Completed { + response_id, token_usage, end_turn, - .. } => { flush_assistant_text_segments_all( &sess, @@ -2287,6 +2289,14 @@ async fn try_run_sampling_request( &mut assistant_message_stream_parsers, ) .await; + sess.send_event( + &turn_context, + EventMsg::RawResponseCompleted(RawResponseCompletedEvent { + response_id, + token_usage: token_usage.clone(), + }), + ) + .await; let budget_result = sess .record_token_usage_info(&turn_context, token_usage.as_ref()) .await; diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index cbc113ff79..ada17e92b7 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -358,6 +358,7 @@ async fn run_codex_tool_session_inner( | EventMsg::ImageGenerationEnd(_) | EventMsg::ViewImageToolCall(_) | EventMsg::RawResponseItem(_) + | EventMsg::RawResponseCompleted(_) | EventMsg::EnteredReviewMode(_) | EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 4f19670ad0..8ccdf59032 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1440,6 +1440,7 @@ pub enum EventMsg { ExitedReviewMode(ExitedReviewModeEvent), RawResponseItem(RawResponseItemEvent), + RawResponseCompleted(RawResponseCompletedEvent), ItemStarted(ItemStartedEvent), ItemCompleted(ItemCompletedEvent), @@ -1796,6 +1797,15 @@ pub struct RawResponseItemEvent { pub item: ResponseItem, } +/// Exact usage reported by one upstream Responses API completion. +/// +/// Unlike TokenCountEvent, this is not accumulated, estimated, or replayed. +#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)] +pub struct RawResponseCompletedEvent { + pub response_id: String, + pub token_usage: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)] pub struct ItemStartedEvent { pub thread_id: ThreadId, diff --git a/codex-rs/rollout-trace/src/protocol_event.rs b/codex-rs/rollout-trace/src/protocol_event.rs index 42e0db04d2..64b3824f37 100644 --- a/codex-rs/rollout-trace/src/protocol_event.rs +++ b/codex-rs/rollout-trace/src/protocol_event.rs @@ -395,6 +395,7 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option Option<&'static s | EventMsg::EnteredReviewMode(_) | EventMsg::ExitedReviewMode(_) | EventMsg::RawResponseItem(_) + | EventMsg::RawResponseCompleted(_) | EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) | EventMsg::HookStarted(_) diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 75ddec35ae..347bfd7c73 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -138,6 +138,7 @@ pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode) | EventMsg::TurnModerationMetadata(_) | EventMsg::AgentReasoningSectionBreak(_) | EventMsg::RawResponseItem(_) + | EventMsg::RawResponseCompleted(_) | EventMsg::SessionConfigured(_) | EventMsg::McpToolCallBegin(_) | EventMsg::ExecCommandBegin(_) diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 6c55850e4d..d71c382bd5 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -87,6 +87,9 @@ pub(super) fn server_notification_thread_target( ServerNotification::RawResponseItemCompleted(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::RawResponseCompleted(notification) => { + Some(notification.thread_id.as_str()) + } ServerNotification::AgentMessageDelta(notification) => { Some(notification.thread_id.as_str()) } diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index f60f9c41da..8c0ba14816 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -200,6 +200,7 @@ impl ChatWidget { | ServerNotification::ThreadDeleted(_) | ServerNotification::ThreadUnarchived(_) | ServerNotification::RawResponseItemCompleted(_) + | ServerNotification::RawResponseCompleted(_) | ServerNotification::CommandExecOutputDelta(_) | ServerNotification::ProcessOutputDelta(_) | ServerNotification::ProcessExited(_)