diff --git a/backend/internal/service/openai_alpha_search.go b/backend/internal/service/openai_alpha_search.go index c80673405f..ac6ed3ee21 100644 --- a/backend/internal/service/openai_alpha_search.go +++ b/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 { diff --git a/backend/internal/service/openai_alpha_search_completion_test.go b/backend/internal/service/openai_alpha_search_completion_test.go new file mode 100644 index 0000000000..97ab5cff4d --- /dev/null +++ b/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)) + } +} diff --git a/backend/internal/service/openai_alpha_search_test.go b/backend/internal/service/openai_alpha_search_test.go index 8521c2160e..378378b2fc 100644 --- a/backend/internal/service/openai_alpha_search_test.go +++ b/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{