vtcode_llm/open_responses/bridge/
items.rs

1//! ThreadEvent item handling: started/updated/completed item conversion.
2
3use super::*;
4
5fn tool_output_text(output: &ToolOutputItem) -> String {
6    if !output.output.is_empty() {
7        return output.output.clone();
8    }
9
10    output
11        .spool_path
12        .as_deref()
13        .map(|path| format!("Output saved to {path}"))
14        .unwrap_or_default()
15}
16
17impl ResponseBuilder {
18    pub fn process_event<E: StreamEventEmitter>(&mut self, event: &ThreadEvent, emitter: &mut E) {
19        match event {
20            ThreadEvent::ThreadStarted(_) => {
21                emitter.response_created(self.response.clone());
22                self.response.status = ResponseStatus::InProgress;
23                emitter.response_in_progress(self.response.clone());
24                self.normalized.response_started = true;
25            }
26
27            ThreadEvent::TurnStarted(_) => {
28                // Turn started is internal to VT Code; no direct Open Responses equivalent
29                // The response is already in progress from ThreadStarted
30            }
31
32            ThreadEvent::TurnCompleted(evt) => {
33                if self.response.status.is_terminal() {
34                    return;
35                }
36                self.response.usage = Some(OpenUsage::from_exec_usage(&evt.usage).into());
37                self.response.status = ResponseStatus::Completed;
38                self.response.complete();
39                emitter.response_completed(self.response.clone());
40            }
41
42            ThreadEvent::TurnFailed(evt) => {
43                if self.response.status.is_terminal() {
44                    return;
45                }
46                self.response.fail(OpenResponseError::model_error(&evt.message));
47                emitter.response_failed(self.response.clone());
48            }
49
50            ThreadEvent::TurnBlocked(evt) => {
51                self.emit_custom_event(
52                    emitter,
53                    "vtcode.turn_blocked",
54                    json!({
55                        "completed_at": evt.completed_at,
56                        "message": evt.message,
57                        "last_tool": evt.last_tool,
58                        "blocked_streak": evt.blocked_streak,
59                        "blocked_total": evt.blocked_total,
60                        "consecutive_cap": evt.consecutive_cap,
61                        "total_cap": evt.total_cap,
62                        "recovery_active": evt.recovery_active,
63                    }),
64                );
65            }
66
67            ThreadEvent::AppSandboxDenied(evt) => {
68                self.emit_custom_event(
69                    emitter,
70                    "vtcode.app_sandbox_denied",
71                    json!({
72                        "tool": evt.tool,
73                        "tier": evt.tier,
74                        "uris": evt.uris,
75                    }),
76                );
77            }
78
79            ThreadEvent::ThreadCompleted(evt) => {
80                self.emit_custom_event(
81                    emitter,
82                    "vtcode.thread_completed",
83                    json!({
84                        "completed_at": evt.completed_at,
85                        "thread_id": evt.thread_id,
86                        "session_id": evt.session_id,
87                        "subtype": evt.subtype.as_str(),
88                        "outcome_code": evt.outcome_code,
89                        "result": evt.result,
90                        "stop_reason": evt.stop_reason,
91                        "usage": evt.usage,
92                        "total_cost_usd": evt.total_cost_usd,
93                        "num_turns": evt.num_turns,
94                    }),
95                );
96            }
97
98            ThreadEvent::ThreadCompactBoundary(evt) => {
99                self.emit_custom_event(
100                    emitter,
101                    "vtcode.thread_compact_boundary",
102                    json!({
103                        "thread_id": evt.thread_id,
104                        "trigger": evt.trigger.as_str(),
105                        "mode": evt.mode.as_str(),
106                        "original_message_count": evt.original_message_count,
107                        "compacted_message_count": evt.compacted_message_count,
108                        "history_artifact_path": evt.history_artifact_path,
109                    }),
110                );
111            }
112
113            ThreadEvent::ContextReset(evt) => {
114                self.emit_custom_event(
115                    emitter,
116                    "vtcode.context_reset",
117                    json!({
118                        "thread_id": evt.thread_id,
119                        "turn_id": evt.turn_id,
120                        "trigger": evt.trigger,
121                        "plan_preserved": evt.plan_preserved,
122                        "previous_context_usage_percent": evt.previous_context_usage_percent,
123                        "tool_budget_reset": evt.tool_budget_reset,
124                    }),
125                );
126            }
127
128            ThreadEvent::ItemStarted(evt) => {
129                self.handle_item_started(&evt.item, emitter);
130            }
131
132            ThreadEvent::ItemUpdated(evt) => {
133                self.handle_item_updated(&evt.item, emitter);
134            }
135
136            ThreadEvent::ItemCompleted(evt) => {
137                self.handle_item_completed(&evt.item, emitter);
138            }
139            ThreadEvent::PlanDelta(_) => {
140                // Plan deltas are VT Code-specific extension events and are intentionally
141                // ignored by the Open Responses bridge. The completed Plan item carries
142                // the full final plan content.
143            }
144
145            ThreadEvent::PlanApprovalRequested(evt) => {
146                self.emit_custom_event(
147                    emitter,
148                    "vtcode.plan_approval_requested",
149                    json!({
150                        "thread_id": evt.thread_id,
151                        "turn_id": evt.turn_id,
152                        "plan_file": evt.plan_file,
153                    }),
154                );
155            }
156
157            ThreadEvent::PlanApprovalResolved(evt) => {
158                self.emit_custom_event(
159                    emitter,
160                    "vtcode.plan_approval_resolved",
161                    json!({
162                        "thread_id": evt.thread_id,
163                        "turn_id": evt.turn_id,
164                        "decision": evt.decision,
165                        "automatic": evt.automatic,
166                    }),
167                );
168            }
169
170            ThreadEvent::Error(evt) => {
171                if self.response.status.is_terminal() {
172                    return;
173                }
174                self.response.fail(OpenResponseError::server_error(&evt.message));
175                emitter.response_failed(self.response.clone());
176            }
177
178            // Unknown events from newer schema versions are silently skipped.
179            ThreadEvent::Unknown
180            | ThreadEvent::PermissionRequested(_)
181            | ThreadEvent::PermissionResolved(_)
182            | ThreadEvent::Interjected(_)
183            | ThreadEvent::MatrixUpdated(_) => {}
184        }
185    }
186
187    fn handle_item_started<E: StreamEventEmitter>(&mut self, item: &ThreadItem, emitter: &mut E) {
188        let output_index = self.next_output_index;
189        self.next_output_index += 1;
190        self.item_id_to_index.insert(item.id.clone(), output_index);
191
192        let output_item = self.convert_thread_item(item, ItemStatus::InProgress);
193
194        // Track active item state for streaming
195        // Initialize prev_text from initial content to prevent duplicate deltas
196        let initial_text = match &item.details {
197            ThreadItemDetails::AgentMessage(msg) => msg.text.clone(),
198            ThreadItemDetails::Plan(plan) => plan.text.clone(),
199            ThreadItemDetails::Reasoning(r) => r.text.clone(),
200            ThreadItemDetails::ToolOutput(output) => tool_output_text(output),
201            _ => String::new(),
202        };
203        let active_state = ActiveItemState {
204            output_index,
205            content_index: 0,
206            prev_text: initial_text,
207        };
208        self.active_items.insert(item.id.clone(), active_state);
209
210        self.response.add_output(output_item.clone());
211        emitter.output_item_added(&self.response.id, output_index, output_item.clone());
212
213        // Emit ContentPartAdded for items with text content
214        if let OutputItem::Message(ref msg) = output_item
215            && !msg.content.is_empty()
216        {
217            emitter.emit(ResponseStreamEvent::ContentPartAdded {
218                response_id: self.response.id.clone(),
219                item_id: item.id.clone(),
220                output_index,
221                content_index: 0,
222                part: msg.content[0].clone(),
223            });
224        }
225    }
226
227    fn emit_custom_event<E: StreamEventEmitter>(&self, emitter: &mut E, event_type: &str, data: serde_json::Value) {
228        emitter.emit(ResponseStreamEvent::CustomEvent {
229            response_id: self.response.id.clone(),
230            event_type: event_type.to_string(),
231            sequence_number: self.next_output_index as u64,
232            data,
233        });
234    }
235
236    fn handle_item_updated<E: StreamEventEmitter>(&mut self, item: &ThreadItem, emitter: &mut E) {
237        // Handle updates for items not yet started (implicit start)
238        let state = if let Some(state) = self.active_items.get_mut(&item.id) {
239            state
240        } else {
241            // Implicit start: create item and emit Added event
242            self.handle_item_started(item, emitter);
243            match self.active_items.get_mut(&item.id) {
244                Some(s) => s,
245                None => return,
246            }
247        };
248
249        match &item.details {
250            ThreadItemDetails::AgentMessage(msg) => {
251                // Use strip_prefix for safe UTF-8 delta computation
252                let delta = if let Some(suffix) = msg.text.strip_prefix(&state.prev_text) {
253                    suffix
254                } else {
255                    // Non-append update: emit full text as delta (fallback)
256                    &msg.text
257                };
258
259                if !delta.is_empty() {
260                    emitter.output_text_delta(
261                        &self.response.id,
262                        &item.id,
263                        state.output_index,
264                        state.content_index,
265                        delta,
266                    );
267                    state.prev_text = msg.text.clone();
268                }
269            }
270
271            ThreadItemDetails::Reasoning(r) => {
272                // Use strip_prefix for safe UTF-8 delta computation
273                let delta = if let Some(suffix) = r.text.strip_prefix(&state.prev_text) {
274                    suffix
275                } else {
276                    // Non-append update: emit full text as delta (fallback)
277                    &r.text
278                };
279
280                if !delta.is_empty() {
281                    emitter.reasoning_delta(&self.response.id, &item.id, state.output_index, delta);
282                    state.prev_text = r.text.clone();
283                }
284            }
285
286            ThreadItemDetails::ToolOutput(output) => {
287                let current_text = tool_output_text(output);
288                let delta = if let Some(suffix) = current_text.strip_prefix(&state.prev_text) {
289                    suffix
290                } else {
291                    current_text.as_str()
292                };
293
294                if !delta.is_empty() {
295                    emitter.output_text_delta(
296                        &self.response.id,
297                        &item.id,
298                        state.output_index,
299                        state.content_index,
300                        delta,
301                    );
302                    state.prev_text = current_text;
303                }
304            }
305
306            _ => {
307                // Other item types don't have incremental updates
308            }
309        }
310    }
311
312    fn handle_item_completed<E: StreamEventEmitter>(&mut self, item: &ThreadItem, emitter: &mut E) {
313        let (was_started, output_index) = match self.item_id_to_index.get(&item.id) {
314            Some(&idx) => (true, idx),
315            None => {
316                // Item was completed without being started (atomic item)
317                let idx = self.next_output_index;
318                self.next_output_index += 1;
319                self.item_id_to_index.insert(item.id.clone(), idx);
320                (false, idx)
321            }
322        };
323
324        // Determine final status
325        let status = self.determine_item_status(&item.details);
326        let output_item = self.convert_thread_item(item, status);
327
328        // For atomic completions (never started), emit Added first, then ContentPartAdded
329        if !was_started {
330            emitter.output_item_added(&self.response.id, output_index, output_item.clone());
331
332            // Emit ContentPartAdded for Message and Reasoning items
333            match &output_item {
334                OutputItem::Message(msg) => {
335                    if !msg.content.is_empty() {
336                        emitter.emit(ResponseStreamEvent::ContentPartAdded {
337                            response_id: self.response.id.clone(),
338                            item_id: item.id.clone(),
339                            output_index,
340                            content_index: 0,
341                            part: msg.content[0].clone(),
342                        });
343                    }
344                }
345                OutputItem::Reasoning(r) => {
346                    let text = r.content.clone().unwrap_or_default();
347                    emitter.emit(ResponseStreamEvent::ContentPartAdded {
348                        response_id: self.response.id.clone(),
349                        item_id: item.id.clone(),
350                        output_index,
351                        content_index: 0,
352                        part: ContentPart::output_text(text),
353                    });
354                }
355                _ => {}
356            }
357        }
358
359        // Update the response output
360        if output_index < self.response.output.len() {
361            self.response.output[output_index] = output_item.clone();
362        } else {
363            self.response.add_output(output_item.clone());
364        }
365
366        // Emit content-specific "done" events based on item type
367        match &output_item {
368            OutputItem::Message(msg) => {
369                // Emit OutputTextDone for text content
370                if let Some(ContentPart::OutputText(text_content)) = msg.content.first() {
371                    emitter.emit(ResponseStreamEvent::OutputTextDone {
372                        response_id: self.response.id.clone(),
373                        item_id: item.id.clone(),
374                        output_index,
375                        content_index: 0,
376                        text: text_content.text.clone(),
377                    });
378                    emitter.emit(ResponseStreamEvent::ContentPartDone {
379                        response_id: self.response.id.clone(),
380                        item_id: item.id.clone(),
381                        output_index,
382                        content_index: 0,
383                        part: msg.content[0].clone(),
384                    });
385                }
386            }
387            OutputItem::Reasoning(r) => {
388                // Emit ReasoningDone then ContentPartDone
389                emitter.emit(ResponseStreamEvent::ReasoningDone {
390                    response_id: self.response.id.clone(),
391                    item_id: item.id.clone(),
392                    output_index,
393                    item: output_item.clone(),
394                });
395                let text = r.content.clone().unwrap_or_default();
396                emitter.emit(ResponseStreamEvent::ContentPartDone {
397                    response_id: self.response.id.clone(),
398                    item_id: item.id.clone(),
399                    output_index,
400                    content_index: 0,
401                    part: ContentPart::output_text(text),
402                });
403            }
404            OutputItem::FunctionCall(fc) => {
405                // Emit FunctionCallArgumentsDone
406                if let Ok(args_str) = serde_json::to_string(&fc.arguments) {
407                    emitter.emit(ResponseStreamEvent::FunctionCallArgumentsDone {
408                        response_id: self.response.id.clone(),
409                        item_id: item.id.clone(),
410                        output_index,
411                        arguments: args_str,
412                    });
413                }
414            }
415            OutputItem::FunctionCallOutput(fco) if !fco.output.is_empty() => {
416                emitter.emit(ResponseStreamEvent::OutputTextDone {
417                    response_id: self.response.id.clone(),
418                    item_id: item.id.clone(),
419                    output_index,
420                    content_index: 0,
421                    text: fco.output.clone(),
422                });
423            }
424            _ => {}
425        }
426
427        // Clean up active state
428        self.active_items.remove(&item.id);
429
430        emitter.output_item_done(&self.response.id, output_index, output_item);
431    }
432
433    fn determine_item_status(&self, details: &ThreadItemDetails) -> ItemStatus {
434        match details {
435            ThreadItemDetails::CommandExecution(cmd) => match cmd.status {
436                CommandExecutionStatus::Completed => ItemStatus::Completed,
437                CommandExecutionStatus::Failed => ItemStatus::Failed,
438                CommandExecutionStatus::InProgress => ItemStatus::InProgress,
439            },
440            ThreadItemDetails::ToolInvocation(invocation) => match invocation.status {
441                vtcode_exec_events::ToolCallStatus::Completed => ItemStatus::Completed,
442                vtcode_exec_events::ToolCallStatus::Failed => ItemStatus::Failed,
443                vtcode_exec_events::ToolCallStatus::InProgress => ItemStatus::InProgress,
444            },
445            ThreadItemDetails::ToolOutput(output) => match output.status {
446                vtcode_exec_events::ToolCallStatus::Completed => ItemStatus::Completed,
447                vtcode_exec_events::ToolCallStatus::Failed => ItemStatus::Failed,
448                vtcode_exec_events::ToolCallStatus::InProgress => ItemStatus::InProgress,
449            },
450            ThreadItemDetails::FileChange(fc) => match fc.status {
451                PatchApplyStatus::Completed => ItemStatus::Completed,
452                PatchApplyStatus::Failed => ItemStatus::Failed,
453            },
454            ThreadItemDetails::McpToolCall(tc) => match tc.status {
455                Some(McpToolCallStatus::Completed) => ItemStatus::Completed,
456                Some(McpToolCallStatus::Failed) => ItemStatus::Failed,
457                Some(McpToolCallStatus::Started) | None => ItemStatus::InProgress,
458            },
459            ThreadItemDetails::Error(_) => ItemStatus::Failed,
460            _ => ItemStatus::Completed,
461        }
462    }
463
464    fn resolve_tool_call_correlation_id(&mut self, harness_call_id: &str, raw_tool_call_id: Option<&str>) -> String {
465        if let Some(existing) = self.tool_call_correlation_ids.get(harness_call_id) {
466            return existing.clone();
467        }
468
469        let correlation_id = match raw_tool_call_id {
470            Some(raw_id) if self.used_tool_call_ids.insert(raw_id.to_string()) => raw_id.to_string(),
471            _ => harness_call_id.to_string(),
472        };
473        self.tool_call_correlation_ids
474            .insert(harness_call_id.to_string(), correlation_id.clone());
475        correlation_id
476    }
477
478    fn convert_thread_item(&mut self, item: &ThreadItem, status: ItemStatus) -> OutputItem {
479        match &item.details {
480            ThreadItemDetails::Decision(decision) => OutputItem::Custom(CustomItem {
481                id: item.id.clone().into(),
482                status,
483                custom_type: "vtcode:decision".into(),
484                data: json!({"decision": decision, "context": item.context}),
485            }),
486            ThreadItemDetails::AgentMessage(msg) => OutputItem::Message(MessageItem {
487                id: item.id.clone().into(),
488                status,
489                role: MessageRole::Assistant,
490                content: vec![ContentPart::output_text(&msg.text)],
491            }),
492
493            ThreadItemDetails::Reasoning(r) => OutputItem::Reasoning(ReasoningItem {
494                id: item.id.clone().into(),
495                status,
496                summary: None,
497                content: Some(r.text.clone()),
498                encrypted_content: None,
499            }),
500
501            ThreadItemDetails::Plan(plan) => OutputItem::Custom(CustomItem {
502                id: item.id.clone().into(),
503                status,
504                custom_type: "vtcode:plan".to_string(),
505                data: json!({
506                    "text": plan.text,
507                }),
508            }),
509
510            ThreadItemDetails::CommandExecution(cmd) => OutputItem::Custom(CustomItem {
511                id: item.id.clone().into(),
512                status,
513                custom_type: "vtcode:command_execution".to_string(),
514                data: json!({
515                    "command": cmd.command,
516                    "arguments": cmd.arguments,
517                    "aggregated_output": cmd.aggregated_output,
518                    "exit_code": cmd.exit_code,
519                    "status": serde_json::to_value(&cmd.status).unwrap_or(serde_json::Value::Null),
520                }),
521            }),
522
523            ThreadItemDetails::ToolInvocation(invocation) => OutputItem::FunctionCall(FunctionCallItem {
524                id: item.id.clone().into(),
525                status,
526                name: invocation.tool_name.clone(),
527                arguments: invocation.arguments.clone().unwrap_or(json!({})),
528                call_id: Some(self.resolve_tool_call_correlation_id(&item.id, invocation.tool_call_id.as_deref())),
529            }),
530
531            ThreadItemDetails::ToolOutput(output) => {
532                OutputItem::FunctionCallOutput(crate::open_responses::FunctionCallOutputItem {
533                    id: item.id.clone().into(),
534                    status,
535                    call_id: Some(
536                        self.resolve_tool_call_correlation_id(&output.call_id, output.tool_call_id.as_deref()),
537                    ),
538                    output: tool_output_text(output),
539                })
540            }
541
542            ThreadItemDetails::FileChange(fc) => {
543                let changes: Vec<_> = fc
544                    .changes
545                    .iter()
546                    .map(|c| {
547                        json!({
548                            "path": c.path,
549                            "kind": format!("{:?}", c.kind).to_lowercase(),
550                        })
551                    })
552                    .collect();
553
554                OutputItem::Custom(CustomItem {
555                    id: item.id.clone().into(),
556                    status,
557                    custom_type: "vtcode:file_change".to_string(),
558                    data: json!({
559                        "changes": changes,
560                        "status": format!("{:?}", fc.status).to_lowercase(),
561                    }),
562                })
563            }
564
565            ThreadItemDetails::McpToolCall(tc) => OutputItem::FunctionCall(FunctionCallItem {
566                id: item.id.clone().into(),
567                status,
568                name: tc.tool_name.clone(),
569                arguments: tc.arguments.clone().unwrap_or(json!({})),
570                call_id: Some(item.id.clone()),
571            }),
572
573            ThreadItemDetails::WebSearch(ws) => OutputItem::Custom(CustomItem {
574                id: item.id.clone().into(),
575                status,
576                custom_type: "vtcode:web_search".to_string(),
577                data: json!({
578                    "query": ws.query,
579                    "provider": ws.provider,
580                    "results": ws.results,
581                }),
582            }),
583
584            ThreadItemDetails::Harness(event) => OutputItem::Custom(CustomItem {
585                id: item.id.clone().into(),
586                status,
587                custom_type: "vtcode:harness_event".to_string(),
588                data: json!({
589                    "event": serde_json::to_value(&event.event).unwrap_or(serde_json::Value::Null),
590                    "message": event.message,
591                    "command": event.command,
592                    "path": event.path,
593                    "exit_code": event.exit_code,
594                }),
595            }),
596
597            ThreadItemDetails::Error(err) => {
598                // Errors are represented as custom items
599                OutputItem::Custom(CustomItem {
600                    id: item.id.clone().into(),
601                    status: ItemStatus::Failed,
602                    custom_type: "vtcode:error".to_string(),
603                    data: json!({
604                        "message": err.message,
605                    }),
606                })
607            }
608        }
609    }
610}
vtcode_llm/open_responses/bridge/items.rs

vtcode_llm/open_responses/bridge/
items.rs
