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

Preserve user input when MCP startup is interrupted (#34839)

Browse Source 
## Why

Interrupting a turn while MCP tools were still starting could abort before the
submitted user input was recorded in conversation history.

## What changed

- Build and retain the MCP tool list and tool router as part of each step
  snapshot, using the turn cancellation token during tool discovery.
- Record turn hooks and submitted input when cancellation occurs during startup
  prewarming, pre-sampling compaction, or first-step capture.
- Reuse the captured router for sampling, compaction, and prompt debugging so
  advertised and executable tools share the same request-scoped view.

## Testing

Add coverage that interrupts each affected MCP startup phase and verifies that
the user prompt remains in history before the turn-aborted marker.

GitOrigin-RevId: 6e96c6814357a55545d69b580e28f35933cfadf7
This commit is contained in:
Ahmed Ibrahim
2026-07-22 22:50:13 +00:00
committed by copyberry
parent bd5b55e403
commit d7e8f4c3dc
21 changed files with 359 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex_thread.rs
+23 -5
View File
@@ -56,6 +56,7 @@ use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use codex_rollout::state_db::StateDbHandle;
@@ -488,8 +489,8 @@ impl CodexThread {
// This history-only API runs without run_turn, so it owns its initial step.
let step_context = self
.session
.capture_step_context(Arc::clone(&turn_context))
.await;
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await?;
self.session
.record_context_updates_and_set_reference_context_item(step_context.as_ref())
.await;
@@ -608,11 +609,28 @@ impl CodexThread {
/// Returns the exact MCP config, environment bindings, and manager most recently published.
pub async fn current_mcp_runtime(&self) -> Arc<crate::session::McpRuntimeSnapshot> {
let turn_context = self.session.new_default_turn().await;
let environments = turn_context.environments.refresh_readiness();
let selected_capability_roots = self
.session
.resolve_selected_capability_roots_for_step(&environments)
.await;
let ready_selected_capability_roots =
Session::ready_selected_capability_roots(&selected_capability_roots);
let executor_capability_discovery = self
.session
.executor_capability_discovery_for_step(
&turn_context.config,
&ready_selected_capability_roots,
)
.await;
self.session
.capture_step_context(turn_context)
.mcp_runtime_for_step(
turn_context.as_ref(),
&environments,
&selected_capability_roots,
executor_capability_discovery.as_deref(),
)
.await
.mcp
.clone()
}
pub fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
codex-rs/core/src/compact.rs
-1
View File
@@ -63,7 +63,6 @@ const COMPACT_USER_MESSAGE_MAX_TOKENS: usize = 20_000;
/// Mid-turn compaction must use `BeforeLastUserMessage` because the model is trained to see the
/// compaction summary as the last item in history after mid-turn compaction; we therefore inject
/// initial context into the replacement history just above the last real user message.
#[derive(Debug)]
pub(crate) enum InitialContextInjection {
BeforeLastUserMessage {
world_state: Arc<WorldState>,
codex-rs/core/src/compact_remote.rs
+4 -1
View File
@@ -37,6 +37,7 @@ use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnStartedEvent;
use codex_rollout_trace::CompactionCheckpointTracePayload;
use codex_utils_output_truncation::approx_token_count;
use tokio_util::sync::CancellationToken;
#[path = "compact_remote_request.rs"]
mod request;
@@ -78,7 +79,9 @@ pub(crate) async fn run_remote_compact_task(
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
// Standalone compaction is its own request boundary, so it captures a fresh step.
let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
let step_context = sess
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await?;
let start_event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
trace_id: turn_context.trace_id.clone(),
codex-rs/core/src/compact_remote_request.rs
+1 -8
View File
@@ -9,12 +9,10 @@ use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_metadata::CompactionTurnMetadata;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn::built_tools;
use codex_protocol::auth::AuthMode;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ResponseItem;
use codex_rollout_trace::CompactionTraceContext;
use tokio_util::sync::CancellationToken;
use tracing::info;
pub(super) struct RemoteCompactAttempt {
@@ -61,12 +59,7 @@ pub(super) async fn run_remote_compact_attempt(
.is_enabled()
.then(|| history.raw_items().to_vec());
let prompt_input = history.for_prompt(&turn_context.model_info.input_modalities);
let tool_router = built_tools(
sess.as_ref(),
step_context.as_ref(),
&CancellationToken::new(),
)
.await?;
let tool_router = &step_context.tool_router;
let prompt = Prompt {
input: prompt_input,
tools: tool_router.model_visible_specs(),
codex-rs/core/src/compact_remote_v2.rs
+4 -1
View File
@@ -43,6 +43,7 @@ use codex_rollout_trace::InferenceTraceContext;
use codex_utils_output_truncation::approx_token_count;
use codex_utils_output_truncation::truncate_text;
use futures::StreamExt;
use tokio_util::sync::CancellationToken;
#[path = "compact_remote_v2_attempt.rs"]
mod attempt;
@@ -87,7 +88,9 @@ pub(crate) async fn run_remote_compact_task(
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
// Standalone compaction is its own request boundary, so it captures a fresh step.
let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
let step_context = sess
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await?;
let start_event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
trace_id: turn_context.trace_id.clone(),
codex-rs/core/src/compact_remote_v2_attempt.rs
+1 -8
View File
@@ -10,14 +10,12 @@ use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_metadata::CompactionTurnMetadata;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn::built_tools;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RawResponseCompletedEvent;
use codex_protocol::protocol::TokenUsage;
use codex_rollout_trace::CompactionTraceContext;
use tokio_util::sync::CancellationToken;
use tracing::info;
pub(super) struct RemoteCompactV2Attempt {
@@ -69,12 +67,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
.is_enabled()
.then(|| history.raw_items().to_vec());
let mut input = history.for_prompt(&turn_context.model_info.input_modalities);
let tool_router = built_tools(
sess.as_ref(),
step_context.as_ref(),
&CancellationToken::new(),
)
.await?;
let tool_router = &step_context.tool_router;
input.push(ResponseItem::CompactionTrigger {});
let prompt = Prompt {
input,
codex-rs/core/src/compact_token_budget.rs
+4 -1
View File
@@ -16,6 +16,7 @@ use codex_protocol::items::ContextCompactionItem;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnStartedEvent;
use tokio_util::sync::CancellationToken;
/// Runs token-budget manual compaction as a normal compaction lifecycle.
///
@@ -36,7 +37,9 @@ pub(crate) async fn run_manual_compact_task(
sess.send_event(&turn_context, start_event).await;
// Manual compaction runs outside run_turn, so it captures its own current step.
let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
let step_context = sess
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await?;
let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await);
run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Manual).await
}
codex-rs/core/src/prompt_debug.rs
+4 -4
View File
@@ -16,7 +16,6 @@ use crate::config::Config;
use crate::resolve_installation_id;
use crate::session::session::Session;
use crate::session::turn::build_prompt;
use crate::session::turn::built_tools;
use crate::state_db_bridge::StateDbHandle;
use crate::thread_manager::StartThreadOptions;
use crate::thread_manager::ThreadManager;
@@ -84,7 +83,9 @@ pub(crate) async fn build_prompt_input_from_session(
) -> CodexResult<Vec<ResponseItem>> {
let turn_context = sess.new_default_turn().await;
// Prompt debugging builds a standalone request without entering run_turn.
let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
let step_context = sess
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await?;
sess.record_context_updates_and_set_reference_context_item(step_context.as_ref())
.await;
@@ -98,11 +99,10 @@ pub(crate) async fn build_prompt_input_from_session(
.clone_history()
.await
.for_prompt(&turn_context.model_info.input_modalities);
let router = built_tools(sess, step_context.as_ref(), &CancellationToken::new()).await?;
let base_instructions = sess.get_base_instructions().await;
let prompt = build_prompt(
prompt_input,
router.as_ref(),
step_context.tool_router.as_ref(),
turn_context.as_ref(),
base_instructions,
);
codex-rs/core/src/session/mod.rs
+15 -4
View File
@@ -2920,7 +2920,8 @@ impl Session {
pub(crate) async fn capture_step_context(
self: &Arc<Self>,
turn_context: Arc<TurnContext>,
) -> Arc<StepContext> {
cancellation_token: &CancellationToken,
) -> CodexResult<Arc<StepContext>> {
// Keep selections fixed for the turn while allowing their startup work to finish.
let environments = turn_context.environments.refresh_readiness();
self.services
@@ -2947,14 +2948,24 @@ impl Session {
executor_capability_discovery.as_deref(),
)
.await;
Arc::new(StepContext::new(
turn_context,
let (mcp_tools, tool_router) = turn::built_tools(
self.as_ref(),
turn_context.as_ref(),
&environments,
mcp.as_ref(),
cancellation_token,
)
.await?;
Ok(Arc::new(StepContext {
turn: turn_context,
environments,
selected_capability_roots,
executor_capability_discovery,
mcp,
mcp_tools,
tool_router,
loaded_agents_md,
))
}))
}
pub(crate) async fn record_inter_agent_communication(
codex-rs/core/src/session/step_context.rs
+4 -30
View File
@@ -4,13 +4,12 @@ use crate::agents_md::LoadedAgentsMd;
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::session::McpRuntimeSnapshot;
use crate::session::turn_context::TurnContext;
use crate::tools::router::ToolRouter;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_mcp::ToolInfo;
use tokio::sync::OnceCell;
/// Request-scoped state that may change between model sampling requests.
#[derive(Debug)]
pub(crate) struct StepContext {
pub(crate) turn: Arc<TurnContext>,
pub(crate) environments: TurnEnvironmentSnapshot,
@@ -21,34 +20,9 @@ pub(crate) struct StepContext {
/// The exact MCP config and manager used to advertise and execute tools for this step.
pub(crate) mcp: Arc<McpRuntimeSnapshot>,
/// The fixed MCP tool list used for this exact sampling request.
mcp_tool_snapshot: OnceCell<Vec<ToolInfo>>,
pub(crate) mcp_tools: Vec<ToolInfo>,
/// The finalized tool plan advertised and executed for this exact sampling request.
pub(crate) tool_router: Arc<ToolRouter>,
/// The canonical AGENTS.md value observed with this environment snapshot.
pub(crate) loaded_agents_md: Option<Arc<LoadedAgentsMd>>,
}
impl StepContext {
pub(crate) fn new(
turn: Arc<TurnContext>,
environments: TurnEnvironmentSnapshot,
selected_capability_roots: Vec<ResolvedSelectedCapabilityRoot>,
executor_capability_discovery: Option<Arc<ExecutorCapabilityDiscoverySnapshot>>,
mcp: Arc<McpRuntimeSnapshot>,
loaded_agents_md: Option<Arc<LoadedAgentsMd>>,
) -> Self {
Self {
turn,
environments,
selected_capability_roots,
executor_capability_discovery,
mcp,
mcp_tool_snapshot: OnceCell::new(),
loaded_agents_md,
}
}
pub(crate) async fn mcp_tools(&self) -> &[ToolInfo] {
self.mcp_tool_snapshot
.get_or_init(|| self.mcp.manager().list_all_tools())
.await
}
}
codex-rs/core/src/session/tests.rs
+29 -20
View File
@@ -17,6 +17,7 @@ use crate::skills::SkillRenderSideEffects;
use crate::skills::render::SkillMetadataBudget;
use crate::test_support::models_manager_with_provider;
use crate::tools::format_exec_output_str;
use crate::tools::registry::ToolRegistry;
use codex_config::ConfigLayerStack;
use codex_config::ConfigLayerStackOrdering;
use codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID;
@@ -197,14 +198,19 @@ use std::time::Duration as StdDuration;
impl StepContext {
pub(crate) fn for_test(turn: Arc<TurnContext>) -> Arc<Self> {
let environments = turn.environments.clone();
Arc::new(Self::new(
Arc::clone(&turn),
Arc::new(Self {
turn: Arc::clone(&turn),
environments,
Vec::new(),
/*executor_capability_discovery*/ None,
crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config),
/*loaded_agents_md*/ None,
))
selected_capability_roots: Vec::new(),
executor_capability_discovery: None,
mcp: crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config),
mcp_tools: Vec::new(),
tool_router: Arc::new(ToolRouter::from_parts(
ToolRegistry::empty_for_test(),
Vec::new(),
)),
loaded_agents_md: None,
})
}
}
@@ -652,7 +658,9 @@ async fn preview_session_start_hooks(
fn test_tool_runtime(session: Arc<Session>, turn_context: Arc<TurnContext>) -> ToolCallRuntime {
let step_context = StepContext::for_test(Arc::clone(&turn_context));
let router = Arc::new(ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
crate::tools::router::ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
@@ -2895,8 +2903,9 @@ async fn start_new_context_window_assigns_and_persists_item_ids() {
let rollout_path =
attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await;
let step_context = session
.capture_step_context(Arc::clone(&turn_context))
.await;
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await
.expect("a fresh cancellation token cannot be cancelled");
let world_state = Arc::new(session.build_world_state_for_step(&step_context).await);
session
@@ -7737,8 +7746,9 @@ async fn refresh_mcp_servers_keeps_the_previous_runtime_alive() {
let turn_context = Arc::new(turn_context);
let old_runtime = session.services.latest_mcp_runtime();
let step_context = session
.capture_step_context(Arc::clone(&turn_context))
.await;
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
.await
.expect("a fresh cancellation token cannot be cancelled");
assert!(Arc::ptr_eq(&step_context.mcp, &old_runtime));
let old_token = session.mcp_startup_cancellation_token().await;
assert!(!old_token.is_cancelled());
@@ -8015,7 +8025,9 @@ async fn built_tools_uses_the_step_mcp_runtime() -> anyhow::Result<()> {
let (session, turn_context) = make_session_and_context().await;
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let step_context = session.capture_step_context(turn_context).await;
let step_context = session
.capture_step_context(turn_context, &CancellationToken::new())
.await?;
let mut refresh_config = step_context.turn.config.as_ref().clone();
refresh_config.mcp_servers.set(HashMap::from([(
@@ -8053,12 +8065,7 @@ async fn built_tools_uses_the_step_mcp_runtime() -> anyhow::Result<()> {
)
.await;
let router = crate::session::turn::built_tools(
session.as_ref(),
step_context.as_ref(),
&CancellationToken::new(),
)
.await?;
let router = &step_context.tool_router;
assert!(
!router
.registered_tool_names_for_test()
@@ -10554,7 +10561,9 @@ async fn fatal_tool_error_stops_turn_and_reports_error() {
let (session, turn_context, _rx) = make_session_and_context_with_rx().await;
let step_context = StepContext::for_test(Arc::clone(&turn_context));
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
crate::tools::router::ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
codex-rs/core/src/session/turn.rs
+69 -30
View File
@@ -16,6 +16,7 @@ use crate::compact_remote::run_inline_remote_auto_compact_task;
use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2;
use crate::connectors;
use crate::context::ContextualUserFragment;
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::feedback_tags;
use crate::hook_runtime::inspect_pending_input;
use crate::hook_runtime::record_additional_contexts;
@@ -38,6 +39,7 @@ use crate::responses_metadata::CodexResponsesMetadata;
use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_retry::ResponsesStreamRequest;
use crate::responses_retry::handle_retryable_response_stream_error;
use crate::session::McpRuntimeSnapshot;
use crate::session::PreviousTurnSettings;
use crate::session::TurnInput;
use crate::session::session::Session;
@@ -80,6 +82,7 @@ use codex_extension_api::TurnInputEnvironment;
use codex_features::Feature;
use codex_file_system::FindUpErrorPolicy;
use codex_file_system::find_nearest_ancestor_with_markers;
use codex_mcp::ToolInfo;
use codex_protocol::ResponseItemId;
use codex_protocol::config_types::AutoCompactTokenLimitScope;
use codex_protocol::config_types::ModeKind;
@@ -158,8 +161,16 @@ pub(crate) async fn run_turn(
// new user message are recorded. Estimate pending incoming items (context
// diffs/full reinjection + user input) and trigger compaction preemptively
// when they would push the thread over the compaction threshold.
if let Err(err) = run_pre_sampling_compact(&sess, &turn_context, &mut client_session).await {
if let Err(err) = run_pre_sampling_compact(
&sess,
&turn_context,
&mut client_session,
&cancellation_token,
)
.await
{
if matches!(err, CodexErr::TurnAborted) {
run_hooks_and_record_inputs(&sess, &turn_context, &input).await;
return Err(err);
}
let error = err.to_codex_protocol_error();
@@ -170,7 +181,17 @@ pub(crate) async fn run_turn(
}
// run_turn owns the step used to seed context and make the first sampling request.
let first_step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
let first_step_context = match sess
.capture_step_context(Arc::clone(&turn_context), &cancellation_token)
.await
{
Ok(step_context) => step_context,
Err(err @ CodexErr::TurnAborted) => {
run_hooks_and_record_inputs(&sess, &turn_context, &input).await;
return Err(err);
}
Err(err) => return Err(err),
};
// Keep the exact model-visible state used by this turn and its inline compactions.
let (mut world_state, display_roots) = tokio::join!(
sess.record_context_updates_and_set_reference_context_item(first_step_context.as_ref()),
@@ -252,7 +273,10 @@ pub(crate) async fn run_turn(
// Capture once so context, advertised tools, and tool calls share one request view.
let step_context = match next_step_context.take() {
Some(step_context) => step_context,
None => sess.capture_step_context(Arc::clone(&turn_context)).await,
None => {
sess.capture_step_context(Arc::clone(&turn_context), &cancellation_token)
.await?
}
};
let sampling_request_result: CodexResult<_> = async {
super::time_reminder::maybe_record_current_time_reminder(
@@ -507,7 +531,7 @@ async fn turn_diff_display_roots(step_context: &StepContext) -> Vec<(String, Pat
}
#[instrument(level = "trace", skip_all)]
async fn run_hooks_and_record_inputs(
pub(crate) async fn run_hooks_and_record_inputs(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
input: &[TurnInput],
@@ -827,15 +851,19 @@ async fn run_pre_sampling_compact(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
client_session: &mut ModelClientSession,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
maybe_run_previous_model_inline_compact(sess, turn_context, client_session).await?;
maybe_run_previous_model_inline_compact(sess, turn_context, client_session, cancellation_token)
.await?;
let token_status =
super::context_window::context_window_token_status(sess.as_ref(), turn_context.as_ref())
.await;
// Compact if the configured auto-compaction budget or usable context window is exhausted.
if token_status.token_limit_reached {
// Pre-turn compaction runs before run_turn creates the normal sampling step.
let step_context = sess.capture_step_context(Arc::clone(turn_context)).await;
let step_context = sess
.capture_step_context(Arc::clone(turn_context), cancellation_token)
.await?;
run_auto_compact(
sess,
step_context,
@@ -866,7 +894,8 @@ async fn capture_current_model_fallback_step_context(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
previous_model: &str,
) -> Option<Arc<StepContext>> {
cancellation_token: &CancellationToken,
) -> CodexResult<Option<Arc<StepContext>>> {
let uses_codex_backend = turn_context
.auth_manager
.as_deref()
@@ -875,9 +904,11 @@ async fn capture_current_model_fallback_step_context(
|| !turn_context.provider.info().is_openai()
|| previous_model == turn_context.model_info.slug
{
return None;
return Ok(None);
}
Some(sess.capture_step_context(Arc::clone(turn_context)).await)
sess.capture_step_context(Arc::clone(turn_context), cancellation_token)
.await
.map(Some)
}
/// Runs pre-sampling compaction against the previous model when its compaction compatibility
@@ -888,6 +919,7 @@ async fn maybe_run_previous_model_inline_compact(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
client_session: &mut ModelClientSession,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
let Some(previous_turn_settings) = sess.previous_turn_settings().await else {
return Ok(());
@@ -905,14 +937,15 @@ async fn maybe_run_previous_model_inline_compact(
if should_compact_for_comp_hash_change {
let step_context = sess
.capture_step_context(Arc::clone(&previous_model_turn_context))
.await;
.capture_step_context(Arc::clone(&previous_model_turn_context), cancellation_token)
.await?;
let fallback_step_context = capture_current_model_fallback_step_context(
sess,
turn_context,
previous_model.as_str(),
cancellation_token,
)
.await;
.await?;
run_auto_compact(
sess,
step_context,
@@ -952,14 +985,15 @@ async fn maybe_run_previous_model_inline_compact(
&& old_context_window > new_context_window;
if should_run {
let step_context = sess
.capture_step_context(Arc::clone(&previous_model_turn_context))
.await;
.capture_step_context(Arc::clone(&previous_model_turn_context), cancellation_token)
.await?;
let fallback_step_context = capture_current_model_fallback_step_context(
sess,
turn_context,
previous_model.as_str(),
cancellation_token,
)
.await;
.await?;
run_auto_compact(
sess,
step_context,
@@ -1149,7 +1183,7 @@ async fn run_sampling_request(
cancellation_token: CancellationToken,
) -> CodexResult<(SamplingRequestResult, Vec<ResponseItem>)> {
let turn_context = Arc::clone(&step_context.turn);
let router = built_tools(sess.as_ref(), step_context.as_ref(), &cancellation_token).await?;
let router = Arc::clone(&step_context.tool_router);
let base_instructions = sess.get_base_instructions().await;
@@ -1238,19 +1272,21 @@ async fn run_sampling_request(
#[instrument(level = "trace",
skip_all,
fields(
turn_id = %step_context.turn.sub_id,
model = %step_context.turn.model_info.slug,
apps_enabled = step_context.turn.apps_enabled()
turn_id = %turn_context.sub_id,
model = %turn_context.model_info.slug,
apps_enabled = turn_context.apps_enabled()
)
)]
pub(crate) async fn built_tools(
sess: &Session,
step_context: &StepContext,
turn_context: &TurnContext,
environments: &TurnEnvironmentSnapshot,
mcp: &McpRuntimeSnapshot,
cancellation_token: &CancellationToken,
) -> CodexResult<Arc<ToolRouter>> {
let turn_context = step_context.turn.as_ref();
let all_mcp_tools = step_context
.mcp_tools()
) -> CodexResult<(Vec<ToolInfo>, Arc<ToolRouter>)> {
let all_mcp_tools = mcp
.manager()
.list_all_tools()
.or_cancel(cancellation_token)
.await?;
let loaded_plugins = sess
@@ -1259,11 +1295,11 @@ pub(crate) async fn built_tools(
.plugins_for_config(&turn_context.config.plugins_config_input())
.instrument(trace_span!("built_tools.load_plugins"))
.await;
let connector_snapshot = step_context.mcp.config().connector_snapshot.clone();
let connector_snapshot = mcp.config().connector_snapshot.clone();
let apps_enabled = turn_context.apps_enabled();
let accessible_connectors =
apps_enabled.then(|| connectors::accessible_connectors_from_mcp_tools(all_mcp_tools));
apps_enabled.then(|| connectors::accessible_connectors_from_mcp_tools(&all_mcp_tools));
let accessible_connectors_with_enabled_state =
accessible_connectors.as_ref().map(|connectors| {
connectors::with_app_enabled_state(connectors.clone(), &turn_context.config)
@@ -1356,13 +1392,15 @@ pub(crate) async fn built_tools(
.await
};
let mcp_tool_runtimes = build_mcp_tool_runtimes(
all_mcp_tools,
&all_mcp_tools,
connectors.as_deref(),
&turn_context.config,
search_tool_enabled(turn_context),
);
Ok(Arc::new(ToolRouter::from_context(
step_context,
let tool_router = Arc::new(ToolRouter::from_context(
turn_context,
environments,
mcp,
ToolRouterParams {
tool_runtimes: mcp_tool_runtimes,
tool_suggest_candidates,
@@ -1370,7 +1408,8 @@ pub(crate) async fn built_tools(
dynamic_tools: turn_context.dynamic_tools.as_slice(),
},
&sess.services.tool_search_handler_cache,
)))
));
Ok((all_mcp_tools, tool_router))
}
#[derive(Debug)]
codex-rs/core/src/session/world_state.rs
+1 -2
View File
@@ -91,9 +91,8 @@ impl Session {
));
let apps_available =
if turn_context.config.include_apps_instructions && turn_context.apps_enabled() {
let tools = step_context.mcp_tools().await;
connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(tools),
connectors::accessible_connectors_from_mcp_tools(&step_context.mcp_tools),
&turn_context.config,
)
.into_iter()
codex-rs/core/src/session_startup_prewarm.rs
+6 -9
View File
@@ -17,7 +17,6 @@ use crate::responses_metadata::CodexResponsesRequestKind;
use crate::session::INITIAL_SUBMIT_ID;
use crate::session::session::Session;
use crate::session::turn::build_prompt;
use crate::session::turn::built_tools;
use codex_otel::STARTUP_PREWARM_AGE_AT_FIRST_TURN_METRIC;
use codex_otel::STARTUP_PREWARM_DURATION_METRIC;
use codex_otel::SessionTelemetry;
@@ -278,14 +277,12 @@ async fn schedule_startup_prewarm_inner(
let built_tools_started_at = Instant::now();
// Startup prewarm runs before run_turn and needs its own tool-building snapshot.
let step_context = session
.capture_step_context(Arc::clone(&startup_turn_context))
.await;
let startup_router = built_tools(
session.as_ref(),
step_context.as_ref(),
&startup_cancellation_token,
)
.await?;
.capture_step_context(
Arc::clone(&startup_turn_context),
&startup_cancellation_token,
)
.await?;
let startup_router = Arc::clone(&step_context.tool_router);
startup_turn_context.session_telemetry.record_startup_phase(
"startup_prewarm_build_tools",
built_tools_started_at.elapsed(),
codex-rs/core/src/stream_events_utils_tests.rs
+3 -1
View File
@@ -277,7 +277,9 @@ async fn handle_output_item_done_returns_contributed_last_agent_message() {
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(Arc::clone(&turn_context));
let router = Arc::new(ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
crate::tools::router::ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
codex-rs/core/src/tasks/regular.rs
+5 -1
View File
@@ -3,6 +3,7 @@ use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use crate::session::TurnInput;
use crate::session::turn::run_hooks_and_record_inputs;
use crate::session::turn::run_turn;
use crate::session::turn_context::TurnContext;
use crate::session_startup_prewarm::SessionStartupPrewarmResolution;
@@ -62,7 +63,10 @@ impl SessionTask for RegularTask {
.instrument(trace_span!("regular_task.prepare_run_turn"))
.await;
let prewarmed_client_session = match prewarmed_client_session {
SessionStartupPrewarmResolution::Cancelled => return Ok(None),
SessionStartupPrewarmResolution::Cancelled => {
run_hooks_and_record_inputs(&sess, &ctx, &input).await;
return Ok(None);
}
SessionStartupPrewarmResolution::Unavailable { .. } => None,
SessionStartupPrewarmResolution::Ready(prewarmed_client_session) => {
Some(*prewarmed_client_session)
codex-rs/core/src/tools/router.rs
+13 -2
View File
@@ -1,6 +1,9 @@
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::function_tool::FunctionCallError;
use crate::session::McpRuntimeSnapshot;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
@@ -58,11 +61,19 @@ pub(crate) struct ToolSuggestCandidates {
impl ToolRouter {
pub(crate) fn from_context(
step_context: &StepContext,
turn_context: &TurnContext,
environments: &TurnEnvironmentSnapshot,
mcp: &McpRuntimeSnapshot,
params: ToolRouterParams<'_>,
tool_search_handler_cache: &ToolSearchHandlerCache,
) -> Self {
build_tool_router(step_context, params, tool_search_handler_cache)
build_tool_router(
turn_context,
environments,
mcp,
params,
tool_search_handler_cache,
)
}
pub(crate) fn from_parts(registry: ToolRegistry, model_visible_specs: Vec<ToolSpec>) -> Self {
codex-rs/core/src/tools/router_tests.rs
+15 -6
View File
@@ -112,7 +112,9 @@ async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
@@ -210,7 +212,9 @@ async fn mcp_parallel_support_uses_handler_data() -> anyhow::Result<()> {
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: vec![
@@ -260,7 +264,9 @@ async fn tools_without_handlers_do_not_support_parallel() -> anyhow::Result<()>
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
@@ -316,7 +322,9 @@ async fn specs_filter_deferred_dynamic_tools() -> anyhow::Result<()> {
})];
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
@@ -330,7 +338,6 @@ async fn specs_filter_deferred_dynamic_tools() -> anyhow::Result<()> {
namespace_function_names(&router.model_visible_specs(), "codex_app"),
vec![visible_tool.to_string()]
);
Ok(())
}
@@ -387,7 +394,9 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow
expected_history_item.set_turn_id_if_missing(&turn.sub_id);
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: None,
tool_runtimes: Vec::new(),
codex-rs/core/src/tools/spec_plan.rs
+40 -28
View File
@@ -1,6 +1,7 @@
use crate::agent::exceeds_thread_spawn_depth_limit;
use crate::agent::next_thread_spawn_depth;
use crate::session::step_context::StepContext;
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::session::McpRuntimeSnapshot;
use crate::session::turn_context::TurnContext;
use crate::tools::code_mode::default_exec_yield_time_override_ms;
use crate::tools::code_mode::execute_spec::create_code_mode_tool;
@@ -139,7 +140,9 @@ impl PlannedTools {
#[derive(Clone, Copy)]
struct CoreToolPlanContext<'a> {
step_context: &'a StepContext,
turn_context: &'a TurnContext,
environments: &'a TurnEnvironmentSnapshot,
mcp: &'a McpRuntimeSnapshot,
tool_runtimes: &'a [PlannedRuntime],
tool_suggest_candidates: Option<&'a crate::tools::router::ToolSuggestCandidates>,
extension_tool_executors: &'a [Arc<dyn ToolExecutor<ExtensionToolCall>>],
@@ -151,22 +154,30 @@ struct CoreToolPlanContext<'a> {
#[instrument(level = "trace", skip_all)]
pub(crate) fn build_tool_router(
step_context: &StepContext,
turn_context: &TurnContext,
environments: &TurnEnvironmentSnapshot,
mcp: &McpRuntimeSnapshot,
params: ToolRouterParams<'_>,
tool_search_handler_cache: &ToolSearchHandlerCache,
) -> ToolRouter {
let (model_visible_specs, registry) =
build_tool_specs_and_registry(step_context, params, tool_search_handler_cache);
let (model_visible_specs, registry) = build_tool_specs_and_registry(
turn_context,
environments,
mcp,
params,
tool_search_handler_cache,
);
ToolRouter::from_parts(registry, model_visible_specs)
}
#[instrument(level = "trace", skip_all)]
fn build_tool_specs_and_registry(
step_context: &StepContext,
turn_context: &TurnContext,
environments: &TurnEnvironmentSnapshot,
mcp: &McpRuntimeSnapshot,
params: ToolRouterParams<'_>,
tool_search_handler_cache: &ToolSearchHandlerCache,
) -> (Vec<ToolSpec>, ToolRegistry) {
let turn_context = step_context.turn.as_ref();
let ToolRouterParams {
tool_runtimes,
tool_suggest_candidates,
@@ -176,7 +187,9 @@ fn build_tool_specs_and_registry(
let default_agent_type_description =
crate::agent::role::spawn_tool_spec::build(&std::collections::BTreeMap::new());
let context = CoreToolPlanContext {
step_context,
turn_context,
environments,
mcp,
tool_runtimes: &tool_runtimes,
tool_suggest_candidates: tool_suggest_candidates.as_ref(),
extension_tool_executors: &extension_tool_executors,
@@ -283,7 +296,7 @@ fn spec_for_model_request(
#[instrument(level = "trace", skip_all)]
fn hosted_model_tool_specs(context: &CoreToolPlanContext<'_>) -> Vec<ToolSpec> {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
// Responses Lite accepts schemas for client-executed tools, not hosted Responses tools.
if turn_context.model_info.use_responses_lite {
return Vec::new();
@@ -565,9 +578,9 @@ fn code_mode_namespace_descriptions(
fn add_tool_sources(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
// Guardian reviewers receive only `exec_command`, `write_stdin`, and `view_image`
// when an environment is available; all general tool sources stay excluded.
if crate::guardian::is_guardian_reviewer_source(&context.step_context.turn.session_source) {
let turn_context = context.step_context.turn.as_ref();
let environment_mode = tool_environment_mode(context.step_context);
if crate::guardian::is_guardian_reviewer_source(&context.turn_context.session_source) {
let turn_context = context.turn_context;
let environment_mode = tool_environment_mode(context.environments);
if environment_mode.has_environment() {
let include_environment_id = matches!(environment_mode, ToolEnvironmentMode::Multiple);
planned_tools.add(ExecCommandHandler::new(ExecCommandHandlerOptions {
@@ -576,7 +589,7 @@ fn add_tool_sources(context: &CoreToolPlanContext<'_>, planned_tools: &mut Plann
include_environment_id,
include_shell_parameter: unified_exec_should_include_shell_parameter(
turn_context,
context.step_context,
context.environments,
),
}));
planned_tools.add(WriteStdinHandler);
@@ -614,15 +627,15 @@ fn standalone_web_search_enabled(turn_context: &TurnContext) -> bool {
.enabled(Feature::StandaloneWebSearch))
}
fn tool_environment_mode(step_context: &StepContext) -> ToolEnvironmentMode {
ToolEnvironmentMode::from_count(step_context.environments.turn_environments().count())
fn tool_environment_mode(environments: &TurnEnvironmentSnapshot) -> ToolEnvironmentMode {
ToolEnvironmentMode::from_count(environments.turn_environments().count())
}
#[instrument(level = "trace", skip_all)]
fn add_shell_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
let features = turn_context.config.features.get();
let environment_mode = tool_environment_mode(context.step_context);
let environment_mode = tool_environment_mode(context.environments);
if !environment_mode.has_environment() {
return;
}
@@ -644,7 +657,7 @@ fn add_shell_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut Planne
include_environment_id,
include_shell_parameter: unified_exec_should_include_shell_parameter(
turn_context,
context.step_context,
context.environments,
),
}));
planned_tools.add(WriteStdinHandler);
@@ -664,20 +677,19 @@ fn add_shell_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut Planne
fn unified_exec_should_include_shell_parameter(
turn_context: &TurnContext,
step_context: &StepContext,
environments: &TurnEnvironmentSnapshot,
) -> bool {
!matches!(
&turn_context.unified_exec_shell_mode,
UnifiedExecShellMode::ZshFork(_)
) || step_context
.environments
) || environments
.turn_environments()
.any(|environment| environment.environment.is_remote())
}
#[instrument(level = "trace", skip_all)]
fn add_mcp_resource_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
if context.step_context.mcp.manager().has_servers() {
if context.mcp.manager().has_servers() {
planned_tools.add(ListMcpResourcesHandler);
planned_tools.add(ListMcpResourceTemplatesHandler);
planned_tools.add(ReadMcpResourceHandler);
@@ -686,9 +698,9 @@ fn add_mcp_resource_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut
#[instrument(level = "trace", skip_all)]
fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
let features = turn_context.config.features.get();
let environment_mode = tool_environment_mode(context.step_context);
let environment_mode = tool_environment_mode(context.environments);
planned_tools.add(PlanHandler);
@@ -770,7 +782,7 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut
#[instrument(level = "trace", skip_all)]
fn add_collaboration_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
if collab_tools_enabled(turn_context) {
if multi_agent_v2_enabled(turn_context) {
let exposure = if turn_context.config.multi_agent_v2.non_code_mode_only {
@@ -897,7 +909,7 @@ fn add_extension_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut Pl
// Extension ToolContributor implementations are resolved into executors
// before planning. Core only adapts those executors into its runtime set.
append_extension_tool_executors(
context.step_context.turn.as_ref(),
context.turn_context,
context.extension_tool_executors,
planned_tools,
);
@@ -908,7 +920,7 @@ fn append_tool_search_executor(
context: &CoreToolPlanContext<'_>,
planned_tools: &mut PlannedTools,
) {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
if !search_tool_enabled(turn_context) {
return;
}
@@ -931,7 +943,7 @@ fn prepend_code_mode_executors(
context: &CoreToolPlanContext<'_>,
planned_tools: &mut PlannedTools,
) {
let turn_context = context.step_context.turn.as_ref();
let turn_context = context.turn_context;
let code_mode_executors = build_code_mode_executors(turn_context, planned_tools.runtimes());
planned_tools.runtimes.splice(0..0, code_mode_executors);
}
codex-rs/core/src/tools/spec_plan_tests.rs
+13 -12
View File
@@ -185,7 +185,9 @@ async fn probe_with(
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
let router = ToolRouter::from_context(
step_context.as_ref(),
step_context.turn.as_ref(),
&step_context.environments,
step_context.mcp.as_ref(),
ToolRouterParams {
tool_suggest_candidates: inputs.tool_suggest_candidates,
tool_runtimes: inputs.tool_runtimes,
@@ -679,17 +681,12 @@ async fn environment_tools_follow_the_step_context() {
let environments = turn.environments.clone();
turn.environments.environments.clear();
let turn = Arc::new(turn);
let step_context = Arc::new(StepContext::new(
Arc::clone(&turn),
environments,
Vec::new(),
/*executor_capability_discovery*/ None,
crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config),
/*loaded_agents_md*/ None,
));
let mcp = crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config);
let plan = ToolPlanProbe::from_router(ToolRouter::from_context(
step_context.as_ref(),
turn.as_ref(),
&environments,
mcp.as_ref(),
ToolRouterParams {
tool_runtimes: Vec::new(),
tool_suggest_candidates: None,
@@ -835,7 +832,9 @@ async fn tool_search_cache_rebuilds_when_deferred_sources_change() {
let first_turn = Arc::new(first_turn);
let first_step_context = StepContext::for_test(Arc::clone(&first_turn));
let first_router = ToolRouter::from_context(
first_step_context.as_ref(),
first_step_context.turn.as_ref(),
&first_step_context.environments,
first_step_context.mcp.as_ref(),
ToolRouterParams {
tool_runtimes: vec![mcp_runtime(
"first",
@@ -856,7 +855,9 @@ async fn tool_search_cache_rebuilds_when_deferred_sources_change() {
let second_turn = Arc::new(second_turn);
let second_step_context = StepContext::for_test(Arc::clone(&second_turn));
let second_router = ToolRouter::from_context(
second_step_context.as_ref(),
second_step_context.turn.as_ref(),
&second_step_context.environments,
second_step_context.mcp.as_ref(),
ToolRouterParams {
tool_runtimes: vec![mcp_runtime(
"second",
codex-rs/core/tests/suite/rmcp_client.rs
+105
View File
@@ -33,7 +33,9 @@ use codex_models_manager::manager::RefreshStrategy;
use codex_utils_path_uri::LegacyAppPathString;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::InputModality;
use codex_protocol::openai_models::ModelInfo;
@@ -46,6 +48,7 @@ use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::McpInvocation;
use codex_protocol::protocol::McpToolCallBeginEvent;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::user_input::UserInput;
use codex_utils_cargo_bin::cargo_bin;
use codex_utils_path_uri::PathUri;
@@ -77,6 +80,7 @@ use serde_json::json;
use serial_test::serial;
use std::io::Cursor;
use tempfile::tempdir;
use test_case::test_case;
use tokio::process::Child;
use tokio::process::Command;
use tokio::time::Instant;
@@ -737,6 +741,107 @@ async fn shutdown_cancels_startup_prewarm_waiting_for_mcp_startup() -> anyhow::R
Ok(())
}
#[derive(Clone, Copy)]
enum InterruptedMcpStartupPhase {
PreSamplingCompaction,
FirstStep,
StartupPrewarm,
}
#[test_case(InterruptedMcpStartupPhase::PreSamplingCompaction; "pre sampling compaction")]
#[test_case(InterruptedMcpStartupPhase::FirstStep; "first step")]
#[test_case(InterruptedMcpStartupPhase::StartupPrewarm; "startup prewarm")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn interrupt_during_mcp_startup_preserves_user_input_in_history(
startup_phase: InterruptedMcpStartupPhase,
) -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let pending_mcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
let pending_mcp_url = format!("http://{}/mcp", pending_mcp_listener.local_addr()?);
let fixture = test_codex()
.with_config(move |config| {
config.model_provider.supports_websockets =
matches!(startup_phase, InterruptedMcpStartupPhase::StartupPrewarm);
if matches!(
startup_phase,
InterruptedMcpStartupPhase::PreSamplingCompaction
) {
config.model_auto_compact_token_limit = Some(0);
}
insert_mcp_server(
config,
"interrupted_startup",
McpServerTransportConfig::StreamableHttp {
url: pending_mcp_url,
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
},
TestMcpServerOptions::default(),
);
})
.build_with_auto_env(&server)
.await?;
let (_pending_mcp_connection, _) =
tokio::time::timeout(Duration::from_secs(5), pending_mcp_listener.accept())
.await
.context("MCP startup should connect before the turn is interrupted")??;
let prompt = "keep this interrupted prompt in conversation history";
fixture
.codex
.submit(read_only_user_turn(&fixture, prompt))
.await?;
wait_for_event(&fixture.codex, |event| {
matches!(event, EventMsg::TurnStarted(_))
})
.await;
fixture.codex.submit(Op::Interrupt).await?;
wait_for_event(&fixture.codex, |event| {
matches!(event, EventMsg::TurnAborted(_))
})
.await;
let history = fixture
.codex
.load_history(/*include_archived*/ false)
.await?;
let user_prompt_index = history
.items
.iter()
.position(|item| {
matches!(
item,
RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. })
if role == "user"
&& content.iter().any(|item| {
matches!(item, ContentItem::InputText { text } if text == prompt)
})
)
})
.expect("an interrupted turn should retain its submitted user prompt");
let interruption_marker_index = history
.items
.iter()
.position(|item| {
matches!(
item,
RolloutItem::ResponseItem(ResponseItem::Message { content, .. })
if content.iter().any(|item| {
matches!(item, ContentItem::InputText { text } if text.contains("<turn_aborted>"))
})
)
})
.expect("an interrupted turn should retain its interruption marker");
assert!(user_prompt_index < interruption_marker_index);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial(mcp_cwd)]
async fn stdio_server_uses_configured_cwd_before_runtime_fallback() -> anyhow::Result<()> {