Prevent stream hangs when SSE headers are missing - #3928
Conversation
## Summary Removes the redundant changelog header and version label from `cli/changelog.md` and fixes a missing newline at the end of the file. ## Changes - Removed the `# Bifrost CLI Changelog` heading and `## v0.10.5` version label from the top of the changelog - Added a trailing newline to the final line of the file to resolve the "no newline at end of file" issue ## Type of change - [ ] Bug fix - [ ] Feature - [ ] Refactor - [ ] Documentation - [x] Chore/CI ## Affected areas - [ ] Core (Go) - [ ] Transports (HTTP) - [ ] Providers/Integrations - [ ] Plugins - [ ] UI (React) - [x] Docs ## How to test Verify the changelog renders correctly and that no trailing newline warning appears in diff tooling. ```sh cat -A cli/changelog.md | tail -5 ``` The last line should end with `$` (indicating a proper newline terminator). ## Screenshots/Recordings N/A ## Breaking changes - [ ] Yes - [x] No ## Related issues N/A ## Security considerations None. ## Checklist - [ ] I read `docs/contributing/README.md` and followed the guidelines - [ ] I added/updated tests where appropriate - [x] I updated documentation where needed - [x] I verified builds succeed (Go and UI) - [ ] I verified the CI pipeline passes locally if applicable <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Release Notes * **Documentation** * Updated changelog to document CLI command overlay improvements enabling arrow key navigation in tab popups. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
Some OpenAI-compatible backends emit valid SSE frames without declaring text/event-stream. The previous non-SSE guard relied only on the response header and drained those streams before the scanner could observe terminal events like response.completed. This keeps the existing drain behavior for JSON/non-line-delimited error bodies, but lets wrapped streaming readers sniff a small prefix and replay it when the bytes look like SSE frames. Constraint: Codex backend responses can be valid SSE without Content-Type: text/event-stream Rejected: Disable the non-SSE drain entirely | it protects scanners from large non-line-delimited error bodies Confidence: high Scope-risk: narrow Directive: Keep stream guards reader-preserving after gzip/idle-timeout wrapping; do not read directly from resp.BodyStream in wrapped streaming paths Tested: docker go test ./providers/utils -run TestDrainNonSSEStream Tested: docker go test ./providers/openai ./providers/utils Co-authored-by: OmX <omx@oh-my-codex.dev>
|
Duong M. CUONG seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
📝 WalkthroughWalkthroughThe PR refactors OpenAI streaming handlers to detect and handle non-SSE responses using a new reader-based drain helper with prefix sniffing. Instead of relying solely on the ChangesSSE Non-Drain Detection via Reader-Based Sniffing
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/providers/utils/utils.go`:
- Around line 1024-1039: DrainNonSSEStreamReader currently decides SSE vs
non‑SSE after a single reader.Read which can return a short partial chunk;
instead loop to accumulate reads into buf (using reader.Read into buf[:]) until
you either have enough bytes to reliably call looksLikeSSEStreamPrefix(prefix)
(use the same prefix length checks as looksLikeSSEStreamPrefix expects), hit EOF
or an error, or fill the 512‑byte buffer; once you have a sufficient prefix,
reconstruct the reader with io.MultiReader(bytes.NewReader(prefix), reader) and
only return (reader, false) if looksLikeSSEStreamPrefix(prefix) is true,
otherwise drain and return (reader, true); ensure you still handle readErr !=
nil (and treat non‑EOF errors as draining case) and avoid consuming the stream
prematurely.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: a761f0cd-ab8f-40bc-828e-55dcc39a55f1
📒 Files selected for processing (3)
core/providers/openai/openai.gocore/providers/utils/utils.gocore/providers/utils/utils_test.go
| var buf [512]byte | ||
| n, readErr := reader.Read(buf[:]) | ||
| if n > 0 { | ||
| prefix := append([]byte(nil), buf[:n]...) | ||
| reader = io.MultiReader(bytes.NewReader(prefix), reader) | ||
| if looksLikeSSEStreamPrefix(prefix) { | ||
| return reader, false | ||
| } | ||
| } | ||
| return true | ||
|
|
||
| if readErr != nil && readErr != io.EOF { | ||
| return reader, true | ||
| } | ||
|
|
||
| _, _ = io.Copy(io.Discard, reader) | ||
| return reader, true |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and show the current implementation around the reported lines
rg -n "func DrainNonSSEStreamReader|DrainNonSSEStreamReader\(" core/providers/utils/utils.go
rg -n "looksLikeSSEStreamPrefix|func looksLikeSSEStreamPrefix" core/providers/utils/utils.go
# Print the relevant section (keep it small)
nl -ba core/providers/utils/utils.go | sed -n '980,1100p'Repository: maximhq/bifrost
Length of output: 393
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show exact code around DrainNonSSEStreamReader and looksLikeSSEStreamPrefix
sed -n '990,1065p' core/providers/utils/utils.go
echo "----"
sed -n '1065,1125p' core/providers/utils/utils.go
echo "----"
# Also show the call site context around line ~1004
sed -n '960,1035p' core/providers/utils/utils.goRepository: maximhq/bifrost
Length of output: 7439
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Get the rest of the function
sed -n '1030,1100p' core/providers/utils/utils.go
echo "---- call sites ----"
rg -n "DrainNonSSEStream(Response|Reader)\(" -S core | head -n 50
echo "---- SSE scanner usage ----"
rg -n "bufio\.NewScanner|Scan\(|SplitFunc|scanner" core/providers -S | head -n 50Repository: maximhq/bifrost
Length of output: 7324
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect call-site logic around one of the DrainNonSSEStreamReader uses
sed -n '520,610p' core/providers/openai/openai.go
echo "----"
sed -n '1080,1145p' core/providers/openai/openai.go
echo "---- tests ----"
sed -n '680,880p' core/providers/utils/utils_test.goRepository: maximhq/bifrost
Length of output: 12545
Don’t classify SSE from a single short Read in DrainNonSSEStreamReader
core/providers/utils/utils.go DrainNonSSEStreamReader does only one reader.Read(buf[:]) sniff and drains the stream unless looksLikeSSEStreamPrefix(prefix) matches. io.Reader.Read is allowed to return partial data on the first call, so a valid SSE stream (with missing/incorrect Content-Type) can be misclassified and fully drained (e.g., first read yields d / da / eve, so the prefix doesn’t match).
Current tests cover full data:/event: bodies from bytes.NewReader (single-read), but don’t exercise fragmented/short-read behavior.
Suggested direction
func DrainNonSSEStreamReader(resp *fasthttp.Response, reader io.Reader) (io.Reader, bool) {
if resp == nil || reader == nil {
return reader, true
}
ct := strings.ToLower(string(resp.Header.ContentType()))
if strings.Contains(ct, "text/event-stream") {
return reader, false
}
- var buf [512]byte
- n, readErr := reader.Read(buf[:])
- if n > 0 {
- prefix := append([]byte(nil), buf[:n]...)
- reader = io.MultiReader(bytes.NewReader(prefix), reader)
- if looksLikeSSEStreamPrefix(prefix) {
- return reader, false
- }
- }
-
- if readErr != nil && readErr != io.EOF {
- return reader, true
- }
-
- _, _ = io.Copy(io.Discard, reader)
- return reader, true
+ var buf [512]byte
+ prefix := make([]byte, 0, len(buf))
+
+ for len(prefix) < len(buf) {
+ n, readErr := reader.Read(buf[len(prefix):])
+ if n > 0 {
+ prefix = append(prefix, buf[len(prefix):len(prefix)+n]...)
+ if looksLikeSSEStreamPrefix(prefix) {
+ return io.MultiReader(bytes.NewReader(prefix), reader), false
+ }
+ if bytes.ContainsRune(prefix, '\n') {
+ break
+ }
+ }
+ if readErr != nil {
+ reader = io.MultiReader(bytes.NewReader(prefix), reader)
+ if readErr != io.EOF {
+ return reader, true
+ }
+ break
+ }
+ }
+
+ reader = io.MultiReader(bytes.NewReader(prefix), reader)
+ if !bytes.ContainsRune(prefix, '\n') {
+ return reader, false // inconclusive prefix: preserve rather than false-drain
+ }
+
+ _, _ = io.Copy(io.Discard, reader)
+ return reader, true
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/providers/utils/utils.go` around lines 1024 - 1039,
DrainNonSSEStreamReader currently decides SSE vs non‑SSE after a single
reader.Read which can return a short partial chunk; instead loop to accumulate
reads into buf (using reader.Read into buf[:]) until you either have enough
bytes to reliably call looksLikeSSEStreamPrefix(prefix) (use the same prefix
length checks as looksLikeSSEStreamPrefix expects), hit EOF or an error, or fill
the 512‑byte buffer; once you have a sufficient prefix, reconstruct the reader
with io.MultiReader(bytes.NewReader(prefix), reader) and only return (reader,
false) if looksLikeSSEStreamPrefix(prefix) is true, otherwise drain and return
(reader, true); ensure you still handle readErr != nil (and treat non‑EOF errors
as draining case) and avoid consuming the stream prematurely.
Confidence Score: 4/5The core fix is sound: all 7 streaming handlers correctly pass the pre-wrapped reader through the sniff, and the MultiReader reconstruction is watertight. The two flagged items are minor — an unnecessary SetBodyStream in a compat shim with no production callers, and a missing drain-path test for the new function — neither of which affects the happy or error paths in production. The streaming logic change is well-scoped and the 512-byte sniff + MultiReader pattern is correct. The only concerns are a small behavioral change in the legacy compat shim (SetBodyStream called when it wasn't before, but with no callers harmed) and missing direct test coverage for the non-SSE drain branch of the new function. core/providers/utils/utils.go — specifically the DrainNonSSEStreamResponse compat shim's unconditional SetBodyStream call Important Files Changed
Reviews (1): Last reviewed commit: "Prevent stream hangs when SSE headers ar..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| func DrainNonSSEStreamResponse(resp *fasthttp.Response) bool { | ||
| reader, drained := DrainNonSSEStreamReader(resp, resp.BodyStream()) | ||
| if !drained && reader != nil { | ||
| resp.SetBodyStream(reader, -1) | ||
| } | ||
| return drained | ||
| } |
There was a problem hiding this comment.
SetBodyStream even when no bytes were sniffed
When Content-Type: text/event-stream is present, DrainNonSSEStreamReader returns the original reader immediately without consuming any bytes. The shim then unconditionally calls resp.SetBodyStream(reader, -1) (since !drained && reader != nil is true), replacing the stream with the same reader but dropping the original known size. The pre-PR DrainNonSSEStreamResponse was a pure check that left the response entirely unmodified for confirmed-SSE streams. The guard should skip the SetBodyStream when the reader is the unmodified original.
| func TestDrainNonSSEStreamReader_SSEWithoutContentTypeDoesNotDrain(t *testing.T) { | ||
| resp := fasthttp.AcquireResponse() | ||
| defer fasthttp.ReleaseResponse(resp) | ||
|
|
||
| body := []byte("event: response.completed\n" + | ||
| `data: {"type":"response.completed","sequence_number":1}` + "\n\n") | ||
| resp.SetBodyStream(bytes.NewReader(body), len(body)) | ||
|
|
||
| reader, drained := DrainNonSSEStreamReader(resp, resp.BodyStream()) | ||
| if drained { | ||
| t.Fatal("expected SSE-looking response without content type to remain readable") | ||
| } | ||
|
|
||
| remaining, err := io.ReadAll(reader) | ||
| if err != nil { | ||
| t.Fatalf("failed to read SSE body after guard: %v", err) | ||
| } | ||
| if string(remaining) != string(body) { | ||
| t.Fatalf("expected SSE body to remain intact, got %q", string(remaining)) | ||
| } |
There was a problem hiding this comment.
DrainNonSSEStreamReader non-SSE drain path
The two new tests only exercise the SSE-detection (returns false) branch of DrainNonSSEStreamReader. The complementary case — a JSON/non-line-delimited body passed directly to DrainNonSSEStreamReader — is not covered. Without it, a future regression in the drain path (e.g., accidentally returning false for a plain JSON body) would go undetected until a stream hang reappears in production.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
We can close this since you already merged #3956 |
3 participants
Summary
Content-Type: text/event-streamValidation
docker run --rm -v /root/ws/ai/ai-gateway/bifrost:/src -w /src/core golang:1.26.3-alpine3.23 sh -lc '/usr/local/go/bin/gofmt -w providers/utils/utils.go providers/openai/openai.go providers/utils/utils_test.go && /usr/local/go/bin/go test ./providers/utils -run "TestDrainNonSSEStream"'\n-docker run --rm -v /root/ws/ai/ai-gateway/bifrost:/src -w /src/core golang:1.26.3-alpine3.23 sh -lc '/usr/local/go/bin/go test ./providers/openai ./providers/utils'\n- manual gateway replay against patched container returned HTTP 200 andevent: response.completedfor/v1/responsesrouted tocodex-backend\n\n## Notes\n- local request fixtures were not committed because they contain secrets\nSummary by CodeRabbit
Release Notes
Bug Fixes
Tests