1use schemars::JsonSchema;
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use ts_rs::TS;
5
6use crate::events::{ApprovalAction, RunEvent, RunEventPayload, ToolStatus};
7
8use super::approval::{ApprovalDecision, ApprovalRequestParams, ApprovalResolveParams};
9use super::ServerNotification;
10
11#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, TS)]
12#[serde(rename_all = "camelCase")]
13pub struct ItemStartedParams {
14 #[serde(flatten)]
15 pub item: AppItem,
16}
17
18#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, TS)]
19#[serde(rename_all = "camelCase")]
20pub struct AgentMessageDeltaParams {
21 #[serde(flatten)]
22 pub item: AppItem,
23 pub delta: String,
24}
25
26#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, TS)]
27#[serde(rename_all = "camelCase")]
28pub struct ToolCallDeltaParams {
29 #[serde(flatten)]
30 pub item: AppItem,
31 pub delta: Value,
32}
33
34#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, TS)]
35#[serde(rename_all = "camelCase")]
36pub struct ItemCompletedParams {
37 #[serde(flatten)]
38 pub item: AppItem,
39}
40
41#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, TS)]
42#[serde(rename_all = "camelCase")]
43pub struct AppItem {
44 pub item_id: String,
45 pub thread_id: String,
46 pub turn_id: String,
47 #[serde(rename = "type")]
48 pub kind: AppItemKind,
49 pub status: AppItemStatus,
50 #[serde(default)]
51 pub payload: Value,
52 pub created_at: f64,
53 pub updated_at: f64,
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
57#[serde(rename_all = "camelCase")]
58pub enum AppItemKind {
59 UserMessage,
60 AgentMessage,
61 ToolCall,
62 Approval,
63 Error,
64 MemoryCompact,
65 SubRun,
66 Handoff,
67 RunStatus,
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
71#[serde(rename_all = "camelCase")]
72pub enum AppItemStatus {
73 Queued,
74 Started,
75 InProgress,
76 Completed,
77 Failed,
78 Interrupted,
79}
80
81pub fn map_run_event_to_notifications(
82 thread_id: &str,
83 turn_id: &str,
84 event: &RunEvent,
85) -> Vec<ServerNotification> {
86 match event.payload() {
87 RunEventPayload::RunStarted { input } => {
88 let item = item(
89 thread_id,
90 turn_id,
91 event,
92 AppItemKind::UserMessage,
93 AppItemStatus::Completed,
94 serde_json::json!({ "text": input }),
95 );
96 vec![ServerNotification::ItemCompleted(ItemCompletedParams {
97 item,
98 })]
99 }
100 RunEventPayload::AssistantDelta { delta, .. } => {
101 let item = item(
102 thread_id,
103 turn_id,
104 event,
105 AppItemKind::AgentMessage,
106 AppItemStatus::InProgress,
107 serde_json::json!({ "delta": delta }),
108 );
109 vec![ServerNotification::AgentMessageDelta(
110 AgentMessageDeltaParams {
111 item,
112 delta: delta.clone(),
113 },
114 )]
115 }
116 RunEventPayload::ModelToolCallProgress {
117 tool_call_id,
118 tool_call_index,
119 tool_name,
120 arguments_chars,
121 ..
122 } => {
123 let item = item(
124 thread_id,
125 turn_id,
126 event,
127 AppItemKind::ToolCall,
128 AppItemStatus::InProgress,
129 serde_json::json!({
130 "toolCallId": tool_call_id,
131 "toolName": tool_name,
132 }),
133 );
134 vec![ServerNotification::ToolCallDelta(ToolCallDeltaParams {
135 item,
136 delta: serde_json::json!({
137 "toolCallId": tool_call_id,
138 "toolCallIndex": tool_call_index,
139 "toolName": tool_name,
140 "argumentsChars": arguments_chars,
141 }),
142 })]
143 }
144 RunEventPayload::ToolCallPlanned { .. } => Vec::new(),
145 RunEventPayload::ToolCallStarted {
146 tool_call_id,
147 tool_name,
148 arguments,
149 } => {
150 let mut payload = serde_json::json!({
151 "toolCallId": tool_call_id,
152 "toolName": tool_name,
153 });
154 if let Some(tool_metadata) = event.tool_metadata() {
155 payload["toolMetadata"] = tool_metadata_payload(&tool_metadata);
156 }
157 let item = item(
158 thread_id,
159 turn_id,
160 event,
161 AppItemKind::ToolCall,
162 AppItemStatus::Started,
163 payload,
164 );
165 vec![
166 ServerNotification::ItemStarted(ItemStartedParams { item: item.clone() }),
167 ServerNotification::ToolCallDelta(ToolCallDeltaParams {
168 item,
169 delta: arguments.clone(),
170 }),
171 ]
172 }
173 RunEventPayload::ToolCallCompleted {
174 tool_call_id,
175 tool_name,
176 status,
177 ..
178 } => {
179 let mut payload = serde_json::json!({
180 "toolCallId": tool_call_id,
181 "toolName": tool_name,
182 "status": status,
183 });
184 payload["directive"] = event
185 .tool_directive()
186 .map(|value| serde_json::to_value(value).expect("directive serializes"))
187 .expect("tool completion directive is present");
188 payload["errorCode"] = event
189 .tool_error_code()
190 .map_or(Value::Null, |value| Value::String(value.to_string()));
191 payload["executionStarted"] = event
192 .tool_execution_started()
193 .map(Value::Bool)
194 .expect("tool completion execution_started is present");
195 payload["durationMs"] = event.tool_duration_ms().map_or(Value::Null, Value::from);
196 if let Some(tool_metadata) = event.tool_metadata() {
197 payload["toolMetadata"] = tool_metadata_payload(&tool_metadata);
198 }
199 let item = item(
200 thread_id,
201 turn_id,
202 event,
203 AppItemKind::ToolCall,
204 tool_status_to_item_status(*status),
205 payload,
206 );
207 vec![ServerNotification::ItemCompleted(ItemCompletedParams {
208 item,
209 })]
210 }
211 RunEventPayload::ApprovalRequested {
212 request_id,
213 tool_call_id,
214 tool_name,
215 message,
216 } => {
217 let item = item(
218 thread_id,
219 turn_id,
220 event,
221 AppItemKind::Approval,
222 AppItemStatus::Started,
223 serde_json::json!({
224 "requestId": request_id,
225 "toolCallId": tool_call_id,
226 "toolName": tool_name,
227 "message": message,
228 "arguments": event
229 .metadata()
230 .get("arguments")
231 .cloned()
232 .unwrap_or_else(|| serde_json::json!({})),
233 }),
234 );
235 vec![
236 ServerNotification::ItemStarted(ItemStartedParams { item }),
237 ServerNotification::ApprovalRequested(ApprovalRequestParams {
238 request_id: request_id.clone(),
239 thread_id: thread_id.to_string(),
240 turn_id: turn_id.to_string(),
241 tool_call_id: tool_call_id.clone(),
242 tool_name: tool_name.clone(),
243 preview: message.clone(),
244 arguments: event
245 .metadata()
246 .get("arguments")
247 .cloned()
248 .unwrap_or_else(|| serde_json::json!({})),
249 }),
250 ]
251 }
252 RunEventPayload::ApprovalResolved {
253 request_id,
254 tool_call_id,
255 tool_name,
256 action,
257 } => {
258 let decision = match action {
259 ApprovalAction::Allow => ApprovalDecision::Allow,
260 ApprovalAction::AllowSession => ApprovalDecision::AllowSession,
261 ApprovalAction::Deny => ApprovalDecision::Deny,
262 ApprovalAction::Timeout => ApprovalDecision::Timeout,
263 };
264 let reason = event
265 .metadata()
266 .get("reason")
267 .and_then(Value::as_str)
268 .unwrap_or_default()
269 .to_string();
270 let decision_metadata = event
271 .metadata()
272 .get("decision_metadata")
273 .and_then(Value::as_object)
274 .map(|metadata| {
275 metadata
276 .iter()
277 .map(|(key, value)| (key.clone(), value.clone()))
278 .collect()
279 })
280 .unwrap_or_default();
281 let item = item(
282 thread_id,
283 turn_id,
284 event,
285 AppItemKind::Approval,
286 AppItemStatus::Completed,
287 serde_json::json!({
288 "requestId": request_id,
289 "toolCallId": tool_call_id,
290 "toolName": tool_name,
291 "action": action.as_str(),
292 "approved": action.is_approved(),
293 "reason": reason,
294 "decisionMetadata": decision_metadata,
295 }),
296 );
297 vec![
298 ServerNotification::ItemCompleted(ItemCompletedParams { item }),
299 ServerNotification::ApprovalResolved(ApprovalResolveParams {
300 thread_id: thread_id.to_string(),
301 turn_id: turn_id.to_string(),
302 request_id: request_id.clone(),
303 decision,
304 reason,
305 metadata: decision_metadata,
306 }),
307 ]
308 }
309 RunEventPayload::RunCompleted { .. } => {
310 event.final_output().map_or_else(Vec::new, |output| {
311 let item = item(
312 thread_id,
313 turn_id,
314 event,
315 AppItemKind::AgentMessage,
316 AppItemStatus::Completed,
317 serde_json::json!({ "text": output }),
318 );
319 vec![ServerNotification::ItemCompleted(ItemCompletedParams {
320 item,
321 })]
322 })
323 }
324 RunEventPayload::RunFailed { error } => {
325 let item = item(
326 thread_id,
327 turn_id,
328 event,
329 AppItemKind::Error,
330 AppItemStatus::Completed,
331 serde_json::json!({ "message": error }),
332 );
333 vec![
334 ServerNotification::ItemCompleted(ItemCompletedParams { item }),
335 ServerNotification::ErrorWarning(super::WarningParams {
336 message: error.clone(),
337 code: Some("run_failed".to_string()),
338 }),
339 ]
340 }
341 _ => Vec::new(),
342 }
343}
344
345fn item(
346 thread_id: &str,
347 turn_id: &str,
348 event: &RunEvent,
349 kind: AppItemKind,
350 status: AppItemStatus,
351 payload: Value,
352) -> AppItem {
353 AppItem {
354 item_id: format!("item_{}", event.event_id().as_str()),
355 thread_id: thread_id.to_string(),
356 turn_id: turn_id.to_string(),
357 kind,
358 status,
359 payload,
360 created_at: event_timestamp(event),
361 updated_at: event_timestamp(event),
362 }
363}
364
365fn event_timestamp(event: &RunEvent) -> f64 {
366 event.created_at()
367}
368
369fn tool_status_to_item_status(status: ToolStatus) -> AppItemStatus {
370 match status {
371 ToolStatus::Started => AppItemStatus::Started,
372 ToolStatus::Success => AppItemStatus::Completed,
373 ToolStatus::Error => AppItemStatus::Failed,
374 ToolStatus::WaitResponse | ToolStatus::Running | ToolStatus::PendingCompress => {
375 AppItemStatus::InProgress
376 }
377 }
378}
379
380fn tool_metadata_payload(metadata: &crate::tools::ToolMetadata) -> Value {
381 serde_json::json!({
382 "sideEffect": metadata.side_effect,
383 "idempotency": metadata.idempotency,
384 "terminal": metadata.terminal,
385 "capabilityTags": metadata.capability_tags,
386 "costDimensions": metadata.cost_dimensions,
387 })
388}