fix(antigravity): surface backend errors and support multi-line stream payloads

- Buffer and assemble multi-line or split JSON chunks until a valid JSON payload is formed
- Intercept upstream backend error objects during streaming and surface them as terminal errors with appropriate status codes

Closes: #6357
This commit is contained in:
Luis Pater
2026-10-05 04:32:35 +08:00
parent d31b61cb87
commit 4e4dede481
2 changed files with 95 additions and 0 deletions

View File

@@ -70,6 +70,68 @@ func collectAntigravityStream(t *testing.T, baseURL string, sourceFormat, respon
return chunks
}
func TestAntigravityStreamSurfacesTrailingBackendError(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = io.WriteString(w, `data: {"response":{"candidates":[{"content":{"parts":[{"text":"partial"}]}}],"modelVersion":"gemini-3.7-flash"}}
{
"error": {
"code": 503,
"message": "backend capacity exhausted",
"status": "UNAVAILABLE"
}
}
`)
}))
defer server.Close()
for _, format := range []sdktranslator.Format{sdktranslator.FormatClaude, sdktranslator.FormatOpenAI} {
t.Run(format.String(), func(t *testing.T) {
executor := NewAntigravityExecutor(&config.Config{
Antigravity: config.AntigravityConfig{},
RequestRetry: 1,
})
result, errExecute := executor.ExecuteStream(context.Background(), &cliproxyauth.Auth{
Metadata: map[string]any{
"access_token": "token-123",
"expired": time.Now().Add(24 * time.Hour).Format(time.RFC3339),
"project_id": "project-1",
},
Attributes: map[string]string{"base_url": server.URL},
}, cliproxyexecutor.Request{
Model: "gemini-3.7-flash",
Payload: []byte(`{"model":"gemini-3.7-flash","stream":true,"messages":[{"role":"user","content":"hello"}]}`),
}, cliproxyexecutor.Options{
SourceFormat: format,
ResponseFormat: format,
Stream: true,
})
if errExecute != nil {
t.Fatalf("ExecuteStream() error = %v", errExecute)
}
var streamErr error
for chunk := range result.Chunks {
if chunk.Err != nil {
streamErr = chunk.Err
continue
}
if bytes.Contains(chunk.Payload, []byte("message_stop")) || bytes.Contains(chunk.Payload, []byte(`"finish_reason":"stop"`)) {
t.Fatalf("trailing backend error became successful completion: %s", chunk.Payload)
}
}
if streamErr == nil || !bytes.Contains([]byte(streamErr.Error()), []byte("backend capacity exhausted")) {
t.Fatalf("trailing backend error = %v, want surfaced backend message", streamErr)
}
statusError, ok := streamErr.(interface{ StatusCode() int })
if !ok || statusError.StatusCode() != http.StatusServiceUnavailable {
t.Fatalf("trailing backend status code = %v, want %d", streamErr, http.StatusServiceUnavailable)
}
})
}
}
// TestAntigravityStreamDoesNotEmitClaudeMessageStopOnReadError locks in the
// executor ordering fix on the response format where it is observable: the
// Claude translator has always synthesized a terminal event on [DONE], so

View File

@@ -205,6 +205,7 @@ func (e *AntigravityExecutor) ExecuteStream(ctx context.Context, auth *cliproxya
claudeInputTokens := helps.NewClaudeInputTokenState(from, to, responseFormat, originalPayload)
var streamUsage helps.StreamUsageBuffer
var streamErr error
var pendingJSON []byte
var replayCommitted bool
var param any
commitReplay := func() {
@@ -256,9 +257,41 @@ func (e *AntigravityExecutor) ExecuteStream(ctx context.Context, auth *cliproxya
line = helps.FilterSSEUsageMetadata(line)
payload := helps.JSONPayload(line)
if len(pendingJSON) > 0 {
trimmedLine := bytes.TrimSpace(line)
if bytes.HasPrefix(trimmedLine, []byte("data:")) {
trimmedLine = bytes.TrimSpace(trimmedLine[len("data:"):])
}
if len(trimmedLine) > 0 {
pendingJSON = append(pendingJSON, '\n')
pendingJSON = append(pendingJSON, trimmedLine...)
}
if !gjson.ValidBytes(pendingJSON) {
continue
}
payload = pendingJSON
pendingJSON = nil
} else if payload != nil && !gjson.ValidBytes(payload) {
pendingJSON = append([]byte(nil), payload...)
continue
}
if payload == nil {
continue
}
if errorResult := gjson.GetBytes(payload, "error"); errorResult.Exists() {
statusCode := int(errorResult.Get("code").Int())
if statusCode < http.StatusBadRequest || statusCode > 599 {
statusCode = http.StatusBadGateway
}
streamErr = newAntigravityStatusErr(statusCode, payload)
stopDeliverySupport()
helps.RecordAPIResponseError(ctx, e.cfg, streamErr)
select {
case out <- cliproxyexecutor.StreamChunk{Err: streamErr}:
case <-ctx.Done():
}
return
}
reporter.ObserveResponseModel(payload)
payload = e.resolveWebSearchGroundingURLs(ctx, auth, from, originalPayload, translated, payload)