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 matches!(event, s::StreamEvent::Unknown) {
123 return Ok(Converted {
124 value: Vec::new(),
125 report: Report::default(),
126 });
127 }
128 if self.failed || self.terminal {
129 self.failed = true;
130 return Err(invalid("event after terminal or failure"));
131 }
132 let result = self.inner(declared(event));
133 if result.is_err() {
134 self.failed = true;
135 }
136 result
137 }
138 fn inner(
139 &mut self,
140 event: s::StreamEvent,
141 ) -> Result<Converted<Vec<cs::StreamEvent>>, TransformError> {
142 let event = crate::transform::generate::multi_agent::attribute_event(event)?;
143 self.budget.input(&event)?;
144 self.source
145 .as_mut()
146 .ok_or_else(|| invalid("source consumed"))?
147 .push(event.clone())?;
148 let mut out = Vec::new();
149 match event {
150 s::StreamEvent::Metadata(_) | s::StreamEvent::Unknown => {}
151 s::StreamEvent::Keepalive => out.push(cs::StreamEvent::Ping(cs::PingEvent {
152 rest: Default::default(),
153 })),
154 s::StreamEvent::Created(v) => {
155 self.response_id = Some(id(
156 &mut self.flow,
157 &self.policy,
158 crate::Dialect::OpenAi,
159 IdentityRole::Response,
160 IdentityRole::Response,
161 Some(v.response.id.clone()),
162 0,
163 )?);
164 self.model = Some(v.response.model.clone());
165 self.observe(&v.response)?;
166 self.start(&mut out)?;
167 for (index, item) in v.response.output.into_iter().enumerate() {
168 self.add_item(index as i64, item)?;
169 }
170 }
171 s::StreamEvent::Queued(v) => {
172 self.observe(&v.response)?;
173 self.start(&mut out)?;
174 }
175 s::StreamEvent::InProgress(v) => {
176 self.observe(&v.response)?;
177 self.start(&mut out)?;
178 }
179 s::StreamEvent::OutputItemAdded(v) => self.add_item(v.output_index, v.item)?,
180 s::StreamEvent::ContentPartAdded(v) => {
181 self.bind(v.output_index, Some(&v.item_id))?;
182 self.add_part(v.output_index, v.content_index, v.part)?;
183 }
184 s::StreamEvent::ContentPartDone(v) => {
185 self.part_done(v.output_index, v.content_index, &mut out)?
186 }
187 s::StreamEvent::OutputTextDelta(v) => self.append_text(
188 v.output_index,
189 v.content_index,
190 &v.item_id,
191 v.delta,
192 false,
193 &mut out,
194 )?,
195 s::StreamEvent::RefusalDelta(v) => self.append_text(
196 v.output_index,
197 v.content_index,
198 &v.item_id,
199 v.delta,
200 true,
201 &mut out,
202 )?,
203 s::StreamEvent::FunctionCallArgumentsDelta(v) => {
204 self.append_args(v.output_index, &v.item_id, v.delta, &mut out)?
205 }
206 s::StreamEvent::FunctionCallArgumentsDone(v) => {
207 self.bind(v.output_index, Some(&v.item_id))?;
208 self.args_done(v.output_index, &v.arguments, &mut out)?;
209 }
210 s::StreamEvent::OutputItemDone(v) => {
211 self.item_done(v.output_index, v.item, &mut out)?
212 }
213 s::StreamEvent::Completed(v) => {
214 self.observe(&v.response)?;
215 self.terminal = true;
216 self.start(&mut out)?;
217 }
218 s::StreamEvent::Incomplete(v) => {
219 self.observe(&v.response)?;
220 self.terminal = true;
221 self.start(&mut out)?;
222 }
223 s::StreamEvent::ReasoningSummaryPartAdded(_) => {}
224 s::StreamEvent::OutputTextDone(_)
225 | s::StreamEvent::RefusalDone(_)
226 | s::StreamEvent::OutputTextAnnotationAdded(_)
227 | s::StreamEvent::ReasoningTextDelta(_)
228 | s::StreamEvent::ReasoningTextDone(_)
229 | s::StreamEvent::ReasoningSummaryTextDelta(_)
230 | s::StreamEvent::ReasoningSummaryTextDone(_)
231 | s::StreamEvent::ReasoningSummaryPartDone(_)
232 | s::StreamEvent::McpArgumentsDelta(_)
233 | s::StreamEvent::McpArgumentsDone(_)
234 | s::StreamEvent::McpInProgress(_)
235 | s::StreamEvent::McpCompleted(_)
236 | s::StreamEvent::McpFailed(_) => {}
237 s::StreamEvent::Failed(_) | s::StreamEvent::Error(_) => {
238 return Err(invalid("native source failure"));
239 }
240 _ => {}
241 }
242 self.flush_items(&mut out)?;
243 Ok(Converted {
244 value: out,
245 report: std::mem::take(&mut self.report),
246 })
247 }
248 fn observe(&mut self, response: &r::GenerateContentResponseBody) -> Result<(), TransformError> {
249 if !self.started {
250 self.tier = claude_tier(response.service_tier.flatten()).or(self.tier);
251 }
252 if let Some(usage) = response.usage.as_ref().and_then(Option::as_ref) {
253 let native = super::super::response::usage::to_claude(usage.clone())?;
254 if let Some(previous) = &self.observed_usage
255 && (usage.input_tokens < previous.input_tokens
256 || usage.output_tokens < previous.output_tokens
257 || usage.total_tokens < previous.total_tokens
258 || usage.input_tokens_details.cached_tokens
259 < previous.input_tokens_details.cached_tokens
260 || usage.input_tokens_details.cache_write_tokens
261 < previous.input_tokens_details.cache_write_tokens
262 || usage.output_tokens_details.reasoning_tokens
263 < previous.output_tokens_details.reasoning_tokens)
264 {
265 return Err(invalid("native cumulative usage decreased"));
266 }
267 self.observed_usage = Some(usage.clone());
268 if self.usage.is_none() {
269 self.usage = Some(native);
270 }
271 }
272 Ok(())
273 }
274 fn start(&mut self, out: &mut Vec<cs::StreamEvent>) -> Result<(), TransformError> {
275 if self.started {
276 return Ok(());
277 }
278 let (Some(id), Some(model)) = (&self.response_id, &self.model) else {
279 return Ok(());
280 };
281 let usage = self
285 .usage
286 .get_or_insert_with(|| c::Usage::builder(0, 0).build());
287 if let Some(tier) = self.tier {
288 if usage
289 .service_tier
290 .flatten()
291 .zip(tier)
292 .is_some_and(|(old, new)| old != new)
293 {
294 return Err(invalid("initial service-tier facts contradict source"));
295 }
296 usage.service_tier = Some(tier);
297 }
298 let message = cs::StreamMessage {
299 input_transformations: None,
300 type_: c::GenerateContentResponseBodyType::Tag,
301 id: id.clone(),
302 container: None,
303 content: Vec::new(),
304 context_management: None,
305 diagnostics: None,
306 model: model.clone(),
307 role: c::ResponseRole::Assistant,
308 stop_details: None,
309 stop_reason: None,
310 stop_sequence: None,
311 usage: usage.clone(),
312 rest: Default::default(),
313 };
314 let event = cs::StreamEvent::MessageStart(Box::new(
315 cs::MessageStartEvent::builder(message).build(),
316 ));
317 self.budget.output(&event)?;
318 self.deliver(out, event)?;
319 self.started = true;
320 for event in std::mem::take(&mut self.pending) {
321 self.deliver(out, event)?;
322 }
323 self.pending_bytes = 0;
324 Ok(())
325 }
326 pub(super) fn reserve_held(&mut self, bytes: usize) -> Result<(), TransformError> {
327 let held = self.held_bytes.checked_add(bytes).ok_or_else(limit)?;
328 if held
329 .checked_add(self.pending_bytes)
330 .is_none_or(|n| n > self.limits.max_pending)
331 {
332 return Err(limit());
333 }
334 self.held_bytes = held;
335 Ok(())
336 }
337 pub(super) fn allocate_block(&mut self) -> Result<i64, TransformError> {
338 let index = i64::try_from(self.next_block).map_err(|_| limit())?;
339 self.next_block += 1;
340 Ok(index)
341 }
342 pub(super) fn complete_block(
343 &mut self,
344 out: &mut Vec<cs::StreamEvent>,
345 block: c::ResponseContentBlock,
346 ) -> Result<(), TransformError> {
347 let index = self.allocate_block()?;
348 self.emit(
349 out,
350 cs::StreamEvent::ContentBlockStart(
351 cs::ContentBlockStartEvent::builder(index, block).build(),
352 ),
353 )?;
354 self.emit(
355 out,
356 cs::StreamEvent::ContentBlockStop(cs::ContentBlockStopEvent::builder(index).build()),
357 )
358 }
359 pub(super) fn emit(
360 &mut self,
361 out: &mut Vec<cs::StreamEvent>,
362 event: cs::StreamEvent,
363 ) -> Result<(), TransformError> {
364 let n = self.budget.output(&event)?;
365 if self.started {
366 self.deliver(out, event)
367 } else {
368 let bytes = self.pending_bytes.checked_add(n).ok_or_else(limit)?;
369 if bytes
370 .checked_add(self.held_bytes)
371 .is_none_or(|n| n > self.limits.max_pending)
372 {
373 return Err(limit());
374 }
375 self.pending.push(event);
376 self.pending_bytes = bytes;
377 Ok(())
378 }
379 }
380 fn deliver(
381 &mut self,
382 out: &mut Vec<cs::StreamEvent>,
383 event: cs::StreamEvent,
384 ) -> Result<(), TransformError> {
385 self.target
386 .as_mut()
387 .ok_or_else(|| invalid("target consumed"))?
388 .push(event.clone())?;
389 out.push(event);
390 Ok(())
391 }
392 pub fn finish(mut self) -> Result<StreamEnd<cs::StreamEvent>, TransformError> {
393 if self.failed || !self.terminal {
394 return Err(invalid("failed stream or premature EOF"));
395 }
396 let source = self
397 .source
398 .take()
399 .ok_or_else(|| invalid("source consumed"))?
400 .finish()?
401 .value;
402 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(_)))));
403 let converted = super::super::responses_to_claude_response(
404 source,
405 &mut self.flow.clone(),
406 &self.policy,
407 )?;
408 let mut expected = converted.value;
409 if part_refusal && expected.stop_reason != c::StopReason::Refusal {
410 self.report.changed("output.refusal", "refusal text is preserved; Claude has no separate per-part refusal marker alongside tool continuation or an incomplete stop");
411 }
412 self.report.diagnostics.extend(converted.report.diagnostics);
413 let mut out = Vec::new();
414 if !self.started {
415 self.usage = Some(expected.usage.clone());
416 self.start(&mut out)?;
417 }
418 let initial = self
419 .usage
420 .as_ref()
421 .ok_or_else(|| invalid("missing start usage"))?;
422
423 if expected.usage.service_tier != initial.service_tier {
424 self.report.omitted(
425 "usage.service_tier",
426 "late effective tier cannot update Claude message_start",
427 );
428 }
429 expected.usage.service_tier = initial.service_tier;
430 expected.usage.cache_creation = initial.cache_creation.clone();
431 expected.usage.fallback_credit = initial.fallback_credit.clone();
432 expected.usage.inference_geo = initial.inference_geo.clone();
433 expected.usage.iterations = initial.iterations.clone();
434 expected.usage.server_tool_use = initial.server_tool_use.clone();
435 expected.usage.speed = initial.speed;
436 measure(&expected, self.limits.max_bytes)?;
437 let usage = &expected.usage;
438 let delta = cs::MessageDelta::builder()
439 .stop_reason(Some(expected.stop_reason))
440 .build();
441 let usage = cs::MessageDeltaUsage {
442 output_tokens: usage.output_tokens,
443 input_tokens: Some(Some(usage.input_tokens)),
444 cache_creation_input_tokens: usage.cache_creation_input_tokens,
445 cache_read_input_tokens: usage.cache_read_input_tokens,
446 output_tokens_details: usage.output_tokens_details.clone(),
447 server_tool_use: usage.server_tool_use.clone(),
448 iterations: usage.iterations.clone(),
449 fallback_credit: usage.fallback_credit.clone(),
450 rest: Default::default(),
451 };
452 self.emit(
453 &mut out,
454 cs::StreamEvent::MessageDelta(Box::new(
455 cs::MessageDeltaEvent::builder(delta, usage).build(),
456 )),
457 )?;
458 self.emit(
459 &mut out,
460 cs::StreamEvent::MessageStop(cs::MessageStopEvent::builder().build()),
461 )?;
462 let actual = self
463 .target
464 .take()
465 .ok_or_else(|| invalid("target consumed"))?
466 .finish()?
467 .value;
468 if actual != expected {
469 return Err(invalid(
470 "native Claude output differs from canonical Responses conversion",
471 ));
472 }
473 Ok(StreamEnd {
474 chunks: out,
475 identities: self.flow,
476 report: self.report,
477 })
478 }
479}