lywsvip/CLIProxyAPI
1
0
Fork 0
mirror of https://github.com/router-for-me/CLIProxyAPI.git synced 2026-10-04 14:49:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(responses): filter upstream private and telemetry events in SSE streams

Browse Source 
- Filter internal `responsesapi.` telemetry and private `codex.` events from SSE output
- Preserve Codex response metadata for official Codex clients while dropping rate limits
- Discard frames matching filtered event names or payload types in the SSE framer

Closes: #6007
This commit is contained in:
Luis Pater
2026-09-21 21:35:48 +08:00
parent 7b6fafce1b
commit dd013f9e29
2 changed files with 224 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
sdk/api/handlers/openai/openai_responses_handlers.go
View File

@@ -57,6 +57,7 @@ type responsesSSEFramer struct {
terminalEvent string
terminalError *interfaces.ErrorMessage
failureEvent string
isCodexClient bool
dataFrames int
}
@@ -119,8 +120,46 @@ func (f *responsesSSEFramer) writeFrame(w io.Writer, frame []byte) {
writeResponsesSSEChunk(w, f.repairFrame(frame))
}
func (f *responsesSSEFramer) shouldFilterPrivateEvent(streamEvent, payloadType string) bool {
check := func(name string) bool {
name = strings.TrimSpace(name)
if name == "" {
return false
}
if responsesSSEErrorEvent(name) {
return false
}
// Always filter internal WebSocket timing telemetry from SSE streams.
if strings.HasPrefix(name, "responsesapi.") {
return true
}
// If official Codex client, preserve codex.response.metadata but filter rate limits.
if f != nil && f.isCodexClient {
if name == "codex.rate_limits" {
return true
}
return false
}
// For standard Responses API clients: filter any codex.* private events.
if strings.HasPrefix(name, "codex.") {
return true
}
return false
}
return check(streamEvent) || check(payloadType)
}
func (f *responsesSSEFramer) repairFrame(frame []byte) []byte {
payload, ok := responsesSSEDataPayload(frame)
streamEvent := responsesSSEEventName(frame)
if streamEvent != "" && f.shouldFilterPrivateEvent(streamEvent, "") {
return nil
}
if !ok || len(payload) == 0 {
return frame
}
@@ -131,16 +170,20 @@ func (f *responsesSSEFramer) repairFrame(frame []byte) []byte {
if !json.Valid(payload) {
return frame
}
f.dataFrames++
payloadType := gjson.GetBytes(payload, "type").String()
if f.shouldFilterPrivateEvent(streamEvent, payloadType) {
return nil
}
f.dataFrames++
if responsesSSEErrorEvent(payloadType) || responsesSSEPayloadHasError(payload) {
if payloadType != "" {
f.lastEvent = sanitizeResponsesStreamEventName(payloadType)
}
return f.repairErrorPayload(payload)
}
streamEvent := responsesSSEEventName(frame)
eventType := payloadType
if responsesSSETerminalEvent(streamEvent) {
eventType = streamEvent
@@ -668,11 +711,12 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ
c.Header("Connection", "keep-alive")
c.Header("Access-Control-Allow-Origin", "*")
}
isCodexClient := isCodexResponsesClientRequest(c)
failureEvent := "error"
if isCodexResponsesClientRequest(c) {
if isCodexClient {
failureEvent = "response.failed"
}
framer := &responsesSSEFramer{failureEvent: failureEvent}
framer := &responsesSSEFramer{failureEvent: failureEvent, isCodexClient: isCodexClient}
var initialOutput bytes.Buffer
// Peek at the first complete SSE data frame.
@@ -968,8 +1012,10 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flush
}
if isCodexResponsesClientRequest(c) {
framer.failureEvent = "response.failed"
framer.isCodexClient = true
} else {
framer.failureEvent = "error"
framer.isCodexClient = false
}
writeTerminalError := func(errMsg *interfaces.ErrorMessage) {
framer.Flush(c.Writer)

174
sdk/api/handlers/openai/openai_responses_handlers_stream_test.go
View File

@@ -526,3 +526,177 @@ func TestResponsesStreamErrorTextPreservesTokenCountersAndLargeInts(t *testing.T
t.Fatalf("token_limit should remain number 8192, got %v", parsed.Error["token_limit"])
}
}
func TestResponsesSSEFramer_FiltersUpstreamPrivateEvents(t *testing.T) {
var output bytes.Buffer
framer := &responsesSSEFramer{failureEvent: "error"}
rateLimitsFrame := []byte("event: codex.rate_limits\ndata: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":42}}}\n\n")
framer.WriteChunk(&output, rateLimitsFrame)
metadataFrame := []byte("event: codex.response.metadata\ndata: {\"type\":\"codex.response.metadata\",\"headers\":{\"x-turn-state\":\"turn-1\"}}\n\n")
framer.WriteChunk(&output, metadataFrame)
createdFrame := []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n")
framer.WriteChunk(&output, createdFrame)
timingFrame := []byte("event: responsesapi.websocket_timing\ndata: {\"type\":\"responsesapi.websocket_timing\",\"timing\":{\"duration_ms\":100}}\n\n")
framer.WriteChunk(&output, timingFrame)
completedFrame := []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"status\":\"completed\"}}\n\n")
framer.WriteChunk(&output, completedFrame)
got := output.String()
if strings.Contains(got, "codex.rate_limits") {
t.Fatalf("expected codex.rate_limits to be stripped, got: %q", got)
}
if strings.Contains(got, "codex.response.metadata") {
t.Fatalf("expected codex.response.metadata to be stripped, got: %q", got)
}
if strings.Contains(got, "responsesapi.websocket_timing") {
t.Fatalf("expected responsesapi.websocket_timing to be stripped, got: %q", got)
}
if !strings.Contains(got, "event: response.created") {
t.Fatalf("expected response.created in output, got: %q", got)
}
if !strings.Contains(got, "event: response.completed") {
t.Fatalf("expected response.completed in output, got: %q", got)
}
}
func TestForwardResponsesStream_FiltersUpstreamPrivateEvents(t *testing.T) {
h, recorder, c, flusher := newResponsesStreamTestHandler(t)
data := make(chan []byte, 5)
errs := make(chan *interfaces.ErrorMessage)
data <- []byte("event: codex.rate_limits\ndata: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":42}}}\n\n")
data <- []byte("event: codex.response.metadata\ndata: {\"type\":\"codex.response.metadata\",\"headers\":{\"x-turn-state\":\"turn-1\"}}\n\n")
data <- []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n")
data <- []byte("event: responsesapi.websocket_timing\ndata: {\"type\":\"responsesapi.websocket_timing\",\"timing\":{\"duration_ms\":100}}\n\n")
data <- []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"status\":\"completed\"}}\n\n")
close(data)
close(errs)
h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil)
got := recorder.Body.String()
if strings.Contains(got, "codex.rate_limits") {
t.Fatalf("expected codex.rate_limits to be stripped in stream, got: %q", got)
}
if strings.Contains(got, "codex.response.metadata") {
t.Fatalf("expected codex.response.metadata to be stripped in stream, got: %q", got)
}
if strings.Contains(got, "responsesapi.websocket_timing") {
t.Fatalf("expected responsesapi.websocket_timing to be stripped in stream, got: %q", got)
}
if !strings.Contains(got, "event: response.created") {
t.Fatalf("expected response.created in stream output, got: %q", got)
}
if !strings.Contains(got, "event: response.completed") {
t.Fatalf("expected response.completed in stream output, got: %q", got)
}
}
func TestResponsesSSEFramer_PreservesMetadataForCodexClient(t *testing.T) {
var output bytes.Buffer
framer := &responsesSSEFramer{failureEvent: "response.failed", isCodexClient: true}
metadataFrame := []byte("event: codex.response.metadata\ndata: {\"type\":\"codex.response.metadata\",\"headers\":{\"x-turn-state\":\"turn-1\"}}\n\n")
framer.WriteChunk(&output, metadataFrame)
createdFrame := []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n")
framer.WriteChunk(&output, createdFrame)
completedFrame := []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"status\":\"completed\"}}\n\n")
framer.WriteChunk(&output, completedFrame)
got := output.String()
if !strings.Contains(got, "codex.response.metadata") {
t.Fatalf("expected codex.response.metadata to be preserved for official Codex client, got: %q", got)
}
if !strings.Contains(got, "event: response.created") {
t.Fatalf("expected response.created in output, got: %q", got)
}
if !strings.Contains(got, "event: response.completed") {
t.Fatalf("expected response.completed in output, got: %q", got)
}
}
func TestResponsesSSEFramer_FiltersDataOnlyPrivateEventsAndNonJSON(t *testing.T) {
var output bytes.Buffer
framer := &responsesSSEFramer{failureEvent: "error"}
// Data-only frame with private type
dataOnlyPrivateFrame := []byte("data: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":42}}}\n\n")
framer.WriteChunk(&output, dataOnlyPrivateFrame)
// Non-JSON payload with private event line
nonJSONPrivateFrame := []byte("event: codex.rate_limits\ndata: not-json-data\n\n")
framer.WriteChunk(&output, nonJSONPrivateFrame)
// Event-only frame without data
eventOnlyPrivateFrame := []byte("event: codex.response.metadata\n\n")
framer.WriteChunk(&output, eventOnlyPrivateFrame)
// Standard response.created
createdFrame := []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n")
framer.WriteChunk(&output, createdFrame)
// Standard response.completed
completedFrame := []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"status\":\"completed\"}}\n\n")
framer.WriteChunk(&output, completedFrame)
got := output.String()
if strings.Contains(got, "codex.rate_limits") {
t.Fatalf("expected codex.rate_limits to be stripped, got: %q", got)
}
if strings.Contains(got, "not-json-data") {
t.Fatalf("expected not-json-data to be stripped, got: %q", got)
}
if strings.Contains(got, "codex.response.metadata") {
t.Fatalf("expected codex.response.metadata to be stripped, got: %q", got)
}
if !strings.Contains(got, "event: response.created") {
t.Fatalf("expected response.created in output, got: %q", got)
}
if !strings.Contains(got, "event: response.completed") {
t.Fatalf("expected response.completed in output, got: %q", got)
}
}
func TestResponsesSSEFramer_CodexClientFiltersRateLimitsAndTiming(t *testing.T) {
var output bytes.Buffer
framer := &responsesSSEFramer{failureEvent: "response.failed", isCodexClient: true}
rateLimitsFrame := []byte("event: codex.rate_limits\ndata: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":42}}}\n\n")
framer.WriteChunk(&output, rateLimitsFrame)
metadataFrame := []byte("event: codex.response.metadata\ndata: {\"type\":\"codex.response.metadata\",\"headers\":{\"x-turn-state\":\"turn-1\"}}\n\n")
framer.WriteChunk(&output, metadataFrame)
timingFrame := []byte("event: responsesapi.websocket_timing\ndata: {\"type\":\"responsesapi.websocket_timing\",\"timing\":{\"duration_ms\":100}}\n\n")
framer.WriteChunk(&output, timingFrame)
createdFrame := []byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-1\"}}\n\n")
framer.WriteChunk(&output, createdFrame)
completedFrame := []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"status\":\"completed\"}}\n\n")
framer.WriteChunk(&output, completedFrame)
got := output.String()
if strings.Contains(got, "codex.rate_limits") {
t.Fatalf("expected codex.rate_limits to be stripped for Codex client, got: %q", got)
}
if strings.Contains(got, "responsesapi.websocket_timing") {
t.Fatalf("expected responsesapi.websocket_timing to be stripped for Codex client, got: %q", got)
}
if !strings.Contains(got, "codex.response.metadata") {
t.Fatalf("expected codex.response.metadata to be preserved for official Codex client, got: %q", got)
}
if !strings.Contains(got, "event: response.created") {
t.Fatalf("expected response.created in output, got: %q", got)
}
if !strings.Contains(got, "event: response.completed") {
t.Fatalf("expected response.completed in output, got: %q", got)
}
}