Add paginated cloud thread discovery to the cloud client (#59514)

## What changed Add `Client::list_threads` with cursor pagination, page limits, sorting, and originator filtering for unarchived threads. Return thread snapshots, optional read states, and forward and backward cursors while preserving server order and duplicate entries. Allow stored thread snapshots to contain empty session IDs and model providers while retaining nonempty requirements for activation. Preserve the original snapshot protobuf bytes. ## Testing Add fixture tests for pagination requests, default query encoding, snapshot and read-state preservation, legacy empty metadata, invalid responses, and activation validation. GitOrigin-RevId: 6492cb60573d13cc7b7d0aa583d1dab6bccafc4d
2026-11-05 20:10:33 +00:00 · 2026-11-04 23:18:18 +00:00
parent f8c4060b68
commit 763faa6ee4
5 changed files with 392 additions and 6 deletions
@@ -267,10 +267,10 @@ impl Client {
 impl ThreadActivation {
    pub(crate) fn from_settings(response: wire::UpdateConfigResponse) -> Result<Self, Error> {
        let raw_thread = required(response.thread)?;
-        let thread = ThreadSnapshot::try_from_wire(
-            wire::Thread::decode(raw_thread.clone()).map_err(|_| Error::InvalidResponse)?,
-            raw_thread,
-        )?;
+        let thread = ThreadSnapshot::decode(raw_thread)?;
+        if thread.session_id.is_empty() || thread.model_provider.is_empty() {
+            return Err(Error::InvalidResponse);
+        }
        Ok(Self {
            thread,
            collaboration_mode: response
@@ -310,10 +310,16 @@ impl ThreadActivation {
 }

 impl ThreadSnapshot {
+    pub(crate) fn decode(raw_protobuf: Bytes) -> Result<Self, Error> {
+        let thread =
+            wire::Thread::decode(raw_protobuf.clone()).map_err(|_| Error::InvalidResponse)?;
+        Self::try_from_wire(thread, raw_protobuf)
+    }
+
    fn try_from_wire(thread: wire::Thread, raw_protobuf: Bytes) -> Result<Self, Error> {
        Ok(Self {
            id: nonempty(thread.id)?,
-            session_id: nonempty(thread.session_id)?,
+            session_id: required(thread.session_id)?,
            forked_from_id: thread.forked_from_id,
            parent_thread_id: thread.parent_thread_id,
            preview: required(thread.preview)?,
@@ -321,7 +327,7 @@ impl ThreadSnapshot {
            daybreak_enabled: thread.daybreak_enabled,
            project_id: thread.project_id,
            history_mode: parse_history_mode(thread.history_mode)?,
-            model_provider: nonempty(thread.model_provider)?,
+            model_provider: required(thread.model_provider)?,
            model: thread.model,
            reasoning_effort: thread.reasoning_effort,
            created_at: required(thread.created_at)?,
@@ -8,6 +8,7 @@ use hyper::service::service_fn;
 use hyper_util::rt::TokioExecutor;
 use hyper_util::rt::TokioIo;
 use pretty_assertions::assert_eq;
+use prost::Message;
 use std::convert::Infallible;
 use tokio::sync::mpsc;

@@ -156,6 +157,23 @@ async fn activation_preserves_native_settings_and_thread_metadata() {
        (response_bytes, thread)
    };

+    // Activation retains stricter runtime requirements than stored list metadata.
+    for (session_id, model_provider) in [("", "p"), ("s", "")] {
+        let (response_bytes, _) = encode_response(1, Some(true), 2);
+        let mut settings =
+            crate::activation_wire::UpdateConfigResponse::decode(Bytes::from(response_bytes))
+                .unwrap();
+        let mut thread =
+            crate::activation_wire::Thread::decode(settings.thread.take().unwrap()).unwrap();
+        thread.session_id = Some(session_id.into());
+        thread.model_provider = Some(model_provider.into());
+        settings.thread = Some(thread.encode_to_vec().into());
+        assert!(matches!(
+            ThreadActivation::from_settings(settings),
+            Err(Error::InvalidResponse)
+        ));
+    }
+
    for (mode, daybreak) in [(1, Some(true)), (2, Some(false)), (2, None)] {
        let (response_bytes, thread) = encode_response(mode, daybreak, 2);
        let mut server = fixture(response(
@@ -562,3 +580,6 @@ async fn https_enforces_unmanaged_endpoint_restrictions_before_connecting() {
 }
 #[path = "transcript_tests.rs"]
 mod transcript;
+
+#[path = "discovery_tests.rs"]
+mod discovery;
@@ -0,0 +1,226 @@
+//! Typed, paginated cloud thread discovery over the native ThreadService.
+
+use crate::Client;
+use crate::Error;
+use crate::ThreadSnapshot;
+use bytes::Bytes;
+use prost::Message;
+use std::collections::HashMap;
+use std::fmt;
+
+#[derive(Clone, Copy, Debug, Default)]
+pub enum ThreadSortKey {
+    #[default]
+    CreatedAt,
+    UpdatedAt,
+    RecencyAt,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+pub enum ThreadSortDirection {
+    Asc,
+    #[default]
+    Desc,
+}
+
+/// Lists unarchived threads with the server's default source selection.
+/// Reuse the same originators and order when continuing with an opaque cursor.
+#[derive(Clone, Default)]
+pub struct ThreadListQuery {
+    pub cursor: Option<String>,
+    pub limit: Option<u32>,
+    pub sort_key: ThreadSortKey,
+    pub sort_direction: ThreadSortDirection,
+    /// Empty means all originators, including conversations created by other clients.
+    pub originators: Vec<String>,
+}
+
+/// One bounded page in server order; no automatic pagination or deduplication.
+#[derive(Clone, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ThreadPage {
+    pub data: Vec<ThreadSnapshot>,
+    pub read_states: Option<HashMap<String, ThreadReadState>>,
+    pub next_cursor: Option<String>,
+    pub backwards_cursor: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ThreadReadState {
+    pub first_unread: Option<ThreadUnreadPosition>,
+    pub revision: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum ThreadUnreadPosition {
+    ThreadStart,
+    Turn {
+        #[serde(rename = "turnId")]
+        turn_id: String,
+    },
+}
+
+macro_rules! redacted_debug {
+    ($($ty:ty),*) => {$(impl fmt::Debug for $ty {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            f.write_str(concat!(stringify!($ty), "([redacted])"))
+        }
+    })*};
+}
+redacted_debug!(ThreadListQuery, ThreadPage);
+
+impl Client {
+    /// Returns one bounded native page. Read failures are not retried automatically.
+    pub async fn list_threads(&self, query: &ThreadListQuery) -> Result<ThreadPage, Error> {
+        let bytes = self
+            .raw_unary(
+                wire::ListThreadsRequest {
+                    cursor: query.cursor.clone(),
+                    limit: query.limit,
+                    sort_key: Some(match query.sort_key {
+                        ThreadSortKey::CreatedAt => 1,
+                        ThreadSortKey::UpdatedAt => 2,
+                        ThreadSortKey::RecencyAt => 3,
+                    }),
+                    sort_direction: Some(match query.sort_direction {
+                        ThreadSortDirection::Asc => 1,
+                        ThreadSortDirection::Desc => 2,
+                    }),
+                    archived: Some(false),
+                    originators: Some(wire::StringList {
+                        values: query.originators.clone(),
+                    }),
+                },
+                "/oaiproto.codex_app_server_backend.ThreadService/ListThreads",
+            )
+            .await?;
+        let response =
+            wire::ListThreadsResponse::decode(bytes).map_err(|_| Error::InvalidResponse)?;
+        Ok(ThreadPage {
+            data: response
+                .data
+                .ok_or(Error::InvalidResponse)?
+                .values
+                .into_iter()
+                .map(ThreadSnapshot::decode)
+                .collect::<Result<_, _>>()?,
+            read_states: response
+                .read_states
+                .map(|states| {
+                    states
+                        .values
+                        .into_iter()
+                        .map(|(thread_id, state)| {
+                            let revision = state.revision.ok_or(Error::InvalidResponse)?;
+                            if revision.is_empty() {
+                                return Err(Error::InvalidResponse);
+                            }
+                            let first_unread = state
+                                .first_unread
+                                .map(|position| match position.value {
+                                    Some(wire::thread_unread_position::Value::ThreadStart(_)) => {
+                                        Ok(ThreadUnreadPosition::ThreadStart)
+                                    }
+                                    Some(wire::thread_unread_position::Value::Turn(turn)) => {
+                                        let turn_id = turn.turn_id.ok_or(Error::InvalidResponse)?;
+                                        if turn_id.is_empty() {
+                                            Err(Error::InvalidResponse)
+                                        } else {
+                                            Ok(ThreadUnreadPosition::Turn { turn_id })
+                                        }
+                                    }
+                                    None => Err(Error::InvalidResponse),
+                                })
+                                .transpose()?;
+                            Ok((
+                                thread_id,
+                                ThreadReadState {
+                                    first_unread,
+                                    revision,
+                                },
+                            ))
+                        })
+                        .collect::<Result<_, _>>()
+                })
+                .transpose()?,
+            next_cursor: response.next_cursor,
+            backwards_cursor: response.backwards_cursor,
+        })
+    }
+}
+
+mod wire {
+    use super::*;
+
+    #[derive(Clone, Message)]
+    pub(super) struct ListThreadsRequest {
+        #[prost(string, optional, tag = "1")]
+        pub cursor: Option<String>,
+        #[prost(uint32, optional, tag = "2")]
+        pub limit: Option<u32>,
+        #[prost(int32, optional, tag = "3")]
+        pub sort_key: Option<i32>,
+        #[prost(int32, optional, tag = "4")]
+        pub sort_direction: Option<i32>,
+        #[prost(bool, optional, tag = "5")]
+        pub archived: Option<bool>,
+        #[prost(message, optional, tag = "6")]
+        pub originators: Option<StringList>,
+    }
+    #[derive(Clone, Message)]
+    pub(super) struct StringList {
+        #[prost(string, repeated, tag = "1")]
+        pub values: Vec<String>,
+    }
+    #[derive(Clone, Message)]
+    pub(super) struct ListThreadsResponse {
+        #[prost(message, optional, tag = "1")]
+        pub data: Option<ThreadList>,
+        #[prost(string, optional, tag = "2")]
+        pub next_cursor: Option<String>,
+        #[prost(string, optional, tag = "3")]
+        pub backwards_cursor: Option<String>,
+        #[prost(message, optional, tag = "4")]
+        pub read_states: Option<ThreadReadStateMap>,
+    }
+    #[derive(Clone, Message)]
+    pub(super) struct ThreadList {
+        #[prost(bytes = "bytes", repeated, tag = "1")]
+        pub values: Vec<Bytes>,
+    }
+    #[derive(Clone, Message)]
+    pub(super) struct ThreadReadStateMap {
+        #[prost(map = "string, message", tag = "1")]
+        pub values: HashMap<String, ThreadReadState>,
+    }
+    #[derive(Clone, Message, PartialEq)]
+    pub(super) struct ThreadReadState {
+        #[prost(message, optional, tag = "1")]
+        pub first_unread: Option<ThreadUnreadPosition>,
+        #[prost(string, optional, tag = "2")]
+        pub revision: Option<String>,
+    }
+    #[derive(Clone, Message, PartialEq)]
+    pub(super) struct ThreadUnreadPosition {
+        #[prost(oneof = "thread_unread_position::Value", tags = "1, 2")]
+        pub value: Option<thread_unread_position::Value>,
+    }
+    pub(super) mod thread_unread_position {
+        #[derive(Clone, prost::Oneof, PartialEq)]
+        pub enum Value {
+            #[prost(message, tag = "1")]
+            ThreadStart(super::ThreadUnreadPositionThreadStart),
+            #[prost(message, tag = "2")]
+            Turn(super::ThreadUnreadPositionTurn),
+        }
+    }
+    #[derive(Clone, Message, PartialEq)]
+    pub(super) struct ThreadUnreadPositionThreadStart {}
+    #[derive(Clone, Message, PartialEq)]
+    pub(super) struct ThreadUnreadPositionTurn {
+        #[prost(string, optional, tag = "1")]
+        pub turn_id: Option<String>,
+    }
+}
@@ -0,0 +1,126 @@
+#![cfg(test)]
+
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn discovery_preserves_pagination_and_order_on_each_request() {
+    let payload = b"\x0a\x00\x12\x04next\x1a\x04back";
+    let mut fixture = fixture_responses(vec![
+        response(vec![frame(payload)], tonic::Status::ok("")),
+        response(vec![frame(b"\x0a\x00")], tonic::Status::ok("")),
+    ])
+    .await;
+    let mut query = ThreadListQuery {
+        cursor: Some("cursor".into()),
+        limit: Some(7),
+        sort_key: ThreadSortKey::UpdatedAt,
+        sort_direction: ThreadSortDirection::Asc,
+        originators: vec!["tui".into()],
+    };
+    let page = fixture.client.list_threads(&query).await.unwrap();
+    assert_eq!(
+        (
+            page.data,
+            page.next_cursor.clone(),
+            page.backwards_cursor,
+            page.read_states
+        ),
+        (Vec::new(), Some("next".into()), Some("back".into()), None)
+    );
+    assert_request(
+        &mut fixture,
+        "ListThreads",
+        b"\x0a\x06cursor\x10\x07\x18\x02\x20\x01\x28\x00\x32\x05\x0a\x03tui",
+    )
+    .await;
+    query.cursor = page.next_cursor;
+    let last = fixture.client.list_threads(&query).await.unwrap();
+    assert_eq!((last.data, last.next_cursor), (Vec::new(), None));
+    assert_request(
+        &mut fixture,
+        "ListThreads",
+        b"\x0a\x04next\x10\x07\x18\x02\x20\x01\x28\x00\x32\x05\x0a\x03tui",
+    )
+    .await;
+    assert!(fixture.requests.try_recv().is_err());
+}
+
+#[tokio::test]
+async fn discovery_preserves_native_snapshots_and_read_state() {
+    // Independent native Thread encodings: modern future field (99) and legacy empty metadata.
+    let thread = Bytes::from_static(b"\x22\x01v\x28\x01\x32\x01/\x48\x00\x72\x01t\x82\x01\x01p\xaa\x01\x00\xda\x01\x01s\xe2\x01\x02\x08\x63\xea\x01\x02\x12\x00\xfa\x01\x00\x80\x02\x02\x98\x06\x01");
+    let legacy = Bytes::from_static(b"\x22\x00\x28\x00\x32\x01/\x48\x00\x72\x06legacy\x82\x01\x00\xaa\x01\x00\xda\x01\x00\xe2\x01\x02\x08\x01\xea\x01\x02\x12\x00\xfa\x01\x00\x80\x02\x00");
+    let mut data = Vec::new();
+    for row in [&thread, &legacy, &thread] {
+        prost::encoding::bytes::encode(1, row, &mut data);
+    }
+    let mut payload = Vec::new();
+    prost::encoding::bytes::encode(1, &Bytes::from(data), &mut payload);
+    // Read state map: t -> {first_unread: thread_start, revision: r}.
+    payload.extend_from_slice(b"\x22\x0e\x0a\x0c\x0a\x01t\x12\x07\x0a\x02\x0a\x00\x12\x01r");
+    let mut fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
+    let page = fixture
+        .client
+        .list_threads(&ThreadListQuery::default())
+        .await
+        .unwrap();
+    assert_eq!(
+        page.data
+            .iter()
+            .map(ThreadSnapshot::protobuf)
+            .collect::<Vec<_>>(),
+        vec![&thread, &legacy, &thread]
+    );
+    assert_eq!(
+        (
+            page.data[1].id.as_str(),
+            page.data[1].session_id.as_str(),
+            page.data[1].model_provider.as_str(),
+            page.data[1].cli_version.as_str()
+        ),
+        ("legacy", "", "", "")
+    );
+    assert_eq!(page.data[0].status, ThreadStatus::Idle);
+    assert_eq!(
+        page.read_states,
+        Some(std::collections::HashMap::from([(
+            "t".into(),
+            ThreadReadState {
+                first_unread: Some(ThreadUnreadPosition::ThreadStart),
+                revision: "r".into(),
+            }
+        )]))
+    );
+    assert_request(
+        &mut fixture,
+        "ListThreads",
+        b"\x18\x01\x20\x02\x28\x00\x32\x00",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn discovery_rejects_missing_data_invalid_snapshot_and_empty_revision() {
+    for payload in [
+        &b""[..],
+        &b"\x0a\x02\x0a\x00"[..],
+        &b"\x0a\x00\x22\x07\x0a\x05\x0a\x01t\x12\x00"[..],
+    ] {
+        let mut fixture = fixture(response(vec![frame(payload)], tonic::Status::ok(""))).await;
+        assert!(matches!(
+            fixture
+                .client
+                .list_threads(&ThreadListQuery::default())
+                .await,
+            Err(Error::InvalidResponse)
+        ));
+        assert_request(
+            &mut fixture,
+            "ListThreads",
+            b"\x18\x01\x20\x02\x28\x00\x32\x00",
+        )
+        .await;
+        assert!(fixture.requests.try_recv().is_err());
+    }
+}
@@ -6,6 +6,7 @@

 mod activation;
 mod activation_wire;
+mod discovery;
 mod history;
 mod history_wire;
 mod item_wire;
@@ -33,6 +34,12 @@ pub use activation::ThreadEnvironment;
 pub use activation::ThreadHistoryMode;
 pub use activation::ThreadSnapshot;
 pub use activation::ThreadStatus;
+pub use discovery::ThreadListQuery;
+pub use discovery::ThreadPage;
+pub use discovery::ThreadReadState;
+pub use discovery::ThreadSortDirection;
+pub use discovery::ThreadSortKey;
+pub use discovery::ThreadUnreadPosition;
 pub use history::HistoryAccess;
 pub use history::HistoryItem;
 pub use history::HistoryOrder;