mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
Add typed item lifecycle events to the cloud client (#59066)
## What changed Decode live item start and completion notifications as `TranscriptUpdate::ItemStarted` and `TranscriptUpdate::ItemCompleted`, with turn identity, lifecycle timestamps, and summaries for user messages, agent messages, reasoning, commands, and file changes. Preserve structured agent questions even when message text is empty, optional command output and exit details, and command or patch status independently of lifecycle completion. Keep unsupported or incomplete items as opaque notifications and redact summary fields in debug output. ## Testing Add fixtures covering lifecycle ordering alongside deltas, identity and timestamps, status and optional-field preservation, structured questions, opaque fallback, malformed payloads, thread ownership, and debug redaction. GitOrigin-RevId: d6bf2d25ee85be17d895726c5072850a9851b55f
This commit is contained in:
Benjamin Carlsson
committed by
copyberry
copyberry
parent
dbaead0106
commit
7ee5efad4a
@@ -26,8 +26,10 @@ 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.
|
||||
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.
|
||||
The minimal envelopes in `wire.rs` and `ResumeRequest` mirror ThreadService field
|
||||
numbers; no monorepo checkout or protobuf compiler is required to build this crate.
|
||||
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
//! Item fixtures cover lifecycle order, outcomes, presence, and opaque fallback.
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
use prost::Message;
|
||||
|
||||
const AGENT: &[u8] = b"\x1a\x06\x12\x01i\x32\x01d";
|
||||
|
||||
const COMMAND: &[u8] =
|
||||
b"\x0a\x00\x12\x01c\x1a\x07\x0a\x05\x22\x03\x0a\x01c\x22\x01w\x28\x00\x30\x00\x3a\x01i\x60\x03";
|
||||
|
||||
fn lifecycle(tag: u32, item: &[u8]) -> Vec<u8> {
|
||||
let (at_ms, emitted_at_ms) = if tag == 27 { (3, 7) } else { (5, 11) };
|
||||
let params = if tag == 27 {
|
||||
[
|
||||
message(/*tag*/ 1, item),
|
||||
vec![16, at_ms as u8],
|
||||
b"\x1a\x01t\x22\x01u".to_vec(),
|
||||
]
|
||||
.concat()
|
||||
} else {
|
||||
[
|
||||
vec![8, at_ms as u8],
|
||||
message(/*tag*/ 2, item),
|
||||
b"\x1a\x01t\x22\x01u".to_vec(),
|
||||
]
|
||||
.concat()
|
||||
};
|
||||
message(
|
||||
tag,
|
||||
&[vec![8, emitted_at_ms as u8], message(/*tag*/ 2, ¶ms)].concat(),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn item_lifecycle_keeps_identity_and_timestamps() {
|
||||
let payloads = [
|
||||
lifecycle(/*tag*/ 27, AGENT),
|
||||
notification(/*tag*/ 32, DELTA),
|
||||
lifecycle(/*tag*/ 31, AGENT),
|
||||
];
|
||||
let fixture = fixture(response(
|
||||
payloads
|
||||
.iter()
|
||||
.map(|payload| frame(&message(/*tag*/ 1, payload)))
|
||||
.collect(),
|
||||
tonic::Status::ok(""),
|
||||
))
|
||||
.await;
|
||||
let mut stream = fixture.client.attach("t").await.unwrap();
|
||||
let item = |at_ms| ItemLifecycle {
|
||||
turn_id: "u".into(),
|
||||
at_ms,
|
||||
item: ItemSummary::AgentMessage {
|
||||
id: "i".into(),
|
||||
text: "d".into(),
|
||||
questions: None,
|
||||
},
|
||||
};
|
||||
let expected = [
|
||||
(Some(7), TranscriptUpdate::ItemStarted(item(3))),
|
||||
(
|
||||
None,
|
||||
TranscriptUpdate::AgentMessageDelta(ItemDelta {
|
||||
item_id: "i".into(),
|
||||
turn_id: "u".into(),
|
||||
delta: "d".into(),
|
||||
}),
|
||||
),
|
||||
(Some(11), TranscriptUpdate::ItemCompleted(item(5))),
|
||||
];
|
||||
for (emitted_at_ms, update) in expected {
|
||||
let Event::Transcript(event) = stream.next().await.unwrap().unwrap() else {
|
||||
panic!("expected transcript event")
|
||||
};
|
||||
assert_eq!(
|
||||
(&event.thread_id, event.emitted_at_ms, &event.update),
|
||||
(&"t".to_owned(), emitted_at_ms, &update)
|
||||
);
|
||||
}
|
||||
assert!(stream.next().await.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_message_preserves_client_id_and_text() {
|
||||
let user = b"\x0a\x23\x0a\x09client-id\x12\x0b\x0a\x09\x0a\x07\x0a\x05hello\x1a\x09user-item";
|
||||
assert_eq!(
|
||||
summary(user),
|
||||
Some(ItemSummary::UserMessage {
|
||||
id: "user-item".into(),
|
||||
client_id: Some("client-id".into()),
|
||||
text: vec!["hello".into()],
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
fn summary(bytes: &[u8]) -> Option<ItemSummary> {
|
||||
crate::items::summary(crate::item_wire::Item::decode(bytes).unwrap())
|
||||
.unwrap_or_else(|_| panic!("expected supported item fields"))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn command_and_patch_summaries_preserve_status_and_presence() {
|
||||
let command = CommandSummary {
|
||||
id: "i".into(),
|
||||
command: "c".into(),
|
||||
cwd: "w".into(),
|
||||
actions: vec![CommandAction::Unknown {
|
||||
command: "c".into(),
|
||||
}],
|
||||
source: CommandExecutionSource::Agent,
|
||||
aggregated_output: Some(String::new()),
|
||||
duration_ms: Some(0),
|
||||
exit_code: Some(0),
|
||||
status: ItemStatus::Failed,
|
||||
};
|
||||
assert_eq!(
|
||||
summary(&message(/*tag*/ 7, &[COMMAND, b"\x58\x02"].concat())),
|
||||
Some(ItemSummary::CommandExecution(CommandSummary {
|
||||
source: CommandExecutionSource::UserShell,
|
||||
..command.clone()
|
||||
}))
|
||||
);
|
||||
let change = b"\x0a\x01d\x12\x02\x0a\x00\x1a\x01p";
|
||||
for (wire_status, status) in [
|
||||
(1, ItemStatus::InProgress),
|
||||
(2, ItemStatus::Completed),
|
||||
(3, ItemStatus::Failed),
|
||||
(4, ItemStatus::Declined),
|
||||
(99, ItemStatus::Unknown(99)),
|
||||
] {
|
||||
assert_eq!(
|
||||
summary(&message(
|
||||
/*tag*/ 7,
|
||||
&[COMMAND, &[96, wire_status]].concat()
|
||||
)),
|
||||
Some(ItemSummary::CommandExecution(CommandSummary {
|
||||
status,
|
||||
..command.clone()
|
||||
}))
|
||||
);
|
||||
let patch = [
|
||||
message(/*tag*/ 1, &message(/*tag*/ 1, change)),
|
||||
b"\x12\x01i".to_vec(),
|
||||
vec![24, wire_status],
|
||||
]
|
||||
.concat();
|
||||
assert_eq!(
|
||||
summary(&message(/*tag*/ 8, &patch)),
|
||||
Some(ItemSummary::FileChange {
|
||||
id: "i".into(),
|
||||
changes: vec![FileUpdate {
|
||||
path: "p".into(),
|
||||
diff: "d".into(),
|
||||
kind: PatchKind::Add
|
||||
}],
|
||||
status,
|
||||
})
|
||||
);
|
||||
}
|
||||
// Lifecycle completion must not rewrite a failed command's outcome.
|
||||
for tag in [27, 31] {
|
||||
let event = TranscriptEvent::decode("t", &lifecycle(tag, &message(/*tag*/ 7, COMMAND)))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let (TranscriptUpdate::ItemStarted(item) | TranscriptUpdate::ItemCompleted(item)) =
|
||||
event.update
|
||||
else {
|
||||
panic!("expected item lifecycle");
|
||||
};
|
||||
assert_eq!(item.item, ItemSummary::CommandExecution(command.clone()));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reasoning_preserves_summary_and_content() {
|
||||
let reasoning = [
|
||||
message(/*tag*/ 1, &message(/*tag*/ 1, b"detail")),
|
||||
b"\x12\x01i".to_vec(),
|
||||
message(/*tag*/ 3, &message(/*tag*/ 1, b"summary")),
|
||||
]
|
||||
.concat();
|
||||
assert_eq!(
|
||||
summary(&message(/*tag*/ 6, &reasoning)),
|
||||
Some(ItemSummary::Reasoning {
|
||||
id: "i".into(),
|
||||
summary: vec!["summary".into()],
|
||||
content: vec!["detail".into()],
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_messages_keep_questions_with_empty_text() {
|
||||
let choices = message(/*tag*/ 1, b"Continue");
|
||||
let questions = [
|
||||
message(/*tag*/ 1, &message(/*tag*/ 2, b"Freeform")),
|
||||
message(/*tag*/ 1, b"\x0a\x00\x12\x05Empty"),
|
||||
message(
|
||||
/*tag*/ 1,
|
||||
&[
|
||||
message(/*tag*/ 1, &choices),
|
||||
message(/*tag*/ 2, b"Proceed?"),
|
||||
]
|
||||
.concat(),
|
||||
),
|
||||
]
|
||||
.concat();
|
||||
let expected = vec![
|
||||
AgentQuestion {
|
||||
title: "Freeform".into(),
|
||||
options: None,
|
||||
},
|
||||
AgentQuestion {
|
||||
title: "Empty".into(),
|
||||
options: Some(vec![]),
|
||||
},
|
||||
AgentQuestion {
|
||||
title: "Proceed?".into(),
|
||||
options: Some(vec!["Continue".into()]),
|
||||
},
|
||||
];
|
||||
for (questions, expected) in [(questions.as_slice(), expected), (&[], vec![])] {
|
||||
let item = message(
|
||||
/*tag*/ 3,
|
||||
&[b"\x12\x01i\x32\x00".to_vec(), message(/*tag*/ 5, questions)].concat(),
|
||||
);
|
||||
assert_eq!(
|
||||
summary(&item),
|
||||
Some(ItemSummary::AgentMessage {
|
||||
id: "i".into(),
|
||||
text: String::new(),
|
||||
questions: Some(expected),
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unsupported_and_incomplete_items_stay_opaque() {
|
||||
let absent_time = [message(/*tag*/ 1, AGENT), b"\x1a\x01t\x22\x01u".to_vec()].concat();
|
||||
// A later audio variant replaces text in the same native UserInput oneof.
|
||||
let text_then_audio = b"\x0a\x03\x0a\x01x\x22\x03\x0a\x01u";
|
||||
let user_content = message(/*tag*/ 2, &message(/*tag*/ 1, text_then_audio));
|
||||
let opaque = [
|
||||
message(
|
||||
/*tag*/ 1,
|
||||
&[user_content, message(/*tag*/ 3, b"i")].concat(),
|
||||
),
|
||||
message(/*tag*/ 7, &[COMMAND, b"\x58\x63"].concat()), // Future command source (99).
|
||||
message(/*tag*/ 9, b""),
|
||||
message(/*tag*/ 20, b""),
|
||||
[AGENT.to_vec(), message(/*tag*/ 19, b"")].concat(),
|
||||
message(/*tag*/ 3, b"\x12\x01i"),
|
||||
message(/*tag*/ 7, b"\x12\x01c\x22\x01w\x3a\x01i"),
|
||||
message(/*tag*/ 8, b"\x12\x01i\x18\x01"),
|
||||
];
|
||||
for item in opaque {
|
||||
assert!(
|
||||
TranscriptEvent::decode("t", &lifecycle(/*tag*/ 27, &item))
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
TranscriptEvent::decode("t", ¬ification(/*tag*/ 27, &absent_time))
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn item_decoder_rejects_malformed_items_and_foreign_owners() {
|
||||
assert!(TranscriptEvent::decode("t", &lifecycle(/*tag*/ 27, &[0xff])).is_err());
|
||||
assert!(TranscriptEvent::decode("other", &lifecycle(/*tag*/ 31, AGENT)).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn item_summaries_redact_sensitive_fields() {
|
||||
let item = ItemSummary::AgentMessage {
|
||||
questions: None,
|
||||
id: "secret".into(),
|
||||
text: "secret".into(),
|
||||
};
|
||||
let text = format!("{item:?}");
|
||||
assert_eq!(text, "ItemSummary([redacted])");
|
||||
assert!(!text.contains("secret"));
|
||||
let action = CommandAction::Unknown {
|
||||
command: "secret".into(),
|
||||
};
|
||||
assert_eq!(format!("{action:?}"), "CommandAction([redacted])");
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
//! Native item lifecycle envelopes and the item fields used by typed events.
|
||||
use prost::Message;
|
||||
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Started {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
pub item: Option<Item>,
|
||||
#[prost(int64, optional, tag = "2")]
|
||||
pub at_ms: Option<i64>,
|
||||
#[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 Completed {
|
||||
#[prost(int64, optional, tag = "1")]
|
||||
pub at_ms: Option<i64>,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
pub item: Option<Item>,
|
||||
#[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 Item {
|
||||
#[prost(
|
||||
oneof = "Value",
|
||||
tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19"
|
||||
)]
|
||||
pub value: Option<Value>,
|
||||
}
|
||||
#[derive(Clone, prost::Oneof)]
|
||||
pub(super) enum Value {
|
||||
#[prost(message, tag = "1")]
|
||||
UserMessage(UserMessage),
|
||||
#[prost(bytes, tag = "2")]
|
||||
HookPrompt(Vec<u8>),
|
||||
#[prost(message, tag = "3")]
|
||||
AgentMessage(AgentMessage),
|
||||
#[prost(bytes, tag = "4")]
|
||||
FunctionCallOutput(Vec<u8>),
|
||||
#[prost(bytes, tag = "5")]
|
||||
Plan(Vec<u8>),
|
||||
#[prost(message, tag = "6")]
|
||||
Reasoning(Reasoning),
|
||||
#[prost(message, tag = "7")]
|
||||
Command(Command),
|
||||
#[prost(message, tag = "8")]
|
||||
FileChange(FileChange),
|
||||
#[prost(bytes, tag = "9")]
|
||||
Mcp(Vec<u8>),
|
||||
#[prost(bytes, tag = "10")]
|
||||
DynamicTool(Vec<u8>),
|
||||
#[prost(bytes, tag = "11")]
|
||||
CollabAgent(Vec<u8>),
|
||||
#[prost(bytes, tag = "12")]
|
||||
SubAgent(Vec<u8>),
|
||||
#[prost(bytes, tag = "13")]
|
||||
WebSearch(Vec<u8>),
|
||||
#[prost(bytes, tag = "14")]
|
||||
ImageView(Vec<u8>),
|
||||
#[prost(bytes, tag = "15")]
|
||||
Sleep(Vec<u8>),
|
||||
#[prost(bytes, tag = "16")]
|
||||
ImageGeneration(Vec<u8>),
|
||||
#[prost(bytes, tag = "17")]
|
||||
EnterReview(Vec<u8>),
|
||||
#[prost(bytes, tag = "18")]
|
||||
ExitReview(Vec<u8>),
|
||||
#[prost(bytes, tag = "19")]
|
||||
Compaction(Vec<u8>),
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct AgentMessage {
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub id: Option<String>,
|
||||
#[prost(message, optional, tag = "5")]
|
||||
pub questions: Option<AgentQuestionList>,
|
||||
#[prost(string, optional, tag = "6")]
|
||||
pub text: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct AgentQuestionList {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
pub values: Vec<AgentQuestion>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct AgentQuestion {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
pub options: Option<StringList>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub title: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Command {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub output: Option<String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub command: Option<String>,
|
||||
#[prost(message, optional, tag = "3")]
|
||||
pub actions: Option<CommandActionList>,
|
||||
#[prost(string, optional, tag = "4")]
|
||||
pub cwd: Option<String>,
|
||||
#[prost(int64, optional, tag = "5")]
|
||||
pub duration_ms: Option<i64>,
|
||||
#[prost(int64, optional, tag = "6")]
|
||||
pub exit_code: Option<i64>,
|
||||
#[prost(string, optional, tag = "7")]
|
||||
pub id: Option<String>,
|
||||
#[prost(int32, optional, tag = "11")]
|
||||
pub source: Option<i32>,
|
||||
#[prost(int32, optional, tag = "12")]
|
||||
pub status: Option<i32>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Reasoning {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
pub content: Option<StringList>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub id: Option<String>,
|
||||
#[prost(message, optional, tag = "3")]
|
||||
pub summary: Option<StringList>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct StringList {
|
||||
#[prost(string, repeated, tag = "1")]
|
||||
pub values: Vec<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct CommandActionList {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
pub values: Vec<CommandAction>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct CommandAction {
|
||||
#[prost(oneof = "command_action::Value", tags = "1, 2, 3, 4")]
|
||||
pub value: Option<command_action::Value>,
|
||||
}
|
||||
pub(super) mod command_action {
|
||||
#[derive(Clone, prost::Oneof)]
|
||||
pub enum Value {
|
||||
#[prost(message, tag = "1")]
|
||||
Read(super::ReadAction),
|
||||
#[prost(message, tag = "2")]
|
||||
ListFiles(super::ListFilesAction),
|
||||
#[prost(message, tag = "3")]
|
||||
Search(super::SearchAction),
|
||||
#[prost(message, tag = "4")]
|
||||
Unknown(super::UnknownAction),
|
||||
}
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct ReadAction {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub command: Option<String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub name: Option<String>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub path: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct ListFilesAction {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub command: Option<String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub path: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct SearchAction {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub command: Option<String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub path: Option<String>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub query: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct UnknownAction {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub command: Option<String>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct FileChange {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
pub changes: Option<crate::transcript_wire::Changes>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub id: Option<String>,
|
||||
#[prost(int32, optional, tag = "3")]
|
||||
pub status: Option<i32>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct UserMessage {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub client_id: Option<String>,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
pub content: Option<crate::turns_wire::Input>,
|
||||
#[prost(string, optional, tag = "3")]
|
||||
pub id: Option<String>,
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
//! Selected item summaries for live start/completion events, without interpreting executor paths.
|
||||
use crate::FileUpdate;
|
||||
use crate::TranscriptUpdate;
|
||||
use crate::item_wire as wire;
|
||||
use crate::transcript::DecodeFailure;
|
||||
use crate::transcript::ITEM_STARTED;
|
||||
use crate::transcript::required;
|
||||
use crate::transcript_wire::Envelope;
|
||||
use crate::transcript_wire::SelectedMessages;
|
||||
use crate::transcript_wire::merge_messages;
|
||||
use std::fmt;
|
||||
|
||||
/// Lifecycle identity and timestamp for a typed item event.
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub struct ItemLifecycle {
|
||||
pub turn_id: String,
|
||||
/// Start or completion time in Unix milliseconds, according to the event variant.
|
||||
pub at_ms: i64,
|
||||
pub item: ItemSummary,
|
||||
}
|
||||
/// Selected fields only; this is not a complete display model.
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub enum ItemSummary {
|
||||
UserMessage {
|
||||
id: String,
|
||||
client_id: Option<String>,
|
||||
text: Vec<String>,
|
||||
},
|
||||
AgentMessage {
|
||||
id: String,
|
||||
text: String,
|
||||
questions: Option<Vec<AgentQuestion>>,
|
||||
},
|
||||
Reasoning {
|
||||
id: String,
|
||||
summary: Vec<String>,
|
||||
content: Vec<String>,
|
||||
},
|
||||
CommandExecution(CommandSummary),
|
||||
FileChange {
|
||||
id: String,
|
||||
changes: Vec<FileUpdate>,
|
||||
status: ItemStatus,
|
||||
},
|
||||
}
|
||||
/// A structured question attached to an agent message.
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub struct AgentQuestion {
|
||||
pub title: String,
|
||||
pub options: Option<Vec<String>>,
|
||||
}
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub struct CommandSummary {
|
||||
pub id: String,
|
||||
pub command: String,
|
||||
/// Executor-native path. Do not interpret it as a path on the client machine.
|
||||
pub cwd: String,
|
||||
pub actions: Vec<CommandAction>,
|
||||
pub source: CommandExecutionSource,
|
||||
pub aggregated_output: Option<String>,
|
||||
pub duration_ms: Option<i64>,
|
||||
pub exit_code: Option<i64>,
|
||||
pub status: ItemStatus,
|
||||
}
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub enum CommandAction {
|
||||
Read {
|
||||
command: String,
|
||||
name: String,
|
||||
path: String,
|
||||
},
|
||||
ListFiles {
|
||||
command: String,
|
||||
path: Option<String>,
|
||||
},
|
||||
Search {
|
||||
command: String,
|
||||
query: Option<String>,
|
||||
path: Option<String>,
|
||||
},
|
||||
Unknown {
|
||||
command: String,
|
||||
},
|
||||
}
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum CommandExecutionSource {
|
||||
Agent,
|
||||
UserShell,
|
||||
UnifiedExecStartup,
|
||||
UnifiedExecInteraction,
|
||||
}
|
||||
/// CommandExecutionStatus and PatchApplyStatus share these wire values.
|
||||
/// Completion of the item lifecycle does not imply the command or patch succeeded.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum ItemStatus {
|
||||
InProgress,
|
||||
Completed,
|
||||
Failed,
|
||||
Declined,
|
||||
Unknown(i32),
|
||||
}
|
||||
impl From<i32> for ItemStatus {
|
||||
fn from(value: i32) -> Self {
|
||||
match value {
|
||||
1 => Self::InProgress,
|
||||
2 => Self::Completed,
|
||||
3 => Self::Failed,
|
||||
4 => Self::Declined,
|
||||
value => Self::Unknown(value),
|
||||
}
|
||||
}
|
||||
}
|
||||
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!(
|
||||
ItemLifecycle,
|
||||
ItemSummary,
|
||||
AgentQuestion,
|
||||
CommandSummary,
|
||||
CommandAction
|
||||
);
|
||||
|
||||
pub(crate) fn decode(
|
||||
tag: u32,
|
||||
values: SelectedMessages<'_>,
|
||||
) -> Result<Option<(String, Option<i64>, TranscriptUpdate)>, DecodeFailure> {
|
||||
let (thread_id, turn_id, at_ms, item, emitted_at_ms) = if tag == ITEM_STARTED {
|
||||
let envelope = merge_messages::<Envelope<wire::Started>>(values)?;
|
||||
let params = required(envelope.params)?;
|
||||
(
|
||||
params.thread_id,
|
||||
params.turn_id,
|
||||
params.at_ms,
|
||||
params.item,
|
||||
envelope.emitted_at_ms,
|
||||
)
|
||||
} else {
|
||||
let envelope = merge_messages::<Envelope<wire::Completed>>(values)?;
|
||||
let params = required(envelope.params)?;
|
||||
(
|
||||
params.thread_id,
|
||||
params.turn_id,
|
||||
params.at_ms,
|
||||
params.item,
|
||||
envelope.emitted_at_ms,
|
||||
)
|
||||
};
|
||||
let Some(item) = summary(required(item)?)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let lifecycle = ItemLifecycle {
|
||||
turn_id: required(turn_id)?,
|
||||
at_ms: required(at_ms)?,
|
||||
item,
|
||||
};
|
||||
let update = if tag == ITEM_STARTED {
|
||||
TranscriptUpdate::ItemStarted(lifecycle)
|
||||
} else {
|
||||
TranscriptUpdate::ItemCompleted(lifecycle)
|
||||
};
|
||||
Ok(Some((required(thread_id)?, emitted_at_ms, update)))
|
||||
}
|
||||
|
||||
pub(crate) fn summary(item: wire::Item) -> Result<Option<ItemSummary>, DecodeFailure> {
|
||||
let summary = match item.value {
|
||||
Some(wire::Value::UserMessage(value)) => {
|
||||
let mut text = Vec::new();
|
||||
for input in required(value.content)?.values {
|
||||
let Some(crate::turns_wire::Value::Text(input)) = input.value else {
|
||||
return Ok(None);
|
||||
};
|
||||
text.push(required(input.text)?);
|
||||
}
|
||||
ItemSummary::UserMessage {
|
||||
id: required(value.id)?,
|
||||
client_id: value.client_id,
|
||||
text,
|
||||
}
|
||||
}
|
||||
Some(wire::Value::AgentMessage(value)) => ItemSummary::AgentMessage {
|
||||
id: required(value.id)?,
|
||||
text: required(value.text)?,
|
||||
questions: value
|
||||
.questions
|
||||
.map(|questions| {
|
||||
questions
|
||||
.values
|
||||
.into_iter()
|
||||
.map(|question| {
|
||||
Ok(AgentQuestion {
|
||||
title: required(question.title)?,
|
||||
options: question.options.map(|options| options.values),
|
||||
})
|
||||
})
|
||||
.collect::<Result<_, DecodeFailure>>()
|
||||
})
|
||||
.transpose()?,
|
||||
},
|
||||
Some(wire::Value::Reasoning(value)) => ItemSummary::Reasoning {
|
||||
id: required(value.id)?,
|
||||
summary: value.summary.map_or_else(Vec::new, |value| value.values),
|
||||
content: value.content.map_or_else(Vec::new, |value| value.values),
|
||||
},
|
||||
Some(wire::Value::Command(value)) => ItemSummary::CommandExecution(CommandSummary {
|
||||
id: required(value.id)?,
|
||||
command: required(value.command)?,
|
||||
cwd: required(value.cwd)?,
|
||||
actions: required(value.actions)?
|
||||
.values
|
||||
.into_iter()
|
||||
.map(command_action)
|
||||
.collect::<Result<_, _>>()?,
|
||||
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),
|
||||
},
|
||||
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)?,
|
||||
changes: required(value.changes)?
|
||||
.values
|
||||
.into_iter()
|
||||
.map(FileUpdate::try_from)
|
||||
.collect::<Result<_, _>>()?,
|
||||
status: required(value.status)?.into(),
|
||||
},
|
||||
_ => return Ok(None),
|
||||
};
|
||||
Ok(Some(summary))
|
||||
}
|
||||
|
||||
fn command_action(action: wire::CommandAction) -> Result<CommandAction, DecodeFailure> {
|
||||
Ok(match required(action.value)? {
|
||||
wire::command_action::Value::Read(value) => CommandAction::Read {
|
||||
command: required(value.command)?,
|
||||
name: required(value.name)?,
|
||||
path: required(value.path)?,
|
||||
},
|
||||
wire::command_action::Value::ListFiles(value) => CommandAction::ListFiles {
|
||||
command: required(value.command)?,
|
||||
path: value.path,
|
||||
},
|
||||
wire::command_action::Value::Search(value) => CommandAction::Search {
|
||||
command: required(value.command)?,
|
||||
query: value.query,
|
||||
path: value.path,
|
||||
},
|
||||
wire::command_action::Value::Unknown(value) => CommandAction::Unknown {
|
||||
command: required(value.command)?,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -4,11 +4,21 @@
|
||||
//! 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 item_wire;
|
||||
mod items;
|
||||
mod transcript;
|
||||
mod transcript_wire;
|
||||
mod turns_wire;
|
||||
mod types;
|
||||
mod wire;
|
||||
|
||||
pub use items::AgentQuestion;
|
||||
pub use items::CommandAction;
|
||||
pub use items::CommandExecutionSource;
|
||||
pub use items::CommandSummary;
|
||||
pub use items::ItemLifecycle;
|
||||
pub use items::ItemStatus;
|
||||
pub use items::ItemSummary;
|
||||
pub use transcript::FileUpdate;
|
||||
pub use transcript::ItemDelta;
|
||||
pub use transcript::PatchKind;
|
||||
|
||||
@@ -14,6 +14,8 @@ pub struct TranscriptEvent {
|
||||
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
pub enum TranscriptUpdate {
|
||||
ItemStarted(crate::ItemLifecycle),
|
||||
ItemCompleted(crate::ItemLifecycle),
|
||||
AgentMessageDelta(ItemDelta),
|
||||
CommandOutputDelta(ItemDelta),
|
||||
FileChangeOutputDelta(ItemDelta),
|
||||
@@ -97,13 +99,16 @@ redacted_debug!(
|
||||
// Field numbers in the ASB Notification.value oneof.
|
||||
const TURN_STARTED: u32 = 21;
|
||||
const TURN_COMPLETED: u32 = 23;
|
||||
pub(super) const ITEM_STARTED: u32 = 27;
|
||||
const ITEM_COMPLETED: u32 = 31;
|
||||
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 {
|
||||
pub(super) enum DecodeFailure {
|
||||
MissingField,
|
||||
UnsupportedCommandSource,
|
||||
InvalidResponse,
|
||||
}
|
||||
|
||||
@@ -113,14 +118,14 @@ impl From<prost::DecodeError> for DecodeFailure {
|
||||
}
|
||||
}
|
||||
|
||||
fn required<T>(value: Option<T>) -> Result<T, DecodeFailure> {
|
||||
pub(super) 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::MissingField | DecodeFailure::UnsupportedCommandSource) => Ok(None),
|
||||
Err(DecodeFailure::InvalidResponse) => Err(Error::InvalidResponse),
|
||||
}
|
||||
}
|
||||
@@ -133,6 +138,10 @@ impl TranscriptEvent {
|
||||
return Ok(None);
|
||||
};
|
||||
let (owner, emitted_at_ms, update) = match tag {
|
||||
ITEM_STARTED | ITEM_COMPLETED => match crate::items::decode(tag, values)? {
|
||||
Some(decoded) => decoded,
|
||||
None => return Ok(None),
|
||||
},
|
||||
ITEM_AGENT_MESSAGE_DELTA
|
||||
| ITEM_COMMAND_EXECUTION_OUTPUT_DELTA
|
||||
| ITEM_FILE_CHANGE_OUTPUT_DELTA => {
|
||||
@@ -159,22 +168,9 @@ impl TranscriptEvent {
|
||||
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,
|
||||
})
|
||||
})
|
||||
.map(FileUpdate::try_from)
|
||||
.collect::<Result<_, DecodeFailure>>()?;
|
||||
|
||||
let update = PatchUpdate {
|
||||
item_id: required(params.item_id)?,
|
||||
turn_id: required(params.turn_id)?,
|
||||
@@ -239,3 +235,22 @@ impl TranscriptEvent {
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<wire::Change> for FileUpdate {
|
||||
type Error = DecodeFailure;
|
||||
fn try_from(change: wire::Change) -> Result<Self, DecodeFailure> {
|
||||
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(Self {
|
||||
path: required(change.path)?,
|
||||
diff: required(change.diff)?,
|
||||
kind,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -160,6 +160,10 @@ 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*/ 27,
|
||||
b"\x0a\x02\x4a\x00\x10\x03\x1a\x01t\x22\x01u",
|
||||
), // Unsupported MCP item.
|
||||
notification(/*tag*/ 21, foreign_turn),
|
||||
notification(/*tag*/ 23, foreign_turn),
|
||||
notification(/*tag*/ 32, &DELTA[..9]), // missing turn ID
|
||||
@@ -288,3 +292,6 @@ fn notification_selection_handles_many_fragments_and_obsolete_malformed_messages
|
||||
payload.extend(message(/*tag*/ 89, b"\xff"));
|
||||
assert!(TranscriptEvent::decode("t", &payload).unwrap().is_none());
|
||||
}
|
||||
|
||||
#[path = "item_tests.rs"]
|
||||
mod items;
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
//! Native user-input envelopes shared by item decoding and later turn submission.
|
||||
use prost::Message;
|
||||
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Input {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
pub values: Vec<UserInput>,
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct UserInput {
|
||||
#[prost(oneof = "Value", tags = "1, 2, 3, 4, 5, 6, 7")]
|
||||
pub value: Option<Value>,
|
||||
}
|
||||
#[derive(Clone, prost::Oneof)]
|
||||
pub(super) enum Value {
|
||||
#[prost(message, tag = "1")]
|
||||
Text(Text),
|
||||
#[prost(bytes, tag = "2")]
|
||||
Image(Vec<u8>),
|
||||
#[prost(bytes, tag = "3")]
|
||||
LocalImage(Vec<u8>),
|
||||
#[prost(bytes, tag = "4")]
|
||||
Audio(Vec<u8>),
|
||||
#[prost(bytes, tag = "5")]
|
||||
LocalAudio(Vec<u8>),
|
||||
#[prost(bytes, tag = "6")]
|
||||
Skill(Vec<u8>),
|
||||
#[prost(bytes, tag = "7")]
|
||||
Mention(Vec<u8>),
|
||||
}
|
||||
#[derive(Clone, Message)]
|
||||
pub(super) struct Text {
|
||||
#[prost(string, optional, tag = "1")]
|
||||
pub text: Option<String>,
|
||||
}
|
||||
@@ -103,4 +103,6 @@ pub enum Error {
|
||||
Rpc(RpcStatus),
|
||||
#[error("invalid gRPC event envelope")]
|
||||
InvalidResponse,
|
||||
#[error("cloud response contains an unsupported {0}")]
|
||||
UnsupportedResponse(&'static str),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user