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

Add a command to queue messages for existing sessions (#39092)

Browse Source 
## What changed

- Add `codex queue --thread <THREAD> --message <TEXT>` to submit a text message through the `thread/queue/add` app-server API.
- Resolve active sessions by UUID or exact name across interactive, exec, and custom sources, and reject ambiguous names.
- Support local and explicit remote app servers while reporting incompatible servers and configuration overrides instead of silently changing the target.
- Reject empty messages and image attachments.

## Testing

- Add CLI coverage for remote submission, validation, unsupported servers, and local-daemon routing.
- Add session lookup and queue tests for stale, non-interactive, custom, and duplicate session names.

GitOrigin-RevId: 44c233f752b63a85c85b92fa6da303336e4ddeca
This commit is contained in:
Eric Traut
2026-08-17 22:44:37 +00:00
committed by copyberry
parent 31f23b6022
commit 83d015375e
9 changed files with 772 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/cli/src/main.rs
+20 -1
View File
@@ -55,6 +55,7 @@ mod marketplace_cmd;
mod mcp_cmd;
mod migrate_rollouts;
mod plugin_cmd;
mod queue_cmd;
mod remote_control_cmd;
#[cfg(target_os = "windows")]
mod sandbox_setup;
@@ -65,6 +66,7 @@ mod wsl_paths;
use crate::mcp_cmd::McpCli;
use crate::plugin_cmd::PluginCli;
use crate::plugin_cmd::PluginSubcommand;
use crate::queue_cmd::QueueCommand;
use crate::remote_control_cmd::RemoteControlCommand;
use doctor::DoctorCommand;
use state_db_recovery as local_state_db;
@@ -187,6 +189,9 @@ enum Subcommand {
/// Resume a previous interactive session (picker by default; use --last to continue the most recent).
Resume(ResumeCommand),
/// Queue a message for an existing session.
Queue(QueueCommand),
/// Archive a saved session by id or session name.
Archive(SessionArchiveCommand),
@@ -1324,6 +1329,18 @@ async fn cli_main(
.await?;
println!("{output}");
}
Some(Subcommand::Queue(cmd)) => {
let output = queue_cmd::run_queue_command(
cmd,
interactive,
root_config_overrides.clone(),
root_remote.clone(),
root_remote_auth_token_env.clone(),
arg0_paths.clone(),
)
.await?;
println!("{output}");
}
Some(Subcommand::Delete(DeleteCommand { session, force })) => {
let action = delete_action(&session.target, force)?;
let output = run_session_archive_cli_command(
@@ -1707,6 +1724,7 @@ fn profile_v2_for_subcommand<'a>(
Subcommand::Exec(_)
| Subcommand::Review(_)
| Subcommand::Resume(_)
| Subcommand::Queue(_)
| Subcommand::Archive(_)
| Subcommand::Delete(_)
| Subcommand::Unarchive(_)
@@ -1717,7 +1735,7 @@ fn profile_v2_for_subcommand<'a>(
subcommand: DebugSubcommand::PromptInput(_),
}) => Ok(Some(profile_v2)),
_ => anyhow::bail!(
"--profile only applies to runtime commands and `codex mcp`: `codex`, `codex exec`, `codex review`, `codex resume`, `codex archive`, `codex delete`, `codex unarchive`, `codex fork`, `codex mcp`, `codex sandbox`, and `codex debug prompt-input`."
"--profile only applies to runtime commands and `codex mcp`: `codex`, `codex exec`, `codex review`, `codex resume`, `codex queue`, `codex archive`, `codex delete`, `codex unarchive`, `codex fork`, `codex mcp`, `codex sandbox`, and `codex debug prompt-input`."
),
}
}
@@ -2275,6 +2293,7 @@ fn unsupported_subcommand_name_for_strict_config(
| Some(Subcommand::McpServer(_))
| Some(Subcommand::ExecServer(_))
| Some(Subcommand::Resume(_))
| Some(Subcommand::Queue(_))
| Some(Subcommand::Archive(_))
| Some(Subcommand::Delete(_))
| Some(Subcommand::Unarchive(_))
codex-rs/cli/src/queue_cmd.rs
+63
View File
@@ -0,0 +1,63 @@
use clap::Parser;
use clap::builder::NonEmptyStringValueParser;
use codex_arg0::Arg0DispatchPaths;
use codex_tui::Cli as TuiCli;
use codex_utils_cli::CliConfigOverrides;
use crate::InteractiveRemoteOptions;
use crate::SessionArchiveConfigOverrides;
use crate::finalize_session_archive_interactive;
use crate::resolve_remote_endpoint;
#[derive(Debug, Parser)]
pub(crate) struct QueueCommand {
/// Session UUID or exact session name.
#[arg(long, value_name = "THREAD")]
thread: String,
/// Message text to queue.
#[arg(long, value_name = "TEXT", value_parser = NonEmptyStringValueParser::new())]
message: String,
#[clap(flatten)]
remote: InteractiveRemoteOptions,
#[clap(flatten)]
config_overrides: SessionArchiveConfigOverrides,
}
pub(crate) async fn run_queue_command(
command: QueueCommand,
interactive: TuiCli,
root_config_overrides: CliConfigOverrides,
root_remote: Option<String>,
root_remote_auth_token_env: Option<String>,
arg0_paths: Arg0DispatchPaths,
) -> anyhow::Result<String> {
let QueueCommand {
thread,
message,
remote,
config_overrides,
} = command;
let cli =
finalize_session_archive_interactive(interactive, root_config_overrides, config_overrides);
if !cli.images.is_empty() {
anyhow::bail!("`codex queue` does not support image attachments");
}
let explicit_remote_endpoint = resolve_remote_endpoint(
remote.remote.or(root_remote),
remote.remote_auth_token_env.or(root_remote_auth_token_env),
)?;
codex_tui::run_session_queue_command(
thread,
message,
codex_tui::SessionArchiveCommandOptions {
cli,
arg0_paths,
explicit_remote_endpoint,
},
)
.await
.map_err(|error| anyhow::anyhow!("{error:#}"))
}
codex-rs/cli/tests/queue.rs
+271
View File
@@ -0,0 +1,271 @@
use std::path::Path;
use std::process::Output;
use anyhow::Context;
use anyhow::Result;
use futures::SinkExt;
use futures::StreamExt;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
const THREAD_ID: &str = "123e4567-e89b-12d3-a456-426614174000";
enum QueueResponse {
Success,
MethodNotFound,
#[cfg(unix)]
UnknownMethodVariant,
}
async fn respond_to_queue_request<S>(
stream: S,
codex_home: &Path,
response: QueueResponse,
) -> Result<Value>
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let mut websocket = accept_async(stream).await?;
let initialize = websocket
.next()
.await
.context("missing initialize request")??;
let initialize: Value = serde_json::from_str(initialize.to_text()?)?;
assert_eq!(initialize["method"], "initialize");
assert_eq!(
initialize["params"]["capabilities"]["experimentalApi"],
true
);
let initialized_response = json!({
"id": initialize["id"],
"result": {
"userAgent": "codex_cli_rs/0.0.0-test",
"codexHome": codex_home,
},
});
websocket
.send(Message::Text(initialized_response.to_string().into()))
.await?;
let initialized = websocket
.next()
.await
.context("missing initialized notification")??;
let initialized: Value = serde_json::from_str(initialized.to_text()?)?;
assert_eq!(initialized["method"], "initialized");
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");
let result = match response {
QueueResponse::Success => json!({
"id": request["id"],
"result": {
"queuedSubmission": {
"id": "queued-submission-id",
"input": request["params"]["input"],
"clientUserMessageId": request["params"]["clientUserMessageId"],
},
},
}),
QueueResponse::MethodNotFound => json!({
"id": request["id"],
"error": { "code": -32601, "message": "Method not found" },
}),
#[cfg(unix)]
QueueResponse::UnknownMethodVariant => json!({
"id": request["id"],
"error": {
"code": -32600,
"message": "Invalid request: unknown variant `thread/queue/add`, expected `thread/list`",
},
}),
};
websocket
.send(Message::Text(result.to_string().into()))
.await?;
Ok(request)
}
#[tokio::test]
async fn queue_submits_message_to_remote_app_server() -> Result<()> {
let (output, request) = run_remote_queue_command(QueueResponse::Success).await?;
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
assert_eq!(request["params"]["threadId"], THREAD_ID);
assert_eq!(request["params"]["input"][0]["text"], "do the thing");
assert_eq!(
String::from_utf8(output.stdout)?,
format!("Queued message queued-submission-id for thread {THREAD_ID}.\n")
);
Ok(())
}
#[tokio::test]
async fn queue_does_not_fallback_from_unsupported_explicit_remote() -> Result<()> {
let (output, _) = run_remote_queue_command(QueueResponse::MethodNotFound).await?;
assert!(!output.status.success());
assert!(
String::from_utf8(output.stderr)?
.contains("remote app server does not support thread/queue/add")
);
Ok(())
}
#[test]
fn queue_rejects_empty_message() -> Result<()> {
let output = std::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.args(["queue", "--thread", THREAD_ID, "--message", ""])
.output()?;
assert!(!output.status.success());
Ok(())
}
#[test]
fn queue_rejects_image_attachments() -> Result<()> {
let output = std::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.args([
"queue",
"--thread",
THREAD_ID,
"--message",
"do the thing",
"--image",
"screenshot.png",
])
.output()?;
assert!(!output.status.success());
assert!(String::from_utf8(output.stderr)?.contains("does not support image attachments"));
Ok(())
}
async fn run_remote_queue_command(response: QueueResponse) -> Result<(Output, Value)> {
let codex_home = TempDir::new()?;
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 server_home = codex_home.path().to_path_buf();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await?;
respond_to_queue_request(stream, server_home.as_path(), response).await
});
let remote_args = if remote_first {
[
"--remote",
endpoint.as_str(),
"--remote-auth-token-env",
"CODEX_REMOTE_TOKEN",
"queue",
]
} else {
[
"queue",
"--remote",
endpoint.as_str(),
"--remote-auth-token-env",
"CODEX_REMOTE_TOKEN",
]
};
let output = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.env("CODEX_HOME", codex_home.path())
.env("CODEX_REMOTE_TOKEN", "test-token")
.args(remote_args)
.args(["--thread", THREAD_ID, "--message", "do the thing"])
.output()
.await?;
Ok((output, server.await??))
}
#[cfg(unix)]
#[tokio::test]
async fn queue_rejects_local_daemon_that_does_not_support_queueing() -> Result<()> {
let codex_home = tempfile::tempdir_in("/tmp")?;
let socket_path = codex_app_server::app_server_control_socket_path(codex_home.path())?;
std::fs::create_dir_all(
socket_path
.as_path()
.parent()
.context("missing socket parent")?,
)?;
let listener = tokio::net::UnixListener::bind(socket_path.as_path())?;
let server_home = codex_home.path().to_path_buf();
let server = tokio::spawn(async move {
let (probe, _) = listener.accept().await?;
drop(probe);
let (stream, _) = listener.accept().await?;
respond_to_queue_request(
stream,
server_home.as_path(),
QueueResponse::UnknownMethodVariant,
)
.await
});
let output = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.env("CODEX_HOME", codex_home.path())
.args(["queue", "--thread", THREAD_ID, "--message", "do the thing"])
.output()
.await?;
server.await??;
assert!(!output.status.success());
assert!(
String::from_utf8(output.stderr)?
.contains("local app-server daemon does not support thread/queue/add")
);
assert!(!codex_home.path().join("queue_1.sqlite").exists());
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn queue_rejects_overrides_that_bypass_local_daemon() -> Result<()> {
let codex_home = tempfile::tempdir_in("/tmp")?;
let socket_path = codex_app_server::app_server_control_socket_path(codex_home.path())?;
std::fs::create_dir_all(
socket_path
.as_path()
.parent()
.context("missing socket parent")?,
)?;
let listener = tokio::net::UnixListener::bind(socket_path.as_path())?;
let server = tokio::spawn(async move {
let (probe, _) = listener.accept().await?;
drop(probe);
Ok::<_, std::io::Error>(())
});
let output = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
.env("CODEX_HOME", codex_home.path())
.args([
"queue",
"-c",
"model=\"test-model\"",
"--thread",
THREAD_ID,
"--message",
"do the thing",
])
.output()
.await?;
server.await??;
assert!(!output.status.success());
assert!(
String::from_utf8(output.stderr)?
.contains("embedded app server while a local app-server daemon is running")
);
Ok(())
}
codex-rs/tui/src/lib.rs
+2
View File
@@ -73,6 +73,7 @@ pub use session_archive_commands::DeleteConfirmation;
pub use session_archive_commands::SessionArchiveAction;
pub use session_archive_commands::SessionArchiveCommandOptions;
pub use session_archive_commands::run_session_archive_command;
pub use session_queue_commands::run_session_queue_command;
use std::fs::OpenOptions;
use std::io::IsTerminal;
use std::path::Path;
@@ -168,6 +169,7 @@ mod selection_list;
mod service_tier_resolution;
mod session_archive_commands;
mod session_log;
mod session_queue_commands;
mod session_resume;
mod session_state;
mod skills_helpers;
codex-rs/tui/src/named_session_lookup.rs
+7 -4
View File
@@ -7,6 +7,7 @@ use codex_app_server_protocol::Thread as AppServerThread;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadSortKey;
use codex_app_server_protocol::ThreadSourceKind;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionSource;
@@ -42,6 +43,7 @@ pub(super) struct NamedSessionCandidates<'a> {
collection: SessionCollection,
mode: SessionNameLookupMode,
search_term: Option<&'a str>,
source_kinds: Vec<ThreadSourceKind>,
cursor: Option<String>,
pending: std::vec::IntoIter<AppServerThread>,
has_more: bool,
@@ -138,7 +140,7 @@ async fn lookup_legacy_index_target(
}
/// Allows a missing SQLite name only for legacy threads, whose name may exist only in the index.
fn current_name_is_compatible(thread: &AppServerThread, name: &str) -> bool {
pub(super) fn current_name_is_compatible(thread: &AppServerThread, name: &str) -> bool {
match thread.history_mode {
ThreadHistoryMode::Legacy => thread.name.as_deref().is_none_or(|current| current == name),
ThreadHistoryMode::Paginated => thread.name.as_deref() == Some(name),
@@ -160,6 +162,7 @@ async fn lookup_from_app_server(
SessionCollection::Active,
mode,
Some(name),
super::resume_source_kinds(/*include_non_interactive*/ false),
);
while let Some(candidate) = candidates.next(app_server).await? {
if let Some(session_meta) = candidate.session_meta
@@ -196,6 +199,7 @@ impl<'a> NamedSessionCandidates<'a> {
collection: SessionCollection,
mode: SessionNameLookupMode,
search_term: Option<&'a str>,
source_kinds: Vec<ThreadSourceKind>,
) -> Self {
Self {
name,
@@ -203,6 +207,7 @@ impl<'a> NamedSessionCandidates<'a> {
collection,
mode,
search_term,
source_kinds,
cursor: None,
pending: Vec::new().into_iter(),
has_more: true,
@@ -274,9 +279,7 @@ impl<'a> NamedSessionCandidates<'a> {
sort_key: Some(sort_key),
sort_direction: None,
model_providers: None,
source_kinds: Some(super::resume_source_kinds(
/*include_non_interactive*/ false,
)),
source_kinds: Some(self.source_kinds.clone()),
archived: Some(self.collection == SessionCollection::Archived),
section_id: None,
project_id: None,
codex-rs/tui/src/session_archive_commands.rs
+83 -26
View File
@@ -19,6 +19,7 @@ use crate::legacy_core::config::resolve_profile_v2_config_path;
use crate::named_session_lookup::NamedSessionCandidates;
use crate::named_session_lookup::SessionCollection;
use crate::named_session_lookup::SessionNameLookupMode;
use crate::named_session_lookup::current_name_is_compatible;
use codex_app_server_protocol::Thread as AppServerThread;
use codex_arg0::Arg0DispatchPaths;
use codex_config::CloudConfigBundleLoader;
@@ -55,6 +56,12 @@ pub struct SessionArchiveCommandOptions {
pub explicit_remote_endpoint: Option<RemoteAppServerEndpoint>,
}
#[derive(Clone, Copy, PartialEq, Eq)]
pub(super) enum SessionNameMatch {
First,
Unique,
}
fn success_message(
action: SessionArchiveAction,
session_id: ThreadId,
@@ -83,7 +90,7 @@ pub async fn run_session_archive_command(
) -> Result<String> {
let codex_home = find_codex_home().wrap_err("failed to find Codex home")?;
let mut app_server =
start_app_server_for_archive_command(options, codex_home.to_path_buf()).await?;
start_app_server_for_session_command(options, codex_home.to_path_buf()).await?;
run_session_archive_action_with_app_server(
&mut app_server,
codex_home.as_path(),
@@ -160,8 +167,14 @@ async fn resolve_session_target(
SessionArchiveAction::Unarchive => ("archived", &[true]),
};
for &archived in archived_values {
if let Some(thread) =
lookup_session_by_exact_name(app_server, codex_home, target, archived).await?
if let Some(thread) = lookup_session_by_exact_name(
app_server,
codex_home,
target,
archived,
SessionNameMatch::First,
)
.await?
{
return session_target_from_app_server_thread(thread);
}
@@ -171,11 +184,12 @@ async fn resolve_session_target(
))
}
async fn lookup_session_by_exact_name(
pub(super) async fn lookup_session_by_exact_name(
app_server: &mut AppServerSession,
codex_home: &Path,
name: &str,
archived: bool,
match_policy: SessionNameMatch,
) -> Result<Option<AppServerThread>> {
// Remote workspaces stay on their existing server-side path. Local workspaces trust SQLite
// names, then scan and repair only after a miss or an unusable rollout path.
@@ -187,30 +201,73 @@ async fn lookup_session_by_exact_name(
SessionNameLookupMode::ScanAndRepair,
][..]
};
for &lookup_mode in lookup_modes {
// Search is the fast path, but legacy stores attach renamed titles after filtering.
for search_term in [Some(name), None] {
let mut candidates = NamedSessionCandidates::new(
name,
codex_home,
if archived {
SessionCollection::Archived
} else {
SessionCollection::Active
},
lookup_mode,
search_term,
);
if let Some(candidate) = candidates
.next(app_server)
.await
.wrap_err("failed to list sessions while resolving session name")?
{
return Ok(Some(candidate.thread));
let source_kind_filters = if match_policy == SessionNameMatch::Unique {
// An empty filter includes Atlas/ChatGPT sessions; explicit kinds additionally include exec.
vec![
super::resume_source_kinds(/*include_non_interactive*/ true),
Vec::new(),
]
} else {
vec![super::resume_source_kinds(
/*include_non_interactive*/ false,
)]
};
let mut unique_match: Option<AppServerThread> = None;
for source_kinds in source_kind_filters {
for &lookup_mode in lookup_modes {
// Search is the fast path, but legacy stores attach renamed titles after filtering.
for search_term in [Some(name), None] {
let mut candidates = NamedSessionCandidates::new(
name,
codex_home,
if archived {
SessionCollection::Archived
} else {
SessionCollection::Active
},
lookup_mode,
search_term,
source_kinds.clone(),
);
while let Some(candidate) = candidates
.next(app_server)
.await
.wrap_err("failed to list sessions while resolving session name")?
{
let thread = if match_policy == SessionNameMatch::Unique
&& lookup_mode == SessionNameLookupMode::ScanAndRepair
&& !app_server.uses_remote_workspace()
{
let thread = app_server
.thread_read(
ThreadId::from_string(&candidate.thread.id)?,
/*include_turns*/ false,
)
.await?;
if !current_name_is_compatible(&thread, name) {
continue;
}
thread
} else {
candidate.thread
};
if match_policy == SessionNameMatch::First {
return Ok(Some(thread));
}
if unique_match
.as_ref()
.is_some_and(|existing| existing.id != thread.id)
{
return Err(eyre!(
"More than one active session is named '{name}'; use a session UUID instead."
));
}
unique_match = Some(thread);
}
}
}
}
Ok(None)
Ok(unique_match)
}
fn session_target_from_app_server_thread(thread: AppServerThread) -> Result<ResolvedSessionTarget> {
@@ -251,7 +308,7 @@ fn confirm_session_delete(target: &ResolvedSessionTarget) -> Result<bool> {
Ok(answer.eq_ignore_ascii_case("y") || answer.eq_ignore_ascii_case("yes"))
}
async fn start_app_server_for_archive_command(
pub(super) async fn start_app_server_for_session_command(
options: SessionArchiveCommandOptions,
codex_home: PathBuf,
) -> Result<AppServerSession> {
codex-rs/tui/src/session_archive_commands_tests.rs
+136 -1
View File
@@ -19,6 +19,7 @@ use crate::app_server_session::AppServerSession;
use crate::app_server_session::ThreadParamsMode;
use crate::legacy_core::config::Config;
use crate::legacy_core::config::ConfigBuilder;
use crate::session_queue_commands::run_session_queue_action_with_app_server;
use crate::tests::start_test_embedded_app_server;
async fn build_config(temp_dir: &TempDir) -> std::io::Result<Config> {
@@ -57,6 +58,7 @@ fn write_rollout(
archived: bool,
timestamp: &str,
preview: &str,
source: SessionSource,
) -> color_eyre::Result<PathBuf> {
let subdir = if archived {
"archived_sessions"
@@ -76,7 +78,7 @@ fn write_rollout(
cwd: config.codex_home.join("project").to_path_buf(),
originator: "codex".to_string(),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
source,
model_provider: Some(config.model_provider_id.clone()),
..Default::default()
},
@@ -157,6 +159,7 @@ async fn archives_by_sqlite_name() -> color_eyre::Result<()> {
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Cli,
)?;
runtime
.upsert_thread(&thread_metadata(
@@ -208,6 +211,7 @@ async fn unarchives_by_sqlite_name() -> color_eyre::Result<()> {
/*archived*/ true,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Cli,
)?;
runtime
.upsert_thread(&thread_metadata(
@@ -259,6 +263,7 @@ async fn deletes_valid_duplicate_after_stale_sqlite_hit() -> color_eyre::Result<
/*archived*/ true,
"2025-02-01T10:00:00Z",
"stale preview",
SessionSource::Cli,
)?;
runtime
.upsert_thread(&thread_metadata(
@@ -278,6 +283,7 @@ async fn deletes_valid_duplicate_after_stale_sqlite_hit() -> color_eyre::Result<
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Cli,
)?;
codex_rollout::append_thread_name(config.codex_home.as_path(), thread_id, "saved-session")
.await?;
@@ -330,6 +336,7 @@ async fn trusts_sqlite_name_over_legacy_index_for_delete() -> color_eyre::Result
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Cli,
)?;
runtime
.upsert_thread(&thread_metadata(
@@ -345,6 +352,19 @@ async fn trusts_sqlite_name_over_legacy_index_for_delete() -> color_eyre::Result
.await?;
let mut app_server = start_app_server(config.clone()).await?;
let error = run_session_queue_action_with_app_server(
&mut app_server,
config.codex_home.as_path(),
"old-session",
"do the thing",
"stable-client-message-id",
)
.await
.expect_err("stale legacy names must not select renamed threads");
assert_eq!(
error.to_string(),
"No active session found matching 'old-session'."
);
let message = run_session_archive_action_with_app_server(
&mut app_server,
config.codex_home.as_path(),
@@ -371,3 +391,118 @@ async fn trusts_sqlite_name_over_legacy_index_for_delete() -> color_eyre::Result
);
Ok(())
}
#[tokio::test]
async fn queues_non_interactive_and_custom_sessions_and_rejects_ambiguous_name()
-> color_eyre::Result<()> {
let temp_dir = TempDir::new()?;
let config = build_config(&temp_dir).await?;
let runtime = state_runtime(&config).await?;
let thread_id = ThreadId::new();
let rollout_path = write_rollout(
&config,
thread_id,
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Exec,
)?;
let mut metadata = thread_metadata(
&config,
thread_id,
rollout_path,
"saved-session",
/*archived*/ false,
);
metadata.source = "exec".to_string();
runtime
.upsert_thread(&metadata)
.await
.map_err(std::io::Error::other)?;
let mut app_server = start_app_server(config.clone()).await?;
let (resolved_thread_id, response) = run_session_queue_action_with_app_server(
&mut app_server,
config.codex_home.as_path(),
"saved-session",
"do the thing",
"stable-client-message-id",
)
.await?;
let queued = runtime
.thread_queue()
.list_page(thread_id, /*offset*/ 0, /*limit*/ 10)
.await
.map_err(|error| color_eyre::eyre::eyre!("{error}"))?;
assert_eq!(resolved_thread_id, thread_id);
assert_eq!(queued.len(), 1);
assert_eq!(queued[0].id, response.queued_submission.id);
assert_eq!(
response.queued_submission.client_user_message_id,
"stable-client-message-id"
);
let custom_thread_id = ThreadId::new();
let custom_source = SessionSource::Custom("atlas".to_string());
let custom_rollout_path = write_rollout(
&config,
custom_thread_id,
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
custom_source.clone(),
)?;
let mut custom_metadata = thread_metadata(
&config,
custom_thread_id,
custom_rollout_path,
"atlas-session",
/*archived*/ false,
);
custom_metadata.source = serde_json::to_string(&custom_source)?;
runtime
.upsert_thread(&custom_metadata)
.await
.map_err(std::io::Error::other)?;
let (resolved_custom_thread_id, _) = run_session_queue_action_with_app_server(
&mut app_server,
config.codex_home.as_path(),
"atlas-session",
"do the thing",
"custom-client-message-id",
)
.await?;
assert_eq!(resolved_custom_thread_id, custom_thread_id);
let legacy_thread_id = ThreadId::new();
write_rollout(
&config,
legacy_thread_id,
/*archived*/ false,
"2025-02-01T10:00:00Z",
"preview",
SessionSource::Cli,
)?;
codex_rollout::append_thread_name(
config.codex_home.as_path(),
legacy_thread_id,
"saved-session",
)
.await?;
let error = run_session_queue_action_with_app_server(
&mut app_server,
config.codex_home.as_path(),
"saved-session",
"do the thing",
"stable-client-message-id",
)
.await
.expect_err("ambiguous session names should be rejected");
app_server.shutdown().await?;
assert_eq!(
error.to_string(),
"More than one active session is named 'saved-session'; use a session UUID instead."
);
Ok(())
}
codex-rs/tui/src/session_queue_commands.rs
+137
View File
@@ -0,0 +1,137 @@
//! Queue user messages through a local or remote app server.
use std::path::Path;
use crate::app_server_session::AppServerSession;
use crate::session_archive_commands::SessionArchiveCommandOptions;
use crate::session_archive_commands::SessionNameMatch;
use crate::session_archive_commands::lookup_session_by_exact_name;
use crate::session_archive_commands::start_app_server_for_session_command;
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::UserInput;
use codex_app_server_protocol::experimental_required_message;
use codex_protocol::ThreadId;
use codex_utils_home_dir::find_codex_home;
use color_eyre::Report;
use color_eyre::eyre::Result;
use color_eyre::eyre::WrapErr;
use color_eyre::eyre::eyre;
use uuid::Uuid;
const JSONRPC_INVALID_REQUEST: i64 = -32600;
const JSONRPC_METHOD_NOT_FOUND: i64 = -32601;
const THREAD_QUEUE_ADD_METHOD: &str = "thread/queue/add";
pub async fn run_session_queue_command(
target: String,
message: String,
options: SessionArchiveCommandOptions,
) -> Result<String> {
let codex_home = find_codex_home().wrap_err("failed to find Codex home")?;
let explicit_remote = options.explicit_remote_endpoint.is_some();
let mut app_server =
start_app_server_for_session_command(options, codex_home.to_path_buf()).await?;
if !explicit_remote
&& app_server.uses_embedded_app_server()
&& super::maybe_probe_default_daemon_socket(codex_home.as_path())
.await
.is_some()
{
return Err(eyre!(
"cannot queue through an embedded app server while a local app-server daemon is running; remove configuration overrides or use --remote"
));
}
let implicit_local_daemon = !explicit_remote && !app_server.uses_embedded_app_server();
let client_message_id = Uuid::now_v7().to_string();
let (thread_id, response) = match run_session_queue_action_with_app_server(
&mut app_server,
codex_home.as_path(),
&target,
&message,
&client_message_id,
)
.await
{
Err(error)
if (implicit_local_daemon || explicit_remote) && is_unsupported_queue_error(&error) =>
{
let server = if explicit_remote {
"remote app server"
} else {
"local app-server daemon"
};
return Err(error.wrap_err(format!(
"the {server} does not support thread/queue/add; update or restart the {server}"
)));
}
result => result?,
};
Ok(format!(
"Queued message {} for thread {}.",
response.queued_submission.id, thread_id
))
}
pub(super) async fn run_session_queue_action_with_app_server(
app_server: &mut AppServerSession,
codex_home: &Path,
target: &str,
message: &str,
client_message_id: &str,
) -> Result<(ThreadId, ThreadQueueAddResponse)> {
let thread_id = if let Ok(thread_id) = ThreadId::from_string(target) {
thread_id
} else {
let thread = lookup_session_by_exact_name(
app_server,
codex_home,
target,
/*archived*/ false,
SessionNameMatch::Unique,
)
.await?
.ok_or_else(|| eyre!("No active session found matching '{target}'."))?;
ThreadId::from_string(&thread.id)
.wrap_err_with(|| format!("app server returned invalid session id `{}`", thread.id))?
};
let request_id = app_server.next_request_id();
let response = app_server
.request_handle()
.request_typed(ClientRequest::ThreadQueueAdd {
request_id,
params: ThreadQueueAddParams {
thread_id: thread_id.to_string(),
input: vec![UserInput::Text {
text: message.to_string(),
text_elements: Vec::new(),
}],
client_user_message_id: client_message_id.to_string(),
},
})
.await
.wrap_err("failed to queue session message")?;
Ok((thread_id, response))
}
fn is_unsupported_queue_error(error: &Report) -> bool {
matches!(
error.downcast_ref::<TypedRequestError>(),
Some(TypedRequestError::Server { method, source })
if method == THREAD_QUEUE_ADD_METHOD
&& (source.code == JSONRPC_METHOD_NOT_FOUND
|| (source.code == JSONRPC_INVALID_REQUEST
&& (source.message == experimental_required_message(THREAD_QUEUE_ADD_METHOD)
|| source.message.starts_with(
"Invalid request: unknown variant `thread/queue/add`",
))))
)
}
#[cfg(test)]
#[path = "session_queue_commands_tests.rs"]
mod tests;
codex-rs/tui/src/session_queue_commands_tests.rs
+53
View File
@@ -0,0 +1,53 @@
use codex_app_server_client::TypedRequestError;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::experimental_required_message;
use color_eyre::Report;
use pretty_assertions::assert_eq;
use super::JSONRPC_INVALID_REQUEST;
use super::JSONRPC_METHOD_NOT_FOUND;
use super::THREAD_QUEUE_ADD_METHOD;
use super::is_unsupported_queue_error;
#[test]
fn recognizes_only_definitively_unsupported_queue_errors() {
for (code, message, unsupported) in [
(
JSONRPC_METHOD_NOT_FOUND,
"Method not found".to_string(),
true,
),
(
JSONRPC_INVALID_REQUEST,
experimental_required_message(THREAD_QUEUE_ADD_METHOD),
true,
),
(
JSONRPC_INVALID_REQUEST,
"Invalid request: unknown variant `thread/queue/add`, expected `thread/list`"
.to_string(),
true,
),
(
JSONRPC_INVALID_REQUEST,
"queue cannot contain more than 100 submissions".to_string(),
false,
),
] {
let error = Report::new(TypedRequestError::Server {
method: THREAD_QUEUE_ADD_METHOD.to_string(),
source: JSONRPCErrorError {
code,
message,
data: None,
},
});
assert_eq!(is_unsupported_queue_error(&error), unsupported);
}
let transport_error = Report::new(TypedRequestError::Transport {
method: THREAD_QUEUE_ADD_METHOD.to_string(),
source: std::io::Error::new(std::io::ErrorKind::TimedOut, "request timed out"),
});
assert!(!is_unsupported_queue_error(&transport_error));
}