From 1f4bde181707938043e27a80ee0ef63234fc4adf Mon Sep 17 00:00:00 2001 From: Dizzle <112548150+maxmad64bis@users.noreply.github.com> Date: Fri, 21 Aug 2026 18:58:56 +0200 Subject: [PATCH] fix(sse): guard reasoning-cache write by the same predicate its readers use (#10978) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ⭐5 — Cache de reasoning-replay escrevia em toda resposta com reasoning_content, mesmo quando nenhum read-path jamais consumiria (install sem provider de replay). Guard com requiresReasoningReplay() nos dois write-sites, superset seguro do que os readers checam. Testes cobrindo o predicate isoladamente e o wiring real via handleChatCore. --- docs/routing/REASONING_REPLAY.md | 4 +- open-sse/handlers/chatCore.ts | 25 ++- .../chatCore-reasoning-cache-guard.test.ts | 27 +++ ...atcore-reasoning-cache-write-guard.test.ts | 200 ++++++++++++++++++ 4 files changed, 246 insertions(+), 10 deletions(-) create mode 100644 tests/unit/chatCore-reasoning-cache-guard.test.ts create mode 100644 tests/unit/chatcore-reasoning-cache-write-guard.test.ts diff --git a/docs/routing/REASONING_REPLAY.md b/docs/routing/REASONING_REPLAY.md index c83f1edc5f..46ed738e81 100644 --- a/docs/routing/REASONING_REPLAY.md +++ b/docs/routing/REASONING_REPLAY.md @@ -26,7 +26,8 @@ But typical clients (Cursor, Cline, Roo Code, OpenAI SDK) strip `reasoning_conte ``` Turn N (assistant generates): → response contains reasoning_content + tool_calls - → cacheReasoningFromAssistantMessage() writes (memory + DB), keyed by every tool_call.id + → if requiresReasoningReplay(provider, model): cacheReasoningFromAssistantMessage() + writes (memory + DB), keyed by every tool_call.id → forward response to client (which may or may not retain reasoning) Turn N+1 (client sends follow-up): @@ -157,6 +158,7 @@ The cache exposes two endpoints under `src/app/api/cache/reasoning/route.ts`. Bo - **Cleanup:** `cleanupReasoningCache()` purges expired memory entries and runs `DELETE FROM reasoning_cache WHERE expires_at <= unixepoch('now')`. Health-check workers call this periodically. - **Crash recovery:** After a restart, memory is empty but the DB still holds unexpired entries. The first lookup for a given `tool_call_id` is a DB hit; subsequent lookups are memory hits. - **No reasoning, no cache:** `cacheReasoningFromAssistantMessage` returns `0` when the assistant message has no `reasoning_content` / `reasoning` field, so non-thinking responses cost nothing. +- **Write is gated too:** both call sites in `chatCore.ts` (non-streaming and streaming) only call `cacheReasoningFromAssistantMessage()` when `requiresReasoningReplay(provider, model)` is `true` — the same predicate the read side checks. Installs that never touch a replay provider stop paying for the write, the index update, and the try/catch on every reasoning-bearing response. - **Non-strict providers:** When `requiresReasoningReplay` is `false` and the target format is OpenAI, the translator **strips** any `reasoning_content` field from outgoing messages — OpenAI Chat Completions does not accept it. ## See Also diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 37362c472d..8f11373b1f 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -367,7 +367,10 @@ import { resolveReportedServiceTier as resolveReportedServiceTierFor, type EffectiveServiceTier, } from "./chatCore/serviceTier.ts"; -import { cacheReasoningFromAssistantMessage } from "../services/reasoningCache.ts"; +import { + cacheReasoningFromAssistantMessage, + requiresReasoningReplay, +} from "../services/reasoningCache.ts"; import { sanitizeOpenAITool } from "../services/toolSchemaSanitizer.ts"; import { isCompactResponsesEndpoint } from "../executors/codex.ts"; import { persistCodexChildQuotaResponse } from "../services/codexAccount/index.ts"; @@ -4934,10 +4937,12 @@ export async function handleChatCore({ const msg = firstChoice?.message; const historyMessages = (translatedBody as { messages?: unknown[] } | null | undefined) ?.messages; - cacheReasoningFromAssistantMessage(msg, provider, model, { - scope: reasoningCacheScope, - historyMessages: Array.isArray(historyMessages) ? historyMessages : [], - }); + if (requiresReasoningReplay({ provider, model })) { + cacheReasoningFromAssistantMessage(msg, provider, model, { + scope: reasoningCacheScope, + historyMessages: Array.isArray(historyMessages) ? historyMessages : [], + }); + } } catch { // Cache capture is non-critical — never block the response } @@ -5456,10 +5461,12 @@ export async function handleChatCore({ const msg = choices?.[0]?.message; const historyMessages = (translatedBody as { messages?: unknown[] } | null | undefined) ?.messages; - cacheReasoningFromAssistantMessage(msg, provider, model, { - scope: reasoningCacheScope, - historyMessages: Array.isArray(historyMessages) ? historyMessages : [], - }); + if (requiresReasoningReplay({ provider, model })) { + cacheReasoningFromAssistantMessage(msg, provider, model, { + scope: reasoningCacheScope, + historyMessages: Array.isArray(historyMessages) ? historyMessages : [], + }); + } } catch { // Cache capture is non-critical — never block the stream } diff --git a/tests/unit/chatCore-reasoning-cache-guard.test.ts b/tests/unit/chatCore-reasoning-cache-guard.test.ts new file mode 100644 index 0000000000..b26460f24c --- /dev/null +++ b/tests/unit/chatCore-reasoning-cache-guard.test.ts @@ -0,0 +1,27 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { requiresReasoningReplay } from "../../open-sse/services/reasoningCache.ts"; + +// Regression guard: the reasoning-cache write in chatCore.ts must be gated by the same +// predicate the read sites already use, so an installation whose traffic never touches a +// replay provider stops writing to the cache on every reasoning-bearing response. +test("requiresReasoningReplay recognizes every provider chatCore.ts's write guard must cover", () => { + // Providers explicitly in REASONING_REPLAY_PROVIDERS + assert.equal(requiresReasoningReplay({ provider: "deepseek", model: "deepseek-chat" }), true); + assert.equal(requiresReasoningReplay({ provider: "kimi-coding", model: "kimi-k2" }), true); + assert.equal(requiresReasoningReplay({ provider: "xiaomi-mimo", model: "mimo-v1" }), true); + + // Providers NOT in the set, caught by the model-pattern fallback (the same fallback + // that covers claudeHelper.ts's non-Anthropic Claude-shape read site, site C in the + // design decision above) + assert.equal( + requiresReasoningReplay({ provider: "some-glm-host", model: "glm-4-thinking" }), + true + ); + assert.equal(requiresReasoningReplay({ provider: "zai", model: "kimi-k2" }), true); + + // A provider/model that should NOT trigger a write — proves the guard actually skips + // something (otherwise this write-guard change is a no-op) + assert.equal(requiresReasoningReplay({ provider: "openai", model: "gpt-5.1" }), false); + assert.equal(requiresReasoningReplay({ provider: "anthropic", model: "claude-opus-5" }), false); +}); diff --git a/tests/unit/chatcore-reasoning-cache-write-guard.test.ts b/tests/unit/chatcore-reasoning-cache-write-guard.test.ts new file mode 100644 index 0000000000..d53008e454 --- /dev/null +++ b/tests/unit/chatcore-reasoning-cache-write-guard.test.ts @@ -0,0 +1,200 @@ +// Integration guard for the reasoning-cache write gate. +// The predicate is tested in isolation in chatCore-reasoning-cache-guard.test.ts; this +// file proves handleChatCore's ACTUAL call sites are wired to it, for both the +// non-streaming and streaming response paths, via the cache's own observable side +// effect (no spying on cacheReasoningFromAssistantMessage — same convention as +// tests/unit/chatcore-sanitization.test.ts and +// tests/unit/combo-context-overflow-compression-probe.test.ts: mock fetch, call the +// real handleChatCore, assert real behavior). +// +// Deepseek is also a replay provider (proven by the predicate test), but its wire +// format is openai-responses — a plain openai chat.completion mock would hit +// MALFORMED-200 and never reach the cache write. xiaomi-mimo serves the same +// predicate (REASONING_REPLAY_PROVIDERS member) while staying on the openai wire +// format, so both the non-streaming JSON mock and the streaming chat.completion.chunk +// SSE mock exercise the passthrough path with minimal translation noise. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync( + path.join(os.tmpdir(), "omniroute-chatcore-reasoning-cache-write-guard-") +); +process.env.DATA_DIR = TEST_DATA_DIR; + +const { handleChatCore } = await import("../../open-sse/handlers/chatCore.ts"); +const { lookupReasoning, clearReasoningCacheAll } = + await import("../../open-sse/services/reasoningCache.ts"); +const core = await import("../../src/lib/db/core.ts"); + +function noopLog() { + return { debug() {}, info() {}, warn() {}, error() {} }; +} + +function nonStreamingUpstreamResponse(toolCallId: string, model: string) { + return new Response( + JSON.stringify({ + id: "chatcmpl-reasoning-cache-guard", + object: "chat.completion", + model, + choices: [ + { + index: 0, + message: { + role: "assistant", + content: null, + reasoning_content: "because the guard test says so", + tool_calls: [ + { id: toolCallId, type: "function", function: { name: "noop", arguments: "{}" } }, + ], + }, + finish_reason: "tool_calls", + }, + ], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }), + { status: 200, headers: { "content-type": "application/json" } } + ); +} + +function streamingUpstreamResponse(toolCallId: string) { + const firstChunk = { + id: "chatcmpl-reasoning-cache-stream-guard", + object: "chat.completion.chunk", + model: "probe", + choices: [ + { + index: 0, + delta: { + role: "assistant", + reasoning_content: "because the guard test says so", + tool_calls: [ + { + index: 0, + id: toolCallId, + type: "function", + function: { name: "noop", arguments: "{}" }, + }, + ], + }, + finish_reason: null, + }, + ], + }; + const secondChunk = { + id: "chatcmpl-reasoning-cache-stream-guard", + object: "chat.completion.chunk", + model: "probe", + choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], + }; + const sseBody = + `data: ${JSON.stringify(firstChunk)}\n\n` + + `data: ${JSON.stringify(secondChunk)}\n\n` + + "data: [DONE]\n\n"; + return new Response(sseBody, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +async function invokeChatCoreNonStreaming(provider: string, model: string, toolCallId: string) { + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => nonStreamingUpstreamResponse(toolCallId, model); + try { + const body = { model, messages: [{ role: "user", content: "call the tool" }], stream: false }; + await handleChatCore({ + body, + modelInfo: { provider, model, extendedContext: false }, + credentials: { apiKey: "sk-test", providerSpecificData: {} }, + log: noopLog(), + clientRawRequest: { + endpoint: "/v1/chat/completions", + body, + headers: new Headers({ accept: "application/json" }), + }, + userAgent: "unit-test", + } as never); + } finally { + globalThis.fetch = originalFetch; + } +} + +async function invokeChatCoreStreaming(provider: string, model: string, toolCallId: string) { + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => streamingUpstreamResponse(toolCallId); + try { + const body = { model, messages: [{ role: "user", content: "call the tool" }], stream: true }; + const result = await handleChatCore({ + body, + modelInfo: { provider, model, extendedContext: false }, + credentials: { apiKey: "sk-test", providerSpecificData: {} }, + log: noopLog(), + clientRawRequest: { + endpoint: "/v1/chat/completions", + body, + headers: new Headers({ accept: "text/event-stream" }), + }, + userAgent: "unit-test", + } as never); + // Drain the streaming response to trigger onStreamComplete (the cache write callback fires on flush/close) + if (result.success && result.response?.body) { + const reader = result.response.body.getReader(); + const decoder = new TextDecoder(); + let text = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (value) text += decoder.decode(value as Uint8Array, { stream: true }); + } + await new Promise((resolve) => setImmediate(resolve)); + void text; + } else if (result.success) { + try { + await result.response.text(); + await new Promise((resolve) => setImmediate(resolve)); + } catch {} + } + } finally { + globalThis.fetch = originalFetch; + } +} + +test.after(() => { + try { + core.resetDbInstance(); + } catch {} + try { + clearReasoningCacheAll(); + } catch {} + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("non-streaming: a replay provider (xiaomi-mimo) populates the reasoning cache", async () => { + const id = "tc-reasoning-cache-nonstream-mimo"; + assert.equal(lookupReasoning(id), null); + await invokeChatCoreNonStreaming("xiaomi-mimo", "mimo-v1", id); + assert.equal(lookupReasoning(id), "because the guard test says so"); +}); + +test("non-streaming: a non-replay provider (openai) does NOT populate the reasoning cache", async () => { + const id = "tc-reasoning-cache-nonstream-openai"; + assert.equal(lookupReasoning(id), null); + await invokeChatCoreNonStreaming("openai", "gpt-5.1", id); + assert.equal(lookupReasoning(id), null); +}); + +test("streaming: a replay provider (xiaomi-mimo) populates the reasoning cache", async () => { + const id = "tc-reasoning-cache-stream-mimo"; + assert.equal(lookupReasoning(id), null); + await invokeChatCoreStreaming("xiaomi-mimo", "mimo-v1", id); + assert.equal(lookupReasoning(id), "because the guard test says so"); +}); + +test("streaming: a non-replay provider (openai) does NOT populate the reasoning cache", async () => { + const id = "tc-reasoning-cache-stream-openai"; + assert.equal(lookupReasoning(id), null); + await invokeChatCoreStreaming("openai", "gpt-5.1", id); + assert.equal(lookupReasoning(id), null); +});