fix(openai): HTTP 经 WS 池转发按执行作用域取会话状态键

This commit is contained in:
xlplbo
2026-09-11 10:25:50 +08:00
parent cd1ee1d1ad
commit e4c369bd56
4 changed files with 134 additions and 1 deletions
@@ -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 == "" {
@@ -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)
@@ -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
@@ -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)
}