gproxy_protocol/adapt/generate/stream/
synthesize.rs

1//! Encode an already converted complete response as a native client lifecycle.
2//! No new client identity is allocated after buffered invocation state was saved.
3
4use super::super::GenerationOutcome;
5use super::{
6    event::{EventLimits, NativeEvent},
7    output,
8    reader::SourceFraming,
9};
10use crate::{
11    Dialect, HttpBody, WireResponse,
12    codec::CodecLimits,
13    transform::{
14        Converted, Report, TransformError,
15        generate::stream as native,
16        identity::{IdNamespace, IdentityFlow},
17    },
18    wire::{
19        DeclaredFields,
20        claude::{generate_content as c, stream as cs},
21        gemini as g,
22        openai::{
23            chat::{self as h, stream as hs},
24            responses::{self as r, stream as rs},
25        },
26    },
27};
28use bytes::Bytes;
29
30mod sealed {
31    pub trait Complete {}
32}
33
34pub trait CompleteResponse:
35    sealed::Complete + DeclaredFields + serde::Serialize + Clone + Sized
36{
37    type Event: NativeEvent<Full = Self>;
38    fn events(
39        self,
40        namespace: IdNamespace,
41        limits: EventLimits,
42    ) -> Result<Converted<Vec<Self::Event>>, TransformError>;
43}
44
45pub enum GenerationStreamOutcome {
46    Success {
47        response: WireResponse<HttpBody>,
48        report: Report,
49    },
50    Rejected(WireResponse<Bytes>),
51}
52
53/// The buffered adapter must have completed its required state writes first.
54/// This helper preserves all its client IDs and pre-encodes under the supplied
55/// aggregate byte limit before exposing any stream bytes.
56pub fn synthesize<C: CompleteResponse>(
57    outcome: GenerationOutcome<C>,
58    namespace: IdNamespace,
59    framing: SourceFraming,
60    codec: CodecLimits,
61    events: EventLimits,
62) -> Result<GenerationStreamOutcome, TransformError> {
63    let (response, mut report) = match outcome {
64        GenerationOutcome::Success { response, report } => (response, report),
65        GenerationOutcome::Rejected(raw) => return Ok(GenerationStreamOutcome::Rejected(raw)),
66    };
67    if C::Event::DIALECT != Dialect::Gemini && framing != SourceFraming::Sse {
68        return Err(TransformError::unsupported(
69            "generation.synthesis.framing",
70            "this native dialect requires SSE",
71        ));
72    }
73    let converted = response.body.into_declared().events(namespace, events)?;
74    let mut encoder = output::Encoder::new(framing, codec);
75    let mut chunks = Vec::new();
76    for event in converted.value {
77        chunks.push(encoder.event(&event, codec)?);
78    }
79    let tail = encoder.finish::<C::Event>()?;
80    if !tail.is_empty() {
81        chunks.push(tail);
82    }
83    report.diagnostics.extend(converted.report.diagnostics);
84    Ok(GenerationStreamOutcome::Success {
85        response: WireResponse {
86            status: response.status,
87            headers: output::headers(response.headers, framing),
88            body: HttpBody::Stream(Box::pin(futures_util::stream::iter(
89                chunks.into_iter().map(Ok),
90            ))),
91        },
92        report,
93    })
94}
95
96/// Synthesize the complete Chat result using the actual client's usage option.
97/// The generic `synthesize` utility instead preserves every field of its DTO.
98pub fn synthesize_chat(
99    mut outcome: GenerationOutcome<h::GenerateContentResponseBody>,
100    request: &h::GenerateContentRequestBody,
101    namespace: IdNamespace,
102    framing: SourceFraming,
103    codec: CodecLimits,
104    events: EventLimits,
105) -> Result<GenerationStreamOutcome, TransformError> {
106    if request.stream.flatten() == Some(true)
107        && request
108            .stream_options
109            .as_ref()
110            .and_then(Option::as_ref)
111            .and_then(|options| options.include_usage)
112            != Some(true)
113        && let GenerationOutcome::Success { response, .. } = &mut outcome
114    {
115        response.body.usage = None;
116    }
117    synthesize(outcome, namespace, framing, codec, events)
118}
119
120impl sealed::Complete for c::GenerateContentResponseBody {}
121impl CompleteResponse for c::GenerateContentResponseBody {
122    type Event = cs::StreamEvent;
123    fn events(
124        self,
125        _: IdNamespace,
126        limits: EventLimits,
127    ) -> Result<Converted<Vec<Self::Event>>, TransformError> {
128        native::claude::synthesize_claude_stream(
129            self,
130            native::claude::ClaudeStreamLimits {
131                max_json_bytes: limits.max_bytes,
132                max_text_bytes: limits.max_bytes,
133            },
134        )
135    }
136}
137
138impl sealed::Complete for h::GenerateContentResponseBody {}
139impl CompleteResponse for h::GenerateContentResponseBody {
140    type Event = hs::ChatCompletionChunk;
141    fn events(
142        self,
143        _: IdNamespace,
144        limits: EventLimits,
145    ) -> Result<Converted<Vec<Self::Event>>, TransformError> {
146        native::chat::synthesize_chat_stream(
147            self,
148            native::chat::ChatStreamLimits {
149                max_bytes: limits.max_bytes,
150            },
151        )
152    }
153}
154
155impl sealed::Complete for g::GenerateContentResponseBody {}
156impl CompleteResponse for g::GenerateContentResponseBody {
157    type Event = Self;
158    fn events(
159        self,
160        _: IdNamespace,
161        limits: EventLimits,
162    ) -> Result<Converted<Vec<Self::Event>>, TransformError> {
163        native::gemini::synthesize_gemini_stream(
164            self,
165            native::gemini::GeminiStreamLimits {
166                max_bytes: limits.max_bytes,
167            },
168        )
169    }
170}
171
172impl sealed::Complete for r::GenerateContentResponseBody {}
173impl CompleteResponse for r::GenerateContentResponseBody {
174    type Event = rs::StreamEvent;
175    fn events(
176        self,
177        namespace: IdNamespace,
178        limits: EventLimits,
179    ) -> Result<Converted<Vec<Self::Event>>, TransformError> {
180        for item in &self.output {
181            let id = match item {
182                r::ResponseOutputItem::FunctionCall(v) => &v.id,
183                r::ResponseOutputItem::CustomToolCall(v) => &v.id,
184                _ => continue,
185            };
186            if id.as_ref().is_none_or(|v| v.is_empty()) {
187                return Err(super::missing(
188                    "complete Responses output needs persisted item IDs before stream synthesis",
189                ));
190            }
191        }
192        native::responses::synthesize_responses_stream(
193            self,
194            &mut IdentityFlow::new(namespace),
195            native::responses::ResponsesStreamLimits {
196                max_bytes: limits.max_bytes,
197
198                max_text_bytes: limits.max_bytes,
199                max_json_bytes: limits.max_bytes,
200            },
201        )
202    }
203}
gproxy_protocol/adapt/generate/stream/synthesize.rs

gproxy_protocol/adapt/generate/stream/
synthesize.rs
