diff --git a/devlog/_plan/260906_a_runtime_stack/041_affinity_refresh.md b/devlog/_plan/260906_a_runtime_stack/041_affinity_refresh.md new file mode 100644 index 0000000000..38515e7742 --- /dev/null +++ b/devlog/_plan/260906_a_runtime_stack/041_affinity_refresh.md @@ -0,0 +1,5 @@ +# Affinity layer P refresh + +Consume 040 on prepared recovery parent332a30e6d. Original #3581 remains f60397d3408e0339ffc66acdcaca8133e40866c2, with SB Yoon attribution preserved. Retain new recovery cache/history logic and termination WeakMap rebind when applying the two core hunks. The new cohort flag must survive initial parse and both fresh/cache-only reparse; true/undefined never authorize cache-key-based session identity. No changes to OAuth command-code cache-key forwarding; enable the existing API-key commandcode registry capability only. + +Scoped regression worker after carry owns tests/helpers/agent-task-recovery.ts, tests/server/server-agent-task-recovery-replay.test.ts and tests/providers/command-code-provider.test.ts. Use the actual ADAPTER_REGISTRY openai-chat create seam already proven in the parent regression to observe parsed fields at real buildRequest. Main owns production and adapters documentation. Remote helper asserts project Bun1.4.0; no local suites/typecheck/build. Full exact-head CI and --admin integration remain final gates. diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 2848fed564..3760ad2869 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -148,6 +148,19 @@ of the HTTP retry loop. ChatGPT account id, and the OpenAI beta/originator/session headers. This is the ChatGPT-login path that also powers the [sidecars](/guides/sidecars/). +## Command Code session affinity + +The OAuth `command-code` adapter derives an opaque `x-session-id` from the client +thread identity, then the reasoning-replay conversation identity. When neither is +available, it uses a prompt-cache key only if the integration has explicitly +classified that key as belonging to one conversation. Shared or unclassified cache +keys do not establish session affinity; requests without a usable identity receive +a fresh session ID. Recovery and cached-history replay preserve this classification. + +The API-key `commandcode` provider uses the `openai-chat` adapter and supports +forwarding `prompt_cache_key`. This is separate from the OAuth adapter's session +header and does not guarantee a provider cache hit. + ## `anthropic` **Targets:** Anthropic **Messages** (`/v1/messages`). diff --git a/src/adapters/command-code.ts b/src/adapters/command-code.ts index a9b429bc99..c20dc88be6 100644 --- a/src/adapters/command-code.ts +++ b/src/adapters/command-code.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import { opendir } from "node:fs/promises"; @@ -213,6 +213,27 @@ function projectSlug(cwd: string): string { return cwd.replace(/[^a-zA-Z0-9]+/g, "-").replace(/^-|-$/g, "").toLowerCase().slice(0, 64) || "workspace"; } +export function commandCodeSessionId(parsed: OcxParsedRequest): string { + // Shared prompt-cache cohorts identify a cache population, not one conversation. Using one + // for session affinity would pin unrelated conversations to the same upstream worker. + const threadId = parsed._clientThreadId?.trim(); + const replayId = parsed._reasoningReplayScope?.clientThreadId?.trim(); + const cacheKey = parsed._promptCacheKeyIsSharedCohort === false + ? parsed.options.promptCacheKey?.trim() + : undefined; + const identity = threadId + ? ["thread", threadId] + : replayId + ? ["replay", replayId] + : cacheKey + ? ["cache", cacheKey] + : undefined; + if (!identity) return randomUUID(); + const hex = createHash("sha256").update(`command-code:${identity[0]}\0${identity[1]}`).digest("hex"); + // Replace the digest nibbles at the UUID version and variant positions; the skipped hex characters are intentional. + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-4${hex.slice(13, 16)}-8${hex.slice(17, 20)}-${hex.slice(20, 32)}`; +} + interface GitWorkspaceInfo { isGitRepo: boolean; currentBranch: string; @@ -525,7 +546,7 @@ export function createCommandCodeAdapter(provider: OcxProviderConfig): ProviderA "x-cli-environment": "production", "x-taste-learning": "false", "x-co-flag": "false", - "x-session-id": randomUUID(), + "x-session-id": commandCodeSessionId(parsed), }; if (cwd) headers["x-project-slug"] = projectSlug(cwd); return { diff --git a/src/providers/registry.ts b/src/providers/registry.ts index 64c01dbb10..82f70a6a6a 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -2175,6 +2175,7 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [ liveModels: true, preserveCustomDestination: true, defaultModel: "deepseek/deepseek-v4-flash", + promptCacheKey: true, // The default is also the cold-start seed: live discovery failure must not empty the catalog // for a freshly configured provider with no stale cache (issue #308 pattern). models: ["deepseek/deepseek-v4-flash"], diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 0471617da0..fc0eaae543 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2953,6 +2953,7 @@ async function handleResponsesInner( let toolBridgeMaps: ReturnType; try { parsed = parseRequest(body); + parsed._promptCacheKeyIsSharedCohort = options.promptCacheKeyIsSharedCohort; // Captured before any parser mutates it, so both grammars see the client's id. const { fastRow, effortRow } = parseSyntheticRowId(parsed.modelId, config); if (fastRow) { @@ -3271,6 +3272,7 @@ async function handleResponsesInner( "_providerContinuationOwner", "_cursorConversationId", "_clientThreadId", + "_promptCacheKeyIsSharedCohort", "_cursorClientThreadId", "_reasoningReplayScope", "_cursorIsolateConversation", diff --git a/src/types/request.ts b/src/types/request.ts index ffee4eb8a3..1c6a5294da 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -68,6 +68,8 @@ export interface OcxParsedRequest { _cursorConversationId?: string; /** Stable upstream client thread identity, used only to derive provider-scoped continuation ids. */ _clientThreadId?: string; + /** True when promptCacheKey identifies a shared cache cohort rather than one conversation. */ + _promptCacheKeyIsSharedCohort?: boolean; /** Cursor-only thread owner; may be an opaque process-local Desktop session/thread identity. */ _cursorClientThreadId?: string; /** Conversation/provider/account/model-bound namespace for reasoning replay state. */ diff --git a/tests/claude-integration/claude-code-thought-signature-scope.test.ts b/tests/claude-integration/claude-code-thought-signature-scope.test.ts index 2437a8d157..eb544dce97 100644 --- a/tests/claude-integration/claude-code-thought-signature-scope.test.ts +++ b/tests/claude-integration/claude-code-thought-signature-scope.test.ts @@ -97,16 +97,19 @@ describe("Claude Code Anthropic inbound reasoning-replay scope", () => { const parsed = await drive({ promptCacheKey: "session-key-123", promptCacheKeyIsSharedCohort: false }); expect(parsed._clientThreadId).toBeUndefined(); expect(parsed._reasoningReplayScope?.clientThreadId).toBe("session-key-123"); + expect(parsed._promptCacheKeyIsSharedCohort).toBe(false); }); test("the shared Desktop prompt_cache_key cohort does not create a scope", async () => { const parsed = await drive({ promptCacheKey: "shared-cohort-key", promptCacheKeyIsSharedCohort: true }); expect(parsed._reasoningReplayScope).toBeUndefined(); + expect(parsed._promptCacheKeyIsSharedCohort).toBe(true); }); test("an Anthropic replay without prompt_cache_key does not create a scope", async () => { const parsed = await drive({}); expect(parsed._reasoningReplayScope).toBeUndefined(); + expect(parsed._promptCacheKeyIsSharedCohort).toBeUndefined(); }); test("an overlong prompt_cache_key is hashed, not stored raw", async () => { diff --git a/tests/helpers/agent-task-recovery.ts b/tests/helpers/agent-task-recovery.ts index bf8a173f70..4a6a95c5ae 100644 --- a/tests/helpers/agent-task-recovery.ts +++ b/tests/helpers/agent-task-recovery.ts @@ -147,7 +147,7 @@ export async function post( input: unknown[], headers: HeadersInit = {}, abortSignal?: AbortSignal, - options: { tools?: unknown[]; translatorBudget?: TranslatorBudget } = {}, + options: { tools?: unknown[]; translatorBudget?: TranslatorBudget; promptCacheKeyIsSharedCohort?: boolean } = {}, ): Promise { return handleResponses(new Request("http://localhost/v1/responses", { method: "POST", @@ -156,7 +156,11 @@ export async function post( ...Object.fromEntries(new Headers(headers)), }, body: JSON.stringify({ model, input, stream: false, ...(options.tools ? { tools: options.tools } : {}) }), - }), config, { model: "", provider: "" }, { abortSignal, translatorBudget: options.translatorBudget }); + }), config, { model: "", provider: "" }, { + abortSignal, + translatorBudget: options.translatorBudget, + promptCacheKeyIsSharedCohort: options.promptCacheKeyIsSharedCohort, + }); } export function encryptedInput(options: { diff --git a/tests/providers/command-code-provider.test.ts b/tests/providers/command-code-provider.test.ts index 3e369a9c6b..a3b81e408d 100644 --- a/tests/providers/command-code-provider.test.ts +++ b/tests/providers/command-code-provider.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { createCommandCodeAdapter } from "../../src/adapters/command-code"; +import { commandCodeSessionId, createCommandCodeAdapter } from "../../src/adapters/command-code"; import { loginCommandCode, parseCommandCodeCallback, shouldImportLocalCommandCodeAuth } from "../../src/oauth/command-code"; import { buildModelsRequest, OAUTH_PROVIDERS } from "../../src/oauth"; import { @@ -796,4 +796,138 @@ describe("Command Code provider", () => { const built = await builtRequest({ ...parsed(), stream: false }); expect(JSON.parse(built.body).params.stream).toBe(true); }); + + test("derives an opaque stable session id from trusted conversation identity", async () => { + const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-8[0-9a-f]{3}-[0-9a-f]{12}$/i; + const identities = { + thread: "thread-secret-value", + replay: "replay-secret-value", + cache: "cache-secret-value", + }; + const thread = { + ...parsed(), + _clientThreadId: ` ${identities.thread} `, + _reasoningReplayScope: { clientThreadId: identities.replay }, + options: { ...parsed().options, promptCacheKey: identities.cache }, + }; + const sameThread = { + ...thread, + _reasoningReplayScope: { clientThreadId: "different-replay" }, + options: { ...thread.options, promptCacheKey: "different-cache" }, + }; + const replay = { + ...parsed(), + _reasoningReplayScope: { clientThreadId: identities.replay }, + options: { ...parsed().options, promptCacheKey: identities.cache }, + }; + const sameReplay = { ...replay, options: { ...replay.options, promptCacheKey: "different-cache" } }; + const cache = { + ...parsed(), + options: { ...parsed().options, promptCacheKey: ` ${identities.cache} ` }, + _promptCacheKeyIsSharedCohort: false, + }; + const sameCache = { + ...cache, + options: { ...cache.options, promptCacheKey: identities.cache }, + }; + + const threadId = commandCodeSessionId(thread); + expect(threadId).toBe(commandCodeSessionId(sameThread)); + expect(threadId).not.toBe(commandCodeSessionId({ ...thread, _clientThreadId: "different-thread" })); + expect(commandCodeSessionId(replay)).toBe(commandCodeSessionId(sameReplay)); + expect(commandCodeSessionId(cache)).toBe(commandCodeSessionId(sameCache)); + expect(commandCodeSessionId(replay)).not.toBe(commandCodeSessionId(cache)); + expect(threadId).toMatch(uuid); + expect(commandCodeSessionId(replay)).toMatch(uuid); + expect(commandCodeSessionId(cache)).toMatch(uuid); + for (const raw of Object.values(identities)) expect(threadId).not.toContain(raw); + + const built = await builtRequest(thread); + expect(built.headers["x-session-id"]).toBe(threadId); + }); + + test("whitespace thread and replay identities fall through to the next trusted identity at the wire", async () => { + const replay: OcxParsedRequest = { + ...parsed(), + _clientThreadId: " \t\n ", + _reasoningReplayScope: { clientThreadId: " replay-after-blank-thread " }, + _promptCacheKeyIsSharedCohort: false, + options: { ...parsed().options, promptCacheKey: "distinct-cache-fallback" }, + }; + const cache: OcxParsedRequest = { + ...replay, + _reasoningReplayScope: { clientThreadId: " \t\n " }, + options: { ...parsed().options, promptCacheKey: " cache-after-blank-replay " }, + }; + const cleanReplay: OcxParsedRequest = { + ...parsed(), + _reasoningReplayScope: { clientThreadId: "replay-after-blank-thread" }, + }; + const cleanCache: OcxParsedRequest = { + ...parsed(), + _promptCacheKeyIsSharedCohort: false, + options: { ...parsed().options, promptCacheKey: "cache-after-blank-replay" }, + }; + const cases: Array<[OcxParsedRequest, OcxParsedRequest]> = [[replay, cleanReplay], [cache, cleanCache]]; + for (const [withWhitespace, clean] of cases) { + const built = await builtRequest(withWhitespace); + const expected = await builtRequest(clean); + expect(built.headers["x-session-id"]).toBe(expected.headers["x-session-id"]); + expect(commandCodeSessionId(withWhitespace)).toBe(built.headers["x-session-id"]); + } + }); + + test("whitespace-only trusted identities produce fresh session headers", async () => { + const blank: OcxParsedRequest = { + ...parsed(), + _clientThreadId: " \t ", + _reasoningReplayScope: { clientThreadId: "\n " }, + _promptCacheKeyIsSharedCohort: false, + options: { ...parsed().options, promptCacheKey: " \t\n " }, + }; + const first = (await builtRequest(blank)).headers["x-session-id"]; + const second = (await builtRequest(blank)).headers["x-session-id"]; + const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + expect(first).toMatch(uuid); + expect(second).toMatch(uuid); + expect(first).not.toBe(second); + }); + + test("the same literal in thread, replay and cache namespaces yields distinct stable session headers", async () => { + const literal = "same-identity-in-every-kind"; + const requests: OcxParsedRequest[] = [ + { ...parsed(), _clientThreadId: literal }, + { ...parsed(), _reasoningReplayScope: { clientThreadId: literal } }, + { + ...parsed(), + _promptCacheKeyIsSharedCohort: false, + options: { ...parsed().options, promptCacheKey: literal }, + }, + ]; + const ids: string[] = []; + for (const request of requests) { + const id = (await builtRequest(request)).headers["x-session-id"]!; + expect(id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-8[0-9a-f]{3}-[0-9a-f]{12}$/i); + expect(id).not.toContain(literal); + expect((await builtRequest(request)).headers["x-session-id"]).toBe(id); + expect(commandCodeSessionId(request)).toBe(id); + ids.push(id); + } + expect(new Set(ids).size).toBe(3); + }); + + test("does not derive affinity from a shared cohort or prompt text", () => { + const shared = { + ...parsed(), + options: { ...parsed().options, promptCacheKey: "shared-cache-key" }, + _promptCacheKeyIsSharedCohort: true, + }; + expect(commandCodeSessionId(shared)).not.toBe(commandCodeSessionId(shared)); + const unclassifiedCache = { + ...parsed(), + options: { ...parsed().options, promptCacheKey: "possibly-shared-cache-key" }, + }; + expect(commandCodeSessionId(unclassifiedCache)).not.toBe(commandCodeSessionId(unclassifiedCache)); + expect(commandCodeSessionId(parsed())).not.toBe(commandCodeSessionId(parsed())); + }); }); diff --git a/tests/providers/commandcode-provider.test.ts b/tests/providers/commandcode-provider.test.ts index e76355dc4f..f70df51093 100644 --- a/tests/providers/commandcode-provider.test.ts +++ b/tests/providers/commandcode-provider.test.ts @@ -67,6 +67,7 @@ describe("Command Code provider", () => { liveModels: true, preserveCustomDestination: true, defaultModel: "deepseek/deepseek-v4-flash", + promptCacheKey: true, apiKeyValidation: "unknown", reasoningEfforts: [], modelReasoningEfforts: { @@ -176,6 +177,20 @@ describe("Command Code provider", () => { expect(body).not.toHaveProperty("parallel_tool_calls"); }); + test("forwards the enabled prompt cache key to chat completions", () => { + const route = routeModel( + commandcodeConfig(), + "commandcode/deepseek/deepseek-v4-flash", + ); + const request = createOpenAIChatAdapter(route.provider).buildRequest({ + modelId: route.modelId, + context: { messages: [{ role: "user", content: "ping", timestamp: 0 }] }, + stream: true, + options: { promptCacheKey: "command-code-session-cache" }, + }); + expect(JSON.parse(String(request.body)).prompt_cache_key).toBe("command-code-session-cache"); + }); + test("discovers the live catalog with context windows and preserves slash ids", async () => { globalThis.fetch = (async (input, init) => { expect(String(input)).toBe("https://api.commandcode.ai/provider/v1/models"); diff --git a/tests/server/server-agent-task-recovery-replay.test.ts b/tests/server/server-agent-task-recovery-replay.test.ts index 42821ac30f..28050c5369 100644 --- a/tests/server/server-agent-task-recovery-replay.test.ts +++ b/tests/server/server-agent-task-recovery-replay.test.ts @@ -73,6 +73,85 @@ function encryptedMessage(): unknown[] { return JSON.parse(JSON.stringify(encryptedInput()).replace("Message Type: NEW_TASK", "Message Type: MESSAGE")); } +test.each([true, false, undefined])("fresh recovery and cache-only reparse preserve cohort marker %s and replay metadata", async (cohort) => { + const { post, providerResponse } = await import("../helpers/agent-task-recovery"); + const parentThread = `affinity-parent-${crypto.randomUUID()}`; + const headers = codexHeaders("acct-caller", { + "x-codex-parent-thread-id": parentThread, + "thread-id": "distinct-child-thread", + session_id: "distinct-session", + }); + const config = routedConfig({ enabled: true }); + let recoveries = 0; + const recoveryBodies: string[] = []; + const providerBodies: string[] = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + const body = String(init?.body); + if (String(url).includes("chatgpt.com")) { + recoveries++; + recoveryBodies.push(body); + return new Response(recoverySse("Read the affinity assignment.")); + } + providerBodies.push(body); + return providerResponse(); + }) as typeof fetch; + + const observations: Array<{ + cohort: boolean | undefined; + thread: string | undefined; + replay: OcxParsedRequest["_reasoningReplayScope"]; + raw: string; + }> = []; + const createChat = ADAPTER_REGISTRY["openai-chat"].create; + const factory = spyOn(ADAPTER_REGISTRY["openai-chat"], "create").mockImplementation((provider, context) => { + const adapter = createChat(provider, context); + return { + ...adapter, + buildRequest(...[parsed, incoming]: Parameters) { + observations.push({ + cohort: parsed._promptCacheKeyIsSharedCohort, + thread: parsed._clientThreadId, + replay: structuredClone(parsed._reasoningReplayScope), + raw: JSON.stringify(parsed._rawBody), + }); + return adapter.buildRequest(parsed, incoming); + }, + }; + }); + try { + const turns = [ + encryptedInput(), + [...encryptedInput(), { type: "message", role: "user", content: "Continue the affinity assignment." }], + ]; + for (const [index, input] of turns.entries()) { + const response = await post(config, "xai/grok-4.5", input, headers, undefined, { + promptCacheKeyIsSharedCohort: cohort, + }); + expect(response.status).toBe(200); + await response.text(); + expect(recoveries).toBe(1); + expect(observations).toHaveLength(index + 1); + expect(providerBodies).toHaveLength(index + 1); + const observed = observations[index]!; + expect(observed.cohort).toBe(cohort); + expect(observed.thread).toBe(parentThread); + expect(observed.replay).toMatchObject({ clientThreadId: parentThread }); + expect(observed.replay).toEqual(observations[0]!.replay); + for (const body of [observed.raw, providerBodies[index]!]) { + expect(body).toContain("Read the affinity assignment."); + expect(body).not.toContain(FERNET_TASK); + expect(body).not.toContain("promptCacheKeyIsSharedCohort"); + } + } + expect(providerBodies[1]).toContain("Continue the affinity assignment."); + expect(recoveryBodies).toHaveLength(1); + expect(recoveryBodies[0]).toContain(FERNET_TASK); + expect(recoveryBodies[0]).not.toContain("promptCacheKeyIsSharedCohort"); + } finally { + factory.mockRestore(); + } +}); + test("MESSAGE recovery reaches the provider and survives tool-result replay", async () => { const { post, providerResponse } = await import("../helpers/agent-task-recovery"); let recoveries = 0;