1use super::{
2 common::{
3 Budget, StreamEnd, StreamLimits, declared, id, initial_usage, invalid, limit, measure,
4 },
5 response_items::Item,
6 usage::claude_tier,
7};
8use crate::{
9 transform::generate::stream::{
10 claude::{ClaudeStreamCollector, ClaudeStreamLimits},
11 responses::{ResponsesStreamCollector, ResponsesStreamLimits},
12 },
13 transform::{
14 Converted, Report, TransformError,
15 identity::{IdentityFlow, IdentityRole, TargetIdPolicy},
16 },
17 wire::{
18 claude::{generate_content as c, stream as cs},
19 openai::responses::{response as r, stream as s},
20 },
21};
22use std::collections::BTreeMap;
23
24#[derive(Default)]
25pub struct ResponsesToClaudeContext {
26 pub usage: Option<c::Usage>,
27}
28
29pub struct ResponsesToClaudeStream {
30 source: Option<ResponsesStreamCollector>,
31 target: Option<ClaudeStreamCollector>,
32 pub(super) flow: IdentityFlow,
33 pub(super) policy: TargetIdPolicy,
34 pub(super) limits: StreamLimits,
35 budget: Budget,
36 usage: Option<c::Usage>,
37 observed_usage: Option<r::ResponseUsage>,
38 pub(super) model: Option<String>,
39 response_id: Option<String>,
40 tier: Option<Option<c::ResponseServiceTier>>,
41 started: bool,
42 failed: bool,
43 terminal: bool,
44 pending: Vec<cs::StreamEvent>,
45 pending_bytes: usize,
46 pub(super) held_bytes: usize,
47 pub(super) items: BTreeMap<i64, Item>,
48 pub(super) cursor: i64,
49
50 next_block: usize,
51 pub(super) report: Report,
52}
53
54impl ResponsesToClaudeStream {
55 pub fn new(
56 context: ResponsesToClaudeContext,
57 flow: IdentityFlow,
58 limits: StreamLimits,
59 ) -> Result<Self, TransformError> {
60 Self::new_with_policy(context, flow, limits, super::common::claude_policy())
61 }
62 pub fn new_with_policy(
63 mut context: ResponsesToClaudeContext,
64 flow: IdentityFlow,
65 limits: StreamLimits,
66 policy: TargetIdPolicy,
67 ) -> Result<Self, TransformError> {
68 if policy.dialect != crate::Dialect::Claude {
69 return Err(TransformError::shape(
70 "stream.policy",
71 "target policy dialect mismatch",
72 ));
73 }
74
75 context.usage = context
76 .usage
77 .map(initial_usage)
78 .map(crate::transform::optional)
79 .transpose()?
80 .flatten();
81 measure(&context.usage, limits.max_bytes)?;
82 Ok(Self {
83 source: Some(ResponsesStreamCollector::new(ResponsesStreamLimits {
84 max_bytes: limits.max_bytes,
85
86 max_text_bytes: limits.max_bytes,
87 max_json_bytes: limits.max_bytes,
88 })),
89 target: Some(ClaudeStreamCollector::new(ClaudeStreamLimits {
90 max_text_bytes: limits.max_bytes,
91 max_json_bytes: limits.max_bytes,
92 })),
93 flow,
94 policy,
95 limits,
96 budget: Budget::new(limits),
97 usage: context.usage,
98 observed_usage: None,
99 model: None,
100 response_id: None,
101 tier: None,
102 started: false,
103 failed: false,
104 terminal: false,
105 pending: Vec::new(),
106 pending_bytes: 0,
107 held_bytes: 0,
108 items: BTreeMap::new(),
109 cursor: 0,
110
111 next_block: 0,
112 report: Default::default(),
113 })
114 }
115 pub fn identities(&self) -> &IdentityFlow {
116 &self.flow
117 }
118 pub fn push(
119 &mut self,
120 event: s::StreamEvent,
121 ) -> Result<Converted<Vec<cs::StreamEvent>>, TransformError> {
122 if self.failed || self.terminal {
123 self.failed = true;
124 return Err(invalid("event after terminal or failure"));
125 }
126 let result = self.inner(declared(event));
127 if result.is_err() {
128 self.failed = true;
129 }
130 result
131 }
132 fn inner(
133 &mut self,
134 event: s::StreamEvent,
135 ) -> Result<Converted<Vec<cs::StreamEvent>>, TransformError> {
136 let event = crate::transform::generate::multi_agent::attribute_event(event)?;
137 self.budget.input(&event)?;
138 self.source
139 .as_mut()
140 .ok_or_else(|| invalid("source consumed"))?
141 .push(event.clone())?;
142 let mut out = Vec::new();
143 match event {
144 s::StreamEvent::Keepalive => out.push(cs::StreamEvent::Ping(cs::PingEvent {
145 rest: Default::default(),
146 })),
147 s::StreamEvent::Created(v) => {
148 self.response_id = Some(id(
149 &mut self.flow,
150 &self.policy,
151 crate::Dialect::OpenAi,
152 IdentityRole::Response,
153 IdentityRole::Response,
154 Some(v.response.id.clone()),
155 0,
156 )?);
157 self.model = Some(v.response.model.clone());
158 self.observe(&v.response)?;
159 self.start(&mut out)?;
160 for (index, item) in v.response.output.into_iter().enumerate() {
161 self.add_item(index as i64, item)?;
162 }
163 }
164 s::StreamEvent::Queued(v) => {
165 self.observe(&v.response)?;
166 self.start(&mut out)?;
167 }
168 s::StreamEvent::InProgress(v) => {
169 self.observe(&v.response)?;
170 self.start(&mut out)?;
171 }
172 s::StreamEvent::OutputItemAdded(v) => self.add_item(v.output_index, v.item)?,
173 s::StreamEvent::ContentPartAdded(v) => {
174 self.bind(v.output_index, Some(&v.item_id))?;
175 self.add_part(v.output_index, v.content_index, v.part)?;
176 }
177 s::StreamEvent::ContentPartDone(v) => {
178 self.part_done(v.output_index, v.content_index, &mut out)?
179 }
180 s::StreamEvent::OutputTextDelta(v) => self.append_text(
181 v.output_index,
182 v.content_index,
183 &v.item_id,
184 v.delta,
185 false,
186 &mut out,
187 )?,
188 s::StreamEvent::RefusalDelta(v) => self.append_text(
189 v.output_index,
190 v.content_index,
191 &v.item_id,
192 v.delta,
193 true,
194 &mut out,
195 )?,
196 s::StreamEvent::FunctionCallArgumentsDelta(v) => {
197 self.append_args(v.output_index, &v.item_id, v.delta, &mut out)?
198 }
199 s::StreamEvent::FunctionCallArgumentsDone(v) => {
200 self.bind(v.output_index, Some(&v.item_id))?;
201 self.args_done(v.output_index, &v.arguments, &mut out)?;
202 }
203 s::StreamEvent::OutputItemDone(v) => {
204 self.item_done(v.output_index, v.item, &mut out)?
205 }
206 s::StreamEvent::Completed(v) => {
207 self.observe(&v.response)?;
208 self.terminal = true;
209 self.start(&mut out)?;
210 }
211 s::StreamEvent::Incomplete(v) => {
212 self.observe(&v.response)?;
213 self.terminal = true;
214 self.start(&mut out)?;
215 }
216 s::StreamEvent::ReasoningSummaryPartAdded(_) => {}
217 s::StreamEvent::OutputTextDone(_)
218 | s::StreamEvent::RefusalDone(_)
219 | s::StreamEvent::OutputTextAnnotationAdded(_)
220 | s::StreamEvent::ReasoningTextDelta(_)
221 | s::StreamEvent::ReasoningTextDone(_)
222 | s::StreamEvent::ReasoningSummaryTextDelta(_)
223 | s::StreamEvent::ReasoningSummaryTextDone(_)
224 | s::StreamEvent::ReasoningSummaryPartDone(_)
225 | s::StreamEvent::McpArgumentsDelta(_)
226 | s::StreamEvent::McpArgumentsDone(_)
227 | s::StreamEvent::McpInProgress(_)
228 | s::StreamEvent::McpCompleted(_)
229 | s::StreamEvent::McpFailed(_) => {}
230 s::StreamEvent::Failed(_) | s::StreamEvent::Error(_) => {
231 return Err(invalid("native source failure"));
232 }
233 _ => {}
234 }
235 self.flush_items(&mut out)?;
236 Ok(Converted {
237 value: out,
238 report: std::mem::take(&mut self.report),
239 })
240 }
241 fn observe(&mut self, response: &r::GenerateContentResponseBody) -> Result<(), TransformError> {
242 if !self.started {
243 self.tier = claude_tier(response.service_tier.flatten()).or(self.tier);
244 }
245 if let Some(usage) = response.usage.as_ref().and_then(Option::as_ref) {
246 let native = super::super::response::usage::to_claude(usage.clone())?;
247 if let Some(previous) = &self.observed_usage
248 && (usage.input_tokens < previous.input_tokens
249 || usage.output_tokens < previous.output_tokens
250 || usage.total_tokens < previous.total_tokens
251 || usage.input_tokens_details.cached_tokens
252 < previous.input_tokens_details.cached_tokens
253 || usage.input_tokens_details.cache_write_tokens
254 < previous.input_tokens_details.cache_write_tokens
255 || usage.output_tokens_details.reasoning_tokens
256 < previous.output_tokens_details.reasoning_tokens)
257 {
258 return Err(invalid("native cumulative usage decreased"));
259 }
260 self.observed_usage = Some(usage.clone());
261 if self.usage.is_none() {
262 self.usage = Some(native);
263 }
264 }
265 Ok(())
266 }
267 fn start(&mut self, out: &mut Vec<cs::StreamEvent>) -> Result<(), TransformError> {
268 if self.started {
269 return Ok(());
270 }
271 let (Some(id), Some(model), Some(usage)) =
272 (&self.response_id, &self.model, &mut self.usage)
273 else {
274 return Ok(());
275 };
276 if let Some(tier) = self.tier {
277 if usage
278 .service_tier
279 .flatten()
280 .zip(tier)
281 .is_some_and(|(old, new)| old != new)
282 {
283 return Err(invalid("initial service-tier facts contradict source"));
284 }
285 usage.service_tier = Some(tier);
286 }
287 let message = cs::StreamMessage {
288 input_transformations: None,
289 type_: c::GenerateContentResponseBodyType::Tag,
290 id: id.clone(),
291 container: None,
292 content: Vec::new(),
293 context_management: None,
294 diagnostics: None,
295 model: model.clone(),
296 role: c::ResponseRole::Assistant,
297 stop_details: None,
298 stop_reason: None,
299 stop_sequence: None,
300 usage: usage.clone(),
301 rest: Default::default(),
302 };
303 let event = cs::StreamEvent::MessageStart(Box::new(
304 cs::MessageStartEvent::builder(message).build(),
305 ));
306 self.budget.output(&event)?;
307 self.deliver(out, event)?;
308 self.started = true;
309 for event in std::mem::take(&mut self.pending) {
310 self.deliver(out, event)?;
311 }
312 self.pending_bytes = 0;
313 Ok(())
314 }
315 pub(super) fn reserve_held(&mut self, bytes: usize) -> Result<(), TransformError> {
316 let held = self.held_bytes.checked_add(bytes).ok_or_else(limit)?;
317 if held
318 .checked_add(self.pending_bytes)
319 .is_none_or(|n| n > self.limits.max_pending)
320 {
321 return Err(limit());
322 }
323 self.held_bytes = held;
324 Ok(())
325 }
326 pub(super) fn allocate_block(&mut self) -> Result<i64, TransformError> {
327 let index = i64::try_from(self.next_block).map_err(|_| limit())?;
328 self.next_block += 1;
329 Ok(index)
330 }
331 pub(super) fn complete_block(
332 &mut self,
333 out: &mut Vec<cs::StreamEvent>,
334 block: c::ResponseContentBlock,
335 ) -> Result<(), TransformError> {
336 let index = self.allocate_block()?;
337 self.emit(
338 out,
339 cs::StreamEvent::ContentBlockStart(
340 cs::ContentBlockStartEvent::builder(index, block).build(),
341 ),
342 )?;
343 self.emit(
344 out,
345 cs::StreamEvent::ContentBlockStop(cs::ContentBlockStopEvent::builder(index).build()),
346 )
347 }
348 pub(super) fn emit(
349 &mut self,
350 out: &mut Vec<cs::StreamEvent>,
351 event: cs::StreamEvent,
352 ) -> Result<(), TransformError> {
353 let n = self.budget.output(&event)?;
354 if self.started {
355 self.deliver(out, event)
356 } else {
357 let bytes = self.pending_bytes.checked_add(n).ok_or_else(limit)?;
358 if bytes
359 .checked_add(self.held_bytes)
360 .is_none_or(|n| n > self.limits.max_pending)
361 {
362 return Err(limit());
363 }
364 self.pending.push(event);
365 self.pending_bytes = bytes;
366 Ok(())
367 }
368 }
369 fn deliver(
370 &mut self,
371 out: &mut Vec<cs::StreamEvent>,
372 event: cs::StreamEvent,
373 ) -> Result<(), TransformError> {
374 self.target
375 .as_mut()
376 .ok_or_else(|| invalid("target consumed"))?
377 .push(event.clone())?;
378 out.push(event);
379 Ok(())
380 }
381 pub fn finish(mut self) -> Result<StreamEnd<cs::StreamEvent>, TransformError> {
382 if self.failed || !self.terminal {
383 return Err(invalid("failed stream or premature EOF"));
384 }
385 let source = self
386 .source
387 .take()
388 .ok_or_else(|| invalid("source consumed"))?
389 .finish()?
390 .value;
391 let part_refusal = source.output.iter().any(|item| matches!(item, r::ResponseOutputItem::Message(message) if message.content.iter().any(|part| matches!(part, crate::wire::openai::responses::input::OutputContent::Refusal(_)))));
392 let converted = super::super::responses_to_claude_response(
393 source,
394 &mut self.flow.clone(),
395 &self.policy,
396 )?;
397 let mut expected = converted.value;
398 if part_refusal && expected.stop_reason != c::StopReason::Refusal {
399 self.report.changed("output.refusal", "refusal text is preserved; Claude has no separate per-part refusal marker alongside tool continuation or an incomplete stop");
400 }
401 self.report.diagnostics.extend(converted.report.diagnostics);
402 let mut out = Vec::new();
403 if !self.started {
404 self.usage = Some(expected.usage.clone());
405 self.start(&mut out)?;
406 }
407 let initial = self
408 .usage
409 .as_ref()
410 .ok_or_else(|| invalid("missing start usage"))?;
411
412 if expected.usage.service_tier != initial.service_tier {
413 self.report.omitted(
414 "usage.service_tier",
415 "late effective tier cannot update Claude message_start",
416 );
417 }
418 expected.usage.service_tier = initial.service_tier;
419 expected.usage.cache_creation = initial.cache_creation.clone();
420 expected.usage.fallback_credit = initial.fallback_credit.clone();
421 expected.usage.inference_geo = initial.inference_geo.clone();
422 expected.usage.iterations = initial.iterations.clone();
423 expected.usage.server_tool_use = initial.server_tool_use.clone();
424 expected.usage.speed = initial.speed;
425 measure(&expected, self.limits.max_bytes)?;
426 let usage = &expected.usage;
427 let delta = cs::MessageDelta::builder()
428 .stop_reason(Some(expected.stop_reason))
429 .build();
430 let usage = cs::MessageDeltaUsage {
431 output_tokens: usage.output_tokens,
432 input_tokens: Some(Some(usage.input_tokens)),
433 cache_creation_input_tokens: usage.cache_creation_input_tokens,
434 cache_read_input_tokens: usage.cache_read_input_tokens,
435 output_tokens_details: usage.output_tokens_details.clone(),
436 server_tool_use: usage.server_tool_use.clone(),
437 iterations: usage.iterations.clone(),
438 fallback_credit: usage.fallback_credit.clone(),
439 rest: Default::default(),
440 };
441 self.emit(
442 &mut out,
443 cs::StreamEvent::MessageDelta(Box::new(
444 cs::MessageDeltaEvent::builder(delta, usage).build(),
445 )),
446 )?;
447 self.emit(
448 &mut out,
449 cs::StreamEvent::MessageStop(cs::MessageStopEvent::builder().build()),
450 )?;
451 let actual = self
452 .target
453 .take()
454 .ok_or_else(|| invalid("target consumed"))?
455 .finish()?
456 .value;
457 if actual != expected {
458 return Err(invalid(
459 "native Claude output differs from canonical Responses conversion",
460 ));
461 }
462 Ok(StreamEnd {
463 chunks: out,
464 identities: self.flow,
465 report: self.report,
466 })
467 }
468}