fix: preserve SSE event boundaries for Responses streams

sdk/api/handlers/openai/openai_responses_handlers.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
@@ -21,6 +22,21 @@ import (
 	"github.com/tidwall/sjson"
 )
 
+func writeResponsesSSEChunk(w io.Writer, chunk []byte) {
+	if w == nil || len(chunk) == 0 {
+		return
+	}
+	_, _ = w.Write(chunk)
+	switch {
+	case bytes.HasSuffix(chunk, []byte("\n\n")):
+		return
+	case bytes.HasSuffix(chunk, []byte("\n")):
+		_, _ = w.Write([]byte("\n"))
+	default:
+		_, _ = w.Write([]byte("\n\n"))
+	}
+}
+
 // OpenAIResponsesAPIHandler contains the handlers for OpenAIResponses API endpoints.
 // It holds a pool of clients to interact with the backend service.
 type OpenAIResponsesAPIHandler struct {
@@ -230,11 +246,7 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ
 			handlers.WriteUpstreamHeaders(c.Writer.Header(), upstreamHeaders)
 
 			// Write first chunk logic (matching forwardResponsesStream)
-			if bytes.HasPrefix(chunk, []byte("event:")) {
+			writeResponsesSSEChunk(c.Writer, chunk)
-				_, _ = c.Writer.Write([]byte("\n"))
-			}
-			_, _ = c.Writer.Write(chunk)
-			_, _ = c.Writer.Write([]byte("\n"))
 			flusher.Flush()
 
 			// Continue
@@ -247,11 +259,7 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ
 func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flusher http.Flusher, cancel func(error), data <-chan []byte, errs <-chan *interfaces.ErrorMessage) {
 	h.ForwardStream(c, flusher, cancel, data, errs, handlers.StreamForwardOptions{
 		WriteChunk: func(chunk []byte) {
-			if bytes.HasPrefix(chunk, []byte("event:")) {
+			writeResponsesSSEChunk(c.Writer, chunk)
-				_, _ = c.Writer.Write([]byte("\n"))
-			}
-			_, _ = c.Writer.Write(chunk)
-			_, _ = c.Writer.Write([]byte("\n"))
 		},
 		WriteTerminalError: func(errMsg *interfaces.ErrorMessage) {
 			if errMsg == nil {

sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go

@@ -41,3 +41,38 @@ func TestForwardResponsesStreamTerminalErrorUsesResponsesErrorChunk(t *testing.T
 		t.Fatalf("expected streaming error chunk (top-level type), got HTTP error body: %q", body)
 	}
 }
+
+func TestForwardResponsesStreamSeparatesDataOnlySSEChunks(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, nil)
+	h := NewOpenAIResponsesAPIHandler(base)
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
+
+	flusher, ok := c.Writer.(http.Flusher)
+	if !ok {
+		t.Fatalf("expected gin writer to implement http.Flusher")
+	}
+
+	data := make(chan []byte, 2)
+	errs := make(chan *interfaces.ErrorMessage)
+	data <- []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"arguments\":\"{}\"}}")
+	data <- []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"output\":[]}}")
+	close(data)
+	close(errs)
+
+	h.forwardResponsesStream(c, flusher, func(error) {}, data, errs)
+	body := recorder.Body.String()
+
+	if !strings.Contains(body, "data: {\"type\":\"response.output_item.done\"") {
+		t.Fatalf("expected first SSE data chunk, got: %q", body)
+	}
+	if !strings.Contains(body, "\n\ndata: {\"type\":\"response.completed\"") {
+		t.Fatalf("expected blank-line separation before second SSE event, got: %q", body)
+	}
+	if strings.Contains(body, "arguments\":\"{}\"}}data: {\"type\":\"response.completed\"") {
+		t.Fatalf("second SSE event was concatenated onto first event body: %q", body)
+	}
+}