chore(internal): refactor sse event parsing

Author: stainless-app[bot]

betathread.go

@@ -127,7 +127,7 @@ func (r *BetaThreadService) NewAndRunStreaming(ctx context.Context, body BetaThr
 	opts = append(opts, option.WithJSONSet("stream", true))
 	path := "threads/runs"
 	err = requestconfig.ExecuteNewRequest(ctx, http.MethodPost, path, body, &raw, opts...)
-	return ssestream.NewStream[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
+	return ssestream.NewStreamWithSynthesizeEventData[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
 }
 
 // AssistantResponseFormatOptionUnion contains all possible properties and values

betathreadrun.go

@@ -78,7 +78,7 @@ func (r *BetaThreadRunService) NewStreaming(ctx context.Context, threadID string
 	}
 	path := fmt.Sprintf("threads/%s/runs", threadID)
 	err = requestconfig.ExecuteNewRequest(ctx, http.MethodPost, path, params, &raw, opts...)
-	return ssestream.NewStream[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
+	return ssestream.NewStreamWithSynthesizeEventData[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
 }
 
 // Retrieves a run.
@@ -215,7 +215,7 @@ func (r *BetaThreadRunService) SubmitToolOutputsStreaming(ctx context.Context, t
 	}
 	path := fmt.Sprintf("threads/%s/runs/%s/submit_tool_outputs", threadID, runID)
 	err = requestconfig.ExecuteNewRequest(ctx, http.MethodPost, path, body, &raw, opts...)
-	return ssestream.NewStream[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
+	return ssestream.NewStreamWithSynthesizeEventData[AssistantStreamEventUnion](ssestream.NewDecoder(raw), err)
 }
 
 // Tool call objects

packages/ssestream/ssestream.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"strings"
 
+	shimjson "github.com/openai/openai-go/v3/internal/encoding/json"
 	"github.com/tidwall/gjson"
 )
 
@@ -134,10 +135,11 @@ func (s *eventStreamDecoder) Err() error {
 }
 
 type Stream[T any] struct {
-	decoder Decoder
-	cur     T
-	err     error
-	done    bool
+	decoder             Decoder
+	cur                 T
+	err                 error
+	done                bool
+	synthesizeEventData bool
 }
 
 func NewStream[T any](decoder Decoder, err error) *Stream[T] {
@@ -147,6 +149,14 @@ func NewStream[T any](decoder Decoder, err error) *Stream[T] {
 	}
 }
 
+func NewStreamWithSynthesizeEventData[T any](decoder Decoder, err error) *Stream[T] {
+	return &Stream[T]{
+		decoder:             decoder,
+		err:                 err,
+		synthesizeEventData: true,
+	}
+}
+
 // Next returns false if the stream has ended or an error occurred.
 // Call Stream.Current() to get the current value.
 // Call Stream.Err() to get the error.
@@ -183,34 +193,32 @@ func (s *Stream[T]) Next() bool {
 			return false
 		}
 		var nxt T
-
-		if s.decoder.Event().Type == "" || !strings.HasPrefix(s.decoder.Event().Type, "thread.") {
-			ep := gjson.GetBytes(s.decoder.Event().Data, "error")
-			if ep.Exists() {
-				s.err = fmt.Errorf("received error while streaming: %s", ep.String())
-				return false
+		data := s.decoder.Event().Data
+		if s.decoder.Event().Type != "" && strings.HasPrefix(s.decoder.Event().Type, "thread.") {
+			synthesized := map[string]any{
+				"event": s.decoder.Event().Type,
+				"data":  json.RawMessage(data),
 			}
-			s.err = json.Unmarshal(s.decoder.Event().Data, &nxt)
+			data, s.err = shimjson.Marshal(synthesized)
 			if s.err != nil {
 				return false
 			}
-			s.cur = nxt
-			return true
-		} else {
-			ep := gjson.GetBytes(s.decoder.Event().Data, "error")
-			if ep.Exists() {
-				s.err = fmt.Errorf("received error while streaming: %s", ep.String())
-				return false
+		} else if s.synthesizeEventData {
+			synthesized := map[string]any{
+				"event": s.decoder.Event().Type,
+				"data":  json.RawMessage(data),
 			}
-			event := s.decoder.Event().Type
-			data := s.decoder.Event().Data
-			s.err = json.Unmarshal([]byte(fmt.Sprintf(`{ "event": %q, "data": %s }`, event, data)), &nxt)
+			data, s.err = shimjson.Marshal(synthesized)
 			if s.err != nil {
 				return false
 			}
-			s.cur = nxt
-			return true
 		}
+		s.err = json.Unmarshal(data, &nxt)
+		if s.err != nil {
+			return false
+		}
+		s.cur = nxt
+		return true
 	}
 
 	// decoder.Next() may be false because of an error