From b1236df88ea0b32e39ba41259ff6888a08266464 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sun, 12 Apr 2026 16:34:38 -0700 Subject: [PATCH] codex: add queued external message delivery --- .../app-server/src/codex_message_processor.rs | 4 + .../tests/suite/v2/thread_archive.rs | 88 +++++ codex-rs/cli/src/main.rs | 16 + codex-rs/cli/src/queue_cmd.rs | 301 ++++++++++++++++++ codex-rs/core/config.schema.json | 6 + codex-rs/core/src/codex/timer_runtime.rs | 162 +++++++++- codex-rs/core/src/injected_message.rs | 17 + codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/rollout.rs | 1 + codex-rs/core/tests/suite/timers.rs | 250 +++++++++++++++ codex-rs/features/src/lib.rs | 8 + codex-rs/features/src/tests.rs | 10 + codex-rs/rollout/src/lib.rs | 1 + codex-rs/rollout/src/session_index.rs | 42 +++ codex-rs/rollout/src/session_index_tests.rs | 41 +++ 15 files changed, 933 insertions(+), 15 deletions(-) create mode 100644 codex-rs/cli/src/queue_cmd.rs diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 6762b9e129..17f50850f7 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -5739,6 +5739,10 @@ impl CodexMessageProcessor { let _ = ctx .mark_archived(thread_id, archived_path.as_path(), Utc::now()) .await; + let thread_id_str = thread_id.to_string(); + if let Err(err) = ctx.delete_thread_delivery_state(&thread_id_str).await { + warn!("failed to delete delivery state for archived thread {thread_id}: {err}"); + } } Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 20fd6fd884..0235a9a133 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -20,6 +20,9 @@ use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_core::find_thread_path_by_id_str; +use codex_state::StateRuntime; +use codex_state::ThreadMessageCreateParams; +use codex_state::ThreadTimerCreateParams; use pretty_assertions::assert_eq; use std::path::Path; use tempfile::TempDir; @@ -118,6 +121,26 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { .expect("expected rollout path for thread id to exist after materialization"); assert_paths_match_on_disk(&discovered_path, &rollout_path)?; + let state_db = StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".to_string()) + .await + .expect("initialize state db"); + state_db + .create_thread_message(&message_params("message-1", &thread.id)) + .await + .expect("create archived thread message"); + state_db + .create_thread_message(&message_params("message-2", "other-thread")) + .await + .expect("create other thread message"); + state_db + .create_thread_timer(&timer_params("timer-1", &thread.id)) + .await + .expect("create archived thread timer"); + state_db + .create_thread_timer(&timer_params("timer-2", "other-thread")) + .await + .expect("create other thread timer"); + let archive_id = mcp .send_thread_archive_request(ThreadArchiveParams { thread_id: thread.id.clone(), @@ -156,6 +179,40 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { "expected archived rollout path {} to exist", archived_rollout_path.display() ); + assert_eq!( + state_db + .list_thread_messages(&thread.id) + .await + .expect("list archived thread messages"), + Vec::new() + ); + assert_eq!( + state_db + .list_thread_timers(&thread.id) + .await + .expect("list archived thread timers"), + Vec::new() + ); + assert_eq!( + state_db + .list_thread_messages("other-thread") + .await + .expect("list other thread messages") + .into_iter() + .map(|message| message.id) + .collect::>(), + vec!["message-2".to_string()] + ); + assert_eq!( + state_db + .list_thread_timers("other-thread") + .await + .expect("list other thread timers") + .into_iter() + .map(|timer| timer.id) + .collect::>(), + vec!["timer-2".to_string()] + ); Ok(()) } @@ -323,3 +380,34 @@ fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result assert_eq!(actual, expected); Ok(()) } + +fn message_params(id: &str, thread_id: &str) -> ThreadMessageCreateParams { + ThreadMessageCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "external".to_string(), + content: "do something".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + queued_at: 100, + } +} + +fn timer_params(id: &str, thread_id: &str) -> ThreadTimerCreateParams { + ThreadTimerCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "agent".to_string(), + client_id: "codex-tui".to_string(), + trigger_json: r#"{"kind":"delay","seconds":10,"repeat":false}"#.to_string(), + content: "run tests".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + created_at: 100, + next_run_at: Some(110), + last_run_at: None, + pending_run: false, + } +} diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index b581898606..4753b84d01 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -40,11 +40,14 @@ mod app_cmd; mod desktop_app; mod marketplace_cmd; mod mcp_cmd; +mod queue_cmd; #[cfg(not(windows))] mod wsl_paths; use crate::marketplace_cmd::MarketplaceCli; use crate::mcp_cmd::McpCli; +use crate::queue_cmd::QueueCommand; +use crate::queue_cmd::run_queue_command; use codex_core::config::Config; use codex_core::config::ConfigOverrides; @@ -66,6 +69,8 @@ use codex_terminal_detection::TerminalName; version, // If a sub‑command is given, ignore requirements of the default args. subcommand_negates_reqs = true, + // Prefer a recognized subcommand over the default interactive prompt positional. + subcommand_precedence_over_arg = true, // The executable is sometimes invoked via a platform‑specific name like // `codex-x86_64-unknown-linux-musl`, but the help output should always use // the generic `codex` command name that users run. @@ -140,6 +145,9 @@ enum Subcommand { /// Resume a previous interactive session (picker by default; use --last to continue the most recent). Resume(ResumeCommand), + /// Queue a message to an existing thread. + Queue(QueueCommand), + /// Fork a previous interactive session (picker by default; use --last to fork the most recent). Fork(ForkCommand), @@ -807,6 +815,14 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> { .await?; handle_app_exit(exit_info)?; } + Some(Subcommand::Queue(queue_cli)) => { + reject_remote_mode_for_subcommand( + root_remote.as_deref(), + root_remote_auth_token_env.as_deref(), + "queue", + )?; + run_queue_command(queue_cli, &root_config_overrides, &interactive).await?; + } Some(Subcommand::Fork(ForkCommand { session_id, last, diff --git a/codex-rs/cli/src/queue_cmd.rs b/codex-rs/cli/src/queue_cmd.rs new file mode 100644 index 0000000000..57f4488a85 --- /dev/null +++ b/codex-rs/cli/src/queue_cmd.rs @@ -0,0 +1,301 @@ +//! Implementation for the `codex queue` command. +//! +//! The top-level CLI module owns command routing; this module owns the +//! queue-specific policy for resolving target threads and writing immediate +//! messages into the SQLite state database. + +use clap::Parser; +use codex_core::config::Config; +use codex_core::config::ConfigOverrides; +use codex_core::timers::TimerDelivery; +use codex_features::Feature; +use codex_features::Features; +use codex_protocol::ThreadId; +use codex_state::StateRuntime; +use codex_tui::Cli as TuiCli; +use codex_utils_cli::CliConfigOverrides; +use std::path::Path; + +#[derive(Debug, Parser)] +pub(crate) struct QueueCommand { + /// Target thread id. + #[arg(long = "thread", value_name = "THREAD_ID")] + thread: String, + + /// Message text. + #[arg(long = "message", value_name = "TEXT")] + message: String, +} + +pub(crate) async fn run_queue_command( + cmd: QueueCommand, + root_config_overrides: &CliConfigOverrides, + interactive: &TuiCli, +) -> anyhow::Result<()> { + let cli_kv_overrides = root_config_overrides + .parse_overrides() + .map_err(anyhow::Error::msg)?; + let overrides = ConfigOverrides { + config_profile: interactive.config_profile.clone(), + ..Default::default() + }; + let config = + Config::load_with_cli_overrides_and_harness_overrides(cli_kv_overrides, overrides).await?; + validate_queue_feature_flags(&config.features)?; + let thread_id = resolve_queue_thread_id(config.codex_home.as_path(), &cmd.thread).await?; + let state_db = + StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()).await?; + let delivery = TimerDelivery::AfterTurn; + + let message_params = codex_state::ThreadMessageCreateParams::new( + thread_id, + "external".to_string(), + cmd.message, + /*instructions*/ None, + "{}".to_string(), + delivery.as_str().to_string(), + unix_timestamp_now()?, + ); + state_db.create_thread_message(&message_params).await?; + remove_queued_message_if_thread_missing( + config.codex_home.as_path(), + &state_db, + &message_params.thread_id, + &message_params.id, + ) + .await?; + println!( + "Queued message {} for thread {}.", + message_params.id, message_params.thread_id + ); + Ok(()) +} + +async fn remove_queued_message_if_thread_missing( + codex_home: &Path, + state_db: &StateRuntime, + thread_id: &str, + message_id: &str, +) -> anyhow::Result<()> { + if codex_core::find_thread_path_by_id_str(codex_home, thread_id) + .await? + .is_some() + { + return Ok(()); + } + + state_db + .delete_thread_message(thread_id, message_id) + .await?; + anyhow::bail!("thread `{thread_id}` was archived before queued work could be created"); +} + +fn validate_queue_feature_flags(features: &Features) -> anyhow::Result<()> { + if !features.enabled(Feature::QueuedMessages) { + anyhow::bail!("codex queue requires the queued_messages feature"); + } + Ok(()) +} + +async fn resolve_queue_thread_id(codex_home: &Path, target: &str) -> anyhow::Result { + if let Ok(thread_id) = ThreadId::from_string(target) { + if codex_core::find_thread_path_by_id_str(codex_home, &thread_id.to_string()) + .await? + .is_none() + { + anyhow::bail!("no thread with id `{thread_id}`"); + } + return Ok(thread_id.to_string()); + } + + let mut active_thread_ids = Vec::new(); + for thread_id in codex_core::find_thread_ids_by_name(codex_home, target).await? { + if codex_core::find_thread_path_by_id_str(codex_home, &thread_id.to_string()) + .await? + .is_some() + { + active_thread_ids.push(thread_id); + } + } + + match active_thread_ids.as_slice() { + [] => anyhow::bail!("no thread named `{target}`"), + [thread_id] => Ok(thread_id.to_string()), + _ => anyhow::bail!("more than one thread is named `{target}`; use a thread id instead"), + } +} + +fn unix_timestamp_now() -> anyhow::Result { + let duration = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|err| anyhow::anyhow!("system clock is before unix epoch: {err}"))?; + i64::try_from(duration.as_secs()).map_err(|_| anyhow::anyhow!("current time is out of range")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::MultitoolCli; + use crate::Subcommand; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + + fn write_test_rollout(codex_home: &Path, thread_id: ThreadId) { + let sessions_dir = codex_home + .join("sessions") + .join("2026") + .join("04") + .join("10"); + std::fs::create_dir_all(&sessions_dir).expect("create sessions dir"); + std::fs::write( + sessions_dir.join(format!("rollout-2026-04-10T12-00-00-{thread_id}.jsonl")), + "", + ) + .expect("write rollout"); + } + + #[test] + fn queue_command_parses_immediate_message() { + let cli = MultitoolCli::try_parse_from([ + "codex", + "queue", + "--thread", + "thread-1", + "--message", + "do work", + ]) + .expect("parse"); + let Some(Subcommand::Queue(cmd)) = cli.subcommand else { + unreachable!() + }; + + assert_eq!(cmd.thread, "thread-1"); + assert_eq!(cmd.message, "do work"); + } + + #[test] + fn queue_without_required_args_is_subcommand_error() { + let err = MultitoolCli::try_parse_from(["codex", "queue"]) + .expect_err("queue should be parsed as a subcommand, not as an interactive prompt"); + assert_eq!(err.kind(), clap::error::ErrorKind::MissingRequiredArgument); + } + + #[test] + fn queue_requires_queued_messages_feature() { + let mut features = Features::with_defaults(); + + let err = validate_queue_feature_flags(&features) + .expect_err("queue should require queued_messages"); + + assert_eq!( + err.to_string(), + "codex queue requires the queued_messages feature" + ); + + features.enable(Feature::QueuedMessages); + validate_queue_feature_flags(&features) + .expect("queued messages feature should permit immediate queue command"); + } + + #[tokio::test] + async fn queue_thread_resolves_thread_name() { + let temp = TempDir::new().expect("tempdir"); + let thread_id = ThreadId::new(); + write_test_rollout(temp.path(), thread_id); + codex_core::append_thread_name(temp.path(), thread_id, "named-thread") + .await + .expect("append thread name"); + + assert_eq!( + resolve_queue_thread_id(temp.path(), "named-thread") + .await + .expect("resolve"), + thread_id.to_string() + ); + } + + #[tokio::test] + async fn queue_thread_id_requires_existing_thread() { + let temp = TempDir::new().expect("tempdir"); + let thread_id = ThreadId::new(); + write_test_rollout(temp.path(), thread_id); + + assert_eq!( + resolve_queue_thread_id(temp.path(), &thread_id.to_string()) + .await + .expect("resolve"), + thread_id.to_string() + ); + + let missing = ThreadId::new(); + assert_eq!( + resolve_queue_thread_id(temp.path(), &missing.to_string()) + .await + .expect_err("missing id should fail") + .to_string(), + format!("no thread with id `{missing}`") + ); + } + + #[tokio::test] + async fn queue_thread_name_rejects_missing_and_ambiguous_names() { + let temp = TempDir::new().expect("tempdir"); + let first = ThreadId::new(); + let second = ThreadId::new(); + write_test_rollout(temp.path(), first); + write_test_rollout(temp.path(), second); + codex_core::append_thread_name(temp.path(), first, "same") + .await + .expect("append first name"); + codex_core::append_thread_name(temp.path(), second, "same") + .await + .expect("append second name"); + + assert_eq!( + resolve_queue_thread_id(temp.path(), "missing") + .await + .expect_err("missing name should fail") + .to_string(), + "no thread named `missing`" + ); + assert_eq!( + resolve_queue_thread_id(temp.path(), "same") + .await + .expect_err("ambiguous name should fail") + .to_string(), + "more than one thread is named `same`; use a thread id instead" + ); + } + + #[tokio::test] + async fn queue_thread_name_ignores_names_without_rollouts() { + let temp = TempDir::new().expect("tempdir"); + let stale = ThreadId::new(); + let active = ThreadId::new(); + write_test_rollout(temp.path(), active); + codex_core::append_thread_name(temp.path(), stale, "same") + .await + .expect("append stale name"); + codex_core::append_thread_name(temp.path(), stale, "stale") + .await + .expect("append stale-only name"); + codex_core::append_thread_name(temp.path(), active, "same") + .await + .expect("append active name"); + + assert_eq!( + resolve_queue_thread_id(temp.path(), "same") + .await + .expect("resolve"), + active.to_string() + ); + + assert_eq!( + resolve_queue_thread_id(temp.path(), "stale") + .await + .expect_err("stale name should fail") + .to_string(), + "no thread named `stale`" + ); + } +} diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 718164a3c2..a49dd0b63c 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -434,6 +434,9 @@ "prevent_idle_sleep": { "type": "boolean" }, + "queued_messages": { + "type": "boolean" + }, "realtime_conversation": { "type": "boolean" }, @@ -2286,6 +2289,9 @@ "prevent_idle_sleep": { "type": "boolean" }, + "queued_messages": { + "type": "boolean" + }, "realtime_conversation": { "type": "boolean" }, diff --git a/codex-rs/core/src/codex/timer_runtime.rs b/codex-rs/core/src/codex/timer_runtime.rs index f5557a9889..f3c66d633e 100644 --- a/codex-rs/core/src/codex/timer_runtime.rs +++ b/codex-rs/core/src/codex/timer_runtime.rs @@ -1,26 +1,29 @@ -//! SQLite-backed runtime bridge for thread timers. +//! SQLite-backed runtime bridge for thread timers and queued thread messages. //! //! This module connects [`Session`] to the persistent state database, keeps the //! in-memory timer scheduler reconciled with cross-instance changes, and -//! converts claimed timers into generated model input plus transcript-safe -//! delivery events. +//! converts claimed timers/messages into generated model input plus +//! transcript-safe delivery events. //! -//! Timer delivery must be single-consumer across all harness instances for a -//! thread, even though those instances share the same SQLite state database. In -//! other words, if two app or CLI processes are attached to the same thread, a -//! due timer should be injected by at most one of them. +//! Timer and queued-message delivery must be single-consumer across all harness +//! instances for a thread, even though those instances share the same SQLite +//! state database. In other words, if two app or CLI processes are attached to +//! the same thread, a due timer or queued message should be injected by at most +//! one of them. //! //! The database is the authority for that guarantee. Before this module -//! Timers are first selected from local memory, but delivery proceeds only if -//! the matching SQLite claim also wins: one-shot timers are deleted as part of -//! the claim, and recurring timers are updated with the expected previous run -//! timestamp so competing instances cannot both observe and persist the same -//! run. If another instance wins the database race, this runtime refreshes its -//! local timer view from SQLite and skips delivery. +//! delivers a queued message, it calls into the state layer to atomically claim +//! and remove the next eligible row. Timers are first selected from local +//! memory, but delivery proceeds only if the matching SQLite claim also wins: +//! one-shot timers are deleted as part of the claim, and recurring timers are +//! updated with the expected previous run timestamp so competing instances +//! cannot both observe and persist the same run. If another instance wins the +//! database race, this runtime refreshes its local timer view from SQLite and +//! skips delivery. //! //! The local `timer_start_in_progress` flag is still useful, but only as an //! in-process guard. It prevents this [`Session`] from starting multiple pending -//! timer deliveries concurrently; cross-process exclusivity comes from +//! timer/message deliveries concurrently; cross-process exclusivity comes from //! the SQLite claim operations above. use super::BackgroundEventEvent; @@ -28,6 +31,7 @@ use super::Event; use super::EventMsg; use super::INITIAL_SUBMIT_ID; use super::Session; +use crate::injected_message::InjectedMessage; use crate::injected_message::MessagePayload; use crate::pending_input::PendingInputItem; use crate::timers::ClaimedTimer; @@ -66,6 +70,17 @@ enum TimerDbSyncStatus { Changed, } +enum PendingMessageStart { + Started, + NotReady, + None, +} + +enum PendingMessageClaim { + Claimed(Box, TimerDelivery), + NotReady, +} + fn db_timer_to_persisted_timer(row: codex_state::ThreadTimer) -> Option { let trigger = match serde_json::from_str(&row.trigger_json) { Ok(trigger) => trigger, @@ -219,6 +234,10 @@ impl Session { { return; } + match self.maybe_start_pending_message().await { + PendingMessageStart::Started | PendingMessageStart::NotReady => return, + PendingMessageStart::None => {} + } self.try_start_pending_timer(RecurringTimerPolicy::IncludeAll) .await; } @@ -263,6 +282,111 @@ impl Session { true } + async fn maybe_start_pending_message(self: &Arc) -> PendingMessageStart { + let Some(claim) = self.claim_next_message_for_delivery().await else { + return PendingMessageStart::None; + }; + let PendingMessageClaim::Claimed(input_item, delivery) = claim else { + return PendingMessageStart::NotReady; + }; + let input_item = *input_item; + + match delivery { + TimerDelivery::SteerCurrentTurn => { + if !self + .inject_message_into_active_turn(input_item.clone()) + .await + { + self.queue_pending_input_for_next_turn(vec![input_item]) + .await; + self.maybe_start_turn_for_pending_work().await; + } + } + TimerDelivery::AfterTurn => { + self.queue_pending_input_for_next_turn(vec![input_item]) + .await; + self.maybe_start_turn_for_pending_work().await; + } + } + *self.timer_start_in_progress.lock().await = false; + PendingMessageStart::Started + } + + async fn claim_next_message_for_delivery(self: &Arc) -> Option { + if !self.queued_messages_feature_enabled() { + return None; + } + let mut timer_start_in_progress = self.timer_start_in_progress.lock().await; + if *timer_start_in_progress { + return None; + } + *timer_start_in_progress = true; + drop(timer_start_in_progress); + + let has_pending_turn_inputs = self.has_queued_response_items_for_next_turn().await + || self.has_trigger_turn_mailbox_items().await; + let (has_active_turn, active_turn_is_regular) = { + let active_turn = self.active_turn.lock().await; + let has_active_turn = active_turn.is_some(); + let active_turn_is_regular = active_turn + .as_ref() + .and_then(|turn| turn.tasks.first()) + .is_some_and(|(_, task)| matches!(task.kind, crate::state::TaskKind::Regular)); + (has_active_turn, active_turn_is_regular) + }; + let can_after_turn = !has_active_turn && !has_pending_turn_inputs; + let can_steer_current_turn = active_turn_is_regular; + let state_db = match self.timer_state_db().await { + Ok(state_db) => state_db, + Err(err) => { + warn!("failed to claim queued message from sqlite: {err}"); + *self.timer_start_in_progress.lock().await = false; + return None; + } + }; + self.start_timer_db_sync_task(state_db.clone()); + + loop { + let claim = match state_db + .claim_next_thread_message( + &self.thread_id_string(), + can_after_turn, + can_steer_current_turn, + ) + .await + { + Ok(claim) => claim, + Err(err) => { + warn!("failed to claim queued message from sqlite: {err}"); + *self.timer_start_in_progress.lock().await = false; + return None; + } + }; + match claim { + Some(codex_state::ThreadMessageClaim::Claimed(row)) => { + let (message, delivery) = match InjectedMessage::from_external_row(row) { + Ok(parsed) => parsed, + Err(err) => { + warn!("{err}"); + continue; + } + }; + let input_item = + PendingInputItem::injected(message.prompt_input_item(), message.event()); + return Some(PendingMessageClaim::Claimed(Box::new(input_item), delivery)); + } + Some(codex_state::ThreadMessageClaim::Invalid { id, reason }) => { + warn!("dropped invalid queued message {id}: {reason}"); + continue; + } + Some(codex_state::ThreadMessageClaim::NotReady) | None => { + *self.timer_start_in_progress.lock().await = false; + return claim.map(|_| PendingMessageClaim::NotReady); + } + } + } + } + async fn claim_next_timer_for_delivery( self: &Arc, recurring_timer_policy: RecurringTimerPolicy, @@ -310,6 +434,10 @@ impl Session { } async fn inject_timer_into_active_turn(&self, item: PendingInputItem) -> bool { + self.inject_message_into_active_turn(item).await + } + + async fn inject_message_into_active_turn(&self, item: PendingInputItem) -> bool { let active = self.active_turn.lock().await; let Some(active_turn) = active.as_ref() else { return false; @@ -634,8 +762,12 @@ impl Session { self.features.enabled(Feature::Timers) } + fn queued_messages_feature_enabled(&self) -> bool { + self.features.enabled(Feature::QueuedMessages) + } + fn timer_db_sync_feature_enabled(&self) -> bool { - self.timers_feature_enabled() + self.timers_feature_enabled() || self.queued_messages_feature_enabled() } fn spawn_restored_timer_tasks(self: &Arc, restored_tasks: Vec) { diff --git a/codex-rs/core/src/injected_message.rs b/codex-rs/core/src/injected_message.rs index ecea47104e..6034f8de92 100644 --- a/codex-rs/core/src/injected_message.rs +++ b/codex-rs/core/src/injected_message.rs @@ -8,6 +8,7 @@ #![allow(dead_code)] +use crate::timers::TimerDelivery; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseInputItem; use codex_protocol::protocol::InjectedMessageEvent; @@ -43,6 +44,22 @@ pub(crate) enum InjectedMessage { } impl InjectedMessage { + pub(crate) fn from_external_row( + row: codex_state::ThreadMessage, + ) -> Result<(Self, TimerDelivery), String> { + let delivery = serde_json::from_value::(serde_json::Value::String( + row.delivery.clone(), + )) + .map_err(|err| format!("invalid message delivery `{}`: {err}", row.delivery))?; + Ok(( + Self::External { + source: row.source, + content: row.content, + }, + delivery, + )) + } + pub(crate) fn prompt_input_item(&self) -> ResponseInputItem { ResponseInputItem::Message { role: "user".to_string(), diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 04b9d17394..d6fcd37cb1 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -171,6 +171,7 @@ pub use rollout::append_thread_name; pub use rollout::find_archived_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use rollout::find_conversation_path_by_id_str; +pub use rollout::find_thread_ids_by_name; pub use rollout::find_thread_meta_by_name_str; pub use rollout::find_thread_name_by_id; pub use rollout::find_thread_names_by_ids; diff --git a/codex-rs/core/src/rollout.rs b/codex-rs/core/src/rollout.rs index 26e3a84d5c..7a902da062 100644 --- a/codex-rs/core/src/rollout.rs +++ b/codex-rs/core/src/rollout.rs @@ -14,6 +14,7 @@ pub use codex_rollout::append_thread_name; pub use codex_rollout::find_archived_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use codex_rollout::find_conversation_path_by_id_str; +pub use codex_rollout::find_thread_ids_by_name; pub use codex_rollout::find_thread_meta_by_name_str; pub use codex_rollout::find_thread_name_by_id; pub use codex_rollout::find_thread_names_by_ids; diff --git a/codex-rs/core/tests/suite/timers.rs b/codex-rs/core/tests/suite/timers.rs index 51016e0eb5..9723e28307 100644 --- a/codex-rs/core/tests/suite/timers.rs +++ b/codex-rs/core/tests/suite/timers.rs @@ -377,3 +377,253 @@ async fn list_timers_discovers_externally_inserted_timer() -> Result<()> { Ok(()) } + +#[cfg_attr( + target_os = "windows", + ignore = "timer/message integration tests currently exceed the Windows Bazel job timeout" +)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn queued_messages_feature_consumes_messages_without_timers() -> Result<()> { + let server = start_mock_server().await; + let mock = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "queued turn"), + ev_completed("resp-1"), + ]), + ) + .await; + + let mut builder = test_codex().with_config(|config| { + config + .features + .enable(Feature::QueuedMessages) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + config + .features + .enable(Feature::Sqlite) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + }); + let test = builder.build(&server).await?; + let db = test.codex.state_db().expect("state db enabled"); + let thread_id = test.session_configured.session_id.to_string(); + db.create_thread_message(&codex_state::ThreadMessageCreateParams::new( + thread_id, + "external".to_string(), + "queued hello".to_string(), + /*instructions*/ None, + "{}".to_string(), + TimerDelivery::AfterTurn.as_str().to_string(), + Utc::now().timestamp(), + )) + .await?; + + wait_for_event_with_timeout( + &test.codex, + |event| match event { + EventMsg::InjectedMessage(event) => { + event.source == "external" && event.content == "queued hello" + } + _ => false, + }, + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + wait_for_event_with_timeout( + &test.codex, + |event| matches!(event, EventMsg::TurnComplete(_)), + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + + let requests = mock.requests(); + assert_eq!(requests.len(), 1); + assert!(requests.iter().any(|request| { + request + .message_input_texts("user") + .iter() + .any(|message| message.contains("\nqueued hello\n")) + })); + + Ok(()) +} + +#[cfg_attr( + target_os = "windows", + ignore = "timer/message integration tests currently exceed the Windows Bazel job timeout" +)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn queued_message_runs_after_idle_recurring_timer() -> Result<()> { + let server = start_mock_server().await; + let mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "timer turn"), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-2", "queued turn"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + + let mut builder = test_codex().with_config(|config| { + config + .features + .enable(Feature::Timers) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + config + .features + .enable(Feature::QueuedMessages) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + config + .features + .enable(Feature::Sqlite) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + }); + let test = builder.build(&server).await?; + let db = test.codex.state_db().expect("state db enabled"); + let timer = test + .codex + .create_timer( + ThreadTimerTrigger::Delay { + seconds: 0, + repeat: Some(true), + }, + MessagePayload { + content: "keep going".to_string(), + instructions: None, + meta: Default::default(), + }, + TimerDelivery::AfterTurn, + ) + .await + .map_err(|err| anyhow!("{err}"))?; + wait_for_event_with_timeout( + &test.codex, + |event| match event { + EventMsg::InjectedMessage(event) => event.source == format!("timer {}", timer.id), + _ => false, + }, + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + let thread_id = test.session_configured.session_id.to_string(); + db.create_thread_message(&codex_state::ThreadMessageCreateParams::new( + thread_id, + "external".to_string(), + "queued hello".to_string(), + /*instructions*/ None, + "{}".to_string(), + TimerDelivery::AfterTurn.as_str().to_string(), + Utc::now().timestamp(), + )) + .await?; + assert!( + test.codex + .delete_timer(&timer.id) + .await + .map_err(|err| anyhow!("{err}"))?, + "test should delete the idle recurring timer before it can schedule another turn" + ); + + wait_for_event_with_timeout( + &test.codex, + |event| matches!(event, EventMsg::TurnComplete(_)), + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + wait_for_event_with_timeout( + &test.codex, + |event| match event { + EventMsg::InjectedMessage(event) => event.source == "external", + _ => false, + }, + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + wait_for_event_with_timeout( + &test.codex, + |event| matches!(event, EventMsg::TurnComplete(_)), + TIMER_INTEGRATION_TIMEOUT, + ) + .await; + + let requests = mock.requests(); + assert!( + requests.len() >= 2, + "expected timer and queued-message turns to run" + ); + assert!( + requests[0] + .message_input_texts("user") + .iter() + .any(|message| message.contains("\nTimer fired: keep going\n")) + ); + assert!( + requests[1] + .message_input_texts("user") + .iter() + .any(|message| message.contains("\nqueued hello\n")) + ); + + Ok(()) +} + +#[cfg_attr( + target_os = "windows", + ignore = "timer/message integration tests currently exceed the Windows Bazel job timeout" +)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn queued_messages_feature_disabled_leaves_messages_queued() -> Result<()> { + let server = start_mock_server().await; + let mock = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "first turn"), + ev_completed("resp-1"), + ]), + ) + .await; + + let mut builder = test_codex().with_config(|config| { + config + .features + .enable(Feature::Sqlite) + .unwrap_or_else(|err| panic!("test config should allow feature update: {err}")); + }); + let test = builder.build(&server).await?; + let db = test.codex.state_db().expect("state db enabled"); + let thread_id = test.session_configured.session_id.to_string(); + db.create_thread_message(&codex_state::ThreadMessageCreateParams::new( + thread_id.clone(), + "external".to_string(), + "queued hello".to_string(), + /*instructions*/ None, + "{}".to_string(), + TimerDelivery::AfterTurn.as_str().to_string(), + Utc::now().timestamp(), + )) + .await?; + + test.submit_turn("start").await?; + tokio::time::sleep(Duration::from_millis(100)).await; + + assert_eq!(mock.requests().len(), 1); + assert!( + db.claim_next_thread_message( + &thread_id, /*can_after_turn*/ true, /*can_steer_current_turn*/ true, + ) + .await? + .is_some() + ); + + Ok(()) +} diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 3efd78cc56..df55295c1e 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -174,6 +174,8 @@ pub enum Feature { Artifact, /// Enable per-thread persistent timer scheduling tools and APIs. Timers, + /// Enable queued message creation and delivery. + QueuedMessages, /// Enable Fast mode selection in the TUI and request layer. FastMode, /// Enable experimental realtime voice conversation mode in the TUI. @@ -865,6 +867,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::QueuedMessages, + key: "queued_messages", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::FastMode, key: "fast_mode", diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index 257419adf9..e125106fd0 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -159,6 +159,16 @@ fn timers_are_under_development() { assert_eq!(Feature::Timers.default_enabled(), false); } +#[test] +fn queued_messages_are_under_development() { + assert_eq!( + feature_for_key("queued_messages"), + Some(Feature::QueuedMessages) + ); + assert_eq!(Feature::QueuedMessages.stage(), Stage::UnderDevelopment); + assert_eq!(Feature::QueuedMessages.default_enabled(), false); +} + #[test] fn tool_call_mcp_elicitation_is_stable_and_enabled_by_default() { assert_eq!(Feature::ToolCallMcpElicitation.stage(), Stage::Stable); diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 31fcf47240..4ffb219f0b 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -56,6 +56,7 @@ pub use recorder::RolloutRecorder; pub use recorder::RolloutRecorderParams; pub use recorder::append_rollout_item_to_path; pub use session_index::append_thread_name; +pub use session_index::find_thread_ids_by_name; pub use session_index::find_thread_meta_by_name_str; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_names_by_ids; diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index 039ce27f24..88532a4557 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -112,6 +112,48 @@ pub async fn find_thread_names_by_ids( Ok(names) } +/// Find all thread ids whose latest indexed thread name exactly matches `name`. +pub async fn find_thread_ids_by_name( + codex_home: &Path, + name: &str, +) -> std::io::Result> { + let name = name.trim(); + if name.is_empty() { + return Ok(Vec::new()); + } + let path = session_index_path(codex_home); + if !path.exists() { + return Ok(Vec::new()); + } + + let file = tokio::fs::File::open(&path).await?; + let reader = tokio::io::BufReader::new(file); + let mut lines = reader.lines(); + let mut latest_names = HashMap::::new(); + + while let Some(line) = lines.next_line().await? { + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(entry) = serde_json::from_str::(trimmed) else { + continue; + }; + let thread_name = entry.thread_name.trim(); + if thread_name.is_empty() { + continue; + } + latest_names.insert(entry.id, thread_name.to_string()); + } + + let mut ids = latest_names + .into_iter() + .filter_map(|(thread_id, thread_name)| (thread_name == name).then_some(thread_id)) + .collect::>(); + ids.sort_by_key(ToString::to_string); + Ok(ids) +} + /// Locate a recorded thread rollout and read its session metadata by thread name. /// Returns the newest indexed name that still has a readable rollout header. pub async fn find_thread_meta_by_name_str( diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index c6a539fb28..a3b7faf87a 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -265,6 +265,47 @@ async fn find_thread_names_by_ids_prefers_latest_entry() -> std::io::Result<()> Ok(()) } +#[tokio::test] +async fn find_thread_ids_by_name_uses_latest_name_per_thread() -> std::io::Result<()> { + let temp = TempDir::new()?; + let path = session_index_path(temp.path()); + let id1 = ThreadId::new(); + let id2 = ThreadId::new(); + let id3 = ThreadId::new(); + let lines = vec![ + SessionIndexEntry { + id: id1, + thread_name: "target".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id: id2, + thread_name: "target".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id: id1, + thread_name: "renamed".to_string(), + updated_at: "2024-01-02T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id: id3, + thread_name: "target".to_string(), + updated_at: "2024-01-03T00:00:00Z".to_string(), + }, + ]; + write_index(&path, &lines)?; + + let found = find_thread_ids_by_name(temp.path(), "target").await?; + let expected = { + let mut ids = vec![id2, id3]; + ids.sort_by_key(ToString::to_string); + ids + }; + assert_eq!(found, expected); + Ok(()) +} + #[test] fn scan_index_finds_latest_match_among_mixed_entries() -> std::io::Result<()> { let temp = TempDir::new()?;