1use super::{
2 common::{Budget, StreamEnd, chat_finish, id},
3 *,
4};
5use crate::transform::generate::reasoning_details as rd;
6use crate::transform::generate::stream::{
7 chat::{self, ChatStreamCollector, ChatStreamLimits},
8 claude::{ClaudeStreamCollector, ClaudeStreamLimits},
9};
10
11#[derive(Debug, Clone, Copy)]
12pub struct ClaudeToChatContext {
13 pub created: i64,
14}
15
16enum Block {
17 Omitted,
18 Text {
19 pending: Option<String>,
20 closed: bool,
21 },
22 Reasoning(crate::wire::claude::content::ThinkingBlock),
23 Tool {
24 index: i64,
25 has_arguments: bool,
26 },
27}
28
29pub struct ClaudeToChatStream {
30 source: Option<ClaudeStreamCollector>,
31 target: Option<ChatStreamCollector>,
32 policy: TargetIdPolicy,
33 flow: IdentityFlow,
34 budget: Budget,
35 context: ClaudeToChatContext,
36 blocks: BTreeMap<i64, Block>,
37 text_order: std::collections::VecDeque<i64>,
38 next_tool: usize,
39 id: Option<String>,
40 model: Option<String>,
41 stopped: bool,
42 failed: bool,
43 report: Report,
44}
45
46impl ClaudeToChatStream {
47 pub fn new(
48 context: ClaudeToChatContext,
49 flow: IdentityFlow,
50 limits: StreamLimits,
51 ) -> Result<Self, TransformError> {
52 Self::new_with_policy(
53 context,
54 flow,
55 limits,
56 TargetIdPolicy::new(crate::Dialect::OpenAiChat),
57 )
58 }
59 pub fn new_with_policy(
60 context: ClaudeToChatContext,
61 flow: IdentityFlow,
62 limits: StreamLimits,
63 policy: TargetIdPolicy,
64 ) -> Result<Self, TransformError> {
65 if policy.dialect != crate::Dialect::OpenAiChat {
66 return Err(TransformError::shape(
67 "stream.policy",
68 "target policy dialect mismatch",
69 ));
70 }
71
72 if context.created < 0 {
73 return Err(invalid("negative factual creation timestamp"));
74 }
75 let mut collector_policy = policy.clone();
79 collector_policy.preserve_source_ids = true;
80 Ok(Self {
81 target: Some(ChatStreamCollector::with_limits(
82 IdentityFlow::new(flow.namespace()),
83 collector_policy,
84 ChatStreamLimits {
85 max_bytes: limits.max_bytes,
86 },
87 )),
88 source: Some(ClaudeStreamCollector::new(ClaudeStreamLimits {
89 max_json_bytes: limits.max_bytes,
90 max_text_bytes: limits.max_bytes,
91 })),
92 policy,
93 flow,
94 budget: Budget::new(limits),
95 context,
96 blocks: BTreeMap::new(),
97 text_order: Default::default(),
98 next_tool: 0,
99 id: None,
100 model: None,
101 stopped: false,
102 failed: false,
103 report: Default::default(),
104 })
105 }
106 pub(crate) fn reserve_external_ids(
107 &mut self,
108 role: crate::transform::identity::IdentityRole,
109 ids: &std::collections::BTreeSet<String>,
110 ) -> Result<(), TransformError> {
111 self.flow.reserve_external_ids(role, ids).map_err(|e| {
112 TransformError::new(
113 crate::transform::TransformErrorKind::Conflict,
114 "fanout.ids",
115 e.to_string(),
116 )
117 })
118 }
119 pub fn identities(&self) -> &IdentityFlow {
120 &self.flow
121 }
122 pub fn push(
123 &mut self,
124 event: s::StreamEvent,
125 ) -> Result<Converted<Vec<q::ChatCompletionChunk>>, TransformError> {
126 if self.failed || self.stopped {
127 self.failed = true;
128 return Err(invalid("event after message_stop or failed stream"));
129 }
130 let result = self.inner(event.into_declared());
131 if result.is_err() {
132 self.failed = true;
133 }
134 result
135 }
136 fn inner(
137 &mut self,
138 event: s::StreamEvent,
139 ) -> Result<Converted<Vec<q::ChatCompletionChunk>>, TransformError> {
140 self.budget.input(&event)?;
141 self.source
142 .as_mut()
143 .ok_or_else(|| invalid("source consumed"))?
144 .push(event.clone())?;
145 let mut out = Vec::new();
146 match event {
147 s::StreamEvent::MessageStart(v) => {
148 self.id = Some(id(
149 &mut self.flow,
150 &self.policy,
151 IdentityRole::Response,
152 crate::Dialect::Claude,
153 Some(v.message.id),
154 0,
155 )?);
156 self.model = Some(v.message.model);
157 self.emit(
158 &mut out,
159 q::Delta::builder().role(q::DeltaRole::Assistant).build(),
160 )?;
161 }
162 s::StreamEvent::ContentBlockStart(v) => match v.content_block {
163 c::ResponseContentBlock::Text(text) => {
164 self.blocks.insert(
165 v.index,
166 Block::Text {
167 pending: Some(text.text),
168 closed: false,
169 },
170 );
171 self.text_order.push_back(v.index);
172 self.flush_text(&mut out)?;
173 if text.citations.is_some() {
174 self.report.omitted(
175 "content.citations",
176 "Chat deltas have no typed citation field",
177 );
178 }
179 }
180 c::ResponseContentBlock::ToolUse(tool)
181 if tool.toolset_name.as_ref().is_none_or(Option::is_none) =>
182 {
183 if tool.name.is_empty() {
184 return Err(TransformError::missing_metadata("tool.name"));
185 }
186 let index = i64::try_from(self.next_tool).map_err(|_| limit())?;
187 self.next_tool += 1;
188 let call_id = id(
189 &mut self.flow,
190 &self.policy,
191 IdentityRole::ToolCall,
192 crate::Dialect::Claude,
193 Some(tool.id),
194 index as u64,
195 )?;
196 let has_arguments = !tool.input.is_empty();
197 let mut function = q::DeltaFunctionCall::builder().name(tool.name).build();
198 if has_arguments {
199 function.arguments = Some(Some(serde_json::to_string(&tool.input)?));
200 }
201 let mut call = q::DeltaToolCall::builder(index).build();
202 call.id = Some(Some(call_id));
203 call.type_ = Some(Some(q::DeltaToolCallType::Function));
204 call.function = Some(Some(function));
205 self.blocks.insert(
206 v.index,
207 Block::Tool {
208 index,
209 has_arguments,
210 },
211 );
212 let mut delta = q::Delta::builder().build();
213 delta.tool_calls = Some(Some(vec![call]));
214 self.emit(&mut out, delta)?;
215 if tool.caller.is_some() {
216 self.report
217 .omitted("tool.caller", "Chat stream has no Claude caller field");
218 }
219 }
220 c::ResponseContentBlock::Thinking(block) => {
221 self.blocks.insert(v.index, Block::Reasoning(block.clone()));
222 if !block.thinking.is_empty() {
223 let mut delta = q::Delta::builder().build();
224 delta.reasoning_content = Some(Some(block.thinking));
225 self.emit(&mut out, delta)?;
226 }
227 }
228 c::ResponseContentBlock::RedactedThinking(block) => {
229 self.blocks.insert(v.index, Block::Omitted);
230 let mut delta = q::Delta::builder().build();
231 delta.reasoning_details = Some(Some(vec![rd::from_redacted(&block, v.index)]));
232 self.emit(&mut out, delta)?;
233 }
234 _ => {
235 self.blocks.insert(v.index, Block::Omitted);
236 self.report
237 .omitted("content_block", "block has no target representation");
238 }
239 },
240 s::StreamEvent::ContentBlockDelta(v) => {
241 let index = v.index;
242 if matches!(self.blocks.get(&index), Some(Block::Omitted)) {
243 return Ok(Converted {
244 value: out,
245 report: Report::default(),
246 });
247 }
248 match v.delta {
249 s::ContentBlockDelta::Text(text) => {
250 let Some(Block::Text { pending, .. }) = self.blocks.get_mut(&index) else {
251 return Err(invalid("text delta on nontext block"));
252 };
253 pending.get_or_insert_default().push_str(&text.text);
254 self.flush_text(&mut out)?;
255 }
256 s::ContentBlockDelta::InputJson(json) => {
257 let (tool, has) = match self.blocks.get_mut(&index) {
258 Some(Block::Tool {
259 index,
260 has_arguments,
261 }) => (*index, has_arguments),
262 _ => return Err(invalid("JSON delta on nonclient tool")),
263 };
264 *has |= !json.partial_json.is_empty();
265 self.emit(&mut out, arguments(tool, json.partial_json))?;
266 }
267 s::ContentBlockDelta::Thinking(block) => {
268 let Some(Block::Reasoning(original)) = self.blocks.get_mut(&index) else {
269 return Err(invalid("thinking delta on nonreasoning block"));
270 };
271 original.thinking.push_str(&block.thinking);
272 let mut delta = q::Delta::builder().build();
273 delta.reasoning_content = Some(Some(block.thinking));
274 self.emit(&mut out, delta)?;
275 }
276 s::ContentBlockDelta::Signature(signature) => {
277 let Some(Block::Reasoning(original)) = self.blocks.get_mut(&index) else {
278 return Err(invalid("signature on nonreasoning block"));
279 };
280 original.signature = signature.signature;
281 }
282 s::ContentBlockDelta::Citations(_) => self
283 .report
284 .omitted("content.citations", "Chat has no typed citation delta"),
285 s::ContentBlockDelta::Compaction(_) => {
286 self.report
287 .omitted("content.compaction", "block has no target representation");
288 }
289 }
290 }
291 s::StreamEvent::ContentBlockStop(v) => {
292 if let Some(Block::Reasoning(block)) = self.blocks.get(&v.index) {
293 let mut delta = q::Delta::builder().build();
294 delta.reasoning_details = Some(Some(vec![rd::from_thinking(block, v.index)]));
295 self.emit(&mut out, delta)?;
296 }
297 if let Some(Block::Text { closed, .. }) = self.blocks.get_mut(&v.index) {
298 *closed = true;
299 self.flush_text(&mut out)?;
300 }
301 if let Some(Block::Tool {
302 index,
303 has_arguments: false,
304 }) = self.blocks.get(&v.index)
305 {
306 let delta = arguments(*index, "{}".into());
307 self.emit(&mut out, delta)?;
308 }
309 }
310 s::StreamEvent::MessageDelta(v) => {
311 if let Some(Some(reason)) = v.delta.stop_reason {
312 chat_finish(reason)?;
313 }
314 }
315 s::StreamEvent::MessageStop(_) => self.stopped = true,
316 s::StreamEvent::Ping(_) => {}
317 s::StreamEvent::Error(_) => return Err(invalid("upstream Claude error")),
318 }
319 Ok(Converted {
320 value: out,
321 report: std::mem::take(&mut self.report),
322 })
323 }
324 fn emit(
325 &mut self,
326 out: &mut Vec<q::ChatCompletionChunk>,
327 delta: q::Delta,
328 ) -> Result<(), TransformError> {
329 let chunk = self.chunk(vec![q::StreamChoice::builder(0, delta).build()])?;
330 self.emit_chunk(out, chunk)
331 }
332 fn emit_chunk(
333 &mut self,
334 out: &mut Vec<q::ChatCompletionChunk>,
335 chunk: q::ChatCompletionChunk,
336 ) -> Result<(), TransformError> {
337 self.budget.emit(out, chunk)?;
338 self.target
339 .as_mut()
340 .ok_or_else(|| invalid("target consumed"))?
341 .push(out.last().unwrap().clone())?;
342 Ok(())
343 }
344 fn flush_text(&mut self, out: &mut Vec<q::ChatCompletionChunk>) -> Result<(), TransformError> {
347 while let Some(index) = self.text_order.front().copied() {
348 let (text, closed) = match self.blocks.get_mut(&index) {
349 Some(Block::Text { pending, closed }) => (pending.take(), *closed),
350 _ => return Err(invalid("missing ordered text block")),
351 };
352 if let Some(text) = text {
353 self.emit(out, q::Delta::builder().content(text).build())?;
354 }
355 if !closed {
356 break;
357 }
358 self.text_order.pop_front();
359 }
360 Ok(())
361 }
362 fn chunk(
363 &self,
364 choices: Vec<q::StreamChoice>,
365 ) -> Result<q::ChatCompletionChunk, TransformError> {
366 Ok(q::ChatCompletionChunk::builder(
367 self.id.clone().ok_or_else(|| invalid("missing start ID"))?,
368 choices,
369 self.context.created,
370 self.model.clone().ok_or_else(|| invalid("missing model"))?,
371 q::ChunkObject::ChatCompletionChunk,
372 )
373 .build())
374 }
375 pub fn finish(mut self) -> Result<StreamEnd<q::ChatCompletionChunk>, TransformError> {
376 if self.failed || !self.stopped {
377 return Err(invalid("Claude stream ended before valid message_stop"));
378 }
379 let source = self
380 .source
381 .take()
382 .ok_or_else(|| invalid("source consumed"))?
383 .finish()?
384 .value;
385 let finish = if source.stop_reason == c::StopReason::ToolUse && self.next_tool == 0
386 && source.content.iter().any(|block| matches!(block, c::ResponseContentBlock::ToolUse(v) if v.toolset_name.as_ref().is_some_and(Option::is_some))) {
387 o::FinishReason::Stop
388 } else {
389 chat_finish(source.stop_reason)?
390 };
391 if source.stop_reason == c::StopReason::Refusal {
392 self.report.changed("refusal","Claude refusal is learned after text delivery; streamed text remains content and terminal is content_filter");
393 }
394 let usage = super::super::usage::to_chat(&source.usage, &mut self.report)?;
395 let mut out = Vec::new();
396 let choice = q::StreamChoice::builder(0, q::Delta::builder().build())
397 .finish_reason(finish)
398 .build();
399 let chunk = self.chunk(vec![choice])?;
400 self.emit_chunk(&mut out, chunk)?;
401 let mut tail = self.chunk(Vec::new())?;
402 tail.usage = Some(Some(chat::usage::synthesize(usage)?));
403 self.emit_chunk(&mut out, tail)?;
404 let mut target = self
405 .target
406 .take()
407 .ok_or_else(|| invalid("target consumed"))?;
408 target.push_done()?;
409 target.finish()?;
410 for (present, field) in [
411 (source.container.is_some(), "container"),
412 (source.context_management.is_some(), "context_management"),
413 (source.diagnostics.is_some(), "diagnostics"),
414 (source.stop_details.is_some(), "stop_details"),
415 (source.stop_sequence.is_some(), "stop_sequence"),
416 ] {
417 if present {
418 self.report
419 .omitted(field, "Chat stream has no equivalent Claude metadata field");
420 }
421 }
422 Ok(StreamEnd {
423 chunks: out,
424 identities: self.flow,
425 report: self.report,
426 })
427 }
428}
429
430fn arguments(index: i64, text: String) -> q::Delta {
431 let mut tool = q::DeltaToolCall::builder(index).build();
432 tool.function = Some(Some(
433 q::DeltaFunctionCall::builder().arguments(text).build(),
434 ));
435 let mut delta = q::Delta::builder().build();
436 delta.tool_calls = Some(Some(vec![tool]));
437 delta
438}