From cc2a75290ead80bd717ac8839ef8779c8b42a5fb Mon Sep 17 00:00:00 2001 From: Vadym O Date: Thu, 15 Oct 2026 21:14:59 +0800 Subject: [PATCH] fix(anthropic): rebind reasoning-replay owner after pool rotation The Anthropic OAuth 429/403/401 rotation arms in adapter dispatch and the terminal continuation committed account B but left the reasoning-replay scope, native-reasoning owner, and provider-continuation owner bound to the refused account A. B was sent A's private continuation, the turn's state was persisted as owned by A, and the next turn on B stripped the replayed reasoning and dropped B's continuation as foreign. Rebind with the admitted credential snapshot, as the key-429 and generic OAuth arms already do; the continuation arm rebinds both parsed requests. Co-Authored-By: Claude Code --- src/server/responses/adapter-continuation.ts | 10 ++ src/server/responses/adapter-dispatch.ts | 7 ++ ...rminal-continuation-owner-rotation.test.ts | 106 +++++++++++++++++- 3 files changed, 120 insertions(+), 3 deletions(-) diff --git a/src/server/responses/adapter-continuation.ts b/src/server/responses/adapter-continuation.ts index 8bef1f9a7c4..582b8ee2850 100644 --- a/src/server/responses/adapter-continuation.ts +++ b/src/server/responses/adapter-continuation.ts @@ -440,6 +440,16 @@ export function createAdapterContinuations( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire, route.staticPolicy), config.cacheRetention, ); + // Rebind both the continuation clone and the outer request that persists the response. + for (const requestParsed of [nextParsed, parsed]) { + bindRouteReasoningReplayScope({ + parsed: requestParsed, + providerName: route.providerName, + provider: route.provider, + adapterName: transportState.activeAdapter.name, + oauthCredentialSnapshot: transportState.replayOAuthCredentialSnapshot, + }); + } sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, transportState.activeAdapter.name, logCtx.accountLogLabel); recordAttemptCredentialSource(logCtx.activeAttempt, route.providerName, route.provider, transportState.activeAdapter.name); nextContinuationRecoveryKind = response.status === 401 ? "oauth-401" : "anthropic-oauth-429"; diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index 6f418529a3f..99df2071ddd 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/src/server/responses/adapter-dispatch.ts @@ -1346,6 +1346,13 @@ export async function prepareAdapterExchange( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire, route.staticPolicy), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: route.provider, + adapterName: transportState.activeAdapter.name, + oauthCredentialSnapshot: transportState.replayOAuthCredentialSnapshot, + }); sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, transportState.activeAdapter.name, logCtx.accountLogLabel); recordAttemptCredentialSource(logCtx.activeAttempt, route.providerName, route.provider, transportState.activeAdapter.name); const result = await rebuildAndRefetch(upstreamResponse.status === 401 ? "oauth-401" : "anthropic-oauth-429"); diff --git a/tests/adapters/terminal-continuation-owner-rotation.test.ts b/tests/adapters/terminal-continuation-owner-rotation.test.ts index 69ce097aca4..74673b1891a 100644 --- a/tests/adapters/terminal-continuation-owner-rotation.test.ts +++ b/tests/adapters/terminal-continuation-owner-rotation.test.ts @@ -5,7 +5,10 @@ import { join } from "node:path"; import type { ProviderAdapter } from "../../src/adapters/base"; import { saveConfig } from "../../src/config"; import { clearKeyCooldowns } from "../../src/providers/key-failover"; -import { reasoningReplayKeyCredentialIdentity } from "../../src/responses/reasoning-replay-cache"; +import { clearAnthropicAccountPoolState } from "../../src/oauth/anthropic-routing"; +import { getValidAccessSnapshotForAccount } from "../../src/oauth"; +import { getAccountSet, saveCredential, setActiveAccount } from "../../src/oauth/store"; +import { reasoningReplayKeyCredentialIdentity, reasoningReplayOAuthCredentialIdentity } from "../../src/responses/reasoning-replay-cache"; import { clearResponseStateForTests, previousResponseProviderState, @@ -22,6 +25,7 @@ import { removeTreeWithRetry } from "../helpers/remove-tree"; interface BuildObservation { key: string; continuation?: string; + stripReasoning?: true; } let builds: BuildObservation[] = []; @@ -76,7 +80,8 @@ const actualResolveAdapter = actualResolver.resolveAdapter; mock.module("../../src/server/adapter-resolve", () => ({ ...actualResolver, resolveAdapter(provider: OcxProviderConfig, cacheRetention?: "none" | "short" | "long") { - if (provider.adapter !== "test-terminal-owned") { + // Match on the destination: the "anthropic" provider is wire-pinned to the anthropic adapter. + if (provider.baseUrl !== "https://owned-terminal.test/v1") { return actualResolveAdapter(provider, cacheRetention); } const key = provider.apiKey ?? ""; @@ -86,7 +91,11 @@ mock.module("../../src/server/adapter-resolve", () => ({ name: "anthropic", buildRequest(parsed: OcxParsedRequest) { const continuation = parsed._providerContinuation?.kiro?.conversationId; - builds.push({ key, ...(continuation ? { continuation } : {}) }); + builds.push({ + key, + ...(continuation ? { continuation } : {}), + ...(parsed._stripReasoningEncryptedContent ? { stripReasoning: true as const } : {}), + }); return { url: "https://owned-terminal.test/v1/messages", method: "POST", @@ -121,6 +130,7 @@ describe("terminal continuation provider-owner rotation", () => { // Take the writer lease after this case installs its home so direct handler dispatch can open the spend journal. releaseSpendHome = acquireOwnedSpendHome(); builds = []; + clearAnthropicAccountPoolState(); clearKeyCooldowns(); clearResponseStateForTests(); }); @@ -130,6 +140,7 @@ describe("terminal continuation provider-owner rotation", () => { releaseSpendHome?.(); releaseSpendHome = undefined; globalThis.fetch = originalFetch; + clearAnthropicAccountPoolState(); if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; clearKeyCooldowns(); @@ -250,4 +261,93 @@ describe("terminal continuation provider-owner rotation", () => { ]); expect(phases).toEqual([]); }); + + test.each([ + ["initial send", false], + ["terminal continuation", true], + ] as const)("Anthropic pool rotation on the %s rebinds the replay owner to the rotated account", async (_label, continuation) => { + for (let index = 0; index < 2; index += 1) { + await saveCredential("anthropic", { + access: `anthropic-access-${index}`, + refresh: `anthropic-refresh-${index}`, + expires: Date.now() + 3_600_000, + accountId: `anthropic-account-${index}`, + }, { addAccount: true }); + } + const [idB, idA] = getAccountSet("anthropic")?.accounts.map(account => account.id) ?? []; + expect(await setActiveAccount("anthropic", idA!)).toBe(true); + const ownerA = reasoningReplayOAuthCredentialIdentity(await getValidAccessSnapshotForAccount("anthropic", idA!)); + const ownerB = reasoningReplayOAuthCredentialIdentity(await getValidAccessSnapshotForAccount("anthropic", idB!)); + expect(ownerB).toBeDefined(); + expect(ownerB).not.toBe(ownerA); + const config: OcxConfig = { + port: 0, + defaultProvider: "anthropic", + providers: { + anthropic: { + adapter: "test-terminal-owned", + baseUrl: "https://owned-terminal.test/v1", + authMode: "oauth", + models: ["model"], + }, + }, + } as OcxConfig; + saveConfig(config); + + const phases = ["seed", ...(continuation ? ["plan"] : []), "rate-limit", "rotated", "follow"]; + const seenAuthorization: string[] = []; + globalThis.fetch = (async (_input, init) => { + seenAuthorization.push(new Headers(init?.headers).get("authorization") ?? ""); + const phase = phases.shift(); + if (!phase) throw new Error("unexpected extra upstream request"); + if (phase === "rate-limit") { + return Response.json({ error: { message: "rotate" } }, { status: 429, headers: { "retry-after": "30" } }); + } + return new Response("", { headers: { "x-test-phase": phase } }); + }) as typeof fetch; + // A stable thread binds the reasoning-replay serving identity across turns. + const post = (body: Record) => handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json", "thread-id": `thread-anthropic-rotation-${continuation}` }, + body: JSON.stringify({ model: "anthropic/model", stream: false, store: true, ...body }), + }), + config, + { model: "", provider: "" }, + ); + + const seed = await post({ input: "seed" }); + expect(seed.status).toBe(200); + const seedJson = await seed.json() as { id: string }; + expect(previousResponseProviderState(seedJson.id)).toMatchObject({ __ocxOwner: { credentialIdentity: ownerA } }); + + const rotated = await post({ + previous_response_id: seedJson.id, + input: continuation ? "Please modify the file now" : "hello", + tools: [{ type: "function", name: "read_file", description: "read a file", parameters: { type: "object" } }], + }); + expect(rotated.status).toBe(200); + const rotatedJson = await rotated.json() as { id: string }; + // The turn was served by B, so B owns the state persisted for the next turn. + expect(previousResponseProviderState(rotatedJson.id)).toMatchObject({ + __ocxOwner: { credentialIdentity: ownerB }, + kiro: { conversationId: "private-b" }, + }); + + const follow = await post({ previous_response_id: rotatedJson.id, input: "follow up" }); + expect(follow.status).toBe(200); + const a = "anthropic-access-1"; + const b = "anthropic-access-0"; + expect(seenAuthorization).toEqual([a, ...(continuation ? [a] : []), a, b, b].map(key => `Bearer ${key}`)); + // B never receives A's continuation or reasoning, and the follow-up on B keeps B's + // continuation and its reasoning replay instead of stripping it as foreign. + expect(builds).toEqual([ + { key: a }, + ...(continuation ? [{ key: a, continuation: "private-a" }] : []), + { key: a, continuation: "private-a" }, + { key: b, stripReasoning: true }, + { key: b, continuation: "private-b" }, + ]); + expect(phases).toEqual([]); + }); });