diff --git a/codex-rs/cloud-client/README.md b/codex-rs/cloud-client/README.md index 7d357f8fcf..0763ab662e 100644 --- a/codex-rs/cloud-client/README.md +++ b/codex-rs/cloud-client/README.md @@ -20,11 +20,14 @@ stopping the thread or answering approvals. Streams end after their first error. Resume and Attach setup have a 150-second local deadline; live streams have no idle deadline. Setup does not send a gRPC deadline that would expire a healthy stream. -Event variants distinguish notifications and server requests; payloads remain the -complete nested protobuf message bytes, preserving unknown fields. These are not -ProtoJSON or app-server JSON-RPC. Native status code, message and binary details -are available explicitly; credentials, event and error diagnostics are redacted. -Full payload typing, history, approval replies and reconnect are subsequent slices. +`Event::Transcript` decodes selected text, command output, patch, and turn updates. +It validates the thread ID and exposes the decoded fields. Turn updates include +start time, completion time, and duration when the server provides them. +Other notifications and server requests remain opaque protobuf payloads, preserving +unknown fields. These are not ProtoJSON or app-server JSON-RPC. Native status +code, message and binary details are available explicitly; credentials, event +and error diagnostics are redacted. History, 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. diff --git a/codex-rs/cloud-client/src/client_tests.rs b/codex-rs/cloud-client/src/client_tests.rs index f571f4ce38..931c269754 100644 --- a/codex-rs/cloud-client/src/client_tests.rs +++ b/codex-rs/cloud-client/src/client_tests.rs @@ -346,3 +346,5 @@ async fn https_enforces_unmanaged_endpoint_restrictions_before_connecting() { .await .unwrap(); } +#[path = "transcript_tests.rs"] +mod transcript; diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs index ff1fdd5540..4b09016dd8 100644 --- a/codex-rs/cloud-client/src/lib.rs +++ b/codex-rs/cloud-client/src/lib.rs @@ -4,9 +4,20 @@ //! 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 transcript; +mod transcript_wire; mod types; mod wire; +pub use transcript::FileUpdate; +pub use transcript::ItemDelta; +pub use transcript::PatchKind; +pub use transcript::PatchUpdate; +pub use transcript::TranscriptEvent; +pub use transcript::TranscriptUpdate; +pub use transcript::TurnFailure; +pub use transcript::TurnStatus; +pub use transcript::TurnSummary; pub use types::Credentials; pub use types::Error; pub use types::Event; @@ -160,10 +171,16 @@ impl Client { .map_err(|_| Error::Timeout)? .map_err(Error::from)?; let mut stream = response.into_inner(); + let thread_id = thread_id.to_owned(); Ok(async_stream::try_stream! { while let Some(message) = stream.message().await.map_err(Error::from)? { yield match message.event.ok_or(Error::InvalidResponse)? { - wire::Payload::Notification(bytes) => Event::Notification(bytes), + wire::Payload::Notification(bytes) => { + match TranscriptEvent::decode(&thread_id, &bytes)? { + Some(event) => Event::Transcript(Box::new(event)), + None => Event::Notification(bytes), + } + }, wire::Payload::ServerRequest(bytes) => Event::ServerRequest(bytes), }; } diff --git a/codex-rs/cloud-client/src/transcript.rs b/codex-rs/cloud-client/src/transcript.rs new file mode 100644 index 0000000000..030cc73e20 --- /dev/null +++ b/codex-rs/cloud-client/src/transcript.rs @@ -0,0 +1,241 @@ +//! Typed live transcript updates for the attached thread. +use crate::Error; +use crate::transcript_wire as wire; +use std::fmt; + +/// A live update, not a history snapshot or a replay guarantee. +#[derive(Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct TranscriptEvent { + pub thread_id: String, + pub emitted_at_ms: Option, + pub update: TranscriptUpdate, +} + +#[derive(Clone, PartialEq, Eq)] +pub enum TranscriptUpdate { + AgentMessageDelta(ItemDelta), + CommandOutputDelta(ItemDelta), + FileChangeOutputDelta(ItemDelta), + FileChangePatchUpdated(PatchUpdate), + TurnStarted(TurnSummary), + TurnCompleted(TurnSummary), +} +#[derive(Clone, PartialEq, Eq)] +pub struct ItemDelta { + pub item_id: String, + pub turn_id: String, + pub delta: String, +} +#[derive(Clone, PartialEq, Eq)] +pub struct PatchUpdate { + pub item_id: String, + pub turn_id: String, + pub changes: Vec, +} +#[derive(Clone, PartialEq, Eq)] +pub struct FileUpdate { + pub path: String, + pub diff: String, + pub kind: PatchKind, +} +#[derive(Clone, PartialEq, Eq)] +pub enum PatchKind { + Add, + Delete, + Update { + move_path: Option, + }, + /// The server reported an unrecognized change kind. + Unknown, +} +/// Status, timing, and failure details for a turn. +#[derive(Clone, PartialEq, Eq)] +pub struct TurnSummary { + pub id: String, + pub status: TurnStatus, + /// Turn start time in Unix seconds, if reported by the server. + pub started_at: Option, + /// Turn completion time in Unix seconds, if reported by the server. + pub completed_at: Option, + /// Turn duration in milliseconds, if reported by the server. + pub duration_ms: Option, + pub error: Option, +} +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TurnStatus { + Completed, + Interrupted, + Failed, + InProgress, + Unknown(i32), +} +#[derive(Clone, PartialEq, Eq)] +pub struct TurnFailure { + pub message: String, + pub additional_details: Option, +} +// Each public payload type is safe to format independently, not just through Event. +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!( + TranscriptEvent, + TranscriptUpdate, + ItemDelta, + PatchUpdate, + FileUpdate, + PatchKind, + TurnSummary, + TurnFailure +); + +// Field numbers in the ASB Notification.value oneof. +const TURN_STARTED: u32 = 21; +const TURN_COMPLETED: u32 = 23; +const ITEM_AGENT_MESSAGE_DELTA: u32 = 32; +const ITEM_COMMAND_EXECUTION_OUTPUT_DELTA: u32 = 37; +const ITEM_FILE_CHANGE_OUTPUT_DELTA: u32 = 39; +const ITEM_FILE_CHANGE_PATCH_UPDATED: u32 = 40; + +enum DecodeFailure { + MissingField, + InvalidResponse, +} + +impl From for DecodeFailure { + fn from(_: prost::DecodeError) -> Self { + Self::InvalidResponse + } +} + +fn required(value: Option) -> Result { + value.ok_or(DecodeFailure::MissingField) +} +impl TranscriptEvent { + pub(crate) fn decode(thread_id: &str, payload: &[u8]) -> Result, Error> { + match Self::decode_present_fields(thread_id, payload) { + Ok(event) => Ok(event), + Err(DecodeFailure::MissingField) => Ok(None), + Err(DecodeFailure::InvalidResponse) => Err(Error::InvalidResponse), + } + } + + fn decode_present_fields( + thread_id: &str, + payload: &[u8], + ) -> Result, DecodeFailure> { + let Some((tag, values)) = wire::notification(payload)? else { + return Ok(None); + }; + let (owner, emitted_at_ms, update) = match tag { + ITEM_AGENT_MESSAGE_DELTA + | ITEM_COMMAND_EXECUTION_OUTPUT_DELTA + | ITEM_FILE_CHANGE_OUTPUT_DELTA => { + let envelope = wire::merge_messages::>(values)?; + let params = required(envelope.params)?; + let delta = ItemDelta { + item_id: required(params.item_id)?, + turn_id: required(params.turn_id)?, + delta: required(params.delta)?, + }; + let update = match tag { + ITEM_AGENT_MESSAGE_DELTA => TranscriptUpdate::AgentMessageDelta(delta), + ITEM_COMMAND_EXECUTION_OUTPUT_DELTA => { + TranscriptUpdate::CommandOutputDelta(delta) + } + ITEM_FILE_CHANGE_OUTPUT_DELTA => TranscriptUpdate::FileChangeOutputDelta(delta), + _ => return Ok(None), + }; + (required(params.thread_id)?, envelope.emitted_at_ms, update) + } + ITEM_FILE_CHANGE_PATCH_UPDATED => { + let envelope = wire::merge_messages::>(values)?; + let params = required(envelope.params)?; + let changes = required(params.changes)? + .values + .into_iter() + .map(|change| { + let kind = match required(change.kind)?.value { + Some(wire::Kind::Add(_)) => PatchKind::Add, + Some(wire::Kind::Delete(_)) => PatchKind::Delete, + Some(wire::Kind::Update(value)) => PatchKind::Update { + move_path: value.move_path, + }, + None => PatchKind::Unknown, + }; + Ok(FileUpdate { + path: required(change.path)?, + diff: required(change.diff)?, + kind, + }) + }) + .collect::>()?; + let update = PatchUpdate { + item_id: required(params.item_id)?, + turn_id: required(params.turn_id)?, + changes, + }; + ( + required(params.thread_id)?, + envelope.emitted_at_ms, + TranscriptUpdate::FileChangePatchUpdated(update), + ) + } + TURN_STARTED | TURN_COMPLETED => { + let envelope = wire::merge_messages::>(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 update = if tag == TURN_STARTED { + TranscriptUpdate::TurnStarted(summary) + } else { + TranscriptUpdate::TurnCompleted(summary) + }; + (required(params.thread_id)?, envelope.emitted_at_ms, update) + } + _ => return Ok(None), + }; + if owner != thread_id { + // Attach also carries account-wide lifecycle notifications. Preserve them + // as opaque events rather than ending this thread's transcript stream. + return if matches!(tag, TURN_STARTED | TURN_COMPLETED) { + Ok(None) + } else { + Err(DecodeFailure::InvalidResponse) + }; + } + Ok(Some(Self { + thread_id: owner, + emitted_at_ms, + update, + })) + } +} diff --git a/codex-rs/cloud-client/src/transcript_tests.rs b/codex-rs/cloud-client/src/transcript_tests.rs new file mode 100644 index 0000000000..f46e657b9e --- /dev/null +++ b/codex-rs/cloud-client/src/transcript_tests.rs @@ -0,0 +1,290 @@ +//! Independent protobuf fixtures cover the public transcript contract. +use super::*; +use pretty_assertions::assert_eq; + +fn message(tag: u32, payload: &[u8]) -> Vec { + let mut out = Vec::new(); + prost::encoding::encode_key(tag, prost::encoding::WireType::LengthDelimited, &mut out); + prost::encoding::encode_varint(payload.len() as u64, &mut out); + out.extend_from_slice(payload); + out +} + +fn notification(tag: u32, params: &[u8]) -> Vec { + message(tag, &message(/*tag*/ 2, params)) +} + +const DELTA: &[u8] = b"\x0a\x01d\x12\x01i\x1a\x01t\x22\x01u"; + +#[tokio::test] +async fn attach_decodes_live_deltas_with_unknown_fields() { + let fixture = fixture(response( + [32, 37, 39] + .into_iter() + .map(|tag| { + let mut payload = notification(tag, DELTA); + payload.extend([0xa0, 0x06, 1]); // unknown field 100 + frame(&message(/*tag*/ 1, &payload)) + }) + .collect(), + tonic::Status::ok(""), + )) + .await; + let mut stream = fixture.client.attach("t").await.unwrap(); + for expected in [ + TranscriptUpdate::AgentMessageDelta as fn(ItemDelta) -> TranscriptUpdate, + TranscriptUpdate::CommandOutputDelta, + TranscriptUpdate::FileChangeOutputDelta, + ] { + let Event::Transcript(event) = stream.next().await.unwrap().unwrap() else { + panic!("expected transcript event") + }; + assert_eq!(format!("{event:?}"), "TranscriptEvent([redacted])"); + assert_eq!(event.thread_id, "t"); + assert_eq!(event.emitted_at_ms, None); + assert_eq!( + event.update, + expected(ItemDelta { + item_id: "i".into(), + turn_id: "u".into(), + delta: "d".into(), + }) + ); + } + assert!(stream.next().await.is_none()); +} + +#[test] +fn patches_preserve_order_kinds_and_move_paths() { + let mut changes = Vec::new(); + for kind in [ + b"\x0a\x00".as_slice(), + b"\x12\x00", + b"\x1a\x03\x0a\x01m", + b"\x22\x00", + ] { + changes.extend(message( + /*tag*/ 1, + &[ + b"\x0a\x01d".to_vec(), + message(/*tag*/ 2, kind), + b"\x1a\x01p".to_vec(), + ] + .concat(), + )); + } + let params = [message(/*tag*/ 1, &changes), DELTA[3..].to_vec()].concat(); + let payload = notification(/*tag*/ 40, ¶ms); + let event = TranscriptEvent::decode("t", &payload).unwrap().unwrap(); + assert_eq!( + event.update, + TranscriptUpdate::FileChangePatchUpdated(PatchUpdate { + item_id: "i".into(), + turn_id: "u".into(), + changes: [ + PatchKind::Add, + PatchKind::Delete, + PatchKind::Update { + move_path: Some("m".into()) + }, + PatchKind::Unknown, + ] + .into_iter() + .map(|kind| FileUpdate { + path: "p".into(), + diff: "d".into(), + kind, + }) + .collect(), + }) + ); +} + +#[test] +fn turn_events_preserve_status_failure_and_timing() { + for (tag, status, expected) in [ + (23, 1, TurnStatus::Completed), + (23, 2, TurnStatus::Interrupted), + (21, 4, TurnStatus::InProgress), + (23, 3, TurnStatus::Failed), + (23, 99, TurnStatus::Unknown(99)), + ] { + let (timing, started_at, completed_at, duration_ms) = match expected { + TurnStatus::InProgress => (b"\x38\x00".as_slice(), Some(0), None, None), + TurnStatus::Completed => ( + b"\x08\x0d\x10\xf4\x03\x38\x0b".as_slice(), + Some(11), + Some(13), + Some(500), + ), + _ => (&[][..], None, None, None), + }; + let turn = [ + b"\x1a\x06\x0a\x01a\x1a\x01e\x22\x01u\x2a\x00\x40".to_vec(), + vec![status], + timing.to_vec(), + ] + .concat(); + let params = [b"\x0a\x01t".to_vec(), message(/*tag*/ 2, &turn)].concat(); + let event = TranscriptEvent::decode( + "t", + &message(tag, &[vec![8, 7], message(/*tag*/ 2, ¶ms)].concat()), + ) + .unwrap() + .unwrap(); + let turn = TurnSummary { + id: "u".into(), + status: expected, + started_at, + completed_at, + duration_ms, + error: Some(TurnFailure { + message: "e".into(), + additional_details: Some("a".into()), + }), + }; + assert_eq!(event.emitted_at_ms, Some(7)); + assert_eq!( + event.update, + if tag == 21 { + TranscriptUpdate::TurnStarted(turn) + } else { + TranscriptUpdate::TurnCompleted(turn) + } + ); + } +} + +#[tokio::test] +async fn attach_preserves_opaque_notifications_and_continues() { + let foreign_turn = b"\x0a\x01x\x12\x05\x22\x01u\x40\x01"; + let opaque = [ + message(/*tag*/ 100, b"unknown notification"), + notification(/*tag*/ 21, foreign_turn), + notification(/*tag*/ 23, foreign_turn), + notification(/*tag*/ 32, &DELTA[..9]), // missing turn ID + ]; + let valid = notification(/*tag*/ 32, DELTA); + let fixture = fixture(response( + opaque + .iter() + .chain([&valid]) + .map(|payload| frame(&message(/*tag*/ 1, payload))) + .collect(), + tonic::Status::ok(""), + )) + .await; + let mut stream = fixture.client.attach("t").await.unwrap(); + for payload in opaque { + let Some(Ok(Event::Notification(bytes))) = stream.next().await else { + panic!("expected opaque notification"); + }; + assert_eq!(bytes.as_ref(), payload.as_slice()); + } + let Some(Ok(Event::Transcript(event))) = stream.next().await else { + panic!("expected transcript after opaque notifications"); + }; + assert_eq!(event.thread_id, "t"); + assert!(stream.next().await.is_none()); +} + +#[test] +fn decoder_keeps_incomplete_events_opaque_and_rejects_foreign_deltas() { + let mut wrong_owner = DELTA.to_vec(); + wrong_owner[8] = b'x'; + for payload in [ + notification(/*tag*/ 32, &DELTA[3..]), + notification(/*tag*/ 40, &DELTA[3..]), + notification(/*tag*/ 23, b"\x0a\x01t"), + ] { + assert!(TranscriptEvent::decode("t", &payload).unwrap().is_none()); + } + assert!(TranscriptEvent::decode("t", ¬ification(/*tag*/ 32, &wrong_owner)).is_err()); + assert!( + TranscriptEvent::decode("t", &message(/*tag*/ 100, b"")) + .unwrap() + .is_none() + ); +} + +#[test] +fn notification_selection_merges_only_repeated_variants() { + let partial = notification(/*tag*/ 32, &DELTA[..6]); + let tail = notification(/*tag*/ 32, &DELTA[6..]); + assert!( + TranscriptEvent::decode("t", &[partial.clone(), tail.clone()].concat()) + .unwrap() + .is_some() + ); + assert!( + TranscriptEvent::decode("t", &[partial, message(/*tag*/ 85, b""), tail].concat()) + .unwrap() + .is_none() + ); + assert!( + TranscriptEvent::decode( + "t", + &[notification(/*tag*/ 32, DELTA), message(/*tag*/ 89, b"")].concat() + ) + .unwrap() + .is_none() + ); + + let envelope = message(/*tag*/ 2, DELTA); + let split_inside_params = [ + message(/*tag*/ 32, &envelope[..4]), + message(/*tag*/ 32, &envelope[4..]), + ] + .concat(); + assert!(TranscriptEvent::decode("t", &split_inside_params).is_err()); +} + +#[test] +fn public_payload_debug_is_redacted() { + for formatted in [ + format!( + "{:?}", + FileUpdate { + path: "secret".into(), + diff: "secret".into(), + kind: PatchKind::Update { + move_path: Some("secret".into()), + }, + } + ), + format!( + "{:?}", + TurnFailure { + message: "secret".into(), + additional_details: Some("secret".into()), + } + ), + ] { + assert!(formatted.ends_with("([redacted])")); + assert!(!formatted.contains("secret")); + } +} + +#[test] +fn notification_selection_handles_many_fragments_and_obsolete_malformed_messages() { + let mut payload = message(/*tag*/ 32, b"\xff"); + payload.extend(message(/*tag*/ 85, b"")); + for _ in 0..100_000 { + payload.extend(message(/*tag*/ 32, b"")); + } + payload.extend(notification(/*tag*/ 32, &DELTA[..6])); + payload.extend(message(/*tag*/ 100, b"\xff")); + payload.extend(notification(/*tag*/ 32, &DELTA[6..])); + let event = TranscriptEvent::decode("t", &payload).unwrap().unwrap(); + assert_eq!( + event.update, + TranscriptUpdate::AgentMessageDelta(ItemDelta { + item_id: "i".into(), + turn_id: "u".into(), + delta: "d".into(), + }) + ); + + payload.extend(message(/*tag*/ 89, b"\xff")); + assert!(TranscriptEvent::decode("t", &payload).unwrap().is_none()); +} diff --git a/codex-rs/cloud-client/src/transcript_wire.rs b/codex-rs/cloud-client/src/transcript_wire.rs new file mode 100644 index 0000000000..670a6b134b --- /dev/null +++ b/codex-rs/cloud-client/src/transcript_wire.rs @@ -0,0 +1,154 @@ +//! Mirrors the ASB notification fields exposed by this client so it can build without +//! the monorepo's generated bindings. Keep the tag numbers in sync with `Notification.value` +//! in `native_notification.proto`. Only the final oneof variant is decoded on a second pass. +use prost::Message; + +type SelectedNotification<'a> = (u32, SelectedMessages<'a>); + +pub(super) struct SelectedMessages<'a> { + tag: u32, + input: &'a [u8], +} + +// Known oneof members in native_notification.proto replace earlier variants, even +// when unsupported. Fields outside that schema leave the selected variant intact. +pub(super) fn notification( + input: &[u8], +) -> Result>, prost::DecodeError> { + let mut cursor = input; + let mut selected = None; + while !cursor.is_empty() { + let start = input.len() - cursor.len(); + let (tag, kind) = prost::encoding::decode_key(&mut cursor)?; + if (1..=89).contains(&tag) { + prost::encoding::check_wire_type(prost::encoding::WireType::LengthDelimited, kind)?; + if selected.is_none_or(|(previous, _)| previous != tag) { + selected = Some((tag, start)); + } + } + prost::encoding::skip_field(kind, tag, &mut cursor, Default::default())?; + } + Ok(selected.map(|(tag, start)| { + ( + tag, + SelectedMessages { + tag, + input: &input[start..], + }, + ) + })) +} + +pub(super) fn merge_messages( + values: SelectedMessages<'_>, +) -> Result { + let mut message = T::default(); + let mut input = values.input; + while !input.is_empty() { + let (tag, kind) = prost::encoding::decode_key(&mut input)?; + if tag == values.tag { + // The first pass validated the wire type and bounded every length. + let length = prost::encoding::decode_varint(&mut input)? as usize; + let (value, tail) = input.split_at(length); + message.merge(value)?; + input = tail; + } else { + prost::encoding::skip_field(kind, tag, &mut input, Default::default())?; + } + } + Ok(message) +} + +#[derive(Clone, Message)] +pub(super) struct Envelope { + #[prost(int64, optional, tag = "1")] + pub emitted_at_ms: Option, + #[prost(message, optional, tag = "2")] + pub params: Option, +} +#[derive(Clone, Message)] +pub(super) struct PatchUpdate { + #[prost(message, optional, tag = "1")] + pub changes: Option, + #[prost(string, optional, tag = "2")] + pub item_id: Option, + #[prost(string, optional, tag = "3")] + pub thread_id: Option, + #[prost(string, optional, tag = "4")] + pub turn_id: Option, +} +#[derive(Clone, Message)] +pub(super) struct Delta { + #[prost(string, optional, tag = "1")] + pub delta: Option, + #[prost(string, optional, tag = "2")] + pub item_id: Option, + #[prost(string, optional, tag = "3")] + pub thread_id: Option, + #[prost(string, optional, tag = "4")] + pub turn_id: Option, +} +#[derive(Clone, Message)] +pub(super) struct Changes { + #[prost(message, repeated, tag = "1")] + pub values: Vec, +} +#[derive(Clone, Message)] +pub(super) struct Change { + #[prost(string, optional, tag = "1")] + pub diff: Option, + #[prost(message, optional, tag = "2")] + pub kind: Option, + #[prost(string, optional, tag = "3")] + pub path: Option, +} +#[derive(Clone, Message)] +pub(super) struct ChangeKind { + #[prost(oneof = "Kind", tags = "1, 2, 3")] + pub value: Option, +} +#[derive(Clone, prost::Oneof)] +pub(super) enum Kind { + #[prost(message, tag = "1")] + Add(Empty), + #[prost(message, tag = "2")] + Delete(Empty), + #[prost(message, tag = "3")] + Update(Move), +} +#[derive(Clone, Message)] +pub(super) struct Empty {} +#[derive(Clone, Message)] +pub(super) struct Move { + #[prost(string, optional, tag = "1")] + pub move_path: Option, +} +#[derive(Clone, Message)] +pub(super) struct TurnUpdate { + #[prost(string, optional, tag = "1")] + pub thread_id: Option, + #[prost(message, optional, tag = "2")] + pub turn: Option, +} +#[derive(Clone, Message)] +pub(super) struct Turn { + #[prost(int64, optional, tag = "1")] + pub completed_at: Option, + #[prost(int64, optional, tag = "2")] + pub duration_ms: Option, + #[prost(message, optional, tag = "3")] + pub error: Option, + #[prost(string, optional, tag = "4")] + pub id: Option, + #[prost(int64, optional, tag = "7")] + pub started_at: Option, + #[prost(int32, optional, tag = "8")] + pub status: Option, +} +#[derive(Clone, Message)] +pub(super) struct TurnError { + #[prost(string, optional, tag = "1")] + pub additional_details: Option, + #[prost(string, optional, tag = "3")] + pub message: Option, +} diff --git a/codex-rs/cloud-client/src/types.rs b/codex-rs/cloud-client/src/types.rs index 209113f57b..0c09cb4a8f 100644 --- a/codex-rs/cloud-client/src/types.rs +++ b/codex-rs/cloud-client/src/types.rs @@ -46,11 +46,11 @@ pub struct ResumeRequest { pub shared_access: bool, } -/// An Attach event containing the complete nested protobuf message bytes. -/// Payloads are native Notification/ServerRequest messages, not JSON-RPC or ProtoJSON. -/// Keeping them opaque preserves unknown fields until full payload typing is added. +/// An Attach event with selected typed transcript updates and otherwise opaque payloads. +/// The opaque variants carry the original native Notification or ServerRequest protobuf bytes. pub enum Event { Notification(Bytes), + Transcript(Box), ServerRequest(Bytes), } @@ -58,6 +58,7 @@ impl fmt::Debug for Event { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str(match self { Self::Notification(_) => "Notification([redacted])", + Self::Transcript(_) => "Transcript([redacted])", Self::ServerRequest(_) => "ServerRequest([redacted])", }) }