From 70b7a3700c91f5651e5f2c93b801d10e1c40cebe Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sun, 12 Apr 2026 19:09:48 -0700 Subject: [PATCH] codex: address PR review feedback (#17581) --- ...essages.sql => 0026_external_messages.sql} | 10 +- codex-rs/state/src/lib.rs | 6 +- ...{thread_message.rs => external_message.rs} | 16 +-- codex-rs/state/src/model/mod.rs | 10 +- codex-rs/state/src/runtime.rs | 8 +- codex-rs/state/src/runtime/delivery_state.rs | 20 +-- .../{messages.rs => external_messages.rs} | 132 +++++++++--------- 7 files changed, 102 insertions(+), 100 deletions(-) rename codex-rs/state/migrations/{0026_thread_messages.sql => 0026_external_messages.sql} (85%) rename codex-rs/state/src/model/{thread_message.rs => external_message.rs} (84%) rename codex-rs/state/src/runtime/{messages.rs => external_messages.rs} (74%) diff --git a/codex-rs/state/migrations/0026_thread_messages.sql b/codex-rs/state/migrations/0026_external_messages.sql similarity index 85% rename from codex-rs/state/migrations/0026_thread_messages.sql rename to codex-rs/state/migrations/0026_external_messages.sql index 0ecf1ed2dd..1f30cec8c1 100644 --- a/codex-rs/state/migrations/0026_thread_messages.sql +++ b/codex-rs/state/migrations/0026_external_messages.sql @@ -58,7 +58,7 @@ CREATE INDEX idx_thread_timers_thread_pending CREATE INDEX idx_thread_timers_thread_next_run ON thread_timers(thread_id, next_run_at); -CREATE TABLE thread_messages ( +CREATE TABLE external_messages ( seq INTEGER PRIMARY KEY, id TEXT NOT NULL UNIQUE, thread_id TEXT NOT NULL, @@ -70,8 +70,8 @@ CREATE TABLE thread_messages ( queued_at INTEGER NOT NULL ); -CREATE INDEX thread_messages_thread_order_idx - ON thread_messages(thread_id, queued_at, seq); +CREATE INDEX external_messages_thread_order_idx + ON external_messages(thread_id, queued_at, seq); -CREATE INDEX thread_messages_thread_delivery_order_idx - ON thread_messages(thread_id, delivery, queued_at, seq); +CREATE INDEX external_messages_thread_delivery_order_idx + ON external_messages(thread_id, delivery, queued_at, seq); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 8d4a9cd2a6..f9a1394ade 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -36,6 +36,9 @@ pub use model::BackfillState; pub use model::BackfillStats; pub use model::BackfillStatus; pub use model::DirectionalThreadSpawnEdgeStatus; +pub use model::ExternalMessage; +pub use model::ExternalMessageClaim; +pub use model::ExternalMessageCreateParams; pub use model::ExtractionOutcome; pub use model::SortKey; pub use model::Stage1JobClaim; @@ -43,9 +46,6 @@ pub use model::Stage1JobClaimOutcome; pub use model::Stage1Output; pub use model::Stage1OutputRef; pub use model::Stage1StartupClaimParams; -pub use model::ThreadMessage; -pub use model::ThreadMessageClaim; -pub use model::ThreadMessageCreateParams; pub use model::ThreadMetadata; pub use model::ThreadMetadataBuilder; pub use model::ThreadTimer; diff --git a/codex-rs/state/src/model/thread_message.rs b/codex-rs/state/src/model/external_message.rs similarity index 84% rename from codex-rs/state/src/model/thread_message.rs rename to codex-rs/state/src/model/external_message.rs index 9cd3f2461b..f7683c0fbe 100644 --- a/codex-rs/state/src/model/thread_message.rs +++ b/codex-rs/state/src/model/external_message.rs @@ -1,7 +1,7 @@ use sqlx::FromRow; #[derive(Clone, Debug, PartialEq, Eq)] -pub struct ThreadMessageCreateParams { +pub struct ExternalMessageCreateParams { pub id: String, pub thread_id: String, pub source: String, @@ -12,7 +12,7 @@ pub struct ThreadMessageCreateParams { pub queued_at: i64, } -impl ThreadMessageCreateParams { +impl ExternalMessageCreateParams { pub fn new( thread_id: String, source: String, @@ -36,7 +36,7 @@ impl ThreadMessageCreateParams { } #[derive(Clone, Debug, PartialEq, Eq)] -pub struct ThreadMessage { +pub struct ExternalMessage { pub seq: i64, pub id: String, pub thread_id: String, @@ -49,14 +49,14 @@ pub struct ThreadMessage { } #[derive(Clone, Debug, PartialEq, Eq)] -pub enum ThreadMessageClaim { - Claimed(ThreadMessage), +pub enum ExternalMessageClaim { + Claimed(ExternalMessage), Invalid { id: String, reason: String }, NotReady, } #[derive(Debug, FromRow)] -pub(crate) struct ThreadMessageRow { +pub(crate) struct ExternalMessageRow { pub seq: i64, pub id: String, pub thread_id: String, @@ -68,8 +68,8 @@ pub(crate) struct ThreadMessageRow { pub queued_at: i64, } -impl From for ThreadMessage { - fn from(row: ThreadMessageRow) -> Self { +impl From for ExternalMessage { + fn from(row: ExternalMessageRow) -> Self { Self { seq: row.seq, id: row.id, diff --git a/codex-rs/state/src/model/mod.rs b/codex-rs/state/src/model/mod.rs index 9f5318d923..c7684ef930 100644 --- a/codex-rs/state/src/model/mod.rs +++ b/codex-rs/state/src/model/mod.rs @@ -1,9 +1,9 @@ mod agent_job; mod backfill_state; +mod external_message; mod graph; mod log; mod memories; -mod thread_message; mod thread_metadata; mod thread_timer; @@ -16,6 +16,9 @@ pub use agent_job::AgentJobProgress; pub use agent_job::AgentJobStatus; pub use backfill_state::BackfillState; pub use backfill_state::BackfillStatus; +pub use external_message::ExternalMessage; +pub use external_message::ExternalMessageClaim; +pub use external_message::ExternalMessageCreateParams; pub use graph::DirectionalThreadSpawnEdgeStatus; pub use log::LogEntry; pub use log::LogQuery; @@ -27,9 +30,6 @@ pub use memories::Stage1JobClaimOutcome; pub use memories::Stage1Output; pub use memories::Stage1OutputRef; pub use memories::Stage1StartupClaimParams; -pub use thread_message::ThreadMessage; -pub use thread_message::ThreadMessageClaim; -pub use thread_message::ThreadMessageCreateParams; pub use thread_metadata::Anchor; pub use thread_metadata::BackfillStats; pub use thread_metadata::ExtractionOutcome; @@ -43,9 +43,9 @@ pub use thread_timer::ThreadTimerUpdateParams; pub(crate) use agent_job::AgentJobItemRow; pub(crate) use agent_job::AgentJobRow; +pub(crate) use external_message::ExternalMessageRow; pub(crate) use memories::Stage1OutputRow; pub(crate) use memories::stage1_output_ref_from_parts; -pub(crate) use thread_message::ThreadMessageRow; pub(crate) use thread_metadata::ThreadRow; pub(crate) use thread_metadata::anchor_from_item; pub(crate) use thread_metadata::datetime_to_epoch_seconds; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index bf494bf6a5..93dc3be7fc 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -5,6 +5,9 @@ use crate::AgentJobItemCreateParams; use crate::AgentJobItemStatus; use crate::AgentJobProgress; use crate::AgentJobStatus; +use crate::ExternalMessage; +use crate::ExternalMessageClaim; +use crate::ExternalMessageCreateParams; use crate::LOGS_DB_FILENAME; use crate::LOGS_DB_VERSION; use crate::LogEntry; @@ -13,9 +16,6 @@ use crate::LogRow; use crate::STATE_DB_FILENAME; use crate::STATE_DB_VERSION; use crate::SortKey; -use crate::ThreadMessage; -use crate::ThreadMessageClaim; -use crate::ThreadMessageCreateParams; use crate::ThreadMetadata; use crate::ThreadMetadataBuilder; use crate::ThreadTimer; @@ -59,9 +59,9 @@ use tracing::warn; mod agent_jobs; mod backfill; mod delivery_state; +mod external_messages; mod logs; mod memories; -mod messages; mod remote_control; #[cfg(test)] mod test_support; diff --git a/codex-rs/state/src/runtime/delivery_state.rs b/codex-rs/state/src/runtime/delivery_state.rs index a78f821d87..e28cb757e8 100644 --- a/codex-rs/state/src/runtime/delivery_state.rs +++ b/codex-rs/state/src/runtime/delivery_state.rs @@ -1,16 +1,16 @@ //! Cleanup operations for per-thread delivery state. //! -//! Timers and queued messages are stored independently because they have +//! Timers and queued external messages are stored independently because they have //! different runtime behavior, but thread lifecycle operations need to treat //! them as one unit. This module owns that cross-table cleanup. use super::*; impl StateRuntime { - /// Delete all queued messages and timers associated with `thread_id`. + /// Delete all queued external messages and timers associated with `thread_id`. pub async fn delete_thread_delivery_state(&self, thread_id: &str) -> anyhow::Result<()> { let mut tx = self.pool.begin().await?; - sqlx::query("DELETE FROM thread_messages WHERE thread_id = ?") + sqlx::query("DELETE FROM external_messages WHERE thread_id = ?") .bind(thread_id) .execute(&mut *tx) .await?; @@ -27,12 +27,12 @@ impl StateRuntime { mod tests { use super::StateRuntime; use super::test_support::unique_temp_dir; - use crate::ThreadMessageCreateParams; + use crate::ExternalMessageCreateParams; use crate::ThreadTimerCreateParams; use pretty_assertions::assert_eq; - 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(), @@ -72,11 +72,11 @@ mod tests { async fn delete_thread_delivery_state_removes_messages_and_timers_for_thread() { let runtime = test_runtime().await; runtime - .create_thread_message(&message_params("message-1", "thread-1")) + .create_external_message(&message_params("message-1", "thread-1")) .await .expect("create thread-1 message"); runtime - .create_thread_message(&message_params("message-2", "thread-2")) + .create_external_message(&message_params("message-2", "thread-2")) .await .expect("create thread-2 message"); runtime @@ -95,7 +95,7 @@ mod tests { assert_eq!( runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list thread-1 messages"), Vec::new() @@ -109,7 +109,7 @@ mod tests { ); assert_eq!( runtime - .list_thread_messages("thread-2") + .list_external_messages("thread-2") .await .expect("list thread-2 messages") .into_iter() diff --git a/codex-rs/state/src/runtime/messages.rs b/codex-rs/state/src/runtime/external_messages.rs similarity index 74% rename from codex-rs/state/src/runtime/messages.rs rename to codex-rs/state/src/runtime/external_messages.rs index efe1e9b278..451b7c60b9 100644 --- a/codex-rs/state/src/runtime/messages.rs +++ b/codex-rs/state/src/runtime/external_messages.rs @@ -1,4 +1,4 @@ -//! SQLite-backed state operations for queued thread messages. +//! SQLite-backed state operations for queued external messages. //! //! This module extends [`StateRuntime`] with the storage APIs used by message //! producers and active threads. Claiming a message deletes the row inside the @@ -6,19 +6,19 @@ //! once. use super::*; -use crate::model::ThreadMessageRow; +use crate::model::ExternalMessageRow; const DELIVERY_AFTER_TURN: &str = "after-turn"; const DELIVERY_STEER_CURRENT_TURN: &str = "steer-current-turn"; impl StateRuntime { - pub async fn create_thread_message( + pub async fn create_external_message( &self, - params: &ThreadMessageCreateParams, + params: &ExternalMessageCreateParams, ) -> anyhow::Result<()> { sqlx::query( r#" -INSERT INTO thread_messages ( +INSERT INTO external_messages ( id, thread_id, source, @@ -43,11 +43,11 @@ INSERT INTO thread_messages ( Ok(()) } - pub async fn list_thread_messages( + pub async fn list_external_messages( &self, thread_id: &str, - ) -> anyhow::Result> { - let rows = sqlx::query_as::<_, ThreadMessageRow>( + ) -> anyhow::Result> { + let rows = sqlx::query_as::<_, ExternalMessageRow>( r#" SELECT seq, @@ -59,7 +59,7 @@ SELECT meta_json, delivery, queued_at -FROM thread_messages +FROM external_messages WHERE thread_id = ? ORDER BY queued_at ASC, seq ASC "#, @@ -67,11 +67,11 @@ ORDER BY queued_at ASC, seq ASC .bind(thread_id) .fetch_all(self.pool.as_ref()) .await?; - Ok(rows.into_iter().map(ThreadMessage::from).collect()) + Ok(rows.into_iter().map(ExternalMessage::from).collect()) } - pub async fn delete_thread_message(&self, thread_id: &str, id: &str) -> anyhow::Result { - let result = sqlx::query("DELETE FROM thread_messages WHERE thread_id = ? AND id = ?") + pub async fn delete_external_message(&self, thread_id: &str, id: &str) -> anyhow::Result { + let result = sqlx::query("DELETE FROM external_messages WHERE thread_id = ? AND id = ?") .bind(thread_id) .bind(id) .execute(self.pool.as_ref()) @@ -79,18 +79,18 @@ ORDER BY queued_at ASC, seq ASC Ok(result.rows_affected() > 0) } - pub async fn claim_next_thread_message( + pub async fn claim_next_external_message( &self, thread_id: &str, can_after_turn: bool, can_steer_current_turn: bool, - ) -> anyhow::Result> { - let row = sqlx::query_as::<_, ThreadMessageRow>( + ) -> anyhow::Result> { + let row = sqlx::query_as::<_, ExternalMessageRow>( r#" -DELETE FROM thread_messages +DELETE FROM external_messages WHERE seq = ( SELECT seq - FROM thread_messages + FROM external_messages WHERE thread_id = ? ORDER BY queued_at ASC, seq ASC LIMIT 1 @@ -124,10 +124,10 @@ RETURNING if let Some(row) = row { return match row.delivery.as_str() { - DELIVERY_AFTER_TURN | DELIVERY_STEER_CURRENT_TURN => { - Ok(Some(ThreadMessageClaim::Claimed(ThreadMessage::from(row)))) - } - delivery => Ok(Some(ThreadMessageClaim::Invalid { + DELIVERY_AFTER_TURN | DELIVERY_STEER_CURRENT_TURN => Ok(Some( + ExternalMessageClaim::Claimed(ExternalMessage::from(row)), + )), + delivery => Ok(Some(ExternalMessageClaim::Invalid { id: row.id, reason: format!("invalid delivery `{delivery}`"), })), @@ -137,7 +137,7 @@ RETURNING let oldest_delivery = sqlx::query_scalar::<_, String>( r#" SELECT delivery -FROM thread_messages +FROM external_messages WHERE thread_id = ? ORDER BY queued_at ASC, seq ASC LIMIT 1 @@ -148,9 +148,11 @@ LIMIT 1 .await?; match oldest_delivery.as_deref() { - Some(DELIVERY_AFTER_TURN) if !can_after_turn => Ok(Some(ThreadMessageClaim::NotReady)), + Some(DELIVERY_AFTER_TURN) if !can_after_turn => { + Ok(Some(ExternalMessageClaim::NotReady)) + } Some(DELIVERY_STEER_CURRENT_TURN) if !(can_steer_current_turn || can_after_turn) => { - Ok(Some(ThreadMessageClaim::NotReady)) + Ok(Some(ExternalMessageClaim::NotReady)) } None | Some(_) => Ok(None), } @@ -161,12 +163,12 @@ LIMIT 1 mod tests { use super::StateRuntime; use super::test_support::unique_temp_dir; - use crate::ThreadMessageClaim; - use crate::ThreadMessageCreateParams; + use crate::ExternalMessageClaim; + use crate::ExternalMessageCreateParams; use pretty_assertions::assert_eq; - fn message_params(id: &str, thread_id: &str, queued_at: i64) -> ThreadMessageCreateParams { - ThreadMessageCreateParams { + fn message_params(id: &str, thread_id: &str, queued_at: i64) -> ExternalMessageCreateParams { + ExternalMessageCreateParams { id: id.to_string(), thread_id: thread_id.to_string(), source: "external".to_string(), @@ -185,13 +187,13 @@ mod tests { } #[tokio::test] - async fn thread_messages_table_and_indexes_exist() { + async fn external_messages_table_and_indexes_exist() { let runtime = test_runtime().await; let names = sqlx::query_scalar::<_, String>( r#" SELECT name FROM sqlite_master -WHERE tbl_name = 'thread_messages' +WHERE tbl_name = 'external_messages' AND name NOT LIKE 'sqlite_autoindex_%' ORDER BY name "#, @@ -203,24 +205,24 @@ ORDER BY name assert_eq!( names, vec![ - "thread_messages", - "thread_messages_thread_delivery_order_idx", - "thread_messages_thread_order_idx", + "external_messages", + "external_messages_thread_delivery_order_idx", + "external_messages_thread_order_idx", ] ); } #[tokio::test] - async fn thread_message_rows_round_trip() { + async fn external_message_rows_round_trip() { let runtime = test_runtime().await; let params = message_params("message-1", "thread-1", /*queued_at*/ 100); runtime - .create_thread_message(¶ms) + .create_external_message(¶ms) .await .expect("create message"); let messages = runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list messages"); @@ -237,10 +239,10 @@ ORDER BY name } #[tokio::test] - async fn delete_thread_message_is_scoped_to_thread_id() { + async fn delete_external_message_is_scoped_to_thread_id() { let runtime = test_runtime().await; runtime - .create_thread_message(&message_params( + .create_external_message(&message_params( "message-1", "thread-1", /*queued_at*/ 100, @@ -248,7 +250,7 @@ ORDER BY name .await .expect("create thread-1 message"); runtime - .create_thread_message(&message_params( + .create_external_message(&message_params( "message-2", "thread-2", /*queued_at*/ 100, @@ -257,25 +259,25 @@ ORDER BY name .expect("create thread-2 message"); let deleted_wrong_thread = runtime - .delete_thread_message("thread-2", "message-1") + .delete_external_message("thread-2", "message-1") .await - .expect("delete wrong-thread message"); + .expect("delete wrong-external message"); assert!(!deleted_wrong_thread); let deleted = runtime - .delete_thread_message("thread-1", "message-1") + .delete_external_message("thread-1", "message-1") .await .expect("delete thread-1 message"); assert!(deleted); assert_eq!( runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list thread-1 messages"), Vec::new() ); assert_eq!( runtime - .list_thread_messages("thread-2") + .list_external_messages("thread-2") .await .expect("list thread-2 messages") .into_iter() @@ -289,30 +291,30 @@ ORDER BY name async fn claim_is_scoped_to_thread_id_and_ordered() { let runtime = test_runtime().await; runtime - .create_thread_message(&message_params("newer", "thread-1", /*queued_at*/ 200)) + .create_external_message(&message_params("newer", "thread-1", /*queued_at*/ 200)) .await .expect("create newer message"); runtime - .create_thread_message(&message_params( + .create_external_message(&message_params( "other-thread", "thread-2", /*queued_at*/ 50, )) .await - .expect("create other thread message"); + .expect("create other external message"); runtime - .create_thread_message(&message_params("older", "thread-1", /*queued_at*/ 100)) + .create_external_message(&message_params("older", "thread-1", /*queued_at*/ 100)) .await .expect("create older message"); let claim = runtime - .claim_next_thread_message( + .claim_next_external_message( "thread-1", /*can_after_turn*/ true, /*can_steer_current_turn*/ true, ) .await .expect("claim message"); - let Some(ThreadMessageClaim::Claimed(claimed)) = claim else { + let Some(ExternalMessageClaim::Claimed(claimed)) = claim else { panic!("expected claimed message"); }; assert_eq!(claimed.id, "older"); @@ -320,7 +322,7 @@ ORDER BY name assert_eq!(claimed.queued_at, 100); assert_eq!( runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list remaining thread-1 messages") .into_iter() @@ -330,7 +332,7 @@ ORDER BY name ); assert_eq!( runtime - .list_thread_messages("thread-2") + .list_external_messages("thread-2") .await .expect("list thread-2 messages") .into_iter() @@ -344,7 +346,7 @@ ORDER BY name async fn claim_consumes_message_once() { let runtime = test_runtime().await; runtime - .create_thread_message(&message_params( + .create_external_message(&message_params( "message-1", "thread-1", /*queued_at*/ 100, @@ -354,16 +356,16 @@ ORDER BY name assert!(matches!( runtime - .claim_next_thread_message( + .claim_next_external_message( "thread-1", /*can_after_turn*/ true, /*can_steer_current_turn*/ true, ) .await .expect("claim message"), - Some(ThreadMessageClaim::Claimed(_)) + Some(ExternalMessageClaim::Claimed(_)) )); assert_eq!( runtime - .claim_next_thread_message( + .claim_next_external_message( "thread-1", /*can_after_turn*/ true, /*can_steer_current_turn*/ true, ) .await @@ -378,27 +380,27 @@ ORDER BY name let mut steer = message_params("steer", "thread-1", /*queued_at*/ 100); steer.delivery = "steer-current-turn".to_string(); runtime - .create_thread_message(&steer) + .create_external_message(&steer) .await .expect("create steer message"); runtime - .create_thread_message(&message_params("after", "thread-1", /*queued_at*/ 200)) + .create_external_message(&message_params("after", "thread-1", /*queued_at*/ 200)) .await .expect("create after-turn message"); assert_eq!( runtime - .claim_next_thread_message( + .claim_next_external_message( "thread-1", /*can_after_turn*/ false, /*can_steer_current_turn*/ false, ) .await .expect("claim message"), - Some(ThreadMessageClaim::NotReady) + Some(ExternalMessageClaim::NotReady) ); assert_eq!( runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list messages") .into_iter() @@ -414,25 +416,25 @@ ORDER BY name let mut params = message_params("bad", "thread-1", /*queued_at*/ 100); params.delivery = "bad-delivery".to_string(); runtime - .create_thread_message(¶ms) + .create_external_message(¶ms) .await .expect("create message"); assert_eq!( runtime - .claim_next_thread_message( + .claim_next_external_message( "thread-1", /*can_after_turn*/ true, /*can_steer_current_turn*/ true, ) .await .expect("claim message"), - Some(ThreadMessageClaim::Invalid { + Some(ExternalMessageClaim::Invalid { id: "bad".to_string(), reason: "invalid delivery `bad-delivery`".to_string(), }) ); assert!( runtime - .list_thread_messages("thread-1") + .list_external_messages("thread-1") .await .expect("list messages") .is_empty()