From abc293c6423334a47f935203ed53b3f9792455a5 Mon Sep 17 00:00:00 2001 From: davidwushi1145 Date: Thu, 2 Apr 2026 20:17:45 +0800 Subject: [PATCH 1/2] Prevent malformed Responses SSE frames from breaking stream clients Line-oriented upstream executors can emit `event:` and `data:` as separate chunks, but the Responses handler had started terminating each incoming chunk as a full SSE event. That split `response.created` into an empty event plus a later data block, which broke downstream clients like OpenClaw. This keeps the fix in the handler layer: a small stateful framer now buffers standalone `event:` lines until the matching `data:` arrives, preserves already-framed events, and ignores delimiter-only leftovers. The regression suite now covers split event/data framing, full-event passthrough, terminal errors, and the bootstrap path that forwards line-oriented openai-response streams from non-Codex executors too. Constraint: Keep the fix localized to Responses handler framing instead of patching every executor Rejected: Revert to v6.9.7 chunk writing | would reintroduce data-only framing regressions Rejected: Patch each line-oriented executor separately | duplicates fragile SSE assembly logic Confidence: high Scope-risk: narrow Reversibility: clean Directive: Do not assume incoming Responses stream chunks are already complete SSE events; preserve handler-layer reassembly for split `event:`/`data:` inputs Tested: /tmp/go1.26.1/go/bin/go test ./sdk/api/handlers/openai -count=1 Tested: /tmp/go1.26.1/go/bin/go test ./sdk/api/handlers -count=1 Tested: /tmp/go1.26.1/go test ./sdk/api/handlers/... -count=1 Tested: /tmp/go1.26.1/go/bin/go vet ./sdk/api/handlers/... Tested: Temporary patched server on 127.0.0.1:18317 -> /v1/models 200, /v1/responses non-stream 200, /v1/responses stream emitted combined `event:` + `data:` frames Not-tested: Full repository test suite outside sdk/api/handlers packages --- .../handlers_stream_bootstrap_test.go | 81 +++++++++++ .../openai/openai_responses_handlers.go | 126 +++++++++++++++++- ...ai_responses_handlers_stream_error_test.go | 2 +- .../openai_responses_handlers_stream_test.go | 50 ++++++- 4 files changed, 250 insertions(+), 9 deletions(-) diff --git a/sdk/api/handlers/handlers_stream_bootstrap_test.go b/sdk/api/handlers/handlers_stream_bootstrap_test.go index b08e3a99..61c03332 100644 --- a/sdk/api/handlers/handlers_stream_bootstrap_test.go +++ b/sdk/api/handlers/handlers_stream_bootstrap_test.go @@ -136,6 +136,8 @@ type authAwareStreamExecutor struct { type invalidJSONStreamExecutor struct{} +type splitResponsesEventStreamExecutor struct{} + func (e *invalidJSONStreamExecutor) Identifier() string { return "codex" } func (e *invalidJSONStreamExecutor) Execute(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (coreexecutor.Response, error) { @@ -165,6 +167,36 @@ func (e *invalidJSONStreamExecutor) HttpRequest(ctx context.Context, auth *corea } } +func (e *splitResponsesEventStreamExecutor) Identifier() string { return "split-sse" } + +func (e *splitResponsesEventStreamExecutor) Execute(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (coreexecutor.Response, error) { + return coreexecutor.Response{}, &coreauth.Error{Code: "not_implemented", Message: "Execute not implemented"} +} + +func (e *splitResponsesEventStreamExecutor) ExecuteStream(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (*coreexecutor.StreamResult, error) { + ch := make(chan coreexecutor.StreamChunk, 2) + ch <- coreexecutor.StreamChunk{Payload: []byte("event: response.completed")} + ch <- coreexecutor.StreamChunk{Payload: []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"output\":[]}}")} + close(ch) + return &coreexecutor.StreamResult{Chunks: ch}, nil +} + +func (e *splitResponsesEventStreamExecutor) Refresh(ctx context.Context, auth *coreauth.Auth) (*coreauth.Auth, error) { + return auth, nil +} + +func (e *splitResponsesEventStreamExecutor) CountTokens(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (coreexecutor.Response, error) { + return coreexecutor.Response{}, &coreauth.Error{Code: "not_implemented", Message: "CountTokens not implemented"} +} + +func (e *splitResponsesEventStreamExecutor) HttpRequest(ctx context.Context, auth *coreauth.Auth, req *http.Request) (*http.Response, error) { + return nil, &coreauth.Error{ + Code: "not_implemented", + Message: "HttpRequest not implemented", + HTTPStatus: http.StatusNotImplemented, + } +} + func (e *authAwareStreamExecutor) Identifier() string { return "codex" } func (e *authAwareStreamExecutor) Execute(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (coreexecutor.Response, error) { @@ -607,3 +639,52 @@ func TestExecuteStreamWithAuthManager_ValidatesOpenAIResponsesStreamDataJSON(t * t.Fatalf("expected terminal error") } } + +func TestExecuteStreamWithAuthManager_AllowsSplitOpenAIResponsesSSEEventLines(t *testing.T) { + executor := &splitResponsesEventStreamExecutor{} + manager := coreauth.NewManager(nil, nil, nil) + manager.RegisterExecutor(executor) + + auth1 := &coreauth.Auth{ + ID: "auth1", + Provider: "split-sse", + Status: coreauth.StatusActive, + Metadata: map[string]any{"email": "test1@example.com"}, + } + if _, err := manager.Register(context.Background(), auth1); err != nil { + t.Fatalf("manager.Register(auth1): %v", err) + } + + registry.GetGlobalRegistry().RegisterClient(auth1.ID, auth1.Provider, []*registry.ModelInfo{{ID: "test-model"}}) + t.Cleanup(func() { + registry.GetGlobalRegistry().UnregisterClient(auth1.ID) + }) + + handler := NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, manager) + dataChan, _, errChan := handler.ExecuteStreamWithAuthManager(context.Background(), "openai-response", "test-model", []byte(`{"model":"test-model"}`), "") + if dataChan == nil || errChan == nil { + t.Fatalf("expected non-nil channels") + } + + var got []string + for chunk := range dataChan { + got = append(got, string(chunk)) + } + + for msg := range errChan { + if msg != nil { + t.Fatalf("unexpected error: %+v", msg) + } + } + + if len(got) != 2 { + t.Fatalf("expected 2 forwarded chunks, got %d: %#v", len(got), got) + } + if got[0] != "event: response.completed" { + t.Fatalf("unexpected first chunk: %q", got[0]) + } + expectedData := "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"output\":[]}}" + if got[1] != expectedData { + t.Fatalf("unexpected second chunk.\nGot: %q\nWant: %q", got[1], expectedData) + } +} diff --git a/sdk/api/handlers/openai/openai_responses_handlers.go b/sdk/api/handlers/openai/openai_responses_handlers.go index d1ba68c7..cdb8bfdf 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers.go +++ b/sdk/api/handlers/openai/openai_responses_handlers.go @@ -29,11 +29,13 @@ func writeResponsesSSEChunk(w io.Writer, chunk []byte) { if _, err := w.Write(chunk); err != nil { return } - if bytes.HasSuffix(chunk, []byte("\n\n")) { + if bytes.HasSuffix(chunk, []byte("\n\n")) || bytes.HasSuffix(chunk, []byte("\r\n\r\n")) { return } suffix := []byte("\n\n") - if bytes.HasSuffix(chunk, []byte("\n")) { + if bytes.HasSuffix(chunk, []byte("\r\n")) { + suffix = []byte("\r\n") + } else if bytes.HasSuffix(chunk, []byte("\n")) { suffix = []byte("\n") } if _, err := w.Write(suffix); err != nil { @@ -41,6 +43,112 @@ func writeResponsesSSEChunk(w io.Writer, chunk []byte) { } } +type responsesSSEFramer struct { + pending []byte +} + +func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) { + if len(chunk) == 0 { + return + } + if responsesSSENeedsLineBreak(f.pending, chunk) { + f.pending = append(f.pending, '\n') + } + f.pending = append(f.pending, chunk...) + for { + frameLen := responsesSSEFrameLen(f.pending) + if frameLen == 0 { + break + } + writeResponsesSSEChunk(w, f.pending[:frameLen]) + copy(f.pending, f.pending[frameLen:]) + f.pending = f.pending[:len(f.pending)-frameLen] + } + if len(bytes.TrimSpace(f.pending)) == 0 { + f.pending = f.pending[:0] + return + } + if len(f.pending) == 0 || responsesSSENeedsMoreData(f.pending) { + return + } + writeResponsesSSEChunk(w, f.pending) + f.pending = f.pending[:0] +} + +func (f *responsesSSEFramer) Flush(w io.Writer) { + if len(f.pending) == 0 { + return + } + if len(bytes.TrimSpace(f.pending)) == 0 { + f.pending = f.pending[:0] + return + } + if responsesSSENeedsMoreData(f.pending) { + f.pending = f.pending[:0] + return + } + writeResponsesSSEChunk(w, f.pending) + f.pending = f.pending[:0] +} + +func responsesSSEFrameLen(chunk []byte) int { + if len(chunk) == 0 { + return 0 + } + lf := bytes.Index(chunk, []byte("\n\n")) + crlf := bytes.Index(chunk, []byte("\r\n\r\n")) + switch { + case lf < 0: + if crlf < 0 { + return 0 + } + return crlf + 4 + case crlf < 0: + return lf + 2 + case lf < crlf: + return lf + 2 + default: + return crlf + 4 + } +} + +func responsesSSENeedsMoreData(chunk []byte) bool { + trimmed := bytes.TrimSpace(chunk) + if len(trimmed) == 0 { + return false + } + return responsesSSEHasField(trimmed, []byte("event:")) && !responsesSSEHasField(trimmed, []byte("data:")) +} + +func responsesSSEHasField(chunk []byte, prefix []byte) bool { + for _, line := range bytes.Split(chunk, []byte("\n")) { + line = bytes.TrimSpace(line) + if bytes.HasPrefix(line, prefix) { + return true + } + } + return false +} + +func responsesSSENeedsLineBreak(pending, chunk []byte) bool { + if len(pending) == 0 || len(chunk) == 0 { + return false + } + if bytes.HasSuffix(pending, []byte("\n")) || bytes.HasSuffix(pending, []byte("\r")) { + return false + } + trimmed := bytes.TrimSpace(chunk) + if len(trimmed) == 0 { + return true + } + for _, prefix := range [][]byte{[]byte("data:"), []byte("event:"), []byte("id:"), []byte("retry:"), []byte(":")} { + if bytes.HasPrefix(trimmed, prefix) { + return true + } + } + return false +} + // OpenAIResponsesAPIHandler contains the handlers for OpenAIResponses API endpoints. // It holds a pool of clients to interact with the backend service. type OpenAIResponsesAPIHandler struct { @@ -213,6 +321,7 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ c.Header("Connection", "keep-alive") c.Header("Access-Control-Allow-Origin", "*") } + framer := &responsesSSEFramer{} // Peek at the first chunk for { @@ -250,22 +359,26 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ handlers.WriteUpstreamHeaders(c.Writer.Header(), upstreamHeaders) // Write first chunk logic (matching forwardResponsesStream) - writeResponsesSSEChunk(c.Writer, chunk) + framer.WriteChunk(c.Writer, chunk) flusher.Flush() // Continue - h.forwardResponsesStream(c, flusher, func(err error) { cliCancel(err) }, dataChan, errChan) + h.forwardResponsesStream(c, flusher, func(err error) { cliCancel(err) }, dataChan, errChan, framer) return } } } -func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flusher http.Flusher, cancel func(error), data <-chan []byte, errs <-chan *interfaces.ErrorMessage) { +func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flusher http.Flusher, cancel func(error), data <-chan []byte, errs <-chan *interfaces.ErrorMessage, framer *responsesSSEFramer) { + if framer == nil { + framer = &responsesSSEFramer{} + } h.ForwardStream(c, flusher, cancel, data, errs, handlers.StreamForwardOptions{ WriteChunk: func(chunk []byte) { - writeResponsesSSEChunk(c.Writer, chunk) + framer.WriteChunk(c.Writer, chunk) }, WriteTerminalError: func(errMsg *interfaces.ErrorMessage) { + framer.Flush(c.Writer) if errMsg == nil { return } @@ -281,6 +394,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flush _, _ = fmt.Fprintf(c.Writer, "\nevent: error\ndata: %s\n\n", string(chunk)) }, WriteDone: func() { + framer.Flush(c.Writer) _, _ = c.Writer.Write([]byte("\n")) }, }) diff --git a/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go b/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go index dce73807..771e46b8 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go @@ -32,7 +32,7 @@ func TestForwardResponsesStreamTerminalErrorUsesResponsesErrorChunk(t *testing.T errs <- &interfaces.ErrorMessage{StatusCode: http.StatusInternalServerError, Error: errors.New("unexpected EOF")} close(errs) - h.forwardResponsesStream(c, flusher, func(error) {}, data, errs) + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) body := recorder.Body.String() if !strings.Contains(body, `"type":"error"`) { t.Fatalf("expected responses error chunk, got: %q", body) diff --git a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go index 185a455a..e6efaa4a 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go @@ -12,7 +12,9 @@ import ( sdkconfig "github.com/router-for-me/CLIProxyAPI/v6/sdk/config" ) -func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) { +func newResponsesStreamTestHandler(t *testing.T) (*OpenAIResponsesAPIHandler, *httptest.ResponseRecorder, *gin.Context, http.Flusher) { + t.Helper() + gin.SetMode(gin.TestMode) base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, nil) h := NewOpenAIResponsesAPIHandler(base) @@ -26,6 +28,12 @@ func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) { t.Fatalf("expected gin writer to implement http.Flusher") } + return h, recorder, c, flusher +} + +func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) { + h, recorder, c, flusher := newResponsesStreamTestHandler(t) + data := make(chan []byte, 2) errs := make(chan *interfaces.ErrorMessage) data <- []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"arguments\":\"{}\"}}") @@ -33,7 +41,7 @@ func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) { close(data) close(errs) - h.forwardResponsesStream(c, flusher, func(error) {}, data, errs) + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) body := recorder.Body.String() parts := strings.Split(strings.TrimSpace(body), "\n\n") if len(parts) != 2 { @@ -50,3 +58,41 @@ func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) { t.Errorf("unexpected second event.\nGot: %q\nWant: %q", parts[1], expectedPart2) } } + +func TestForwardResponsesStreamReassemblesSplitSSEEventChunks(t *testing.T) { + h, recorder, c, flusher := newResponsesStreamTestHandler(t) + + data := make(chan []byte, 3) + errs := make(chan *interfaces.ErrorMessage) + data <- []byte("event: response.created") + data <- []byte("data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}") + data <- []byte("\n") + close(data) + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) + + got := strings.TrimSuffix(recorder.Body.String(), "\n") + want := "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n" + if got != want { + t.Fatalf("unexpected split-event framing.\nGot: %q\nWant: %q", got, want) + } +} + +func TestForwardResponsesStreamPreservesValidFullSSEEventChunks(t *testing.T) { + h, recorder, c, flusher := newResponsesStreamTestHandler(t) + + data := make(chan []byte, 1) + errs := make(chan *interfaces.ErrorMessage) + chunk := []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n") + data <- chunk + close(data) + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) + + got := strings.TrimSuffix(recorder.Body.String(), "\n") + if got != string(chunk) { + t.Fatalf("unexpected full-event framing.\nGot: %q\nWant: %q", got, string(chunk)) + } +} From 108895fc04cce5d9e55fb2f0cad60807b3bef4b9 Mon Sep 17 00:00:00 2001 From: davidwushi1145 Date: Thu, 2 Apr 2026 20:39:49 +0800 Subject: [PATCH 2/2] Harden Responses SSE framing against partial chunk boundaries Follow-up review found two real framing hazards in the handler-layer framer: it could flush a partial `data:` payload before the JSON was complete, and it could inject an extra newline before chunks that already began with `\n`/`\r\n`. This commit tightens the framer so it only emits undelimited events when the buffered `data:` payload is already valid JSON (or `[DONE]`), skips newline injection for chunks that already start with a line break, and avoids the heavier `bytes.Split` path while scanning SSE fields. The regression suite now covers split `data:` payload chunks, newline-prefixed chunks, and dropping incomplete trailing data on flush, so the original Responses fix remains intact while the review concerns are explicitly locked down. Constraint: Keep the follow-up limited to handler-layer framing and tests Rejected: Ignore the review and rely on current executor chunk shapes | leaves partial data payload corruption possible Rejected: Build a fully generic SSE parser | wider change than needed for the identified risks Confidence: high Scope-risk: narrow Reversibility: clean Directive: Do not emit undelimited Responses SSE events unless buffered `data:` content is already complete and valid Tested: /tmp/go1.26.1/go/bin/go test ./sdk/api/handlers/openai -count=1 Tested: /tmp/go1.26.1/go/bin/go test ./sdk/api/handlers -count=1 Tested: /tmp/go1.26.1/go/bin/go vet ./sdk/api/handlers/... Not-tested: Full repository test suite outside sdk/api/handlers packages --- .../openai/openai_responses_handlers.go | 55 +++++++++++++++++-- .../openai_responses_handlers_stream_test.go | 44 +++++++++++++++ 2 files changed, 94 insertions(+), 5 deletions(-) diff --git a/sdk/api/handlers/openai/openai_responses_handlers.go b/sdk/api/handlers/openai/openai_responses_handlers.go index cdb8bfdf..8969ce2f 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers.go +++ b/sdk/api/handlers/openai/openai_responses_handlers.go @@ -9,6 +9,7 @@ package openai import ( "bytes" "context" + "encoding/json" "fmt" "io" "net/http" @@ -68,7 +69,7 @@ func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) { f.pending = f.pending[:0] return } - if len(f.pending) == 0 || responsesSSENeedsMoreData(f.pending) { + if len(f.pending) == 0 || !responsesSSECanEmitWithoutDelimiter(f.pending) { return } writeResponsesSSEChunk(w, f.pending) @@ -83,7 +84,7 @@ func (f *responsesSSEFramer) Flush(w io.Writer) { f.pending = f.pending[:0] return } - if responsesSSENeedsMoreData(f.pending) { + if !responsesSSECanEmitWithoutDelimiter(f.pending) { f.pending = f.pending[:0] return } @@ -121,7 +122,15 @@ func responsesSSENeedsMoreData(chunk []byte) bool { } func responsesSSEHasField(chunk []byte, prefix []byte) bool { - for _, line := range bytes.Split(chunk, []byte("\n")) { + s := chunk + for len(s) > 0 { + line := s + if i := bytes.IndexByte(s, '\n'); i >= 0 { + line = s[:i] + s = s[i+1:] + } else { + s = nil + } line = bytes.TrimSpace(line) if bytes.HasPrefix(line, prefix) { return true @@ -130,6 +139,39 @@ func responsesSSEHasField(chunk []byte, prefix []byte) bool { return false } +func responsesSSECanEmitWithoutDelimiter(chunk []byte) bool { + trimmed := bytes.TrimSpace(chunk) + if len(trimmed) == 0 || responsesSSENeedsMoreData(trimmed) || !responsesSSEHasField(trimmed, []byte("data:")) { + return false + } + return responsesSSEDataLinesValid(trimmed) +} + +func responsesSSEDataLinesValid(chunk []byte) bool { + s := chunk + for len(s) > 0 { + line := s + if i := bytes.IndexByte(s, '\n'); i >= 0 { + line = s[:i] + s = s[i+1:] + } else { + s = nil + } + line = bytes.TrimSpace(line) + if len(line) == 0 || !bytes.HasPrefix(line, []byte("data:")) { + continue + } + data := bytes.TrimSpace(line[len("data:"):]) + if len(data) == 0 || bytes.Equal(data, []byte("[DONE]")) { + continue + } + if !json.Valid(data) { + return false + } + } + return true +} + func responsesSSENeedsLineBreak(pending, chunk []byte) bool { if len(pending) == 0 || len(chunk) == 0 { return false @@ -137,9 +179,12 @@ func responsesSSENeedsLineBreak(pending, chunk []byte) bool { if bytes.HasSuffix(pending, []byte("\n")) || bytes.HasSuffix(pending, []byte("\r")) { return false } - trimmed := bytes.TrimSpace(chunk) + if chunk[0] == '\n' || chunk[0] == '\r' { + return false + } + trimmed := bytes.TrimLeft(chunk, " \t") if len(trimmed) == 0 { - return true + return false } for _, prefix := range [][]byte{[]byte("data:"), []byte("event:"), []byte("id:"), []byte("retry:"), []byte(":")} { if bytes.HasPrefix(trimmed, prefix) { diff --git a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go index e6efaa4a..ef16fe80 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go @@ -96,3 +96,47 @@ func TestForwardResponsesStreamPreservesValidFullSSEEventChunks(t *testing.T) { t.Fatalf("unexpected full-event framing.\nGot: %q\nWant: %q", got, string(chunk)) } } + +func TestForwardResponsesStreamBuffersSplitDataPayloadChunks(t *testing.T) { + h, recorder, c, flusher := newResponsesStreamTestHandler(t) + + data := make(chan []byte, 2) + errs := make(chan *interfaces.ErrorMessage) + data <- []byte("data: {\"type\":\"response.created\"") + data <- []byte(",\"response\":{\"id\":\"resp-1\"}}") + close(data) + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) + + got := recorder.Body.String() + want := "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n\n" + if got != want { + t.Fatalf("unexpected split-data framing.\nGot: %q\nWant: %q", got, want) + } +} + +func TestResponsesSSENeedsLineBreakSkipsChunksThatAlreadyStartWithNewline(t *testing.T) { + if responsesSSENeedsLineBreak([]byte("event: response.created"), []byte("\n")) { + t.Fatal("expected no injected newline before newline-only chunk") + } + if responsesSSENeedsLineBreak([]byte("event: response.created"), []byte("\r\n")) { + t.Fatal("expected no injected newline before CRLF chunk") + } +} + +func TestForwardResponsesStreamDropsIncompleteTrailingDataChunkOnFlush(t *testing.T) { + h, recorder, c, flusher := newResponsesStreamTestHandler(t) + + data := make(chan []byte, 1) + errs := make(chan *interfaces.ErrorMessage) + data <- []byte("data: {\"type\":\"response.created\"") + close(data) + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) + + if got := recorder.Body.String(); got != "\n" { + t.Fatalf("expected incomplete trailing data to be dropped on flush.\nGot: %q", got) + } +}