diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs b/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs index bb2633b91..3207c2744 100644 --- a/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs +++ b/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs @@ -56,7 +56,7 @@ impl StreamState { } /// Process a parsed SSE chunk and return events to emit, if any. - fn process_chunk(&mut self, mut chunk: StreamChunk) -> Option> { + fn process_chunk(&mut self, mut chunk: StreamChunk) -> Result>, Error> { // Capture response metadata from the first chunk. if let Some(id) = &chunk.id { if self.response_id.is_empty() { @@ -80,8 +80,12 @@ impl StreamState { .or_else(|| chunk.cost.as_ref().and_then(|cost| cost.usd)); self.cost_usd = cost_usd.or(self.cost_usd); - let choices = chunk.choices.as_mut()?; - let choice = choices.first_mut()?; + let Some(choices) = chunk.choices.as_mut() else { + return Ok(None); + }; + let Some(choice) = choices.first_mut() else { + return Ok(None); + }; let mut events = Vec::new(); @@ -90,7 +94,9 @@ impl StreamState { self.finish_reason = map_finish_reason(Some(reason.as_str())); } - let delta = choice.delta.as_mut()?; + let Some(delta) = choice.delta.as_mut() else { + return Ok(None); + }; // Accumulate reasoning/thinking content (Kimi, etc.). if let Some(reasoning) = delta.reasoning() { @@ -121,8 +127,22 @@ impl StreamState { for tc in tool_calls { let index = tc.index; - // Grow the accumulated tool calls vector if needed. - while self.tool_calls.len() <= index { + // A delta may only continue an already-started tool call or + // open the next slot. Padding a skipped slot would materialize + // a phantom tool call with an empty id and name, which poisons + // the conversation once echoed back to the provider. + if index > self.tool_calls.len() { + return Err(Error::Stream { + message: format!( + "malformed tool call stream from {}: delta for tool_calls[{index}] \ + arrived before tool_calls[{}] was started", + self.provider_name, + self.tool_calls.len() + ), + source: None, + }); + } + if index == self.tool_calls.len() { self.tool_calls.push(AccumulatedToolCall { id: String::new(), name: String::new(), @@ -163,9 +183,9 @@ impl StreamState { } if events.is_empty() { - None + Ok(None) } else { - Some(events) + Ok(Some(events)) } } @@ -262,7 +282,7 @@ impl StreamDecoder for StreamState { let chunk: StreamChunk = serde_json::from_str(ev.data) .map_err(|e| Error::stream_error(format!("failed to parse SSE chunk: {e}"), e))?; - Ok(self.process_chunk(chunk).unwrap_or_default()) + Ok(self.process_chunk(chunk)?.unwrap_or_default()) } fn finish(&mut self) -> Vec { @@ -373,7 +393,7 @@ mod tests { let chunk1: StreamChunk = serde_json::from_str( r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#, ).unwrap(); - let events1 = state.process_chunk(chunk1).unwrap(); + let events1 = state.process_chunk(chunk1).unwrap().unwrap(); assert_eq!(events1.len(), 2); assert!(matches!(events1[0], StreamEvent::TextStart { .. })); assert!(matches!(events1[1], StreamEvent::TextDelta { .. })); @@ -381,7 +401,7 @@ mod tests { let chunk2: StreamChunk = serde_json::from_str( r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":" world"},"finish_reason":null}]}"#, ).unwrap(); - let events2 = state.process_chunk(chunk2).unwrap(); + let events2 = state.process_chunk(chunk2).unwrap().unwrap(); assert_eq!(events2.len(), 1); assert!(matches!(events2[0], StreamEvent::TextDelta { .. })); @@ -395,14 +415,14 @@ mod tests { let chunk1: StreamChunk = serde_json::from_str( r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"fn1","arguments":"{\"k"}}]},"finish_reason":null}]}"#, ).unwrap(); - let events1 = state.process_chunk(chunk1).unwrap(); + let events1 = state.process_chunk(chunk1).unwrap().unwrap(); assert_eq!(events1.len(), 1); assert!(matches!(events1[0], StreamEvent::ToolCallStart { .. })); let chunk2: StreamChunk = serde_json::from_str( r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ey\"}"}}]},"finish_reason":null}]}"#, ).unwrap(); - let events2 = state.process_chunk(chunk2).unwrap(); + let events2 = state.process_chunk(chunk2).unwrap().unwrap(); assert_eq!(events2.len(), 1); assert!(matches!(events2[0], StreamEvent::ToolCallDelta { .. })); @@ -479,6 +499,37 @@ mod tests { } } + // Reproduces run 01M11JZVT7V507R56BCJJHZB1B: venice (proxying Anthropic) + // numbered tool_calls[].index by content block, so the first tool call + // arrived with index 1 when text preceded it. Padding the skipped slot + // used to materialize a phantom tool call with an empty id and name that + // the provider rejected once echoed back (tool_use.id must match + // '^[a-zA-Z0-9_-]+$'). A gap in the index sequence is indistinguishable + // from lost chunks, so the stream must fail instead. + #[test] + fn sparse_tool_call_index_is_a_stream_error() { + let mut state = test_state("venice", "claude-opus-5"); + + let text_chunk: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"claude-opus-5","choices":[{"delta":{"content":"I'll start by reading the state file."},"finish_reason":null}]}"#, + ) + .unwrap(); + state.process_chunk(text_chunk).unwrap(); + + let tool_chunk: StreamChunk = serde_json::from_str( + r#"{"id":"c1","model":"claude-opus-5","choices":[{"delta":{"tool_calls":[{"index":1,"id":"toolu_01EgMidFVtGhitWE22jXQ9Eo","function":{"name":"Read","arguments":"{\"file_path\":\"state.json\"}"}}]},"finish_reason":null}]}"#, + ) + .unwrap(); + let err = state.process_chunk(tool_chunk).unwrap_err(); + + assert!(err.retryable(), "malformed stream should be retryable"); + let message = err.to_string(); + assert!( + message.contains("tool_calls[1]") && message.contains("venice"), + "unexpected error message: {message}" + ); + } + #[test] fn uses_request_model_as_fallback() { let mut state = test_state("test", "fallback-model");