gproxy_protocol/transform/generate/chat_responses/stream/
responses_to_chat.rs

1mod content;
2mod reasoning;
3use super::common::{Budget, StreamEnd, StreamLimits, declared, invalid, unsupported};
4use crate::{
5    Dialect,
6    transform::generate::stream::{
7        chat::{ChatStreamCollector, ChatStreamLimits},
8        responses::{ResponsesStreamCollector, ResponsesStreamLimits},
9    },
10    transform::{
11        Converted, Report, TransformError,
12        identity::{IdentityFlow, IdentityRole, SourceIdentity, TargetIdPolicy},
13    },
14    wire::openai::{
15        chat::{response as c, stream as cs},
16        responses::{input as i, response as r, stream as rs},
17    },
18};
19use std::collections::BTreeMap;
20
21pub(super) struct TextPart {
22    pub pending: String,
23    pub refusal: bool,
24    pub done: bool,
25    pub emitted: bool,
26}
27
28pub(super) enum ItemKind {
29    Excluded,
30    Message {
31        parts: BTreeMap<i64, TextPart>,
32        cursor: i64,
33        done: bool,
34    },
35    Function {
36        call_id: String,
37        name: String,
38        tool_index: i64,
39        argument_bytes: usize,
40        sent: bool,
41    },
42    Reasoning(reasoning::ReasoningState),
43}
44
45pub(super) struct ItemState {
46    pub id: Option<String>,
47    pub kind: ItemKind,
48}
49
50/// Each push returns only newly emitted target chunks. Source item/part order is
51/// retained when concurrent Responses messages feed Chat's single content field.
52pub struct ResponsesToChatStream {
53    pub(super) source: Option<ResponsesStreamCollector>,
54    pub(super) target: Option<ChatStreamCollector>,
55    pub(super) flow: IdentityFlow,
56    pub(super) policy: TargetIdPolicy,
57    pub(super) budget: Budget,
58    pub(super) response_id: Option<String>,
59    pub(super) model: Option<String>,
60    pub(super) created: Option<i64>,
61    pub(super) items: BTreeMap<i64, ItemState>,
62    pub(super) content_cursor: i64,
63    pub(super) tools: usize,
64    reasoning_cursor: i64,
65    reasoning_parts: usize,
66    reasoning_text: String,
67    reasoning_details: usize,
68    pub(super) report: Report,
69    pub(super) expected: Option<c::GenerateContentResponseBody>,
70    pub(super) terminal: bool,
71    pub(super) failed: bool,
72}
73
74impl ResponsesToChatStream {
75    pub fn new(flow: IdentityFlow, limits: StreamLimits) -> Self {
76        Self::new_with_policy(
77            flow,
78            limits,
79            TargetIdPolicy::new(crate::Dialect::OpenAiChat),
80        )
81        .expect("default policy matches target dialect")
82    }
83    pub fn new_with_policy(
84        flow: IdentityFlow,
85        limits: StreamLimits,
86        policy: TargetIdPolicy,
87    ) -> Result<Self, TransformError> {
88        if policy.dialect != crate::Dialect::OpenAiChat {
89            return Err(TransformError::shape(
90                "stream.policy",
91                "target policy dialect mismatch",
92            ));
93        }
94
95        // These IDs were already allocated under the selected policy. The
96        // native target collector must preserve them while checking syntax;
97        // preserve_source_ids=false must not trigger a second allocation.
98        let mut collector_policy = policy.clone();
99        collector_policy.preserve_source_ids = true;
100        Ok(Self {
101            source: Some(ResponsesStreamCollector::new(ResponsesStreamLimits {
102                max_bytes: limits.max_bytes,
103
104                max_text_bytes: limits.max_bytes,
105                max_json_bytes: limits.max_bytes,
106            })),
107            target: Some(ChatStreamCollector::with_limits(
108                IdentityFlow::new(flow.namespace()),
109                collector_policy,
110                ChatStreamLimits {
111                    max_bytes: limits.max_bytes,
112                },
113            )),
114            flow,
115            policy,
116            budget: Budget::new(limits),
117            response_id: None,
118            model: None,
119            created: None,
120            items: BTreeMap::new(),
121            content_cursor: 0,
122            tools: 0,
123            reasoning_cursor: 0,
124            reasoning_parts: 0,
125            reasoning_text: String::new(),
126            reasoning_details: 0,
127            report: Report::default(),
128            expected: None,
129            terminal: false,
130            failed: false,
131        })
132    }
133    pub(crate) fn reserve_external_ids(
134        &mut self,
135        role: crate::transform::identity::IdentityRole,
136        ids: &std::collections::BTreeSet<String>,
137    ) -> Result<(), TransformError> {
138        self.flow.reserve_external_ids(role, ids).map_err(|e| {
139            TransformError::new(
140                crate::transform::TransformErrorKind::Conflict,
141                "fanout.ids",
142                e.to_string(),
143            )
144        })
145    }
146    pub fn identities(&self) -> &IdentityFlow {
147        &self.flow
148    }
149    pub fn push(
150        &mut self,
151        event: rs::StreamEvent,
152    ) -> Result<Converted<Vec<cs::ChatCompletionChunk>>, TransformError> {
153        if matches!(event, rs::StreamEvent::Unknown) {
154            return Ok(Converted {
155                value: Vec::new(),
156                report: Report::default(),
157            });
158        }
159        if self.failed || self.terminal {
160            self.failed = true;
161            return Err(invalid("event after terminal or failed stream"));
162        }
163        let result = self.push_inner(declared(event));
164        if result.is_err() {
165            self.failed = true;
166        }
167        result
168    }
169    fn push_inner(
170        &mut self,
171        event: rs::StreamEvent,
172    ) -> Result<Converted<Vec<cs::ChatCompletionChunk>>, TransformError> {
173        let event = crate::transform::generate::multi_agent::attribute_event(event)?;
174        self.budget.input(&event)?;
175        self.source
176            .as_mut()
177            .ok_or_else(|| invalid("source consumed"))?
178            .push(event.clone())?;
179        let mut out = Vec::new();
180        match event {
181            rs::StreamEvent::Keepalive
182            | rs::StreamEvent::Metadata(_)
183            | rs::StreamEvent::Unknown => {}
184            rs::StreamEvent::Created(v) => {
185                let id = self
186                    .flow
187                    .resolve_or_allocate(
188                        IdentityRole::Response,
189                        SourceIdentity::new(Dialect::OpenAi, Some(v.response.id), 0),
190                        &self.policy,
191                    )
192                    .map_err(|e| {
193                        TransformError::invalid_result("identity.response", e.to_string())
194                    })?
195                    .emitted_id;
196                self.response_id = Some(id);
197                self.model = Some(v.response.model);
198                self.created = Some(v.response.created_at);
199                self.emit_delta(
200                    cs::Delta::builder().role(cs::DeltaRole::Assistant).build(),
201                    &mut out,
202                )?;
203                for (index, item) in v.response.output.into_iter().enumerate() {
204                    self.add_item(index as i64, item, &mut out)?;
205                }
206            }
207            rs::StreamEvent::OutputItemAdded(v) => {
208                self.add_item(v.output_index, v.item, &mut out)?
209            }
210            rs::StreamEvent::ContentPartAdded(v) => {
211                self.add_part(v.output_index, v.content_index, v.part)?
212            }
213            rs::StreamEvent::OutputTextDelta(v) => {
214                self.append_text(v.output_index, v.content_index, &v.item_id, v.delta, false)?
215            }
216            rs::StreamEvent::RefusalDelta(v) => {
217                self.append_text(v.output_index, v.content_index, &v.item_id, v.delta, true)?
218            }
219            rs::StreamEvent::ContentPartDone(v) => {
220                if let Some(ItemState {
221                    kind: ItemKind::Message { parts, .. },
222                    ..
223                }) = self.items.get_mut(&v.output_index)
224                {
225                    parts
226                        .get_mut(&v.content_index)
227                        .ok_or_else(|| invalid("part done without part"))?
228                        .done = true;
229                }
230            }
231            rs::StreamEvent::OutputItemDone(v) => {
232                self.bind_item(v.output_index, item_id(&v.item))?;
233                if let r::ResponseOutputItem::Reasoning(item) = &v.item
234                    && let Some(state) = self.reasoning_mut(v.output_index, &item.id)?
235                {
236                    for part in state.summary.values_mut() {
237                        part.done = true;
238                    }
239                    state.final_item = Some(item.clone());
240                }
241                if let Some(ItemState {
242                    kind: ItemKind::Message { parts, done, .. },
243                    ..
244                }) = self.items.get_mut(&v.output_index)
245                {
246                    *done = true;
247                    for p in parts.values_mut() {
248                        p.done = true;
249                    }
250                }
251            }
252            rs::StreamEvent::FunctionCallArgumentsDelta(v) => {
253                self.bind_item(v.output_index, Some(&v.item_id))?;
254                self.emit_tool(v.output_index, v.delta, &mut out)?;
255            }
256            rs::StreamEvent::FunctionCallArgumentsDone(v) => {
257                self.bind_item(v.output_index, Some(&v.item_id))?;
258                if let Some(ItemState {
259                    kind: ItemKind::Function { argument_bytes, .. },
260                    ..
261                }) = self.items.get(&v.output_index)
262                {
263                    let suffix = v.arguments[*argument_bytes..].to_owned();
264                    if !suffix.is_empty() {
265                        self.emit_tool(v.output_index, suffix, &mut out)?;
266                    }
267                }
268            }
269            rs::StreamEvent::Completed(_) | rs::StreamEvent::Incomplete(_) => {
270                self.finish_source(&mut out)?;
271            }
272            rs::StreamEvent::Queued(_)
273            | rs::StreamEvent::InProgress(_)
274            | rs::StreamEvent::OutputTextDone(_)
275            | rs::StreamEvent::RefusalDone(_) => {}
276            rs::StreamEvent::OutputTextAnnotationAdded(_) => self.report.omitted(
277                "response.annotations",
278                "Chat stream Delta has no annotation field",
279            ),
280            rs::StreamEvent::ReasoningSummaryPartAdded(v) => {
281                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
282                    state.summary.insert(
283                        v.summary_index,
284                        TextPart {
285                            pending: v.part.text,
286                            refusal: false,
287                            done: false,
288                            emitted: false,
289                        },
290                    );
291                }
292            }
293            rs::StreamEvent::ReasoningSummaryTextDelta(v) => {
294                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
295                    state
296                        .summary
297                        .get_mut(&v.summary_index)
298                        .ok_or_else(|| invalid("reasoning delta without summary part"))?
299                        .pending
300                        .push_str(&v.delta);
301                }
302            }
303            rs::StreamEvent::ReasoningSummaryPartDone(v) => {
304                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
305                    state
306                        .summary
307                        .get_mut(&v.summary_index)
308                        .ok_or_else(|| invalid("reasoning done without summary part"))?
309                        .done = true;
310                }
311            }
312            // Raw content follows all summaries in the canonical Chat view.
313            // Until item completion, a later summary could still precede it.
314            rs::StreamEvent::ReasoningTextDelta(_)
315            | rs::StreamEvent::ReasoningTextDone(_)
316            | rs::StreamEvent::ReasoningSummaryTextDone(_) => {}
317            rs::StreamEvent::Failed(_) | rs::StreamEvent::Error(_) => {
318                return Err(invalid("Responses stream failed"));
319            }
320            rs::StreamEvent::CustomToolInputDelta(_) | rs::StreamEvent::CustomToolInputDone(_) => {
321                return Err(unsupported(
322                    "custom_tool",
323                    "Chat streaming tool schema supports only function calls; buffered Chat has a separate custom-tool shape",
324                ));
325            }
326            rs::StreamEvent::AudioDelta(_)
327            | rs::StreamEvent::AudioDone(_)
328            | rs::StreamEvent::AudioTranscriptDelta(_)
329            | rs::StreamEvent::AudioTranscriptDone(_) => {
330                return Err(unsupported(
331                    "audio",
332                    "this native stream wire has no canonical output audio item",
333                ));
334            }
335            _ => {
336                return Err(unsupported(
337                    "response.execution",
338                    "native execution events require an invocation adapter",
339                ));
340            }
341        }
342        if !self.terminal {
343            self.flush_text(&mut out)?;
344            self.flush_reasoning(&mut out)?;
345        }
346        Ok(Converted {
347            value: out,
348            report: std::mem::take(&mut self.report),
349        })
350    }
351    fn finish_source(
352        &mut self,
353        out: &mut Vec<cs::ChatCompletionChunk>,
354    ) -> Result<(), TransformError> {
355        let actual = self
356            .source
357            .take()
358            .ok_or_else(|| invalid("source consumed"))?
359            .finish()?;
360        self.report.diagnostics.extend(actual.report.diagnostics);
361        let mapped =
362            super::super::responses_to_chat_response(actual.value, &mut self.flow, &self.policy)?;
363        self.report.diagnostics.extend(mapped.report.diagnostics);
364        for item in self.items.values_mut() {
365            if let ItemKind::Message { parts, done, .. } = &mut item.kind {
366                *done = true;
367                for p in parts.values_mut() {
368                    p.done = true;
369                }
370            }
371        }
372        self.flush_text(out)?;
373        let mut expected = mapped.value;
374        let message = &expected.choices[0].message;
375        if message.reasoning_details.is_some() || message.reasoning_content.is_some() {
376            let mut delta = cs::Delta::builder().build();
377            if let Some(Some(details)) = &message.reasoning_details {
378                let remaining = details
379                    .get(self.reasoning_details..)
380                    .ok_or_else(|| invalid("reasoning details exceed final result"))?;
381                if !remaining.is_empty() {
382                    delta.reasoning_details = Some(Some(remaining.to_vec()));
383                }
384            }
385            if let Some(Some(text)) = &message.reasoning_content {
386                let suffix = text
387                    .strip_prefix(&self.reasoning_text)
388                    .ok_or_else(|| invalid("streamed reasoning differs from final result"))?;
389                if !suffix.is_empty() || self.reasoning_parts == 0 {
390                    delta.reasoning_content = Some(Some(suffix.to_owned()));
391                }
392            }
393            if delta.reasoning_details.is_some() || delta.reasoning_content.is_some() {
394                self.emit_delta(delta, out)?;
395            }
396        }
397        if expected.choices[0].message.annotations.take().is_some() {
398            self.report.omitted(
399                "message.annotations",
400                "Chat streaming Delta has no annotation field",
401            );
402        }
403        let logs = expected.choices[0]
404            .logprobs
405            .clone()
406            .map(crate::transform::generate::stream::chat::logs::synthesize)
407            .map(crate::transform::optional)
408            .transpose()?
409            .flatten();
410        let mut choice = cs::StreamChoice::builder(0, cs::Delta::builder().build())
411            .finish_reason(expected.choices[0].finish_reason)
412            .build();
413        choice.logprobs = logs.map(Some);
414        let mut terminal = self.chunk(vec![choice])?;
415        terminal.service_tier = expected.service_tier;
416        terminal.moderation = expected.moderation.clone();
417        self.emit(terminal, out)?;
418        if let Some(usage) = expected.usage.clone() {
419            let usage = crate::transform::generate::stream::chat::usage::synthesize(usage)?;
420            let mut chunk = self.chunk(Vec::new())?;
421            chunk.usage = Some(Some(usage));
422            self.emit(chunk, out)?;
423        }
424        self.expected = Some(expected);
425        self.terminal = true;
426        Ok(())
427    }
428    fn emit_delta(
429        &mut self,
430        delta: cs::Delta,
431        out: &mut Vec<cs::ChatCompletionChunk>,
432    ) -> Result<(), TransformError> {
433        let chunk = self.chunk(vec![cs::StreamChoice::builder(0, delta).build()])?;
434        self.emit(chunk, out)
435    }
436    fn chunk(
437        &self,
438        choices: Vec<cs::StreamChoice>,
439    ) -> Result<cs::ChatCompletionChunk, TransformError> {
440        Ok(cs::ChatCompletionChunk::builder(
441            self.response_id
442                .clone()
443                .ok_or_else(|| TransformError::missing_metadata("response.id"))?,
444            choices,
445            self.created
446                .ok_or_else(|| TransformError::missing_metadata("response.created_at"))?,
447            self.model
448                .clone()
449                .ok_or_else(|| TransformError::missing_metadata("response.model"))?,
450            cs::ChunkObject::ChatCompletionChunk,
451        )
452        .build())
453    }
454    fn emit(
455        &mut self,
456        chunk: cs::ChatCompletionChunk,
457        out: &mut Vec<cs::ChatCompletionChunk>,
458    ) -> Result<(), TransformError> {
459        self.budget.output(&chunk)?;
460        self.target
461            .as_mut()
462            .ok_or_else(|| invalid("target consumed"))?
463            .push(chunk.clone())?;
464        out.push(chunk);
465        Ok(())
466    }
467    pub fn finish(mut self) -> Result<StreamEnd<cs::ChatCompletionChunk>, TransformError> {
468        if self.failed || !self.terminal {
469            return Err(invalid("Responses EOF without terminal"));
470        }
471        let mut target = self
472            .target
473            .take()
474            .ok_or_else(|| invalid("target consumed"))?;
475        target.push_done()?;
476        let actual = target.finish()?;
477        if Some(&actual.value) != self.expected.as_ref() {
478            return Err(invalid(
479                "emitted Chat content/identity/metadata differs from native terminal response",
480            ));
481        }
482        self.report.diagnostics.extend(actual.report.diagnostics);
483        Ok(StreamEnd {
484            chunks: Vec::new(),
485            identities: self.flow,
486            report: self.report,
487        })
488    }
489}
490
491fn item_id(item: &r::ResponseOutputItem) -> Option<&str> {
492    match item {
493        r::ResponseOutputItem::Message(v) => Some(&v.id),
494        r::ResponseOutputItem::FunctionCall(v) => v.id.as_deref(),
495        r::ResponseOutputItem::Reasoning(v) => Some(&v.id),
496        _ => None,
497    }
498}
gproxy_protocol/transform/generate/chat_responses/stream/responses_to_chat.rs

gproxy_protocol/transform/generate/chat_responses/stream/
responses_to_chat.rs
