diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index c3b607eb66..6bc53a49fb 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2589,6 +2589,7 @@ dependencies = [ "ctor 0.6.3", "futures", "pretty_assertions", + "reqwest", "serde", "serde_json", "serial_test", @@ -2597,6 +2598,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-tungstenite", + "tokio-util", "tracing", "uuid", ] diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 167c637918..aa320f8b41 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -1704,6 +1704,23 @@ ], "type": "string" }, + "GuardianWarningNotification": { + "properties": { + "message": { + "description": "Concise guardian warning message for the user.", + "type": "string" + }, + "threadId": { + "description": "Thread target for the guardian warning.", + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "type": "object" + }, "HookCompletedNotification": { "properties": { "run": { @@ -5342,6 +5359,26 @@ "title": "WarningNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "guardianWarning" + ], + "title": "GuardianWarningNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/GuardianWarningNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "GuardianWarningNotification", + "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 fbc853f20b..4064e83115 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 @@ -4437,6 +4437,26 @@ "title": "WarningNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "guardianWarning" + ], + "title": "GuardianWarningNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/GuardianWarningNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "GuardianWarningNotification", + "type": "object" + }, { "properties": { "method": { @@ -9299,6 +9319,25 @@ ], "type": "string" }, + "GuardianWarningNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "message": { + "description": "Concise guardian warning message for the user.", + "type": "string" + }, + "threadId": { + "description": "Thread target for the guardian warning.", + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "title": "GuardianWarningNotification", + "type": "object" + }, "HookCompletedNotification": { "$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 3fc4f39199..7c4e7d99f3 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 @@ -6017,6 +6017,25 @@ ], "type": "string" }, + "GuardianWarningNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "message": { + "description": "Concise guardian warning message for the user.", + "type": "string" + }, + "threadId": { + "description": "Thread target for the guardian warning.", + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "title": "GuardianWarningNotification", + "type": "object" + }, "HookCompletedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -10797,6 +10816,26 @@ "title": "WarningNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "guardianWarning" + ], + "title": "GuardianWarningNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/GuardianWarningNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "GuardianWarningNotification", + "type": "object" + }, { "properties": { "method": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/GuardianWarningNotification.json b/codex-rs/app-server-protocol/schema/json/v2/GuardianWarningNotification.json new file mode 100644 index 0000000000..5a4ef82f0a --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/GuardianWarningNotification.json @@ -0,0 +1,19 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "message": { + "description": "Concise guardian warning message for the user.", + "type": "string" + }, + "threadId": { + "description": "Thread target for the guardian warning.", + "type": "string" + } + }, + "required": [ + "message", + "threadId" + ], + "title": "GuardianWarningNotification", + "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 f55700bf54..614dd611d2 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -18,6 +18,7 @@ import type { ExternalAgentConfigImportCompletedNotification } from "./v2/Extern import type { FileChangeOutputDeltaNotification } from "./v2/FileChangeOutputDeltaNotification"; import type { FileChangePatchUpdatedNotification } from "./v2/FileChangePatchUpdatedNotification"; import type { FsChangedNotification } from "./v2/FsChangedNotification"; +import type { GuardianWarningNotification } from "./v2/GuardianWarningNotification"; import type { HookCompletedNotification } from "./v2/HookCompletedNotification"; import type { HookStartedNotification } from "./v2/HookStartedNotification"; import type { ItemCompletedNotification } from "./v2/ItemCompletedNotification"; @@ -62,4 +63,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/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "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": "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": "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": "warning", "params": WarningNotification } | { "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/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "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": "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": "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": "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/GuardianWarningNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/GuardianWarningNotification.ts new file mode 100644 index 0000000000..1659f62f1b --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/GuardianWarningNotification.ts @@ -0,0 +1,13 @@ +// 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 GuardianWarningNotification = { +/** + * Thread target for the guardian warning. + */ +threadId: string, +/** + * Concise guardian warning message for the user. + */ +message: string, }; 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 837b265fc8..aa0ed36d25 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -147,6 +147,7 @@ export type { GuardianApprovalReviewStatus } from "./GuardianApprovalReviewStatu export type { GuardianCommandSource } from "./GuardianCommandSource"; export type { GuardianRiskLevel } from "./GuardianRiskLevel"; export type { GuardianUserAuthorization } from "./GuardianUserAuthorization"; +export type { GuardianWarningNotification } from "./GuardianWarningNotification"; export type { HookCompletedNotification } from "./HookCompletedNotification"; export type { HookEventName } from "./HookEventName"; export type { HookExecutionMode } from "./HookExecutionMode"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 503cb6bc17..a751f11ec0 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1061,6 +1061,7 @@ server_notification_definitions! { ContextCompacted => "thread/compacted" (v2::ContextCompactedNotification), ModelRerouted => "model/rerouted" (v2::ModelReroutedNotification), Warning => "warning" (v2::WarningNotification), + GuardianWarning => "guardianWarning" (v2::GuardianWarningNotification), DeprecationNotice => "deprecationNotice" (v2::DeprecationNoticeNotification), ConfigWarning => "configWarning" (v2::ConfigWarningNotification), FuzzyFileSearchSessionUpdated => "fuzzyFileSearch/sessionUpdated" (FuzzyFileSearchSessionUpdatedNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index c2dfbcc6b9..76825877e0 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -7273,6 +7273,16 @@ pub struct WarningNotification { pub message: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct GuardianWarningNotification { + /// Thread target for the guardian warning. + pub thread_id: String, + /// Concise guardian warning message for the user. + pub message: String, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 12129f7ee5..4730d31679 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -45,6 +45,7 @@ use codex_app_server_protocol::FileChangeRequestApprovalParams; use codex_app_server_protocol::FileChangeRequestApprovalResponse; use codex_app_server_protocol::FileUpdateChange; use codex_app_server_protocol::GrantedPermissionProfile as V2GrantedPermissionProfile; +use codex_app_server_protocol::GuardianWarningNotification; use codex_app_server_protocol::HookCompletedNotification; use codex_app_server_protocol::HookStartedNotification; use codex_app_server_protocol::InterruptConversationResponse; @@ -286,6 +287,22 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } } + EventMsg::GuardianWarning(warning_event) => { + if let ApiVersion::V2 = api_version { + let notification = GuardianWarningNotification { + thread_id: conversation_id.to_string(), + message: warning_event.message, + }; + if let Some(analytics_events_client) = analytics_events_client.as_ref() { + analytics_events_client.track_notification( + ServerNotification::GuardianWarning(notification.clone()), + ); + } + outgoing + .send_server_notification(ServerNotification::GuardianWarning(notification)) + .await; + } + } EventMsg::GuardianAssessment(assessment) => { if let ApiVersion::V2 = api_version { let pending_command_execution = match build_item_from_guardian_event( diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 3c5bef3ea4..e3c92f70b5 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -657,6 +657,7 @@ mod tests { use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::DynamicToolCallParams; use codex_app_server_protocol::FileChangeRequestApprovalParams; + use codex_app_server_protocol::GuardianWarningNotification; use codex_app_server_protocol::ModelRerouteReason; use codex_app_server_protocol::ModelReroutedNotification; use codex_app_server_protocol::RateLimitSnapshot; @@ -812,6 +813,28 @@ mod tests { ); } + #[test] + fn verify_guardian_warning_notification_serialization() { + let notification = ServerNotification::GuardianWarning(GuardianWarningNotification { + thread_id: "thread-1".to_string(), + message: "Automatic approval review denied the requested action.".to_string(), + }); + + let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification); + assert_eq!( + json!({ + "method": "guardianWarning", + "params": { + "threadId": "thread-1", + "message": "Automatic approval review denied the requested action.", + }, + }), + serde_json::to_value(jsonrpc_notification) + .expect("ensure the notification serializes correctly"), + "ensure the notification serializes correctly" + ); + } + #[test] fn verify_model_rerouted_notification_serialization() { let notification = ServerNotification::ModelRerouted(ModelReroutedNotification { diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index a2c9b45c8d..155db6f1f4 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -30,6 +30,8 @@ pub(crate) use review::guardian_rejection_message; pub(crate) use review::guardian_timeout_message; pub(crate) use review::is_guardian_reviewer_source; pub(crate) use review::new_guardian_review_id; +#[cfg(test)] +pub(crate) use review::record_guardian_denial_for_test; pub(crate) use review::review_approval_request; #[cfg(test)] pub(crate) use review::review_approval_request_with_cancel; @@ -40,6 +42,8 @@ pub(crate) use review_session::GuardianReviewSessionManager; const GUARDIAN_PREFERRED_MODEL: &str = "codex-auto-review"; pub(crate) const GUARDIAN_REVIEW_TIMEOUT: Duration = Duration::from_secs(90); pub(crate) const GUARDIAN_REVIEWER_NAME: &str = "guardian"; +pub(crate) const MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN: u32 = 3; +pub(crate) const MAX_TOTAL_GUARDIAN_DENIALS_PER_TURN: u32 = 10; const GUARDIAN_MAX_MESSAGE_TRANSCRIPT_TOKENS: usize = 10_000; const GUARDIAN_MAX_TOOL_TRANSCRIPT_TOKENS: usize = 10_000; const GUARDIAN_MAX_MESSAGE_ENTRY_TOKENS: usize = 2_000; @@ -63,6 +67,56 @@ pub(crate) struct GuardianRejection { pub(crate) source: GuardianAssessmentDecisionSource, } +#[derive(Debug, Default)] +pub(crate) struct GuardianRejectionCircuitBreaker { + turns: std::collections::HashMap, +} + +#[derive(Debug, Default)] +struct GuardianRejectionCircuitBreakerTurn { + consecutive_denials: u32, + total_denials: u32, + interrupt_triggered: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum GuardianRejectionCircuitBreakerAction { + Continue, + InterruptTurn { + consecutive_denials: u32, + total_denials: u32, + }, +} + +impl GuardianRejectionCircuitBreaker { + pub(crate) fn clear_turn(&mut self, turn_id: &str) { + self.turns.remove(turn_id); + } + + pub(crate) fn record_denial(&mut self, turn_id: &str) -> GuardianRejectionCircuitBreakerAction { + let turn = self.turns.entry(turn_id.to_string()).or_default(); + turn.consecutive_denials = turn.consecutive_denials.saturating_add(1); + turn.total_denials = turn.total_denials.saturating_add(1); + if !turn.interrupt_triggered + && (turn.consecutive_denials >= MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN + || turn.total_denials >= MAX_TOTAL_GUARDIAN_DENIALS_PER_TURN) + { + turn.interrupt_triggered = true; + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: turn.consecutive_denials, + total_denials: turn.total_denials, + } + } else { + GuardianRejectionCircuitBreakerAction::Continue + } + } + + pub(crate) fn record_non_denial(&mut self, turn_id: &str) { + let turn = self.turns.entry(turn_id.to_string()).or_default(); + turn.consecutive_denials = 0; + } +} + #[cfg(test)] use approval_request::format_guardian_action_pretty; #[cfg(test)] diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index ce01ac970e..3b5c63b4dc 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -17,6 +17,7 @@ use codex_protocol::protocol::GuardianRiskLevel; use codex_protocol::protocol::GuardianUserAuthorization; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::WarningEvent; use tokio::sync::oneshot; use tokio_util::sync::CancellationToken; @@ -30,6 +31,7 @@ use super::GuardianApprovalRequest; use super::GuardianAssessment; use super::GuardianAssessmentOutcome; use super::GuardianRejection; +use super::GuardianRejectionCircuitBreakerAction; use super::approval_request::guardian_assessment_action; use super::approval_request::guardian_request_target_item_id; use super::approval_request::guardian_request_turn_id; @@ -172,6 +174,64 @@ fn track_guardian_review( .track_guardian_review(tracking, result); } +async fn record_guardian_non_denial(session: &Arc, turn_id: &str) { + session + .services + .guardian_rejection_circuit_breaker + .lock() + .await + .record_non_denial(turn_id); +} + +async fn record_guardian_denial(session: &Arc, turn: &Arc, turn_id: &str) { + let action = session + .services + .guardian_rejection_circuit_breaker + .lock() + .await + .record_denial(turn_id); + let GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials, + total_denials, + } = action + else { + return; + }; + + if session.turn_context_for_sub_id(turn_id).await.is_none() { + return; + } + + session + .send_event( + turn.as_ref(), + EventMsg::GuardianWarning(WarningEvent { + message: format!( + "Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {total_denials} total); interrupting the turn." + ), + }), + ) + .await; + + let runtime_handle = session.services.runtime_handle.clone(); + let session = Arc::clone(session); + let turn_id = turn_id.to_string(); + let _abort_task = runtime_handle.spawn(async move { + session + .abort_turn_if_active(&turn_id, TurnAbortReason::Interrupted) + .await; + }); +} + +#[cfg(test)] +pub(crate) async fn record_guardian_denial_for_test( + session: &Arc, + turn: &Arc, + turn_id: &str, +) { + record_guardian_denial(session, turn, turn_id).await; +} + /// This function always fails closed: timeouts, review-session failures, and /// parse failures all block execution, but timeouts are still surfaced to the /// caller as distinct from explicit guardian denials. @@ -234,7 +294,7 @@ async fn run_guardian_review( EventMsg::GuardianAssessment(GuardianAssessmentEvent { id: review_id, target_item_id, - turn_id: assessment_turn_id, + turn_id: assessment_turn_id.clone(), status: GuardianAssessmentStatus::Aborted, risk_level: None, user_authorization: None, @@ -244,6 +304,7 @@ async fn run_guardian_review( }), ) .await; + record_guardian_non_denial(&session, &assessment_turn_id).await; return ReviewDecision::Abort; } @@ -259,7 +320,7 @@ async fn run_guardian_review( )) .await; - let assessment = match outcome { + let (assessment, count_denial_for_circuit_breaker) = match outcome { GuardianReviewOutcome::Completed(assessment) => { let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow); track_guardian_review( @@ -284,7 +345,9 @@ async fn run_guardian_review( ..analytics_result }, ); - assessment + let count_denial_for_circuit_breaker = + matches!(assessment.outcome, GuardianAssessmentOutcome::Deny); + (assessment, count_denial_for_circuit_breaker) } GuardianReviewOutcome::Error(error) => match error { GuardianReviewError::Timeout => { @@ -305,7 +368,7 @@ async fn run_guardian_review( session .send_event( turn.as_ref(), - EventMsg::Warning(WarningEvent { + EventMsg::GuardianWarning(WarningEvent { message: rationale.clone(), }), ) @@ -316,7 +379,7 @@ async fn run_guardian_review( EventMsg::GuardianAssessment(GuardianAssessmentEvent { id: review_id, target_item_id, - turn_id: assessment_turn_id, + turn_id: assessment_turn_id.clone(), status: GuardianAssessmentStatus::TimedOut, risk_level: None, user_authorization: None, @@ -326,6 +389,7 @@ async fn run_guardian_review( }), ) .await; + record_guardian_non_denial(&session, &assessment_turn_id).await; return ReviewDecision::TimedOut; } GuardianReviewError::Cancelled => { @@ -346,7 +410,7 @@ async fn run_guardian_review( EventMsg::GuardianAssessment(GuardianAssessmentEvent { id: review_id, target_item_id, - turn_id: assessment_turn_id, + turn_id: assessment_turn_id.clone(), status: GuardianAssessmentStatus::Aborted, risk_level: None, user_authorization: None, @@ -356,6 +420,7 @@ async fn run_guardian_review( }), ) .await; + record_guardian_non_denial(&session, &assessment_turn_id).await; return ReviewDecision::Abort; } GuardianReviewError::PromptBuild { .. } @@ -381,12 +446,15 @@ async fn run_guardian_review( ..analytics_result }, ); - GuardianAssessment { - risk_level: GuardianRiskLevel::High, - user_authorization: GuardianUserAuthorization::Unknown, - outcome: GuardianAssessmentOutcome::Deny, - rationale, - } + ( + GuardianAssessment { + risk_level: GuardianRiskLevel::High, + user_authorization: GuardianUserAuthorization::Unknown, + outcome: GuardianAssessmentOutcome::Deny, + rationale, + }, + false, + ) } }, }; @@ -410,7 +478,7 @@ async fn run_guardian_review( session .send_event( turn.as_ref(), - EventMsg::Warning(WarningEvent { message: warning }), + EventMsg::GuardianWarning(WarningEvent { message: warning }), ) .await; let status = if approved { @@ -436,7 +504,7 @@ async fn run_guardian_review( EventMsg::GuardianAssessment(GuardianAssessmentEvent { id: review_id, target_item_id, - turn_id: assessment_turn_id, + turn_id: assessment_turn_id.clone(), status, risk_level: Some(assessment.risk_level), user_authorization: Some(assessment.user_authorization), @@ -447,6 +515,12 @@ async fn run_guardian_review( ) .await; + if count_denial_for_circuit_breaker { + record_guardian_denial(&session, &turn, &assessment_turn_id).await; + } else { + record_guardian_non_denial(&session, &assessment_turn_id).await; + } + if approved { ReviewDecision::Approved } else { diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 02ff93dd06..cf29c54112 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -67,6 +67,74 @@ fn fixed_guardian_parent_session_id() -> ThreadId { .expect("fixed parent session id should be a valid UUID") } +#[test] +fn guardian_rejection_circuit_breaker_interrupts_after_three_consecutive_denials() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 3, + total_denials: 3, + } + ); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); +} + +#[test] +fn guardian_rejection_circuit_breaker_resets_consecutive_denials_on_non_denial() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + circuit_breaker.record_non_denial("turn-1"); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 3, + total_denials: 4, + } + ); +} + +#[test] +fn guardian_rejection_circuit_breaker_interrupts_after_ten_total_denials() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + for _ in 0..9 { + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::Continue + ); + circuit_breaker.record_non_denial("turn-1"); + } + assert_eq!( + circuit_breaker.record_denial("turn-1"), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 1, + total_denials: 10, + } + ); +} + async fn guardian_test_session_and_turn( server: &wiremock::MockServer, ) -> (Arc, Arc) { @@ -739,7 +807,7 @@ async fn cancelled_guardian_review_emits_terminal_abort_without_warning() { while let Ok(event) = rx.try_recv() { match event.msg { EventMsg::GuardianAssessment(event) => guardian_statuses.push(event.status), - EventMsg::Warning(event) => warnings.push(event.message), + EventMsg::GuardianWarning(event) => warnings.push(event.message), _ => {} } } @@ -1491,7 +1559,7 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() -> let mut denial_rationales = Vec::new(); while let Ok(event) = rx.try_recv() { match event.msg { - EventMsg::Warning(event) => warnings.push(event.message), + EventMsg::GuardianWarning(event) => warnings.push(event.message), EventMsg::GuardianAssessment(event) if event.status == GuardianAssessmentStatus::Denied => { diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 33714dbec7..e3f7d1b6c2 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -730,6 +730,8 @@ impl Session { models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejections: Mutex::new(HashMap::new()), + guardian_rejection_circuit_breaker: Mutex::new(Default::default()), + runtime_handle: tokio::runtime::Handle::current(), skills_manager, plugins_manager: Arc::clone(&plugins_manager), mcp_manager: Arc::clone(&mcp_manager), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 41283e5763..075b851a68 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -3143,6 +3143,8 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejections: Mutex::new(std::collections::HashMap::new()), + guardian_rejection_circuit_breaker: Mutex::new(Default::default()), + runtime_handle: tokio::runtime::Handle::current(), skills_manager, plugins_manager, mcp_manager, @@ -4387,6 +4389,8 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), guardian_rejections: Mutex::new(std::collections::HashMap::new()), + guardian_rejection_circuit_breaker: Mutex::new(Default::default()), + runtime_handle: tokio::runtime::Handle::current(), skills_manager, plugins_manager, mcp_manager, @@ -5627,6 +5631,125 @@ impl SessionTask for NeverEndingTask { } } +#[derive(Clone, Copy)] +struct GuardianDeniedApprovalTask; + +impl SessionTask for GuardianDeniedApprovalTask { + fn kind(&self) -> TaskKind { + TaskKind::Regular + } + + fn span_name(&self) -> &'static str { + "session_task.guardian_denied_approval" + } + + async fn run( + self: Arc, + session: Arc, + ctx: Arc, + _input: Vec, + cancellation_token: CancellationToken, + ) -> Option { + let session = session.clone_session(); + for _ in 0..3 { + crate::guardian::record_guardian_denial_for_test(&session, &ctx, &ctx.sub_id).await; + } + + cancellation_token.cancelled().await; + None + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_auto_review_interrupts_after_three_consecutive_denials() { + let (sess, tc, rx) = make_session_and_context_with_rx().await; + let input = vec![UserInput::Text { + text: "trigger guardian denials".to_string(), + text_elements: Vec::new(), + }]; + sess.spawn_task(Arc::clone(&tc), input, GuardianDeniedApprovalTask) + .await; + + let mut observed = Vec::new(); + let aborted = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let event = rx.recv().await.expect("event"); + if let EventMsg::TurnAborted(event) = &event.msg { + let event = event.clone(); + observed.push(EventMsg::TurnAborted(event.clone())); + break event; + } + observed.push(event.msg); + } + }) + .await + .unwrap_or_else(|_| { + panic!( + "guardian denial circuit breaker should interrupt the turn; observed events: {observed:?}" + ) + }); + assert_eq!(aborted.reason, TurnAbortReason::Interrupted); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_helper_review_interrupts_after_three_consecutive_denials() { + let (sess, tc, rx) = make_session_and_context_with_rx().await; + let input = vec![UserInput::Text { + text: "keep turn active for helper reviews".to_string(), + text_elements: Vec::new(), + }]; + sess.spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + let session_for_review = Arc::clone(&sess); + let turn_for_review = Arc::clone(&tc); + let turn_id = tc.sub_id.clone(); + let review_thread = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("helper review runtime"); + runtime.block_on(async move { + for _ in 0..3 { + crate::guardian::record_guardian_denial_for_test( + &session_for_review, + &turn_for_review, + &turn_id, + ) + .await; + } + }); + }); + review_thread.join().expect("helper review thread"); + + let mut observed = Vec::new(); + let aborted = timeout(StdDuration::from_secs(5), async { + loop { + let event = rx.recv().await.expect("event"); + if let EventMsg::TurnAborted(event) = &event.msg { + let event = event.clone(); + observed.push(EventMsg::TurnAborted(event.clone())); + break event; + } + observed.push(event.msg); + } + }) + .await + .unwrap_or_else(|_| { + panic!( + "helper review circuit breaker should interrupt the turn; observed events: {observed:?}" + ) + }); + assert_eq!(aborted.reason, TurnAbortReason::Interrupted); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[test_log::test] async fn abort_regular_task_emits_turn_aborted_only() { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 780cee5f87..15b48f3922 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1484,6 +1484,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { }, EventMsg::Error(_) | EventMsg::Warning(_) + | EventMsg::GuardianWarning(_) | EventMsg::RealtimeConversationStarted(_) | EventMsg::RealtimeConversationSdp(_) | EventMsg::RealtimeConversationRealtime(_) diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 29f4433ce0..f93a44d7fe 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -8,6 +8,7 @@ use crate::client::ModelClient; use crate::config::StartedNetworkProxy; use crate::exec_policy::ExecPolicyManager; use crate::guardian::GuardianRejection; +use crate::guardian::GuardianRejectionCircuitBreaker; use crate::mcp::McpManager; use crate::plugins::PluginsManager; use crate::skills_watcher::SkillsWatcher; @@ -26,6 +27,7 @@ use codex_rollout::state_db::StateDbHandle; use codex_rollout_trace::RolloutTraceRecorder; use codex_thread_store::LocalThreadStore; use std::path::PathBuf; +use tokio::runtime::Handle; use tokio::sync::Mutex; use tokio::sync::RwLock; use tokio::sync::watch; @@ -52,6 +54,8 @@ pub(crate) struct SessionServices { pub(crate) session_telemetry: SessionTelemetry, pub(crate) tool_approvals: Mutex, pub(crate) guardian_rejections: Mutex>, + pub(crate) guardian_rejection_circuit_breaker: Mutex, + pub(crate) runtime_handle: Handle, pub(crate) skills_manager: Arc, pub(crate) plugins_manager: Arc, pub(crate) mcp_manager: Arc, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index d9dfeb7db4..b9621d8fab 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -269,6 +269,12 @@ impl Session { let cancellation_token = CancellationToken::new(); let done = Arc::new(Notify::new()); + self.services + .guardian_rejection_circuit_breaker + .lock() + .await + .clear_turn(&turn_context.sub_id); + let queued_response_items = self.take_queued_response_items_for_next_turn().await; let mailbox_items = self.get_pending_input().await; let turn_state = { @@ -410,6 +416,40 @@ impl Session { } } + pub(crate) async fn abort_turn_if_active( + self: &Arc, + turn_id: &str, + reason: TurnAbortReason, + ) -> bool { + let active_turn = { + let mut active = self.active_turn.lock().await; + if active + .as_ref() + .is_some_and(|active_turn| active_turn.tasks.contains_key(turn_id)) + { + active.take() + } else { + None + } + }; + let Some(mut active_turn) = active_turn else { + return false; + }; + + for task in active_turn.drain_tasks() { + self.handle_task_abort(task, reason.clone()).await; + } + // Let interrupted tasks observe cancellation before dropping pending approvals, or an + // in-flight approval wait can surface as a model-visible rejection before TurnAborted. + active_turn.clear_pending().await; + + if reason == TurnAbortReason::Interrupted { + self.maybe_start_turn_for_pending_work().await; + } + + true + } + pub async fn on_task_finished( self: &Arc, turn_context: Arc, @@ -568,6 +608,11 @@ impl Session { time_to_first_token_ms, }); self.send_event(turn_context.as_ref(), event).await; + self.services + .guardian_rejection_circuit_breaker + .lock() + .await + .clear_turn(&turn_context.sub_id); if should_clear_active_turn { let session = Arc::clone(self); @@ -654,6 +699,11 @@ impl Session { duration_ms, }); self.send_event(task.turn_context.as_ref(), event).await; + self.services + .guardian_rejection_circuit_breaker + .lock() + .await + .clear_turn(&task.turn_context.sub_id); } } diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index 5ca265c6b6..9688063278 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/codex-rs/exec-server/Cargo.toml @@ -21,6 +21,7 @@ codex-sandboxing = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-pty = { workspace = true } futures = { workspace = true } +reqwest = { workspace = true, features = ["rustls-tls", "stream"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } thiserror = { workspace = true } @@ -35,6 +36,7 @@ tokio = { workspace = true, features = [ "sync", "time", ] } +tokio-util = { workspace = true, features = ["rt"] } tokio-tungstenite = { workspace = true } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/exec-server/src/client/http_client.rs b/codex-rs/exec-server/src/client/http_client.rs index a02a8be9a3..1e91fa4489 100644 --- a/codex-rs/exec-server/src/client/http_client.rs +++ b/codex-rs/exec-server/src/client/http_client.rs @@ -1,7 +1,15 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::Ordering; +use std::time::Duration; +use codex_app_server_protocol::JSONRPCErrorError; +use futures::StreamExt; +use reqwest::Method; +use reqwest::Url; +use reqwest::header::HeaderMap; +use reqwest::header::HeaderName; +use reqwest::header::HeaderValue; use serde_json::Value; use serde_json::from_value; use tokio::runtime::Handle; @@ -12,14 +20,28 @@ use tracing::debug; use super::ExecServerClient; use super::ExecServerError; use super::Inner; +use crate::protocol::HTTP_REQUEST_BODY_DELTA_METHOD; use crate::protocol::HTTP_REQUEST_METHOD; +use crate::protocol::HttpHeader; use crate::protocol::HttpRequestBodyDeltaNotification; use crate::protocol::HttpRequestParams; use crate::protocol::HttpRequestResponse; +use crate::rpc::RpcNotificationSender; +use crate::rpc::internal_error; +use crate::rpc::invalid_params; /// Maximum queued body frames per streamed executor HTTP response. const HTTP_BODY_DELTA_CHANNEL_CAPACITY: usize = 256; +pub(crate) struct ExecutorPendingHttpBodyStream { + pub(crate) request_id: String, + response: reqwest::Response, +} + +pub(crate) struct ExecutorHttpRequestRunner { + client: reqwest::Client, +} + /// Request-scoped stream of body chunks for an executor HTTP response. /// /// The initial `http/request` call returns status and headers. This stream then @@ -35,6 +57,60 @@ pub struct HttpResponseBodyStream { closed: bool, } +impl ExecServerClient { + /// Performs an executor-side HTTP request and buffers the response body. + pub async fn http_request( + &self, + mut params: HttpRequestParams, + ) -> Result { + params.stream_response = false; + self.call(HTTP_REQUEST_METHOD, ¶ms).await + } + + /// Performs an executor-side HTTP request and returns a body stream. + /// + /// The method sets `stream_response` and replaces any caller-supplied + /// `request_id` with a connection-local id, so late deltas from abandoned + /// streams cannot be confused with later requests. + pub async fn http_request_stream( + &self, + mut params: HttpRequestParams, + ) -> Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError> { + params.stream_response = true; + let request_id = self.inner.next_http_body_stream_request_id(); + params.request_id = request_id.clone(); + let (tx, rx) = mpsc::channel(HTTP_BODY_DELTA_CHANNEL_CAPACITY); + self.inner + .insert_http_body_stream(request_id.clone(), tx) + .await?; + let mut registration = HttpBodyStreamRegistration { + inner: Arc::clone(&self.inner), + request_id: request_id.clone(), + active: true, + }; + let response = match self.call(HTTP_REQUEST_METHOD, ¶ms).await { + Ok(response) => response, + Err(error) => { + self.inner.remove_http_body_stream(&request_id).await; + registration.active = false; + return Err(error); + } + }; + registration.active = false; + Ok(( + response, + HttpResponseBodyStream { + inner: Arc::clone(&self.inner), + request_id, + next_seq: 1, + rx, + pending_eof: false, + closed: false, + }, + )) + } +} + impl HttpResponseBodyStream { /// Receives the next response-body chunk. /// @@ -109,75 +185,165 @@ impl Drop for HttpResponseBodyStream { } } -/// Active route registration owned while `http_request_stream` awaits headers. -struct HttpBodyStreamRegistration { - inner: Arc, - request_id: String, - active: bool, -} - -impl Drop for HttpBodyStreamRegistration { - /// Removes the route if the stream request future is cancelled before headers return. - fn drop(&mut self) { - if self.active { - spawn_remove_http_body_stream(Arc::clone(&self.inner), self.request_id.clone()); - } - } -} - -impl ExecServerClient { - /// Performs an executor-side HTTP request and buffers the response body. - pub async fn http_request( - &self, - mut params: HttpRequestParams, - ) -> Result { - params.stream_response = false; - params.request_id = None; - self.call(HTTP_REQUEST_METHOD, ¶ms).await - } - - /// Performs an executor-side HTTP request and returns a body stream. - /// - /// The method sets `stream_response` and replaces any caller-supplied - /// `request_id` with a connection-local id, so late deltas from abandoned - /// streams cannot be confused with later requests. - pub async fn http_request_stream( - &self, - mut params: HttpRequestParams, - ) -> Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError> { - params.stream_response = true; - let request_id = self.inner.next_http_body_stream_request_id(); - params.request_id = Some(request_id.clone()); - let (tx, rx) = mpsc::channel(HTTP_BODY_DELTA_CHANNEL_CAPACITY); - self.inner - .insert_http_body_stream(request_id.clone(), tx) - .await?; - let mut registration = HttpBodyStreamRegistration { - inner: Arc::clone(&self.inner), - request_id: request_id.clone(), - active: true, - }; - let response = match self.call(HTTP_REQUEST_METHOD, ¶ms).await { - Ok(response) => response, - Err(error) => { - self.inner.remove_http_body_stream(&request_id).await; - registration.active = false; - return Err(error); +impl ExecutorHttpRequestRunner { + pub(crate) fn new(timeout_ms: Option) -> Result { + let client = match timeout_ms { + None => reqwest::Client::builder(), + Some(timeout_ms) => { + reqwest::Client::builder().timeout(Duration::from_millis(timeout_ms)) } - }; - registration.active = false; + } + .build() + .map_err(|err| internal_error(format!("failed to build http/request client: {err}")))?; + Ok(Self { client }) + } + + pub(crate) async fn run( + &self, + params: HttpRequestParams, + ) -> Result<(HttpRequestResponse, Option), JSONRPCErrorError> + { + let method = Method::from_bytes(params.method.as_bytes()) + .map_err(|err| invalid_params(format!("http/request method is invalid: {err}")))?; + let url = Url::parse(¶ms.url) + .map_err(|err| invalid_params(format!("http/request url is invalid: {err}")))?; + match url.scheme() { + "http" | "https" => {} + scheme => { + return Err(invalid_params(format!( + "http/request only supports http and https URLs, got {scheme}" + ))); + } + } + + let headers = Self::build_headers(params.headers)?; + let mut request = self.client.request(method, url).headers(headers); + if let Some(body) = params.body { + request = request.body(body.into_inner()); + } + + let response = request + .send() + .await + .map_err(|err| internal_error(format!("http/request failed: {err}")))?; + let status = response.status().as_u16(); + let headers = Self::response_headers(response.headers()); + + if params.stream_response { + return Ok(( + HttpRequestResponse { + status, + headers, + body: Vec::new().into(), + }, + Some(ExecutorPendingHttpBodyStream { + request_id: params.request_id, + response, + }), + )); + } + + let body = response.bytes().await.map_err(|err| { + internal_error(format!("failed to read http/request response body: {err}")) + })?; + Ok(( - response, - HttpResponseBodyStream { - inner: Arc::clone(&self.inner), - request_id, - next_seq: 1, - rx, - pending_eof: false, - closed: false, + HttpRequestResponse { + status, + headers, + body: body.to_vec().into(), }, + None, )) } + + fn build_headers(headers: Vec) -> Result { + let mut header_map = HeaderMap::new(); + for header in headers { + let name = HeaderName::from_bytes(header.name.as_bytes()).map_err(|err| { + invalid_params(format!("http/request header name is invalid: {err}")) + })?; + let value = HeaderValue::from_str(&header.value).map_err(|err| { + invalid_params(format!( + "http/request header value is invalid for {}: {err}", + header.name + )) + })?; + header_map.append(name, value); + } + Ok(header_map) + } + + fn response_headers(headers: &HeaderMap) -> Vec { + headers + .iter() + .filter_map(|(name, value)| { + Some(HttpHeader { + name: name.as_str().to_string(), + value: value.to_str().ok()?.to_string(), + }) + }) + .collect() + } + + pub(crate) async fn stream_body( + pending_stream: ExecutorPendingHttpBodyStream, + notifications: RpcNotificationSender, + ) { + let ExecutorPendingHttpBodyStream { + request_id, + response, + } = pending_stream; + let mut seq = 1; + let mut body = response.bytes_stream(); + while let Some(chunk) = body.next().await { + match chunk { + Ok(bytes) => { + if !send_executor_body_delta( + ¬ifications, + HttpRequestBodyDeltaNotification { + request_id: request_id.clone(), + seq, + delta: bytes.to_vec().into(), + done: false, + error: None, + }, + ) + .await + { + return; + } + seq += 1; + } + Err(err) => { + let _ = send_executor_body_delta( + ¬ifications, + HttpRequestBodyDeltaNotification { + request_id, + seq, + delta: Vec::new().into(), + done: true, + error: Some(err.to_string()), + }, + ) + .await; + return; + } + } + } + + let _ = send_executor_body_delta( + ¬ifications, + HttpRequestBodyDeltaNotification { + request_id, + seq, + delta: Vec::new().into(), + done: true, + error: None, + }, + ) + .await; + } } impl Inner { @@ -231,13 +397,21 @@ impl Inner { let streams = streams.as_ref().clone(); self.http_body_streams.store(Arc::new(HashMap::new())); for (request_id, tx) in streams { - let _ = tx.try_send(HttpRequestBodyDeltaNotification { - request_id, - seq: 1, - delta: Vec::new().into(), - done: true, - error: Some(message.clone()), - }); + if tx + .try_send(HttpRequestBodyDeltaNotification { + request_id: request_id.clone(), + seq: 1, + delta: Vec::new().into(), + done: true, + error: Some(message.clone()), + }) + .is_err() + { + let mut next_failures = self.http_body_stream_failures.load().as_ref().clone(); + next_failures.insert(request_id, message.clone()); + self.http_body_stream_failures + .store(Arc::new(next_failures)); + } } } @@ -312,6 +486,22 @@ impl Inner { } } +/// Active route registration owned while `http_request_stream` awaits headers. +struct HttpBodyStreamRegistration { + inner: Arc, + request_id: String, + active: bool, +} + +impl Drop for HttpBodyStreamRegistration { + /// Removes the route if the stream request future is cancelled before headers return. + fn drop(&mut self) { + if self.active { + spawn_remove_http_body_stream(Arc::clone(&self.inner), self.request_id.clone()); + } + } +} + /// Schedules HTTP body route removal from synchronous drop paths. fn spawn_remove_http_body_stream(inner: Arc, request_id: String) { if let Ok(handle) = Handle::try_current() { @@ -320,3 +510,13 @@ fn spawn_remove_http_body_stream(inner: Arc, request_id: String) { }); } } + +async fn send_executor_body_delta( + notifications: &RpcNotificationSender, + delta: HttpRequestBodyDeltaNotification, +) -> bool { + notifications + .notify(HTTP_REQUEST_BODY_DELTA_METHOD, &delta) + .await + .is_ok() +} diff --git a/codex-rs/exec-server/src/protocol.rs b/codex-rs/exec-server/src/protocol.rs index 24d1ee1c1e..435187d05a 100644 --- a/codex-rs/exec-server/src/protocol.rs +++ b/codex-rs/exec-server/src/protocol.rs @@ -286,15 +286,19 @@ pub struct HttpRequestParams { /// Optional request body bytes. #[serde(default, rename = "bodyBase64")] pub body: Option, - /// Optional request timeout in milliseconds. - #[serde(default)] + /// Request timeout in milliseconds. + /// + /// Omitted or `null` disables the timeout. A number applies that exact + /// millisecond deadline. + #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout_ms: Option, /// Caller-chosen stream id for `http/request/bodyDelta` notifications. /// /// The id must remain unique on a connection until the terminal body delta - /// arrives, even if the caller stops reading the stream earlier. - #[serde(default)] - pub request_id: Option, + /// arrives, even if the caller stops reading the stream earlier. Buffered + /// requests still send an id so callers can keep one consistent request + /// envelope shape. + pub request_id: String, /// Return after response headers and stream the response body as deltas. #[serde(default)] pub stream_response: bool, @@ -391,3 +395,49 @@ mod base64_bytes { .map_err(serde::de::Error::custom) } } + +#[cfg(test)] +mod tests { + use super::HttpRequestParams; + use pretty_assertions::assert_eq; + + #[test] + fn http_request_timeout_treats_omitted_and_null_as_no_timeout() { + let omitted: HttpRequestParams = serde_json::from_value(serde_json::json!({ + "method": "GET", + "url": "https://example.test", + "requestId": "req-omitted-timeout", + })) + .expect("omitted timeout should deserialize"); + let null_timeout: HttpRequestParams = serde_json::from_value(serde_json::json!({ + "method": "GET", + "url": "https://example.test", + "requestId": "req-null-timeout", + "timeoutMs": null, + })) + .expect("null timeout should deserialize"); + let explicit_timeout: HttpRequestParams = serde_json::from_value(serde_json::json!({ + "method": "GET", + "url": "https://example.test", + "requestId": "req-explicit-timeout", + "timeoutMs": 1234, + })) + .expect("numeric timeout should deserialize"); + + assert_eq!( + (omitted.request_id.as_str(), omitted.timeout_ms), + ("req-omitted-timeout", None) + ); + assert_eq!( + (null_timeout.request_id.as_str(), null_timeout.timeout_ms), + ("req-null-timeout", None) + ); + assert_eq!( + ( + explicit_timeout.request_id.as_str(), + explicit_timeout.timeout_ms + ), + ("req-explicit-timeout", Some(1234)) + ); + } +} diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index e82b4a0ea2..723b99f502 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -36,8 +36,9 @@ pub(crate) enum RpcCallError { type PendingRequest = oneshot::Sender>; type BoxFuture = Pin + Send + 'static>>; -type RequestRoute = - Box, JSONRPCRequest) -> BoxFuture + Send + Sync>; +type RequestRoute = Box< + dyn Fn(Arc, JSONRPCRequest) -> BoxFuture> + Send + Sync, +>; type NotificationRoute = Box, JSONRPCNotification) -> BoxFuture> + Send + Sync>; @@ -72,6 +73,17 @@ impl RpcNotificationSender { Self { outgoing_tx } } + pub(crate) async fn response( + &self, + request_id: RequestId, + result: Value, + ) -> Result<(), JSONRPCErrorError> { + self.outgoing_tx + .send(RpcServerOutboundMessage::Response { request_id, result }) + .await + .map_err(|_| internal_error("RPC connection closed while sending response".into())) + } + #[allow(dead_code)] pub(crate) async fn notify( &self, @@ -131,10 +143,10 @@ where let response = match response { Ok(response) => response.await, Err(error) => { - return RpcServerOutboundMessage::Error { request_id, error }; + return Some(RpcServerOutboundMessage::Error { request_id, error }); } }; - match response { + Some(match response { Ok(result) => match serde_json::to_value(result) { Ok(result) => RpcServerOutboundMessage::Response { request_id, result }, Err(err) => RpcServerOutboundMessage::Error { @@ -143,6 +155,34 @@ where }, }, Err(error) => RpcServerOutboundMessage::Error { request_id, error }, + }) + }) + }), + ); + } + + pub(crate) fn request_with_id(&mut self, method: &'static str, handler: F) + where + P: DeserializeOwned + Send + 'static, + F: Fn(Arc, RequestId, P) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + self.request_routes.insert( + method, + Box::new(move |state, request| { + let request_id = request.id; + let params = decode_request_params::

(request.params) + .map(|params| handler(state, request_id.clone(), params)); + Box::pin(async move { + let response = match params { + Ok(response) => response.await, + Err(error) => { + return Some(RpcServerOutboundMessage::Error { request_id, error }); + } + }; + match response { + Ok(()) => None, + Err(error) => Some(RpcServerOutboundMessage::Error { request_id, error }), } }) }), diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 46f7af90a1..15035335ea 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -4,8 +4,16 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::RequestId; +use serde_json::to_value; +use std::collections::HashSet; +use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use crate::ExecServerRuntimePaths; +use crate::client::http_client::ExecutorHttpRequestRunner; +use crate::client::http_client::ExecutorPendingHttpBodyStream; use crate::protocol::ExecParams; use crate::protocol::ExecResponse; use crate::protocol::FsCopyParams; @@ -22,6 +30,7 @@ use crate::protocol::FsRemoveParams; use crate::protocol::FsRemoveResponse; use crate::protocol::FsWriteFileParams; use crate::protocol::FsWriteFileResponse; +use crate::protocol::HttpRequestParams; use crate::protocol::InitializeParams; use crate::protocol::InitializeResponse; use crate::protocol::ReadParams; @@ -31,6 +40,8 @@ use crate::protocol::TerminateResponse; use crate::protocol::WriteParams; use crate::protocol::WriteResponse; use crate::rpc::RpcNotificationSender; +use crate::rpc::internal_error; +use crate::rpc::invalid_params; use crate::rpc::invalid_request; use crate::server::file_system_handler::FileSystemHandler; use crate::server::session_registry::SessionHandle; @@ -40,6 +51,9 @@ pub(crate) struct ExecServerHandler { session_registry: Arc, notifications: RpcNotificationSender, session: StdMutex>, + active_body_stream_ids: Mutex>, + background_task_shutdown: CancellationToken, + background_tasks: TaskTracker, file_system: FileSystemHandler, initialize_requested: AtomicBool, initialized: AtomicBool, @@ -55,6 +69,9 @@ impl ExecServerHandler { session_registry, notifications, session: StdMutex::new(None), + active_body_stream_ids: Mutex::new(HashSet::new()), + background_task_shutdown: CancellationToken::new(), + background_tasks: TaskTracker::new(), file_system: FileSystemHandler::new(runtime_paths), initialize_requested: AtomicBool::new(false), initialized: AtomicBool::new(false), @@ -62,6 +79,9 @@ impl ExecServerHandler { } pub(crate) async fn shutdown(&self) { + self.background_task_shutdown.cancel(); + self.background_tasks.close(); + self.background_tasks.wait().await; if let Some(session) = self.session() { session.detach().await; } @@ -147,6 +167,47 @@ impl ExecServerHandler { session.process().terminate(params).await } + pub(crate) async fn http_request( + self: &Arc, + request_id: RequestId, + params: HttpRequestParams, + ) -> Result<(), JSONRPCErrorError> { + self.require_initialized_for("http")?; + let stream_response = params.stream_response; + let http_request_id = params.request_id.clone(); + if stream_response { + self.reserve_http_body_stream(&http_request_id).await?; + } + let response = ExecutorHttpRequestRunner::new(params.timeout_ms)? + .run(params) + .await; + if response.is_err() && stream_response { + self.release_http_body_stream(&http_request_id).await; + } + let (response, mut pending_stream) = response?; + let result = match to_value(response) { + Ok(result) => result, + Err(err) => { + if let Some(pending_stream) = pending_stream.take() { + self.release_http_body_stream(&pending_stream.request_id) + .await; + } + return Err(internal_error(err.to_string())); + } + }; + if let Err(error) = self.notifications.response(request_id, result).await { + if let Some(pending_stream) = pending_stream.take() { + self.release_http_body_stream(&pending_stream.request_id) + .await; + } + return Err(error); + } + if let Some(pending_stream) = pending_stream { + self.start_http_body_stream(pending_stream).await; + } + Ok(()) + } + pub(crate) async fn fs_read_file( &self, params: FsReadFileParams, @@ -242,6 +303,44 @@ impl ExecServerHandler { .unwrap_or_else(std::sync::PoisonError::into_inner) .clone() } + + async fn start_http_body_stream( + self: &Arc, + pending_stream: ExecutorPendingHttpBodyStream, + ) { + let request_id = pending_stream.request_id.clone(); + if self.background_task_shutdown.is_cancelled() { + self.release_http_body_stream(&request_id).await; + return; + } + let finished_request_id = request_id.clone(); + let handler = Arc::clone(self); + let notifications = self.notifications.clone(); + let shutdown = self.background_task_shutdown.clone(); + self.background_tasks.spawn(async move { + tokio::select! { + _ = shutdown.cancelled() => {} + _ = ExecutorHttpRequestRunner::stream_body(pending_stream, notifications) => {} + } + handler.release_http_body_stream(&finished_request_id).await; + }); + } + + async fn release_http_body_stream(&self, request_id: &str) { + let mut active_body_stream_ids = self.active_body_stream_ids.lock().await; + active_body_stream_ids.remove(request_id); + } + + async fn reserve_http_body_stream(&self, request_id: &str) -> Result<(), JSONRPCErrorError> { + let mut active_body_stream_ids = self.active_body_stream_ids.lock().await; + if active_body_stream_ids.contains(request_id) { + return Err(invalid_params(format!( + "http/request streamResponse requestId `{request_id}` is already active" + ))); + } + active_body_stream_ids.insert(request_id.to_string()); + Ok(()) + } } #[cfg(test)] diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 1153bc83e2..dc1a9b9ffe 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -103,7 +103,9 @@ async fn run_connection( break; } }; - if outgoing_tx.send(message).await.is_err() { + if let Some(message) = message + && outgoing_tx.send(message).await.is_err() + { break; } } else if outgoing_tx diff --git a/codex-rs/exec-server/src/server/registry.rs b/codex-rs/exec-server/src/server/registry.rs index a57704c505..87dee6aa58 100644 --- a/codex-rs/exec-server/src/server/registry.rs +++ b/codex-rs/exec-server/src/server/registry.rs @@ -19,6 +19,8 @@ use crate::protocol::FsReadDirectoryParams; use crate::protocol::FsReadFileParams; use crate::protocol::FsRemoveParams; use crate::protocol::FsWriteFileParams; +use crate::protocol::HTTP_REQUEST_METHOD; +use crate::protocol::HttpRequestParams; use crate::protocol::INITIALIZE_METHOD; use crate::protocol::INITIALIZED_METHOD; use crate::protocol::InitializeParams; @@ -42,6 +44,12 @@ pub(crate) fn build_router() -> RpcRouter { handler.initialize(params).await }, ); + router.request_with_id( + HTTP_REQUEST_METHOD, + |handler: Arc, request_id, params: HttpRequestParams| async move { + handler.http_request(request_id, params).await + }, + ); router.request( EXEC_METHOD, |handler: Arc, params: ExecParams| async move { handler.exec(params).await }, diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs index 1462c80168..6a087dd115 100644 --- a/codex-rs/exec-server/tests/http_client.rs +++ b/codex-rs/exec-server/tests/http_client.rs @@ -42,6 +42,7 @@ const HTTP_REQUEST_BODY_DELTA_METHOD: &str = "http/request/bodyDelta"; const INITIALIZE_METHOD: &str = "initialize"; const INITIALIZED_METHOD: &str = "initialized"; const TEST_TIMEOUT: Duration = Duration::from_secs(5); +const HTTP_BODY_DELTA_CHANNEL_CAPACITY: u64 = 256; const OVERFLOWING_BODY_DELTA_FRAMES: u64 = 1_024; /// What this tests: the buffered HTTP helper always sends a buffered @@ -51,8 +52,8 @@ async fn http_request_forces_buffered_request_params() -> Result<()> { // Phase 1: start a fake WebSocket exec-server so the test covers the // public client connection path without depending on the HTTP runner. let server = spawn_scripted_exec_server(|mut peer| async move { - // Phase 2: verify the buffered helper strips streaming-only fields - // before it sends the JSON-RPC call. + // Phase 2: verify the buffered helper forces buffered mode before it + // sends the JSON-RPC call. let (request_id, params) = peer.read_http_request().await?; assert_eq!( params, @@ -62,7 +63,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> { headers: Vec::new(), body: None, timeout_ms: None, - request_id: None, + request_id: "ignored-stream-id".to_string(), stream_response: false, } ); @@ -90,7 +91,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> { headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("ignored-stream-id".to_string()), + request_id: "ignored-stream-id".to_string(), stream_response: true, }), ) @@ -130,7 +131,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta }], body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -185,7 +186,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-2".to_string()), + request_id: "http-2".to_string(), stream_response: true, } ); @@ -214,7 +215,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta }], body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -252,7 +253,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -288,7 +289,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -321,7 +322,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-2".to_string()), + request_id: "http-2".to_string(), stream_response: true, } ); @@ -347,7 +348,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -371,7 +372,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }; let (reuse_response, _reuse_body_stream) = @@ -410,7 +411,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -429,7 +430,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-2".to_string()), + request_id: "http-2".to_string(), stream_response: true, } ); @@ -473,7 +474,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }) .await; @@ -494,7 +495,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -540,7 +541,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -579,7 +580,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-2".to_string()), + request_id: "http-2".to_string(), stream_response: true, } ); @@ -614,7 +615,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -646,7 +647,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -690,7 +691,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result< headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -716,7 +717,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result< headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) @@ -742,6 +743,98 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result< Ok(()) } +/// What this tests: transport disconnect still records a terminal stream +/// failure even when the client-side body-delta queue is already full. +#[tokio::test] +async fn http_response_body_stream_reports_disconnect_when_queue_is_full() -> Result<()> { + // Phase 1: fill the queued body-delta route exactly to capacity before the + // response headers arrive, then drop the transport without sending EOF. + let server = spawn_scripted_exec_server(|mut peer| async move { + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/disconnect-full-queue".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: "http-1".to_string(), + stream_response: true, + } + ); + for seq in 1..=HTTP_BODY_DELTA_CHANNEL_CAPACITY { + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq, + delta: b"x".to_vec().into(), + done: false, + error: None, + }) + .await?; + } + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 2: start the streaming request and receive headers while the + // queue is already full. + let (_response, mut body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/disconnect-full-queue".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: "caller-stream-id".to_string(), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + + // Phase 3: drain the queued chunks and assert the transport disconnect is + // still reported as an error rather than a clean EOF. + let mut chunks = 0; + let error = loop { + match timeout(TEST_TIMEOUT, body_stream.recv()) + .await + .context("disconnect should wake the full queued body stream")? + { + Ok(Some(_chunk)) => { + chunks += 1; + } + Ok(None) => bail!("disconnect with a full queue should not look like clean EOF"), + Err(error) => break error, + } + }; + assert_eq!( + ( + chunks, + error + .to_string() + .starts_with( + "exec-server protocol error: http response stream `http-1` failed: exec-server transport disconnected", + ), + ), + (HTTP_BODY_DELTA_CHANNEL_CAPACITY as usize, true) + ); + + drop(client); + server.finish().await?; + Ok(()) +} + /// What this tests: body-delta backpressure closes the public body stream as /// an error rather than letting callers accept a truncated body as clean EOF. #[tokio::test] @@ -759,7 +852,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<( headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("http-1".to_string()), + request_id: "http-1".to_string(), stream_response: true, } ); @@ -801,7 +894,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<( headers: Vec::new(), body: None, timeout_ms: None, - request_id: Some("caller-stream-id".to_string()), + request_id: "caller-stream-id".to_string(), stream_response: false, }), ) diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs new file mode 100644 index 0000000000..f45261dad7 --- /dev/null +++ b/codex-rs/exec-server/tests/http_request.rs @@ -0,0 +1,568 @@ +#![cfg(unix)] + +mod common; + +use std::collections::BTreeMap; +use std::io::ErrorKind; +use std::time::Duration; + +use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRequestBodyDeltaNotification; +use codex_exec_server::HttpRequestParams; +use codex_exec_server::HttpRequestResponse; +use codex_exec_server::InitializeParams; +use common::exec_server::ExecServerHarness; +use common::exec_server::exec_server; +use pretty_assertions::assert_eq; +use serde::de::DeserializeOwned; +use serde_json::Value; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::sync::oneshot; +use tokio::time::timeout; + +/// HTTP request captured by the ad-hoc TCP server in these integration tests. +#[derive(Debug)] +struct CapturedHttpRequest { + stream: TcpStream, + request_line: String, + headers: BTreeMap, + body: Vec, +} + +/// What this tests: a real exec-server websocket `http/request` performs one +/// HTTP request through the runner and returns the complete response body in +/// the JSON-RPC response. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> { + // Phase 1: start exec-server and complete the JSON-RPC handshake. + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + // Phase 2: start a local HTTP peer and ask exec-server to POST to it. + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("http://{}/mcp?case=buffered", listener.local_addr()?); + let http_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "POST".to_string(), + url, + headers: vec![HttpHeader { + name: "x-codex-test".to_string(), + value: "buffered".to_string(), + }], + body: Some(b"request-body".to_vec().into()), + timeout_ms: Some(5_000), + request_id: "buffered-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + // Phase 3: assert the HTTP peer observes the expected method, path, + // headers, and body before returning a fixed-length response. + let captured = accept_http_request(&listener).await?; + assert_eq!( + ( + captured.request_line.as_str(), + captured.headers.get("x-codex-test").map(String::as_str), + captured.body.as_slice(), + ), + ( + "POST /mcp?case=buffered HTTP/1.1", + Some("buffered"), + b"request-body".as_slice(), + ) + ); + respond_with_status_and_headers( + captured.stream, + "201 Created", + &[("x-mcp-test", "buffered")], + b"response-body", + ) + .await?; + + // Phase 4: assert exec-server returns status, response headers, and the + // full response body in the JSON-RPC result. + let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?; + assert_eq!( + ( + response.status, + response_header(&response.headers, "x-mcp-test"), + response.body.into_inner(), + ), + (201, Some("buffered".to_string()), b"response-body".to_vec(),) + ); + + server.shutdown().await?; + Ok(()) +} + +/// What this tests: a real exec-server websocket `http/request` can return +/// response headers immediately and stream the response body as ordered +/// `http/request/bodyDelta` notifications. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_streams_response_body_notifications() -> anyhow::Result<()> { + // Phase 1: start exec-server and complete the JSON-RPC handshake. + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + // Phase 2: start a local HTTP peer and ask exec-server for a streamed GET. + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("http://{}/mcp?case=streaming", listener.local_addr()?); + let http_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url, + headers: vec![HttpHeader { + name: "accept".to_string(), + value: "text/event-stream".to_string(), + }], + body: None, + timeout_ms: Some(5_000), + request_id: "stream-1".to_string(), + stream_response: true, + })?, + ) + .await?; + + // Phase 3: assert the HTTP peer observes the expected request and then + // respond with chunked transfer encoding to exercise streaming. + let captured = accept_http_request(&listener).await?; + assert_eq!( + ( + captured.request_line.as_str(), + captured.headers.get("accept").map(String::as_str), + captured.body, + ), + ( + "GET /mcp?case=streaming HTTP/1.1", + Some("text/event-stream"), + Vec::new(), + ) + ); + respond_with_chunked_body( + captured.stream, + &[("x-mcp-test", "streaming")], + &[b"hello ".as_slice(), b"world".as_slice()], + ) + .await?; + + // Phase 4: assert the JSON-RPC response reaches the wire before any body + // delta notifications, and that it contains status and headers but no + // buffered body when streaming is requested. + let first_event = server.next_event().await?; + let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = first_event else { + anyhow::bail!("expected http/request response before body deltas, got {first_event:?}"); + }; + assert_eq!(id, http_request_id); + let response: HttpRequestResponse = serde_json::from_value(result)?; + assert_eq!( + ( + response.status, + response_header(&response.headers, "x-mcp-test"), + response.body.into_inner(), + ), + (200, Some("streaming".to_string()), Vec::new()) + ); + + // Phase 5: assert the body notifications are contiguous, ordered, and end + // with a clean terminal frame. + let deltas = collect_response_body_deltas(&mut server, "stream-1").await?; + let seqs = deltas.iter().map(|delta| delta.seq).collect::>(); + let body = deltas + .iter() + .flat_map(|delta| delta.delta.clone().into_inner()) + .collect::>(); + let terminal = deltas.last().map(|delta| (delta.done, delta.error.clone())); + let expected_seqs = (1..=deltas.len() as u64).collect::>(); + assert_eq!( + (seqs, body, terminal), + (expected_seqs, b"hello world".to_vec(), Some((true, None))) + ); + + server.shutdown().await?; + Ok(()) +} + +/// What this tests: streamed `requestId`s stay reserved until the body stream +/// finishes, so a second in-flight request cannot reuse the same id. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_rejects_duplicate_stream_request_ids() -> anyhow::Result<()> { + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!( + "http://{}/mcp?case=duplicate-stream-id", + listener.local_addr()? + ); + let first_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: url.clone(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: "stream-dup".to_string(), + stream_response: true, + })?, + ) + .await?; + + let captured = accept_http_request(&listener).await?; + let (finish_tx, finish_rx) = oneshot::channel(); + let response_task = tokio::spawn(async move { + respond_with_chunked_body_until_finish(captured.stream, &[], &[b"hello"], finish_rx).await + }); + + let _: HttpRequestResponse = wait_for_response(&mut server, first_request_id).await?; + + let duplicate_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url, + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: "stream-dup".to_string(), + stream_response: true, + })?, + ) + .await?; + + let duplicate_response = server + .wait_for_event(|event| { + matches!( + event, + JSONRPCMessage::Error(JSONRPCError { id, .. }) if id == &duplicate_request_id + ) + }) + .await?; + let JSONRPCMessage::Error(JSONRPCError { error, .. }) = duplicate_response else { + anyhow::bail!("expected duplicate requestId error response"); + }; + assert_eq!(error.code, -32602); + assert_eq!( + error.message, + "http/request streamResponse requestId `stream-dup` is already active" + ); + + finish_tx + .send(()) + .expect("response task should still be waiting"); + response_task.await??; + let _ = collect_response_body_deltas(&mut server, "stream-dup").await?; + + server.shutdown().await?; + Ok(()) +} + +/// What this tests: omitting `timeoutMs` leaves the request unbounded, while +/// an explicit short timeout still fails the same delayed response. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_honors_optional_timeout() -> anyhow::Result<()> { + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let delayed_url = format!( + "http://{}/mcp?case=optional-timeout", + listener.local_addr()? + ); + let no_timeout_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: delayed_url.clone(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: "buffered-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + let captured = accept_http_request(&listener).await?; + let delayed_response = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + respond_with_status_and_headers(captured.stream, "200 OK", &[], b"slow-success").await + }); + let response: HttpRequestResponse = + wait_for_response(&mut server, no_timeout_request_id).await?; + assert_eq!(response.body.into_inner(), b"slow-success".to_vec()); + delayed_response.await??; + + let timeout_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: delayed_url, + headers: Vec::new(), + body: None, + timeout_ms: Some(10), + request_id: "buffered-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + let captured = accept_http_request(&listener).await?; + let delayed_timeout_response = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + respond_with_status_and_headers(captured.stream, "200 OK", &[], b"too-late").await + }); + let error = wait_for_error_response(&mut server, timeout_request_id).await?; + assert_eq!(error.code, -32603); + assert!( + error.message.starts_with("http/request failed: "), + "unexpected timeout error: {}", + error.message + ); + match delayed_timeout_response.await? { + Ok(()) => {} + Err(err) if is_expected_peer_disconnect(&err) => {} + Err(err) => return Err(err), + } + + server.shutdown().await?; + Ok(()) +} + +/// Performs the JSON-RPC initialize handshake required before executor methods. +async fn initialize_exec_server(server: &mut ExecServerHarness) -> anyhow::Result<()> { + let initialize_id = server + .send_request( + "initialize", + serde_json::to_value(InitializeParams { + client_name: "exec-server-http-test".to_string(), + resume_session_id: None, + })?, + ) + .await?; + let _: Value = wait_for_response(server, initialize_id).await?; + server + .send_notification("initialized", serde_json::json!({})) + .await?; + Ok(()) +} + +/// Waits for a typed JSON-RPC response with the requested id. +async fn wait_for_response( + server: &mut ExecServerHarness, + request_id: RequestId, +) -> anyhow::Result +where + T: DeserializeOwned, +{ + let response = server + .wait_for_event(|event| { + matches!( + event, + JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &request_id + ) + }) + .await?; + let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = response else { + anyhow::bail!("expected JSON-RPC response for {request_id:?}"); + }; + Ok(serde_json::from_value(result)?) +} + +/// Waits for a JSON-RPC error with the requested id. +async fn wait_for_error_response( + server: &mut ExecServerHarness, + request_id: RequestId, +) -> anyhow::Result { + let response = server + .wait_for_event(|event| { + matches!( + event, + JSONRPCMessage::Error(JSONRPCError { id, .. }) if id == &request_id + ) + }) + .await?; + let JSONRPCMessage::Error(JSONRPCError { error, .. }) = response else { + anyhow::bail!("expected JSON-RPC error for {request_id:?}"); + }; + Ok(error) +} + +/// Accepts one HTTP/1.1 request and captures its wire-visible fields. +async fn accept_http_request(listener: &TcpListener) -> anyhow::Result { + let (stream, _) = timeout(Duration::from_secs(5), listener.accept()).await??; + let mut reader = BufReader::new(stream); + + let mut request_line = String::new(); + reader.read_line(&mut request_line).await?; + let request_line = request_line.trim_end_matches("\r\n").to_string(); + + let mut headers = BTreeMap::new(); + loop { + let mut line = String::new(); + reader.read_line(&mut line).await?; + if line == "\r\n" { + break; + } + let line = line.trim_end_matches("\r\n"); + let (name, value) = line + .split_once(':') + .ok_or_else(|| anyhow::anyhow!("HTTP header should contain colon: {line}"))?; + headers.insert(name.to_ascii_lowercase(), value.trim().to_string()); + } + + let content_length = headers + .get("content-length") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0); + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).await?; + + Ok(CapturedHttpRequest { + stream: reader.into_inner(), + request_line, + headers, + body, + }) +} + +/// Writes a fixed-length HTTP response to the captured request stream. +async fn respond_with_status_and_headers( + mut stream: TcpStream, + status: &str, + headers: &[(&str, &str)], + body: &[u8], +) -> anyhow::Result<()> { + let extra_headers = headers + .iter() + .map(|(name, value)| format!("{name}: {value}\r\n")) + .collect::(); + let response = format!( + "HTTP/1.1 {status}\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n{extra_headers}\r\n", + body.len(), + ); + stream.write_all(response.as_bytes()).await?; + stream.write_all(body).await?; + stream.flush().await?; + Ok(()) +} + +fn is_expected_peer_disconnect(err: &anyhow::Error) -> bool { + err.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|io_err| { + matches!( + io_err.kind(), + ErrorKind::BrokenPipe | ErrorKind::ConnectionReset | ErrorKind::UnexpectedEof + ) + }) + }) +} + +/// Writes a chunked HTTP response so reqwest must drive the streaming path. +async fn respond_with_chunked_body( + mut stream: TcpStream, + headers: &[(&str, &str)], + chunks: &[&[u8]], +) -> anyhow::Result<()> { + let extra_headers = headers + .iter() + .map(|(name, value)| format!("{name}: {value}\r\n")) + .collect::(); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\nconnection: close\r\n{extra_headers}\r\n", + ); + stream.write_all(response.as_bytes()).await?; + for chunk in chunks { + stream + .write_all(format!("{:x}\r\n", chunk.len()).as_bytes()) + .await?; + stream.write_all(chunk).await?; + stream.write_all(b"\r\n").await?; + stream.flush().await?; + } + stream.write_all(b"0\r\n\r\n").await?; + stream.flush().await?; + Ok(()) +} + +/// Writes a chunked response and keeps the stream open until the test allows EOF. +async fn respond_with_chunked_body_until_finish( + mut stream: TcpStream, + headers: &[(&str, &str)], + chunks: &[&[u8]], + finish_rx: oneshot::Receiver<()>, +) -> anyhow::Result<()> { + let extra_headers = headers + .iter() + .map(|(name, value)| format!("{name}: {value}\r\n")) + .collect::(); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\nconnection: close\r\n{extra_headers}\r\n", + ); + stream.write_all(response.as_bytes()).await?; + for chunk in chunks { + stream + .write_all(format!("{:x}\r\n", chunk.len()).as_bytes()) + .await?; + stream.write_all(chunk).await?; + stream.write_all(b"\r\n").await?; + stream.flush().await?; + } + finish_rx.await?; + stream.write_all(b"0\r\n\r\n").await?; + stream.flush().await?; + Ok(()) +} + +/// Collects streamed response-body notifications until the terminal frame. +async fn collect_response_body_deltas( + server: &mut ExecServerHarness, + request_id: &str, +) -> anyhow::Result> { + let mut deltas = Vec::new(); + loop { + let event = server.next_event().await?; + let JSONRPCMessage::Notification(JSONRPCNotification { method, params }) = event else { + anyhow::bail!("expected http/request body delta notification, got {event:?}"); + }; + assert_eq!(method, "http/request/bodyDelta"); + let delta: HttpRequestBodyDeltaNotification = + serde_json::from_value(params.unwrap_or(Value::Null))?; + assert_eq!(delta.request_id, request_id); + + let done = delta.done; + deltas.push(delta); + if done { + return Ok(deltas); + } + } +} + +/// Returns a response header value without depending on header-name casing. +fn response_header(headers: &[HttpHeader], name: &str) -> Option { + headers + .iter() + .find(|header| header.name.eq_ignore_ascii_case(name)) + .map(|header| header.value.clone()) +} diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 5ad6b160b1..039aab66b7 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -263,7 +263,7 @@ async fn run_codex_tool_session_inner( outgoing.send_response(request_id.clone(), result).await; break; } - EventMsg::Warning(_) => { + EventMsg::Warning(_) | EventMsg::GuardianWarning(_) => { continue; } EventMsg::GuardianAssessment(_) => { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 999080be66..da4071c291 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1527,6 +1527,9 @@ pub enum EventMsg { /// indicates the turn continued but the user should still be notified. Warning(WarningEvent), + /// Warning issued by the guardian automatic approval reviewer. + GuardianWarning(WarningEvent), + /// Realtime conversation lifecycle start event. RealtimeConversationStarted(RealtimeConversationStartedEvent), diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index df41d2c766..897cd0bbf2 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -132,6 +132,7 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option { | EventMsg::DynamicToolCallRequest(_) | EventMsg::DynamicToolCallResponse(_) => Some(EventPersistenceMode::Extended), EventMsg::Warning(_) + | EventMsg::GuardianWarning(_) | EventMsg::RealtimeConversationStarted(_) | EventMsg::RealtimeConversationSdp(_) | EventMsg::RealtimeConversationRealtime(_) diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index db5aa1314d..f7793a55ae 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -425,6 +425,7 @@ fn server_notification_thread_target( Some(notification.thread_id.as_str()) } ServerNotification::Warning(notification) => notification.thread_id.as_deref(), + ServerNotification::GuardianWarning(notification) => Some(notification.thread_id.as_str()), ServerNotification::SkillsChanged(_) | ServerNotification::McpServerStatusUpdated(_) | ServerNotification::McpServerOauthLoginCompleted(_) @@ -1067,6 +1068,7 @@ mod tests { use codex_app_server_protocol::CommandExecutionOutputDeltaNotification; use codex_app_server_protocol::CommandExecutionSource; use codex_app_server_protocol::CommandExecutionStatus; + use codex_app_server_protocol::GuardianWarningNotification; use codex_app_server_protocol::ItemCompletedNotification; use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification; @@ -1686,4 +1688,17 @@ mod tests { assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id)); } + + #[test] + fn guardian_warning_notifications_route_to_threads() { + let thread_id = ThreadId::new(); + let notification = ServerNotification::GuardianWarning(GuardianWarningNotification { + thread_id: thread_id.to_string(), + message: "warning".to_string(), + }); + + let target = server_notification_thread_target(¬ification); + + assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id)); + } } diff --git a/codex-rs/tui/src/app/replay_filter.rs b/codex-rs/tui/src/app/replay_filter.rs index ec9176ffaf..273ca0dd5a 100644 --- a/codex-rs/tui/src/app/replay_filter.rs +++ b/codex-rs/tui/src/app/replay_filter.rs @@ -25,7 +25,9 @@ pub(super) fn event_is_notice(event: &ThreadBufferedEvent) -> bool { matches!( event, ThreadBufferedEvent::Notification( - ServerNotification::Warning(_) | ServerNotification::ConfigWarning(_) + ServerNotification::Warning(_) + | ServerNotification::GuardianWarning(_) + | ServerNotification::ConfigWarning(_) ) ) } diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index e7adb1cd27..79f3bbc903 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -6589,6 +6589,9 @@ impl ChatWidget { } ServerNotification::ModelRerouted(_) => {} ServerNotification::Warning(notification) => self.on_warning(notification.message), + ServerNotification::GuardianWarning(notification) => { + self.on_warning(notification.message) + } ServerNotification::DeprecationNotice(notification) => { self.on_deprecation_notice(DeprecationNoticeEvent { summary: notification.summary, @@ -7086,7 +7089,8 @@ impl ChatWidget { self.set_token_info(ev.info); self.on_rate_limit_snapshot(ev.rate_limits); } - EventMsg::Warning(WarningEvent { message }) => self.on_warning(message), + EventMsg::Warning(WarningEvent { message }) + | EventMsg::GuardianWarning(WarningEvent { message }) => self.on_warning(message), EventMsg::GuardianAssessment(ev) => self.on_guardian_assessment(ev), EventMsg::ModelReroute(_) => {} EventMsg::Error(ErrorEvent { diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index d81544e4f7..f001c3afdc 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -51,6 +51,7 @@ pub(super) use codex_app_server_protocol::GuardianApprovalReviewStatus; pub(super) use codex_app_server_protocol::GuardianCommandSource as AppServerGuardianCommandSource; pub(super) use codex_app_server_protocol::GuardianRiskLevel as AppServerGuardianRiskLevel; pub(super) use codex_app_server_protocol::GuardianUserAuthorization as AppServerGuardianUserAuthorization; +pub(super) use codex_app_server_protocol::GuardianWarningNotification; pub(super) use codex_app_server_protocol::HookCompletedNotification as AppServerHookCompletedNotification; pub(super) use codex_app_server_protocol::HookEventName as AppServerHookEventName; pub(super) use codex_app_server_protocol::HookExecutionMode as AppServerHookExecutionMode; diff --git a/codex-rs/tui/src/chatwidget/tests/app_server.rs b/codex-rs/tui/src/chatwidget/tests/app_server.rs index 95ec729403..06acb16d07 100644 --- a/codex-rs/tui/src/chatwidget/tests/app_server.rs +++ b/codex-rs/tui/src/chatwidget/tests/app_server.rs @@ -212,6 +212,27 @@ async fn live_app_server_warning_notification_renders_message() { ); } +#[tokio::test] +async fn live_app_server_guardian_warning_notification_renders_message() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + + chat.handle_server_notification( + ServerNotification::GuardianWarning(GuardianWarningNotification { + thread_id: "thread-1".to_string(), + message: "Automatic approval review denied the requested action.".to_string(), + }), + /*replay_kind*/ None, + ); + + let cells = drain_insert_history(&mut rx); + assert_eq!(cells.len(), 1, "expected one warning history cell"); + let rendered = lines_to_single_string(&cells[0]); + assert!( + rendered.contains("Automatic approval review denied the requested action."), + "expected guardian warning notification message, got {rendered}" + ); +} + #[tokio::test] async fn live_app_server_config_warning_prefixes_summary() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/guardian.rs b/codex-rs/tui/src/chatwidget/tests/guardian.rs index ffdefa534b..dfb74a5d5c 100644 --- a/codex-rs/tui/src/chatwidget/tests/guardian.rs +++ b/codex-rs/tui/src/chatwidget/tests/guardian.rs @@ -28,7 +28,7 @@ async fn guardian_denied_exec_renders_warning_and_denied_request() { }); chat.handle_codex_event(Event { id: "guardian-warning".into(), - msg: EventMsg::Warning(WarningEvent { + msg: EventMsg::GuardianWarning(WarningEvent { message: "Automatic approval review denied (risk: high): The planned action would transmit the full contents of a workspace source file (`core/src/codex.rs`) to `https://example.com`, which is an external and untrusted endpoint.".into(), }), }); @@ -228,7 +228,7 @@ async fn guardian_timed_out_exec_renders_warning_and_timed_out_request() { }); chat.handle_codex_event(Event { id: "guardian-warning".into(), - msg: EventMsg::Warning(WarningEvent { + msg: EventMsg::GuardianWarning(WarningEvent { message: "Automatic approval review timed out while evaluating the requested approval." .into(), }),