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 self.failed || self.terminal {
154            self.failed = true;
155            return Err(invalid("event after terminal or failed stream"));
156        }
157        let result = self.push_inner(declared(event));
158        if result.is_err() {
159            self.failed = true;
160        }
161        result
162    }
163    fn push_inner(
164        &mut self,
165        event: rs::StreamEvent,
166    ) -> Result<Converted<Vec<cs::ChatCompletionChunk>>, TransformError> {
167        let event = crate::transform::generate::multi_agent::attribute_event(event)?;
168        self.budget.input(&event)?;
169        self.source
170            .as_mut()
171            .ok_or_else(|| invalid("source consumed"))?
172            .push(event.clone())?;
173        let mut out = Vec::new();
174        match event {
175            rs::StreamEvent::Keepalive => {}
176            rs::StreamEvent::Created(v) => {
177                let id = self
178                    .flow
179                    .resolve_or_allocate(
180                        IdentityRole::Response,
181                        SourceIdentity::new(Dialect::OpenAi, Some(v.response.id), 0),
182                        &self.policy,
183                    )
184                    .map_err(|e| {
185                        TransformError::invalid_result("identity.response", e.to_string())
186                    })?
187                    .emitted_id;
188                self.response_id = Some(id);
189                self.model = Some(v.response.model);
190                self.created = Some(v.response.created_at);
191                self.emit_delta(
192                    cs::Delta::builder().role(cs::DeltaRole::Assistant).build(),
193                    &mut out,
194                )?;
195                for (index, item) in v.response.output.into_iter().enumerate() {
196                    self.add_item(index as i64, item, &mut out)?;
197                }
198            }
199            rs::StreamEvent::OutputItemAdded(v) => {
200                self.add_item(v.output_index, v.item, &mut out)?
201            }
202            rs::StreamEvent::ContentPartAdded(v) => {
203                self.add_part(v.output_index, v.content_index, v.part)?
204            }
205            rs::StreamEvent::OutputTextDelta(v) => {
206                self.append_text(v.output_index, v.content_index, &v.item_id, v.delta, false)?
207            }
208            rs::StreamEvent::RefusalDelta(v) => {
209                self.append_text(v.output_index, v.content_index, &v.item_id, v.delta, true)?
210            }
211            rs::StreamEvent::ContentPartDone(v) => {
212                if let Some(ItemState {
213                    kind: ItemKind::Message { parts, .. },
214                    ..
215                }) = self.items.get_mut(&v.output_index)
216                {
217                    parts
218                        .get_mut(&v.content_index)
219                        .ok_or_else(|| invalid("part done without part"))?
220                        .done = true;
221                }
222            }
223            rs::StreamEvent::OutputItemDone(v) => {
224                self.bind_item(v.output_index, item_id(&v.item))?;
225                if let r::ResponseOutputItem::Reasoning(item) = &v.item
226                    && let Some(state) = self.reasoning_mut(v.output_index, &item.id)?
227                {
228                    for part in state.summary.values_mut() {
229                        part.done = true;
230                    }
231                    state.final_item = Some(item.clone());
232                }
233                if let Some(ItemState {
234                    kind: ItemKind::Message { parts, done, .. },
235                    ..
236                }) = self.items.get_mut(&v.output_index)
237                {
238                    *done = true;
239                    for p in parts.values_mut() {
240                        p.done = true;
241                    }
242                }
243            }
244            rs::StreamEvent::FunctionCallArgumentsDelta(v) => {
245                self.bind_item(v.output_index, Some(&v.item_id))?;
246                self.emit_tool(v.output_index, v.delta, &mut out)?;
247            }
248            rs::StreamEvent::FunctionCallArgumentsDone(v) => {
249                self.bind_item(v.output_index, Some(&v.item_id))?;
250                if let Some(ItemState {
251                    kind: ItemKind::Function { argument_bytes, .. },
252                    ..
253                }) = self.items.get(&v.output_index)
254                {
255                    let suffix = v.arguments[*argument_bytes..].to_owned();
256                    if !suffix.is_empty() {
257                        self.emit_tool(v.output_index, suffix, &mut out)?;
258                    }
259                }
260            }
261            rs::StreamEvent::Completed(_) | rs::StreamEvent::Incomplete(_) => {
262                self.finish_source(&mut out)?;
263            }
264            rs::StreamEvent::Queued(_)
265            | rs::StreamEvent::InProgress(_)
266            | rs::StreamEvent::OutputTextDone(_)
267            | rs::StreamEvent::RefusalDone(_) => {}
268            rs::StreamEvent::OutputTextAnnotationAdded(_) => self.report.omitted(
269                "response.annotations",
270                "Chat stream Delta has no annotation field",
271            ),
272            rs::StreamEvent::ReasoningSummaryPartAdded(v) => {
273                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
274                    state.summary.insert(
275                        v.summary_index,
276                        TextPart {
277                            pending: v.part.text,
278                            refusal: false,
279                            done: false,
280                            emitted: false,
281                        },
282                    );
283                }
284            }
285            rs::StreamEvent::ReasoningSummaryTextDelta(v) => {
286                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
287                    state
288                        .summary
289                        .get_mut(&v.summary_index)
290                        .ok_or_else(|| invalid("reasoning delta without summary part"))?
291                        .pending
292                        .push_str(&v.delta);
293                }
294            }
295            rs::StreamEvent::ReasoningSummaryPartDone(v) => {
296                if let Some(state) = self.reasoning_mut(v.output_index, &v.item_id)? {
297                    state
298                        .summary
299                        .get_mut(&v.summary_index)
300                        .ok_or_else(|| invalid("reasoning done without summary part"))?
301                        .done = true;
302                }
303            }
304            // Raw content follows all summaries in the canonical Chat view.
305            // Until item completion, a later summary could still precede it.
306            rs::StreamEvent::ReasoningTextDelta(_)
307            | rs::StreamEvent::ReasoningTextDone(_)
308            | rs::StreamEvent::ReasoningSummaryTextDone(_) => {}
309            rs::StreamEvent::Failed(_) | rs::StreamEvent::Error(_) => {
310                return Err(invalid("Responses stream failed"));
311            }
312            rs::StreamEvent::CustomToolInputDelta(_) | rs::StreamEvent::CustomToolInputDone(_) => {
313                return Err(unsupported(
314                    "custom_tool",
315                    "Chat streaming tool schema supports only function calls; buffered Chat has a separate custom-tool shape",
316                ));
317            }
318            rs::StreamEvent::AudioDelta(_)
319            | rs::StreamEvent::AudioDone(_)
320            | rs::StreamEvent::AudioTranscriptDelta(_)
321            | rs::StreamEvent::AudioTranscriptDone(_) => {
322                return Err(unsupported(
323                    "audio",
324                    "this native stream wire has no canonical output audio item",
325                ));
326            }
327            _ => {
328                return Err(unsupported(
329                    "response.execution",
330                    "native execution events require an invocation adapter",
331                ));
332            }
333        }
334        if !self.terminal {
335            self.flush_text(&mut out)?;
336            self.flush_reasoning(&mut out)?;
337        }
338        Ok(Converted {
339            value: out,
340            report: std::mem::take(&mut self.report),
341        })
342    }
343    fn finish_source(
344        &mut self,
345        out: &mut Vec<cs::ChatCompletionChunk>,
346    ) -> Result<(), TransformError> {
347        let actual = self
348            .source
349            .take()
350            .ok_or_else(|| invalid("source consumed"))?
351            .finish()?;
352        self.report.diagnostics.extend(actual.report.diagnostics);
353        let mapped =
354            super::super::responses_to_chat_response(actual.value, &mut self.flow, &self.policy)?;
355        self.report.diagnostics.extend(mapped.report.diagnostics);
356        for item in self.items.values_mut() {
357            if let ItemKind::Message { parts, done, .. } = &mut item.kind {
358                *done = true;
359                for p in parts.values_mut() {
360                    p.done = true;
361                }
362            }
363        }
364        self.flush_text(out)?;
365        let mut expected = mapped.value;
366        let message = &expected.choices[0].message;
367        if message.reasoning_details.is_some() || message.reasoning_content.is_some() {
368            let mut delta = cs::Delta::builder().build();
369            if let Some(Some(details)) = &message.reasoning_details {
370                let remaining = details
371                    .get(self.reasoning_details..)
372                    .ok_or_else(|| invalid("reasoning details exceed final result"))?;
373                if !remaining.is_empty() {
374                    delta.reasoning_details = Some(Some(remaining.to_vec()));
375                }
376            }
377            if let Some(Some(text)) = &message.reasoning_content {
378                let suffix = text
379                    .strip_prefix(&self.reasoning_text)
380                    .ok_or_else(|| invalid("streamed reasoning differs from final result"))?;
381                if !suffix.is_empty() || self.reasoning_parts == 0 {
382                    delta.reasoning_content = Some(Some(suffix.to_owned()));
383                }
384            }
385            if delta.reasoning_details.is_some() || delta.reasoning_content.is_some() {
386                self.emit_delta(delta, out)?;
387            }
388        }
389        if expected.choices[0].message.annotations.take().is_some() {
390            self.report.omitted(
391                "message.annotations",
392                "Chat streaming Delta has no annotation field",
393            );
394        }
395        let logs = expected.choices[0]
396            .logprobs
397            .clone()
398            .map(crate::transform::generate::stream::chat::logs::synthesize)
399            .map(crate::transform::optional)
400            .transpose()?
401            .flatten();
402        let mut choice = cs::StreamChoice::builder(0, cs::Delta::builder().build())
403            .finish_reason(expected.choices[0].finish_reason)
404            .build();
405        choice.logprobs = logs.map(Some);
406        let mut terminal = self.chunk(vec![choice])?;
407        terminal.service_tier = expected.service_tier;
408        terminal.moderation = expected.moderation.clone();
409        self.emit(terminal, out)?;
410        if let Some(usage) = expected.usage.clone() {
411            let usage = crate::transform::generate::stream::chat::usage::synthesize(usage)?;
412            let mut chunk = self.chunk(Vec::new())?;
413            chunk.usage = Some(Some(usage));
414            self.emit(chunk, out)?;
415        }
416        self.expected = Some(expected);
417        self.terminal = true;
418        Ok(())
419    }
420    fn emit_delta(
421        &mut self,
422        delta: cs::Delta,
423        out: &mut Vec<cs::ChatCompletionChunk>,
424    ) -> Result<(), TransformError> {
425        let chunk = self.chunk(vec![cs::StreamChoice::builder(0, delta).build()])?;
426        self.emit(chunk, out)
427    }
428    fn chunk(
429        &self,
430        choices: Vec<cs::StreamChoice>,
431    ) -> Result<cs::ChatCompletionChunk, TransformError> {
432        Ok(cs::ChatCompletionChunk::builder(
433            self.response_id
434                .clone()
435                .ok_or_else(|| TransformError::missing_metadata("response.id"))?,
436            choices,
437            self.created
438                .ok_or_else(|| TransformError::missing_metadata("response.created_at"))?,
439            self.model
440                .clone()
441                .ok_or_else(|| TransformError::missing_metadata("response.model"))?,
442            cs::ChunkObject::ChatCompletionChunk,
443        )
444        .build())
445    }
446    fn emit(
447        &mut self,
448        chunk: cs::ChatCompletionChunk,
449        out: &mut Vec<cs::ChatCompletionChunk>,
450    ) -> Result<(), TransformError> {
451        self.budget.output(&chunk)?;
452        self.target
453            .as_mut()
454            .ok_or_else(|| invalid("target consumed"))?
455            .push(chunk.clone())?;
456        out.push(chunk);
457        Ok(())
458    }
459    pub fn finish(mut self) -> Result<StreamEnd<cs::ChatCompletionChunk>, TransformError> {
460        if self.failed || !self.terminal {
461            return Err(invalid("Responses EOF without terminal"));
462        }
463        let mut target = self
464            .target
465            .take()
466            .ok_or_else(|| invalid("target consumed"))?;
467        target.push_done()?;
468        let actual = target.finish()?;
469        if Some(&actual.value) != self.expected.as_ref() {
470            return Err(invalid(
471                "emitted Chat content/identity/metadata differs from native terminal response",
472            ));
473        }
474        self.report.diagnostics.extend(actual.report.diagnostics);
475        Ok(StreamEnd {
476            chunks: Vec::new(),
477            identities: self.flow,
478            report: self.report,
479        })
480    }
481}
482
483fn item_id(item: &r::ResponseOutputItem) -> Option<&str> {
484    match item {
485        r::ResponseOutputItem::Message(v) => Some(&v.id),
486        r::ResponseOutputItem::FunctionCall(v) => v.id.as_deref(),
487        r::ResponseOutputItem::Reasoning(v) => Some(&v.id),
488        _ => None,
489    }
490}
gproxy_protocol/transform/generate/chat_responses/stream/responses_to_chat.rs

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