diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 0fb65079a6..3e5c3ad948 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -4,6 +4,7 @@ use crate::elicitation::ElicitationRegistration; use crate::session::SessionIo; use crate::session::SessionSettingsUpdate; use crate::session::SteerInputError; +use crate::session::TurnInput; use crate::session::session::Session; use crate::user_message_admission::UserMessageAdmission; use codex_exec_server::SelectedCapabilityRootsStatus; @@ -93,8 +94,8 @@ pub enum TryStartTurnIfIdleRejectionReason { /// User/client-triggered mailbox work is already queued and must take /// priority over extension-initiated idle work. PendingTriggerTurn, - /// The thread is in Plan mode, where automatic idle work must not start a - /// new model turn. + /// The thread is in Plan mode, where idle work without user input must not + /// start a new model turn. PlanMode, /// Another turn or task is active, or the idle reservation was lost before /// the automatic turn could start. @@ -106,11 +107,11 @@ pub enum TryStartTurnIfIdleRejectionReason { #[derive(Debug)] pub struct TryStartTurnIfIdleError { reason: TryStartTurnIfIdleRejectionReason, - input: Vec, + input: Vec, } impl TryStartTurnIfIdleError { - pub(crate) fn new(reason: TryStartTurnIfIdleRejectionReason, input: Vec) -> Self { + pub(crate) fn new(reason: TryStartTurnIfIdleRejectionReason, input: Vec) -> Self { Self { reason, input } } @@ -119,9 +120,9 @@ impl TryStartTurnIfIdleError { 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. - pub fn into_input(self) -> Vec { + pub fn into_input(self) -> Vec { self.input } } @@ -371,22 +372,22 @@ impl CodexThread { self.session.inject_if_running(items).await } - /// Starts an automatic regular turn with model-visible items only when idle - /// work is allowed for this thread. + /// Starts an automatic regular turn with response items or user input only + /// when idle work is allowed for this thread. /// /// This is the required entry point for extensions that want to launch /// model-visible work from `ThreadLifecycleContributor::on_thread_idle`. - /// The call succeeds only if no user/client-triggered turn is queued, no - /// task is currently active, and the thread is not in Plan mode. Active - /// Review tasks are rejected by the active-task check because Review turns - /// are not steerable. + /// The call succeeds only if no user/client-triggered turn is queued and no + /// task is currently active. Work without user input is also rejected in + /// Plan mode. Active Review tasks are rejected by the active-task check + /// because Review turns are not steerable. /// /// On rejection, the returned error includes a stable reason and carries /// the original `items` unchanged so the caller can decide whether to drop /// them, retry later, or log why no automatic turn was started. pub async fn try_start_turn_if_idle( &self, - items: Vec, + items: Vec, ) -> Result<(), TryStartTurnIfIdleError> { self.session.try_start_turn_if_idle(items).await } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 612879cedc..f1ba9c705e 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -17,6 +17,7 @@ mod responses_retry; pub(crate) mod session; pub use responses_metadata::CodexResponsesMetadata; pub use session::SteerInputError; +pub use session::TurnInput; pub use turn_metadata::detached_memory_responses_metadata; mod codex_thread; mod compact_model_fallback; diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index e772513d00..9b73b05370 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -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. /// /// 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 - /// active, or the session is currently in Plan mode. Active Review tasks are - /// covered by the active-task check because Review turns are not steerable. + /// start a turn when user/client-triggered work is queued or any task is + /// still active. Work without user input is also rejected in Plan mode. + /// 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( self: &Arc, - input: Vec, + input: Vec, ) -> Result<(), TryStartTurnIfIdleError> { if input.is_empty() { 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 { return Err(TryStartTurnIfIdleError::new( TryStartTurnIfIdleRejectionReason::PendingTriggerTurn, input, )); } - if self.collaboration_mode().await.mode == ModeKind::Plan { + if !has_user_input && self.collaboration_mode().await.mode == ModeKind::Plan { return Err(TryStartTurnIfIdleError::new( TryStartTurnIfIdleRejectionReason::PlanMode, input, @@ -87,7 +91,7 @@ impl Session { let turn_context = self .new_default_turn_with_sub_id(uuid::Uuid::new_v4().to_string()) .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.maybe_start_turn_for_pending_work().await; return Err(TryStartTurnIfIdleError::new( @@ -119,15 +123,23 @@ impl Session { )); } - self.input_queue - .extend_pending_input_for_turn_state( - turn_state.as_ref(), - input.into_iter().map(TurnInput::ResponseItem).collect(), - ) - .await; + let task_input = if has_user_input { + self.clear_connector_selection().await; + for item in &input { + if let TurnInput::UserInput { content, .. } = item { + turn_context.session_telemetry.user_prompt(content); + } + } + input + } else { + self.input_queue + .extend_pending_input_for_turn_state(turn_state.as_ref(), input) + .await; + Vec::new() + }; self.start_task( turn_context, - Vec::new(), + task_input, RegularTask::new(), MailboxParentProvenance::Ignore, ) diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 34f6f778d7..d70dbe6306 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -9,8 +9,9 @@ use std::sync::Arc; use tokio::sync::Mutex; use tokio::sync::watch; +/// Input consumed by a regular turn. #[derive(Clone, Debug, PartialEq)] -pub(crate) enum TurnInput { +pub enum TurnInput { UserInput { content: Vec, client_id: Option, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index ad31042d3a..db73bdbfa4 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -228,7 +228,7 @@ use self::config_lock::validate_config_lock_if_configured; use self::handlers::submission_dispatch_span; use self::handlers::submission_loop; 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; use self::review::spawn_review_thread; use self::session::AppServerClientMetadata; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 38e2d7daaf..7c6f2fce4c 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -10317,7 +10317,7 @@ async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() { ) .await; - let item = user_message("synthetic idle input"); + let item = TurnInput::ResponseItem(user_message("synthetic idle input")); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await @@ -10343,7 +10343,7 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() { 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 .try_start_turn_if_idle(vec![item.clone()]) .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] async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() { 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; - let item = user_message("synthetic idle input"); + let item = TurnInput::ResponseItem(user_message("synthetic idle input")); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await @@ -10402,7 +10455,7 @@ async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() { ) .await; - let item = user_message("synthetic idle input"); + let item = TurnInput::ResponseItem(user_message("synthetic idle input")); let err = sess .try_start_turn_if_idle(vec![item.clone()]) .await diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index a5a3df2022..136996723e 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -2,11 +2,15 @@ use core_test_support::test_codex::local_selections; use std::sync::Arc; use codex_core::CodexThread; +use codex_core::TurnInput; use codex_core::config::CurrentTimeReminderConfig; use codex_extension_items::ExtensionItem; use codex_extension_items::sleep::SleepItem; use codex_features::Feature; 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::models::PermissionProfile; use codex_protocol::protocol::AskForApproval; @@ -15,6 +19,7 @@ use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::Op; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; @@ -41,6 +46,138 @@ use serde_json::from_slice; use serde_json::json; 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 { serde_json::json!({ "type": "response.output_item.done", diff --git a/codex-rs/core/tests/suite/skills.rs b/codex-rs/core/tests/suite/skills.rs index a6a4716879..84407b5b86 100644 --- a/codex-rs/core/tests/suite/skills.rs +++ b/codex-rs/core/tests/suite/skills.rs @@ -2,6 +2,7 @@ #![allow(clippy::unwrap_used)] use anyhow::Result; +use codex_core::TurnInput; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::ExecutorFileSystem; use codex_protocol::models::PermissionProfile; @@ -133,3 +134,68 @@ async fn user_turn_includes_skill_instructions() -> Result<()> { 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("\nqueued-demo") + && text.contains("") + && text.contains(skill_body) + && text.contains(skill_path_str.as_ref()) + }), + "expected queued skill instructions in the first request, got {user_texts:?}" + ); + + Ok(()) +} diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index 31b8b2eb8e..324e7a23d2 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/codex-rs/ext/goal/src/runtime.rs @@ -4,6 +4,7 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use codex_core::ThreadManager; +use codex_core::TurnInput; use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::ThreadGoal; @@ -402,7 +403,10 @@ impl GoalRuntimeHandle { } 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(); tracing::debug!( ?reason,