From e4c369bd56305c9984c01cf9eb2bfb6f1cb8da1a Mon Sep 17 00:00:00 2001 From: xlplbo Date: Fri, 11 Sep 2026 10:25:30 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai):=20HTTP=20=E7=BB=8F=20WS=20?= =?UTF-8?q?=E6=B1=A0=E8=BD=AC=E5=8F=91=E6=8C=89=E6=89=A7=E8=A1=8C=E4=BD=9C?= =?UTF-8?q?=E7=94=A8=E5=9F=9F=E5=8F=96=E4=BC=9A=E8=AF=9D=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E9=94=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_ws_execution_scope.go | 22 ++++ .../openai_ws_forwarder_success_test.go | 4 +- .../service/openai_ws_forwarder_v2.go | 5 + ...ai_ws_forwarder_v2_execution_scope_test.go | 104 ++++++++++++++++++ 4 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go diff --git a/backend/internal/service/openai_ws_execution_scope.go b/backend/internal/service/openai_ws_execution_scope.go index 730e86cb01..fbd1994058 100644 --- a/backend/internal/service/openai_ws_execution_scope.go +++ b/backend/internal/service/openai_ws_execution_scope.go @@ -40,6 +40,28 @@ 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_success_test.go b/backend/internal/service/openai_ws_forwarder_success_test.go index 6923352d7d..0401f59f55 100644 --- a/backend/internal/service/openai_ws_forwarder_success_test.go +++ b/backend/internal/service/openai_ws_forwarder_success_test.go @@ -1403,7 +1403,9 @@ func TestOpenAIGatewayService_Forward_WSv2_TurnStateAndMetadataReplayOnReconnect require.NoError(t, err) require.NotNil(t, result1) - sessionHash := svc.GenerateSessionHash(c1, reqBody) + // 会话级状态按执行作用域取键（显式 session_id 也在其中），不再是原会话哈希。 + sessionHash, _ := resolveOpenAIWSExecutionScope(c1, reqBody, getAPIKeyIDFromContext(c1)) + require.NotEmpty(t, sessionHash) store := svc.getOpenAIWSStateStore() turnState, ok := store.GetSessionTurnState(0, sessionHash) require.True(t, ok) diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 9f04d42640..30f897928d 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -124,6 +124,11 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( sessionHash, legacySessionHash = openAIWSSessionHashesFromID(promptCacheKey) attachOpenAILegacySessionHashToGin(c, legacySessionHash) } + // 与 WS 接入路径共用执行作用域：codex 多智能体共用 session-id，turn state 与 + // store=false 的连接绑定必须按线程隔离，同一线程在两条路径之间也才能共享状态。 + if scope, _ := resolveOpenAIWSExecutionScope(c, openAIWSExecutionScopeBodyFromRequest(reqBody), getAPIKeyIDFromContext(c)); scope != "" { + sessionHash = scope + } if turnState == "" && stateStore != nil && sessionHash != "" { if savedTurnState, ok := stateStore.GetSessionTurnState(groupID, sessionHash); ok { turnState = savedTurnState 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 new file mode 100644 index 0000000000..59d22036c9 --- /dev/null +++ b/backend/internal/service/openai_ws_forwarder_v2_execution_scope_test.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "github.com/stretchr/testify/require" +) + +// 场景：codex 子智能体经 HTTP 入站、网关内部走 WS 池转发时，与父线程共用 session_id 头。 +// turn state 必须落在执行作用域键下，而不是按 session_id 算出的会话哈希下， +// 否则子智能体会覆盖父线程的绑定；同一线程在 WS 接入与 HTTP 路径之间也才能共享状态。 +func TestOpenAIGatewayService_Forward_WSv2_TurnStateBoundToExecutionScope(t *testing.T) { + gin.SetMode(gin.TestMode) + + upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} + wsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + respHeader := http.Header{} + respHeader.Set("x-codex-turn-state", "turn-state-from-upstream") + conn, err := upgrader.Upgrade(w, r, respHeader) + if err != nil { + t.Errorf("upgrade websocket failed: %v", err) + return + } + defer func() { _ = conn.Close() }() + var request map[string]any + if err := conn.ReadJSON(&request); err != nil { + t.Errorf("read ws request failed: %v", err) + return + } + if err := conn.WriteJSON(map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": "resp_exec_scope_http", + "model": "gpt-5.1", + "usage": map[string]any{"input_tokens": 2, "output_tokens": 1}, + }, + }); err != nil { + t.Errorf("write response.completed failed: %v", err) + } + })) + defer wsServer.Close() + + 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 = 0 + + svc := &OpenAIGatewayService{ + cfg: cfg, + httpUpstream: &httpUpstreamRecorder{}, + cache: &stubGatewayCache{}, + openaiWSResolver: NewOpenAIWSProtocolResolver(cfg), + toolCorrector: NewCodexToolCorrector(), + } + groupID := int64(9) + account := &Account{ + ID: 456, + Name: "openai-http-exec-scope", + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{"api_key": "sk-test", "base_url": wsServer.URL}, + Extra: map[string]any{"responses_websockets_v2_enabled": true}, + } + + reqBody := []byte(`{"model":"gpt-5.1","stream":false,"input":[{"type":"input_text","text":"hello"}]}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", nil) + c.Request.Header.Set("session_id", "root-session") + c.Request.Header.Set(openAIWSTurnMetadataHeader, `{"session_id":"root-session","thread_id":"child-thread"}`) + c.Set("api_key", &APIKey{ID: 21, GroupID: &groupID}) + + result, err := svc.Forward(context.Background(), c, account, reqBody) + require.NoError(t, err) + require.NotNil(t, result) + + legacyHash := svc.GenerateSessionHash(c, reqBody) + scope, _ := resolveOpenAIWSExecutionScope(c, reqBody, 21) + require.NotEmpty(t, legacyHash) + require.Len(t, scope, 16) + require.NotEqual(t, legacyHash, scope) + + store := svc.getOpenAIWSStateStore() + _, turnStateOnLegacy := store.GetSessionTurnState(groupID, legacyHash) + require.False(t, turnStateOnLegacy, "turn state 不得绑定到按 session_id 算出的会话哈希") + turnState, turnStateOnScope := store.GetSessionTurnState(groupID, scope) + require.True(t, turnStateOnScope, "turn state 应绑定到执行作用域") + require.Equal(t, "turn-state-from-upstream", turnState) +}