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