Merge pull request #4394 from siyuan-123/codex/pr-gpt56-pat-alpha-search

fix(openai): route PAT alpha search via responses web_search
This commit is contained in:
Wesley Liddick
2026-07-16 09:32:59 +08:00
committed by GitHub
5 changed files with 795 additions and 20 deletions
@@ -118,7 +118,7 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) {
requestedModel,
failedAccountIDs,
service.OpenAIUpstreamTransportHTTPSSE,
service.OpenAIEndpointCapabilityChatCompletions,
service.OpenAIEndpointCapabilityAlphaSearch,
false,
false,
service.PlatformOpenAI,
+11
View File
@@ -87,6 +87,7 @@ const openAILongContextBillingEnabledKey = "openai_long_context_billing_enabled"
const (
OpenAIEndpointCapabilityChatCompletions OpenAIEndpointCapability = "chat_completions"
OpenAIEndpointCapabilityEmbeddings OpenAIEndpointCapability = "embeddings"
OpenAIEndpointCapabilityAlphaSearch OpenAIEndpointCapability = "alpha_search"
)
const openAIEndpointCapabilitiesCredentialKey = "openai_capabilities"
@@ -1397,6 +1398,13 @@ func (a *Account) SupportsOpenAIEndpointCapability(capability OpenAIEndpointCapa
}
switch capability {
case OpenAIEndpointCapabilityChatCompletions:
case OpenAIEndpointCapabilityAlphaSearch:
// Codex alpha/search 是 ChatGPT/Codex 后端工具端点,必须使用
// OAuth/PAT/AgentIdentity 这类 ChatGPT 账号凭据;API key 被发往
// chatgpt.com/backend-api/codex/alpha/search 会稳定 401。
if a.Type != AccountTypeOAuth {
return false
}
case OpenAIEndpointCapabilityEmbeddings:
if a.Type != AccountTypeAPIKey {
return false
@@ -1409,6 +1417,9 @@ func (a *Account) SupportsOpenAIEndpointCapability(capability OpenAIEndpointCapa
if !found {
return true
}
if capability == OpenAIEndpointCapabilityAlphaSearch && configured[string(OpenAIEndpointCapabilityChatCompletions)] {
return true
}
return configured[string(capability)]
}
+500 -19
View File
@@ -3,6 +3,7 @@ package service
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
@@ -39,21 +40,38 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co
if upstreamModel != "" && upstreamModel != requestedModel {
body = ReplaceModelInBody(body, upstreamModel)
}
sanitizedBody, err := sanitizeOpenAIAlphaSearchBody(body)
if err != nil {
return nil, fmt.Errorf("sanitize alpha search request body: %w", err)
}
body = sanitizedBody
token, _, err := s.GetAccessToken(ctx, account)
if err != nil {
return nil, err
}
req, err := s.buildOpenAIAlphaSearchRequest(ctx, c, account, body, token)
if err != nil {
return nil, err
}
proxyURL := ""
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
if err := s.ensureOpenAIAlphaSearchAuthMetadata(ctx, account, token, proxyURL); err != nil {
return nil, err
}
// Codex Personal Access Token(at-...)目前可访问 ChatGPT Codex
// /responses,但会被 standalone /alpha/search 的 access enforcement
// 拒绝为 no_matching_rule。对 PAT 账号使用等价的 hosted web_search
// Responses 路径兜底,避免把可用账号误判为搜索不可用。
if account.IsOpenAIPersonalAccessToken() {
return s.forwardAlphaSearchViaResponsesWebSearch(ctx, c, account, body, token, proxyURL, requestedModel, upstreamModel)
}
req, err := s.buildOpenAIAlphaSearchRequest(ctx, c, account, body, token)
if err != nil {
return nil, err
}
upstreamStart := time.Now()
resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
@@ -71,7 +89,14 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co
upstreamMessage := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody)))
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMessage, respBody) {
resp.Body = io.NopCloser(bytes.NewReader(respBody))
s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel)
// alpha/search 是独立的工具端点,单次 401 不能证明账号的模型调用
// 凭据全局失效。若沿用通用 401 逻辑,PAT 会因没有 refresh_token
// 被永久标记为 error;历史导入且缺少 auth_mode 标记的 at- token 也会
// 漏过 PAT 类型判断。这里仍允许本次请求换号,但不修改任何账号状态;
// 真正的凭据失效由普通 Responses 请求或 whoami 校验判定。
if shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(resp.StatusCode) {
s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel)
}
return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode,
ResponseBody: respBody,
@@ -102,16 +127,212 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co
}, nil
}
func (s *OpenAIGatewayService) buildOpenAIAlphaSearchRequest(ctx context.Context, c *gin.Context, account *Account, body []byte, token string) (*http.Request, error) {
clientBeta := ""
if c != nil {
clientBeta = c.GetHeader("OpenAI-Beta")
func (s *OpenAIGatewayService) forwardAlphaSearchViaResponsesWebSearch(
ctx context.Context,
c *gin.Context,
account *Account,
alphaBody []byte,
token string,
proxyURL string,
requestedModel string,
upstreamModel string,
) (*OpenAIForwardResult, error) {
if upstreamModel == "" {
upstreamModel = requestedModel
}
req, err := s.buildUpstreamRequestOpenAIPassthrough(ctx, c, account, body, token)
responsesBody, err := buildOpenAIAlphaSearchResponsesWebSearchBody(alphaBody, upstreamModel)
if err != nil {
return nil, err
}
req, err := s.buildOpenAIAlphaSearchResponsesWebSearchRequest(ctx, c, account, alphaBody, responsesBody, token)
if err != nil {
return nil, err
}
SetActualOpenAIUpstreamEndpoint(c, "/v1/responses")
upstreamStart := time.Now()
resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
if err != nil {
return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true)
}
defer func() { _ = resp.Body.Close() }()
respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
if err != nil {
return nil, fmt.Errorf("read alpha search responses fallback response: %w", err)
}
if resp.StatusCode >= http.StatusBadRequest {
upstreamMessage := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody)))
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMessage, respBody) {
resp.Body = io.NopCloser(bytes.NewReader(respBody))
// 仍按 alpha/search 工具请求处理:PAT 的工具链路失败不能直接永久置错。
if shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(resp.StatusCode) {
s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel)
}
return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode,
ResponseBody: respBody,
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
}
}
}
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
writeOpenAIPassthroughResponseHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
contentType := resp.Header.Get("Content-Type")
if contentType == "" {
contentType = "application/json"
}
c.Data(resp.StatusCode, contentType, respBody)
return nil, nil
}
if !account.IsShadow() {
s.UpdateCodexUsageSnapshotFromHeaders(ctx, account.ID, resp.Header)
}
alphaRespBody, err := openAIAlphaSearchResponseFromResponsesSSE(respBody)
if err != nil {
return nil, err
}
c.Data(http.StatusOK, "application/json", alphaRespBody)
return &OpenAIForwardResult{
RequestID: strings.TrimSpace(resp.Header.Get("x-request-id")),
Model: requestedModel,
UpstreamModel: upstreamModel,
UpstreamEndpoint: "/v1/responses",
ResponseHeaders: resp.Header.Clone(),
Duration: time.Since(upstreamStart),
WebSearchCalls: 1,
}, nil
}
func (s *OpenAIGatewayService) buildOpenAIAlphaSearchResponsesWebSearchRequest(ctx context.Context, c *gin.Context, account *Account, alphaBody []byte, body []byte, token string) (*http.Request, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, bytes.NewReader(body))
if err != nil {
return nil, err
}
req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
if err != nil {
return nil, fmt.Errorf("build openai authentication headers: %w", err)
}
for key, values := range authHeaders {
for _, value := range values {
req.Header.Add(key, value)
}
}
req.Host = "chatgpt.com"
if err := resolveAndSetOpenAIChatGPTAccountHeaders(ctx, s.accountRepo, req.Header, account); err != nil {
return nil, fmt.Errorf("resolve chatgpt account headers: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("OpenAI-Beta", "responses=experimental")
if turnMetadata := openAIAlphaSearchInboundHeader(c, "X-Codex-Turn-Metadata"); turnMetadata != "" {
req.Header.Set("X-Codex-Turn-Metadata", turnMetadata)
}
if version := openAIAlphaSearchInboundHeader(c, "Version"); version != "" {
req.Header.Set("Version", version)
} else {
req.Header.Set("Version", codexCLIVersion)
}
if originator := openAIAlphaSearchInboundHeader(c, "Originator"); originator != "" {
req.Header.Set("Originator", originator)
} else {
req.Header.Set("Originator", "codex_cli_rs")
}
if customUA := account.GetOpenAIUserAgent(); customUA != "" {
req.Header.Set("User-Agent", customUA)
} else if userAgent := openAIAlphaSearchInboundHeader(c, "User-Agent"); userAgent != "" {
req.Header.Set("User-Agent", userAgent)
} else {
req.Header.Set("User-Agent", codexCLIUserAgent)
}
if s.cfg != nil && s.cfg.Gateway.ForceCodexCLI {
req.Header.Set("User-Agent", codexCLIUserAgent)
}
apiKeyID := getAPIKeyIDFromContext(c)
if sessionID := strings.TrimSpace(gjson.GetBytes(alphaBody, "id").String()); sessionID != "" {
isolated := isolateOpenAISessionID(apiKeyID, sessionID)
req.Header.Set("Session_ID", isolated)
req.Header.Set("Conversation_ID", isolated)
}
s.overrideBrowserUserAgent(ctx, account, req)
enforceCodexIdentityHeaders(req.Header)
account.ApplyHeaderOverrides(req.Header)
return req, nil
}
func buildOpenAIAlphaSearchResponsesWebSearchBody(alphaBody []byte, model string) ([]byte, error) {
if strings.TrimSpace(model) == "" {
return nil, fmt.Errorf("model is required")
}
tool := map[string]any{"type": "web_search"}
if contextSize := strings.TrimSpace(gjson.GetBytes(alphaBody, "settings.search_context_size").String()); contextSize != "" {
tool["search_context_size"] = contextSize
}
if userLocation := gjson.GetBytes(alphaBody, "settings.user_location"); userLocation.IsObject() {
var loc map[string]any
if err := json.Unmarshal([]byte(userLocation.Raw), &loc); err == nil && len(loc) > 0 {
tool["user_location"] = loc
}
}
payload := map[string]any{
"model": model,
"stream": true,
"store": false,
"input": []any{
map[string]any{
"role": "user",
"content": []any{
map[string]any{
"type": "input_text",
"text": openAIAlphaSearchResponsesWebSearchPrompt(alphaBody),
},
},
},
},
"tools": []any{tool},
}
return json.Marshal(payload)
}
func openAIAlphaSearchResponsesWebSearchPrompt(alphaBody []byte) string {
var b strings.Builder
_, _ = b.WriteString("Execute this Codex standalone web.run request for another model.\n")
_, _ = b.WriteString("Use the hosted web_search tool when web/current information is needed.\n")
_, _ = b.WriteString("Return concise source-backed results. Include titles, URLs, dates, and direct answers when available.\n")
if commands := strings.TrimSpace(gjson.GetBytes(alphaBody, "commands").Raw); commands != "" {
_, _ = b.WriteString("\nCommands JSON:\n")
_, _ = b.WriteString(truncateOpenAIAlphaSearchPromptJSON(commands, 12000))
}
if settings := strings.TrimSpace(gjson.GetBytes(alphaBody, "settings").Raw); settings != "" {
_, _ = b.WriteString("\n\nSearch settings JSON:\n")
_, _ = b.WriteString(truncateOpenAIAlphaSearchPromptJSON(settings, 4000))
}
if input := strings.TrimSpace(gjson.GetBytes(alphaBody, "input").Raw); input != "" {
_, _ = b.WriteString("\n\nRecent conversation/input JSON:\n")
_, _ = b.WriteString(truncateOpenAIAlphaSearchPromptJSON(input, 8000))
}
if b.Len() == 0 {
return "Execute the requested web search and return concise source-backed results."
}
return b.String()
}
func truncateOpenAIAlphaSearchPromptJSON(value string, limit int) string {
value = strings.TrimSpace(value)
if limit <= 0 || len(value) <= limit {
return value
}
return value[:limit] + "\n...<truncated>"
}
func (s *OpenAIGatewayService) buildOpenAIAlphaSearchRequest(ctx context.Context, c *gin.Context, account *Account, body []byte, token string) (*http.Request, error) {
targetURL, err := s.openAIAlphaSearchURL(account)
if err != nil {
return nil, err
@@ -129,19 +350,279 @@ func (s *OpenAIGatewayService) buildOpenAIAlphaSearchRequest(ctx context.Context
}
parsedURL.RawQuery = query.Encode()
}
req.URL = parsedURL
req, err := http.NewRequestWithContext(ctx, http.MethodPost, parsedURL.String(), bytes.NewReader(body))
if err != nil {
return nil, err
}
req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
if err != nil {
return nil, fmt.Errorf("build openai authentication headers: %w", err)
}
for key, values := range authHeaders {
for _, value := range values {
req.Header.Add(key, value)
}
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
if clientBeta == "" {
req.Header.Del("OpenAI-Beta")
}
if version := strings.TrimSpace(c.GetHeader("Version")); version != "" {
req.Header.Set("Version", version)
} else if account.Type == AccountTypeOAuth {
req.Header.Set("Version", codexCLIVersion)
if account.Type == AccountTypeOAuth {
req.Host = "chatgpt.com"
if err := resolveAndSetOpenAIChatGPTAccountHeaders(ctx, s.accountRepo, req.Header, account); err != nil {
return nil, fmt.Errorf("resolve chatgpt account headers: %w", err)
}
if turnMetadata := openAIAlphaSearchInboundHeader(c, "X-Codex-Turn-Metadata"); turnMetadata != "" {
req.Header.Set("X-Codex-Turn-Metadata", turnMetadata)
}
if version := openAIAlphaSearchInboundHeader(c, "Version"); version != "" {
req.Header.Set("Version", version)
} else {
req.Header.Set("Version", codexCLIVersion)
}
if originator := openAIAlphaSearchInboundHeader(c, "Originator"); originator != "" {
req.Header.Set("Originator", originator)
} else {
req.Header.Set("Originator", "codex_cli_rs")
}
if customUA := account.GetOpenAIUserAgent(); customUA != "" {
req.Header.Set("User-Agent", customUA)
} else if userAgent := openAIAlphaSearchInboundHeader(c, "User-Agent"); userAgent != "" {
req.Header.Set("User-Agent", userAgent)
} else {
req.Header.Set("User-Agent", codexCLIUserAgent)
}
if s.cfg != nil && s.cfg.Gateway.ForceCodexCLI {
req.Header.Set("User-Agent", codexCLIUserAgent)
}
s.overrideBrowserUserAgent(ctx, account, req)
enforceCodexIdentityHeaders(req.Header)
}
account.ApplyHeaderOverrides(req.Header)
stripOpenAIAlphaSearchResponsesHeaders(req.Header)
return req, nil
}
// stripOpenAIAlphaSearchResponsesHeaders 让独立搜索请求与官方 Codex
// SearchClient 的线协议保持一致。alpha/search 不是 /responses 的子请求:官方
// 客户端仅在 Provider/Auth 基础头之外附加 x-codex-turn-metadata,不发送
// OpenAI-Beta、会话隔离或 Responses Lite 状态头。originator 与 User-Agent
// 属于官方默认客户端头,必须保留。
//
// alpha/search 使用专用构造器生成官方 SearchClient 的最小线协议形态;
// 该函数作为最后一道防线,避免账号 header 覆写或后续改动重新带入
// Responses 专用头,使 PAT 的 alpha/search 被上游按错误认证路径处理。
func stripOpenAIAlphaSearchResponsesHeaders(headers http.Header) {
if headers == nil {
return
}
for _, key := range []string{
"OpenAI-Beta",
"Session_ID",
"Conversation_ID",
"X-Codex-Beta-Features",
"X-Codex-Turn-State",
responsesLiteHeaderKey,
} {
headers.Del(key)
}
}
func openAIAlphaSearchInboundHeader(c *gin.Context, key string) string {
if c == nil {
return ""
}
return strings.TrimSpace(c.GetHeader(key))
}
var openAIAlphaSearchUnsupportedBodyFields = [...]string{
// Codex alpha/search 是 SearchRequest 独立协议,不是 /responses 子请求。
// 新版 Codex/第三方代理可能把 Responses 公共字段误带到搜索请求里;ChatGPT
// alpha/search 会对这些字段返回 Unknown parameter(例如 prompt_cache_key)。
"prompt_cache_key",
"prompt_cache_retention",
}
func sanitizeOpenAIAlphaSearchBody(body []byte) ([]byte, error) {
if len(body) == 0 {
return body, nil
}
var obj map[string]json.RawMessage
if err := json.Unmarshal(body, &obj); err != nil || obj == nil {
return body, nil
}
changed := false
for _, field := range openAIAlphaSearchUnsupportedBodyFields {
if _, ok := obj[field]; ok {
delete(obj, field)
changed = true
}
}
if !changed {
return body, nil
}
out, err := json.Marshal(obj)
if err != nil {
return nil, err
}
return out, nil
}
func (s *OpenAIGatewayService) ensureOpenAIAlphaSearchAuthMetadata(ctx context.Context, account *Account, token string, proxyURL string) error {
if s == nil || account == nil || !account.IsOpenAIPersonalAccessToken() {
return nil
}
if strings.TrimSpace(account.GetChatGPTAccountID()) != "" {
return nil
}
var oauthService *OpenAIOAuthService
if s.openAITokenProvider != nil {
oauthService = s.openAITokenProvider.openAIOAuthService
}
if oauthService == nil {
return nil
}
tokenInfo, err := oauthService.ValidateCodexPersonalAccessToken(ctx, token, proxyURL)
if err != nil {
return fmt.Errorf("validate Codex PAT metadata for alpha/search: %w", err)
}
credentials := shallowCopyMap(account.Credentials)
for key, value := range oauthService.BuildAccountCredentials(tokenInfo) {
credentials[key] = value
}
credentials = NormalizeOpenAIPersonalAccessTokenCredentials(account, tokenInfo, credentials)
account.Credentials = shallowCopyMap(credentials)
if s.accountRepo != nil {
if err := persistAccountCredentials(ctx, s.accountRepo, account, credentials); err != nil {
return fmt.Errorf("persist Codex PAT metadata for alpha/search: %w", err)
}
}
return nil
}
func shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(statusCode int) bool {
return statusCode != http.StatusUnauthorized
}
func openAIAlphaSearchResponseFromResponsesSSE(body []byte) ([]byte, error) {
output, results := parseOpenAIResponsesSSEForAlphaSearch(body)
resp := map[string]any{
"output": output,
}
if len(results) > 0 {
resp["results"] = results
}
return json.Marshal(resp)
}
func parseOpenAIResponsesSSEForAlphaSearch(body []byte) (string, []any) {
text := strings.ReplaceAll(string(body), "\r\n", "\n")
var output strings.Builder
var completedResponse any
results := make([]any, 0)
seenURLs := make(map[string]struct{})
for _, block := range strings.Split(text, "\n\n") {
data := openAIAlphaSearchSSEData(block)
if data == "" || data == "[DONE]" {
continue
}
var event map[string]any
if err := json.Unmarshal([]byte(data), &event); err != nil {
continue
}
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)
}
out := output.String()
if strings.TrimSpace(out) == "" && completedResponse != nil {
out = extractOpenAIResponsesCompletedText(completedResponse)
collectOpenAIAlphaSearchURLCitations(completedResponse, &results, seenURLs)
}
return out, results
}
func openAIAlphaSearchSSEData(block string) string {
var lines []string
for _, line := range strings.Split(block, "\n") {
line = strings.TrimRight(line, "\r")
if !strings.HasPrefix(line, "data:") {
continue
}
lines = append(lines, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
}
return strings.TrimSpace(strings.Join(lines, "\n"))
}
func extractOpenAIResponsesCompletedText(response any) string {
resp, ok := response.(map[string]any)
if !ok {
return ""
}
outputItems, _ := resp["output"].([]any)
var b strings.Builder
for _, item := range outputItems {
itemMap, ok := item.(map[string]any)
if !ok || itemMap["type"] != "message" {
continue
}
contentItems, _ := itemMap["content"].([]any)
for _, content := range contentItems {
contentMap, ok := content.(map[string]any)
if !ok {
continue
}
if contentMap["type"] == "output_text" {
if text, _ := contentMap["text"].(string); text != "" {
_, _ = b.WriteString(text)
}
}
}
}
return b.String()
}
func collectOpenAIAlphaSearchURLCitations(value any, results *[]any, seen map[string]struct{}) {
switch typed := value.(type) {
case map[string]any:
if typed["type"] == "url_citation" {
if urlValue, _ := typed["url"].(string); strings.TrimSpace(urlValue) != "" {
urlValue = strings.TrimSpace(urlValue)
if _, exists := seen[urlValue]; !exists {
seen[urlValue] = struct{}{}
result := map[string]any{
"type": "text_result",
"ref_id": fmt.Sprintf("turn0search%d", len(*results)),
"url": urlValue,
}
if title, _ := typed["title"].(string); strings.TrimSpace(title) != "" {
result["title"] = strings.TrimSpace(title)
}
*results = append(*results, result)
}
}
}
for _, child := range typed {
collectOpenAIAlphaSearchURLCitations(child, results, seen)
}
case []any:
for _, child := range typed {
collectOpenAIAlphaSearchURLCitations(child, results, seen)
}
}
}
func (s *OpenAIGatewayService) openAIAlphaSearchURL(account *Account) (string, error) {
if account == nil {
return "", fmt.Errorf("account is required")
@@ -6,7 +6,9 @@ import (
"io"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"sync/atomic"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
@@ -15,6 +17,27 @@ import (
"github.com/tidwall/gjson"
)
type alphaSearchAccountStateRepo struct {
AccountRepository
setErrorCalls int
lastError string
}
func (r *alphaSearchAccountStateRepo) SetError(_ context.Context, _ int64, errorMsg string) error {
r.setErrorCalls++
r.lastError = errorMsg
return nil
}
func alphaSearchResponsesSSE(output string) string {
return "event: response.output_text.delta\n" +
`data: {"type":"response.output_text.delta","delta":` + strconv.Quote(output) + `}` + "\n\n" +
"event: response.output_text.annotation.added\n" +
`data: {"type":"response.output_text.annotation.added","annotation":{"type":"url_citation","url":"https://example.com/news","title":"Example News"}}` + "\n\n" +
"event: response.completed\n" +
`data: {"type":"response.completed","response":{"output":[{"type":"message","content":[{"type":"output_text","text":` + strconv.Quote(output) + `}]}]}}` + "\n\n"
}
func TestForwardAlphaSearchOAuthPreservesWire(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{
@@ -70,6 +93,143 @@ func TestForwardAlphaSearchOAuthPreservesWire(t *testing.T) {
require.JSONEq(t, string(body), string(upstream.lastBody))
}
func TestForwardAlphaSearchPATUsesResponsesWebSearchFallback(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{
"id":"search-session",
"model":"gpt-5.6-sol",
"commands":{"search_query":[{"q":"OpenAI news"}]},
"prompt_cache_key":"responses-cache-key",
"prompt_cache_retention":"24h"
}`)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/alpha/search", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
c.Request.Header.Set("User-Agent", codexCLIUserAgent)
c.Request.Header.Set("Originator", "codex_cli_rs")
c.Request.Header.Set("Version", "0.144.1")
c.Request.Header.Set("OpenAI-Beta", "responses=experimental")
c.Request.Header.Set("Accept-Language", "zh-CN")
c.Request.Header.Set("Authorization", "Bearer client-token")
c.Request.Header.Set("Session_ID", "session-client")
c.Request.Header.Set("Conversation_ID", "conversation-client")
c.Request.Header.Set("X-Codex-Beta-Features", "feature-a")
c.Request.Header.Set("X-Codex-Turn-State", "turn-state")
c.Request.Header.Set(responsesLiteHeaderKey, "true")
c.Request.Header.Set("X-Codex-Turn-Metadata", `{"turn_id":"turn-1"}`)
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"req-search"}},
Body: io.NopCloser(strings.NewReader(alphaSearchResponsesSSE("search result"))),
}}
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": "chatgpt-account",
"chatgpt_account_is_fedramp": true,
},
}
result, err := service.ForwardAlphaSearch(context.Background(), c, account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 1, result.WebSearchCalls)
require.Equal(t, "/v1/responses", result.UpstreamEndpoint)
require.Equal(t, http.StatusOK, recorder.Code)
require.JSONEq(t, `{"output":"search result","results":[{"type":"text_result","ref_id":"turn0search0","url":"https://example.com/news","title":"Example News"}]}`, recorder.Body.String())
require.Equal(t, chatgptCodexURL, upstream.lastReq.URL.String())
require.Equal(t, "Bearer at-test-token", upstream.lastReq.Header.Get("Authorization"))
require.Equal(t, "chatgpt-account", upstream.lastReq.Header.Get("ChatGPT-Account-ID"))
require.Equal(t, "true", upstream.lastReq.Header.Get("X-OpenAI-Fedramp"))
require.Equal(t, "application/json", upstream.lastReq.Header.Get("Content-Type"))
require.Equal(t, "text/event-stream", upstream.lastReq.Header.Get("Accept"))
require.Equal(t, "responses=experimental", upstream.lastReq.Header.Get("OpenAI-Beta"))
require.Equal(t, "0.144.1", upstream.lastReq.Header.Get("Version"))
require.Equal(t, `{"turn_id":"turn-1"}`, upstream.lastReq.Header.Get("X-Codex-Turn-Metadata"))
require.Equal(t, "codex_cli_rs", upstream.lastReq.Header.Get("Originator"))
require.Empty(t, upstream.lastReq.Header.Get("X-Codex-Beta-Features"))
require.Empty(t, upstream.lastReq.Header.Get("X-Codex-Turn-State"))
require.Empty(t, upstream.lastReq.Header.Get(responsesLiteHeaderKey))
require.Empty(t, upstream.lastReq.Header.Get("Accept-Language"))
require.False(t, gjson.GetBytes(upstream.lastBody, "prompt_cache_key").Exists())
require.False(t, gjson.GetBytes(upstream.lastBody, "prompt_cache_retention").Exists())
require.Equal(t, "gpt-5.6-sol", gjson.GetBytes(upstream.lastBody, "model").String())
require.True(t, gjson.GetBytes(upstream.lastBody, "stream").Bool())
require.False(t, gjson.GetBytes(upstream.lastBody, "store").Bool())
require.Equal(t, "web_search", gjson.GetBytes(upstream.lastBody, "tools.0.type").String())
require.Contains(t, gjson.GetBytes(upstream.lastBody, "input.0.content.0.text").String(), `"search_query"`)
}
func TestForwardAlphaSearchPATBackfillsMissingChatGPTAccountMetadata(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"id":"search-session","model":"gpt-5.6-sol","commands":{"search_query":[{"q":"OpenAI news"}]}}`)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/alpha/search", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
var whoamiCalls int32
whoamiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&whoamiCalls, 1)
require.Equal(t, "Bearer at-test-token", r.Header.Get("Authorization"))
require.Equal(t, "application/json", r.Header.Get("Accept"))
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{
"email":"pat@example.com",
"chatgpt_user_id":"user-123",
"chatgpt_account_id":"acct-123",
"chatgpt_plan_type":"plus",
"chatgpt_account_is_fedramp":true
}`))
}))
defer whoamiServer.Close()
oldWhoamiURL := openAICodexPATWhoamiURL
openAICodexPATWhoamiURL = whoamiServer.URL
defer func() { openAICodexPATWhoamiURL = oldWhoamiURL }()
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(`{"output":"search result"}`)),
}}
oauthService := NewOpenAIOAuthService(nil, nil)
service := &OpenAIGatewayService{
cfg: &config.Config{},
httpUpstream: upstream,
openAITokenProvider: NewOpenAITokenProvider(nil, nil, oauthService),
}
account := &Account{
ID: 45,
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
"access_token": "at-test-token",
"auth_mode": OpenAIAuthModePersonalAccessToken,
},
}
result, err := service.ForwardAlphaSearch(context.Background(), c, account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, int32(1), atomic.LoadInt32(&whoamiCalls))
require.Equal(t, "acct-123", upstream.lastReq.Header.Get("ChatGPT-Account-ID"))
require.Equal(t, "true", upstream.lastReq.Header.Get("X-OpenAI-Fedramp"))
require.Equal(t, "acct-123", account.Credentials["chatgpt_account_id"])
require.Equal(t, "user-123", account.Credentials["chatgpt_user_id"])
require.Equal(t, OpenAIAuthModePersonalAccessToken, account.Credentials["auth_mode"])
}
func TestForwardAlphaSearchAPIKeyMapsModelAndPassesThroughError(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"id":"search-session","model":"gpt-5.6-sol","commands":{"search_query":[{"q":"news"}]}}`)
@@ -143,3 +303,98 @@ func TestForwardAlphaSearchReturnsFailoverBeforeWriting(t *testing.T) {
require.False(t, c.Writer.Written())
require.Empty(t, recorder.Body.String())
}
func TestForwardAlphaSearchUnauthorizedDoesNotMarkAccountError(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"id":"search-session","model":"gpt-5.6-sol","commands":{"search_query":[{"q":"news"}]}}`)
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.StatusUnauthorized,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(`{"detail":"Unauthorized"}`)),
}}
repo := &alphaSearchAccountStateRepo{}
cfg := &config.Config{}
service := &OpenAIGatewayService{
cfg: cfg,
httpUpstream: upstream,
accountRepo: repo,
rateLimitService: NewRateLimitService(repo, nil, cfg, nil, nil),
}
account := &Account{
ID: 44,
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
// 刻意不设置 auth_mode:覆盖历史上把 at- token 当普通 OAuth 导入的账号。
"access_token": "at-test-token",
"chatgpt_account_id": "chatgpt-account",
},
}
result, err := service.ForwardAlphaSearch(context.Background(), c, account, body)
require.Nil(t, result)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, http.StatusUnauthorized, failoverErr.StatusCode)
require.Zero(t, repo.setErrorCalls)
require.Empty(t, repo.lastError)
require.False(t, c.Writer.Written())
}
func TestForwardAlphaSearchPATResponsesFallbackUnauthorizedDoesNotMarkAccountError(t *testing.T) {
gin.SetMode(gin.TestMode)
body := []byte(`{"id":"search-session","model":"gpt-5.6-sol","commands":{"search_query":[{"q":"news"}]}}`)
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.StatusUnauthorized,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(`{"detail":"Unauthorized"}`)),
}}
repo := &alphaSearchAccountStateRepo{}
cfg := &config.Config{}
service := &OpenAIGatewayService{
cfg: cfg,
httpUpstream: upstream,
accountRepo: repo,
rateLimitService: NewRateLimitService(repo, nil, cfg, nil, nil),
}
account := &Account{
ID: 46,
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
"access_token": "at-test-token",
"auth_mode": OpenAIAuthModePersonalAccessToken,
"chatgpt_account_id": "chatgpt-account",
},
}
result, err := service.ForwardAlphaSearch(context.Background(), c, account, body)
require.Nil(t, result)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, http.StatusUnauthorized, failoverErr.StatusCode)
require.Equal(t, chatgptCodexURL, upstream.lastReq.URL.String())
require.Equal(t, "text/event-stream", upstream.lastReq.Header.Get("Accept"))
require.Equal(t, "responses=experimental", upstream.lastReq.Header.Get("OpenAI-Beta"))
require.Zero(t, repo.setErrorCalls)
require.Empty(t, repo.lastError)
require.False(t, c.Writer.Written())
}
func TestShouldApplyOpenAIAlphaSearchAccountErrorSideEffects(t *testing.T) {
require.False(t, shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(http.StatusUnauthorized))
require.True(t, shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(http.StatusForbidden))
require.True(t, shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(http.StatusTooManyRequests))
}
@@ -492,9 +492,24 @@ func TestAccountSupportsOpenAIEndpointCapability(t *testing.T) {
}
require.True(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityChatCompletions))
require.True(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityAlphaSearch))
require.False(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityEmbeddings))
})
t.Run("alpha search 仅允许 OpenAI OAuth/PAT 类账号", func(t *testing.T) {
apiKey := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
}
oauth := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
}
require.False(t, apiKey.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityAlphaSearch))
require.True(t, oauth.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityAlphaSearch))
})
t.Run("显式列表支持同时声明 chat 和 embeddings", func(t *testing.T) {
account := &Account{
Platform: PlatformOpenAI,
@@ -518,9 +533,22 @@ func TestAccountSupportsOpenAIEndpointCapability(t *testing.T) {
}
require.True(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityChatCompletions))
require.False(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityAlphaSearch))
require.False(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityEmbeddings))
})
t.Run("OAuth 显式列表沿用 chat 能力放行 alpha search", func(t *testing.T) {
account := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Credentials: map[string]any{
"openai_capabilities": []any{"chat_completions"},
},
}
require.True(t, account.SupportsOpenAIEndpointCapability(OpenAIEndpointCapabilityAlphaSearch))
})
t.Run("显式 map 支持单独关闭 chat 并开启 embeddings", func(t *testing.T) {
account := &Account{
Platform: PlatformOpenAI,