Files
codex/codex-rs/cloud-client/src/history_wire.rs
T
Benjamin Carlsson bc4116e4cd Add paginated history reads to the cloud client (#59075)
## Why

`Attach` delivers live events only. Clients also need access to stored turn metadata and item content.

## What changed

- Add `Client::list_turns` and `Client::list_items` using native gRPC history endpoints, with optional turn filtering for items.
- Expose page limits, sort order, access mode, and forward/backward cursors. Each call returns one page in server order without automatic pagination or retries.
- Preserve raw response and row payloads alongside typed summaries and item timestamps. Retain unknown item variants and command sources without typed summaries.
- Document cursor reuse and history/live overlap: history pages do not provide an atomic boundary with `Attach`, and coordinated reconciliation still requires caller-managed state.

## Testing

Add wire-fixture tests for request encoding, explicit pagination, raw payload preservation, unknown data, malformed rows, empty pages, RPC errors without retries, and diagnostic redaction.

GitOrigin-RevId: f507530b8e431236bc7e19d867517af56f74644f
2026-11-03 06:34:51 +00:00

87 lines
2.7 KiB
Rust

//! Native paginated history envelopes. Raw rows and unary responses retain unknown fields.
use bytes::Buf;
use bytes::Bytes;
use prost::Message;
use std::marker::PhantomData;
use tonic::codec::Codec;
#[derive(Clone, Message)]
pub(super) struct TurnsRequest {
#[prost(string, tag = "1")]
pub thread_id: String,
#[prost(string, optional, tag = "2")]
pub cursor: Option<String>,
#[prost(uint32, optional, tag = "3")]
pub limit: Option<u32>,
#[prost(int32, optional, tag = "4")]
pub order: Option<i32>,
#[prost(int32, optional, tag = "6")]
pub access: Option<i32>,
}
#[derive(Clone, Message)]
pub(super) struct ItemsRequest {
#[prost(string, tag = "1")]
pub thread_id: String,
#[prost(string, optional, tag = "2")]
pub turn_id: Option<String>,
#[prost(string, optional, tag = "3")]
pub cursor: Option<String>,
#[prost(uint32, optional, tag = "4")]
pub limit: Option<u32>,
#[prost(int32, optional, tag = "5")]
pub order: Option<i32>,
#[prost(int32, optional, tag = "6")]
pub access: Option<i32>,
}
#[derive(Clone, Message)]
pub(super) struct Page {
#[prost(message, optional, tag = "1")]
pub data: Option<Rows>,
#[prost(string, optional, tag = "2")]
pub next_cursor: Option<String>,
#[prost(string, optional, tag = "3")]
pub backwards_cursor: Option<String>,
}
#[derive(Clone, Message)]
pub(super) struct Rows {
#[prost(bytes = "bytes", repeated, tag = "1")]
pub values: Vec<Bytes>,
}
#[derive(Clone, Message)]
pub(super) struct ItemEntry {
#[prost(string, optional, tag = "1")]
pub turn_id: Option<String>,
#[prost(message, optional, tag = "2")]
pub item: Option<crate::item_wire::Item>,
#[prost(int64, optional, tag = "3")]
pub started_at_ms: Option<i64>,
#[prost(int64, optional, tag = "4")]
pub completed_at_ms: Option<i64>,
}
// Tonic still enforces message bounds and status trailers. Only payload decoding
// is delayed so callers can retain the original complete protobuf response.
pub(super) struct RawResponseCodec<T>(pub PhantomData<T>);
impl<T: Message + Send + 'static> Codec for RawResponseCodec<T> {
type Encode = T;
type Decode = Bytes;
type Encoder = tonic_prost::ProstEncoder<T>;
type Decoder = RawDecoder;
fn encoder(&mut self) -> Self::Encoder {
tonic_prost::ProstCodec::<T, Page>::default().encoder()
}
fn decoder(&mut self) -> Self::Decoder {
RawDecoder
}
}
pub(super) struct RawDecoder;
impl tonic::codec::Decoder for RawDecoder {
type Item = Bytes;
type Error = tonic::Status;
fn decode(
&mut self,
buffer: &mut tonic::codec::DecodeBuf<'_>,
) -> Result<Option<Bytes>, tonic::Status> {
Ok(Some(buffer.copy_to_bytes(buffer.remaining())))
}
}