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