From 4e4dede4811ab99c6ce20bdea3e45041e6304b29 Mon Sep 17 00:00:00 2001
From: Luis Pater <webmaster@idotorg.org>
Date: Mon, 5 Oct 2026 04:32:35 +0800
Subject: [PATCH] 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
---
 ...antigravity_executor_finish_reason_test.go | 62 +++++++++++++++++++
 .../executor/antigravity_executor_stream.go   | 33 ++++++++++
 2 files changed, 95 insertions(+)

diff --git a/internal/runtime/executor/antigravity_executor_finish_reason_test.go b/internal/runtime/executor/antigravity_executor_finish_reason_test.go
index 64377c178..c54065f65 100644
--- a/internal/runtime/executor/antigravity_executor_finish_reason_test.go
+++ b/internal/runtime/executor/antigravity_executor_finish_reason_test.go
@@ -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
diff --git a/internal/runtime/executor/antigravity_executor_stream.go b/internal/runtime/executor/antigravity_executor_stream.go
index 904b333cd..de64ce733 100644
--- a/internal/runtime/executor/antigravity_executor_stream.go
+++ b/internal/runtime/executor/antigravity_executor_stream.go
@@ -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)