diff --git a/src/responses/provider-continuation.ts b/src/responses/provider-continuation.ts new file mode 100644 index 0000000000..cbb6efc7ab --- /dev/null +++ b/src/responses/provider-continuation.ts @@ -0,0 +1,98 @@ +import type { + OcxProviderContinuationOwner, + OcxReasoningReplayIdentity, +} from "../types"; + +const bounded = (value: unknown, max: number): value is string => + typeof value === "string" && value.length > 0 && value.length <= max; + +const isPlainRecord = (value: unknown): value is Record => { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +}; + +const defineOwn = (target: Record, key: string, value: unknown): void => { + Object.defineProperty(target, key, { + value, + enumerable: true, + configurable: true, + writable: true, + }); +}; + +/** Recursively merge provider-owned plain objects; arrays and scalar values replace prior state. */ +export function mergeProviderContinuationPayload( + inherited: Readonly>, + emitted: Readonly>, +): Record { + const merged: Record = {}; + for (const [key, value] of Object.entries(inherited)) defineOwn(merged, key, value); + for (const [key, value] of Object.entries(emitted)) { + const prior = Object.hasOwn(merged, key) ? merged[key] : undefined; + defineOwn( + merged, + key, + isPlainRecord(prior) && isPlainRecord(value) + ? mergeProviderContinuationPayload(prior, value) + : value, + ); + } + return merged; +} + +/** Validate proxy-authored continuation ownership before trusting persisted state. */ +export function isValidProviderContinuationOwner( + value: unknown, +): value is OcxProviderContinuationOwner { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const owner = value as Record; + return owner.version === 1 + && bounded(owner.providerName, 256) + && typeof owner.providerDestinationIdentity === "string" + && /^destination:[0-9a-f]{64}$/.test(owner.providerDestinationIdentity) + && bounded(owner.adapterName, 128) + && bounded(owner.modelId, 512) + && typeof owner.credentialIdentity === "string" + && /^(key|oauth|codex):[0-9a-f]{64}$/.test(owner.credentialIdentity); +} + +/** Reuse the exact process-local route identity already established for reasoning replay. */ +export function providerContinuationOwnerFromReplayIdentity( + identity: Readonly | undefined, +): OcxProviderContinuationOwner | undefined { + if (!identity) return undefined; + const owner: OcxProviderContinuationOwner = { + version: 1, + providerName: identity.providerName, + providerDestinationIdentity: identity.providerDestinationIdentity, + adapterName: identity.adapterName, + modelId: identity.modelId, + credentialIdentity: identity.credentialIdentity, + }; + return isValidProviderContinuationOwner(owner) ? owner : undefined; +} + +export function sameProviderContinuationOwner( + left: OcxProviderContinuationOwner, + right: OcxProviderContinuationOwner, +): boolean { + return left.version === right.version + && left.providerName === right.providerName + && left.providerDestinationIdentity === right.providerDestinationIdentity + && left.adapterName === right.adapterName + && left.modelId === right.modelId + && left.credentialIdentity === right.credentialIdentity; +} + +/** Cursor hashes this non-secret namespace with the client thread id. */ +export function providerContinuationRouteScope(owner: OcxProviderContinuationOwner): string { + return JSON.stringify([ + "provider-continuation-v1", + owner.providerName, + owner.providerDestinationIdentity, + owner.adapterName, + owner.modelId, + owner.credentialIdentity, + ]); +} diff --git a/src/responses/spill-store.ts b/src/responses/spill-store.ts index dbf5502018..825a3da3a3 100644 --- a/src/responses/spill-store.ts +++ b/src/responses/spill-store.ts @@ -18,6 +18,7 @@ import { createHash, randomBytes } from "node:crypto"; import { join } from "node:path"; import { getConfigDir } from "../config"; import { forgetEphemeralSecretPath, forgetHardenedSecretPath, hardenSecretDir, hardenSecretPath } from "../lib/windows-secret-acl"; +import { isValidProviderContinuationOwner } from "./provider-continuation"; import type { OcxProviderContinuationState } from "../types"; export const RESPONSE_SPILL_VERSION = 1; @@ -287,7 +288,11 @@ function validPayload(value: unknown, responseId: string): value is ResponseSpil } if (payload.providers !== undefined) { if (!payload.providers || typeof payload.providers !== "object" || Array.isArray(payload.providers)) return false; - for (const providerState of Object.values(payload.providers)) { + const providers = payload.providers as Record; + if (providers.__ocxOwner !== undefined + && !isValidProviderContinuationOwner(providers.__ocxOwner)) return false; + for (const [provider, providerState] of Object.entries(providers)) { + if (provider === "__ocxOwner") continue; if (!providerState || typeof providerState !== "object" || Array.isArray(providerState)) return false; } } diff --git a/src/responses/state.ts b/src/responses/state.ts index 39c83b7022..71f0e7d74e 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -1216,6 +1216,17 @@ export function previousResponseReplayPrefixLength(body: unknown): number { return replayedInputPrefixLengths.get(body) ?? 0; } +/** Copy proxy-private replay provenance to an internal clone with the same materialized input. */ +export function copyPreviousResponseReplayProvenance(source: unknown, target: unknown): void { + if (!source || typeof source !== "object" || Array.isArray(source)) return; + if (!target || typeof target !== "object" || Array.isArray(target)) return; + const prefixLength = replayedInputPrefixLengths.get(source); + if (!prefixLength) return; + const input = (target as { input?: unknown }).input; + if (!Array.isArray(input) || prefixLength > input.length) return; + replayedInputPrefixLengths.set(target, prefixLength); +} + /** True when a stale or foreign previous_response_id was removed from this exact request body. */ export function previousResponseScopeMismatch(body: unknown): boolean { return !!body && typeof body === "object" && replayScopeMismatches.has(body as object); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 3c7773e4e4..e2c13abe6e 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1,4 +1,5 @@ import type { Server } from "bun"; +import { randomUUID } from "node:crypto"; import { bridgeToResponsesSSE, buildResponseJSON, formatErrorResponse, type ResponsesTerminalStatus } from "../../bridge"; import { formatPassthroughUpstreamError } from "./passthrough-error"; import { @@ -26,6 +27,7 @@ import { awaitThoughtSignatureDurability, thoughtSignatureReplaySalt } from "../ import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractCompactUserMessages } from "../../responses/compaction"; import { FORWARD_HEADERS, sanitizeReasoningInputContent } from "../../adapters/openai-responses"; import { + copyPreviousResponseReplayProvenance, expandPreviousResponseInput, markBodyNonPersistable, previousResponseProviderState, @@ -33,6 +35,13 @@ import { previousResponseScopeMismatch, rememberResponseState, } from "../../responses/state"; +import { + isValidProviderContinuationOwner, + mergeProviderContinuationPayload, + providerContinuationOwnerFromReplayIdentity, + providerContinuationRouteScope, + sameProviderContinuationOwner, +} from "../../responses/provider-continuation"; import { comboRouteDecisionTrace, NoEligiblePolicyCandidateError, @@ -62,7 +71,17 @@ import { injectionDebugLog } from "../../lib/injection-debug-log"; import { resolveClientRetryAfter } from "../../lib/retry-after"; import { enrichOpenCodeZenRateLimitMessage } from "../../providers/opencode-zen-rate-limit"; import { modelInList, namespacedToolName } from "../../types"; -import type { AdapterEvent, OcxConfig, OcxParsedRequest, OcxProviderConfig, OcxProviderContinuationState, OcxUsage, TierDecision } from "../../types"; +import type { + AdapterEvent, + OcxConfig, + OcxParsedRequest, + OcxProviderConfig, + OcxProviderContinuationOwner, + OcxProviderContinuationState, + OcxReasoningReplayIdentity, + OcxUsage, + TierDecision, +} from "../../types"; import { forceRefreshOAuthAccessSnapshot, getOAuthCredentialApiBaseUrl, @@ -325,6 +344,62 @@ export function codexLogAccountId(authCtx: CodexAuthContext): string | null { return authCtx.kind === "pool" || authCtx.kind === "main-pool" ? authCtx.accountId : null; } +type ContinuationOwnerRead = + | { kind: "missing" } + | { kind: "invalid" } + | { kind: "valid"; owner: OcxProviderContinuationOwner }; + +function readProviderContinuationOwner( + state: OcxProviderContinuationState | undefined, +): ContinuationOwnerRead { + if (!state || state.__ocxOwner === undefined) return { kind: "missing" }; + const owner = state.__ocxOwner; + if (!isValidProviderContinuationOwner(owner)) return { kind: "invalid" }; + return { kind: "valid", owner: { ...owner } }; +} + +function providerContinuationPayload( + state: OcxProviderContinuationState | undefined, +): OcxProviderContinuationState | undefined { + if (!state) return undefined; + const cloned = structuredClone(state); + delete cloned.__ocxOwner; + return Object.keys(cloned).length > 0 ? cloned : undefined; +} + +function bindProviderContinuationForRoute( + parsed: OcxParsedRequest, + currentOwner: OcxProviderContinuationOwner | undefined, +): void { + const candidate = parsed._providerContinuationCandidate; + const storedOwner = readProviderContinuationOwner(candidate); + const mayRestore = storedOwner.kind === "valid" + && !!currentOwner + && sameProviderContinuationOwner(storedOwner.owner, currentOwner); + const restored = mayRestore ? providerContinuationPayload(candidate) : undefined; + if (restored) parsed._providerContinuation = restored; + else delete parsed._providerContinuation; + const cursorConversationId = restored?.cursor?.conversationId; + if (cursorConversationId) parsed._cursorConversationId = cursorConversationId; + else delete parsed._cursorConversationId; + if (currentOwner) parsed._providerContinuationOwner = { ...currentOwner }; + else delete parsed._providerContinuationOwner; +} + +function providerContinuationDestinationIdentity( + parsed: OcxParsedRequest, + provider: OcxProviderConfig, +): string | undefined { + const kiroContext = parsed._kiroAuthContext; + return reasoningReplayDestinationIdentity(JSON.stringify([ + provider.baseUrl.trim().replace(/\/+$/, ""), + provider.responsesPath ?? "", + kiroContext?.profileArn ?? "", + kiroContext?.apiRegion ?? "", + kiroContext?.ssoRegion ?? "", + ])); +} + function bindRouteReasoningReplayScope(args: { parsed: OcxParsedRequest; providerName: string; @@ -393,20 +468,39 @@ function bindRouteReasoningReplayScope(args: { ); } const providerDestinationIdentity = reasoningReplayDestinationIdentity(provider.baseUrl); + const replayIdentity: OcxReasoningReplayIdentity | undefined = credentialIdentity && providerDestinationIdentity + ? { + providerName, + providerDestinationIdentity, + providerDestinationDurableIdentity: durableReplayDestinationIdentity(provider.baseUrl), + adapterName, + modelId: parsed.modelId, + credentialIdentity, + ...(credentialDurableIdentity ? { credentialDurableIdentity } : {}), + } + : undefined; + const continuationDestinationIdentity = providerContinuationDestinationIdentity(parsed, provider); + const continuationOwner = providerContinuationOwnerFromReplayIdentity( + replayIdentity && continuationDestinationIdentity + ? { ...replayIdentity, providerDestinationIdentity: continuationDestinationIdentity } + : undefined, + ); + if (adapterName === "cursor") { + // The final route owner is authoritative for Cursor and supersedes the account-derived + // seed assigned before route binding. A Cursor conversation must be scoped to the exact + // provider/destination/adapter/model/credential that serves it. + if (continuationOwner) parsed._cursorIdentityScope = providerContinuationRouteScope(continuationOwner); + else if (!parsed._cursorIdentityScope?.startsWith("cursor-unowned:")) { + // Prevent the adapter's token-only fallback from recreating a provider-private id after the + // route owner failed closed. The sentinel is per parsed request and contains no credential. + parsed._cursorIdentityScope = `cursor-unowned:${randomUUID()}`; + } + } bindReasoningReplayScope( parsed._reasoningReplayScope, - credentialIdentity && providerDestinationIdentity - ? { - providerName, - providerDestinationIdentity, - providerDestinationDurableIdentity: durableReplayDestinationIdentity(provider.baseUrl), - adapterName, - modelId: parsed.modelId, - credentialIdentity, - ...(credentialDurableIdentity ? { credentialDurableIdentity } : {}), - } - : undefined, + replayIdentity, ); + bindProviderContinuationForRoute(parsed, continuationOwner); } function nonEmptyProviderApiKey(provider: OcxProviderConfig): string | undefined { @@ -943,6 +1037,12 @@ export interface HandleResponsesOptions { stripClaudeMainAuthForNoncanonicalForward?: boolean; /** Internal recursion guard; callers outside this module must not set it. */ comboAttempt?: boolean; + /** Internal combo handoff for one parent-validated continuation snapshot. */ + comboReplaySnapshot?: { + sourceBody: unknown; + previousResponseInputExpanded: boolean; + providerContinuation: OcxProviderContinuationState | undefined; + }; /** Internal combo handoff: allow a later same-provider model after a reset-derived 429/402. */ deferCodexResetDerivedCooldown?: boolean; /** 030-owned handoff when a child consumed the original failure under bounds. */ @@ -1435,7 +1535,12 @@ export async function handleComboResponses( // Expand previous_response_id before image policy and child dispatch so a // continuation that only references prior images still fails closed when // imageInput is disabled (and so targets see the full replayed input). - const body = expandPreviousResponseInput(rawBody); + const inboundClientThreadId = req.headers.get("x-codex-parent-thread-id")?.trim() || undefined; + const body = expandPreviousResponseInput(rawBody, inboundClientThreadId); + const scopeMismatch = previousResponseScopeMismatch(body); + if (scopeMismatch) { + console.warn("[opencodex] dropped a previous_response_id with a mismatched client task scope; continuing fresh"); + } if (previousResponseReplayFailure(body)) { return formatErrorResponse( 400, @@ -1462,11 +1567,14 @@ export async function handleComboResponses( if (combo.imageInput === "disabled" && comboRequestHasImageInput(body)) { return formatErrorResponse(400, "invalid_request_error", `Combo "${comboId}" does not accept image input`); } - // Expansion already materialised prior input. Drop the id so the child - // handleResponses path does not expand again and double-prepend history. - if (body !== rawBody && body && typeof body === "object" && !Array.isArray(body)) { - delete (body as Record).previous_response_id; - } + const comboReplaySnapshot = { + sourceBody: body, + previousResponseInputExpanded: body !== rawBody + && typeof (body as { previous_response_id?: unknown }).previous_response_id === "string", + providerContinuation: !scopeMismatch && body !== rawBody && requestedPreviousId + ? previousResponseProviderState(requestedPreviousId) + : undefined, + }; const adoptFailedChildLog = (childLog: RequestLogContext): void => { // Attempts remain the complete physical history; the logical row mirrors the most recent // failed target so an exhausted combo still has useful top-level reasoning diagnostics. @@ -1573,6 +1681,7 @@ export async function handleComboResponses( response = await handleResponses(childRequest, config, childLog, { ...options, comboAttempt: true, + comboReplaySnapshot, deferCodexResetDerivedCooldown, // Attempt-relative TTFT is recorded HERE (not via childLog.firstOutputMs — a later // Object.assign(logCtx, childLog) would overwrite the request-relative value). @@ -1842,19 +1951,24 @@ async function handleResponsesInner( ); const inboundClientThreadId = req.headers.get("x-codex-parent-thread-id")?.trim() || undefined; const originalBody = body; - body = expandPreviousResponseInput(body, inboundClientThreadId); - if (previousResponseScopeMismatch(body)) { - console.warn("[opencodex] dropped a previous_response_id with a mismatched client task scope; continuing fresh"); - } - if (previousResponseReplayFailure(body)) { - return formatErrorResponse( - 400, - "previous_response_not_found", - "Continuation state is unavailable or corrupt; resend the full conversation without previous_response_id.", - ); + if (options.comboReplaySnapshot) { + copyPreviousResponseReplayProvenance(options.comboReplaySnapshot.sourceBody, body); + } else { + body = expandPreviousResponseInput(body, inboundClientThreadId); + if (previousResponseScopeMismatch(body)) { + console.warn("[opencodex] dropped a previous_response_id with a mismatched client task scope; continuing fresh"); + } + if (previousResponseReplayFailure(body)) { + return formatErrorResponse( + 400, + "previous_response_not_found", + "Continuation state is unavailable or corrupt; resend the full conversation without previous_response_id.", + ); + } } - const previousResponseInputExpanded = body !== originalBody - && typeof (body as { previous_response_id?: unknown }).previous_response_id === "string"; + const previousResponseInputExpanded = options.comboReplaySnapshot?.previousResponseInputExpanded + ?? (body !== originalBody + && typeof (body as { previous_response_id?: unknown }).previous_response_id === "string"); // Spawn-message compatibility (both directions): agent_message task payloads ride in // encrypted_content slots as plaintext. Rewrite them to input_text on the RAW body BEFORE @@ -1877,8 +1991,10 @@ async function handleResponsesInner( parsed = parseRequest(body); toolBridgeMaps = buildToolBridgeMaps(parsed, translatorBudget); if (previousResponseInputExpanded) parsed._previousResponseInputExpanded = true; - parsed._providerContinuation = previousResponseProviderState(parsed.previousResponseId); - parsed._cursorConversationId = parsed._providerContinuation?.cursor?.conversationId; + const providerContinuationCandidate = options.comboReplaySnapshot + ? options.comboReplaySnapshot.providerContinuation + : previousResponseProviderState(parsed.previousResponseId); + if (providerContinuationCandidate) parsed._providerContinuationCandidate = providerContinuationCandidate; if (inboundClientThreadId) { parsed._clientThreadId = inboundClientThreadId; parsed._reasoningReplayScope = { clientThreadId: inboundClientThreadId }; @@ -2077,6 +2193,8 @@ async function handleResponsesInner( const kept: Array = [ "_previousResponseInputExpanded", "_providerContinuation", + "_providerContinuationCandidate", + "_providerContinuationOwner", "_cursorConversationId", "_clientThreadId", "_reasoningReplayScope", @@ -2230,8 +2348,9 @@ async function handleResponsesInner( ? `${route.providerName}-${route.codexAccountNamespace}` : formatCodexProviderForLog(route.providerName, codexLogAccountId(authCtx), config); logCtx.accountLogLabel = codexAuthContextLogLabel(authCtx, config); - // Prefer Codex pool account as the Cursor thread namespace when present. Cursor routes without - // codexAccountMode still get a credential-derived scope inside the Cursor adapter. + // Seed an account-derived scope before final adapter binding. Cursor never treats it as + // authoritative: bindRouteReasoningReplayScope replaces it with the exact route owner or a + // per-request fail-closed sentinel after the final provider and credential are known. const identityScope = codexLogAccountId(authCtx); if (identityScope) parsed._cursorIdentityScope = identityScope; subagentFallbackAccountId = authCtx.kind === "pool" || authCtx.kind === "main-pool" @@ -2345,6 +2464,9 @@ async function handleResponsesInner( codexAuthContext: authCtx, forwardHeaders: selectedForwardHeaders, }); + if (!logCtx.conversationId && parsed._cursorConversationId) { + logCtx.conversationId = normalizeLogConversationId(parsed._cursorConversationId); + } logCtx.providerAdapter = adapter.name; // Ordinary requests receive one durable attempt only after their final initial // adapter is resolved. Combo children own their attempt and retries keep it. @@ -2445,24 +2567,19 @@ async function handleResponsesInner( emitted?: OcxProviderContinuationState, ): OcxProviderContinuationState | undefined => { const cursorConversationId = parsed._cursorConversationId; - const inherited = parsed._providerContinuation; - if (!emitted && !inherited && !cursorConversationId) return undefined; - return { - ...(inherited ?? {}), - ...(emitted ?? {}), - ...((inherited?.kiro || emitted?.kiro) - ? { kiro: { ...(inherited?.kiro ?? {}), ...(emitted?.kiro ?? {}) } } - : {}), - ...(cursorConversationId - ? { - cursor: { - ...(inherited?.cursor ?? {}), - ...(emitted?.cursor ?? {}), - conversationId: cursorConversationId, - }, - } - : {}), - }; + const inherited = providerContinuationPayload(parsed._providerContinuation); + const emittedPayload = providerContinuationPayload(emitted); + if (!emittedPayload && !inherited && !cursorConversationId) return undefined; + const merged = mergeProviderContinuationPayload( + inherited ?? {}, + emittedPayload ?? {}, + ) as OcxProviderContinuationState; + if (cursorConversationId) { + merged.cursor = { ...(merged.cursor ?? {}), conversationId: cursorConversationId }; + } + return parsed._providerContinuationOwner + ? { ...merged, __ocxOwner: { ...parsed._providerContinuationOwner } } + : merged; }; // Remote compaction v2 on a ROUTED model: Codex sent `compaction_trigger` and requires exactly @@ -4475,6 +4592,14 @@ async function handleResponsesInner( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed: nextParsed, + providerName: route.providerName, + provider: route.provider, + adapterName: activeAdapter.name, + }); + // Response persistence closes over the outer parsed request; keep its owner binding in + // sync with the terminal-guard clone that builds the rotated continuation request. bindRouteReasoningReplayScope({ parsed, providerName: route.providerName, diff --git a/src/types.ts b/src/types.ts index f474695ca7..cf045ca891 100644 --- a/src/types.ts +++ b/src/types.ts @@ -43,6 +43,7 @@ export type { OcxAssistantContentPart, OcxRequestOptions, OcxMessagePhase, + OcxProviderContinuationOwner, OcxProviderContinuationState, AdapterEvent, OcxUrlCitation, diff --git a/src/types/request.ts b/src/types/request.ts index c01d6d3614..6fa4815add 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -79,6 +79,10 @@ export interface OcxParsedRequest { _kiroAuthContext?: Pick; /** Provider-private continuation metadata resolved from the Responses previous_response_id chain. */ _providerContinuation?: OcxProviderContinuationState; + /** Persisted continuation considered only after the final physical route is known. */ + _providerContinuationCandidate?: OcxProviderContinuationState; + /** Exact process-local route owner attached to newly persisted provider state. */ + _providerContinuationOwner?: OcxProviderContinuationOwner; /** * The hosted `{type:"web_search", ...}` tool config, stashed when Codex enables web search. Routed * (non-OpenAI) providers can't run it server-side, so the proxy re-exposes it as a function tool and @@ -251,16 +255,31 @@ export interface OcxRequestOptions { export type OcxMessagePhase = "commentary" | "final_answer"; +/** Non-secret, process-local owner fence for provider-private continuation state. */ +export interface OcxProviderContinuationOwner { + [field: string]: string | number; + version: 1; + providerName: string; + providerDestinationIdentity: string; + adapterName: string; + modelId: string; + credentialIdentity: string; +} + /** * Provider-private state that must follow a locally expanded `previous_response_id` chain. * Kept out of public Responses output and persisted only in the bounded local continuation cache. */ export interface OcxProviderContinuationState { + /** Proxy-authored owner metadata; stripped before provider adapters receive the state. */ + __ocxOwner?: OcxProviderContinuationOwner; cursor?: { + [field: string]: unknown; conversationId?: string; checkpointUsable?: boolean; }; kiro?: { + [field: string]: unknown; conversationId?: string; }; [provider: string]: Record | undefined; diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index 07d23a57a9..e3866c149c 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -21,6 +21,7 @@ import { buildResponseJSON } from "../src/bridge"; import { createCursorRequest } from "../src/adapters/cursor/request-builder"; import { createCursorContextUsageTracker } from "../src/adapters/cursor/protobuf-events"; import { parseRequest } from "../src/responses/parser"; +import { mergeProviderContinuationPayload } from "../src/responses/provider-continuation"; import { createSseInspector } from "../src/server/relay"; import { clearResponseStateForTests, @@ -774,6 +775,54 @@ describe("Responses previous_response_id state", () => { expect(expanded.input.at(-1)).toMatchObject({ type: "function_call_output", call_id: "call_1" }); }); + test("validates reserved continuation ownership separately from provider spill state", () => { + const owner = { + version: 1 as const, + providerName: "kiro", + providerDestinationIdentity: `destination:${"a".repeat(64)}`, + adapterName: "kiro", + modelId: "gpt-5.6-sol", + credentialIdentity: `oauth:${"b".repeat(64)}`, + }; + const valid = writeResponseSpillDurably("resp_valid_spill_owner", { + createdAt: Date.now(), + items: ["valid"], + providers: { __ocxOwner: owner, kiro: { conversationId: "kiro-valid" } }, + }); + expect(readResponseSpill("resp_valid_spill_owner", valid).ok).toBe(true); + + const invalid = writeResponseSpillDurably("resp_invalid_spill_owner", { + createdAt: Date.now(), + items: ["invalid"], + providers: { + __ocxOwner: { ...owner, version: 2 }, + kiro: { conversationId: "must-not-load" }, + } as never, + }); + expect(readResponseSpill("resp_invalid_spill_owner", invalid)).toEqual({ + ok: false, + reason: "corrupt", + }); + }); + + test("deep provider-state merge keeps __proto__ as data", () => { + const inherited = JSON.parse( + '{"__proto__":{"stable":"keep"},"future":{"metadata":{"stable":"keep","list":["old"]}}}', + ) as Record; + const emitted = JSON.parse( + '{"__proto__":{"changed":"new"},"future":{"metadata":{"changed":"new","list":["new"]}}}', + ) as Record; + + const merged = mergeProviderContinuationPayload(inherited, emitted); + + expect(Object.getPrototypeOf(merged)).toBe(Object.prototype); + expect(Object.hasOwn(merged, "__proto__")).toBe(true); + expect(merged["__proto__"]).toEqual({ stable: "keep", changed: "new" }); + expect(merged.future).toEqual({ + metadata: { stable: "keep", changed: "new", list: ["new"] }, + }); + }); + test("replays a durable spill after simulated process restart", async () => { setResponseStateByteCapForTests(1_024); rememberResponseState( @@ -1877,6 +1926,14 @@ describe("Responses previous_response_id state", () => { { model: "kiro/gpt-5.6-sol", input: "hello" }, first, { + __ocxOwner: { + version: 1, + providerName: "kiro", + providerDestinationIdentity: `destination:${"a".repeat(64)}`, + adapterName: "kiro", + modelId: "gpt-5.6-sol", + credentialIdentity: `oauth:${"b".repeat(64)}`, + }, cursor: { conversationId: "cursor_conv_2" }, kiro: { conversationId: "kiro_conv_2" }, }, @@ -1885,6 +1942,14 @@ describe("Responses previous_response_id state", () => { clearResponseStateMemoryForTests(); expect(previousResponseProviderState(first.id as string)).toEqual({ + __ocxOwner: { + version: 1, + providerName: "kiro", + providerDestinationIdentity: `destination:${"a".repeat(64)}`, + adapterName: "kiro", + modelId: "gpt-5.6-sol", + credentialIdentity: `oauth:${"b".repeat(64)}`, + }, cursor: { conversationId: "cursor_conv_2", checkpointUsable: true }, kiro: { conversationId: "kiro_conv_2" }, }); diff --git a/tests/server-combo-failover-e2e.test.ts b/tests/server-combo-failover-e2e.test.ts index ca46ada550..81b96f4834 100644 --- a/tests/server-combo-failover-e2e.test.ts +++ b/tests/server-combo-failover-e2e.test.ts @@ -1,7 +1,7 @@ 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"; -import { mkdtempSync, rmSync } from "node:fs"; +import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -15,7 +15,7 @@ import { handleManagementAPI } from "../src/server/management-api"; import { saveCredential } from "../src/oauth/store"; import { XAI_OAUTH_DISCOVERY_URL } from "../src/oauth/xai"; import { XAI_GROK_CLI_BASE_URL } from "../src/providers/xai-transport"; -import type { AdapterEvent, OcxConfig, OcxProviderConfig } from "../src/types"; +import type { AdapterEvent, OcxConfig, OcxProviderConfig, OcxProviderContinuationState } from "../src/types"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home"; import { clearRequestLogsForTests, hydrateRequestLogsFromDisk, type RequestLogContext } from "../src/server/request-log"; import { responseWithDeferredRequestLog } from "../src/server/relay"; @@ -29,6 +29,13 @@ import { import { startServer } from "../src/server"; import { fakeChatGptJwt } from "./helpers/fake-chatgpt-jwt"; import { catalogConvergenceFactory } from "./helpers/catalog-convergence"; +import { removeTreeWithRetry } from "./helpers/remove-tree"; +import { + clearResponseStateForTests, + flushResponseState, + responseStatePersistPendingForTests, +} from "../src/responses/state"; +import { clearCursorThreadContinuityForTests } from "../src/adapters/cursor/thread-continuity"; // 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). @@ -54,9 +61,13 @@ mock.module("../src/server/adapter-resolve", () => ({ // tests can drive the genuine continuation/persistence policy without a live socket. return createCursorAdapter(provider, { createTransport: customCursorTransportFactory }); } - if (provider.adapter === "test-run-turn") { + if ( + provider.adapter === "test-run-turn" + || provider.adapter === "test-kiro" + || provider.adapter === "test-owned" + ) { const adapter: ProviderAdapter = { - name: "test-run-turn", + name: provider.adapter === "test-kiro" ? "kiro" : provider.adapter, buildRequest: () => ({ url: provider.baseUrl, method: "POST", headers: {}, body: "" }), async *parseStream(): AsyncGenerator { yield { type: "error", message: "test runTurn adapter does not use parseStream" }; @@ -132,23 +143,34 @@ beforeEach(() => { customUsageEstimate = undefined; customCursorTransportFactory = undefined; clearRequestLogsForTests(); + clearResponseStateForTests(); + clearCursorThreadContinuityForTests(); }); afterEach(async () => { - globalThis.fetch = originalFetch; - Date.now = originalNow; - for (const server of servers.splice(0)) await server.stop(true); - if (previousHome === undefined) delete process.env.OPENCODEX_HOME; - else process.env.OPENCODEX_HOME = previousHome; - if (previousCursorToken === undefined) delete process.env.OPENCODEX_CURSOR_TEST_TOKEN; - else process.env.OPENCODEX_CURSOR_TEST_TOKEN = previousCursorToken; - isolatedCodexHome?.restore(); - isolatedCodexHome = null; - if (testDir) rmSync(testDir, { recursive: true, force: true }); - clearComboSelectionState(); - clearComboTargetCooldowns(); - clearCodexUpstreamHealth(); - clearRequestLogsForTests(); + let responseStatePending = true; + try { + for (const server of servers.splice(0)) await server.stop(true); + await flushResponseState(); + responseStatePending = responseStatePersistPendingForTests(); + } finally { + clearResponseStateForTests(); + clearCursorThreadContinuityForTests(); + globalThis.fetch = originalFetch; + Date.now = originalNow; + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousCursorToken === undefined) delete process.env.OPENCODEX_CURSOR_TEST_TOKEN; + else process.env.OPENCODEX_CURSOR_TEST_TOKEN = previousCursorToken; + isolatedCodexHome?.restore(); + isolatedCodexHome = null; + if (testDir) removeTreeWithRetry(testDir); + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearCodexUpstreamHealth(); + clearRequestLogsForTests(); + } + expect(responseStatePending).toBe(false); }); function serve(handler: (request: Request) => Response | Promise) { @@ -1403,6 +1425,564 @@ describe("server combo failover 030 activation matrix", () => { expect(inputText.split("next turn")).toHaveLength(2); }); + test("combo continuation expansion respects the client task scope", async () => { + const { rememberResponseState } = await import("../src/responses/state"); + rememberResponseState( + { model: "combo/free", input: "legacy private history" }, + { + id: "resp_combo_legacy_unscoped", + status: "completed", + output: [{ type: "message", role: "assistant", content: "legacy reply" }], + }, + ); + rememberResponseState( + { model: "combo/free", input: "scoped private history" }, + { + id: "resp_combo_scoped", + status: "completed", + output: [{ type: "message", role: "assistant", content: "scoped reply" }], + }, + undefined, + { clientThreadId: "combo-task" }, + ); + const bodies: Array> = []; + const a = serve(async request => { + bodies.push(await request.json() as Record); + return chatSuccess("continued", "m1"); + }); + const config = comboConfig({ a: provider("openai-chat", baseUrl(a), "key-a") }); + const headers = { "x-codex-parent-thread-id": "combo-task" }; + + const legacyResponse = await post(config, { + previous_response_id: "resp_combo_legacy_unscoped", + input: "fresh scoped input", + }, {}, headers); + const scopedResponse = await post(config, { + previous_response_id: "resp_combo_scoped", + input: "continue scoped task", + }, {}, headers); + + expect(legacyResponse.status).toBe(200); + expect(scopedResponse.status).toBe(200); + expect(bodies).toHaveLength(2); + expect(JSON.stringify(bodies[0])).not.toContain("legacy private history"); + expect(JSON.stringify(bodies[0])).toContain("fresh scoped input"); + expect(JSON.stringify(bodies[1])).toContain("scoped private history"); + expect(JSON.stringify(bodies[1])).toContain("continue scoped task"); + }); + + test("combo child preserves replay provenance for compaction and generated guidance", async () => { + const { rememberResponseState } = await import("../src/responses/state"); + const { multiAgentGuidanceText, PROACTIVE_MULTI_AGENT_MODE_TEXT } = await import("../src/server/responses/collaboration"); + const guidance = `${PROACTIVE_MULTI_AGENT_MODE_TEXT}`; + const tools = ["spawn_agent", "send_input"].map(name => ({ + type: "function", + name, + namespace: "multi_agent_v1", + description: "Collaborate on work", + parameters: { type: "object", properties: {} }, + })); + rememberResponseState( + { + model: "combo/free", + input: [ + { type: "context_compaction" }, + { + type: "message", + role: "developer", + content: [{ type: "input_text", text: guidance }], + }, + { type: "message", role: "user", content: "prior task" }, + ], + reasoning: { effort: "max" }, + tools, + }, + { + id: "resp_combo_replay_provenance", + status: "completed", + output: [{ + id: "msg_combo_replay_provenance", + type: "message", + role: "assistant", + content: "prior answer", + }], + }, + undefined, + { clientThreadId: "combo-provenance-task" }, + ); + + let observed: { + replayPrefixLength: number; + contextCompactionBoundary: boolean | undefined; + generatedGuidance: string | null; + taggedGuidance: string[]; + } | undefined; + const guidanceOptions = { multiAgentGuidanceEnabled: true }; + const config = comboConfig({ + a: provider("test-run-turn", "https://a.test/v1", "key-a"), + }); + Object.assign(config, guidanceOptions); + customRunTurn = async (parsed, _incoming, emit) => { + const rawInput = (parsed._rawBody as { input?: unknown[] } | undefined)?.input ?? []; + const taggedGuidance = rawInput.flatMap(item => { + if (!item || typeof item !== "object" || Array.isArray(item)) return []; + const record = item as Record; + if (record.type !== "message" || record.role !== "developer" || !Array.isArray(record.content)) return []; + return record.content.flatMap(part => !!part && typeof part === "object" + && !Array.isArray(part) + && (part as Record).type === "input_text" + && typeof (part as Record).text === "string" + && ((part as Record).text as string).startsWith("") + && ((part as Record).text as string).endsWith("") + ? [(part as Record).text as string] + : []); + }); + observed = { + replayPrefixLength: parsed._replayPrefixLen ?? 0, + contextCompactionBoundary: parsed._contextCompactionBoundary, + generatedGuidance: await multiAgentGuidanceText(parsed, guidanceOptions), + taggedGuidance, + }; + emit({ type: "text_delta", text: "continued" }); + emit({ type: "done" }); + }; + + const response = await post(config, { + previous_response_id: "resp_combo_replay_provenance", + input: [{ type: "message", role: "user", content: "current turn" }], + reasoning: { effort: "max" }, + tools, + }, {}, { "x-codex-parent-thread-id": "combo-provenance-task" }); + + expect(response.status).toBe(200); + expect(observed).toEqual({ + replayPrefixLength: expect.any(Number), + contextCompactionBoundary: undefined, + generatedGuidance: guidance, + taggedGuidance: [guidance], + }); + expect(observed!.replayPrefixLength).toBeGreaterThan(0); + }); + + test("combo failover dispatches the one parent-validated continuation snapshot", async () => { + const { clearResponseStateForTests, rememberResponseState } = await import("../src/responses/state"); + rememberResponseState( + { model: "combo/free", input: [{ role: "user", content: "stable prior history" }] }, + { + id: "resp_combo_stable_snapshot", + status: "completed", + output: [{ type: "message", role: "assistant", content: "stable prior answer" }], + }, + ); + const a = serve(() => { + clearResponseStateForTests(); + return Response.json({ error: { message: "retry" } }, { status: 503 }); + }); + let backupParsed: { + previousResponseId?: string; + replayPrefixLength: number; + rawInput: unknown[]; + } | undefined; + customRunTurn = async (parsed, _incoming, emit) => { + backupParsed = { + previousResponseId: parsed.previousResponseId, + replayPrefixLength: parsed._replayPrefixLen ?? 0, + rawInput: (parsed._rawBody as { input?: unknown[] } | undefined)?.input ?? [], + }; + emit({ type: "text_delta", text: "continued" }); + emit({ type: "done" }); + }; + const config = comboConfig({ + a: provider("openai-chat", baseUrl(a), "key-a"), + b: provider("test-run-turn", "https://b.test/v1", "key-b"), + }); + + const response = await post(config, { + previous_response_id: "resp_combo_stable_snapshot", + input: [{ role: "user", content: "stable current turn" }], + }); + + expect(response.status).toBe(200); + expect(backupParsed?.previousResponseId).toBe("resp_combo_stable_snapshot"); + expect(backupParsed?.replayPrefixLength).toBeGreaterThan(0); + const requestText = JSON.stringify(backupParsed?.rawInput); + expect(backupParsed?.rawInput).toHaveLength(3); + expect(requestText.split("stable prior history")).toHaveLength(2); + expect(requestText.split("stable prior answer")).toHaveLength(2); + expect(requestText.split("stable current turn")).toHaveLength(2); + }); + + test("combo keeps an explicitly empty provider-state snapshot across failover", async () => { + const { previousResponseProviderState, rememberResponseState } = await import("../src/responses/state"); + customRunTurn = async (_parsed, _incoming, emit) => { + emit({ type: "text_delta", text: "seed" }); + emit({ type: "done", providerState: { kiro: { conversationId: "late-owned-state" } } }); + }; + const config = comboConfig({ + b: provider("test-owned", "https://provider-b.test/v1", "key-b"), + }, [{ provider: "b", model: "m2" }]); + const seed = await post(config, { input: "seed owner" }); + expect(seed.status).toBe(200); + const seedJson = await seed.json() as { id: string }; + const ownedState = previousResponseProviderState(seedJson.id); + expect(ownedState?.__ocxOwner?.providerName).toBe("b"); + + config.providers.a = provider("test-owned", "https://provider-a.test/v1", "key-a"); + config.combos!.free!.targets = [ + { provider: "a", model: "m1" }, + { provider: "b", model: "m2" }, + ]; + let backupObserved: string | undefined; + customRunTurn = async (parsed, _incoming, emit) => { + if (parsed.modelId === "m1") { + rememberResponseState( + { model: "combo/free", input: "late state" }, + { + id: "resp_combo_late_provider_state", + status: "completed", + output: [{ type: "message", role: "assistant", content: "late" }], + }, + ownedState, + { force: true }, + ); + emit({ type: "error", message: "retry elsewhere", status: 503, retryable: true }); + return; + } + backupObserved = parsed._providerContinuation?.kiro?.conversationId; + emit({ type: "text_delta", text: "backup" }); + emit({ type: "done" }); + }; + + const response = await post(config, { + previous_response_id: "resp_combo_late_provider_state", + input: "continue", + }); + + expect(response.status).toBe(200); + expect(backupObserved).toBeUndefined(); + }); + + test("combo response state deep-merges provider-private payloads generically", async () => { + const { previousResponseProviderState } = await import("../src/responses/state"); + let turn = 0; + customRunTurn = async (_parsed, _incoming, emit) => { + turn += 1; + emit({ type: "text_delta", text: `turn-${turn}` }); + emit({ + type: "done", + providerState: turn === 1 + ? { + cursor: { checkpointRef: "opaque-ref" }, + future: { + stable: "keep", + changed: "old", + metadata: { + stable: "keep-nested", + changed: "old-nested", + list: ["old"], + scalar: "old", + }, + }, + } + : { + cursor: { checkpointUsable: true }, + future: { + changed: "new", + metadata: { + changed: "new-nested", + list: ["new"], + scalar: 42, + }, + }, + }, + }); + }; + const config = comboConfig({ + a: provider("test-owned", "https://provider-a.test/v1", "key-a"), + }, [{ provider: "a", model: "m1" }]); + + const first = await post(config, { input: "seed future provider state" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + const second = await post(config, { + previous_response_id: firstJson.id, + input: "update future provider state", + }); + expect(second.status).toBe(200); + const secondJson = await second.json() as { id: string }; + + const stored = previousResponseProviderState(secondJson.id); + expect(stored?.future).toEqual({ + stable: "keep", + changed: "new", + metadata: { + stable: "keep-nested", + changed: "new-nested", + list: ["new"], + scalar: 42, + }, + }); + expect(stored?.cursor).toEqual({ checkpointRef: "opaque-ref", checkpointUsable: true }); + expect(stored?.__ocxOwner?.providerName).toBe("a"); + }); + + test("combo child retains the local id without inheriting unbound provider state", async () => { + const { rememberResponseState } = await import("../src/responses/state"); + rememberResponseState( + { model: "combo/free", input: "prior target turn" }, + { + id: "resp_combo_unbound_provider_state", + status: "completed", + output: [{ type: "message", role: "assistant", content: "prior target answer" }], + }, + { + cursor: { conversationId: "cursor_owned_by_another_target" }, + kiro: { conversationId: "kiro_owned_by_another_target" }, + }, + ); + let observed: { + previousResponseId?: string; + providerContinuation: unknown; + cursorConversationId: unknown; + } | undefined; + customRunTurn = async (parsed, _incoming, emit) => { + observed = { + previousResponseId: parsed.previousResponseId, + providerContinuation: parsed._providerContinuation, + cursorConversationId: parsed._cursorConversationId, + }; + emit({ type: "text_delta", text: "continued" }); + emit({ type: "done" }); + }; + const config = comboConfig({ + a: provider("test-run-turn", "https://a.test/v1", "key-a"), + }); + + const response = await post(config, { + previous_response_id: "resp_combo_unbound_provider_state", + input: [{ role: "user", content: "continue" }], + }); + + expect(response.status).toBe(200); + expect(observed?.previousResponseId).toBe("resp_combo_unbound_provider_state"); + expect(observed?.providerContinuation).toBeUndefined(); + expect(observed?.cursorConversationId).toBeUndefined(); + }); + + test("combo rejects malformed provider-continuation owner metadata", async () => { + const { rememberResponseState } = await import("../src/responses/state"); + rememberResponseState( + { model: "combo/free", input: "prior target turn" }, + { + id: "resp_combo_malformed_provider_owner", + status: "completed", + output: [{ type: "message", role: "assistant", content: "prior target answer" }], + }, + { + __ocxOwner: { + version: 2, + providerName: "a", + providerDestinationIdentity: `destination:${"a".repeat(64)}`, + adapterName: "kiro", + modelId: "m1", + credentialIdentity: `key:${"b".repeat(64)}`, + }, + kiro: { conversationId: "must-not-restore" }, + } as unknown as OcxProviderContinuationState, + ); + let observed: string | undefined; + customRunTurn = async (parsed, _incoming, emit) => { + observed = parsed._providerContinuation?.kiro?.conversationId; + emit({ type: "text_delta", text: "continued" }); + emit({ type: "done", providerState: { kiro: { conversationId: "fresh" } } }); + }; + const config = comboConfig({ + a: provider("test-kiro", "https://kiro-a.test/v1", "key-a"), + }); + + const response = await post(config, { + previous_response_id: "resp_combo_malformed_provider_owner", + input: "continue", + }); + + expect(response.status).toBe(200); + expect(observed).toBeUndefined(); + }); + + test("same Kiro combo target and credential retain the provider conversation id", async () => { + const seen: Array = []; + customRunTurn = async (parsed, _incoming, emit) => { + const conversationId = parsed._providerContinuation?.kiro?.conversationId; + seen.push(conversationId); + emit({ type: "text_delta", text: "continued" }); + emit({ + type: "done", + providerState: { kiro: { conversationId: conversationId ?? "kiro-owned-conversation" } }, + }); + }; + const config = comboConfig({ + a: provider("test-kiro", "https://kiro-a.test/v1", "key-a"), + }); + + const first = await post(config, { store: false, input: "first" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + const second = await post(config, { + store: false, + previous_response_id: firstJson.id, + input: "second", + }); + + expect(second.status).toBe(200); + expect(seen).toEqual([undefined, "kiro-owned-conversation"]); + }); + + test("same Cursor combo target without a parent-thread header retains its conversation id", async () => { + const seen: string[] = []; + customCursorTransportFactory = () => ({ + async *run(request) { + seen.push(request.conversationId); + yield { type: "text", text: "cursor ok" }; + yield { type: "done", usage: { inputTokens: 10, outputTokens: 2, estimated: true } }; + }, + writeClient() {}, + close() {}, + }); + const config = comboConfig( + { cursortest: provider("cursor", "https://api2.cursor.sh", "fake-cursor-token") }, + [{ provider: "cursortest", model: "composer-2" }], + ); + + const first = await post(config, { store: false, input: "first" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + const second = await post(config, { + store: false, + previous_response_id: firstJson.id, + input: "second", + }); + + expect(second.status).toBe(200); + expect(seen).toHaveLength(2); + expect(seen[1]).toBe(seen[0]); + }); + + test("combo failover to another provider does not inherit provider continuation state", async () => { + const seen: Array<{ model: string; conversationId?: string }> = []; + customRunTurn = async (parsed, _incoming, emit) => { + const conversationId = parsed._providerContinuation?.kiro?.conversationId; + seen.push({ model: parsed.modelId, ...(conversationId ? { conversationId } : {}) }); + emit({ type: "text_delta", text: "first" }); + emit({ type: "done", providerState: { kiro: { conversationId: "kiro-provider-a" } } }); + }; + const config = comboConfig({ + a: provider("test-kiro", "https://kiro-a.test/v1", "key-a"), + }); + const first = await post(config, { store: false, input: "first" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + + config.providers.b = provider("test-owned", "https://provider-b.test/v1", "key-b"); + config.combos!.free!.targets = [ + { provider: "a", model: "m1" }, + { provider: "b", model: "m2" }, + ]; + customRunTurn = async (parsed, _incoming, emit) => { + const conversationId = parsed._providerContinuation?.kiro?.conversationId; + seen.push({ model: parsed.modelId, ...(conversationId ? { conversationId } : {}) }); + if (parsed.modelId === "m1") { + emit({ type: "error", message: "retry elsewhere", status: 503, retryable: true }); + return; + } + emit({ type: "text_delta", text: "backup" }); + emit({ type: "done", providerState: { kiro: { conversationId: "provider-b" } } }); + }; + + const second = await post(config, { + store: false, + previous_response_id: firstJson.id, + input: "second", + }); + + expect(second.status).toBe(200); + expect(seen.slice(1)).toEqual([ + { model: "m1", conversationId: "kiro-provider-a" }, + { model: "m2" }, + ]); + }); + + test("same provider with a different credential does not inherit provider continuation state", async () => { + const seen: Array = []; + customRunTurn = async (parsed, _incoming, emit) => { + const conversationId = parsed._providerContinuation?.kiro?.conversationId; + seen.push(conversationId); + emit({ type: "text_delta", text: "continued" }); + emit({ + type: "done", + providerState: { kiro: { conversationId: conversationId ?? "credential-one-conversation" } }, + }); + }; + const config = comboConfig({ + a: provider("test-kiro", "https://kiro-a.test/v1", "credential-one"), + }); + const first = await post(config, { store: false, input: "first" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + + config.providers.a!.apiKey = "credential-two"; + const second = await post(config, { + store: false, + previous_response_id: firstJson.id, + input: "second", + }); + + expect(second.status).toBe(200); + expect(seen).toEqual([undefined, undefined]); + }); + + test.each(["provider", "destination", "adapter", "model"] as const)( + "continuation owner rejects an exact %s mismatch", + async mismatch => { + const seen: Array = []; + customRunTurn = async (parsed, _incoming, emit) => { + const conversationId = parsed._providerContinuation?.kiro?.conversationId; + seen.push(conversationId); + emit({ type: "text_delta", text: "continued" }); + emit({ + type: "done", + providerState: { kiro: { conversationId: conversationId ?? "owned-conversation" } }, + }); + }; + const config = comboConfig({ + a: provider("test-kiro", "https://kiro-a.test/v1", "credential-one"), + }); + const first = await post(config, { store: false, input: "first" }); + expect(first.status).toBe(200); + const firstJson = await first.json() as { id: string }; + + if (mismatch === "provider") { + config.providers.b = provider("test-kiro", "https://kiro-a.test/v1", "credential-one"); + config.combos!.free!.targets = [{ provider: "b", model: "m1" }]; + } else if (mismatch === "destination") { + config.providers.a!.baseUrl = "https://kiro-b.test/v1"; + } else if (mismatch === "adapter") { + // Keep provider, destination, credential, and model fixed so only the adapter owner + // component changes. The first test-kiro turn already persisted the owned state. + config.providers.a!.adapter = "test-owned"; + } else { + config.combos!.free!.targets = [{ provider: "a", model: "m2" }]; + } + const second = await post(config, { + store: false, + previous_response_id: firstJson.id, + input: "second", + }); + + expect(second.status).toBe(200); + expect(seen).toEqual([undefined, undefined]); + }, + ); + test("disabled image input rejects an image restored from previous_response_id before dispatch", async () => { const { rememberResponseState } = await import("../src/responses/state"); rememberResponseState( @@ -2020,6 +2600,32 @@ describe("cursor conversation continuity across store:false chains", () => { }; } + test("ownerless legacy Cursor state fails closed before adapter dispatch", async () => { + const { rememberResponseState } = await import("../src/responses/state"); + rememberResponseState( + { model: "cursortest/composer-2", input: "legacy" }, + { + id: "resp_cursor_ownerless_legacy", + status: "completed", + output: [{ type: "message", role: "assistant", content: "legacy reply" }], + }, + { cursor: { conversationId: "legacy-cursor-conversation" } }, + { force: true }, + ); + const seen: string[] = []; + customCursorTransportFactory = fakeCursorTransportFactory(seen); + + const response = await postCursor(cursorConfig(), { + model: "cursortest/composer-2", + previous_response_id: "resp_cursor_ownerless_legacy", + input: "continue", + }); + + expect(response.status).toBe(200); + expect(seen).toHaveLength(1); + expect(seen[0]).not.toBe("legacy-cursor-conversation"); + }); + test("store:false chain reuses the SAME cursor conversationId (native model)", async () => { const seen: string[] = []; customCursorTransportFactory = fakeCursorTransportFactory(seen); diff --git a/tests/terminal-continuation-owner-rotation.test.ts b/tests/terminal-continuation-owner-rotation.test.ts new file mode 100644 index 0000000000..53fa7d82df --- /dev/null +++ b/tests/terminal-continuation-owner-rotation.test.ts @@ -0,0 +1,245 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { ProviderAdapter } from "../src/adapters/base"; +import { saveConfig } from "../src/config"; +import { clearKeyCooldowns } from "../src/providers/key-failover"; +import { reasoningReplayKeyCredentialIdentity } from "../src/responses/reasoning-replay-cache"; +import { + clearResponseStateForTests, + previousResponseProviderState, +} from "../src/responses/state"; +import type { + AdapterEvent, + OcxConfig, + OcxParsedRequest, + OcxProviderConfig, +} from "../src/types"; + +interface BuildObservation { + key: string; + continuation?: string; +} + +let builds: BuildObservation[] = []; + +function eventsForPhase(phase: string): AdapterEvent[] { + if (phase === "seed") { + return [ + { type: "text_delta", text: "seeded" }, + { + type: "done", + stopReason: "end_turn", + providerState: { kiro: { conversationId: "private-a" } }, + }, + ]; + } + if (phase === "plan") { + return [ + { type: "text_delta", text: "I will modify the file now." }, + { + type: "done", + stopReason: "end_turn", + providerState: { kiro: { conversationId: "private-a-plan" } }, + }, + ]; + } + if (phase === "rotated") { + return [ + { type: "text_delta", text: "completed on the rotated key" }, + { + type: "done", + stopReason: "end_turn", + providerState: { kiro: { conversationId: "private-b" } }, + }, + ]; + } + if (phase === "follow") { + return [ + { type: "text_delta", text: "continued on the rotated key" }, + { + type: "done", + stopReason: "end_turn", + providerState: { kiro: { conversationId: "private-b-next" } }, + }, + ]; + } + throw new Error(`unexpected test phase: ${phase}`); +} + +const actualResolver = await import("../src/server/adapter-resolve"); +const actualResolveAdapter = actualResolver.resolveAdapter; + +mock.module("../src/server/adapter-resolve", () => ({ + ...actualResolver, + resolveAdapter(provider: OcxProviderConfig, cacheRetention?: "none" | "short" | "long") { + if (provider.adapter !== "test-terminal-owned") { + return actualResolveAdapter(provider, cacheRetention); + } + const key = provider.apiKey ?? ""; + const adapter: ProviderAdapter = { + // The terminal guard is enabled for Anthropic adapters. The transport is otherwise a + // narrow test double so the test can emit provider-private state deterministically. + name: "anthropic", + buildRequest(parsed: OcxParsedRequest) { + const continuation = parsed._providerContinuation?.kiro?.conversationId; + builds.push({ key, ...(continuation ? { continuation } : {}) }); + return { + url: "https://owned-terminal.test/v1/messages", + method: "POST", + headers: { authorization: `Bearer ${key}` }, + body: "{}", + }; + }, + async *parseStream(response: Response): AsyncGenerator { + yield* eventsForPhase(response.headers.get("x-test-phase") ?? ""); + }, + async parseResponse(response: Response): Promise { + return eventsForPhase(response.headers.get("x-test-phase") ?? ""); + }, + }; + return adapter; + }, +})); + +const { handleResponses } = await import("../src/server/responses"); + +describe("terminal continuation provider-owner rotation", () => { + let originalFetch: typeof fetch; + let previousHome: string | undefined; + let testHome = ""; + + beforeEach(() => { + originalFetch = globalThis.fetch; + previousHome = process.env.OPENCODEX_HOME; + testHome = mkdtempSync(join(tmpdir(), "ocx-terminal-owner-")); + process.env.OPENCODEX_HOME = testHome; + builds = []; + clearKeyCooldowns(); + clearResponseStateForTests(); + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + clearKeyCooldowns(); + clearResponseStateForTests(); + rmSync(testHome, { recursive: true, force: true }); + }); + + test("429 rotation fences inherited state and persists the rotated owner", async () => { + const keyA = "key-alpha-000111222333"; + const keyB = "key-beta-444555666777"; + const config: OcxConfig = { + port: 0, + defaultProvider: "owned", + providers: { + owned: { + adapter: "test-terminal-owned", + baseUrl: "https://owned-terminal.test/v1", + authMode: "key", + apiKey: keyA, + apiKeyPool: [ + { id: "k1", key: keyA, addedAt: 1 }, + { id: "k2", key: keyB, addedAt: 2 }, + ], + }, + }, + } as OcxConfig; + const keyAIdentity = reasoningReplayKeyCredentialIdentity({ apiKey: keyA }); + saveConfig(config); + + const phases = ["seed", "plan", "rate-limit", "rotated", "follow"]; + const seenAuthorization: string[] = []; + globalThis.fetch = (async (_input, init) => { + seenAuthorization.push(new Headers(init?.headers).get("authorization") ?? ""); + const phase = phases.shift(); + if (!phase) throw new Error("unexpected extra upstream request"); + if (phase === "rate-limit") { + return Response.json( + { error: { message: "rotate" } }, + { status: 429, headers: { "retry-after": "30" } }, + ); + } + return new Response("", { headers: { "x-test-phase": phase } }); + }) as typeof fetch; + + const post = (body: Record) => handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + config, + { model: "", provider: "" }, + ); + + const seed = await post({ + model: "owned/model", + input: "seed", + stream: false, + store: true, + }); + expect(seed.status).toBe(200); + const seedJson = await seed.json() as { id: string }; + expect(previousResponseProviderState(seedJson.id)).toMatchObject({ + __ocxOwner: { credentialIdentity: keyAIdentity }, + kiro: { conversationId: "private-a" }, + }); + + const rotated = await post({ + model: "owned/model", + previous_response_id: seedJson.id, + input: "Please modify the file now", + stream: false, + store: true, + tools: [{ + type: "function", + name: "read_file", + description: "read a file", + parameters: { type: "object" }, + }], + }); + expect(rotated.status).toBe(200); + const rotatedJson = await rotated.json() as { id: string }; + const keyBIdentity = reasoningReplayKeyCredentialIdentity(config.providers.owned!); + expect(keyBIdentity).toBeDefined(); + expect(keyBIdentity).not.toBe(keyAIdentity); + expect(previousResponseProviderState(rotatedJson.id)).toMatchObject({ + __ocxOwner: { credentialIdentity: keyBIdentity }, + kiro: { conversationId: "private-b" }, + }); + + const follow = await post({ + model: "owned/model", + previous_response_id: rotatedJson.id, + input: "follow up", + stream: false, + store: true, + }); + expect(follow.status).toBe(200); + const followJson = await follow.json() as { id: string }; + expect(previousResponseProviderState(followJson.id)).toMatchObject({ + __ocxOwner: { credentialIdentity: keyBIdentity }, + kiro: { conversationId: "private-b-next" }, + }); + + expect(seenAuthorization).toEqual([ + `Bearer ${keyA}`, + `Bearer ${keyA}`, + `Bearer ${keyA}`, + `Bearer ${keyB}`, + `Bearer ${keyB}`, + ]); + expect(builds).toEqual([ + { key: keyA }, + { key: keyA, continuation: "private-a" }, + { key: keyA, continuation: "private-a" }, + { key: keyB }, + { key: keyB, continuation: "private-b" }, + ]); + expect(phases).toEqual([]); + }); +});