oadmin/xinghuoapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(openai): emit OpenAI-compatible SSE error events

Browse Source
This commit is contained in:
cyhhao
2026-01-19 13:53:39 +08:00
parent eb7d830296
commit 26298c4a5f
2 changed files with 20 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
backend/internal/service/openai_gateway_service.go
View File

@@ -1067,7 +1067,9 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp
// 记录上次收到上游数据的时间，用于控制 keepalive 发送频率
lastDataAt := time.Now()
// 仅发送一次错误事件，避免多次写入导致协议混乱（写失败时尽力通知客户端）
// 仅发送一次错误事件，避免多次写入导致协议混乱。
// 注意：OpenAI `/v1/responses` streaming 事件必须符合 OpenAI Responses schema；
// 否则下游 SDK（例如 OpenCode）会因为类型校验失败而报错。
errorEventSent := false
clientDisconnected := false // 客户端断开后继续 drain 上游以收集 usage
sendErrorEvent := func(reason string) {
@@ -1075,8 +1077,19 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp
return
}
errorEventSent = true
_, _ = fmt.Fprintf(w, "event: error\ndata: {\"error\":\"%s\"}\n\n", reason)
flusher.Flush()
payload := map[string]any{
"type": "error",
"sequence_number": 0,
"error": map[string]any{
"type": "upstream_error",
"message": reason,
"code": reason,
},
}
if b, err := json.Marshal(payload); err == nil {
_, _ = fmt.Fprintf(w, "data: %s\n\n", b)
flusher.Flush()
}
}
needModelReplace := originalModel != mappedModel

8
backend/internal/service/openai_gateway_service_test.go
View File

@@ -188,8 +188,8 @@ func TestOpenAIStreamingTimeout(t *testing.T) {
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected stream timeout error, got %v", err)
}
if !strings.Contains(rec.Body.String(), "stream_timeout") {
t.Fatalf("expected stream_timeout SSE error, got %q", rec.Body.String())
if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "stream_timeout") {
t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String())
}
}
@@ -305,8 +305,8 @@ func TestOpenAIStreamingTooLong(t *testing.T) {
if !errors.Is(err, bufio.ErrTooLong) {
t.Fatalf("expected ErrTooLong, got %v", err)
}
if !strings.Contains(rec.Body.String(), "response_too_large") {
t.Fatalf("expected response_too_large SSE error, got %q", rec.Body.String())
if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "response_too_large") {
t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String())
}
}