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

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

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