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

1use crate::transform::generate::reasoning_details as rd;
2mod blocks;
3use super::common::{Budget, StreamEnd, StreamLimits, declared, invalid, limit};
4use crate::transform::generate::stream::{
5    chat::{ChatStreamCollector, ChatStreamLimits},
6    responses::ResponsesStreamCollector,
7};
8use crate::{
9    Dialect,
10    transform::{
11        Converted, Report, TransformError,
12        identity::{IdentityFlow, IdentityRole, OutputItemKind, 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
21/// Original Responses request and effective controls retained during request
22/// preparation. A stream cannot invent these fields from a Chat chunk.
23pub struct ChatToResponsesContext {
24    pub response: super::super::response::ResponsesResponseContext,
25}
26
27impl From<super::super::response::ResponsesResponseContext> for ChatToResponsesContext {
28    fn from(response: super::super::response::ResponsesResponseContext) -> Self {
29        Self { response }
30    }
31}
32
33struct MessageState {
34    id: String,
35    index: i64,
36    text: String,
37    refusal: String,
38    text_part: bool,
39    refusal_part: bool,
40    logs: Option<c::Logprobs>,
41}
42
43struct ToolState {
44    source_id: Option<String>,
45    name: String,
46    arguments: String,
47}
48
49pub struct ChatToResponsesStream {
50    source: Option<ChatStreamCollector>,
51    target: Option<ResponsesStreamCollector>,
52    context: Option<super::super::response::ResponsesResponseContext>,
53    flow: IdentityFlow,
54    target_policy: TargetIdPolicy,
55    limits: StreamLimits,
56    budget: Budget,
57    sequence: i64,
58    response_id: Option<String>,
59    model: Option<String>,
60    created: Option<i64>,
61    base: Option<r::GenerateContentResponseBody>,
62    message: Option<MessageState>,
63    reasoning_text: String,
64    reasoning_details: Vec<crate::wire::openai::chat::ReasoningDetail>,
65    tools: BTreeMap<i64, ToolState>,
66    client_tools: super::super::client_tools::Bindings,
67    next_output: i64,
68    finish: Option<c::FinishReason>,
69    events: Vec<rs::StreamEvent>,
70    pending: Vec<cs::ChatCompletionChunk>,
71    report: Report,
72    output: BTreeMap<i64, r::ResponseOutputItem>,
73    usage_facts: super::super::response::ChatUsageSupplement,
74    failed: bool,
75    done: bool,
76}
77
78impl ChatToResponsesStream {
79    pub fn new(
80        context: impl Into<ChatToResponsesContext>,
81        flow: IdentityFlow,
82        limits: StreamLimits,
83    ) -> Self {
84        Self::new_with_policy(
85            context,
86            flow,
87            limits,
88            TargetIdPolicy::new(crate::Dialect::OpenAi),
89        )
90        .expect("default policy matches target dialect")
91    }
92    pub fn new_with_policy(
93        context: impl Into<ChatToResponsesContext>,
94        flow: IdentityFlow,
95        limits: StreamLimits,
96        policy: TargetIdPolicy,
97    ) -> Result<Self, TransformError> {
98        if policy.dialect != crate::Dialect::OpenAi {
99            return Err(TransformError::shape(
100                "stream.policy",
101                "target policy dialect mismatch",
102            ));
103        }
104
105        let mut context = context.into();
106        let client_tools = super::super::client_tools::Bindings::new(&context.response.request)?;
107        context.response.request = declared(context.response.request);
108        context.response.effective_tool_choice = declared(context.response.effective_tool_choice);
109        context.response.effective_prompt_cache_options =
110            declared(context.response.effective_prompt_cache_options);
111        let source = ChatStreamCollector::with_limits(
112            IdentityFlow::new(flow.namespace()),
113            TargetIdPolicy::new(Dialect::OpenAiChat),
114            ChatStreamLimits {
115                max_bytes: limits.max_bytes,
116            },
117        );
118        let usage_facts = context.response.usage;
119        Ok(Self {
120            source: Some(source),
121            target: Some(ResponsesStreamCollector::new(
122                crate::transform::generate::stream::responses::ResponsesStreamLimits {
123                    max_bytes: limits.max_bytes,
124
125                    max_text_bytes: limits.max_bytes,
126                    max_json_bytes: limits.max_bytes,
127                },
128            )),
129            context: Some(context.response),
130            flow,
131            target_policy: policy,
132            limits,
133            budget: Budget::new(limits),
134            sequence: 0,
135            response_id: None,
136            model: None,
137            created: None,
138            base: None,
139            message: None,
140            reasoning_text: String::new(),
141            reasoning_details: Vec::new(),
142            tools: BTreeMap::new(),
143            client_tools,
144            next_output: 0,
145            finish: None,
146            events: Vec::new(),
147            pending: Vec::new(),
148            report: Report::default(),
149            output: BTreeMap::new(),
150            usage_facts,
151            failed: false,
152            done: false,
153        })
154    }
155
156    pub fn identities(&self) -> &IdentityFlow {
157        &self.flow
158    }
159    pub fn push(
160        &mut self,
161        chunk: cs::ChatCompletionChunk,
162    ) -> Result<Converted<Vec<rs::StreamEvent>>, TransformError> {
163        if self.failed || self.done {
164            self.failed = true;
165            return Err(invalid("event after DONE or failed stream"));
166        }
167        let result = self.push_inner(declared(chunk));
168        if result.is_err() {
169            self.failed = true;
170        }
171        result
172    }
173
174    fn push_inner(
175        &mut self,
176        chunk: cs::ChatCompletionChunk,
177    ) -> Result<Converted<Vec<rs::StreamEvent>>, TransformError> {
178        self.budget.input(&chunk)?;
179        self.source
180            .as_mut()
181            .ok_or_else(|| invalid("source consumed"))?
182            .push(chunk.clone())?;
183        if chunk.created < 0 {
184            return Err(TransformError::missing_metadata("chat.created/model"));
185        }
186        if self.created.is_some_and(|v| v != chunk.created) {
187            return Err(invalid("created changed during stream"));
188        }
189        self.created = Some(chunk.created);
190        if !chunk.id.is_empty() {
191            self.response_id = Some(chunk.id.clone());
192        }
193        if !chunk.model.is_empty() {
194            self.model = Some(chunk.model.clone());
195        }
196        // Bind a late upstream response ID without replacing the emitted alias.
197        if self.base.is_some() {
198            self.flow
199                .resolve_or_allocate(
200                    IdentityRole::Response,
201                    SourceIdentity::new(Dialect::OpenAiChat, self.response_id.clone(), 0),
202                    &self.target_policy,
203                )
204                .map_err(|e| invalid_owned(e.to_string()))?;
205        }
206        if self.model.is_none() {
207            self.pending.push(chunk);
208            return Ok(Converted {
209                value: Vec::new(),
210                report: std::mem::take(&mut self.report),
211            });
212        }
213        self.start_if_needed()?;
214        let mut pending = std::mem::take(&mut self.pending);
215        pending.push(chunk);
216        for chunk in pending {
217            if chunk.choices.len() > 1 || chunk.choices.iter().any(|v| v.index != 0) {
218                return Err(TransformError::unsupported(
219                    "choices",
220                    "Chat→Responses stream requires one choice; fan-out must be a separate invocation",
221                ));
222            }
223            for choice in chunk.choices {
224                let choice = declared(choice);
225                self.apply_choice(choice)?;
226            }
227        }
228        Ok(Converted {
229            value: std::mem::take(&mut self.events),
230            report: std::mem::take(&mut self.report),
231        })
232    }
233
234    fn start_if_needed(&mut self) -> Result<(), TransformError> {
235        if self.base.is_some() {
236            return Ok(());
237        }
238        let source_id = self.response_id.clone();
239        let created = self
240            .created
241            .ok_or_else(|| TransformError::missing_metadata("chat.created"))?;
242        let model = self
243            .model
244            .clone()
245            .ok_or_else(|| TransformError::missing_metadata("chat.model"))?;
246        let id = self
247            .flow
248            .resolve_or_allocate(
249                IdentityRole::Response,
250                SourceIdentity::new(Dialect::OpenAiChat, source_id, 0),
251                &self.target_policy,
252            )
253            .map_err(|e| TransformError::invalid_result("identity", e.to_string()))?
254            .emitted_id;
255        let context = self
256            .context
257            .take()
258            .ok_or_else(|| invalid("missing response context"))?;
259        super::common::measure(&context.request, self.limits.max_bytes)?;
260        super::common::measure(&context.effective_tool_choice, self.limits.max_bytes)?;
261        let mut base = context.into_response(id, created, model)?;
262        super::common::measure(&base, self.limits.max_bytes)?;
263        base.status = Some(r::ResponseStatus::InProgress);
264        base.output.clear();
265        base.usage = None;
266        base.output_text = None;
267        base.completed_at = None;
268        base.error = None;
269        base.incomplete_details = None;
270        self.emit(rs::StreamEvent::Created(rs::ResponseCreated {
271            sequence_number: self.sequence,
272            response: base.clone(),
273            rest: Default::default(),
274        }))?;
275        self.base = Some(base);
276        Ok(())
277    }
278
279    fn apply_choice(&mut self, choice: cs::StreamChoice) -> Result<(), TransformError> {
280        if let Some(Some(logs)) = choice.logprobs {
281            self.message_mut()?;
282            crate::transform::generate::stream::chat::logs::append(
283                &mut self.message.as_mut().unwrap().logs,
284                logs,
285            )?;
286        }
287        let d = choice.delta;
288        if let Some(text) =
289            crate::wire::openai::chat::reasoning_text(&d.reasoning_content, &d.reasoning)
290        {
291            self.reasoning_text.push_str(text);
292        }
293        if let Some(Some(details)) = d.reasoning_details {
294            rd::merge(&mut self.reasoning_details, details)?;
295        }
296
297        if let Some(Some(role)) = d.role
298            && role != cs::DeltaRole::Assistant
299        {
300            return Err(invalid("Chat stream delta role must be assistant"));
301        }
302        if let Some(Some(text)) = d.content {
303            self.message_mut()?;
304            self.message_text(text)?;
305        }
306        if let Some(Some(text)) = d.refusal {
307            self.message_mut()?;
308            let state = self.message.as_mut().unwrap();
309            state.refusal_part = true;
310            state.refusal.push_str(&text);
311        }
312        if let Some(Some(function)) = d.function_call {
313            self.tool_delta(-1, None, function)?;
314        }
315        if let Some(Some(calls)) = d.tool_calls {
316            for call in calls {
317                if call.index < 0 {
318                    return Err(invalid("negative tool index"));
319                }
320                self.tool_delta(
321                    call.index,
322                    call.id.flatten(),
323                    call.function
324                        .flatten()
325                        .unwrap_or_else(|| cs::DeltaFunctionCall::builder().build()),
326                )?;
327            }
328        }
329        if let Some(Some(reason)) = choice.finish_reason {
330            self.finish = Some(reason);
331            self.finish_choice()?;
332        }
333        Ok(())
334    }
335
336    fn emit(&mut self, event: rs::StreamEvent) -> Result<(), TransformError> {
337        self.budget.output(&event)?;
338        self.target
339            .as_mut()
340            .ok_or_else(|| invalid("target consumed"))?
341            .push(event.clone())?;
342        self.sequence = self.sequence.checked_add(1).ok_or_else(limit)?;
343        self.events.push(event);
344        Ok(())
345    }
346
347    pub fn push_done(&mut self) -> Result<(), TransformError> {
348        if self.failed || self.done {
349            self.failed = true;
350            return Err(invalid("duplicate DONE or failed stream"));
351        }
352        let result = self
353            .source
354            .as_mut()
355            .ok_or_else(|| invalid("source consumed"))?
356            .push_done();
357        if result.is_err() {
358            self.failed = true;
359        } else {
360            self.done = true;
361        }
362        result
363    }
364
365    pub fn finish(mut self) -> Result<StreamEnd<rs::StreamEvent>, TransformError> {
366        if self.failed || !self.done {
367            return Err(invalid("Chat stream ended without valid SSE DONE"));
368        }
369        let source = self
370            .source
371            .take()
372            .ok_or_else(|| invalid("source consumed"))?
373            .finish()?;
374        self.report.diagnostics.extend(source.report.diagnostics);
375        let final_chat = source.value;
376        if final_chat.choices.len() != 1 {
377            return Err(TransformError::unsupported(
378                "choices",
379                "one Responses response requires one Chat choice",
380            ));
381        }
382        let usage = final_chat
383            .usage
384            .map(|u| {
385                super::super::response::usage::to_responses(u, self.usage_facts, &mut self.report)
386            })
387            .map(crate::transform::optional)
388            .transpose()?
389            .flatten();
390        let mut response = self
391            .base
392            .take()
393            .ok_or_else(|| invalid("missing response.created"))?;
394        response.output = self.current_output();
395        response.usage = usage.map(Some);
396        response.status = Some(
397            if matches!(
398                self.finish,
399                Some(c::FinishReason::Length | c::FinishReason::ContentFilter)
400            ) {
401                r::ResponseStatus::Incomplete
402            } else {
403                r::ResponseStatus::Completed
404            },
405        );
406        response.incomplete_details = match self.finish {
407            Some(c::FinishReason::Length) => Some(r::ResponseIncompleteDetails {
408                reason: Some(r::ResponseIncompleteReason::MaxOutputTokens),
409                rest: Default::default(),
410            }),
411            Some(c::FinishReason::ContentFilter) => Some(r::ResponseIncompleteDetails {
412                reason: Some(r::ResponseIncompleteReason::ContentFilter),
413                rest: Default::default(),
414            }),
415            _ => None,
416        };
417        response.service_tier = final_chat
418            .service_tier
419            .map(|v| v.map(super::super::response::tier_to_responses));
420        response.moderation = final_chat
421            .moderation
422            .map(|v| {
423                v.map(|v| super::super::response::moderation::to_responses(v, &mut self.report))
424                    .transpose()
425            })
426            .map(crate::transform::optional)
427            .transpose()?
428            .flatten();
429        if final_chat.system_fingerprint.is_some() {
430            self.report
431                .omitted("system_fingerprint", "Responses has no fingerprint field");
432        }
433        let terminal = if response.status == Some(r::ResponseStatus::Incomplete) {
434            rs::StreamEvent::Incomplete(rs::ResponseIncomplete {
435                sequence_number: self.sequence,
436                response,
437                rest: Default::default(),
438            })
439        } else {
440            rs::StreamEvent::Completed(rs::ResponseCompleted {
441                sequence_number: self.sequence,
442                response,
443                rest: Default::default(),
444            })
445        };
446        self.emit(terminal)?;
447        self.target
448            .take()
449            .ok_or_else(|| invalid("target consumed"))?
450            .finish()?;
451        Ok(StreamEnd {
452            chunks: self.events,
453            identities: self.flow,
454            report: self.report,
455        })
456    }
457
458    fn current_output(&self) -> Vec<r::ResponseOutputItem> {
459        self.output.values().cloned().collect()
460    }
461}
462
463fn invalid_owned(message: String) -> TransformError {
464    TransformError::invalid_result("chat_responses.stream", message)
465}
gproxy_protocol/transform/generate/chat_responses/stream/chat_to_responses.rs

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