diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 758b592047..07e39c35dc 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -6019,6 +6019,7 @@ pub(crate) async fn run_turn( // one instance across retries within this turn. let mut client_session = prewarmed_client_session.unwrap_or_else(|| sess.services.model_client.new_session()); + let mut can_record_pending_input = input.is_empty(); loop { if run_pending_session_start_hooks(&sess, &turn_context).await { @@ -6028,7 +6029,11 @@ pub(crate) async fn run_turn( // Note that pending_input would be something like a message the user // submitted through the UI while the model was running. Though the UI // may support this, the model might not. - let pending_input = sess.get_pending_input().await; + let pending_input = if can_record_pending_input { + sess.get_pending_input().await + } else { + Vec::new() + }; let mut blocked_pending_input = false; let mut blocked_pending_input_contexts = Vec::new(); @@ -6105,6 +6110,7 @@ pub(crate) async fn run_turn( needs_follow_up, last_agent_message: sampling_request_last_agent_message, } = sampling_request_output; + can_record_pending_input = true; let total_usage_tokens = sess.get_total_token_usage().await; let token_limit_reached = total_usage_tokens >= auto_compact_limit; @@ -6134,6 +6140,7 @@ pub(crate) async fn run_turn( return None; } client_session.reset_websocket_session(); + can_record_pending_input = false; continue; } diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index 2fc5bf4e34..2ba862efd4 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -11,6 +11,7 @@ use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::responses; use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_completed_with_tokens; use core_test_support::responses::ev_function_call; use core_test_support::responses::ev_message_item_added; use core_test_support::responses::ev_output_text_delta; @@ -426,3 +427,98 @@ async fn user_input_does_not_preempt_after_reasoning_item() { server.shutdown().await; } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steered_user_input_waits_for_model_continuation_after_mid_turn_compact() { + let first_chunks = vec![ + chunk(ev_response_created("resp-1")), + chunk(ev_function_call("call-1", "test_tool", "{}")), + chunk(ev_completed_with_tokens( + "resp-1", /*total_tokens*/ 500, + )), + ]; + + let compact_chunks = vec![ + chunk(ev_response_created("resp-compact")), + chunk(ev_message_item_done("msg-compact", "AUTO_COMPACT_SUMMARY")), + chunk(ev_completed_with_tokens( + "resp-compact", + /*total_tokens*/ 50, + )), + ]; + + let post_compact_continuation_chunks = vec![ + chunk(ev_response_created("resp-post-compact")), + chunk(ev_message_item_added("msg-post-compact", "")), + chunk(ev_output_text_delta("resumed old task")), + chunk(ev_message_item_done("msg-post-compact", "resumed old task")), + chunk(ev_completed_with_tokens( + "resp-post-compact", + /*total_tokens*/ 60, + )), + ]; + + let steered_follow_up_chunks = vec![ + chunk(ev_response_created("resp-steered")), + chunk(ev_message_item_done( + "msg-steered", + "processed steered prompt", + )), + chunk(ev_completed_with_tokens( + "resp-steered", + /*total_tokens*/ 70, + )), + ]; + + let (server, _completions) = start_streaming_sse_server(vec![ + first_chunks, + compact_chunks, + post_compact_continuation_chunks, + steered_follow_up_chunks, + ]) + .await; + + let codex = test_codex() + .with_model("gpt-5.1") + .with_config(|config| { + config.model_provider.name = "OpenAI (test)".to_string(); + config.model_provider.supports_websockets = false; + config.model_auto_compact_token_limit = Some(200); + }) + .build_with_streaming_server(&server) + .await + .unwrap_or_else(|err| panic!("build streaming Codex test session: {err}")) + .codex; + + submit_user_input(&codex, "first prompt").await; + submit_user_input(&codex, "second prompt").await; + + wait_for_agent_message(&codex, "resumed old task").await; + wait_for_turn_complete(&codex).await; + + let requests = server.requests().await; + assert_eq!(requests.len(), 4); + + let post_compact_body: Value = + from_slice(&requests[2]).unwrap_or_else(|err| panic!("parse post-compact request: {err}")); + let steered_body: Value = + from_slice(&requests[3]).unwrap_or_else(|err| panic!("parse steered request: {err}")); + + let post_compact_user_texts = message_input_texts(&post_compact_body, "user"); + assert!( + !post_compact_user_texts + .iter() + .any(|text| text == "second prompt"), + "steered input should stay pending until the model resumes after compaction" + ); + + let steered_user_texts = message_input_texts(&steered_body, "user"); + assert!( + steered_user_texts + .iter() + .any(|text| text == "second prompt"), + "steered input should be recorded on the request after the post-compact continuation" + ); + + server.shutdown().await; +}