diff --git a/src/server/responses/combo-session-recall.ts b/src/server/responses/combo-session-recall.ts new file mode 100644 index 0000000000..e7a403484b --- /dev/null +++ b/src/server/responses/combo-session-recall.ts @@ -0,0 +1,71 @@ +/** + * Session-scoped recall of the last successful combo selection (#3891). + * + * When Codex compacts a conversation that was switched to a different combo + * mid-session, it sends the *bare native model* of the new combo's first + * target (e.g. "gpt-5.6-terra") rather than the combo/ selector it + * uses for ordinary turns. Without recall, that bare model hits the router + * and fails with 404 ("requires the canonical openai provider") because no + * canonical route exists for it. + * + * This module remembers, per session lane, which combo last served a + * successful turn and what its concrete target model was. The compaction + * entry points then rewrite a bare model back to the remembered + * combo/ selector only when the bare model exactly matches the + * remembered combo target, so explicit provider/model selectors and + * unrelated models are never touched. + */ + +interface ComboRecallEntry { + comboId: string; + targetModel: string; + at: number; +} + +/** Bounded map: stale entries are dropped, oldest evicted at capacity. */ +const RECALL_CAPACITY = 256; +const RECALL_TTL_MS = 30 * 60 * 1000; + +const recall = new Map(); + +export function rememberComboForLane( + lane: string | undefined, + comboId: string, + targetModel: string, +): void { + if (!lane || !comboId || !targetModel) return; + // Delete-then-set keeps insertion order fresh for eviction. + recall.delete(lane); + recall.set(lane, { comboId, targetModel, at: Date.now() }); + while (recall.size > RECALL_CAPACITY) { + const oldest = recall.keys().next().value; + if (oldest === undefined) break; + recall.delete(oldest); + } +} + +/** + * Returns the remembered combo id when the incoming bare model exactly + * matches the combo target that last succeeded on this lane. Returns + * undefined for explicit provider selectors, stale lanes, and + * non-matching models: those keep ordinary routing. + */ +export function recallComboForLane( + lane: string | undefined, + model: string, +): string | undefined { + if (!lane || !model) return undefined; + const entry = recall.get(lane); + if (!entry) return undefined; + if (Date.now() - entry.at > RECALL_TTL_MS) { + recall.delete(lane); + return undefined; + } + if (entry.targetModel !== model) return undefined; + return entry.comboId; +} + +/** Test-only: clear all recall state. */ +export function clearComboRecallForTests(): void { + recall.clear(); +} diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 4e7481bb2e..e4587678eb 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -152,6 +152,7 @@ import { import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers"; import { mapCodexAuthContextErrorToResponse, nativeMainRefreshFailureResponse } from "./codex-auth-error"; import { sessionLaneIdFromRequest } from "../request-log-conversation"; +import { recallComboForLane } from "./combo-session-recall"; export const COMPACT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; @@ -536,13 +537,31 @@ export async function handleResponsesCompact( // a local rather than written back to `raw.model`: assigning to the property widens it out // of the `string` narrowing the guard above just established. const compactFastRow = parseFastOnlyRowId(config, () => raw.model as string); - const compactModel = compactFastRow ? compactFastRow.baseId : raw.model; + let compactModel = compactFastRow ? compactFastRow.baseId : raw.model; if (compactFastRow) (raw as Record).model = compactModel; // The client's own selector, kept for the request log: `raw.model` is rewritten to the // base id above, and logCtx.requestedModel is assigned from it further down, so without // this the log would lose which id the client actually asked for. const compactRequestedModel = compactFastRow ? compactFastRow.baseId + "--fast" : raw.model; + // A combo switch mid-session leaves Codex sending the bare native model of + // the new combo's target on the compact endpoint (#3891). Rewrite it to the + // remembered combo selector so the combo failover path engages. Only fires + // for bare models (no provider/ prefix) that exactly match the combo target + // last served on this session lane. + if (typeof compactModel === "string" && !compactModel.includes("/") && !compactFastRow) { + const recalledComboId = recallComboForLane(sessionLaneIdFromRequest(req.headers), compactModel); + if (recalledComboId) { + (raw as Record).model = `combo/${recalledComboId}`; + // Keep the routed identity in sync: the bare model can 404 outright (no + // canonical openai provider) or resolve straight onto a native-compact + // provider, both bypassing combo failover. The combo selector resolves + // through tryPickComboModel, whose route.combo skips the native compact + // endpoint. + compactModel = `combo/${recalledComboId}`; + } + } + let route; try { // Compact requests route through the same policy evaluation as normal diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 133ed9cdbc..2a8f3dff05 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -59,6 +59,10 @@ import { providerContinuationRouteScope, sameProviderContinuationOwner, } from "../../responses/provider-continuation"; +import { + rememberComboForLane, + recallComboForLane, +} from "./combo-session-recall"; import { comboRouteDecisionTrace, NoEligiblePolicyCandidateError, @@ -2788,6 +2792,11 @@ export async function handleComboResponses( (logCtx.attempts ??= []).push(attempt); attemptRetained = true; noteComboSuccess(comboId, combo, pick.target, pick.writerGeneration); + rememberComboForLane( + sessionLaneIdFromRequest(req.headers), + comboId, + pick.target.model, + ); Object.assign(logCtx, childLog, { requestedModel, model: requestedModel, @@ -3109,6 +3118,26 @@ async function handleResponsesInner( effort: comboEffortRow.effort, }; } + // A combo switch mid-session leaves Codex sending the bare native model of + // the new combo's target in the compaction request (#3886). Rewrite it to + // the remembered combo selector BEFORE comboIdFromRawBody reads model, so + // the combo dispatch and failover path engage. Only fires for compaction + // requests (compaction_trigger present in input) whose model is bare (no + // provider/ prefix) and exactly matches the combo target last served on + // this session lane. + if (!options.comboAttempt && body && typeof body === "object" && !Array.isArray(body)) { + const rawModel = (body as { model?: unknown }).model; + const rawInput = (body as { input?: unknown }).input; + const isCompactionTrigger = Array.isArray(rawInput) + && rawInput.some((item: unknown) => + typeof item === "object" && item !== null && (item as { type?: string }).type === "compaction_trigger"); + if (typeof rawModel === "string" && !rawModel.includes("/") && isCompactionTrigger) { + const recalledComboId = recallComboForLane(sessionLaneIdFromRequest(req.headers), rawModel); + if (recalledComboId) { + (body as Record).model = `combo/${recalledComboId}`; + } + } + } const comboId = !options.comboAttempt ? comboIdFromRawBody(body, config) : null; if (comboId && Object.hasOwn(config.combos ?? {}, comboId)) { options.onRequestBodyRead?.(); diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index 2a1f69be1e..967a4fac3c 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -35,6 +35,7 @@ import { supportsNativeResponsesCompactEndpoint } from "../../src/providers/open import type { RequestLogContext } from "../../src/server/request-log"; import { acquireNativeMainProfileDrain, tryAdmitTurn } from "../../src/server/lifecycle"; import type { OcxConfig, OcxProviderConfig } from "../../src/types"; +import { clearComboRecallForTests } from "../../src/server/responses/combo-session-recall"; import { removeTreeWithRetry } from "../helpers/remove-tree"; const originalFetch = globalThis.fetch; @@ -1693,6 +1694,278 @@ describe("compact alternate-account attempt (#913)", () => { }); }); +describe("compaction combo recall after combo switch (#3891)", () => { + afterEach(() => clearComboRecallForTests()); + + function comboTestConfig(): OcxConfig { + return { + defaultProvider: "gw", + providers: { + gw: { + adapter: "openai-chat", + baseUrl: "https://gw-primary.example/v1", + authMode: "key", + apiKey: "key-gw", + models: ["gpt-5.6-terra"], + }, + alt: { + adapter: "openai-chat", + baseUrl: "https://gw-alt.example/v1", + authMode: "key", + apiKey: "key-alt", + models: ["gpt-5.6-luna"], + }, + }, + combos: { + terra: { strategy: "failover", targets: [{ provider: "gw", model: "gpt-5.6-terra" }] }, + }, + } as unknown as OcxConfig; + } + + function chatCompletionPayload(text: string): Record { + return { + choices: [{ index: 0, message: { role: "assistant", content: text }, finish_reason: "stop" }], + usage: { prompt_tokens: 10, completion_tokens: 5 }, + }; + } + + // The routed compact turn dispatches combo children as SSE (stream is forced + // when route.combo is set), so streaming-capable mocks answer the chat wire. + function chatStreamResponse(text: string): Response { + return new Response([ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { content: text }, finish_reason: null }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`, + "data: [DONE]\n\n", + ].join(""), { headers: { "content-type": "text/event-stream" } }); + } + + test("bare native model after combo switch routes through the remembered combo", async () => { + const calls: Array<{ url: string; body: Record }> = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + calls.push({ url: String(url), body: JSON.parse(String(init?.body ?? "{}")) as Record }); + return jsonResponse(chatCompletionPayload("handoff summary")); + }) as typeof fetch; + + const config = comboTestConfig(); + const laneHeaders = { "session_id": "lane-combo-recall" }; + + // Step 1: an ordinary combo turn succeeds, populating the recall map. + const comboRes = await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, laneHeaders), + config, + { model: "", provider: "" }, + ); + expect(comboRes.status).toBe(200); + + // Step 2: compaction arrives with the bare native model on the same lane. + const logCtx: RequestLogContext = { model: "", provider: "" }; + const res = await handleResponses( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-terra" }), undefined, laneHeaders), + config, + logCtx, + ); + + expect(res.status).toBe(200); + expect(logCtx.provider).toBe("combo"); + expect(logCtx.comboId).toBe("terra"); + expect(logCtx.requestedModel).toBe("combo/terra"); + const json = await res.json() as { output?: Array<{ type?: string }> }; + expect((json.output ?? []).filter(item => item.type === "compaction").length).toBe(1); + }); + + test("v1 /responses/compact takes the same recall path", async () => { + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + const body = await request.json() as { stream?: boolean }; + return body.stream === true + ? chatStreamResponse("handoff summary") + : jsonResponse(chatCompletionPayload("handoff summary")); + }) as typeof fetch; + + const config = comboTestConfig(); + const laneHeaders = { "session_id": "lane-compact-recall" }; + + const comboRes = await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, laneHeaders), + config, + { model: "", provider: "" }, + ); + expect(comboRes.status).toBe(200); + + const logCtx: RequestLogContext = { model: "", provider: "" }; + const compactRes = await handleResponsesCompact( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-terra" }), undefined, laneHeaders), + config, + logCtx, + ); + + expect(compactRes.status).toBe(200); + expect(logCtx.provider).toBe("combo"); + }); + + test("a different lane does not borrow the remembered combo", async () => { + globalThis.fetch = (async () => jsonResponse(chatCompletionPayload("handoff summary"))) as typeof fetch; + + const config = comboTestConfig(); + + // Populate recall on lane A. + await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, { "session_id": "lane-A" }), + config, + { model: "", provider: "" }, + ); + + // Compaction on lane B: the bare model should NOT be rewritten to the combo. + // It falls through to the compaction default-provider fallback (#2901) and lands on gw. + const logCtx: RequestLogContext = { model: "", provider: "" }; + const res = await handleResponses( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-terra" }), undefined, { "session_id": "lane-B" }), + config, + logCtx, + ); + + expect(res.status).toBe(200); + expect(logCtx.provider).toBe("gw"); + expect(logCtx.comboId).toBeUndefined(); + }); + + test("a non-matching bare model is not rewritten", async () => { + globalThis.fetch = (async () => jsonResponse(chatCompletionPayload("handoff summary"))) as typeof fetch; + + const config = comboTestConfig(); + const laneHeaders = { "session_id": "lane-no-match" }; + + await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, laneHeaders), + config, + { model: "", provider: "" }, + ); + + // Bare model "gpt-5.6-luna" does not match terra combo target "gpt-5.6-terra". + const logCtx: RequestLogContext = { model: "", provider: "" }; + const res = await handleResponses( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-luna" }), undefined, laneHeaders), + config, + logCtx, + ); + + expect(res.status).toBe(200); + expect(logCtx.provider).toBe("gw"); + expect(logCtx.comboId).toBeUndefined(); + }); + + test("recall routes before the bare model can 404 without an openai provider", async () => { + // Maintainer review: with no canonical openai row, the bare model dies in + // routeCompactionModel before any combo logic unless the recall rewrite + // also reaches the routed identity, not only the raw body model. + const config = { + defaultProvider: "openai", + providers: { + gw: { + adapter: "openai-chat", + baseUrl: "https://gw-primary.example/v1", + authMode: "key", + apiKey: "key-gw", + models: ["gpt-5.6-terra"], + }, + }, + combos: { + terra: { strategy: "failover", targets: [{ provider: "gw", model: "gpt-5.6-terra" }] }, + }, + } as unknown as OcxConfig; + const bodies: Array> = []; + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + const body = await request.json() as Record; + bodies.push(body); + return body.stream === true + ? chatStreamResponse("handoff summary") + : jsonResponse(chatCompletionPayload("handoff summary")); + }) as typeof fetch; + + const laneHeaders = { "session_id": "lane-recall-404" }; + const comboRes = await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, laneHeaders), + config, + { model: "", provider: "" }, + ); + expect(comboRes.status).toBe(200); + + const logCtx: RequestLogContext = { model: "", provider: "" }; + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-terra" }), undefined, laneHeaders), + config, + logCtx, + ); + + expect(res.status).toBe(200); + expect(logCtx.provider).toBe("combo"); + expect(logCtx.comboId).toBe("terra"); + // The internal combo turn goes out streaming through the combo dispatch. + expect(bodies[1]!.stream).toBe(true); + await res.text(); + }); + + test("recall keeps a native-compact target on the combo /responses path", async () => { + // CodeRabbit review: the recalled target itself can live on a provider + // that supports the native /responses/compact endpoint. Without the + // routed identity sync, the bare model would go straight to the native + // compact endpoint and bypass combo dispatch entirely. + const config = { + defaultProvider: "openai-apikey", + providers: { + "openai-apikey": { + adapter: "openai-responses", + baseUrl: "https://api.openai.com/v1", + authMode: "key", + apiKey: "test-key", + }, + }, + combos: { + terra: { strategy: "failover", targets: [{ provider: "openai-apikey", model: "gpt-5.6-terra" }] }, + }, + } as unknown as OcxConfig; + const calls: Array<{ url: string; body: Record }> = []; + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + if (request.url.endsWith("/responses/compact")) { + return Response.json({ detail: "Not Found" }, { status: 404 }); + } + calls.push({ url: request.url, body: await request.json() as Record }); + return calls.at(-1)!.body.stream === true + ? sseResponse([{ type: "response.completed", response: completedPayload("handoff summary") }]) + : jsonResponse(completedPayload("handoff summary")); + }) as typeof fetch; + + const laneHeaders = { "session_id": "lane-recall-native" }; + const comboRes = await handleResponses( + compactionRequest({ model: "combo/terra", stream: false, input: "hello" }, undefined, laneHeaders), + config, + { model: "", provider: "" }, + ); + expect(comboRes.status).toBe(200); + + const logCtx: RequestLogContext = { model: "", provider: "" }; + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({ model: "gpt-5.6-terra" }), undefined, laneHeaders), + config, + logCtx, + ); + + expect(res.status).toBe(200); + expect(logCtx.provider).toBe("combo"); + expect(logCtx.comboId).toBe("terra"); + // Both upstream calls take the plain /responses path; the native compact + // endpoint (which this provider supports) must never be hit. + expect(calls.map(call => call.url)).toEqual([ + "https://api.openai.com/v1/responses", + "https://api.openai.com/v1/responses", + ]); + expect(calls[1]!.body.stream).toBe(true); + await res.text(); + }); +}); + test("a no-eligible policy compact request persists the evaluation trace", async () => { const config = { ...keyProviderConfig(),