From cb683abd3efbabbe9f551b63678ff0e16ff6675d Mon Sep 17 00:00:00 2001 From: Agent59353 Date: Thu, 6 Aug 2026 19:29:20 +0800 Subject: [PATCH 1/4] fix(responses): make reasoning replay restart-safe and observable (#950) DeepSeek thinking mode requires the assistant's original reasoning_content on every tool-call continuation. The #971 replay cache is in-memory only, so a proxy restart mid-round still loses recovery, and there is no privacy-safe way to see when a bare tool-call continuation is about to be serialized. - Opt-in disk spill (OPENCODEX_REASONING_REPLAY_PERSIST=1, optional OPENCODEX_REASONING_REPLAY_FILE override): bounded, TTL'd, atomically written with best-effort 0600 perms; rehydrated at boot. Default stays memory-only. - Privacy-safe diagnostics: getReasoningReplayStats() exposes counters and bounds only; recordBareToolCallSerialization() counts the exact 400 shape per model; openai-chat logs a throttled counter line (never reasoning text) when a bare tool-call continuation is serialized for a preserveReasoningContentModels provider. - Regression tests: restart round-trip, TTL filter on reload, corrupt-file tolerance, entry-cap on reload, stats privacy, and the wire-level bare serialization counter. --- src/adapters/openai-chat.ts | 31 +++- src/responses/reasoning-replay-cache.ts | 205 +++++++++++++++++++++- tests/reasoning-replay-robustness.test.ts | 179 +++++++++++++++++++ 3 files changed, 409 insertions(+), 6 deletions(-) create mode 100644 tests/reasoning-replay-robustness.test.ts diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index 577d268d2b..7fe7f3c158 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -8,7 +8,11 @@ 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 { + getReasoningReplayStats, + peekReasoningForCall, + recordBareToolCallSerialization, +} from "../responses/reasoning-replay-cache"; import { buildNonOpenAIToolCatalogNudgeForTools, shouldInjectNonOpenAIToolCatalogNudge } from "./tool-catalog-nudge"; import { openRouterProviderPayload, resolveOpenRouterRouting } from "../providers/openrouter-routing"; import { @@ -193,6 +197,22 @@ function toolResultImageChatParts(content: string | OcxContentPart[]): unknown[] return parts; } +// #950 diagnostics: a bare tool-call continuation for a preserveReasoningContentModels +// provider is the exact 400 shape this fix eliminates. Count every occurrence +// (privacy-safe) and surface a throttled counter line — never reasoning text. +let lastBareToolCallWarnAt = 0; +const BARE_TOOL_CALL_WARN_MIN_INTERVAL_MS = 60_000; +function noteBareToolCallSerialization(modelId: string): void { + recordBareToolCallSerialization(modelId); + const at = Date.now(); + if (at - lastBareToolCallWarnAt < BARE_TOOL_CALL_WARN_MIN_INTERVAL_MS) return; + lastBareToolCallWarnAt = at; + const stats = getReasoningReplayStats(); + console.warn( + `[opencodex] reasoning replay miss: bare tool-call continuation for preserveReasoningContentModels model="${modelId}" (cache hits=${stats.hits}, misses=${stats.misses}); reasoning text is never logged`, + ); +} + function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderConfig): unknown[] { const out: unknown[] = []; const { context, options } = parsed; @@ -346,6 +366,8 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon // recorded under every call id — join unique texts only. if (cached.length > 0) { reasoningContent = [...new Set(cached)].join("\n"); + } else { + noteBareToolCallSerialization(parsed.modelId); } } if (reasoningContent.length > 0 && modelInList(provider.preserveReasoningContentModels, parsed.modelId)) { @@ -412,6 +434,13 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon toolCallId && modelInList(provider.preserveReasoningContentModels, parsed.modelId) ? peekReasoningForCall(toolCallId, replayCacheScope) : undefined; + if ( + !cachedReasoning + && toolCallId + && modelInList(provider.preserveReasoningContentModels, parsed.modelId) + ) { + noteBareToolCallSerialization(parsed.modelId); + } out.push({ role: "assistant", content: emptyAssistantContent(provider), diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index 12c585f2cb..df9bb96ef6 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -14,14 +14,26 @@ * 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. + * Privacy: entries hold reasoning text. By default they live in memory only — + * never logged or exported. Operators may opt into a bounded, TTL'd disk spill + * (OPENCODEX_REASONING_REPLAY_PERSIST=1, optional + * OPENCODEX_REASONING_REPLAY_FILE override) so a proxy restart mid-round can + * still replay a call's reasoning; the spill file is written atomically with + * best-effort 0600 permissions. Diagnostics expose counters only, never + * reasoning text. The store is bounded by entry count, total bytes, and TTL, + * so a long-lived proxy cannot grow without limit. */ +import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { dirname, join } from "node:path"; + const MAX_ENTRIES = 64; const MAX_TOTAL_BYTES = 256 * 1024; const TTL_MS = 60 * 60 * 1000; +const PERSIST_DEBOUNCE_MS = 750; +const PERSIST_ENV = "OPENCODEX_REASONING_REPLAY_PERSIST"; +const PERSIST_FILE_ENV = "OPENCODEX_REASONING_REPLAY_FILE"; interface CacheEntry { text: string; @@ -29,14 +41,56 @@ interface CacheEntry { at: number; } +interface PersistedEntry { + scope: string; + callId: string; + text: string; + at: number; +} + +interface PersistFile { + v: 1; + savedAt: number; + entries: PersistedEntry[]; +} + const entries = new Map(); let totalBytes = 0; let clockForTests: (() => number) | null = null; +let hits = 0; +let misses = 0; +const bareSerializationsByModel = new Map(); +let persistEnabled = persistFlagFromEnv(); +let persistPath = persistPathFromEnv(); +let persistTimer: ReturnType | undefined; +let persistWrites = 0; +let persistLastError: string | undefined; const now = (): number => clockForTests?.() ?? Date.now(); const keyFor = (callId: string, scope: string | undefined): string => `${scope ?? "global"}\u0000${callId}`; +function persistFlagFromEnv(): boolean { + const raw = process.env[PERSIST_ENV]?.trim().toLowerCase(); + return raw === "1" || raw === "true" || raw === "yes" || raw === "on"; +} + +/** Mirror config.getConfigDir() resolution (OPENCODEX_HOME or ~/.opencodex) without importing config. */ +function defaultPersistPath(): string { + const raw = process.env.OPENCODEX_HOME?.trim(); + let base: string; + if (!raw) base = join(homedir(), ".opencodex"); + else if (raw === "~") base = homedir(); + else if (raw.startsWith("~/") || raw.startsWith("~\\")) base = join(homedir(), raw.slice(2)); + else base = raw; + return join(base, "reasoning-replay-cache.json"); +} + +function persistPathFromEnv(): string { + const raw = process.env[PERSIST_FILE_ENV]?.trim(); + return raw && raw.length > 0 ? raw : defaultPersistPath(); +} + /** * Record the raw reasoning text that preceded the given tool call. * @@ -44,11 +98,14 @@ const keyFor = (callId: string, scope: string | undefined): string => * id is never read again. */ export function rememberReasoningForCall(callId: string, text: string, scope?: string): void { + rememberReasoningAt(callId, text, scope, now()); +} + +function rememberReasoningAt(callId: string, text: string, scope: string | undefined, at: number): 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) { @@ -78,6 +135,7 @@ export function rememberReasoningForCall(callId: string, text: string, scope?: s totalBytes -= evicted.bytes; entries.delete(oldestKey); } + if (persistEnabled) schedulePersist(); } /** @@ -88,18 +146,155 @@ export function peekReasoningForCall(callId: string, scope?: string): string | u if (!callId) return undefined; const key = keyFor(callId, scope); const entry = entries.get(key); - if (!entry) return undefined; + if (!entry) { + misses += 1; + return undefined; + } if (now() - entry.at >= TTL_MS) { entries.delete(key); totalBytes -= entry.bytes; + misses += 1; return undefined; } + hits += 1; return entry.text; } +// ── Opt-in disk spill (issue #950 robustness: survive proxy restarts) ──────── + +function schedulePersist(): void { + if (!persistEnabled) return; + if (persistTimer !== undefined) clearTimeout(persistTimer); + persistTimer = setTimeout(() => { + persistTimer = undefined; + writePersisted(); + }, PERSIST_DEBOUNCE_MS); +} + +/** Synchronously flush the opt-in spill file (debounced writes call this too). */ +export function flushReasoningReplayCache(): void { + if (persistTimer !== undefined) { + clearTimeout(persistTimer); + persistTimer = undefined; + } + writePersisted(); +} + +function writePersisted(): void { + if (!persistEnabled) return; + try { + const payload: PersistFile = { + v: 1, + savedAt: now(), + entries: [...entries.entries()].map(([key, entry]) => { + const sep = key.indexOf("\u0000"); + return { + scope: sep === -1 ? "global" : key.slice(0, sep), + callId: sep === -1 ? key : key.slice(sep + 1), + text: entry.text, + at: entry.at, + }; + }), + }; + try { + mkdirSync(dirname(persistPath), { recursive: true }); + } catch { /* best-effort */ } + const tmpPath = `${persistPath}.tmp`; + writeFileSync(tmpPath, JSON.stringify(payload), "utf8"); + renameSync(tmpPath, persistPath); + try { chmodSync(persistPath, 0o600); } catch { /* best-effort on platforms that ignore chmod */ } + persistWrites += 1; + persistLastError = undefined; + } catch (err) { + persistLastError = err instanceof Error ? err.message : String(err); + } +} + +function loadPersisted(): void { + if (!persistEnabled || !existsSync(persistPath)) return; + let raw: string; + try { + raw = readFileSync(persistPath, "utf8"); + } catch { + return; // unreadable spill is not worth failing the proxy over + } + let data: Partial; + try { + data = JSON.parse(raw) as Partial; + } catch { + return; // corrupt spill file: treat as empty, overwrite on next flush + } + if (data?.v !== 1 || !Array.isArray(data.entries)) return; + const at = now(); + for (const entry of data.entries) { + if (!entry || typeof entry.callId !== "string" || typeof entry.text !== "string" || entry.text.length === 0) continue; + const entryAt = typeof entry.at === "number" && Number.isFinite(entry.at) ? entry.at : at; + if (at - entryAt >= TTL_MS) continue; + rememberReasoningAt(entry.callId, entry.text, typeof entry.scope === "string" ? entry.scope : undefined, entryAt); + } +} + +// ── Privacy-safe diagnostics (counters only, never reasoning text) ─────────── + +/** Count a bare tool-call continuation serialized for a preserveReasoningContentModels model (#950). */ +export function recordBareToolCallSerialization(modelId: string): void { + if (!modelId) return; + bareSerializationsByModel.set(modelId, (bareSerializationsByModel.get(modelId) ?? 0) + 1); +} + +export interface ReasoningReplayStats { + entries: number; + totalBytes: number; + hits: number; + misses: number; + bareSerializationsByModel: Record; + persistence: { + enabled: boolean; + path: string; + writes: number; + lastError?: string; + }; +} + +/** Counters and bounds only — never reasoning text (issue #950 privacy checklist). */ +export function getReasoningReplayStats(): ReasoningReplayStats { + return { + entries: entries.size, + totalBytes, + hits, + misses, + bareSerializationsByModel: Object.fromEntries(bareSerializationsByModel), + persistence: { + enabled: persistEnabled, + path: persistPath, + writes: persistWrites, + ...(persistLastError !== undefined ? { lastError: persistLastError } : {}), + }, + }; +} + +// Boot-time rehydration when persistence is opted in. +if (persistEnabled) loadPersisted(); + +// ── Test seams ──────────────────────────────────────────────────────────────── + /** Test-only: reset the cache and optionally pin the clock. */ export function clearReasoningReplayCacheForTests(clock?: (() => number) | null): void { entries.clear(); totalBytes = 0; + hits = 0; + misses = 0; + bareSerializationsByModel.clear(); clockForTests = clock ?? null; } + +/** Test-only: toggle the opt-in spill (loads the file when enabling). */ +export function setReasoningReplayPersistenceForTests(enabled: boolean, path?: string): void { + if (persistTimer !== undefined) { + clearTimeout(persistTimer); + persistTimer = undefined; + } + persistEnabled = enabled; + if (path !== undefined && path.length > 0) persistPath = path; + if (enabled) loadPersisted(); +} diff --git a/tests/reasoning-replay-robustness.test.ts b/tests/reasoning-replay-robustness.test.ts new file mode 100644 index 0000000000..a9851269d3 --- /dev/null +++ b/tests/reasoning-replay-robustness.test.ts @@ -0,0 +1,179 @@ +import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; +import { parseRequest } from "../src/responses/parser"; +import { + clearReasoningReplayCacheForTests, + flushReasoningReplayCache, + getReasoningReplayStats, + peekReasoningForCall, + recordBareToolCallSerialization, + rememberReasoningForCall, + setReasoningReplayPersistenceForTests, +} from "../src/responses/reasoning-replay-cache"; +import { routeModel } from "../src/router"; +import type { OcxConfig, OcxParsedRequest } from "../src/types"; + +/** + * Robustness + privacy coverage for issue #950's enhancement checklist: + * restart-safe (opt-in disk spill), counter-only diagnostics, and the + * bare tool-call serialization invariant. + */ + +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 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", +}); + +let spillDir = ""; +let spillFile = ""; + +beforeAll(() => { + spillDir = mkdtempSync(join(tmpdir(), "ocx-reasoning-replay-")); + spillFile = join(spillDir, "reasoning-replay-cache.json"); +}); + +afterAll(() => { + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(); + rmSync(spillDir, { recursive: true, force: true }); +}); + +afterEach(() => { + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(); + rmSync(spillFile, { force: true }); +}); + +describe("reasoning replay — opt-in disk spill", () => { + test("round-trips reasoning across a simulated restart (memory cleared, file reloaded)", () => { + setReasoningReplayPersistenceForTests(true, spillFile); + rememberReasoningForCall("call_restart", REASONING, "thread-1"); + flushReasoningReplayCache(); + + // Simulate a proxy restart: memory wiped, persistence re-enabled at boot. + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(); + setReasoningReplayPersistenceForTests(true, spillFile); + + expect(peekReasoningForCall("call_restart", "thread-1")).toBe(REASONING); + }); + + test("expired entries are not restored from the spill file", () => { + let clock = 0; + clearReasoningReplayCacheForTests(() => clock); + setReasoningReplayPersistenceForTests(true, spillFile); + rememberReasoningForCall("call_expired", REASONING, "thread-2"); + flushReasoningReplayCache(); + + clock = 61 * 60 * 1000; // past the 60-minute TTL + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(() => clock); + setReasoningReplayPersistenceForTests(true, spillFile); + + expect(peekReasoningForCall("call_expired", "thread-2")).toBeUndefined(); + expect(getReasoningReplayStats().entries).toBe(0); + }); + + test("corrupt or unreadable spill file loads as an empty cache without throwing", () => { + writeFileSync(spillFile, "{not json!!", "utf8"); + setReasoningReplayPersistenceForTests(true, spillFile); + expect(getReasoningReplayStats().entries).toBe(0); + expect(peekReasoningForCall("anything", "thread-3")).toBeUndefined(); + }); + + test("reload respects the entry-count cap", () => { + setReasoningReplayPersistenceForTests(true, spillFile); + for (let i = 0; i < 80; i++) { + rememberReasoningForCall(`call_${i}`, "x", "thread-4"); + } + flushReasoningReplayCache(); + + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(); + setReasoningReplayPersistenceForTests(true, spillFile); + + expect(getReasoningReplayStats().entries).toBeLessThanOrEqual(64); + }); +}); + +describe("reasoning replay — privacy-safe diagnostics", () => { + test("stats expose counters and bounds, never reasoning text", () => { + clearReasoningReplayCacheForTests(); + rememberReasoningForCall("call_stats", REASONING, "thread-5"); + expect(peekReasoningForCall("call_stats", "thread-5")).toBe(REASONING); + expect(peekReasoningForCall("call_missing", "thread-5")).toBeUndefined(); + recordBareToolCallSerialization("deepseek-v4-flash"); + recordBareToolCallSerialization("deepseek-v4-flash"); + + const stats = getReasoningReplayStats(); + expect(stats.hits).toBe(1); + expect(stats.misses).toBe(1); + expect(stats.bareSerializationsByModel["deepseek-v4-flash"]).toBe(2); + expect(JSON.stringify(stats)).not.toContain(REASONING); + }); + + test("persistence stays off by default — no spill file is created", () => { + const untouched = join(spillDir, "must-not-exist.json"); + setReasoningReplayPersistenceForTests(false, untouched); + clearReasoningReplayCacheForTests(); + rememberReasoningForCall("call_default", REASONING, "thread-6"); + flushReasoningReplayCache(); + expect(getReasoningReplayStats().persistence.enabled).toBe(false); + expect(existsSync(untouched)).toBe(false); + }); + + test("bare tool-call continuation increments the invariant counter (wire-level)", () => { + clearReasoningReplayCacheForTests(); + const { messages } = wireFor([userMessage(), functionCallItem(), functionCallOutputItem()]); + const assistant = messages.find(m => m.role === "assistant" && Array.isArray(m.tool_calls)); + expect(assistant).toBeDefined(); + // No reasoning anywhere, no cache entry: the serialization is bare, which + // is exactly the 400 shape the invariant counter must surface. + expect(assistant!["reasoning_content"]).toBeUndefined(); + expect(getReasoningReplayStats().bareSerializationsByModel["deepseek-v4-flash"]).toBe(1); + }); +}); From b8158f1e706c972e66516ea95f31d52ecc304806 Mon Sep 17 00:00:00 2001 From: Agent59353 Date: Thu, 6 Aug 2026 21:30:17 +0800 Subject: [PATCH 2/4] fix(responses): keep reasoning handoff across empty batch text deltas (#950) --- src/bridge.ts | 14 ++-- tests/bridge-reasoning-replay-batch.test.ts | 92 +++++++++++++++++++++ 2 files changed, 101 insertions(+), 5 deletions(-) create mode 100644 tests/bridge-reasoning-replay-batch.test.ts diff --git a/src/bridge.ts b/src/bridge.ts index 45f046f424..699f5c83e3 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -827,8 +827,10 @@ export function bridgeToResponsesSSE( if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - // Reasoning consumed by a text turn, not a tool call: no cache target. - rawReasoningForNextToolCall = ""; + // Reasoning consumed by a REAL text turn, not a tool call: no cache target. + // Empty text deltas must not wipe reasoning that precedes a tool call + // (chat-completions providers emit empty content deltas mid-tool-turn). + if (event.text.length > 0) 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. @@ -880,7 +882,7 @@ export function bridgeToResponsesSSE( if (currentMsg) closeCurrentMessage("commentary"); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); - rawReasoningForNextToolCall = ""; + if (event.thinking.length > 0) rawReasoningForNextToolCall = ""; if (currentToolCall) closeCurrentToolCall(); if (!currentReasoning) { const itemId = `rs_${uuid()}`; @@ -1561,7 +1563,9 @@ function buildResponseJSONWithBudget( if (currentText && e.phase !== undefined && currentTextPhase !== e.phase) flushText("commentary"); if (currentSummaryReasoning) flushSummaryReasoning(); if (currentRawReasoning) flushRawReasoning(); - rawReasoningForNextToolCall = ""; + // Empty text deltas (batch chat responses always carry content, often "") must + // not wipe reasoning that precedes a tool call (#950 non-streaming path). + if (e.text.length > 0) 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. @@ -1580,7 +1584,7 @@ function buildResponseJSONWithBudget( case "thinking_delta": if (currentText) flushText("commentary"); if (currentRawReasoning) flushRawReasoning(); - rawReasoningForNextToolCall = ""; + if (e.thinking.length > 0) rawReasoningForNextToolCall = ""; if (currentToolCallId) flushToolCall(); { ({ value: currentSummaryReasoning, bytes: currentSummaryReasoningBytes } = appendBatchString( diff --git a/tests/bridge-reasoning-replay-batch.test.ts b/tests/bridge-reasoning-replay-batch.test.ts new file mode 100644 index 0000000000..de617128be --- /dev/null +++ b/tests/bridge-reasoning-replay-batch.test.ts @@ -0,0 +1,92 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../src/bridge"; +import { + clearReasoningReplayCacheForTests, + peekReasoningForCall, +} from "../src/responses/reasoning-replay-cache"; +import type { AdapterEvent } from "../src/types"; + +/** + * Regression for issue #950's non-streaming path: chat-completions batch + * responses always carry `content` (often an empty string), and the adapter + * emits that as a text_delta BEFORE tool_call_start. The bridge used to wipe + * the pending reasoning handoff on every text_delta, so batch-born tool rounds + * never entered the replay cache and any later continuation serialized bare → + * upstream 400 ("The `reasoning_content` in the thinking mode must be passed + * back to the API."). All real-world 400s on this proxy were closeReason + * non_stream. + */ + +const REASONING = "I need to inspect files before answering."; +const SCOPE = "thread-batch"; + +function batchOutput(events: AdapterEvent[]): Record { + return buildResponseJSON(events, "opencode-free/deepseek-v4-flash-free", { + replayCacheScope: SCOPE, + }); +} + +async function streamFrames(events: AdapterEvent[]): Promise { + async function* replay(list: AdapterEvent[]): AsyncGenerator { + for (const event of list) yield event; + } + const reader = bridgeToResponsesSSE( + replay(events), + "opencode-free/deepseek-v4-flash-free", + undefined, + undefined, + undefined, + undefined, + undefined, + { replayCacheScope: SCOPE }, + ).getReader(); + const decoder = new TextDecoder(); + while (true) { + const { done, value } = await reader.read(); + if (done) break; + decoder.decode(value, { stream: true }); + } +} + +const toolRoundEvents = (): AdapterEvent[] => [ + { type: "reasoning_raw_delta", text: REASONING }, + { type: "text_delta", text: "" }, + { type: "tool_call_start", id: "call_batch_1", name: "read_file" }, + { type: "tool_call_delta", arguments: '{"path":"README.md"}' }, + { type: "tool_call_end" }, + { type: "done" }, +]; + +describe("reasoning replay survives empty text deltas (both wire modes)", () => { + beforeEach(() => { + clearReasoningReplayCacheForTests(); + }); + afterEach(() => { + clearReasoningReplayCacheForTests(); + }); + + test("batch: reasoning + empty content + tool call is cached for the call id", () => { + const response = batchOutput(toolRoundEvents()); + expect(response.output.some(o => (o as Record).type === "function_call")).toBe(true); + expect(peekReasoningForCall("call_batch_1", SCOPE)).toBe(REASONING); + }); + + 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." }; + batchOutput(events); + expect(peekReasoningForCall("call_batch_1", SCOPE)).toBeUndefined(); + }); + + test("stream: reasoning + empty content delta + tool call is cached for the call id", async () => { + await streamFrames(toolRoundEvents()); + expect(peekReasoningForCall("call_batch_1", SCOPE)).toBe(REASONING); + }); + + 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." }; + await streamFrames(events); + expect(peekReasoningForCall("call_batch_1", SCOPE)).toBeUndefined(); + }); +}); From 43ca009e506aa4a3a87b59ba9fb4bf3a385f2a86 Mon Sep 17 00:00:00 2001 From: Agent59353 Date: Thu, 6 Aug 2026 21:30:17 +0800 Subject: [PATCH 3/4] refactor(responses): share config-dir resolution and harden spill writes (#950) --- src/config.ts | 3 +- src/lib/config-dir.ts | 18 ++++++++ src/responses/reasoning-replay-cache.ts | 40 ++++++++++++------ tests/reasoning-replay-robustness.test.ts | 51 ++++++++++++++++++++++- 4 files changed, 96 insertions(+), 16 deletions(-) create mode 100644 src/lib/config-dir.ts diff --git a/src/config.ts b/src/config.ts index c962f07434..fd3f0df7ce 100644 --- a/src/config.ts +++ b/src/config.ts @@ -5,6 +5,7 @@ import { homedir } from "node:os"; import { dirname, join, resolve } from "node:path"; import { Database } from "bun:sqlite"; import * as z from "zod/v4"; +import { resolveOpenCodexConfigDir } from "./lib/config-dir"; import { bumpConfigGenerationAtPath, bumpCurrentConfigGeneration, @@ -549,7 +550,7 @@ let resolvedConfigDirCache: { raw: string | undefined; path: string } | null = n function resolveConfigDir(): string { const raw = process.env["OPENCODEX_HOME"]?.trim() || undefined; if (resolvedConfigDirCache && resolvedConfigDirCache.raw === raw) return resolvedConfigDirCache.path; - const path = raw ? resolve(expandUserPath(raw)) : join(homedir(), ".opencodex"); + const path = resolveOpenCodexConfigDir(); resolvedConfigDirCache = { raw, path }; return path; } diff --git a/src/lib/config-dir.ts b/src/lib/config-dir.ts new file mode 100644 index 0000000000..357f9c32c8 --- /dev/null +++ b/src/lib/config-dir.ts @@ -0,0 +1,18 @@ +/** + * Dependency-free config-directory resolution shared by config.ts and the + * reasoning replay spill (PR #1126): OPENCODEX_HOME (with `~` expansion) wins, + * otherwise /.opencodex. Kept primitive on purpose so leaf modules can + * import it without pulling in the whole config surface. + */ +import { homedir } from "node:os"; +import { join, resolve } from "node:path"; + +export function resolveOpenCodexConfigDir( + env: Record = process.env, +): string { + const raw = env["OPENCODEX_HOME"]?.trim() || undefined; + if (!raw) return join(homedir(), ".opencodex"); + if (raw === "~") return homedir(); + if (raw.startsWith("~/") || raw.startsWith("~\\")) return join(homedir(), raw.slice(2)); + return resolve(raw); +} diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index df9bb96ef6..2602ee68fa 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -25,12 +25,15 @@ */ import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; -import { homedir } from "node:os"; import { dirname, join } from "node:path"; +import { resolveOpenCodexConfigDir } from "../lib/config-dir"; const MAX_ENTRIES = 64; const MAX_TOTAL_BYTES = 256 * 1024; const TTL_MS = 60 * 60 * 1000; +// Clock-skew tolerance when validating spilled timestamps: entries dated more +// than this far in the future are rejected as invalid rather than trusted. +const MAX_FUTURE_SKEW_MS = 60 * 1000; const PERSIST_DEBOUNCE_MS = 750; const PERSIST_ENV = "OPENCODEX_REASONING_REPLAY_PERSIST"; const PERSIST_FILE_ENV = "OPENCODEX_REASONING_REPLAY_FILE"; @@ -75,15 +78,9 @@ function persistFlagFromEnv(): boolean { return raw === "1" || raw === "true" || raw === "yes" || raw === "on"; } -/** Mirror config.getConfigDir() resolution (OPENCODEX_HOME or ~/.opencodex) without importing config. */ +/** Shared resolution (OPENCODEX_HOME or ~/.opencodex) — see src/lib/config-dir.ts. */ function defaultPersistPath(): string { - const raw = process.env.OPENCODEX_HOME?.trim(); - let base: string; - if (!raw) base = join(homedir(), ".opencodex"); - else if (raw === "~") base = homedir(); - else if (raw.startsWith("~/") || raw.startsWith("~\\")) base = join(homedir(), raw.slice(2)); - else base = raw; - return join(base, "reasoning-replay-cache.json"); + return join(resolveOpenCodexConfigDir(), "reasoning-replay-cache.json"); } function persistPathFromEnv(): string { @@ -199,8 +196,11 @@ function writePersisted(): void { try { mkdirSync(dirname(persistPath), { recursive: true }); } catch { /* best-effort */ } - const tmpPath = `${persistPath}.tmp`; - writeFileSync(tmpPath, JSON.stringify(payload), "utf8"); + // Unique temp name + 0600 at creation, then atomic rename: a concurrent + // reader can never observe a half-written spill, and the final file never + // depends on a post-rename chmod that a rename may not preserve. + const tmpPath = `${persistPath}.${process.pid}.${Math.random().toString(36).slice(2, 10)}`; + writeFileSync(tmpPath, JSON.stringify(payload), { encoding: "utf8", mode: 0o600 }); renameSync(tmpPath, persistPath); try { chmodSync(persistPath, 0o600); } catch { /* best-effort on platforms that ignore chmod */ } persistWrites += 1; @@ -226,12 +226,26 @@ function loadPersisted(): void { } if (data?.v !== 1 || !Array.isArray(data.entries)) return; const at = now(); + // Malformed, expired, or future-dated entries are dropped and the spill is + // rewritten so the file cannot accumulate junk that every reload re-parses. + let dirty = false; for (const entry of data.entries) { - if (!entry || typeof entry.callId !== "string" || typeof entry.text !== "string" || entry.text.length === 0) continue; + if (!entry || typeof entry.callId !== "string" || typeof entry.text !== "string" || entry.text.length === 0) { + dirty = true; + continue; + } const entryAt = typeof entry.at === "number" && Number.isFinite(entry.at) ? entry.at : at; - if (at - entryAt >= TTL_MS) continue; + if (entryAt - at > MAX_FUTURE_SKEW_MS) { + dirty = true; // future-dated timestamps are invalid + continue; + } + if (at - entryAt >= TTL_MS) { + dirty = true; // expired + continue; + } rememberReasoningAt(entry.callId, entry.text, typeof entry.scope === "string" ? entry.scope : undefined, entryAt); } + if (dirty) writePersisted(); } // ── Privacy-safe diagnostics (counters only, never reasoning text) ─────────── diff --git a/tests/reasoning-replay-robustness.test.ts b/tests/reasoning-replay-robustness.test.ts index a9851269d3..8dbfbe7cd8 100644 --- a/tests/reasoning-replay-robustness.test.ts +++ b/tests/reasoning-replay-robustness.test.ts @@ -1,7 +1,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; -import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { basename, join } from "node:path"; import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; import { parseRequest } from "../src/responses/parser"; import { @@ -118,6 +118,53 @@ describe("reasoning replay — opt-in disk spill", () => { expect(getReasoningReplayStats().entries).toBe(0); }); + test("expired spill entries are dropped and the spill file is rewritten after load", () => { + let clock = 0; + clearReasoningReplayCacheForTests(() => clock); + setReasoningReplayPersistenceForTests(true, spillFile); + rememberReasoningForCall("call_ttl_rewrite", REASONING, "thread-ttl-rewrite"); + flushReasoningReplayCache(); + + clock = 61 * 60 * 1000; // past the 60-minute TTL + setReasoningReplayPersistenceForTests(false); + clearReasoningReplayCacheForTests(() => clock); + setReasoningReplayPersistenceForTests(true, spillFile); + + expect(peekReasoningForCall("call_ttl_rewrite", "thread-ttl-rewrite")).toBeUndefined(); + const reloaded = JSON.parse(readFileSync(spillFile, "utf8")) as { entries: unknown[] }; + expect(reloaded.entries).toHaveLength(0); + }); + + test("future-dated spill entries are rejected on load and the file is rewritten", () => { + const clock = 1_000; + const future = clock + 5 * 60 * 1000; + clearReasoningReplayCacheForTests(() => clock); + writeFileSync( + spillFile, + JSON.stringify({ + v: 1, + savedAt: future, + entries: [{ scope: "thread-future", callId: "call_future", text: REASONING, at: future }], + }), + "utf8", + ); + setReasoningReplayPersistenceForTests(true, spillFile); + + expect(peekReasoningForCall("call_future", "thread-future")).toBeUndefined(); + const reloaded = JSON.parse(readFileSync(spillFile, "utf8")) as { entries: unknown[] }; + expect(reloaded.entries).toHaveLength(0); + }); + + test("spill writes use a unique temp file and leave no remnants", () => { + setReasoningReplayPersistenceForTests(true, spillFile); + rememberReasoningForCall("call_tmp", REASONING, "thread-tmp"); + flushReasoningReplayCache(); + + expect(existsSync(spillFile)).toBe(true); + const remnants = readdirSync(spillDir).filter(f => f !== basename(spillFile)); + expect(remnants).toHaveLength(0); + }); + test("corrupt or unreadable spill file loads as an empty cache without throwing", () => { writeFileSync(spillFile, "{not json!!", "utf8"); setReasoningReplayPersistenceForTests(true, spillFile); From abcaf34aa30db1f90a526252649b227bcc87dd57 Mon Sep 17 00:00:00 2001 From: Agent59353 Date: Thu, 6 Aug 2026 21:39:06 +0800 Subject: [PATCH 4/4] fix(responses): flush spill on exit and reclaim stale temp files (#950) --- src/responses/reasoning-replay-cache.ts | 29 +++++++++++++++++++++-- tests/reasoning-replay-robustness.test.ts | 16 ++++++++++++- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index 2602ee68fa..349421e335 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -24,8 +24,8 @@ * so a long-lived proxy cannot grow without limit. */ -import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; -import { dirname, join } from "node:path"; +import { chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, statSync, unlinkSync, writeFileSync } from "node:fs"; +import { basename, dirname, join } from "node:path"; import { resolveOpenCodexConfigDir } from "../lib/config-dir"; const MAX_ENTRIES = 64; @@ -205,11 +205,29 @@ function writePersisted(): void { try { chmodSync(persistPath, 0o600); } catch { /* best-effort on platforms that ignore chmod */ } persistWrites += 1; persistLastError = undefined; + cleanupStaleTempFiles(); } catch (err) { persistLastError = err instanceof Error ? err.message : String(err); } } +/** Remove temp files left behind by a crashed writer (unique names mean they + * can never be confused with a live write, but they should still be reclaimed). */ +function cleanupStaleTempFiles(): void { + try { + const dir = dirname(persistPath); + const base = basename(persistPath); + const cutoff = now() - 10 * 60 * 1000; + for (const name of readdirSync(dir)) { + if (!name.startsWith(`${base}.`) || name === base) continue; + const full = join(dir, name); + try { + if (statSync(full).mtimeMs < cutoff) unlinkSync(full); + } catch { /* best-effort per file */ } + } + } catch { /* best-effort */ } +} + function loadPersisted(): void { if (!persistEnabled || !existsSync(persistPath)) return; let raw: string; @@ -290,6 +308,13 @@ export function getReasoningReplayStats(): ReasoningReplayStats { // Boot-time rehydration when persistence is opted in. if (persistEnabled) loadPersisted(); +// Synchronous best-effort flush so a normal process exit does not strand a +// just-recorded call's reasoning when persistence is opted in (P2: flush +// before exit). Bounded by the same caps as every other write. +process.on("exit", () => { + if (persistEnabled) writePersisted(); +}); + // ── Test seams ──────────────────────────────────────────────────────────────── /** Test-only: reset the cache and optionally pin the clock. */ diff --git a/tests/reasoning-replay-robustness.test.ts b/tests/reasoning-replay-robustness.test.ts index 8dbfbe7cd8..31945598f3 100644 --- a/tests/reasoning-replay-robustness.test.ts +++ b/tests/reasoning-replay-robustness.test.ts @@ -1,5 +1,5 @@ import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; -import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, utimesSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, join } from "node:path"; import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; @@ -165,6 +165,20 @@ describe("reasoning replay — opt-in disk spill", () => { expect(remnants).toHaveLength(0); }); + test("stale spill temp files from a crashed writer are reclaimed on the next write", () => { + setReasoningReplayPersistenceForTests(true, spillFile); + const stale = `${spillFile}.9999.deadbeef`; + writeFileSync(stale, "junk", "utf8"); + const old = new Date(Date.now() - 60 * 60 * 1000); + utimesSync(stale, old, old); + + rememberReasoningForCall("call_cleanup", REASONING, "thread-cleanup"); + flushReasoningReplayCache(); + + expect(existsSync(stale)).toBe(false); + expect(existsSync(spillFile)).toBe(true); + }); + test("corrupt or unreadable spill file loads as an empty cache without throwing", () => { writeFileSync(spillFile, "{not json!!", "utf8"); setReasoningReplayPersistenceForTests(true, spillFile);