Add durable thread read state with revision-checked updates (#52337)

## Why

A stale read acknowledgement must not clear unread results published after the client’s snapshot or overwrite another read-state edit. Read state also needs to survive thread metadata rebuilds.

## What changed

- Add `thread_read_receipts` and state APIs to query read state, publish thread attention, and mark threads read or unread.
- Track the first unread turn and require matching revisions for explicit edits, returning the current state on conflicts. Allow manually marking even an empty thread unread.
- Retain the last published turn across acknowledgements to suppress duplicate delivery. Keep receipts independent of rebuildable metadata and remove them on thread deletion.
- Return initial read states without creating receipt rows.

## Testing

Add tests covering stale edits, thread-specific initial revisions, repeated unread marks, duplicate publication, persistence across metadata rebuilds and restarts, clearing a reverted turn’s unread position, and deleted-thread handling.

GitOrigin-RevId: fee8c05c20903b1d7aea7e8b61c9248c7f623a72

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
+);

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;

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:

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<String>,
+    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<HashMap<ThreadId, ThreadReadState>> {
+        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<bool> {
+        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<ReadStateUpdate> {
+        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<HashMap<ThreadId, ThreadReadState>> {
+    let mut result = HashMap::new();
+    for ids in ids.chunks(998) {
+        let mut query = QueryBuilder::<Sqlite>::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<String> = 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;

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(())
+}

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)