diff --git a/src/bridge.ts b/src/bridge.ts index 85802118f6..c34e91734a 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -11,7 +11,10 @@ import { adapterFailureFromMessage, classifyError, CYBER_POLICY_ERROR_CODE, isCy import { encodeCompactionSummary } from "./responses/compaction"; import { encodeReasoningEnvelope, type ReasoningEnvelope } from "./responses/reasoning-envelope"; import { rememberReasoningForCall } from "./responses/reasoning-replay-cache"; -import { responsesExtraContentFromProviderMetadata } from "./responses/provider-opaque-metadata"; +import { + rememberAndSerializeExtraContent, + rememberExtraContentForReplay, +} from "./responses/thought-signature-replay"; import { resolveStallTimeoutSec } from "./stall-timeout"; import { usageDisplayTotalTokens } from "./usage/totals"; import { @@ -606,6 +609,9 @@ export function bridgeToResponsesSSE( input: freeformInput(currentToolCall.args), }); } + // Freeform tools serialize as custom_tool_call without extra_content; remember the + // signature server-side regardless so the replayed call can be re-signed (#1735). + void rememberExtraContentForReplay(currentToolCall.callId, currentToolCall.providerMetadata, replayCacheScope); const item = currentToolCall.toolSearch ? { type: "tool_search_call", id: currentToolCall.itemId, @@ -624,8 +630,9 @@ export function bridgeToResponsesSSE( arguments: argsStr, status: "completed", ...(currentToolCall.namespace ? { namespace: currentToolCall.namespace } : {}), // Provider-opaque metadata (issue #1735) rides the item so a client that replays - // this history can hand the signature back on the part it belongs to. - ...(responsesExtraContentFromProviderMetadata(currentToolCall.providerMetadata) ?? {}), + // this history can hand the signature back on the part it belongs to. The proxy + // also remembers it server-side for clients that never echo extra_content. + ...(rememberAndSerializeExtraContent(currentToolCall.callId, currentToolCall.providerMetadata, replayCacheScope).extra ?? {}), }; emit("response.output_item.done", { output_index: currentToolCall.outputIndex, item }); retainFinishedItem(item as OutputItem); @@ -643,6 +650,7 @@ export function bridgeToResponsesSSE( const failCurrentToolCall = () => { if (!currentToolCall) return; const argsStr = currentToolCall.args || "{}"; + void rememberExtraContentForReplay(currentToolCall.callId, currentToolCall.providerMetadata, replayCacheScope); const item = currentToolCall.toolSearch ? { type: "tool_search_call", id: currentToolCall.itemId, @@ -663,7 +671,7 @@ export function bridgeToResponsesSSE( // An incomplete call can still be persisted and replayed (max_output_tokens), so it // carries the same metadata as the completed item — otherwise SSE and buffered JSON // would disagree about whether the signature survives. - ...(responsesExtraContentFromProviderMetadata(currentToolCall.providerMetadata) ?? {}), + ...(rememberAndSerializeExtraContent(currentToolCall.callId, currentToolCall.providerMetadata, replayCacheScope).extra ?? {}), }; emit("response.output_item.done", { output_index: currentToolCall.outputIndex, item }); retainFinishedItem(item as OutputItem); @@ -1576,6 +1584,9 @@ function buildResponseJSONWithBudget( currentToolCallArgs, options?.toolParameterSchemas?.get(currentToolCallName), ); + // Freeform tools serialize as custom_tool_call without extra_content; remember the + // signature server-side regardless so the replayed call can be re-signed (#1735). + void rememberExtraContentForReplay(currentToolCallId, currentToolCallProviderMetadata, replayCacheScope); if (toolSearch) { pushOutput({ type: "tool_search_call", id: `tsc_${uuid()}`, @@ -1594,7 +1605,7 @@ function buildResponseJSONWithBudget( call_id: currentToolCallId, name: realName, arguments: coercedArgs || "{}", status, ...(ns ? { namespace: ns } : {}), - ...(responsesExtraContentFromProviderMetadata(currentToolCallProviderMetadata) ?? {}), + ...(rememberAndSerializeExtraContent(currentToolCallId, currentToolCallProviderMetadata, replayCacheScope).extra ?? {}), }); } budget?.closeCall(currentToolCallId); diff --git a/src/lib/config-ownership.ts b/src/lib/config-ownership.ts index 553581e2d7..27a8f823c3 100644 --- a/src/lib/config-ownership.ts +++ b/src/lib/config-ownership.ts @@ -69,6 +69,7 @@ const INITIAL_OWNED_PATHS = [ "service-state.json", "service.log", "system-env-port", + "thought-signature-replay.json", "tray-heartbeat.json", "tray-state.json", "update-job.json", diff --git a/src/responses/parser.ts b/src/responses/parser.ts index 225bb89e1a..068c2cc52d 100644 --- a/src/responses/parser.ts +++ b/src/responses/parser.ts @@ -9,10 +9,12 @@ import type { OcxThinkingContent, OcxTool, OcxToolCall, + OcxReasoningReplayScopeRef, } from "../types"; import { namespacedToolName } from "../types"; import { responsesRequestSchema } from "./schema"; import { providerMetadataFromResponsesFunctionCall } from "./provider-opaque-metadata"; +import { lookupReplayThoughtSignature } from "./thought-signature-replay"; import { compactionItemToText } from "./compaction"; import { previousResponseReplayPrefixLength } from "./state"; import { decodeReasoningEnvelope } from "./reasoning-envelope"; @@ -23,6 +25,21 @@ function isObj(v: unknown): v is Record { return typeof v === "object" && v !== null && !Array.isArray(v); } +/** + * Wrap a remembered proxy-side signature as provider metadata for a replayed tool call. + * + * The scope is REQUIRED for a hit. `parseRequest` runs before the route and account are + * chosen, so a caller that has not yet bound a replay scope gets nothing rather than a + * signature belonging to some other thread that happened to reuse the same `call_id`. + */ +function replayThoughtSignatureMetadata( + callId: string, + scope: OcxReasoningReplayScopeRef | undefined, +): { google: { thoughtSignature: string } } | undefined { + const signature = lookupReplayThoughtSignature(callId, scope); + return signature ? { google: { thoughtSignature: signature } } : undefined; +} + type InputBlock = | { type: "input_text"; text: string } | { type: "text"; text: string } @@ -294,7 +311,11 @@ function attachPendingReasoningToCallOwner( const REASONING_EFFORTS = new Set(["none", "minimal", "low", "medium", "high", "xhigh", "max"]); -export function parseRequest(body: unknown): OcxParsedRequest { +export function parseRequest( + body: unknown, + parseOptions?: { replayCacheScope?: OcxReasoningReplayScopeRef }, +): OcxParsedRequest { + const replayCacheScope = parseOptions?.replayCacheScope; const replayedInputPrefixLength = previousResponseReplayPrefixLength(body); const parsed = responsesRequestSchema.safeParse(body); if (!parsed.success) { @@ -522,8 +543,12 @@ export function parseRequest(body: unknown): OcxParsedRequest { }; // Provider-opaque metadata (e.g. a Gemini thought signature) travels with the call so a // history-replayed or previous_response_id turn rebuilds the same signed part instead of - // depending on the same-process replay cache (issue #1735). - const providerMetadata = providerMetadataFromResponsesFunctionCall(call); + // depending on the same-process replay cache (issue #1735). Real clients do not echo + // extra_content on replay, so fall back to the proxy-side store keyed by call_id. + const providerMetadata = providerMetadataFromResponsesFunctionCall(call) + ?? (typeof call.call_id === "string" + ? replayThoughtSignatureMetadata(call.call_id, replayCacheScope) + : undefined); if (providerMetadata) toolCall.providerMetadata = providerMetadata; assistantHolderWithReasoning().content.push(toolCall); continue; @@ -531,10 +556,12 @@ export function parseRequest(body: unknown): OcxParsedRequest { if (effectiveType === "custom_tool_call") { const call = item as { id?: string; call_id: string; name: string; input: string }; + const remembered = typeof call.call_id === "string" ? replayThoughtSignatureMetadata(call.call_id, replayCacheScope) : undefined; const toolCall: OcxToolCall = { type: "toolCall", id: call.call_id, name: call.name, arguments: { input: call.input ?? "" }, customWireName: call.name, + ...(remembered ? { providerMetadata: remembered } : {}), }; assistantHolderWithReasoning().content.push(toolCall); continue; @@ -547,9 +574,11 @@ export function parseRequest(body: unknown): OcxParsedRequest { const callId = call.call_id ?? call.id; if (callId) { const command = Array.isArray(call.action?.command) ? call.action.command : []; + const remembered = replayThoughtSignatureMetadata(callId, replayCacheScope); assistantHolderWithReasoning().content.push({ type: "toolCall", id: callId, name: "shell", arguments: command.length > 0 ? { command } : {}, + ...(remembered ? { providerMetadata: remembered } : {}), }); } continue; @@ -568,9 +597,11 @@ export function parseRequest(body: unknown): OcxParsedRequest { // history stays complete (otherwise the model re-issues tool_search forever). const call = item as { id?: string; call_id?: string; arguments?: unknown }; const callId = call.call_id ?? call.id ?? ""; + const remembered = callId ? replayThoughtSignatureMetadata(callId, replayCacheScope) : undefined; assistantHolderWithReasoning().content.push({ type: "toolCall", id: callId, name: "tool_search", arguments: isObj(call.arguments) ? call.arguments : {}, + ...(remembered ? { providerMetadata: remembered } : {}), }); continue; } diff --git a/src/responses/provider-opaque-metadata.ts b/src/responses/provider-opaque-metadata.ts index c52e30951f..070982b722 100644 --- a/src/responses/provider-opaque-metadata.ts +++ b/src/responses/provider-opaque-metadata.ts @@ -30,7 +30,7 @@ function isObj(value: unknown): value is Record { */ const MAX_SIGNATURE_BYTES = 64 * 1024; -function isCarryableSignature(value: unknown): value is string { +export function isCarryableSignature(value: unknown): value is string { if (typeof value !== "string" || value.length === 0) return false; // Cheap length pre-check: UTF-8 is at most 3 bytes per UTF-16 code unit for the BMP, so this // skips the encode for the overwhelmingly common short case. diff --git a/src/responses/thought-signature-replay.ts b/src/responses/thought-signature-replay.ts new file mode 100644 index 0000000000..421305cde4 --- /dev/null +++ b/src/responses/thought-signature-replay.ts @@ -0,0 +1,261 @@ +/** + * Server-side thought-signature replay store (issue #1735 follow-up). + * + * Gemini issues a thoughtSignature on the function-call part of a response and requires it + * back when the same call is replayed in a later request. The Responses wire carries the + * signature in extra_content.google.thought_signature, and a conforming client echoes it on + * the replay. Real clients (codex-rs 0.144.x, Codex desktop) do NOT echo extra_content: + * they replay history as bare function_call / custom_tool_call items keyed by call_id. + * Without the signature Gemini rejects the replayed part with + * "Function call is missing a thought_signature in functionCall parts". + * + * This module is the proxy-side fallback: remember the signature we handed out and re-attach + * it on replay even when the client never echoes it. Values stay opaque (never parsed or + * re-encoded) and are bounded like the wire metadata. + * + * SCOPE: a client-visible `call_id` is NOT unique across conversations, accounts, providers + * or models. Keying on it alone let one thread's signature overwrite another's, and let a + * lookup hand a signature from a different account's turn to the current one. The key is the + * same identity the in-process reasoning cache already uses — thread plus exact provider + * destination, adapter, model and credential — so a signature can only ever be replayed into + * the turn that produced it. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { atomicWriteFileAsync, getConfigDir } from "../config"; +import type { OcxProviderOpaqueToolCallMetadata, OcxReasoningReplayScopeRef } from "../types"; +import { isCarryableSignature, responsesExtraContentFromProviderMetadata } from "./provider-opaque-metadata"; + +const STORE_FILE_NAME = "thought-signature-replay.json"; + +/** Bound on remembered entries; real signatures are a few hundred bytes, so this stays small. */ +const MAX_ENTRIES = 16_384; +/** + * Total bytes of remembered signature material. + * + * An entry count alone is not a memory bound: a single signature may be 64KiB, so 16,384 + * entries is a ~1GiB ceiling. This is the bound that actually holds. + */ +const MAX_TOTAL_BYTES = 32 * 1024 * 1024; +/** A signature is needed for the immediate next turn; a long TTL also covers resumed threads. */ +const TTL_MS = 7 * 24 * 60 * 60 * 1000; + +type StoredEntry = { sig: string; savedAt: number }; + +/** Outcome of a remember attempt. `conflict` is a real signal, not a no-op. */ +export type ThoughtSignatureRememberResult = + | "stored" + | "already-equal" + | "conflict" + | "unscoped" + | "ignored"; + +let entries = new Map(); +let totalBytes = 0; +let loaded = false; +let persistChain: Promise = Promise.resolve(); + +function storePath(): string { + return join(getConfigDir(), STORE_FILE_NAME); +} + +function nonEmpty(value: unknown): value is string { + return typeof value === "string" && value.trim().length > 0; +} + +/** + * Durable key for one call, or `undefined` when the scope is incomplete. + * + * Incomplete scope means "do not remember" rather than "remember globally": a partially + * identified entry is exactly the cross-thread collision this store exists to prevent. + * The reasoning cache's identities are process-local HMACs, so this key deliberately uses + * only the stable, non-secret fields that survive a restart. + */ +function keyFor(callId: string, scope: OcxReasoningReplayScopeRef | undefined): string | undefined { + const identity = scope?.current; + if ( + !nonEmpty(callId) + || !nonEmpty(scope?.clientThreadId) + || !nonEmpty(identity?.providerName) + || !nonEmpty(identity?.adapterName) + || !nonEmpty(identity?.modelId) + ) return undefined; + return JSON.stringify([ + scope.clientThreadId, + identity.providerName, + identity.adapterName, + identity.modelId, + callId, + ]); +} + +function load(): void { + if (loaded) return; + loaded = true; + let raw: string; + try { + raw = readFileSync(storePath(), "utf8"); + } catch { + return; // First run or unreadable file: start empty. + } + try { + const parsed: unknown = JSON.parse(raw); + if (typeof parsed !== "object" || parsed === null || !Array.isArray((parsed as { entries?: unknown }).entries)) { + return; + } + const nowMs = Date.now(); + for (const entry of (parsed as { entries: unknown[] }).entries) { + if (typeof entry !== "object" || entry === null) continue; + const { key, sig, savedAt } = entry as { key?: unknown; sig?: unknown; savedAt?: unknown }; + if (typeof key !== "string" || typeof sig !== "string" || typeof savedAt !== "number") continue; + if (savedAt <= nowMs - TTL_MS) continue; + if (!isCarryableSignature(sig)) continue; + entries.set(key, { sig, savedAt }); + totalBytes += sig.length; + } + } catch { + // Corrupt store: ignore it; a later remember() rewrites a clean snapshot. + } + // A loaded snapshot can already exceed the bounds if they were lowered, so enforce them + // here rather than waiting for the next write. + prune(Date.now()); +} + +function prune(nowMs: number): void { + for (const [key, entry] of entries) { + if (nowMs - entry.savedAt > TTL_MS) { + entries.delete(key); + totalBytes -= entry.sig.length; + } + } + if (entries.size <= MAX_ENTRIES && totalBytes <= MAX_TOTAL_BYTES) return; + const sorted = [...entries.entries()].sort((a, b) => a[1].savedAt - b[1].savedAt); + for (const [key, entry] of sorted) { + if (entries.size <= MAX_ENTRIES && totalBytes <= MAX_TOTAL_BYTES) break; + entries.delete(key); + totalBytes -= entry.sig.length; + } +} + +function persist(): Promise { + persistChain = persistChain + .then(async () => { + const snapshot = JSON.stringify({ + version: 2, + entries: [...entries].map(([key, entry]) => ({ key, sig: entry.sig, savedAt: entry.savedAt })), + }); + await atomicWriteFileAsync(storePath(), snapshot); + }) + .catch(() => { + // Best-effort persistence: the in-memory store still serves the running process. + }); + return persistChain; +} + +/** + * Record the signature that left the proxy on a function-call response item. + * + * Returns the outcome so a caller can await durability before exposing the item, and so a + * genuine conflict is observable instead of silently overwriting. A different signature under + * the SAME complete key means two different upstream turns claimed one identity: that is a + * corruption signal, and keeping the first value is the fail-closed choice. + */ +export function rememberThoughtSignatureForReplay( + callId: string, + signature: string, + scope: OcxReasoningReplayScopeRef | undefined, +): { result: ThoughtSignatureRememberResult; durable: Promise } { + if (!callId || !isCarryableSignature(signature)) { + return { result: "ignored", durable: Promise.resolve() }; + } + const key = keyFor(callId, scope); + if (key === undefined) return { result: "unscoped", durable: Promise.resolve() }; + load(); + const existing = entries.get(key); + if (existing) { + if (existing.sig === signature) return { result: "already-equal", durable: Promise.resolve() }; + return { result: "conflict", durable: Promise.resolve() }; + } + entries.set(key, { sig: signature, savedAt: Date.now() }); + totalBytes += signature.length; + prune(Date.now()); + return { result: "stored", durable: persist() }; +} + +/** + * Serialize provider metadata onto an outbound Responses function_call item AND remember the + * signature server-side, so a client that replays the call without echoing extra_content can + * still be served from the store. + */ +export function rememberAndSerializeExtraContent( + callId: string, + metadata: OcxProviderOpaqueToolCallMetadata | undefined, + scope: OcxReasoningReplayScopeRef | undefined, +): { + extra?: { extra_content: { google: { thought_signature: string } } }; + durable: Promise; +} { + const extra = responsesExtraContentFromProviderMetadata(metadata); + if (!extra) return { durable: Promise.resolve() }; + const { durable } = rememberThoughtSignatureForReplay( + callId, + extra.extra_content.google.thought_signature, + scope, + ); + return { extra, durable }; +} + +/** + * Remember the signature without serializing it onto the item. Used for freeform tools, whose + * Responses items are custom_tool_call blocks that cannot carry extra_content — the signature + * still must be stored so the replayed call (which comes back as custom_tool_call and never + * echoes metadata) can be re-signed server-side. + */ +export function rememberExtraContentForReplay( + callId: string, + metadata: OcxProviderOpaqueToolCallMetadata | undefined, + scope: OcxReasoningReplayScopeRef | undefined, +): Promise { + const extra = responsesExtraContentFromProviderMetadata(metadata); + if (!extra) return Promise.resolve(); + return rememberThoughtSignatureForReplay( + callId, + extra.extra_content.google.thought_signature, + scope, + ).durable; +} + +/** Look up a signature previously handed out for this call in THIS scope, if still fresh. */ +export function lookupReplayThoughtSignature( + callId: string, + scope: OcxReasoningReplayScopeRef | undefined, +): string | undefined { + const key = keyFor(callId, scope); + if (key === undefined) return undefined; + load(); + const entry = entries.get(key); + if (!entry) return undefined; + if (Date.now() - entry.savedAt > TTL_MS) { + entries.delete(key); + totalBytes -= entry.sig.length; + return undefined; + } + return entry.sig; +} + +/** Test seams: clear in-memory state and the loaded flag without touching the file. */ +export function resetThoughtSignatureReplayForTests(): void { + entries = new Map(); + totalBytes = 0; + loaded = false; + persistChain = Promise.resolve(); +} + +export function thoughtSignatureReplayCountForTests(): number { + return entries.size; +} + +/** Test seam: resolve after the queued snapshot write settles. */ +export function flushThoughtSignatureReplayForTests(): Promise { + return persistChain; +} diff --git a/tests/google-signature-history-roundtrip.test.ts b/tests/google-signature-history-roundtrip.test.ts index 4557613717..049ea50f36 100644 --- a/tests/google-signature-history-roundtrip.test.ts +++ b/tests/google-signature-history-roundtrip.test.ts @@ -2,10 +2,19 @@ * #1735: a Gemini thought signature must survive a HISTORY-driven turn, where the same-process * replay cache is not available — the exact case the cache was masking. */ -import { beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { createGoogleAdapter as createGoogleAdapterProduction } from "../src/adapters/google"; import { __resetAntigravityReplayCache } from "../src/adapters/google-antigravity-replay"; import { parseRequest } from "../src/responses/parser"; +import { + flushThoughtSignatureReplayForTests, + lookupReplayThoughtSignature, + rememberThoughtSignatureForReplay, + resetThoughtSignatureReplayForTests, +} from "../src/responses/thought-signature-replay"; import type { AdapterEvent, OcxParsedRequest, OcxProviderConfig } from "../src/types"; import { withTestTranslatorBudget } from "./helpers/translator-budget"; @@ -23,6 +32,29 @@ const provider = { apiKey: "vertex-test-key", } as OcxProviderConfig; + +/** + * A replay scope is now REQUIRED for the store to remember or return anything: a + * client-visible call_id is not unique across threads, accounts, providers or models, + * so keying on it alone let one conversation's signature reach another's turn. + */ +function scopeFor(threadId = "thread-a", modelId = MODEL, providerName = "google") { + return { + clientThreadId: threadId, + current: { + providerName, + providerDestinationIdentity: `dest-${providerName}`, + adapterName: "google", + modelId, + credentialIdentity: `cred-${providerName}`, + }, + }; +} + +/** parseRequest with the replay scope bound, as the server does after route selection. */ +function parseRequestScoped(body: unknown, scope = scopeFor()) { + return parseRequest(body, { replayCacheScope: scope }); +} function firstTurn(): OcxParsedRequest { return { modelId: MODEL, @@ -49,7 +81,22 @@ function modelParts(body: string): Record[] { } describe("#1735 thought signature survives history replay", () => { - beforeEach(() => __resetAntigravityReplayCache()); + let previousHome: string | undefined; + let testDir: string; + + beforeEach(() => { + __resetAntigravityReplayCache(); + resetThoughtSignatureReplayForTests(); + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-thought-sig-")); + process.env.OPENCODEX_HOME = testDir; + }); + + afterEach(() => { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + rmSync(testDir, { recursive: true, force: true }); + }); test("the adapter attaches the signature to the tool call that produced it", async () => { const adapter = createGoogleAdapter(provider); @@ -78,7 +125,7 @@ describe("#1735 thought signature survives history replay", () => { test("a signature replayed through Responses history reaches the rebuilt Google part", async () => { // No cache is warmed here: this is a cold process replaying client-supplied history. - const parsed = parseRequest({ + const parsed = parseRequestScoped({ model: MODEL, input: [ { type: "message", role: "user", content: [{ type: "input_text", text: "run pwd" }] }, @@ -100,7 +147,7 @@ describe("#1735 thought signature survives history replay", () => { }); test("history without a signature stays unsigned rather than borrowing one", async () => { - const parsed = parseRequest({ + const parsed = parseRequestScoped({ model: MODEL, input: [ { type: "message", role: "user", content: [{ type: "input_text", text: "run pwd" }] }, @@ -113,4 +160,142 @@ describe("#1735 thought signature survives history replay", () => { const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); expect(part?.thoughtSignature).toBeUndefined(); }); + + test("a signature the proxy remembered re-signs a replay the client sent without extra_content", async () => { + // The proxy handed out SIGNATURE for call_shell_9 in a previous turn; the client replays + // the call as a bare function_call item (codex-rs/desktop never echo extra_content). + rememberThoughtSignatureForReplay("call_shell_9", SIGNATURE, scopeFor()); + const parsed = parseRequestScoped({ + model: MODEL, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "run pwd" }] }, + { type: "function_call", call_id: "call_shell_9", name: "shell_command", arguments: JSON.stringify({ command: "pwd" }) }, + { type: "function_call_output", call_id: "call_shell_9", output: "/workspace" }, + ], + tools: [{ type: "function", name: "shell_command", description: "run", parameters: { type: "object" } }], + }); + const request = await createGoogleAdapter(provider).buildRequest(parsed); + const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); + expect(part?.thoughtSignature).toBe(SIGNATURE); + }); + + test("a custom_tool_call replay is re-signed from the proxy-side store", async () => { + rememberThoughtSignatureForReplay("call_custom_1", SIGNATURE_B, scopeFor()); + const parsed = parseRequestScoped({ + model: MODEL, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "run pwd" }] }, + { type: "custom_tool_call", call_id: "call_custom_1", name: "shell_command", input: JSON.stringify({ command: "pwd" }) }, + { type: "custom_tool_call_output", call_id: "call_custom_1", output: "/workspace" }, + ], + tools: [{ type: "function", name: "shell_command", description: "run", parameters: { type: "object" } }], + }); + const request = await createGoogleAdapter(provider).buildRequest(parsed); + const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); + expect(part?.thoughtSignature).toBe(SIGNATURE_B); + }); + + test("a tool_search_call replay is re-signed from the proxy-side store", async () => { + rememberThoughtSignatureForReplay("call_ts_1", SIGNATURE, scopeFor()); + const parsed = parseRequestScoped({ + model: MODEL, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "search tool" }] }, + { type: "tool_search_call", call_id: "call_ts_1", arguments: { query: "grep" } }, + { type: "tool_search_output", call_id: "call_ts_1", tools: [] }, + ], + tools: [{ type: "function", name: "tool_search", description: "search", parameters: { type: "object" } }], + }); + const request = await createGoogleAdapter(provider).buildRequest(parsed); + const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); + expect(part?.thoughtSignature).toBe(SIGNATURE); + }); + + test("a local_shell_call replay is re-signed from the proxy-side store", async () => { + rememberThoughtSignatureForReplay("call_lsh_1", SIGNATURE, scopeFor()); + const parsed = parseRequestScoped({ + model: MODEL, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "shell command" }] }, + { type: "local_shell_call", call_id: "call_lsh_1", action: { type: "exec", command: ["ls"] } }, + { type: "function_call_output", call_id: "call_lsh_1", output: "file.txt" }, + ], + tools: [{ type: "function", name: "shell", description: "run", parameters: { type: "object" } }], + }); + const request = await createGoogleAdapter(provider).buildRequest(parsed); + const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); + expect(part?.thoughtSignature).toBe(SIGNATURE); + }); + + test("an unknown call_id stays unsigned", async () => { + const parsed = parseRequestScoped({ + model: MODEL, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "run pwd" }] }, + { type: "function_call", call_id: "call_never_seen", name: "shell_command", arguments: JSON.stringify({ command: "pwd" }) }, + { type: "function_call_output", call_id: "call_never_seen", output: "/workspace" }, + ], + tools: [{ type: "function", name: "shell_command", description: "run", parameters: { type: "object" } }], + }); + const request = await createGoogleAdapter(provider).buildRequest(parsed); + const part = modelParts(request.body as string).find(candidate => "functionCall" in candidate); + expect(part?.thoughtSignature).toBeUndefined(); + }); + + + test("the same call_id in a different thread does not borrow the signature (#1823)", () => { + // A client-visible call_id is not unique. Keyed on it alone, one conversation's + // signature was handed to another's replay -- and a second thread writing the same + // id silently overwrote the first. + rememberThoughtSignatureForReplay("call_shared", SIGNATURE, scopeFor("thread-a")); + + expect(lookupReplayThoughtSignature("call_shared", scopeFor("thread-a"))).toBe(SIGNATURE); + expect(lookupReplayThoughtSignature("call_shared", scopeFor("thread-b"))).toBeUndefined(); + }); + + test("a different account or model is a different scope (#1823)", () => { + rememberThoughtSignatureForReplay("call_scoped", SIGNATURE, scopeFor("thread-a", MODEL, "google")); + + expect(lookupReplayThoughtSignature("call_scoped", scopeFor("thread-a", MODEL, "google"))).toBe(SIGNATURE); + // Same thread and call id, different provider identity: opaque signatures are not + // portable across providers, so this must miss rather than cross-contaminate. + expect(lookupReplayThoughtSignature("call_scoped", scopeFor("thread-a", MODEL, "antigravity"))).toBeUndefined(); + // Same thread and provider, different model. + expect(lookupReplayThoughtSignature("call_scoped", scopeFor("thread-a", "gemini-3.6-pro", "google"))).toBeUndefined(); + }); + + test("a conflicting signature under one key fails closed instead of overwriting (#1823)", () => { + expect(rememberThoughtSignatureForReplay("call_conflict", SIGNATURE, scopeFor()).result).toBe("stored"); + // Re-remembering the same value is a no-op, not a conflict: retries are ordinary. + expect(rememberThoughtSignatureForReplay("call_conflict", SIGNATURE, scopeFor()).result).toBe("already-equal"); + // A DIFFERENT value under the same complete key means two upstream turns claimed one + // identity. Keeping the first is the fail-closed choice; last-write-wins would let a + // later turn silently invalidate an earlier replay. + expect(rememberThoughtSignatureForReplay("call_conflict", SIGNATURE_B, scopeFor()).result).toBe("conflict"); + expect(lookupReplayThoughtSignature("call_conflict", scopeFor())).toBe(SIGNATURE); + }); + + test("an incomplete scope remembers nothing rather than remembering globally (#1823)", () => { + // A partially identified entry is exactly the cross-thread collision this store exists + // to prevent, so it must not be stored under a degraded key. + expect(rememberThoughtSignatureForReplay("call_unscoped", SIGNATURE, undefined).result).toBe("unscoped"); + expect(rememberThoughtSignatureForReplay("call_unscoped", SIGNATURE, { clientThreadId: "t" }).result).toBe("unscoped"); + expect(lookupReplayThoughtSignature("call_unscoped", scopeFor())).toBeUndefined(); + }); + + test("a store write reports when it is durable (#1823)", async () => { + // The caller can await this before exposing the tool-call item, so a client cannot + // observe a call whose signature was never persisted. + const { result, durable } = rememberThoughtSignatureForReplay("call_durable", SIGNATURE, scopeFor()); + expect(result).toBe("stored"); + await durable; + expect(lookupReplayThoughtSignature("call_durable", scopeFor())).toBe(SIGNATURE); + }); + test("the proxy-side store survives a process restart via its snapshot", async () => { + rememberThoughtSignatureForReplay("call_disk_1", SIGNATURE, scopeFor()); + await flushThoughtSignatureReplayForTests(); + // Simulate a fresh process: drop in-memory state; lookup must reload from disk. + resetThoughtSignatureReplayForTests(); + expect(lookupReplayThoughtSignature("call_disk_1", scopeFor())).toBe(SIGNATURE); + }); });