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