Skip to content

fix(runtime): don't truncate the SSE stream on tool results >64 KiB #3883

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sayt-0 merged 1 commit into from
Aug 4, 2026
+28 −0
Merged

fix(runtime): don't truncate the SSE stream on tool results >64 KiB #3883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions pkg/runtime/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,10 @@ func (c *Client) runAgentWithAgentName(ctx context.Context, sessionID, agent, ag
defer resp.Body.Close()

scanner := bufio.NewScanner(resp.Body)
// A single SSE line can carry a large tool response; raise the cap
// above bufio's 64 KiB default so an oversized line does not silently
// truncate the stream (bufio.ErrTooLong).
scanner.Buffer(make([]byte, 0, bufio.MaxScanTokenSize), maxSSELineBytes)
for scanner.Scan() {
line := scanner.Bytes()
if len(line) == 0 || line[0] == ':' {
Expand Down Expand Up @@ -438,7 +442,12 @@ func (c *Client) runAgentWithAgentName(ctx context.Context, sessionID, agent, ag
eventChan <- e
}

// Surface a read failure (e.g. an over-long line) instead of ending
// the stream silently — otherwise the run appears to stop with no
// error after the last event that fit.
if err := scanner.Err(); err != nil {
slog.DebugContext(ctx, "event", "scanner_error", err)
eventChan <- Error(fmt.Sprintf("reading event stream: %v", err))

@docker-agent docker-agent Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[low] Blocking eventChan <- Error(...) send after scanner error can leak the goroutine if the consumer has already exited

After scanner.Scan() returns due to an error (including context.Canceled from HTTP-request cancellation), the new code sends one Error event on eventChan before the deferred close(eventChan) runs. The send is a plain blocking operation on a buffered channel (capacity 128). If the consumer (for streamEvent := range streamChan) has already stopped draining the channel — for example because its own context was cancelled and it returned before seeing this final event — the goroutine parks on the blocked send indefinitely, defer close(eventChan) never executes, and the goroutine leaks.

The same risk already exists for every eventChan <- e inside the for scanner.Scan() loop, so this PR does not introduce a new class of bug. However, the context-cancellation path is a new specific scenario where this send is reached after the consumer is most likely gone: when ctx is cancelled, scanner.Scan() returns false, scanner.Err() returns context.Canceled, and the new code immediately attempts the send — right when the consumer has already left. In practice the 128-slot buffer almost always has room, but under back-pressure (slow consumer, burst of events) it is not guaranteed.

Consider guarding the send with a select to avoid blocking:

if err := scanner.Err(); err != nil {
    slog.DebugContext(ctx, "event", "scanner_error", err)
    select {
    case eventChan <- Error(fmt.Sprintf("reading event stream: %v", err)):
    default:
        // consumer gone or buffer full; drop the error event
        slog.WarnContext(ctx, "event channel full, dropping scanner error", "err", err)
    }
    return
}

The same pattern applies to the identical block in StreamSessionEvents (line 600).

Confidence Score
🟡 moderate 70/100

return
}
}()
Expand Down Expand Up @@ -541,6 +550,10 @@ func (c *Client) StreamSessionEvents(ctx context.Context, sessionID string) (<-c
defer resp.Body.Close()

scanner := bufio.NewScanner(resp.Body)
// A single SSE line can carry a large tool response; raise the cap
// above bufio's 64 KiB default so an oversized line does not silently
// truncate the stream (bufio.ErrTooLong).
scanner.Buffer(make([]byte, 0, bufio.MaxScanTokenSize), maxSSELineBytes)
for scanner.Scan() {
line := scanner.Bytes()
if len(line) == 0 || line[0] == ':' {
Expand Down Expand Up @@ -579,8 +592,12 @@ func (c *Client) StreamSessionEvents(ctx context.Context, sessionID string) (<-c
eventChan <- e
}

// Surface a read failure (e.g. an over-long line) instead of ending
// the stream silently — otherwise the run appears to stop with no
// error after the last event that fit.
if err := scanner.Err(); err != nil {
slog.DebugContext(ctx, "scanner error", "error", err)
eventChan <- Error(fmt.Sprintf("reading event stream: %v", err))
}
}()

Expand Down
11 changes: 11 additions & 0 deletions pkg/runtime/defaults.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,17 @@ import "time"
// so the buffer size is set in exactly one place.
const defaultEventChannelCapacity = 128

// maxSSELineBytes bounds a single Server-Sent-Events line the client SSE
// reader will accept. Each event is delivered as one `data: {json}` line, and
// a tool result can be large (e.g. a big tool-response payload), so the default
// bufio.Scanner cap of 64 KiB (bufio.MaxScanTokenSize) truncates the stream:
// the oversized line trips bufio.ErrTooLong, the scan ends, and the run appears
// to stop silently after the preceding event. The reader raises its buffer to
// this ceiling and surfaces any remaining scanner error as an Error event
// rather than closing the stream without a trace. Sized to comfortably hold a
// large tool response while still bounding per-line memory.
const maxSSELineBytes = 16 * 1024 * 1024

// defaultMaxOverflowCompactions caps the number of consecutive
// context-overflow auto-compactions that the run loop will attempt before
// giving up and surfacing the error to the caller. The runtime's
Expand Down
Loading