mirror of
https://github.com/openai/codex.git
synced 2026-11-07 20:30:33 +00:00
Decode live transcript updates in the cloud client (#59047)
## Why Attach notifications previously exposed only opaque protobuf bytes. Typed transcript events let callers access live text, command output, file changes, and turn state directly. ## What changed - Add `Event::Transcript` with typed agent-message, command-output, and file-change deltas; patch updates; and turn start/completion events. - Expose patch kinds and move paths, turn status and failure details, and server-provided timestamps and duration. - Validate transcript thread IDs. Preserve unsupported or incomplete notifications and other threads' turn lifecycle events as opaque payloads; reject deltas for another thread. - Keep transcript payloads redacted in `Debug` output. ## Testing Add independent protobuf fixtures covering Attach decoding, opaque fallback and stream continuation, patch ordering, turn timing and failures, thread validation, repeated oneof variants, unknown fields, and debug redaction. GitOrigin-RevId: f26b8d4a5b355bb12159136e83310cb57c1546ca
This commit is contained in:
Benjamin Carlsson
committed by
copyberry
copyberry
parent
7e8143a897
commit
fbf5f0d3a7
@@ -346,3 +346,5 @@ async fn https_enforces_unmanaged_endpoint_restrictions_before_connecting() {
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
#[path = "transcript_tests.rs"]
|
||||
mod transcript;
|
||||
|
||||
@@ -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),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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<i64>,
|
||||
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<FileUpdate>,
|
||||
}
|
||||
#[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<String>,
|
||||
},
|
||||
/// 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<i64>,
|
||||
/// Turn completion time in Unix seconds, if reported by the server.
|
||||
pub completed_at: Option<i64>,
|
||||
/// Turn duration in milliseconds, if reported by the server.
|
||||
pub duration_ms: Option<i64>,
|
||||
pub error: Option<TurnFailure>,
|
||||
}
|
||||
#[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<String>,
|
||||
}
|
||||
// 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<prost::DecodeError> for DecodeFailure {
|
||||
fn from(_: prost::DecodeError) -> Self {
|
||||
Self::InvalidResponse
|
||||
}
|
||||
}
|
||||
|
||||
fn required<T>(value: Option<T>) -> Result<T, DecodeFailure> {
|
||||
value.ok_or(DecodeFailure::MissingField)
|
||||
}
|
||||
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) => Ok(None),
|
||||
Err(DecodeFailure::InvalidResponse) => Err(Error::InvalidResponse),
|
||||
}
|
||||
}
|
||||
|
||||
fn decode_present_fields(
|
||||
thread_id: &str,
|
||||
payload: &[u8],
|
||||
) -> Result<Option<Self>, 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::<wire::Envelope<wire::Delta>>(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::<wire::Envelope<wire::PatchUpdate>>(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::<Result<_, DecodeFailure>>()?;
|
||||
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::<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 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,
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -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<u8> {
|
||||
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<u8> {
|
||||
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());
|
||||
}
|
||||
@@ -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<Option<SelectedNotification<'_>>, 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<T: Message + Default>(
|
||||
values: SelectedMessages<'_>,
|
||||
) -> Result<T, prost::DecodeError> {
|
||||
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<T: Message + Default + std::fmt::Debug> {
|
||||
#[prost(int64, optional, tag = "1")]
|
||||
pub emitted_at_ms: Option<i64>,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
pub params: Option<T>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct PatchUpdate {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
pub changes: Option<Changes>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub item_id: Option<String>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub thread_id: Option<String>,
|
||||
#[prost(string, optional, tag = "4")]
|
||||
pub turn_id: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Delta {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub delta: Option<String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub item_id: Option<String>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub thread_id: Option<String>,
|
||||
#[prost(string, optional, tag = "4")]
|
||||
pub turn_id: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Changes {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
pub values: Vec<Change>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Change {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub diff: Option<String>,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
pub kind: Option<ChangeKind>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub path: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct ChangeKind {
|
||||
#[prost(oneof = "Kind", tags = "1, 2, 3")]
|
||||
pub value: Option<Kind>,
|
||||
}
|
||||
#[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<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct TurnUpdate {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub thread_id: Option<String>,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
pub turn: Option<Turn>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Turn {
|
||||
#[prost(int64, optional, tag = "1")]
|
||||
pub completed_at: Option<i64>,
|
||||
#[prost(int64, optional, tag = "2")]
|
||||
pub duration_ms: Option<i64>,
|
||||
#[prost(message, optional, tag = "3")]
|
||||
pub error: Option<TurnError>,
|
||||
#[prost(string, optional, tag = "4")]
|
||||
pub id: Option<String>,
|
||||
#[prost(int64, optional, tag = "7")]
|
||||
pub started_at: Option<i64>,
|
||||
#[prost(int32, optional, tag = "8")]
|
||||
pub status: Option<i32>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct TurnError {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub additional_details: Option<String>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub message: Option<String>,
|
||||
}
|
||||
@@ -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<crate::TranscriptEvent>),
|
||||
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])",
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user