gproxy_transform/transform/dispatch/
mod.rs

1//! Bytes-level dispatch from a resolved [`TransformPair`] to its typed pair
2//! functions. The private `content` module holds content-generation pairs;
3//! `other` holds count_tokens/models/embeddings/images/compact.
4//! Streaming is wired for content pairs only.
5
6mod content;
7mod other;
8
9use serde::Serialize;
10use serde::de::DeserializeOwned;
11
12use super::{TransformContext, TransformError, TransformOutput, TransformPair};
13
14/// Whether the bytes dispatch has arms for this pair.
15pub fn is_wired(pair: TransformPair) -> bool {
16    content::is_content(pair) || other::is_wired(pair)
17}
18
19/// Convert a request body (inbound wire JSON → upstream wire JSON).
20pub fn request_bytes(
21    pair: TransformPair,
22    ctx: &TransformContext,
23    body: &[u8],
24) -> Result<Vec<u8>, TransformError> {
25    validate_pair(pair, ctx)?;
26    ctx.scope(|| {
27        if content::is_content(pair) {
28            content::request_bytes(pair, ctx, body)
29        } else {
30            other::request_bytes(pair, ctx, body)
31        }
32    })
33}
34
35/// Convert a request and return non-fatal semantic-loss diagnostics.
36pub fn request_bytes_detailed(
37    pair: TransformPair,
38    ctx: &TransformContext,
39    body: &[u8],
40) -> Result<TransformOutput<Vec<u8>>, TransformError> {
41    let scoped = ctx.isolated();
42    let value = request_bytes(pair, &scoped, body)?;
43    Ok(TransformOutput::new(value, scoped.take_diagnostics()))
44}
45
46/// Convert a response body (upstream wire JSON → inbound wire JSON). The pair
47/// here is the REVERSE pair (`resolve(upstream_key, inbound_key)`).
48pub fn response_bytes(
49    pair: TransformPair,
50    ctx: &TransformContext,
51    body: &[u8],
52) -> Result<Vec<u8>, TransformError> {
53    validate_pair(pair, ctx)?;
54    ctx.scope(|| {
55        if content::is_content(pair) {
56            content::response_bytes(pair, ctx, body)
57        } else {
58            other::response_bytes(pair, ctx, body)
59        }
60    })
61}
62
63/// Convert a response and return non-fatal semantic-loss diagnostics.
64pub fn response_bytes_detailed(
65    pair: TransformPair,
66    ctx: &TransformContext,
67    body: &[u8],
68) -> Result<TransformOutput<Vec<u8>>, TransformError> {
69    let scoped = ctx.isolated();
70    let value = response_bytes(pair, &scoped, body)?;
71    Ok(TransformOutput::new(value, scoped.take_diagnostics()))
72}
73
74/// One converted stream event: pre-encoded inbound frame payload, or the
75/// typed Responses event when the inbound side runs the aggregation state
76/// machine.
77pub enum StreamEventOut {
78    Encoded { event: Option<String>, data: String },
79    Responses(Box<crate::protocol::openai::ResponseStreamEvent>),
80}
81
82/// Stateful `0..N` stream-event converter for one resolved pair.
83///
84/// Create one converter per upstream response and retain it until
85/// [`finish`](Self::finish). This preserves pair-specific state such as tool
86/// call arguments split across multiple frames.
87pub struct StreamConverter {
88    inner: content::ContentStreamConverter,
89}
90
91impl StreamConverter {
92    pub fn new(pair: TransformPair, ctx: TransformContext) -> Result<Self, TransformError> {
93        validate_pair(pair, &ctx)?;
94        Ok(Self {
95            inner: content::ContentStreamConverter::new(pair, ctx)?,
96        })
97    }
98
99    /// Convert one decoded upstream event into zero or more inbound events.
100    pub fn push(&mut self, data: &str) -> Result<Vec<StreamEventOut>, TransformError> {
101        Ok(self.push_detailed(data)?.value)
102    }
103
104    /// Convert one event and return semantic-loss diagnostics produced by it.
105    pub fn push_detailed(
106        &mut self,
107        data: &str,
108    ) -> Result<TransformOutput<Vec<StreamEventOut>>, TransformError> {
109        let value = self.inner.push(data)?;
110        Ok(TransformOutput::new(value, self.inner.take_diagnostics()))
111    }
112
113    /// Flush pair-specific state into zero or more final inbound events.
114    pub fn finish(&mut self) -> Result<Vec<StreamEventOut>, TransformError> {
115        Ok(self.finish_detailed()?.value)
116    }
117
118    /// Flush state and return any final semantic-loss diagnostics.
119    pub fn finish_detailed(
120        &mut self,
121    ) -> Result<TransformOutput<Vec<StreamEventOut>>, TransformError> {
122        let value = self.inner.finish()?;
123        Ok(TransformOutput::new(value, self.inner.take_diagnostics()))
124    }
125}
126
127/// Convert one decoded stream event (upstream wire JSON text → inbound event).
128/// Same reverse-pair convention as [`response_bytes`]. Only content-generation
129/// pairs stream; the other groups are buffered. This convenience call retains
130/// no cross-frame state; use [`StreamConverter`] for an actual response stream.
131pub fn stream_event(
132    pair: TransformPair,
133    ctx: &TransformContext,
134    data: &str,
135) -> Result<Vec<StreamEventOut>, TransformError> {
136    if content::is_content(pair) {
137        let mut converter = StreamConverter::new(pair, ctx.clone())?;
138        converter.push(data)
139    } else {
140        Err(not_wired(pair))
141    }
142}
143
144fn validate_pair(pair: TransformPair, ctx: &TransformContext) -> Result<(), TransformError> {
145    let resolved = super::resolve(ctx.source, ctx.target)?;
146    if resolved == pair {
147        Ok(())
148    } else {
149        Err(TransformError::InvalidInput {
150            reason: format!(
151                "transform pair {pair:?} does not match context {:?} -> {:?} (resolved {resolved:?})",
152                ctx.source, ctx.target
153            ),
154        })
155    }
156}
157
158fn run<S, T>(
159    f: impl Fn(S, &TransformContext) -> Result<T, TransformError>,
160    ctx: &TransformContext,
161    body: &[u8],
162) -> Result<Vec<u8>, TransformError>
163where
164    S: DeserializeOwned,
165    T: Serialize,
166{
167    let input: S = serde_json::from_slice(body).map_err(|e| TransformError::InvalidInput {
168        reason: format!("decode source body: {e}"),
169    })?;
170    let out = f(input, ctx)?;
171    serde_json::to_vec(&out).map_err(|e| TransformError::Serialization {
172        reason: e.to_string(),
173    })
174}
175
176/// [`run`] for infallible pair functions (plain return, no `Result`).
177fn run_ok<S, T>(
178    f: impl Fn(S, &TransformContext) -> T,
179    ctx: &TransformContext,
180    body: &[u8],
181) -> Result<Vec<u8>, TransformError>
182where
183    S: DeserializeOwned,
184    T: Serialize,
185{
186    run(
187        |input, ctx| Ok::<_, TransformError>(f(input, ctx)),
188        ctx,
189        body,
190    )
191}
192
193fn not_wired(pair: TransformPair) -> TransformError {
194    TransformError::InvalidInput {
195        reason: format!("bytes dispatch not wired for {pair:?}"),
196    }
197}
198
199#[cfg(test)]
200mod tests {
201    use serde_json::Value;
202
203    use super::*;
204    use crate::protocol::{ContentGenerationKind, Operation, OperationKey, Provider};
205    use crate::transform::TransformDiagnosticKind;
206
207    #[test]
208    fn claude_to_openai_chat_request_roundtrip() {
209        let source = OperationKey::content_generation(
210            Operation::GenerateContent,
211            ContentGenerationKind::ClaudeMessages,
212        );
213        let target = OperationKey::content_generation(
214            Operation::GenerateContent,
215            ContentGenerationKind::OpenAiChatCompletions,
216        );
217        let ctx = TransformContext::new(source, target);
218        let body = br#"{"model":"m","max_tokens":16,"messages":[{"role":"user","content":"hi"}]}"#;
219        let out = request_bytes(TransformPair::ClaudeMessagesToOpenAiChat, &ctx, body).unwrap();
220        let v: Value = serde_json::from_slice(&out).unwrap();
221        assert_eq!(v["messages"][0]["role"], "user");
222        assert!(v.get("max_tokens").is_some() || v.get("max_completion_tokens").is_some());
223    }
224
225    #[test]
226    fn detailed_request_returns_structured_semantic_loss() {
227        let source = OperationKey::content_generation(
228            Operation::GenerateContent,
229            ContentGenerationKind::OpenAiChatCompletions,
230        );
231        let target = OperationKey::content_generation(
232            Operation::GenerateContent,
233            ContentGenerationKind::ClaudeMessages,
234        );
235        let pair = crate::transform::resolve(source, target).unwrap();
236        let ctx = TransformContext::new(source, target);
237        let body = br#"{
238            "model":"m",
239            "messages":[{"role":"user","content":[{
240                "type":"text",
241                "text":"",
242                "prompt_cache_breakpoint":{"mode":"explicit"}
243            }]}]
244        }"#;
245
246        let output = request_bytes_detailed(pair, &ctx, body).unwrap();
247
248        assert_eq!(output.diagnostics.len(), 1);
249        assert_eq!(
250            output.diagnostics[0].kind,
251            TransformDiagnosticKind::LossyField
252        );
253        assert_eq!(
254            output.diagnostics[0].field,
255            "messages[].content[].text.prompt_cache_breakpoint"
256        );
257        assert!(ctx.diagnostics().is_empty(), "detailed calls are isolated");
258    }
259
260    #[test]
261    fn openai_responses_to_websocket_request_roundtrip() {
262        let source = OperationKey::content_generation(
263            Operation::GenerateContent,
264            ContentGenerationKind::OpenAiResponses,
265        );
266        let target = OperationKey::content_generation(
267            Operation::GenerateContent,
268            ContentGenerationKind::OpenAiResponsesWebSocket,
269        );
270        let pair = crate::transform::resolve(source, target).unwrap();
271        assert!(is_wired(pair));
272
273        let ctx = TransformContext::new(source, target);
274        let body = br#"{"model":"m","input":"hi","stream":true}"#;
275        let out = request_bytes(pair, &ctx, body).unwrap();
276        let v: Value = serde_json::from_slice(&out).unwrap();
277
278        assert_eq!(v["type"], "response.create");
279        assert_eq!(v["model"], "m");
280        assert_eq!(v["stream"], true);
281    }
282
283    #[test]
284    fn claude_to_openai_responses_websocket_request_roundtrip() {
285        let source = OperationKey::content_generation(
286            Operation::GenerateContent,
287            ContentGenerationKind::ClaudeMessages,
288        );
289        let target = OperationKey::content_generation(
290            Operation::GenerateContent,
291            ContentGenerationKind::OpenAiResponsesWebSocket,
292        );
293        let pair = crate::transform::resolve(source, target).unwrap();
294        assert!(is_wired(pair));
295
296        let ctx = TransformContext::new(source, target);
297        let body = br#"{"model":"m","max_tokens":16,"messages":[{"role":"user","content":"hi"}]}"#;
298        let out = request_bytes(pair, &ctx, body).unwrap();
299        let v: Value = serde_json::from_slice(&out).unwrap();
300
301        assert_eq!(v["type"], "response.create");
302        assert_eq!(v["model"], "m");
303        assert!(v.get("input").is_some());
304    }
305
306    #[test]
307    fn claude_to_openai_count_tokens_request_roundtrip() {
308        let source = OperationKey::provider(Operation::CountTokens, Provider::Claude);
309        let target = OperationKey::provider(Operation::CountTokens, Provider::OpenAi);
310        let ctx = TransformContext::new(source, target);
311        let body = br#"{"model":"m","messages":[{"role":"user","content":"hi"}]}"#;
312        let out = request_bytes(TransformPair::ClaudeToOpenAiCountTokens, &ctx, body).unwrap();
313        let v: Value = serde_json::from_slice(&out).unwrap();
314        assert_eq!(v["model"], "m");
315        assert!(v.get("input").is_some());
316    }
317
318    #[test]
319    fn compact_to_responses_is_resolved_and_wired() {
320        let source = OperationKey::provider(Operation::CompactContent, Provider::OpenAi);
321        let target = OperationKey::content_generation(
322            Operation::GenerateContent,
323            ContentGenerationKind::OpenAiResponses,
324        );
325        let pair = crate::transform::resolve(source, target).unwrap();
326        assert_eq!(pair, TransformPair::OpenAiCompactToOpenAiResponses);
327        assert!(is_wired(pair));
328
329        let ctx = TransformContext::new(source, target);
330        let body = br#"{"model":"m","input":"summarize this"}"#;
331        let out = request_bytes(pair, &ctx, body).unwrap();
332        let value: Value = serde_json::from_slice(&out).unwrap();
333        assert_eq!(value["model"], "m");
334        assert!(value.get("input").is_some());
335    }
336
337    #[test]
338    fn openai_to_claude_models_list_response_roundtrip() {
339        let source = OperationKey::provider(Operation::ListModels, Provider::OpenAi);
340        let target = OperationKey::provider(Operation::ListModels, Provider::Claude);
341        let ctx = TransformContext::new(source, target);
342        let body = br#"{"object":"list","data":[{"id":"gpt-x","created":1,"object":"model","owned_by":"openai"}]}"#;
343        let out = response_bytes(TransformPair::OpenAiToClaudeModels, &ctx, body).unwrap();
344        let v: Value = serde_json::from_slice(&out).unwrap();
345        assert_eq!(v["data"][0]["id"], "gpt-x");
346    }
347}
gproxy_transform/transform/dispatch/mod.rs

gproxy_transform/transform/dispatch/
mod.rs
