mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 16:17:25 +00:00
fix(providers): unbounded SSE line reader to stop image_generation "token too long" crash (#1349)
bufio.Scanner has a fixed max-token size (SSEScanBufMax, 1MB). Codex's native image_generation streams a full base64 image in a SINGLE SSE data line (response.image_generation_call.partial_image and the final image_generation_call item), which regularly exceeds 1MB for anything larger than a thumbnail. Scanner.Scan() then fails with: stream read error: bufio.Scanner: token too long killing the whole turn after the model already generated the image. Replace bufio.Scanner with bufio.Reader.ReadString, which grows to fit a line of any length instead of erroring past a fixed cap. SSEScanBufMax is now unused and removed; SSEScanBufInit is reused as the reader's initial buffer size.
This commit is contained in:
1 parent
cde8940501
commit
016263eb84
3 files changed
+104
-37
No files matched your search
@@ -11,9 +11,10 @@ const (
|
||||
// ctx cancellation from unblocking bufio.Scanner. Use NewDefaultHTTPClient() instead.
|
||||
DefaultHTTPTimeout = 300 * time.Second
|
||||
|
||||
// SSE stream scanner buffer sizes (OpenAI-compat, Anthropic, Codex).
|
||||
SSEScanBufInit = 64 * 1024 // 64KB initial buffer
|
||||
SSEScanBufMax = 1024 * 1024 // 1MB max line for large tool call / thinking chunks
|
||||
// SSE stream reader initial buffer size (OpenAI-compat, Anthropic, Codex).
|
||||
// No max: bufio.Reader.ReadString grows to fit a line of any length — a full
|
||||
// base64 image in a single image-generation SSE data line can exceed several MB.
|
||||
SSEScanBufInit = 64 * 1024 // 64KB initial buffer
|
||||
|
||||
// Stdio/JSONRPC scanner buffer sizes (Claude CLI, ACP).
|
||||
StdioScanBufInit = 256 * 1024 // 256KB initial buffer
|
||||
|
||||
@@ -11,55 +11,76 @@ import (
|
||||
// SSEScanner reads an SSE (Server-Sent Events) stream line by line,
|
||||
// extracting data payloads. Shared by OpenAI, Anthropic, and Codex providers.
|
||||
type SSEScanner struct {
|
||||
scanner *bufio.Scanner
|
||||
reader *bufio.Reader
|
||||
data string
|
||||
eventType string
|
||||
err error
|
||||
done bool
|
||||
}
|
||||
|
||||
// NewSSEScanner creates an SSE scanner with appropriate buffer sizes.
|
||||
// NewSSEScanner creates an SSE scanner.
|
||||
//
|
||||
// It reads lines with bufio.Reader.ReadString, which grows to fit a line of
|
||||
// any length, instead of bufio.Scanner, whose fixed max-token size (formerly
|
||||
// SSEScanBufMax) overflowed with "token too long" on a single data line
|
||||
// carrying a full base64 image — image generation streams multi-MB frames in
|
||||
// a single SSE line. SSEScanBufInit is reused as the reader's initial size.
|
||||
func NewSSEScanner(r io.Reader) *SSEScanner {
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 0, SSEScanBufInit), SSEScanBufMax)
|
||||
return &SSEScanner{scanner: sc}
|
||||
return &SSEScanner{reader: bufio.NewReaderSize(r, SSEScanBufInit)}
|
||||
}
|
||||
|
||||
// Next advances to the next data line. Returns false when the stream ends
|
||||
// or "[DONE]" is encountered. After Next returns false, call Err() for errors.
|
||||
func (s *SSEScanner) Next() bool {
|
||||
for s.scanner.Scan() {
|
||||
line := s.scanner.Text()
|
||||
if s.done {
|
||||
return false
|
||||
}
|
||||
for {
|
||||
line, err := s.reader.ReadString('\n')
|
||||
if len(line) > 0 {
|
||||
// Trim the trailing newline ("\n" or "\r\n").
|
||||
line = strings.TrimSuffix(line, "\n")
|
||||
line = strings.TrimSuffix(line, "\r")
|
||||
|
||||
// Track event type (e.g. "event: message_start")
|
||||
if after, ok := strings.CutPrefix(line, "event: "); ok {
|
||||
s.eventType = after
|
||||
continue
|
||||
if after, ok := strings.CutPrefix(line, "event: "); ok {
|
||||
s.eventType = after
|
||||
} else if after, ok := strings.CutPrefix(line, "event:"); ok {
|
||||
s.eventType = strings.TrimSpace(after)
|
||||
} else {
|
||||
payload, isData := "", false
|
||||
if after, ok := strings.CutPrefix(line, "data: "); ok {
|
||||
payload, isData = after, true
|
||||
} else if after, ok := strings.CutPrefix(line, "data:"); ok {
|
||||
payload, isData = after, true
|
||||
}
|
||||
if isData {
|
||||
// "[DONE]" is the OpenAI/Codex stream terminator.
|
||||
if payload == "[DONE]" {
|
||||
s.done = true
|
||||
return false
|
||||
}
|
||||
s.data = payload
|
||||
// A final line may arrive together with io.EOF (no trailing
|
||||
// newline). Deliver it now; the next call reports the end.
|
||||
if err != nil {
|
||||
s.done = true
|
||||
if err != io.EOF {
|
||||
s.err = err
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
// Non event/data line (blank line, ":" comment, other field): skip.
|
||||
}
|
||||
}
|
||||
if after, ok := strings.CutPrefix(line, "event:"); ok {
|
||||
s.eventType = strings.TrimSpace(after)
|
||||
continue
|
||||
}
|
||||
|
||||
// Extract data payload
|
||||
var payload string
|
||||
if after, ok := strings.CutPrefix(line, "data: "); ok {
|
||||
payload = after
|
||||
} else if after, ok := strings.CutPrefix(line, "data:"); ok {
|
||||
payload = after
|
||||
} else {
|
||||
continue // skip empty lines, comments, other fields
|
||||
}
|
||||
|
||||
// "[DONE]" is the OpenAI/Codex stream terminator
|
||||
if payload == "[DONE]" {
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
s.err = err
|
||||
}
|
||||
s.done = true
|
||||
return false
|
||||
}
|
||||
|
||||
s.data = payload
|
||||
return true
|
||||
}
|
||||
s.err = s.scanner.Err()
|
||||
return false
|
||||
}
|
||||
|
||||
// Data returns the current data payload (valid after Next returns true).
|
||||
@@ -78,7 +99,7 @@ func (s *SSEScanner) Err() error {
|
||||
}
|
||||
|
||||
// CtxBody wraps an http.Response.Body so that ctx cancellation closes the
|
||||
// underlying socket, unblocking a goroutine stuck inside bufio.Scanner.Scan().
|
||||
// underlying socket, unblocking a goroutine stuck inside a blocking read.
|
||||
// Safe for concurrent Close (sync.Once). Caller MUST defer Close() to release
|
||||
// the watchdog goroutine even on success.
|
||||
type CtxBody struct {
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package providers
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Reproduces the image-generation failure: a single SSE data line carrying a
|
||||
// multi-MB base64 image used to overflow bufio.Scanner ("token too long").
|
||||
func TestSSEScannerHandlesMultiMBLine(t *testing.T) {
|
||||
big := strings.Repeat("A", 3*1024*1024) // 3MB, well over the old 1MB cap
|
||||
stream := "event: response.image_generation_call.partial_image\n" +
|
||||
"data: " + big + "\n" +
|
||||
"data: [DONE]\n"
|
||||
sc := NewSSEScanner(strings.NewReader(stream))
|
||||
if !sc.Next() {
|
||||
t.Fatalf("expected first data line; err=%v", sc.Err())
|
||||
}
|
||||
if len(sc.Data()) != len(big) {
|
||||
t.Fatalf("data len=%d want %d", len(sc.Data()), len(big))
|
||||
}
|
||||
if sc.EventType() != "response.image_generation_call.partial_image" {
|
||||
t.Fatalf("eventType=%q", sc.EventType())
|
||||
}
|
||||
if sc.Next() {
|
||||
t.Fatalf("expected [DONE] to end stream, got %q", sc.Data())
|
||||
}
|
||||
if sc.Err() != nil {
|
||||
t.Fatalf("unexpected err: %v", sc.Err())
|
||||
}
|
||||
}
|
||||
|
||||
// Final data line arriving with EOF and no trailing newline must be delivered.
|
||||
func TestSSEScannerFinalLineNoNewline(t *testing.T) {
|
||||
sc := NewSSEScanner(strings.NewReader("data: hello"))
|
||||
if !sc.Next() || sc.Data() != "hello" {
|
||||
t.Fatalf("data=%q err=%v", sc.Data(), sc.Err())
|
||||
}
|
||||
if sc.Next() {
|
||||
t.Fatalf("expected end")
|
||||
}
|
||||
if sc.Err() != nil {
|
||||
t.Fatalf("err=%v", sc.Err())
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user