mirror of
https://github.com/openai/codex.git
synced 2026-10-31 19:28:49 +00:00
Split timer and queued message flags
This commit is contained in:
@@ -1,20 +1,20 @@
|
||||
//! 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, validating message
|
||||
//! metadata, and writing either immediate messages or one-shot timers into the
|
||||
//! SQLite state database.
|
||||
//! queue-specific policy for resolving target threads and writing either
|
||||
//! immediate messages or one-shot timers into the SQLite state database.
|
||||
|
||||
use clap::Parser;
|
||||
use codex_core::config::Config;
|
||||
use codex_core::config::ConfigOverrides;
|
||||
use codex_core::messages::MessagePayload;
|
||||
use codex_core::messages::validate_meta_key;
|
||||
use codex_core::timers::ThreadTimerStorageCreateParams;
|
||||
use codex_core::timers::ThreadTimerTrigger;
|
||||
use codex_core::timers::TimerDelivery;
|
||||
use codex_core::timers::build_thread_timer_create_params;
|
||||
use codex_core::timers::normalize_thread_timer_dtstart_input;
|
||||
use codex_features::Feature;
|
||||
use codex_features::Features;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_state::StateRuntime;
|
||||
use codex_tui::Cli as TuiCli;
|
||||
@@ -32,10 +32,6 @@ pub(crate) struct QueueCommand {
|
||||
#[arg(long = "message", value_name = "TEXT")]
|
||||
message: String,
|
||||
|
||||
/// Message metadata as key=value. May be repeated.
|
||||
#[arg(long = "meta", value_name = "KEY=VALUE")]
|
||||
meta: Vec<String>,
|
||||
|
||||
/// Queue a one-shot timer for a local datetime or time, e.g. 2026-04-10T09:30:00 or 09:30.
|
||||
#[arg(long = "at", value_name = "WHEN")]
|
||||
at: Option<String>,
|
||||
@@ -46,7 +42,6 @@ pub(crate) async fn run_queue_command(
|
||||
root_config_overrides: &CliConfigOverrides,
|
||||
interactive: &TuiCli,
|
||||
) -> anyhow::Result<()> {
|
||||
let meta = parse_queue_meta(&cmd.meta)?;
|
||||
let cli_kv_overrides = root_config_overrides
|
||||
.parse_overrides()
|
||||
.map_err(anyhow::Error::msg)?;
|
||||
@@ -56,6 +51,12 @@ pub(crate) async fn run_queue_command(
|
||||
};
|
||||
let config =
|
||||
Config::load_with_cli_overrides_and_harness_overrides(cli_kv_overrides, overrides).await?;
|
||||
let queue_mode = if cmd.at.is_some() {
|
||||
QueueMode::Timer
|
||||
} else {
|
||||
QueueMode::ImmediateMessage
|
||||
};
|
||||
validate_queue_feature_flags(&config.features, queue_mode)?;
|
||||
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?;
|
||||
@@ -74,7 +75,7 @@ pub(crate) async fn run_queue_command(
|
||||
payload: MessagePayload {
|
||||
content: cmd.message,
|
||||
instructions: None,
|
||||
meta,
|
||||
meta: BTreeMap::new(),
|
||||
},
|
||||
delivery,
|
||||
})
|
||||
@@ -92,13 +93,12 @@ pub(crate) async fn run_queue_command(
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let meta_json = serde_json::to_string(&meta)?;
|
||||
let message_params = codex_state::ThreadMessageCreateParams::new(
|
||||
thread_id,
|
||||
"external".to_string(),
|
||||
cmd.message,
|
||||
None,
|
||||
meta_json,
|
||||
"{}".to_string(),
|
||||
delivery.as_str().to_string(),
|
||||
unix_timestamp_now()?,
|
||||
);
|
||||
@@ -110,6 +110,22 @@ pub(crate) async fn run_queue_command(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum QueueMode {
|
||||
ImmediateMessage,
|
||||
Timer,
|
||||
}
|
||||
|
||||
fn validate_queue_feature_flags(features: &Features, mode: QueueMode) -> anyhow::Result<()> {
|
||||
if !features.enabled(Feature::QueuedMessages) {
|
||||
anyhow::bail!("codex queue requires the queued_messages feature");
|
||||
}
|
||||
if mode == QueueMode::Timer && !features.enabled(Feature::Timers) {
|
||||
anyhow::bail!("codex queue --at requires the timers feature");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn queue_timer_success_message(
|
||||
timer_id: &str,
|
||||
thread_id: &str,
|
||||
@@ -146,20 +162,6 @@ fn unix_timestamp_now() -> anyhow::Result<i64> {
|
||||
i64::try_from(duration.as_secs()).map_err(|_| anyhow::anyhow!("current time is out of range"))
|
||||
}
|
||||
|
||||
fn parse_queue_meta(entries: &[String]) -> anyhow::Result<BTreeMap<String, String>> {
|
||||
let mut meta = BTreeMap::new();
|
||||
for entry in entries {
|
||||
let Some((key, value)) = entry.split_once('=') else {
|
||||
anyhow::bail!("metadata entry `{entry}` must use key=value syntax");
|
||||
};
|
||||
validate_meta_key(key).map_err(anyhow::Error::msg)?;
|
||||
if meta.insert(key.to_string(), value.to_string()).is_some() {
|
||||
anyhow::bail!("duplicate metadata key `{key}`");
|
||||
}
|
||||
}
|
||||
Ok(meta)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -177,8 +179,6 @@ mod tests {
|
||||
"thread-1",
|
||||
"--message",
|
||||
"do work",
|
||||
"--meta",
|
||||
"ticket=ABC_123",
|
||||
])
|
||||
.expect("parse");
|
||||
let Some(Subcommand::Queue(cmd)) = cli.subcommand else {
|
||||
@@ -187,7 +187,6 @@ mod tests {
|
||||
|
||||
assert_eq!(cmd.thread, "thread-1");
|
||||
assert_eq!(cmd.message, "do work");
|
||||
assert_eq!(cmd.meta, vec!["ticket=ABC_123".to_string()]);
|
||||
assert_eq!(cmd.at, None);
|
||||
}
|
||||
|
||||
@@ -214,9 +213,20 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn queue_meta_rejects_invalid_and_duplicate_keys() {
|
||||
assert!(parse_queue_meta(&["bad-key=value".to_string()]).is_err());
|
||||
assert!(parse_queue_meta(&["ticket=one".to_string(), "ticket=two".to_string()]).is_err());
|
||||
fn queue_rejects_meta_flag() {
|
||||
assert!(
|
||||
MultitoolCli::try_parse_from([
|
||||
"codex",
|
||||
"queue",
|
||||
"--thread",
|
||||
"thread-1",
|
||||
"--message",
|
||||
"do work",
|
||||
"--meta",
|
||||
"ticket=ABC_123",
|
||||
])
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -272,6 +282,41 @@ mod tests {
|
||||
assert_eq!(cmd.at, Some("2026-04-10T12:00:00".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn queue_requires_queued_messages_feature() {
|
||||
let mut features = Features::with_defaults();
|
||||
|
||||
let err = validate_queue_feature_flags(&features, QueueMode::ImmediateMessage)
|
||||
.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, QueueMode::ImmediateMessage)
|
||||
.expect("queued messages feature should permit immediate queue command");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn queue_at_requires_timers_feature() {
|
||||
let mut features = Features::with_defaults();
|
||||
features.enable(Feature::QueuedMessages);
|
||||
|
||||
let err = validate_queue_feature_flags(&features, QueueMode::Timer)
|
||||
.expect_err("queue --at should require timers");
|
||||
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
"codex queue --at requires the timers feature"
|
||||
);
|
||||
|
||||
features.enable(Feature::Timers);
|
||||
validate_queue_feature_flags(&features, QueueMode::Timer)
|
||||
.expect("timers feature should permit queue --at");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn queued_timer_output_reports_fire_time() {
|
||||
assert_eq!(
|
||||
|
||||
@@ -434,6 +434,9 @@
|
||||
"prevent_idle_sleep": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"queued_messages": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"realtime_conversation": {
|
||||
"type": "boolean"
|
||||
},
|
||||
@@ -485,7 +488,7 @@
|
||||
"steer": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"timer_tool": {
|
||||
"timers": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"tool_call_mcp_elicitation": {
|
||||
@@ -2286,6 +2289,9 @@
|
||||
"prevent_idle_sleep": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"queued_messages": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"realtime_conversation": {
|
||||
"type": "boolean"
|
||||
},
|
||||
@@ -2337,7 +2343,7 @@
|
||||
"steer": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"timer_tool": {
|
||||
"timers": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"tool_call_mcp_elicitation": {
|
||||
|
||||
@@ -98,6 +98,9 @@ fn db_timer_to_persisted_timer(row: codex_state::ThreadTimer) -> Option<Persiste
|
||||
|
||||
impl Session {
|
||||
pub(crate) async fn list_timers(self: &Arc<Self>) -> Vec<ThreadTimer> {
|
||||
if !self.timers_feature_enabled() {
|
||||
return Vec::new();
|
||||
}
|
||||
self.sync_timers_from_db(/*emit_update*/ false).await;
|
||||
self.list_timers_from_memory().await
|
||||
}
|
||||
@@ -112,6 +115,9 @@ impl Session {
|
||||
payload: MessagePayload,
|
||||
delivery: TimerDelivery,
|
||||
) -> Result<ThreadTimer, String> {
|
||||
if !self.timers_feature_enabled() {
|
||||
return Err("timers feature is disabled".to_string());
|
||||
}
|
||||
validate_meta(&payload.meta)?;
|
||||
self.ensure_rollout_materialized().await;
|
||||
let state_db = self.timer_state_db().await?;
|
||||
@@ -155,6 +161,9 @@ impl Session {
|
||||
}
|
||||
|
||||
pub(crate) async fn delete_timer(self: &Arc<Self>, id: &str) -> Result<bool, String> {
|
||||
if !self.timers_feature_enabled() {
|
||||
return Err("timers feature is disabled".to_string());
|
||||
}
|
||||
self.sync_timers_from_db(/*emit_update*/ false).await;
|
||||
let state_db = self.timer_state_db().await?;
|
||||
self.start_timer_db_sync_task(state_db.clone());
|
||||
@@ -255,7 +264,7 @@ impl Session {
|
||||
async fn claim_next_message_for_delivery(
|
||||
self: &Arc<Self>,
|
||||
) -> Option<(PendingInputItem, TimerDelivery)> {
|
||||
if !self.features.enabled(Feature::TimerScheduler) {
|
||||
if !self.queued_messages_feature_enabled() {
|
||||
return None;
|
||||
}
|
||||
let mut timer_start_in_progress = self.timer_start_in_progress.lock().await;
|
||||
@@ -340,6 +349,9 @@ impl Session {
|
||||
}
|
||||
|
||||
async fn claim_next_timer_for_delivery(self: &Arc<Self>) -> Option<ClaimedTimer> {
|
||||
if !self.timers_feature_enabled() {
|
||||
return None;
|
||||
}
|
||||
let mut timer_start_in_progress = self.timer_start_in_progress.lock().await;
|
||||
if *timer_start_in_progress {
|
||||
return None;
|
||||
@@ -552,19 +564,21 @@ impl Session {
|
||||
}
|
||||
|
||||
pub(crate) async fn restore_timers_from_db(self: &Arc<Self>) {
|
||||
if !self.features.enabled(Feature::TimerScheduler) {
|
||||
if !self.timer_db_sync_feature_enabled() {
|
||||
return;
|
||||
}
|
||||
let Ok(state_db) = self.timer_state_db().await else {
|
||||
return;
|
||||
};
|
||||
self.start_timer_db_sync_task(state_db);
|
||||
self.sync_timers_from_db(/*emit_update*/ true).await;
|
||||
if self.timers_feature_enabled() {
|
||||
self.sync_timers_from_db(/*emit_update*/ true).await;
|
||||
}
|
||||
self.maybe_start_pending_timer().await;
|
||||
}
|
||||
|
||||
fn start_timer_db_sync_task(self: &Arc<Self>, state_db: state_db::StateDbHandle) {
|
||||
if !self.features.enabled(Feature::TimerScheduler) {
|
||||
if !self.timer_db_sync_feature_enabled() {
|
||||
return;
|
||||
}
|
||||
if self
|
||||
@@ -623,7 +637,7 @@ impl Session {
|
||||
}
|
||||
|
||||
async fn sync_timers_from_db(self: &Arc<Self>, emit_update: bool) -> bool {
|
||||
if !self.features.enabled(Feature::TimerScheduler) {
|
||||
if !self.timers_feature_enabled() {
|
||||
return false;
|
||||
}
|
||||
let Ok(state_db) = self.timer_state_db().await else {
|
||||
@@ -654,6 +668,18 @@ impl Session {
|
||||
changed
|
||||
}
|
||||
|
||||
fn timers_feature_enabled(&self) -> bool {
|
||||
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.queued_messages_feature_enabled()
|
||||
}
|
||||
|
||||
fn spawn_restored_timer_tasks(self: &Arc<Self>, restored_tasks: Vec<RestoredTimerTask>) {
|
||||
for RestoredTimerTask {
|
||||
id,
|
||||
|
||||
@@ -2,6 +2,7 @@ use super::make_session_and_context_with_rx;
|
||||
use crate::messages::MessagePayload;
|
||||
use crate::timers::ThreadTimerTrigger;
|
||||
use crate::timers::TimerDelivery;
|
||||
use codex_features::Feature;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
@@ -16,50 +17,79 @@ async fn dropping_session_cancels_timer_tasks() {
|
||||
|
||||
#[tokio::test]
|
||||
async fn maybe_start_pending_timer_claims_only_one_timer_while_start_is_in_progress() {
|
||||
let (session, _, _) = make_session_and_context_with_rx().await;
|
||||
let (mut session, _, _) = make_session_and_context_with_rx().await;
|
||||
Arc::get_mut(&mut session)
|
||||
.expect("session should have no other references")
|
||||
.features
|
||||
.enable(Feature::Timers)
|
||||
.expect("test config should allow feature update");
|
||||
let config = {
|
||||
let state = session.state.lock().await;
|
||||
state
|
||||
.session_configuration
|
||||
.original_config_do_not_use
|
||||
.clone()
|
||||
};
|
||||
let state_db = codex_state::StateRuntime::init(
|
||||
config.sqlite_home.clone(),
|
||||
config.model_provider_id.clone(),
|
||||
)
|
||||
.await
|
||||
.expect("state db should open");
|
||||
let now = chrono::Utc::now();
|
||||
let trigger = ThreadTimerTrigger::Delay {
|
||||
seconds: 10,
|
||||
repeat: Some(true),
|
||||
};
|
||||
{
|
||||
let mut timers = session.timers.lock().await;
|
||||
timers
|
||||
.create_timer(
|
||||
crate::timers::CreateTimer {
|
||||
id: "timer-1".to_string(),
|
||||
trigger: ThreadTimerTrigger::Delay {
|
||||
seconds: 10,
|
||||
repeat: Some(true),
|
||||
for (id, content) in [("timer-1", "first"), ("timer-2", "second")] {
|
||||
timers
|
||||
.create_timer(
|
||||
crate::timers::CreateTimer {
|
||||
id: id.to_string(),
|
||||
trigger: trigger.clone(),
|
||||
payload: MessagePayload {
|
||||
content: content.to_string(),
|
||||
instructions: None,
|
||||
meta: Default::default(),
|
||||
},
|
||||
delivery: TimerDelivery::AfterTurn,
|
||||
now,
|
||||
},
|
||||
payload: MessagePayload {
|
||||
content: "first".to_string(),
|
||||
instructions: None,
|
||||
meta: Default::default(),
|
||||
},
|
||||
delivery: TimerDelivery::AfterTurn,
|
||||
now,
|
||||
},
|
||||
/*timer_cancel*/ None,
|
||||
)
|
||||
.expect("first timer should be created");
|
||||
timers
|
||||
.create_timer(
|
||||
crate::timers::CreateTimer {
|
||||
id: "timer-2".to_string(),
|
||||
trigger: ThreadTimerTrigger::Delay {
|
||||
seconds: 10,
|
||||
repeat: Some(true),
|
||||
},
|
||||
payload: MessagePayload {
|
||||
content: "second".to_string(),
|
||||
instructions: None,
|
||||
meta: Default::default(),
|
||||
},
|
||||
delivery: TimerDelivery::AfterTurn,
|
||||
now,
|
||||
},
|
||||
/*timer_cancel*/ None,
|
||||
)
|
||||
.expect("second timer should be created");
|
||||
timers.mark_timer_due("timer-1", now);
|
||||
timers.mark_timer_due("timer-2", now);
|
||||
/*timer_cancel*/ None,
|
||||
)
|
||||
.expect("timer should be created");
|
||||
timers.mark_timer_due(id, now);
|
||||
}
|
||||
}
|
||||
for timer_id in ["timer-1", "timer-2"] {
|
||||
let persisted_timer = session
|
||||
.timers
|
||||
.lock()
|
||||
.await
|
||||
.persisted_timer(timer_id)
|
||||
.expect("timer should be in memory");
|
||||
let timer = persisted_timer.timer;
|
||||
state_db
|
||||
.create_thread_timer(&codex_state::ThreadTimerCreateParams {
|
||||
id: timer.id,
|
||||
thread_id: session.conversation_id.to_string(),
|
||||
source: "agent".to_string(),
|
||||
client_id: "codex-cli".to_string(),
|
||||
trigger_json: serde_json::to_string(&timer.trigger)
|
||||
.expect("trigger should serialize"),
|
||||
content: timer.content,
|
||||
instructions: timer.instructions,
|
||||
meta_json: serde_json::to_string(&timer.meta).expect("metadata should serialize"),
|
||||
delivery: timer.delivery.as_str().to_string(),
|
||||
created_at: timer.created_at,
|
||||
next_run_at: timer.next_run_at,
|
||||
last_run_at: timer.last_run_at,
|
||||
pending_run: persisted_timer.pending_run,
|
||||
})
|
||||
.await
|
||||
.expect("timer should be persisted");
|
||||
}
|
||||
|
||||
let first = Arc::clone(&session);
|
||||
|
||||
@@ -21,8 +21,6 @@ use crate::tools::registry::ToolKind;
|
||||
struct CreateTimerArgs {
|
||||
trigger: ThreadTimerTrigger,
|
||||
content: Option<String>,
|
||||
prompt: Option<String>,
|
||||
instructions: Option<String>,
|
||||
#[serde(default)]
|
||||
meta: BTreeMap<String, String>,
|
||||
delivery: TimerDelivery,
|
||||
@@ -49,7 +47,7 @@ impl ToolHandler for CreateTimerHandler {
|
||||
));
|
||||
};
|
||||
let args: CreateTimerArgs = parse_arguments(&arguments)?;
|
||||
let content = args.content.or(args.prompt).ok_or_else(|| {
|
||||
let content = args.content.ok_or_else(|| {
|
||||
FunctionCallError::RespondToModel("create_timer requires `content`".to_string())
|
||||
})?;
|
||||
let timer = invocation
|
||||
@@ -58,7 +56,7 @@ impl ToolHandler for CreateTimerHandler {
|
||||
args.trigger,
|
||||
MessagePayload {
|
||||
content,
|
||||
instructions: args.instructions,
|
||||
instructions: None,
|
||||
meta: args.meta,
|
||||
},
|
||||
args.delivery,
|
||||
|
||||
@@ -12,12 +12,15 @@ use core_test_support::responses::ev_assistant_message;
|
||||
use core_test_support::responses::ev_completed;
|
||||
use core_test_support::responses::ev_response_created;
|
||||
use core_test_support::responses::mount_sse_once;
|
||||
use core_test_support::responses::mount_sse_sequence;
|
||||
use core_test_support::responses::sse;
|
||||
use core_test_support::responses::start_mock_server;
|
||||
use core_test_support::test_codex::test_codex;
|
||||
use core_test_support::wait_for_event;
|
||||
use core_test_support::wait_for_event_match;
|
||||
use core_test_support::wait_for_event_with_timeout;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn create_timer_emits_fired_background_event_when_timer_starts() -> Result<()> {
|
||||
@@ -44,7 +47,7 @@ async fn assert_after_turn_timer_starts_and_emits_fired_event() -> Result<()> {
|
||||
let mut builder = test_codex().with_config(|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::TimerScheduler)
|
||||
.enable(Feature::Timers)
|
||||
.unwrap_or_else(|err| panic!("test config should allow feature update: {err}"));
|
||||
config
|
||||
.features
|
||||
@@ -107,7 +110,7 @@ async fn create_timer_persists_source_and_client_metadata() -> Result<()> {
|
||||
let mut builder = test_codex().with_config(|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::TimerScheduler)
|
||||
.enable(Feature::Timers)
|
||||
.unwrap_or_else(|err| panic!("test config should allow feature update: {err}"));
|
||||
config
|
||||
.features
|
||||
@@ -153,7 +156,7 @@ async fn create_timer_lazily_opens_sqlite_for_ephemeral_thread() -> Result<()> {
|
||||
config.ephemeral = true;
|
||||
config
|
||||
.features
|
||||
.enable(Feature::TimerScheduler)
|
||||
.enable(Feature::Timers)
|
||||
.unwrap_or_else(|err| panic!("test config should allow feature update: {err}"));
|
||||
});
|
||||
let test = builder.build(&server).await?;
|
||||
@@ -190,7 +193,7 @@ async fn list_timers_discovers_externally_inserted_timer() -> Result<()> {
|
||||
let mut builder = test_codex().with_config(|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::TimerScheduler)
|
||||
.enable(Feature::Timers)
|
||||
.unwrap_or_else(|err| panic!("test config should allow feature update: {err}"));
|
||||
config
|
||||
.features
|
||||
@@ -236,3 +239,119 @@ async fn list_timers_discovers_externally_inserted_timer() -> Result<()> {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[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_sequence(
|
||||
&server,
|
||||
vec![
|
||||
sse(vec![
|
||||
ev_response_created("resp-1"),
|
||||
ev_assistant_message("msg-1", "first 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::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(),
|
||||
None,
|
||||
"{}".to_string(),
|
||||
TimerDelivery::AfterTurn.as_str().to_string(),
|
||||
Utc::now().timestamp(),
|
||||
))
|
||||
.await?;
|
||||
|
||||
test.submit_turn("start").await?;
|
||||
wait_for_event_with_timeout(
|
||||
&test.codex,
|
||||
|event| matches!(event, EventMsg::InjectedMessage(_)),
|
||||
Duration::from_secs(20),
|
||||
)
|
||||
.await;
|
||||
wait_for_event_with_timeout(
|
||||
&test.codex,
|
||||
|event| matches!(event, EventMsg::TurnComplete(_)),
|
||||
Duration::from_secs(20),
|
||||
)
|
||||
.await;
|
||||
|
||||
let requests = mock.requests();
|
||||
assert_eq!(requests.len(), 2);
|
||||
assert!(
|
||||
requests[1]
|
||||
.message_input_texts("user")
|
||||
.iter()
|
||||
.any(|message| message.contains("<content>\nqueued hello\n</content>"))
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[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(),
|
||||
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, true, true)
|
||||
.await?
|
||||
.is_some()
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -173,7 +173,9 @@ pub enum Feature {
|
||||
/// Enable native artifact tools.
|
||||
Artifact,
|
||||
/// Enable per-thread persistent timer scheduling tools and APIs.
|
||||
TimerScheduler,
|
||||
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.
|
||||
@@ -860,8 +862,14 @@ pub const FEATURES: &[FeatureSpec] = &[
|
||||
default_enabled: false,
|
||||
},
|
||||
FeatureSpec {
|
||||
id: Feature::TimerScheduler,
|
||||
key: "timer_tool",
|
||||
id: Feature::Timers,
|
||||
key: "timers",
|
||||
stage: Stage::UnderDevelopment,
|
||||
default_enabled: false,
|
||||
},
|
||||
FeatureSpec {
|
||||
id: Feature::QueuedMessages,
|
||||
key: "queued_messages",
|
||||
stage: Stage::UnderDevelopment,
|
||||
default_enabled: false,
|
||||
},
|
||||
|
||||
@@ -181,6 +181,20 @@ fn image_detail_original_feature_is_experimental_and_user_toggleable() {
|
||||
assert_eq!(Feature::ImageDetailOriginal.default_enabled(), false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timers_and_queued_messages_are_under_development() {
|
||||
assert_eq!(feature_for_key("timers"), Some(Feature::Timers));
|
||||
assert_eq!(
|
||||
feature_for_key("queued_messages"),
|
||||
Some(Feature::QueuedMessages)
|
||||
);
|
||||
assert_eq!(feature_for_key("timer_tool"), None);
|
||||
assert_eq!(Feature::Timers.stage(), Stage::UnderDevelopment);
|
||||
assert_eq!(Feature::Timers.default_enabled(), false);
|
||||
assert_eq!(Feature::QueuedMessages.stage(), Stage::UnderDevelopment);
|
||||
assert_eq!(Feature::QueuedMessages.default_enabled(), false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn collab_is_legacy_alias_for_multi_agent() {
|
||||
assert_eq!(feature_for_key("multi_agent"), Some(Feature::Collab));
|
||||
|
||||
@@ -56,12 +56,6 @@ pub fn create_timer_tool() -> ToolSpec {
|
||||
"Message content to execute when the timer fires.".to_string(),
|
||||
)),
|
||||
),
|
||||
(
|
||||
"instructions".to_string(),
|
||||
JsonSchema::string(Some(
|
||||
"Optional extra instructions for the timer message.".to_string(),
|
||||
)),
|
||||
),
|
||||
(
|
||||
"meta".to_string(),
|
||||
JsonSchema::object(BTreeMap::new(), None, Some(true.into())),
|
||||
|
||||
@@ -14,6 +14,18 @@ fn timer_create_tool_uses_expected_name() {
|
||||
assert_eq!(name, "create_timer");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timer_create_tool_exposes_only_core_timer_payload_fields() {
|
||||
let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = create_timer_tool() else {
|
||||
panic!("expected function tool");
|
||||
};
|
||||
let properties = parameters.properties.expect("create_timer properties");
|
||||
|
||||
assert!(properties.contains_key("content"));
|
||||
assert!(!properties.contains_key("prompt"));
|
||||
assert!(!properties.contains_key("instructions"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timer_delete_tool_uses_expected_name() {
|
||||
let ToolSpec::Function(ResponsesApiTool { name, .. }) = create_delete_timer_tool() else {
|
||||
|
||||
@@ -147,7 +147,7 @@ impl ToolsConfig {
|
||||
let include_collab_tools = features.enabled(Feature::Collab);
|
||||
let include_multi_agent_v2 = features.enabled(Feature::MultiAgentV2);
|
||||
let include_agent_jobs = features.enabled(Feature::SpawnCsv);
|
||||
let include_timer_scheduler = features.enabled(Feature::TimerScheduler);
|
||||
let include_timers = features.enabled(Feature::Timers);
|
||||
let include_default_mode_request_user_input =
|
||||
features.enabled(Feature::DefaultModeRequestUserInput);
|
||||
let include_search_tool =
|
||||
@@ -228,7 +228,7 @@ impl ToolsConfig {
|
||||
collab_tools: include_collab_tools,
|
||||
multi_agent_v2: include_multi_agent_v2,
|
||||
hide_spawn_agent_metadata: false,
|
||||
timer_scheduler: include_timer_scheduler,
|
||||
timer_scheduler: include_timers,
|
||||
spawn_agent_usage_hint: true,
|
||||
spawn_agent_usage_hint_text: None,
|
||||
default_mode_request_user_input: include_default_mode_request_user_input,
|
||||
|
||||
Reference in New Issue
Block a user