Merge pull request #816 from fabro-sh/reject-tool-call-index-gaps

Reject tool call index gaps in Chat Completions streams

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<Vec<StreamEvent>> {
+    fn process_chunk(&mut self, mut chunk: StreamChunk) -> Result<Option<Vec<StreamEvent>>, 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<StreamEvent> {
@@ -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");