Forward cloud message item lifecycle notifications (#59524)

## Why

The cloud backend discarded item start and completion events, preventing clients from receiving user message receipts and complete assistant messages through those notifications.

## What changed

Convert cloud user and agent message summaries into `ItemStarted` and `ItemCompleted` notifications. Preserve user submission IDs and text, assistant text and questions, and lifecycle timestamps. Continue ignoring unsupported item types.

## Testing

Add unit tests covering user receipt identity and text, plus completed assistant message text, questions, and timestamp.

GitOrigin-RevId: 3e88129553844a90eea2a006d35f32a0b7d9e53a
This commit is contained in:
Benjamin Carlsson
2026-11-05 00:05:34 +00:00
committed by copyberry
parent a5ed46ca37
commit b2a47c5cdd
3 changed files with 151 additions and 7 deletions
@@ -33,11 +33,11 @@ pub(super) fn turn(summary: cloud::TurnSummary) -> Result<protocol::Turn> {
}
pub(super) fn transcript(
event: cloud::TranscriptEvent,
thread_id: String,
update: cloud::TranscriptUpdate,
) -> Result<Option<protocol::ServerNotification>> {
use protocol::ServerNotification as Notification;
let thread_id = event.thread_id;
Ok(Some(match event.update {
Ok(Some(match update {
cloud::TranscriptUpdate::TurnStarted(summary) => {
Notification::TurnStarted(protocol::TurnStartedNotification {
thread_id,
@@ -58,10 +58,79 @@ pub(super) fn transcript(
delta: delta.delta,
})
}
cloud::TranscriptUpdate::ItemStarted(_)
| cloud::TranscriptUpdate::ItemCompleted(_)
| cloud::TranscriptUpdate::CommandOutputDelta(_)
cloud::TranscriptUpdate::ItemStarted(item) => {
let Some(content) = item_summary(item.item) else {
return Ok(None);
};
Notification::ItemStarted(protocol::ItemStartedNotification::new(
thread_id,
item.turn_id,
content.payload,
item.at_ms,
))
}
cloud::TranscriptUpdate::ItemCompleted(item) => {
let Some(content) = item_summary(item.item) else {
return Ok(None);
};
Notification::ItemCompleted(protocol::ItemCompletedNotification::new(
thread_id,
item.turn_id,
content.payload,
/*started_at_ms*/ None,
item.at_ms,
))
}
cloud::TranscriptUpdate::CommandOutputDelta(_)
| cloud::TranscriptUpdate::FileChangeOutputDelta(_)
| cloud::TranscriptUpdate::FileChangePatchUpdated(_) => return Ok(None),
}))
}
pub(super) fn item_summary(item: cloud::ItemSummary) -> Option<protocol::ThreadItem> {
let payload = match item {
cloud::ItemSummary::UserMessage {
id,
client_id,
text,
} => protocol::ThreadItemPayload::UserMessage {
id,
client_id,
content: text
.into_iter()
.map(|text| protocol::UserInput::Text {
text,
text_elements: Vec::new(),
})
.collect(),
},
cloud::ItemSummary::AgentMessage {
id,
text,
questions,
} => protocol::ThreadItemPayload::AgentMessage {
id,
text,
phase: None,
memory_citation: None,
delivery: None,
questions: questions.map(|questions| {
questions
.into_iter()
.map(|question| protocol::AsyncUserInputQuestion {
title: question.title,
options: question.options,
})
.collect()
}),
},
cloud::ItemSummary::Reasoning { .. }
| cloud::ItemSummary::CommandExecution(_)
| cloud::ItemSummary::FileChange { .. } => return None,
};
Some(protocol::ThreadItem::untimed(payload))
}
#[cfg(test)]
#[path = "events_tests.rs"]
mod tests;
@@ -0,0 +1,75 @@
#![cfg(test)]
//! Covers the user receipt and final assistant message presented by Cloud.
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn text_receipt_preserves_submission_identity() {
let actual = item_summary(cloud::ItemSummary::UserMessage {
id: "item-1".into(),
client_id: Some("submission-1".into()),
text: vec!["hello".into(), "world".into()],
});
assert_eq!(
actual,
Some(protocol::ThreadItem::untimed(
protocol::ThreadItemPayload::UserMessage {
id: "item-1".into(),
client_id: Some("submission-1".into()),
content: vec![
protocol::UserInput::Text {
text: "hello".into(),
text_elements: Vec::new()
},
protocol::UserInput::Text {
text: "world".into(),
text_elements: Vec::new()
},
],
},
))
);
}
#[test]
fn completed_reply_preserves_the_full_text_questions_and_timestamp() -> Result<()> {
let item = cloud::ItemSummary::AgentMessage {
id: "reply".into(),
text: "Choose one".into(),
questions: Some(vec![cloud::AgentQuestion {
title: "Which?".into(),
options: Some(vec!["One".into(), "Two".into()]),
}]),
};
let update = cloud::TranscriptUpdate::ItemCompleted(cloud::ItemLifecycle {
turn_id: "turn".into(),
at_ms: 45,
item,
});
let Some(protocol::ServerNotification::ItemCompleted(actual)) =
transcript("thread".into(), update)?
else {
panic!("expected a completed message");
};
assert_eq!(
actual,
protocol::ItemCompletedNotification::new(
"thread".into(),
"turn".into(),
protocol::ThreadItemPayload::AgentMessage {
id: "reply".into(),
text: "Choose one".into(),
phase: None,
memory_citation: None,
delivery: None,
questions: Some(vec![protocol::AsyncUserInputQuestion {
title: "Which?".into(),
options: Some(vec!["One".into(), "Two".into()]),
}]),
},
/*started_at_ms*/ None,
/*completed_at_ms*/ 45
)
);
Ok(())
}
@@ -238,7 +238,7 @@ fn forward(
return;
}
let notification = match event {
cloud::Event::Transcript(event) => match events::transcript(*event) {
cloud::Event::Transcript(event) => match events::transcript(event.thread_id, event.update) {
Ok(Some(notification)) => notification,
Ok(None) => return,
Err(_) => cloud_error(id, "The cloud sent an unsupported transcript event."),