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::EnvironmentConnectionNotification;
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::ModelSafetyBufferingUpdatedNotification;
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::RawResponseCompletedNotification;
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_protocol::ThreadId;
use codex_protocol::items::CollabAgentTool as CoreCollabAgentTool;
use codex_protocol::items::TurnItem as CoreTurnItem;
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::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 codex_utils_path_uri::LegacyAppPathString;
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: LegacyAppPathString,
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) => {
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) => {
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, failure_reason) = match update.status {
codex_protocol::protocol::McpStartupStatus::Starting => {
(McpServerStartupState::Starting, None, None)
}
codex_protocol::protocol::McpStartupStatus::Ready => {
(McpServerStartupState::Ready, None, None)
}
codex_protocol::protocol::McpStartupStatus::Failed { error, reason } => (
McpServerStartupState::Failed,
Some(error),
reason.map(Into::into),
),
codex_protocol::protocol::McpStartupStatus::Cancelled => {
(McpServerStartupState::Cancelled, None, None)
}
};
let notification = McpServerStatusUpdatedNotification {
thread_id: Some(conversation_id.to_string()),
name: update.server,
status,
error,
failure_reason,
};
outgoing
.send_server_notification(ServerNotification::McpServerStatusUpdated(notification))
.await;
}
EventMsg::EnvironmentConnected(event) => {
outgoing
.send_server_notification(ServerNotification::EnvironmentConnected(
EnvironmentConnectionNotification {
thread_id: conversation_id.to_string(),
environment_id: event.environment_id,
},
))
.await;
}
EventMsg::EnvironmentDisconnected(event) => {
outgoing
.send_server_notification(ServerNotification::EnvironmentDisconnected(
EnvironmentConnectionNotification {
thread_id: conversation_id.to_string(),
environment_id: event.environment_id,
},
))
.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,
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::SafetyBuffering(event) => {
let notification = ModelSafetyBufferingUpdatedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id.clone(),
model: event.model,
use_cases: event.use_cases,
reasons: event.reasons,
show_buffering_ui: event.show_buffering_ui,
faster_model: event.faster_model,
};
outgoing
.send_server_notification(ServerNotification::ModelSafetyBufferingUpdated(
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,
environment_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().into(),
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(),
environment_id,
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(_)
| EventMsg::DynamicToolCallResponse(_)
| EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_)
| EventMsg::SubAgentActivity(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::ExecCommandEnd(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_) => {
}
EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_) => {
}
msg @ (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::ContextCompacted(..) => {
}
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 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) => {
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::ItemStarted(event) => {
let should_emit = match &event.item {
CoreTurnItem::CommandExecution(item) => thread_state
.lock()
.await
.turn_summary
.command_execution_started
.insert(item.id.clone()),
_ => true,
};
let dynamic_tool_call_params = match &event.item {
CoreTurnItem::DynamicToolCall(item) => Some(DynamicToolCallParams {
thread_id: conversation_id.to_string(),
turn_id: event.turn_id.clone(),
call_id: item.id.clone(),
namespace: item.namespace.clone(),
tool: item.tool.clone(),
arguments: item.arguments.clone(),
}),
_ => None,
};
if should_emit {
let notification = item_event_to_server_notification(
EventMsg::ItemStarted(event),
&conversation_id.to_string(),
&event_turn_id,
);
outgoing.send_server_notification(notification).await;
}
if let Some(params) = dynamic_tool_call_params {
let call_id = params.call_id.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::ItemCompleted(event) => {
apply_canonical_item_completed_side_effects(
&thread_manager,
&thread_watch_manager,
&thread_state,
&event.item,
)
.await;
let notification = item_event_to_server_notification(
EventMsg::ItemCompleted(event),
&conversation_id.to_string(),
&event_turn_id,
);
outgoing.send_server_notification(notification).await;
}
msg @ (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::RawResponseItem(raw_response_item_event) => {
maybe_emit_raw_response_item_completed(
conversation_id,
&event_turn_id,
raw_response_item_event.item,
&outgoing,
)
.await;
}
EventMsg::RawResponseCompleted(raw_response_completed_event) => {
let notification = RawResponseCompletedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id,
response_id: raw_response_completed_event.response_id,
usage: raw_response_completed_event.token_usage.map(Into::into),
};
outgoing
.send_server_notification(ServerNotification::RawResponseCompleted(notification))
.await;
}
EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) => {
}
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::TurnAborted(turn_aborted_event) => {
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(
true, 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,
) {
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;
}
async fn apply_canonical_item_completed_side_effects(
thread_manager: &Arc<ThreadManager>,
thread_watch_manager: &ThreadWatchManager,
thread_state: &Arc<Mutex<ThreadState>>,
item: &CoreTurnItem,
) {
match item {
CoreTurnItem::CommandExecution(item) => {
thread_state
.lock()
.await
.turn_summary
.command_execution_started
.remove(&item.id);
}
CoreTurnItem::SubAgentActivity(activity)
if activity.kind == SubAgentActivityKind::Interrupted =>
{
remove_missing_thread_watch(
thread_manager,
thread_watch_manager,
activity.agent_thread_id,
)
.await;
}
CoreTurnItem::CollabAgentToolCall(item) if item.tool == CoreCollabAgentTool::CloseAgent => {
for thread_id in &item.receiver_thread_ids {
remove_missing_thread_watch(thread_manager, thread_watch_manager, *thread_id).await;
}
}
_ => {}
}
}
async fn remove_missing_thread_watch(
thread_manager: &Arc<ThreadManager>,
thread_watch_manager: &ThreadWatchManager,
thread_id: ThreadId,
) {
if thread_manager.get_thread(thread_id).await.is_err() {
thread_watch_manager
.remove_thread(&thread_id.to_string())
.await;
}
}
#[allow(clippy::too_many_arguments)]
async fn start_command_execution_item(
conversation_id: &ThreadId,
turn_id: String,
item_id: String,
command: String,
cwd: LegacyAppPathString,
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: LegacyAppPathString,
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;
}
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, 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,
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,
}))
}
fn map_file_change_approval_decision(decision: FileChangeApprovalDecision) -> ReviewDecision {
match decision {
FileChangeApprovalDecision::Accept => ReviewDecision::Approved,
FileChangeApprovalDecision::AcceptForSession => ReviewDecision::ApprovedForSession,
FileChangeApprovalDecision::Decline => ReviewDecision::denied("rejected by user"),
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)) => match serde_json::from_value::<FileChangeRequestApprovalResponse>(value) {
Ok(response) => map_file_change_approval_decision(response.decision),
Err(err) => {
error!("failed to deserialize FileChangeRequestApprovalResponse: {err}");
ReviewDecision::denied("approval request failed")
}
},
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
ReviewDecision::denied("approval request failed")
}
Err(err) => {
error!("request failed: {err:?}");
ReviewDecision::denied("approval request failed")
}
};
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)) => {
match serde_json::from_value::<CommandExecutionRequestApprovalResponse>(value) {
Ok(response) => match response.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("rejected by user"),
Some(CommandExecutionStatus::Declined),
),
CommandExecutionApprovalDecision::Cancel => (
ReviewDecision::Abort,
Some(CommandExecutionStatus::Declined),
),
},
Err(err) => {
error!("failed to deserialize CommandExecutionRequestApprovalResponse: {err}");
(
ReviewDecision::denied("approval request failed"),
Some(CommandExecutionStatus::Failed),
)
}
}
}
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
(
ReviewDecision::denied("approval request failed"),
Some(CommandExecutionStatus::Failed),
)
}
Err(err) => {
error!("request failed: {err:?}");
(
ReviewDecision::denied("approval request failed"),
Some(CommandExecutionStatus::Failed),
)
}
};
let suppress_subcommand_completion_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,
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::ServerRequest;
use codex_app_server_protocol::TurnPlanStepStatus;
use codex_login::CodexAuth;
use codex_protocol::AgentPath;
use codex_protocol::items::DynamicToolCallItem;
use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus;
use codex_protocol::items::SubAgentActivityItem;
use codex_protocol::items::TurnItem as CoreTurnItem;
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::ItemCompletedEvent;
use codex_protocol::protocol::ItemStartedEvent;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::RateLimitWindow;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
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),
}
}
async fn recv_broadcast_notification(
rx: &mut mpsc::Receiver<OutgoingEnvelope>,
) -> Result<ServerNotification> {
let message = recv_broadcast_message(rx).await?;
let OutgoingMessage::AppServerNotification(envelope) = message else {
bail!("unexpected message: {message:?}");
};
Ok(envelope.notification)
}
#[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,
recency_at: created_at,
archived_at: None,
cwd: test_path_buf("/tmp").abs().into(),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
history_mode: Default::default(),
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(),
started_at: None,
last_agent_message: None,
error: 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()),
started_at: None,
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().into(),
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( 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_notification(&mut rx).await?;
match msg {
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_notification(&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(),
None,
CommandExecutionSource::Agent,
completion_item.command_actions.clone(),
CommandExecutionStatus::Declined,
&outgoing,
&thread_state,
)
.await;
let completed = recv_broadcast_notification(&mut rx).await?;
match completed {
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,
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_notification(&mut rx).await?;
match first {
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_notification(&mut rx).await?;
match second {
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_notification(&mut rx).await?;
match third {
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_notification(&mut rx).await?;
match fourth {
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_notification(&mut rx).await?;
match fifth {
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_notification(&mut rx).await?;
match sixth {
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_notification(&mut rx).await?;
match seventh {
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_notification(&mut rx).await?;
match eighth {
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(
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( None),
},
access: FileSystemAccessMode::Write,
missing_path_behavior: None,
}],
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(
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( None),
},
access: FileSystemAccessMode::Write,
missing_path_behavior: None,
}],
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(
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( 1)),
"test-provider".to_string(),
)
.await;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
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::ItemCompleted(ItemCompletedEvent {
thread_id: conversation_id,
turn_id: "turn-1".to_string(),
item: CoreTurnItem::SubAgentActivity(SubAgentActivityItem {
id: "activity-1".to_string(),
kind: SubAgentActivityKind::Interrupted,
agent_thread_id: child_thread_id,
agent_path: AgentPath::try_from("/root/worker")
.expect("agent path should parse"),
}),
completed_at_ms: 42,
}),
},
conversation_id,
conversation,
thread_manager,
outgoing,
new_thread_state(),
thread_watch_manager.clone(),
Arc::new(tokio::sync::Semaphore::new( 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_notification(&mut rx).await?;
let 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 canonical_dynamic_tool_start_emits_item_and_requests_client() -> 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 (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::ItemStarted(ItemStartedEvent {
thread_id: conversation_id,
turn_id: "turn-1".to_string(),
item: CoreTurnItem::DynamicToolCall(DynamicToolCallItem {
id: "dynamic-1".to_string(),
namespace: Some("apps".to_string()),
tool: "lookup".to_string(),
arguments: json!({"id": "123"}),
status: CoreDynamicToolCallStatus::InProgress,
content_items: None,
success: None,
error: None,
duration: None,
}),
started_at_ms: 42,
}),
},
conversation_id,
conversation,
thread_manager,
outgoing,
new_thread_state(),
ThreadWatchManager::new(),
Arc::new(tokio::sync::Semaphore::new( 1)),
"test-provider".to_string(),
)
.await;
let item_started = recv_broadcast_notification(&mut rx).await?;
let ServerNotification::ItemStarted(payload) = item_started else {
bail!("unexpected message: {item_started:?}");
};
assert_eq!(payload.item.id(), "dynamic-1");
let request = recv_broadcast_message(&mut rx).await?;
let OutgoingMessage::Request(ServerRequest::DynamicToolCall { params, .. }) = request
else {
bail!("unexpected message: {request:?}");
};
assert_eq!(
params,
DynamicToolCallParams {
thread_id: conversation_id.to_string(),
turn_id: "turn-1".to_string(),
call_id: "dynamic-1".to_string(),
namespace: Some("apps".to_string()),
tool: "lookup".to_string(),
arguments: json!({"id": "123"}),
}
);
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_notification(&mut rx).await?;
match msg {
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_notification(&mut rx).await?;
match msg {
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_notification(&mut rx).await?;
match msg {
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_notification(&mut rx).await?;
match msg {
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,
cache_write_input_tokens: 0,
output_tokens: 50,
reasoning_output_tokens: 9,
total_tokens: 200,
},
last_token_usage: TokenUsage {
input_tokens: 10,
cached_input_tokens: 5,
cache_write_input_tokens: 0,
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,
spend_control_reached: 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_notification(&mut rx).await?;
match first {
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_notification(&mut rx).await?;
match second {
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<()> {
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(),
);
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;
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;
let a_turn2 = "a_turn2".to_string();
handle_turn_complete(
conversation_a,
a_turn2.clone(),
turn_complete_event(&a_turn2),
&outgoing,
&thread_state,
)
.await;
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
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:?}"),
}
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
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:?}"),
}
let msg = recv_broadcast_notification(&mut rx).await?;
match msg {
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_notification(&mut rx).await?;
match msg {
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(())
}
}