codex / codex-rs /core /src /session /handlers.rs

Add files using upload-large-folder tool

52a9af3 verified about 1 month ago

24 kB

use crate::realtime_conversation::handle_audio as handle_realtime_conversation_audio;
use crate::realtime_conversation::handle_close as handle_realtime_conversation_close;
use crate::realtime_conversation::handle_speech as handle_realtime_conversation_speech;
use crate::realtime_conversation::handle_start as handle_realtime_conversation_start;
use crate::realtime_conversation::handle_text as handle_realtime_conversation_text;
use async_channel::Receiver;
use codex_otel::set_parent_from_w3c_trace_context;
use codex_protocol::protocol::Submission;
use tracing::Instrument;
use tracing::debug_span;
use tracing::info_span;

use crate::session::session::Session;
use crate::session::thread_settings;
use crate::session::turn_input;

use crate::config::Config;
use crate::context::ContextualUserFragment;
use crate::context::GuardianApprovedAction;
use crate::review_prompts::resolve_review_request;
use crate::session::spawn_review_thread;
use crate::tasks::CompactTask;
use crate::tasks::UserShellCommandMode;
use crate::tasks::UserShellCommandTask;
use crate::tasks::execute_user_shell_command;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GuardianAssessmentEvent;
use codex_protocol::protocol::GuardianAssessmentStatus;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;
use codex_protocol::protocol::RealtimeVoicesList;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::request_permissions::RequestPermissionsResponse;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_thread_store::PersistContext;

use crate::context_manager::is_user_turn_boundary;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::mcp::RequestId as ProtocolRequestId;
use codex_rmcp_client::ElicitationAction;
use codex_rmcp_client::ElicitationResponse;
use serde_json::Value;
use std::sync::Arc;
use tracing::debug;
use tracing::info;
use tracing::warn;

pub async fn interrupt(sess: &Arc<Session>) {
    sess.interrupt_task().await;
}

pub async fn clean_background_terminals(sess: &Arc<Session>) {
    sess.close_unified_exec_processes().await;
}

pub async fn realtime_conversation_list_voices(sess: &Session, sub_id: String) {
    sess.send_event_raw(Event {
        id: sub_id,
        msg: EventMsg::RealtimeConversationListVoicesResponse(
            RealtimeConversationListVoicesResponseEvent {
                voices: RealtimeVoicesList::builtin(),
            },
        ),
    })
    .await;
}

/// Queues an inter-agent message, then lets the shared pending-work scheduler
/// decide whether an idle session should start a regular turn.
pub async fn inter_agent_communication(
    sess: &Arc<Session>,
    sub_id: String,
    communication: InterAgentCommunication,
    start_options: codex_protocol::turn_input::TurnStartOptions,
) {
    let trigger_turn = communication.trigger_turn;
    sess.input_queue
        .enqueue_mailbox_communication(communication, start_options)
        .await;
    crate::agent_communication::emit_agent_communication_receive(&sub_id);
    if trigger_turn || sess.has_outstanding_durable_sleep() {
        sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
            .await;
    }
}

pub async fn run_user_shell_command(
    sess: &Arc<Session>,
    sub_id: String,
    command: String,
    timeout_ms: Option<u64>,
) {
    if let Some((turn_context, cancellation_token)) =
        sess.active_turn_context_and_cancellation_token().await
    {
        let session = Arc::clone(sess);
        tokio::spawn(async move {
            execute_user_shell_command(
                session,
                turn_context,
                command,
                timeout_ms,
                cancellation_token,
                UserShellCommandMode::ActiveTurnAuxiliary,
            )
            .await;
        });
        return;
    }

    let turn_context = sess
        .new_turn_with_default_settings(sub_id, Default::default())
        .await;
    sess.spawn_task(
        turn_context,
        Vec::new(),
        UserShellCommandTask::new(command, timeout_ms),
    )
    .await;
}

pub async fn resolve_elicitation(
    sess: &Arc<Session>,
    server_name: String,
    request_id: ProtocolRequestId,
    decision: codex_protocol::approvals::ElicitationAction,
    content: Option<Value>,
    meta: Option<Value>,
) {
    let action = match decision {
        codex_protocol::approvals::ElicitationAction::Accept => ElicitationAction::Accept,
        codex_protocol::approvals::ElicitationAction::Decline => ElicitationAction::Decline,
        codex_protocol::approvals::ElicitationAction::Cancel => ElicitationAction::Cancel,
    };
    let content = match action {
        // Preserve the legacy fallback for clients that only send an action.
        ElicitationAction::Accept => Some(content.unwrap_or_else(|| serde_json::json!({}))),
        ElicitationAction::Decline | ElicitationAction::Cancel => None,
        _ => None,
    };
    let response = ElicitationResponse {
        action,
        content,
        meta,
    };
    let request_id = match request_id {
        ProtocolRequestId::String(value) => {
            rmcp::model::NumberOrString::String(std::sync::Arc::from(value))
        }
        ProtocolRequestId::Integer(value) => rmcp::model::NumberOrString::Number(value),
    };
    if let Err(err) = sess
        .resolve_elicitation(server_name, request_id, response)
        .await
    {
        warn!(
            error = %err,
            "failed to resolve elicitation request in session"
        );
    }
}

/// Propagate a user's exec approval decision to the session.
/// Also optionally applies an execpolicy amendment.
pub async fn exec_approval(
    sess: &Arc<Session>,
    approval_id: String,
    turn_id: Option<String>,
    decision: ReviewDecision,
) {
    let event_turn_id = turn_id.unwrap_or_else(|| approval_id.clone());
    if let ReviewDecision::ApprovedExecpolicyAmendment {
        proposed_execpolicy_amendment,
    } = &decision
        && let Err(err) = sess
            .persist_execpolicy_amendment(proposed_execpolicy_amendment)
            .await
    {
        let message = format!("Failed to apply execpolicy amendment: {err}");
        tracing::warn!("{message}");
        let warning = EventMsg::Warning(WarningEvent { message });
        sess.send_event_raw(Event {
            id: event_turn_id.clone(),
            msg: warning,
        })
        .await;
    }
    match decision {
        ReviewDecision::Abort => {
            sess.interrupt_task().await;
        }
        other => sess.notify_approval(&approval_id, other).await,
    }
}

pub async fn patch_approval(sess: &Arc<Session>, id: String, decision: ReviewDecision) {
    match decision {
        ReviewDecision::Abort => {
            sess.interrupt_task().await;
        }
        other => sess.notify_approval(&id, other).await,
    }
}

pub async fn request_user_input_response(
    sess: &Arc<Session>,
    id: String,
    response: RequestUserInputResponse,
) {
    sess.notify_user_input_response(&id, response).await;
}

pub async fn request_permissions_response(
    sess: &Arc<Session>,
    id: String,
    response: RequestPermissionsResponse,
) {
    sess.notify_request_permissions_response(&id, response)
        .await;
}

pub async fn dynamic_tool_response(sess: &Arc<Session>, id: String, response: DynamicToolResponse) {
    sess.notify_dynamic_tool_response(&id, response).await;
}

pub fn refresh_mcp_servers(sess: &Session) {
    sess.services.mcp_runtime.reconnect_on_next_refresh();
    sess.request_mcp_runtime_refresh();
}

pub async fn reload_user_config(sess: &Arc<Session>) {
    sess.reload_user_config_layer().await;
}

pub async fn compact(sess: &Arc<Session>, sub_id: String) {
    let turn_context = sess
        .new_turn_with_default_settings(sub_id, Default::default())
        .await;

    sess.spawn_task(turn_context, Vec::new(), CompactTask).await;
}

pub(super) async fn persist_thread_memory_mode_update(
    sess: &Arc<Session>,
    mode: ThreadMemoryMode,
) -> anyhow::Result<()> {
    let live_thread = sess.live_thread_for_persistence("update thread memory mode")?;
    live_thread.persist(PersistContext::Standard).await?;
    live_thread.flush().await?;
    live_thread
        .update_memory_mode(mode, /*include_archived*/ false)
        .await?;
    live_thread.flush().await?;
    Ok(())
}

/// Persists thread-level memory mode metadata for the active session.
///
/// This does not involve the model and only affects whether the thread is
/// eligible for future memory generation.
pub async fn set_thread_memory_mode(sess: &Arc<Session>, sub_id: String, mode: ThreadMemoryMode) {
    if let Err(err) = persist_thread_memory_mode_update(sess, mode).await {
        warn!("Failed to persist thread memory mode update to rollout: {err}");
        let event = Event {
            id: sub_id,
            msg: EventMsg::Error(ErrorEvent {
                misalignment: None,
                message: err.to_string(),
                codex_error_info: Some(CodexErrorInfo::Other),
            }),
        };
        sess.send_event_raw(event).await;
    }
}

pub(super) async fn shutdown_session_runtime(sess: &Arc<Session>) {
    if let Some(startup_prewarm) = sess.take_session_startup_prewarm().await {
        startup_prewarm.abort().await;
    }
    let _ = sess.conversation.shutdown().await;
    sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
    let shell_snapshot_prewarm = sess.state.lock().await.shell_snapshot_prewarm.take();
    if let Some(shell_snapshot_prewarm) = shell_snapshot_prewarm {
        shell_snapshot_prewarm.abort();
        let _ = shell_snapshot_prewarm.await;
    }
    sess.hooks().shutdown().await;
    sess.async_hook_results.close();
    while sess.async_hook_results.try_recv().is_ok() {}
    sess.services
        .unified_exec_manager
        .terminate_all_processes()
        .await;
    if let Err(err) = sess.services.code_mode_service.shutdown().await {
        warn!("failed to shutdown code mode session: {err}");
    }
    sess.stop_mcp_prewarm_worker().await;
    {
        let _refresh = sess.mcp_refresh.acquire().await;
        sess.mcp_refresh.close();
        sess.services.mcp_runtime.shutdown().await;
    }

    crate::hook_runtime::run_session_end_hooks(sess).await;
    emit_thread_stop_lifecycle(sess).await;
}

async fn emit_thread_stop_lifecycle(sess: &Session) {
    for contributor in sess.services.extensions.thread_lifecycle_contributors() {
        contributor
            .on_thread_stop(codex_extension_api::ThreadStopInput {
                session_store: &sess.services.session_extension_data,
                thread_store: &sess.services.thread_extension_data,
            })
            .await;
    }
}

pub async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool {
    shutdown_session_runtime(sess).await;
    info!("Shutting down Codex instance");
    let history = sess.clone_history().await;
    let turn_count = history
        .raw_items()
        .filter(|item| is_user_turn_boundary(item))
        .count();
    sess.services.session_telemetry.counter(
        "codex.conversation.turn.count",
        i64::try_from(turn_count).unwrap_or(0),
        &[],
    );

    // Gracefully flush and shutdown thread persistence on session end so tests
    // that inspect durable state do not race with the background writer.
    if let Some(live_thread) = sess.live_thread()
        && let Err(e) = live_thread.shutdown().await
    {
        warn!("failed to shutdown thread persistence: {e}");
        let event = Event {
            id: sub_id.clone(),
            msg: EventMsg::Error(ErrorEvent {
                misalignment: None,
                message: "Failed to shutdown thread persistence".to_string(),
                codex_error_info: Some(CodexErrorInfo::Other),
            }),
        };
        sess.send_event_raw(event).await;
    }

    let event = Event {
        id: sub_id,
        msg: EventMsg::ShutdownComplete,
    };
    sess.services
        .rollout_thread_trace
        .record_protocol_event(&event.msg);
    sess.deliver_event_raw(event).await;
    sess.services
        .rollout_thread_trace
        .record_ended(codex_rollout_trace::RolloutStatus::Completed);
    true
}

pub async fn review(
    sess: &Arc<Session>,
    config: &Arc<Config>,
    sub_id: String,
    review_request: ReviewRequest,
) {
    let turn_context = sess
        .new_turn_with_default_settings(sub_id.clone(), Default::default())
        .await;
    sess.maybe_emit_model_warnings_for_turn(turn_context.as_ref())
        .await;
    #[allow(deprecated)]
    match resolve_review_request(review_request, &turn_context.cwd) {
        Ok(resolved) => {
            spawn_review_thread(
                Arc::clone(sess),
                Arc::clone(config),
                turn_context.clone(),
                sub_id,
                resolved,
            )
            .await;
        }
        Err(err) => {
            let event = Event {
                id: sub_id,
                msg: EventMsg::Error(ErrorEvent {
                    misalignment: None,
                    message: err.to_string(),
                    codex_error_info: Some(CodexErrorInfo::Other),
                }),
            };
            sess.send_event(&turn_context, event.msg).await;
        }
    }
}

pub(super) async fn submission_loop(
    sess: Arc<Session>,
    config: Arc<Config>,
    rx_sub: Receiver<Submission>,
) {
    // To break out of this loop, send Op::Shutdown.
    let mut shutdown_received = false;
    while let Ok(sub) = rx_sub.recv().await {
        if matches!(sub.op, Op::ResolveElicitation { .. }) {
            debug!(submission_id = %sub.id, operation = sub.op.kind(), "Submission");
        } else {
            debug!(?sub, "Submission");
        }
        let dispatch_span = submission_dispatch_span(&sub);
        let should_exit = async {
            match sub.op {
                Op::Interrupt => {
                    interrupt(&sess).await;
                    false
                }
                Op::CleanBackgroundTerminals => {
                    clean_background_terminals(&sess).await;
                    false
                }
                Op::RealtimeConversationStart(params) => {
                    if let Err(err) =
                        handle_realtime_conversation_start(&sess, sub.id.clone(), params).await
                    {
                        sess.send_event_raw(Event {
                            id: sub.id.clone(),
                            msg: EventMsg::Error(ErrorEvent {
                                misalignment: None,
                                message: err.to_string(),
                                codex_error_info: Some(CodexErrorInfo::Other),
                            }),
                        })
                        .await;
                    }
                    false
                }
                Op::RealtimeConversationAudio(params) => {
                    handle_realtime_conversation_audio(&sess, sub.id.clone(), params).await;
                    false
                }
                Op::RealtimeConversationText(params) => {
                    handle_realtime_conversation_text(&sess, sub.id.clone(), params).await;
                    false
                }
                Op::RealtimeConversationSpeech(params) => {
                    handle_realtime_conversation_speech(&sess, sub.id.clone(), params).await;
                    false
                }
                Op::RealtimeConversationClose => {
                    handle_realtime_conversation_close(&sess, sub.id.clone()).await;
                    false
                }
                Op::RealtimeConversationListVoices => {
                    realtime_conversation_list_voices(&sess, sub.id.clone()).await;
                    false
                }
                Op::TurnInput {
                    request,
                    mode,
                    reply,
                } => {
                    let result = turn_input::handle(&sess, *request, mode, sub.id.clone()).await;
                    let _ = reply.send(result);
                    false
                }
                Op::RecoverTurn {
                    thread_settings,
                    start_options,
                    reply,
                } => {
                    let result = turn_input::handle_recovery(
                        &sess,
                        thread_settings,
                        start_options,
                        sub.id.clone(),
                    )
                    .await;
                    let _ = reply.send(result);
                    false
                }
                Op::SuspendTurnAndShutdown { reply } => {
                    let result =
                        super::turn_suspension::suspend_turn_and_shutdown(&sess, sub.id.clone())
                            .await;
                    // Exit only after history is durable and its writer has closed; an error
                    // must leave responsibility for the thread with the current worker.
                    let should_exit = matches!(
                        &result,
                        Ok(codex_protocol::turn_input::SuspendTurnOutcome::Suspended { .. })
                    );
                    let _ = reply.send(result);
                    should_exit
                }
                Op::ThreadSettings { thread_settings } => {
                    thread_settings::update(&sess, sub.id.clone(), thread_settings).await;
                    false
                }
                Op::TurnSettings {
                    turn_id,
                    update,
                    reply,
                } => {
                    let outcome = sess.apply_turn_settings(&turn_id, update).await;
                    let _ = reply.send(outcome);
                    false
                }
                Op::InterAgentCommunication {
                    communication,
                    start_options,
                } => {
                    inter_agent_communication(&sess, sub.id.clone(), communication, start_options)
                        .await;
                    false
                }
                Op::ExecApproval {
                    id: approval_id,
                    turn_id,
                    decision,
                } => {
                    exec_approval(&sess, approval_id, turn_id, decision).await;
                    false
                }
                Op::PatchApproval { id, decision } => {
                    patch_approval(&sess, id, decision).await;
                    false
                }
                Op::UserInputAnswer { id, response } => {
                    request_user_input_response(&sess, id, response).await;
                    false
                }
                Op::RequestPermissionsResponse { id, response } => {
                    request_permissions_response(&sess, id, response).await;
                    false
                }
                Op::DynamicToolResponse { id, response } => {
                    dynamic_tool_response(&sess, id, response).await;
                    false
                }
                Op::RefreshMcpServers => {
                    refresh_mcp_servers(&sess);
                    false
                }
                Op::ReloadUserConfig => {
                    reload_user_config(&sess).await;
                    false
                }
                Op::Compact => {
                    compact(&sess, sub.id.clone()).await;
                    false
                }
                Op::SetThreadMemoryMode { mode } => {
                    set_thread_memory_mode(&sess, sub.id.clone(), mode).await;
                    false
                }
                Op::RunUserShellCommand {
                    command,
                    timeout_ms,
                } => {
                    run_user_shell_command(&sess, sub.id.clone(), command, timeout_ms).await;
                    false
                }
                Op::ResolveElicitation {
                    server_name,
                    request_id,
                    decision,
                    content,
                    meta,
                } => {
                    resolve_elicitation(&sess, server_name, request_id, decision, content, meta)
                        .await;
                    false
                }
                Op::Shutdown => shutdown(&sess, sub.id.clone()).await,
                Op::Review { review_request } => {
                    review(&sess, &config, sub.id.clone(), review_request).await;
                    false
                }
                Op::ApproveGuardianDeniedAction { event } => {
                    approve_guardian_denied_action(&sess, event).await;
                    false
                }
                _ => false, // Ignore unknown ops; enum is non_exhaustive to allow extensions.
            }
        }
        .instrument(dispatch_span)
        .await;
        if should_exit {
            shutdown_received = true;
            break;
        }
    }
    // If the submission loop exits because the channel closed without an
    // explicit shutdown op, still run session teardown.
    if !shutdown_received {
        shutdown_session_runtime(&sess).await;
        if let Some(live_thread) = sess.live_thread()
            && let Err(err) = live_thread.shutdown().await
        {
            warn!("failed to shutdown thread persistence after submission channel closed: {err}");
        }
    }
    debug!("Agent loop exited");
}

async fn approve_guardian_denied_action(sess: &Arc<Session>, event: GuardianAssessmentEvent) {
    if event.status != GuardianAssessmentStatus::Denied {
        warn!(
            review_id = event.id.as_str(),
            "ignoring approval for non-denied Guardian assessment"
        );
        return;
    }

    let approved_action = serde_json::json!({
        "action": &event.action,
        "outcome": "allowed",
    });
    let approved_action_json = match serde_json::to_string_pretty(&approved_action) {
        Ok(approved_action_json) => approved_action_json,
        Err(error) => {
            warn!(%error, review_id = event.id.as_str(), "failed to serialize approved Guardian action");
            return;
        }
    };
    let items = vec![ContextualUserFragment::into(GuardianApprovedAction::new(
        approved_action_json,
    ))];

    sess.inject_no_new_turn(items, /*current_turn_context*/ None)
        .await;
}

pub(super) fn submission_dispatch_span(sub: &Submission) -> tracing::Span {
    let op_name = sub.op.kind();
    let span_name = format!("op.dispatch.{op_name}");
    let dispatch_span = match &sub.op {
        Op::RealtimeConversationAudio(_) => {
            debug_span!(
                "submission_dispatch",
                otel.name = span_name.as_str(),
                submission.id = sub.id.as_str(),
                codex.op = op_name
            )
        }
        _ => info_span!(
            "submission_dispatch",
            otel.name = span_name.as_str(),
            submission.id = sub.id.as_str(),
            codex.op = op_name
        ),
    };
    if let Some(trace) = sub.trace.as_ref()
        && !set_parent_from_w3c_trace_context(&dispatch_span, trace)
    {
        warn!(
            submission.id = sub.id.as_str(),
            "ignoring invalid submission trace carrier"
        );
    }
    dispatch_span
}

	use crate::realtime_conversation::handle_audio as handle_realtime_conversation_audio;
	use crate::realtime_conversation::handle_close as handle_realtime_conversation_close;
	use crate::realtime_conversation::handle_speech as handle_realtime_conversation_speech;
	use crate::realtime_conversation::handle_start as handle_realtime_conversation_start;
	use crate::realtime_conversation::handle_text as handle_realtime_conversation_text;
	use async_channel::Receiver;
	use codex_otel::set_parent_from_w3c_trace_context;
	use codex_protocol::protocol::Submission;
	use tracing::Instrument;
	use tracing::debug_span;
	use tracing::info_span;

	use crate::session::session::Session;
	use crate::session::thread_settings;
	use crate::session::turn_input;

	use crate::config::Config;
	use crate::context::ContextualUserFragment;
	use crate::context::GuardianApprovedAction;
	use crate::review_prompts::resolve_review_request;
	use crate::session::spawn_review_thread;
	use crate::tasks::CompactTask;
	use crate::tasks::UserShellCommandMode;
	use crate::tasks::UserShellCommandTask;
	use crate::tasks::execute_user_shell_command;
	use codex_protocol::protocol::CodexErrorInfo;
	use codex_protocol::protocol::ErrorEvent;
	use codex_protocol::protocol::Event;
	use codex_protocol::protocol::EventMsg;
	use codex_protocol::protocol::GuardianAssessmentEvent;
	use codex_protocol::protocol::GuardianAssessmentStatus;
	use codex_protocol::protocol::InterAgentCommunication;
	use codex_protocol::protocol::Op;
	use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;
	use codex_protocol::protocol::RealtimeVoicesList;
	use codex_protocol::protocol::ReviewDecision;
	use codex_protocol::protocol::ReviewRequest;
	use codex_protocol::protocol::ThreadMemoryMode;
	use codex_protocol::protocol::TurnAbortReason;
	use codex_protocol::protocol::WarningEvent;
	use codex_protocol::request_permissions::RequestPermissionsResponse;
	use codex_protocol::request_user_input::RequestUserInputResponse;
	use codex_thread_store::PersistContext;

	use crate::context_manager::is_user_turn_boundary;
	use codex_protocol::dynamic_tools::DynamicToolResponse;
	use codex_protocol::mcp::RequestId as ProtocolRequestId;
	use codex_rmcp_client::ElicitationAction;
	use codex_rmcp_client::ElicitationResponse;
	use serde_json::Value;
	use std::sync::Arc;
	use tracing::debug;
	use tracing::info;
	use tracing::warn;

	pub async fn interrupt(sess: &Arc<Session>) {
	sess.interrupt_task().await;
	}

	pub async fn clean_background_terminals(sess: &Arc<Session>) {
	sess.close_unified_exec_processes().await;
	}

	pub async fn realtime_conversation_list_voices(sess: &Session, sub_id: String) {
	sess.send_event_raw(Event {
	id: sub_id,
	msg: EventMsg::RealtimeConversationListVoicesResponse(
	RealtimeConversationListVoicesResponseEvent {
	voices: RealtimeVoicesList::builtin(),
	},
	),
	})
	.await;
	}

	/// Queues an inter-agent message, then lets the shared pending-work scheduler
	/// decide whether an idle session should start a regular turn.
	pub async fn inter_agent_communication(
	sess: &Arc<Session>,
	sub_id: String,
	communication: InterAgentCommunication,
	start_options: codex_protocol::turn_input::TurnStartOptions,
	) {
	let trigger_turn = communication.trigger_turn;
	sess.input_queue
	.enqueue_mailbox_communication(communication, start_options)
	.await;
	crate::agent_communication::emit_agent_communication_receive(&sub_id);
	if trigger_turn \|\| sess.has_outstanding_durable_sleep() {
	sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
	.await;
	}
	}

	pub async fn run_user_shell_command(
	sess: &Arc<Session>,
	sub_id: String,
	command: String,
	timeout_ms: Option<u64>,
	) {
	if let Some((turn_context, cancellation_token)) =
	sess.active_turn_context_and_cancellation_token().await
	{
	let session = Arc::clone(sess);
	tokio::spawn(async move {
	execute_user_shell_command(
	session,
	turn_context,
	command,
	timeout_ms,
	cancellation_token,
	UserShellCommandMode::ActiveTurnAuxiliary,
	)
	.await;
	});
	return;
	}

	let turn_context = sess
	.new_turn_with_default_settings(sub_id, Default::default())
	.await;
	sess.spawn_task(
	turn_context,
	Vec::new(),
	UserShellCommandTask::new(command, timeout_ms),
	)
	.await;
	}

	pub async fn resolve_elicitation(
	sess: &Arc<Session>,
	server_name: String,
	request_id: ProtocolRequestId,
	decision: codex_protocol::approvals::ElicitationAction,
	content: Option<Value>,
	meta: Option<Value>,
	) {
	let action = match decision {
	codex_protocol::approvals::ElicitationAction::Accept => ElicitationAction::Accept,
	codex_protocol::approvals::ElicitationAction::Decline => ElicitationAction::Decline,
	codex_protocol::approvals::ElicitationAction::Cancel => ElicitationAction::Cancel,
	};
	let content = match action {
	// Preserve the legacy fallback for clients that only send an action.
	ElicitationAction::Accept => Some(content.unwrap_or_else(\|\| serde_json::json!({}))),
	ElicitationAction::Decline \| ElicitationAction::Cancel => None,
	_ => None,
	};
	let response = ElicitationResponse {
	action,
	content,
	meta,
	};
	let request_id = match request_id {
	ProtocolRequestId::String(value) => {
	rmcp::model::NumberOrString::String(std::sync::Arc::from(value))
	}
	ProtocolRequestId::Integer(value) => rmcp::model::NumberOrString::Number(value),
	};
	if let Err(err) = sess
	.resolve_elicitation(server_name, request_id, response)
	.await
	{
	warn!(
	error = %err,
	"failed to resolve elicitation request in session"
	);
	}
	}

	/// Propagate a user's exec approval decision to the session.
	/// Also optionally applies an execpolicy amendment.
	pub async fn exec_approval(
	sess: &Arc<Session>,
	approval_id: String,
	turn_id: Option<String>,
	decision: ReviewDecision,
	) {
	let event_turn_id = turn_id.unwrap_or_else(\|\| approval_id.clone());
	if let ReviewDecision::ApprovedExecpolicyAmendment {
	proposed_execpolicy_amendment,
	} = &decision
	&& let Err(err) = sess
	.persist_execpolicy_amendment(proposed_execpolicy_amendment)
	.await
	{
	let message = format!("Failed to apply execpolicy amendment: {err}");
	tracing::warn!("{message}");
	let warning = EventMsg::Warning(WarningEvent { message });
	sess.send_event_raw(Event {
	id: event_turn_id.clone(),
	msg: warning,
	})
	.await;
	}
	match decision {
	ReviewDecision::Abort => {
	sess.interrupt_task().await;
	}
	other => sess.notify_approval(&approval_id, other).await,
	}
	}

	pub async fn patch_approval(sess: &Arc<Session>, id: String, decision: ReviewDecision) {
	match decision {
	ReviewDecision::Abort => {
	sess.interrupt_task().await;
	}
	other => sess.notify_approval(&id, other).await,
	}
	}

	pub async fn request_user_input_response(
	sess: &Arc<Session>,
	id: String,
	response: RequestUserInputResponse,
	) {
	sess.notify_user_input_response(&id, response).await;
	}

	pub async fn request_permissions_response(
	sess: &Arc<Session>,
	id: String,
	response: RequestPermissionsResponse,
	) {
	sess.notify_request_permissions_response(&id, response)
	.await;
	}

	pub async fn dynamic_tool_response(sess: &Arc<Session>, id: String, response: DynamicToolResponse) {
	sess.notify_dynamic_tool_response(&id, response).await;
	}

	pub fn refresh_mcp_servers(sess: &Session) {
	sess.services.mcp_runtime.reconnect_on_next_refresh();
	sess.request_mcp_runtime_refresh();
	}

	pub async fn reload_user_config(sess: &Arc<Session>) {
	sess.reload_user_config_layer().await;
	}

	pub async fn compact(sess: &Arc<Session>, sub_id: String) {
	let turn_context = sess
	.new_turn_with_default_settings(sub_id, Default::default())
	.await;

	sess.spawn_task(turn_context, Vec::new(), CompactTask).await;
	}

	pub(super) async fn persist_thread_memory_mode_update(
	sess: &Arc<Session>,
	mode: ThreadMemoryMode,
	) -> anyhow::Result<()> {
	let live_thread = sess.live_thread_for_persistence("update thread memory mode")?;
	live_thread.persist(PersistContext::Standard).await?;
	live_thread.flush().await?;
	live_thread
	.update_memory_mode(mode, /include_archived/ false)
	.await?;
	live_thread.flush().await?;
	Ok(())
	}

	/// Persists thread-level memory mode metadata for the active session.
	///
	/// This does not involve the model and only affects whether the thread is
	/// eligible for future memory generation.
	pub async fn set_thread_memory_mode(sess: &Arc<Session>, sub_id: String, mode: ThreadMemoryMode) {
	if let Err(err) = persist_thread_memory_mode_update(sess, mode).await {
	warn!("Failed to persist thread memory mode update to rollout: {err}");
	let event = Event {
	id: sub_id,
	msg: EventMsg::Error(ErrorEvent {
	misalignment: None,
	message: err.to_string(),
	codex_error_info: Some(CodexErrorInfo::Other),
	}),
	};
	sess.send_event_raw(event).await;
	}
	}

	pub(super) async fn shutdown_session_runtime(sess: &Arc<Session>) {
	if let Some(startup_prewarm) = sess.take_session_startup_prewarm().await {
	startup_prewarm.abort().await;
	}
	let _ = sess.conversation.shutdown().await;
	sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
	let shell_snapshot_prewarm = sess.state.lock().await.shell_snapshot_prewarm.take();
	if let Some(shell_snapshot_prewarm) = shell_snapshot_prewarm {
	shell_snapshot_prewarm.abort();
	let _ = shell_snapshot_prewarm.await;
	}
	sess.hooks().shutdown().await;
	sess.async_hook_results.close();
	while sess.async_hook_results.try_recv().is_ok() {}
	sess.services
	.unified_exec_manager
	.terminate_all_processes()
	.await;
	if let Err(err) = sess.services.code_mode_service.shutdown().await {
	warn!("failed to shutdown code mode session: {err}");
	}
	sess.stop_mcp_prewarm_worker().await;
	{
	let _refresh = sess.mcp_refresh.acquire().await;
	sess.mcp_refresh.close();
	sess.services.mcp_runtime.shutdown().await;
	}

	crate::hook_runtime::run_session_end_hooks(sess).await;
	emit_thread_stop_lifecycle(sess).await;
	}

	async fn emit_thread_stop_lifecycle(sess: &Session) {
	for contributor in sess.services.extensions.thread_lifecycle_contributors() {
	contributor
	.on_thread_stop(codex_extension_api::ThreadStopInput {
	session_store: &sess.services.session_extension_data,
	thread_store: &sess.services.thread_extension_data,
	})
	.await;
	}
	}

	pub async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool {
	shutdown_session_runtime(sess).await;
	info!("Shutting down Codex instance");
	let history = sess.clone_history().await;
	let turn_count = history
	.raw_items()
	.filter(\|item\| is_user_turn_boundary(item))
	.count();
	sess.services.session_telemetry.counter(
	"codex.conversation.turn.count",
	i64::try_from(turn_count).unwrap_or(0),
	&[],
	);

	// Gracefully flush and shutdown thread persistence on session end so tests
	// that inspect durable state do not race with the background writer.
	if let Some(live_thread) = sess.live_thread()
	&& let Err(e) = live_thread.shutdown().await
	{
	warn!("failed to shutdown thread persistence: {e}");
	let event = Event {
	id: sub_id.clone(),
	msg: EventMsg::Error(ErrorEvent {
	misalignment: None,
	message: "Failed to shutdown thread persistence".to_string(),
	codex_error_info: Some(CodexErrorInfo::Other),
	}),
	};
	sess.send_event_raw(event).await;
	}

	let event = Event {
	id: sub_id,
	msg: EventMsg::ShutdownComplete,
	};
	sess.services
	.rollout_thread_trace
	.record_protocol_event(&event.msg);
	sess.deliver_event_raw(event).await;
	sess.services
	.rollout_thread_trace
	.record_ended(codex_rollout_trace::RolloutStatus::Completed);
	true
	}

	pub async fn review(
	sess: &Arc<Session>,
	config: &Arc<Config>,
	sub_id: String,
	review_request: ReviewRequest,
	) {
	let turn_context = sess
	.new_turn_with_default_settings(sub_id.clone(), Default::default())
	.await;
	sess.maybe_emit_model_warnings_for_turn(turn_context.as_ref())
	.await;
	#[allow(deprecated)]
	match resolve_review_request(review_request, &turn_context.cwd) {
	Ok(resolved) => {
	spawn_review_thread(
	Arc::clone(sess),
	Arc::clone(config),
	turn_context.clone(),
	sub_id,
	resolved,
	)
	.await;
	}
	Err(err) => {
	let event = Event {
	id: sub_id,
	msg: EventMsg::Error(ErrorEvent {
	misalignment: None,
	message: err.to_string(),
	codex_error_info: Some(CodexErrorInfo::Other),
	}),
	};
	sess.send_event(&turn_context, event.msg).await;
	}
	}
	}

	pub(super) async fn submission_loop(
	sess: Arc<Session>,
	config: Arc<Config>,
	rx_sub: Receiver<Submission>,
	) {
	// To break out of this loop, send Op::Shutdown.
	let mut shutdown_received = false;
	while let Ok(sub) = rx_sub.recv().await {
	if matches!(sub.op, Op::ResolveElicitation { .. }) {
	debug!(submission_id = %sub.id, operation = sub.op.kind(), "Submission");
	} else {
	debug!(?sub, "Submission");
	}
	let dispatch_span = submission_dispatch_span(&sub);
	let should_exit = async {
	match sub.op {
	Op::Interrupt => {
	interrupt(&sess).await;
	false
	}
	Op::CleanBackgroundTerminals => {
	clean_background_terminals(&sess).await;
	false
	}
	Op::RealtimeConversationStart(params) => {
	if let Err(err) =
	handle_realtime_conversation_start(&sess, sub.id.clone(), params).await
	{
	sess.send_event_raw(Event {
	id: sub.id.clone(),
	msg: EventMsg::Error(ErrorEvent {
	misalignment: None,
	message: err.to_string(),
	codex_error_info: Some(CodexErrorInfo::Other),
	}),
	})
	.await;
	}
	false
	}
	Op::RealtimeConversationAudio(params) => {
	handle_realtime_conversation_audio(&sess, sub.id.clone(), params).await;
	false
	}
	Op::RealtimeConversationText(params) => {
	handle_realtime_conversation_text(&sess, sub.id.clone(), params).await;
	false
	}
	Op::RealtimeConversationSpeech(params) => {
	handle_realtime_conversation_speech(&sess, sub.id.clone(), params).await;
	false
	}
	Op::RealtimeConversationClose => {
	handle_realtime_conversation_close(&sess, sub.id.clone()).await;
	false
	}
	Op::RealtimeConversationListVoices => {
	realtime_conversation_list_voices(&sess, sub.id.clone()).await;
	false
	}
	Op::TurnInput {
	request,
	mode,
	reply,
	} => {
	let result = turn_input::handle(&sess, *request, mode, sub.id.clone()).await;
	let _ = reply.send(result);
	false
	}
	Op::RecoverTurn {
	thread_settings,
	start_options,
	reply,
	} => {
	let result = turn_input::handle_recovery(
	&sess,
	thread_settings,
	start_options,
	sub.id.clone(),
	)
	.await;
	let _ = reply.send(result);
	false
	}
	Op::SuspendTurnAndShutdown { reply } => {
	let result =
	super::turn_suspension::suspend_turn_and_shutdown(&sess, sub.id.clone())
	.await;
	// Exit only after history is durable and its writer has closed; an error
	// must leave responsibility for the thread with the current worker.
	let should_exit = matches!(
	&result,
	Ok(codex_protocol::turn_input::SuspendTurnOutcome::Suspended { .. })
	);
	let _ = reply.send(result);
	should_exit
	}
	Op::ThreadSettings { thread_settings } => {
	thread_settings::update(&sess, sub.id.clone(), thread_settings).await;
	false
	}
	Op::TurnSettings {
	turn_id,
	update,
	reply,
	} => {
	let outcome = sess.apply_turn_settings(&turn_id, update).await;
	let _ = reply.send(outcome);
	false
	}
	Op::InterAgentCommunication {
	communication,
	start_options,
	} => {
	inter_agent_communication(&sess, sub.id.clone(), communication, start_options)
	.await;
	false
	}
	Op::ExecApproval {
	id: approval_id,
	turn_id,
	decision,
	} => {
	exec_approval(&sess, approval_id, turn_id, decision).await;
	false
	}
	Op::PatchApproval { id, decision } => {
	patch_approval(&sess, id, decision).await;
	false
	}
	Op::UserInputAnswer { id, response } => {
	request_user_input_response(&sess, id, response).await;
	false
	}
	Op::RequestPermissionsResponse { id, response } => {
	request_permissions_response(&sess, id, response).await;
	false
	}
	Op::DynamicToolResponse { id, response } => {
	dynamic_tool_response(&sess, id, response).await;
	false
	}
	Op::RefreshMcpServers => {
	refresh_mcp_servers(&sess);
	false
	}
	Op::ReloadUserConfig => {
	reload_user_config(&sess).await;
	false
	}
	Op::Compact => {
	compact(&sess, sub.id.clone()).await;
	false
	}
	Op::SetThreadMemoryMode { mode } => {
	set_thread_memory_mode(&sess, sub.id.clone(), mode).await;
	false
	}
	Op::RunUserShellCommand {
	command,
	timeout_ms,
	} => {
	run_user_shell_command(&sess, sub.id.clone(), command, timeout_ms).await;
	false
	}
	Op::ResolveElicitation {
	server_name,
	request_id,
	decision,
	content,
	meta,
	} => {
	resolve_elicitation(&sess, server_name, request_id, decision, content, meta)
	.await;
	false
	}
	Op::Shutdown => shutdown(&sess, sub.id.clone()).await,
	Op::Review { review_request } => {
	review(&sess, &config, sub.id.clone(), review_request).await;
	false
	}
	Op::ApproveGuardianDeniedAction { event } => {
	approve_guardian_denied_action(&sess, event).await;
	false
	}
	_ => false, // Ignore unknown ops; enum is non_exhaustive to allow extensions.
	}
	}
	.instrument(dispatch_span)
	.await;
	if should_exit {
	shutdown_received = true;
	break;
	}
	}
	// If the submission loop exits because the channel closed without an
	// explicit shutdown op, still run session teardown.
	if !shutdown_received {
	shutdown_session_runtime(&sess).await;
	if let Some(live_thread) = sess.live_thread()
	&& let Err(err) = live_thread.shutdown().await
	{
	warn!("failed to shutdown thread persistence after submission channel closed: {err}");
	}
	}
	debug!("Agent loop exited");
	}

	async fn approve_guardian_denied_action(sess: &Arc<Session>, event: GuardianAssessmentEvent) {
	if event.status != GuardianAssessmentStatus::Denied {
	warn!(
	review_id = event.id.as_str(),
	"ignoring approval for non-denied Guardian assessment"
	);
	return;
	}

	let approved_action = serde_json::json!({
	"action": &event.action,
	"outcome": "allowed",
	});
	let approved_action_json = match serde_json::to_string_pretty(&approved_action) {
	Ok(approved_action_json) => approved_action_json,
	Err(error) => {
	warn!(%error, review_id = event.id.as_str(), "failed to serialize approved Guardian action");
	return;
	}
	};
	let items = vec![ContextualUserFragment::into(GuardianApprovedAction::new(
	approved_action_json,
	))];

	sess.inject_no_new_turn(items, /current_turn_context/ None)
	.await;
	}

	pub(super) fn submission_dispatch_span(sub: &Submission) -> tracing::Span {
	let op_name = sub.op.kind();
	let span_name = format!("op.dispatch.{op_name}");
	let dispatch_span = match &sub.op {
	Op::RealtimeConversationAudio(_) => {
	debug_span!(
	"submission_dispatch",
	otel.name = span_name.as_str(),
	submission.id = sub.id.as_str(),
	codex.op = op_name
	)
	}
	_ => info_span!(
	"submission_dispatch",
	otel.name = span_name.as_str(),
	submission.id = sub.id.as_str(),
	codex.op = op_name
	),
	};
	if let Some(trace) = sub.trace.as_ref()
	&& !set_parent_from_w3c_trace_context(&dispatch_span, trace)
	{
	warn!(
	submission.id = sub.id.as_str(),
	"ignoring invalid submission trace carrier"
	);
	}
	dispatch_span
	}