openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 4696169fe9 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-04-22 13:57:47 -07:00
committed by GitHub
parent 7bbd21837d 4696169fe9
commit 4382a5037c
38 changed files with 1814 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -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",
]
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+37
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+39
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+39
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/v2/GuardianWarningNotification.json
Generated
+19
View File
@@ -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"
}
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts
Generated
+2 -1
View File
@@ -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 };
codex-rs/app-server-protocol/schema/typescript/v2/GuardianWarningNotification.ts
Generated
+13
View File
@@ -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, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+1
View File
@@ -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";
codex-rs/app-server-protocol/src/protocol/common.rs
+1
View File
@@ -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),
codex-rs/app-server-protocol/src/protocol/v2.rs
+10
View File
@@ -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/")]
codex-rs/app-server/src/bespoke_event_handling.rs
+17
View File
@@ -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(
codex-rs/app-server/src/outgoing_message.rs
+23
View File
@@ -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 {
codex-rs/core/src/guardian/mod.rs
+54
View File
@@ -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<String, GuardianRejectionCircuitBreakerTurn>,
}
#[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)]
codex-rs/core/src/guardian/review.rs
+88 -14
View File
@@ -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<Session>, turn_id: &str) {
session
.services
.guardian_rejection_circuit_breaker
.lock()
.await
.record_non_denial(turn_id);
}
async fn record_guardian_denial(session: &Arc<Session>, turn: &Arc<TurnContext>, 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<Session>,
turn: &Arc<TurnContext>,
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 {
codex-rs/core/src/guardian/tests.rs
+70 -2
View File
@@ -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<Session>, Arc<TurnContext>) {
@@ -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 =>
{
codex-rs/core/src/session/session.rs
+2
View File
@@ -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),
codex-rs/core/src/session/tests.rs
+123
View File
@@ -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<Self>,
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
_input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> Option<String> {
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() {
codex-rs/core/src/session/turn.rs
+1
View File
@@ -1484,6 +1484,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
},
EventMsg::Error(_)
| EventMsg::Warning(_)
| EventMsg::GuardianWarning(_)
| EventMsg::RealtimeConversationStarted(_)
| EventMsg::RealtimeConversationSdp(_)
| EventMsg::RealtimeConversationRealtime(_)
codex-rs/core/src/state/service.rs
+4
View File
@@ -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<ApprovalStore>,
pub(crate) guardian_rejections: Mutex<HashMap<String, GuardianRejection>>,
pub(crate) guardian_rejection_circuit_breaker: Mutex<GuardianRejectionCircuitBreaker>,
pub(crate) runtime_handle: Handle,
pub(crate) skills_manager: Arc<SkillsManager>,
pub(crate) plugins_manager: Arc<PluginsManager>,
pub(crate) mcp_manager: Arc<McpManager>,
codex-rs/core/src/tasks/mod.rs
+50
View File
@@ -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<Self>,
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<Self>,
turn_context: Arc<TurnContext>,
@@ -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);
}
}
codex-rs/exec-server/Cargo.toml
+2
View File
@@ -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"] }
codex-rs/exec-server/src/client/http_client.rs
+271 -71
View File
@@ -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<HttpRequestResponse, ExecServerError> {
params.stream_response = false;
self.call(HTTP_REQUEST_METHOD, &params).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, &params).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<Inner>,
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<HttpRequestResponse, ExecServerError> {
params.stream_response = false;
params.request_id = None;
self.call(HTTP_REQUEST_METHOD, &params).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, &params).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<u64>) -> Result<Self, JSONRPCErrorError> {
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<ExecutorPendingHttpBodyStream>), 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(&params.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<HttpHeader>) -> Result<HeaderMap, JSONRPCErrorError> {
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<HttpHeader> {
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(
&notifications,
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(
&notifications,
HttpRequestBodyDeltaNotification {
request_id,
seq,
delta: Vec::new().into(),
done: true,
error: Some(err.to_string()),
},
)
.await;
return;
}
}
}
let _ = send_executor_body_delta(
&notifications,
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<Inner>,
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<Inner>, request_id: String) {
if let Ok(handle) = Handle::try_current() {
@@ -320,3 +510,13 @@ fn spawn_remove_http_body_stream(inner: Arc<Inner>, 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()
}
codex-rs/exec-server/src/protocol.rs
+55 -5
View File
@@ -286,15 +286,19 @@ pub struct HttpRequestParams {
/// Optional request body bytes.
#[serde(default, rename = "bodyBase64")]
pub body: Option<ByteChunk>,
/// 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<u64>,
/// 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<String>,
/// 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))
);
}
}
codex-rs/exec-server/src/rpc.rs
+44 -4
View File
@@ -36,8 +36,9 @@ pub(crate) enum RpcCallError {
type PendingRequest = oneshot::Sender<Result<Value, RpcCallError>>;
type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
type RequestRoute<S> =
Box<dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<RpcServerOutboundMessage> + Send + Sync>;
type RequestRoute<S> = Box<
dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<Option<RpcServerOutboundMessage>> + Send + Sync,
>;
type NotificationRoute<S> =
Box<dyn Fn(Arc<S>, JSONRPCNotification) -> BoxFuture<Result<(), String>> + 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<P: Serialize>(
&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<P, F, Fut>(&mut self, method: &'static str, handler: F)
where
P: DeserializeOwned + Send + 'static,
F: Fn(Arc<S>, RequestId, P) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), JSONRPCErrorError>> + Send + 'static,
{
self.request_routes.insert(
method,
Box::new(move |state, request| {
let request_id = request.id;
let params = decode_request_params::<P>(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 }),
}
})
}),
codex-rs/exec-server/src/server/handler.rs
+99
View File
@@ -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<SessionRegistry>,
notifications: RpcNotificationSender,
session: StdMutex<Option<SessionHandle>>,
active_body_stream_ids: Mutex<HashSet<String>>,
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<Self>,
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<Self>,
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)]
codex-rs/exec-server/src/server/processor.rs
+3 -1
View File
@@ -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
codex-rs/exec-server/src/server/registry.rs
+8
View File
@@ -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<ExecServerHandler> {
handler.initialize(params).await
},
);
router.request_with_id(
HTTP_REQUEST_METHOD,
|handler: Arc<ExecServerHandler>, request_id, params: HttpRequestParams| async move {
handler.http_request(request_id, params).await
},
);
router.request(
EXEC_METHOD,
|handler: Arc<ExecServerHandler>, params: ExecParams| async move { handler.exec(params).await },
codex-rs/exec-server/tests/http_client.rs
+117 -24
View File
@@ -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,
}),
)
codex-rs/exec-server/tests/http_request.rs
+568
View File
@@ -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<String, String>,
body: Vec<u8>,
}
/// 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::<Vec<_>>();
let body = deltas
.iter()
.flat_map(|delta| delta.delta.clone().into_inner())
.collect::<Vec<_>>();
let terminal = deltas.last().map(|delta| (delta.done, delta.error.clone()));
let expected_seqs = (1..=deltas.len() as u64).collect::<Vec<_>>();
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<T>(
server: &mut ExecServerHarness,
request_id: RequestId,
) -> anyhow::Result<T>
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<codex_app_server_protocol::JSONRPCErrorError> {
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<CapturedHttpRequest> {
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::<usize>().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::<String>();
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::<std::io::Error>()
.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::<String>();
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::<String>();
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<Vec<HttpRequestBodyDeltaNotification>> {
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<String> {
headers
.iter()
.find(|header| header.name.eq_ignore_ascii_case(name))
.map(|header| header.value.clone())
}
codex-rs/mcp-server/src/codex_tool_runner.rs
+1 -1
View File
@@ -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(_) => {
codex-rs/protocol/src/protocol.rs
+3
View File
@@ -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),
codex-rs/rollout/src/policy.rs
+1
View File
@@ -132,6 +132,7 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option<EventPersistenceMode> {
| EventMsg::DynamicToolCallRequest(_)
| EventMsg::DynamicToolCallResponse(_) => Some(EventPersistenceMode::Extended),
EventMsg::Warning(_)
| EventMsg::GuardianWarning(_)
| EventMsg::RealtimeConversationStarted(_)
| EventMsg::RealtimeConversationSdp(_)
| EventMsg::RealtimeConversationRealtime(_)
codex-rs/tui/src/app/app_server_adapter.rs
+15
View File
@@ -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(&notification);
assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id));
}
}
codex-rs/tui/src/app/replay_filter.rs
+3 -1
View File
@@ -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(_)
)
)
}
codex-rs/tui/src/chatwidget.rs
+5 -1
View File
@@ -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 {
codex-rs/tui/src/chatwidget/tests.rs
+1
View File
@@ -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;
codex-rs/tui/src/chatwidget/tests/app_server.rs
+21
View File
@@ -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;
codex-rs/tui/src/chatwidget/tests/guardian.rs
+2 -2
View File
@@ -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(),
}),