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 0235a9a133..ee765cabef 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -20,8 +20,8 @@ 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::ExternalMessageCreateParams; use codex_state::StateRuntime; -use codex_state::ThreadMessageCreateParams; use codex_state::ThreadTimerCreateParams; use pretty_assertions::assert_eq; use std::path::Path; @@ -125,11 +125,11 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { .await .expect("initialize state db"); state_db - .create_thread_message(&message_params("message-1", &thread.id)) + .create_external_message(&message_params("message-1", &thread.id)) .await .expect("create archived thread message"); state_db - .create_thread_message(&message_params("message-2", "other-thread")) + .create_external_message(&message_params("message-2", "other-thread")) .await .expect("create other thread message"); state_db @@ -181,7 +181,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { ); assert_eq!( state_db - .list_thread_messages(&thread.id) + .list_external_messages(&thread.id) .await .expect("list archived thread messages"), Vec::new() @@ -195,7 +195,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { ); assert_eq!( state_db - .list_thread_messages("other-thread") + .list_external_messages("other-thread") .await .expect("list other thread messages") .into_iter() @@ -381,8 +381,8 @@ fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result Ok(()) } -fn message_params(id: &str, thread_id: &str) -> ThreadMessageCreateParams { - ThreadMessageCreateParams { +fn message_params(id: &str, thread_id: &str) -> ExternalMessageCreateParams { + ExternalMessageCreateParams { id: id.to_string(), thread_id: thread_id.to_string(), source: "external".to_string(), diff --git a/codex-rs/cli/src/queue_cmd.rs b/codex-rs/cli/src/queue_cmd.rs index 57f4488a85..bad4d2940b 100644 --- a/codex-rs/cli/src/queue_cmd.rs +++ b/codex-rs/cli/src/queue_cmd.rs @@ -47,7 +47,7 @@ pub(crate) async fn run_queue_command( StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()).await?; let delivery = TimerDelivery::AfterTurn; - let message_params = codex_state::ThreadMessageCreateParams::new( + let message_params = codex_state::ExternalMessageCreateParams::new( thread_id, "external".to_string(), cmd.message, @@ -56,7 +56,7 @@ pub(crate) async fn run_queue_command( delivery.as_str().to_string(), unix_timestamp_now()?, ); - state_db.create_thread_message(&message_params).await?; + state_db.create_external_message(&message_params).await?; remove_queued_message_if_thread_missing( config.codex_home.as_path(), &state_db, @@ -85,7 +85,7 @@ async fn remove_queued_message_if_thread_missing( } state_db - .delete_thread_message(thread_id, message_id) + .delete_external_message(thread_id, message_id) .await?; anyhow::bail!("thread `{thread_id}` was archived before queued work could be created"); } diff --git a/codex-rs/core/src/codex/timer_runtime.rs b/codex-rs/core/src/codex/timer_runtime.rs index 167d7cde4c..3a69afdfe4 100644 --- a/codex-rs/core/src/codex/timer_runtime.rs +++ b/codex-rs/core/src/codex/timer_runtime.rs @@ -348,7 +348,7 @@ impl Session { loop { let claim = match state_db - .claim_next_thread_message( + .claim_next_external_message( &self.thread_id_string(), can_after_turn, can_steer_current_turn, @@ -363,7 +363,7 @@ impl Session { } }; match claim { - Some(codex_state::ThreadMessageClaim::Claimed(row)) => { + Some(codex_state::ExternalMessageClaim::Claimed(row)) => { let (message, delivery) = match InjectedMessage::from_external_row(row) { Ok(parsed) => parsed, Err(err) => { @@ -375,11 +375,11 @@ impl Session { PendingInputItem::injected(message.prompt_input_item(), message.event()); return Some(PendingMessageClaim::Claimed(Box::new(input_item), delivery)); } - Some(codex_state::ThreadMessageClaim::Invalid { id, reason }) => { + Some(codex_state::ExternalMessageClaim::Invalid { id, reason }) => { warn!("dropped invalid queued message {id}: {reason}"); continue; } - Some(codex_state::ThreadMessageClaim::NotReady) | None => { + Some(codex_state::ExternalMessageClaim::NotReady) | None => { *self.timer_start_in_progress.lock().await = false; return claim.map(|_| PendingMessageClaim::NotReady); } diff --git a/codex-rs/core/src/injected_message.rs b/codex-rs/core/src/injected_message.rs index 6034f8de92..d7a0cc0607 100644 --- a/codex-rs/core/src/injected_message.rs +++ b/codex-rs/core/src/injected_message.rs @@ -45,7 +45,7 @@ pub(crate) enum InjectedMessage { impl InjectedMessage { pub(crate) fn from_external_row( - row: codex_state::ThreadMessage, + row: codex_state::ExternalMessage, ) -> Result<(Self, TimerDelivery), String> { let delivery = serde_json::from_value::(serde_json::Value::String( row.delivery.clone(), diff --git a/codex-rs/core/tests/suite/timers.rs b/codex-rs/core/tests/suite/timers.rs index 9723e28307..7fa5c79ad8 100644 --- a/codex-rs/core/tests/suite/timers.rs +++ b/codex-rs/core/tests/suite/timers.rs @@ -408,7 +408,7 @@ async fn queued_messages_feature_consumes_messages_without_timers() -> Result<() 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( + db.create_external_message(&codex_state::ExternalMessageCreateParams::new( thread_id, "external".to_string(), "queued hello".to_string(), @@ -515,7 +515,7 @@ async fn queued_message_runs_after_idle_recurring_timer() -> Result<()> { ) .await; let thread_id = test.session_configured.session_id.to_string(); - db.create_thread_message(&codex_state::ThreadMessageCreateParams::new( + db.create_external_message(&codex_state::ExternalMessageCreateParams::new( thread_id, "external".to_string(), "queued hello".to_string(), @@ -602,7 +602,7 @@ async fn queued_messages_feature_disabled_leaves_messages_queued() -> Result<()> 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( + db.create_external_message(&codex_state::ExternalMessageCreateParams::new( thread_id.clone(), "external".to_string(), "queued hello".to_string(), @@ -618,7 +618,7 @@ async fn queued_messages_feature_disabled_leaves_messages_queued() -> Result<()> assert_eq!(mock.requests().len(), 1); assert!( - db.claim_next_thread_message( + db.claim_next_external_message( &thread_id, /*can_after_turn*/ true, /*can_steer_current_turn*/ true, ) .await?