gproxy_protocol/adapt/guardian/
stream.rs

1//! Restore the Guardian client's Responses SSE contract after validation.
2
3use super::{GuardianClassifyInvocation, GuardianNativeResponse, GuardianReviewInvocation};
4use crate::{
5    HttpBody, WireResponse,
6    codec::{self, CodecLimits, SseEncoder, SseEvent},
7    transform::{
8        Converted, Report, TransformError, TransformErrorKind,
9        generate::{
10            chat_responses::{self, ResponsesResponseContext},
11            claude_responses::{self, ClaudeResponseContext},
12            gemini_responses::{self, GeminiResponseContext},
13            stream::responses::{self, ResponsesStreamLimits},
14        },
15        guardian::{self, GuardianExtraction, GuardianOperation},
16        identity::{IdentityFlow, TargetIdPolicy},
17    },
18    wire::{
19        DeclaredFields,
20        openai::responses::{input as i, response as r, stream as s},
21    },
22};
23
24/// Concrete invocation facts used by the existing direct response pair. Claude
25/// and Gemini need a factual timestamp, and missing usage details need actual
26/// host facts. No timestamp or zero counter is supplied by this adapter.
27#[derive(Debug)]
28#[allow(clippy::large_enum_variant)]
29pub enum GuardianStreamContext {
30    Responses,
31    Chat(ResponsesResponseContext),
32    Claude(ClaudeResponseContext),
33    Gemini(GeminiResponseContext),
34}
35
36#[derive(Debug)]
37pub enum GuardianStreamInvocation {
38    Success {
39        result: GuardianExtraction,
40        response: WireResponse<HttpBody>,
41        report: Report,
42    },
43    Rejected(WireResponse<HttpBody>),
44    Invalid {
45        error: TransformError,
46        native: Box<GuardianNativeResponse>,
47    },
48}
49
50#[derive(Debug, Clone, Copy)]
51pub struct GuardianStreamLimits {
52    pub codec: CodecLimits,
53    pub stream: ResponsesStreamLimits,
54}
55
56impl GuardianReviewInvocation {
57    pub fn into_stream(
58        self,
59        context: GuardianStreamContext,
60        flow: &mut IdentityFlow,
61        limits: GuardianStreamLimits,
62    ) -> Result<GuardianStreamInvocation, TransformError> {
63        match self {
64            Self::Rejected(response) => Ok(GuardianStreamInvocation::Rejected(response)),
65            Self::Invalid { error, native } => Ok(GuardianStreamInvocation::Invalid {
66                error,
67                native: Box::new(native),
68            }),
69            Self::Success { result, native } => build(
70                native,
71                context,
72                GuardianExtraction::Review(result),
73                flow,
74                limits,
75            ),
76        }
77    }
78}
79
80impl GuardianClassifyInvocation {
81    pub fn into_stream(
82        self,
83        context: GuardianStreamContext,
84        flow: &mut IdentityFlow,
85        limits: GuardianStreamLimits,
86    ) -> Result<GuardianStreamInvocation, TransformError> {
87        match self {
88            Self::Rejected(response) => Ok(GuardianStreamInvocation::Rejected(response)),
89            Self::Invalid { error, native } => Ok(GuardianStreamInvocation::Invalid {
90                error,
91                native: Box::new(native),
92            }),
93            Self::Success { result, native } => build(
94                native,
95                context,
96                GuardianExtraction::Classify(result),
97                flow,
98                limits,
99            ),
100        }
101    }
102}
103
104fn clean<T: DeclaredFields + serde::Serialize>(
105    body: T,
106    limits: CodecLimits,
107) -> Result<T, TransformError> {
108    let body = body.into_declared();
109    codec::encode_json(&body, limits).map_err(encoding)?;
110    Ok(body)
111}
112
113fn build(
114    native: GuardianNativeResponse,
115    context: GuardianStreamContext,
116    expected: GuardianExtraction,
117    flow: &mut IdentityFlow,
118    limits: GuardianStreamLimits,
119) -> Result<GuardianStreamInvocation, TransformError> {
120    let mut ids = flow.clone();
121    let policy = TargetIdPolicy::new(crate::Dialect::OpenAi);
122    let (status, mut headers, original, converted) = match (native, context) {
123        (GuardianNativeResponse::Responses(response), GuardianStreamContext::Responses) => {
124            let body = clean(response.body, limits.codec)?;
125            let text = crate::transform::memory::responses_text(body.clone())?;
126            (
127                response.status,
128                response.headers,
129                text,
130                Converted {
131                    value: body,
132                    report: Report::default(),
133                },
134            )
135        }
136        (GuardianNativeResponse::Chat(response), GuardianStreamContext::Chat(context)) => {
137            let body = clean(response.body, limits.codec)?;
138            let text = crate::transform::memory::chat_text(body.clone())?;
139            (
140                response.status,
141                response.headers,
142                text,
143                chat_responses::chat_to_responses_response(body, context, &mut ids, &policy)?,
144            )
145        }
146        (GuardianNativeResponse::Claude(response), GuardianStreamContext::Claude(context)) => {
147            let body = clean(response.body, limits.codec)?;
148            let text = crate::transform::memory::claude_text(body.clone())?;
149            (
150                response.status,
151                response.headers,
152                text,
153                claude_responses::claude_to_responses_response(body, context, &mut ids, &policy)?,
154            )
155        }
156        (GuardianNativeResponse::Gemini(response), GuardianStreamContext::Gemini(context)) => {
157            let body = clean(response.body, limits.codec)?;
158            let text = crate::transform::memory::gemini_text(body.clone())?;
159            (
160                response.status,
161                response.headers,
162                text,
163                gemini_responses::gemini_to_responses_response(body, context, &mut ids, &policy)?,
164            )
165        }
166        _ => {
167            return Err(TransformError::shape(
168                "guardian.stream.context",
169                "response dialect and concrete context must match",
170            ));
171        }
172    };
173    if !status.is_success() {
174        return Err(TransformError::invalid_result(
175            "guardian.status",
176            "successful result contains non-success HTTP status",
177        ));
178    }
179    let mut response = clean(converted.value, limits.codec)?;
180    let mut report = converted.report;
181    let operation = match &expected {
182        GuardianExtraction::Review(_) => GuardianOperation::Review,
183        GuardianExtraction::Classify(_) => GuardianOperation::Classify,
184    };
185    let text = match operation {
186        GuardianOperation::Review => original,
187        GuardianOperation::Classify => original.trim().to_owned(),
188    };
189    // Generic response pairs may represent separate text blocks with separators.
190    // Guardian consumes one formal JSON document or one full label, so retain
191    // the actual target text concatenation before applying its client projection.
192    collapse_text(&mut response, text, &mut report)?;
193    let actual = guardian::extract_responses(response.clone(), operation)?;
194    if actual != expected {
195        return Err(TransformError::invalid_result(
196            "guardian.result",
197            "validated label/result contradicts native response text",
198        ));
199    }
200    let synthesized = responses::synthesize_responses_stream(response, &mut ids, limits.stream)?;
201    report.diagnostics.extend(synthesized.report.diagnostics);
202    let mut encoder = SseEncoder::new(limits.codec);
203    let mut chunks = Vec::new();
204    for event in synthesized.value {
205        let Some(name) = event_name(&event) else {
206            continue;
207        };
208        let data = codec::encode_json(&event, limits.codec).map_err(encoding)?;
209        let frame = SseEvent {
210            event: Some(name.into()),
211            id: None,
212            data: String::from_utf8(data.to_vec())
213                .map_err(|e| TransformError::invalid_result("guardian.sse", e.to_string()))?,
214            retry: None,
215        };
216        chunks.push(encoder.event(&frame).map_err(encoding)?);
217    }
218    for name in [
219        http::header::CONTENT_LENGTH,
220        http::header::CONTENT_ENCODING,
221        http::header::TRANSFER_ENCODING,
222    ] {
223        headers.remove(name);
224    }
225    headers.insert(
226        http::header::CONTENT_TYPE,
227        http::HeaderValue::from_static("text/event-stream"),
228    );
229    let body = HttpBody::Stream(Box::pin(futures_util::stream::iter(
230        chunks.into_iter().map(Ok),
231    )));
232    *flow = ids;
233    Ok(GuardianStreamInvocation::Success {
234        result: actual,
235        response: WireResponse {
236            status,
237            headers,
238            body,
239        },
240        report,
241    })
242}
243
244fn collapse_text(
245    response: &mut r::GenerateContentResponseBody,
246    text: String,
247    report: &mut Report,
248) -> Result<(), TransformError> {
249    let mut found = false;
250    let mut changed = false;
251    let mut output = Vec::new();
252    for item in std::mem::take(&mut response.output) {
253        match item {
254            r::ResponseOutputItem::Message(mut message) => {
255                if found {
256                    changed = true;
257                    continue;
258                }
259                found = true;
260                let exact = message.content.len() == 1
261                    && matches!(&message.content[0],i::OutputContent::Text(v) if v.text==text);
262                if !exact {
263                    changed = true;
264                    message.content = vec![i::OutputContent::Text(
265                        i::ResponseOutputText::builder(
266                            i::ResponseOutputTextType::ResponseOutputText,
267                            text.clone(),
268                            Vec::new(),
269                            Vec::new(),
270                        )
271                        .build(),
272                    )];
273                }
274                output.push(r::ResponseOutputItem::Message(message));
275            }
276            item => output.push(item),
277        }
278    }
279    if !found {
280        return Err(TransformError::invalid_result(
281            "guardian.output",
282            "missing assistant text message",
283        ));
284    }
285    response.output = output;
286    if response.output_text.is_some() {
287        response.output_text = Some(Some(text));
288    }
289    if changed {
290        report.changed("guardian.output_text","joined validated text into one complete client text event; displaced per-part annotations/logprobs are omitted");
291    }
292    Ok(())
293}
294
295fn encoding(error: crate::codec::CodecError) -> TransformError {
296    TransformError::new(
297        if error.kind() == crate::codec::CodecErrorKind::Limit {
298            TransformErrorKind::Limit
299        } else {
300            TransformErrorKind::InvalidResult
301        },
302        "guardian.sse",
303        error.to_string(),
304    )
305}
306
307fn event_name(event: &s::StreamEvent) -> Option<&'static str> {
308    Some(match event {
309        s::StreamEvent::Unknown => return None,
310        s::StreamEvent::Keepalive => "keepalive",
311        s::StreamEvent::Metadata(_) => "response.metadata",
312        s::StreamEvent::Created(_) => "response.created",
313        s::StreamEvent::Queued(_) => "response.queued",
314        s::StreamEvent::InProgress(_) => "response.in_progress",
315        s::StreamEvent::Completed(_) => "response.completed",
316        s::StreamEvent::Failed(_) => "response.failed",
317        s::StreamEvent::Incomplete(_) => "response.incomplete",
318        s::StreamEvent::OutputItemAdded(_) => "response.output_item.added",
319        s::StreamEvent::OutputItemDone(_) => "response.output_item.done",
320        s::StreamEvent::ContentPartAdded(_) => "response.content_part.added",
321        s::StreamEvent::ContentPartDone(_) => "response.content_part.done",
322        s::StreamEvent::OutputTextDelta(_) => "response.output_text.delta",
323        s::StreamEvent::OutputTextDone(_) => "response.output_text.done",
324        s::StreamEvent::OutputTextAnnotationAdded(_) => "response.output_text.annotation.added",
325        s::StreamEvent::RefusalDelta(_) => "response.refusal.delta",
326        s::StreamEvent::RefusalDone(_) => "response.refusal.done",
327        s::StreamEvent::ReasoningTextDelta(_) => "response.reasoning_text.delta",
328        s::StreamEvent::ReasoningTextDone(_) => "response.reasoning_text.done",
329        s::StreamEvent::ReasoningSummaryPartAdded(_) => "response.reasoning_summary_part.added",
330        s::StreamEvent::ReasoningSummaryPartDone(_) => "response.reasoning_summary_part.done",
331        s::StreamEvent::ReasoningSummaryTextDelta(_) => "response.reasoning_summary_text.delta",
332        s::StreamEvent::ReasoningSummaryTextDone(_) => "response.reasoning_summary_text.done",
333        s::StreamEvent::FunctionCallArgumentsDelta(_) => "response.function_call_arguments.delta",
334        s::StreamEvent::FunctionCallArgumentsDone(_) => "response.function_call_arguments.done",
335        s::StreamEvent::CustomToolInputDelta(_) => "response.custom_tool_call_input.delta",
336        s::StreamEvent::CustomToolInputDone(_) => "response.custom_tool_call_input.done",
337        s::StreamEvent::CodeInterpreterCodeDelta(_) => "response.code_interpreter_call_code.delta",
338        s::StreamEvent::CodeInterpreterCodeDone(_) => "response.code_interpreter_call_code.done",
339        s::StreamEvent::AudioDelta(_) => "response.audio.delta",
340        s::StreamEvent::AudioDone(_) => "response.audio.done",
341        s::StreamEvent::AudioTranscriptDelta(_) => "response.audio.transcript.delta",
342        s::StreamEvent::AudioTranscriptDone(_) => "response.audio.transcript.done",
343        s::StreamEvent::ImagePartial(_) => "response.image_generation_call.partial_image",
344        s::StreamEvent::ImageCall(_) => "response.image_generation_call.in_progress",
345        s::StreamEvent::ImageGenerating(_) => "response.image_generation_call.generating",
346        s::StreamEvent::ImageCompleted(_) => "response.image_generation_call.completed",
347        s::StreamEvent::CodeInterpreterInProgress(_) => {
348            "response.code_interpreter_call.in_progress"
349        }
350        s::StreamEvent::CodeInterpreterInterpreting(_) => {
351            "response.code_interpreter_call.interpreting"
352        }
353        s::StreamEvent::CodeInterpreterCompleted(_) => "response.code_interpreter_call.completed",
354        s::StreamEvent::FileSearchInProgress(_) => "response.file_search_call.in_progress",
355        s::StreamEvent::FileSearchSearching(_) => "response.file_search_call.searching",
356        s::StreamEvent::FileSearchCompleted(_) => "response.file_search_call.completed",
357        s::StreamEvent::WebSearchInProgress(_) => "response.web_search_call.in_progress",
358        s::StreamEvent::WebSearchSearching(_) => "response.web_search_call.searching",
359        s::StreamEvent::WebSearchCompleted(_) => "response.web_search_call.completed",
360        s::StreamEvent::McpArgumentsDelta(_) => "response.mcp_call_arguments.delta",
361        s::StreamEvent::McpArgumentsDone(_) => "response.mcp_call_arguments.done",
362        s::StreamEvent::McpInProgress(_) => "response.mcp_call.in_progress",
363        s::StreamEvent::McpCompleted(_) => "response.mcp_call.completed",
364        s::StreamEvent::McpFailed(_) => "response.mcp_call.failed",
365        s::StreamEvent::McpListToolsInProgress(_) => "response.mcp_list_tools.in_progress",
366        s::StreamEvent::McpListToolsCompleted(_) => "response.mcp_list_tools.completed",
367        s::StreamEvent::McpListToolsFailed(_) => "response.mcp_list_tools.failed",
368        s::StreamEvent::Error(_) => "error",
369    })
370}
gproxy_protocol/adapt/guardian/stream.rs

gproxy_protocol/adapt/guardian/
stream.rs
