Accept user input when starting idle turns (#36947)

## What changed

- Generalize `CodexThread::try_start_turn_if_idle` to accept exported
  `TurnInput` values, preserving user content and client IDs in the first model
  request.
- Allow non-empty user input to start an idle turn in Plan mode while retaining
  the Plan-mode rejection for automatic response items and empty user input.
- Process idle user input through the normal user-turn path so connector
  selections are cleared, prompt telemetry is recorded, and referenced skill
  instructions are included.

## Testing

- Cover idle user input in Default and Plan modes, response-item mailbox
  merging, empty-input rejection, and skill instruction injection.

GitOrigin-RevId: 116667d160d4464b525d9e4c7e8cc05d23a94008
This commit is contained in:
efrazer-oai authored and copyberry committed 2026-08-04 18:52:41 +00:00
1 parent ceaa818898
commit 989a0b053e
9 files changed
+309 -34

No files matched your search

+14 -13
View File
@@ -4,6 +4,7 @@ use crate::elicitation::ElicitationRegistration;
use crate::session::SessionIo; use crate::session::SessionIo;
use crate::session::SessionSettingsUpdate; use crate::session::SessionSettingsUpdate;
use crate::session::SteerInputError; use crate::session::SteerInputError;
use crate::session::TurnInput;
use crate::session::session::Session; use crate::session::session::Session;
use crate::user_message_admission::UserMessageAdmission; use crate::user_message_admission::UserMessageAdmission;
use codex_exec_server::SelectedCapabilityRootsStatus; use codex_exec_server::SelectedCapabilityRootsStatus;
@@ -93,8 +94,8 @@ pub enum TryStartTurnIfIdleRejectionReason {
/// User/client-triggered mailbox work is already queued and must take /// User/client-triggered mailbox work is already queued and must take
/// priority over extension-initiated idle work. /// priority over extension-initiated idle work.
PendingTriggerTurn, PendingTriggerTurn,
/// The thread is in Plan mode, where automatic idle work must not start a /// The thread is in Plan mode, where idle work without user input must not
/// new model turn. /// start a new model turn.
PlanMode, PlanMode,
/// Another turn or task is active, or the idle reservation was lost before /// Another turn or task is active, or the idle reservation was lost before
/// the automatic turn could start. /// the automatic turn could start.
@@ -106,11 +107,11 @@ pub enum TryStartTurnIfIdleRejectionReason {
#[derive(Debug)] #[derive(Debug)]
pub struct TryStartTurnIfIdleError { pub struct TryStartTurnIfIdleError {
reason: TryStartTurnIfIdleRejectionReason, reason: TryStartTurnIfIdleRejectionReason,
input: Vec<ResponseItem>, input: Vec<TurnInput>,
} }
impl TryStartTurnIfIdleError { impl TryStartTurnIfIdleError {
pub(crate) fn new(reason: TryStartTurnIfIdleRejectionReason, input: Vec<ResponseItem>) -> Self { pub(crate) fn new(reason: TryStartTurnIfIdleRejectionReason, input: Vec<TurnInput>) -> Self {
Self { reason, input } Self { reason, input }
} }
@@ -119,9 +120,9 @@ impl TryStartTurnIfIdleError {
self.reason self.reason
} }
/// Consumes the rejection and returns the original model-visible input /// Consumes the rejection and returns the original turn input
/// unchanged, so callers can retry, drop, or log it explicitly. /// unchanged, so callers can retry, drop, or log it explicitly.
pub fn into_input(self) -> Vec<ResponseItem> { pub fn into_input(self) -> Vec<TurnInput> {
self.input self.input
} }
} }
@@ -371,22 +372,22 @@ impl CodexThread {
self.session.inject_if_running(items).await self.session.inject_if_running(items).await
} }
/// Starts an automatic regular turn with model-visible items only when idle /// Starts an automatic regular turn with response items or user input only
/// work is allowed for this thread. /// when idle work is allowed for this thread.
/// ///
/// This is the required entry point for extensions that want to launch /// This is the required entry point for extensions that want to launch
/// model-visible work from `ThreadLifecycleContributor::on_thread_idle`. /// model-visible work from `ThreadLifecycleContributor::on_thread_idle`.
/// The call succeeds only if no user/client-triggered turn is queued, no /// The call succeeds only if no user/client-triggered turn is queued and no
/// task is currently active, and the thread is not in Plan mode. Active /// task is currently active. Work without user input is also rejected in
/// Review tasks are rejected by the active-task check because Review turns /// Plan mode. Active Review tasks are rejected by the active-task check
/// are not steerable. /// because Review turns are not steerable.
/// ///
/// On rejection, the returned error includes a stable reason and carries /// On rejection, the returned error includes a stable reason and carries
/// the original `items` unchanged so the caller can decide whether to drop /// the original `items` unchanged so the caller can decide whether to drop
/// them, retry later, or log why no automatic turn was started. /// them, retry later, or log why no automatic turn was started.
pub async fn try_start_turn_if_idle( pub async fn try_start_turn_if_idle(
&self, &self,
items: Vec<ResponseItem>, items: Vec<TurnInput>,
) -> Result<(), TryStartTurnIfIdleError> { ) -> Result<(), TryStartTurnIfIdleError> {
self.session.try_start_turn_if_idle(items).await self.session.try_start_turn_if_idle(items).await
} }
+1
View File
@@ -17,6 +17,7 @@ mod responses_retry;
pub(crate) mod session; pub(crate) mod session;
pub use responses_metadata::CodexResponsesMetadata; pub use responses_metadata::CodexResponsesMetadata;
pub use session::SteerInputError; pub use session::SteerInputError;
pub use session::TurnInput;
pub use turn_metadata::detached_memory_responses_metadata; pub use turn_metadata::detached_memory_responses_metadata;
mod codex_thread; mod codex_thread;
mod compact_model_fallback; mod compact_model_fallback;
+26 -14
View File
@@ -36,27 +36,31 @@ impl Session {
} }
} }
/// Starts a regular turn with the provided items only if automatic idle work /// Starts a regular turn with the provided input only if automatic idle work
/// is allowed for the current session state. /// is allowed for the current session state.
/// ///
/// This is the shared gate for extension-initiated idle work. It refuses to /// This is the shared gate for extension-initiated idle work. It refuses to
/// start a turn when user/client-triggered work is queued, any task is still /// start a turn when user/client-triggered work is queued or any task is
/// active, or the session is currently in Plan mode. Active Review tasks are /// still active. Work without user input is also rejected in Plan mode.
/// covered by the active-task check because Review turns are not steerable. /// Active Review tasks are covered by the active-task check because Review
/// turns are not steerable.
pub(crate) async fn try_start_turn_if_idle( pub(crate) async fn try_start_turn_if_idle(
self: &Arc<Self>, self: &Arc<Self>,
input: Vec<ResponseItem>, input: Vec<TurnInput>,
) -> Result<(), TryStartTurnIfIdleError> { ) -> Result<(), TryStartTurnIfIdleError> {
if input.is_empty() { if input.is_empty() {
return Ok(()); return Ok(());
} }
let has_user_input = input.iter().any(
|item| matches!(item, TurnInput::UserInput { content, .. } if !content.is_empty()),
);
if self.input_queue.has_trigger_turn_mailbox_items().await { if self.input_queue.has_trigger_turn_mailbox_items().await {
return Err(TryStartTurnIfIdleError::new( return Err(TryStartTurnIfIdleError::new(
TryStartTurnIfIdleRejectionReason::PendingTriggerTurn, TryStartTurnIfIdleRejectionReason::PendingTriggerTurn,
input, input,
)); ));
} }
if self.collaboration_mode().await.mode == ModeKind::Plan { if !has_user_input && self.collaboration_mode().await.mode == ModeKind::Plan {
return Err(TryStartTurnIfIdleError::new( return Err(TryStartTurnIfIdleError::new(
TryStartTurnIfIdleRejectionReason::PlanMode, TryStartTurnIfIdleRejectionReason::PlanMode,
input, input,
@@ -87,7 +91,7 @@ impl Session {
let turn_context = self let turn_context = self
.new_default_turn_with_sub_id(uuid::Uuid::new_v4().to_string()) .new_default_turn_with_sub_id(uuid::Uuid::new_v4().to_string())
.await; .await;
if turn_context.mode == ModeKind::Plan { if !has_user_input && turn_context.mode == ModeKind::Plan {
self.clear_reserved_idle_turn(&turn_state).await; self.clear_reserved_idle_turn(&turn_state).await;
self.maybe_start_turn_for_pending_work().await; self.maybe_start_turn_for_pending_work().await;
return Err(TryStartTurnIfIdleError::new( return Err(TryStartTurnIfIdleError::new(
@@ -119,15 +123,23 @@ impl Session {
)); ));
} }
self.input_queue let task_input = if has_user_input {
.extend_pending_input_for_turn_state( self.clear_connector_selection().await;
turn_state.as_ref(), for item in &input {
input.into_iter().map(TurnInput::ResponseItem).collect(), if let TurnInput::UserInput { content, .. } = item {
) turn_context.session_telemetry.user_prompt(content);
.await; }
}
input
} else {
self.input_queue
.extend_pending_input_for_turn_state(turn_state.as_ref(), input)
.await;
Vec::new()
};
self.start_task( self.start_task(
turn_context, turn_context,
Vec::new(), task_input,
RegularTask::new(), RegularTask::new(),
MailboxParentProvenance::Ignore, MailboxParentProvenance::Ignore,
) )
+2 -1
View File
@@ -9,8 +9,9 @@ use std::sync::Arc;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tokio::sync::watch; use tokio::sync::watch;
/// Input consumed by a regular turn.
#[derive(Clone, Debug, PartialEq)] #[derive(Clone, Debug, PartialEq)]
pub(crate) enum TurnInput { pub enum TurnInput {
UserInput { UserInput {
content: Vec<UserInput>, content: Vec<UserInput>,
client_id: Option<String>, client_id: Option<String>,
+1 -1
View File
@@ -228,7 +228,7 @@ use self::config_lock::validate_config_lock_if_configured;
use self::handlers::submission_dispatch_span; use self::handlers::submission_dispatch_span;
use self::handlers::submission_loop; use self::handlers::submission_loop;
pub(crate) use self::input_queue::InputQueueActivity; pub(crate) use self::input_queue::InputQueueActivity;
pub(crate) use self::input_queue::TurnInput; pub use self::input_queue::TurnInput;
pub(crate) use self::input_queue::TurnInputQueue; pub(crate) use self::input_queue::TurnInputQueue;
use self::review::spawn_review_thread; use self::review::spawn_review_thread;
use self::session::AppServerClientMetadata; use self::session::AppServerClientMetadata;
+57 -4
View File
@@ -10317,7 +10317,7 @@ async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() {
) )
.await; .await;
let item = user_message("synthetic idle input"); let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
let err = sess let err = sess
.try_start_turn_if_idle(vec![item.clone()]) .try_start_turn_if_idle(vec![item.clone()])
.await .await
@@ -10343,7 +10343,7 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() {
state.session_configuration.collaboration_mode = collaboration_mode; state.session_configuration.collaboration_mode = collaboration_mode;
} }
let item = user_message("synthetic idle input"); let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
let err = sess let err = sess
.try_start_turn_if_idle(vec![item.clone()]) .try_start_turn_if_idle(vec![item.clone()])
.await .await
@@ -10358,6 +10358,59 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() {
); );
} }
#[tokio::test]
async fn try_start_turn_if_idle_accepts_user_input_in_plan_mode() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
let mut collaboration_mode = sess.collaboration_mode().await;
collaboration_mode.mode = ModeKind::Plan;
{
let mut state = sess.state.lock().await;
state.session_configuration.collaboration_mode = collaboration_mode;
state.merge_connector_selection(["calendar".to_string()]);
}
sess.try_start_turn_if_idle(vec![TurnInput::UserInput {
content: vec![UserInput::Text {
text: "queued user input".to_string(),
text_elements: Vec::new(),
}],
client_id: Some("queued-user-message".to_string()),
}])
.await
.expect("plan mode should accept user-authored idle input");
assert!(sess.state.lock().await.get_connector_selection().is_empty());
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
}
#[tokio::test]
async fn try_start_turn_if_idle_rejects_empty_user_input_in_plan_mode() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
let mut collaboration_mode = sess.collaboration_mode().await;
collaboration_mode.mode = ModeKind::Plan;
{
let mut state = sess.state.lock().await;
state.session_configuration.collaboration_mode = collaboration_mode;
}
let input = vec![
TurnInput::UserInput {
content: Vec::new(),
client_id: Some("empty-queued-user-message".to_string()),
},
TurnInput::ResponseItem(user_message("automatic idle input")),
];
let error = sess
.try_start_turn_if_idle(input.clone())
.await
.expect_err("empty user input should not bypass plan mode");
assert_eq!(TryStartTurnIfIdleRejectionReason::PlanMode, error.reason());
assert_eq!(input, error.into_input());
assert!(sess.active_turn.lock().await.is_none());
}
#[tokio::test] #[tokio::test]
async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() { async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await; let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
@@ -10374,7 +10427,7 @@ async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting()
) )
.await; .await;
let item = user_message("synthetic idle input"); let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
let err = sess let err = sess
.try_start_turn_if_idle(vec![item.clone()]) .try_start_turn_if_idle(vec![item.clone()])
.await .await
@@ -10402,7 +10455,7 @@ async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() {
) )
.await; .await;
let item = user_message("synthetic idle input"); let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
let err = sess let err = sess
.try_start_turn_if_idle(vec![item.clone()]) .try_start_turn_if_idle(vec![item.clone()])
.await .await
+137
View File
@@ -2,11 +2,15 @@ use core_test_support::test_codex::local_selections;
use std::sync::Arc; use std::sync::Arc;
use codex_core::CodexThread; use codex_core::CodexThread;
use codex_core::TurnInput;
use codex_core::config::CurrentTimeReminderConfig; use codex_core::config::CurrentTimeReminderConfig;
use codex_extension_items::ExtensionItem; use codex_extension_items::ExtensionItem;
use codex_extension_items::sleep::SleepItem; use codex_extension_items::sleep::SleepItem;
use codex_features::Feature; use codex_features::Feature;
use codex_protocol::AgentPath; use codex_protocol::AgentPath;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::items::TurnItem; use codex_protocol::items::TurnItem;
use codex_protocol::models::PermissionProfile; use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::AskForApproval;
@@ -15,6 +19,7 @@ use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op; use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::user_input::UserInput; use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot; use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::context_snapshot::ContextSnapshotOptions;
@@ -41,6 +46,138 @@ use serde_json::from_slice;
use serde_json::json; use serde_json::json;
use tokio::sync::oneshot; use tokio::sync::oneshot;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_user_input_reaches_the_first_model_request() -> anyhow::Result<()> {
assert_idle_user_input_reaches_the_first_model_request(ModeKind::Default).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_user_input_reaches_the_first_model_request_in_plan_mode() -> anyhow::Result<()> {
assert_idle_user_input_reaches_the_first_model_request(ModeKind::Plan).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyhow::Result<()> {
let server = responses::start_mock_server().await;
let response = responses::mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("idle-response-items"),
ev_completed("idle-response-items"),
]),
)
.await;
let test = test_codex().build_with_auto_env(&server).await?;
submit_queue_only_agent_mail(test.codex.as_ref(), "pending mailbox input").await;
test.codex
.try_start_turn_if_idle(vec![TurnInput::ResponseItem(responses::user_message_item(
"automatic response item",
))])
.await
.map_err(|error| {
anyhow::anyhow!("idle response items were rejected: {:?}", error.reason())
})?;
wait_for_turn_complete(test.codex.as_ref()).await;
let request = response.single_request();
let user_messages = request.message_input_texts("user");
assert!(
user_messages
.iter()
.any(|message| message == "automatic response item")
);
assert!(
request
.inputs_of_type("agent_message")
.iter()
.any(|message| {
message["author"] == "/root/worker"
&& message["recipient"] == "/root"
&& message["content"].as_array().is_some_and(|content| {
content.iter().any(|item| {
item["type"] == "input_text" && item["text"] == "pending mailbox input"
})
})
})
);
Ok(())
}
async fn assert_idle_user_input_reaches_the_first_model_request(
mode: ModeKind,
) -> anyhow::Result<()> {
let server = responses::start_mock_server().await;
let response = responses::mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("idle-user-input"),
ev_completed("idle-user-input"),
]),
)
.await;
let test = test_codex().build_with_auto_env(&server).await?;
if mode == ModeKind::Plan {
core_test_support::submit_thread_settings(
test.codex.as_ref(),
ThreadSettingsOverrides {
collaboration_mode: Some(CollaborationMode {
mode,
settings: Settings {
model: test.session_configured.model.clone(),
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
},
)
.await?;
}
let expected_input = vec![UserInput::Text {
text: "queued user input reaches the first request".to_string(),
text_elements: Vec::new(),
}];
test.codex
.try_start_turn_if_idle(vec![TurnInput::UserInput {
content: expected_input.clone(),
client_id: Some("queued-user-message".to_string()),
}])
.await
.map_err(|error| anyhow::anyhow!("idle user input was rejected: {:?}", error.reason()))?;
let user_message = core_test_support::wait_for_event_match(test.codex.as_ref(), |event| {
let EventMsg::ItemCompleted(event) = event else {
return None;
};
let TurnItem::UserMessage(item) = &event.item else {
return None;
};
Some(item.clone())
})
.await;
assert_eq!(
Some("queued-user-message".to_string()),
user_message.client_id
);
assert_eq!(expected_input, user_message.content);
wait_for_turn_complete(test.codex.as_ref()).await;
let request = response.single_request();
assert!(
request
.message_input_texts("user")
.iter()
.any(|text| text == "queued user input reaches the first request"),
"the first Responses request should contain the queued user message"
);
Ok(())
}
fn ev_message_item_done(id: &str, text: &str) -> Value { fn ev_message_item_done(id: &str, text: &str) -> Value {
serde_json::json!({ serde_json::json!({
"type": "response.output_item.done", "type": "response.output_item.done",
+66
View File
@@ -2,6 +2,7 @@
#![allow(clippy::unwrap_used)] #![allow(clippy::unwrap_used)]
use anyhow::Result; use anyhow::Result;
use codex_core::TurnInput;
use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::CreateDirectoryOptions;
use codex_exec_server::ExecutorFileSystem; use codex_exec_server::ExecutorFileSystem;
use codex_protocol::models::PermissionProfile; use codex_protocol::models::PermissionProfile;
@@ -133,3 +134,68 @@ async fn user_turn_includes_skill_instructions() -> Result<()> {
Ok(()) Ok(())
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_user_turn_includes_skill_instructions_in_the_first_request() -> Result<()> {
skip_if_target_windows!(Ok(()), "requires native cross-OS skill paths");
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let skill_body = "queued skill body";
let mut builder = test_codex().with_workspace_setup(move |cwd, fs| async move {
write_repo_skill(cwd, fs, "queued-demo", "queued demo skill", skill_body).await
});
let test = builder.build_with_auto_env(&server).await?;
let skill_path = test
.config
.cwd
.join(".agents/skills/queued-demo/SKILL.md")
.canonicalize()
.unwrap_or_else(|_| test.config.cwd.join(".agents/skills/queued-demo/SKILL.md"))
.to_path_buf();
let mock = mount_sse_once(
&server,
sse(vec![
ev_response_created("queued-skill-response"),
ev_assistant_message("queued-skill-message", "done"),
ev_completed("queued-skill-response"),
]),
)
.await;
test.codex
.try_start_turn_if_idle(vec![TurnInput::UserInput {
content: vec![
UserInput::Text {
text: "please use $queued-demo".to_string(),
text_elements: Vec::new(),
},
UserInput::Skill {
name: "queued-demo".to_string(),
path: skill_path.clone(),
},
],
client_id: Some("queued-skill-user-message".to_string()),
}])
.await
.map_err(|error| anyhow::anyhow!("idle skill input was rejected: {:?}", error.reason()))?;
core_test_support::wait_for_event(test.codex.as_ref(), |event| {
matches!(event, codex_protocol::protocol::EventMsg::TurnComplete(_))
})
.await;
let user_texts = mock.single_request().message_input_texts("user");
let skill_path_str = skill_path.to_string_lossy();
assert!(
user_texts.iter().any(|text| {
text.contains("<skill>\n<name>queued-demo</name>")
&& text.contains("<path>")
&& text.contains(skill_body)
&& text.contains(skill_path_str.as_ref())
}),
"expected queued skill instructions in the first request, got {user_texts:?}"
);
Ok(())
}
+5 -1
View File
@@ -4,6 +4,7 @@ use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use codex_core::ThreadManager; use codex_core::ThreadManager;
use codex_core::TurnInput;
use codex_protocol::ThreadId; use codex_protocol::ThreadId;
use codex_protocol::models::ResponseItem; use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::ThreadGoal; use codex_protocol::protocol::ThreadGoal;
@@ -402,7 +403,10 @@ impl GoalRuntimeHandle {
} }
let item = continuation_steering_item(&protocol_goal_from_state(goal)); let item = continuation_steering_item(&protocol_goal_from_state(goal));
if let Err(err) = thread.try_start_turn_if_idle(vec![item]).await { if let Err(err) = thread
.try_start_turn_if_idle(vec![TurnInput::ResponseItem(item)])
.await
{
let reason = err.reason(); let reason = err.reason();
tracing::debug!( tracing::debug!(
?reason, ?reason,