diff --git a/internal/providers/defaults.go b/internal/providers/defaults.go index bb464703..13a4420b 100644 --- a/internal/providers/defaults.go +++ b/internal/providers/defaults.go @@ -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 diff --git a/internal/providers/sse_reader.go b/internal/providers/sse_reader.go index d1a4e408..0f9ed170 100644 --- a/internal/providers/sse_reader.go +++ b/internal/providers/sse_reader.go @@ -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 { diff --git a/internal/providers/sse_reader_bigline_test.go b/internal/providers/sse_reader_bigline_test.go new file mode 100644 index 00000000..79333a5b --- /dev/null +++ b/internal/providers/sse_reader_bigline_test.go @@ -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()) + } +}