From c9bbe8b3a8f888eb54e10313225d015d2befbdcd Mon Sep 17 00:00:00 2001
From: kanylbullen <x@xuper.fun>
Date: Tue, 14 Apr 2026 08:28:58 +0000
Subject: [PATCH] Fix SSE stream parser dropping tool calls on EOF

The read loop in ChatStream breaks immediately on io.EOF without
processing remaining buffered data. Per Go's io.Reader contract,
Read may return both n > 0 and io.EOF simultaneously, so the final
bytes (which may contain tool call deltas and [DONE]) are silently
discarded.

This causes the agentic loop to see tool_calls=0 even though the
model correctly produced tool calls in the stream.

Changes:
- Process pendingData when EOF is received before breaking
- Add fallback: emit accumulated tool calls if [DONE] was never
  reached (server closed connection early)

Fixes #1411

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 internal/ai/providers/openai.go | 50 +++++++++++++++++++++++++++++++--
 1 file changed, 47 insertions(+), 3 deletions(-)

diff --git a/internal/ai/providers/openai.go b/internal/ai/providers/openai.go
index 62eca76e9..6ec506d57 100644
--- a/internal/ai/providers/openai.go
+++ b/internal/ai/providers/openai.go
@@ -714,16 +714,25 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback
 	var inputTokens, outputTokens int
 	var finishReason string
 
+	atEOF := false
 	for {
 		n, err := reader.Read(buf)
+		if n > 0 {
+			pendingData += string(buf[:n])
+		}
 		if err != nil {
-			if err == io.EOF {
+			if err != io.EOF {
+				return fmt.Errorf("stream read error: %w", err)
+			}
+			atEOF = true
+			// At EOF, process any remaining pendingData then break
+			if pendingData == "" {
 				break
 			}
-			return fmt.Errorf("stream read error: %w", err)
+			// Ensure trailing data is processed by appending a newline
+			pendingData += "\n"
 		}
 
-		pendingData += string(buf[:n])
 		lines := strings.Split(pendingData, "\n")
 
 		// Keep the last incomplete line for next iteration
@@ -839,6 +848,41 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback
 				}
 			}
 		}
+
+		if atEOF {
+			break
+		}
+	}
+
+	// If we exited the loop without hitting [DONE] (e.g. server closed connection),
+	// still build and emit any accumulated tool calls so they aren't silently dropped.
+	if len(toolCallBuilders) > 0 && len(toolCalls) == 0 {
+		for _, builder := range toolCallBuilders {
+			var input map[string]interface{}
+			if err := json.Unmarshal([]byte(builder.args.String()), &input); err != nil {
+				input = map[string]interface{}{"raw": builder.args.String()}
+			}
+			toolCalls = append(toolCalls, ToolCall{
+				ID:    builder.id,
+				Name:  builder.name,
+				Input: input,
+			})
+		}
+		stopReason := finishReason
+		if len(toolCalls) > 0 {
+			stopReason = "tool_use"
+		} else if stopReason == "stop" || stopReason == "" {
+			stopReason = "end_turn"
+		}
+		callback(StreamEvent{
+			Type: "done",
+			Data: DoneEvent{
+				StopReason:   stopReason,
+				ToolCalls:    toolCalls,
+				InputTokens:  inputTokens,
+				OutputTokens: outputTokens,
+			},
+		})
 	}
 
 	return nil