Merge pull request #7572 from SPHINX998/fix/openai-alpha-search-require-completion

fix(openai): require a successful completion before billing an alpha search fallback

backend/internal/service/openai_alpha_search.go

@@ -553,7 +553,10 @@ func shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(statusCode int) bool {
 }
 
 func openAIAlphaSearchResponseFromResponsesSSE(body []byte) ([]byte, error) {
-	output, results := parseOpenAIResponsesSSEForAlphaSearch(body)
+	output, results, err := parseOpenAIResponsesSSEForAlphaSearch(body)
+	if err != nil {
+		return nil, err
+	}
 	resp := map[string]any{
 		"output": output,
 	}
@@ -563,7 +566,7 @@ func openAIAlphaSearchResponseFromResponsesSSE(body []byte) ([]byte, error) {
 	return json.Marshal(resp)
 }
 
-func parseOpenAIResponsesSSEForAlphaSearch(body []byte) (string, []any) {
+func parseOpenAIResponsesSSEForAlphaSearch(body []byte) (string, []any, error) {
 	text := strings.ReplaceAll(string(body), "\r\n", "\n")
 	var output strings.Builder
 	var completedResponse any
@@ -579,21 +582,36 @@ func parseOpenAIResponsesSSEForAlphaSearch(body []byte) (string, []any) {
 		if err := json.Unmarshal([]byte(data), &event); err != nil {
 			continue
 		}
+		switch event["type"] {
+		case "error", "response.failed", "response.incomplete":
+			return "", nil, fmt.Errorf("alpha search responses stream ended with %s", event["type"])
+		case "response.completed":
+			response, ok := event["response"].(map[string]any)
+			if !ok || response == nil {
+				return "", nil, fmt.Errorf("alpha search responses completion is missing its response")
+			}
+			if status, present := response["status"]; present && status != "completed" {
+				return "", nil, fmt.Errorf("alpha search responses completion has a non-success status")
+			}
+			completedResponse = response
+		}
 		if delta, _ := event["delta"].(string); delta != "" && event["type"] == "response.output_text.delta" {
 			_, _ = output.WriteString(delta)
 		}
-		if event["type"] == "response.completed" {
-			completedResponse = event["response"]
-		}
 		collectOpenAIAlphaSearchURLCitations(event, &results, seenURLs)
 	}
+	// A transport-level 200, deltas or [DONE] alone are not a successful
+	// search. Do not return partial content or a billable result on EOF.
+	if completedResponse == nil {
+		return "", nil, fmt.Errorf("alpha search responses stream ended before completion")
+	}
 
 	out := output.String()
 	if strings.TrimSpace(out) == "" && completedResponse != nil {
 		out = extractOpenAIResponsesCompletedText(completedResponse)
 		collectOpenAIAlphaSearchURLCitations(completedResponse, &results, seenURLs)
 	}
-	return out, results
+	return out, results, nil
 }
 
 func openAIAlphaSearchSSEData(block string) string {

backend/internal/service/openai_alpha_search_completion_test.go

@@ -0,0 +1,61 @@
+package service
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAlphaSearchFallbackRequiresSuccessfulCompletion(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	delta := "data: {\"type\":\"response.output_text.delta\",\"delta\":\"unfinished answer\"}\n\n"
+	cases := map[string]string{
+		"empty":             "",
+		"truncated":         delta,
+		"done_only":         delta + "data: [DONE]\n\n",
+		"failed":            delta + "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\"}}\n\n",
+		"incomplete":        delta + "data: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\"}}\n\n",
+		"error":             delta + "data: {\"type\":\"error\",\"error\":{\"message\":\"unavailable\"}}\n\n",
+		"invalid_completed": delta + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"failed\"}}\n\n",
+		"missing_response":  delta + "data: {\"type\":\"response.completed\"}\n\n",
+	}
+	for name, wire := range cases {
+		t.Run(name, func(t *testing.T) {
+			body := []byte(`{"model":"gpt-5.5","commands":{"search_query":[{"q":"example"}]}}`)
+			recorder := httptest.NewRecorder()
+			c, _ := gin.CreateTestContext(recorder)
+			c.Request = httptest.NewRequest(http.MethodPost, "/v1/alpha/search", bytes.NewReader(body))
+			upstream := &httpUpstreamRecorder{resp: &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+				Body:       io.NopCloser(strings.NewReader(wire)),
+			}}
+			service := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream}
+			account := &Account{ID: 43, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1,
+				Credentials: map[string]any{"access_token": "at-test-token", "auth_mode": OpenAIAuthModePersonalAccessToken, "chatgpt_account_id": "fixture-account"}}
+			result, err := service.ForwardAlphaSearch(context.Background(), c, account, body)
+			require.Error(t, err)
+			require.Nil(t, result, "a failed search must not produce a billable WebSearchCalls result")
+			require.False(t, c.Writer.Written(), "do not write partial content as a successful search")
+		})
+	}
+}
+
+func TestAlphaSearchFallbackCompletionPreservesOutputAndCitations(t *testing.T) {
+	for _, wire := range []string{
+		alphaSearchResponsesSSE("search result"),
+		"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[{\"type\":\"message\",\"content\":[{\"type\":\"output_text\",\"text\":\"search result\",\"annotations\":[{\"type\":\"url_citation\",\"url\":\"https://example.com/news\",\"title\":\"Example News\"}]}]}]}}\n\n",
+	} {
+		body, err := openAIAlphaSearchResponseFromResponsesSSE([]byte(wire))
+		require.NoError(t, err)
+		require.JSONEq(t, `{"output":"search result","results":[{"type":"text_result","ref_id":"turn0search0","url":"https://example.com/news","title":"Example News"}]}`, string(body))
+	}
+}

backend/internal/service/openai_alpha_search_test.go

@@ -212,8 +212,8 @@ func TestForwardAlphaSearchPATBackfillsMissingChatGPTAccountMetadata(t *testing.
 
 	upstream := &httpUpstreamRecorder{resp: &http.Response{
 		StatusCode: http.StatusOK,
-		Header:     http.Header{"Content-Type": []string{"application/json"}},
-		Body:       io.NopCloser(strings.NewReader(`{"output":"search result"}`)),
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(alphaSearchResponsesSSE("search result"))),
 	}}
 	oauthService := NewOpenAIOAuthService(nil, nil)
 	service := &OpenAIGatewayService{