gproxy_protocol/transform/generate/claude_gemini/stream/
gemini_to_claude.rs

1use super::super::ClaudeGeminiUsageFacts;
2use super::{
3    common::{Budget, StreamEnd, StreamLimits, bound, invalid, limit, response_id},
4    usage::{GeminiUsageProgress, gemini_usage, prepare_initial},
5};
6use crate::transform::generate::stream::{
7    claude::{ClaudeStreamCollector, ClaudeStreamLimits},
8    gemini::{GeminiStreamCollector, GeminiStreamLimits},
9};
10use crate::{
11    transform::{
12        Converted, Report, TransformError, TransformErrorKind,
13        generate::signature,
14        identity::{IdentityFlow, TargetIdPolicy},
15    },
16    wire::{
17        DeclaredFields,
18        claude::{
19            content::{ThinkingBlock, ThinkingBlockType},
20            generate_content as c, stream as s,
21        },
22        gemini as g,
23    },
24};
25
26#[derive(Debug, Default)]
27pub struct GeminiToClaudeContext {
28    /// Actual model fallback, used only until a native model is observed.
29    pub model: Option<String>,
30    /// Actual source identity fallback; None permits a stable generated alias.
31    pub response_id: Option<String>,
32    pub usage: Option<c::Usage>,
33    pub facts: ClaudeGeminiUsageFacts,
34}
35
36pub struct GeminiToClaudeStream {
37    source: Option<GeminiStreamCollector>,
38    target: Option<ClaudeStreamCollector>,
39    flow: IdentityFlow,
40    policy: TargetIdPolicy,
41    budget: Budget,
42    limits: StreamLimits,
43    ctx: GeminiToClaudeContext,
44    fixed_start: bool,
45    usage: GeminiUsageProgress,
46    source_id: Option<String>,
47    model: Option<String>,
48    started: bool,
49    failed: bool,
50    pending: Vec<s::StreamEvent>,
51    pending_bytes: usize,
52    next_block: usize,
53    text_block: Option<i64>,
54    /// The open thinking block, closed by its run's signature part.
55    thinking_block: Option<i64>,
56
57    calls: super::super::history::Calls,
58}
59
60impl GeminiToClaudeStream {
61    pub fn new(
62        ctx: GeminiToClaudeContext,
63        flow: IdentityFlow,
64        limits: StreamLimits,
65    ) -> Result<Self, TransformError> {
66        Self::new_with_policy(ctx, flow, limits, super::common::claude_policy())
67    }
68    pub fn new_with_policy(
69        mut ctx: GeminiToClaudeContext,
70        flow: IdentityFlow,
71        limits: StreamLimits,
72        policy: TargetIdPolicy,
73    ) -> Result<Self, TransformError> {
74        if policy.dialect != crate::Dialect::Claude {
75            return Err(TransformError::shape(
76                "stream.policy",
77                "target policy dialect mismatch",
78            ));
79        }
80
81        ctx.usage = ctx.usage.into_declared();
82
83        if ctx.model.as_ref().is_some_and(|v| v.trim().is_empty())
84            || ctx.response_id.as_ref().is_some_and(String::is_empty)
85        {
86            return Err(invalid(
87                "context",
88                "empty actual model or response identity",
89            ));
90        }
91        if let Some(usage) = &mut ctx.usage {
92            prepare_initial(usage, ctx.facts)?;
93        }
94        bound(
95            &(&ctx.model, &ctx.response_id, &ctx.usage),
96            limits.max_bytes,
97        )?;
98        Ok(Self {
99            source: Some(GeminiStreamCollector::new(GeminiStreamLimits {
100                max_bytes: limits.max_bytes,
101            })),
102            target: Some(ClaudeStreamCollector::new(ClaudeStreamLimits {
103                max_text_bytes: limits.max_bytes,
104                max_json_bytes: limits.max_bytes,
105            })),
106            flow,
107            policy,
108            budget: Budget::new(limits),
109            limits,
110            source_id: ctx.response_id.clone(),
111            model: ctx.model.clone(),
112            fixed_start: ctx.usage.is_some(),
113            ctx,
114            usage: Default::default(),
115            started: false,
116            failed: false,
117            pending: Vec::new(),
118            pending_bytes: 0,
119            next_block: 0,
120            text_block: None,
121            thinking_block: None,
122
123            calls: Default::default(),
124        })
125    }
126    pub fn identities(&self) -> &IdentityFlow {
127        &self.flow
128    }
129    pub fn push(
130        &mut self,
131        chunk: g::GenerateContentResponseBody,
132    ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
133        if self.failed {
134            return Err(invalid("stream", "stream failed"));
135        }
136        let result = self.inner(chunk.into_declared());
137        if result.is_err() {
138            self.failed = true;
139        }
140        result
141    }
142    fn inner(
143        &mut self,
144        chunk: g::GenerateContentResponseBody,
145    ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
146        self.budget.input(&chunk)?;
147        self.source
148            .as_mut()
149            .ok_or_else(|| invalid("stream", "source consumed"))?
150            .push(chunk.clone())?;
151        if let Some(id) = chunk.response_id {
152            if self.started {
153                response_id(
154                    &mut self.flow,
155                    &self.policy,
156                    crate::Dialect::Gemini,
157                    Some(id.clone()),
158                )?;
159            }
160            self.source_id = Some(id);
161        }
162        if let Some(model) = chunk.model_version {
163            if self.started && self.model.as_ref() != Some(&model) {
164                return Err(invalid(
165                    "model_version",
166                    "actual model contradicts emitted Claude model",
167                ));
168            }
169            self.model = Some(model);
170        }
171        if let Some(usage) = chunk.usage_metadata {
172            self.usage.observe(&usage);
173        }
174        if self.ctx.usage.is_none() {
175            let mut snapshot = self.usage.usage().clone();
176            let mut current_facts = self.ctx.facts;
177            current_facts.thinking_tokens = None;
178            let observed = self
179                .usage
180                .effective(&mut snapshot, None)
181                .and_then(|_| gemini_usage(snapshot, current_facts, None, &mut Report::default()));
182            match observed {
183                Ok((usage, _)) => self.ctx.usage = Some(usage),
184                // Partial usage fields may precede a coherent total snapshot.
185                // Native negative/decreasing counters already failed above.
186                Err(e)
187                    if matches!(
188                        e.kind(),
189                        TransformErrorKind::MissingMetadata | TransformErrorKind::InvalidResult
190                    ) => {}
191                Err(e) => return Err(e),
192            }
193        }
194        let mut out = Vec::new();
195        let mut report = Report::default();
196        self.ensure_start(&mut out)?;
197        for candidate in chunk.candidates.unwrap_or_default() {
198            if candidate.index.is_some_and(|v| v != 0) {
199                continue;
200            }
201            if candidate
202                .content
203                .as_ref()
204                .and_then(|v| v.role.as_deref())
205                .is_some_and(|v| v != "model")
206            {
207                return Err(invalid("candidate.role", "model role required"));
208            }
209            for part in candidate.content.and_then(|v| v.parts).unwrap_or_default() {
210                crate::transform::generate::gemini_controls::omitted(&part, &mut report);
211                self.part(part, &mut out)?;
212            }
213        }
214        Ok(Converted { value: out, report })
215    }
216    fn ensure_start(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
217        if self.started {
218            return Ok(());
219        }
220        let (Some(model), Some(usage)) = (&self.model, &self.ctx.usage) else {
221            return Ok(());
222        };
223        let id = response_id(
224            &mut self.flow,
225            &self.policy,
226            crate::Dialect::Gemini,
227            self.source_id.clone(),
228        )?;
229        let message = s::StreamMessage {
230            input_transformations: None,
231            type_: c::GenerateContentResponseBodyType::Tag,
232            id,
233            container: None,
234            content: Vec::new(),
235            context_management: None,
236            diagnostics: None,
237            model: model.clone(),
238            role: c::ResponseRole::Assistant,
239            stop_details: None,
240            stop_reason: None,
241            stop_sequence: None,
242            usage: usage.clone(),
243            rest: Default::default(),
244        };
245        let event =
246            s::StreamEvent::MessageStart(Box::new(s::MessageStartEvent::builder(message).build()));
247        self.budget.output(&event)?;
248        self.deliver(out, event)?;
249        self.started = true;
250        for event in std::mem::take(&mut self.pending) {
251            self.deliver(out, event)?;
252        }
253        self.pending_bytes = 0;
254        Ok(())
255    }
256    fn allocate_block(&mut self) -> Result<i64, TransformError> {
257        let index = i64::try_from(self.next_block).map_err(|_| limit())?;
258        self.next_block += 1;
259        Ok(index)
260    }
261    fn part(&mut self, part: g::Part, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
262        if super::super::thinking::is_empty_text(&part) {
263            return Ok(());
264        }
265        if super::super::thinking::is_run_part(&part) {
266            return self.thought(part, out);
267        }
268        self.close_thinking(out)?;
269        let carried = super::super::thinking::call_signature(&part);
270        if let Some(text) = part.text.filter(|_| part.thought != Some(true)) {
271            // Gemini chunks extend the current text; they are not separate
272            // Claude messages. In particular, a trailing empty chunk must not
273            // replace the CLI's final assistant result with an empty block.
274            let index = if let Some(index) = self.text_block {
275                index
276            } else {
277                let index = self.allocate_block()?;
278                let block = c::ResponseContentBlock::Text(
279                    c::ResponseTextBlock::builder(c::ResponseTextBlockType::Tag, String::new())
280                        .build(),
281                );
282                self.emit(
283                    out,
284                    s::StreamEvent::ContentBlockStart(
285                        s::ContentBlockStartEvent::builder(index, block).build(),
286                    ),
287                )?;
288                self.text_block = Some(index);
289                index
290            };
291            self.emit(
292                out,
293                s::StreamEvent::ContentBlockDelta(
294                    s::ContentBlockDeltaEvent::builder(
295                        index,
296                        s::ContentBlockDelta::Text(s::TextDelta::builder(text).build()),
297                    )
298                    .build(),
299                ),
300            )?;
301        }
302        if let Some(call) = part.function_call {
303            self.close_text(out)?;
304
305            if let Some(carried) = carried {
306                // The call's own signature travels on an empty thinking block
307                // right before it; see `thinking`.
308                self.open_thinking(out)?;
309                self.end_thinking(carried, out)?;
310            }
311
312            let id = self.calls.call(
313                call.id,
314                &call.name,
315                crate::Dialect::Gemini,
316                &mut self.flow,
317                &self.policy,
318            )?;
319            let index = self.allocate_block()?;
320            let block = c::ResponseContentBlock::ToolUse(
321                c::ResponseToolUseBlock::builder(
322                    c::ResponseToolUseBlockType::Tag,
323                    id,
324                    Default::default(),
325                    call.name,
326                )
327                .build(),
328            );
329            self.emit(
330                out,
331                s::StreamEvent::ContentBlockStart(
332                    s::ContentBlockStartEvent::builder(index, block).build(),
333                ),
334            )?;
335            let json = serde_json::to_string(&call.args.unwrap_or_default())?;
336            self.emit(
337                out,
338                s::StreamEvent::ContentBlockDelta(
339                    s::ContentBlockDeltaEvent::builder(
340                        index,
341                        s::ContentBlockDelta::InputJson(s::InputJsonDelta::builder(json).build()),
342                    )
343                    .build(),
344                ),
345            )?;
346            self.emit(
347                out,
348                s::StreamEvent::ContentBlockStop(s::ContentBlockStopEvent::builder(index).build()),
349            )?;
350        }
351        Ok(())
352    }
353    /// Thought text streams as thinking deltas; the run's signature part
354    /// sends its carried signature and closes the block.
355    fn thought(
356        &mut self,
357        part: g::Part,
358        out: &mut Vec<s::StreamEvent>,
359    ) -> Result<(), TransformError> {
360        self.close_text(out)?;
361        let index = self.open_thinking(out)?;
362        if let Some(text) = part.text.filter(|text| !text.is_empty()) {
363            self.emit(
364                out,
365                s::StreamEvent::ContentBlockDelta(
366                    s::ContentBlockDeltaEvent::builder(
367                        index,
368                        s::ContentBlockDelta::Thinking(s::ThinkingDelta::builder(text).build()),
369                    )
370                    .build(),
371                ),
372            )?;
373        }
374        if let Some(native) = part.thought_signature.as_deref() {
375            self.end_thinking(signature::gemini_thought(Some(native)), out)?;
376        }
377        Ok(())
378    }
379    /// The open thinking block, started empty when none is open.
380    fn open_thinking(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<i64, TransformError> {
381        if let Some(index) = self.thinking_block {
382            return Ok(index);
383        }
384        let index = self.allocate_block()?;
385        let block = c::ResponseContentBlock::Thinking(
386            ThinkingBlock::builder(ThinkingBlockType::Tag, String::new(), String::new()).build(),
387        );
388        self.emit(
389            out,
390            s::StreamEvent::ContentBlockStart(
391                s::ContentBlockStartEvent::builder(index, block).build(),
392            ),
393        )?;
394        self.thinking_block = Some(index);
395        Ok(index)
396    }
397    /// A run that ends without a signature is an unsigned summary.
398    fn close_thinking(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
399        if self.thinking_block.is_some() {
400            self.end_thinking(signature::gemini_thought(None), out)?;
401        }
402        Ok(())
403    }
404    fn end_thinking(
405        &mut self,
406        signature: String,
407        out: &mut Vec<s::StreamEvent>,
408    ) -> Result<(), TransformError> {
409        let Some(index) = self.thinking_block.take() else {
410            return Ok(());
411        };
412        self.emit(
413            out,
414            s::StreamEvent::ContentBlockDelta(
415                s::ContentBlockDeltaEvent::builder(
416                    index,
417                    s::ContentBlockDelta::Signature(s::SignatureDelta::builder(signature).build()),
418                )
419                .build(),
420            ),
421        )?;
422        self.emit(
423            out,
424            s::StreamEvent::ContentBlockStop(s::ContentBlockStopEvent::builder(index).build()),
425        )
426    }
427    fn close_text(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
428        if let Some(index) = self.text_block.take() {
429            self.emit(
430                out,
431                s::StreamEvent::ContentBlockStop(s::ContentBlockStopEvent::builder(index).build()),
432            )?;
433        }
434        Ok(())
435    }
436    fn emit(
437        &mut self,
438        out: &mut Vec<s::StreamEvent>,
439        event: s::StreamEvent,
440    ) -> Result<(), TransformError> {
441        let size = self.budget.output(&event)?;
442        if self.started {
443            self.deliver(out, event)
444        } else {
445            let bytes = self
446                .pending_bytes
447                .checked_add(size)
448                .filter(|n| *n <= self.limits.max_pending)
449                .ok_or_else(limit)?;
450            self.pending.push(event);
451            self.pending_bytes = bytes;
452            Ok(())
453        }
454    }
455    fn deliver(
456        &mut self,
457        out: &mut Vec<s::StreamEvent>,
458        event: s::StreamEvent,
459    ) -> Result<(), TransformError> {
460        self.target
461            .as_mut()
462            .ok_or_else(|| invalid("stream", "target consumed"))?
463            .push(event.clone())?;
464        out.push(event);
465        Ok(())
466    }
467    pub fn finish(mut self) -> Result<StreamEnd<s::StreamEvent>, TransformError> {
468        if self.failed {
469            return Err(invalid("stream", "stream failed"));
470        }
471        let mut source = self
472            .source
473            .take()
474            .ok_or_else(|| invalid("stream", "source consumed"))?
475            .finish()?
476            .value;
477        let normalized = self.usage.effective(
478            source
479                .usage_metadata
480                .as_mut()
481                .ok_or_else(|| TransformError::missing_metadata("usage_metadata"))?,
482            self.ctx.facts.thinking_tokens,
483        )?;
484        let (usage, facts) = gemini_usage(
485            source
486                .usage_metadata
487                .clone()
488                .ok_or_else(|| TransformError::missing_metadata("usage_metadata"))?,
489            self.ctx.facts,
490            self.ctx.usage.as_ref().filter(|_| self.fixed_start),
491            &mut Report::default(),
492        )?;
493
494        response_id(
495            &mut self.flow,
496            &self.policy,
497            crate::Dialect::Gemini,
498            self.source_id.clone(),
499        )?;
500        let mut converted = super::super::gemini_to_claude_response(
501            source,
502            self.model.clone(),
503            facts,
504            &mut self.flow.clone(),
505            &self.policy,
506        )?;
507        if normalized {
508            converted.report.changed("usage.output_components", "stale prefix components resolved from current totals, components and explicit final facts");
509        }
510        let mut expected = converted.value;
511        expected.usage = usage;
512        bound(&expected, self.limits.max_bytes)?;
513        let mut out = Vec::new();
514        if !self.started {
515            self.ctx.usage = Some(expected.usage.clone());
516            self.model = Some(expected.model.clone());
517            self.ensure_start(&mut out)?;
518        }
519        self.close_thinking(&mut out)?;
520        self.close_text(&mut out)?;
521        let delta = s::MessageDelta::builder()
522            .stop_reason(Some(expected.stop_reason))
523            .build();
524        let usage = &expected.usage;
525        let usage = s::MessageDeltaUsage {
526            output_tokens: usage.output_tokens,
527            input_tokens: Some(Some(usage.input_tokens)),
528            cache_creation_input_tokens: usage.cache_creation_input_tokens,
529            cache_read_input_tokens: usage.cache_read_input_tokens,
530            output_tokens_details: usage.output_tokens_details.clone(),
531            server_tool_use: usage.server_tool_use.clone(),
532            iterations: usage.iterations.clone(),
533            fallback_credit: usage.fallback_credit.clone(),
534            rest: Default::default(),
535        };
536        self.emit(
537            &mut out,
538            s::StreamEvent::MessageDelta(Box::new(
539                s::MessageDeltaEvent::builder(delta, usage).build(),
540            )),
541        )?;
542        self.emit(
543            &mut out,
544            s::StreamEvent::MessageStop(s::MessageStopEvent::builder().build()),
545        )?;
546        let actual = self
547            .target
548            .take()
549            .ok_or_else(|| invalid("stream", "target consumed"))?
550            .finish()?
551            .value;
552        if actual != expected {
553            return Err(invalid(
554                "stream.parity",
555                "native Claude output differs from completed Gemini conversion",
556            ));
557        }
558        Ok(StreamEnd {
559            chunks: out,
560            identities: self.flow,
561            report: converted.report,
562        })
563    }
564}
gproxy_protocol/transform/generate/claude_gemini/stream/gemini_to_claude.rs

gproxy_protocol/transform/generate/claude_gemini/stream/
gemini_to_claude.rs
