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::ToolCallStarted {
117 tool_call_id,
118 tool_name,
119 arguments,
120 } => {
121 let item = item(
122 thread_id,
123 turn_id,
124 event,
125 AppItemKind::ToolCall,
126 AppItemStatus::Started,
127 serde_json::json!({
128 "toolCallId": tool_call_id,
129 "toolName": tool_name,
130 }),
131 );
132 vec![
133 ServerNotification::ItemStarted(ItemStartedParams { item: item.clone() }),
134 ServerNotification::ToolCallDelta(ToolCallDeltaParams {
135 item,
136 delta: arguments.clone(),
137 }),
138 ]
139 }
140 RunEventPayload::ToolCallCompleted {
141 tool_call_id,
142 tool_name,
143 status,
144 } => {
145 let item = item(
146 thread_id,
147 turn_id,
148 event,
149 AppItemKind::ToolCall,
150 tool_status_to_item_status(*status),
151 serde_json::json!({
152 "toolCallId": tool_call_id,
153 "toolName": tool_name,
154 "status": status,
155 }),
156 );
157 vec![ServerNotification::ItemCompleted(ItemCompletedParams {
158 item,
159 })]
160 }
161 RunEventPayload::ApprovalRequested {
162 request_id,
163 tool_call_id,
164 tool_name,
165 message,
166 } => {
167 let item = item(
168 thread_id,
169 turn_id,
170 event,
171 AppItemKind::Approval,
172 AppItemStatus::Started,
173 serde_json::json!({
174 "requestId": request_id,
175 "toolCallId": tool_call_id,
176 "toolName": tool_name,
177 "message": message,
178 "arguments": event
179 .metadata()
180 .get("arguments")
181 .cloned()
182 .unwrap_or_else(|| serde_json::json!({})),
183 }),
184 );
185 vec![
186 ServerNotification::ItemStarted(ItemStartedParams { item }),
187 ServerNotification::ApprovalRequested(ApprovalRequestParams {
188 request_id: request_id.clone(),
189 thread_id: thread_id.to_string(),
190 turn_id: turn_id.to_string(),
191 tool_call_id: tool_call_id.clone(),
192 tool_name: tool_name.clone(),
193 preview: message.clone(),
194 arguments: event
195 .metadata()
196 .get("arguments")
197 .cloned()
198 .unwrap_or_else(|| serde_json::json!({})),
199 }),
200 ]
201 }
202 RunEventPayload::ApprovalResolved {
203 request_id,
204 tool_call_id,
205 tool_name,
206 approved,
207 } => {
208 let action = event
209 .approval_action()
210 .unwrap_or_else(|| ApprovalAction::from_approved(*approved));
211 let decision = match action {
212 ApprovalAction::Allow => ApprovalDecision::Allow,
213 ApprovalAction::AllowSession => ApprovalDecision::AllowSession,
214 ApprovalAction::Deny => ApprovalDecision::Deny,
215 ApprovalAction::Timeout => ApprovalDecision::Timeout,
216 };
217 let reason = event
218 .metadata()
219 .get("reason")
220 .and_then(Value::as_str)
221 .unwrap_or_default()
222 .to_string();
223 let decision_metadata = event
224 .metadata()
225 .get("decision_metadata")
226 .and_then(Value::as_object)
227 .map(|metadata| {
228 metadata
229 .iter()
230 .map(|(key, value)| (key.clone(), value.clone()))
231 .collect()
232 })
233 .unwrap_or_default();
234 let item = item(
235 thread_id,
236 turn_id,
237 event,
238 AppItemKind::Approval,
239 AppItemStatus::Completed,
240 serde_json::json!({
241 "requestId": request_id,
242 "toolCallId": tool_call_id,
243 "toolName": tool_name,
244 "action": action.as_str(),
245 "approved": approved,
246 "reason": reason,
247 "decisionMetadata": decision_metadata,
248 }),
249 );
250 vec![
251 ServerNotification::ItemCompleted(ItemCompletedParams { item }),
252 ServerNotification::ApprovalResolved(ApprovalResolveParams {
253 thread_id: thread_id.to_string(),
254 turn_id: turn_id.to_string(),
255 request_id: request_id.clone(),
256 decision,
257 reason,
258 metadata: decision_metadata,
259 }),
260 ]
261 }
262 RunEventPayload::RunCompleted { .. } => {
263 event.final_output().map_or_else(Vec::new, |output| {
264 let item = item(
265 thread_id,
266 turn_id,
267 event,
268 AppItemKind::AgentMessage,
269 AppItemStatus::Completed,
270 serde_json::json!({ "text": output }),
271 );
272 vec![ServerNotification::ItemCompleted(ItemCompletedParams {
273 item,
274 })]
275 })
276 }
277 RunEventPayload::RunFailed { error } => {
278 let item = item(
279 thread_id,
280 turn_id,
281 event,
282 AppItemKind::Error,
283 AppItemStatus::Completed,
284 serde_json::json!({ "message": error }),
285 );
286 vec![
287 ServerNotification::ItemCompleted(ItemCompletedParams { item }),
288 ServerNotification::ErrorWarning(super::WarningParams {
289 message: error.clone(),
290 code: Some("run_failed".to_string()),
291 }),
292 ]
293 }
294 _ => Vec::new(),
295 }
296}
297
298fn item(
299 thread_id: &str,
300 turn_id: &str,
301 event: &RunEvent,
302 kind: AppItemKind,
303 status: AppItemStatus,
304 payload: Value,
305) -> AppItem {
306 AppItem {
307 item_id: format!("item_{}", event.event_id().as_str()),
308 thread_id: thread_id.to_string(),
309 turn_id: turn_id.to_string(),
310 kind,
311 status,
312 payload,
313 created_at: event_timestamp(event),
314 updated_at: event_timestamp(event),
315 }
316}
317
318fn event_timestamp(event: &RunEvent) -> f64 {
319 event.created_at_ms() as f64 / 1000.0
320}
321
322fn tool_status_to_item_status(status: ToolStatus) -> AppItemStatus {
323 match status {
324 ToolStatus::Started => AppItemStatus::Started,
325 ToolStatus::Success => AppItemStatus::Completed,
326 ToolStatus::Error => AppItemStatus::Failed,
327 ToolStatus::WaitResponse => AppItemStatus::InProgress,
328 }
329}