From 4fc522f5e0d3187af8047c4d666d67da3a27ef9b Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Mon, 26 Oct 2026 07:07:34 +0000 Subject: [PATCH] Add optional thread resumption to `codex queue` (#57232) ## What changed Add `codex queue --resume` to resume an unloaded thread after queueing a message through the local daemon or selected `--remote` server. Already loaded threads are left running. If resumption fails, emit a warning while preserving the successful queue result. Read only the thread status when checking whether to resume, allowing remote metadata to contain paths from another operating system. ## Testing Extend CLI integration coverage for unloaded threads consuming queued messages, archived thread rejection, and successful queueing when another server owns the thread and rejects resumption. GitOrigin-RevId: cc8aa3c11a37e990da36e6ff9a17934f73322e7b --- codex-rs/cli/src/queue_cmd.rs | 7 ++ codex-rs/cli/tests/queue.rs | 118 ++++++++++++++++++--- codex-rs/tui/src/session_queue_commands.rs | 57 ++++++++++ 3 files changed, 168 insertions(+), 14 deletions(-) diff --git a/codex-rs/cli/src/queue_cmd.rs b/codex-rs/cli/src/queue_cmd.rs index c8bf9f707d..8a1b997f39 100644 --- a/codex-rs/cli/src/queue_cmd.rs +++ b/codex-rs/cli/src/queue_cmd.rs @@ -19,6 +19,11 @@ pub(crate) struct QueueCommand { #[arg(long, value_name = "TEXT", value_parser = NonEmptyStringValueParser::new())] message: String, + /// After queueing, try to resume the thread on the local daemon or selected --remote server. + /// A failed resume produces a warning and the command still succeeds. + #[arg(long)] + resume: bool, + #[clap(flatten)] remote: InteractiveRemoteOptions, @@ -37,6 +42,7 @@ pub(crate) async fn run_queue_command( let QueueCommand { thread, message, + resume, remote, config_overrides, } = command; @@ -52,6 +58,7 @@ pub(crate) async fn run_queue_command( codex_tui::run_session_queue_command( thread, message, + resume, codex_tui::SessionArchiveCommandOptions { cli, arg0_paths, diff --git a/codex-rs/cli/tests/queue.rs b/codex-rs/cli/tests/queue.rs index 25aafe2e82..7d15124cb1 100644 --- a/codex-rs/cli/tests/queue.rs +++ b/codex-rs/cli/tests/queue.rs @@ -17,6 +17,7 @@ const THREAD_ID: &str = "123e4567-e89b-12d3-a456-426614174000"; enum QueueResponse { Success, + ResumeRejected, MethodNotFound, #[cfg(unix)] UnknownMethodVariant, @@ -72,9 +73,10 @@ where let request = websocket.next().await.context("missing queue request")??; let request: Value = serde_json::from_str(request.to_text()?)?; assert_eq!(request["method"], "thread/queue/add"); + assert_eq!(request["params"]["threadId"], THREAD_ID); let result = match response { - QueueResponse::Success => json!({ + QueueResponse::Success | QueueResponse::ResumeRejected => json!({ "id": request["id"], "result": { "queuedSubmission": { @@ -100,6 +102,27 @@ where websocket .send(Message::Text(result.to_string().into())) .await?; + if matches!(response, QueueResponse::ResumeRejected) { + for (method, mut reply) in [ + ( + "thread/read", + json!({"result": {"thread": {"status": {"type": "notLoaded"}, "cwd": r"C:\workspace"}}}), + ), + ( + "thread/resume", + json!({"error": {"code": -32603, "message": "thread already has an active writer"}}), + ), + ] { + let rpc = websocket.next().await.context("missing resume request")??; + let rpc: Value = serde_json::from_str(rpc.to_text()?)?; + assert_eq!(rpc["method"], method); + assert_eq!(rpc["params"]["threadId"], THREAD_ID); + reply["id"] = rpc["id"].clone(); + websocket + .send(Message::Text(reply.to_string().into())) + .await?; + } + } Ok(request) } @@ -121,6 +144,22 @@ async fn queue_submits_message_to_remote_app_server() -> Result<()> { Ok(()) } +#[tokio::test] +async fn queue_succeeds_when_another_server_owns_the_thread() -> Result<()> { + let (output, _) = run_remote_queue_command(QueueResponse::ResumeRejected).await?; + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!( + String::from_utf8(output.stdout)?, + format!("Queued message queued-submission-id for thread {THREAD_ID}.\n") + ); + assert!(String::from_utf8(output.stderr)?.contains("thread already has an active writer")); + Ok(()) +} + #[tokio::test] async fn queue_does_not_fallback_from_unsupported_explicit_remote() -> Result<()> { let (output, _) = run_remote_queue_command(QueueResponse::MethodNotFound).await?; @@ -165,6 +204,7 @@ async fn run_remote_queue_command(response: QueueResponse) -> Result<(Output, Va let listener = TcpListener::bind("127.0.0.1:0").await?; let endpoint = format!("ws://{}", listener.local_addr()?); let remote_first = matches!(response, QueueResponse::MethodNotFound); + let resume = matches!(response, QueueResponse::ResumeRejected); let server_home = codex_home.path().to_path_buf(); let server = tokio::spawn(async move { let (stream, _) = listener.accept().await?; @@ -192,6 +232,7 @@ async fn run_remote_queue_command(response: QueueResponse) -> Result<(Output, Va .env("CODEX_HOME", codex_home.path()) .env("CODEX_REMOTE_TOKEN", "test-token") .args(remote_args) + .args(resume.then_some("--resume")) .args(["--thread", THREAD_ID, "--message", "do the thing"]) .output() .await?; @@ -206,7 +247,6 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::create_final_assistant_message_sse_response; - use app_test_support::create_mock_responses_server_sequence_unchecked; use codex_app_server_client::AppServerEvent; use codex_app_server_client::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY; use codex_app_server_client::RemoteAppServerClient; @@ -223,18 +263,28 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res use codex_app_server_protocol::ThreadSetNameResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; + use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::TurnStatus; use codex_protocol::shell_environment::OPENAI_FEDERATION_RULE_ID_ENV_VAR; use codex_protocol::shell_environment::OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR; + use core_test_support::responses; use tokio::process::Command; use tokio::time::timeout; + use wiremock::Mock; + use wiremock::matchers::method; + use wiremock::matchers::path; let codex_home = socket_test_home()?; let codex = codex_utils_cargo_bin::cargo_bin("codex")?; - let model = create_mock_responses_server_sequence_unchecked(vec![ - create_final_assistant_message_sse_response("queued message consumed")?, - ]) - .await; + let model = responses::start_mock_server().await; + let model_requests = Mock::given(method("POST")) + .and(path("/v1/responses")) + .respond_with(responses::sse_response( + create_final_assistant_message_sse_response("queued message consumed")?, + )) + .expect(2) + .mount_as_scoped(&model) + .await; MockResponsesConfig::new(&model.uri()) .with_root_config("features.plugins = false\nanalytics.enabled = false") .with_provider_config("env_key = \"CODEX_QUEUE_SERVER_API_KEY\"") @@ -294,7 +344,7 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res ) .env_remove("CODEX_QUEUE_SERVER_API_KEY") .kill_on_drop(true) - .args(["queue", "--remote", "unix://"]) + .args(["queue", "--resume", "--remote", "unix://"]) .args(["--thread", &thread_id, "--message", message]) .output(), ) @@ -329,13 +379,9 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res .await?; assert!(queue.data.is_empty()); - let requests = model - .received_requests() - .await - .context("model request capture unavailable")?; + let requests = model_requests.received_requests().await; let request = requests - .iter() - .find(|request| request.url.path().ends_with("/responses")) + .first() .context("queued message did not reach the model")?; assert!( request.body_json::()?["input"] @@ -384,6 +430,7 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res assert!(!output.status.success()); assert!(String::from_utf8(output.stderr)?.contains("cannot confirm session deletion")); + let cold_message = "Continue after the thread was unloaded"; let cases: &[(&[&str], Option)] = &[ (&["archive", name], Some(true)), (&["unarchive", name], Some(false)), @@ -425,11 +472,54 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res }, }) .await?; + let thread = threads.data.iter().find(|thread| thread.id == thread_id); assert_eq!( - threads.data.iter().any(|thread| thread.id == thread_id), + thread.is_some(), expected_archived == Some(archived), "{args:?}: archived={archived}" ); + if expected_archived == Some(false) + && let Some(thread) = thread + { + assert_eq!(thread.status, ThreadStatus::NotLoaded); + } + } + let Some(archived) = expected_archived else { + continue; + }; + // Archived threads are reachable by ID; active threads can also be targeted by name. + let (target, text) = if archived { + (thread_id.as_str(), "still archived") + } else { + (name, cold_message) + }; + let output = remote_command() + .args(["queue", "--resume", "--thread", target, "--message", text]) + .output() + .await?; + if archived { + assert!(!output.status.success()); + assert!( + String::from_utf8(output.stderr)? + .contains(&format!("session {thread_id} is archived")) + ); + } else { + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + timeout( + Duration::from_secs(30), + model_requests.wait_until_satisfied(), + ) + .await?; + let requests = model_requests.received_requests().await; + assert!( + requests[1].body_json::()?["input"] + .to_string() + .contains(cold_message) + ); } } Ok(()) diff --git a/codex-rs/tui/src/session_queue_commands.rs b/codex-rs/tui/src/session_queue_commands.rs index 9752a42817..21181f7336 100644 --- a/codex-rs/tui/src/session_queue_commands.rs +++ b/codex-rs/tui/src/session_queue_commands.rs @@ -9,6 +9,9 @@ use codex_app_server_client::TypedRequestError; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ThreadQueueAddParams; use codex_app_server_protocol::ThreadQueueAddResponse; +use codex_app_server_protocol::ThreadReadParams; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::UserInput; use codex_app_server_protocol::experimental_required_message; use codex_protocol::ThreadId; @@ -17,6 +20,9 @@ use color_eyre::Report; use color_eyre::eyre::Result; use color_eyre::eyre::WrapErr; use color_eyre::eyre::eyre; +use serde::Deserialize; +use serde::de::IgnoredAny; +use std::io::Write; use std::path::Path; use uuid::Uuid; @@ -27,6 +33,7 @@ const THREAD_QUEUE_ADD_METHOD: &str = "thread/queue/add"; pub async fn run_session_queue_command( target: String, message: String, + resume: bool, options: SessionArchiveCommandOptions, ) -> Result { if options.cli.no_daemon && options.explicit_remote_endpoint.is_none() { @@ -74,6 +81,12 @@ pub async fn run_session_queue_command( } result => result?, }; + if resume && let Err(err) = resume_if_unloaded(&mut app_server, thread_id).await { + let _ = writeln!( + std::io::stderr().lock(), + "Warning: message was queued, but could not resume thread {thread_id}: {err:#}" + ); + } Ok(format!( "Queued message {} for thread {}.", @@ -127,6 +140,50 @@ pub(super) async fn run_session_queue_action_with_app_server( Ok((thread_id, response)) } +async fn resume_if_unloaded(app_server: &mut AppServerSession, thread_id: ThreadId) -> Result<()> { + if app_server.uses_embedded_app_server() { + return Err(eyre!( + "start a local daemon or specify a server with --remote" + )); + } + // Only decode the status: remote thread metadata can contain paths from another OS. + #[derive(Deserialize)] + struct ReadResponse { + thread: Status, + } + #[derive(Deserialize)] + struct Status { + status: ThreadStatus, + } + + let handle = app_server.request_handle(); + let response: ReadResponse = handle + .request_typed(ClientRequest::ThreadRead { + request_id: app_server.next_request_id(), + params: ThreadReadParams { + thread_id: thread_id.to_string(), + include_turns: false, + }, + }) + .await + .wrap_err_with(|| format!("failed to read thread {thread_id}"))?; + if response.thread.status != ThreadStatus::NotLoaded { + // A fresh loaded thread can accept input before it has any history to resume. + return Ok(()); + } + handle + .request_typed::(ClientRequest::ThreadResume { + request_id: app_server.next_request_id(), + params: ThreadResumeParams { + thread_id: thread_id.to_string(), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + Ok(()) +} + fn is_unsupported_queue_error(error: &Report) -> bool { matches!( error.downcast_ref::(),