use crate::error_code::internal_error;use crate::error_code::invalid_request;use crate::outgoing_message::ClientRequestResult;use crate::outgoing_message::ThreadScopedOutgoingMessageSender;use crate::request_processors::populate_thread_turns_from_history;use crate::request_processors::thread_from_stored_thread;use crate::request_processors::thread_settings_from_core_snapshot;use crate::server_request_error::is_turn_transition_server_request_error;use crate::thread_state::ThreadState;use crate::thread_state::TurnSummary;use crate::thread_state::resolve_server_request_on_thread_listener;use crate::thread_status::ThreadWatchActiveGuard;use crate::thread_status::ThreadWatchManager;use codex_app_server_protocol::AccountRateLimitsUpdatedNotification;use codex_app_server_protocol::AdditionalPermissionProfile as V2AdditionalPermissionProfile;use codex_app_server_protocol::CodexErrorInfo as V2CodexErrorInfo;use codex_app_server_protocol::CommandAction as V2ParsedCommand;use codex_app_server_protocol::CommandExecutionApprovalDecision;use codex_app_server_protocol::CommandExecutionRequestApprovalParams;use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;use codex_app_server_protocol::CommandExecutionSource;use codex_app_server_protocol::CommandExecutionStatus;use codex_app_server_protocol::DeprecationNoticeNotification;use codex_app_server_protocol::DynamicToolCallParams;use codex_app_server_protocol::DynamicToolCallStatus;use codex_app_server_protocol::ErrorNotification;use codex_app_server_protocol::ExecPolicyAmendment as V2ExecPolicyAmendment;use codex_app_server_protocol::FileChangeApprovalDecision;use codex_app_server_protocol::FileChangeRequestApprovalParams;use codex_app_server_protocol::FileChangeRequestApprovalResponse;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::ItemCompletedNotification;use codex_app_server_protocol::ItemStartedNotification;use codex_app_server_protocol::McpServerElicitationAction;use codex_app_server_protocol::McpServerElicitationRequestParams;use codex_app_server_protocol::McpServerElicitationRequestResponse;use codex_app_server_protocol::McpServerStartupState;use codex_app_server_protocol::McpServerStatusUpdatedNotification;use codex_app_server_protocol::ModelReroutedNotification;use codex_app_server_protocol::ModelVerificationNotification;use codex_app_server_protocol::NetworkApprovalContext as V2NetworkApprovalContext;use codex_app_server_protocol::NetworkPolicyAmendment as V2NetworkPolicyAmendment;use codex_app_server_protocol::NetworkPolicyRuleAction as V2NetworkPolicyRuleAction;use codex_app_server_protocol::PermissionsRequestApprovalParams;use codex_app_server_protocol::PermissionsRequestApprovalResponse;use codex_app_server_protocol::RawResponseItemCompletedNotification;use codex_app_server_protocol::RequestId;use codex_app_server_protocol::ServerNotification;use codex_app_server_protocol::ServerRequestPayload;use codex_app_server_protocol::ThreadGoalUpdatedNotification;use codex_app_server_protocol::ThreadItem;use codex_app_server_protocol::ThreadRealtimeClosedNotification;use codex_app_server_protocol::ThreadRealtimeErrorNotification;use codex_app_server_protocol::ThreadRealtimeItemAddedNotification;use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification;use codex_app_server_protocol::ThreadRealtimeSdpNotification;use codex_app_server_protocol::ThreadRealtimeStartedNotification;use codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification;use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification;use codex_app_server_protocol::ThreadRollbackResponse;use codex_app_server_protocol::ThreadSettingsUpdatedNotification;use codex_app_server_protocol::ThreadStatus;use codex_app_server_protocol::ThreadTokenUsage;use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;use codex_app_server_protocol::ToolRequestUserInputOption;use codex_app_server_protocol::ToolRequestUserInputParams;use codex_app_server_protocol::ToolRequestUserInputQuestion;use codex_app_server_protocol::ToolRequestUserInputResponse;use codex_app_server_protocol::Turn;use codex_app_server_protocol::TurnCompletedNotification;use codex_app_server_protocol::TurnDiffUpdatedNotification;use codex_app_server_protocol::TurnError;use codex_app_server_protocol::TurnInterruptResponse;use codex_app_server_protocol::TurnItemsView;use codex_app_server_protocol::TurnModerationMetadataNotification;use codex_app_server_protocol::TurnPlanStep;use codex_app_server_protocol::TurnPlanUpdatedNotification;use codex_app_server_protocol::TurnStartedNotification;use codex_app_server_protocol::TurnStatus;use codex_app_server_protocol::WarningNotification;use codex_app_server_protocol::build_item_from_guardian_event;use codex_app_server_protocol::guardian_auto_approval_review_notification;use codex_app_server_protocol::item_event_to_server_notification;use codex_core::CodexThread;use codex_core::ThreadManager;use codex_core::review_format::format_review_findings_block;use codex_core::review_prompts;use codex_protocol::ThreadId;use codex_protocol::items::parse_hook_prompt_message;use codex_protocol::models::AdditionalPermissionProfile as CoreAdditionalPermissionProfile;use codex_protocol::plan_tool::UpdatePlanArgs;use codex_protocol::protocol::CodexErrorInfo as CoreCodexErrorInfo;use codex_protocol::protocol::Event;use codex_protocol::protocol::EventMsg;use codex_protocol::protocol::ExecApprovalRequestEvent;use codex_protocol::protocol::Op;use codex_protocol::protocol::RealtimeEvent;use codex_protocol::protocol::ReviewDecision;use codex_protocol::protocol::ReviewOutputEvent;use codex_protocol::protocol::SubAgentActivityKind;use codex_protocol::protocol::TokenCountEvent;use codex_protocol::protocol::TurnAbortedEvent;use codex_protocol::protocol::TurnCompleteEvent;use codex_protocol::protocol::TurnDiffEvent;use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope;use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile;use codex_protocol::request_permissions::RequestPermissionsResponse as CoreRequestPermissionsResponse;use codex_protocol::request_user_input::RequestUserInputAnswer as CoreRequestUserInputAnswer;use codex_protocol::request_user_input::RequestUserInputResponse as CoreRequestUserInputResponse;use codex_sandboxing::policy_transforms::intersect_permission_profiles;use codex_shell_command::parse_command::shlex_join;use codex_utils_absolute_path::AbsolutePathBuf;use std::collections::HashMap;use std::sync::Arc;use std::time::SystemTime;use std::time::UNIX_EPOCH;use tokio::sync::Mutex;use tokio::sync::oneshot;use tracing::error;enum CommandExecutionApprovalPresentation { Network(V2NetworkApprovalContext), Command(CommandExecutionCompletionItem),}#[derive(Debug, PartialEq)]struct CommandExecutionCompletionItem { command: String, cwd: AbsolutePathBuf, command_actions: Vec<V2ParsedCommand>,}#[allow(clippy::too_many_arguments)]pub(crate) async fn apply_bespoke_event_handling( event: Event, conversation_id: ThreadId, conversation: Arc<CodexThread>, thread_manager: Arc<ThreadManager>, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc<tokio::sync::Mutex<ThreadState>>, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc<tokio::sync::Semaphore>, fallback_model_provider: String,) { let Event { id: event_turn_id, msg, } = event; match msg { EventMsg::TurnStarted(payload) => { // While not technically necessary as it was already done on TurnComplete, be extra cautios and abort any pending server requests. outgoing.abort_pending_server_requests().await; thread_watch_manager .note_turn_started(&conversation_id.to_string()) .await; let turn = { let state = thread_state.lock().await; let mut turn = state.active_turn_snapshot().unwrap_or_else(|| Turn { id: payload.turn_id.clone(), items: Vec::new(), items_view: TurnItemsView::NotLoaded, error: None, status: TurnStatus::InProgress, started_at: payload.started_at, completed_at: None, duration_ms: None, }); turn.items.clear(); turn.items_view = TurnItemsView::NotLoaded; turn }; let notification = TurnStartedNotification { thread_id: conversation_id.to_string(), turn, }; outgoing .send_server_notification(ServerNotification::TurnStarted(notification)) .await; } EventMsg::TurnComplete(turn_complete_event) => { // All per-thread requests are bound to a turn, so abort them. outgoing.abort_pending_server_requests().await; respond_to_pending_interrupts(&thread_state, &outgoing).await; let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some(); thread_watch_manager .note_turn_completed(&conversation_id.to_string(), turn_failed) .await; handle_turn_complete( conversation_id, event_turn_id, turn_complete_event, &outgoing, &thread_state, ) .await; } EventMsg::McpStartupUpdate(update) => { let (status, error) = match update.status { codex_protocol::protocol::McpStartupStatus::Starting => { (McpServerStartupState::Starting, None) } codex_protocol::protocol::McpStartupStatus::Ready => { (McpServerStartupState::Ready, None) } codex_protocol::protocol::McpStartupStatus::Failed { error } => { (McpServerStartupState::Failed, Some(error)) } codex_protocol::protocol::McpStartupStatus::Cancelled => { (McpServerStartupState::Cancelled, None) } }; let notification = McpServerStatusUpdatedNotification { thread_id: Some(conversation_id.to_string()), name: update.server, status, error, }; outgoing .send_server_notification(ServerNotification::McpServerStatusUpdated(notification)) .await; } EventMsg::Warning(warning_event) => { let notification = WarningNotification { thread_id: Some(conversation_id.to_string()), message: warning_event.message, }; outgoing .send_server_notification(ServerNotification::Warning(notification)) .await; } EventMsg::GuardianWarning(warning_event) => { let notification = GuardianWarningNotification { thread_id: conversation_id.to_string(), message: warning_event.message, }; outgoing .send_server_notification(ServerNotification::GuardianWarning(notification)) .await; } EventMsg::GuardianAssessment(assessment) => { let pending_command_execution = match build_item_from_guardian_event( &assessment, CommandExecutionStatus::InProgress, ) { Some(ThreadItem::CommandExecution { id, command, cwd, command_actions, .. }) => Some(( id, CommandExecutionCompletionItem { command, cwd, command_actions, }, )), Some(_) | None => None, }; let assessment_turn_id = if assessment.turn_id.is_empty() { event_turn_id.clone() } else { assessment.turn_id.clone() }; if assessment.status == codex_protocol::protocol::GuardianAssessmentStatus::InProgress && let Some((target_item_id, completion_item)) = pending_command_execution.as_ref() { start_command_execution_item( &conversation_id, assessment_turn_id.clone(), target_item_id.clone(), completion_item.command.clone(), completion_item.cwd.clone(), completion_item.command_actions.clone(), CommandExecutionSource::Agent, &outgoing, &thread_state, ) .await; } let notification = guardian_auto_approval_review_notification( &conversation_id, &event_turn_id, &assessment, ); outgoing.send_server_notification(notification).await; let completion_status = match assessment.status { codex_protocol::protocol::GuardianAssessmentStatus::Denied | codex_protocol::protocol::GuardianAssessmentStatus::Aborted => { Some(CommandExecutionStatus::Declined) } codex_protocol::protocol::GuardianAssessmentStatus::TimedOut => { Some(CommandExecutionStatus::Failed) } codex_protocol::protocol::GuardianAssessmentStatus::InProgress | codex_protocol::protocol::GuardianAssessmentStatus::Approved => None, }; if let Some(completion_status) = completion_status && let Some((target_item_id, completion_item)) = pending_command_execution { complete_command_execution_item( &conversation_id, assessment_turn_id, target_item_id, completion_item.command, completion_item.cwd, /*process_id*/ None, CommandExecutionSource::Agent, completion_item.command_actions, completion_status, &outgoing, &thread_state, ) .await; } } EventMsg::ModelReroute(event) => { let notification = ModelReroutedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), from_model: event.from_model, to_model: event.to_model, reason: event.reason.into(), }; outgoing .send_server_notification(ServerNotification::ModelRerouted(notification)) .await; } EventMsg::ModelVerification(event) => { let notification = ModelVerificationNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), verifications: event.verifications.into_iter().map(Into::into).collect(), }; outgoing .send_server_notification(ServerNotification::ModelVerification(notification)) .await; } EventMsg::TurnModerationMetadata(event) => { let notification = TurnModerationMetadataNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), metadata: event.metadata, }; outgoing .send_server_notification(ServerNotification::TurnModerationMetadata(notification)) .await; } EventMsg::RealtimeConversationStarted(event) => { let notification = ThreadRealtimeStartedNotification { thread_id: conversation_id.to_string(), realtime_session_id: event.realtime_session_id, version: event.version, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeStarted(notification)) .await; } EventMsg::RealtimeConversationSdp(event) => { let notification = ThreadRealtimeSdpNotification { thread_id: conversation_id.to_string(), sdp: event.sdp, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeSdp(notification)) .await; } EventMsg::RealtimeConversationRealtime(event) => match event.payload { RealtimeEvent::SessionUpdated { .. } => {} RealtimeEvent::InputAudioSpeechStarted(event) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), item: serde_json::json!({ "type": "input_audio_buffer.speech_started", "item_id": event.item_id, }), }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( notification, )) .await; } RealtimeEvent::InputTranscriptDelta(event) => { let notification = ThreadRealtimeTranscriptDeltaNotification { thread_id: conversation_id.to_string(), role: "user".to_string(), delta: event.delta, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeTranscriptDelta( notification, )) .await; } RealtimeEvent::InputTranscriptDone(event) => { let notification = ThreadRealtimeTranscriptDoneNotification { thread_id: conversation_id.to_string(), role: "user".to_string(), text: event.text, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeTranscriptDone( notification, )) .await; } RealtimeEvent::OutputTranscriptDelta(event) => { let notification = ThreadRealtimeTranscriptDeltaNotification { thread_id: conversation_id.to_string(), role: "assistant".to_string(), delta: event.delta, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeTranscriptDelta( notification, )) .await; } RealtimeEvent::OutputTranscriptDone(event) => { let notification = ThreadRealtimeTranscriptDoneNotification { thread_id: conversation_id.to_string(), role: "assistant".to_string(), text: event.text, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeTranscriptDone( notification, )) .await; } RealtimeEvent::AudioOut(audio) => { let notification = ThreadRealtimeOutputAudioDeltaNotification { thread_id: conversation_id.to_string(), audio: audio.into(), }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeOutputAudioDelta( notification, )) .await; } RealtimeEvent::ResponseCreated(_) => {} RealtimeEvent::ResponseCancelled(event) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), item: serde_json::json!({ "type": "response.cancelled", "response_id": event.response_id, }), }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( notification, )) .await; } RealtimeEvent::ResponseDone(_) => {} RealtimeEvent::ConversationItemAdded(item) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), item, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( notification, )) .await; } RealtimeEvent::ConversationItemDone { .. } | RealtimeEvent::NoopRequested(_) => {} RealtimeEvent::HandoffRequested(handoff) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), item: serde_json::json!({ "type": "handoff_request", "handoff_id": handoff.handoff_id, "item_id": handoff.item_id, "input_transcript": handoff.input_transcript, "active_transcript": handoff.active_transcript, }), }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( notification, )) .await; } RealtimeEvent::Error(message) => { let notification = ThreadRealtimeErrorNotification { thread_id: conversation_id.to_string(), message, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeError(notification)) .await; } }, EventMsg::RealtimeConversationClosed(event) => { let notification = ThreadRealtimeClosedNotification { thread_id: conversation_id.to_string(), reason: event.reason, }; outgoing .send_server_notification(ServerNotification::ThreadRealtimeClosed(notification)) .await; } EventMsg::ApplyPatchApprovalRequest(event) => { let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; let item_id = event.call_id.clone(); let params = FileChangeRequestApprovalParams { thread_id: conversation_id.to_string(), turn_id: event.turn_id.clone(), item_id: item_id.clone(), started_at_ms: event.started_at_ms, reason: event.reason.clone(), grant_root: event.grant_root.clone(), }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::FileChangeRequestApproval(params)) .await; tokio::spawn(async move { on_file_change_request_approval_response( item_id, pending_request_id, rx, conversation, thread_state.clone(), permission_guard, ) .await; }); } EventMsg::ExecApprovalRequest(ev) => { let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; let available_decisions = ev .effective_available_decisions() .into_iter() .map(CommandExecutionApprovalDecision::from) .collect::<Vec<_>>(); let ExecApprovalRequestEvent { call_id, approval_id, turn_id, started_at_ms, command, cwd, reason, network_approval_context, proposed_execpolicy_amendment, proposed_network_policy_amendments, additional_permissions, parsed_cmd, .. } = ev; let command_actions = parsed_cmd .iter() .cloned() .map(|parsed| V2ParsedCommand::from_core_with_cwd(parsed, &cwd)) .collect::<Vec<_>>(); let presentation = if let Some(network_approval_context) = network_approval_context.map(V2NetworkApprovalContext::from) { CommandExecutionApprovalPresentation::Network(network_approval_context) } else { let command_string = shlex_join(&command); let completion_item = CommandExecutionCompletionItem { command: command_string, cwd: cwd.clone(), command_actions: command_actions.clone(), }; CommandExecutionApprovalPresentation::Command(completion_item) }; let (network_approval_context, command, cwd, command_actions, completion_item) = match presentation { CommandExecutionApprovalPresentation::Network(network_approval_context) => { (Some(network_approval_context), None, None, None, None) } CommandExecutionApprovalPresentation::Command(completion_item) => ( None, Some(completion_item.command.clone()), Some(completion_item.cwd.clone()), Some(completion_item.command_actions.clone()), Some(completion_item), ), }; if approval_id.is_none() && let Some(completion_item) = completion_item.as_ref() { start_command_execution_item( &conversation_id, event_turn_id.clone(), call_id.clone(), completion_item.command.clone(), completion_item.cwd.clone(), completion_item.command_actions.clone(), CommandExecutionSource::Agent, &outgoing, &thread_state, ) .await; } let proposed_execpolicy_amendment_v2 = proposed_execpolicy_amendment.map(V2ExecPolicyAmendment::from); let proposed_network_policy_amendments_v2 = proposed_network_policy_amendments.map(|amendments| { amendments .into_iter() .map(V2NetworkPolicyAmendment::from) .collect() }); let additional_permissions = additional_permissions.map(V2AdditionalPermissionProfile::from); let params = CommandExecutionRequestApprovalParams { thread_id: conversation_id.to_string(), turn_id: turn_id.clone(), item_id: call_id.clone(), started_at_ms, approval_id: approval_id.clone(), reason, network_approval_context, command, cwd, command_actions, additional_permissions, proposed_execpolicy_amendment: proposed_execpolicy_amendment_v2, proposed_network_policy_amendments: proposed_network_policy_amendments_v2, available_decisions: Some(available_decisions), }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::CommandExecutionRequestApproval( params, )) .await; tokio::spawn(async move { on_command_execution_request_approval_response( event_turn_id, conversation_id, approval_id, call_id, completion_item, pending_request_id, rx, conversation, outgoing, thread_state.clone(), permission_guard, ) .await; }); } EventMsg::RequestUserInput(request) => { let user_input_guard = thread_watch_manager .note_user_input_requested(&conversation_id.to_string()) .await; let questions = request .questions .into_iter() .map(|question| ToolRequestUserInputQuestion { id: question.id, header: question.header, question: question.question, is_other: question.is_other, is_secret: question.is_secret, options: question.options.map(|options| { options .into_iter() .map(|option| ToolRequestUserInputOption { label: option.label, description: option.description, }) .collect() }), }) .collect(); let params = ToolRequestUserInputParams { thread_id: conversation_id.to_string(), turn_id: request.turn_id, item_id: request.call_id, questions, auto_resolution_ms: request.auto_resolution_ms, }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::ToolRequestUserInput(params)) .await; tokio::spawn(async move { on_request_user_input_response( event_turn_id, pending_request_id, rx, conversation, thread_state, user_input_guard, ) .await; }); } EventMsg::ElicitationRequest(request) => { let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; let turn_id = match request.turn_id.clone() { Some(turn_id) => Some(turn_id), None => { let state = thread_state.lock().await; state.active_turn_snapshot().map(|turn| turn.id) } }; let server_name = request.server_name.clone(); let request_body = match request.request.try_into() { Ok(request_body) => request_body, Err(err) => { error!( error = %err, server_name, request_id = ?request.id, "failed to parse typed MCP elicitation schema" ); if let Err(err) = conversation .submit(Op::ResolveElicitation { server_name: request.server_name, request_id: request.id, decision: codex_protocol::approvals::ElicitationAction::Cancel, content: None, meta: None, }) .await { error!("failed to submit ResolveElicitation: {err}"); } return; } }; let params = McpServerElicitationRequestParams { thread_id: conversation_id.to_string(), turn_id, server_name: request.server_name.clone(), request: request_body, }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::McpServerElicitationRequest(params)) .await; tokio::spawn(async move { on_mcp_server_elicitation_response( request.server_name, request.id, pending_request_id, rx, conversation, thread_state, permission_guard, ) .await; }); } EventMsg::RequestPermissions(request) => { let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; let requested_permissions = request.permissions.clone(); let request_cwd = match request.cwd.clone() { Some(cwd) => cwd, None => conversation.config_snapshot().await.cwd().clone(), }; let params = PermissionsRequestApprovalParams { thread_id: conversation_id.to_string(), turn_id: request.turn_id.clone(), item_id: request.call_id.clone(), environment_id: request.environment_id.clone(), started_at_ms: request.started_at_ms, cwd: request_cwd.clone(), reason: request.reason, permissions: request.permissions.into(), }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::PermissionsRequestApproval(params)) .await; let pending_response = PendingRequestPermissionsResponse { call_id: request.call_id, conversation_id, turn_id: request.turn_id, requested_permissions, request_cwd, pending_request_id, outgoing, receiver: rx, request_permissions_guard: permission_guard, }; tokio::spawn(async move { on_request_permissions_response(pending_response, conversation, thread_state).await; }); } EventMsg::DynamicToolCallRequest(request) => { let call_id = request.call_id; let turn_id = request.turn_id; let namespace = request.namespace; let tool = request.tool; let arguments = request.arguments; let item = ThreadItem::DynamicToolCall { id: call_id.clone(), namespace: namespace.clone(), tool: tool.clone(), arguments: arguments.clone(), status: DynamicToolCallStatus::InProgress, content_items: None, success: None, duration_ms: None, }; let notification = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id: turn_id.clone(), started_at_ms: request.started_at_ms, item, }; outgoing .send_server_notification(ServerNotification::ItemStarted(notification)) .await; let params = DynamicToolCallParams { thread_id: conversation_id.to_string(), turn_id: turn_id.clone(), call_id: call_id.clone(), namespace, tool: tool.clone(), arguments: arguments.clone(), }; let (_pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::DynamicToolCall(params)) .await; tokio::spawn(async move { crate::dynamic_tools::on_call_response(call_id, rx, conversation).await; }); } EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_) => { // Deprecated MCP tool-call events are still fanned out for legacy clients. // App-server v2 receives the canonical TurnItem::McpToolCall lifecycle instead. } msg @ (EventMsg::DynamicToolCallResponse(_) | EventMsg::CollabAgentSpawnBegin(_) | EventMsg::CollabAgentSpawnEnd(_) | EventMsg::CollabAgentInteractionBegin(_) | EventMsg::CollabAgentInteractionEnd(_) | EventMsg::CollabWaitingBegin(_) | EventMsg::CollabWaitingEnd(_) | EventMsg::CollabCloseBegin(_) | EventMsg::CollabResumeBegin(_) | EventMsg::CollabResumeEnd(_) | EventMsg::AgentMessageContentDelta(_) | EventMsg::PlanDelta(_) | EventMsg::ReasoningContentDelta(_) | EventMsg::ReasoningRawContentDelta(_) | EventMsg::AgentReasoningSectionBreak(_)) => { let notification = item_event_to_server_notification( msg, &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } EventMsg::SubAgentActivity(activity) => { if activity.kind == SubAgentActivityKind::Interrupted && thread_manager .get_thread(activity.agent_thread_id) .await .is_err() { thread_watch_manager .remove_thread(&activity.agent_thread_id.to_string()) .await; } let notification = item_event_to_server_notification( EventMsg::SubAgentActivity(activity), &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } EventMsg::CollabCloseEnd(end_event) => { if thread_manager .get_thread(end_event.receiver_thread_id) .await .is_err() { thread_watch_manager .remove_thread(&end_event.receiver_thread_id.to_string()) .await; } let notification = item_event_to_server_notification( EventMsg::CollabCloseEnd(end_event), &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } EventMsg::ContextCompacted(..) => { // Core still fans out this deprecated event for legacy clients; // v2 clients receive the canonical ContextCompaction item instead. } EventMsg::DeprecationNotice(event) => { let notification = DeprecationNoticeNotification { summary: event.summary, details: event.details, }; outgoing .send_server_notification(ServerNotification::DeprecationNotice(notification)) .await; } EventMsg::TokenCount(token_count_event) => { handle_token_count_event(conversation_id, event_turn_id, token_count_event, &outgoing) .await; } EventMsg::Error(ev) => { thread_watch_manager .note_system_error(&conversation_id.to_string()) .await; let message = ev.message.clone(); let codex_error_info = ev.codex_error_info.clone(); // If this error belongs to an in-flight `thread/rollback` request, fail that request // (and clear pending state) so subsequent rollbacks are unblocked. // // Don't send a notification for this error. if matches!( codex_error_info, Some(CoreCodexErrorInfo::ThreadRollbackFailed) ) { return handle_thread_rollback_failed( conversation_id, message, &thread_state, &outgoing, ) .await; }; if !ev.affects_turn_status() { return; } let turn_error = TurnError { message: ev.message, codex_error_info: ev.codex_error_info.map(V2CodexErrorInfo::from), additional_details: None, }; handle_error_notification( conversation_id, &event_turn_id, turn_error, &outgoing, &thread_state, ) .await; } EventMsg::StreamError(ev) => { // We don't need to update the turn summary store for stream errors as they are intermediate error states for retries, // but we notify the client. let turn_error = TurnError { message: ev.message, codex_error_info: ev.codex_error_info.map(V2CodexErrorInfo::from), additional_details: ev.additional_details, }; outgoing .send_server_notification(ServerNotification::Error(ErrorNotification { error: turn_error, will_retry: true, thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), })) .await; } EventMsg::ViewImageToolCall(_) => {} EventMsg::EnteredReviewMode(review_request) => { let review = review_request .user_facing_hint .unwrap_or_else(|| review_prompts::user_facing_hint(&review_request.target)); let item = ThreadItem::EnteredReviewMode { id: event_turn_id.clone(), review, }; let started = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), started_at_ms: now_unix_timestamp_ms(), item: item.clone(), }; outgoing .send_server_notification(ServerNotification::ItemStarted(started)) .await; let completed = ItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), completed_at_ms: now_unix_timestamp_ms(), item, }; outgoing .send_server_notification(ServerNotification::ItemCompleted(completed)) .await; } msg @ (EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) | EventMsg::PatchApplyUpdated(_) | EventMsg::TerminalInteraction(_)) => { let notification = item_event_to_server_notification( msg, &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } EventMsg::HookStarted(event) => { let notification = HookStartedNotification { thread_id: conversation_id.to_string(), turn_id: event.turn_id, run: event.run.into(), }; outgoing .send_server_notification(ServerNotification::HookStarted(notification)) .await; } EventMsg::HookCompleted(event) => { let notification = HookCompletedNotification { thread_id: conversation_id.to_string(), turn_id: event.turn_id, run: event.run.into(), }; outgoing .send_server_notification(ServerNotification::HookCompleted(notification)) .await; } EventMsg::ExitedReviewMode(review_event) => { let review = match review_event.review_output { Some(output) => render_review_output_text(&output), None => REVIEW_FALLBACK_MESSAGE.to_string(), }; let item = ThreadItem::ExitedReviewMode { id: event_turn_id.clone(), review, }; let started = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), started_at_ms: now_unix_timestamp_ms(), item: item.clone(), }; outgoing .send_server_notification(ServerNotification::ItemStarted(started)) .await; let completed = ItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), completed_at_ms: now_unix_timestamp_ms(), item, }; outgoing .send_server_notification(ServerNotification::ItemCompleted(completed)) .await; } EventMsg::RawResponseItem(raw_response_item_event) => { maybe_emit_hook_prompt_item_completed( conversation_id, &event_turn_id, &raw_response_item_event.item, &outgoing, ) .await; maybe_emit_raw_response_item_completed( conversation_id, &event_turn_id, raw_response_item_event.item, &outgoing, ) .await; } EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) => { // Core still fans out these deprecated events for legacy clients; // v2 clients receive the canonical FileChange item instead. } EventMsg::ExecCommandBegin(exec_command_begin_event) => { if matches!( exec_command_begin_event.source, codex_protocol::protocol::ExecCommandSource::UnifiedExecInteraction ) { // TerminalInteraction is the v2 surface for unified exec // stdin/poll events. Suppress the legacy CommandExecution // item so clients do not render the same wait twice. return; } let item_id = exec_command_begin_event.call_id.clone(); let first_start = { let mut state = thread_state.lock().await; state .turn_summary .command_execution_started .insert(item_id.clone()) }; if first_start { let notification = item_event_to_server_notification( EventMsg::ExecCommandBegin(exec_command_begin_event), &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } } EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event) => { let notification = item_event_to_server_notification( EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event), &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } EventMsg::ExecCommandEnd(exec_command_end_event) => { let call_id = exec_command_end_event.call_id.clone(); { let mut state = thread_state.lock().await; state .turn_summary .command_execution_started .remove(&call_id); } if matches!( exec_command_end_event.source, codex_protocol::protocol::ExecCommandSource::UnifiedExecInteraction ) { // The paired begin event is suppressed above; keep the // completion out of v2 as well so no orphan legacy item is // emitted for unified exec interactions. return; } let notification = item_event_to_server_notification( EventMsg::ExecCommandEnd(exec_command_end_event), &conversation_id.to_string(), &event_turn_id, ); outgoing.send_server_notification(notification).await; } // If this is a TurnAborted, reply to any pending interrupt requests. EventMsg::TurnAborted(turn_aborted_event) => { // All per-thread requests are bound to a turn, so abort them. outgoing.abort_pending_server_requests().await; respond_to_pending_interrupts(&thread_state, &outgoing).await; thread_watch_manager .note_turn_interrupted(&conversation_id.to_string()) .await; handle_turn_interrupted( conversation_id, event_turn_id, turn_aborted_event, &outgoing, &thread_state, ) .await; } EventMsg::ThreadRolledBack(_rollback_event) => { let pending = { let mut state = thread_state.lock().await; state.pending_rollbacks.take() }; if let Some(request_id) = pending { let _thread_list_state_permit = match thread_list_state_permit.acquire().await { Ok(permit) => permit, Err(err) => { outgoing .send_error( request_id, internal_error(format!( "failed to acquire thread list state permit: {err}" )), ) .await; return; } }; let fallback_cwd = conversation.config_snapshot().await.cwd().clone(); let stored_thread = match conversation .read_thread( /*include_archived*/ true, /*include_history*/ true, ) .await { Ok(stored_thread) => stored_thread, Err(err) => { outgoing .send_error( request_id.clone(), internal_error(format!( "failed to read thread {conversation_id} after rollback: {err}" )), ) .await; return; } }; let loaded_status = thread_watch_manager .loaded_status_for_thread(&conversation_id.to_string()) .await; let response = match thread_rollback_response_from_stored_thread( stored_thread, conversation.session_configured().session_id.to_string(), fallback_model_provider.as_str(), &fallback_cwd, loaded_status, ) { Ok(response) => response, Err(err) => { outgoing .send_error(request_id.clone(), internal_error(err)) .await; return; } }; outgoing.send_response(request_id, response).await; } } EventMsg::ThreadGoalUpdated(thread_goal_event) => { let notification = ThreadGoalUpdatedNotification { thread_id: thread_goal_event.thread_id.to_string(), turn_id: thread_goal_event.turn_id, goal: thread_goal_event.goal.clone().into(), }; outgoing .send_global_server_notification(ServerNotification::ThreadGoalUpdated( notification, )) .await; } EventMsg::ThreadSettingsApplied(thread_settings_event) => { let thread_settings = thread_settings_from_core_snapshot(thread_settings_event.thread_settings); let changed = { let mut state = thread_state.lock().await; state.note_thread_settings(thread_settings.clone()) }; if changed { outgoing .send_server_notification(ServerNotification::ThreadSettingsUpdated( ThreadSettingsUpdatedNotification { thread_id: conversation_id.to_string(), thread_settings, }, )) .await; } } EventMsg::TurnDiff(turn_diff_event) => { handle_turn_diff(conversation_id, &event_turn_id, turn_diff_event, &outgoing).await; } EventMsg::PlanUpdate(plan_update_event) => { handle_turn_plan_update( conversation_id, &event_turn_id, plan_update_event, &outgoing, ) .await; } EventMsg::ShutdownComplete => { thread_watch_manager .note_thread_shutdown(&conversation_id.to_string()) .await; } _ => {} }}async fn handle_turn_diff( conversation_id: ThreadId, event_turn_id: &str, turn_diff_event: TurnDiffEvent, outgoing: &ThreadScopedOutgoingMessageSender,) { let notification = TurnDiffUpdatedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.to_string(), diff: turn_diff_event.unified_diff, }; outgoing .send_server_notification(ServerNotification::TurnDiffUpdated(notification)) .await;}async fn handle_turn_plan_update( conversation_id: ThreadId, event_turn_id: &str, plan_update_event: UpdatePlanArgs, outgoing: &ThreadScopedOutgoingMessageSender,) { // `update_plan` is a todo/checklist tool; it is not related to plan-mode updates let notification = TurnPlanUpdatedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.to_string(), explanation: plan_update_event.explanation, plan: plan_update_event .plan .into_iter() .map(TurnPlanStep::from) .collect(), }; outgoing .send_server_notification(ServerNotification::TurnPlanUpdated(notification)) .await;}struct TurnCompletionMetadata { status: TurnStatus, error: Option<TurnError>, started_at: Option<i64>, completed_at: Option<i64>, duration_ms: Option<i64>,}async fn emit_turn_completed_with_status( conversation_id: ThreadId, event_turn_id: String, turn_completion_metadata: TurnCompletionMetadata, outgoing: &ThreadScopedOutgoingMessageSender,) { let notification = TurnCompletedNotification { thread_id: conversation_id.to_string(), turn: Turn { id: event_turn_id, items: vec![], items_view: TurnItemsView::NotLoaded, error: turn_completion_metadata.error, status: turn_completion_metadata.status, started_at: turn_completion_metadata.started_at, completed_at: turn_completion_metadata.completed_at, duration_ms: turn_completion_metadata.duration_ms, }, }; outgoing .send_server_notification(ServerNotification::TurnCompleted(notification)) .await;}#[allow(clippy::too_many_arguments)]async fn start_command_execution_item( conversation_id: &ThreadId, turn_id: String, item_id: String, command: String, cwd: AbsolutePathBuf, command_actions: Vec<V2ParsedCommand>, source: CommandExecutionSource, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc<Mutex<ThreadState>>,) -> bool { let first_start = { let mut state = thread_state.lock().await; state .turn_summary .command_execution_started .insert(item_id.clone()) }; if first_start { let notification = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id, started_at_ms: now_unix_timestamp_ms(), item: ThreadItem::CommandExecution { id: item_id, command, cwd, process_id: None, source, status: CommandExecutionStatus::InProgress, command_actions, aggregated_output: None, exit_code: None, duration_ms: None, }, }; outgoing .send_server_notification(ServerNotification::ItemStarted(notification)) .await; } first_start}#[allow(clippy::too_many_arguments)]async fn complete_command_execution_item( conversation_id: &ThreadId, turn_id: String, item_id: String, command: String, cwd: AbsolutePathBuf, process_id: Option<String>, source: CommandExecutionSource, command_actions: Vec<V2ParsedCommand>, status: CommandExecutionStatus, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc<Mutex<ThreadState>>,) { let should_emit = thread_state .lock() .await .turn_summary .command_execution_started .remove(&item_id); if !should_emit { return; } let item = ThreadItem::CommandExecution { id: item_id, command, cwd, process_id, source, status, command_actions, aggregated_output: None, exit_code: None, duration_ms: None, }; let notification = ItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id, completed_at_ms: now_unix_timestamp_ms(), item, }; outgoing .send_server_notification(ServerNotification::ItemCompleted(notification)) .await;}async fn maybe_emit_raw_response_item_completed( conversation_id: ThreadId, turn_id: &str, item: codex_protocol::models::ResponseItem, outgoing: &ThreadScopedOutgoingMessageSender,) { let notification = RawResponseItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id: turn_id.to_string(), item, }; outgoing .send_server_notification(ServerNotification::RawResponseItemCompleted(notification)) .await;}pub(crate) async fn maybe_emit_hook_prompt_item_completed( conversation_id: ThreadId, turn_id: &str, item: &codex_protocol::models::ResponseItem, outgoing: &ThreadScopedOutgoingMessageSender,) { let codex_protocol::models::ResponseItem::Message { role, content, id, .. } = item else { return; }; if role != "user" { return; } let Some(hook_prompt) = parse_hook_prompt_message(id.as_ref(), content) else { return; }; let notification = ItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id: turn_id.to_string(), completed_at_ms: now_unix_timestamp_ms(), item: ThreadItem::HookPrompt { id: hook_prompt.id, fragments: hook_prompt .fragments .into_iter() .map(codex_app_server_protocol::HookPromptFragment::from) .collect(), }, }; outgoing .send_server_notification(ServerNotification::ItemCompleted(notification)) .await;}async fn find_and_remove_turn_summary( _conversation_id: ThreadId, thread_state: &Arc<Mutex<ThreadState>>,) -> TurnSummary { let mut state = thread_state.lock().await; std::mem::take(&mut state.turn_summary)}async fn handle_turn_complete( conversation_id: ThreadId, event_turn_id: String, turn_complete_event: TurnCompleteEvent, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc<Mutex<ThreadState>>,) { let turn_summary = find_and_remove_turn_summary(conversation_id, thread_state).await; let (status, error) = match turn_summary.last_error { Some(error) => (TurnStatus::Failed, Some(error)), None => (TurnStatus::Completed, None), }; emit_turn_completed_with_status( conversation_id, event_turn_id, TurnCompletionMetadata { status, error, started_at: turn_summary.started_at, completed_at: turn_complete_event.completed_at, duration_ms: turn_complete_event.duration_ms, }, outgoing, ) .await;}async fn handle_turn_interrupted( conversation_id: ThreadId, event_turn_id: String, turn_aborted_event: TurnAbortedEvent, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc<Mutex<ThreadState>>,) { let turn_summary = find_and_remove_turn_summary(conversation_id, thread_state).await; emit_turn_completed_with_status( conversation_id, event_turn_id, TurnCompletionMetadata { status: TurnStatus::Interrupted, error: None, started_at: turn_summary.started_at, completed_at: turn_aborted_event.completed_at, duration_ms: turn_aborted_event.duration_ms, }, outgoing, ) .await;}async fn handle_thread_rollback_failed( _conversation_id: ThreadId, message: String, thread_state: &Arc<Mutex<ThreadState>>, outgoing: &ThreadScopedOutgoingMessageSender,) { let pending_rollback = thread_state.lock().await.pending_rollbacks.take(); if let Some(request_id) = pending_rollback { outgoing .send_error(request_id, invalid_request(message)) .await; }}fn thread_rollback_response_from_stored_thread( stored_thread: codex_thread_store::StoredThread, session_id: String, fallback_model_provider: &str, fallback_cwd: &AbsolutePathBuf, loaded_status: ThreadStatus,) -> std::result::Result<ThreadRollbackResponse, String> { let thread_id = stored_thread.thread_id; let (mut thread, history) = thread_from_stored_thread(stored_thread, fallback_model_provider, fallback_cwd); thread.session_id = session_id; let Some(history) = history else { return Err(format!( "thread {thread_id} did not include persisted history after rollback" )); }; populate_thread_turns_from_history(&mut thread, &history.items, /*active_turn*/ None); thread.status = loaded_status; Ok(ThreadRollbackResponse { thread })}async fn respond_to_pending_interrupts( thread_state: &Arc<Mutex<ThreadState>>, outgoing: &ThreadScopedOutgoingMessageSender,) { let pending = { let mut state = thread_state.lock().await; std::mem::take(&mut state.pending_interrupts) }; for request_id in pending { outgoing .send_response(request_id, TurnInterruptResponse {}) .await; }}async fn handle_token_count_event( conversation_id: ThreadId, turn_id: String, token_count_event: TokenCountEvent, outgoing: &ThreadScopedOutgoingMessageSender,) { let TokenCountEvent { info, rate_limits } = token_count_event; if let Some(token_usage) = info.map(ThreadTokenUsage::from) { let notification = ThreadTokenUsageUpdatedNotification { thread_id: conversation_id.to_string(), turn_id, token_usage, }; outgoing .send_server_notification(ServerNotification::ThreadTokenUsageUpdated(notification)) .await; } if let Some(rate_limits) = rate_limits { outgoing .send_server_notification(ServerNotification::AccountRateLimitsUpdated( AccountRateLimitsUpdatedNotification { rate_limits: rate_limits.into(), }, )) .await; }}async fn handle_error( _conversation_id: ThreadId, error: TurnError, thread_state: &Arc<Mutex<ThreadState>>,) { let mut state = thread_state.lock().await; state.turn_summary.last_error = Some(error);}async fn handle_error_notification( conversation_id: ThreadId, event_turn_id: &str, error: TurnError, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc<Mutex<ThreadState>>,) { handle_error(conversation_id, error.clone(), thread_state).await; outgoing .send_server_notification(ServerNotification::Error(ErrorNotification { error, will_retry: false, thread_id: conversation_id.to_string(), turn_id: event_turn_id.to_string(), })) .await;}async fn on_request_user_input_response( event_turn_id: String, pending_request_id: RequestId, receiver: oneshot::Receiver<ClientRequestResult>, conversation: Arc<CodexThread>, thread_state: Arc<Mutex<ThreadState>>, user_input_guard: ThreadWatchActiveGuard,) { let response = receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await; drop(user_input_guard); let value = match response { Ok(Ok(value)) => value, Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return, Ok(Err(err)) => { error!("request failed with client error: {err:?}"); let empty = CoreRequestUserInputResponse { answers: HashMap::new(), }; if let Err(err) = conversation .submit(Op::UserInputAnswer { id: event_turn_id, response: empty, }) .await { error!("failed to submit UserInputAnswer: {err}"); } return; } Err(err) => { error!("request failed: {err:?}"); let empty = CoreRequestUserInputResponse { answers: HashMap::new(), }; if let Err(err) = conversation .submit(Op::UserInputAnswer { id: event_turn_id, response: empty, }) .await { error!("failed to submit UserInputAnswer: {err}"); } return; } }; let response = serde_json::from_value::<ToolRequestUserInputResponse>(value).unwrap_or_else(|err| { error!("failed to deserialize ToolRequestUserInputResponse: {err}"); ToolRequestUserInputResponse { answers: HashMap::new(), } }); let response = CoreRequestUserInputResponse { answers: response .answers .into_iter() .map(|(id, answer)| { ( id, CoreRequestUserInputAnswer { answers: answer.answers, }, ) }) .collect(), }; if let Err(err) = conversation .submit(Op::UserInputAnswer { id: event_turn_id, response, }) .await { error!("failed to submit UserInputAnswer: {err}"); }}async fn on_mcp_server_elicitation_response( server_name: String, request_id: codex_protocol::mcp::RequestId, pending_request_id: RequestId, receiver: oneshot::Receiver<ClientRequestResult>, conversation: Arc<CodexThread>, thread_state: Arc<Mutex<ThreadState>>, permission_guard: ThreadWatchActiveGuard,) { let response = receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await; drop(permission_guard); let response = mcp_server_elicitation_response_from_client_result(response); if let Err(err) = conversation .submit(Op::ResolveElicitation { server_name, request_id, decision: response.action.to_core(), content: response.content, meta: response.meta, }) .await { error!("failed to submit ResolveElicitation: {err}"); }}fn mcp_server_elicitation_response_from_client_result( response: std::result::Result<ClientRequestResult, oneshot::error::RecvError>,) -> McpServerElicitationRequestResponse { match response { Ok(Ok(value)) => serde_json::from_value::<McpServerElicitationRequestResponse>(value) .unwrap_or_else(|err| { error!("failed to deserialize McpServerElicitationRequestResponse: {err}"); McpServerElicitationRequestResponse { action: McpServerElicitationAction::Decline, content: None, meta: None, } }), Ok(Err(err)) if is_turn_transition_server_request_error(&err) => { McpServerElicitationRequestResponse { action: McpServerElicitationAction::Cancel, content: None, meta: None, } } Ok(Err(err)) => { error!("request failed with client error: {err:?}"); McpServerElicitationRequestResponse { action: McpServerElicitationAction::Decline, content: None, meta: None, } } Err(err) => { error!("request failed: {err:?}"); McpServerElicitationRequestResponse { action: McpServerElicitationAction::Decline, content: None, meta: None, } } }}async fn on_request_permissions_response( pending_response: PendingRequestPermissionsResponse, conversation: Arc<CodexThread>, thread_state: Arc<Mutex<ThreadState>>,) { let PendingRequestPermissionsResponse { call_id, conversation_id, turn_id, requested_permissions, request_cwd, pending_request_id, outgoing, receiver, request_permissions_guard, } = pending_response; let response = receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id.clone()).await; drop(request_permissions_guard); let response = match request_permissions_response_from_client_result( requested_permissions, response, request_cwd.as_path(), ) { Ok(Some(response)) => response, Ok(None) => return, // TODO(anp): Remove this native-path localization error path once core permission paths // remain PathUri after crossing the app-server boundary. Err(err) => { let message = format!("failed to localize granted filesystem paths: {err}"); handle_error_notification( conversation_id, &turn_id, TurnError { message, codex_error_info: None, additional_details: None, }, &outgoing, &thread_state, ) .await; if let Err(err) = conversation.submit(Op::Interrupt).await { error!("failed to interrupt turn after invalid permission paths: {err}"); } return; } }; outgoing.track_effective_permissions_approval_response(pending_request_id, response.clone()); if let Err(err) = conversation .submit(Op::RequestPermissionsResponse { id: call_id, response, }) .await { error!("failed to submit RequestPermissionsResponse: {err}"); }}struct PendingRequestPermissionsResponse { call_id: String, conversation_id: ThreadId, turn_id: String, requested_permissions: CoreRequestPermissionProfile, request_cwd: AbsolutePathBuf, pending_request_id: RequestId, outgoing: ThreadScopedOutgoingMessageSender, receiver: oneshot::Receiver<ClientRequestResult>, request_permissions_guard: ThreadWatchActiveGuard,}fn request_permissions_response_from_client_result( requested_permissions: CoreRequestPermissionProfile, response: std::result::Result<ClientRequestResult, oneshot::error::RecvError>, cwd: &std::path::Path,) -> std::io::Result<Option<CoreRequestPermissionsResponse>> { let value = match response { Ok(Ok(value)) => value, Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return Ok(None), Ok(Err(err)) => { error!("request failed with client error: {err:?}"); return Ok(Some(CoreRequestPermissionsResponse { permissions: Default::default(), scope: CorePermissionGrantScope::Turn, strict_auto_review: false, })); } Err(err) => { error!("request failed: {err:?}"); return Ok(Some(CoreRequestPermissionsResponse { permissions: Default::default(), scope: CorePermissionGrantScope::Turn, strict_auto_review: false, })); } }; let response = serde_json::from_value::<PermissionsRequestApprovalResponse>(value) .unwrap_or_else(|err| { error!("failed to deserialize PermissionsRequestApprovalResponse: {err}"); PermissionsRequestApprovalResponse { permissions: V2GrantedPermissionProfile::default(), scope: codex_app_server_protocol::PermissionGrantScope::Turn, strict_auto_review: None, } }); let strict_auto_review = response.strict_auto_review.unwrap_or(false); if strict_auto_review && matches!( response.scope, codex_app_server_protocol::PermissionGrantScope::Session ) { error!("strict auto review is only supported for turn-scoped permission grants"); return Ok(Some(CoreRequestPermissionsResponse { permissions: Default::default(), scope: CorePermissionGrantScope::Turn, strict_auto_review: false, })); } let granted_permissions: CoreAdditionalPermissionProfile = response.permissions.try_into()?; let permissions = if granted_permissions.is_empty() { CoreRequestPermissionProfile::default() } else { intersect_permission_profiles(requested_permissions.into(), granted_permissions, cwd).into() }; Ok(Some(CoreRequestPermissionsResponse { permissions, scope: response.scope.to_core(), strict_auto_review, }))}const REVIEW_FALLBACK_MESSAGE: &str = "Reviewer failed to output a response.";fn render_review_output_text(output: &ReviewOutputEvent) -> String { let mut sections = Vec::new(); let explanation = output.overall_explanation.trim(); if !explanation.is_empty() { sections.push(explanation.to_string()); } if !output.findings.is_empty() { let findings = format_review_findings_block(&output.findings, /*selection*/ None); let trimmed = findings.trim(); if !trimmed.is_empty() { sections.push(trimmed.to_string()); } } if sections.is_empty() { REVIEW_FALLBACK_MESSAGE.to_string() } else { sections.join("\n\n") }}fn map_file_change_approval_decision(decision: FileChangeApprovalDecision) -> ReviewDecision { match decision { FileChangeApprovalDecision::Accept => ReviewDecision::Approved, FileChangeApprovalDecision::AcceptForSession => ReviewDecision::ApprovedForSession, FileChangeApprovalDecision::Decline => ReviewDecision::Denied, FileChangeApprovalDecision::Cancel => ReviewDecision::Abort, }}#[allow(clippy::too_many_arguments)]async fn on_file_change_request_approval_response( item_id: String, pending_request_id: RequestId, receiver: oneshot::Receiver<ClientRequestResult>, codex: Arc<CodexThread>, thread_state: Arc<Mutex<ThreadState>>, permission_guard: ThreadWatchActiveGuard,) { let response = receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await; drop(permission_guard); let decision = match response { Ok(Ok(value)) => { let response = serde_json::from_value::<FileChangeRequestApprovalResponse>(value) .unwrap_or_else(|err| { error!("failed to deserialize FileChangeRequestApprovalResponse: {err}"); FileChangeRequestApprovalResponse { decision: FileChangeApprovalDecision::Decline, } }); map_file_change_approval_decision(response.decision) } Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return, Ok(Err(err)) => { error!("request failed with client error: {err:?}"); ReviewDecision::Denied } Err(err) => { error!("request failed: {err:?}"); ReviewDecision::Denied } }; if let Err(err) = codex .submit(Op::PatchApproval { id: item_id, decision, }) .await { error!("failed to submit PatchApproval: {err}"); }}#[allow(clippy::too_many_arguments)]async fn on_command_execution_request_approval_response( event_turn_id: String, conversation_id: ThreadId, approval_id: Option<String>, item_id: String, completion_item: Option<CommandExecutionCompletionItem>, pending_request_id: RequestId, receiver: oneshot::Receiver<ClientRequestResult>, conversation: Arc<CodexThread>, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc<Mutex<ThreadState>>, permission_guard: ThreadWatchActiveGuard,) { let response = receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await; drop(permission_guard); let (decision, completion_status) = match response { Ok(Ok(value)) => { let response = serde_json::from_value::<CommandExecutionRequestApprovalResponse>(value) .unwrap_or_else(|err| { error!("failed to deserialize CommandExecutionRequestApprovalResponse: {err}"); CommandExecutionRequestApprovalResponse { decision: CommandExecutionApprovalDecision::Decline, } }); let decision = response.decision; let (decision, completion_status) = match decision { CommandExecutionApprovalDecision::Accept => (ReviewDecision::Approved, None), CommandExecutionApprovalDecision::AcceptForSession => { (ReviewDecision::ApprovedForSession, None) } CommandExecutionApprovalDecision::AcceptWithExecpolicyAmendment { execpolicy_amendment, } => ( ReviewDecision::ApprovedExecpolicyAmendment { proposed_execpolicy_amendment: execpolicy_amendment.into_core(), }, None, ), CommandExecutionApprovalDecision::ApplyNetworkPolicyAmendment { network_policy_amendment, } => { let completion_status = match network_policy_amendment.action { V2NetworkPolicyRuleAction::Allow => None, V2NetworkPolicyRuleAction::Deny => Some(CommandExecutionStatus::Declined), }; ( ReviewDecision::NetworkPolicyAmendment { network_policy_amendment: network_policy_amendment.into_core(), }, completion_status, ) } CommandExecutionApprovalDecision::Decline => ( ReviewDecision::Denied, Some(CommandExecutionStatus::Declined), ), CommandExecutionApprovalDecision::Cancel => ( ReviewDecision::Abort, Some(CommandExecutionStatus::Declined), ), }; (decision, completion_status) } Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return, Ok(Err(err)) => { error!("request failed with client error: {err:?}"); (ReviewDecision::Denied, Some(CommandExecutionStatus::Failed)) } Err(err) => { error!("request failed: {err:?}"); (ReviewDecision::Denied, Some(CommandExecutionStatus::Failed)) } }; let suppress_subcommand_completion_item = { // For regular shell/unified_exec approvals, approval_id is null. // For zsh-fork subcommand approvals, approval_id is present and // item_id points to the parent command item. if approval_id.is_some() { let state = thread_state.lock().await; state .turn_summary .command_execution_started .contains(&item_id) } else { false } }; if let Some(status) = completion_status && !suppress_subcommand_completion_item && let Some(completion_item) = completion_item { complete_command_execution_item( &conversation_id, event_turn_id.clone(), item_id.clone(), completion_item.command, completion_item.cwd, /*process_id*/ None, CommandExecutionSource::Agent, completion_item.command_actions, status, &outgoing, &thread_state, ) .await; } if let Err(err) = conversation .submit(Op::ExecApproval { id: approval_id.unwrap_or_else(|| item_id.clone()), turn_id: Some(event_turn_id), decision, }) .await { error!("failed to submit ExecApproval: {err}"); }}fn now_unix_timestamp_ms() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|duration| duration.as_millis() as i64) .unwrap_or_default()}#[cfg(test)]mod tests { use super::*; use crate::CHANNEL_CAPACITY; use crate::outgoing_message::ConnectionId; use crate::outgoing_message::OutgoingEnvelope; use crate::outgoing_message::OutgoingMessage; use crate::outgoing_message::OutgoingMessageSender; use anyhow::Result; use anyhow::anyhow; use anyhow::bail; use chrono::Utc; use codex_app_server_protocol::AutoReviewDecisionSource; use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::TurnPlanStepStatus; use codex_login::CodexAuth; use codex_protocol::AgentPath; use codex_protocol::items::HookPromptFragment; use codex_protocol::items::build_hook_prompt_message; use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions; use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSandboxEntry; use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::plan_tool::PlanItemArg; use codex_protocol::plan_tool::StepStatus; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentActivityEvent; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::UserMessageEvent; use codex_thread_store::StoredThread; use codex_thread_store::StoredThreadHistory; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use core_test_support::load_default_config_for_test; use pretty_assertions::assert_eq; use serde_json::json; use tempfile::TempDir; use tokio::sync::Mutex; use tokio::sync::mpsc; fn new_thread_state() -> Arc<Mutex<ThreadState>> { Arc::new(Mutex::new(ThreadState::default())) } const TEST_TURN_COMPLETED_AT: i64 = 1_716_000_456; const TEST_TURN_DURATION_MS: i64 = 1_234; async fn recv_broadcast_message( rx: &mut mpsc::Receiver<OutgoingEnvelope>, ) -> Result<OutgoingMessage> { let envelope = rx .recv() .await .ok_or_else(|| anyhow!("should send one message"))?; match envelope { OutgoingEnvelope::Broadcast { message } => Ok(message), OutgoingEnvelope::ToConnection { message, .. } => Ok(message), } } #[test] fn rollback_response_rebuilds_pathless_thread_from_stored_history() -> Result<()> { let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000789")?; let created_at = Utc::now(); let history_items = vec![ RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "before rollback".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() })), RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { message: "after rollback".to_string(), phase: None, memory_citation: None, })), ]; let stored_thread = StoredThread { thread_id, extra_config: None, rollout_path: None, forked_from_id: None, parent_thread_id: None, preview: "fallback preview".to_string(), name: Some("Rollback thread".to_string()), model_provider: "openai".to_string(), model: None, reasoning_effort: None, created_at, updated_at: created_at, archived_at: None, cwd: test_path_buf("/tmp").abs().into(), cli_version: "0.0.0".to_string(), source: SessionSource::Cli, thread_source: None, agent_nickname: None, agent_role: None, agent_path: None, git_info: None, approval_mode: AskForApproval::OnRequest, permission_profile: PermissionProfile::read_only(), token_usage: None, first_user_message: Some("before rollback".to_string()), history: Some(StoredThreadHistory { thread_id, items: history_items, }), }; let fallback_cwd = test_path_buf("/tmp").abs(); let response = thread_rollback_response_from_stored_thread( stored_thread, thread_id.to_string(), "fallback-provider", &fallback_cwd, ThreadStatus::NotLoaded, ) .expect("rollback response should rebuild from stored history"); assert_eq!(response.thread.id, thread_id.to_string()); assert_eq!(response.thread.path, None); assert_eq!(response.thread.preview, "fallback preview"); assert_eq!(response.thread.name.as_deref(), Some("Rollback thread")); assert_eq!(response.thread.status, ThreadStatus::NotLoaded); assert_eq!(response.thread.turns.len(), 1); assert_eq!(response.thread.turns[0].items.len(), 2); Ok(()) } fn turn_complete_event(turn_id: &str) -> TurnCompleteEvent { TurnCompleteEvent { turn_id: turn_id.to_string(), last_agent_message: None, completed_at: Some(TEST_TURN_COMPLETED_AT), duration_ms: Some(TEST_TURN_DURATION_MS), time_to_first_token_ms: None, } } fn turn_aborted_event(turn_id: &str) -> TurnAbortedEvent { TurnAbortedEvent { turn_id: Some(turn_id.to_string()), reason: codex_protocol::protocol::TurnAbortReason::Interrupted, completed_at: Some(TEST_TURN_COMPLETED_AT), duration_ms: Some(TEST_TURN_DURATION_MS), } } fn command_execution_completion_item(command: &str) -> CommandExecutionCompletionItem { CommandExecutionCompletionItem { command: command.to_string(), cwd: test_path_buf("/tmp").abs(), command_actions: vec![V2ParsedCommand::Unknown { command: command.to_string(), }], } } fn guardian_command_assessment( id: &str, turn_id: &str, status: GuardianAssessmentStatus, ) -> GuardianAssessmentEvent { let (risk_level, user_authorization, rationale) = match status { GuardianAssessmentStatus::InProgress => (None, None, None), GuardianAssessmentStatus::Approved => ( Some(codex_protocol::protocol::GuardianRiskLevel::Low), Some(codex_protocol::protocol::GuardianUserAuthorization::High), Some("looks safe".to_string()), ), GuardianAssessmentStatus::Denied => ( Some(codex_protocol::protocol::GuardianRiskLevel::High), Some(codex_protocol::protocol::GuardianUserAuthorization::Low), Some("too risky".to_string()), ), GuardianAssessmentStatus::TimedOut => { (None, None, Some("review timed out".to_string())) } GuardianAssessmentStatus::Aborted => (None, None, None), }; GuardianAssessmentEvent { id: format!("review-{id}"), target_item_id: Some(id.to_string()), turn_id: turn_id.to_string(), started_at_ms: 1_000, completed_at_ms: (!matches!(status, GuardianAssessmentStatus::InProgress)) .then_some(1_042), status, risk_level, user_authorization, rationale, decision_source: if matches!(status, GuardianAssessmentStatus::InProgress) { None } else { Some(codex_protocol::protocol::GuardianAssessmentDecisionSource::Agent) }, action: serde_json::from_value(json!({ "type": "command", "source": "shell", "command": format!("rm -f /tmp/{id}.sqlite"), "cwd": test_path_buf("/tmp"), })) .expect("guardian action"), } } struct GuardianAssessmentTestContext { conversation_id: ThreadId, conversation: Arc<CodexThread>, thread_manager: Arc<ThreadManager>, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc<Mutex<ThreadState>>, thread_watch_manager: ThreadWatchManager, } impl GuardianAssessmentTestContext { async fn apply_guardian_assessment_event(&self, assessment: GuardianAssessmentEvent) { let event_turn_id = assessment.turn_id.clone(); apply_bespoke_event_handling( Event { id: event_turn_id, msg: EventMsg::GuardianAssessment(assessment), }, self.conversation_id, self.conversation.clone(), self.thread_manager.clone(), self.outgoing.clone(), self.thread_state.clone(), self.thread_watch_manager.clone(), Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)), "test-provider".to_string(), ) .await; } } #[test] fn guardian_assessment_started_uses_event_turn_id_fallback() { let conversation_id = ThreadId::new(); let action = codex_protocol::protocol::GuardianAssessmentAction::Command { source: codex_protocol::protocol::GuardianCommandSource::Shell, command: "rm -rf /tmp/example.sqlite".to_string(), cwd: test_path_buf("/tmp").abs(), }; let notification = guardian_auto_approval_review_notification( &conversation_id, "turn-from-event", &GuardianAssessmentEvent { id: "review-1".to_string(), target_item_id: Some("item-1".to_string()), turn_id: String::new(), started_at_ms: 1_000, completed_at_ms: None, status: codex_protocol::protocol::GuardianAssessmentStatus::InProgress, risk_level: None, user_authorization: None, rationale: None, decision_source: None, action: action.clone(), }, ); match notification { ServerNotification::ItemGuardianApprovalReviewStarted(payload) => { assert_eq!(payload.thread_id, conversation_id.to_string()); assert_eq!(payload.turn_id, "turn-from-event"); assert_eq!(payload.started_at_ms, 1_000); assert_eq!(payload.review_id, "review-1"); assert_eq!(payload.target_item_id.as_deref(), Some("item-1")); assert_eq!( payload.review.status, GuardianApprovalReviewStatus::InProgress ); assert_eq!(payload.review.risk_level, None); assert_eq!(payload.review.user_authorization, None); assert_eq!(payload.review.rationale, None); assert_eq!(payload.action, action.into()); } other => panic!("unexpected notification: {other:?}"), } } #[test] fn guardian_assessment_completed_emits_review_payload() { let conversation_id = ThreadId::new(); let action = codex_protocol::protocol::GuardianAssessmentAction::Command { source: codex_protocol::protocol::GuardianCommandSource::Shell, command: "rm -rf /tmp/example.sqlite".to_string(), cwd: test_path_buf("/tmp").abs(), }; let notification = guardian_auto_approval_review_notification( &conversation_id, "turn-from-event", &GuardianAssessmentEvent { id: "review-2".to_string(), target_item_id: Some("item-2".to_string()), turn_id: "turn-from-assessment".to_string(), started_at_ms: 1_000, completed_at_ms: Some(1_042), status: codex_protocol::protocol::GuardianAssessmentStatus::Denied, risk_level: Some(codex_protocol::protocol::GuardianRiskLevel::High), user_authorization: Some(codex_protocol::protocol::GuardianUserAuthorization::Low), rationale: Some("too risky".to_string()), decision_source: Some( codex_protocol::protocol::GuardianAssessmentDecisionSource::Agent, ), action: action.clone(), }, ); match notification { ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => { assert_eq!(payload.thread_id, conversation_id.to_string()); assert_eq!(payload.turn_id, "turn-from-assessment"); assert_eq!(payload.started_at_ms, 1_000); assert_eq!(payload.completed_at_ms, 1_042); assert_eq!(payload.review_id, "review-2"); assert_eq!(payload.target_item_id.as_deref(), Some("item-2")); assert_eq!(payload.decision_source, AutoReviewDecisionSource::Agent); assert_eq!(payload.review.status, GuardianApprovalReviewStatus::Denied); assert_eq!( payload.review.risk_level, Some(codex_app_server_protocol::GuardianRiskLevel::High) ); assert_eq!( payload.review.user_authorization, Some(codex_app_server_protocol::GuardianUserAuthorization::Low) ); assert_eq!(payload.review.rationale.as_deref(), Some("too risky")); assert_eq!(payload.action, action.into()); } other => panic!("unexpected notification: {other:?}"), } } #[test] fn guardian_assessment_aborted_emits_completed_review_payload() { let conversation_id = ThreadId::new(); let action = codex_protocol::protocol::GuardianAssessmentAction::NetworkAccess { target: "api.openai.com:443".to_string(), host: "api.openai.com".to_string(), protocol: codex_protocol::protocol::NetworkApprovalProtocol::Https, port: 443, }; let notification = guardian_auto_approval_review_notification( &conversation_id, "turn-from-event", &GuardianAssessmentEvent { id: "review-3".to_string(), target_item_id: None, turn_id: "turn-from-assessment".to_string(), started_at_ms: 1_000, completed_at_ms: Some(1_042), status: codex_protocol::protocol::GuardianAssessmentStatus::Aborted, risk_level: None, user_authorization: None, rationale: None, decision_source: Some( codex_protocol::protocol::GuardianAssessmentDecisionSource::Agent, ), action: action.clone(), }, ); match notification { ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => { assert_eq!(payload.thread_id, conversation_id.to_string()); assert_eq!(payload.turn_id, "turn-from-assessment"); assert_eq!(payload.review_id, "review-3"); assert_eq!(payload.target_item_id, None); assert_eq!(payload.decision_source, AutoReviewDecisionSource::Agent); assert_eq!(payload.review.status, GuardianApprovalReviewStatus::Aborted); assert_eq!(payload.review.risk_level, None); assert_eq!(payload.review.user_authorization, None); assert_eq!(payload.review.rationale, None); assert_eq!(payload.action, action.into()); } other => panic!("unexpected notification: {other:?}"), } } #[tokio::test] async fn command_execution_started_helper_emits_once() -> Result<()> { let conversation_id = ThreadId::new(); let thread_state = new_thread_state(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let completion_item = command_execution_completion_item("printf hi"); let first_start = start_command_execution_item( &conversation_id, "turn-1".to_string(), "cmd-1".to_string(), completion_item.command.clone(), completion_item.cwd.clone(), completion_item.command_actions.clone(), CommandExecutionSource::Agent, &outgoing, &thread_state, ) .await; assert!(first_start); let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { assert_eq!(payload.thread_id, conversation_id.to_string()); assert_eq!(payload.turn_id, "turn-1"); assert_eq!( payload.item, ThreadItem::CommandExecution { id: "cmd-1".to_string(), command: completion_item.command.clone(), cwd: completion_item.cwd.clone(), process_id: None, source: CommandExecutionSource::Agent, status: CommandExecutionStatus::InProgress, command_actions: completion_item.command_actions.clone(), aggregated_output: None, exit_code: None, duration_ms: None, } ); } other => bail!("unexpected message: {other:?}"), } let second_start = start_command_execution_item( &conversation_id, "turn-1".to_string(), "cmd-1".to_string(), completion_item.command.clone(), completion_item.cwd.clone(), completion_item.command_actions.clone(), CommandExecutionSource::Agent, &outgoing, &thread_state, ) .await; assert!(!second_start); assert!(rx.try_recv().is_err(), "duplicate start should not emit"); Ok(()) } #[tokio::test] async fn complete_command_execution_item_emits_declined_once_for_pending_command() -> Result<()> { let conversation_id = ThreadId::new(); let thread_state = new_thread_state(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let completion_item = command_execution_completion_item("printf hi"); start_command_execution_item( &conversation_id, "turn-1".to_string(), "cmd-1".to_string(), completion_item.command.clone(), completion_item.cwd.clone(), completion_item.command_actions.clone(), CommandExecutionSource::Agent, &outgoing, &thread_state, ) .await; let _started = recv_broadcast_message(&mut rx).await?; complete_command_execution_item( &conversation_id, "turn-1".to_string(), "cmd-1".to_string(), completion_item.command.clone(), completion_item.cwd.clone(), /*process_id*/ None, CommandExecutionSource::Agent, completion_item.command_actions.clone(), CommandExecutionStatus::Declined, &outgoing, &thread_state, ) .await; let completed = recv_broadcast_message(&mut rx).await?; match completed { OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => { let ThreadItem::CommandExecution { id, status, .. } = payload.item else { bail!("expected command execution completion"); }; assert_eq!(id, "cmd-1"); assert_eq!(status, CommandExecutionStatus::Declined); } other => bail!("unexpected message: {other:?}"), } complete_command_execution_item( &conversation_id, "turn-1".to_string(), "cmd-1".to_string(), completion_item.command, completion_item.cwd, /*process_id*/ None, CommandExecutionSource::Agent, completion_item.command_actions, CommandExecutionStatus::Declined, &outgoing, &thread_state, ) .await; assert!( rx.try_recv().is_err(), "completion should not emit after the pending item is cleared" ); Ok(()) } #[tokio::test] async fn guardian_command_execution_notifications_wrap_review_lifecycle() -> Result<()> { let codex_home = TempDir::new()?; let config = load_default_config_for_test(&codex_home).await; let thread_manager = Arc::new( codex_core::test_support::thread_manager_with_models_provider_and_home( CodexAuth::create_dummy_chatgpt_auth_for_testing(), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ), ); let codex_core::NewThread { thread_id: conversation_id, thread: conversation, .. } = thread_manager.start_thread(config.clone()).await?; let thread_state = new_thread_state(); let thread_watch_manager = ThreadWatchManager::new(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], conversation_id, ); let guardian_context = GuardianAssessmentTestContext { conversation_id, conversation: conversation.clone(), thread_manager: thread_manager.clone(), outgoing: outgoing.clone(), thread_state: thread_state.clone(), thread_watch_manager: thread_watch_manager.clone(), }; guardian_context .apply_guardian_assessment_event(guardian_command_assessment( "cmd-guardian-approved", "turn-guardian-approved", GuardianAssessmentStatus::InProgress, )) .await; let first = recv_broadcast_message(&mut rx).await?; match first { OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { assert_eq!(payload.turn_id, "turn-guardian-approved"); let ThreadItem::CommandExecution { id, status, .. } = payload.item else { bail!("expected command execution item"); }; assert_eq!(id, "cmd-guardian-approved"); assert_eq!(status, CommandExecutionStatus::InProgress); } other => bail!("unexpected message: {other:?}"), } let second = recv_broadcast_message(&mut rx).await?; match second { OutgoingMessage::AppServerNotification( ServerNotification::ItemGuardianApprovalReviewStarted(payload), ) => { assert_eq!(payload.review_id, "review-cmd-guardian-approved"); assert_eq!( payload.target_item_id.as_deref(), Some("cmd-guardian-approved") ); assert_eq!( payload.review.status, GuardianApprovalReviewStatus::InProgress ); } other => bail!("unexpected message: {other:?}"), } guardian_context .apply_guardian_assessment_event(guardian_command_assessment( "cmd-guardian-approved", "turn-guardian-approved", GuardianAssessmentStatus::Approved, )) .await; let third = recv_broadcast_message(&mut rx).await?; match third { OutgoingMessage::AppServerNotification( ServerNotification::ItemGuardianApprovalReviewCompleted(payload), ) => { assert_eq!(payload.review_id, "review-cmd-guardian-approved"); assert_eq!( payload.target_item_id.as_deref(), Some("cmd-guardian-approved") ); assert_eq!(payload.decision_source, AutoReviewDecisionSource::Agent); assert_eq!( payload.review.status, GuardianApprovalReviewStatus::Approved ); } other => bail!("unexpected message: {other:?}"), } assert!( rx.try_recv().is_err(), "approved review should not complete the command item" ); guardian_context .apply_guardian_assessment_event(guardian_command_assessment( "cmd-guardian-denied", "turn-guardian-denied", GuardianAssessmentStatus::InProgress, )) .await; let fourth = recv_broadcast_message(&mut rx).await?; match fourth { OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { assert_eq!(payload.turn_id, "turn-guardian-denied"); let ThreadItem::CommandExecution { id, status, .. } = payload.item else { bail!("expected command execution item"); }; assert_eq!(id, "cmd-guardian-denied"); assert_eq!(status, CommandExecutionStatus::InProgress); } other => bail!("unexpected message: {other:?}"), } let fifth = recv_broadcast_message(&mut rx).await?; match fifth { OutgoingMessage::AppServerNotification( ServerNotification::ItemGuardianApprovalReviewStarted(payload), ) => { assert_eq!(payload.review_id, "review-cmd-guardian-denied"); assert_eq!( payload.target_item_id.as_deref(), Some("cmd-guardian-denied") ); assert_eq!( payload.review.status, GuardianApprovalReviewStatus::InProgress ); } other => bail!("unexpected message: {other:?}"), } guardian_context .apply_guardian_assessment_event(guardian_command_assessment( "cmd-guardian-denied", "turn-guardian-denied", GuardianAssessmentStatus::Denied, )) .await; let sixth = recv_broadcast_message(&mut rx).await?; match sixth { OutgoingMessage::AppServerNotification( ServerNotification::ItemGuardianApprovalReviewCompleted(payload), ) => { assert_eq!(payload.review_id, "review-cmd-guardian-denied"); assert_eq!( payload.target_item_id.as_deref(), Some("cmd-guardian-denied") ); assert_eq!(payload.decision_source, AutoReviewDecisionSource::Agent); assert_eq!(payload.review.status, GuardianApprovalReviewStatus::Denied); } other => bail!("unexpected message: {other:?}"), } let seventh = recv_broadcast_message(&mut rx).await?; match seventh { OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => { let ThreadItem::CommandExecution { id, status, .. } = payload.item else { bail!("expected command execution completion"); }; assert_eq!(id, "cmd-guardian-denied"); assert_eq!(status, CommandExecutionStatus::Declined); } other => bail!("unexpected message: {other:?}"), } let mut missing_target = guardian_command_assessment( "cmd-guardian-missing-target", "turn-guardian-missing-target", GuardianAssessmentStatus::InProgress, ); missing_target.target_item_id = None; guardian_context .apply_guardian_assessment_event(missing_target) .await; let eighth = recv_broadcast_message(&mut rx).await?; match eighth { OutgoingMessage::AppServerNotification( ServerNotification::ItemGuardianApprovalReviewStarted(payload), ) => { assert_eq!(payload.review_id, "review-cmd-guardian-missing-target"); assert_eq!(payload.target_item_id, None); assert_eq!( payload.review.status, GuardianApprovalReviewStatus::InProgress ); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); conversation.shutdown_and_wait().await?; Ok(()) } #[test] fn file_change_accept_for_session_maps_to_approved_for_session() { let decision = map_file_change_approval_decision(FileChangeApprovalDecision::AcceptForSession); assert_eq!(decision, ReviewDecision::ApprovedForSession); } #[test] fn mcp_server_elicitation_turn_transition_error_maps_to_cancel() { let error = JSONRPCErrorError { code: -1, message: "client request resolved because the turn state was changed".to_string(), data: Some(serde_json::json!({ "reason": "turnTransition" })), }; let response = mcp_server_elicitation_response_from_client_result(Ok(Err(error))); assert_eq!( response, McpServerElicitationRequestResponse { action: McpServerElicitationAction::Cancel, content: None, meta: None, } ); } #[test] fn request_permissions_turn_transition_error_is_ignored() { let error = JSONRPCErrorError { code: -1, message: "client request resolved because the turn state was changed".to_string(), data: Some(serde_json::json!({ "reason": "turnTransition" })), }; let response = request_permissions_response_from_client_result( CoreRequestPermissionProfile::default(), Ok(Err(error)), std::env::current_dir().expect("current dir").as_path(), ) .expect("paths should localize"); assert_eq!(response, None); } #[test] fn request_permissions_response_accepts_partial_network_and_file_system_grants() { let input_path = if cfg!(target_os = "windows") { r"C:\tmp\input" } else { "/tmp/input" }; let output_path = if cfg!(target_os = "windows") { r"C:\tmp\output" } else { "/tmp/output" }; let ignored_path = if cfg!(target_os = "windows") { r"C:\tmp\ignored" } else { "/tmp/ignored" }; let absolute_path = |path: &str| { AbsolutePathBuf::try_from(std::path::PathBuf::from(path)).expect("absolute path") }; let requested_permissions = CoreRequestPermissionProfile { network: Some(CoreNetworkPermissions { enabled: Some(true), }), file_system: Some(CoreFileSystemPermissions::from_read_write_roots( Some(vec![absolute_path(input_path)]), Some(vec![absolute_path(output_path)]), )), }; let cases = vec![ ( serde_json::json!({}), CoreRequestPermissionProfile::default(), ), ( serde_json::json!({ "network": { "enabled": true, }, }), CoreRequestPermissionProfile { network: Some(CoreNetworkPermissions { enabled: Some(true), }), ..CoreRequestPermissionProfile::default() }, ), ( serde_json::json!({ "fileSystem": { "write": [output_path], }, }), CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions::from_read_write_roots( /*read*/ None, Some(vec![absolute_path(output_path)]), )), ..CoreRequestPermissionProfile::default() }, ), ( serde_json::json!({ "fileSystem": { "read": [input_path], "write": [output_path, ignored_path], }, "macos": { "calendar": true, }, }), CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions::from_read_write_roots( Some(vec![absolute_path(input_path)]), Some(vec![absolute_path(output_path)]), )), ..CoreRequestPermissionProfile::default() }, ), ]; let cwd = std::env::current_dir().expect("current dir"); for (granted_permissions, expected_permissions) in cases { let response = request_permissions_response_from_client_result( requested_permissions.clone(), Ok(Ok(serde_json::json!({ "permissions": granted_permissions, }))), cwd.as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response, CoreRequestPermissionsResponse { permissions: expected_permissions, scope: CorePermissionGrantScope::Turn, strict_auto_review: false, } ); } } #[test] fn request_permissions_response_preserves_session_scope() { let response = request_permissions_response_from_client_result( CoreRequestPermissionProfile::default(), Ok(Ok(serde_json::json!({ "scope": "session", "permissions": {}, }))), std::env::current_dir().expect("current dir").as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response, CoreRequestPermissionsResponse { permissions: CoreRequestPermissionProfile::default(), scope: CorePermissionGrantScope::Session, strict_auto_review: false, } ); } #[test] fn request_permissions_response_rejects_session_scoped_strict_auto_review() { let response = request_permissions_response_from_client_result( CoreRequestPermissionProfile::default(), Ok(Ok(serde_json::json!({ "scope": "session", "strictAutoReview": true, "permissions": { "network": { "enabled": true, }, }, }))), std::env::current_dir().expect("current dir").as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response, CoreRequestPermissionsResponse { permissions: CoreRequestPermissionProfile::default(), scope: CorePermissionGrantScope::Turn, strict_auto_review: false, } ); } #[test] fn request_permissions_response_preserves_turn_scoped_strict_auto_review() { let response = request_permissions_response_from_client_result( CoreRequestPermissionProfile { network: Some(codex_protocol::models::NetworkPermissions { enabled: Some(true), }), ..Default::default() }, Ok(Ok(serde_json::json!({ "strictAutoReview": true, "permissions": { "network": { "enabled": true, }, }, }))), std::env::current_dir().expect("current dir").as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!(response.scope, CorePermissionGrantScope::Turn); assert!(response.strict_auto_review); } #[test] fn request_permissions_response_accepts_explicit_child_grant_for_requested_cwd_scope() { let temp_dir = TempDir::new().expect("temp dir"); let cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path()).expect("absolute cwd"); let child = cwd.join("child"); let requested_permissions = CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions { entries: vec![FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }], glob_scan_max_depth: None, }), ..Default::default() }; let response = request_permissions_response_from_client_result( requested_permissions, Ok(Ok(serde_json::json!({ "permissions": { "fileSystem": { "write": [child], }, }, }))), cwd.as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response.permissions, CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions::from_read_write_roots( /*read*/ None, Some(vec![child]), )), ..Default::default() } ); } #[test] fn request_permissions_response_rejects_child_grant_outside_requested_cwd_scope() { let temp_dir = TempDir::new().expect("temp dir"); let request_cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path().join("request-cwd")) .expect("absolute request cwd"); let later_cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path().join("later-cwd")) .expect("absolute later cwd"); let later_child = later_cwd.join("child"); let requested_permissions = CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions { entries: vec![FileSystemSandboxEntry { path: FileSystemPath::Special { value: FileSystemSpecialPath::project_roots(/*subpath*/ None), }, access: FileSystemAccessMode::Write, }], glob_scan_max_depth: None, }), ..Default::default() }; let response = request_permissions_response_from_client_result( requested_permissions, Ok(Ok(serde_json::json!({ "permissions": { "fileSystem": { "write": [later_child], }, }, }))), request_cwd.as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response.permissions, CoreRequestPermissionProfile::default() ); } #[test] fn request_permissions_response_ignores_broader_cwd_grant_for_requested_child_path() { let temp_dir = TempDir::new().expect("temp dir"); let cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path()).expect("absolute cwd"); let child = cwd.join("child"); let requested_permissions = CoreRequestPermissionProfile { file_system: Some(CoreFileSystemPermissions::from_read_write_roots( /*read*/ None, Some(vec![child]), )), ..Default::default() }; let response = request_permissions_response_from_client_result( requested_permissions, Ok(Ok(serde_json::json!({ "permissions": { "fileSystem": { "entries": [{ "path": { "type": "special", "value": { "kind": "project_roots", "subpath": null } }, "access": "write" }], }, }, }))), cwd.as_path(), ) .expect("paths should localize") .expect("response should be accepted"); assert_eq!( response.permissions, CoreRequestPermissionProfile::default() ); } #[tokio::test] async fn test_handle_error_records_message() -> Result<()> { let conversation_id = ThreadId::new(); let thread_state = new_thread_state(); handle_error( conversation_id, TurnError { message: "boom".to_string(), codex_error_info: Some(V2CodexErrorInfo::InternalServerError), additional_details: None, }, &thread_state, ) .await; let turn_summary = find_and_remove_turn_summary(conversation_id, &thread_state).await; assert_eq!( turn_summary.last_error, Some(TurnError { message: "boom".to_string(), codex_error_info: Some(V2CodexErrorInfo::InternalServerError), additional_details: None, }) ); Ok(()) } #[tokio::test] async fn turn_started_omits_active_snapshot_items() -> Result<()> { let codex_home = TempDir::new()?; let config = load_default_config_for_test(&codex_home).await; let thread_manager = Arc::new( codex_core::test_support::thread_manager_with_models_provider_and_home( CodexAuth::create_dummy_chatgpt_auth_for_testing(), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ), ); let codex_core::NewThread { thread_id: conversation_id, thread: conversation, .. } = thread_manager.start_thread(config.clone()).await?; let thread_state = new_thread_state(); { let mut state = thread_state.lock().await; state.track_current_turn_event( "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), trace_id: None, started_at: Some(42), model_context_window: None, collaboration_mode_kind: Default::default(), }), ); state.track_current_turn_event( "turn-1", &EventMsg::UserMessage(codex_protocol::protocol::UserMessageEvent { client_id: None, message: "already tracked".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }), ); } let thread_watch_manager = ThreadWatchManager::new(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], conversation_id, ); apply_bespoke_event_handling( Event { id: "turn-1".to_string(), msg: EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), trace_id: None, started_at: Some(42), model_context_window: None, collaboration_mode_kind: Default::default(), }), }, conversation_id, conversation, thread_manager, outgoing, thread_state, thread_watch_manager, Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)), "test-provider".to_string(), ) .await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnStarted(n)) => { assert_eq!(n.turn.id, "turn-1"); assert_eq!(n.turn.items_view, TurnItemsView::NotLoaded); assert!(n.turn.items.is_empty()); } other => bail!("unexpected message: {other:?}"), } Ok(()) } #[tokio::test] async fn interrupted_subagent_activity_removes_missing_thread_watch() -> Result<()> { let codex_home = TempDir::new()?; let config = load_default_config_for_test(&codex_home).await; let thread_manager = Arc::new( codex_core::test_support::thread_manager_with_models_provider_and_home( CodexAuth::create_dummy_chatgpt_auth_for_testing(), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ), ); let codex_core::NewThread { thread_id: conversation_id, thread: conversation, .. } = thread_manager.start_thread(config).await?; let child_thread_id = ThreadId::new(); let child_thread_id_string = child_thread_id.to_string(); let thread_watch_manager = ThreadWatchManager::new(); thread_watch_manager .note_turn_started(&child_thread_id_string) .await; assert_eq!(thread_watch_manager.running_turn_count().await, 1); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], conversation_id, ); apply_bespoke_event_handling( Event { id: "turn-1".to_string(), msg: EventMsg::SubAgentActivity(SubAgentActivityEvent { event_id: "activity-1".to_string(), occurred_at_ms: 42, agent_thread_id: child_thread_id, agent_path: AgentPath::try_from("/root/worker") .expect("agent path should parse"), kind: SubAgentActivityKind::Interrupted, }), }, conversation_id, conversation, thread_manager, outgoing, new_thread_state(), thread_watch_manager.clone(), Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)), "test-provider".to_string(), ) .await; assert_eq!( thread_watch_manager .loaded_status_for_thread(&child_thread_id_string) .await, ThreadStatus::NotLoaded ); assert_eq!(thread_watch_manager.running_turn_count().await, 0); let message = recv_broadcast_message(&mut rx).await?; let OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) = message else { bail!("unexpected message: {message:?}"); }; assert_eq!( payload, ItemCompletedNotification { item: ThreadItem::SubAgentActivity { id: "activity-1".to_string(), kind: codex_app_server_protocol::SubAgentActivityKind::Interrupted, agent_thread_id: child_thread_id_string, agent_path: "/root/worker".to_string(), }, thread_id: conversation_id.to_string(), turn_id: "turn-1".to_string(), completed_at_ms: 42, } ); Ok(()) } #[tokio::test] async fn test_handle_turn_complete_emits_completed_without_error() -> Result<()> { let conversation_id = ThreadId::new(); let event_turn_id = "complete1".to_string(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let thread_state = new_thread_state(); { let mut state = thread_state.lock().await; state.track_current_turn_event( &event_turn_id, &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: event_turn_id.clone(), trace_id: None, started_at: Some(42), model_context_window: None, collaboration_mode_kind: Default::default(), }), ); state.track_current_turn_event( &event_turn_id, &EventMsg::TurnComplete(turn_complete_event(&event_turn_id)), ); } handle_turn_complete( conversation_id, event_turn_id.clone(), turn_complete_event(&event_turn_id), &outgoing, &thread_state, ) .await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, event_turn_id); assert_eq!(n.turn.status, TurnStatus::Completed); assert_eq!(n.turn.items_view, TurnItemsView::NotLoaded); assert!(n.turn.items.is_empty()); assert_eq!(n.turn.error, None); assert_eq!(n.turn.started_at, Some(42)); assert_eq!(n.turn.completed_at, Some(TEST_TURN_COMPLETED_AT)); assert_eq!(n.turn.duration_ms, Some(TEST_TURN_DURATION_MS)); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_handle_turn_interrupted_emits_interrupted_with_error() -> Result<()> { let conversation_id = ThreadId::new(); let event_turn_id = "interrupt1".to_string(); let thread_state = new_thread_state(); handle_error( conversation_id, TurnError { message: "oops".to_string(), codex_error_info: None, additional_details: None, }, &thread_state, ) .await; let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); handle_turn_interrupted( conversation_id, event_turn_id.clone(), turn_aborted_event(&event_turn_id), &outgoing, &thread_state, ) .await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, event_turn_id); assert_eq!(n.turn.status, TurnStatus::Interrupted); assert_eq!(n.turn.error, None); assert_eq!(n.turn.completed_at, Some(TEST_TURN_COMPLETED_AT)); assert_eq!(n.turn.duration_ms, Some(TEST_TURN_DURATION_MS)); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_handle_turn_complete_emits_failed_with_error() -> Result<()> { let conversation_id = ThreadId::new(); let event_turn_id = "complete_err1".to_string(); let thread_state = new_thread_state(); handle_error( conversation_id, TurnError { message: "bad".to_string(), codex_error_info: Some(V2CodexErrorInfo::Other), additional_details: None, }, &thread_state, ) .await; let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); handle_turn_complete( conversation_id, event_turn_id.clone(), turn_complete_event(&event_turn_id), &outgoing, &thread_state, ) .await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, event_turn_id); assert_eq!(n.turn.status, TurnStatus::Failed); assert_eq!( n.turn.error, Some(TurnError { message: "bad".to_string(), codex_error_info: Some(V2CodexErrorInfo::Other), additional_details: None, }) ); assert_eq!(n.turn.completed_at, Some(TEST_TURN_COMPLETED_AT)); assert_eq!(n.turn.duration_ms, Some(TEST_TURN_DURATION_MS)); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_handle_turn_plan_update_emits_notification_for_v2() -> Result<()> { let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let update = UpdatePlanArgs { explanation: Some("need plan".to_string()), plan: vec![ PlanItemArg { step: "first".to_string(), status: StepStatus::Pending, }, PlanItemArg { step: "second".to_string(), status: StepStatus::Completed, }, ], }; let conversation_id = ThreadId::new(); handle_turn_plan_update(conversation_id, "turn-123", update, &outgoing).await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(n)) => { assert_eq!(n.thread_id, conversation_id.to_string()); assert_eq!(n.turn_id, "turn-123"); assert_eq!(n.explanation.as_deref(), Some("need plan")); assert_eq!(n.plan.len(), 2); assert_eq!(n.plan[0].step, "first"); assert_eq!(n.plan[0].status, TurnPlanStepStatus::Pending); assert_eq!(n.plan[1].step, "second"); assert_eq!(n.plan[1].status, TurnPlanStepStatus::Completed); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_handle_token_count_event_emits_usage_and_rate_limits() -> Result<()> { let conversation_id = ThreadId::new(); let turn_id = "turn-123".to_string(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let info = TokenUsageInfo { total_token_usage: TokenUsage { input_tokens: 100, cached_input_tokens: 25, output_tokens: 50, reasoning_output_tokens: 9, total_tokens: 200, }, last_token_usage: TokenUsage { input_tokens: 10, cached_input_tokens: 5, output_tokens: 7, reasoning_output_tokens: 1, total_tokens: 23, }, model_context_window: Some(4096), }; let rate_limits = RateLimitSnapshot { limit_id: Some("codex".to_string()), limit_name: None, primary: Some(RateLimitWindow { used_percent: 42.5, window_minutes: Some(15), resets_at: Some(1700000000), }), secondary: None, credits: Some(CreditsSnapshot { has_credits: true, unlimited: false, balance: Some("5".to_string()), }), individual_limit: None, plan_type: None, rate_limit_reached_type: None, }; handle_token_count_event( conversation_id, turn_id.clone(), TokenCountEvent { info: Some(info), rate_limits: Some(rate_limits), }, &outgoing, ) .await; let first = recv_broadcast_message(&mut rx).await?; match first { OutgoingMessage::AppServerNotification( ServerNotification::ThreadTokenUsageUpdated(payload), ) => { assert_eq!(payload.thread_id, conversation_id.to_string()); assert_eq!(payload.turn_id, turn_id); let usage = payload.token_usage; assert_eq!(usage.total.total_tokens, 200); assert_eq!(usage.total.cached_input_tokens, 25); assert_eq!(usage.last.output_tokens, 7); assert_eq!(usage.model_context_window, Some(4096)); } other => bail!("unexpected notification: {other:?}"), } let second = recv_broadcast_message(&mut rx).await?; match second { OutgoingMessage::AppServerNotification( ServerNotification::AccountRateLimitsUpdated(payload), ) => { assert_eq!(payload.rate_limits.limit_id.as_deref(), Some("codex")); assert_eq!(payload.rate_limits.limit_name, None); assert!(payload.rate_limits.primary.is_some()); assert!(payload.rate_limits.credits.is_some()); } other => bail!("unexpected notification: {other:?}"), } Ok(()) } #[tokio::test] async fn test_handle_token_count_event_without_usage_info() -> Result<()> { let conversation_id = ThreadId::new(); let turn_id = "turn-456".to_string(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); handle_token_count_event( conversation_id, turn_id.clone(), TokenCountEvent { info: None, rate_limits: None, }, &outgoing, ) .await; assert!( rx.try_recv().is_err(), "no notifications should be emitted when token usage info is absent" ); Ok(()) } #[tokio::test] async fn test_handle_turn_complete_emits_error_multiple_turns() -> Result<()> { // Conversation A will have two turns; Conversation B will have one turn. let conversation_a = ThreadId::new(); let conversation_b = ThreadId::new(); let thread_state = new_thread_state(); let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); // Turn 1 on conversation A let a_turn1 = "a_turn1".to_string(); handle_error( conversation_a, TurnError { message: "a1".to_string(), codex_error_info: Some(V2CodexErrorInfo::BadRequest), additional_details: None, }, &thread_state, ) .await; handle_turn_complete( conversation_a, a_turn1.clone(), turn_complete_event(&a_turn1), &outgoing, &thread_state, ) .await; // Turn 1 on conversation B let b_turn1 = "b_turn1".to_string(); handle_error( conversation_b, TurnError { message: "b1".to_string(), codex_error_info: None, additional_details: None, }, &thread_state, ) .await; handle_turn_complete( conversation_b, b_turn1.clone(), turn_complete_event(&b_turn1), &outgoing, &thread_state, ) .await; // Turn 2 on conversation A let a_turn2 = "a_turn2".to_string(); handle_turn_complete( conversation_a, a_turn2.clone(), turn_complete_event(&a_turn2), &outgoing, &thread_state, ) .await; // Verify: A turn 1 let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, a_turn1); assert_eq!(n.turn.status, TurnStatus::Failed); assert_eq!( n.turn.error, Some(TurnError { message: "a1".to_string(), codex_error_info: Some(V2CodexErrorInfo::BadRequest), additional_details: None, }) ); } other => bail!("unexpected message: {other:?}"), } // Verify: B turn 1 let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, b_turn1); assert_eq!(n.turn.status, TurnStatus::Failed); assert_eq!( n.turn.error, Some(TurnError { message: "b1".to_string(), codex_error_info: None, additional_details: None, }) ); } other => bail!("unexpected message: {other:?}"), } // Verify: A turn 2 let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnCompleted(n)) => { assert_eq!(n.turn.id, a_turn2); assert_eq!(n.turn.status, TurnStatus::Completed); assert_eq!(n.turn.error, None); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_handle_turn_diff_emits_v2_notification() -> Result<()> { let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], ThreadId::new(), ); let unified_diff = "--- a\n+++ b\n".to_string(); let conversation_id = ThreadId::new(); handle_turn_diff( conversation_id, "turn-1", TurnDiffEvent { unified_diff: unified_diff.clone(), }, &outgoing, ) .await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::TurnDiffUpdated( notification, )) => { assert_eq!(notification.thread_id, conversation_id.to_string()); assert_eq!(notification.turn_id, "turn-1"); assert_eq!(notification.diff, unified_diff); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) } #[tokio::test] async fn test_hook_prompt_raw_response_emits_item_completed() -> Result<()> { let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); let outgoing = Arc::new(OutgoingMessageSender::new( tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let conversation_id = ThreadId::new(); let outgoing = ThreadScopedOutgoingMessageSender::new( outgoing, vec![ConnectionId(1)], conversation_id, ); let item = build_hook_prompt_message(&[ HookPromptFragment::from_single_hook("Retry with tests.", "hook-run-1"), HookPromptFragment::from_single_hook("Then summarize cleanly.", "hook-run-2"), ]) .expect("hook prompt message"); maybe_emit_hook_prompt_item_completed(conversation_id, "turn-1", &item, &outgoing).await; let msg = recv_broadcast_message(&mut rx).await?; match msg { OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted( notification, )) => { assert_eq!(notification.thread_id, conversation_id.to_string()); assert_eq!(notification.turn_id, "turn-1"); assert_eq!( notification.item, ThreadItem::HookPrompt { id: notification.item.id().to_string(), fragments: vec![ codex_app_server_protocol::HookPromptFragment { text: "Retry with tests.".into(), hook_run_id: "hook-run-1".into(), }, codex_app_server_protocol::HookPromptFragment { text: "Then summarize cleanly.".into(), hook_run_id: "hook-run-2".into(), }, ], } ); } other => bail!("unexpected message: {other:?}"), } assert!(rx.try_recv().is_err(), "no extra messages expected"); Ok(()) }}