mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
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
This commit is contained in:
Benjamin Carlsson
committed by
copyberry
copyberry
parent
f9e7b19df8
commit
bc4116e4cd
@@ -29,7 +29,7 @@ code, message and binary details are available explicitly; credentials, event
|
||||
and error diagnostics are redacted. Item lifecycle updates report supported
|
||||
message, command, and file-change summaries. Agent message summaries retain
|
||||
structured questions even when their text is empty. Unsupported item kinds stay opaque.
|
||||
History, approval replies and reconnect are subsequent slices.
|
||||
Approval replies and reconnect are subsequent slices.
|
||||
The minimal envelopes in `wire.rs` and `ResumeRequest` mirror ThreadService field
|
||||
numbers; no monorepo checkout or protobuf compiler is required to build this crate.
|
||||
|
||||
@@ -43,3 +43,19 @@ cargo run -p codex-cloud-client --example attach -- "$CODEX_CLOUD_GRPC_ENDPOINT"
|
||||
Use an existing authorized thread and native gRPC origin. No production URL is
|
||||
assumed. Public native ingress is a rollout assumption, not proved by fixture
|
||||
tests. This change does not publish a package to a registry.
|
||||
|
||||
History reads use native `ListTurns` (stored turn metadata) and `ListItems` (one
|
||||
turn or the whole thread). Calls return one page with forward/backward opaque
|
||||
cursors, preserve server order and retain raw responses and rows. Reuse a cursor
|
||||
with the same account, thread and filters. Keep the sort order for `next_cursor`
|
||||
and reverse it for `backwards_cursor`. Calls never paginate or retry automatically.
|
||||
Items with an unknown variant or command source remain in the page with no typed summary.
|
||||
|
||||
History pages and Attach have no shared revision or replay cursor. A page is not
|
||||
an atomic history/live boundary, and timestamps are not event offsets. Do not
|
||||
append buffered deltas to a fetched snapshot: it may already include that text.
|
||||
Full item snapshots can replace matching account/host/thread/turn/item identities;
|
||||
they must not create duplicate transcript rows. Disconnects, untyped recovery
|
||||
notifications and uncertain overlap require authoritative history reload. This
|
||||
crate exposes the read primitives; a coordinated resync/display state machine is
|
||||
still required before claiming lossless resume or live/history reconciliation.
|
||||
|
||||
@@ -22,15 +22,20 @@ impl Drop for Fixture {
|
||||
}
|
||||
}
|
||||
async fn fixture(response: http::Response<Body>) -> Fixture {
|
||||
fixture_responses(vec![response]).await
|
||||
}
|
||||
async fn fixture_responses(responses: Vec<http::Response<Body>>) -> Fixture {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let endpoint = format!("http://{}", listener.local_addr().unwrap());
|
||||
let (sender, requests) = mpsc::unbounded_channel();
|
||||
let response = std::sync::Arc::new(std::sync::Mutex::new(Some(response)));
|
||||
let responses = std::sync::Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from(
|
||||
responses,
|
||||
)));
|
||||
let task = tokio::spawn(async move {
|
||||
let (socket, _) = listener.accept().await.unwrap();
|
||||
let service = service_fn(move |request: http::Request<hyper::body::Incoming>| {
|
||||
let sender = sender.clone();
|
||||
let response = response.lock().unwrap().take().unwrap();
|
||||
let response = responses.lock().unwrap().pop_front().unwrap();
|
||||
async move {
|
||||
let (parts, body) = request.into_parts();
|
||||
sender
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
//! Bounded history page reads. Pagination cursors are not live-event replay offsets.
|
||||
use crate::Client;
|
||||
use crate::Error;
|
||||
use crate::ItemSummary;
|
||||
use crate::TurnSummary;
|
||||
use crate::history_wire as wire;
|
||||
use crate::transcript::DecodeFailure;
|
||||
use crate::transcript::required;
|
||||
use bytes::Bytes;
|
||||
use prost::Message;
|
||||
use std::fmt;
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
pub enum HistoryOrder {
|
||||
Ascending,
|
||||
#[default]
|
||||
Descending,
|
||||
}
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
pub enum HistoryAccess {
|
||||
#[default]
|
||||
Default,
|
||||
Shared,
|
||||
}
|
||||
/// A cursor belongs to its original thread and filters. Keep the order for
|
||||
/// `next_cursor` and reverse it for `backwards_cursor`.
|
||||
#[derive(Clone, Default)]
|
||||
pub struct HistoryQuery {
|
||||
pub cursor: Option<String>,
|
||||
pub limit: Option<u32>,
|
||||
pub order: HistoryOrder,
|
||||
pub access: HistoryAccess,
|
||||
}
|
||||
#[derive(Clone)]
|
||||
pub struct ListTurnsRequest {
|
||||
pub thread_id: String,
|
||||
pub page: HistoryQuery,
|
||||
}
|
||||
#[derive(Clone)]
|
||||
pub struct ListItemsRequest {
|
||||
pub thread_id: String,
|
||||
pub turn_id: Option<String>,
|
||||
pub page: HistoryQuery,
|
||||
}
|
||||
/// One server page in server order. No automatic pagination, deduplication or retries.
|
||||
pub struct HistoryPage<T> {
|
||||
pub thread_id: String,
|
||||
pub data: Vec<T>,
|
||||
pub next_cursor: Option<String>,
|
||||
pub backwards_cursor: Option<String>,
|
||||
payload: Bytes,
|
||||
}
|
||||
impl<T> HistoryPage<T> {
|
||||
pub fn payload(&self) -> &Bytes {
|
||||
&self.payload
|
||||
}
|
||||
}
|
||||
pub struct HistoryTurn {
|
||||
pub summary: TurnSummary,
|
||||
payload: Bytes,
|
||||
}
|
||||
impl HistoryTurn {
|
||||
/// Original Turn protobuf; includes metadata omitted from the summary.
|
||||
pub fn payload(&self) -> &Bytes {
|
||||
&self.payload
|
||||
}
|
||||
}
|
||||
pub struct HistoryItem {
|
||||
pub turn_id: String,
|
||||
/// Unsupported items and command sources have no summary and retain their original entry.
|
||||
pub summary: Option<ItemSummary>,
|
||||
pub started_at_ms: Option<i64>,
|
||||
pub completed_at_ms: Option<i64>,
|
||||
payload: Bytes,
|
||||
}
|
||||
impl HistoryItem {
|
||||
/// Original ThreadItemEntry protobuf, including the item and untyped metadata.
|
||||
pub fn payload(&self) -> &Bytes {
|
||||
&self.payload
|
||||
}
|
||||
}
|
||||
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!(
|
||||
HistoryQuery,
|
||||
ListTurnsRequest,
|
||||
ListItemsRequest,
|
||||
HistoryTurn,
|
||||
HistoryItem
|
||||
);
|
||||
impl<T> fmt::Debug for HistoryPage<T> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str("HistoryPage([redacted])")
|
||||
}
|
||||
}
|
||||
impl HistoryQuery {
|
||||
fn order(&self) -> Option<i32> {
|
||||
Some(match self.order {
|
||||
HistoryOrder::Ascending => 1,
|
||||
HistoryOrder::Descending => 2,
|
||||
})
|
||||
}
|
||||
fn access(&self) -> Option<i32> {
|
||||
match self.access {
|
||||
HistoryAccess::Default => None,
|
||||
HistoryAccess::Shared => Some(1),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl Client {
|
||||
/// Reads stored turn metadata; use `list_items` for paginated item content.
|
||||
pub async fn list_turns(
|
||||
&self,
|
||||
request: &ListTurnsRequest,
|
||||
) -> Result<HistoryPage<HistoryTurn>, Error> {
|
||||
let payload = self
|
||||
.history_rpc(
|
||||
wire::TurnsRequest {
|
||||
thread_id: request.thread_id.clone(),
|
||||
cursor: request.page.cursor.clone(),
|
||||
limit: request.page.limit,
|
||||
order: request.page.order(),
|
||||
access: request.page.access(),
|
||||
},
|
||||
"/oaiproto.codex_app_server_backend.ThreadService/ListTurns",
|
||||
)
|
||||
.await?;
|
||||
decode_page(&request.thread_id, payload, |payload| {
|
||||
let turn = crate::transcript_wire::Turn::decode(payload.clone())?;
|
||||
Ok(HistoryTurn {
|
||||
summary: turn.try_into()?,
|
||||
payload,
|
||||
})
|
||||
})
|
||||
.map_err(|_| Error::InvalidResponse)
|
||||
}
|
||||
/// Reads one page, across the thread or restricted to one turn. This is not an
|
||||
/// atomic snapshot synchronized with Attach; do not append buffered deltas to it.
|
||||
pub async fn list_items(
|
||||
&self,
|
||||
request: &ListItemsRequest,
|
||||
) -> Result<HistoryPage<HistoryItem>, Error> {
|
||||
let payload = self
|
||||
.history_rpc(
|
||||
wire::ItemsRequest {
|
||||
thread_id: request.thread_id.clone(),
|
||||
turn_id: request.turn_id.clone(),
|
||||
cursor: request.page.cursor.clone(),
|
||||
limit: request.page.limit,
|
||||
order: request.page.order(),
|
||||
access: request.page.access(),
|
||||
},
|
||||
"/oaiproto.codex_app_server_backend.ThreadService/ListItems",
|
||||
)
|
||||
.await?;
|
||||
decode_page(&request.thread_id, payload, |payload| {
|
||||
let entry = wire::ItemEntry::decode(payload.clone())?;
|
||||
let turn_id = required(entry.turn_id)?;
|
||||
if request
|
||||
.turn_id
|
||||
.as_ref()
|
||||
.is_some_and(|expected| expected != &turn_id)
|
||||
{
|
||||
return Err(DecodeFailure::InvalidResponse);
|
||||
}
|
||||
Ok(HistoryItem {
|
||||
turn_id,
|
||||
summary: crate::items::summary(required(entry.item)?)?,
|
||||
started_at_ms: entry.started_at_ms,
|
||||
completed_at_ms: entry.completed_at_ms,
|
||||
payload,
|
||||
})
|
||||
})
|
||||
.map_err(|_| Error::InvalidResponse)
|
||||
}
|
||||
async fn history_rpc<T: Message + Send + 'static>(
|
||||
&self,
|
||||
value: T,
|
||||
path: &'static str,
|
||||
) -> Result<Bytes, Error> {
|
||||
let mut grpc = self.grpc.clone();
|
||||
let mut request = tonic::Request::new(value);
|
||||
*request.metadata_mut() =
|
||||
tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone());
|
||||
let response = tokio::time::timeout(
|
||||
crate::REQUEST_TIMEOUT,
|
||||
grpc.unary(
|
||||
request,
|
||||
http::uri::PathAndQuery::from_static(path),
|
||||
wire::RawResponseCodec::<T>(std::marker::PhantomData),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| Error::Timeout)?
|
||||
.map_err(Error::from)?;
|
||||
Ok(response.into_inner())
|
||||
}
|
||||
}
|
||||
fn decode_page<T>(
|
||||
thread_id: &str,
|
||||
payload: Bytes,
|
||||
decode: impl Fn(Bytes) -> Result<T, DecodeFailure>,
|
||||
) -> Result<HistoryPage<T>, DecodeFailure> {
|
||||
let page = wire::Page::decode(payload.clone())?;
|
||||
let data = required(page.data)?
|
||||
.values
|
||||
.into_iter()
|
||||
.map(decode)
|
||||
.collect::<Result<_, _>>()?;
|
||||
Ok(HistoryPage {
|
||||
thread_id: thread_id.to_owned(),
|
||||
data,
|
||||
next_cursor: page.next_cursor,
|
||||
backwards_cursor: page.backwards_cursor,
|
||||
payload,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,245 @@
|
||||
//! Independent wire fixtures cover page requests, presence, provenance and unknown data.
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
fn page(rows: &[&[u8]]) -> Vec<u8> {
|
||||
message(
|
||||
/*tag*/ 1,
|
||||
&rows
|
||||
.iter()
|
||||
.flat_map(|row| message(/*tag*/ 1, row))
|
||||
.collect::<Vec<_>>(),
|
||||
)
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn item_history_paginates_with_explicit_filters_and_preserves_rows() {
|
||||
let entry =
|
||||
b"\x0a\x01u\x12\x0f\x1a\x0d\x12\x01i\x32\x01d\x2a\x05\x0a\x03\x12\x01Q\x18\x65\x20\x67";
|
||||
// A future command source alongside a known agent message.
|
||||
let command = b"\x12\x01c\x1a\x00\x22\x01w\x3a\x01i\x58\x63\x60\x03";
|
||||
let item = message(/*tag*/ 7, command);
|
||||
let opaque = [message(/*tag*/ 1, b"u"), message(/*tag*/ 2, &item)].concat();
|
||||
let mut first = page(&[entry, &opaque]);
|
||||
first.extend(b"\x12\x01n\x1a\x01b\xa0\x06\x01");
|
||||
let unknown = b"\x0a\x01u\x12\x03\xa2\x01\x00";
|
||||
let second = page(&[unknown]);
|
||||
let mut fixture = fixture_responses(vec![
|
||||
response(vec![frame(&first)], tonic::Status::ok("")),
|
||||
response(vec![frame(&second)], tonic::Status::ok("")),
|
||||
])
|
||||
.await;
|
||||
let mut request = ListItemsRequest {
|
||||
thread_id: "t".into(),
|
||||
turn_id: Some("u".into()),
|
||||
page: HistoryQuery {
|
||||
cursor: None,
|
||||
limit: Some(2),
|
||||
order: HistoryOrder::Ascending,
|
||||
access: HistoryAccess::Shared,
|
||||
},
|
||||
};
|
||||
let first_page = fixture.client.list_items(&request).await.unwrap();
|
||||
assert_request(
|
||||
&mut fixture,
|
||||
"ListItems",
|
||||
b"\x0a\x01t\x12\x01u\x20\x02\x28\x01\x30\x01",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
(
|
||||
&first_page.thread_id,
|
||||
&first_page.next_cursor,
|
||||
&first_page.backwards_cursor,
|
||||
first_page.payload()
|
||||
),
|
||||
(
|
||||
&"t".to_owned(),
|
||||
&Some("n".into()),
|
||||
&Some("b".into()),
|
||||
&Bytes::from(first)
|
||||
)
|
||||
);
|
||||
let [item, unknown_source] = first_page.data.as_slice() else {
|
||||
panic!("expected two items")
|
||||
};
|
||||
assert_eq!(unknown_source.summary, None);
|
||||
assert_eq!(unknown_source.payload().as_ref(), opaque.as_slice());
|
||||
assert_eq!(
|
||||
(
|
||||
&item.turn_id,
|
||||
&item.summary,
|
||||
item.started_at_ms,
|
||||
item.completed_at_ms,
|
||||
item.payload()
|
||||
),
|
||||
(
|
||||
&"u".to_owned(),
|
||||
&Some(ItemSummary::AgentMessage {
|
||||
id: "i".into(),
|
||||
text: "d".into(),
|
||||
questions: Some(vec![AgentQuestion {
|
||||
title: "Q".into(),
|
||||
options: None
|
||||
}]),
|
||||
}),
|
||||
Some(101),
|
||||
Some(103),
|
||||
&Bytes::from_static(entry)
|
||||
)
|
||||
);
|
||||
request.page.cursor = first_page.next_cursor;
|
||||
let second_page = fixture.client.list_items(&request).await.unwrap();
|
||||
assert_request(
|
||||
&mut fixture,
|
||||
"ListItems",
|
||||
b"\x0a\x01t\x12\x01u\x1a\x01n\x20\x02\x28\x01\x30\x01",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
(second_page.next_cursor, second_page.backwards_cursor),
|
||||
(None, None)
|
||||
);
|
||||
assert_eq!(second_page.data.len(), 1);
|
||||
assert_eq!(
|
||||
(
|
||||
&second_page.data[0].summary,
|
||||
second_page.data[0].started_at_ms,
|
||||
second_page.data[0].payload()
|
||||
),
|
||||
(&None, None, &Bytes::from_static(unknown))
|
||||
);
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn turn_history_reuses_summary_and_retains_omitted_metadata() {
|
||||
let turn = b"\x08\x65\x10\x02\x22\x01u\x2a\x00\x38\x63\x40\x03";
|
||||
let payload = page(&[turn]);
|
||||
let mut fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
|
||||
let result = fixture
|
||||
.client
|
||||
.list_turns(&ListTurnsRequest {
|
||||
thread_id: "t".into(),
|
||||
page: HistoryQuery {
|
||||
cursor: Some("c".into()),
|
||||
..Default::default()
|
||||
},
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_request(&mut fixture, "ListTurns", b"\x0a\x01t\x12\x01c\x20\x02").await;
|
||||
assert_eq!(result.data.len(), 1);
|
||||
assert_eq!(
|
||||
(
|
||||
&result.thread_id,
|
||||
&result.data[0].summary,
|
||||
result.data[0].payload(),
|
||||
result.payload()
|
||||
),
|
||||
(
|
||||
&"t".to_owned(),
|
||||
&TurnSummary {
|
||||
id: "u".into(),
|
||||
status: TurnStatus::Failed,
|
||||
started_at: Some(99),
|
||||
completed_at: Some(101),
|
||||
duration_ms: Some(2),
|
||||
error: None
|
||||
},
|
||||
&Bytes::from_static(turn),
|
||||
&Bytes::from(payload)
|
||||
)
|
||||
);
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn history_distinguishes_empty_pages_from_missing_wrappers_and_bad_rows() {
|
||||
let valid_empty = page(&[]);
|
||||
let fixture = fixture(response(vec![frame(&valid_empty)], tonic::Status::ok(""))).await;
|
||||
let request = ListItemsRequest {
|
||||
thread_id: "t".into(),
|
||||
turn_id: Some("u".into()),
|
||||
page: HistoryQuery::default(),
|
||||
};
|
||||
assert!(
|
||||
fixture
|
||||
.client
|
||||
.list_items(&request)
|
||||
.await
|
||||
.unwrap()
|
||||
.data
|
||||
.is_empty()
|
||||
);
|
||||
for payload in [
|
||||
vec![],
|
||||
page(&[b"\x0a\x01u"]),
|
||||
page(&[b"\x12\x00"]),
|
||||
page(&[b"\x0a\x01u\x12\x05\x1a\x03\x12\x01i"]), // Agent message without text.
|
||||
page(&[b"\x0a\x01x\x12\x00"]),
|
||||
// Unknown command source and missing required status.
|
||||
page(&[b"\x0a\x01u\x12\x0f\x3a\x0d\x12\x01c\x1a\x00\x22\x01w\x3a\x01i\x58\x63"]),
|
||||
vec![10, 127],
|
||||
] {
|
||||
let server = super::fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
|
||||
assert!(matches!(
|
||||
server.client.list_items(&request).await,
|
||||
Err(Error::InvalidResponse)
|
||||
));
|
||||
}
|
||||
for turn in [b"\x22\x01u".as_slice(), b"\x40\x01"] {
|
||||
let server =
|
||||
super::fixture(response(vec![frame(&page(&[turn]))], tonic::Status::ok(""))).await;
|
||||
assert!(matches!(
|
||||
server
|
||||
.client
|
||||
.list_turns(&ListTurnsRequest {
|
||||
thread_id: "t".into(),
|
||||
page: HistoryQuery::default()
|
||||
})
|
||||
.await,
|
||||
Err(Error::InvalidResponse)
|
||||
));
|
||||
}
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn history_errors_preserve_status_without_retrying() {
|
||||
let mut fixture = fixture(response(
|
||||
vec![],
|
||||
tonic::Status::permission_denied("private error"),
|
||||
))
|
||||
.await;
|
||||
let error = fixture
|
||||
.client
|
||||
.list_items(&ListItemsRequest {
|
||||
thread_id: "t".into(),
|
||||
turn_id: None,
|
||||
page: HistoryQuery::default(),
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(!format!("{error:?}").contains("private error"));
|
||||
assert!(matches!(error, Error::Rpc(status) if status.code == 7));
|
||||
assert_request(&mut fixture, "ListItems", b"\x0a\x01t\x28\x02").await;
|
||||
assert!(fixture.requests.try_recv().is_err());
|
||||
}
|
||||
#[test]
|
||||
fn history_diagnostics_redact_each_public_payload_type() {
|
||||
let query = HistoryQuery {
|
||||
cursor: Some("secret".into()),
|
||||
..Default::default()
|
||||
};
|
||||
let items = ListItemsRequest {
|
||||
thread_id: "secret".into(),
|
||||
turn_id: Some("secret".into()),
|
||||
page: query.clone(),
|
||||
};
|
||||
let turns = ListTurnsRequest {
|
||||
thread_id: "secret".into(),
|
||||
page: query.clone(),
|
||||
};
|
||||
for value in [
|
||||
format!("{query:?}"),
|
||||
format!("{items:?}"),
|
||||
format!("{turns:?}"),
|
||||
] {
|
||||
assert!(value.ends_with("([redacted])"));
|
||||
assert!(!value.contains("secret"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
//! 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())))
|
||||
}
|
||||
}
|
||||
@@ -215,17 +215,17 @@ pub(crate) fn summary(item: wire::Item) -> Result<Option<ItemSummary>, DecodeFai
|
||||
.into_iter()
|
||||
.map(command_action)
|
||||
.collect::<Result<_, _>>()?,
|
||||
status: required(value.status)?.into(),
|
||||
source: match value.source {
|
||||
None | Some(1) => CommandExecutionSource::Agent,
|
||||
Some(2) => CommandExecutionSource::UserShell,
|
||||
Some(3) => CommandExecutionSource::UnifiedExecStartup,
|
||||
Some(4) => CommandExecutionSource::UnifiedExecInteraction,
|
||||
Some(_) => return Err(DecodeFailure::UnsupportedCommandSource),
|
||||
Some(_) => return Ok(None),
|
||||
},
|
||||
aggregated_output: value.output,
|
||||
duration_ms: value.duration_ms,
|
||||
exit_code: value.exit_code,
|
||||
status: required(value.status)?.into(),
|
||||
}),
|
||||
Some(wire::Value::FileChange(value)) => ItemSummary::FileChange {
|
||||
id: required(value.id)?,
|
||||
|
||||
@@ -4,6 +4,8 @@
|
||||
//! a timeout or disconnect can leave Resume admitted. Attach delivers live events only;
|
||||
//! dropping its stream detaches without interrupting the thread or answering approvals.
|
||||
|
||||
mod history;
|
||||
mod history_wire;
|
||||
mod item_wire;
|
||||
mod items;
|
||||
mod transcript;
|
||||
@@ -12,6 +14,14 @@ mod turns_wire;
|
||||
mod types;
|
||||
mod wire;
|
||||
|
||||
pub use history::HistoryAccess;
|
||||
pub use history::HistoryItem;
|
||||
pub use history::HistoryOrder;
|
||||
pub use history::HistoryPage;
|
||||
pub use history::HistoryQuery;
|
||||
pub use history::HistoryTurn;
|
||||
pub use history::ListItemsRequest;
|
||||
pub use history::ListTurnsRequest;
|
||||
pub use items::AgentQuestion;
|
||||
pub use items::CommandAction;
|
||||
pub use items::CommandExecutionSource;
|
||||
|
||||
@@ -108,7 +108,6 @@ const ITEM_FILE_CHANGE_PATCH_UPDATED: u32 = 40;
|
||||
|
||||
pub(super) enum DecodeFailure {
|
||||
MissingField,
|
||||
UnsupportedCommandSource,
|
||||
InvalidResponse,
|
||||
}
|
||||
|
||||
@@ -125,7 +124,7 @@ impl TranscriptEvent {
|
||||
pub(crate) fn decode(thread_id: &str, payload: &[u8]) -> Result<Option<Self>, Error> {
|
||||
match Self::decode_present_fields(thread_id, payload) {
|
||||
Ok(event) => Ok(event),
|
||||
Err(DecodeFailure::MissingField | DecodeFailure::UnsupportedCommandSource) => Ok(None),
|
||||
Err(DecodeFailure::MissingField) => Ok(None),
|
||||
Err(DecodeFailure::InvalidResponse) => Err(Error::InvalidResponse),
|
||||
}
|
||||
}
|
||||
@@ -185,31 +184,7 @@ impl TranscriptEvent {
|
||||
TURN_STARTED | TURN_COMPLETED => {
|
||||
let envelope = wire::merge_messages::<wire::Envelope<wire::TurnUpdate>>(values)?;
|
||||
let params = required(envelope.params)?;
|
||||
let turn = required(params.turn)?;
|
||||
let status = match required(turn.status)? {
|
||||
1 => TurnStatus::Completed,
|
||||
2 => TurnStatus::Interrupted,
|
||||
3 => TurnStatus::Failed,
|
||||
4 => TurnStatus::InProgress,
|
||||
value => TurnStatus::Unknown(value),
|
||||
};
|
||||
let error = turn
|
||||
.error
|
||||
.map(|error| {
|
||||
Ok::<_, DecodeFailure>(TurnFailure {
|
||||
message: required(error.message)?,
|
||||
additional_details: error.additional_details,
|
||||
})
|
||||
})
|
||||
.transpose()?;
|
||||
let summary = TurnSummary {
|
||||
id: required(turn.id)?,
|
||||
status,
|
||||
started_at: turn.started_at,
|
||||
completed_at: turn.completed_at,
|
||||
duration_ms: turn.duration_ms,
|
||||
error,
|
||||
};
|
||||
let summary = TurnSummary::try_from(required(params.turn)?)?;
|
||||
let update = if tag == TURN_STARTED {
|
||||
TranscriptUpdate::TurnStarted(summary)
|
||||
} else {
|
||||
@@ -254,3 +229,34 @@ impl TryFrom<wire::Change> for FileUpdate {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<wire::Turn> for TurnSummary {
|
||||
type Error = DecodeFailure;
|
||||
|
||||
fn try_from(turn: wire::Turn) -> Result<Self, DecodeFailure> {
|
||||
let status = match required(turn.status)? {
|
||||
1 => TurnStatus::Completed,
|
||||
2 => TurnStatus::Interrupted,
|
||||
3 => TurnStatus::Failed,
|
||||
4 => TurnStatus::InProgress,
|
||||
value => TurnStatus::Unknown(value),
|
||||
};
|
||||
let error = turn
|
||||
.error
|
||||
.map(|error| {
|
||||
Ok::<_, DecodeFailure>(TurnFailure {
|
||||
message: required(error.message)?,
|
||||
additional_details: error.additional_details,
|
||||
})
|
||||
})
|
||||
.transpose()?;
|
||||
Ok(Self {
|
||||
id: required(turn.id)?,
|
||||
status,
|
||||
started_at: turn.started_at,
|
||||
completed_at: turn.completed_at,
|
||||
duration_ms: turn.duration_ms,
|
||||
error,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -295,3 +295,6 @@ fn notification_selection_handles_many_fragments_and_obsolete_malformed_messages
|
||||
|
||||
#[path = "item_tests.rs"]
|
||||
mod items;
|
||||
|
||||
#[path = "history_tests.rs"]
|
||||
mod history;
|
||||
|
||||
Reference in New Issue
Block a user