From 566f7bf6314cbf213de523a0268d8df89f93ef62 Mon Sep 17 00:00:00 2001 From: Francis Chalissery Date: Sun, 21 Jun 2026 20:39:14 -0700 Subject: [PATCH] Propagate safety buffering events to app-server clients (#29371) Responses API safety buffering metadata currently stops at the transport boundary, so app-server clients cannot render the in-progress safety review state. This change: - decodes and deduplicates `safety_buffering` metadata from Responses API SSE and WebSocket events without suppressing the original response event - emits a typed core event containing the requested model plus backend use cases and reasons - forwards that event as `turn/safetyBuffering/updated` through app-server v2 and updates generated protocol schemas - keeps the side-channel event out of persisted rollouts and turn timing This supports the Codex Apps buffering UX and depends on the Responses API backend work in https://github.com/openai/openai/pull/1044569 and https://github.com/openai/openai/pull/1044571. Validation: - focused `codex-core` safety-buffering integration test passes - `cargo check -p codex-core -p codex-app-server -p codex-app-server-protocol` - `just fix -p codex-api -p codex-protocol -p codex-core -p codex-app-server-protocol -p codex-app-server -p codex-rollout -p codex-rollout-trace -p codex-otel` - `just fmt` - broad package test run: 4,430/4,492 passed; 62 unrelated local-environment/concurrency failures involved unavailable test binaries, MCP subprocess setup, and app-server timeouts --- .../schema/json/ServerNotification.json | 53 ++++++++++++++ .../codex_app_server_protocol.schemas.json | 55 ++++++++++++++ .../codex_app_server_protocol.v2.schemas.json | 55 ++++++++++++++ ...delSafetyBufferingUpdatedNotification.json | 35 +++++++++ .../schema/typescript/ServerNotification.ts | 3 +- ...ModelSafetyBufferingUpdatedNotification.ts | 5 ++ .../schema/typescript/v2/index.ts | 1 + .../src/protocol/common.rs | 28 +++++++ .../src/protocol/v2/model.rs | 11 +++ codex-rs/app-server/README.md | 1 + .../app-server/src/bespoke_event_handling.rs | 15 ++++ codex-rs/codex-api/src/common.rs | 7 ++ .../src/endpoint/responses_websocket.rs | 11 +++ codex-rs/codex-api/src/sse/responses.rs | 73 +++++++++++++++++++ codex-rs/core/src/session/turn.rs | 13 ++++ codex-rs/core/src/turn_timing.rs | 1 + codex-rs/core/tests/suite/mod.rs | 1 + codex-rs/core/tests/suite/safety_buffering.rs | 63 ++++++++++++++++ codex-rs/mcp-server/src/codex_tool_runner.rs | 1 + codex-rs/otel/src/events/session_telemetry.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 + 25 files changed, 449 insertions(+), 1 deletion(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ModelSafetyBufferingUpdatedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ModelSafetyBufferingUpdatedNotification.ts create mode 100644 codex-rs/core/tests/suite/safety_buffering.rs diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 6ba699dce..f1fbee5c5 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2583,6 +2583,39 @@ ], "type": "object" }, + "ModelSafetyBufferingUpdatedNotification": { + "properties": { + "model": { + "type": "string" + }, + "reasons": { + "items": { + "type": "string" + }, + "type": "array" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "useCases": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "model", + "reasons", + "threadId", + "turnId", + "useCases" + ], + "type": "object" + }, "ModelVerification": { "enum": [ "trustedAccessForCyber" @@ -6608,6 +6641,26 @@ "title": "Turn/moderationMetadataNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "model/safetyBuffering/updated" + ], + "title": "Model/safetyBuffering/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ModelSafetyBufferingUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Model/safetyBuffering/updatedNotification", + "type": "object" + }, { "properties": { "method": { 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 f8f8db58b..bf7f0e5b3 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 @@ -5091,6 +5091,26 @@ "title": "Turn/moderationMetadataNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "model/safetyBuffering/updated" + ], + "title": "Model/safetyBuffering/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ModelSafetyBufferingUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Model/safetyBuffering/updatedNotification", + "type": "object" + }, { "properties": { "method": { @@ -12559,6 +12579,41 @@ "title": "ModelReroutedNotification", "type": "object" }, + "ModelSafetyBufferingUpdatedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "model": { + "type": "string" + }, + "reasons": { + "items": { + "type": "string" + }, + "type": "array" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "useCases": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "model", + "reasons", + "threadId", + "turnId", + "useCases" + ], + "title": "ModelSafetyBufferingUpdatedNotification", + "type": "object" + }, "ModelServiceTier": { "properties": { "description": { 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 81a79c0c9..7a080b56b 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 @@ -8983,6 +8983,41 @@ "title": "ModelReroutedNotification", "type": "object" }, + "ModelSafetyBufferingUpdatedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "model": { + "type": "string" + }, + "reasons": { + "items": { + "type": "string" + }, + "type": "array" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "useCases": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "model", + "reasons", + "threadId", + "turnId", + "useCases" + ], + "title": "ModelSafetyBufferingUpdatedNotification", + "type": "object" + }, "ModelServiceTier": { "properties": { "description": { @@ -13412,6 +13447,26 @@ "title": "Turn/moderationMetadataNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "model/safetyBuffering/updated" + ], + "title": "Model/safetyBuffering/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ModelSafetyBufferingUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Model/safetyBuffering/updatedNotification", + "type": "object" + }, { "properties": { "method": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ModelSafetyBufferingUpdatedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ModelSafetyBufferingUpdatedNotification.json new file mode 100644 index 000000000..5ed692480 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ModelSafetyBufferingUpdatedNotification.json @@ -0,0 +1,35 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "model": { + "type": "string" + }, + "reasons": { + "items": { + "type": "string" + }, + "type": "array" + }, + "threadId": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "useCases": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "model", + "reasons", + "threadId", + "turnId", + "useCases" + ], + "title": "ModelSafetyBufferingUpdatedNotification", + "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 82814111d..f3f828b8a 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -30,6 +30,7 @@ import type { McpServerOauthLoginCompletedNotification } from "./v2/McpServerOau import type { McpServerStatusUpdatedNotification } from "./v2/McpServerStatusUpdatedNotification"; import type { McpToolCallProgressNotification } from "./v2/McpToolCallProgressNotification"; import type { ModelReroutedNotification } from "./v2/ModelReroutedNotification"; +import type { ModelSafetyBufferingUpdatedNotification } from "./v2/ModelSafetyBufferingUpdatedNotification"; import type { ModelVerificationNotification } from "./v2/ModelVerificationNotification"; import type { PlanDeltaNotification } from "./v2/PlanDeltaNotification"; import type { ProcessExitedNotification } from "./v2/ProcessExitedNotification"; @@ -73,4 +74,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": "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": "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/ModelSafetyBufferingUpdatedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ModelSafetyBufferingUpdatedNotification.ts new file mode 100644 index 000000000..484f8c614 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ModelSafetyBufferingUpdatedNotification.ts @@ -0,0 +1,5 @@ +// 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. + +export type ModelSafetyBufferingUpdatedNotification = { threadId: string, turnId: string, model: string, useCases: Array, reasons: Array, }; 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 9b1bd3b6e..419cf8c3c 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -260,6 +260,7 @@ export type { ModelProviderCapabilitiesReadParams } from "./ModelProviderCapabil export type { ModelProviderCapabilitiesReadResponse } from "./ModelProviderCapabilitiesReadResponse"; export type { ModelRerouteReason } from "./ModelRerouteReason"; export type { ModelReroutedNotification } from "./ModelReroutedNotification"; +export type { ModelSafetyBufferingUpdatedNotification } from "./ModelSafetyBufferingUpdatedNotification"; export type { ModelServiceTier } from "./ModelServiceTier"; export type { ModelUpgradeInfo } from "./ModelUpgradeInfo"; export type { ModelVerification } from "./ModelVerification"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 2f322aa5f..eb3e6c82f 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1655,6 +1655,7 @@ server_notification_definitions! { ModelVerification => "model/verification" (v2::ModelVerificationNotification), #[experimental("turn/moderationMetadata")] TurnModerationMetadata => "turn/moderationMetadata" (v2::TurnModerationMetadataNotification), + ModelSafetyBufferingUpdated => "model/safetyBuffering/updated" (v2::ModelSafetyBufferingUpdatedNotification), Warning => "warning" (v2::WarningNotification), GuardianWarning => "guardianWarning" (v2::GuardianWarningNotification), DeprecationNotice => "deprecationNotice" (v2::DeprecationNoticeNotification), @@ -3347,6 +3348,33 @@ mod tests { Ok(()) } + #[test] + fn serialize_model_safety_buffering_updated_notification() -> Result<()> { + let notification = ServerNotification::ModelSafetyBufferingUpdated( + v2::ModelSafetyBufferingUpdatedNotification { + thread_id: "thr_123".to_string(), + turn_id: "turn_123".to_string(), + model: "gpt-5.4".to_string(), + use_cases: vec!["cyber".to_string()], + reasons: vec!["user_risk".to_string()], + }, + ); + assert_eq!( + json!({ + "method": "model/safetyBuffering/updated", + "params": { + "threadId": "thr_123", + "turnId": "turn_123", + "model": "gpt-5.4", + "useCases": ["cyber"], + "reasons": ["user_risk"] + } + }), + serde_json::to_value(¬ification)?, + ); + Ok(()) + } + #[test] fn serialize_thread_realtime_output_audio_delta_notification() -> Result<()> { let notification = ServerNotification::ThreadRealtimeOutputAudioDelta( diff --git a/codex-rs/app-server-protocol/src/protocol/v2/model.rs b/codex-rs/app-server-protocol/src/protocol/v2/model.rs index d8fa24958..2ce181902 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/model.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/model.rs @@ -162,3 +162,14 @@ pub struct TurnModerationMetadataNotification { pub turn_id: String, pub metadata: JsonValue, } + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ModelSafetyBufferingUpdatedNotification { + pub thread_id: String, + pub turn_id: String, + pub model: String, + pub use_cases: Vec, + pub reasons: Vec, +} diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index f8ceb69d4..8de6e3b8c 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1353,6 +1353,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`. +- `model/safetyBuffering/updated` — `{ threadId, turnId, model, useCases, reasons }` when a response enters safety buffering. 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`. - `turn/moderationMetadata` — experimental; `{ threadId, turnId, metadata }` when a first-party backend supplies turn-scoped moderation metadata for client-side presentation. diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index a77b0f466..bc6950506 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -40,6 +40,7 @@ use codex_app_server_protocol::McpServerElicitationRequestResponse; use codex_app_server_protocol::McpServerStartupState; use codex_app_server_protocol::McpServerStatusUpdatedNotification; use codex_app_server_protocol::ModelReroutedNotification; +use codex_app_server_protocol::ModelSafetyBufferingUpdatedNotification; use codex_app_server_protocol::ModelVerificationNotification; use codex_app_server_protocol::NetworkApprovalContext as V2NetworkApprovalContext; use codex_app_server_protocol::NetworkPolicyAmendment as V2NetworkPolicyAmendment; @@ -351,6 +352,20 @@ pub(crate) async fn apply_bespoke_event_handling( .send_server_notification(ServerNotification::TurnModerationMetadata(notification)) .await; } + EventMsg::SafetyBuffering(event) => { + let notification = ModelSafetyBufferingUpdatedNotification { + thread_id: conversation_id.to_string(), + turn_id: event_turn_id.clone(), + model: event.model, + use_cases: event.use_cases, + reasons: event.reasons, + }; + outgoing + .send_server_notification(ServerNotification::ModelSafetyBufferingUpdated( + notification, + )) + .await; + } EventMsg::RealtimeConversationStarted(event) => { let notification = ThreadRealtimeStartedNotification { thread_id: conversation_id.to_string(), diff --git a/codex-rs/codex-api/src/common.rs b/codex-rs/codex-api/src/common.rs index 17753e799..aeea85e7f 100644 --- a/codex-rs/codex-api/src/common.rs +++ b/codex-rs/codex-api/src/common.rs @@ -72,6 +72,7 @@ pub struct MemorySummarizeOutput { #[derive(Debug)] pub enum ResponseEvent { Created, + SafetyBuffering(SafetyBuffering), OutputItemDone(ResponseItem), OutputItemAdded(ResponseItem), /// Emitted when the server includes `OpenAI-Model` on the stream response. @@ -113,6 +114,12 @@ pub enum ResponseEvent { ModelsEtag(String), } +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct SafetyBuffering { + pub use_cases: Vec, + pub reasons: Vec, +} + #[derive(Debug, Serialize, Clone, PartialEq)] #[serde(rename_all = "snake_case")] pub enum ReasoningContext { diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 4ea564f13..fa0153010 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -690,6 +690,7 @@ async fn run_websocket_response_stream( } let model_verifications = event.model_verifications(); let turn_moderation_metadata = event.turn_moderation_metadata(); + let safety_buffering = event.safety_buffering(); if event.kind() == "codex.rate_limits" { if let Some(snapshot) = parse_rate_limit_event(&text) { let _ = tx_event.send(Ok(ResponseEvent::RateLimits(snapshot))).await; @@ -724,6 +725,16 @@ async fn run_websocket_response_stream( "response event consumer dropped".to_string(), )); } + if let Some(buffering) = safety_buffering + && tx_event + .send(Ok(ResponseEvent::SafetyBuffering(buffering))) + .await + .is_err() + { + return Err(ApiError::Stream( + "response event consumer dropped".to_string(), + )); + } match process_responses_event(event) { Ok(Some(event)) => { let is_completed = matches!(event, ResponseEvent::Completed { .. }); diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index a451f67af..0c20e1039 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -1,5 +1,6 @@ use crate::common::ResponseEvent; use crate::common::ResponseStream; +use crate::common::SafetyBuffering; use crate::error::ApiError; use crate::rate_limits::parse_all_rate_limits; use crate::telemetry::SseTelemetry; @@ -157,6 +158,7 @@ pub struct ResponsesStreamEvent { delta: Option, summary_index: Option, content_index: Option, + safety_buffering: Option, } impl ResponsesStreamEvent { @@ -217,6 +219,10 @@ impl ResponsesStreamEvent { .cloned() .map(|metadata| TurnModerationMetadataEvent { metadata }) } + + pub(crate) fn safety_buffering(&self) -> Option { + serde_json::from_value(self.safety_buffering.as_ref()?.clone()).ok() + } } fn header_openai_model_value_from_json(value: &Value) -> Option { @@ -480,6 +486,7 @@ pub async fn process_sse( }; let model_verifications = event.model_verifications(); let turn_moderation_metadata = event.turn_moderation_metadata(); + let safety_buffering = event.safety_buffering(); if let Some(model) = event.response_model() && last_server_model.as_deref() != Some(model.as_str()) @@ -509,6 +516,14 @@ pub async fn process_sse( { return; } + if let Some(buffering) = safety_buffering + && tx_event + .send(Ok(ResponseEvent::SafetyBuffering(buffering))) + .await + .is_err() + { + return; + } match process_responses_event(event) { Ok(Some(event)) => { @@ -1294,6 +1309,64 @@ mod tests { ); } + #[tokio::test] + async fn process_sse_emits_all_safety_buffering_notifications_without_dropping_response_events() + { + let events = run_sse(vec![ + json!({ + "type": "response.created", + "response": { "id": "resp-1" }, + "safety_buffering": false + }), + json!({ + "type": "response.output_text.delta", + "delta": "hello", + "safety_buffering": { + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + }), + json!({ + "type": "response.output_text.delta", + "delta": " world", + "safety_buffering": { + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + }), + json!({ + "type": "response.completed", + "response": { "id": "resp-1" }, + "safety_buffering": { + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + }), + ]) + .await; + + assert_eq!(events.len(), 7); + assert_matches!(&events[0], ResponseEvent::Created); + assert_matches!( + &events[1], + ResponseEvent::SafetyBuffering(buffering) + if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"] + ); + assert_matches!(&events[2], ResponseEvent::OutputTextDelta(delta) if delta == "hello"); + assert_matches!( + &events[3], + ResponseEvent::SafetyBuffering(buffering) + if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"] + ); + assert_matches!(&events[4], ResponseEvent::OutputTextDelta(delta) if delta == " world"); + assert_matches!( + &events[5], + ResponseEvent::SafetyBuffering(buffering) + if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"] + ); + assert_matches!(&events[6], ResponseEvent::Completed { response_id, .. } if response_id == "resp-1"); + } + #[test] fn responses_stream_event_response_model_reads_top_level_headers() { let ev: ResponsesStreamEvent = serde_json::from_value(json!({ diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 909048ce6..2630ba1cc 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::EventMsg; use codex_protocol::protocol::PlanDeltaEvent; use codex_protocol::protocol::ReasoningContentDeltaEvent; use codex_protocol::protocol::ReasoningRawContentDeltaEvent; +use codex_protocol::protocol::SafetyBufferingEvent; use codex_protocol::protocol::TurnDiffEvent; use codex_protocol::protocol::WarningEvent; use codex_protocol::user_input::UserInput; @@ -1527,6 +1528,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<(String, Option< | EventMsg::ModelReroute(_) | EventMsg::ModelVerification(_) | EventMsg::TurnModerationMetadata(_) + | EventMsg::SafetyBuffering(_) | EventMsg::ContextCompacted(_) | EventMsg::ThreadRolledBack(_) | EventMsg::TurnStarted(_) @@ -2199,6 +2201,17 @@ async fn try_run_sampling_request( sess.emit_turn_moderation_metadata(&turn_context, metadata) .await; } + ResponseEvent::SafetyBuffering(buffering) => { + sess.send_event( + &turn_context, + EventMsg::SafetyBuffering(SafetyBufferingEvent { + model: turn_context.model_info.slug.clone(), + use_cases: buffering.use_cases, + reasons: buffering.reasons, + }), + ) + .await; + } ResponseEvent::ServerReasoningIncluded(included) => { sess.set_server_reasoning_included(included).await; } diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 8a93a9183..e7a90800b 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -339,6 +339,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool { | ResponseEvent::ServerModel(_) | ResponseEvent::ModelVerifications(_) | ResponseEvent::TurnModerationMetadata(_) + | ResponseEvent::SafetyBuffering(_) | ResponseEvent::ServerReasoningIncluded(_) | ResponseEvent::ToolCallInputDelta { .. } | ResponseEvent::Completed { .. } diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 492f30979..a9298ec6d 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -104,6 +104,7 @@ mod review; mod rmcp_client; mod rollout_budget; mod rollout_list_find; +mod safety_buffering; mod safety_check_downgrade; mod search_tool; mod shell_command; diff --git a/codex-rs/core/tests/suite/safety_buffering.rs b/codex-rs/core/tests/suite/safety_buffering.rs new file mode 100644 index 000000000..6c2df1b6d --- /dev/null +++ b/codex-rs/core/tests/suite/safety_buffering.rs @@ -0,0 +1,63 @@ +use anyhow::Ok; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::SafetyBufferingEvent; +use codex_protocol::user_input::UserInput; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_sse_once; +use core_test_support::responses::sse; +use core_test_support::responses::start_mock_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn emits_safety_buffering_with_the_requested_model() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let mut created = ev_response_created("resp-1"); + created["safety_buffering"] = json!({ + "use_cases": ["cyber"], + "reasons": ["policy-check"], + }); + mount_sse_once(&server, sse(vec![created, ev_completed("resp-1")])).await; + + let test = test_codex().build(&server).await?; + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "Check this request".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + let event = wait_for_event_match(&test.codex, |event| match event { + EventMsg::SafetyBuffering(event) => Some(event.clone()), + _ => None, + }) + .await; + assert_eq!( + event, + SafetyBufferingEvent { + model: test.session_configured.model.clone(), + use_cases: vec!["cyber".to_string()], + reasons: vec!["policy-check".to_string()], + } + ); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + Ok(()) +} diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index d605b640c..cbc113ff7 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -266,6 +266,7 @@ async fn run_codex_tool_session_inner( EventMsg::Warning(_) | EventMsg::GuardianWarning(_) | EventMsg::ModelVerification(_) + | EventMsg::SafetyBuffering(_) | EventMsg::TurnModerationMetadata(_) => { continue; } diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 8f0471088..c8e35b9f7 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -1209,6 +1209,7 @@ impl SessionTelemetry { ResponseEvent::ServerModel(_) => "server_model".into(), ResponseEvent::ModelVerifications(_) => "model_verifications".into(), ResponseEvent::TurnModerationMetadata(_) => "turn_moderation_metadata".into(), + ResponseEvent::SafetyBuffering(_) => "safety_buffering".into(), ResponseEvent::ServerReasoningIncluded(_) => "server_reasoning_included".into(), ResponseEvent::RateLimits(_) => "rate_limits".into(), ResponseEvent::ModelsEtag(_) => "models_etag".into(), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 081649b7a..483618f06 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1251,6 +1251,9 @@ pub enum EventMsg { /// Backend moderation metadata intended for first-party turn presentation. TurnModerationMetadata(TurnModerationMetadataEvent), + /// Backend indicates that response output is waiting on a safety review. + SafetyBuffering(SafetyBufferingEvent), + /// Conversation history was compacted (either automatically or manually). ContextCompacted(ContextCompactedEvent), @@ -1929,6 +1932,13 @@ pub struct TurnModerationMetadataEvent { pub metadata: Value, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct SafetyBufferingEvent { + pub model: String, + pub use_cases: Vec, + pub reasons: Vec, +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct ContextCompactedEvent; diff --git a/codex-rs/rollout-trace/src/protocol_event.rs b/codex-rs/rollout-trace/src/protocol_event.rs index a998c98fc..42e0db04d 100644 --- a/codex-rs/rollout-trace/src/protocol_event.rs +++ b/codex-rs/rollout-trace/src/protocol_event.rs @@ -344,6 +344,7 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option Option<&'static s EventMsg::Warning(_) => Some("warning"), EventMsg::ShutdownComplete => Some("shutdown_complete"), EventMsg::GuardianWarning(_) + | EventMsg::SafetyBuffering(_) | EventMsg::RealtimeConversationStarted(_) | EventMsg::RealtimeConversationRealtime(_) | EventMsg::RealtimeConversationClosed(_) diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 169f4360d..b342d699b 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -121,6 +121,7 @@ pub fn should_persist_event_msg(ev: &EventMsg) -> bool { | EventMsg::RealtimeConversationSdp(_) | EventMsg::RealtimeConversationRealtime(_) | EventMsg::RealtimeConversationClosed(_) + | EventMsg::SafetyBuffering(_) | EventMsg::ModelReroute(_) | EventMsg::ModelVerification(_) | EventMsg::TurnModerationMetadata(_) 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 7e84e4227..212c45b4b 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -123,6 +123,9 @@ pub(super) fn server_notification_thread_target( ServerNotification::ModelVerification(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::ModelSafetyBufferingUpdated(notification) => { + Some(notification.thread_id.as_str()) + } ServerNotification::TurnModerationMetadata(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 e2cf31b03..085255a1c 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -208,6 +208,7 @@ impl ChatWidget { | ServerNotification::ExternalAgentConfigImportProgress(_) | ServerNotification::ExternalAgentConfigImportCompleted(_) | ServerNotification::FsChanged(_) + | ServerNotification::ModelSafetyBufferingUpdated(_) | ServerNotification::TurnModerationMetadata(_) | ServerNotification::FuzzyFileSearchSessionUpdated(_) | ServerNotification::FuzzyFileSearchSessionCompleted(_)