gproxy_protocol/transform/generate/claude_responses/stream/
responses_to_claude.rs

1use super::{
2    common::{
3        Budget, StreamEnd, StreamLimits, declared, id, initial_usage, invalid, limit, measure,
4    },
5    response_items::Item,
6    usage::claude_tier,
7};
8use crate::{
9    transform::generate::stream::{
10        claude::{ClaudeStreamCollector, ClaudeStreamLimits},
11        responses::{ResponsesStreamCollector, ResponsesStreamLimits},
12    },
13    transform::{
14        Converted, Report, TransformError,
15        identity::{IdentityFlow, IdentityRole, TargetIdPolicy},
16    },
17    wire::{
18        claude::{generate_content as c, stream as cs},
19        openai::responses::{response as r, stream as s},
20    },
21};
22use std::collections::BTreeMap;
23
24#[derive(Default)]
25pub struct ResponsesToClaudeContext {
26    pub usage: Option<c::Usage>,
27}
28
29pub struct ResponsesToClaudeStream {
30    source: Option<ResponsesStreamCollector>,
31    target: Option<ClaudeStreamCollector>,
32    pub(super) flow: IdentityFlow,
33    pub(super) policy: TargetIdPolicy,
34    pub(super) limits: StreamLimits,
35    budget: Budget,
36    usage: Option<c::Usage>,
37    observed_usage: Option<r::ResponseUsage>,
38    pub(super) model: Option<String>,
39    response_id: Option<String>,
40    tier: Option<Option<c::ResponseServiceTier>>,
41    started: bool,
42    failed: bool,
43    terminal: bool,
44    pending: Vec<cs::StreamEvent>,
45    pending_bytes: usize,
46    pub(super) held_bytes: usize,
47    pub(super) items: BTreeMap<i64, Item>,
48    pub(super) cursor: i64,
49
50    next_block: usize,
51    pub(super) report: Report,
52}
53
54impl ResponsesToClaudeStream {
55    pub fn new(
56        context: ResponsesToClaudeContext,
57        flow: IdentityFlow,
58        limits: StreamLimits,
59    ) -> Result<Self, TransformError> {
60        Self::new_with_policy(context, flow, limits, super::common::claude_policy())
61    }
62    pub fn new_with_policy(
63        mut context: ResponsesToClaudeContext,
64        flow: IdentityFlow,
65        limits: StreamLimits,
66        policy: TargetIdPolicy,
67    ) -> Result<Self, TransformError> {
68        if policy.dialect != crate::Dialect::Claude {
69            return Err(TransformError::shape(
70                "stream.policy",
71                "target policy dialect mismatch",
72            ));
73        }
74
75        context.usage = context
76            .usage
77            .map(initial_usage)
78            .map(crate::transform::optional)
79            .transpose()?
80            .flatten();
81        measure(&context.usage, limits.max_bytes)?;
82        Ok(Self {
83            source: Some(ResponsesStreamCollector::new(ResponsesStreamLimits {
84                max_bytes: limits.max_bytes,
85
86                max_text_bytes: limits.max_bytes,
87                max_json_bytes: limits.max_bytes,
88            })),
89            target: Some(ClaudeStreamCollector::new(ClaudeStreamLimits {
90                max_text_bytes: limits.max_bytes,
91                max_json_bytes: limits.max_bytes,
92            })),
93            flow,
94            policy,
95            limits,
96            budget: Budget::new(limits),
97            usage: context.usage,
98            observed_usage: None,
99            model: None,
100            response_id: None,
101            tier: None,
102            started: false,
103            failed: false,
104            terminal: false,
105            pending: Vec::new(),
106            pending_bytes: 0,
107            held_bytes: 0,
108            items: BTreeMap::new(),
109            cursor: 0,
110
111            next_block: 0,
112            report: Default::default(),
113        })
114    }
115    pub fn identities(&self) -> &IdentityFlow {
116        &self.flow
117    }
118    pub fn push(
119        &mut self,
120        event: s::StreamEvent,
121    ) -> Result<Converted<Vec<cs::StreamEvent>>, TransformError> {
122        if self.failed || self.terminal {
123            self.failed = true;
124            return Err(invalid("event after terminal or failure"));
125        }
126        let result = self.inner(declared(event));
127        if result.is_err() {
128            self.failed = true;
129        }
130        result
131    }
132    fn inner(
133        &mut self,
134        event: s::StreamEvent,
135    ) -> Result<Converted<Vec<cs::StreamEvent>>, TransformError> {
136        let event = crate::transform::generate::multi_agent::attribute_event(event)?;
137        self.budget.input(&event)?;
138        self.source
139            .as_mut()
140            .ok_or_else(|| invalid("source consumed"))?
141            .push(event.clone())?;
142        let mut out = Vec::new();
143        match event {
144            s::StreamEvent::Keepalive => out.push(cs::StreamEvent::Ping(cs::PingEvent {
145                rest: Default::default(),
146            })),
147            s::StreamEvent::Created(v) => {
148                self.response_id = Some(id(
149                    &mut self.flow,
150                    &self.policy,
151                    crate::Dialect::OpenAi,
152                    IdentityRole::Response,
153                    IdentityRole::Response,
154                    Some(v.response.id.clone()),
155                    0,
156                )?);
157                self.model = Some(v.response.model.clone());
158                self.observe(&v.response)?;
159                self.start(&mut out)?;
160                for (index, item) in v.response.output.into_iter().enumerate() {
161                    self.add_item(index as i64, item)?;
162                }
163            }
164            s::StreamEvent::Queued(v) => {
165                self.observe(&v.response)?;
166                self.start(&mut out)?;
167            }
168            s::StreamEvent::InProgress(v) => {
169                self.observe(&v.response)?;
170                self.start(&mut out)?;
171            }
172            s::StreamEvent::OutputItemAdded(v) => self.add_item(v.output_index, v.item)?,
173            s::StreamEvent::ContentPartAdded(v) => {
174                self.bind(v.output_index, Some(&v.item_id))?;
175                self.add_part(v.output_index, v.content_index, v.part)?;
176            }
177            s::StreamEvent::ContentPartDone(v) => {
178                self.part_done(v.output_index, v.content_index, &mut out)?
179            }
180            s::StreamEvent::OutputTextDelta(v) => self.append_text(
181                v.output_index,
182                v.content_index,
183                &v.item_id,
184                v.delta,
185                false,
186                &mut out,
187            )?,
188            s::StreamEvent::RefusalDelta(v) => self.append_text(
189                v.output_index,
190                v.content_index,
191                &v.item_id,
192                v.delta,
193                true,
194                &mut out,
195            )?,
196            s::StreamEvent::FunctionCallArgumentsDelta(v) => {
197                self.append_args(v.output_index, &v.item_id, v.delta, &mut out)?
198            }
199            s::StreamEvent::FunctionCallArgumentsDone(v) => {
200                self.bind(v.output_index, Some(&v.item_id))?;
201                self.args_done(v.output_index, &v.arguments, &mut out)?;
202            }
203            s::StreamEvent::OutputItemDone(v) => {
204                self.item_done(v.output_index, v.item, &mut out)?
205            }
206            s::StreamEvent::Completed(v) => {
207                self.observe(&v.response)?;
208                self.terminal = true;
209                self.start(&mut out)?;
210            }
211            s::StreamEvent::Incomplete(v) => {
212                self.observe(&v.response)?;
213                self.terminal = true;
214                self.start(&mut out)?;
215            }
216            s::StreamEvent::ReasoningSummaryPartAdded(_) => {}
217            s::StreamEvent::OutputTextDone(_)
218            | s::StreamEvent::RefusalDone(_)
219            | s::StreamEvent::OutputTextAnnotationAdded(_)
220            | s::StreamEvent::ReasoningTextDelta(_)
221            | s::StreamEvent::ReasoningTextDone(_)
222            | s::StreamEvent::ReasoningSummaryTextDelta(_)
223            | s::StreamEvent::ReasoningSummaryTextDone(_)
224            | s::StreamEvent::ReasoningSummaryPartDone(_)
225            | s::StreamEvent::McpArgumentsDelta(_)
226            | s::StreamEvent::McpArgumentsDone(_)
227            | s::StreamEvent::McpInProgress(_)
228            | s::StreamEvent::McpCompleted(_)
229            | s::StreamEvent::McpFailed(_) => {}
230            s::StreamEvent::Failed(_) | s::StreamEvent::Error(_) => {
231                return Err(invalid("native source failure"));
232            }
233            _ => {}
234        }
235        self.flush_items(&mut out)?;
236        Ok(Converted {
237            value: out,
238            report: std::mem::take(&mut self.report),
239        })
240    }
241    fn observe(&mut self, response: &r::GenerateContentResponseBody) -> Result<(), TransformError> {
242        if !self.started {
243            self.tier = claude_tier(response.service_tier.flatten()).or(self.tier);
244        }
245        if let Some(usage) = response.usage.as_ref().and_then(Option::as_ref) {
246            let native = super::super::response::usage::to_claude(usage.clone())?;
247            if let Some(previous) = &self.observed_usage
248                && (usage.input_tokens < previous.input_tokens
249                    || usage.output_tokens < previous.output_tokens
250                    || usage.total_tokens < previous.total_tokens
251                    || usage.input_tokens_details.cached_tokens
252                        < previous.input_tokens_details.cached_tokens
253                    || usage.input_tokens_details.cache_write_tokens
254                        < previous.input_tokens_details.cache_write_tokens
255                    || usage.output_tokens_details.reasoning_tokens
256                        < previous.output_tokens_details.reasoning_tokens)
257            {
258                return Err(invalid("native cumulative usage decreased"));
259            }
260            self.observed_usage = Some(usage.clone());
261            if self.usage.is_none() {
262                self.usage = Some(native);
263            }
264        }
265        Ok(())
266    }
267    fn start(&mut self, out: &mut Vec<cs::StreamEvent>) -> Result<(), TransformError> {
268        if self.started {
269            return Ok(());
270        }
271        let (Some(id), Some(model)) = (&self.response_id, &self.model) else {
272            return Ok(());
273        };
274        // Responses may only report usage at completion. Do not hold content
275        // until then: message_delta replaces these provisional counters with
276        // the actual final usage, which finish() still requires from the source.
277        let usage = self
278            .usage
279            .get_or_insert_with(|| c::Usage::builder(0, 0).build());
280        if let Some(tier) = self.tier {
281            if usage
282                .service_tier
283                .flatten()
284                .zip(tier)
285                .is_some_and(|(old, new)| old != new)
286            {
287                return Err(invalid("initial service-tier facts contradict source"));
288            }
289            usage.service_tier = Some(tier);
290        }
291        let message = cs::StreamMessage {
292            input_transformations: None,
293            type_: c::GenerateContentResponseBodyType::Tag,
294            id: id.clone(),
295            container: None,
296            content: Vec::new(),
297            context_management: None,
298            diagnostics: None,
299            model: model.clone(),
300            role: c::ResponseRole::Assistant,
301            stop_details: None,
302            stop_reason: None,
303            stop_sequence: None,
304            usage: usage.clone(),
305            rest: Default::default(),
306        };
307        let event = cs::StreamEvent::MessageStart(Box::new(
308            cs::MessageStartEvent::builder(message).build(),
309        ));
310        self.budget.output(&event)?;
311        self.deliver(out, event)?;
312        self.started = true;
313        for event in std::mem::take(&mut self.pending) {
314            self.deliver(out, event)?;
315        }
316        self.pending_bytes = 0;
317        Ok(())
318    }
319    pub(super) fn reserve_held(&mut self, bytes: usize) -> Result<(), TransformError> {
320        let held = self.held_bytes.checked_add(bytes).ok_or_else(limit)?;
321        if held
322            .checked_add(self.pending_bytes)
323            .is_none_or(|n| n > self.limits.max_pending)
324        {
325            return Err(limit());
326        }
327        self.held_bytes = held;
328        Ok(())
329    }
330    pub(super) fn allocate_block(&mut self) -> Result<i64, TransformError> {
331        let index = i64::try_from(self.next_block).map_err(|_| limit())?;
332        self.next_block += 1;
333        Ok(index)
334    }
335    pub(super) fn complete_block(
336        &mut self,
337        out: &mut Vec<cs::StreamEvent>,
338        block: c::ResponseContentBlock,
339    ) -> Result<(), TransformError> {
340        let index = self.allocate_block()?;
341        self.emit(
342            out,
343            cs::StreamEvent::ContentBlockStart(
344                cs::ContentBlockStartEvent::builder(index, block).build(),
345            ),
346        )?;
347        self.emit(
348            out,
349            cs::StreamEvent::ContentBlockStop(cs::ContentBlockStopEvent::builder(index).build()),
350        )
351    }
352    pub(super) fn emit(
353        &mut self,
354        out: &mut Vec<cs::StreamEvent>,
355        event: cs::StreamEvent,
356    ) -> Result<(), TransformError> {
357        let n = self.budget.output(&event)?;
358        if self.started {
359            self.deliver(out, event)
360        } else {
361            let bytes = self.pending_bytes.checked_add(n).ok_or_else(limit)?;
362            if bytes
363                .checked_add(self.held_bytes)
364                .is_none_or(|n| n > self.limits.max_pending)
365            {
366                return Err(limit());
367            }
368            self.pending.push(event);
369            self.pending_bytes = bytes;
370            Ok(())
371        }
372    }
373    fn deliver(
374        &mut self,
375        out: &mut Vec<cs::StreamEvent>,
376        event: cs::StreamEvent,
377    ) -> Result<(), TransformError> {
378        self.target
379            .as_mut()
380            .ok_or_else(|| invalid("target consumed"))?
381            .push(event.clone())?;
382        out.push(event);
383        Ok(())
384    }
385    pub fn finish(mut self) -> Result<StreamEnd<cs::StreamEvent>, TransformError> {
386        if self.failed || !self.terminal {
387            return Err(invalid("failed stream or premature EOF"));
388        }
389        let source = self
390            .source
391            .take()
392            .ok_or_else(|| invalid("source consumed"))?
393            .finish()?
394            .value;
395        let part_refusal = source.output.iter().any(|item| matches!(item, r::ResponseOutputItem::Message(message) if message.content.iter().any(|part| matches!(part, crate::wire::openai::responses::input::OutputContent::Refusal(_)))));
396        let converted = super::super::responses_to_claude_response(
397            source,
398            &mut self.flow.clone(),
399            &self.policy,
400        )?;
401        let mut expected = converted.value;
402        if part_refusal && expected.stop_reason != c::StopReason::Refusal {
403            self.report.changed("output.refusal", "refusal text is preserved; Claude has no separate per-part refusal marker alongside tool continuation or an incomplete stop");
404        }
405        self.report.diagnostics.extend(converted.report.diagnostics);
406        let mut out = Vec::new();
407        if !self.started {
408            self.usage = Some(expected.usage.clone());
409            self.start(&mut out)?;
410        }
411        let initial = self
412            .usage
413            .as_ref()
414            .ok_or_else(|| invalid("missing start usage"))?;
415
416        if expected.usage.service_tier != initial.service_tier {
417            self.report.omitted(
418                "usage.service_tier",
419                "late effective tier cannot update Claude message_start",
420            );
421        }
422        expected.usage.service_tier = initial.service_tier;
423        expected.usage.cache_creation = initial.cache_creation.clone();
424        expected.usage.fallback_credit = initial.fallback_credit.clone();
425        expected.usage.inference_geo = initial.inference_geo.clone();
426        expected.usage.iterations = initial.iterations.clone();
427        expected.usage.server_tool_use = initial.server_tool_use.clone();
428        expected.usage.speed = initial.speed;
429        measure(&expected, self.limits.max_bytes)?;
430        let usage = &expected.usage;
431        let delta = cs::MessageDelta::builder()
432            .stop_reason(Some(expected.stop_reason))
433            .build();
434        let usage = cs::MessageDeltaUsage {
435            output_tokens: usage.output_tokens,
436            input_tokens: Some(Some(usage.input_tokens)),
437            cache_creation_input_tokens: usage.cache_creation_input_tokens,
438            cache_read_input_tokens: usage.cache_read_input_tokens,
439            output_tokens_details: usage.output_tokens_details.clone(),
440            server_tool_use: usage.server_tool_use.clone(),
441            iterations: usage.iterations.clone(),
442            fallback_credit: usage.fallback_credit.clone(),
443            rest: Default::default(),
444        };
445        self.emit(
446            &mut out,
447            cs::StreamEvent::MessageDelta(Box::new(
448                cs::MessageDeltaEvent::builder(delta, usage).build(),
449            )),
450        )?;
451        self.emit(
452            &mut out,
453            cs::StreamEvent::MessageStop(cs::MessageStopEvent::builder().build()),
454        )?;
455        let actual = self
456            .target
457            .take()
458            .ok_or_else(|| invalid("target consumed"))?
459            .finish()?
460            .value;
461        if actual != expected {
462            return Err(invalid(
463                "native Claude output differs from canonical Responses conversion",
464            ));
465        }
466        Ok(StreamEnd {
467            chunks: out,
468            identities: self.flow,
469            report: self.report,
470        })
471    }
472}
gproxy_protocol/transform/generate/claude_responses/stream/responses_to_claude.rs

gproxy_protocol/transform/generate/claude_responses/stream/
responses_to_claude.rs
