gproxy_protocol/adapt/generate/stream/prepare/
claude_responses.rs

1use std::future::Future;
2
3use super::super::super::{
4    Endpoint, GenerationIdentity, GenerationResources, GenerationStateAccess,
5    claude_responses::{ClaudeViaResponses, ResponsesViaClaude},
6};
7use super::super::{ResponsesHistoryCache, StreamInvocation, StreamSettings, StreamTarget};
8use super::{RequestMode, start};
9use crate::{
10    capability::{ResourceAccess, StateStore},
11    transform::{
12        TransformError,
13        generate::{claude_responses as pair, claude_responses::stream as p},
14    },
15    wire::{DeclaredFields, claude::generate_content as c, openai::responses as r},
16};
17
18pub struct ResponsesViaClaudeStreamFacts {
19    pub request: pair::ClaudeRequestContext,
20    pub response: p::ClaudeToResponsesContext,
21}
22
23impl ClaudeViaResponses {
24    pub async fn prepare_stream<S: StateStore>(
25        input: c::GenerateContentRequestBody,
26        target: StreamTarget,
27        context: p::ResponsesToClaudeContext,
28        settings: StreamSettings,
29        state: &GenerationStateAccess<'_, S>,
30    ) -> Result<StreamInvocation<p::ResponsesToClaudeStream>, TransformError> {
31        Self::stream(
32            input,
33            target,
34            context,
35            settings,
36            state,
37            |input, endpoint, ids| Self::prepare_with_state(input, endpoint, ids, state),
38        )
39        .await
40    }
41    pub async fn prepare_stream_with_capabilities<S: StateStore, R: ResourceAccess>(
42        input: c::GenerateContentRequestBody,
43        target: StreamTarget,
44        context: p::ResponsesToClaudeContext,
45        settings: StreamSettings,
46        state: &GenerationStateAccess<'_, S>,
47        resources: &GenerationResources<'_, R>,
48    ) -> Result<StreamInvocation<p::ResponsesToClaudeStream>, TransformError> {
49        Self::stream(
50            input,
51            target,
52            context,
53            settings,
54            state,
55            |input, endpoint, ids| {
56                Self::prepare_with_capabilities(input, endpoint, ids, state, resources)
57            },
58        )
59        .await
60    }
61    async fn stream<S: StateStore, Fut: Future<Output = Result<Self, TransformError>>>(
62        input: c::GenerateContentRequestBody,
63        target: StreamTarget,
64        context: p::ResponsesToClaudeContext,
65        settings: StreamSettings,
66        state: &GenerationStateAccess<'_, S>,
67        prepare: impl FnOnce(c::GenerateContentRequestBody, Endpoint, GenerationIdentity) -> Fut,
68    ) -> Result<StreamInvocation<p::ResponsesToClaudeStream>, TransformError> {
69        let original = input.into_declared();
70        let prepared = prepare(
71            original.clone().buffered(),
72            target.endpoint.clone(),
73            target.identities.clone(),
74        )
75        .await?;
76        start(
77            original,
78            prepared,
79            target,
80            settings,
81            state,
82            |_, target, settings| {
83                p::ResponsesToClaudeStream::new_with_policy(
84                    context,
85                    target.identities.response.clone(),
86                    settings.events.into(),
87                    target.identities.response_policy.clone(),
88                )
89            },
90        )
91        .await
92    }
93}
94
95impl ResponsesViaClaude {
96    pub async fn prepare_stream<S: StateStore>(
97        input: r::GenerateContentRequestBody,
98        target: StreamTarget,
99        context: ResponsesViaClaudeStreamFacts,
100        settings: StreamSettings,
101        state: &GenerationStateAccess<'_, S>,
102    ) -> Result<StreamInvocation<p::ClaudeToResponsesStream>, TransformError> {
103        Self::stream(
104            input,
105            target,
106            context,
107            settings,
108            state,
109            None,
110            |input, endpoint, ids, request| {
111                Self::prepare_with_state(input, endpoint, ids, state, request)
112            },
113        )
114        .await
115    }
116    pub async fn prepare_stream_with_history_cache<S: StateStore>(
117        input: r::GenerateContentRequestBody,
118        target: StreamTarget,
119        context: ResponsesViaClaudeStreamFacts,
120        settings: StreamSettings,
121        state: &GenerationStateAccess<'_, S>,
122        history_cache: &ResponsesHistoryCache,
123    ) -> Result<StreamInvocation<p::ClaudeToResponsesStream>, TransformError> {
124        let cache = Some(history_cache);
125        Self::stream(
126            input,
127            target,
128            context,
129            settings,
130            state,
131            cache,
132            |input, endpoint, ids, request| {
133                Self::prepare_with_state(input, endpoint, ids, state, request)
134            },
135        )
136        .await
137    }
138    pub async fn prepare_stream_with_capabilities<S: StateStore, R: ResourceAccess>(
139        input: r::GenerateContentRequestBody,
140        target: StreamTarget,
141        context: ResponsesViaClaudeStreamFacts,
142        settings: StreamSettings,
143        state: &GenerationStateAccess<'_, S>,
144        resources: &GenerationResources<'_, R>,
145    ) -> Result<StreamInvocation<p::ClaudeToResponsesStream>, TransformError> {
146        Self::stream(
147            input,
148            target,
149            context,
150            settings,
151            state,
152            None,
153            |input, endpoint, ids, request| {
154                Self::prepare_with_capabilities(input, endpoint, ids, state, resources, request)
155            },
156        )
157        .await
158    }
159    pub async fn prepare_stream_with_capabilities_and_history_cache<
160        S: StateStore,
161        R: ResourceAccess,
162    >(
163        input: r::GenerateContentRequestBody,
164        target: StreamTarget,
165        context: ResponsesViaClaudeStreamFacts,
166        settings: StreamSettings,
167        state: &GenerationStateAccess<'_, S>,
168        resources: &GenerationResources<'_, R>,
169        history_cache: &ResponsesHistoryCache,
170    ) -> Result<StreamInvocation<p::ClaudeToResponsesStream>, TransformError> {
171        let cache = Some(history_cache);
172        Self::stream(
173            input,
174            target,
175            context,
176            settings,
177            state,
178            cache,
179            |input, endpoint, ids, request| {
180                Self::prepare_with_capabilities(input, endpoint, ids, state, resources, request)
181            },
182        )
183        .await
184    }
185    async fn stream<S: StateStore, Fut: Future<Output = Result<Self, TransformError>>>(
186        input: r::GenerateContentRequestBody,
187        target: StreamTarget,
188        mut context: ResponsesViaClaudeStreamFacts,
189        settings: StreamSettings,
190        state: &GenerationStateAccess<'_, S>,
191        history_cache: Option<&ResponsesHistoryCache>,
192        prepare: impl FnOnce(
193            r::GenerateContentRequestBody,
194            Endpoint,
195            GenerationIdentity,
196            pair::ClaudeRequestContext,
197        ) -> Fut,
198    ) -> Result<StreamInvocation<p::ClaudeToResponsesStream>, TransformError> {
199        let original = input.into_declared();
200        let (history, expanded) = super::super::history::History::prepare_with_cache(
201            &original,
202            state,
203            settings.codec,
204            history_cache,
205        )
206        .await?;
207        context.response.response.request = original.clone();
208        context.response.response.request.input = expanded.input.clone();
209        let prepared = prepare(
210            expanded.buffered(),
211            target.endpoint.clone(),
212            target.identities.clone(),
213            context.request,
214        )
215        .await?;
216        let mut invocation = start(
217            original,
218            prepared,
219            target,
220            settings,
221            state,
222            |_, target, settings| {
223                history.claim_response_id(&mut target.identities, crate::Dialect::Claude)?;
224                p::ClaudeToResponsesStream::new_with_policy(
225                    context.response,
226                    target.identities.response.clone(),
227                    settings.events.into(),
228                    target.identities.response_policy.clone(),
229                )
230            },
231        )
232        .await?;
233        history.bind(&invocation.binding, state)?;
234        invocation.history = Some(history);
235        Ok(invocation)
236    }
237}
gproxy_protocol/adapt/generate/stream/prepare/claude_responses.rs

gproxy_protocol/adapt/generate/stream/prepare/
claude_responses.rs
