From 0456daa6a299dad7b396baa1615fb4fca3a9bb64 Mon Sep 17 00:00:00 2001 From: x3M3x Date: Mon, 7 Sep 2026 15:17:20 +0400 Subject: [PATCH 1/9] fix(responses): recall last combo on compaction after a mid-session combo switch When Codex compacts a conversation that was switched to a different combo mid-session, it sends the bare native model of the new combo target (e.g. gpt-5.6-terra) rather than the combo/ selector. Without recall, the bare model hits routeCompactionModel and lands on the configured default provider (or 404s) instead of routing through the combo failover path. Added a bounded session-lane recall map that records the combo target on every successful combo turn. Both compaction entry points (v2 compaction_trigger in handleResponsesInner and v1 /responses/compact in handleResponsesCompact) rewrite a bare model back to the remembered combo selector when it exactly matches the last-served combo target on that session lane. Review round 1 (#3891): the v1 compact endpoint now also syncs the routed identity (compactModel) with the recall rewrite - previously only raw.model was rewritten, so a bare model with no canonical openai row could still 404, and a bare model whose target lives on a native-compact provider (openai / openai-apikey) resolved straight onto the native /responses/compact endpoint, bypassing combo dispatch. Two regression tests cover both routing holes. Safety properties: - Only fires for compaction requests (compaction_trigger present) - Only fires for bare models (no provider/ prefix) - Only fires when the bare model exactly matches the combo target - Different session lanes never borrow each others recall - Explicit combo/provider selectors are never touched The rewrite in handleResponsesInner runs BEFORE comboIdFromRawBody so the combo dispatch path engages. The compact endpoint rewrites raw.model before the non-native dispatch falls through to handleResponses. (cherry picked from commit 52e66f672d9d30e5dd1188590e669c2bac119540) --- src/server/responses/combo-session-recall.ts | 71 +++++ src/server/responses/compact.ts | 21 +- src/server/responses/core.ts | 29 ++ .../responses-compaction-routing.test.ts | 273 ++++++++++++++++++ 4 files changed, 393 insertions(+), 1 deletion(-) create mode 100644 src/server/responses/combo-session-recall.ts 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(), From 175ecbac717a49605930769ab140b84da985f1db Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:50:48 +0900 Subject: [PATCH 2/9] fix(responses): fence combo recall on accepted completed responses Use existing parsed completion callbacks and an independent child gate slot. Preserve explicit combo aliases, current target ownership, config-generation rejection, and final client-visible model identity across failover and compaction. Verification: source review and git diff --check only. Local product tests, typecheck, builds, installs, scripts, servers, and live API calls NOT RUN by user instruction. Co-authored-by: x3M3x --- src/lib/state-store-registrations.ts | 2 + src/server/responses/combo-session-recall.ts | 88 +++--- src/server/responses/compact.ts | 13 +- src/server/responses/core.ts | 203 +++++++++----- tests/oauth/state-store-sweeper.test.ts | 59 +++- .../responses-compaction-routing.test.ts | 213 ++++++++++++++- .../server/server-combo-failover-e2e.test.ts | 258 +++++++++++++++++- 7 files changed, 724 insertions(+), 112 deletions(-) diff --git a/src/lib/state-store-registrations.ts b/src/lib/state-store-registrations.ts index 433a8989b4..13a22bfce0 100644 --- a/src/lib/state-store-registrations.ts +++ b/src/lib/state-store-registrations.ts @@ -21,6 +21,7 @@ import { } from "../combos/failover"; import { reconcileComboWarningMemos } from "../combos/request"; import { reconcileComboRotationState } from "../combos/resolve"; +import { reconcileComboRecall } from "../server/responses/combo-session-recall"; import { listLiveComboTargetKeys } from "../combos/types"; import { listLiveConfigOwnershipRoots, @@ -111,6 +112,7 @@ export const STATE_STORE_REGISTRATIONS = [ { name: "model-cache-history", reconcileGeneration: reconcileModelCacheGeneration }, { name: "pool-rotation", reconcileGeneration: reconcilePoolRotationState }, { name: "combo-rotation", reconcileGeneration: reconcileComboRotationState }, + { name: "combo-session-recall", reconcileGeneration: reconcileComboRecall }, { name: "guardian-backoff", reconcileGeneration: reconcileGuardianBackoff }, { name: "codex-reauth", reconcileGeneration: reconcileCodexReauthState }, { name: "oauth-reauth", reconcileGeneration: reconcileOAuthReauthState }, diff --git a/src/server/responses/combo-session-recall.ts b/src/server/responses/combo-session-recall.ts index e7a403484b..84dfd8d466 100644 --- a/src/server/responses/combo-session-recall.ts +++ b/src/server/responses/combo-session-recall.ts @@ -1,42 +1,41 @@ -/** - * 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. - */ +/** Process-local recall of the last completed combo response on an explicit session lane. */ +import { getCombo, targetKey } from "../../combos/types"; +import { captureConfigGeneration, type GenerationContext } from "../../lib/state-store-sweeper"; +import type { OcxConfig, OcxComboTarget } from "../../types"; interface ComboRecallEntry { comboId: string; - targetModel: string; + target: Pick; + responseModel: 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(); +let lastReconciledGeneration = 0; +let liveOwners: Pick | undefined; + +function ownsEntry(context: Pick, entry: ComboRecallEntry): boolean { + return context.comboIds.has(entry.comboId) + && context.providerNames.has(entry.target.provider) + && context.comboTargets.has(`${entry.comboId}::${targetKey(entry.target)}`); +} export function rememberComboForLane( lane: string | undefined, comboId: string, - targetModel: string, + target: Pick, + responseModel: string, + writerGeneration: number, ): void { - if (!lane || !comboId || !targetModel) return; - // Delete-then-set keeps insertion order fresh for eviction. + if (!lane || !comboId || !responseModel.trim()) return; + // Reject even a same-named recreated owner: its previous in-flight turn is obsolete. + if (writerGeneration < Math.max(lastReconciledGeneration, captureConfigGeneration())) return; + const entry = { comboId, target: { provider: target.provider, model: target.model }, responseModel, at: Date.now() }; + if (liveOwners && !ownsEntry(liveOwners, entry)) return; recall.delete(lane); - recall.set(lane, { comboId, targetModel, at: Date.now() }); + recall.set(lane, entry); while (recall.size > RECALL_CAPACITY) { const oldest = recall.keys().next().value; if (oldest === undefined) break; @@ -44,28 +43,47 @@ export function rememberComboForLane( } } -/** - * 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( + config: OcxConfig, lane: string | undefined, model: string, ): string | undefined { - if (!lane || !model) return undefined; + if (!lane || !model || model.includes("/")) return undefined; const entry = recall.get(lane); if (!entry) return undefined; - if (Date.now() - entry.at > RECALL_TTL_MS) { + const combo = getCombo(config, entry.comboId); + const provider = config.providers[entry.target.provider]; + if (Date.now() - entry.at >= RECALL_TTL_MS + || !Object.hasOwn(config.providers, entry.target.provider) + || !provider || provider.disabled === true + || !combo?.targets.some(target => targetKey(target) === targetKey(entry.target))) { recall.delete(lane); return undefined; } - if (entry.targetModel !== model) return undefined; - return entry.comboId; + return entry.responseModel === model ? entry.comboId : undefined; +} + +export function reconcileComboRecall(context: GenerationContext): number { + if (context.generation <= lastReconciledGeneration) return 0; + lastReconciledGeneration = context.generation; + liveOwners = { + comboIds: new Set(context.comboIds), + comboTargets: new Set(context.comboTargets), + providerNames: new Set(context.providerNames), + }; + let removed = 0; + for (const [lane, entry] of recall) { + if (!ownsEntry(context, entry) || Date.now() - entry.at >= RECALL_TTL_MS) { + recall.delete(lane); + removed += 1; + } + } + return removed; } -/** Test-only: clear all recall state. */ +/** Test-only reset, alongside the combo rotation/cooldown resets. */ export function clearComboRecallForTests(): void { recall.clear(); + lastReconciledGeneration = 0; + liveOwners = undefined; } diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index e4587678eb..567012a0f5 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -18,6 +18,7 @@ import { comboIdFromRawBody, concreteComboRequestBody, getCombo, + resolveComboId, isComboTargetInCooldown, NoAvailableComboTargetsError, noteComboSuccess, @@ -544,13 +545,11 @@ export async function handleResponsesCompact( // 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); + // Recall the last completed client-visible bare model after a combo switch (#3891). + // Configured selectors take precedence over this implicit session hint. + if (typeof compactModel === "string" && !compactModel.includes("/") && !compactFastRow + && !resolveComboId(config, compactModel)) { + const recalledComboId = recallComboForLane(config, 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 diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 2a8f3dff05..a86053d7af 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -83,6 +83,7 @@ import { comboRequestHasImageInput, concreteComboRequestBody, getCombo, + resolveComboId, isComboTargetInCooldown, NoAvailableComboTargetsError, noteComboSuccess, @@ -315,6 +316,7 @@ import { consumeForInspection, consumeForResponseLogMetadata, createSseInspector, + terminalStatusFromParsed, isEagerRelaySseResponse, isNativePassthroughSseResponse, markEagerRelaySseResponse, @@ -1670,6 +1672,8 @@ export interface HandleResponsesOptions { onCodexAuthContextResolved?: (context: CodexAuthContext | undefined) => void; /** Internal deterministic seam for account-gated native fallback tests. */ resolveCodexModelEntitlements?: typeof resolveCodexModelEntitlements; + /** Internal: validated final client-visible model, after completed terminal success only. */ + onResponseComplete?: (model: string) => void; recordTerminalOutcomes?: boolean; setTerminalOutcomeRecorder?: (recorder: ((status: ResponsesTerminalStatus, httpStatusOverride?: number) => void) | undefined) => void; onNativePassthroughTerminal?: (status: ResponsesTerminalStatus) => void; @@ -1870,34 +1874,54 @@ export function createChildPassthroughCallbackGate(options: HandleResponsesOptio let state: "pending" | "committed" | "discarded" = "pending"; let pending: Pending | undefined; let accepted = false; + let pendingModel: string | undefined; + let completionAccepted = false; + let completionRejected = false; const publish = (value: Pending): void => { if (value.kind === "terminal") options.onNativePassthroughTerminal?.(value.status); else options.onNativePassthroughCancel?.(); }; + const publishCompletion = (): void => { + if (state !== "committed" || completionRejected || pendingModel === undefined) return; + const model = pendingModel; + pendingModel = undefined; + options.onResponseComplete?.(model); + }; const receive = (value: Pending): void => { if (state === "discarded" || accepted) return; accepted = true; + if (value.kind === "cancel" || value.status !== "completed") { + completionRejected = true; + pendingModel = undefined; + } if (state === "committed") return publish(value); pending ??= value; }; return { onTerminal: (status: ResponsesTerminalStatus) => receive({ kind: "terminal", status }), onCancel: () => receive({ kind: "cancel" }), + onResponseComplete: (model: string) => { + if (state === "discarded" || completionRejected || completionAccepted || !model.trim()) return; + completionAccepted = true; + pendingModel = model; + publishCompletion(); + }, commit: () => { if (state !== "pending") return; state = "committed"; if (pending) publish(pending); pending = undefined; + publishCompletion(); }, discard: () => { state = "discarded"; pending = undefined; + pendingModel = undefined; }, }; } - export function buildComboChildHeaders(parentHeaders: HeadersInit): Headers { const childHeaders = new Headers(parentHeaders); // Combo children re-serialize already-decoded JSON. Keeping transport metadata from @@ -2698,9 +2722,22 @@ export async function handleComboResponses( (logCtx.attempts ??= []).push(attempt); attemptRetained = true; }; + const completedTarget = { provider: pick.target.provider, model: pick.target.model }; + const writerGeneration = pick.writerGeneration; let consumedChildFailure: ConsumedComboFailure | undefined; const callbackGate = createChildPassthroughCallbackGate({ ...options, + onResponseComplete: model => { + // The live config can change while the child is streaming. Never retain credentials. + const currentCombo = getCombo(config, comboId); + const provider = config.providers[completedTarget.provider]; + if (Object.hasOwn(config.providers, completedTarget.provider) + && provider && provider.disabled !== true + && currentCombo?.targets.some(target => targetKey(target) === targetKey(completedTarget))) { + rememberComboForLane(sessionLaneIdFromRequest(req.headers), comboId, completedTarget, model, writerGeneration); + } + options.onResponseComplete?.(model); + }, onNativePassthroughTerminal: status => { // A committed stream can acquire terminal metadata after preflight copied // the child log. Publish it before the outer logger finalizes, but only @@ -2741,6 +2778,7 @@ export async function handleComboResponses( onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; }, onNativePassthroughTerminal: callbackGate.onTerminal, onNativePassthroughCancel: callbackGate.onCancel, + onResponseComplete: callbackGate.onResponseComplete, }); } catch (error) { callbackGate.discard(); @@ -2792,11 +2830,6 @@ 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, @@ -3118,21 +3151,18 @@ 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. + // Compaction may send the last client-visible bare model after a combo switch. + // Configured selectors take precedence; otherwise recall before combo dispatch (#3891). 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 (typeof rawModel === "string" && !rawModel.includes("/") && isCompactionTrigger + && !comboRows.fastRow && !comboEffortRow + && !resolveComboId(config, rawModel)) { + const recalledComboId = recallComboForLane(config, sessionLaneIdFromRequest(req.headers), rawModel); if (recalledComboId) { (body as Record).model = `combo/${recalledComboId}`; } @@ -4341,6 +4371,17 @@ async function handleResponsesInner( } const recordTerminalOutcomes = options.recordTerminalOutcomes !== false; + let responseCompletionNotified = false; + let responseCompletionCancelled = false; + const cancelResponseCompletion = (): void => { responseCompletionCancelled = true; }; + const notifyResponseComplete = (response: { status?: unknown; model?: unknown }): void => { + if (responseCompletionNotified || responseCompletionCancelled + || options.abortSignal?.aborted || req.signal.aborted + || response.status !== "completed" + || typeof response.model !== "string" || !response.model.trim()) return; + responseCompletionNotified = true; + options.onResponseComplete?.(response.model); + }; const continuationStateForResponse = ( emitted?: OcxProviderContinuationState, @@ -4618,9 +4659,13 @@ async function handleResponsesInner( // check sees nothing undeclared, and the refused turn enters continuation state anyway. So the // rejection is sticky for the whole turn, set from every parsed payload on the inspection side. let inspectionSawUndeclaredTool = false; + let inspectedTerminal: ResponsesTerminalStatus | null = null; const passiveQuotaObserved = hasPassiveAccountQuota(route.providerName) && route.provider.authMode === "oauth"; const noteInspectedPayload = (payload: unknown) => { + // First terminal stays authoritative even in metadata-only inspection, which + // intentionally continues parsing after a failed/incomplete terminal. + inspectedTerminal ??= terminalStatusFromParsed(payload); // Meta reports subscription usage ONLY as an in-stream event; there is no endpoint // to poll (003 §E probed 17 paths, all 404). Observed here rather than behind a // dedicated inspector handler because onParsedPayload already reaches every @@ -4642,8 +4687,7 @@ async function handleResponsesInner( // Gated on the same flag as the guard itself: with no readable catalog (or a forward-auth // provider) every name looks undeclared, and flipping this would stop recording continuation // state for exactly the passthrough traffic the guard deliberately stands down for. - if (!undeclaredToolGuardActive || inspectionSawUndeclaredTool) return; - if (undeclaredToolCallName( + if (undeclaredToolGuardActive && !inspectionSawUndeclaredTool && undeclaredToolCallName( restoreAuthorizedBareNamespaceToolCalls(payload), declaredWireToolNames, declaredNamelessClientCallTypes, @@ -4651,33 +4695,52 @@ async function handleResponsesInner( ) !== undefined) { inspectionSawUndeclaredTool = true; } + // The snapshot callback opts the inspector into output reconstruction. Compaction + // has no continuation cache, so use the parsed terminal here without adding retention. + if (!rememberPassthroughResponse && payload && typeof payload === "object" + && "type" in payload && payload.type === "response.completed" + && "response" in payload && payload.response && typeof payload.response === "object" + && !Array.isArray(payload.response)) { + rememberPassthroughResponseChecked(payload.response as Record); + } }; - const rememberPassthroughResponseChecked = rememberPassthroughResponse - ? (response: { id?: unknown; output?: unknown; status?: unknown }) => { - if (inspectionSawUndeclaredTool) return; - const restored = restoreRoutedCustomCalls( - restoreAuthorizedBareNamespaceToolCalls(restoreRoutedNamespaceCalls(response, routedNamespaceToolAliases).value), - routedCustomToolNames, - routedCustomToolRepairNames, + const rememberPassthroughResponseChecked = ( + response: { id?: unknown; output?: unknown; status?: unknown; model?: unknown }, + ) => { + if (inspectionSawUndeclaredTool) return; + const restored = restoreRoutedCustomCalls( + restoreAuthorizedBareNamespaceToolCalls(restoreRoutedNamespaceCalls(response, routedNamespaceToolAliases).value), + routedCustomToolNames, + routedCustomToolRepairNames, + declaredWireToolNames, + ).value; + const restoredResponse = (functionRepairSchemas.size > 0 + ? JSON.parse(normalizeFunctionCompletionJson(JSON.stringify(restored))) + : restored) as { id?: unknown; output?: unknown; status?: unknown }; + if ( + undeclaredToolGuardActive + && undeclaredToolCallNameInResponse( + restoredResponse, declaredWireToolNames, - ).value; - const restoredResponse = (functionRepairSchemas.size > 0 - ? JSON.parse(normalizeFunctionCompletionJson(JSON.stringify(restored))) - : restored) as { id?: unknown; output?: unknown; status?: unknown }; - if ( - undeclaredToolGuardActive - && undeclaredToolCallNameInResponse( - restoredResponse, - declaredWireToolNames, - declaredNamelessClientCallTypes, - providerExecutedCallTypes, - ) !== undefined - ) { - return; + declaredNamelessClientCallTypes, + providerExecutedCallTypes, + ) !== undefined + ) { + return; + } + rememberPassthroughResponse?.(restoredResponse); + if (inspectedTerminal === null || inspectedTerminal === "completed") { + // Native inspection sees the pre-rewrite model. Only an actual terminal + // model can seed recall; an absent model never falls back to the pick. + if (typeof response.model === "string" && response.model.trim()) { + notifyResponseComplete({ + status: response.status, + model: parsed._responseModelId !== undefined && parsed._responseModelId !== parsed.modelId + ? parsed._responseModelId : response.model, + }); } - rememberPassthroughResponse(restoredResponse); } - : undefined; + }; recordAdapterReasoning(logCtx, request); recordAdapterTier(logCtx, request); const actualHostKey = upstreamHostHealthKey( @@ -5685,7 +5748,7 @@ async function handleResponsesInner( const inspector = createSseInspector({ onTerminal: reportNativeTerminal, logCtx, - onCompletedResponse: rememberPassthroughResponseChecked, + onCompletedResponse: rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, onParsedPayload: noteInspectedPayload, onFirstOutput: options.onFirstOutput, pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled, @@ -5715,7 +5778,10 @@ async function handleResponsesInner( reportNativeTerminal("failed", 502); } }, - onClientCancel: () => options.onNativePassthroughCancel?.(), + onClientCancel: () => { + responseCompletionCancelled = true; + options.onNativePassthroughCancel?.(); + }, onDone: () => unregisterTurn(turnAc), }, { clientGoneSignal: options.abortSignal, @@ -5777,8 +5843,11 @@ async function handleResponsesInner( turnAc.signal, () => unregisterTurn(turnAc), logCtx, - () => options.onNativePassthroughCancel?.(), - rememberPassthroughResponseChecked, + () => { + responseCompletionCancelled = true; + options.onNativePassthroughCancel?.(); + }, + rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, options.onFirstOutput, inspectionConsumerOptions, ); @@ -5788,7 +5857,7 @@ async function handleResponsesInner( logCtx, turnAc.signal, () => unregisterTurn(turnAc), - rememberPassthroughResponseChecked, + rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, options.onFirstOutput, inspectionConsumerOptions, ); @@ -5803,7 +5872,10 @@ async function handleResponsesInner( const clientBody = relaySseWithFailedTail( rewrittenBody, upstream, - reason => clientGone.abort(reason), + reason => { + responseCompletionCancelled = true; + clientGone.abort(reason); + }, { upstreamError: logCtx.upstreamError }, ); return markNativePassthroughSseResponse(new Response(clientBody, { @@ -5883,13 +5955,11 @@ async function handleResponsesInner( } } commitReasoningReplayServingRoute(); - if (rememberPassthroughResponseChecked) { - try { - rememberPassthroughResponseChecked( - JSON.parse(text) as { id?: unknown; output?: unknown; status?: unknown }, - ); - } catch { /* non-JSON despite content-type; recording is best-effort */ } - } + try { + rememberPassthroughResponseChecked( + JSON.parse(text) as { id?: unknown; output?: unknown; status?: unknown; model?: unknown }, + ); + } catch { /* non-JSON despite content-type; recording is best-effort */ } // #875: the transport-neutral reliability policy forced a bounded JSON // upstream for a client that asked for SSE. Reframe the completed JSON // as the canonical terminal SSE sequence (created → output_item.done → @@ -6223,10 +6293,12 @@ async function handleResponsesInner( continuationStateForResponse(providerState), responseStateOptions(adapterNeedsForcedContinuation(adapter.name)), ); + notifyResponseComplete(response); }, }); if (imgResponse.body) { const imgTurnAc = new AbortController(); + imgTurnAc.signal.addEventListener("abort", cancelResponseCompletion, { once: true }); return new Response(trackStreamLifetime(imgResponse.body, imgTurnAc, undefined, options.turnAdmissionLease), { status: imgResponse.status, headers: imgResponse.headers, @@ -6297,12 +6369,16 @@ async function handleResponsesInner( streamRoutedModelOutput: wsPlan.streamRoutedModelOutput, on429: rotateSidecarProviderOn429, retryOn429Policy: rateLimitRetryPolicyFor(route.provider), - onCompletedResponse: commitReasoningReplayServingRoute, + onCompletedResponse: response => { + commitReasoningReplayServingRoute(); + notifyResponseComplete(response); + }, }); // Register the sidecar stream as an active turn so drainAndShutdown waits for (or aborts) // in-flight web-search turns instead of skipping them during graceful shutdown. if (wsResponse.body) { const wsTurnAc = new AbortController(); + wsTurnAc.signal.addEventListener("abort", cancelResponseCompletion, { once: true }); return new Response(trackStreamLifetime(wsResponse.body, wsTurnAc, undefined, options.turnAdmissionLease), { status: wsResponse.status, headers: wsResponse.headers, @@ -6526,6 +6602,7 @@ async function handleResponsesInner( const sseStream = bridgeToResponsesSSE( guardedSource, parsed._responseModelId ?? parsed.modelId, toolNsMap, freeformToolNames, toolSearchToolNames, () => { + cancelResponseCompletion(); runTurnAbort.abort(); queue.close(); }, 2_000, @@ -6562,6 +6639,7 @@ async function handleResponsesInner( responseStateOptions(adapterNeedsForcedContinuation(adapter.name)), ); } + notifyResponseComplete(response); }, }, ); @@ -6638,6 +6716,7 @@ async function handleResponsesInner( if (adapterResponseReachedServingTerminal(events, json)) { commitReasoningReplayServingRoute(); } + notifyResponseComplete(json); return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } @@ -6685,10 +6764,11 @@ async function handleResponsesInner( toolBridgeMaps.toolNsMap, toolBridgeMaps.freeformToolNames, toolBridgeMaps.toolSearchToolNames, - undefined, + cancelResponseCompletion, 2_000, { translatorBudget, + onCompletedResponse: notifyResponseComplete, ...(options.forceEmptyResponseId ? { responseId: "" } : {}), ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), }, @@ -6709,12 +6789,9 @@ async function handleResponsesInner( }, ); } - return new Response( - JSON.stringify(buildResponseJSON(terminalEvents, parsed._responseModelId ?? parsed.modelId, { - translatorBudget, - })), - { headers: { "Content-Type": "application/json" } }, - ); + const json = buildResponseJSON(terminalEvents, parsed._responseModelId ?? parsed.modelId, { translatorBudget }); + notifyResponseComplete(json); + return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } // One request-scoped transient-retry budget owner, declared here so BOTH the initial send // and the later recovery refetches (429, key/account rotation, OAuth replay) share it. A @@ -7696,7 +7773,7 @@ async function handleResponsesInner( const { toolNsMap, declaredToolNames, toolParameterSchemas, freeformToolNames, toolSearchToolNames } = toolBridgeMaps; const sseStream = bridgeToResponsesSSE( guardedEventStream, parsed._responseModelId ?? parsed.modelId, toolNsMap, freeformToolNames, toolSearchToolNames, - () => upstream.abort(), 2_000, + () => { cancelResponseCompletion(); upstream.abort(); }, 2_000, { translatorBudget, replayCacheScope: parsed._reasoningReplayScope, @@ -7731,6 +7808,7 @@ async function handleResponsesInner( responseStateOptions(activeAdapter.name === "kiro"), ); } + notifyResponseComplete(response); }, }, ); @@ -7810,6 +7888,7 @@ async function handleResponsesInner( if (adapterResponseReachedServingTerminal(events, json)) { commitReasoningReplayServingRoute(); } + notifyResponseComplete(json); return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } diff --git a/tests/oauth/state-store-sweeper.test.ts b/tests/oauth/state-store-sweeper.test.ts index 36ede716d4..b162e2214b 100644 --- a/tests/oauth/state-store-sweeper.test.ts +++ b/tests/oauth/state-store-sweeper.test.ts @@ -22,7 +22,8 @@ import { setOcxStartProcessProbeForTests, sweepDeadOcxStartProcessCache, } from "../../src/config"; -import { STATE_STORE_REGISTRATIONS } from "../../src/lib/state-store-registrations"; +import { STATE_STORE_REGISTRATIONS, setLiveStateStoreConfig, reconcileLiveStateStores } from "../../src/lib/state-store-registrations"; +import { clearComboRecallForTests, recallComboForLane, rememberComboForLane } from "../../src/server/responses/combo-session-recall"; import { getAccountSet, saveCredential } from "../../src/oauth/store"; import { clearAccountQuotaCache, @@ -78,12 +79,14 @@ beforeEach(() => { sweeperHome = mkdtempSync(join(tmpdir(), "ocx-sweeper-home-")); process.env.OPENCODEX_HOME = sweeperHome; resetStateStoreSweeperForTests(); + clearComboRecallForTests(); resetAppOwnedMemoryForTests(); clearResponseStateMemoryForTests(); __resetAntigravityReplayCache(); }); afterEach(() => { resetStateStoreSweeperForTests(); + clearComboRecallForTests(); resetAppOwnedMemoryForTests(); clearResponseStateMemoryForTests(); __resetAntigravityReplayCache(); @@ -147,6 +150,7 @@ describe("state-store sweeper", () => { "model-cache-history", "pool-rotation", "combo-rotation", + "combo-session-recall", "guardian-backoff", "codex-reauth", "oauth-reauth", @@ -157,6 +161,59 @@ describe("state-store sweeper", () => { ]); }); + test("registered combo recall cleanup rejects an old completion after delete and recreate while retaining another owner", () => { + registerStateStore(STATE_STORE_REGISTRATIONS.find(row => row.name === "combo-session-recall")!); + const config: OcxConfig = { + port: 0, defaultProvider: "a", + providers: { a: { adapter: "openai-chat", baseUrl: "https://a.example/v1" } }, + combos: { + first: { targets: [{ provider: "a", model: "m1" }] }, + other: { targets: [{ provider: "a", model: "m2" }] }, + }, + }; + setLiveStateStoreConfig(config); + const staleGeneration = captureConfigGeneration(); + rememberComboForLane("first-lane", "first", { provider: "a", model: "m1" }, "visible-first", staleGeneration); + rememberComboForLane("other-lane", "other", { provider: "a", model: "m2" }, "visible-other", staleGeneration); + delete config.combos!.first; + expect(reconcileLiveStateStores()).toEqual({ storesVisited: 1, rowsRemoved: 1 }); + config.combos!.first = { targets: [{ provider: "a", model: "m1" }] }; + expect(reconcileLiveStateStores()).toEqual({ storesVisited: 1, rowsRemoved: 0 }); + rememberComboForLane("first-lane", "first", { provider: "a", model: "m1" }, "visible-first", staleGeneration); + expect(recallComboForLane(config, "first-lane", "visible-first")).toBeUndefined(); + expect(recallComboForLane(config, "other-lane", "visible-other")).toBe("other"); + rememberComboForLane("first-lane", "first", { provider: "a", model: "m1" }, "visible-new", captureConfigGeneration()); + expect(recallComboForLane(config, "first-lane", "visible-new")).toBe("first"); + delete config.providers.a; + expect(reconcileLiveStateStores()).toEqual({ storesVisited: 1, rowsRemoved: 2 }); + }); + + test("combo recall watermark rejects writers after a partially failed generation", () => { + registerStateStore(STATE_STORE_REGISTRATIONS.find(row => row.name === "combo-session-recall")!); + const warning = spyOn(console, "warn").mockImplementation(() => {}); + const unregisterFailure = registerStateStore({ name: "failed-owner", reconcileGeneration: () => { throw new Error("retry"); } }); + const owners = context(0, { + comboIds: new Set(["first"]), comboTargets: new Set(["first::a/m1"]), providerNames: new Set(["a"]), + }); + const config: OcxConfig = { + port: 0, defaultProvider: "a", providers: { a: { adapter: "openai-chat", baseUrl: "https://a.example/v1" } }, + combos: { first: { targets: [{ provider: "a", model: "m1" }] } }, + }; + try { + reconcileStateGeneration(owners); + expect(captureConfigGeneration()).toBe(0); + rememberComboForLane("lane", "first", { provider: "a", model: "m1" }, "m1", 0); + expect(recallComboForLane(config, "lane", "m1")).toBeUndefined(); + unregisterFailure(); + reconcileStateGeneration(owners); + rememberComboForLane("lane", "first", { provider: "a", model: "m1" }, "m1", captureConfigGeneration()); + expect(recallComboForLane(config, "lane", "m1")).toBe("first"); + } finally { + unregisterFailure(); + warning.mockRestore(); + } + }); + test("a sweeper tick expires continuation and Antigravity rows without store traffic", () => { rememberResponseState({ input: "old" }, { id: "resp_sweeper_ttl", output: [], status: "completed" }); observeAntigravityReplay("gemini-3-pro", "session-old", [{ diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index 967a4fac3c..63a734bca9 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -35,7 +35,8 @@ 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 { clearComboRecallForTests, recallComboForLane, rememberComboForLane } from "../../src/server/responses/combo-session-recall"; +import { captureConfigGeneration } from "../../src/lib/state-store-sweeper"; import { removeTreeWithRetry } from "../helpers/remove-tree"; const originalFetch = globalThis.fetch; @@ -950,6 +951,71 @@ describe("compact alternate-account attempt (#913)", () => { }); } + for (const version of ["v1", "v2"] as const) { + test(`${version} recalled native combo reselects the current account and respects admission refusal`, async () => { + await withPoolEnv("ocx-combo-recall-account-", async config => { + clearComboRecallForTests(); + config.combos = { native: { targets: [{ provider: "openai", model: "gpt-5.5" }] } }; + config.codexAccountNamespaces = { side: "pool-a" }; + const headers = { session_id: "account-recall" }; + const accounts: Array = []; + // Fix the selected account deterministically while retaining the real credential + // and admission owner; an explicit namespace still owns its account selection. + const resolver = authContextModule.resolveCodexAuthContext; + const authSpy = spyOn(authContextModule, "resolveCodexAuthContext").mockImplementation( + (incoming, liveConfig, mode, options = {}) => resolver(incoming, liveConfig, mode, { + ...options, accountId: options.accountId ?? liveConfig.activeCodexAccountId, + }), + ); + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + accounts.push(request.headers.get("chatgpt-account-id")); + const body = await request.json() as { input?: Array<{ type?: string }> }; + if (request.url.endsWith("/responses/compact")) { + return jsonResponse({ output: [{ type: "compaction", encrypted_content: "native-recall-ciphertext" }] }); + } + const compact = Array.isArray(body.input) && body.input.some(item => item.type === "compaction_trigger"); + return sseResponse([{ type: "response.completed", response: { + ...completedPayload("native answer"), model: "gpt-5.5", + ...(compact ? { output: [{ type: "compaction", encrypted_content: "native-recall-ciphertext" }] } : {}), + } }]); + }) as typeof fetch; + try { + let complete!: () => void; + const completed = new Promise(resolve => { complete = resolve; }); + const seed = await handleResponses(compactionRequest({ model: "combo/native", stream: true, input: "hello" }, undefined, headers), + config, { model: "", provider: "" }, { onResponseComplete: complete }); + await seed.text(); + await completed; + expect(recallComboForLane(config, "account-recall", "gpt-5.5")).toBe("native"); + config.activeCodexAccountId = "pool-b"; + const compact = version === "v1" ? handleResponsesCompact : handleResponses; + const log: RequestLogContext = { model: "", provider: "" }; + const response = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), undefined, headers), config, log); + expect(response.status).toBe(200); + await response.text(); + expect(log.comboId).toBe("native"); + expect(accounts).toEqual(["pool_acc_a", "pool_acc_b"]); + + const explicitLog: RequestLogContext = { model: "", provider: "" }; + const explicit = await compact(compactionRequest(baseCompactionBody({ model: "side/gpt-5.5", stream: true }), undefined, headers), config, explicitLog); + expect(explicit.status).toBe(200); + await explicit.text(); + expect(explicitLog.comboId).toBeUndefined(); + expect(accounts.at(-1)).toBe("pool_acc_a"); + const sends = accounts.length; + authSpy.mockRejectedValue(new authContextModule.CodexMainProfileDrainingError()); + const refused = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), undefined, headers), config, { model: "", provider: "" }); + expect(refused.status).toBe(503); + expect(accounts).toHaveLength(sends); + } finally { + authSpy.mockRestore(); + clearComboRecallForTests(); + } + }); + }); + } + for (const [model, account] of [["gpt-5.5", "pool-a"], ["side/gpt-5.5", "pool-b"]] as const) { test(`native 404 falls back to canonical SSE with ${model} account and session identity`, async () => { await withPoolEnv("ocx-compact-404-canonical-", async config => { @@ -1933,8 +1999,8 @@ describe("compaction combo recall after combo switch (#3891)", () => { } 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")); + ? sseResponse([{ type: "response.completed", response: { ...completedPayload("handoff summary"), model: "gpt-5.6-terra" } }]) + : jsonResponse({ ...completedPayload("handoff summary"), model: "gpt-5.6-terra" }); }) as typeof fetch; const laneHeaders = { "session_id": "lane-recall-native" }; @@ -1964,6 +2030,147 @@ describe("compaction combo recall after combo switch (#3891)", () => { expect(calls[1]!.body.stream).toBe(true); await res.text(); }); + + function installRecallChatFixture(): void { + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const body = await new Request(input, init).json() as { stream?: boolean }; + return body.stream ? chatStreamResponse("summary") : jsonResponse(chatCompletionPayload("answer")); + }) as typeof fetch; + } + + async function seedRecall(config: OcxConfig, lane: string | undefined = "recall-lane"): Promise { + const response = await handleResponses(compactionRequest( + { model: "combo/terra", stream: false, input: "hello" }, undefined, + lane ? { session_id: lane } : {}, + ), config, { model: "", provider: "" }); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ status: "completed", model: "gpt-5.6-terra" }); + } + + for (const version of ["v1", "v2"] as const) { + const compact = version === "v1" ? handleResponsesCompact : handleResponses; + const dispatch = async (config: OcxConfig, model: string, lane: string | undefined = "recall-lane") => { + const log: RequestLogContext = { model: "", provider: "" }; + const response = await compact(compactionRequest(baseCompactionBody({ model }), undefined, + lane ? { session_id: lane } : {}), config, log); + expect(response.status).toBe(200); + await response.text(); + return log; + }; + + test(`${version} explicit bare nativeAlias beats a different remembered combo`, async () => { + installRecallChatFixture(); + const config = comboTestConfig(); + config.combos!.explicit = { + alias: "gpt-5.6-terra", nativeAlias: true, + targets: [{ provider: "alt", model: "gpt-5.6-luna" }], + }; + await seedRecall(config); + expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra")).toBe("terra"); + const log = await dispatch(config, "gpt-5.6-terra"); + expect(log.comboId).toBe("explicit"); + expect(log.resolvedModel).toBe("gpt-5.6-luna"); + }); + + test(`${version} explicit provider and combo selectors beat recall`, async () => { + installRecallChatFixture(); + const config = comboTestConfig(); + config.combos!.explicit = { targets: [{ provider: "alt", model: "gpt-5.6-luna" }] }; + await seedRecall(config); + expect((await dispatch(config, "alt/gpt-5.6-luna")).provider).toBe("alt"); + expect((await dispatch(config, "combo/explicit")).comboId).toBe("explicit"); + }); + + for (const mutation of ["delete", "rename", "replace-target", "delete-provider", "disable-provider"] as const) { + test(`${version} ${mutation} invalidates remembered ownership before fallback`, async () => { + installRecallChatFixture(); + const config = comboTestConfig(); + await seedRecall(config); + // The default is distinct from the original target and remains usable. + config.defaultProvider = "alt"; + if (mutation === "rename") config.combos!.renamed = config.combos!.terra!; + if (mutation === "delete" || mutation === "rename") delete config.combos!.terra; + if (mutation === "replace-target") config.combos!.terra!.targets = [{ provider: "alt", model: "gpt-5.6-luna" }]; + if (mutation === "delete-provider") delete config.providers.gw; + if (mutation === "disable-provider") config.providers.gw!.disabled = true; + const log = await dispatch(config, "gpt-5.6-terra"); + expect(log.comboId).toBeUndefined(); + expect(log.provider).toBe("alt"); + }); + } + + test(`${version} missing and sibling lanes cannot borrow a completed selection`, async () => { + installRecallChatFixture(); + const config = comboTestConfig(); + await seedRecall(config); + expect((await dispatch(config, "gpt-5.6-terra", "sibling")).comboId).toBeUndefined(); + // Empty lane explicitly omits the header (undefined would use the helper default). + expect((await dispatch(config, "gpt-5.6-terra", "")).comboId).toBeUndefined(); + clearComboRecallForTests(); + await seedRecall(config, ""); + expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBeUndefined(); + }); + + test(`${version} recall expires at thirty minutes and evicts the oldest of 257 lanes`, async () => { + installRecallChatFixture(); + const config = comboTestConfig(); + let now = 100_000; + const clock = spyOn(Date, "now").mockImplementation(() => now); + try { + await seedRecall(config); + now += 30 * 60 * 1000 - 1; + expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra")).toBe("terra"); + now += 1; + expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBeUndefined(); + const target = { provider: "gw", model: "gpt-5.6-terra" }; + for (let index = 0; index < 257; index += 1) { + rememberComboForLane(`lane-${index}`, "terra", target, "gpt-5.6-terra", captureConfigGeneration()); + } + expect((await dispatch(config, "gpt-5.6-terra", "lane-0")).comboId).toBeUndefined(); + expect((await dispatch(config, "gpt-5.6-terra", "lane-1")).comboId).toBe("terra"); + expect((await dispatch(config, "gpt-5.6-terra", "lane-256")).comboId).toBe("terra"); + } finally { + clock.mockRestore(); + } + }); + + test(`${version} virtual Pro target recalls the emitted base model`, async () => { + const config = comboTestConfig(); + config.providers["openai-apikey"] = { + adapter: "openai-responses", baseUrl: "https://api.openai.com/v1", authMode: "key", apiKey: "test-key", + }; + config.combos!.terra!.targets = [{ provider: "openai-apikey", model: "gpt-5.6-terra-pro" }]; + const calls: Array<{ url: string; body: Record }> = []; + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + const body = await request.json() as Record; + calls.push({ url: request.url, body }); + const completed = { ...completedPayload("summary"), model: "gpt-5.6-terra" }; + return body.stream ? sseResponse([{ type: "response.completed", response: completed }]) : jsonResponse(completed); + }) as typeof fetch; + await seedRecall(config); + expect(calls[0]!.body).toMatchObject({ model: "gpt-5.6-terra", reasoning: { mode: "pro" } }); + expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra-pro")).toBeUndefined(); + expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBe("terra"); + expect(calls.every(call => call.url.endsWith("/responses"))).toBe(true); + }); + + test(`${version} recalled combo resolves the current key rather than retaining a credential`, async () => { + const config = comboTestConfig(); + const auth: Array = []; + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init); + auth.push(request.headers.get("authorization")); + const body = await request.json() as { stream?: boolean }; + return body.stream ? chatStreamResponse("summary") : jsonResponse(chatCompletionPayload("answer")); + }) as typeof fetch; + await seedRecall(config); + config.providers.gw!.apiKey = "key-current"; + expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBe("terra"); + expect(auth).toEqual(["Bearer key-gw", "Bearer key-current"]); + }); + } + }); test("a no-eligible policy compact request persists the evaluation trace", async () => { diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index e4523aabb3..c3e2fa1809 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -38,7 +38,9 @@ import { import { clearCursorThreadContinuityForTests } from "../../src/adapters/cursor/thread-continuity"; import { COMPACT_PROMPT, encodeCompactionSummary } from "../../src/responses/compaction"; import { clearKeyCooldowns } from "../../src/providers/key-failover"; -import { consumeComboFailure } from "../../src/server/responses/core"; +import { consumeComboFailure, createChildPassthroughCallbackGate } from "../../src/server/responses/core"; +import { clearComboRecallForTests, recallComboForLane, reconcileComboRecall } from "../../src/server/responses/combo-session-recall"; +import { captureConfigGeneration } from "../../src/lib/state-store-sweeper"; // Full-suite Windows load: startServer + combo rename/delete management flows exceed the // default 5s per-test budget (same flake class as 810fa115 / claude-management-api). @@ -139,6 +141,7 @@ beforeEach(() => { testDir = mkdtempSync(join(tmpdir(), "ocx-combo-030-")); process.env.OPENCODEX_HOME = testDir; clearComboSelectionState(); + clearComboRecallForTests(); clearComboTargetCooldowns(); clearKeyCooldowns(); clearCodexUpstreamHealth(); @@ -171,6 +174,7 @@ afterEach(async () => { isolatedCodexHome = null; if (testDir) removeTreeWithRetry(testDir); clearComboSelectionState(); + clearComboRecallForTests(); clearComboTargetCooldowns(); clearKeyCooldowns(); clearCodexUpstreamHealth(); @@ -432,6 +436,241 @@ function heldNativeTerminal(payload: Record) { } describe("server combo failover 030 activation matrix", () => { + test("recall completion has an independent gate slot and publishes once only on commit", () => { + const calls: string[] = []; + const gate = createChildPassthroughCallbackGate({ + onNativePassthroughTerminal: status => calls.push(status), + onResponseComplete: model => calls.push(model), + }); + gate.onTerminal("completed"); + gate.onResponseComplete("final-model"); + expect(calls).toEqual([]); + gate.commit(); + gate.commit(); + gate.onResponseComplete("duplicate-model"); + expect(calls).toEqual(["completed", "final-model"]); + }); + + for (const rejection of ["discard", "failed", "incomplete", "cancel"] as const) { + test(`recall gate drops pre-commit completion on ${rejection}`, () => { + const models: string[] = []; + const gate = createChildPassthroughCallbackGate({ onResponseComplete: model => models.push(model) }); + gate.onResponseComplete("unaccepted-model"); + if (rejection === "discard") gate.discard(); + else if (rejection === "cancel") gate.onCancel(); + else gate.onTerminal(rejection); + gate.commit(); + gate.onResponseComplete("late-model"); + expect(models).toEqual([]); + }); + } + + for (const wire of ["native", "chat", "runTurn"] as const) { + for (const stream of [false, true]) { + for (const terminal of ["completed", "failed", "incomplete"] as const) { + test(`${wire} ${stream ? "SSE" : "JSON"} ${terminal} B replaces A only after completed response`, async () => { + const encode = (events: Array>) => events.map(event => + `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""); + const upstream = serve(async request => { + const body = await request.json() as { model: string; stream?: boolean }; + if (body.model === "m1") return Response.json(responsesSuccess("A", "m1")); + if (wire === "native") { + const response = { ...responsesSuccess("B output", "final-b"), status: terminal }; + return stream ? new Response(encode([ + { type: "response.output_text.delta", delta: "B output", item_id: "msg_b", output_index: 0, content_index: 0 }, + { type: `response.${terminal}`, response }, + ]), { headers: { "content-type": "text/event-stream" } }) : Response.json(response); + } + if (stream) { + if (terminal === "failed") return chatErrorStream("failed after output", "B output"); + return new Response([ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { content: "B output" }, finish_reason: null }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: terminal === "incomplete" ? "length" : "stop" }] })}\n\n`, + "data: [DONE]\n\n", + ].join(""), { headers: { "content-type": "text/event-stream" } }); + } + if (terminal === "failed") return Response.json({ error: { message: "failed B" } }); + return Response.json({ choices: [{ index: 0, message: { role: "assistant", content: "B output" }, finish_reason: terminal === "incomplete" ? "length" : "stop" }] }); + }); + customRunTurn = async (_parsed, _incoming, emit) => { + emit({ type: "text_delta", text: "B output" }); + if (terminal === "failed") emit({ type: "error", message: "failed after output" }); + else emit({ type: "done", ...(terminal === "incomplete" ? { stopReason: "length" } : {}) }); + }; + const config = comboConfig({ + a: provider("openai-responses", baseUrl(upstream), "key-a"), + b: provider(wire === "native" ? "openai-responses" : wire === "chat" ? "openai-chat" : "test-run-turn", baseUrl(upstream), "key-b"), + }); + config.combos = { + alpha: { targets: [{ provider: "a", model: "m1" }] }, + beta: { targets: [{ provider: "b", model: "m2" }] }, + }; + const headers = { session_id: "terminal-recall" }; + const a = await post(config, { model: "combo/alpha" }, {}, headers); + expect(await a.json()).toMatchObject({ status: "completed", model: "m1" }); + expect(recallComboForLane(config, "terminal-recall", "m1")).toBe("alpha"); + const models: string[] = []; + const completed = deferred(); + const b = await post(config, { model: "combo/beta", stream }, { + onResponseComplete: model => { models.push(model); completed.resolve(); }, + }, headers); + const body = await b.text(); + if (terminal === "completed") { + await within(completed.promise); + const expected = wire === "native" ? "final-b" : "m2"; + expect(body).toContain(`"model":"${expected}"`); + expect(models).toEqual([expected]); + expect(recallComboForLane(config, "terminal-recall", expected)).toBe("beta"); + expect(recallComboForLane(config, "terminal-recall", "m1")).toBeUndefined(); + } else { + expect(models).toEqual([]); + expect(recallComboForLane(config, "terminal-recall", "m1")).toBe("alpha"); + } + }); + } + } + } + + for (const recordTerminalOutcomes of [true, false]) { + for (const scenario of ["completed", "missing-model", "empty-model", "failed-first", "incomplete-first", "undeclared-tool"] as const) { + test(`native SSE recall ${scenario} with terminal recording ${recordTerminalOutcomes}`, async () => { + const upstream = serve(() => { + const response = responsesSuccess("B", "final-b"); + if (scenario === "missing-model") delete response.model; + if (scenario === "empty-model") response.model = ""; + const events: Array> = [{ type: "response.output_text.delta", delta: "B", item_id: "msg_b", output_index: 0, content_index: 0 }]; + if (scenario === "failed-first" || scenario === "incomplete-first") { + const status = scenario === "failed-first" ? "failed" : "incomplete"; + events.push({ type: `response.${status}`, response: { ...response, status } }); + } + if (scenario === "undeclared-tool") events.push({ + type: "response.output_item.added", output_index: 0, + item: { type: "function_call", id: "fc_bad", call_id: "bad", name: "not_declared", arguments: "{}" }, + }); + // Empty terminal output cannot erase an earlier rejected tool call. + events.push({ type: "response.completed", response: { ...response, output: [] } }); + return new Response(events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""), { + headers: { "content-type": "text/event-stream" }, + }); + }); + const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(seed), "key-a"), + b: provider("openai-responses", baseUrl(upstream), "key-b"), + }); + config.combos = { + alpha: { targets: [{ provider: "a", model: "m1" }] }, + beta: { targets: [{ provider: "b", model: "m2" }] }, + }; + const headers = { session_id: "native-sticky" }; + await (await post(config, { model: "combo/alpha" }, {}, headers)).text(); + const completed = deferred(); + const models: string[] = []; + const response = await post(config, { model: "combo/beta", stream: true, tools: [] }, { + recordTerminalOutcomes, + onResponseComplete: model => { models.push(model); completed.resolve(); }, + }, headers); + await response.text(); + if (scenario === "completed") { + await within(completed.promise); + expect(models).toEqual(["final-b"]); + expect(recallComboForLane(config, "native-sticky", "final-b")).toBe("beta"); + } else { + expect(models).toEqual([]); + expect(recallComboForLane(config, "native-sticky", "m1")).toBe("alpha"); + } + }); + } + } + + test("native output before cancellation preserves A and cannot record late B completion", async () => { + const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); + const held = heldNativeTerminal({ type: "response.completed", response: responsesSuccess("B", "final-b") }); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(seed), "key-a"), + b: provider("openai-responses", baseUrl(held.upstream), "key-b"), + }); + config.combos = { + alpha: { targets: [{ provider: "a", model: "m1" }] }, + beta: { targets: [{ provider: "b", model: "m2" }] }, + }; + const headers = { session_id: "cancel-recall" }; + await (await post(config, { model: "combo/alpha" }, {}, headers)).text(); + const abort = new AbortController(); + const models: string[] = []; + const response = await post(config, { model: "combo/beta", stream: true }, { + abortSignal: abort.signal, onResponseComplete: model => models.push(model), + }, headers); + expect(recallComboForLane(config, "cancel-recall", "m1")).toBe("alpha"); + abort.abort(); + held.release(); + await response.text().catch(() => undefined); + expect(models).toEqual([]); + expect(recallComboForLane(config, "cancel-recall", "m1")).toBe("alpha"); + }); + + test("a held completed response cannot resurrect recall across combo delete and recreate", async () => { + const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); + const held = heldNativeTerminal({ type: "response.completed", response: responsesSuccess("B", "final-b") }); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(seed), "key-a"), + b: provider("openai-responses", baseUrl(held.upstream), "key-b"), + }); + config.combos = { + alpha: { targets: [{ provider: "a", model: "m1" }] }, + beta: { targets: [{ provider: "b", model: "m2" }] }, + }; + const headers = { session_id: "recreated-recall" }; + await (await post(config, { model: "combo/alpha" }, {}, headers)).text(); + const completed = deferred(); + const response = await post(config, { model: "combo/beta", stream: true }, { + onResponseComplete: () => completed.resolve(), + }, headers); + const generation = captureConfigGeneration(); + delete config.combos.beta; + const owners = { + generation: generation + 1, + providerNames: new Set(["a", "b"]), comboIds: new Set(["alpha"]), comboTargets: new Set(["alpha::a/m1"]), + codexAccountIds: new Set(), oauthAccountKeys: new Set(), configRoots: new Set(), + }; + reconcileComboRecall(owners); + config.combos.beta = { targets: [{ provider: "b", model: "m2" }] }; + reconcileComboRecall({ + ...owners, generation: generation + 2, + comboIds: new Set(["alpha", "beta"]), comboTargets: new Set(["alpha::a/m1", "beta::b/m2"]), + }); + held.release(); + await response.text(); + await within(completed.promise); + expect(recallComboForLane(config, "recreated-recall", "m1")).toBe("alpha"); + expect(recallComboForLane(config, "recreated-recall", "final-b")).toBeUndefined(); + }); + + for (const media of ["image", "video"] as const) { + test(`${media} bridge completion records the final response model`, async () => { + const tools: string[] = []; + const routed = serve(async request => { + const body = await request.json() as { tools?: Array<{ function?: { name?: string } }> }; + tools.push(...(body.tools ?? []).map(tool => tool.function?.name ?? "")); + return chatStream("media bridge answer"); + }); + const config = comboConfig({ + a: provider("openai-chat", baseUrl(routed), "key-a"), + xai: provider("openai-chat", "https://api.x.ai/v1", "synthetic-xai-key"), + }, [{ provider: "a", model: "m1" }]); + config.images = media === "image" ? { bridgeEnabled: true } : { videoBridgeEnabled: true }; + const models: string[] = []; + const response = await post(config, { + stream: true, ...(media === "image" ? { tools: [{ type: "image_generation" }] } : {}), + }, { onResponseComplete: model => models.push(model) }, { session_id: "media-recall" }); + const frames = await collectSse(response); + expect(tools).toContain(media === "image" ? "image_gen" : "video_gen"); + expect(frames.some(frame => frame.event === "response.completed")).toBe(true); + expect(models).toEqual(["m1"]); + expect(recallComboForLane(config, "media-recall", "m1")).toBe("free"); + }); + } + test("dispatches a selected concrete target despite a shadowing combo alias", async () => { const hits: string[] = []; const a = serve(async request => { @@ -1596,10 +1835,12 @@ describe("server combo failover 030 activation matrix", () => { { provider: "b", model: "m2" }, ]); config.webSearchSidecar = { enabled: true, backend: "openai" }; + const models: string[] = []; const response = await post(config, { stream: true, tools: [{ type: "web_search" }], - }, {}, { + }, { onResponseComplete: model => models.push(model) }, { + session_id: "web-search-recall", authorization: `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "acct-combo-search" })}`, "chatgpt-account-id": "acct-combo-search", }); @@ -1607,6 +1848,8 @@ describe("server combo failover 030 activation matrix", () => { expect(JSON.stringify(await collectSse(response))).toContain("web loop backup"); expect(modelHits.map(hit => hit.model)).toEqual(["m1", "m2"]); expect(modelHits.every(hit => hit.hasWebTool)).toBe(true); + expect(models).toEqual(["m2"]); + expect(recallComboForLane(config, "web-search-recall", "m2")).toBe("free"); }); test("context 400 stops while exhausted retryable targets return the sanitized last status", async () => { @@ -2881,7 +3124,7 @@ describe("server combo failover 030 activation matrix", () => { const terminalFrame = (status: "failed" | "completed") => [ `event: response.${status}`, `data: ${JSON.stringify({ type: `response.${status}`, response: { - id: `resp_${status}`, status, output: [], + id: `resp_${status}`, status, model: status === "completed" ? "final-b" : "failed-a", output: [], ...(status === "failed" ? { error: { code: "rate_limit_exceeded", message: "discarded quota failure" } } : {}), } })}`, "", @@ -2900,13 +3143,16 @@ describe("server combo failover 030 activation matrix", () => { }); const finalized = deferred(); const statuses: string[] = []; + const models: string[] = []; + const completed = deferred(); let cancels = 0; const parent: RequestLogContext = { model: "", provider: "" }; const snapshots: RequestLogContext[] = []; const response = await handleResponses(new Request("http://localhost/v1/responses", { - method: "POST", headers: { "content-type": "application/json" }, + method: "POST", headers: { "content-type": "application/json", session_id: "hop-recall" }, body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), }), config, parent, { + onResponseComplete: model => { models.push(model); completed.resolve(); }, onNativePassthroughTerminal: status => { statuses.push(status); snapshots.push({ ...parent }); @@ -2917,6 +3163,10 @@ describe("server combo failover 030 activation matrix", () => { expect(response.status).toBe(200); await response.text(); await within(finalized.promise); + await within(completed.promise); + expect(models).toEqual(["final-b"]); + expect(recallComboForLane(config, "hop-recall", "final-b")).toBe("free"); + expect(recallComboForLane(config, "hop-recall", "m1")).toBeUndefined(); expect(statuses).toEqual(["completed"]); expect(cancels).toBe(0); expect(snapshots).toHaveLength(1); From cfe687d9af15e977912e774092615403cf3ebb45 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:51:47 +0900 Subject: [PATCH 3/9] docs(combos): describe bounded successful compaction recall --- docs-site/src/content/docs/guides/combos.md | 13 +++++++++++++ docs-site/src/content/docs/ko/guides/combos.md | 6 ++++++ structure/04_transports-and-sidecars.md | 7 +++++++ 3 files changed, 26 insertions(+) diff --git a/docs-site/src/content/docs/guides/combos.md b/docs-site/src/content/docs/guides/combos.md index 031766373e..db94da045f 100644 --- a/docs-site/src/content/docs/guides/combos.md +++ b/docs-site/src/content/docs/guides/combos.md @@ -74,6 +74,19 @@ Aliases change the public name clients request; they do not change the combo's s concrete provider/model selectors behind it. ::: +## Compaction after switching combos + +When a client compacts using a bare model name after switching combos, opencodex can recall the +combo that most recently completed successfully on that conversation lane. The model must match +the completed response, and the combo and its target must still exist in the current configuration. +The request then follows normal combo selection and failover. + +Explicit provider/combo selectors and configured combo aliases take precedence over this recall. +Failed, incomplete, or cancelled responses do not replace the last successful selection. Recall is +process-local and bounded to 256 lanes for 30 minutes; it does not store account credentials. +Without usable conversation identity or valid remembered state, normal compaction routing applies. +A restart clears the remembered state. + ## Codex Desktop native-allowlist compatibility Some Codex Desktop releases apply a remote native-only `available_models` allowlist after the diff --git a/docs-site/src/content/docs/ko/guides/combos.md b/docs-site/src/content/docs/ko/guides/combos.md index aef5ca1cc1..633feb838b 100644 --- a/docs-site/src/content/docs/ko/guides/combos.md +++ b/docs-site/src/content/docs/ko/guides/combos.md @@ -62,6 +62,12 @@ alias를 설정해도 정식 `combo/` 형식은 계속 해석됩니다. 정 alias는 클라이언트가 요청하는 공개 이름만 바꿉니다. 콤보에 저장된 ID나 그 뒤의 실제 공급자/모델 선택자는 바꾸지 않습니다. ::: +## 콤보를 바꾼 뒤 대화 압축 + +클라이언트가 콤보를 바꾼 뒤 공급자 접두사 없는 모델 이름으로 압축을 요청하면, opencodex는 같은 대화에서 가장 최근에 응답을 성공적으로 마친 콤보를 기억해 사용할 수 있습니다. 모델 이름이 완료된 응답과 일치하고, 현재 설정에 해당 콤보와 대상이 남아 있어야 합니다. 압축 요청도 일반 콤보 선택과 페일오버를 따릅니다. + +명시한 공급자·콤보 선택자와 설정된 콤보 별칭이 기억한 값보다 우선합니다. 실패·미완료·취소된 응답은 마지막 성공 기록을 덮어쓰지 않습니다. 기록은 프로세스 안에서 최대 256개 대화, 30분 동안 유지하며 계정 자격증명을 저장하지 않습니다. 유효한 대화 식별자나 기록이 없으면 일반 압축 라우팅을 사용합니다. 재시작하면 기록은 사라집니다. + ## 전략 선택 ### 페일오버: 순서가 있는 기본값과 예비값 diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 256bd1eae5..d18adb198d 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -242,6 +242,13 @@ alone never opt a gateway in. and before the `/v1/*` guard. Unknown `/v1/*` paths return JSON 404 errors instead of falling through to GUI static serving. +Combo compaction recall uses accepted completed-response callbacks to record the final client-visible +model and originating combo target. The existing child callback gate defers publication until an +attempt is accepted and drops discarded/failed attempts. Both compaction entry points preserve +explicit configured selectors before consulting bounded lane state. The existing state-store +reconciliation owns removal of obsolete targets and generation fencing; core imports no registration +composition root or Lab code. Recall retains routing identity only, never account credentials. + [Decision Log] - 목적과 의도: Complete Cursor turns at the protocol terminal instead of waiting for a separate HTTP-body EOF that may never arrive. - 기존 구현 및 제약 조건: Cursor can send turnEnded followed by a clean Connect END_STREAM envelope while RunSSE remains open or later closes through an abort-shaped transport error. The adapter logged the clean envelope but did not settle its terminal owner, so a completed-looking turn could remain open until the Responses stall watchdog. From d6f013698cf2a509aa0aff770848667155c3686e Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:54:46 +0900 Subject: [PATCH 4/9] test(responses): use normalized request lanes in combo recall assertions --- .../responses-compaction-routing.test.ts | 11 +++---- .../server/server-combo-failover-e2e.test.ts | 29 ++++++++++--------- 2 files changed, 21 insertions(+), 19 deletions(-) diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index 63a734bca9..ec014470ed 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -1,3 +1,4 @@ +import { sessionLaneIdFromRequest } from "../../src/server/request-log-conversation"; /** * Issue #422: a Responses-shaped wire does not imply support for Codex's private * `compaction_trigger` item. Only the canonical ChatGPT backend speaks that @@ -987,7 +988,7 @@ describe("compact alternate-account attempt (#913)", () => { config, { model: "", provider: "" }, { onResponseComplete: complete }); await seed.text(); await completed; - expect(recallComboForLane(config, "account-recall", "gpt-5.5")).toBe("native"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "account-recall" })), "gpt-5.5")).toBe("native"); config.activeCodexAccountId = "pool-b"; const compact = version === "v1" ? handleResponsesCompact : handleResponses; const log: RequestLogContext = { model: "", provider: "" }; @@ -2066,7 +2067,7 @@ describe("compaction combo recall after combo switch (#3891)", () => { targets: [{ provider: "alt", model: "gpt-5.6-luna" }], }; await seedRecall(config); - expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra")).toBe("terra"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "recall-lane" })), "gpt-5.6-terra")).toBe("terra"); const log = await dispatch(config, "gpt-5.6-terra"); expect(log.comboId).toBe("explicit"); expect(log.resolvedModel).toBe("gpt-5.6-luna"); @@ -2119,12 +2120,12 @@ describe("compaction combo recall after combo switch (#3891)", () => { try { await seedRecall(config); now += 30 * 60 * 1000 - 1; - expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra")).toBe("terra"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "recall-lane" })), "gpt-5.6-terra")).toBe("terra"); now += 1; expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBeUndefined(); const target = { provider: "gw", model: "gpt-5.6-terra" }; for (let index = 0; index < 257; index += 1) { - rememberComboForLane(`lane-${index}`, "terra", target, "gpt-5.6-terra", captureConfigGeneration()); + rememberComboForLane(sessionLaneIdFromRequest(new Headers({ session_id: `lane-${index}` })), "terra", target, "gpt-5.6-terra", captureConfigGeneration()); } expect((await dispatch(config, "gpt-5.6-terra", "lane-0")).comboId).toBeUndefined(); expect((await dispatch(config, "gpt-5.6-terra", "lane-1")).comboId).toBe("terra"); @@ -2150,7 +2151,7 @@ describe("compaction combo recall after combo switch (#3891)", () => { }) as typeof fetch; await seedRecall(config); expect(calls[0]!.body).toMatchObject({ model: "gpt-5.6-terra", reasoning: { mode: "pro" } }); - expect(recallComboForLane(config, "recall-lane", "gpt-5.6-terra-pro")).toBeUndefined(); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "recall-lane" })), "gpt-5.6-terra-pro")).toBeUndefined(); expect((await dispatch(config, "gpt-5.6-terra")).comboId).toBe("terra"); expect(calls.every(call => call.url.endsWith("/responses"))).toBe(true); }); diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index c3e2fa1809..2aa90bd8aa 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -1,3 +1,4 @@ +import { sessionLaneIdFromRequest } from "../../src/server/request-log-conversation"; import { afterEach, beforeEach, describe, expect, mock, setDefaultTimeout, test } from "bun:test"; import { logsFromApiBody } from "../helpers/logs-api"; import { managementFetch as fetch, ManagementRequest as Request } from "../helpers/management-auth"; @@ -508,7 +509,7 @@ describe("server combo failover 030 activation matrix", () => { const headers = { session_id: "terminal-recall" }; const a = await post(config, { model: "combo/alpha" }, {}, headers); expect(await a.json()).toMatchObject({ status: "completed", model: "m1" }); - expect(recallComboForLane(config, "terminal-recall", "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "terminal-recall" })), "m1")).toBe("alpha"); const models: string[] = []; const completed = deferred(); const b = await post(config, { model: "combo/beta", stream }, { @@ -520,11 +521,11 @@ describe("server combo failover 030 activation matrix", () => { const expected = wire === "native" ? "final-b" : "m2"; expect(body).toContain(`"model":"${expected}"`); expect(models).toEqual([expected]); - expect(recallComboForLane(config, "terminal-recall", expected)).toBe("beta"); - expect(recallComboForLane(config, "terminal-recall", "m1")).toBeUndefined(); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "terminal-recall" })), expected)).toBe("beta"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "terminal-recall" })), "m1")).toBeUndefined(); } else { expect(models).toEqual([]); - expect(recallComboForLane(config, "terminal-recall", "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "terminal-recall" })), "m1")).toBe("alpha"); } }); } @@ -574,10 +575,10 @@ describe("server combo failover 030 activation matrix", () => { if (scenario === "completed") { await within(completed.promise); expect(models).toEqual(["final-b"]); - expect(recallComboForLane(config, "native-sticky", "final-b")).toBe("beta"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "native-sticky" })), "final-b")).toBe("beta"); } else { expect(models).toEqual([]); - expect(recallComboForLane(config, "native-sticky", "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "native-sticky" })), "m1")).toBe("alpha"); } }); } @@ -601,12 +602,12 @@ describe("server combo failover 030 activation matrix", () => { const response = await post(config, { model: "combo/beta", stream: true }, { abortSignal: abort.signal, onResponseComplete: model => models.push(model), }, headers); - expect(recallComboForLane(config, "cancel-recall", "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "cancel-recall" })), "m1")).toBe("alpha"); abort.abort(); held.release(); await response.text().catch(() => undefined); expect(models).toEqual([]); - expect(recallComboForLane(config, "cancel-recall", "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "cancel-recall" })), "m1")).toBe("alpha"); }); test("a held completed response cannot resurrect recall across combo delete and recreate", async () => { @@ -642,8 +643,8 @@ describe("server combo failover 030 activation matrix", () => { held.release(); await response.text(); await within(completed.promise); - expect(recallComboForLane(config, "recreated-recall", "m1")).toBe("alpha"); - expect(recallComboForLane(config, "recreated-recall", "final-b")).toBeUndefined(); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "recreated-recall" })), "m1")).toBe("alpha"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "recreated-recall" })), "final-b")).toBeUndefined(); }); for (const media of ["image", "video"] as const) { @@ -667,7 +668,7 @@ describe("server combo failover 030 activation matrix", () => { expect(tools).toContain(media === "image" ? "image_gen" : "video_gen"); expect(frames.some(frame => frame.event === "response.completed")).toBe(true); expect(models).toEqual(["m1"]); - expect(recallComboForLane(config, "media-recall", "m1")).toBe("free"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "media-recall" })), "m1")).toBe("free"); }); } @@ -1849,7 +1850,7 @@ describe("server combo failover 030 activation matrix", () => { expect(modelHits.map(hit => hit.model)).toEqual(["m1", "m2"]); expect(modelHits.every(hit => hit.hasWebTool)).toBe(true); expect(models).toEqual(["m2"]); - expect(recallComboForLane(config, "web-search-recall", "m2")).toBe("free"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "web-search-recall" })), "m2")).toBe("free"); }); test("context 400 stops while exhausted retryable targets return the sanitized last status", async () => { @@ -3165,8 +3166,8 @@ describe("server combo failover 030 activation matrix", () => { await within(finalized.promise); await within(completed.promise); expect(models).toEqual(["final-b"]); - expect(recallComboForLane(config, "hop-recall", "final-b")).toBe("free"); - expect(recallComboForLane(config, "hop-recall", "m1")).toBeUndefined(); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "hop-recall" })), "final-b")).toBe("free"); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "hop-recall" })), "m1")).toBeUndefined(); expect(statuses).toEqual(["completed"]); expect(cancels).toBe(0); expect(snapshots).toHaveLength(1); From 7c0c997d636c2b3dd8fd1af2ed87751ad88d10c4 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:56:22 +0900 Subject: [PATCH 5/9] fix(responses): bind combo recall to the first visible completion --- src/server/responses/core.ts | 7 ++- .../server/server-combo-failover-e2e.test.ts | 45 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index a86053d7af..5bdfc4b98c 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -4660,6 +4660,7 @@ async function handleResponsesInner( // rejection is sticky for the whole turn, set from every parsed payload on the inspection side. let inspectionSawUndeclaredTool = false; let inspectedTerminal: ResponsesTerminalStatus | null = null; + let inspectedCompletionSeen = false; const passiveQuotaObserved = hasPassiveAccountQuota(route.providerName) && route.provider.authMode === "oauth"; const noteInspectedPayload = (payload: unknown) => { @@ -4729,7 +4730,11 @@ async function handleResponsesInner( return; } rememberPassthroughResponse?.(restoredResponse); - if (inspectedTerminal === null || inspectedTerminal === "completed") { + const firstCompletion = !inspectedCompletionSeen; + inspectedCompletionSeen = true; + if (firstCompletion && (inspectedTerminal === null || inspectedTerminal === "completed")) { + // A model-less first completion permanently declines recall; later terminal + // frames are hidden by the client boundary and cannot supply its identity. // Native inspection sees the pre-rewrite model. Only an actual terminal // model can seed recall; an absent model never falls back to the pick. if (typeof response.model === "string" && response.model.trim()) { diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index 2aa90bd8aa..5903e38d64 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -584,6 +584,51 @@ describe("server combo failover 030 activation matrix", () => { } } + for (const streamMode of ["legacy-tee", "eager-relay"] as const) { + for (const recordTerminalOutcomes of [true, false]) { + for (const firstModel of [undefined, ""] as const) { + test(`native ${streamMode} ignores hidden completion after ${firstModel === undefined ? "missing" : "empty"} model with recording ${recordTerminalOutcomes}`, async () => { + const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); + const upstream = serve(() => { + const first = responsesSuccess("first", "ignored"); + if (firstModel === undefined) delete first.model; + else first.model = firstModel; + const events = [ + { type: "response.output_text.delta", delta: "B", item_id: "msg_b", output_index: 0, content_index: 0 }, + { type: "response.completed", response: first }, + { type: "response.completed", response: responsesSuccess("hidden", "final-b") }, + ]; + return new Response(events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""), { + headers: { "content-type": "text/event-stream" }, + }); + }); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(seed), "key-a"), + b: provider("openai-responses", baseUrl(upstream), "key-b"), + }); + config.streamMode = streamMode; + config.combos = { + alpha: { targets: [{ provider: "a", model: "m1" }] }, + beta: { targets: [{ provider: "b", model: "m2" }] }, + }; + const headers = { session_id: "first-terminal-recall" }; + const lane = sessionLaneIdFromRequest(new Headers(headers)); + await (await post(config, { model: "combo/alpha" }, {}, headers)).text(); + expect(recallComboForLane(config, lane, "m1")).toBe("alpha"); + const completedModels: string[] = []; + const response = await post(config, { model: "combo/beta", stream: true }, { + recordTerminalOutcomes, onResponseComplete: model => { completedModels.push(model); }, + }, headers); + const wire = await response.text(); + expect(wire).not.toContain("final-b"); + expect(completedModels).toEqual([]); + expect(recallComboForLane(config, lane, "m1")).toBe("alpha"); + expect(recallComboForLane(config, lane, "final-b")).toBeUndefined(); + }); + } + } + } + test("native output before cancellation preserves A and cannot record late B completion", async () => { const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); const held = heldNativeTerminal({ type: "response.completed", response: responsesSuccess("B", "final-b") }); From f26b734f5f08dc08e7e63e524a846f79ac4f832a Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:58:52 +0900 Subject: [PATCH 6/9] fix(responses): consume recall eligibility at first terminal observation --- src/server/responses/core.ts | 16 ++++++++++++++-- tests/server/server-combo-failover-e2e.test.ts | 11 +++++++---- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 5bdfc4b98c..7c1302178a 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -4661,12 +4661,24 @@ async function handleResponsesInner( let inspectionSawUndeclaredTool = false; let inspectedTerminal: ResponsesTerminalStatus | null = null; let inspectedCompletionSeen = false; + let firstTerminalAllowsRecall = false; const passiveQuotaObserved = hasPassiveAccountQuota(route.providerName) && route.provider.authMode === "oauth"; const noteInspectedPayload = (payload: unknown) => { // First terminal stays authoritative even in metadata-only inspection, which // intentionally continues parsing after a failed/incomplete terminal. - inspectedTerminal ??= terminalStatusFromParsed(payload); + const terminal = terminalStatusFromParsed(payload); + if (inspectedTerminal === null && terminal !== null) { + inspectedTerminal = terminal; + // The client boundary accepts a terminal by event type, even without a + // response object. Such a terminal must permanently decline recall. + if (terminal === "completed" && payload && typeof payload === "object" + && "response" in payload && payload.response && typeof payload.response === "object" + && !Array.isArray(payload.response) && "model" in payload.response) { + firstTerminalAllowsRecall = typeof payload.response.model === "string" + && payload.response.model.trim().length > 0; + } + } // Meta reports subscription usage ONLY as an in-stream event; there is no endpoint // to poll (003 §E probed 17 paths, all 404). Observed here rather than behind a // dedicated inspector handler because onParsedPayload already reaches every @@ -4732,7 +4744,7 @@ async function handleResponsesInner( rememberPassthroughResponse?.(restoredResponse); const firstCompletion = !inspectedCompletionSeen; inspectedCompletionSeen = true; - if (firstCompletion && (inspectedTerminal === null || inspectedTerminal === "completed")) { + if (firstCompletion && (inspectedTerminal === null || firstTerminalAllowsRecall)) { // A model-less first completion permanently declines recall; later terminal // frames are hidden by the client boundary and cannot supply its identity. // Native inspection sees the pre-rewrite model. Only an actual terminal diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index 5903e38d64..38bfc4267b 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -586,16 +586,19 @@ describe("server combo failover 030 activation matrix", () => { for (const streamMode of ["legacy-tee", "eager-relay"] as const) { for (const recordTerminalOutcomes of [true, false]) { - for (const firstModel of [undefined, ""] as const) { - test(`native ${streamMode} ignores hidden completion after ${firstModel === undefined ? "missing" : "empty"} model with recording ${recordTerminalOutcomes}`, async () => { + for (const firstModel of [undefined, "", "missing-response", "null-response"] as const) { + test(`native ${streamMode} ignores hidden completion after ${firstModel === undefined ? "missing-model" : firstModel || "empty-model"} with recording ${recordTerminalOutcomes}`, async () => { const seed = serve(() => Response.json(responsesSuccess("A", "m1"))); const upstream = serve(() => { const first = responsesSuccess("first", "ignored"); if (firstModel === undefined) delete first.model; - else first.model = firstModel; + else if (firstModel === "") first.model = firstModel; + const firstEvent: Record = { type: "response.completed", response: first }; + if (firstModel === "missing-response") delete firstEvent.response; + else if (firstModel === "null-response") firstEvent.response = null; const events = [ { type: "response.output_text.delta", delta: "B", item_id: "msg_b", output_index: 0, content_index: 0 }, - { type: "response.completed", response: first }, + firstEvent, { type: "response.completed", response: responsesSuccess("hidden", "final-b") }, ]; return new Response(events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""), { From de638a3d07ed29409b7cb3b86717e39bab11a6f9 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:01:19 +0900 Subject: [PATCH 7/9] test(responses): retain abort guard with completion cancellation callback --- tests/responses/passthrough-abort.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/responses/passthrough-abort.test.ts b/tests/responses/passthrough-abort.test.ts index 6fdc468b1b..46100c6902 100644 --- a/tests/responses/passthrough-abort.test.ts +++ b/tests/responses/passthrough-abort.test.ts @@ -79,7 +79,7 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(sseBranch).toContain("rewriteBlocks: clientBlockRewrite"); // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed. expect(sseBranch).toMatch( - /relaySseWithFailedTail\(\s*rewrittenBody,\s*upstream,\s*reason\s*=>\s*clientGone\.abort\(reason\),\s*\{\s*upstreamError:\s*logCtx\.upstreamError\s*\},\s*\)/, + /relaySseWithFailedTail\(\s*rewrittenBody,\s*upstream,\s*reason\s*=>\s*\{\s*responseCompletionCancelled\s*=\s*true;\s*clientGone\.abort\(reason\);\s*\},\s*\{\s*upstreamError:\s*logCtx\.upstreamError\s*\},\s*\)/, ); expect(sseBranch).toContain("new Response(clientBody"); expect(sseBranch).toContain("markNativePassthroughSseResponse"); From e8a167bd5016ed76e3fd5456508c753c8d3bf9fd Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:10:09 +0900 Subject: [PATCH 8/9] test(responses): distinguish configured and emitted failover models --- tests/server/server-combo-failover-e2e.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index 38bfc4267b..c08c706bac 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -3219,7 +3219,7 @@ describe("server combo failover 030 activation matrix", () => { expect(statuses).toEqual(["completed"]); expect(cancels).toBe(0); expect(snapshots).toHaveLength(1); - expect(snapshots[0]).toMatchObject({ provider: "combo", model: "combo/free", resolvedModel: "m2" }); + expect(snapshots[0]).toMatchObject({ provider: "combo", model: "combo/free", resolvedModel: "final-b" }); for (const field of ["terminalHttpStatus", "terminalIncompleteReason", "terminalErrorCode", "upstreamError"] as const) { expect(snapshots[0]![field]).toBeUndefined(); } From 797a74c646b554a1716905c3721dd13dcef0d579 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:20:01 +0900 Subject: [PATCH 9/9] test(responses): isolate combo account recall fixture lifecycle --- .../responses-compaction-routing.test.ts | 34 +++++++++++++++---- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index ec014470ed..fafbbd6806 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -1,3 +1,4 @@ +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; import { sessionLaneIdFromRequest } from "../../src/server/request-log-conversation"; /** * Issue #422: a Responses-shaped wire does not imply support for Codex's private @@ -956,6 +957,8 @@ describe("compact alternate-account attempt (#913)", () => { test(`${version} recalled native combo reselects the current account and respects admission refusal`, async () => { await withPoolEnv("ocx-combo-recall-account-", async config => { clearComboRecallForTests(); + clearComboSelectionState(); + clearComboTargetCooldowns(); config.combos = { native: { targets: [{ provider: "openai", model: "gpt-5.5" }] } }; config.codexAccountNamespaces = { side: "pool-a" }; const headers = { session_id: "account-recall" }; @@ -981,37 +984,54 @@ describe("compact alternate-account attempt (#913)", () => { ...(compact ? { output: [{ type: "compaction", encrypted_content: "native-recall-ciphertext" }] } : {}), } }]); }) as typeof fetch; + const client = new AbortController(); + let completionTimer: ReturnType | undefined; try { let complete!: () => void; const completed = new Promise(resolve => { complete = resolve; }); - const seed = await handleResponses(compactionRequest({ model: "combo/native", stream: true, input: "hello" }, undefined, headers), - config, { model: "", provider: "" }, { onResponseComplete: complete }); - await seed.text(); - await completed; + const seedWork = (async () => { + const seed = await handleResponses(compactionRequest({ model: "combo/native", stream: true, input: "hello" }, client.signal, headers), + config, { model: "", provider: "" }, { onResponseComplete: complete, abortSignal: client.signal }); + expect(seed.status).toBe(200); + await seed.text(); + await completed; + })(); + await Promise.race([ + seedWork, + new Promise((_, reject) => { + completionTimer = setTimeout(() => reject(new Error("native combo seed did not complete")), 10_000); + }), + ]); + clearTimeout(completionTimer); + completionTimer = undefined; expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "account-recall" })), "gpt-5.5")).toBe("native"); config.activeCodexAccountId = "pool-b"; const compact = version === "v1" ? handleResponsesCompact : handleResponses; const log: RequestLogContext = { model: "", provider: "" }; - const response = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), undefined, headers), config, log); + const response = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), client.signal, headers), config, log); expect(response.status).toBe(200); await response.text(); expect(log.comboId).toBe("native"); expect(accounts).toEqual(["pool_acc_a", "pool_acc_b"]); const explicitLog: RequestLogContext = { model: "", provider: "" }; - const explicit = await compact(compactionRequest(baseCompactionBody({ model: "side/gpt-5.5", stream: true }), undefined, headers), config, explicitLog); + const explicit = await compact(compactionRequest(baseCompactionBody({ model: "side/gpt-5.5", stream: true }), client.signal, headers), config, explicitLog); expect(explicit.status).toBe(200); await explicit.text(); expect(explicitLog.comboId).toBeUndefined(); expect(accounts.at(-1)).toBe("pool_acc_a"); const sends = accounts.length; authSpy.mockRejectedValue(new authContextModule.CodexMainProfileDrainingError()); - const refused = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), undefined, headers), config, { model: "", provider: "" }); + const refused = await compact(compactionRequest(baseCompactionBody({ model: "gpt-5.5", stream: true }), client.signal, headers), config, { model: "", provider: "" }); expect(refused.status).toBe(503); expect(accounts).toHaveLength(sends); } finally { + if (completionTimer !== undefined) clearTimeout(completionTimer); + client.abort(); authSpy.mockRestore(); clearComboRecallForTests(); + clearComboSelectionState(); + clearComboTargetCooldowns(); } }); });