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 data = codec::encode_json(&event, limits.codec).map_err(encoding)?;
206        let frame = SseEvent {
207            event: Some(event_name(&event).into()),
208            id: None,
209            data: String::from_utf8(data.to_vec())
210                .map_err(|e| TransformError::invalid_result("guardian.sse", e.to_string()))?,
211            retry: None,
212        };
213        chunks.push(encoder.event(&frame).map_err(encoding)?);
214    }
215    for name in [
216        http::header::CONTENT_LENGTH,
217        http::header::CONTENT_ENCODING,
218        http::header::TRANSFER_ENCODING,
219    ] {
220        headers.remove(name);
221    }
222    headers.insert(
223        http::header::CONTENT_TYPE,
224        http::HeaderValue::from_static("text/event-stream"),
225    );
226    let body = HttpBody::Stream(Box::pin(futures_util::stream::iter(
227        chunks.into_iter().map(Ok),
228    )));
229    *flow = ids;
230    Ok(GuardianStreamInvocation::Success {
231        result: actual,
232        response: WireResponse {
233            status,
234            headers,
235            body,
236        },
237        report,
238    })
239}
240
241fn collapse_text(
242    response: &mut r::GenerateContentResponseBody,
243    text: String,
244    report: &mut Report,
245) -> Result<(), TransformError> {
246    let mut found = false;
247    let mut changed = false;
248    let mut output = Vec::new();
249    for item in std::mem::take(&mut response.output) {
250        match item {
251            r::ResponseOutputItem::Message(mut message) => {
252                if found {
253                    changed = true;
254                    continue;
255                }
256                found = true;
257                let exact = message.content.len() == 1
258                    && matches!(&message.content[0],i::OutputContent::Text(v) if v.text==text);
259                if !exact {
260                    changed = true;
261                    message.content = vec![i::OutputContent::Text(
262                        i::ResponseOutputText::builder(
263                            i::ResponseOutputTextType::ResponseOutputText,
264                            text.clone(),
265                            Vec::new(),
266                            Vec::new(),
267                        )
268                        .build(),
269                    )];
270                }
271                output.push(r::ResponseOutputItem::Message(message));
272            }
273            item => output.push(item),
274        }
275    }
276    if !found {
277        return Err(TransformError::invalid_result(
278            "guardian.output",
279            "missing assistant text message",
280        ));
281    }
282    response.output = output;
283    if response.output_text.is_some() {
284        response.output_text = Some(Some(text));
285    }
286    if changed {
287        report.changed("guardian.output_text","joined validated text into one complete client text event; displaced per-part annotations/logprobs are omitted");
288    }
289    Ok(())
290}
291
292fn encoding(error: crate::codec::CodecError) -> TransformError {
293    TransformError::new(
294        if error.kind() == crate::codec::CodecErrorKind::Limit {
295            TransformErrorKind::Limit
296        } else {
297            TransformErrorKind::InvalidResult
298        },
299        "guardian.sse",
300        error.to_string(),
301    )
302}
303
304fn event_name(event: &s::StreamEvent) -> &'static str {
305    match event {
306        s::StreamEvent::Created(_) => "response.created",
307        s::StreamEvent::Queued(_) => "response.queued",
308        s::StreamEvent::InProgress(_) => "response.in_progress",
309        s::StreamEvent::Completed(_) => "response.completed",
310        s::StreamEvent::Failed(_) => "response.failed",
311        s::StreamEvent::Incomplete(_) => "response.incomplete",
312        s::StreamEvent::OutputItemAdded(_) => "response.output_item.added",
313        s::StreamEvent::OutputItemDone(_) => "response.output_item.done",
314        s::StreamEvent::ContentPartAdded(_) => "response.content_part.added",
315        s::StreamEvent::ContentPartDone(_) => "response.content_part.done",
316        s::StreamEvent::OutputTextDelta(_) => "response.output_text.delta",
317        s::StreamEvent::OutputTextDone(_) => "response.output_text.done",
318        s::StreamEvent::OutputTextAnnotationAdded(_) => "response.output_text.annotation.added",
319        s::StreamEvent::RefusalDelta(_) => "response.refusal.delta",
320        s::StreamEvent::RefusalDone(_) => "response.refusal.done",
321        s::StreamEvent::ReasoningTextDelta(_) => "response.reasoning_text.delta",
322        s::StreamEvent::ReasoningTextDone(_) => "response.reasoning_text.done",
323        s::StreamEvent::ReasoningSummaryPartAdded(_) => "response.reasoning_summary_part.added",
324        s::StreamEvent::ReasoningSummaryPartDone(_) => "response.reasoning_summary_part.done",
325        s::StreamEvent::ReasoningSummaryTextDelta(_) => "response.reasoning_summary_text.delta",
326        s::StreamEvent::ReasoningSummaryTextDone(_) => "response.reasoning_summary_text.done",
327        s::StreamEvent::FunctionCallArgumentsDelta(_) => "response.function_call_arguments.delta",
328        s::StreamEvent::FunctionCallArgumentsDone(_) => "response.function_call_arguments.done",
329        s::StreamEvent::CustomToolInputDelta(_) => "response.custom_tool_call_input.delta",
330        s::StreamEvent::CustomToolInputDone(_) => "response.custom_tool_call_input.done",
331        s::StreamEvent::CodeInterpreterCodeDelta(_) => "response.code_interpreter_call_code.delta",
332        s::StreamEvent::CodeInterpreterCodeDone(_) => "response.code_interpreter_call_code.done",
333        s::StreamEvent::AudioDelta(_) => "response.audio.delta",
334        s::StreamEvent::AudioDone(_) => "response.audio.done",
335        s::StreamEvent::AudioTranscriptDelta(_) => "response.audio.transcript.delta",
336        s::StreamEvent::AudioTranscriptDone(_) => "response.audio.transcript.done",
337        s::StreamEvent::ImagePartial(_) => "response.image_generation_call.partial_image",
338        s::StreamEvent::ImageCall(_) => "response.image_generation_call.in_progress",
339        s::StreamEvent::ImageGenerating(_) => "response.image_generation_call.generating",
340        s::StreamEvent::ImageCompleted(_) => "response.image_generation_call.completed",
341        s::StreamEvent::CodeInterpreterInProgress(_) => {
342            "response.code_interpreter_call.in_progress"
343        }
344        s::StreamEvent::CodeInterpreterInterpreting(_) => {
345            "response.code_interpreter_call.interpreting"
346        }
347        s::StreamEvent::CodeInterpreterCompleted(_) => "response.code_interpreter_call.completed",
348        s::StreamEvent::FileSearchInProgress(_) => "response.file_search_call.in_progress",
349        s::StreamEvent::FileSearchSearching(_) => "response.file_search_call.searching",
350        s::StreamEvent::FileSearchCompleted(_) => "response.file_search_call.completed",
351        s::StreamEvent::WebSearchInProgress(_) => "response.web_search_call.in_progress",
352        s::StreamEvent::WebSearchSearching(_) => "response.web_search_call.searching",
353        s::StreamEvent::WebSearchCompleted(_) => "response.web_search_call.completed",
354        s::StreamEvent::McpArgumentsDelta(_) => "response.mcp_call_arguments.delta",
355        s::StreamEvent::McpArgumentsDone(_) => "response.mcp_call_arguments.done",
356        s::StreamEvent::McpInProgress(_) => "response.mcp_call.in_progress",
357        s::StreamEvent::McpCompleted(_) => "response.mcp_call.completed",
358        s::StreamEvent::McpFailed(_) => "response.mcp_call.failed",
359        s::StreamEvent::McpListToolsInProgress(_) => "response.mcp_list_tools.in_progress",
360        s::StreamEvent::McpListToolsCompleted(_) => "response.mcp_list_tools.completed",
361        s::StreamEvent::McpListToolsFailed(_) => "response.mcp_list_tools.failed",
362        s::StreamEvent::Error(_) => "error",
363    }
364}
gproxy_protocol/adapt/guardian/stream.rs

gproxy_protocol/adapt/guardian/
stream.rs
