1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
use super::{
common::{invalid, limit},
identity,
response_items::{Kind, Part},
responses_to_gemini::ResponsesToGeminiStream,
};
use crate::{transform::TransformError, wire::gemini as g};
use std::collections::BTreeMap;
impl ResponsesToGeminiStream {
pub(super) fn flush(
&mut self,
out: &mut Vec<g::GenerateContentResponseBody>,
) -> Result<(), TransformError> {
while let Some(mut item) = self.items.remove(&self.cursor) {
let advance = match &mut item.kind {
Kind::Excluded => item.done,
Kind::Image { value, emitted } => {
if item.done && !*emitted
&& value.status != crate::wire::openai::responses::input::ImageGenerationStatus::Completed {
// A nonterminal image is not an image chunk. Hold its
// disposition until the actual response terminal; the
// canonical converter distinguishes incomplete from a
// malformed completed response without fabricating bytes.
if self.terminal {
self.release(item.held);
item.held = 0;
*emitted = true;
}
} else if item.done && !*emitted {
let _model = self
.model
.as_deref()
.ok_or_else(|| invalid("missing native model"))?;
let part = super::super::images::to_gemini(
value,
self.limits.max_bytes as u64,
)?;
if self.jpeg_only {
let mime = part
.inline_data
.as_ref()
.map(|b| b.mime_type.as_str())
.or_else(|| {
part.file_data.as_ref().and_then(|f| f.mime_type.as_deref())
});
if mime != Some("image/jpeg") {
return Err(invalid(
"actual image MIME differs from requested JPEG output",
));
}
}
self.emit_part(part, out)?;
self.release(item.held);
item.held = 0;
*emitted = true;
}
*emitted
}
Kind::Message { parts, next } => {
self.flush_parts(parts, next, false, &mut item.held, out)?;
item.done && *next == parts.len() as i64
}
Kind::Reasoning {
parts,
next,
has_content,
summaries,
summary_next,
summary_emitted,
} => {
if *has_content {
self.flush_parts(parts, next, true, &mut item.held, out)?;
item.done && *next == parts.len() as i64
} else {
let before = out.len();
self.flush_parts(summaries, summary_next, true, &mut item.held, out)?;
*summary_emitted |= out.len() != before;
item.done && *summary_next == summaries.len() as i64
}
}
Kind::Function {
value,
ready,
emitted,
} => {
if *ready && !*emitted {
let _model = self
.model
.as_deref()
.ok_or_else(|| invalid("missing native model"))?;
let mut part = super::super::identity::function((**value).clone())?;
identity::call(
&mut part,
value,
self.cursor,
(&mut self.flow, &self.policy),
&mut self.used,
)?;
self.emit_part(part, out)?;
self.release(item.held);
item.held = 0;
*emitted = true;
}
*emitted
}
};
if advance {
self.release(item.held);
item.held = 0;
}
self.items.insert(self.cursor, item);
if !advance {
break;
}
self.cursor = self.cursor.checked_add(1).ok_or_else(limit)?;
}
Ok(())
}
fn flush_parts(
&mut self,
parts: &mut BTreeMap<i64, Part>,
next: &mut i64,
thought: bool,
held: &mut usize,
out: &mut Vec<g::GenerateContentResponseBody>,
) -> Result<(), TransformError> {
while let Some(part) = parts.get_mut(next) {
if !part.emitted || !part.pending.is_empty() {
let text = std::mem::take(&mut part.pending);
let bytes = text.len();
// The buffered pair omits unsigned empty reasoning. Empty
// message text remains a real native part; reasoning summaries
// and raw content both omit empty thought fragments.
if !thought || !text.is_empty() {
let mut value = g::Part::builder().text(text).build();
if thought {
value.thought = Some(true);
}
self.emit_part(value, out)?;
}
self.release(bytes);
*held -= bytes;
part.emitted = true;
}
if !part.done {
break;
}
*next = next.checked_add(1).ok_or_else(limit)?;
}
Ok(())
}
}