1use std::collections::HashMap;
2
3use crate::anthropic::sse::encode_sse_event;
4use crate::providers::codex::events::is_terminal_rate_limit_event;
5use crate::traffic::TrafficCapture;
6
7use super::read_rewrite::sanitize_read_args;
8use super::reasoning_signature::{PendingReasoning, encode_reasoning_signature};
9use super::reducer::{
10 CodexUsage, STOP_END_TURN, STOP_MAX_TOKENS, STOP_TOOL_USE, map_codex_usage_to_anthropic,
11};
12
13const BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES: usize = 1_024;
14const BUFFERED_TOOL_MAX_ARGS_BYTES: usize = 5_000_000;
15
16enum LiveBlock {
17 Text {
18 index: usize,
19 text: String,
20 deferred: bool,
21 },
22 Tool {
23 index: usize,
24 call_id: String,
25 name: String,
26 args_accum: String,
27 had_delta: bool,
28 buffer_until_done: bool,
29 emitted_args: bool,
30 },
31}
32
33struct LiveWebSearch {
34 index: usize,
35 result_index: usize,
36 id: String,
37 query: String,
38}
39
40#[derive(Clone)]
41struct LiveWebSearchResult {
42 title: String,
43 url: String,
44}
45
46#[derive(Clone, Copy)]
47struct LiveThinking {
48 output_index: usize,
49 anthropic_index: usize,
50}
51
52pub struct LiveStreamTranslator {
53 message_id: String,
54 model: String,
55 message_started: bool,
56 blocks_by_output_index: HashMap<usize, LiveBlock>,
57 item_id_to_output_index: HashMap<String, usize>,
58 anthropic_index: usize,
59 thinking: Option<LiveThinking>,
60 reasoning_by_output_index: HashMap<usize, PendingReasoning>,
61 saw_tool_use: bool,
62 web_search_requests: usize,
63 web_searches: Vec<LiveWebSearch>,
64 web_search_results: Vec<LiveWebSearchResult>,
65 deferred_text: Vec<(usize, String)>,
66 semantic_output_started: bool,
67 estimated_input_tokens: u64,
70 finished: bool,
71}
72
73impl LiveStreamTranslator {
74 pub fn new(message_id: impl Into<String>, model: impl Into<String>) -> Self {
75 Self::with_estimated_input_tokens(message_id, model, 0)
76 }
77
78 pub fn with_estimated_input_tokens(
79 message_id: impl Into<String>,
80 model: impl Into<String>,
81 estimated_input_tokens: u64,
82 ) -> Self {
83 Self {
84 message_id: message_id.into(),
85 model: model.into(),
86 message_started: false,
87 blocks_by_output_index: HashMap::new(),
88 item_id_to_output_index: HashMap::new(),
89 anthropic_index: 0,
90 thinking: None,
91 reasoning_by_output_index: HashMap::new(),
92 saw_tool_use: false,
93 web_search_requests: 0,
94 web_searches: Vec::new(),
95 web_search_results: Vec::new(),
96 deferred_text: Vec::new(),
97 semantic_output_started: false,
98 estimated_input_tokens,
99 finished: false,
100 }
101 }
102
103 pub fn accept(
104 &mut self,
105 payload: &serde_json::Value,
106 traffic: Option<&TrafficCapture>,
107 ) -> Result<Vec<u8>, String> {
108 if self.finished {
109 return Ok(Vec::new());
110 }
111
112 let kind = payload.get("type").and_then(|v| v.as_str()).unwrap_or("");
113 let mut out = Vec::new();
114
115 match kind {
116 "codex.rate_limits" => {
117 if is_terminal_rate_limit_event(payload) {
118 return Err("rate limit reached".to_string());
119 }
120 self.emit_ping(traffic, &mut out);
121 }
122 "keepalive" | "response.created" | "response.in_progress" => {
123 self.emit_ping(traffic, &mut out);
124 }
125 "response.failed" | "response.error" | "error" => {
126 return Err(error_message(payload));
127 }
128 "response.web_search_call.in_progress"
129 | "response.web_search_call.searching"
130 | "response.web_search_call.completed" => {}
131 "response.output_item.added" => {
132 self.output_item_added(payload, traffic, &mut out);
133 }
134 "response.reasoning_summary_part.added" => {
135 let output_index = output_index(payload);
136 if let Some(thinking) = self
137 .thinking
138 .filter(|thinking| thinking.output_index == output_index)
139 {
140 self.emit(
141 traffic,
142 &mut out,
143 "content_block_delta",
144 &serde_json::json!({
145 "type": "content_block_delta",
146 "index": thinking.anthropic_index,
147 "delta": {"type": "thinking_delta", "thinking": "\n\n"}
148 }),
149 );
150 }
151 }
152 "response.reasoning_summary_text.delta" => {
153 self.reasoning_delta(payload, traffic, &mut out);
154 }
155 "response.output_text.delta" => {
156 self.text_delta(payload, traffic, &mut out);
157 }
158 "response.output_text.annotation.added" => {
159 self.web_search_annotation(payload);
160 }
161 "response.function_call_arguments.delta" => {
162 self.tool_delta(payload, traffic, &mut out)?;
163 }
164 "response.function_call_arguments.done" => {
165 self.tool_arguments_done(payload);
166 }
167 "response.output_item.done" => {
168 self.output_item_done(payload, traffic, &mut out);
169 }
170 "response.completed" | "response.incomplete" | "response.done" => {
171 self.finish(payload, traffic, &mut out);
172 }
173 _ => {}
174 }
175
176 Ok(out)
177 }
178
179 pub fn is_finished(&self) -> bool {
180 self.finished
181 }
182
183 pub fn has_semantic_output(&self) -> bool {
184 self.semantic_output_started
185 }
186
187 pub fn ping_chunk(&mut self, traffic: Option<&TrafficCapture>) -> Vec<u8> {
188 let mut out = Vec::new();
189 if !self.finished {
190 self.emit_ping(traffic, &mut out);
191 }
192 out
193 }
194
195 pub fn finish_after_closed_completed_tool_call(
196 &mut self,
197 traffic: Option<&TrafficCapture>,
198 ) -> Vec<u8> {
199 let mut out = Vec::new();
200 if self.finished || !self.saw_tool_use || !self.blocks_by_output_index.is_empty() {
201 return out;
202 }
203 self.close_thinking(traffic, &mut out);
204 self.ensure_message_start(traffic, &mut out);
205 self.emit_finish(STOP_TOOL_USE, None, traffic, &mut out);
206 out
207 }
208
209 pub fn error_chunk(
210 &mut self,
211 message: &str,
212 error_type: &str,
213 traffic: Option<&TrafficCapture>,
214 ) -> Vec<u8> {
215 let mut out = Vec::new();
216 if self.finished {
217 return out;
218 }
219 self.close_open_blocks(traffic, &mut out);
220 self.ensure_message_start(traffic, &mut out);
221 self.emit(
222 traffic,
223 &mut out,
224 "error",
225 &serde_json::json!({
226 "type": "error",
227 "error": {
228 "type": error_type,
229 "message": message,
230 }
231 }),
232 );
233 self.finished = true;
234 out
235 }
236
237 fn ensure_message_start(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec<u8>) {
238 if self.message_started {
239 return;
240 }
241 self.message_started = true;
242 self.emit(
243 traffic,
244 out,
245 "message_start",
246 &serde_json::json!({
247 "type": "message_start",
248 "message": {
249 "id": self.message_id,
250 "type": "message",
251 "role": "assistant",
252 "model": self.model,
253 "content": [],
254 "stop_reason": null,
255 "stop_sequence": null,
256 "usage": {
257 "input_tokens": self.estimated_input_tokens,
258 "output_tokens": 0
259 }
260 }
261 }),
262 );
263 }
264
265 fn emit_ping(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec<u8>) {
266 self.ensure_message_start(traffic, out);
267 self.emit(traffic, out, "ping", &serde_json::json!({"type": "ping"}));
268 }
269
270 fn emit(
271 &self,
272 traffic: Option<&TrafficCapture>,
273 out: &mut Vec<u8>,
274 event: &str,
275 data: &serde_json::Value,
276 ) {
277 if let Some(traffic) = traffic {
278 traffic.write_json_event(
279 "050-downstream-event",
280 &serde_json::json!({
281 "event": event,
282 "data": data,
283 }),
284 );
285 }
286 out.extend_from_slice(&encode_sse_event(Some(event), &data.to_string()));
287 }
288
289 fn output_item_added(
290 &mut self,
291 payload: &serde_json::Value,
292 traffic: Option<&TrafficCapture>,
293 out: &mut Vec<u8>,
294 ) {
295 let Some(item) = payload.get("item") else {
296 return;
297 };
298 let output_index = output_index(payload);
299 let item_type = item.get("type").and_then(|v| v.as_str()).unwrap_or("");
300
301 match item_type {
302 "reasoning" => {
303 self.reasoning_by_output_index
304 .entry(output_index)
305 .or_default()
306 .capture(item);
307 }
308 "message" => {
309 self.close_thinking(traffic, out);
310 let index = self.anthropic_index;
311 self.anthropic_index += 1;
312 if let Some(id) = item.get("id").and_then(|v| v.as_str()) {
313 self.item_id_to_output_index
314 .insert(id.to_string(), output_index);
315 }
316 let deferred = !self.web_searches.is_empty();
317 self.blocks_by_output_index.insert(
318 output_index,
319 LiveBlock::Text {
320 index,
321 text: String::new(),
322 deferred,
323 },
324 );
325 if !deferred {
326 self.ensure_message_start(traffic, out);
327 self.emit(
328 traffic,
329 out,
330 "content_block_start",
331 &serde_json::json!({
332 "type": "content_block_start",
333 "index": index,
334 "content_block": {"type": "text", "text": ""}
335 }),
336 );
337 }
338 }
339 "function_call" => {
340 self.close_thinking(traffic, out);
341 self.saw_tool_use = true;
342 self.semantic_output_started = true;
343 let index = self.anthropic_index;
344 self.anthropic_index += 1;
345 let call_id = item
346 .get("call_id")
347 .and_then(|v| v.as_str())
348 .unwrap_or("")
349 .to_string();
350 let name = item
351 .get("name")
352 .and_then(|v| v.as_str())
353 .unwrap_or("")
354 .to_string();
355 self.blocks_by_output_index.insert(
356 output_index,
357 LiveBlock::Tool {
358 index,
359 call_id: call_id.clone(),
360 name: name.clone(),
361 args_accum: String::new(),
362 had_delta: false,
363 buffer_until_done: name == "Read",
364 emitted_args: false,
365 },
366 );
367 self.ensure_message_start(traffic, out);
368 self.emit(
369 traffic,
370 out,
371 "content_block_start",
372 &serde_json::json!({
373 "type": "content_block_start",
374 "index": index,
375 "content_block": {
376 "type": "tool_use",
377 "id": call_id,
378 "name": name,
379 "input": {}
380 }
381 }),
382 );
383 }
384 "web_search_call" => {
385 self.web_search_requests += 1;
386 }
387 _ => {}
388 }
389 }
390
391 fn reasoning_delta(
392 &mut self,
393 payload: &serde_json::Value,
394 traffic: Option<&TrafficCapture>,
395 out: &mut Vec<u8>,
396 ) {
397 let output_index = output_index(payload);
398 let delta = payload.get("delta").and_then(|v| v.as_str()).unwrap_or("");
399 if delta.is_empty() {
400 return;
401 }
402 self.semantic_output_started = true;
403 if self.thinking.map(|thinking| thinking.output_index) != Some(output_index) {
404 self.close_thinking(traffic, out);
405 let index = self.anthropic_index;
406 self.anthropic_index += 1;
407 self.thinking = Some(LiveThinking {
408 output_index,
409 anthropic_index: index,
410 });
411 self.ensure_message_start(traffic, out);
412 self.emit(
413 traffic,
414 out,
415 "content_block_start",
416 &serde_json::json!({
417 "type": "content_block_start",
418 "index": index,
419 "content_block": {"type": "thinking", "thinking": "", "signature": ""}
420 }),
421 );
422 }
423 let index = self
424 .thinking
425 .expect("thinking block was started")
426 .anthropic_index;
427 self.emit(
428 traffic,
429 out,
430 "content_block_delta",
431 &serde_json::json!({
432 "type": "content_block_delta",
433 "index": index,
434 "delta": {"type": "thinking_delta", "thinking": delta}
435 }),
436 );
437 }
438
439 fn text_delta(
440 &mut self,
441 payload: &serde_json::Value,
442 traffic: Option<&TrafficCapture>,
443 out: &mut Vec<u8>,
444 ) {
445 self.close_thinking(traffic, out);
446 let delta = payload.get("delta").and_then(|v| v.as_str()).unwrap_or("");
447 if delta.is_empty() {
448 return;
449 }
450 self.semantic_output_started = true;
451
452 let output_index = payload
453 .get("output_index")
454 .and_then(|v| v.as_u64())
455 .map(|v| v as usize)
456 .or_else(|| {
457 payload
458 .get("item_id")
459 .and_then(|v| v.as_str())
460 .and_then(|id| self.item_id_to_output_index.get(id).copied())
461 })
462 .unwrap_or(0);
463
464 if !self.blocks_by_output_index.contains_key(&output_index) {
465 let index = self.anthropic_index;
466 self.anthropic_index += 1;
467 let deferred = !self.web_searches.is_empty();
468 self.blocks_by_output_index.insert(
469 output_index,
470 LiveBlock::Text {
471 index,
472 text: String::new(),
473 deferred,
474 },
475 );
476 if !deferred {
477 self.ensure_message_start(traffic, out);
478 self.emit(
479 traffic,
480 out,
481 "content_block_start",
482 &serde_json::json!({
483 "type": "content_block_start",
484 "index": index,
485 "content_block": {"type": "text", "text": ""}
486 }),
487 );
488 }
489 }
490
491 let Some(LiveBlock::Text {
492 index,
493 text,
494 deferred,
495 }) = self.blocks_by_output_index.get_mut(&output_index)
496 else {
497 return;
498 };
499 text.push_str(delta);
500 if *deferred {
501 return;
502 }
503 let index = *index;
504 self.emit(
505 traffic,
506 out,
507 "content_block_delta",
508 &serde_json::json!({
509 "type": "content_block_delta",
510 "index": index,
511 "delta": {"type": "text_delta", "text": delta}
512 }),
513 );
514 }
515
516 fn tool_delta(
517 &mut self,
518 payload: &serde_json::Value,
519 traffic: Option<&TrafficCapture>,
520 out: &mut Vec<u8>,
521 ) -> Result<(), String> {
522 let Some(output_index) = payload
523 .get("output_index")
524 .and_then(|v| v.as_u64())
525 .map(|v| v as usize)
526 else {
527 return Ok(());
528 };
529 let delta = payload.get("delta").and_then(|v| v.as_str()).unwrap_or("");
530 if delta.is_empty() {
531 return Ok(());
532 }
533 let mut repaired_read: Option<(usize, String)> = None;
534 let Some(LiveBlock::Tool {
535 index,
536 call_id,
537 name,
538 args_accum,
539 had_delta,
540 buffer_until_done,
541 emitted_args,
542 ..
543 }) = self.blocks_by_output_index.get_mut(&output_index)
544 else {
545 return Ok(());
546 };
547 args_accum.push_str(delta);
548 *had_delta = true;
549 if *buffer_until_done {
550 if args_accum.len() > BUFFERED_TOOL_MAX_ARGS_BYTES {
551 return Err(format!(
552 "Buffered {name} tool arguments exceeded safe limits"
553 ));
554 }
555 if let Some(repaired) =
556 repair_whitespace_stalled_read_args(name, args_accum, Some(call_id.as_str()))
557 {
558 *args_accum = repaired.clone();
559 *emitted_args = true;
560 repaired_read = Some((*index, repaired));
561 }
562 } else {
563 *emitted_args = true;
564 let index = *index;
565 self.emit(
566 traffic,
567 out,
568 "content_block_delta",
569 &serde_json::json!({
570 "type": "content_block_delta",
571 "index": index,
572 "delta": {
573 "type": "input_json_delta",
574 "partial_json": delta
575 }
576 }),
577 );
578 return Ok(());
579 }
580 if let Some((index, repaired)) = repaired_read {
581 self.blocks_by_output_index.remove(&output_index);
582 self.emit(
583 traffic,
584 out,
585 "content_block_delta",
586 &serde_json::json!({
587 "type": "content_block_delta",
588 "index": index,
589 "delta": {
590 "type": "input_json_delta",
591 "partial_json": repaired
592 }
593 }),
594 );
595 self.emit(
596 traffic,
597 out,
598 "content_block_stop",
599 &serde_json::json!({
600 "type": "content_block_stop",
601 "index": index,
602 }),
603 );
604 self.ensure_message_start(traffic, out);
605 self.emit_finish(STOP_TOOL_USE, None, traffic, out);
606 }
607 Ok(())
608 }
609
610 fn tool_arguments_done(&mut self, payload: &serde_json::Value) {
611 let Some(output_index) = payload
612 .get("output_index")
613 .and_then(|v| v.as_u64())
614 .map(|v| v as usize)
615 else {
616 return;
617 };
618 let Some(args) = payload.get("arguments").and_then(|v| v.as_str()) else {
619 return;
620 };
621 let Some(LiveBlock::Tool { args_accum, .. }) =
622 self.blocks_by_output_index.get_mut(&output_index)
623 else {
624 return;
625 };
626 if args_accum.is_empty() {
627 *args_accum = args.to_string();
628 }
629 }
630
631 fn output_item_done(
632 &mut self,
633 payload: &serde_json::Value,
634 traffic: Option<&TrafficCapture>,
635 out: &mut Vec<u8>,
636 ) {
637 let output_index = output_index(payload);
638 if let Some(item) = payload
639 .get("item")
640 .and_then(|item| item.get("type"))
641 .and_then(|v| v.as_str())
642 .filter(|item_type| *item_type == "reasoning")
643 .and_then(|_| payload.get("item"))
644 {
645 self.reasoning_by_output_index
646 .entry(output_index)
647 .or_default()
648 .capture(item);
649 let had_active_summary = self
650 .thinking
651 .is_some_and(|thinking| thinking.output_index == output_index);
652 self.close_thinking(traffic, out);
653 if !had_active_summary {
654 self.emit_signature_only_reasoning(output_index, traffic, out);
655 }
656 return;
657 }
658
659 if payload
660 .get("item")
661 .and_then(|item| item.get("type"))
662 .and_then(|v| v.as_str())
663 == Some("web_search_call")
664 {
665 self.close_thinking(traffic, out);
666 self.semantic_output_started = true;
667 let item = &payload["item"];
668 let index = self.anthropic_index;
669 self.anthropic_index += 1;
670 let result_index = self.anthropic_index;
671 self.anthropic_index += 1;
672 let raw_id = item.get("id").and_then(|v| v.as_str()).unwrap_or("");
673 self.web_searches.push(LiveWebSearch {
674 index,
675 result_index,
676 id: super::web_search_compat::server_tool_use_id_from_codex_web_search_id(raw_id),
677 query: web_search_query(item),
678 });
679 return;
680 }
681
682 let Some(mut state) = self.blocks_by_output_index.remove(&output_index) else {
683 return;
684 };
685
686 match &mut state {
687 LiveBlock::Text {
688 index,
689 text,
690 deferred,
691 } => {
692 if *deferred {
693 self.deferred_text.push((*index, std::mem::take(text)));
694 } else {
695 self.emit(
696 traffic,
697 out,
698 "content_block_stop",
699 &serde_json::json!({
700 "type": "content_block_stop",
701 "index": index,
702 }),
703 );
704 }
705 }
706 LiveBlock::Tool {
707 index,
708 name,
709 call_id,
710 args_accum,
711 had_delta,
712 buffer_until_done,
713 emitted_args,
714 ..
715 } => {
716 if let Some(final_args) = payload
717 .get("item")
718 .and_then(|item| item.get("arguments"))
719 .and_then(|v| v.as_str())
720 .filter(|s| !s.is_empty())
721 && (args_accum.is_empty() || (!*had_delta && !*emitted_args))
722 {
723 *args_accum = final_args.to_string();
724 }
725 if !args_accum.is_empty() {
726 *args_accum = sanitize_read_args(name, args_accum, Some(call_id.as_str()));
727 if *buffer_until_done || !*emitted_args {
728 *emitted_args = true;
729 self.emit(
730 traffic,
731 out,
732 "content_block_delta",
733 &serde_json::json!({
734 "type": "content_block_delta",
735 "index": index,
736 "delta": {
737 "type": "input_json_delta",
738 "partial_json": args_accum
739 }
740 }),
741 );
742 }
743 }
744 self.emit(
745 traffic,
746 out,
747 "content_block_stop",
748 &serde_json::json!({
749 "type": "content_block_stop",
750 "index": index,
751 }),
752 );
753 }
754 }
755 }
756
757 fn web_search_annotation(&mut self, payload: &serde_json::Value) {
758 let Some(annotation) = payload.get("annotation") else {
759 return;
760 };
761 if annotation.get("type").and_then(|v| v.as_str()) != Some("url_citation") {
762 return;
763 }
764 let Some(url) = annotation.get("url").and_then(|v| v.as_str()) else {
765 return;
766 };
767 if self
768 .web_search_results
769 .iter()
770 .any(|result| result.url == url)
771 {
772 return;
773 }
774 let title = annotation
775 .get("title")
776 .and_then(|v| v.as_str())
777 .filter(|title| !title.is_empty())
778 .unwrap_or(url);
779 self.web_search_results.push(LiveWebSearchResult {
780 title: title.to_string(),
781 url: url.to_string(),
782 });
783 }
784
785 fn emit_web_searches(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec<u8>) {
786 let searches = std::mem::take(&mut self.web_searches);
787 for search in searches {
788 self.ensure_message_start(traffic, out);
789 self.emit(
790 traffic,
791 out,
792 "content_block_start",
793 &serde_json::json!({
794 "type": "content_block_start",
795 "index": search.index,
796 "content_block": {
797 "type": "server_tool_use",
798 "id": search.id,
799 "name": "web_search",
800 "input": {}
801 }
802 }),
803 );
804 self.emit(
805 traffic,
806 out,
807 "content_block_delta",
808 &serde_json::json!({
809 "type": "content_block_delta",
810 "index": search.index,
811 "delta": {
812 "type": "input_json_delta",
813 "partial_json": serde_json::to_string(&serde_json::json!({"query": search.query})).unwrap_or_default()
814 }
815 }),
816 );
817 self.emit(
818 traffic,
819 out,
820 "content_block_stop",
821 &serde_json::json!({"type": "content_block_stop", "index": search.index}),
822 );
823 let results: Vec<_> = self
824 .web_search_results
825 .iter()
826 .map(|result| {
827 serde_json::json!({
828 "type": "web_search_result",
829 "title": result.title,
830 "url": result.url,
831 })
832 })
833 .collect();
834 self.emit(
835 traffic,
836 out,
837 "content_block_start",
838 &serde_json::json!({
839 "type": "content_block_start",
840 "index": search.result_index,
841 "content_block": {
842 "type": "web_search_tool_result",
843 "tool_use_id": search.id,
844 "content": results
845 }
846 }),
847 );
848 self.emit(
849 traffic,
850 out,
851 "content_block_stop",
852 &serde_json::json!({"type": "content_block_stop", "index": search.result_index}),
853 );
854 }
855
856 for (index, text) in std::mem::take(&mut self.deferred_text) {
857 self.emit(
858 traffic,
859 out,
860 "content_block_start",
861 &serde_json::json!({
862 "type": "content_block_start",
863 "index": index,
864 "content_block": {"type": "text", "text": ""}
865 }),
866 );
867 if !text.is_empty() {
868 self.emit(
869 traffic,
870 out,
871 "content_block_delta",
872 &serde_json::json!({
873 "type": "content_block_delta",
874 "index": index,
875 "delta": {"type": "text_delta", "text": text}
876 }),
877 );
878 }
879 self.emit(
880 traffic,
881 out,
882 "content_block_stop",
883 &serde_json::json!({"type": "content_block_stop", "index": index}),
884 );
885 }
886 }
887
888 fn finish(
889 &mut self,
890 payload: &serde_json::Value,
891 traffic: Option<&TrafficCapture>,
892 out: &mut Vec<u8>,
893 ) {
894 self.close_thinking(traffic, out);
895 self.close_open_blocks(traffic, out);
896 self.emit_web_searches(traffic, out);
897 self.ensure_message_start(traffic, out);
898 let usage = payload.get("response").map(parse_codex_usage);
899 let incomplete = response_is_incomplete(payload);
900 let stop_reason = if incomplete {
901 STOP_MAX_TOKENS
902 } else if self.saw_tool_use {
903 STOP_TOOL_USE
904 } else {
905 STOP_END_TURN
906 };
907 self.emit_finish(stop_reason, usage, traffic, out);
908 }
909
910 fn emit_finish(
911 &mut self,
912 stop_reason: &str,
913 usage: Option<CodexUsage>,
914 traffic: Option<&TrafficCapture>,
915 out: &mut Vec<u8>,
916 ) {
917 let mapped = map_codex_usage_to_anthropic(&usage, Some(self.web_search_requests));
918 self.emit(
919 traffic,
920 out,
921 "message_delta",
922 &serde_json::json!({
923 "type": "message_delta",
924 "delta": {
925 "stop_reason": stop_reason,
926 "stop_sequence": null
927 },
928 "usage": mapped,
929 }),
930 );
931 self.emit(
932 traffic,
933 out,
934 "message_stop",
935 &serde_json::json!({"type": "message_stop"}),
936 );
937 self.finished = true;
938 }
939
940 fn close_open_blocks(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec<u8>) {
941 self.close_thinking(traffic, out);
942 let open: Vec<usize> = self.blocks_by_output_index.keys().copied().collect();
943 for output_index in open {
944 let Some(state) = self.blocks_by_output_index.remove(&output_index) else {
945 continue;
946 };
947 let index = match state {
948 LiveBlock::Text {
949 index,
950 text,
951 deferred: true,
952 } => {
953 self.deferred_text.push((index, text));
954 continue;
955 }
956 LiveBlock::Text { index, .. } => index,
957 LiveBlock::Tool { index, .. } => index,
958 };
959 self.emit(
960 traffic,
961 out,
962 "content_block_stop",
963 &serde_json::json!({
964 "type": "content_block_stop",
965 "index": index,
966 }),
967 );
968 }
969 }
970
971 fn emit_signature_only_reasoning(
972 &mut self,
973 output_index: usize,
974 traffic: Option<&TrafficCapture>,
975 out: &mut Vec<u8>,
976 ) {
977 let Some(replay) = self
978 .reasoning_by_output_index
979 .remove(&output_index)
980 .and_then(|pending| pending.replay())
981 else {
982 return;
983 };
984 let Some(signature) = encode_reasoning_signature(&replay) else {
985 return;
986 };
987 self.semantic_output_started = true;
988 let index = self.anthropic_index;
989 self.anthropic_index += 1;
990 self.ensure_message_start(traffic, out);
991 self.emit(
992 traffic,
993 out,
994 "content_block_start",
995 &serde_json::json!({
996 "type": "content_block_start",
997 "index": index,
998 "content_block": {"type": "thinking", "thinking": "", "signature": ""}
999 }),
1000 );
1001 self.emit(
1002 traffic,
1003 out,
1004 "content_block_delta",
1005 &serde_json::json!({
1006 "type": "content_block_delta",
1007 "index": index,
1008 "delta": {"type": "signature_delta", "signature": signature}
1009 }),
1010 );
1011 self.emit(
1012 traffic,
1013 out,
1014 "content_block_stop",
1015 &serde_json::json!({
1016 "type": "content_block_stop",
1017 "index": index,
1018 }),
1019 );
1020 }
1021
1022 fn close_thinking(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec<u8>) {
1023 let Some(thinking) = self.thinking.take() else {
1024 return;
1025 };
1026 if let Some(signature) = self
1027 .reasoning_by_output_index
1028 .remove(&thinking.output_index)
1029 .and_then(|pending| pending.replay())
1030 .and_then(|replay| encode_reasoning_signature(&replay))
1031 {
1032 self.emit(
1033 traffic,
1034 out,
1035 "content_block_delta",
1036 &serde_json::json!({
1037 "type": "content_block_delta",
1038 "index": thinking.anthropic_index,
1039 "delta": {"type": "signature_delta", "signature": signature}
1040 }),
1041 );
1042 }
1043 self.emit(
1044 traffic,
1045 out,
1046 "content_block_stop",
1047 &serde_json::json!({
1048 "type": "content_block_stop",
1049 "index": thinking.anthropic_index,
1050 }),
1051 );
1052 }
1053}
1054
1055fn web_search_query(item: &serde_json::Value) -> String {
1056 let Some(action) = item.get("action") else {
1057 return String::new();
1058 };
1059 action
1060 .get("query")
1061 .and_then(|v| v.as_str())
1062 .or_else(|| {
1063 action
1064 .get("queries")
1065 .and_then(|v| v.as_array())
1066 .and_then(|queries| queries.iter().find_map(|query| query.as_str()))
1067 })
1068 .unwrap_or("")
1069 .to_string()
1070}
1071
1072fn output_index(payload: &serde_json::Value) -> usize {
1073 payload
1074 .get("output_index")
1075 .and_then(|v| v.as_u64())
1076 .unwrap_or(0) as usize
1077}
1078
1079fn parse_codex_usage(response: &serde_json::Value) -> CodexUsage {
1080 let usage = match response.get("usage") {
1081 Some(u) => u,
1082 None => return CodexUsage::default(),
1083 };
1084 CodexUsage {
1085 input_tokens: usage.get("input_tokens").and_then(|v| v.as_u64()),
1086 output_tokens: usage.get("output_tokens").and_then(|v| v.as_u64()),
1087 input_tokens_details_cached: usage
1088 .get("input_tokens_details")
1089 .and_then(|d| d.get("cached_tokens"))
1090 .and_then(|v| v.as_u64()),
1091 output_tokens_details_reasoning: usage
1092 .get("output_tokens_details")
1093 .and_then(|d| d.get("reasoning_tokens"))
1094 .and_then(|v| v.as_u64()),
1095 }
1096}
1097
1098fn response_is_incomplete(payload: &serde_json::Value) -> bool {
1099 payload.get("type").and_then(|v| v.as_str()) == Some("response.incomplete")
1100 || payload
1101 .get("response")
1102 .and_then(|r| r.get("status"))
1103 .and_then(|v| v.as_str())
1104 == Some("incomplete")
1105 || payload
1106 .get("response")
1107 .and_then(|r| r.get("incomplete_details"))
1108 .and_then(|d| d.get("reason"))
1109 .and_then(|v| v.as_str())
1110 .is_some()
1111}
1112
1113fn repair_whitespace_stalled_read_args(
1114 name: &str,
1115 args: &str,
1116 call_id: Option<&str>,
1117) -> Option<String> {
1118 if name != "Read" {
1119 return None;
1120 }
1121 let trimmed = args.trim_end();
1122 let trailing_whitespace = args.len().saturating_sub(trimmed.len());
1123 if trailing_whitespace < BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES {
1124 return None;
1125 }
1126 parse_read_args_candidate(trimmed, call_id).or_else(|| {
1127 let with_brace = format!("{trimmed}}}");
1128 parse_read_args_candidate(&with_brace, call_id)
1129 })
1130}
1131
1132fn parse_read_args_candidate(args: &str, call_id: Option<&str>) -> Option<String> {
1133 let parsed: serde_json::Value = serde_json::from_str(args).ok()?;
1134 if !is_valid_read_args(&parsed) {
1135 return None;
1136 }
1137 Some(sanitize_read_args(
1138 "Read",
1139 &serde_json::to_string(&parsed).ok()?,
1140 call_id,
1141 ))
1142}
1143
1144fn is_valid_read_args(value: &serde_json::Value) -> bool {
1145 let Some(obj) = value.as_object() else {
1146 return false;
1147 };
1148 for key in obj.keys() {
1149 if !matches!(key.as_str(), "file_path" | "offset" | "limit" | "pages") {
1150 return false;
1151 }
1152 }
1153 let Some(file_path) = obj.get("file_path").and_then(|v| v.as_str()) else {
1154 return false;
1155 };
1156 if file_path.is_empty() {
1157 return false;
1158 }
1159 if let Some(offset) = obj.get("offset").and_then(|v| v.as_i64())
1160 && offset < 0
1161 {
1162 return false;
1163 }
1164 if let Some(limit) = obj.get("limit").and_then(|v| v.as_i64())
1165 && limit <= 0
1166 {
1167 return false;
1168 }
1169 if obj.get("offset").is_some_and(|v| !v.is_i64()) {
1170 return false;
1171 }
1172 if obj.get("limit").is_some_and(|v| !v.is_i64()) {
1173 return false;
1174 }
1175 if obj.get("pages").is_some_and(|v| !v.is_string()) {
1176 return false;
1177 }
1178 true
1179}
1180
1181fn error_message(payload: &serde_json::Value) -> String {
1182 payload
1183 .get("response")
1184 .and_then(|r| r.get("error"))
1185 .and_then(|e| e.get("message"))
1186 .and_then(|v| v.as_str())
1187 .or_else(|| {
1188 payload
1189 .get("error")
1190 .and_then(|e| e.get("message"))
1191 .and_then(|v| v.as_str())
1192 })
1193 .unwrap_or("Upstream error")
1194 .to_string()
1195}
1196
1197#[cfg(test)]
1198mod tests {
1199 use super::*;
1200 use crate::anthropic::sse::parse_sse_events;
1201 use serde_json::json;
1202
1203 fn render(events: Vec<serde_json::Value>) -> String {
1204 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1205 let mut out = Vec::new();
1206 for event in events {
1207 out.extend(translator.accept(&event, None).unwrap());
1208 }
1209 String::from_utf8(out).unwrap()
1210 }
1211
1212 #[test]
1213 fn emits_text_delta_before_terminal_event() {
1214 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1215 let out = translator
1216 .accept(
1217 &json!({
1218 "type": "response.output_text.delta",
1219 "output_index": 0,
1220 "delta": "hello"
1221 }),
1222 None,
1223 )
1224 .unwrap();
1225 let out = String::from_utf8(out).unwrap();
1226 assert!(out.contains("message_start"));
1227 assert!(out.contains("content_block_start"));
1228 assert!(out.contains("text_delta"));
1229 assert!(out.contains("hello"));
1230 assert!(!out.contains("message_stop"));
1231 assert!(translator.has_semantic_output());
1232 }
1233
1234 #[test]
1235 fn estimated_input_is_visible_at_start_and_provider_usage_is_exact_at_finish() {
1236 let mut translator =
1237 LiveStreamTranslator::with_estimated_input_tokens("msg_1", "gpt-5.5", 321);
1238
1239 let started = translator
1240 .accept(
1241 &json!({
1242 "type": "response.output_text.delta",
1243 "output_index": 0,
1244 "delta": "abcdefgh"
1245 }),
1246 None,
1247 )
1248 .unwrap();
1249 let started = parse_sse_events(&started)
1250 .into_iter()
1251 .filter_map(|event| serde_json::from_str::<serde_json::Value>(&event.data).ok())
1252 .find(|value| {
1253 value.get("type").and_then(serde_json::Value::as_str) == Some("message_start")
1254 })
1255 .unwrap();
1256 assert_eq!(
1257 started.pointer("/message/usage/input_tokens"),
1258 Some(&json!(321))
1259 );
1260
1261 let finished = translator
1262 .accept(
1263 &json!({
1264 "type": "response.completed",
1265 "response": {
1266 "id": "resp_1",
1267 "status": "completed",
1268 "usage": {"input_tokens": 300, "output_tokens": 9}
1269 }
1270 }),
1271 None,
1272 )
1273 .unwrap();
1274 let finished = parse_sse_events(&finished)
1275 .into_iter()
1276 .filter_map(|event| serde_json::from_str::<serde_json::Value>(&event.data).ok())
1277 .find(|value| {
1278 value.get("type").and_then(serde_json::Value::as_str) == Some("message_delta")
1279 })
1280 .unwrap();
1281 assert_eq!(finished.pointer("/usage/input_tokens"), Some(&json!(300)));
1282 assert_eq!(finished.pointer("/usage/output_tokens"), Some(&json!(9)));
1283 }
1284
1285 #[test]
1286 fn finishes_text_stream() {
1287 let out = render(vec![
1288 json!({
1289 "type": "response.output_item.added",
1290 "output_index": 0,
1291 "item": {"type": "message", "id": "msg_up"}
1292 }),
1293 json!({
1294 "type": "response.output_text.delta",
1295 "output_index": 0,
1296 "delta": "hello"
1297 }),
1298 json!({
1299 "type": "response.output_item.done",
1300 "output_index": 0,
1301 "item": {"type": "message"}
1302 }),
1303 json!({
1304 "type": "response.completed",
1305 "response": {"id": "resp_1", "status": "completed", "incomplete_details": null, "usage": {"input_tokens": 2, "output_tokens": 1}}
1306 }),
1307 ]);
1308 assert!(out.contains("content_block_stop"));
1309 assert!(out.contains("message_delta"));
1310 assert!(out.contains(r#""stop_reason":"end_turn""#));
1311 assert!(out.contains("message_stop"));
1312 }
1313
1314 #[test]
1315 fn terminal_only_completion_remains_non_semantic() {
1316 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1317 let out = translator
1318 .accept(
1319 &json!({
1320 "type": "response.completed",
1321 "response": {"id": "resp_1", "status": "completed", "incomplete_details": null, "usage": {}}
1322 }),
1323 None,
1324 )
1325 .unwrap();
1326 let out = String::from_utf8(out).unwrap();
1327 assert!(out.contains(r#""stop_reason":"end_turn""#));
1328 assert!(!out.contains(r#""stop_reason":"max_tokens""#));
1329 assert!(!translator.has_semantic_output());
1330 }
1331
1332 #[test]
1333 fn tool_thinking_and_web_search_events_are_semantic() {
1334 let mut tool = LiveStreamTranslator::new("msg_tool", "gpt-5.5");
1335 tool.accept(
1336 &json!({
1337 "type": "response.output_item.added",
1338 "output_index": 0,
1339 "item": {"type": "function_call", "call_id": "call_1", "name": "Read"}
1340 }),
1341 None,
1342 )
1343 .unwrap();
1344 assert!(tool.has_semantic_output());
1345
1346 let mut thinking = LiveStreamTranslator::new("msg_thinking", "gpt-5.5");
1347 thinking
1348 .accept(
1349 &json!({
1350 "type": "response.reasoning_summary_text.delta",
1351 "output_index": 0,
1352 "delta": "plan"
1353 }),
1354 None,
1355 )
1356 .unwrap();
1357 assert!(thinking.has_semantic_output());
1358
1359 let mut web_search = LiveStreamTranslator::new("msg_search", "gpt-5.5");
1360 web_search
1361 .accept(
1362 &json!({
1363 "type": "response.output_item.added",
1364 "output_index": 0,
1365 "item": {"type": "web_search_call", "id": "ws_1"}
1366 }),
1367 None,
1368 )
1369 .unwrap();
1370 assert!(!web_search.has_semantic_output());
1371 web_search
1372 .accept(
1373 &json!({
1374 "type": "response.output_item.done",
1375 "output_index": 0,
1376 "item": {
1377 "type": "web_search_call",
1378 "id": "ws_1",
1379 "action": {"query": "claude-code-proxy"}
1380 }
1381 }),
1382 None,
1383 )
1384 .unwrap();
1385 assert!(web_search.has_semantic_output());
1386 }
1387
1388 #[test]
1389 fn buffers_read_tool_args_until_done() {
1390 let out = render(vec![
1391 json!({
1392 "type": "response.output_item.added",
1393 "output_index": 0,
1394 "item": {"type": "function_call", "call_id": "call_1", "name": "Read"}
1395 }),
1396 json!({
1397 "type": "response.function_call_arguments.delta",
1398 "output_index": 0,
1399 "delta": "{\"file_path\":\"/tmp/a\",\"pages\":\"\"}"
1400 }),
1401 json!({
1402 "type": "response.output_item.done",
1403 "output_index": 0,
1404 "item": {
1405 "type": "function_call",
1406 "call_id": "call_1",
1407 "name": "Read",
1408 "arguments": "{\"file_path\":\"/tmp/a\",\"pages\":\"\"}"
1409 }
1410 }),
1411 json!({
1412 "type": "response.completed",
1413 "response": {"id": "resp_1", "usage": {}}
1414 }),
1415 ]);
1416 assert!(out.contains("tool_use"));
1417 assert!(out.contains("input_json_delta"));
1418 assert!(out.contains("/tmp/a"));
1419 assert!(!out.contains("pages"));
1420 }
1421
1422 #[test]
1423 fn repairs_whitespace_stalled_read_args_as_tool_use_finish() {
1424 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1425 let mut out = Vec::new();
1426 out.extend(
1427 translator
1428 .accept(
1429 &json!({
1430 "type": "response.output_item.added",
1431 "output_index": 0,
1432 "item": {"type":"function_call","call_id":"call_1","name":"Read"}
1433 }),
1434 None,
1435 )
1436 .unwrap(),
1437 );
1438 out.extend(
1439 translator
1440 .accept(
1441 &json!({
1442 "type": "response.function_call_arguments.delta",
1443 "output_index": 0,
1444 "delta": format!("{{\"file_path\":\"/tmp/a\",\"pages\":\"\"{}", " ".repeat(1024))
1445 }),
1446 None,
1447 )
1448 .unwrap(),
1449 );
1450 let rendered = String::from_utf8(out).unwrap();
1451 assert!(rendered.contains(r#""partial_json":"{\"file_path\":\"/tmp/a\"}""#));
1452 assert!(rendered.contains(r#""stop_reason":"tool_use""#));
1453 assert!(rendered.contains("message_stop"));
1454 assert!(translator.is_finished());
1455 }
1456
1457 #[test]
1458 fn finishes_after_closed_completed_tool_call() {
1459 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1460 let mut out = Vec::new();
1461 for event in [
1462 json!({
1463 "type": "response.output_item.added",
1464 "output_index": 0,
1465 "item": {"type":"function_call","call_id":"call_1","name":"WebSearch"}
1466 }),
1467 json!({
1468 "type": "response.function_call_arguments.done",
1469 "output_index": 0,
1470 "arguments": "{\"query\":\"claude-code-proxy github\"}"
1471 }),
1472 json!({
1473 "type": "response.output_item.done",
1474 "output_index": 0,
1475 "item": {
1476 "type":"function_call",
1477 "call_id":"call_1",
1478 "name":"WebSearch",
1479 "arguments":"{\"query\":\"claude-code-proxy github\"}"
1480 }
1481 }),
1482 ] {
1483 out.extend(translator.accept(&event, None).unwrap());
1484 }
1485 out.extend(translator.finish_after_closed_completed_tool_call(None));
1486 let rendered = String::from_utf8(out).unwrap();
1487 assert!(rendered.contains("content_block_start"));
1488 assert!(rendered.contains("input_json_delta"));
1489 assert!(rendered.contains(r#""stop_reason":"tool_use""#));
1490 assert!(rendered.contains("message_stop"));
1491 assert!(!rendered.contains("event: error"));
1492 }
1493
1494 #[test]
1495 fn emits_web_search_results_from_citations_before_deferred_text() {
1496 let out = render(vec![
1497 json!({
1498 "type": "response.output_item.added",
1499 "output_index": 0,
1500 "item": {"type": "web_search_call", "id": "ws_1"}
1501 }),
1502 json!({
1503 "type": "response.output_item.done",
1504 "output_index": 0,
1505 "item": {
1506 "type": "web_search_call",
1507 "id": "ws_1",
1508 "action": {"query": "grok reasoning effort"}
1509 }
1510 }),
1511 json!({
1512 "type": "response.output_item.added",
1513 "output_index": 1,
1514 "item": {"type": "message", "id": "msg_up"}
1515 }),
1516 json!({
1517 "type": "response.output_text.delta",
1518 "output_index": 1,
1519 "delta": "See the official docs."
1520 }),
1521 json!({
1522 "type": "response.output_text.annotation.added",
1523 "annotation": {
1524 "type": "url_citation",
1525 "title": "Reasoning",
1526 "url": "https://docs.x.ai/docs/guides/reasoning"
1527 }
1528 }),
1529 json!({
1530 "type": "response.output_item.done",
1531 "output_index": 1,
1532 "item": {"type": "message"}
1533 }),
1534 json!({
1535 "type": "response.completed",
1536 "response": {"status": "completed", "usage": {}}
1537 }),
1538 ]);
1539
1540 let tool = out.find("server_tool_use").unwrap();
1541 let result = out.find("web_search_tool_result").unwrap();
1542 let text = out.find("See the official docs.").unwrap();
1543 assert!(tool < result && result < text);
1544 assert!(out.contains("https://docs.x.ai/docs/guides/reasoning"));
1545 assert!(out.contains(r#""web_search_requests":1"#));
1546 }
1547
1548 #[test]
1549 fn rate_limit_event_returns_error() {
1550 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1551 let err = translator
1552 .accept(
1553 &json!({
1554 "type": "codex.rate_limits",
1555 "rate_limits": {"limit_reached": true}
1556 }),
1557 None,
1558 )
1559 .unwrap_err();
1560 assert_eq!(err, "rate limit reached");
1561 }
1562
1563 #[test]
1564 fn progress_events_start_message_and_emit_pings() {
1565 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1566 let first = String::from_utf8(
1567 translator
1568 .accept(&json!({"type": "response.created"}), None)
1569 .unwrap(),
1570 )
1571 .unwrap();
1572 assert_eq!(first.matches("event: message_start").count(), 1);
1573 assert_eq!(first.matches("event: ping").count(), 1);
1574
1575 let second = String::from_utf8(translator.ping_chunk(None)).unwrap();
1576 assert!(!second.contains("event: message_start"));
1577 assert_eq!(second.matches("event: ping").count(), 1);
1578 }
1579
1580 #[test]
1581 fn live_stream_emits_signature_delta_before_thinking_stop() {
1582 let out = render(vec![
1583 json!({
1584 "type":"response.output_item.added",
1585 "output_index":0,
1586 "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"}
1587 }),
1588 json!({
1589 "type":"response.reasoning_summary_text.delta",
1590 "output_index":0,
1591 "delta":"plan"
1592 }),
1593 json!({
1594 "type":"response.output_item.done",
1595 "output_index":0,
1596 "item":{"type":"reasoning","id":"rs_1"}
1597 }),
1598 json!({
1599 "type":"response.completed",
1600 "response":{"id":"resp_1","usage":{}}
1601 }),
1602 ]);
1603 let thinking_delta = out.find(r#""type":"thinking_delta""#).unwrap();
1604 let signature_delta = out.find(r#""type":"signature_delta""#).unwrap();
1605 let thinking_stop = out[signature_delta..]
1606 .find("event: content_block_stop")
1607 .map(|offset| signature_delta + offset)
1608 .unwrap();
1609 assert!(thinking_delta < signature_delta);
1610 assert!(signature_delta < thinking_stop);
1611 assert!(out.contains("ccp:codex:v1:"));
1612 }
1613
1614 #[test]
1615 fn signature_only_reasoning_is_semantic_output() {
1616 let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5");
1617 let mut out = Vec::new();
1618 for event in [
1619 json!({
1620 "type":"response.output_item.added",
1621 "output_index":0,
1622 "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"}
1623 }),
1624 json!({
1625 "type":"response.output_item.done",
1626 "output_index":0,
1627 "item":{"type":"reasoning","id":"rs_1"}
1628 }),
1629 ] {
1630 out.extend(translator.accept(&event, None).unwrap());
1631 }
1632
1633 let out = String::from_utf8(out).unwrap();
1634 assert!(out.contains(r#""type":"signature_delta""#));
1635 assert!(translator.has_semantic_output());
1636 }
1637}