diff --git a/codex-rs/state/migrations/0060_thread_read_state.sql b/codex-rs/state/migrations/0060_thread_read_state.sql new file mode 100644 index 0000000000..9d80bcd76e --- /dev/null +++ b/codex-rs/state/migrations/0060_thread_read_state.sql @@ -0,0 +1,8 @@ +-- Independent of rebuildable thread metadata: only actual thread deletion removes receipts. +-- NULL means read; the empty string means manually unread from the start, even when empty. +CREATE TABLE thread_read_receipts ( + thread_id TEXT PRIMARY KEY NOT NULL, + first_unread_turn TEXT, + revision TEXT NOT NULL, + last_published_turn TEXT +); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index cff7792c93..d1de678a86 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -90,10 +90,13 @@ pub use runtime::GoalAccountingOutcome; pub use runtime::GoalStore; pub use runtime::GoalUpdate; pub use runtime::MemoryStore; +pub use runtime::ReadStateOperation; +pub use runtime::ReadStateUpdate; pub use runtime::RemoteControlEnrollmentRecord; pub use runtime::SqliteIntegrityCheck; pub use runtime::SqliteQueueStore; pub use runtime::ThreadFilterOptions; +pub use runtime::ThreadReadState; pub use runtime::backup_runtime_db_for_fresh_start; pub use runtime::collect_runtime_db_backups; pub use runtime::is_sqlite_corruption_error; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 92f2736423..244f219cd5 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -57,6 +57,7 @@ mod rollout_migration; pub(crate) mod test_support; mod thread_attachments; mod thread_metadata; +mod thread_read_state; mod thread_section_order; mod thread_sections; mod threads; @@ -81,6 +82,9 @@ pub use recovery::is_sqlite_corruption_error; pub use recovery::runtime_db_path_for_corruption_error; pub use recovery::sqlite_error_detail_is_lock; pub use remote_control::RemoteControlEnrollmentRecord; +pub use thread_read_state::ReadStateOperation; +pub use thread_read_state::ReadStateUpdate; +pub use thread_read_state::ThreadReadState; pub use threads::ThreadFilterOptions; // "Partition" is the retained-log-content bucket we cap at 10 MiB: diff --git a/codex-rs/state/src/runtime/thread_read_state.rs b/codex-rs/state/src/runtime/thread_read_state.rs new file mode 100644 index 0000000000..31bb985a0a --- /dev/null +++ b/codex-rs/state/src/runtime/thread_read_state.rs @@ -0,0 +1,144 @@ +//! One durable unread position per local thread, independent of rebuildable metadata. +//! Reads do not materialize state. Live terminal events are delivered serially per thread; +//! keep their last key across acknowledgements and reverts to suppress redelivery. + +use std::collections::HashMap; + +use codex_protocol::ThreadId; +use sqlx::QueryBuilder; +use sqlx::Row; +use sqlx::Sqlite; +use sqlx::SqliteConnection; +use uuid::Uuid; + +use super::StateRuntime; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ThreadReadState { + /// None is read; Some("") marks the whole thread unread, even if it has no turns. + pub first_unread_turn: Option, + pub revision: String, +} + +pub enum ReadStateOperation { + Read, + Unread, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum ReadStateUpdate { + Applied(ThreadReadState), + Conflict(ThreadReadState), + Unavailable, +} + +impl StateRuntime { + /// Read just the requested page. Only durable metadata rows are eligible. + pub async fn thread_read_states( + &self, + ids: &[ThreadId], + ) -> anyhow::Result> { + let mut conn = self.pool.acquire().await?; + read_states(&mut conn, ids).await + } + + /// Publish a newly delivered terminal event only once, never on history replay. + /// Every newer result guards against an acknowledgement based on an older snapshot, + /// even if the first unread position stays the same. + pub async fn publish_thread_attention( + &self, + id: ThreadId, + turn_id: &str, + ) -> anyhow::Result { + let changed = sqlx::query( + "INSERT INTO thread_read_receipts(thread_id,first_unread_turn,revision,last_published_turn) + SELECT id, ?, ?, ? FROM threads WHERE id=? + ON CONFLICT(thread_id) DO UPDATE SET + first_unread_turn=COALESCE(thread_read_receipts.first_unread_turn,excluded.first_unread_turn), + revision=excluded.revision, + last_published_turn=excluded.last_published_turn + WHERE thread_read_receipts.last_published_turn IS NOT excluded.last_published_turn", + ) + .bind(turn_id) + .bind(Uuid::new_v4().to_string()) + .bind(turn_id) + .bind(id.to_string()) + .execute(self.pool.as_ref()) + .await? + .rows_affected() != 0; + Ok(changed) + } + + /// Both read and unread edits compare the snapshot the client actually saw. Never retry a + /// stale edit automatically: its author may not have seen a newer result or another mark. + pub async fn update_thread_read_state( + &self, + id: ThreadId, + expected_revision: &str, + operation: ReadStateOperation, + ) -> anyhow::Result { + let mut tx = self.pool.begin_with("BEGIN IMMEDIATE").await?; + let Some(before) = read_states(&mut tx, &[id]).await?.remove(&id) else { + return Ok(ReadStateUpdate::Unavailable); + }; + if before.revision != expected_revision { + return Ok(ReadStateUpdate::Conflict(before)); + } + let first_unread = match operation { + ReadStateOperation::Read => None, + ReadStateOperation::Unread => Some(""), + }; + let revision = Uuid::new_v4().to_string(); + sqlx::query( + "INSERT INTO thread_read_receipts(thread_id,first_unread_turn,revision) + VALUES(?,?,?) ON CONFLICT(thread_id) DO UPDATE + SET first_unread_turn=excluded.first_unread_turn, revision=excluded.revision", + ) + .bind(id.to_string()) + .bind(first_unread) + .bind(&revision) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(ReadStateUpdate::Applied(ThreadReadState { + first_unread_turn: first_unread.map(str::to_owned), + revision, + })) + } +} + +async fn read_states( + conn: &mut SqliteConnection, + ids: &[ThreadId], +) -> anyhow::Result> { + let mut result = HashMap::new(); + for ids in ids.chunks(998) { + let mut query = QueryBuilder::::new( + "SELECT t.id, r.first_unread_turn, r.revision FROM threads t + LEFT JOIN thread_read_receipts r ON r.thread_id=t.id WHERE t.id IN (", + ); + let mut separated = query.separated(", "); + for id in ids { + separated.push_bind(id.to_string()); + } + separated.push_unseparated(")"); + for row in query.build().fetch_all(&mut *conn).await? { + let id: String = row.try_get("id")?; + let revision: Option = row.try_get("revision")?; + result.insert( + ThreadId::from_string(&id)?, + ThreadReadState { + first_unread_turn: row.try_get("first_unread_turn")?, + // No baseline write. Bind the initial revision to the thread so it cannot + // accidentally be used to edit another thread with no receipt yet. + revision: revision.unwrap_or_else(|| format!("initial:{id}")), + }, + ); + } + } + Ok(result) +} + +#[cfg(test)] +#[path = "thread_read_state_tests.rs"] +mod tests; diff --git a/codex-rs/state/src/runtime/thread_read_state_tests.rs b/codex-rs/state/src/runtime/thread_read_state_tests.rs new file mode 100644 index 0000000000..83c1593f80 --- /dev/null +++ b/codex-rs/state/src/runtime/thread_read_state_tests.rs @@ -0,0 +1,167 @@ +//! Live completion, cross-window edits and rebuildable-metadata isolation. +use codex_protocol::ThreadId; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; + +use super::*; +use crate::SqliteConfig; +use crate::runtime::test_support::test_thread_metadata; +use crate::runtime::test_support::unique_temp_dir; + +#[tokio::test] +async fn a_stale_read_cannot_erase_a_later_completion_or_another_mark() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let runtime = StateRuntime::init( + SqliteConfig::new_for_testing(home.as_path().abs()), + "test-provider".to_string(), + ) + .await?; + let id = ThreadId::new(); + let other = ThreadId::new(); + runtime + .upsert_thread(&test_thread_metadata(&home, id, home.clone())) + .await?; + runtime + .upsert_thread(&test_thread_metadata(&home, other, home.clone())) + .await?; + let baseline = runtime.thread_read_states(&[id]).await?[&id].clone(); + assert_eq!(baseline.first_unread_turn, None); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT count(*) FROM thread_read_receipts") + .fetch_one(runtime.pool.as_ref()) + .await?, + 0 + ); + assert!(matches!( + runtime + .update_thread_read_state(other, &baseline.revision, ReadStateOperation::Unread) + .await?, + ReadStateUpdate::Conflict(_) + )); + // A mark is meaningful even without any completed turn. + let ReadStateUpdate::Applied(marked) = runtime + .update_thread_read_state(id, &baseline.revision, ReadStateOperation::Unread) + .await? + else { + panic!("could not mark an empty thread unread") + }; + assert_eq!(marked.first_unread_turn, Some(String::new())); + assert_eq!( + runtime + .update_thread_read_state(id, &baseline.revision, ReadStateOperation::Read) + .await?, + ReadStateUpdate::Conflict(marked.clone()) + ); + // Even another mark of an already-unread thread guards against an older explicit read. + let ReadStateUpdate::Applied(marked_again) = runtime + .update_thread_read_state(id, &marked.revision, ReadStateOperation::Unread) + .await? + else { + panic!("could not mark again") + }; + assert_eq!( + runtime + .update_thread_read_state(id, &marked.revision, ReadStateOperation::Read) + .await?, + ReadStateUpdate::Conflict(marked_again.clone()) + ); + let ReadStateUpdate::Applied(read) = runtime + .update_thread_read_state(id, &marked_again.revision, ReadStateOperation::Read) + .await? + else { + panic!("could not read at current revision") + }; + assert_eq!(read.first_unread_turn, None); + assert!(runtime.publish_thread_attention(id, "turn-1").await?); + let first = runtime.thread_read_states(&[id]).await?[&id].clone(); + assert_eq!(first.first_unread_turn.as_deref(), Some("turn-1")); + assert!(!runtime.publish_thread_attention(id, "turn-1").await?); + assert_eq!(runtime.thread_read_states(&[id]).await?[&id], first); + assert!(runtime.publish_thread_attention(id, "turn-2").await?); + let second = runtime.thread_read_states(&[id]).await?[&id].clone(); + assert_eq!(second.first_unread_turn.as_deref(), Some("turn-1")); + assert_ne!(second.revision, first.revision); + assert_eq!( + runtime + .update_thread_read_state(id, &first.revision, ReadStateOperation::Read) + .await?, + ReadStateUpdate::Conflict(second.clone()) + ); + assert_eq!( + runtime + .update_thread_read_state(id, &first.revision, ReadStateOperation::Unread) + .await?, + ReadStateUpdate::Conflict(second.clone()) + ); + let ReadStateUpdate::Applied(read) = runtime + .update_thread_read_state(id, &second.revision, ReadStateOperation::Read) + .await? + else { + panic!("could not acknowledge both visible results") + }; + assert_eq!(read.first_unread_turn, None); + assert!(!runtime.publish_thread_attention(id, "turn-2").await?); + assert_eq!(runtime.thread_read_states(&[id]).await?[&id], read); + runtime.close().await; + std::fs::remove_dir_all(home)?; + Ok(()) +} + +#[tokio::test] +async fn position_and_dedupe_survive_metadata_rebuild_restart_and_revert() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); + let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()).await?; + let id = ThreadId::new(); + let metadata = test_thread_metadata(&home, id, home.clone()); + runtime.upsert_thread(&metadata).await?; + let original_metadata = runtime.get_thread(id).await?; + assert!(runtime.publish_thread_attention(id, "removed-turn").await?); + let before = runtime.thread_read_states(&[id]).await?[&id].clone(); + // After a revert the caller checks whether the position still exists, then clears by revision. + let ReadStateUpdate::Applied(removed) = runtime + .update_thread_read_state(id, &before.revision, ReadStateOperation::Read) + .await? + else { + panic!("could not retire the removed position") + }; + assert_eq!(removed.first_unread_turn, None); + assert_eq!(runtime.get_thread(id).await?, original_metadata); + sqlx::query("DELETE FROM threads WHERE id=?") + .bind(id.to_string()) + .execute(runtime.pool.as_ref()) + .await?; + assert!(runtime.thread_read_states(&[id]).await?.is_empty()); + runtime.upsert_thread(&metadata).await?; + runtime.close().await; + let reopened = StateRuntime::init(sqlite, "test-provider".to_string()).await?; + assert_eq!(reopened.thread_read_states(&[id]).await?[&id], removed); + assert!( + !reopened + .publish_thread_attention(id, "removed-turn") + .await? + ); + assert_eq!(reopened.thread_read_states(&[id]).await?[&id], removed); + assert!( + reopened + .publish_thread_attention(id, "new-live-turn") + .await? + ); + assert_eq!( + reopened.thread_read_states(&[id]).await?[&id] + .first_unread_turn + .as_deref(), + Some("new-live-turn") + ); + reopened.delete_thread(id).await?; + assert!(!reopened.publish_thread_attention(id, "late").await?); + assert_eq!( + reopened + .update_thread_read_state(id, &removed.revision, ReadStateOperation::Unread) + .await?, + ReadStateUpdate::Unavailable + ); + reopened.close().await; + std::fs::remove_dir_all(home)?; + Ok(()) +} diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 228a7cd6db..1179126f62 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1198,6 +1198,10 @@ ON CONFLICT(id) DO UPDATE SET } let mut rows_affected = 0; for thread_id_string in &thread_id_strings { + sqlx::query("DELETE FROM thread_read_receipts WHERE thread_id = ?") + .bind(thread_id_string) + .execute(&mut *tx) + .await?; rows_affected += sqlx::query("DELETE FROM threads WHERE id = ?") .bind(thread_id_string) .execute(&mut *tx)