diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index edfca8ca6b8..877117babe3 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -364,12 +364,20 @@ If a canonical ChatGPT forward continuation references expired or missing local opencodex returns `previous_response_not_found` before sending anything upstream. Codex's WebSocket client recognizes this error and can reconnect with its full retained context, including completed tool calls and their results, within its normal stream retry budget. An -idle task therefore does not need a new task solely because the proxy's one-hour cache expired. -The cache remains bounded; this does not extend retention or recover history the client no -longer has. HTTP clients must handle the error explicitly and resend their full context without -`previous_response_id`. Retrying only the same ID cannot recover missing state. - -The same recovery signal applies to routed Responses providers configured with +idle task therefore does not need a new task solely because the proxy's replay cache expired. +Replayed continuation state is retained for 24 hours and stays bounded by its existing memory, +disk, and entry ceilings; this does not recover history the client no longer has. HTTP clients +must handle the error explicitly and resend their full context without `previous_response_id`. +Retrying only the same ID cannot recover missing state. + +The same recovery signal applies to every routed destination, because only the native Responses +passthrough can answer a turn whose history this proxy lost — it forwards `previous_response_id` +to a backend that stored the chain. Every other wire rebuilds the conversation from each request's +own input, so a missed expansion there would otherwise send the current turn alone and silently +lose the conversation. That includes the three that look stateful: Devin re-sends the whole +conversation every turn, Cursor's checkpoint reference lives in the same expired store and falls +back to full replay without it, and Kiro rebuilds its conversation history from the turns it was +handed. It also applies to routed Responses providers configured with `statelessResponses: true`, and to routed requests where a custom tool was lowered to a function but a delta result has no local call to establish its original type. Full replay preserves the call, result, and reasoning together; opencodex does not guess the result type or drop it. diff --git a/docs-site/src/content/docs/ko/guides/codex-integration.md b/docs-site/src/content/docs/ko/guides/codex-integration.md index 7e353c50f96..0d049a01fc0 100644 --- a/docs-site/src/content/docs/ko/guides/codex-integration.md +++ b/docs-site/src/content/docs/ko/guides/codex-integration.md @@ -205,13 +205,20 @@ Windows에서 Orca shell은 `CODEX_HOME`과 `ORCA_CODEX_HOME`을 Orca의 번들 네이티브 ChatGPT forward 요청의 로컬 재생 상태가 만료되었거나 없으면 opencodex는 upstream 요청 전에 `previous_response_not_found`를 반환합니다. Codex WebSocket 클라이언트는 일반 스트림 재시도 한도 안에서 다시 연결하고, 완료된 도구 호출과 결과를 포함한 현재 보유 -컨텍스트 전체를 다시 보낼 수 있습니다. 따라서 프록시의 1시간 캐시가 만료되었다는 이유만으로 -새 작업을 만들 필요는 없습니다. 캐시 한도와 보존 기간은 그대로이며, 클라이언트가 더 이상 -보유하지 않는 기록을 복구하는 기능은 아닙니다. HTTP 클라이언트는 이 오류를 직접 처리하고 -`previous_response_id` 없이 전체 컨텍스트를 다시 보내야 합니다. 같은 ID만 재시도해서는 -누락된 상태를 복구할 수 없습니다. - -`statelessResponses: true`로 설정한 routed Responses provider에도 같은 복구 신호가 적용됩니다. +컨텍스트 전체를 다시 보낼 수 있습니다. 따라서 프록시의 재생 캐시가 만료되었다는 이유만으로 +새 작업을 만들 필요는 없습니다. 재생 상태는 24시간 보존하며 기존 메모리·디스크·항목 수 +상한은 그대로입니다. 클라이언트가 더 이상 보유하지 않는 기록을 복구하는 기능은 아닙니다. +HTTP 클라이언트는 이 오류를 직접 처리하고 `previous_response_id` 없이 전체 컨텍스트를 다시 +보내야 합니다. 같은 ID만 재시도해서는 누락된 상태를 복구할 수 없습니다. + +routed 목적지에는 모두 같은 복구 신호가 적용됩니다. 프록시가 잃어버린 기록을 대신 볼 수 있는 +것은 네이티브 Responses 패스스루뿐입니다. 체인을 저장해 둔 백엔드로 `previous_response_id`를 +그대로 넘기기 때문입니다. 나머지 wire는 매 턴 요청에 담긴 입력만으로 대화를 다시 구성하므로, +재생이 실패한 채 전달하면 이번 턴 한 줄만 올라가고 대화가 조용히 사라집니다. 상태를 들고 +있어 보이는 셋도 마찬가지입니다. Devin은 매 턴 전체 대화를 다시 보내고, Cursor의 체크포인트 +참조는 방금 만료된 그 저장소에 있어 없으면 full-replay로 떨어지며, Kiro는 넘겨받은 턴으로 +conversation history를 다시 만듭니다. +`statelessResponses: true`로 설정한 routed Responses provider에도 같은 신호가 적용됩니다. routed 경로에서 custom 도구를 function으로 낮췄는데 증분 결과에 대응하는 로컬 호출 기록이 없을 때도 전체 기록을 다시 요청합니다. 호출과 결과, reasoning을 함께 재생하며 결과 유형을 추측하거나 버리지 않습니다. 상태를 저장하는 provider의 네이티브 function 및 네이티브 custom diff --git a/src/responses/continuation-ownership.ts b/src/responses/continuation-ownership.ts new file mode 100644 index 00000000000..10650587d97 --- /dev/null +++ b/src/responses/continuation-ownership.ts @@ -0,0 +1,29 @@ +import { effectiveAdapterContract, getAdapterDefinition, type AdapterWire } from "../adapters/registry"; + +/** + * Wires whose upstream holds the conversation itself, so a turn may reference history this + * process no longer has. + * + * The set is empty, and that is the finding rather than an oversight. The three wires that look + * like they belong here do not: + * + * - devin sends `mapOcxMessagesToDevin(parsed)` — the whole conversation — on every turn + * (`src/adapters/devin.ts`). Its session/thread id buys prompt caching, not remembered context. + * - cursor continues from `_providerContinuation.cursor.checkpointRef`, which is read out of the + * very store that just expired; without it `resolveCursorCheckpoint` returns a reason and the + * request falls back to `continuationMode: "full-replay"` over `parsed.context.messages` + * (`src/adapters/cursor/request-builder.ts`). + * - kiro builds `conversationState.history` from the parsed turns it was given + * (`src/adapters/kiro/payload.ts`); a conversation id alone reconstructs nothing. + * + * So for every translated wire a replay miss means the delta travels alone. Only the native + * Responses passthrough, which forwards `previous_response_id` untouched to a backend that stored + * the chain, can answer a turn whose history this process lost. + */ +export const PROVIDER_OWNED_CONTINUATION_WIRES: ReadonlySet = new Set(); + +/** The wire an adapter id resolves to through contract inheritance, or undefined if unknown. */ +export function resolvedAdapterWire(adapterId: unknown): AdapterWire | undefined { + if (typeof adapterId !== "string" || !getAdapterDefinition(adapterId)) return undefined; + return effectiveAdapterContract(adapterId).wire; +} diff --git a/src/responses/state.ts b/src/responses/state.ts index a36435aa0b6..43a1e3e43e9 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -39,7 +39,22 @@ import { } from "./state/spill-queue"; const MAX_STORED_RESPONSES = 1_000; -const RESPONSE_TTL_MS = 60 * 60 * 1_000; +/** + * Retention for locally replayed continuation state. + * + * A Codex client chained by `previous_response_id` sends ONLY the new turn and expects this + * process to hold everything before it, so this constant is the practical memory span of every + * conversation that does not go to the canonical ChatGPT backend. At the original one hour, a + * session resumed after lunch expanded to nothing and the delta — one user line — was all the + * provider ever saw, which reads to the operator as the model losing the conversation. + * + * A day is safe to hold because retention is no longer what bounds this store: the resident cap + * (MAX_STORED_RESPONSE_BYTES), the spill ceiling (MAX_SPILLED_RESPONSE_BYTES) and the entry count + * all evict oldest-first, and every turn re-stores the whole chain under a fresh id, so the live + * conversation is the last thing any of those three caps would drop. Raising the TTL therefore + * moves eviction from the clock to those budgets rather than growing the ceiling. + */ +export const RESPONSE_TTL_MS = 24 * 60 * 60 * 1_000; const SNAPSHOT_DEBOUNCE_MS = 2_000; /** Snapshot size below which the debounce stays at its base value. */ const SNAPSHOT_DEBOUNCE_SCALE_FROM_BYTES = 1 * 1024 * 1024; @@ -1253,7 +1268,7 @@ export function rememberResponseState( // `force` bypasses only the store:false skip: Codex sends `store:false` on every non-Azure // HTTP request (and WS inherits it), yet its WS turns still chain with previous_response_id. // The passthrough branch records with force so those chains can be expanded locally; the - // store stays in-memory with a 1h TTL, so this is a proxy-internal continuation cache, not + // store stays in-memory under RESPONSE_TTL_MS, so this is a proxy-internal continuation cache, not // real server-side response storage. if (request.store === false && !opts?.force) return; if (typeof response.id !== "string" || !Array.isArray(response.output)) return; diff --git a/src/server/index/live-sideband.ts b/src/server/index/live-sideband.ts index 33206468056..44acd372efd 100644 --- a/src/server/index/live-sideband.ts +++ b/src/server/index/live-sideband.ts @@ -12,9 +12,34 @@ import { } from "../ws-bridge"; import type { Server, ServerWebSocket } from "bun"; import { handleLive, logLiveSidebandFrame, parseLiveSidebandTarget, resolveLiveSidebandUpgrade } from "../live"; +import { RESPONSE_TTL_MS } from "../../responses/state"; export const MAX_WS_FRAME_BYTES = 50 * 1024 * 1024; +/** + * 0 means Bun never closes an idle socket, and this one value covers every socket kind the + * server accepts — the live sideband relay, where a quiet call is normal, and the Responses data + * plane, where quiet means the client is simply between turns. + * + * It is coupled to `RESPONSE_TTL_MS` whether or not anyone says so, which is why it is said here. + * A codex-rs client caches its `WebsocketSession` across turns and chains `previous_response_id` + * onto it; it only clears `last_request`/`last_response_rx` when it finds the connection closed. + * So a socket that outlives retention is a client that keeps referencing continuation state this + * process has already evicted. Two settings can hold that line and only these two: + * + * - a FINITE idle timeout below `MAX_WEBSOCKET_IDLE_TIMEOUT_SECONDS`, which closes the socket + * first and lets the client reset its own chain, or + * - this 0, which obliges the proxy to fail closed on the expired reference instead — + * `server/responses/request-prepare.ts` returns `previous_response_not_found`, the error + * codex-rs recognizes on a WebSocket turn and answers by replaying its full input. + * + * What must never happen is neither: an immortal socket plus a destination that silently accepts + * the orphaned delta. `tests/responses/ws-endpoint.test.ts` holds exactly that pair together. + * Raising the timeout off 0 is still worth doing for its own reasons (a dead peer holds a socket + * forever today), and Bun caps the value at 255 seconds, well inside the bound below. + */ export const WEBSOCKET_IDLE_TIMEOUT_SECONDS = 0; +/** Ceiling a finite websocket idle timeout must stay under, in seconds. See above. */ +export const MAX_WEBSOCKET_IDLE_TIMEOUT_SECONDS = Math.floor(RESPONSE_TTL_MS / 1_000); const LIVE_SIDEBAND_PENDING_MAX = 32; const LIVE_SIDEBAND_PENDING_BYTES_MAX = 1024 * 1024; diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 4d69c1ec895..81ffeb013f8 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -96,6 +96,7 @@ import { slugsEquivalent } from "../../providers/slug-codec"; import type { AgentTaskRecoveryFailureReason } from "./agent-task-recovery"; import { resolveWireProtocolOverride } from "../adapter-resolve"; import { hasUnmappedRoutedCustomToolOutput } from "../../responses/custom-tool-compat"; +import { PROVIDER_OWNED_CONTINUATION_WIRES, resolvedAdapterWire } from "../../responses/continuation-ownership"; import { isCodexReserveHelperUnsupported, CODEX_RESERVE_HELPER_UNSUPPORTED_MESSAGE, @@ -799,12 +800,23 @@ export async function prepareResponsesRequest( if (hasUnexpandedPreviousResponse) { const continuationProvider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire); - // Stateless destinations cannot resolve the omitted prefix. Stateful destinations may, - // but a lowered custom result still needs its call to recover the original wire type. - // Native function/custom continuations without lowering keep their upstream-owned state. - if (continuationProvider.adapter === "openai-responses" - && (continuationProvider.statelessResponses === true - || hasUnmappedRoutedCustomToolOutput(parsed._rawBody, continuationProvider.supportsResponsesCustomTools))) { + // Can the DESTINATION see the history this process failed to restore? Only the native + // Responses passthrough can: it forwards previous_response_id to a backend that stored the + // chain. Every translated wire rebuilds the conversation from this request's input alone — + // including the three that look stateful, for the reasons recorded in + // responses/continuation-ownership.ts — so a replay miss there is not a degraded turn. It is + // the entire conversation deleted, with one user line left in its place and nothing in the + // response saying so. Refuse before auth or upstream I/O and let the client resend. + const continuationWire = resolvedAdapterWire(continuationProvider.adapter); + const upstreamOwnsOmittedHistory = continuationWire === "openai-responses" + // Stateless destinations cannot resolve the omitted prefix. Stateful destinations may, + // but a lowered custom result still needs its call to recover the original wire type. + // Native function/custom continuations without lowering keep their upstream-owned state. + ? !(continuationProvider.statelessResponses === true + || hasUnmappedRoutedCustomToolOutput(parsed._rawBody, continuationProvider.supportsResponsesCustomTools)) + // An unknown adapter is left to the resolution error it already raises below. + : continuationWire === undefined || PROVIDER_OWNED_CONTINUATION_WIRES.has(continuationWire); + if (!upstreamOwnsOmittedHistory) { return formatErrorResponse( 400, "previous_response_not_found", diff --git a/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 67f863858f2..c80f2421210 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -639,14 +639,6 @@ export async function prepareResponsesTransport( ); } - if (adapter.name === "kiro" && parsed.previousResponseId && !parsed._previousResponseInputExpanded) { - return formatErrorResponse( - 400, - "invalid_request_error", - "Kiro continuation state is missing; start a new session instead of reusing this previous_response_id.", - ); - } - return { isOAuth401ReplayProvider, get sentOAuthSnapshot(): OAuthAccessSnapshot | undefined { diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 96ddb693a2f..6f0276696b4 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -331,7 +331,20 @@ no new destination-based migration. The existing stateless pass sets `store: fal stored continuation parameters, and repairs orphan calls/results without claiming execution success. A local replay-cache hit supplies history; a miss cannot reconstruct it, so callers receive `previous_response_not_found` before upstream dispatch and must resend complete history -without `previous_response_id`. Routed custom-tool lowering requires the same recovery when a delta +without `previous_response_id`. That refusal is not specific to the stateless flag: it covers every +destination that cannot see the prefix this process failed to restore, which is every destination +except the native Responses passthrough. The passthrough forwards the id and keeps its +upstream-owned state. `PROVIDER_OWNED_CONTINUATION_WIRES` in +`src/responses/continuation-ownership.ts` is deliberately empty and records why the three +candidates do not qualify: devin re-sends the whole conversation each turn, cursor reads its +`checkpointRef` out of the same expired store and otherwise falls back to `full-replay`, and kiro +rebuilds `conversationState.history` from the turns it was handed. A missed expansion on any of +them would forward the current turn alone under a normal 200 — the whole conversation replaced by +one line, with nothing in the response saying so. This also replaces kiro's former +`invalid_request_error`, which told the client to start a new session and therefore skipped the +recovery Codex performs on `previous_response_not_found`. Retention is the other half: local +continuation state is held for `RESPONSE_TTL_MS` (24 hours), long enough that an ordinary idle gap +resumes by expansion rather than by asking the client to replay. Routed custom-tool lowering requires the same recovery when a delta custom result has no local call, because its original wire type cannot be established and guessing it would send an unmatched result upstream. The check resolves the selected wire protocol and the request's own tool declarations after final route selection, so stateful destinations keep their diff --git a/tests/codex-integration/issue-702-expired-replay-state.test.ts b/tests/codex-integration/issue-702-expired-replay-state.test.ts index 84100540d4f..afc233754d0 100644 --- a/tests/codex-integration/issue-702-expired-replay-state.test.ts +++ b/tests/codex-integration/issue-702-expired-replay-state.test.ts @@ -13,6 +13,7 @@ import { flushPendingResponseSpillsForTests, rememberResponseState, responseStateMetrics, + RESPONSE_TTL_MS, setResponseStateByteCapForTests, type ResponseStateMetrics, } from "../../src/responses/state"; @@ -27,8 +28,8 @@ import { removeTreeWithRetry } from "../helpers/remove-tree"; const originalFetch = globalThis.fetch; const previousOpencodexHome = process.env.OPENCODEX_HOME; const previousApiToken = process.env.OPENCODEX_API_AUTH_TOKEN; -const EXPIRED_AGE_MS = 2 * 60 * 60 * 1_000; -const REPLAY_TTL_MS = 60 * 60 * 1_000; +const REPLAY_TTL_MS = RESPONSE_TTL_MS; +const EXPIRED_AGE_MS = REPLAY_TTL_MS + 60 * 60 * 1_000; const FIRST_RESPONSE_ID = "resp_issue_702_first"; const HISTORICAL_USER_SENTINEL = "issue-702 historical user context"; const HISTORICAL_ASSISTANT_SENTINEL = "issue-702 historical assistant context"; @@ -408,6 +409,165 @@ describe("routed replay recovery", () => { await upstream.stop(true); } }, SERVER_BUDGET_MS); + + test("a translated wire refuses an expired continuation instead of sending the delta alone", async () => { + // The reported symptom: Codex chained by previous_response_id, a gap longer than retention, + // and a Chat-wire destination that rebuilds the conversation from this request's input. The + // expansion misses, the id is stripped, and what reaches the model is the single line the + // user just typed -- with a normal 200 hiding it. Refuse, so the client resends everything. + const upstreamRequests: Record[] = []; + const realNow = Date.now; + let server: ReturnType | null = null; + const chunk = (delta: Record, finish: string | null) => + `data: ${JSON.stringify({ + id: "chatcmpl-routed", object: "chat.completion.chunk", created: 1, model: "test-model", + choices: [{ index: 0, delta, finish_reason: finish }], + })}\n\n`; + const upstream = Bun.serve({ + port: 0, + async fetch(request) { + upstreamRequests.push(await request.json() as Record); + return new Response( + chunk({ role: "assistant", content: "recovered" }, null) + chunk({}, "stop") + "data: [DONE]\n\n", + { headers: { "content-type": "text/event-stream" } }, + ); + }, + }); + + try { + Date.now = () => realNow() - EXPIRED_AGE_MS; + rememberResponseState( + { input: [inputMessage(HISTORICAL_USER_SENTINEL)], store: false }, + { + id: FIRST_RESPONSE_ID, + status: "completed", + output: [{ type: "message", role: "assistant", content: [{ type: "output_text", text: HISTORICAL_ASSISTANT_SENTINEL }] }], + }, + undefined, + { force: true }, + ); + Date.now = realNow; + expect(responseStateMetrics().oldestAgeMs).toBeGreaterThan(REPLAY_TTL_MS); + + saveConfig({ + port: 0, + hostname: "127.0.0.1", + defaultProvider: "chat-test", + providers: { + "chat-test": { + adapter: "openai-chat", + baseUrl: `${upstream.url.toString().replace(/\/$/, "")}/v1`, + allowPrivateNetwork: true, + authMode: "key", + apiKey: "synthetic-key", + defaultModel: "test-model", + models: ["test-model"], + }, + }, + } as OcxConfig); + server = startServer(0); + + const refused = await originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "chat-test/test-model", + previous_response_id: FIRST_RESPONSE_ID, + input: [inputMessage(CURRENT_USER_SENTINEL)], + stream: true, + }), + }); + expect(refused.status).toBe(400); + expect(await refused.json()).toMatchObject({ + error: { type: "invalid_request_error", code: "previous_response_not_found" }, + }); + expect(upstreamRequests).toHaveLength(0); + + // What the client does next: resend the whole conversation without the id. + const recovered = await originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "chat-test/test-model", + input: [inputMessage(HISTORICAL_USER_SENTINEL), inputMessage(CURRENT_USER_SENTINEL)], + stream: true, + }), + }); + expect(recovered.status).toBe(200); + await recovered.text(); + expect(upstreamRequests).toHaveLength(1); + const forwarded = JSON.stringify(upstreamRequests[0]); + expect(forwarded).toContain(HISTORICAL_USER_SENTINEL); + expect(forwarded).toContain(CURRENT_USER_SENTINEL); + } finally { + Date.now = realNow; + await server?.stop(true); + await upstream.stop(true); + } + }, SERVER_BUDGET_MS); + + test.each(["kiro", "cursor", "devin", "anthropic"] as const)( + "%s refuses an expired continuation: none of these can resolve the omitted prefix upstream", + async adapter => { + // The three provider-session wires look stateful and are not. devin re-sends the whole + // conversation each turn, cursor's checkpointRef is read from the store that just expired + // and falls back to full replay, and kiro rebuilds conversationState.history from the turns + // it was handed. So the refusal is not limited to the obviously translated wires. + const realNow = Date.now; + let server: ReturnType | null = null; + let upstreamCalls = 0; + try { + Date.now = () => realNow() - EXPIRED_AGE_MS; + rememberResponseState( + { input: [inputMessage(HISTORICAL_USER_SENTINEL)], store: false }, + { id: FIRST_RESPONSE_ID, status: "completed", output: [] }, + undefined, + { force: true }, + ); + Date.now = realNow; + globalThis.fetch = (async () => { + upstreamCalls += 1; + throw new Error("upstream must not be called"); + }) as typeof fetch; + saveConfig({ + port: 0, + hostname: "127.0.0.1", + defaultProvider: "wire-test", + providers: { + "wire-test": { + adapter, + baseUrl: "https://example.invalid/v1", + authMode: "key", + apiKey: "synthetic-key", + defaultModel: "test-model", + models: ["test-model"], + }, + }, + } as OcxConfig); + server = startServer(0); + const response = await originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "wire-test/test-model", + previous_response_id: FIRST_RESPONSE_ID, + input: [inputMessage(CURRENT_USER_SENTINEL)], + stream: true, + }), + }); + expect(response.status).toBe(400); + expect(await response.json()).toMatchObject({ + error: { type: "invalid_request_error", code: "previous_response_not_found" }, + }); + expect(upstreamCalls).toBe(0); + } finally { + Date.now = realNow; + globalThis.fetch = originalFetch; + await server?.stop(true); + } + }, + SERVER_BUDGET_MS, + ); }); describe("Issue #702 expired forward replay state", () => { diff --git a/tests/oauth/state-store-sweeper.test.ts b/tests/oauth/state-store-sweeper.test.ts index b162e2214bb..5b691d28d6d 100644 --- a/tests/oauth/state-store-sweeper.test.ts +++ b/tests/oauth/state-store-sweeper.test.ts @@ -43,6 +43,7 @@ import { clearResponseStateMemoryForTests, rememberResponseState, responseStateMetrics, + RESPONSE_TTL_MS, } from "../../src/responses/state"; import { __resetAntigravityReplayCache, @@ -226,7 +227,9 @@ describe("state-store sweeper", () => { for (const name of ["responses-continuation", "antigravity-replay"]) { registerStateStore(STATE_STORE_REGISTRATIONS.find(registration => registration.name === name)!); } - const result = sweepExpired(Date.now() + 60 * 60 * 1_000 + 1); + // Past both retentions: the Antigravity replay cache expires after an hour, the responses + // continuation store after RESPONSE_TTL_MS. One tick has to clear both rows. + const result = sweepExpired(Date.now() + RESPONSE_TTL_MS + 60 * 60 * 1_000); expect(result.rowsRemoved).toBe(2); expect(responseStateMetrics().count).toBe(0); expect(antigravityReplayMetrics().sessions).toBe(0); diff --git a/tests/responses/responses-state.test.ts b/tests/responses/responses-state.test.ts index 1906d6b1e16..c1eaa5770cd 100644 --- a/tests/responses/responses-state.test.ts +++ b/tests/responses/responses-state.test.ts @@ -2173,7 +2173,7 @@ describe("Responses previous_response_id state", () => { const realNow = Date.now; setResponseStateByteCapForTests(1_024); try { - Date.now = () => realNow() - 2 * 60 * 60 * 1_000; + Date.now = () => realNow() - 25 * 60 * 60 * 1_000; rememberLarge("resp_ttl_spill", "t".repeat(8_000)); const ttlFile = spillFileNames(home)[0]!; Date.now = realNow; @@ -2775,8 +2775,8 @@ describe("Responses previous_response_id state", () => { try { const realNow = Date.now; try { - // Store an old heavy entry, then advance time past the 1h TTL. - Date.now = () => realNow() - 2 * 60 * 60 * 1_000; + // Store an old heavy entry, then advance time past the 24h RESPONSE_TTL_MS. + Date.now = () => realNow() - 25 * 60 * 60 * 1_000; const oldBody = { model: "cursor/grok-4.5", input: "o".repeat(6_000), store: false }; const oldJson = buildResponseJSON([{ type: "text_delta", text: "ok" }, { type: "done" }], "cursor/grok-4.5"); rememberResponseState(oldBody, oldJson, { cursor: { conversationId: "conv_old" } }, { force: true }); @@ -3228,12 +3228,12 @@ describe("Responses previous_response_id state", () => { await flushResponseState(); clearResponseStateMemoryForTests(); - // Rewrite the snapshot with an expired createdAt (2h ago > 1h TTL). + // Rewrite the snapshot with a createdAt past the 24h RESPONSE_TTL_MS. const path = join(home, "responses-state.json"); const snapshot = JSON.parse(readFileSync(path, "utf-8")) as { states: [string, { createdAt: number }][]; }; - for (const [, state] of snapshot.states) state.createdAt = Date.now() - 2 * 60 * 60 * 1_000; + for (const [, state] of snapshot.states) state.createdAt = Date.now() - 25 * 60 * 60 * 1_000; writeFileSync(path, JSON.stringify(snapshot)); const second = { diff --git a/tests/responses/ws-endpoint.test.ts b/tests/responses/ws-endpoint.test.ts index 46de26fdd71..fd9472d747a 100644 --- a/tests/responses/ws-endpoint.test.ts +++ b/tests/responses/ws-endpoint.test.ts @@ -10,6 +10,11 @@ import { sendResponseToWebSocket, type WsData, } from "../../src/server/ws-bridge"; +import { + MAX_WEBSOCKET_IDLE_TIMEOUT_SECONDS, + WEBSOCKET_IDLE_TIMEOUT_SECONDS, +} from "../../src/server/index/live-sideband"; +import { RESPONSE_TTL_MS } from "../../src/responses/state"; import type { ServerWebSocket } from "bun"; function mockWs(sendResult = 1): { ws: ServerWebSocket; sent: string[] } { @@ -56,6 +61,25 @@ describe("WS endpoint re-framer (120/132)", () => { expect(source).toContain("if (!logged) finalizeLog(turnAbort.signal.aborted ? 499 : response.status);"); }); + test("an immortal websocket is paired with a proxy that fails closed on expired continuation state", () => { + // codex-rs reuses its cached WebsocketSession across turns and chains previous_response_id + // onto it, clearing that chain only when it finds the socket closed. So one of two things + // must be true, and this test refuses the third case where neither is. + const idleTimeout = WEBSOCKET_IDLE_TIMEOUT_SECONDS; + if (idleTimeout > 0) { + expect(idleTimeout).toBeLessThan(MAX_WEBSOCKET_IDLE_TIMEOUT_SECONDS); + return; + } + // The socket never closes on its own, so the refusal has to come from the request path. + const gate = readFileSync( + new URL("../../src/server/responses/request-prepare.ts", import.meta.url), + "utf8", + ); + expect(gate).toContain("hasUnexpandedPreviousResponse"); + expect(gate).toContain("previous_response_not_found"); + expect(MAX_WEBSOCKET_IDLE_TIMEOUT_SECONDS).toBe(Math.floor(RESPONSE_TTL_MS / 1_000)); + }); + test("generate=false warmup completes locally without upstream and forces full next request", () => { const frames = buildWarmupCompletionFrames({ model: "gpt-5.5", generate: false }).map(f => JSON.parse(f));