diff --git a/codex-rs/cloud-client/src/activation.rs b/codex-rs/cloud-client/src/activation.rs index 1734c8586f..8a46ceacde 100644 --- a/codex-rs/cloud-client/src/activation.rs +++ b/codex-rs/cloud-client/src/activation.rs @@ -267,10 +267,10 @@ impl Client { impl ThreadActivation { pub(crate) fn from_settings(response: wire::UpdateConfigResponse) -> Result { 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 { + 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 { 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)?, diff --git a/codex-rs/cloud-client/src/client_tests.rs b/codex-rs/cloud-client/src/client_tests.rs index 5216fb02c2..979f861a79 100644 --- a/codex-rs/cloud-client/src/client_tests.rs +++ b/codex-rs/cloud-client/src/client_tests.rs @@ -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; diff --git a/codex-rs/cloud-client/src/discovery.rs b/codex-rs/cloud-client/src/discovery.rs new file mode 100644 index 0000000000..a97d10f96b --- /dev/null +++ b/codex-rs/cloud-client/src/discovery.rs @@ -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, + pub limit: Option, + pub sort_key: ThreadSortKey, + pub sort_direction: ThreadSortDirection, + /// Empty means all originators, including conversations created by other clients. + pub originators: Vec, +} + +/// 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, + pub read_states: Option>, + pub next_cursor: Option, + pub backwards_cursor: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ThreadReadState { + pub first_unread: Option, + 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 { + 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::>()?, + 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::>() + }) + .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, + #[prost(uint32, optional, tag = "2")] + pub limit: Option, + #[prost(int32, optional, tag = "3")] + pub sort_key: Option, + #[prost(int32, optional, tag = "4")] + pub sort_direction: Option, + #[prost(bool, optional, tag = "5")] + pub archived: Option, + #[prost(message, optional, tag = "6")] + pub originators: Option, + } + #[derive(Clone, Message)] + pub(super) struct StringList { + #[prost(string, repeated, tag = "1")] + pub values: Vec, + } + #[derive(Clone, Message)] + pub(super) struct ListThreadsResponse { + #[prost(message, optional, tag = "1")] + pub data: Option, + #[prost(string, optional, tag = "2")] + pub next_cursor: Option, + #[prost(string, optional, tag = "3")] + pub backwards_cursor: Option, + #[prost(message, optional, tag = "4")] + pub read_states: Option, + } + #[derive(Clone, Message)] + pub(super) struct ThreadList { + #[prost(bytes = "bytes", repeated, tag = "1")] + pub values: Vec, + } + #[derive(Clone, Message)] + pub(super) struct ThreadReadStateMap { + #[prost(map = "string, message", tag = "1")] + pub values: HashMap, + } + #[derive(Clone, Message, PartialEq)] + pub(super) struct ThreadReadState { + #[prost(message, optional, tag = "1")] + pub first_unread: Option, + #[prost(string, optional, tag = "2")] + pub revision: Option, + } + #[derive(Clone, Message, PartialEq)] + pub(super) struct ThreadUnreadPosition { + #[prost(oneof = "thread_unread_position::Value", tags = "1, 2")] + pub value: Option, + } + 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, + } +} diff --git a/codex-rs/cloud-client/src/discovery_tests.rs b/codex-rs/cloud-client/src/discovery_tests.rs new file mode 100644 index 0000000000..2a2fcf1f7c --- /dev/null +++ b/codex-rs/cloud-client/src/discovery_tests.rs @@ -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![&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()); + } +} diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs index 367b1bef18..7ca207c1e3 100644 --- a/codex-rs/cloud-client/src/lib.rs +++ b/codex-rs/cloud-client/src/lib.rs @@ -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;