diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index 2ec019a6d..d8fbab0f1 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -8,6 +8,7 @@ import { isCyberPolicyCode } from "../lib/errors"; import { redactSecretString } from "../lib/redact"; import { contentPartsToText } from "./image"; import { identifyRoutedModel } from "./identity"; +import { peekReasoningForCall } from "../responses/reasoning-replay-cache"; import { buildNonOpenAIToolCatalogNudgeForTools, shouldInjectNonOpenAIToolCatalogNudge } from "./tool-catalog-nudge"; import { openRouterProviderPayload, resolveOpenRouterRouting } from "../providers/openrouter-routing"; import { @@ -195,6 +196,9 @@ function toolResultImageChatParts(content: string | OcxContentPart[]): unknown[] function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderConfig): unknown[] { const out: unknown[] = []; const { context, options } = parsed; + // Mirror the bridge's replay-cache scope (issue #950): provider call ids are + // not globally unique, so reasoning must not cross conversation boundaries. + const replayCacheScope = parsed._clientThreadId ?? "global"; // 260718 dangling tool_calls hardening (devlog/_plan/260718_dangling_toolcall_hardening): // strict chat providers (Kimi/Moonshot) 400 when an assistant tool_call is not answered @@ -324,7 +328,26 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon if (textParts.length > 0) { chatMsg.content = textParts.map(p => p.text).join(""); } - const reasoningContent = thinkingParts.map(p => p.thinking).join(""); + let reasoningContent = thinkingParts.map(p => p.thinking).join(""); + // History transformations (compaction, lost assistant turn, resumed + // threads) can strip the reasoning item while the tool round survives. + // Re-attach the reasoning the bridge recorded for these call ids so + // preserveReasoningContentModels providers (DeepSeek thinking mode) + // never receive a bare tool-call continuation (issue #950). + if ( + reasoningContent.length === 0 + && toolCalls.length > 0 + && modelInList(provider.preserveReasoningContentModels, parsed.modelId) + ) { + const cached = toolCalls + .map(tc => (tc.id ? peekReasoningForCall(tc.id, replayCacheScope) : undefined)) + .filter((text): text is string => typeof text === "string" && text.length > 0); + // Parallel calls share one preceding reasoning block, which is + // recorded under every call id — join unique texts only. + if (cached.length > 0) { + reasoningContent = [...new Set(cached)].join("\n"); + } + } if (reasoningContent.length > 0 && modelInList(provider.preserveReasoningContentModels, parsed.modelId)) { chatMsg.reasoning_content = reasoningContent; } @@ -382,9 +405,17 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon // role:"tool" message unless an assistant tool_call with the same id immediately precedes it. flushPendingToolCalls(); const name = safeToolName(msg.toolName); + // The orphan repair synthesizes an assistant tool call for a result + // whose assistant turn was lost; carry the recorded reasoning so the + // replayed round stays valid for thinking-mode providers (#950). + const cachedReasoning = + toolCallId && modelInList(provider.preserveReasoningContentModels, parsed.modelId) + ? peekReasoningForCall(toolCallId, replayCacheScope) + : undefined; out.push({ role: "assistant", content: emptyAssistantContent(provider), + ...(cachedReasoning ? { reasoning_content: cachedReasoning } : {}), tool_calls: [{ id: toolCallId, type: "function", diff --git a/src/bridge.ts b/src/bridge.ts index 21fa85118..945c08177 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -2,6 +2,7 @@ import type { AdapterEvent, OcxMessagePhase, OcxProviderContinuationState, OcxUs import { adapterFailureFromMessage, classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, type OcxErrorPayload } from "./lib/errors"; import { encodeCompactionSummary } from "./responses/compaction"; import { encodeReasoningEnvelope, type ReasoningEnvelope } from "./responses/reasoning-envelope"; +import { rememberReasoningForCall } from "./responses/reasoning-replay-cache"; import { resolveStallTimeoutSec } from "./stall-timeout"; import { usageDisplayTotalTokens } from "./usage/totals"; import { @@ -180,6 +181,12 @@ export function bridgeToResponsesSSE( */ onUsage?: (usage: OcxUsage | undefined) => void; translatorBudget?: TranslatorBudget; + /** + * Conversation identity for the reasoning replay cache (issue #950). + * Provider call ids are not globally unique; scoping by thread keeps one + * conversation's reasoning out of another's continuations. + */ + replayCacheScope?: string; /** * Test seam for the wire/stall beat loop. Production omits this and uses the * global timers; injecting here must not change scheduling semantics. @@ -190,6 +197,7 @@ export function bridgeToResponsesSSE( }; }, ): ReadableStream { + const replayCacheScope = options?.replayCacheScope ?? "global"; 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 @@ -426,8 +434,15 @@ export function bridgeToResponsesSSE( // encodeReasoningEnvelope: takeReasoningEnvelope's sig/red guard would drop txt-only. let hiddenRawReasoningText = ""; let hiddenRawReasoningBytes = 0; + // Raw reasoning text flushed most recently, waiting for the tool call it + // preceded. Recorded into the replay cache on tool_call_start so a later + // continuation can re-attach it when history lost the reasoning item + // (issue #950). Kept until new reasoning/text arrives: parallel tool + // calls share the same preceding reasoning block. + let rawReasoningForNextToolCall = ""; const flushHiddenRawReasoning = () => { if (!hiddenRawReasoningText) return; + rawReasoningForNextToolCall = hiddenRawReasoningText; const previousBytes = hiddenRawReasoningBytes; const encrypted = encodeReasoningEnvelope({ txt: hiddenRawReasoningText }); const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" }); @@ -519,6 +534,7 @@ export function bridgeToResponsesSSE( const closeCurrentRawReasoning = () => { if (!currentRawReasoning) return; + rawReasoningForNextToolCall = currentRawReasoning.text; const item = { type: "reasoning", id: currentRawReasoning.itemId, summary: [], content: [{ type: "reasoning_text", text: currentRawReasoning.text }], @@ -779,6 +795,7 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); + rawReasoningForNextToolCall = ""; if (currentToolCall) closeCurrentToolCall(); flushHiddenReasoningEnvelope(); break; @@ -787,6 +804,8 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); + // Reasoning consumed by a text turn, not a tool call: no cache target. + rawReasoningForNextToolCall = ""; if (currentToolCall) closeCurrentToolCall(); // Only flush on an explicit phase change. A later delta that omits `phase` must // keep appending to the current message rather than wiping the earlier phase. @@ -821,6 +840,12 @@ export function bridgeToResponsesSSE( } case "thinking_delta": { if (options?.hideThinkingSummary) { + // The hidden branch returns early, so flush any raw reasoning + // that preceded the thinking block and clear the replay-cache + // candidate — otherwise a stale reasoning_raw_delta would be + // recorded for a LATER tool call (CodeRabbit on #971). + flushHiddenRawReasoning(); + rawReasoningForNextToolCall = ""; ({ value: hiddenThinkingText, bytes: hiddenThinkingBytes } = appendString( hiddenThinkingText, hiddenThinkingBytes, @@ -832,6 +857,7 @@ export function bridgeToResponsesSSE( if (currentMsg) closeCurrentMessage("commentary"); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); + rawReasoningForNextToolCall = ""; if (currentToolCall) closeCurrentToolCall(); if (!currentReasoning) { const itemId = `rs_${uuid()}`; @@ -905,6 +931,9 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); + if (rawReasoningForNextToolCall) { + rememberReasoningForCall(event.id, rawReasoningForNextToolCall, replayCacheScope); + } if (currentToolCall) closeCurrentToolCall(); const mapped = toolNsMap?.get(event.name); const realName = mapped?.name ?? event.name; @@ -1298,9 +1327,12 @@ function buildResponseJSONWithBudget( /** Raw adapter-reported usage before wire normalization (see bridgeToResponsesSSE onUsage). */ onUsage?: (usage: OcxUsage | undefined) => void; translatorBudget?: TranslatorBudget; + /** Conversation identity for the reasoning replay cache (issue #950). */ + replayCacheScope?: string; }, ): Record { const responseId = `resp_${uuid()}`; + const replayCacheScope = options?.replayCacheScope ?? "global"; const output: OutputItem[] = []; const budget = options?.translatorBudget; const encoder = new TextEncoder(); @@ -1356,6 +1388,9 @@ function buildResponseJSONWithBudget( let currentSummaryReasoningBytes = 0; let currentRawReasoning = ""; let currentRawReasoningBytes = 0; + // Same replay-cache handoff as the streaming path (issue #950): the most + // recently flushed raw reasoning waits for the tool call it preceded. + let rawReasoningForNextToolCall = ""; // Anthropic extended-thinking round-trip (batch): see bridgeToResponsesSSE counterpart. let batchSignature: string | undefined; let batchSignatureBytes = 0; @@ -1423,6 +1458,7 @@ function buildResponseJSONWithBudget( }; const flushRawReasoning = () => { if (!currentRawReasoning) return; + rawReasoningForNextToolCall = currentRawReasoning; if (options?.hideThinkingSummary === true) { // Same contract as the streaming path: no visible reasoning, txt-only envelope round-trip. pushOutput({ @@ -1480,6 +1516,7 @@ function buildResponseJSONWithBudget( flushText("commentary"); flushSummaryReasoning(); flushRawReasoning(); + rawReasoningForNextToolCall = ""; flushToolCall(); break; case "text_delta": @@ -1488,6 +1525,7 @@ function buildResponseJSONWithBudget( if (currentText && e.phase !== undefined && currentTextPhase !== e.phase) flushText("commentary"); if (currentSummaryReasoning) flushSummaryReasoning(); if (currentRawReasoning) flushRawReasoning(); + rawReasoningForNextToolCall = ""; if (currentToolCallId) flushToolCall(); // Compaction turns keep the summary out of normal message output (replay dedup — see // bridgeToResponsesSSE); it ships only inside the synthetic compaction item below. @@ -1506,6 +1544,7 @@ function buildResponseJSONWithBudget( case "thinking_delta": if (currentText) flushText("commentary"); if (currentRawReasoning) flushRawReasoning(); + rawReasoningForNextToolCall = ""; if (currentToolCallId) flushToolCall(); { ({ value: currentSummaryReasoning, bytes: currentSummaryReasoningBytes } = appendBatchString( @@ -1542,6 +1581,9 @@ function buildResponseJSONWithBudget( if (currentText) flushText("commentary"); if (currentSummaryReasoning) flushSummaryReasoning(); if (currentRawReasoning) flushRawReasoning(); + if (rawReasoningForNextToolCall) { + rememberReasoningForCall(e.id, rawReasoningForNextToolCall, replayCacheScope); + } flushToolCall(); currentToolCallId = e.id; budget?.openCall(e.id); diff --git a/src/images/loop.ts b/src/images/loop.ts index 535f295df..039e244fb 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -159,6 +159,7 @@ function extractIterationThinking(events: AdapterEvent[]): OcxThinkingContent[] const parts: OcxThinkingContent[] = []; let thinking = ""; let signature: string | undefined; + let rawReasoning = ""; const flushVisible = () => { if (!thinking && !signature) return; @@ -170,19 +171,33 @@ function extractIterationThinking(events: AdapterEvent[]): OcxThinkingContent[] thinking = ""; signature = undefined; }; + const flushRaw = () => { + if (!rawReasoning) return; + parts.push({ type: "thinking", thinking: rawReasoning }); + rawReasoning = ""; + }; for (const e of events) { if (e.type === "thinking_delta") { + flushRaw(); thinking += e.thinking; + } else if (e.type === "reasoning_raw_delta") { + // OpenAI-compatible providers emit raw reasoning instead of signed + // thinking; DeepSeek thinking mode requires it back alongside replayed + // tool_calls (mirrors src/web-search/loop.ts, issue #950). + flushVisible(); + rawReasoning += e.text; } else if (e.type === "thinking_signature") { signature = e.signature; flushVisible(); } else if (e.type === "redacted_thinking") { flushVisible(); + flushRaw(); parts.push({ type: "thinking", thinking: "", redacted: [e.data] }); } } flushVisible(); + flushRaw(); return parts; } @@ -813,6 +828,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise, +): void { + if (pendingReasoning.length === 0 || !callId) return; + for (let i = messages.length - 1; i >= 0; i--) { + const m = messages[i]; + if (m.role !== "assistant") continue; + for (const part of m.content) { + if (part.type === "toolCall" && part.id === callId) { + // Prepend so thinking still precedes tool_use for adapters that require + // that ordering (Anthropic-style replay). + m.content = [...pendingReasoning.map(entry => entry.part), ...m.content]; + return; + } + } + } +} + const REASONING_EFFORTS = new Set(["none", "minimal", "low", "medium", "high", "xhigh", "max"]); export function parseRequest(body: unknown): OcxParsedRequest { @@ -545,8 +573,9 @@ export function parseRequest(body: unknown): OcxParsedRequest { } if (effectiveType === "function_call_output") { - pendingReasoning.length = 0; const output = item as { call_id: string; output?: string | unknown[] }; + attachPendingReasoningToCallOwner(messages, output.call_id, pendingReasoning); + pendingReasoning.length = 0; const toolInfo = findToolById(messages, output.call_id); messages.push({ role: "toolResult", toolCallId: output.call_id, @@ -558,8 +587,9 @@ export function parseRequest(body: unknown): OcxParsedRequest { } if (effectiveType === "custom_tool_call_output") { - pendingReasoning.length = 0; const output = item as { call_id: string; output: string | unknown[] }; + attachPendingReasoningToCallOwner(messages, output.call_id, pendingReasoning); + pendingReasoning.length = 0; const toolInfo = findToolById(messages, output.call_id); messages.push({ role: "toolResult", toolCallId: output.call_id, diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts new file mode 100644 index 000000000..12c585f2c --- /dev/null +++ b/src/responses/reasoning-replay-cache.ts @@ -0,0 +1,105 @@ +/** + * In-process fallback store pairing raw reasoning text with the tool call it + * preceded (issue #950). + * + * DeepSeek thinking mode requires the assistant's original `reasoning_content` + * to be replayed on every continuation of a tool-call turn. The bridge records + * the raw reasoning here when it closes a reasoning block and a tool call + * follows; the openai-chat adapter re-attaches it when a `tool_calls` + * assistant message is about to serialize without thinking parts (compacted + * history, lost assistant turn, orphan-repaired tool results). + * + * Entries are scoped by an optional conversation identity in addition to the + * call id: provider-generated ids like `call_1` are not globally unique, so a + * process-wide key would let one conversation's reasoning bleed into another + * when ids collide (CodeRabbit P1 on #971). + * + * Privacy: entries hold reasoning text in memory only — never logged, + * serialized, or exported. Bounded by entry count, total bytes, and TTL, so a + * long-lived proxy cannot grow without limit. + */ + +const MAX_ENTRIES = 64; +const MAX_TOTAL_BYTES = 256 * 1024; +const TTL_MS = 60 * 60 * 1000; + +interface CacheEntry { + text: string; + bytes: number; + at: number; +} + +const entries = new Map(); +let totalBytes = 0; +let clockForTests: (() => number) | null = null; + +const now = (): number => clockForTests?.() ?? Date.now(); +const keyFor = (callId: string, scope: string | undefined): string => + `${scope ?? "global"}\u0000${callId}`; + +/** + * Record the raw reasoning text that preceded the given tool call. + * + * Expired entries are swept on insert so the TTL bound holds even when a call + * id is never read again. + */ +export function rememberReasoningForCall(callId: string, text: string, scope?: string): void { + if (!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; + const at = now(); + // Delete every due entry first so expired reasoning cannot linger until a + // later peek or capacity eviction. + for (const [key, entry] of entries) { + if (at - entry.at >= TTL_MS) { + entries.delete(key); + totalBytes -= entry.bytes; + } + } + const key = keyFor(callId, scope); + const previous = entries.get(key); + if (previous) totalBytes -= previous.bytes; + entries.set(key, { text, bytes, at }); + totalBytes += bytes; + // Evict oldest-first until both caps hold (never evict the entry just written + // while it is the only one — the loop guards on size > 1). + while ((totalBytes > MAX_TOTAL_BYTES || entries.size > MAX_ENTRIES) && entries.size > 1) { + let oldestKey: string | undefined; + let oldestAt = Infinity; + for (const [candidateKey, entry] of entries) { + if (entry.at < oldestAt) { + oldestAt = entry.at; + oldestKey = candidateKey; + } + } + if (oldestKey === undefined) break; + const evicted = entries.get(oldestKey)!; + totalBytes -= evicted.bytes; + entries.delete(oldestKey); + } +} + +/** + * Read the recorded reasoning for a call id without removing it: retries after + * a failed continuation reuse the same fallback. + */ +export function peekReasoningForCall(callId: string, scope?: string): string | undefined { + if (!callId) return undefined; + const key = keyFor(callId, scope); + const entry = entries.get(key); + if (!entry) return undefined; + if (now() - entry.at >= TTL_MS) { + entries.delete(key); + totalBytes -= entry.bytes; + return undefined; + } + return entry.text; +} + +/** Test-only: reset the cache and optionally pin the clock. */ +export function clearReasoningReplayCacheForTests(clock?: (() => number) | null): void { + entries.clear(); + totalBytes = 0; + clockForTests = clock ?? null; +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index e4573637e..6edfe7d81 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2325,6 +2325,7 @@ async function handleResponsesInner( }, 2_000, { translatorBudget, + replayCacheScope: parsed._clientThreadId ?? "global", ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -2371,6 +2372,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed.modelId, { translatorBudget, + replayCacheScope: parsed._clientThreadId ?? "global", hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, @@ -2808,6 +2810,7 @@ async function handleResponsesInner( () => upstream.abort(), 2_000, { translatorBudget, + replayCacheScope: parsed._clientThreadId ?? "global", ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -2865,6 +2868,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed.modelId, { translatorBudget, + replayCacheScope: parsed._clientThreadId ?? "global", hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index b4a5e2797..d3aafd643 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -653,6 +653,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise): Promise<{ event?: const sseOpts = (hide: boolean) => ({ hideThinkingSummary: hide }); describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_delta)", () => { + beforeEach(() => { + clearReasoningReplayCacheForTests(); + }); + afterEach(() => { + clearReasoningReplayCacheForTests(); + }); + test("streamed hidden: no reasoning_text deltas, envelope-only item, tool calls untouched", async () => { const frames = await collectSse(bridgeToResponsesSSE(replay([ { type: "reasoning_raw_delta", text: "chain " }, @@ -130,4 +141,52 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del const assistant = body.messages.find(m => m.role === "assistant" && m.reasoning_content !== undefined); expect(assistant?.reasoning_content).toBe("replay me"); }); + + test("streamed hidden: raw reasoning is recorded in the replay cache for the following tool call", async () => { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "reasoning_raw_delta", text: "chain " }, + { type: "reasoning_raw_delta", text: "of thought" }, + { type: "tool_call_start", id: "call_1", name: "read_file" }, + { type: "tool_call_delta", arguments: "{\"path\":\"a.txt\"}" }, + { 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(); + }); + + test("non-streaming hidden: raw reasoning is recorded for the following tool call", () => { + buildResponseJSON([ + { type: "reasoning_raw_delta", text: "quiet" }, + { type: "tool_call_start", id: "call_2", name: "read_file" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done" }, + ], "routed/model", { hideThinkingSummary: true }); + expect(peekReasoningForCall("call_2")).toBe("quiet"); + }); + + test("raw reasoning consumed by a text turn is NOT cached for a later tool call", async () => { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "reasoning_raw_delta", text: "for the text" }, + { type: "text_delta", text: "answer" }, + { type: "tool_call_start", id: "call_later", name: "read_file" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done" }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, sseOpts(true))); + expect(peekReasoningForCall("call_later")).toBeUndefined(); + }); + + test("hidden thinking_delta clears raw reasoning pending for a later tool call", async () => { + await collectSse(bridgeToResponsesSSE(replay([ + { type: "reasoning_raw_delta", text: "stale raw" }, + { type: "thinking_delta", thinking: "signed thinking follows" }, + { type: "tool_call_start", id: "call_after_thinking", name: "read_file" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done" }, + ]), "routed/model", undefined, undefined, undefined, undefined, undefined, sseOpts(true))); + expect(peekReasoningForCall("call_after_thinking")).toBeUndefined(); + }); }); diff --git a/tests/deepseek-reasoning-replay-gaps.test.ts b/tests/deepseek-reasoning-replay-gaps.test.ts new file mode 100644 index 000000000..74787adc5 --- /dev/null +++ b/tests/deepseek-reasoning-replay-gaps.test.ts @@ -0,0 +1,229 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; +import { parseRequest } from "../src/responses/parser"; +import { + clearReasoningReplayCacheForTests, + peekReasoningForCall, + rememberReasoningForCall, +} from "../src/responses/reasoning-replay-cache"; +import { routeModel } from "../src/router"; +import type { OcxConfig, OcxParsedRequest } from "../src/types"; + +/** + * Regression coverage for opencodex issue #950: OpenCode Go DeepSeek V4 Flash + * intermittently drops `reasoning_content` on tool-call continuations and the + * upstream rejects the request with HTTP 400 ("The `reasoning_content` in the + * thinking mode must be passed back to the API"). + * + * The invariant: for every assistant message that contains `tool_calls`, the + * openai-chat adapter must serialize a non-empty `reasoning_content` when the + * provider originally returned one for that turn. + * + * Each test exercises one transformation path that used to break the + * invariant; all four were red against the pre-fix code. + */ + +const MODEL = "opencode-go/deepseek-v4-flash"; +const REASONING = "I need to inspect files before answering."; + +function configFor(): OcxConfig { + return { + port: 10100, + defaultProvider: "opencode-go", + providers: { + "opencode-go": { + adapter: "openai-chat", + baseUrl: "https://opencode.ai/zen/go/v1", + apiKey: "key", + models: ["deepseek-v4-flash"], + }, + }, + }; +} + +function wireFor(input: unknown[]): { messages: Array> } { + const parsed = parseRequest({ model: MODEL, input, stream: true }); + const route = routeModel(configFor(), parsed.modelId); + parsed.modelId = route.modelId; + const req = createOpenAIChatAdapter(route.provider).buildRequest(parsed as OcxParsedRequest); + return JSON.parse(req.body as string) as { messages: Array> }; +} + +const userMessage = () => ({ + type: "message", + role: "user", + content: [{ type: "input_text", text: "inspect the repo" }], +}); + +const reasoningItem = () => ({ + type: "reasoning", + id: "rs_1", + summary: [], + content: [{ type: "reasoning_text", text: REASONING }], +}); + +const functionCallItem = () => ({ + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "read_file", + arguments: '{"path":"README.md"}', +}); + +const functionCallOutputItem = () => ({ + type: "function_call_output", + call_id: "call_1", + output: "contents", +}); + +function toolCallAssistant(messages: Array>): Record | undefined { + return messages.find(m => m.role === "assistant" && Array.isArray(m.tool_calls)); +} + +describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire)", () => { + beforeEach(() => { + clearReasoningReplayCacheForTests(); + }); + afterEach(() => { + clearReasoningReplayCacheForTests(); + }); + + test("CONTROL: canonical full-history tool round keeps reasoning_content", () => { + const { messages } = wireFor([userMessage(), reasoningItem(), functionCallItem(), functionCallOutputItem()]); + const assistant = toolCallAssistant(messages); + expect(assistant).toBeDefined(); + expect(assistant!["reasoning_content"]).toBe(REASONING); + }); + + test("GAP A: reasoning item arriving AFTER its function_call is attached to its turn", () => { + // Reconstructed histories (resume/retry/synthetic) may order the reasoning + // item after the call it belongs to. The parser used to clear the pending + // buffer at function_call_output and serialize the turn bare. + const { messages } = wireFor([userMessage(), functionCallItem(), reasoningItem(), functionCallOutputItem()]); + const assistant = toolCallAssistant(messages); + expect(assistant).toBeDefined(); + expect(assistant!["reasoning_content"]).toBe(REASONING); + }); + + test("GAP B: tool round surviving compaction without its reasoning sibling is re-attached from the replay cache", () => { + // Mid-turn/remote compaction drops all Reasoning items while the open tool + // round (function_call + output) can survive in the in-flight input. The + // bridge recorded the reasoning under the call id on the original turn. + rememberReasoningForCall("call_1", REASONING); + const { messages } = wireFor([ + userMessage(), + { type: "compaction", encrypted_content: "ocx1:c3VtbWFyeQ==" }, + functionCallItem(), + functionCallOutputItem(), + ]); + const assistant = toolCallAssistant(messages); + expect(assistant).toBeDefined(); + expect(assistant!["reasoning_content"]).toBe(REASONING); + }); + + test("GAP C: orphan tool result (lost assistant turn) is repaired WITH the recorded reasoning", () => { + // When previous_response_id expansion misses or history loses the assistant + // turn, the adapter's orphan repair synthesizes an assistant tool_call; it + // must carry the reasoning recorded for that call id — and keep carrying it + // on a retry of the same continuation (peek is non-destructive). + rememberReasoningForCall("call_1", REASONING); + const first = toolCallAssistant(wireFor([userMessage(), functionCallOutputItem()]).messages); + const retry = toolCallAssistant(wireFor([userMessage(), functionCallOutputItem()]).messages); + expect(first).toBeDefined(); + expect(first!["reasoning_content"]).toBe(REASONING); + expect(retry).toBeDefined(); + expect(retry!["reasoning_content"]).toBe(REASONING); + }); + + test("documented non-bug: opaque encrypted-only reasoning is intentionally not replayed", () => { + // Native (non-ocxr1) encrypted reasoning has no readable text; the parser + // deliberately degrades instead of inventing replayable plaintext. Not a + // candidate for the opencode-go path (its reasoning is plaintext/ocxr1). + const { messages } = wireFor([ + userMessage(), + { type: "reasoning", id: "rs_1", encrypted_content: "some-opaque-blob" }, + functionCallItem(), + functionCallOutputItem(), + ]); + const assistant = toolCallAssistant(messages); + expect(assistant).toBeDefined(); + expect(assistant!["reasoning_content"]).toBeUndefined(); + }); +}); + +describe("issue #950 — reasoning replay cache bounds", () => { + beforeEach(() => { + clearReasoningReplayCacheForTests(); + }); + afterEach(() => { + clearReasoningReplayCacheForTests(); + }); + + test("recorded reasoning is readable under the same call id and does not leak across ids", () => { + rememberReasoningForCall("call_a", "alpha reasoning"); + rememberReasoningForCall("call_b", "beta reasoning"); + expect(peekReasoningForCall("call_a")).toBe("alpha reasoning"); + expect(peekReasoningForCall("call_b")).toBe("beta reasoning"); + expect(peekReasoningForCall("call_c")).toBeUndefined(); + }); + + test("conversation scopes isolate entries with the same call id", () => { + rememberReasoningForCall("call_1", "thread alpha reasoning", "thread-a"); + rememberReasoningForCall("call_1", "thread beta reasoning", "thread-b"); + 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(); + }); + + test("entries expire after the TTL", () => { + let clock = 1_000; + clearReasoningReplayCacheForTests(() => clock); + rememberReasoningForCall("call_ttl", "stale reasoning"); + expect(peekReasoningForCall("call_ttl")).toBe("stale reasoning"); + clock += 60 * 60 * 1000 + 1; + expect(peekReasoningForCall("call_ttl")).toBeUndefined(); + clearReasoningReplayCacheForTests(); + }); + + test("expired entries are swept on the next remember without a peek", () => { + let clock = 1_000; + clearReasoningReplayCacheForTests(() => clock); + rememberReasoningForCall("call_stale", "old reasoning"); + clock += 60 * 60 * 1000 + 1; + rememberReasoningForCall("call_fresh", "new reasoning"); + expect(peekReasoningForCall("call_stale")).toBeUndefined(); + expect(peekReasoningForCall("call_fresh")).toBe("new reasoning"); + clearReasoningReplayCacheForTests(); + }); + + test("older entries are evicted when the entry cap is exceeded", () => { + for (let i = 0; i < 70; i++) rememberReasoningForCall(`call_${i}`, `reasoning ${i}`); + const oldest = peekReasoningForCall("call_0"); + // Exactly 70 - 64 = 6 entries must be evicted: call_5 is the last evicted + // entry and call_6 must survive — proving MAX_ENTRIES is 64, not larger. + expect(oldest).toBeUndefined(); + expect(peekReasoningForCall("call_5")).toBeUndefined(); + expect(peekReasoningForCall("call_6")).toBe("reasoning 6"); + expect(peekReasoningForCall("call_63")).toBe("reasoning 63"); + expect(peekReasoningForCall("call_69")).toBe("reasoning 69"); + }); + + test("oldest valid entries are evicted when their combined size exceeds 256 KiB", () => { + const chunk = "x".repeat(65 * 1024); + for (let i = 0; i < 4; i++) rememberReasoningForCall(`call_bytes_${i}`, chunk); + // 4 x 65 KiB = 260 KiB > 256 KiB: exactly the oldest entry is evicted. + expect(peekReasoningForCall("call_bytes_0")).toBeUndefined(); + expect(peekReasoningForCall("call_bytes_1")).toBe(chunk); + expect(peekReasoningForCall("call_bytes_3")).toBe(chunk); + }); + + test("empty and oversized entries are ignored", () => { + rememberReasoningForCall("", "no id"); + rememberReasoningForCall("call_empty", ""); + expect(peekReasoningForCall("call_empty")).toBeUndefined(); + const huge = "x".repeat(300 * 1024); + rememberReasoningForCall("call_huge", huge); + expect(peekReasoningForCall("call_huge")).toBeUndefined(); + }); +}); diff --git a/tests/images/loop-reasoning-replay.test.ts b/tests/images/loop-reasoning-replay.test.ts new file mode 100644 index 000000000..d4fce160e --- /dev/null +++ b/tests/images/loop-reasoning-replay.test.ts @@ -0,0 +1,123 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, mock, test } from "bun:test"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import type { ProviderAdapter } from "../../src/adapters/base"; +import type { AdapterEvent, OcxParsedRequest } from "../../src/types"; +import type { ImageBridgePlan } from "../../src/images/types"; +import type { ImageBridgeDeps } from "../../src/images/loop"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; + +/** + * Issue #950 regression: the image bridge's synthetic tool round must preserve + * raw reasoning (`reasoning_raw_delta`) ahead of the replayed image_gen call, + * exactly like the web-search loop does (#688). Without it, DeepSeek thinking + * mode receives a bare tool-call continuation and rejects it with HTTP 400. + */ + +const REASONING = "I need to inspect files before answering."; + +const PREV_HOME = process.env.OPENCODEX_HOME; +let runWithImageBridgeProduction: typeof import("../../src/images/loop")["runWithImageBridge"]; +let fulfillResult: import("../../src/images/types").ImageCallResult = { + ok: true, model: "grok-imagine-image-quality", prompt: "a cat", + files: ["/test/img.png"], count: 1, markdown: "![image](/test/img.png)", +}; + +beforeAll(async () => { + process.env.OPENCODEX_HOME = join(tmpdir(), "ocx-test-" + randomUUID()); + mock.restore(); + mock.module("../../src/web-search/progress-stream", () => ({ + parseStreamWithProgress: async function* (_resp: Response, parse: (r: Response) => AsyncGenerator, _opts: unknown) { + for await (const e of parse(_resp)) yield e; + }, + RoutedModelInactivityError: class extends Error { readonly timeoutMs = 0; }, + WebSearchStreamProtocolError: class extends Error { /* */ }, + })); + mock.module("../../src/images/fulfill", () => ({ + fulfillImageCall: async (): Promise => fulfillResult, + })); + ({ runWithImageBridge: runWithImageBridgeProduction } = await import("../../src/images/loop")); +}); + +afterAll(() => { + if (PREV_HOME === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = PREV_HOME; + mock.restore(); +}); + +let streamQueue: AdapterEvent[][] = []; + +beforeEach(() => { + fulfillResult = { + ok: true, model: "grok-imagine-image-quality", prompt: "a cat", + files: ["/test/img.png"], count: 1, markdown: "![image](/test/img.png)", + }; + streamQueue = []; +}); + +function runWithImageBridge( + deps: Omit & { incomingMeta?: ImageBridgeDeps["incomingMeta"] }, +): Promise { + return runWithImageBridgeProduction({ + ...deps, + incomingMeta: deps.incomingMeta ?? { + headers: new Headers(), + translatorBudget: createTestTranslatorBudget(), + }, + }); +} + +const mockAdapter: ProviderAdapter = { + name: "test", + buildRequest: async () => ({ url: "https://test/v1/chat", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("{}", { status: 200, headers: { "content-type": "application/json" } }), + parseStream: async function* (): AsyncGenerator { + const events = streamQueue.shift(); + if (events) for (const e of events) yield e; + }, +}; + +const imagePlan = { + provider: {} as never, + auth: { baseUrl: "https://api.x.ai", token: "test-token" }, + model: "grok-imagine-image-quality", + toolNames: new Set(["image_gen"]), +} as ImageBridgePlan; + +function makeParsed(): OcxParsedRequest { + return { modelId: "test-model", context: { messages: [], tools: [] }, stream: true, options: {} } as OcxParsedRequest; +} + +describe("issue #950 — image-bridge synthetic tool round (raw reasoning)", () => { + test("GAP D: raw reasoning preceding an image_gen call survives into the replayed assistant turn", async () => { + const seenAssistants: Array> = []; + const capturingAdapter: ProviderAdapter = { + ...mockAdapter, + buildRequest: async (parsed) => { + const assistant = parsed.context.messages.find(m => m.role === "assistant"); + if (assistant && Array.isArray(assistant.content)) { + seenAssistants.push(assistant.content as Array<{ type?: string; thinking?: string }>); + } + return { url: "https://test/v1/chat", method: "POST", headers: {}, body: "{}" }; + }, + }; + streamQueue = [ + [ + { type: "reasoning_raw_delta", text: REASONING }, + { type: "tool_call_start", id: "call_1", name: "image_gen" }, + { type: "tool_call_delta", arguments: '{"prompt":"a cat"}' }, + { type: "tool_call_end" }, + { type: "done" }, + ], + [{ type: "text_delta", text: "ready" }, { type: "done" }], + ]; + const response = await runWithImageBridge({ + parsed: makeParsed(), adapter: capturingAdapter, plan: imagePlan, maxRounds: 1, + }); + await response.text(); + expect(seenAssistants.length).toBeGreaterThan(0); + expect(seenAssistants[0]!.map(p => p.type)).toEqual(["thinking", "toolCall"]); + expect(seenAssistants[0]!.find(p => p.type === "thinking")).toMatchObject({ thinking: REASONING }); + }); +});