lywsvip/CLIProxyAPI
1
0
Fork 0
mirror of https://github.com/router-for-me/CLIProxyAPI.git synced 2026-05-22 12:19:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): preserve unindexed response output items

Browse Source
This commit is contained in:
edlsh
2026-04-25 18:06:00 -04:00
parent fd45dece7f
commit d36e70e9dc
2 changed files with 54 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -46,9 +46,10 @@ func writeResponsesSSEChunk(w io.Writer, chunk []byte) {
}
type responsesSSEFramer struct {
pending []byte
outputItems map[int][]byte
outputOrder []int
pending []byte
outputItems map[int][]byte
outputOrder []int
unindexedOutputItems [][]byte
}
func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) {
@@ -159,21 +160,23 @@ func (f *responsesSSEFramer) recordOutputItem(payload []byte) {
return
}
index := len(f.outputOrder)
if outputIndex := gjson.GetBytes(payload, "output_index"); outputIndex.Exists() {
index = int(outputIndex.Int())
index := int(outputIndex.Int())
if f.outputItems == nil {
f.outputItems = make(map[int][]byte)
}
if _, exists := f.outputItems[index]; !exists {
f.outputOrder = append(f.outputOrder, index)
}
f.outputItems[index] = append([]byte(nil), item.Raw...)
return
}
if f.outputItems == nil {
f.outputItems = make(map[int][]byte)
}
if _, exists := f.outputItems[index]; !exists {
f.outputOrder = append(f.outputOrder, index)
}
f.outputItems[index] = append([]byte(nil), item.Raw...)
f.unindexedOutputItems = append(f.unindexedOutputItems, append([]byte(nil), item.Raw...))
}
func (f *responsesSSEFramer) repairCompletedPayload(payload []byte) []byte {
if len(f.outputOrder) == 0 {
if len(f.outputOrder) == 0 && len(f.unindexedOutputItems) == 0 {
return payload
}
output := gjson.GetBytes(payload, "response.output")
@@ -197,6 +200,13 @@ func (f *responsesSSEFramer) repairCompletedPayload(payload []byte) []byte {
outputJSON.Write(item)
written++
}
for _, item := range f.unindexedOutputItems {
if written > 0 {
outputJSON.WriteByte(',')
}
outputJSON.Write(item)
written++
}
outputJSON.WriteByte(']')
repaired, err := sjson.SetRawBytes(payload, "response.output", outputJSON.Bytes())

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

@@ -91,6 +91,37 @@ func TestForwardResponsesStreamRepairsEmptyCompletedOutputFromDoneItems(t *testi
}
}
func TestForwardResponsesStreamRepairsMixedIndexedAndUnindexedDoneItems(t *testing.T) {
h, recorder, c, flusher := newResponsesStreamTestHandler(t)
data := make(chan []byte, 3)
errs := make(chan *interfaces.ErrorMessage)
data <- []byte(`data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"fc-1","call_id":"call-1","name":"shell","arguments":"{}","status":"completed"}}`)
data <- []byte(`data: {"type":"response.output_item.done","item":{"type":"message","id":"msg-1","role":"assistant","content":[{"type":"output_text","text":"done"}]}}`)
data <- []byte(`data: {"type":"response.completed","response":{"id":"resp-1","output":[]}}`)
close(data)
close(errs)
h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil)
parts := strings.Split(strings.TrimSpace(recorder.Body.String()), "\n\n")
if len(parts) != 3 {
t.Fatalf("expected 3 SSE events, got %d. Body: %q", len(parts), recorder.Body.String())
}
payload := strings.TrimPrefix(parts[2], "data: ")
output := gjson.Get(payload, "response.output")
if !output.IsArray() || len(output.Array()) != 2 {
t.Fatalf("expected repaired completed output with 2 items, got %s", output.Raw)
}
if got := gjson.Get(payload, "response.output.0.name").String(); got != "shell" {
t.Fatalf("expected indexed function_call to be preserved first, got %q in %s", got, payload)
}
if got := gjson.Get(payload, "response.output.1.id").String(); got != "msg-1" {
t.Fatalf("expected unindexed message to be appended, got %q in %s", got, payload)
}
}
func TestForwardResponsesStreamReassemblesSplitSSEEventChunks(t *testing.T) {
h, recorder, c, flusher := newResponsesStreamTestHandler(t)