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/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 new file mode 100644 index 0000000000..84dfd8d466 --- /dev/null +++ b/src/server/responses/combo-session-recall.ts @@ -0,0 +1,89 @@ +/** 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; + target: Pick; + responseModel: string; + at: number; +} + +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, + target: Pick, + responseModel: string, + writerGeneration: number, +): void { + 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, entry); + while (recall.size > RECALL_CAPACITY) { + const oldest = recall.keys().next().value; + if (oldest === undefined) break; + recall.delete(oldest); + } +} + +export function recallComboForLane( + config: OcxConfig, + lane: string | undefined, + model: string, +): string | undefined { + if (!lane || !model || model.includes("/")) return undefined; + const entry = recall.get(lane); + if (!entry) return undefined; + 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; + } + 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 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 4e7481bb2e..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, @@ -152,6 +153,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 +538,29 @@ 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; + // 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 + // 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..7c1302178a 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, @@ -79,6 +83,7 @@ import { comboRequestHasImageInput, concreteComboRequestBody, getCombo, + resolveComboId, isComboTargetInCooldown, NoAvailableComboTargetsError, noteComboSuccess, @@ -311,6 +316,7 @@ import { consumeForInspection, consumeForResponseLogMetadata, createSseInspector, + terminalStatusFromParsed, isEagerRelaySseResponse, isNativePassthroughSseResponse, markEagerRelaySseResponse, @@ -1666,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; @@ -1866,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 @@ -2694,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 @@ -2737,6 +2778,7 @@ export async function handleComboResponses( onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; }, onNativePassthroughTerminal: callbackGate.onTerminal, onNativePassthroughCancel: callbackGate.onCancel, + onResponseComplete: callbackGate.onResponseComplete, }); } catch (error) { callbackGate.discard(); @@ -3109,6 +3151,23 @@ async function handleResponsesInner( effort: comboEffortRow.effort, }; } + // 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 + && !comboRows.fastRow && !comboEffortRow + && !resolveComboId(config, rawModel)) { + const recalledComboId = recallComboForLane(config, 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?.(); @@ -4312,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, @@ -4589,9 +4659,26 @@ 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; + 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. + 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 @@ -4613,8 +4700,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, @@ -4622,33 +4708,56 @@ 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); + const firstCompletion = !inspectedCompletionSeen; + inspectedCompletionSeen = true; + 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 + // 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( @@ -5656,7 +5765,7 @@ async function handleResponsesInner( const inspector = createSseInspector({ onTerminal: reportNativeTerminal, logCtx, - onCompletedResponse: rememberPassthroughResponseChecked, + onCompletedResponse: rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, onParsedPayload: noteInspectedPayload, onFirstOutput: options.onFirstOutput, pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled, @@ -5686,7 +5795,10 @@ async function handleResponsesInner( reportNativeTerminal("failed", 502); } }, - onClientCancel: () => options.onNativePassthroughCancel?.(), + onClientCancel: () => { + responseCompletionCancelled = true; + options.onNativePassthroughCancel?.(); + }, onDone: () => unregisterTurn(turnAc), }, { clientGoneSignal: options.abortSignal, @@ -5748,8 +5860,11 @@ async function handleResponsesInner( turnAc.signal, () => unregisterTurn(turnAc), logCtx, - () => options.onNativePassthroughCancel?.(), - rememberPassthroughResponseChecked, + () => { + responseCompletionCancelled = true; + options.onNativePassthroughCancel?.(); + }, + rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, options.onFirstOutput, inspectionConsumerOptions, ); @@ -5759,7 +5874,7 @@ async function handleResponsesInner( logCtx, turnAc.signal, () => unregisterTurn(turnAc), - rememberPassthroughResponseChecked, + rememberPassthroughResponse ? rememberPassthroughResponseChecked : undefined, options.onFirstOutput, inspectionConsumerOptions, ); @@ -5774,7 +5889,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, { @@ -5854,13 +5972,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 → @@ -6194,10 +6310,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, @@ -6268,12 +6386,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, @@ -6497,6 +6619,7 @@ async function handleResponsesInner( const sseStream = bridgeToResponsesSSE( guardedSource, parsed._responseModelId ?? parsed.modelId, toolNsMap, freeformToolNames, toolSearchToolNames, () => { + cancelResponseCompletion(); runTurnAbort.abort(); queue.close(); }, 2_000, @@ -6533,6 +6656,7 @@ async function handleResponsesInner( responseStateOptions(adapterNeedsForcedContinuation(adapter.name)), ); } + notifyResponseComplete(response); }, }, ); @@ -6609,6 +6733,7 @@ async function handleResponsesInner( if (adapterResponseReachedServingTerminal(events, json)) { commitReasoningReplayServingRoute(); } + notifyResponseComplete(json); return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } @@ -6656,10 +6781,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 } : {}), }, @@ -6680,12 +6806,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 @@ -7667,7 +7790,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, @@ -7702,6 +7825,7 @@ async function handleResponsesInner( responseStateOptions(activeAdapter.name === "kiro"), ); } + notifyResponseComplete(response); }, }, ); @@ -7781,6 +7905,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/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. 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/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"); diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index 2a1f69be1e..fafbbd6806 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -1,3 +1,5 @@ +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 * `compaction_trigger` item. Only the canonical ChatGPT backend speaks that @@ -35,6 +37,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, 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; @@ -949,6 +953,90 @@ 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(); + clearComboSelectionState(); + clearComboTargetCooldowns(); + 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; + const client = new AbortController(); + let completionTimer: ReturnType | undefined; + try { + let complete!: () => void; + const completed = new Promise(resolve => { complete = resolve; }); + 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 }), 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 }), 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 }), 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(); + } + }); + }); + } + 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 => { @@ -1693,6 +1781,419 @@ 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"), model: "gpt-5.6-terra" } }]) + : jsonResponse({ ...completedPayload("handoff summary"), model: "gpt-5.6-terra" }); + }) 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(); + }); + + 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, 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"); + }); + + 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, 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(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"); + 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, 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); + }); + + 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 () => { const config = { ...keyProviderConfig(), diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index e4523aabb3..c08c706bac 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"; @@ -38,7 +39,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 +142,7 @@ beforeEach(() => { testDir = mkdtempSync(join(tmpdir(), "ocx-combo-030-")); process.env.OPENCODEX_HOME = testDir; clearComboSelectionState(); + clearComboRecallForTests(); clearComboTargetCooldowns(); clearKeyCooldowns(); clearCodexUpstreamHealth(); @@ -171,6 +175,7 @@ afterEach(async () => { isolatedCodexHome = null; if (testDir) removeTreeWithRetry(testDir); clearComboSelectionState(); + clearComboRecallForTests(); clearComboTargetCooldowns(); clearKeyCooldowns(); clearCodexUpstreamHealth(); @@ -432,6 +437,289 @@ 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, 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 }, { + 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, 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, sessionLaneIdFromRequest(new Headers({ session_id: "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, sessionLaneIdFromRequest(new Headers({ session_id: "native-sticky" })), "final-b")).toBe("beta"); + } else { + expect(models).toEqual([]); + expect(recallComboForLane(config, sessionLaneIdFromRequest(new Headers({ session_id: "native-sticky" })), "m1")).toBe("alpha"); + } + }); + } + } + + for (const streamMode of ["legacy-tee", "eager-relay"] as const) { + for (const recordTerminalOutcomes of [true, false]) { + 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 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 }, + 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(""), { + 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") }); + 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, 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, sessionLaneIdFromRequest(new Headers({ session_id: "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, 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) { + 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, sessionLaneIdFromRequest(new Headers({ session_id: "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 +1884,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 +1897,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, 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 () => { @@ -2881,7 +3173,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 +3192,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,10 +3212,14 @@ 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, 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); - 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(); }