From 613722eee434c0a62e9397487b3713c16362e18d Mon Sep 17 00:00:00 2001 From: xlplbo Date: Fri, 11 Sep 2026 11:36:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai):=20HTTP=20=E8=BD=AC=E5=8F=91?= =?UTF-8?q?=E7=9A=84=E6=89=A7=E8=A1=8C=E4=BD=9C=E7=94=A8=E5=9F=9F=E6=94=B9?= =?UTF-8?q?=E4=BB=8E=E5=8E=9F=E5=A7=8B=E8=AF=B7=E6=B1=82=E8=AE=A1=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_gateway_forward.go | 5 + .../service/openai_ws_execution_scope.go | 22 ----- .../service/openai_ws_forwarder_v2.go | 6 +- ...ai_ws_forwarder_v2_execution_scope_test.go | 97 +++++++++++++++++++ 4 files changed, 106 insertions(+), 24 deletions(-) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 6837cb8236..ef716e0f99 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -42,6 +42,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco restrictionResult := s.detectCodexClientRestriction(c, account, body) apiKeyID := getAPIKeyIDFromContext(c) + // 执行作用域必须取自客户端原始身份：后面的账号 namespace 改写与指纹收敛会改掉 + // 请求体里的 client_metadata / prompt_cache_key，用改写后的值取键会让不同会话 + // 落到同一个键，也会与 WS 接入路径按原始报文算出的键对不上。 + wsExecutionScope, _ := resolveOpenAIWSExecutionScope(c, body, apiKeyID) logCodexCLIOnlyDetection(ctx, c, account, apiKeyID, restrictionResult, body) if restrictionResult.Enabled && !restrictionResult.Matched { MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalPolicyDenied) @@ -874,6 +878,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco account, wsReqBody, clientPromptCacheKey, + wsExecutionScope, token, wsDecision, isCodexCLI, diff --git a/backend/internal/service/openai_ws_execution_scope.go b/backend/internal/service/openai_ws_execution_scope.go index fbd1994058..730e86cb01 100644 --- a/backend/internal/service/openai_ws_execution_scope.go +++ b/backend/internal/service/openai_ws_execution_scope.go @@ -40,28 +40,6 @@ func resolveOpenAIWSClientThreadID(c *gin.Context, body []byte) string { return codexTurnMetadataThreadID(gjson.GetBytes(body, "client_metadata."+openAIWSTurnMetadataHeader).String()) } -// openAIWSExecutionScopeBodyFromRequest 只序列化执行作用域会读取的字段， -// 供已解析成 map 的 HTTP 请求体取键，避免为此重新编码整个请求体。 -func openAIWSExecutionScopeBodyFromRequest(reqBody map[string]any) []byte { - if len(reqBody) == 0 { - return nil - } - subset := make(map[string]any, 3) - for _, key := range []string{"client_metadata", "prompt_cache_key", "previous_response_id"} { - if value, ok := reqBody[key]; ok { - subset[key] = value - } - } - if len(subset) == 0 { - return nil - } - raw, err := json.Marshal(subset) - if err != nil { - return nil - } - return raw -} - func codexTurnMetadataThreadID(raw string) string { raw = strings.TrimSpace(raw) if raw == "" { diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 30f897928d..d908b0ef1a 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -23,6 +23,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( account *Account, reqBody map[string]any, clientPromptCacheKey string, + executionScope string, token string, decision OpenAIWSProtocolDecision, isCodexCLI bool, @@ -126,8 +127,9 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( } // 与 WS 接入路径共用执行作用域：codex 多智能体共用 session-id，turn state 与 // store=false 的连接绑定必须按线程隔离，同一线程在两条路径之间也才能共享状态。 - if scope, _ := resolveOpenAIWSExecutionScope(c, openAIWSExecutionScopeBodyFromRequest(reqBody), getAPIKeyIDFromContext(c)); scope != "" { - sessionHash = scope + // 作用域由 Forward 从改写前的原始请求算出后传入，reqBody 此时已带账号 namespace。 + if executionScope = strings.TrimSpace(executionScope); executionScope != "" { + sessionHash = executionScope } if turnState == "" && stateStore != nil && sessionHash != "" { if savedTurnState, ok := stateStore.GetSessionTurnState(groupID, sessionHash); ok { diff --git a/backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go b/backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go index 59d22036c9..da13cd1fcb 100644 --- a/backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go +++ b/backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "net/http" "net/http/httptest" "testing" @@ -102,3 +103,99 @@ func TestOpenAIGatewayService_Forward_WSv2_TurnStateBoundToExecutionScope(t *tes require.True(t, turnStateOnScope, "turn state 应绑定到执行作用域") require.Equal(t, "turn-state-from-upstream", turnState) } + +// Forward 在转发前会按账号 namespace 改写请求体里的 client_metadata，指纹 full 模式还会给每个 +// 请求注入同一个 thread_id。执行作用域必须取自客户端原始请求：否则同一 API key 下不同 +// session_id 的会话会落到同一个键共用 turn state，客户端自带线程标识时也会与 WS 接入路径 +// 按原始报文算出的键对不上。 +func TestOpenAIGatewayService_Forward_WSv2_ExecutionScopeUsesOriginalIdentity(t *testing.T) { + gin.SetMode(gin.TestMode) + + cfg := &config.Config{} + cfg.Security.URLAllowlist.Enabled = false + cfg.Security.URLAllowlist.AllowInsecureHTTP = true + cfg.Gateway.OpenAIWS.Enabled = true + cfg.Gateway.OpenAIWS.OAuthEnabled = true + cfg.Gateway.OpenAIWS.APIKeyEnabled = true + cfg.Gateway.OpenAIWS.ResponsesWebsocketsV2 = true + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1 + + completed := func(id string) []byte { + return []byte(`{"type":"response.completed","response":{"id":"` + id + `","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`) + } + captureConn := &openAIWSCaptureConn{events: [][]byte{completed("resp_a"), completed("resp_b"), completed("resp_c")}} + handshake := http.Header{} + handshake.Set(openAIWSTurnStateHeader, "turn-state-from-upstream") + pool := newOpenAIWSConnPool(cfg) + pool.setClientDialerForTest(&openAIWSCaptureDialer{conn: captureConn, handshake: handshake}) + defer pool.Close() + + stateStore := NewOpenAIWSStateStore(nil) + svc := &OpenAIGatewayService{ + cfg: cfg, + httpUpstream: &httpUpstreamRecorder{}, + cache: &stubGatewayCache{}, + openaiWSResolver: NewOpenAIWSProtocolResolver(cfg), + toolCorrector: NewCodexToolCorrector(), + openaiWSPool: pool, + openaiWSStateStore: stateStore, + } + groupID := int64(9) + const apiKeyID = int64(21) + account := &Account{ + ID: 457, + Name: "openai-oauth-fingerprint-full", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{"access_token": "oauth-token-1"}, + Extra: map[string]any{ + "responses_websockets_v2_enabled": true, + codexFingerprintModeExtraKey: "full", + codexFingerprintSeedExtraKey: "11111111-1111-4111-8111-aaaaaaaaaaaa", + }, + } + + forward := func(sessionID, body string) (*gin.Context, []byte) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", nil) + c.Request.Header.Set("session_id", sessionID) + c.Set("api_key", &APIKey{ID: apiKeyID, GroupID: &groupID}) + raw := []byte(body) + result, err := svc.Forward(context.Background(), c, account, raw) + require.NoError(t, err) + require.NotNil(t, result) + return c, raw + } + + plainBody := `{"model":"gpt-5.1","stream":false,"input":[{"type":"input_text","text":"hello"}]}` + cA, rawA := forward("session-a", plainBody) + cB, rawB := forward("session-b", plainBody) + + scopeA, _ := resolveOpenAIWSExecutionScope(cA, rawA, apiKeyID) + scopeB, _ := resolveOpenAIWSExecutionScope(cB, rawB, apiKeyID) + require.NotEmpty(t, scopeA) + require.NotEqual(t, scopeA, scopeB, "不同 session_id 的会话必须落在不同的作用域") + _, boundA := stateStore.GetSessionTurnState(groupID, scopeA) + require.True(t, boundA, "会话 A 的 turn state 应落在按原始 session_id 算出的作用域") + _, boundB := stateStore.GetSessionTurnState(groupID, scopeB) + require.True(t, boundB, "会话 B 的 turn state 应落在按原始 session_id 算出的作用域") + + injected := resolveCodexFingerprintIDs(account, "", codexFingerprintFull) + require.NotNil(t, injected) + injectedScope, _ := deriveOpenAISessionHashes(fmt.Sprintf("openai_ws_exec:%d|thread=%s", apiKeyID, injected.threadID)) + _, boundToInjected := stateStore.GetSessionTurnState(groupID, injectedScope) + require.False(t, boundToInjected, "指纹收敛注入的固定 thread_id 不得成为状态键") + + threadBody := `{"model":"gpt-5.1","stream":false,"client_metadata":{"thread_id":"child-thread"},"input":[{"type":"input_text","text":"hello"}]}` + cC, rawC := forward("session-c", threadBody) + scopeC, threadC := resolveOpenAIWSExecutionScope(cC, rawC, apiKeyID) + require.Equal(t, "child-thread", threadC) + _, boundC := stateStore.GetSessionTurnState(groupID, scopeC) + require.True(t, boundC, "客户端自带线程标识时，键必须与按原始报文算出的一致，不受账号 namespace 改写影响") +}