mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-11 14:00:29 +00:00
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>
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user