diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 5f3fc99649..84c65c6195 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -949,6 +949,17 @@ receive the default only when the setting is absent; custom renamed entries keep value and do not acquire this default by destination matching. Chat model routes keep their existing protocol. The stateless flag does not force Responses streaming into JSON. +## OpenCode Go session affinity + +Requests routed to OpenCode Go destinations carry session affinity via the `x-opencode-session` header: + +- Operator configuration: if the provider configuration specifies `x-opencode-session`, that value is preserved verbatim. +- Client request header: if an incoming request provides `x-opencode-session`, it is treated as client session identity and derived into a canonical `ocx_` session id. +- Real conversation identity: when an incoming request carries conversation metadata, `conversation_id`, or `parent_message_id`, OpenCodex derives a stable conversation-scoped session lane. +- Sessionless requests: requests without conversation identity (such as capability probes or standalone requests) receive an isolated, request-scoped ephemeral session lane allocated once per request lifecycle. This ephemeral identity remains stable across route retries, policy fallback attempts, and internal request fanout (such as Claude translation or compaction), preventing missing-header 400 errors while avoiding session collision between concurrent requests. + +Non-Go destinations remain unaffected and do not receive the session header. + ## OpenCode Go reasoning efforts Go catalog rows preserve their configured reasoning efforts exactly, including during diff --git a/src/providers/opencode-go-transport.ts b/src/providers/opencode-go-transport.ts index a863d24f37..44260e8790 100644 --- a/src/providers/opencode-go-transport.ts +++ b/src/providers/opencode-go-transport.ts @@ -1,4 +1,4 @@ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import type { OcxProviderConfig } from "../types"; import { registryEntryForProviderDestination } from "./registry"; @@ -22,20 +22,27 @@ export function deriveOpenCodeGoSessionId(sessionLane: string): string { return `ocx_${digest}`; } -/** Add per-conversation Go affinity only to the canonical fixed-key destination. */ +/** + * Add per-conversation Go affinity only to the canonical fixed-key destination. + * + * When an explicit or WeakMap-allocated session lane is provided, it is hashed into + * a stable session id. If an unlinked caller passes undefined, randomUUID() serves + * as a standalone fallback to satisfy Console Go header requirements without asserting + * cross-request stability. + */ export function resolveOpenCodeGoTransport( provider: T, sessionLane: string | undefined, ): T { if (registryEntryForProviderDestination(provider)?.id !== "opencode-go") return provider; - if (!sessionLane) return provider; + const effectiveLane = sessionLane || randomUUID(); if (hasHeaderCaseInsensitive(provider.headers, OPENCODE_GO_SESSION_HEADER)) return provider; return { ...provider, headers: { ...(provider.headers ?? {}), - [OPENCODE_GO_SESSION_HEADER]: deriveOpenCodeGoSessionId(sessionLane), + [OPENCODE_GO_SESSION_HEADER]: deriveOpenCodeGoSessionId(effectiveLane), }, }; } diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 20c29856ec..f8bae6bcfd 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -26,7 +26,7 @@ import { NoEligiblePolicyCandidateError, UnknownRoutingPolicyError, routeModel } import { evidenceFromBody } from "../routing/request-evidence"; import { resolveWireProtocolOverride } from "./adapter-resolve"; import { resolveOpenCodeGoTransport } from "../providers/opencode-go-transport"; -import { normalizeLogConversationId, sessionLaneIdFromRequest } from "./request-log-conversation"; +import { getOrAllocateRequestSessionLane, linkRequestSessionLane, normalizeLogConversationId, sessionLaneIdFromRequest } from "./request-log-conversation"; import type { OcxConfig } from "../types"; import { readJsonRequestBody, resolveInboundBodyLimitBytes } from "./request-decompress"; import { @@ -143,7 +143,7 @@ async function handleChatCompletionsWithBudget( try { const route = routeModel(config, chatBody.model as string, evidenceFromBody(chatBody)); route.provider = resolveOpenCodeGoTransport(route.provider, - sessionLaneIdFromRequest(req.headers) ?? normalizeLogConversationId(req.headers.get("x-opencode-session"))); + getOrAllocateRequestSessionLane(req)); // Settle the wire once so every branch below reads the adapter this model will // actually use, not the provider-wide default (#404). route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, "chat"); @@ -305,6 +305,7 @@ async function handleChatCompletionsWithBudget( headers, body: internalBodyJson, }); + linkRequestSessionLane(req, internalReq); let nativeLogged = false; const finalizeNativeLog = (status: number, meta: { terminalStatus?: RequestLogEntry["terminalStatus"]; closeReason: "terminal" | "client_cancel" | "non_stream" }) => { diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index a16466459b..fa3b5079cd 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -36,7 +36,7 @@ import { resolveWireProtocolOverride } from "./adapter-resolve"; import type { OcxConfig } from "../types"; import { readJsonRequestBody, resolveInboundBodyLimitBytes } from "./request-decompress"; import { addFinalRequestLog, httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext, type RequestLogEntry } from "./request-log"; -import { conversationIdFromClaudeMetadata, normalizeLogConversationId, sessionLaneIdFromRequest } from "./request-log-conversation"; +import { conversationIdFromClaudeMetadata, linkRequestSessionLane, normalizeLogConversationId, sessionLaneIdFromRequest } from "./request-log-conversation"; import { responseWithDeferredRequestLog } from "./relay"; import { handleResponses } from "./responses"; import { @@ -897,6 +897,7 @@ async function handleClaudeMessagesWithBudget( headers, body: JSON.stringify(internalBody), }); + linkRequestSessionLane(req, internalReq); } finally { reservation.release(); } diff --git a/src/server/request-log-conversation.ts b/src/server/request-log-conversation.ts index ed1e2f33d4..326b326526 100644 --- a/src/server/request-log-conversation.ts +++ b/src/server/request-log-conversation.ts @@ -2,7 +2,7 @@ * Best-effort chat/session correlation for Logs / usage.jsonl (#330). * Opaque ids only — never persist raw emails or Claude Desktop system-hash fallbacks. */ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; /** Reject absurdly long client strings before hashing (DoS / JSONL bloat). */ export const LOG_CONVERSATION_ID_INPUT_MAX = 4096; @@ -217,3 +217,32 @@ export function summarizeConversationLogs(entries: readonly TotalsSource[]): Con unmeteredRequests, }; } + +const requestAllocatedSessionLanes = new WeakMap(); + +/** + * Link session lane identity from a source Request to an internal/child Request. + * Preserves ephemeral allocated lanes across internal request translation/fanout. + */ +export function linkRequestSessionLane(sourceReq: Request, targetReq: Request): void { + const lane = getOrAllocateRequestSessionLane(sourceReq); + requestAllocatedSessionLanes.set(targetReq, lane); +} + +/** + * Resolve or allocate a request-scoped session lane identity. + * If the request has an explicit session lane (headers or x-opencode-session), use it. + * Otherwise, allocates an ephemeral UUID once per admitted request, retained across retries. + */ +export function getOrAllocateRequestSessionLane(req: Request): string { + const explicit = sessionLaneIdFromRequest(req.headers) + ?? normalizeLogConversationId(req.headers.get("x-opencode-session")); + if (explicit) return explicit; + + let allocated = requestAllocatedSessionLanes.get(req); + if (!allocated) { + allocated = randomUUID(); + requestAllocatedSessionLanes.set(req, allocated); + } + return allocated; +} diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 3c345a9f1d..0ee04b5f14 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -157,7 +157,7 @@ import { } from "./core"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers"; import { mapCodexAuthContextErrorToResponse, nativeMainRefreshFailureResponse } from "./codex-auth-error"; -import { sessionLaneIdFromRequest } from "../request-log-conversation"; +import { linkRequestSessionLane, sessionLaneIdFromRequest } from "../request-log-conversation"; import { recallComboForLane } from "./combo-session-recall"; export const COMPACT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; @@ -1101,6 +1101,7 @@ export async function handleResponsesCompact( body: JSON.stringify({ ...raw, model: fallbackModel }), signal: req.signal, }); + linkRequestSessionLane(req, fallbackReq); try { const fallback = await handleResponsesCompact( fallbackReq, @@ -1149,6 +1150,7 @@ export async function handleResponsesCompact( headers: internalHeaders, body: JSON.stringify(internalBody), }); + linkRequestSessionLane(req, internalReq); const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, ...(admission ? { admission } : {}) }); if (!response.ok) return response; let json: { output?: unknown[]; status?: unknown; error?: unknown }; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index c1ce136ca4..ccbdd006e8 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -322,6 +322,8 @@ import { conversationIdFromResponsesRequest, normalizeLogConversationId, reasoningReplayConversationIdFromResponsesRequest, + getOrAllocateRequestSessionLane, + linkRequestSessionLane, sessionLaneIdFromRequest, sessionIdHeaderFromRequest, } from "../request-log-conversation"; @@ -2452,7 +2454,7 @@ async function applyFinalRouteRequestNormalization(args: { // Settle the wire once so logging, fast-mode, auth, and sidecars read the adapter // this request will actually use (#404). route.provider = resolveOpenCodeGoTransport(route.provider, - sessionLaneIdFromRequest(req.headers) ?? normalizeLogConversationId(req.headers.get("x-opencode-session"))); + getOrAllocateRequestSessionLane(req)); route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire); if (preserveAnthropicResponseModel) parsed._responseModelId = responseModelId; logCtx.model = route.modelId; @@ -2816,6 +2818,7 @@ export async function handleComboResponses( headers: childHeaders, body: JSON.stringify(childBody), }); + linkRequestSessionLane(req, childRequest); let resolvedAuth: CodexAuthContext | undefined; let terminalRecorder: ((status: ResponsesTerminalStatus, httpStatusOverride?: number) => void) | undefined; const started = Date.now(); diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 13c90bdcee..cb91134b3d 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -8,6 +8,7 @@ import { handleResponses as handleResponsesCore } from "./core"; import { requestPacingOverloadResponse } from "./pacing-overload"; import { captureExplicitOpenAiCallerAuth } from "../../providers/openai-sidecar"; import { captureCallerDirectAuth } from "../../providers/caller-authorization"; +import { linkRequestSessionLane } from "../request-log-conversation"; type CoreHandler = typeof handleResponsesCore; type CoreOptions = Parameters[3]; @@ -56,12 +57,14 @@ function requestWithCandidate( headers.delete("content-encoding"); headers.delete("content-length"); headers.set("content-type", "application/json"); - return new Request(req.url, { + const retryRequest = new Request(req.url, { method: req.method, headers, body: JSON.stringify({ ...rawBody, model: `${candidate.provider}/${candidate.model}` }), signal: req.signal, }); + linkRequestSessionLane(req, retryRequest); + return retryRequest; } function errorCodeFromText(text: string): string | undefined { diff --git a/tests/providers/opencode-go-session-header.test.ts b/tests/providers/opencode-go-session-header.test.ts index c176dc703e..3a2cb23abb 100644 --- a/tests/providers/opencode-go-session-header.test.ts +++ b/tests/providers/opencode-go-session-header.test.ts @@ -1,10 +1,14 @@ import { afterEach, describe, expect, test } from "bun:test"; import { providerConfigSeed } from "../../src/providers/derive"; import { resolveOpenCodeGoTransport } from "../../src/providers/opencode-go-transport"; +import { getOrAllocateRequestSessionLane, linkRequestSessionLane } from "../../src/server/request-log-conversation"; import { getProviderRegistryEntry } from "../../src/providers/registry"; import { handleResponses } from "../../src/server/responses/core"; import { handleChatCompletions } from "../../src/server/chat-completions"; import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { handleResponsesWithPolicyFallback } from "../../src/server/responses/policy-fallback"; +import type { RequestLogContext } from "../../src/server/request-log"; +import type { RouteDecisionTraceV1 } from "../../src/routing/trace"; import type { OcxConfig, OcxProviderConfig } from "../../src/types"; const MUSE_MODEL = "muse-spark-1.3-contributor"; @@ -138,7 +142,8 @@ describe("OpenCode Go session affinity (#3344)", () => { const metadata = await captureRequest({ ...input, metadataUserId: "user_test_account__session_conversation-a" }); const desktop = await captureRequest(input); expect(metadata.headers.get(SESSION_HEADER)).toBe("ocx_a89540229ef781fd5f7adf92a711b436"); - expect(desktop.headers.has(SESSION_HEADER)).toBe(false); + expect(desktop.headers.get(SESSION_HEADER)).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(desktop.headers.get(SESSION_HEADER)).not.toBe("ocx_a89540229ef781fd5f7adf92a711b436"); }); test("Claude explicit Go header precedes metadata and matches native Chat affinity", async () => { @@ -185,11 +190,11 @@ describe("OpenCode Go session affinity (#3344)", () => { } }); - test(`Claude ${model} omits Go affinity without usable metadata identity`, async () => { + test(`Claude ${model} assigns isolated Go affinity without usable metadata identity`, async () => { for (const metadataUserId of [undefined, "", " \t\n ", "invalid\u0000identity", "x".repeat(4097)]) { const captured = await captureRequest({ claude: true, model, metadataUserId }); expect(captured.url).toBe(url); - expect(captured.headers.has(SESSION_HEADER)).toBe(false); + expect(captured.headers.get(SESSION_HEADER)).toMatch(/^ocx_[0-9a-f]{32}$/); expect(captured.headers.has("session_id")).toBe(false); } }); @@ -351,12 +356,175 @@ describe("OpenCode Go session affinity (#3344)", () => { expect([...captured.headers.keys()].filter(name => name === SESSION_HEADER)).toHaveLength(1); }); - test("keeps generated affinity runtime-only and omits it without a stable lane", async () => { + test("assigns isolated affinity without a stable lane and keeps configured headers untouched", async () => { const configured = opencodeGo(); await captureRequest({ provider: configured }); expect(configured.headers?.[SESSION_HEADER]).toBeUndefined(); - expect(resolveOpenCodeGoTransport(configured, undefined)).toBe(configured); - expect(resolveOpenCodeGoTransport(configured, undefined).headers?.[SESSION_HEADER]).toBeUndefined(); + const first = resolveOpenCodeGoTransport(configured, undefined); + const second = resolveOpenCodeGoTransport(configured, undefined); + expect(first.headers?.[SESSION_HEADER]).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(second.headers?.[SESSION_HEADER]).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(first.headers?.[SESSION_HEADER]).not.toBe(second.headers?.[SESSION_HEADER]); + expect(configured.headers?.[SESSION_HEADER]).toBeUndefined(); + }); + + test("two independent sessionless Claude requests receive distinct Go affinities", async () => { + const first = await captureRequest({ claude: true, model: CHAT_MODEL }); + const second = await captureRequest({ claude: true, model: CHAT_MODEL }); + const firstSession = first.headers.get(SESSION_HEADER); + const secondSession = second.headers.get(SESSION_HEADER); + expect(firstSession).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(secondSession).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(firstSession).not.toBe(secondSession); + }); + + test("two independent sessionless native Chat requests receive distinct Go affinities", async () => { + const headers = { "content-type": "application/json" }; + const first = await captureRequest({ nativeChat: true, model: "omen-alpha", headers }); + const second = await captureRequest({ nativeChat: true, model: "omen-alpha", headers }); + const firstSession = first.headers.get(SESSION_HEADER); + const secondSession = second.headers.get(SESSION_HEADER); + expect(firstSession).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(secondSession).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(firstSession).not.toBe(secondSession); + }); + + test("the same admitted request retains the same Go affinity across route normalization and retries", () => { + const configured = opencodeGo(); + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + }); + const lane1 = getOrAllocateRequestSessionLane(req); + const lane2 = getOrAllocateRequestSessionLane(req); + expect(lane1).toBe(lane2); + + const transport1 = resolveOpenCodeGoTransport(configured, lane1); + const transport2 = resolveOpenCodeGoTransport(configured, lane2); + expect(transport1.headers?.[SESSION_HEADER]).toBe(transport2.headers?.[SESSION_HEADER]); + }); + + test("linking an admitted request forwards the allocated fallback lane to internal replay requests", () => { + const sourceReq = new Request("http://localhost/v1/messages", { + method: "POST", + headers: { "content-type": "application/json" }, + }); + const targetReq = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + }); + linkRequestSessionLane(sourceReq, targetReq); + const sourceLane = getOrAllocateRequestSessionLane(sourceReq); + const targetLane = getOrAllocateRequestSessionLane(targetReq); + expect(sourceLane).toBe(targetLane); + }); + + test("handleResponsesWithPolicyFallback retains the same Go affinity across retried candidate attempts (#4172)", async () => { + const trace: RouteDecisionTraceV1 = { + version: 1, + decisionId: "decision-policy-go", + createdAt: 1, + requestedModel: "policy/go-test", + routeKind: "policy", + profile: { id: "go-test", revision: "rev-1" }, + requirements: [], + candidates: [ + { provider: "opencode-go", model: MUSE_MODEL, eligible: true, exclusions: [], score: { total: 0.9, components: {} } }, + { provider: "opencode-go", model: CHAT_MODEL, eligible: true, exclusions: [], score: { total: 0.8, components: {} } }, + ], + selected: { candidateIndex: 0, provider: "opencode-go", model: MUSE_MODEL, reason: "highest-score" }, + }; + + const capturedHeaders: string[] = []; + const capturedUrls: string[] = []; + globalThis.fetch = (async (requestInput: RequestInfo | URL, init?: RequestInit) => { + const url = String(requestInput); + capturedUrls.push(url); + const headers = new Headers(init?.headers); + const session = headers.get(SESSION_HEADER); + if (session) capturedHeaders.push(session); + if (url.endsWith("/responses")) { + return Response.json({ error: { message: "rate limited upstream" } }, { status: 429 }); + } + return upstreamResponse(url); + }) as typeof fetch; + + const initialReq = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "policy/go-test", input: "ping", stream: false }), + }); + + const config = { + providers: { "opencode-go": opencodeGo() }, + } as unknown as OcxConfig; + + const logCtx = { model: "", provider: "" } as RequestLogContext; + const response = await handleResponsesWithPolicyFallback( + initialReq, + config, + logCtx, + { inboundWire: "responses" }, + { + runCore: async (candidateReq, cfg, ctx, options) => { + const body = await candidateReq.clone().json() as { model: string }; + ctx.routeDecision = trace; + // For the initial policy request, forward as candidate 0 + const forwardReq = body.model === "policy/go-test" + ? new Request(candidateReq.url, { + method: candidateReq.method, + headers: candidateReq.headers, + body: JSON.stringify({ ...body, model: `opencode-go/${MUSE_MODEL}` }), + signal: candidateReq.signal, + }) + : candidateReq; + linkRequestSessionLane(candidateReq, forwardReq); + const res = await handleResponses( + forwardReq, + cfg, + ctx, + { ...options, comboAttempt: true }, + ); + ctx.routeDecision = trace; + return res; + }, + }, + ); + + expect(response.status).toBe(200); + expect(capturedUrls).toHaveLength(2); + expect(capturedUrls[0]).toContain("/responses"); + expect(capturedUrls[1]).toContain("/chat/completions"); + expect(capturedHeaders).toHaveLength(2); + expect(capturedHeaders[0]).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(capturedHeaders[1]).toBe(capturedHeaders[0]); + + // An independent sessionless request receives a distinct Go affinity + const independentCaptured: string[] = []; + globalThis.fetch = (async (requestInput: RequestInfo | URL, init?: RequestInit) => { + const headers = new Headers(init?.headers); + const session = headers.get(SESSION_HEADER); + if (session) independentCaptured.push(session); + return upstreamResponse(String(requestInput)); + }) as typeof fetch; + + const secondReq = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "opencode-go/muse-spark-1.3-contributor", input: "ping", stream: false }), + }); + + const secondResponse = await handleResponses( + secondReq, + config, + { model: "", provider: "" } as RequestLogContext, + { inboundWire: "responses" }, + ); + + expect(secondResponse.status).toBe(200); + expect(independentCaptured).toHaveLength(1); + expect(independentCaptured[0]).toMatch(/^ocx_[0-9a-f]{32}$/); + expect(independentCaptured[0]).not.toBe(capturedHeaders[0]); }); test("does not inject the header into a lookalike destination", async () => { diff --git a/tests/routing/routing-policy-fallback.test.ts b/tests/routing/routing-policy-fallback.test.ts index 1883c1277d..7a16a22aef 100644 --- a/tests/routing/routing-policy-fallback.test.ts +++ b/tests/routing/routing-policy-fallback.test.ts @@ -10,6 +10,7 @@ import { handleResponsesWithPolicyFallback, rankPolicyFallbackCandidates, } from "../../src/server/responses/policy-fallback"; +import { getOrAllocateRequestSessionLane } from "../../src/server/request-log-conversation"; function policyTrace(): RouteDecisionTraceV1 { return { @@ -48,6 +49,33 @@ function seedAttempt(logCtx: RequestLogContext, provider: string, model: string) } describe("policy candidate fallback", () => { + test("carries allocated request session lane across candidate retry attempts (#4172)", async () => { + const initial = request(); + const initialLane = getOrAllocateRequestSessionLane(initial); + const observedLanes: string[] = []; + + const response = await handleResponsesWithPolicyFallback( + initial, + { port: 0, defaultProvider: "provider-a", providers: {} } as unknown as OcxConfig, + { model: "", provider: "" } as RequestLogContext, + {}, + { + runCore: async (req, _config, context) => { + observedLanes.push(getOrAllocateRequestSessionLane(req)); + context.routeDecision = policyTrace(); + return observedLanes.length === 1 + ? Response.json({ error: { message: "transient error" } }, { status: 503 }) + : Response.json({ status: "completed" }); + }, + }, + ); + + expect(response.status).toBe(200); + expect(observedLanes).toHaveLength(2); + expect(observedLanes[0]).toBe(initialLane); + expect(observedLanes[1]).toBe(initialLane); + }); + test("policy hops retain only the original sidecar snapshot outside primary headers", async () => { const authorization = `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "sidecar-account" })}`; const initial = request();