1pub use super::context::ClaudeToResponsesContext;
2use super::{
3 common::{
4 Budget, StreamEnd, StreamLimits, declared, id, invalid, item_id, limit, measure,
5 normalize_arguments,
6 },
7 context::{annotations, clean_context, clone_response_context},
8 response_events::ResponseEvents,
9 usage::ClaudeProgress,
10};
11use crate::{
12 transform::generate::stream::claude::{ClaudeStreamCollector, ClaudeStreamLimits},
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::{input as i, response as r, stream as s},
20 },
21};
22use std::collections::{BTreeMap, BTreeSet, VecDeque};
23
24pub(super) struct Block {
25 pub output_index: Option<i64>,
26 pub item: Option<r::ResponseOutputItem>,
27 pub closed: bool,
28 pub deferred: bool,
29}
30
31pub struct ClaudeToResponsesStream {
32 source: Option<ClaudeStreamCollector>,
33 context: ClaudeToResponsesContext,
34 pub(super) events: ResponseEvents,
35 pub(super) flow: IdentityFlow,
36 pub(super) policy: TargetIdPolicy,
37 pub(super) budget: Budget,
38 pub(super) limits: StreamLimits,
39 pub(super) blocks: BTreeMap<i64, Block>,
40 pub(super) output_items: usize,
41
42 pub(super) argument_bytes: usize,
43 pub(super) client_tools: crate::transform::generate::client_tools::Bindings,
44 pub(super) deferred_block: Option<i64>,
45 pub(super) deferred_bytes: usize,
46 pending_events: BTreeMap<usize, (cs::StreamEvent, usize)>,
47 pending_blocks: BTreeMap<i64, VecDeque<usize>>,
48 next_pending: usize,
49 pub(super) mcp_calls: BTreeMap<String, i64>,
50 pub(super) mcp_results: BTreeSet<String>,
51 usage: ClaudeProgress,
52 failed: bool,
53 stopped: bool,
54}
55
56impl ClaudeToResponsesStream {
57 pub fn new(
58 context: impl Into<ClaudeToResponsesContext>,
59 flow: IdentityFlow,
60 limits: StreamLimits,
61 ) -> Result<Self, TransformError> {
62 Self::new_with_policy(
63 context,
64 flow,
65 limits,
66 TargetIdPolicy::new(crate::Dialect::OpenAi),
67 )
68 }
69 pub fn new_with_policy(
70 context: impl Into<ClaudeToResponsesContext>,
71 flow: IdentityFlow,
72 limits: StreamLimits,
73 policy: TargetIdPolicy,
74 ) -> Result<Self, TransformError> {
75 if policy.dialect != crate::Dialect::OpenAi {
76 return Err(TransformError::shape(
77 "stream.policy",
78 "target policy dialect mismatch",
79 ));
80 }
81
82 let context = clean_context(context.into(), limits)?;
83 let client_tools = crate::transform::generate::client_tools::Bindings::for_target(
84 &context.response.request,
85 crate::Dialect::Claude,
86 )?;
87 Ok(Self {
88 source: Some(ClaudeStreamCollector::new(ClaudeStreamLimits {
89 max_text_bytes: limits.max_bytes,
90 max_json_bytes: limits.max_bytes,
91 })),
92 context,
93 events: ResponseEvents::new(limits),
94 flow,
95 policy,
96 budget: Budget::new(limits),
97 limits,
98 blocks: BTreeMap::new(),
99 output_items: 0,
100
101 argument_bytes: 0,
102 client_tools,
103 deferred_block: None,
104 deferred_bytes: 0,
105 pending_events: BTreeMap::new(),
106 pending_blocks: BTreeMap::new(),
107 next_pending: 0,
108 mcp_calls: BTreeMap::new(),
109 mcp_results: BTreeSet::new(),
110 usage: Default::default(),
111 failed: false,
112 stopped: false,
113 })
114 }
115 pub fn identities(&self) -> &IdentityFlow {
116 &self.flow
117 }
118 pub fn push(
119 &mut self,
120 event: cs::StreamEvent,
121 ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
122 if self.failed || self.stopped {
123 self.failed = true;
124 return Err(invalid("event after message_stop 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: cs::StreamEvent,
135 ) -> Result<Converted<Vec<s::StreamEvent>>, TransformError> {
136 self.budget.input(&event)?;
137 self.source
138 .as_mut()
139 .ok_or_else(|| invalid("source consumed"))?
140 .push(event.clone())?;
141 let mut out = Vec::new();
142 if self.deferred_block.is_some() && block_index(&event) != self.deferred_block {
143 let size = measure(
144 &event,
145 self.limits
146 .max_pending
147 .saturating_sub(self.argument_bytes)
148 .saturating_sub(self.deferred_bytes),
149 )?;
150 self.deferred_bytes += size;
151 let ordinal = self.next_pending;
152 self.next_pending = ordinal.checked_add(1).ok_or_else(limit)?;
153 if let Some(index) = block_index(&event) {
154 self.pending_blocks
155 .entry(index)
156 .or_default()
157 .push_back(ordinal);
158 }
159 self.pending_events.insert(ordinal, (event, size));
160 } else {
161 self.process(event, &mut out)?;
162 }
163 loop {
164 let next = match self.deferred_block {
165 Some(index) => self
166 .pending_blocks
167 .get(&index)
168 .and_then(|events| events.front())
169 .copied(),
170 None => self.pending_events.first_key_value().map(|(key, _)| *key),
171 };
172 let Some(next) = next else { break };
173 let (event, bytes) = self
174 .pending_events
175 .remove(&next)
176 .expect("selected queued event");
177 if let Some(index) = block_index(&event) {
178 let events = self
179 .pending_blocks
180 .get_mut(&index)
181 .expect("queued block index");
182 events.pop_front();
183 if events.is_empty() {
184 self.pending_blocks.remove(&index);
185 }
186 }
187 self.deferred_bytes -= bytes;
188 self.process(event, &mut out)?;
189 }
190 Ok(Converted {
191 value: out,
192 report: Report::default(),
193 })
194 }
195 fn process(
196 &mut self,
197 event: cs::StreamEvent,
198 out: &mut Vec<s::StreamEvent>,
199 ) -> Result<(), TransformError> {
200 match event {
201 cs::StreamEvent::MessageStart(v) => {
202 self.usage.start(&v.message.usage)?;
203 let id = id(
204 &mut self.flow,
205 &self.policy,
206 crate::Dialect::Claude,
207 IdentityRole::Response,
208 IdentityRole::Response,
209 Some(v.message.id),
210 0,
211 )?;
212 let mut base = clone_response_context(&self.context.response).into_response(
213 id,
214 self.context.response.created_at,
215 v.message.model,
216 )?;
217 base.status = Some(r::ResponseStatus::InProgress);
218 base.usage = Some(None);
219 measure(&base, self.limits.max_bytes)?;
220 self.events.emit(&mut self.budget, out, |sequence_number| {
221 s::StreamEvent::Created(s::ResponseCreated {
222 sequence_number,
223 response: base,
224 rest: Default::default(),
225 })
226 })?;
227 }
228 cs::StreamEvent::ContentBlockStart(v) => self.start_block(v, out)?,
229 cs::StreamEvent::ContentBlockDelta(v) => match v.delta {
230 cs::ContentBlockDelta::Text(vv) => self.payload(v.index, vv.text, out)?,
231 cs::ContentBlockDelta::Thinking(vv) => self.payload(v.index, vv.thinking, out)?,
232 cs::ContentBlockDelta::InputJson(vv) => {
233 self.payload(v.index, vv.partial_json, out)?
234 }
235 cs::ContentBlockDelta::Signature(_) | cs::ContentBlockDelta::Citations(_) => {}
236 cs::ContentBlockDelta::Compaction(_) => {}
237 },
238 cs::StreamEvent::ContentBlockStop(v) => self.stop_block(v.index, out)?,
239 cs::StreamEvent::MessageDelta(v) => self.usage.delta(&v.usage)?,
240 cs::StreamEvent::MessageStop(_) => self.stopped = true,
241 cs::StreamEvent::Ping(_) => {}
242 cs::StreamEvent::Error(v) => {
243 return Err(invalid(format!("{:?}: {}", v.error.type_, v.error.message)));
244 }
245 }
246 Ok(())
247 }
248 pub fn finish(mut self) -> Result<StreamEnd<s::StreamEvent>, TransformError> {
249 if self.failed
250 || !self.stopped
251 || !self.pending_events.is_empty()
252 || self.blocks.values().any(|v| !v.closed)
253 {
254 return Err(invalid("failed stream or premature EOF"));
255 }
256 let mut source = self
257 .source
258 .take()
259 .ok_or_else(|| invalid("source consumed"))?
260 .finish()?
261 .value;
262 self.usage
263 .finalize(&mut source.usage, self.context.response.usage)?;
264 let refusal = source.stop_reason == c::StopReason::Refusal;
265 let mut citation_report = Report::default();
266 let citations = annotations(
267 std::mem::take(&mut self.context.annotations),
268 &source.content,
269 &mut citation_report,
270 )?;
271 let converted = super::super::claude_to_responses_response(
272 source,
273 self.context.response,
274 &mut self.flow.clone(),
275 &self.policy,
276 )?;
277 let mut expected = converted.value;
278 let mut report = converted.report;
279 report
280 .diagnostics
281 .retain(|v| v.field != "content.citations");
282 report.diagnostics.extend(citation_report.diagnostics);
283 if refusal {
284 lower_refusal(&mut expected);
285 report.changed("stop_reason","late Claude refusal preserves output text and maps the terminal to incomplete/content_filter");
286 }
287 for (source_index, block) in &self.blocks {
288 let Some(index) = block.output_index else {
289 continue;
290 };
291 let item = expected
292 .output
293 .get_mut(index as usize)
294 .ok_or_else(|| invalid("missing canonical output item"))?;
295 if item_id(item) != block.item.as_ref().and_then(item_id) {
296 return Err(invalid("canonical item identity differs from emitted item"));
297 }
298 match (item, block.item.as_ref().unwrap()) {
299 (
300 r::ResponseOutputItem::FunctionCall(target),
301 r::ResponseOutputItem::FunctionCall(actual),
302 ) => target.arguments = checked_raw(&target.arguments, &actual.arguments)?,
303 (
304 r::ResponseOutputItem::McpCall(target),
305 r::ResponseOutputItem::McpCall(actual),
306 ) => target.arguments = checked_raw(&target.arguments, &actual.arguments)?,
307 (r::ResponseOutputItem::Message(target), _) => {
308 if let Some(facts) = citations.get(&(*source_index as usize))
309 && let Some(i::OutputContent::Text(part)) = target.content.first_mut()
310 {
311 part.annotations = facts.clone();
312 }
313 }
314 _ => {}
315 }
316 }
317 measure(&expected, self.limits.max_bytes)?;
318 let mut out = Vec::new();
319 for (index, item) in expected.output.iter().cloned().enumerate() {
320 self.events
321 .item_done(&mut self.budget, &mut out, index as i64, item)?;
322 }
323 self.events
324 .emit(&mut self.budget, &mut out, |sequence_number| {
325 if expected.status == Some(r::ResponseStatus::Incomplete) {
326 s::StreamEvent::Incomplete(s::ResponseIncomplete {
327 sequence_number,
328 response: expected.clone(),
329 rest: Default::default(),
330 })
331 } else {
332 s::StreamEvent::Completed(s::ResponseCompleted {
333 sequence_number,
334 response: expected.clone(),
335 rest: Default::default(),
336 })
337 }
338 })?;
339 let mut actual = self.events.finish()?;
340 normalize_arguments(&mut actual)?;
341 normalize_arguments(&mut expected)?;
342 if actual != expected {
343 return Err(invalid(
344 "native Responses output differs from canonical source conversion",
345 ));
346 }
347 Ok(StreamEnd {
348 chunks: out,
349 identities: self.flow,
350 report,
351 })
352 }
353}
354
355fn lower_refusal(response: &mut r::GenerateContentResponseBody) {
356 for item in &mut response.output {
357 if let r::ResponseOutputItem::Message(message) = item {
358 for part in &mut message.content {
359 if let i::OutputContent::Refusal(value) = part {
360 *part = i::OutputContent::Text(super::response_events::text(
361 std::mem::take(&mut value.refusal),
362 Vec::new(),
363 ));
364 }
365 }
366 }
367 }
368 response.status = Some(r::ResponseStatus::Incomplete);
369 response.incomplete_details = Some(r::ResponseIncompleteDetails {
370 reason: Some(r::ResponseIncompleteReason::ContentFilter),
371 rest: Default::default(),
372 });
373}
374
375fn block_index(event: &cs::StreamEvent) -> Option<i64> {
376 match event {
377 cs::StreamEvent::ContentBlockStart(v) => Some(v.index),
378 cs::StreamEvent::ContentBlockDelta(v) => Some(v.index),
379 cs::StreamEvent::ContentBlockStop(v) => Some(v.index),
380 _ => None,
381 }
382}
383
384fn checked_raw(canonical: &str, actual: &str) -> Result<String, TransformError> {
385 let canonical: serde_json::Map<String, serde_json::Value> =
386 serde_json::from_str(canonical).map_err(|e| invalid(e.to_string()))?;
387 let parsed: serde_json::Map<String, serde_json::Value> =
388 serde_json::from_str(actual).map_err(|e| invalid(e.to_string()))?;
389 if canonical != parsed {
390 return Err(invalid(
391 "streamed arguments differ from actual native tool input",
392 ));
393 }
394 Ok(actual.to_owned())
395}