From 1ecb5a760638d56b6d3262bb79b87e8c5706f74c Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:18:31 +0900 Subject: [PATCH] fix(responses): reject unscoped reasoning replay --- src/adapters/openai-chat.ts | 2 +- src/bridge.ts | 4 +- src/images/loop.ts | 2 +- src/responses/reasoning-replay-cache.ts | 19 ++++--- src/server/responses/core.ts | 8 +-- src/web-search/loop.ts | 2 +- tests/bridge-raw-reasoning-hidden.test.ts | 26 +++++++--- tests/bridge-reasoning-replay-batch.test.ts | 14 +++++- tests/deepseek-reasoning-replay-gaps.test.ts | 52 ++++++++++++++++++-- tests/images/loop-reasoning-replay.test.ts | 8 ++- tests/reasoning-replay-scope-source.test.ts | 27 ++++++++++ tests/web-search.test.ts | 8 ++- 12 files changed, 138 insertions(+), 34 deletions(-) create mode 100644 tests/reasoning-replay-scope-source.test.ts diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index 94cddca397..6350d032aa 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -226,7 +226,7 @@ function toolResultImageChatParts(content: string | OcxContentPart[]): unknown[] function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderConfig): unknown[] { const out: unknown[] = []; const { context, options } = parsed; - const replayCacheScope = parsed._clientThreadId ?? "global"; + const replayCacheScope = parsed._clientThreadId; interface PendingToolCall { id: string; name: string } let pendingToolCalls: PendingToolCall[] = []; diff --git a/src/bridge.ts b/src/bridge.ts index 699f5c83e3..1df0aeba5c 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -197,7 +197,7 @@ export function bridgeToResponsesSSE( }; }, ): ReadableStream { - const replayCacheScope = options?.replayCacheScope ?? "global"; + const replayCacheScope = options?.replayCacheScope; const setBeatInterval = options?.timers?.setInterval ?? ((handler: () => void, ms: number) => setInterval(handler, ms)); const clearBeatInterval = options?.timers?.clearInterval ?? ((id: unknown) => clearInterval(id as ReturnType)); // Freeform/custom tools (apply_patch) carry their body in `input`; the model is given a @@ -1366,7 +1366,7 @@ function buildResponseJSONWithBudget( }, ): Record { const responseId = `resp_${uuid()}`; - const replayCacheScope = options?.replayCacheScope ?? "global"; + const replayCacheScope = options?.replayCacheScope; const output: OutputItem[] = []; const budget = options?.translatorBudget; const encoder = new TextEncoder(); diff --git a/src/images/loop.ts b/src/images/loop.ts index 948f422526..07af9e6bbc 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -907,7 +907,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise number) | null = null; const now = (): number => clockForTests?.() ?? Date.now(); -const keyFor = (callId: string, scope: string | undefined): string => - `${scope ?? "global"}\u0000${callId}`; +const keyFor = (callId: string, scope: string): string => `${scope}\u0000${callId}`; /** * Record the raw reasoning text that preceded the given tool call. @@ -44,8 +43,12 @@ const keyFor = (callId: string, scope: string | undefined): string => * id is never read again. */ export function rememberReasoningForCall(callId: string, text: string, scope?: string): void { + // Never fall back to a process-wide namespace. Call ids are supplied by + // clients/providers and are therefore neither unique nor trustworthy; an + // unscoped entry could be recovered by an unrelated request that reuses the + // same id. // Empty provider deltas are absence of new reasoning, not a request to erase a candidate. - if (!callId || typeof text !== "string" || text.length === 0) return; + if (!scope || !callId || typeof text !== "string" || text.length === 0) return; const bytes = Buffer.byteLength(text, "utf8"); // A single entry larger than the whole budget would immediately evict itself. if (bytes > MAX_TOTAL_BYTES) return; @@ -86,7 +89,7 @@ export function rememberReasoningForCall(callId: string, text: string, scope?: s * a failed continuation reuse the same fallback. */ export function peekReasoningForCall(callId: string, scope?: string): string | undefined { - if (!callId) return undefined; + if (!scope || !callId) return undefined; const key = keyFor(callId, scope); const entry = entries.get(key); if (!entry) return undefined; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 3b8904cd0b..5c6c96180a 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2848,7 +2848,7 @@ async function handleResponsesInner( }, 2_000, { translatorBudget, - replayCacheScope: parsed._clientThreadId ?? "global", + replayCacheScope: parsed._clientThreadId, ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -2895,7 +2895,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed._responseModelId ?? parsed.modelId, { translatorBudget, - replayCacheScope: parsed._clientThreadId ?? "global", + replayCacheScope: parsed._clientThreadId, hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, @@ -3554,7 +3554,7 @@ async function handleResponsesInner( () => upstream.abort(), 2_000, { translatorBudget, - replayCacheScope: parsed._clientThreadId ?? "global", + replayCacheScope: parsed._clientThreadId, ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -3612,7 +3612,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed._responseModelId ?? parsed.modelId, { translatorBudget, - replayCacheScope: parsed._clientThreadId ?? "global", + replayCacheScope: parsed._clientThreadId, hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index e7cc19d888..bf5a7e64ce 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -781,7 +781,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise): Promise<{ event?: }); } -const sseOpts = (hide: boolean) => ({ hideThinkingSummary: hide }); +const REPLAY_SCOPE = "hidden-replay-thread"; +const sseOpts = (hide: boolean) => ({ hideThinkingSummary: hide, replayCacheScope: REPLAY_SCOPE }); describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_delta)", () => { beforeEach(() => { @@ -151,8 +152,19 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del { type: "tool_call_end" }, { type: "done" }, ]), "routed/model", undefined, undefined, undefined, undefined, undefined, sseOpts(true))); - expect(peekReasoningForCall("call_1")).toBe("chain of thought"); - expect(peekReasoningForCall("call_other")).toBeUndefined(); + expect(peekReasoningForCall("call_1", REPLAY_SCOPE)).toBe("chain of thought"); + expect(peekReasoningForCall("call_other", REPLAY_SCOPE)).toBeUndefined(); + }); + + test("streamed hidden: an unscoped bridge never writes a global replay entry", async () => { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "reasoning_raw_delta", text: "private reasoning" }, + { type: "tool_call_start", id: "call_unscoped_stream", name: "read_file" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done" }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, { hideThinkingSummary: true })); + expect(peekReasoningForCall("call_unscoped_stream", "global")).toBeUndefined(); }); test("non-streaming hidden: raw reasoning is recorded for the following tool call", () => { @@ -162,8 +174,8 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del { type: "tool_call_delta", arguments: "{}" }, { type: "tool_call_end" }, { type: "done" }, - ], "routed/model", { hideThinkingSummary: true }); - expect(peekReasoningForCall("call_2")).toBe("quiet"); + ], "routed/model", { hideThinkingSummary: true, replayCacheScope: REPLAY_SCOPE }); + expect(peekReasoningForCall("call_2", REPLAY_SCOPE)).toBe("quiet"); }); test("raw reasoning consumed by a text turn is NOT cached for a later tool call", async () => { @@ -175,7 +187,7 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del { type: "tool_call_end" }, { type: "done" }, ]), "routed/model", undefined, undefined, undefined, undefined, undefined, sseOpts(true))); - expect(peekReasoningForCall("call_later")).toBeUndefined(); + expect(peekReasoningForCall("call_later", REPLAY_SCOPE)).toBeUndefined(); }); test("hidden thinking_delta clears raw reasoning pending for a later tool call", async () => { @@ -187,6 +199,6 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del { type: "tool_call_end" }, { type: "done" }, ]), "routed/model", undefined, undefined, undefined, undefined, undefined, sseOpts(true))); - expect(peekReasoningForCall("call_after_thinking")).toBeUndefined(); + expect(peekReasoningForCall("call_after_thinking", REPLAY_SCOPE)).toBeUndefined(); }); }); diff --git a/tests/bridge-reasoning-replay-batch.test.ts b/tests/bridge-reasoning-replay-batch.test.ts index de617128be..714ad9afcd 100644 --- a/tests/bridge-reasoning-replay-batch.test.ts +++ b/tests/bridge-reasoning-replay-batch.test.ts @@ -26,7 +26,7 @@ function batchOutput(events: AdapterEvent[]): Record { }); } -async function streamFrames(events: AdapterEvent[]): Promise { +async function streamFrames(events: AdapterEvent[], replayScope: string | null = SCOPE): Promise { async function* replay(list: AdapterEvent[]): AsyncGenerator { for (const event of list) yield event; } @@ -38,7 +38,7 @@ async function streamFrames(events: AdapterEvent[]): Promise { undefined, undefined, undefined, - { replayCacheScope: SCOPE }, + replayScope === null ? undefined : { replayCacheScope: replayScope }, ).getReader(); const decoder = new TextDecoder(); while (true) { @@ -71,6 +71,11 @@ describe("reasoning replay survives empty text deltas (both wire modes)", () => expect(peekReasoningForCall("call_batch_1", SCOPE)).toBe(REASONING); }); + test("batch: an unscoped bridge never writes a global replay entry", () => { + buildResponseJSON(toolRoundEvents(), "opencode-free/deepseek-v4-flash-free", {}); + expect(peekReasoningForCall("call_batch_1", "global")).toBeUndefined(); + }); + test("batch: real text between reasoning and the tool call clears the cache target", () => { const events = toolRoundEvents(); events[1] = { type: "text_delta", text: "Let me look at the repo first." }; @@ -83,6 +88,11 @@ describe("reasoning replay survives empty text deltas (both wire modes)", () => expect(peekReasoningForCall("call_batch_1", SCOPE)).toBe(REASONING); }); + test("stream: an unscoped bridge never writes a global replay entry", async () => { + await streamFrames(toolRoundEvents(), null); + expect(peekReasoningForCall("call_batch_1", "global")).toBeUndefined(); + }); + test("stream: real text between reasoning and the tool call clears the cache target", async () => { const events = toolRoundEvents(); events[1] = { type: "text_delta", text: "Let me look at the repo first." }; diff --git a/tests/deepseek-reasoning-replay-gaps.test.ts b/tests/deepseek-reasoning-replay-gaps.test.ts index 6b72f5aceb..73226c2d09 100644 --- a/tests/deepseek-reasoning-replay-gaps.test.ts +++ b/tests/deepseek-reasoning-replay-gaps.test.ts @@ -1,13 +1,14 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; +import { buildResponseJSON } from "../src/bridge"; import { parseRequest } from "../src/responses/parser"; import { clearReasoningReplayCacheForTests, - peekReasoningForCall, - rememberReasoningForCall, + peekReasoningForCall as peekReasoningForCallRaw, + rememberReasoningForCall as rememberReasoningForCallRaw, } from "../src/responses/reasoning-replay-cache"; import { routeModel } from "../src/router"; -import type { OcxConfig, OcxParsedRequest } from "../src/types"; +import type { AdapterEvent, OcxConfig, OcxParsedRequest } from "../src/types"; /** * Regression coverage for opencodex issue #950: OpenCode Go DeepSeek V4 Flash @@ -25,6 +26,11 @@ import type { OcxConfig, OcxParsedRequest } from "../src/types"; const MODEL = "opencode-go/deepseek-v4-flash"; const REASONING = "I need to inspect files before answering."; +const REPLAY_SCOPE = "test-thread"; +const rememberReasoningForCall = (callId: string, text: string, scope = REPLAY_SCOPE): void => + rememberReasoningForCallRaw(callId, text, scope); +const peekReasoningForCall = (callId: string, scope = REPLAY_SCOPE): string | undefined => + peekReasoningForCallRaw(callId, scope); function configFor(): OcxConfig { return { @@ -41,8 +47,12 @@ function configFor(): OcxConfig { }; } -function wireFor(input: unknown[]): { messages: Array> } { +function wireFor( + input: unknown[], + replayScope: string | null = REPLAY_SCOPE, +): { messages: Array> } { const parsed = parseRequest({ model: MODEL, input, stream: true }); + if (replayScope !== null) parsed._clientThreadId = replayScope; const route = routeModel(configFor(), parsed.modelId); parsed.modelId = route.modelId; const req = createOpenAIChatAdapter(route.provider).buildRequest(parsed as OcxParsedRequest); @@ -135,6 +145,31 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) expect(retry!["reasoning_content"]).toBe(REASONING); }); + test("cross-request replay isolates threads and rejects an unscoped producer/consumer pair", () => { + rememberReasoningForCallRaw("call_1", "thread alpha reasoning", "thread-a"); + rememberReasoningForCallRaw("call_1", "thread beta reasoning", "thread-b"); + const unscopedProducer: AdapterEvent[] = [ + { type: "reasoning_raw_delta", text: "unrelated private reasoning" }, + { type: "tool_call_start", id: "call_1", name: "read_file" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done" }, + ]; + buildResponseJSON(unscopedProducer, "routed/model", {}); + const input = [ + userMessage(), + { type: "compaction", encrypted_content: "ocx1:c3VtbWFyeQ==" }, + functionCallItem(), + functionCallOutputItem(), + ]; + const alpha = toolCallAssistant(wireFor(input, "thread-a").messages); + const beta = toolCallAssistant(wireFor(input, "thread-b").messages); + const unscoped = toolCallAssistant(wireFor(input, null).messages); + expect(alpha?.reasoning_content).toBe("thread alpha reasoning"); + expect(beta?.reasoning_content).toBe("thread beta reasoning"); + expect(unscoped?.reasoning_content).toBe(" "); + }); + test("GAP D (issue #1193): replay cache MISS on the main assistant path injects a placeholder", () => { // The replay cache is bounded (64 entries / 256 KiB / 1 h TTL) and always // misses on long sessions. DeepSeek thinking mode rejects ANY tool_call @@ -195,6 +230,7 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) // reasoning still replays via preserveReasoningContentModels. const minimaxWire = (input: unknown[]) => { const parsed = parseRequest({ model: "minimax/MiniMax-M3", input, stream: true }); + parsed._clientThreadId = REPLAY_SCOPE; const config: OcxConfig = { port: 10100, defaultProvider: "minimax", @@ -292,7 +328,13 @@ describe("issue #950 — reasoning replay cache bounds", () => { expect(peekReasoningForCall("call_1", "thread-a")).toBe("thread alpha reasoning"); expect(peekReasoningForCall("call_1", "thread-b")).toBe("thread beta reasoning"); // An unscoped read must not see either scoped entry. - expect(peekReasoningForCall("call_1")).toBeUndefined(); + expect(peekReasoningForCallRaw("call_1")).toBeUndefined(); + }); + + test("unscoped entries are rejected instead of sharing a process-wide namespace", () => { + rememberReasoningForCallRaw("call_collision", "private reasoning"); + expect(peekReasoningForCallRaw("call_collision")).toBeUndefined(); + expect(peekReasoningForCallRaw("call_collision", "global")).toBeUndefined(); }); test("entries expire after the TTL", () => { diff --git a/tests/images/loop-reasoning-replay.test.ts b/tests/images/loop-reasoning-replay.test.ts index d4fce160e0..7ddb7c3901 100644 --- a/tests/images/loop-reasoning-replay.test.ts +++ b/tests/images/loop-reasoning-replay.test.ts @@ -86,7 +86,13 @@ const imagePlan = { } as ImageBridgePlan; function makeParsed(): OcxParsedRequest { - return { modelId: "test-model", context: { messages: [], tools: [] }, stream: true, options: {} } as OcxParsedRequest; + return { + modelId: "test-model", + context: { messages: [], tools: [] }, + stream: true, + options: {}, + _clientThreadId: "image-replay-test", + } as OcxParsedRequest; } describe("issue #950 — image-bridge synthetic tool round (raw reasoning)", () => { diff --git a/tests/reasoning-replay-scope-source.test.ts b/tests/reasoning-replay-scope-source.test.ts new file mode 100644 index 0000000000..c68b0d2bd1 --- /dev/null +++ b/tests/reasoning-replay-scope-source.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +const source = (relative: string): string => + readFileSync(join(import.meta.dir, "..", "src", ...relative.split("/")), "utf8"); + +describe("reasoning replay scope propagation", () => { + test("every production bridge call passes only the explicit client thread scope", () => { + const core = source("server/responses/core.ts"); + const images = source("images/loop.ts"); + const webSearch = source("web-search/loop.ts"); + expect(core.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(4); + expect(images.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(1); + expect(webSearch.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(1); + }); + + test("bridge, adapter, and cache contain no process-wide fallback", () => { + const bridge = source("bridge.ts"); + const adapter = source("adapters/openai-chat.ts"); + const cache = source("responses/reasoning-replay-cache.ts"); + expect(bridge.match(/const replayCacheScope = options\?\.replayCacheScope;/g)).toHaveLength(2); + expect(adapter.match(/const replayCacheScope = parsed\._clientThreadId;/g)).toHaveLength(1); + expect(cache).not.toContain('scope ?? "global"'); + expect(`${bridge}\n${adapter}`).not.toContain('replayCacheScope ?? "global"'); + }); +}); diff --git a/tests/web-search.test.ts b/tests/web-search.test.ts index 8050f0c1f6..ae536e04ae 100644 --- a/tests/web-search.test.ts +++ b/tests/web-search.test.ts @@ -1185,8 +1185,10 @@ describe("web-search sidecar native web_search_call emission", () => { async parseResponse() { throw new Error("parseResponse must be unreachable"); }, }; + const parsed = parseRequest({ model: "routed/model", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }); + parsed._clientThreadId = "web-search-raw-replay"; const response = await runWithWebSearch({ - parsed: parseRequest({ model: "routed/model", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }), + parsed, adapter, forwardProvider, hostedTool: { type: "web_search" }, @@ -1382,8 +1384,10 @@ describe("web-search sidecar native web_search_call emission", () => { preserveReasoningContentModels: ["deepseek-v4-flash"], }; + const parsed = parseRequest({ model: "deepseek-v4-flash", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }); + parsed._clientThreadId = "web-search-deepseek-replay"; const response = await runWithWebSearch({ - parsed: parseRequest({ model: "deepseek-v4-flash", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }), + parsed, adapter: createOpenAIChatAdapter(deepseekProvider), forwardProvider, hostedTool: { type: "web_search" },