streaming support

Signed-off-by: JaredforReal <[email protected]>

Author: JaredforReal

src/semantic-router/pkg/extproc/mapping_responses.go

@@ -144,3 +144,62 @@ func mapChatCompletionToResponses(chatCompletionJSON []byte) ([]byte, error) {
 
 	return json.Marshal(out)
 }
+
+// translateSSEChunkToResponses converts a single OpenAI chat.completion.chunk SSE payload
+// (the JSON after "data: ") into Responses SSE events (delta/stop). Returns empty when not applicable.
+func translateSSEChunkToResponses(chunk []byte) ([][]byte, bool) {
+	// Expect chunk JSON like {"id":"...","object":"chat.completion.chunk","created":...,"model":"...","choices":[{"index":0,"delta":{"role":"assistant","content":"..."},"finish_reason":null}]}
+	var parsed map[string]interface{}
+	if err := json.Unmarshal(chunk, &parsed); err != nil {
+		return nil, false
+	}
+	if parsed["object"] != "chat.completion.chunk" {
+		return nil, false
+	}
+
+	created, _ := parsed["created"].(float64)
+	// Emit a created event only once per stream (handled by caller)
+
+	// Extract content delta and finish_reason
+	var deltaText string
+	var finish string
+	if arr, ok := parsed["choices"].([]interface{}); ok && len(arr) > 0 {
+		if ch, ok := arr[0].(map[string]interface{}); ok {
+			if fr, ok := ch["finish_reason"].(string); ok && fr != "" {
+				finish = fr
+			}
+			if d, ok := ch["delta"].(map[string]interface{}); ok {
+				if c, ok := d["content"].(string); ok {
+					deltaText = c
+				}
+			}
+		}
+	}
+
+	var events [][]byte
+	if deltaText != "" {
+		ev := map[string]interface{}{
+			"type":  "response.output_text.delta",
+			"delta": deltaText,
+		}
+		if created > 0 {
+			ev["created"] = int64(created)
+		}
+		b, _ := json.Marshal(ev)
+		events = append(events, b)
+	}
+
+	if finish != "" {
+		ev := map[string]interface{}{
+			"type":        "response.completed",
+			"stop_reason": finish,
+		}
+		b, _ := json.Marshal(ev)
+		events = append(events, b)
+	}
+
+	if len(events) == 0 {
+		return nil, false
+	}
+	return events, true
+}

src/semantic-router/pkg/extproc/mapping_responses_test.go

@@ -44,3 +44,11 @@ func TestMapChatCompletionToResponses_Minimal(t *testing.T) {
 		t.Fatalf("stop_reason missing")
 	}
 }
+
+func TestTranslateSSEChunkToResponses(t *testing.T) {
+	chunk := []byte(`{"id":"c1","object":"chat.completion.chunk","created":1,"model":"m","choices":[{"index":0,"delta":{"role":"assistant","content":"Hi"},"finish_reason":null}]}`)
+	evs, ok := translateSSEChunkToResponses(chunk)
+	if !ok || len(evs) == 0 {
+		t.Fatalf("expected events")
+	}
+}

src/semantic-router/pkg/extproc/processor_res_header.go

@@ -205,7 +205,54 @@ func (r *OpenAIRouter) handleResponseBody(v *ext_proc.ProcessingRequest_Response
 			}
 		}
 
-		// For streaming chunks, just continue (no token parsing or cache update)
+		// If Responses adapter is active for this request, translate SSE chunks
+		if r.Config != nil && r.Config.EnableResponsesAdapter {
+			if p, ok := ctx.Headers[":path"]; ok && strings.HasPrefix(p, "/v1/responses") {
+				body := v.ResponseBody.Body
+				// Envoy provides raw chunk bytes, typically like: "data: {json}\n\n" or "data: [DONE]\n\n"
+				b := string(body)
+				if strings.Contains(b, "[DONE]") {
+					// Emit a final response.completed if not already concluded
+					response := &ext_proc.ProcessingResponse{
+						Response: &ext_proc.ProcessingResponse_ResponseBody{
+							ResponseBody: &ext_proc.BodyResponse{
+								Response: &ext_proc.CommonResponse{Status: ext_proc.CommonResponse_CONTINUE},
+							},
+						},
+					}
+					return response, nil
+				}
+
+				// Extract JSON after "data: " prefix if present
+				idx := strings.Index(b, "data:")
+				var payload []byte
+				if idx >= 0 {
+					payload = []byte(strings.TrimSpace(b[idx+5:]))
+				} else {
+					payload = v.ResponseBody.Body
+				}
+
+				if len(payload) > 0 && payload[0] == '{' {
+					if !ctx.ResponsesStreamInit {
+						// Emit an initial created event on first chunk
+						ctx.ResponsesStreamInit = true
+						// We don't inject a new chunk here; clients will see deltas below
+					}
+					events, ok := translateSSEChunkToResponses(payload)
+					if ok && len(events) > 0 {
+						// Rebuild body as multiple SSE events in Responses format
+						var sb strings.Builder
+						for _, ev := range events {
+							sb.WriteString("data: ")
+							sb.Write(ev)
+							sb.WriteString("\n\n")
+						}
+						v.ResponseBody.Body = []byte(sb.String())
+					}
+				}
+			}
+		}
+
 		response := &ext_proc.ProcessingResponse{
 			Response: &ext_proc.ProcessingResponse_ResponseBody{
 				ResponseBody: &ext_proc.BodyResponse{

src/semantic-router/pkg/extproc/request_handler.go

@@ -249,6 +249,9 @@ type RequestContext struct {
 	TTFTRecorded bool
 	TTFTSeconds  float64
 
+	// Responses SSE translation state
+	ResponsesStreamInit bool
+
 	// VSR decision tracking
 	VSRSelectedCategory     string // The category selected by VSR
 	VSRReasoningMode        string // "on" or "off" - whether reasoning mode was determined to be used