1use super::{
2 chat_blocks::{self, Tool},
3 common::{Budget, StreamEnd, bounded, claude_finish, id, legacy_id},
4 *,
5};
6use crate::transform::generate::reasoning_details as rd;
7use crate::transform::generate::stream::{
8 chat::{self, ChatStreamCollector, ChatStreamLimits},
9 claude::{ClaudeStreamCollector, ClaudeStreamLimits},
10};
11
12#[derive(Debug, Clone, Default)]
16pub struct ChatToClaudeContext {
17 pub start_usage: Option<c::Usage>,
18}
19
20pub struct ChatToClaudeStream {
21 source: Option<ChatStreamCollector>,
22 target: Option<ClaudeStreamCollector>,
23 policy: TargetIdPolicy,
24 flow: IdentityFlow,
25 budget: Budget,
26 limits: StreamLimits,
27 context: ChatToClaudeContext,
28 pending: Vec<q::ChatCompletionChunk>,
29 source_id: Option<String>,
30 model: Option<String>,
31 started: bool,
32 done: bool,
33 failed: bool,
34 closed_blocks: bool,
35 text: Option<i64>,
36 reasoning_text: String,
37 reasoning_details: Vec<crate::wire::openai::chat::ReasoningDetail>,
38 refusal_text: Option<String>,
39 next_block: i64,
40 tools: BTreeMap<i64, Tool>,
41 legacy: Option<Tool>,
42
43 report: Report,
44}
45
46impl ChatToClaudeStream {
47 pub fn new(
48 context: ChatToClaudeContext,
49 flow: IdentityFlow,
50 limits: StreamLimits,
51 ) -> Result<Self, TransformError> {
52 Self::new_with_policy(context, flow, limits, super::common::claude_policy())
53 }
54 pub fn new_with_policy(
55 mut context: ChatToClaudeContext,
56 flow: IdentityFlow,
57 limits: StreamLimits,
58 policy: TargetIdPolicy,
59 ) -> Result<Self, TransformError> {
60 if policy.dialect != crate::Dialect::Claude {
61 return Err(TransformError::shape(
62 "stream.policy",
63 "target policy dialect mismatch",
64 ));
65 }
66
67 context.start_usage = context.start_usage.into_declared();
68 if let Some(usage) = &context.start_usage {
69 bounded(usage, limits.max_bytes)?;
70 super::super::usage::to_chat(usage, &mut Report::default())?;
71 }
72 let native = ClaudeStreamLimits {
73 max_json_bytes: limits.max_bytes,
74 max_text_bytes: limits.max_bytes,
75 };
76 Ok(Self {
77 source: Some(ChatStreamCollector::with_limits(
78 IdentityFlow::new(flow.namespace()),
79 TargetIdPolicy::new(crate::Dialect::OpenAiChat),
80 ChatStreamLimits {
81 max_bytes: limits.max_bytes,
82 },
83 )),
84 target: Some(ClaudeStreamCollector::new(native)),
85 policy,
86 flow,
87 budget: Budget::new(limits),
88 limits,
89 model: None,
90 context,
91 pending: Vec::new(),
92 source_id: None,
93 started: false,
94 done: false,
95 failed: false,
96 closed_blocks: false,
97 text: None,
98 reasoning_text: String::new(),
99 reasoning_details: Vec::new(),
100 refusal_text: None,
101 next_block: 0,
102 tools: BTreeMap::new(),
103 legacy: None,
104
105 report: Default::default(),
106 })
107 }
108 pub fn identities(&self) -> &IdentityFlow {
109 &self.flow
110 }
111 pub fn push(
112 &mut self,
113 chunk: q::ChatCompletionChunk,
114 ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
115 if self.failed || self.done {
116 self.failed = true;
117 return Err(invalid("event after DONE or failed stream"));
118 }
119 let result = self.inner(chunk.into_declared());
120 if result.is_err() {
121 self.failed = true;
122 }
123 result
124 }
125 fn inner(
126 &mut self,
127 chunk: q::ChatCompletionChunk,
128 ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
129 self.budget.input(&chunk)?;
130 if chunk.choices.iter().any(|v| v.index != 0) {
131 return Err(TransformError::unsupported(
132 "choices",
133 "one Claude message requires one Chat choice; fanout needs separate invocations",
134 ));
135 }
136 self.source
137 .as_mut()
138 .ok_or_else(|| invalid("source consumed"))?
139 .push(chunk.clone())?;
140 if !chunk.id.is_empty() {
141 self.source_id = Some(chunk.id.clone());
142 }
143 if !chunk.model.is_empty() {
144 if self.started && self.model.as_ref().is_some_and(|v| v != &chunk.model) {
145 return Err(invalid("model changed after emission"));
146 }
147 self.model = Some(chunk.model.clone());
148 }
149 if self.context.start_usage.is_none()
150 && let Some(usage) = chunk.usage.as_ref().and_then(Option::as_ref)
151 {
152 let usage = chat::usage::collect(usage.clone(), &mut self.report);
153 self.context.start_usage =
154 Some(super::super::usage::to_claude(&usage, &mut self.report)?);
155 }
156 if self.context.start_usage.is_none() || self.model.is_none() {
157 self.pending.push(chunk);
158 return Ok(Converted {
159 value: Vec::new(),
160 report: Report::default(),
161 });
162 }
163 let mut out = Vec::new();
164 if !self.started {
165 self.start(&mut out)?;
166 }
167 for chunk in std::mem::take(&mut self.pending) {
168 self.content(chunk, &mut out)?;
169 }
170 self.content(chunk, &mut out)?;
171 Ok(Converted {
172 value: out,
173 report: std::mem::take(&mut self.report),
174 })
175 }
176 fn start(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
177 let id = id(
178 &mut self.flow,
179 &self.policy,
180 IdentityRole::Response,
181 crate::Dialect::OpenAiChat,
182 self.source_id.clone(),
183 0,
184 )?;
185 let mut initial_usage = self
186 .context
187 .start_usage
188 .clone()
189 .ok_or_else(|| TransformError::missing_metadata("initial usage"))?;
190 if initial_usage.output_tokens_details.take().is_some() {
191 self.report.changed(
192 "usage.output_tokens_details",
193 "initial output detail is deferred until a final source counter is available",
194 );
195 }
196 let message = s::StreamMessage {
197 input_transformations: None,
198 type_: c::GenerateContentResponseBodyType::Tag,
199 id,
200 container: None,
201 content: Vec::new(),
202 context_management: None,
203 diagnostics: None,
204 model: self
205 .model
206 .clone()
207 .ok_or_else(|| TransformError::missing_metadata("model"))?,
208 role: c::ResponseRole::Assistant,
209 stop_details: None,
210 stop_reason: None,
211 stop_sequence: None,
212 usage: initial_usage,
213 rest: Default::default(),
214 };
215 self.emit(
216 out,
217 s::StreamEvent::MessageStart(Box::new(s::MessageStartEvent::builder(message).build())),
218 )?;
219 self.started = true;
220 Ok(())
221 }
222 fn content(
223 &mut self,
224 chunk: q::ChatCompletionChunk,
225 out: &mut Vec<s::StreamEvent>,
226 ) -> Result<(), TransformError> {
227 if let Some(id) = self.source_id.clone() {
228 common::id(
229 &mut self.flow,
230 &self.policy,
231 IdentityRole::Response,
232 crate::Dialect::OpenAiChat,
233 Some(id),
234 0,
235 )?;
236 }
237 for choice in chunk.choices {
238 let delta = choice.delta;
239 if let Some(text) = crate::wire::openai::chat::reasoning_text(
240 &delta.reasoning_content,
241 &delta.reasoning,
242 ) {
243 self.reasoning_text.push_str(text);
244 }
245 if let Some(Some(details)) = delta.reasoning_details {
246 rd::merge(&mut self.reasoning_details, details)?;
247 }
248 if let Some(text) = delta.content.flatten() {
249 let index = self.text_block(out)?;
250 self.emit(out, chat_blocks::text_delta(index, text))?;
251 }
252 if let Some(refusal) = delta.refusal.flatten() {
253 self.refusal_text.get_or_insert_default().push_str(&refusal);
254 }
255 for call in delta.tool_calls.flatten().into_iter().flatten() {
256 let tool = match self.tools.entry(call.index) {
257 std::collections::btree_map::Entry::Occupied(entry) => entry.into_mut(),
258 std::collections::btree_map::Entry::Vacant(entry) => {
259 entry.insert(Tool::default())
260 }
261 };
262 tool.append(call.id.flatten(), call.function.flatten())?;
263 }
264 if let Some(function) = delta.function_call.flatten() {
265 self.legacy
266 .get_or_insert_default()
267 .append(None, Some(function))?;
268 }
269 if choice.finish_reason.flatten().is_some() {
270 self.close_blocks(out)?;
271 }
272 }
273 Ok(())
274 }
275 fn text_block(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<i64, TransformError> {
276 if let Some(index) = self.text {
277 return Ok(index);
278 }
279 let index = self.allocate_block()?;
280 self.emit(out, chat_blocks::text_start(index))?;
281 self.text = Some(index);
282 Ok(index)
283 }
284 fn allocate_block(&mut self) -> Result<i64, TransformError> {
285 let index = self.next_block;
286 self.next_block = self.next_block.checked_add(1).ok_or_else(limit)?;
287 Ok(index)
288 }
289 fn close_blocks(&mut self, out: &mut Vec<s::StreamEvent>) -> Result<(), TransformError> {
290 if self.closed_blocks {
291 return Err(invalid("duplicate choice finish"));
292 }
293 let mut restored = rd::to_claude(&self.reasoning_details);
294 if restored.is_empty() {
295 let text = crate::wire::openai::chat::visible_reasoning(
296 &Some(Some(std::mem::take(&mut self.reasoning_text))),
297 &None,
298 &Some(Some(std::mem::take(&mut self.reasoning_details))),
299 );
300 if let Some(text) = text.filter(|v| !v.is_empty()) {
301 restored.push(crate::wire::claude::content::ContentBlock::Thinking(
302 crate::wire::claude::content::ThinkingBlock::builder(
303 crate::wire::claude::content::ThinkingBlockType::Tag,
304 String::new(),
305 text,
306 )
307 .build(),
308 ));
309 }
310 }
311 for block in restored {
312 let index = self.allocate_block()?;
313 let block = match block {
314 crate::wire::claude::content::ContentBlock::Thinking(v) => {
315 c::ResponseContentBlock::Thinking(v)
316 }
317 crate::wire::claude::content::ContentBlock::RedactedThinking(v) => {
318 c::ResponseContentBlock::RedactedThinking(v)
319 }
320 _ => unreachable!(),
321 };
322 self.emit(
323 out,
324 s::StreamEvent::ContentBlockStart(
325 s::ContentBlockStartEvent::builder(index, block).build(),
326 ),
327 )?;
328 self.emit(out, chat_blocks::block_stop(index))?;
329 }
330 if let Some(index) = self.text {
331 self.emit(out, chat_blocks::block_stop(index))?;
332 }
333 if let Some(text) = self.refusal_text.take() {
334 let index = self.allocate_block()?;
335 self.emit(out, chat_blocks::text_start(index))?;
336 self.emit(out, chat_blocks::text_delta(index, text))?;
337 self.emit(out, chat_blocks::block_stop(index))?;
338 }
339 let mut calls: Vec<_> = std::mem::take(&mut self.tools)
340 .into_values()
341 .map(|tool| (false, tool))
342 .collect();
343 if let Some(tool) = self.legacy.take() {
344 calls.push((true, tool));
345 }
346 for (ordinal, (legacy, tool)) in calls.into_iter().enumerate() {
347 tool.object()?;
348 let block = self.allocate_block()?;
349 let id = if legacy {
350 legacy_id(&mut self.flow, &self.policy, ordinal as u64)?
351 } else {
352 id(
353 &mut self.flow,
354 &self.policy,
355 IdentityRole::ToolCall,
356 crate::Dialect::OpenAiChat,
357 tool.id,
358 ordinal as u64,
359 )?
360 };
361 let content = c::ResponseContentBlock::ToolUse(
362 c::ResponseToolUseBlock::builder(
363 c::ResponseToolUseBlockType::Tag,
364 id,
365 Default::default(),
366 tool.name,
367 )
368 .build(),
369 );
370 self.emit(
371 out,
372 s::StreamEvent::ContentBlockStart(
373 s::ContentBlockStartEvent::builder(block, content).build(),
374 ),
375 )?;
376 self.emit(
377 out,
378 s::StreamEvent::ContentBlockDelta(
379 s::ContentBlockDeltaEvent::builder(
380 block,
381 s::ContentBlockDelta::InputJson(
382 s::InputJsonDelta::builder(tool.arguments).build(),
383 ),
384 )
385 .build(),
386 ),
387 )?;
388 self.emit(out, chat_blocks::block_stop(block))?;
389 }
390 self.closed_blocks = true;
391 Ok(())
392 }
393 fn emit(
394 &mut self,
395 out: &mut Vec<s::StreamEvent>,
396 event: s::StreamEvent,
397 ) -> Result<(), TransformError> {
398 self.budget.emit(out, event)?;
399 self.target
400 .as_mut()
401 .ok_or_else(|| invalid("target consumed"))?
402 .push(out.last().unwrap().clone())?;
403 Ok(())
404 }
405 pub fn push_done(&mut self) -> Result<(), TransformError> {
406 if self.failed || self.done {
407 self.failed = true;
408 return Err(invalid("duplicate DONE or failed stream"));
409 }
410 let result = self
411 .source
412 .as_mut()
413 .ok_or_else(|| invalid("source consumed"))?
414 .push_done();
415 if result.is_err() {
416 self.failed = true;
417 } else {
418 self.done = true;
419 }
420 result
421 }
422 pub fn finish(self) -> Result<StreamEnd<s::StreamEvent>, TransformError> {
423 self.finish_with_usage(None)
424 }
425 pub fn finish_with_usage(
427 mut self,
428 facts: Option<c::Usage>,
429 ) -> Result<StreamEnd<s::StreamEvent>, TransformError> {
430 let facts = facts.into_declared();
431 if self.failed || !self.done {
432 return Err(invalid("Chat stream lacks valid DONE"));
433 }
434 let source = self
435 .source
436 .take()
437 .ok_or_else(|| invalid("source consumed"))?
438 .finish()?;
439 self.report.diagnostics.extend(source.report.diagnostics);
440 let source = source.value;
441 if let Some(facts) = &facts {
442 bounded(facts, self.limits.max_bytes)?;
443 }
444 let usage = super::usage_facts::resolve(
445 source.usage.as_ref(),
446 facts,
447 self.context.start_usage.as_ref(),
448 &mut self.report,
449 )?;
450 bounded(&usage, self.limits.max_bytes)?;
451 let mut out = Vec::new();
452 if !self.started {
453 self.context.start_usage = Some(usage.clone());
454 self.model = Some(source.model.clone());
455 self.source_id = Some(source.id.clone());
456 self.start(&mut out)?;
457 for chunk in std::mem::take(&mut self.pending) {
458 self.content(chunk, &mut out)?;
459 }
460 }
461 if !self.closed_blocks {
462 self.close_blocks(&mut out)?;
463 }
464 let choice = source
465 .choices
466 .first()
467 .ok_or_else(|| invalid("missing choice"))?;
468 if let Some(initial) = &self.context.start_usage {
469 for (changed, field) in [
470 (
471 usage.cache_creation.is_some()
472 && usage.cache_creation != initial.cache_creation,
473 "cache_creation",
474 ),
475 (
476 usage.service_tier.is_some() && usage.service_tier != initial.service_tier,
477 "service_tier",
478 ),
479 (
480 usage.inference_geo.is_some() && usage.inference_geo != initial.inference_geo,
481 "inference_geo",
482 ),
483 (
484 usage.speed.is_some() && usage.speed != initial.speed,
485 "speed",
486 ),
487 ] {
488 if changed {
489 self.report.omitted(
490 format!("usage.{field}"),
491 "Claude message_delta has no field for this late usage metadata",
492 );
493 }
494 }
495 }
496 let delta = s::MessageDelta::builder()
497 .stop_reason(Some(claude_finish(
498 choice.finish_reason,
499 choice.message.refusal.is_some(),
500 )))
501 .build();
502 let event = s::StreamEvent::MessageDelta(Box::new(
503 s::MessageDeltaEvent::builder(delta, chat_blocks::usage_delta(usage)).build(),
504 ));
505 self.emit(&mut out, event)?;
506 self.emit(
507 &mut out,
508 s::StreamEvent::MessageStop(s::MessageStopEvent::builder().build()),
509 )?;
510 self.target
511 .take()
512 .ok_or_else(|| invalid("target consumed"))?
513 .finish()?;
514 for (present, field) in [
515 (choice.logprobs.is_some(), "logprobs"),
516 (source.service_tier.is_some(), "service_tier"),
517 (source.system_fingerprint.is_some(), "system_fingerprint"),
518 (source.moderation.is_some(), "moderation"),
519 ] {
520 if present {
521 self.report
522 .omitted(field, "Claude stream has no equivalent Chat metadata field");
523 }
524 }
525 Ok(StreamEnd {
526 chunks: out,
527 identities: self.flow,
528 report: self.report,
529 })
530 }
531}