From 612436a5a7cbb7ab9d4ce4174fcdff948fab31fb Mon Sep 17 00:00:00 2001 From: lbyxiaolizi Date: Mon, 17 Aug 2026 16:48:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai-compat):=20Responses=E2=86=92Chat=20?= =?UTF-8?q?=E6=A1=A5=E6=8E=A5=E6=8C=89=20reasoning=20item=20id=20=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E5=9B=9E=E6=B3=A8=20reasoning=5Fcontent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复 #5520：Codex 经 force_chat_completions 桥接到 DeepSeek thinking 上游时， 历史中的 encrypted-only reasoning item（summary 为空 + 不透明 encrypted_content， 远程 compaction / 跨会话恢复后常见）取不出明文，后续 assistant 消息缺 reasoning_content，DeepSeek 400 "The reasoning_content in the thinking mode must be passed back to the API"，客户端仅看到通用 502。 reasoning item 的 id 一定会被客户端回传，以其为 key 做服务端缓存： - GatewayCache 新增 Set/GetReasoningContent（Redis，默认 TTL 7 天） - 响应侧：流式扫 response.output_item.done、非流式扫 output，把 reasoning 全文按 item id 写缓存；客户端断连后 drain 期间用 detached ctx 仍会写完 - 请求侧：apicompat 新增 ResponsesToChatCompletionsRequestWithOptions 与 ReasoningContentByID 钩子，encrypted-only item 查缓存补回 pendingReasoning； 缓存 miss/出错一律 fail-open 维持原行为 - 自愈：历史里带明文 summary 的 reasoning item 顺手刷新缓存，覆盖 Redis flush / 跨实例漂移 测试：apicompat（命中恢复/miss 保持原样/明文优先）、service 端到端（流式 写缓存、请求侧回注+自愈）、repository miniredis 存取。 --- .../chatcompletions_responses_bridge.go | 63 +++++++- ...pletions_responses_reasoning_cache_test.go | 115 ++++++++++++++ backend/internal/repository/gateway_cache.go | 42 ++++++ .../gateway_cache_reasoning_test.go | 41 +++++ .../gateway_hotpath_optimization_test.go | 7 + .../service/gateway_multiplatform_test.go | 7 + backend/internal/service/gateway_service.go | 14 ++ .../service/gemini_multiplatform_test.go | 7 + .../service/openai_account_scheduler_test.go | 7 + .../openai_cyber_session_block_test.go | 7 + .../openai_gateway_responses_chat_fallback.go | 115 +++++++++++++- ...ai_gateway_responses_chat_fallback_test.go | 142 ++++++++++++++++++ .../service/openai_gateway_service_test.go | 7 + .../service/openai_ws_state_store_test.go | 7 + backend/internal/testutil/stubs.go | 7 + 15 files changed, 582 insertions(+), 6 deletions(-) create mode 100644 backend/internal/pkg/apicompat/chatcompletions_responses_reasoning_cache_test.go create mode 100644 backend/internal/repository/gateway_cache_reasoning_test.go diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go index 70dee035d8..05eb95af6e 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go @@ -17,15 +17,37 @@ const ( type toolOutputMediaByCallID map[string][]ChatContentPart +// ResponsesToChatOptions carries optional hooks for +// ResponsesToChatCompletionsRequestWithOptions. All fields are optional; a nil +// *ResponsesToChatOptions behaves exactly like ResponsesToChatCompletionsRequest. +type ResponsesToChatOptions struct { + // ReasoningContentByID looks up the cached reasoning text for a reasoning + // item id. Codex histories may carry reasoning items with no plaintext + // summary (empty summary + opaque encrypted_content, e.g. after remote + // compaction); DeepSeek's thinking mode rejects such histories with 400 + // "The `reasoning_content` in the thinking mode must be passed back to the + // API". The gateway caches the reasoning text it streamed under the item + // id, so the lookup restores the reasoning_content the client can no + // longer provide. Return "" on a miss. A nil lookup keeps the original + // behavior. + ReasoningContentByID func(itemID string) string +} + // ResponsesToChatCompletionsRequest converts a Responses API request into a // Chat Completions request for upstreams that only implement // /v1/chat/completions. func ResponsesToChatCompletionsRequest(req *ResponsesRequest) (*ChatCompletionsRequest, error) { + return ResponsesToChatCompletionsRequestWithOptions(req, nil) +} + +// ResponsesToChatCompletionsRequestWithOptions is ResponsesToChatCompletionsRequest +// with optional hooks (see ResponsesToChatOptions). +func ResponsesToChatCompletionsRequestWithOptions(req *ResponsesRequest, opts *ResponsesToChatOptions) (*ChatCompletionsRequest, error) { if req == nil { return nil, fmt.Errorf("responses request is nil") } - messages, err := responsesInputToChatMessages(req.Instructions, req.Input) + messages, err := responsesInputToChatMessagesWithOptions(req.Instructions, req.Input, opts) if err != nil { return nil, err } @@ -202,6 +224,12 @@ func HasToolSearchTool(tools []ResponsesTool) bool { // scattered across per-item cases, and makes unknown future codex item types // fail safe instead of leaking into the upstream request. func responsesInputToChatMessages(instructions string, inputRaw json.RawMessage) ([]ChatMessage, error) { + return responsesInputToChatMessagesWithOptions(instructions, inputRaw, nil) +} + +// responsesInputToChatMessagesWithOptions is responsesInputToChatMessages with +// optional hooks (see ResponsesToChatOptions). +func responsesInputToChatMessagesWithOptions(instructions string, inputRaw json.RawMessage, opts *ResponsesToChatOptions) ([]ChatMessage, error) { var messages []ChatMessage if strings.TrimSpace(instructions) != "" { content, _ := json.Marshal(instructions) @@ -226,7 +254,7 @@ func responsesInputToChatMessages(instructions string, inputRaw json.RawMessage) return nil, fmt.Errorf("parse responses input: %w", err) } - built, mediaByCallID, err := buildChatMessagesFromItems(messages, rawItems) + built, mediaByCallID, err := buildChatMessagesFromItems(messages, rawItems, opts) if err != nil { return nil, err } @@ -235,7 +263,7 @@ func responsesInputToChatMessages(instructions string, inputRaw json.RawMessage) // buildChatMessagesFromItems walks the Responses input items and appends the // corresponding Chat messages. -func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessage) ([]ChatMessage, toolOutputMediaByCallID, error) { +func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessage, opts *ResponsesToChatOptions) ([]ChatMessage, toolOutputMediaByCallID, error) { // pendingReasoning holds the reasoning text from a reasoning item until the // assistant message it belongs to is emitted. DeepSeek's thinking mode // requires the reasoning_content that produced a tool call to be passed back @@ -269,6 +297,15 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa case "reasoning": if txt := extractResponsesReasoningText(item); txt != "" { pendingReasoning = txt + } else if opts != nil && opts.ReasoningContentByID != nil { + // No plaintext summary (encrypted-only reasoning, e.g. after codex + // remote compaction): fall back to the gateway-side cache keyed + // by the reasoning item id, which always round-trips in history. + if id := rawString(item["id"]); id != "" { + if cached := opts.ReasoningContentByID(id); cached != "" { + pendingReasoning = cached + } + } } continue case "function_call": @@ -687,6 +724,26 @@ func extractResponsesReasoningText(item map[string]json.RawMessage) string { return strings.Join(parts, "\n") } +// ExtractResponsesReasoningItem parses a raw Responses input item and, when it +// is a reasoning item, returns its id and extractable plaintext (summary +// preferred, content fallback). ok is false for non-reasoning items. It exists +// for the gateway-side reasoning cache: items with plaintext get (re)cached so +// later encrypted-only replicas of the same item id can be restored. +func ExtractResponsesReasoningItem(raw json.RawMessage) (id string, text string, ok bool) { + raw = bytesTrimSpace(raw) + if len(raw) == 0 || string(raw) == "null" { + return "", "", false + } + var item map[string]json.RawMessage + if err := json.Unmarshal(raw, &item); err != nil { + return "", "", false + } + if rawString(item["type"]) != "reasoning" { + return "", "", false + } + return rawString(item["id"]), extractResponsesReasoningText(item), true +} + func chatCompletionsBridgeRole(role string) string { trimmed := strings.TrimSpace(role) if trimmed == "" { diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_reasoning_cache_test.go b/backend/internal/pkg/apicompat/chatcompletions_responses_reasoning_cache_test.go new file mode 100644 index 0000000000..33816f7abb --- /dev/null +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_reasoning_cache_test.go @@ -0,0 +1,115 @@ +package apicompat + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +// Encrypted-only reasoning items (empty summary + opaque encrypted_content, +// e.g. after codex remote compaction) carry no plaintext the bridge can map to +// reasoning_content. The gateway-side cache keyed by reasoning item id restores +// it; without the restore, DeepSeek thinking mode rejects the history with 400 +// "The `reasoning_content` in the thinking mode must be passed back to the API". +func TestResponsesToChat_ReasoningCacheLookup_RestoresEncryptedOnlyItem(t *testing.T) { + req := &ResponsesRequest{ + Model: "deepseek-reasoner", + Input: json.RawMessage(`[ + {"type":"reasoning","id":"item_enc1","summary":[],"encrypted_content":"opaque"}, + {"type":"function_call","call_id":"call_1","name":"get_value","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"ok"}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"go on"}]} + ]`), + } + + out, err := ResponsesToChatCompletionsRequestWithOptions(req, &ResponsesToChatOptions{ + ReasoningContentByID: func(itemID string) string { + if itemID == "item_enc1" { + return "cached thinking" + } + return "" + }, + }) + require.NoError(t, err) + require.Len(t, out.Messages, 3) + require.Equal(t, "assistant", out.Messages[0].Role) + require.Equal(t, "cached thinking", out.Messages[0].ReasoningContent) + require.Len(t, out.Messages[0].ToolCalls, 1) + require.Equal(t, "call_1", out.Messages[0].ToolCalls[0].ID) + require.Equal(t, "tool", out.Messages[1].Role) + require.Equal(t, "user", out.Messages[2].Role) +} + +// A cache miss keeps the original behavior: no reasoning_content, no error. +func TestResponsesToChat_ReasoningCacheLookup_MissKeepsOriginalBehavior(t *testing.T) { + req := &ResponsesRequest{ + Model: "deepseek-reasoner", + Input: json.RawMessage(`[ + {"type":"reasoning","id":"item_unknown","summary":[],"encrypted_content":"opaque"}, + {"type":"function_call","call_id":"call_1","name":"get_value","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"ok"}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"go on"}]} + ]`), + } + + out, err := ResponsesToChatCompletionsRequestWithOptions(req, &ResponsesToChatOptions{ + ReasoningContentByID: func(string) string { return "" }, + }) + require.NoError(t, err) + require.Len(t, out.Messages, 3) + require.Empty(t, out.Messages[0].ReasoningContent) + + // Nil options (legacy path) behaves identically. + legacy, err := ResponsesToChatCompletionsRequest(req) + require.NoError(t, err) + require.Equal(t, out.Messages, legacy.Messages) +} + +// Plaintext summary wins and the cache lookup is not consulted. +func TestResponsesToChat_ReasoningCacheLookup_PlaintextPreferred(t *testing.T) { + req := &ResponsesRequest{ + Model: "deepseek-reasoner", + Input: json.RawMessage(`[ + {"type":"reasoning","id":"item_plain","summary":[{"type":"summary_text","text":"plain thinking"}]}, + {"type":"function_call","call_id":"call_1","name":"get_value","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"ok"}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"go on"}]} + ]`), + } + + lookupCalled := false + out, err := ResponsesToChatCompletionsRequestWithOptions(req, &ResponsesToChatOptions{ + ReasoningContentByID: func(string) string { + lookupCalled = true + return "cached thinking" + }, + }) + require.NoError(t, err) + require.Len(t, out.Messages, 3) + require.Equal(t, "plain thinking", out.Messages[0].ReasoningContent) + require.False(t, lookupCalled, "plaintext summary present → cache lookup must not run") +} + +func TestExtractResponsesReasoningItem(t *testing.T) { + id, text, ok := ExtractResponsesReasoningItem(json.RawMessage( + `{"type":"reasoning","id":"item_a","summary":[{"type":"summary_text","text":"think"}]}`)) + require.True(t, ok) + require.Equal(t, "item_a", id) + require.Equal(t, "think", text) + + // Encrypted-only item: ok with id but empty text. + id, text, ok = ExtractResponsesReasoningItem(json.RawMessage( + `{"type":"reasoning","id":"item_b","summary":[],"encrypted_content":"opaque"}`)) + require.True(t, ok) + require.Equal(t, "item_b", id) + require.Empty(t, text) + + // Non-reasoning items are skipped. + _, _, ok = ExtractResponsesReasoningItem(json.RawMessage( + `{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}`)) + require.False(t, ok) + + _, _, ok = ExtractResponsesReasoningItem(json.RawMessage(`"bare string"`)) + require.False(t, ok) +} diff --git a/backend/internal/repository/gateway_cache.go b/backend/internal/repository/gateway_cache.go index dc3d4a980a..c05cbded41 100644 --- a/backend/internal/repository/gateway_cache.go +++ b/backend/internal/repository/gateway_cache.go @@ -131,6 +131,48 @@ func (c *gatewayCache) ReleaseGrokVideoBilled(ctx context.Context, key string) e var _ service.CyberSessionBlockStore = (*gatewayCache)(nil) var _ service.LiveCallStore = (*gatewayCache)(nil) +const reasoningContentPrefix = "reasoning_content:" + +// reasoningContentDefaultTTL 是 reasoning 缓存的默认过期时间。Codex 会话可能 +// 跨多天恢复，取 7 天；调用方传入非正 TTL 时兜底。 +const reasoningContentDefaultTTL = 7 * 24 * time.Hour + +// SetReasoningContent 按 reasoning item id 缓存 reasoning 全文。 +// itemID 或 content 为空时直接返回 nil（无可缓存内容，属正常情况而非错误）。 +func (c *gatewayCache) SetReasoningContent(ctx context.Context, itemID string, content string, ttl time.Duration) error { + if c == nil || c.rdb == nil { + return errors.New("gateway cache unavailable") + } + itemID = strings.TrimSpace(itemID) + if itemID == "" || content == "" { + return nil + } + if ttl <= 0 { + ttl = reasoningContentDefaultTTL + } + return c.rdb.Set(ctx, reasoningContentPrefix+itemID, content, ttl).Err() +} + +// GetReasoningContent 返回缓存的 reasoning 全文；未命中返回 +// service.ErrReasoningContentNotFound。 +func (c *gatewayCache) GetReasoningContent(ctx context.Context, itemID string) (string, error) { + if c == nil || c.rdb == nil { + return "", errors.New("gateway cache unavailable") + } + itemID = strings.TrimSpace(itemID) + if itemID == "" { + return "", service.ErrReasoningContentNotFound + } + val, err := c.rdb.Get(ctx, reasoningContentPrefix+itemID).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return "", service.ErrReasoningContentNotFound + } + return "", err + } + return val, nil +} + const cyberSessionBlockPrefix = "cyber_session_block:" // SetCyberSessionBlocked 把被 cyber_policy 命中的会话写入屏蔽表（TTL 自动过期）。 diff --git a/backend/internal/repository/gateway_cache_reasoning_test.go b/backend/internal/repository/gateway_cache_reasoning_test.go new file mode 100644 index 0000000000..f284da51b1 --- /dev/null +++ b/backend/internal/repository/gateway_cache_reasoning_test.go @@ -0,0 +1,41 @@ +package repository + +import ( + "context" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" +) + +func TestGatewayCacheReasoningContent(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + cache := NewGatewayCache(client) + ctx := context.Background() + + // 未命中返回哨兵错误，区别于真实读取失败。 + _, err := cache.GetReasoningContent(ctx, "item_missing") + require.ErrorIs(t, err, service.ErrReasoningContentNotFound) + + // 写入后可读回。 + require.NoError(t, cache.SetReasoningContent(ctx, "item_abc", "think hard", time.Minute)) + got, err := cache.GetReasoningContent(ctx, "item_abc") + require.NoError(t, err) + require.Equal(t, "think hard", got) + + // ttl<=0 时兜底为默认 7 天。 + require.NoError(t, cache.SetReasoningContent(ctx, "item_ttl", "x", 0)) + ttl := mr.TTL(reasoningContentPrefix + "item_ttl") + require.Greater(t, ttl, 6*24*time.Hour) + require.LessOrEqual(t, ttl, reasoningContentDefaultTTL) + + // 空 itemID / 空 content 是 no-op（无可缓存内容不算错误）。 + require.NoError(t, cache.SetReasoningContent(ctx, "", "x", time.Minute)) + require.NoError(t, cache.SetReasoningContent(ctx, "item_empty", "", time.Minute)) + _, err = cache.GetReasoningContent(ctx, "item_empty") + require.ErrorIs(t, err, service.ErrReasoningContentNotFound) +} diff --git a/backend/internal/service/gateway_hotpath_optimization_test.go b/backend/internal/service/gateway_hotpath_optimization_test.go index 48cfc358a6..c8ca1acf07 100644 --- a/backend/internal/service/gateway_hotpath_optimization_test.go +++ b/backend/internal/service/gateway_hotpath_optimization_test.go @@ -158,6 +158,13 @@ func (s *stickyGatewayCacheHotpathStub) ReleaseGrokVideoBilled(_ context.Context return nil } +func (s *stickyGatewayCacheHotpathStub) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (s *stickyGatewayCacheHotpathStub) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + func (s *modelsListAccountRepoStub) ListSchedulableByGroupID(ctx context.Context, groupID int64) ([]Account, error) { s.listByGroupCalls.Add(1) if s.err != nil { diff --git a/backend/internal/service/gateway_multiplatform_test.go b/backend/internal/service/gateway_multiplatform_test.go index ac734a3642..664b6bf08b 100644 --- a/backend/internal/service/gateway_multiplatform_test.go +++ b/backend/internal/service/gateway_multiplatform_test.go @@ -291,6 +291,13 @@ func (m *mockGatewayCacheForPlatform) ReleaseGrokVideoBilled(_ context.Context, return nil } +func (m *mockGatewayCacheForPlatform) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (m *mockGatewayCacheForPlatform) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + type mockGroupRepoForGateway struct { groups map[int64]*Group getByIDCalls int diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 0f3415fab5..bdc028bb46 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -452,6 +452,10 @@ var allowedHeaders = map[string]bool{ // cache implementation (e.g. redis.Nil), mirroring ErrRefreshTokenNotFound. var ErrStickySessionNotFound = errors.New("sticky session not found") +// ErrReasoningContentNotFound is returned by GatewayCache.GetReasoningContent +// when no cached reasoning content exists for the reasoning item ID. +var ErrReasoningContentNotFound = errors.New("reasoning content not found") + // GatewayCache 定义网关服务的缓存操作接口。 // 提供粘性会话（Sticky Session）的存储、查询、刷新和删除功能。 // @@ -486,6 +490,16 @@ type GatewayCache interface { ClaimGrokVideoBilled(ctx context.Context, key string, ttl time.Duration) (bool, error) // ReleaseGrokVideoBilled clears a claim so a failed RecordUsage can retry billing. ReleaseGrokVideoBilled(ctx context.Context, key string) error + + // Reasoning content cache (Responses→Chat Completions 桥接）。 + // SetReasoningContent 按 reasoning item id 缓存 reasoning 全文，供后续请求 + // 在客户端不回传明文 summary 时回注 reasoning_content（DeepSeek thinking + // mode 要求回传，否则 400）。 + SetReasoningContent(ctx context.Context, itemID string, content string, ttl time.Duration) error + // GetReasoningContent 返回缓存的 reasoning 全文；未命中返回 + // ErrReasoningContentNotFound，使 service 层无需依赖具体缓存实现即可 + // 区分"未缓存"与真实读取失败。 + GetReasoningContent(ctx context.Context, itemID string) (string, error) } // derefGroupID safely dereferences *int64 to int64, returning 0 if nil diff --git a/backend/internal/service/gemini_multiplatform_test.go b/backend/internal/service/gemini_multiplatform_test.go index 63ebc929b9..9a89998d51 100644 --- a/backend/internal/service/gemini_multiplatform_test.go +++ b/backend/internal/service/gemini_multiplatform_test.go @@ -319,6 +319,13 @@ func (m *mockGatewayCacheForGemini) ReleaseGrokVideoBilled(_ context.Context, _ return nil } +func (m *mockGatewayCacheForGemini) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (m *mockGatewayCacheForGemini) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + // TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform 测试 Gemini 单平台选择 func TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform(t *testing.T) { ctx := context.Background() diff --git a/backend/internal/service/openai_account_scheduler_test.go b/backend/internal/service/openai_account_scheduler_test.go index 3b3a64856e..cd914da0f8 100644 --- a/backend/internal/service/openai_account_scheduler_test.go +++ b/backend/internal/service/openai_account_scheduler_test.go @@ -196,6 +196,13 @@ func (c *schedulerTestGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ return nil } +func (c *schedulerTestGatewayCache) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (c *schedulerTestGatewayCache) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + func newSchedulerTestOpenAIWSV2Config() *config.Config { cfg := &config.Config{} cfg.Gateway.OpenAIWS.Enabled = true diff --git a/backend/internal/service/openai_cyber_session_block_test.go b/backend/internal/service/openai_cyber_session_block_test.go index 6d6ad8bad4..66ee99fa77 100644 --- a/backend/internal/service/openai_cyber_session_block_test.go +++ b/backend/internal/service/openai_cyber_session_block_test.go @@ -148,6 +148,13 @@ func (c *comboCacheAndStore) ReleaseGrokVideoBilled(_ context.Context, _ string) return nil } +func (c *comboCacheAndStore) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (c *comboCacheAndStore) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + func (c *comboCacheAndStore) SetCyberSessionBlocked(ctx context.Context, key string, ttl time.Duration) error { return c.store.SetCyberSessionBlocked(ctx, key, ttl) } diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback.go b/backend/internal/service/openai_gateway_responses_chat_fallback.go index 1082b020dd..084f9e5020 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback.go @@ -1,6 +1,7 @@ package service import ( + "bytes" "context" "encoding/json" "errors" @@ -52,7 +53,13 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions( toolSearch := apicompat.HasToolSearchTool(effectiveTools) namespaceTools := apicompat.NamespaceToolNames(effectiveTools) - chatReq, err := apicompat.ResponsesToChatCompletionsRequest(&responsesReq) + // 自愈回写：历史里带明文 summary 的 reasoning item 刷新进缓存，覆盖 Redis + // 被 flush / 跨实例漂移后同 id 的 encrypted-only 副本无法再取明文的情况。 + s.recacheReasoningItemsFromInput(responsesReq.Input) + + chatReq, err := apicompat.ResponsesToChatCompletionsRequestWithOptions(&responsesReq, &apicompat.ResponsesToChatOptions{ + ReasoningContentByID: s.reasoningContentByID, + }) if err != nil { writeOpenAIResponsesFallbackError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) return nil, fmt.Errorf("convert responses to chat completions: %w", err) @@ -136,6 +143,7 @@ func (s *OpenAIGatewayService) bufferChatCompletionsAsResponses( return nil, err } responsesResp := apicompat.ChatCompletionsResponseToResponses(ccResp, originalModel, customTools, toolSearch, namespaceTools) + s.cacheReasoningItemsFromOutput(responsesResp.Output) if s.responseHeaderFilter != nil { responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) @@ -204,7 +212,9 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses( } scan := s.scanCCStream(resp, "openai responses chat fallback", requestID, startTime, func(chunk *apicompat.ChatCompletionsChunk) { - writeEvents(apicompat.ChatCompletionsChunkToResponsesEvents(chunk, state)) + events := apicompat.ChatCompletionsChunkToResponsesEvents(chunk, state) + s.cacheReasoningItemsFromEvents(events) + writeEvents(events) }) if scan.Err != nil { @@ -222,7 +232,9 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses( }, fmt.Errorf("stream usage incomplete: %w", scan.Err) } - writeEvents(apicompat.FinalizeChatCompletionsResponsesStream(state)) + finalEvents := apicompat.FinalizeChatCompletionsResponsesStream(state) + s.cacheReasoningItemsFromEvents(finalEvents) + writeEvents(finalEvents) if !clientDisconnected { writeStreamHeaders() if _, err := fmt.Fprint(c.Writer, "data: [DONE]\n\n"); err != nil { @@ -261,3 +273,100 @@ func chatChunkStartsResponsesOutput(chunk *apicompat.ChatCompletionsChunk) bool } return false } + +// responsesReasoningCacheTTL 是 reasoning 缓存（按 reasoning item id）的过期时间。 +// Codex 会话可能跨多天恢复历史，取 7 天。 +const responsesReasoningCacheTTL = 7 * 24 * time.Hour + +// reasoningContentByID 按 reasoning item id 回查缓存的 reasoning 全文，供 +// Responses→CC 桥接在客户端不回传明文 summary（encrypted-only reasoning +// item）时回注 reasoning_content。任何失败都 fail-open 返回 ""（维持桥接原 +// 行为），因为缓存只是优化而非正确性前提。 +func (s *OpenAIGatewayService) reasoningContentByID(itemID string) string { + if s == nil || s.cache == nil { + return "" + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + content, err := s.cache.GetReasoningContent(ctx, itemID) + if err != nil { + return "" + } + return content +} + +// recacheReasoningItemsFromInput 把请求历史里带明文 summary 的 reasoning item +// 重新写入缓存（best-effort）。Codex 多数时候会原样回传明文 summary，借机 +// 刷新 TTL 并自愈 Redis 被 flush / 跨实例漂移造成的缓存缺失。 +func (s *OpenAIGatewayService) recacheReasoningItemsFromInput(inputRaw json.RawMessage) { + if s == nil || s.cache == nil { + return + } + inputRaw = bytes.TrimSpace(inputRaw) + if len(inputRaw) == 0 || inputRaw[0] != '[' { + return + } + var items []json.RawMessage + if err := json.Unmarshal(inputRaw, &items); err != nil { + return + } + for _, raw := range items { + id, text, ok := apicompat.ExtractResponsesReasoningItem(raw) + if !ok || id == "" || text == "" { + continue + } + s.setReasoningContent(id, text) + } +} + +// cacheReasoningItemsFromEvents 从 Responses 流事件里提取完成的 reasoning +// item 写入缓存（覆盖一个流中的多个 reasoning item）。 +func (s *OpenAIGatewayService) cacheReasoningItemsFromEvents(events []apicompat.ResponsesStreamEvent) { + for _, event := range events { + if event.Type != "response.output_item.done" || event.Item == nil { + continue + } + s.cacheReasoningItem(event.Item) + } +} + +// cacheReasoningItemsFromOutput 从非流式 Responses 响应的 output 里提取 +// reasoning item 写入缓存。 +func (s *OpenAIGatewayService) cacheReasoningItemsFromOutput(output []apicompat.ResponsesOutput) { + for i := range output { + s.cacheReasoningItem(&output[i]) + } +} + +func (s *OpenAIGatewayService) cacheReasoningItem(item *apicompat.ResponsesOutput) { + if item == nil || item.Type != "reasoning" || item.ID == "" { + return + } + var parts []string + for _, sum := range item.Summary { + if t := strings.TrimSpace(sum.Text); t != "" { + parts = append(parts, t) + } + } + if len(parts) == 0 { + return + } + s.setReasoningContent(item.ID, strings.Join(parts, "\n")) +} + +// setReasoningContent 写入缓存，使用 detached ctx：客户端断连后仍在 drain +// 上游流（计费需要），此时的 reasoning 也是后续轮次回注所依赖的，不能随 +// 请求 ctx 一起取消。失败仅记日志，不影响转发。 +func (s *OpenAIGatewayService) setReasoningContent(itemID, content string) { + if s == nil || s.cache == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := s.cache.SetReasoningContent(ctx, itemID, content, responsesReasoningCacheTTL); err != nil { + logger.L().Warn("openai responses chat fallback: cache reasoning content failed", + zap.Error(err), + zap.String("item_id", itemID), + ) + } +} diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go index 3682335c1f..15952f3a0f 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go @@ -9,7 +9,9 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" + "time" "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" "github.com/gin-gonic/gin" @@ -227,3 +229,143 @@ func forceChatResponsesFallbackAccount() *Account { } return account } + +// reasoningRecordingCache 记录 reasoning 缓存写入、并按需响应回查。 +type reasoningRecordingCache struct { + stubGatewayCache + mu sync.Mutex + sets map[string]string + getResp map[string]string +} + +func (c *reasoningRecordingCache) SetReasoningContent(_ context.Context, itemID string, content string, _ time.Duration) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.sets == nil { + c.sets = make(map[string]string) + } + c.sets[itemID] = content + return nil +} + +func (c *reasoningRecordingCache) GetReasoningContent(_ context.Context, itemID string) (string, error) { + if v, ok := c.getResp[itemID]; ok { + return v, nil + } + return "", ErrReasoningContentNotFound +} + +func (c *reasoningRecordingCache) snapshotSets() map[string]string { + c.mu.Lock() + defer c.mu.Unlock() + out := make(map[string]string, len(c.sets)) + for k, v := range c.sets { + out[k] = v + } + return out +} + +// 流式响应里的 reasoning_content 应按 reasoning item id 写入缓存，供后续轮次 +// 客户端不回传明文 summary 时回注（DeepSeek thinking mode 400 修复的写入侧）。 +func TestForwardResponses_ChatFallbackCachesStreamedReasoning(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-reasoner","input":"hello","stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_rc","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_rc","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"reasoning_content":"think "},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_rc","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"reasoning_content":"first"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_rc","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"content":"answer"},"finish_reason":"stop"}]}`, + "", + `data: {"id":"chatcmpl_rc","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[],"usage":{"prompt_tokens":4,"completion_tokens":3,"total_tokens":7}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_reasoning_cache_stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + cache := &reasoningRecordingCache{} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + cache: cache, + } + + result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) + require.NoError(t, err) + require.NotNil(t, result) + + sets := cache.snapshotSets() + require.Len(t, sets, 1, "应恰好缓存一个 reasoning item") + for itemID, content := range sets { + require.NotEmpty(t, itemID) + require.Equal(t, "think first", content) + } +} + +// 请求侧：encrypted-only reasoning item（无明文 summary）经缓存回查补回 +// reasoning_content；带明文 summary 的 item 顺手回写缓存（自愈）。 +func TestForwardResponses_ChatFallbackRestoresReasoningFromCache(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{ + "model":"deepseek-reasoner", + "stream":false, + "input":[ + {"type":"reasoning","id":"item_plain","summary":[{"type":"summary_text","text":"plain thinking"}]}, + {"type":"function_call","call_id":"call_0","name":"get_value","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_0","output":"ok"}, + {"type":"reasoning","id":"item_enc1","summary":[],"encrypted_content":"opaque"}, + {"type":"function_call","call_id":"call_1","name":"get_value","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"ok"}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"go on"}]} + ] + }`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_reasoning_cache_restore"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl_restore","object":"chat.completion","model":"deepseek-reasoner","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}`, + )), + }} + cache := &reasoningRecordingCache{ + getResp: map[string]string{"item_enc1": "cached thinking"}, + } + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + cache: cache, + } + + result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) + require.NoError(t, err) + require.NotNil(t, result) + + // 明文 summary 的 assistant 工具调用消息：reasoning_content 来自 summary 本身。 + require.Equal(t, "plain thinking", gjson.GetBytes(upstream.lastBody, "messages.0.reasoning_content").String()) + require.Equal(t, "call_0", gjson.GetBytes(upstream.lastBody, "messages.0.tool_calls.0.id").String()) + require.Equal(t, "tool", gjson.GetBytes(upstream.lastBody, "messages.1.role").String()) + // encrypted-only 的 assistant 工具调用消息：reasoning_content 来自缓存回查。 + require.Equal(t, "cached thinking", gjson.GetBytes(upstream.lastBody, "messages.2.reasoning_content").String()) + require.Equal(t, "call_1", gjson.GetBytes(upstream.lastBody, "messages.2.tool_calls.0.id").String()) + require.Equal(t, "tool", gjson.GetBytes(upstream.lastBody, "messages.3.role").String()) + + // 明文 summary 的 item 被回写进缓存（自愈）。 + require.Equal(t, "plain thinking", cache.snapshotSets()["item_plain"]) +} diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 617c07c066..36e38c659e 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -710,6 +710,13 @@ func (c *stubGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ string) e return nil } +func (c *stubGatewayCache) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (c *stubGatewayCache) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + func TestOpenAISelectAccountWithLoadAwareness_FiltersUnschedulable(t *testing.T) { now := time.Now() resetAt := now.Add(10 * time.Minute) diff --git a/backend/internal/service/openai_ws_state_store_test.go b/backend/internal/service/openai_ws_state_store_test.go index 9c98a1ea85..869e5e0fd6 100644 --- a/backend/internal/service/openai_ws_state_store_test.go +++ b/backend/internal/service/openai_ws_state_store_test.go @@ -207,6 +207,13 @@ func (c *openAIWSStateStoreTimeoutProbeCache) ReleaseGrokVideoBilled(_ context.C return nil } +func (c *openAIWSStateStoreTimeoutProbeCache) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (c *openAIWSStateStoreTimeoutProbeCache) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", ErrReasoningContentNotFound +} + func TestOpenAIWSStateStore_RedisOpsUseShortTimeout(t *testing.T) { probe := &openAIWSStateStoreTimeoutProbeCache{} store := NewOpenAIWSStateStore(probe) diff --git a/backend/internal/testutil/stubs.go b/backend/internal/testutil/stubs.go index 9890ec14a5..339d57d867 100644 --- a/backend/internal/testutil/stubs.go +++ b/backend/internal/testutil/stubs.go @@ -118,6 +118,13 @@ func (c StubGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ string) er return nil } +func (c StubGatewayCache) SetReasoningContent(_ context.Context, _ string, _ string, _ time.Duration) error { + return nil +} +func (c StubGatewayCache) GetReasoningContent(_ context.Context, _ string) (string, error) { + return "", service.ErrReasoningContentNotFound +} + // ============================================================ // StubSessionLimitCache — service.SessionLimitCache 的空实现 // ============================================================