From dcd2d0740301785ec624168073c7bfc59c10bb9f Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 13:53:36 +0900 Subject: [PATCH 01/10] fix(search): retry clean empty answers without masking truncation Co-authored-by: Cortes Ventures --- .../content/docs/reference/proxy-formats.md | 7 ++ src/web-search/loop.ts | 58 ++++++++- structure/runtime.md | 4 + tests/web-search/web-search.test.ts | 116 ++++++++++++++++++ 4 files changed, 182 insertions(+), 3 deletions(-) diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 8975c944cf..1bd63cac79 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -24,6 +24,13 @@ should select among several targets. Credential-bearing model, image, video, and search requests do not automatically follow HTTP redirects, including same-origin redirects. Configure the final upstream API URL instead of a redirecting alias. A redirect does not cause the server to resend credentials or the request body to its destination. The response owner retains its existing error or relay behavior; native Responses and compact routes can return the original 3xx and `Location` to the client. Client redirect behavior is separate from this server transport policy. +## Empty search answers + +After hosted search, a clean but empty forced-answer pass receives one additional answer +attempt with tools removed and existing results retained. This can incur another model +request. A second empty answer fails; malformed calls and provider refusal or truncation +outcomes are preserved without this retry. + ## Endpoint overview | Client surface | Endpoint | Successful non-stream result | Successful stream or socket result | diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 99b275ed8d..8feceb4c27 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -3,6 +3,7 @@ import type { AdapterEvent, OcxMessage, OcxParsedRequest, OcxProviderConfig, Ocx import { namespacedToolName, toolChoiceToolPredicate } from "../types"; import { cloneProviderOpaqueToolCallMetadata } from "../responses/provider-opaque-metadata"; import type { AttemptRecoveryKind } from "../usage/log"; +import { isTruncatedStopReason } from "../responses/truncated-stop-reason"; import { bridgeToResponsesSSE } from "../bridge"; import { runWebSearch, type SidecarOutcome, type SidecarOutcomeRecorder, type SidecarSettings } from "./executor"; import { runAnthropicWebSearch } from "./anthropic-executor"; @@ -230,6 +231,24 @@ function forcedAnswerNudge(): OcxMessage { }; } +/** + * Transient developer-role nudge for the ONE recovery pass after a forced answer came back empty. + * The recovery also removes every tool, so the model has nothing to call and can only return text; + * this turn says so explicitly rather than relying on the removal alone. Like {@link forcedAnswerNudge} + * it is iteration-local and never touches the persisted `messages`. + */ +function forcedAnswerRetryNudge(): OcxMessage { + return { + role: "developer", + content: + "Your previous response contained no usable answer. Web search has finished for this turn and " + + "no tools are available for this response. Answer the user's question now in assistant text, " + + "using the web search results already gathered above. If those results are insufficient, say " + + "what is missing instead of returning an empty response.", + timestamp: Date.now(), + }; +} + function jsonError(status: number, message: string): Response { return new Response(JSON.stringify({ error: { message, type: "upstream_error", code: null } }), { status, @@ -370,7 +389,9 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise 0 + let iterMessages: OcxMessage[] = forceAnswer && executedSearchCount > 0 ? [...messages, forcedAnswerNudge()] : messages; + // #1001 follow-up: the recovery pass for an empty forced answer. Removing every tool leaves the + // model nothing to call, and the extra developer turn asks it for the text it just failed to + // produce. `toolChoice: "none"` is what drops those definitions in the adapter, so the retry + // cannot repeat the same empty or tool-shaped response. + const recoveringEmptyAnswer = forceAnswer && emptyAnswerRetries > 0; + if (recoveringEmptyAnswer) iterMessages = [...iterMessages, forcedAnswerRetryNudge()]; const iterParsed: OcxParsedRequest = { ...parsed, stream: true, - context: { ...parsed.context, messages: iterMessages, tools: forceAnswer ? toolsNoWebSearch : allTools }, + ...(recoveringEmptyAnswer ? { options: { ...parsed.options, toolChoice: "none" as const } } : {}), + context: { ...parsed.context, messages: iterMessages, tools: recoveringEmptyAnswer ? [] : forceAnswer ? toolsNoWebSearch : allTools }, }; // One cumulative header deadline spans every pool-key 429 rotation in this model iteration. // clear() stops only its timer after final headers; the direct turn signal remains attached to @@ -847,9 +875,33 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise event.type === "done"); + if (terminalEvent?.type === "done" && isTruncatedStopReason(terminalEvent.stopReason)) { + // A provider refusal or truncation is authoritative, even without text. + // Preserve it once; neither an empty-answer retry nor a generic 502 applies. + yield* replay(split.passthrough.slice(split.streamedPassthroughCount)); + return; + } if (terminalEvent?.type === "done" && (split.hasMalformedToolCall || (!split.hasRealToolCall && !hasVisibleAssistantText(split.passthrough)))) { + // #1001 fixed the silent success by failing here. A malformed call still fails: it + // reports a protocol problem, and replaying it would only re-ask an unwell upstream. + // Silence is different — it is recoverable, so retry exactly once with the results + // already gathered before failing the turn. + console.warn("[web-search-loop] unusable forced answer", JSON.stringify({ + model: parsed.modelId, + recoveryAttempt: emptyAnswerRetries, + searchCalls: split.calls.length, + malformed: split.hasMalformedToolCall, + stopReason: terminalEvent.stopReason, + eventTypes: [...new Set(split.passthrough.map(event => event.type))], + })); + if (!split.hasMalformedToolCall && !split.hasRealToolCall && emptyAnswerRetries === 0) { + emptyAnswerRetries++; + console.warn("[web-search-loop] empty forced answer — retrying once without tools"); + yield { type: "heartbeat" }; + continue; + } throw new LoopError(502, "forced-answer pass produced no usable assistant output"); } } diff --git a/structure/runtime.md b/structure/runtime.md index 495745051e..ea8e98e277 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -192,3 +192,7 @@ Codex display-cache expiry, retained main-policy evidence, and reset history fol Chat helper admission in `src/server/responses/core.ts` follows the [deferred stored-main contract](providers/openai-tiers.md): only a needed Direct OpenAI helper claims stored main, after terminal vision, routed vision and search exclusions. + +### Empty forced search answers + +`src/web-search/loop.ts` makes at most one extra answer attempt after a clean forced-answer terminal with no visible output or tool call. The recovery has no tools and reuses gathered search results. Malformed calls fail, and recognized refusal/truncation terminals pass through unchanged, including empty or partial answers. The extra generation may incur provider usage. diff --git a/tests/web-search/web-search.test.ts b/tests/web-search/web-search.test.ts index e182308a0d..5c41ad5e82 100644 --- a/tests/web-search/web-search.test.ts +++ b/tests/web-search/web-search.test.ts @@ -133,6 +133,122 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = expect(frames.some(frame => frame.event === "response.completed")).toBe(true); expect(frames.some(frame => frame.event === "response.failed")).toBe(false); }); + + // #1001 chose to fail rather than complete silently, which turned silence into a dead turn: + // the user sees "stream disconnected before completion: forced-answer pass produced no usable + // assistant output". Silence is recoverable, so the pass is retried once with no tools before + // the same error is reported. Malformed calls still fail immediately. + describe("empty forced answer recovery", () => { + function sequenceAdapter(passes: AdapterEvent[][], seen: OcxParsedRequest[]): ProviderAdapter { + let pass = 0; + return { + name: "sequence", + buildRequest: (request) => { + seen.push(request); + return { url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }; + }, + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + for (const event of passes[Math.min(pass++, passes.length - 1)] ?? []) yield event; + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + } + + async function drivePasses(passes: AdapterEvent[][], seen: OcxParsedRequest[] = [], ordinaryTool = false) { + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }, ...(ordinaryTool ? [{ type: "function", name: "fixture", parameters: { type: "object", properties: {} } }] : [])] }), + adapter: sequenceAdapter(passes, seen), + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + }); + return collectSse(response.body!); + } + + test("an empty forced pass is retried once and completes", async () => { + const frames = await drivePasses([ + webSearchFirstPass, + [{ type: "done" }], + [{ type: "text_delta", text: "recovered answer" }, { type: "done" }], + ]); + expect(frames.some(frame => frame.event === "response.completed")).toBe(true); + expect(frames.some(frame => frame.event === "response.failed")).toBe(false); + }); + + test("the recovery pass asks for text with every tool removed", async () => { + const seen: OcxParsedRequest[] = []; + await drivePasses([ + webSearchFirstPass, + [{ type: "done" }], + [{ type: "text_delta", text: "recovered answer" }, { type: "done" }], + ], seen); + // The search pass plus the empty forced pass plus exactly one recovery — no extra upstream call. + expect(seen).toHaveLength(3); + const recovery = seen[2]!; + expect(recovery.options.toolChoice).toBe("none"); + expect(recovery.context.tools).toEqual([]); + // The results gathered by the search reach the recovery turn as a tool result ... + expect(recovery.context.messages.filter(message => message.role === "toolResult")).toHaveLength(1); + // ... and the recovery turn carries the developer nudge that asks for the missing text. + expect(recovery.context.messages.some(message => + message.role === "developer" && String(message.content).includes("no tools are available"))) + .toBe(true); + }); + + test("recovery removes ordinary tools as well as web search", async () => { + const seen: OcxParsedRequest[] = []; + await drivePasses([webSearchFirstPass, [{ type: "done" }], [{ type: "text_delta", text: "answer" }, { type: "done" }]], seen, true); + expect(seen).toHaveLength(3); + expect(seen[1]!.context.tools.length).toBeGreaterThan(0); + expect(seen[2]!.context.tools).toEqual([]); + expect(seen[2]!.options.toolChoice).toBe("none"); + }); + + for (const [stopReason, reason] of [["refusal", "content_filter"], ["content_filter", "content_filter"], ["max_tokens", "max_output_tokens"], ["length", "max_output_tokens"]]) { + for (const partial of [false, true]) { + test(`${stopReason} partial=${partial} stays authoritative without a retry`, async () => { + const seen: OcxParsedRequest[] = []; + const terminalPass: AdapterEvent[] = [ + ...(partial ? [{ type: "text_delta" as const, text: "partial answer" }] : []), + { type: "done", stopReason }, + ]; + const frames = await drivePasses([webSearchFirstPass, terminalPass, [{ type: "done" }]], seen); + expect(seen).toHaveLength(2); + expect(frames.filter(frame => ["response.incomplete", "response.completed", "response.failed"].includes(frame.event)).map(frame => frame.event)).toEqual(["response.incomplete"]); + expect(frames.find(frame => frame.event === "response.incomplete")!.data.response.incomplete_details.reason).toBe(reason); + if (partial) expect(frames.filter(frame => frame.event === "response.output_text.delta").map(frame => frame.data.delta).join("")).toBe("partial answer"); + }); + } + } + + test("a persistent empty forced pass still fails after the one recovery", async () => { + const seen: OcxParsedRequest[] = []; + const frames = await drivePasses([ + webSearchFirstPass, + [{ type: "done" }], + [{ type: "done" }], + ], seen); + expect(seen).toHaveLength(3); + expect(frames.some(frame => frame.event === "response.failed")).toBe(true); + expect(frames.some(frame => frame.event === "response.completed")).toBe(false); + }); + + test("a malformed forced call is not retried", async () => { + const seen: OcxParsedRequest[] = []; + const frames = await drivePasses([ + webSearchFirstPass, + [{ type: "tool_call_start", id: "", name: "" }, { type: "tool_call_end" }, { type: "done" }], + [{ type: "text_delta", text: "recovered answer" }, { type: "done" }], + ], seen); + expect(seen).toHaveLength(2); + expect(frames.some(frame => frame.event === "response.failed")).toBe(true); + }); + }); }); const routedProvider: OcxProviderConfig = { From 45f703f356b20172ffd9a7301a9ecaf967deccdf Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 13:54:48 +0900 Subject: [PATCH 02/10] test(search): narrow terminal fixture projection --- tests/web-search/web-search.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/web-search/web-search.test.ts b/tests/web-search/web-search.test.ts index 5c41ad5e82..61abb904fd 100644 --- a/tests/web-search/web-search.test.ts +++ b/tests/web-search/web-search.test.ts @@ -219,8 +219,9 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = ]; const frames = await drivePasses([webSearchFirstPass, terminalPass, [{ type: "done" }]], seen); expect(seen).toHaveLength(2); - expect(frames.filter(frame => ["response.incomplete", "response.completed", "response.failed"].includes(frame.event)).map(frame => frame.event)).toEqual(["response.incomplete"]); - expect(frames.find(frame => frame.event === "response.incomplete")!.data.response.incomplete_details.reason).toBe(reason); + expect(frames.filter(frame => ["response.incomplete", "response.completed", "response.failed"].includes(frame.event ?? "")).map(frame => frame.event)).toEqual(["response.incomplete"]); + const terminalResponse = frames.find(frame => frame.event === "response.incomplete")!.data.response as { incomplete_details: { reason: string } }; + expect(terminalResponse.incomplete_details.reason).toBe(reason); if (partial) expect(frames.filter(frame => frame.event === "response.output_text.delta").map(frame => frame.data.delta).join("")).toBe("partial answer"); }); } From 3652da790b58177f5ae7eebecb9c8ba3527e9109 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 13:55:23 +0900 Subject: [PATCH 03/10] test(search): exercise live output truncation without duplicate replay --- tests/web-search/web-search.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/web-search/web-search.test.ts b/tests/web-search/web-search.test.ts index 61abb904fd..9bf7c02444 100644 --- a/tests/web-search/web-search.test.ts +++ b/tests/web-search/web-search.test.ts @@ -157,7 +157,7 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = }; } - async function drivePasses(passes: AdapterEvent[][], seen: OcxParsedRequest[] = [], ordinaryTool = false) { + async function drivePasses(passes: AdapterEvent[][], seen: OcxParsedRequest[] = [], ordinaryTool = false, liveOutput = false) { const response = await runWithWebSearch({ parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }, ...(ordinaryTool ? [{ type: "function", name: "fixture", parameters: { type: "object", properties: {} } }] : [])] }), adapter: sequenceAdapter(passes, seen), @@ -166,6 +166,7 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, maxSearches: 1, + streamRoutedModelOutput: liveOutput, }); return collectSse(response.body!); } @@ -217,7 +218,7 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = ...(partial ? [{ type: "text_delta" as const, text: "partial answer" }] : []), { type: "done", stopReason }, ]; - const frames = await drivePasses([webSearchFirstPass, terminalPass, [{ type: "done" }]], seen); + const frames = await drivePasses([webSearchFirstPass, terminalPass, [{ type: "done" }]], seen, false, true); expect(seen).toHaveLength(2); expect(frames.filter(frame => ["response.incomplete", "response.completed", "response.failed"].includes(frame.event ?? "")).map(frame => frame.event)).toEqual(["response.incomplete"]); const terminalResponse = frames.find(frame => frame.event === "response.incomplete")!.data.response as { incomplete_details: { reason: string } }; From 59ec04b907b56a324971f23fd5350795f3039021 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 14:00:49 +0900 Subject: [PATCH 04/10] fix(cursor): preserve first overflow and bound stable-thread remints Co-authored-by: SB Yoon <44089734+yansigit@users.noreply.github.com> --- .../content/docs/reference/proxy-formats.md | 8 + scripts/test-layout/layout.json | 1 + src/adapters/cursor.ts | 176 ++++++---- src/adapters/cursor/cursor-errors.ts | 12 + src/adapters/cursor/thread-continuity.ts | 87 +++++ structure/providers/cursor.md | 4 + tests/fixtures/test-layout-expected.json | 1 + tests/providers/cursor/cursor-adapter.test.ts | 327 ++++++++++++++++++ .../cursor-continuity-retention.test.ts | 27 ++ 9 files changed, 570 insertions(+), 73 deletions(-) create mode 100644 tests/providers/cursor/cursor-continuity-retention.test.ts diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 8975c944cf..865cb63834 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -24,6 +24,14 @@ should select among several targets. Credential-bearing model, image, video, and search requests do not automatically follow HTTP redirects, including same-origin redirects. Configure the final upstream API URL instead of a redirecting alias. A redirect does not cause the server to resend credentials or the request body to its destination. The response owner retains its existing error or relay behavior; native Responses and compact routes can return the original 3xx and `Location` to the client. Client redirect behavior is separate from this server transport policy. +## Cursor context overflow + +Cursor's first bare context overflow is surfaced to the client. Later eligible requests +with a stable client thread may recover with up to three conversation remints per retained +scope. The in-memory allowance expires after one idle hour, eviction, or restart. Requests +without a stable thread, tool-result resumes, partial output, compaction and quota errors do +not use this recovery. This does not infer whether a task is making progress. + ## Endpoint overview | Client surface | Endpoint | Successful non-stream result | Successful stream or socket result | diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 7241f26266..8a40a91e13 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -538,6 +538,7 @@ "crash-guard.test.ts": "service", "credential-redirect-guard.test.ts": "lib", "cursor-adapter.test.ts": "providers/cursor", + "cursor-continuity-retention.test.ts": "providers/cursor", "cursor-arg-normalize.test.ts": "providers/cursor", "cursor-blob-integrity.test.ts": "providers/cursor", "cursor-blob.test.ts": "providers/cursor", diff --git a/src/adapters/cursor.ts b/src/adapters/cursor.ts index 7382c25d8c..a240ae1982 100644 --- a/src/adapters/cursor.ts +++ b/src/adapters/cursor.ts @@ -3,7 +3,7 @@ import type { AdapterEvent, OcxProviderConfig } from "../types"; import type { ProviderAdapter } from "./base"; import { isTranslatorBudgetExceededError } from "../lib/translator-budget"; import { cursorExecDeniedMessage, cursorRequestDeclaresFullAccess } from "./cursor/exec-policy"; -import { isCursorBenignCancelError, isCursorInvalidArgumentError, isCursorRootEnvelopeError, safeCursorErrorMessage, type CursorSizeContext } from "./cursor/cursor-errors"; +import { isCursorBenignCancelError, isCursorInvalidArgumentError, isCursorOverflowRemintCandidate, isCursorRootEnvelopeError, safeCursorErrorMessage, type CursorSizeContext } from "./cursor/cursor-errors"; import { cursorCheckpointModelAffinityId, inferCursorContextWindow, isCursorExternalWireModel } from "./cursor/discovery"; import { createCursorKvStore, type CursorKvStore } from "./cursor/kv-store"; import { mapCursorServerMessage } from "./cursor/message-mapper"; @@ -31,7 +31,14 @@ import { debugProviderDiagnostic } from "../lib/debug"; import { isDebugEnabled } from "../lib/debug-settings"; import { createAdapterTierMetadata } from "../providers/fastwire"; import { estimateTokens } from "../lib/token-estimate"; -import { rememberCursorThreadConversation } from "./cursor/thread-continuity"; +import { + cursorOverflowRemintScopeKey, + markCursorOverflowSurfaced, + recordCursorOverflowRemint, + rememberCursorThreadConversation, + shouldSkipCursorOverflowRemint, + shouldSurfaceCursorOverflowFirst, +} from "./cursor/thread-continuity"; import { runCursorTurnWithRetry } from "./cursor/transport-retry"; import { cursorRequestHasShellAlias, cursorRequestUsesCodeMode } from "./cursor/tool-definitions"; import { @@ -399,84 +406,107 @@ export function createCursorAdapter(provider: OcxProviderConfig, deps: CursorAda ); }; - try { - await runOnce(request); - } catch (err) { - const outputGuardRetryText = - err instanceof CursorToolResultEchoError - ? CURSOR_ECHO_RETRY_CONTINUATION_TEXT - : err instanceof CursorRoutingCommentaryError - ? CURSOR_ROUTING_COMMENTARY_RETRY_TEXT - : undefined; - // One-shot corrective retry for guarded external output (devlog 260826 gap-10/11). - // The quarantine guarantees no client-visible delta escaped, so a fresh-conversation - // retry is safe. A second rejection propagates as an error rather than looping. - if ( - outputGuardRetryText - && !emittedOutput - && !replayUnsafe - && !incoming.abortSignal?.aborted - ) { - debugProviderDiagnostic( - "cursor", - err instanceof CursorToolResultEchoError - ? "envelope-echo-retry" - : "routing-commentary-retry", - { - wireModel: request.modelId, - conversationHash: request.conversationId.slice(0, 16), - }, + const remintConversationId = (failedConversationId: string) => { + lastTransport = undefined; + _parsed._cursorConversationId = undefined; + const next = createCursorRequest(_parsed, { forceFreshConversation: true }); + rekeyContextUsage(failedConversationId, next.conversationId); + _parsed._cursorConversationId = next.conversationId; + // Persist recovery for store:false clients that send any stable Cursor thread owner, so + // the next turn does not recompute the stale deterministic thread hash. Isolated helper / + // compaction turns must not park their throwaway id under the parent or Desktop owner. + const threadOwner = cursorClientThreadOwner(_parsed); + if (threadOwner && _parsed._cursorIsolateConversation !== true) { + rememberCursorThreadConversation( + threadOwner, + next.conversationId, + _parsed._cursorIdentityScope, ); - const echoedConversationId = request.conversationId; - lastTransport = undefined; - _parsed._cursorConversationId = undefined; - request = { - ...createCursorRequest(_parsed, { forceFreshConversation: true }), - echoRetryContinuationText: outputGuardRetryText, - }; - rekeyContextUsage(echoedConversationId, request.conversationId); - _parsed._cursorConversationId = request.conversationId; - const echoThreadOwner = cursorClientThreadOwner(_parsed); - if (echoThreadOwner && _parsed._cursorIsolateConversation !== true) { - rememberCursorThreadConversation( - echoThreadOwner, - request.conversationId, - _parsed._cursorIdentityScope, - ); - } + } + return next; + }; + + for (;;) { + try { await runOnce(request); - } else { - // One-shot fallback for external-model Connect invalid_argument before any - // non-heartbeat output. Retries apply only to safe plain-user turns; tool-result - // resumes, local exec/MCP side effects, and already-emitted output fail closed. + break; + } catch (err) { + const outputGuardRetryText = + err instanceof CursorToolResultEchoError + ? CURSOR_ECHO_RETRY_CONTINUATION_TEXT + : err instanceof CursorRoutingCommentaryError + ? CURSOR_ROUTING_COMMENTARY_RETRY_TEXT + : undefined; + // One-shot corrective retry for guarded external output (devlog 260826 gap-10/11). + // The quarantine guarantees no client-visible delta escaped, so a fresh-conversation + // retry is safe. A second rejection propagates as an error rather than looping. if ( - !isCursorInvalidArgumentError(err) - || !isCursorExternalWireModel(request.modelId) - || lastRawIsToolResult - || emittedOutput - || replayUnsafe - || incoming.abortSignal?.aborted + outputGuardRetryText + && !emittedOutput + && !replayUnsafe + && !incoming.abortSignal?.aborted ) { - throw err; - } - const failedConversationId = request.conversationId; - lastTransport = undefined; - _parsed._cursorConversationId = undefined; - request = createCursorRequest(_parsed, { forceFreshConversation: true }); - rekeyContextUsage(failedConversationId, request.conversationId); - _parsed._cursorConversationId = request.conversationId; - // Persist recovery for store:false clients that send any stable Cursor thread owner, so - // the next turn does not recompute the stale deterministic thread hash. Isolated helper / - // compaction turns must not park their throwaway id under the parent or Desktop owner. - const threadOwner = cursorClientThreadOwner(_parsed); - if (threadOwner && _parsed._cursorIsolateConversation !== true) { - rememberCursorThreadConversation( - threadOwner, - request.conversationId, + debugProviderDiagnostic( + "cursor", + err instanceof CursorToolResultEchoError + ? "envelope-echo-retry" + : "routing-commentary-retry", + { + wireModel: request.modelId, + conversationHash: request.conversationId.slice(0, 16), + }, + ); + const echoedConversationId = request.conversationId; + request = { + ...remintConversationId(echoedConversationId), + echoRetryContinuationText: outputGuardRetryText, + }; + await runOnce(request); + break; + } else { + const overflowRemintSafe = + !lastRawIsToolResult + && !emittedOutput + && !replayUnsafe + && request.contextUsageStoreCheckpoints !== false + && !incoming.abortSignal?.aborted; + const overflowScopeKey = cursorOverflowRemintScopeKey( + cursorClientThreadOwner(_parsed), _parsed._cursorIdentityScope, ); + if ( + overflowScopeKey + && overflowRemintSafe + && isCursorOverflowRemintCandidate(err, requestSizeContext) + ) { + if (shouldSkipCursorOverflowRemint(overflowScopeKey)) throw err; + if (shouldSurfaceCursorOverflowFirst(overflowScopeKey)) { + markCursorOverflowSurfaced(overflowScopeKey); + throw err; + } + if (!recordCursorOverflowRemint(overflowScopeKey)) throw err; + if (inheritedCheckpointRef) invalidateCursorCheckpoint(inheritedCheckpointRef); + request = remintConversationId(request.conversationId); + continue; + } + + // One-shot fallback for external-model Connect invalid_argument before any + // non-heartbeat output. Retries apply only to safe plain-user turns; tool-result + // resumes, local exec/MCP side effects, and already-emitted output fail closed. + if ( + !isCursorInvalidArgumentError(err) + || !isCursorExternalWireModel(request.modelId) + || lastRawIsToolResult + || emittedOutput + || replayUnsafe + || incoming.abortSignal?.aborted + ) { + throw err; + } + request = remintConversationId(request.conversationId); + await runOnce(request); + break; } - await runOnce(request); } } if ( diff --git a/src/adapters/cursor/cursor-errors.ts b/src/adapters/cursor/cursor-errors.ts index b7005db3c2..fba44b2f99 100644 --- a/src/adapters/cursor/cursor-errors.ts +++ b/src/adapters/cursor/cursor-errors.ts @@ -191,6 +191,18 @@ function bareReLooksLikeOverflow(context?: CursorSizeContext): boolean { return estimatedInputTokens >= OVERFLOW_MIN_FRACTION * contextWindow; } +/** + * True when a transport error is the bare 0-token resource_exhausted overflow shape + * (not quota/rate) that should surface for Codex compact or remint on later hits. + */ +export function isCursorOverflowRemintCandidate(err: unknown, sizeContext?: CursorSizeContext): boolean { + const message = errorMessage(err); + if (!message) return false; + const lower = message.toLowerCase(); + if (!isCursorZeroTokenResourceExhausted(lower)) return false; + return classifyCursorError(message, sizeContext) === "Cursor context limit exceeded"; +} + export function isCursorZeroTokenResourceExhausted(lowerMessage: string): boolean { if (!lowerMessage.includes("resource_exhausted") && !lowerMessage.includes("resource exhausted")) return false; // Any explicit quota/rate cue wins: this is a real 429. diff --git a/src/adapters/cursor/thread-continuity.ts b/src/adapters/cursor/thread-continuity.ts index 6cb0e2cf35..aa3c3dac32 100644 --- a/src/adapters/cursor/thread-continuity.ts +++ b/src/adapters/cursor/thread-continuity.ts @@ -65,3 +65,90 @@ export function lookupCursorThreadConversation( export function clearCursorThreadContinuityForTests(): void { overrides.clear(); } + +/** Max conversation-id remints after the first surfaced overflow per retained scope. */ +export const CURSOR_OVERFLOW_REMINT_MAX = 3; +export const CURSOR_OVERFLOW_REMINT_TTL_MS = 60 * 60 * 1000; +export const CURSOR_OVERFLOW_REMINT_MAX_ENTRIES = 2_048; + +type OverflowRemintState = { + surfaced: boolean; + remintCount: number; + skip: boolean; + updatedAt: number; +}; + +const overflowRemintByScope = new Map(); + +function pruneOverflowRemints(at: number): void { + for (const [scopeKey, entry] of overflowRemintByScope) { + if (at - entry.updatedAt > CURSOR_OVERFLOW_REMINT_TTL_MS) overflowRemintByScope.delete(scopeKey); + } + while (overflowRemintByScope.size > CURSOR_OVERFLOW_REMINT_MAX_ENTRIES) { + const oldest = overflowRemintByScope.keys().next().value; + if (oldest === undefined) break; + overflowRemintByScope.delete(oldest); + } +} + +function overflowRemintEntry(scopeKey: string): OverflowRemintState { + const at = now(); + pruneOverflowRemints(at); + const existing = overflowRemintByScope.get(scopeKey); + if (existing) { + existing.updatedAt = at; + overflowRemintByScope.delete(scopeKey); + overflowRemintByScope.set(scopeKey, existing); + return existing; + } + const fresh: OverflowRemintState = { surfaced: false, remintCount: 0, skip: false, updatedAt: at }; + overflowRemintByScope.set(scopeKey, fresh); + pruneOverflowRemints(at); + return fresh; +} + +/** Stable client-thread ownership survives conversation remints; wire ids alone do not. */ +export function cursorOverflowRemintScopeKey( + threadOwner: string | undefined, + identityScope?: string, +): string | null { + if (!threadOwner) return null; + return `overflow\0${cursorThreadScopeKey(threadOwner, identityScope)}`; +} + +/** True until the first overflow for this scope has been surfaced for Codex compact. */ +export function shouldSurfaceCursorOverflowFirst(scopeKey: string): boolean { + pruneOverflowRemints(now()); + return overflowRemintByScope.get(scopeKey)?.surfaced !== true; +} + +export function markCursorOverflowSurfaced(scopeKey: string): void { + const entry = overflowRemintEntry(scopeKey); + entry.surfaced = true; +} + +export function shouldSkipCursorOverflowRemint(scopeKey: string): boolean { + pruneOverflowRemints(now()); + const entry = overflowRemintByScope.get(scopeKey); + return entry?.skip === true || (entry?.remintCount ?? 0) >= CURSOR_OVERFLOW_REMINT_MAX; +} + +/** Record one overflow remint; returns false when the cap is exhausted. */ +export function recordCursorOverflowRemint(scopeKey: string): boolean { + const entry = overflowRemintEntry(scopeKey); + if (entry.skip || entry.remintCount >= CURSOR_OVERFLOW_REMINT_MAX) { + entry.skip = true; + return false; + } + entry.remintCount += 1; + return true; +} + +export function clearCursorOverflowRemintForTests(): void { + overflowRemintByScope.clear(); +} + +export function cursorOverflowRemintCountForTests(): number { + pruneOverflowRemints(now()); + return overflowRemintByScope.size; +} diff --git a/structure/providers/cursor.md b/structure/providers/cursor.md index be42793e0b..3d734a33ed 100644 --- a/structure/providers/cursor.md +++ b/structure/providers/cursor.md @@ -82,3 +82,7 @@ constraints cannot widen the canonical shape. Bare shell bridge names are reject on the freeform path. Namespaced tools do not acquire bare-shell behavior. Regression coverage lives in `tests/providers/cursor/cursor-tool-definitions.test.ts`. + +## Overflow remint boundary + +`src/adapters/cursor.ts` surfaces the first bare context overflow before attempting conversation remint on later eligible requests. `cursorClientThreadOwner` recognizes both client thread aliases; `src/adapters/cursor/thread-continuity.ts` limits recovery to three remints per retained identity-scoped owner, with a one-hour idle TTL and 2,048-entry bound. Conversation-only requests have no stable owner and do not automatically remint. Quota/rate errors, tool-result resumes, partial output, local side effects and compaction remain fail-closed. Retention expiry, eviction or process restart resets the in-memory allowance; this is not a persistent lifetime cap or semantic-progress policy. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 62724ffed2..cf071d04fc 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -373,6 +373,7 @@ "crash-guard.test.ts": "service", "credential-redirect-guard.test.ts": "lib", "cursor-adapter.test.ts": "providers/cursor", + "cursor-continuity-retention.test.ts": "providers/cursor", "cursor-arg-normalize.test.ts": "providers/cursor", "cursor-blob-integrity.test.ts": "providers/cursor", "cursor-blob.test.ts": "providers/cursor", diff --git a/tests/providers/cursor/cursor-adapter.test.ts b/tests/providers/cursor/cursor-adapter.test.ts index a86df7c243..82049b83f4 100644 --- a/tests/providers/cursor/cursor-adapter.test.ts +++ b/tests/providers/cursor/cursor-adapter.test.ts @@ -4,6 +4,7 @@ import { cursorExecDeniedMessage, } from "../../../src/adapters/cursor"; import { + clearCursorOverflowRemintForTests, clearCursorThreadContinuityForTests, lookupCursorThreadConversation, } from "../../../src/adapters/cursor/thread-continuity"; @@ -817,3 +818,329 @@ describe("Cursor adapter live transport", () => { clearCursorCheckpointsForTests(); }); }); +const LARGE_OVERFLOW_CONTENT = "word ".repeat(100_000); + +function bareOverflowError(): Error { + return Object.assign( + new Error("Cursor context limit exceeded: Cursor Connect error resource_exhausted: Error"), + { code: "resource_exhausted" }, + ); +} + +function overflowTurnBody(threadId?: string): OcxParsedRequest { + return { + modelId: "cursor/auto", + context: { messages: [{ role: "user", content: LARGE_OVERFLOW_CONTENT, timestamp: 1 }] }, + stream: false, + options: {}, + _cursorIdentityScope: "acct-overflow-remint", + ...(threadId ? { _clientThreadId: threadId } : { _cursorConversationId: "cursor_overflow_base" }), + }; +} + +describe("Cursor overflow conversation remint", () => { + test("first bare overflow surfaces without reminting the conversation id", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const seen: string[] = []; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run(request) { + attempts += 1; + seen.push(request.conversationId); + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + + const body = overflowTurnBody("overflow-surface-first"); + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => events.push(event)); + + expect(attempts).toBe(1); + expect(seen).toHaveLength(1); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + type: "error", + message: expect.stringContaining("Cursor context limit exceeded"), + }); + }); + + test("second overflow remints and persists thread override", async () => { + clearCursorOverflowRemintForTests(); + clearCursorThreadContinuityForTests(); + let attempts = 0; + const seen: string[] = []; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run(request) { + attempts += 1; + seen.push(request.conversationId); + if (attempts === 1) { + throw bareOverflowError(); + } + yield { type: "done" } satisfies CursorServerMessage; + }, + writeClient() {}, + }), + rekeyContextUsage: () => {}, + }); + + const threadId = "overflow-remint-thread"; + const body = overflowTurnBody(threadId); + + const surfaceEvents: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => surfaceEvents.push(event)); + expect(attempts).toBe(1); + expect(surfaceEvents.some(event => event.type === "error")).toBe(true); + + seen.length = 0; + attempts = 0; + const remintEvents: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => remintEvents.push(event)); + + expect(attempts).toBe(2); + expect(seen).toHaveLength(2); + expect(seen[1]).not.toBe(seen[0]); + expect(remintEvents.some(event => event.type === "done")).toBe(true); + expect(lookupCursorThreadConversation(threadId, "acct-overflow-remint")).toBe(seen[1]); + expect(body._cursorConversationId).toBe(seen[1]); + }); + + test("fourth overflow skips remint after surface-first and three remints", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run() { + attempts += 1; + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + + const body = overflowTurnBody("overflow-cap-skip"); + await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); + expect(attempts).toBe(1); + + attempts = 0; + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => events.push(event)); + + expect(attempts).toBe(4); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + type: "error", + message: expect.stringContaining("Cursor context limit exceeded"), + }); + }); + + test("quota-cue resource_exhausted does not remint and surfaces as rate limit", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run() { + attempts += 1; + throw Object.assign( + new Error("Cursor rate limit exceeded: resource_exhausted: too many requests"), + { code: "resource_exhausted" }, + ); + }, + writeClient() {}, + }), + }); + + const body = overflowTurnBody("overflow-quota-cue"); + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => events.push(event)); + + expect(attempts).toBe(1); + expect(events[0]).toMatchObject({ + type: "error", + message: expect.stringContaining("Cursor rate limit exceeded"), + }); + }); + + test("does not overflow-remint on tool-result resumes", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run() { + attempts += 1; + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + + const body: OcxParsedRequest = { + modelId: "cursor/auto", + context: { + messages: [ + { role: "user", content: LARGE_OVERFLOW_CONTENT, timestamp: 1 }, + { + role: "assistant", + model: "cursor/auto", + timestamp: 2, + content: [{ type: "toolCall", id: "call_1", name: "read_file", namespace: "mcp__fs", arguments: { path: "a.txt" } }], + }, + { + role: "toolResult", + toolCallId: "call_1", + toolName: "read_file", + toolNamespace: "mcp__fs", + content: "FILE CONTENTS HERE", + isError: false, + timestamp: 3, + }, + ], + }, + stream: false, + options: {}, + _cursorConversationId: "cursor_overflow_tool", + _cursorIdentityScope: "acct-overflow-remint", + }; + + await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); + expect(attempts).toBe(1); + }); + + test("does not overflow-remint compaction turns", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const seen: string[] = []; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run(request) { + attempts += 1; + seen.push(request.conversationId); + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + + const body = overflowTurnBody("overflow-compaction"); + await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); + attempts = 0; + seen.length = 0; + body._compactionRequest = true; + body._cursorIsolateConversation = true; + await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); + + expect(attempts).toBe(1); + expect(seen).toHaveLength(1); + }); + + test("does not overflow-remint after non-heartbeat output was emitted", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ + ...provider, + apiKey: "cursor-token", + }, { + createTransport: () => ({ + async *run() { + attempts += 1; + yield { type: "text", text: "partial" } satisfies CursorServerMessage; + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + + const body = overflowTurnBody("overflow-after-output"); + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body, { headers: new Headers() }, event => events.push(event)); + + expect(attempts).toBe(1); + expect(events.some(event => event.type === "text_delta")).toBe(true); + expect(events.some(event => event.type === "error")).toBe(true); + }); +}); + + +describe("Cursor overflow accounting across requests", () => { + for (const ownerField of ["_clientThreadId", "_cursorClientThreadId"] as const) { + test(`${ownerField} retains the cap across successful remints`, async () => { + clearCursorOverflowRemintForTests(); + clearCursorThreadContinuityForTests(); + let attempts = 0; + let failNext = true; + const seen: string[] = []; + const adapter = createCursorAdapter({ ...provider, apiKey: "cursor-token" }, { + createTransport: () => ({ + async *run(request) { + attempts++; + seen.push(request.conversationId); + if (failNext) { failNext = false; throw bareOverflowError(); } + yield { type: "done" } satisfies CursorServerMessage; + }, + writeClient() {}, + }), + rekeyContextUsage: () => {}, + }); + const body = () => { + const parsed = overflowTurnBody(); + parsed._cursorConversationId = undefined; + parsed[ownerField] = `cross-request-${ownerField}`; + return parsed; + }; + await adapter.runTurn?.(body(), { headers: new Headers() }, () => {}); + expect(attempts).toBe(1); + for (let remint = 0; remint < 3; remint++) { + failNext = true; + const before = attempts; + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body(), { headers: new Headers() }, event => events.push(event)); + expect(attempts - before).toBe(2); + expect(seen[seen.length - 1]).not.toBe(seen[seen.length - 2]); + expect(events.some(event => event.type === "done")).toBe(true); + } + failNext = true; + const before = attempts; + const events: AdapterEvent[] = []; + await adapter.runTurn?.(body(), { headers: new Headers() }, event => events.push(event)); + expect(attempts - before).toBe(1); + expect(events.some(event => event.type === "error")).toBe(true); + }); + } + test("conversation-only clients never gain an automatic remint allowance", async () => { + clearCursorOverflowRemintForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ ...provider, apiKey: "cursor-token" }, { + createTransport: () => ({ + async *run() { attempts++; throw bareOverflowError(); }, + writeClient() {}, + }), + }); + for (let turn = 0; turn < 3; turn++) { + const events: AdapterEvent[] = []; + await adapter.runTurn?.(overflowTurnBody(), { headers: new Headers() }, event => events.push(event)); + expect(attempts).toBe(turn + 1); + expect(events.some(event => event.type === "error")).toBe(true); + } + }); +}); diff --git a/tests/providers/cursor/cursor-continuity-retention.test.ts b/tests/providers/cursor/cursor-continuity-retention.test.ts new file mode 100644 index 0000000000..2d3f833e8f --- /dev/null +++ b/tests/providers/cursor/cursor-continuity-retention.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, test } from "bun:test"; +import { + clearCursorOverflowRemintForTests, + CURSOR_OVERFLOW_REMINT_MAX_ENTRIES, + cursorOverflowRemintCountForTests, + markCursorOverflowSurfaced, + shouldSkipCursorOverflowRemint, + shouldSurfaceCursorOverflowFirst, +} from "../../../src/adapters/cursor/thread-continuity"; + +describe("Cursor overflow remint retention", () => { + test("bounds per-scope state", () => { + clearCursorOverflowRemintForTests(); + for (let index = 0; index < CURSOR_OVERFLOW_REMINT_MAX_ENTRIES + 20; index++) { + markCursorOverflowSurfaced(`scope-${index}`); + } + expect(cursorOverflowRemintCountForTests()).toBe(CURSOR_OVERFLOW_REMINT_MAX_ENTRIES); + clearCursorOverflowRemintForTests(); + }); + + test("read-only checks do not allocate retention entries", () => { + clearCursorOverflowRemintForTests(); + expect(shouldSurfaceCursorOverflowFirst("missing")).toBe(true); + expect(shouldSkipCursorOverflowRemint("missing")).toBe(false); + expect(cursorOverflowRemintCountForTests()).toBe(0); + }); +}); From 36625c78be4ca0ff5a94e145cdffe52a5ba9092d Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 14:01:36 +0900 Subject: [PATCH 05/10] test(cursor): activate remint guards after first overflow --- tests/providers/cursor/cursor-adapter.test.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/providers/cursor/cursor-adapter.test.ts b/tests/providers/cursor/cursor-adapter.test.ts index 82049b83f4..1f42b697dc 100644 --- a/tests/providers/cursor/cursor-adapter.test.ts +++ b/tests/providers/cursor/cursor-adapter.test.ts @@ -1017,9 +1017,12 @@ describe("Cursor overflow conversation remint", () => { stream: false, options: {}, _cursorConversationId: "cursor_overflow_tool", + _clientThreadId: "overflow-tool-result", _cursorIdentityScope: "acct-overflow-remint", }; + await adapter.runTurn?.(overflowTurnBody("overflow-tool-result"), { headers: new Headers() }, () => {}); + attempts = 0; await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); expect(attempts).toBe(1); }); @@ -1057,6 +1060,7 @@ describe("Cursor overflow conversation remint", () => { test("does not overflow-remint after non-heartbeat output was emitted", async () => { clearCursorOverflowRemintForTests(); let attempts = 0; + let emitPartial = false; const adapter = createCursorAdapter({ ...provider, apiKey: "cursor-token", @@ -1064,7 +1068,7 @@ describe("Cursor overflow conversation remint", () => { createTransport: () => ({ async *run() { attempts += 1; - yield { type: "text", text: "partial" } satisfies CursorServerMessage; + if (emitPartial) yield { type: "text", text: "partial" } satisfies CursorServerMessage; throw bareOverflowError(); }, writeClient() {}, @@ -1072,6 +1076,9 @@ describe("Cursor overflow conversation remint", () => { }); const body = overflowTurnBody("overflow-after-output"); + await adapter.runTurn?.(body, { headers: new Headers() }, () => {}); + attempts = 0; + emitPartial = true; const events: AdapterEvent[] = []; await adapter.runTurn?.(body, { headers: new Headers() }, event => events.push(event)); @@ -1113,9 +1120,11 @@ describe("Cursor overflow accounting across requests", () => { for (let remint = 0; remint < 3; remint++) { failNext = true; const before = attempts; + const priorConversation = seen[seen.length - 1]; const events: AdapterEvent[] = []; await adapter.runTurn?.(body(), { headers: new Headers() }, event => events.push(event)); expect(attempts - before).toBe(2); + expect(seen[seen.length - 2]).toBe(priorConversation); expect(seen[seen.length - 1]).not.toBe(seen[seen.length - 2]); expect(events.some(event => event.type === "done")).toBe(true); } From 321b9b1cd1e9031732f46893d6cbbc0c774cfca1 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 14:07:33 +0900 Subject: [PATCH 06/10] fix(live): validate sideband upstream before client upgrade Co-authored-by: Kosta Milovanovic --- .../content/docs/reference/proxy-formats.md | 8 + src/server/index.ts | 368 ++++++++++++- src/server/ws-bridge.ts | 21 + structure/runtime.md | 4 + tests/server/server-live.test.ts | 493 +++++++++++++++++- 5 files changed, 865 insertions(+), 29 deletions(-) diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 8975c944cf..dedd5b83ec 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -24,6 +24,14 @@ should select among several targets. Credential-bearing model, image, video, and search requests do not automatically follow HTTP redirects, including same-origin redirects. Configure the final upstream API URL instead of a redirecting alias. A redirect does not cause the server to resend credentials or the request body to its destination. The response owner retains its existing error or relay behavior; native Responses and compact routes can return the original 3xx and `Location` to the client. Client redirect behavior is separate from this server transport policy. +## Live sideband connection failures + +The proxy completes the upstream live sideband handshake before accepting the client +WebSocket. An upstream rejection fails the upgrade with 502; a ten-second handshake timeout +returns 504. Bun does not expose the exact upstream handshake status, so an upstream 404/410 +cannot currently be forwarded precisely. A successful connection preserves the initial session +frames in order. This handshake policy is separate from the Responses WebSocket transport. + ## Endpoint overview | Client surface | Endpoint | Successful non-stream result | Successful stream or socket result | diff --git a/src/server/index.ts b/src/server/index.ts index 2cb11c1e9f..e0af64d255 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -8,6 +8,8 @@ import { buildResponsesWsData, sendResponseToWebSocket, sendTextFrame, + type LiveSidebandUpstreamFailure, + type LiveSidebandUpstreamHandoff, type WsData, } from "./ws-bridge"; import type { Server, ServerWebSocket } from "bun"; @@ -319,6 +321,29 @@ function withRemoteCatalogKeyId(response: Response, admission: DataPlaneAdmissio const LIVE_SIDEBAND_PENDING_MAX = 32; const LIVE_SIDEBAND_PENDING_BYTES_MAX = 1024 * 1024; const LIVE_SIDEBAND_CLOSE_FALLBACK_MS = 1_000; +/** + * Bound the pre-upgrade upstream handshake. A sideband join that cannot reach 101 + * must fail the client upgrade promptly rather than hold it open indefinitely. + */ +export const LIVE_SIDEBAND_UPSTREAM_OPEN_TIMEOUT_MS = 10_000; + +/** + * Outcome of the upstream sideband handshake performed before the client upgrade. + * + * `ok: false` carries the HTTP status the client upgrade must fail with. Only an + * upgrade failure reaches codex-rs as a connect error, and only a connect error + * ends its sideband reconnect loop (`realtime_conversation/sideband.rs`: the `Err` + * arm always breaks). A 101 followed by a close is instead read as `TransportLost` + * and retried forever against the same, permanently dead call id. + */ +export type LiveSidebandUpstreamOpenResult = + | { + ok: true; + socket: WebSocket; + /** Owns capture and terminal events until the downstream relay attaches. */ + handoff: LiveSidebandUpstreamHandoff; + } + | { ok: false; status: number; code: string; message: string; socket?: WebSocket }; export function exceedsLiveSidebandFrameByteLimit(frameBytes: number): boolean { return frameBytes > MAX_WS_FRAME_BYTES; @@ -416,6 +441,48 @@ function armLiveSidebandCloseFallback(ws: ServerWebSocket, upstream: Web }, LIVE_SIDEBAND_CLOSE_FALLBACK_MS); } +function closeLiveSidebandBeforeUpgrade( + upstream: WebSocket, + release: () => void, + code = 1000, + reason = "", +): void { + // There is no downstream socket to own this transport yet. Mirror + // closeLiveSideband's bounded close contract directly: release only after a + // close event or an observed CLOSED state, never merely after requesting close. + let released = false; + let fallback: ReturnType | undefined; + const releaseOnce = (): void => { + if (released) return; + released = true; + if (fallback !== undefined) clearTimeout(fallback); + release(); + }; + upstream.addEventListener("close", releaseOnce, { once: true }); + if (upstream.readyState === WebSocket.CLOSED) { + releaseOnce(); + return; + } + fallback = setTimeout(() => { + if (upstream.readyState === WebSocket.CLOSED) { + releaseOnce(); + return; + } + try { + upstream.close(1000, "upstream close timeout"); + } catch { + /* retain ownership until CLOSED is observed */ + } + if ((upstream.readyState as number) === 3) releaseOnce(); + }, LIVE_SIDEBAND_CLOSE_FALLBACK_MS); + try { + upstream.close(code, reason); + } catch { + /* the bounded fallback retries without releasing ownership */ + } + if ((upstream.readyState as number) === 3) releaseOnce(); +} + function closeLiveSideband(ws: ServerWebSocket, code = 1000, reason = ""): void { if (ws.data.liveClosing) return; ws.data.liveClosing = true; @@ -448,29 +515,243 @@ function closeLiveSideband(ws: ServerWebSocket, code = 1000, reason = "" } } -function attachLiveSidebandUpstream( +/** + * Dial the upstream sideband and report whether its handshake reached 101. + * + * Bun's client WebSocket does not surface the upstream handshake status, so the + * result is "opened" or "failed" and nothing finer. That is sufficient for the + * property this exists to guarantee: the client is never told the relay is live + * when it is not. Frames the upstream sends before the client socket exists are + * captured and handed back by `drain`, because a session preamble such as + * `session.created` arrives immediately after the upstream opens. + */ +export function openLiveSidebandUpstream( + url: string, + headers: Record, + createWebSocket: LiveSidebandWebSocketFactory = (socketUrl, socketHeaders) => ( + new WebSocket(socketUrl, { headers: socketHeaders } as unknown as string[]) + ), + timeoutMs: number = LIVE_SIDEBAND_UPSTREAM_OPEN_TIMEOUT_MS, + signal?: AbortSignal, +): Promise { + return new Promise(resolve => { + let socket: WebSocket; + try { + socket = createWebSocket(url, headers); + } catch { + resolve({ ok: false, status: 502, code: "upstream_error", message: "voice upstream connect failed" }); + return; + } + + const buffered: Array = []; + let bufferedBytes = 0; + let capturing = true; + let settled = false; + let terminalFailure: LiveSidebandUpstreamFailure | undefined; + let removeAbortListener = (): void => {}; + + const finish = (result: LiveSidebandUpstreamOpenResult): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + removeAbortListener(); + resolve(result); + }; + const timer = setTimeout(() => { + const failure = { status: 504, code: "upstream_timeout", message: "voice upstream did not open in time" }; + terminalFailure = failure; + capturing = false; + buffered.length = 0; + bufferedBytes = 0; + finish({ ok: false, ...failure, socket }); + try { + socket.close(); + } catch { + /* ignore */ + } + }, timeoutMs); + + const failCapture = (failure: LiveSidebandUpstreamFailure): void => { + if (!capturing || terminalFailure) return; + terminalFailure = failure; + capturing = false; + buffered.length = 0; + bufferedBytes = 0; + finish({ ok: false, ...failure, socket }); + try { + socket.close(1009, "sideband preamble overflow"); + } catch { + /* the terminal failure is already retained for the downstream handoff */ + } + }; + const handoff: LiveSidebandUpstreamHandoff = { + failure: () => terminalFailure, + take: () => { + capturing = false; + if (terminalFailure) return { ok: false, failure: terminalFailure }; + const frames = buffered.slice(); + buffered.length = 0; + bufferedBytes = 0; + return { ok: true, frames }; + }, + }; + + socket.addEventListener("message", event => { + if (!capturing) return; + const frameBytes = webSocketFrameBytes(event.data); + if (exceedsLiveSidebandFrameByteLimit(frameBytes)) { + failCapture({ status: 502, code: "upstream_overflow", message: "voice upstream preamble frame is too large" }); + return; + } + if (buffered.length >= LIVE_SIDEBAND_PENDING_MAX) { + failCapture({ status: 502, code: "upstream_overflow", message: "voice upstream sent too many preamble frames" }); + return; + } + if (exceedsLiveSidebandPendingByteLimit(bufferedBytes, frameBytes)) { + failCapture({ status: 502, code: "upstream_overflow", message: "voice upstream preamble is too large" }); + return; + } + if (typeof event.data === "string") buffered.push(event.data); + else if (event.data instanceof ArrayBuffer) buffered.push(Buffer.from(new Uint8Array(event.data))); + else if (ArrayBuffer.isView(event.data)) { + buffered.push(Buffer.from(new Uint8Array(event.data.buffer, event.data.byteOffset, event.data.byteLength))); + } else return; + bufferedBytes += frameBytes; + }); + socket.addEventListener("open", () => { + finish({ + ok: true, + socket, + handoff, + }); + }); + socket.addEventListener("error", () => { + const failure = { status: 502, code: "upstream_error", message: "voice upstream rejected the sideband join" }; + terminalFailure ??= failure; + capturing = false; + buffered.length = 0; + bufferedBytes = 0; + finish({ ok: false, ...terminalFailure, socket }); + try { + socket.close(); + } catch { + /* the terminal failure is already retained */ + } + }); + socket.addEventListener("close", event => { + const failure = { + status: 502, + code: "upstream_error", + message: `voice upstream closed before opening (code ${event.code})`, + closeCode: event.code, + closeReason: event.reason, + }; + terminalFailure ??= failure; + capturing = false; + buffered.length = 0; + bufferedBytes = 0; + finish({ ok: false, ...terminalFailure, socket }); + }); + const abortOpen = (): void => { + const failure = { status: 499, code: "request_cancelled", message: "voice sideband join was cancelled" }; + terminalFailure ??= failure; + capturing = false; + buffered.length = 0; + bufferedBytes = 0; + finish({ ok: false, ...terminalFailure, socket }); + try { + socket.close(); + } catch { + /* the cancelled join no longer owns the socket */ + } + }; + if (signal) { + signal.addEventListener("abort", abortOpen, { once: true }); + removeAbortListener = () => signal.removeEventListener("abort", abortOpen); + if (signal.aborted) abortOpen(); + } + }); +} + +export function attachLiveSidebandUpstream( ws: ServerWebSocket, createWebSocket: LiveSidebandWebSocketFactory = (url, headers) => ( new WebSocket(url, { headers } as unknown as string[]) ), ): void { - const url = ws.data.liveUpstreamUrl; - if (!url) { - closeLiveSideband(ws, 1011, "missing upstream"); - return; - } + // A socket carried in from the upgrade handler already completed its handshake + // before the client was told 101. Reuse it rather than dialing a second upstream. + const preOpened = ws.data.liveUpstream; let upstream: WebSocket; - try { - // Bun accepts per-handshake headers; the DOM lib types only list protocol arrays. - upstream = createWebSocket(url, ws.data.liveUpstreamHeaders ?? {}); - } catch { - closeLiveSideband(ws, 1011, "upstream connect failed"); - return; + if (preOpened) { + upstream = preOpened; + } else { + const url = ws.data.liveUpstreamUrl; + if (!url) { + closeLiveSideband(ws, 1011, "missing upstream"); + return; + } + try { + // Bun accepts per-handshake headers; the DOM lib types only list protocol arrays. + upstream = createWebSocket(url, ws.data.liveUpstreamHeaders ?? {}); + } catch { + closeLiveSideband(ws, 1011, "upstream connect failed"); + return; + } } ws.data.liveUpstream = upstream; ws.data.liveClosing = false; ws.data.cancel = () => closeLiveSideband(ws, 1000, "client closed"); + upstream.addEventListener("close", (event) => { + if (ws.data.liveUpstream !== upstream) return; + ws.data.liveClosing = true; + finalizeLiveSideband(ws, upstream); + try { + ws.close(event.code || 1000, event.reason || ""); + } catch { + /* ignore */ + } + }); + upstream.addEventListener("error", () => { + if (ws.data.liveUpstream !== upstream) return; + closeLiveSideband(ws, 1011, "upstream error"); + }); + + if (preOpened) { + // The upstream opened before this socket existed, so its `open` event has already + // fired and the listener below will never run. Its early frames were captured for + // us; forward the capture now rather than dropping the session preamble. + const handoff = ws.data.liveUpstreamHandoff; + ws.data.liveUpstreamHandoff = undefined; + const takeover = handoff?.take(); + if (!takeover?.ok || preOpened.readyState !== WebSocket.OPEN) { + const failure = takeover && !takeover.ok ? takeover.failure : undefined; + closeLiveSideband( + ws, + failure?.closeCode ?? 1011, + failure?.closeReason ?? "upstream closed before relay attachment", + ); + return; + } + ws.data.liveOpened = true; + for (const frame of takeover.frames) { + try { + // Mirror the live message listener exactly: same ceiling, same diagnostic + // record. These frames are upstream-to-client like any other. + if (exceedsLiveSidebandFrameByteLimit(webSocketFrameBytes(frame))) { + closeLiveSideband(ws, 1009, "message too large"); + return; + } + logLiveSidebandFrame("u2c", frame); + ws.send(frame); + } catch { + closeLiveSideband(ws, 1011, "client send failed"); + return; + } + } + } + upstream.addEventListener("open", () => { if (ws.data.liveUpstream !== upstream || ws.data.liveClosing) return; ws.data.liveOpened = true; @@ -503,20 +784,6 @@ function attachLiveSidebandUpstream( closeLiveSideband(ws, 1011, "client send failed"); } }); - upstream.addEventListener("close", (event) => { - if (ws.data.liveUpstream !== upstream) return; - ws.data.liveClosing = true; - finalizeLiveSideband(ws, upstream); - try { - ws.close(event.code || 1000, event.reason || ""); - } catch { - /* ignore */ - } - }); - upstream.addEventListener("error", () => { - if (ws.data.liveUpstream !== upstream) return; - closeLiveSideband(ws, 1011, "upstream error"); - }); } // GUI static serving extracted to ./server/gui-static. Re-exported below to keep the @@ -2185,19 +2452,64 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server turnAdmissionLease.release()); + } else { + turnAdmissionLease.release(); + } + addFinalRequestLog(requestId, start, logCtx, upstreamHandshake.status); + console.error(`[live] sideband upstream handshake failed: ${upstreamHandshake.message}`); + return withCors( + formatErrorResponse(upstreamHandshake.status, upstreamHandshake.code, upstreamHandshake.message), + req, + policy, + ); + } + const handoffFailure = upstreamHandshake.handoff.failure(); + if (handoffFailure || upstreamHandshake.socket.readyState !== WebSocket.OPEN) { + closeLiveSidebandBeforeUpgrade(upstreamHandshake.socket, () => turnAdmissionLease.release()); + const failure = handoffFailure ?? { + status: 502, + code: "upstream_error", + message: "voice upstream closed before client upgrade", + }; + addFinalRequestLog(requestId, start, logCtx, failure.status); + return withCors(formatErrorResponse(failure.status, failure.code, failure.message), req, policy); + } addFinalRequestLog(requestId, start, logCtx, 101); if (requestServer.upgrade(req, { data: { kind: "live-sideband", + liveUpstream: upstreamHandshake.socket, liveUpstreamUrl: resolved.upstreamWsUrl, liveUpstreamHeaders: resolved.headers, + liveUpstreamHandoff: upstreamHandshake.handoff, livePending: [], livePendingBytes: 0, - liveOpened: false, + liveOpened: true, liveTurnAdmissionLease: turnAdmissionLease, } satisfies WsData, })) return undefined as unknown as Response; - turnAdmissionLease.release(); + // The upgrade was refused after the upstream had already opened; drop it. + try { + upstreamHandshake.handoff.take(); + } catch { + /* ignore */ + } + closeLiveSidebandBeforeUpgrade(upstreamHandshake.socket, () => turnAdmissionLease.release()); return withCors(formatErrorResponse(426, "upgrade_required", "WebSocket upgrade failed"), req, policy); } diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index 5777b45a10..7b4e4c37f8 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -39,6 +39,8 @@ export interface WsData { /** Total encoded bytes retained in livePending while the upstream connects. */ livePendingBytes?: number; liveOpened?: boolean; + /** Owns captured frames and terminal state until the downstream relay attaches. */ + liveUpstreamHandoff?: LiveSidebandUpstreamHandoff; /** Once teardown starts, ignore new client frames until the upstream closes. */ liveClosing?: boolean; /** Schedules one bounded close retry without surrendering native-main ownership. */ @@ -48,6 +50,25 @@ export interface WsData { admissionLease?: AdmissionReservation>; } +export interface LiveSidebandUpstreamFailure { + status: number; + code: string; + message: string; + closeCode?: number; + closeReason?: string; +} + +export type LiveSidebandUpstreamTakeover = + | { ok: true; frames: Array } + | { ok: false; failure: LiveSidebandUpstreamFailure }; + +export interface LiveSidebandUpstreamHandoff { + /** Observe failure before the downstream upgrade without ending capture. */ + failure(): LiveSidebandUpstreamFailure | undefined; + /** Atomically ends capture and transfers buffered frames or terminal state. */ + take(): LiveSidebandUpstreamTakeover; +} + /** * Build the Responses WebSocket upgrade payload. * diff --git a/structure/runtime.md b/structure/runtime.md index 6d733bf8b5..19f7dc3c95 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -212,3 +212,7 @@ cooldowns and response-driven retry remain authoritative. The management quota DTO keeps Combo editing aligned with scoped inference evidence; see [Combo editor routing quota](gui-and-management-api.md#combo-editor-routing-quota). + +### Live sideband handshake + +`src/server/index.ts` establishes the authorized upstream live sideband before accepting the client WebSocket upgrade. `openLiveSidebandUpstream` bounds the handshake to ten seconds and retains at most 32 frames and 1 MiB of preamble within the frame limit. `src/server/ws-bridge.ts` defines the runtime handoff carrying captured frames or terminal state. Failed handshakes return 502/504; exact upstream 404/410 status is unavailable from Bun's client WebSocket. Admission ownership lasts until upstream close/CLOSED, including failed upgrades and failed attachment. The ordinary Responses WebSocket exchange remains separate. diff --git a/tests/server/server-live.test.ts b/tests/server/server-live.test.ts index f6d4da8916..441d6bf01a 100644 --- a/tests/server/server-live.test.ts +++ b/tests/server/server-live.test.ts @@ -15,13 +15,20 @@ import { type ReadinessGate, } from "../../src/server/readiness"; import { + attachLiveSidebandUpstream, enqueueLiveSidebandPendingFrame, exceedsLiveSidebandFrameByteLimit, exceedsLiveSidebandPendingByteLimit, MAX_WS_FRAME_BYTES, + openLiveSidebandUpstream, startServer, } from "../../src/server"; -import { beginShutdownDrain, isDraining, resetLifecycleDrainStateForTests } from "../../src/server/lifecycle"; +import { + activeRegistryMetrics, + beginShutdownDrain, + isDraining, + resetLifecycleDrainStateForTests, +} from "../../src/server/lifecycle"; import type { OcxConfig } from "../../src/types"; import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; @@ -1739,3 +1746,487 @@ describe("GET /readyz while draining", () => { } }); }); + +/** + * A sideband join must not report 101 unless the upstream handshake actually + * succeeded. A 101 followed by a close is read by codex-rs as `TransportLost`, + * which it recovers from by rejoining the same call id indefinitely; a failed + * upgrade is a connect error instead, and that is the only outcome that ends the + * loop. These cases pin the handshake result and its client-visible consequence. + */ +class FakeUpstreamSocket { + private readonly listeners = new Map void>>(); + closed = false; + closeCalls = 0; + closeMode: "closed" | "closing" | "closing-then-close" = "closed"; + readyState = WebSocket.CONNECTING; + + addEventListener(type: string, listener: (event: { code?: number; data?: unknown; reason?: string }) => void): void { + const bucket = this.listeners.get(type) ?? []; + bucket.push(listener); + this.listeners.set(type, bucket); + } + + emit(type: string, event: { code?: number; data?: unknown; reason?: string } = {}): void { + if (type === "open") this.readyState = WebSocket.OPEN; + if (type === "close") this.readyState = WebSocket.CLOSED; + for (const listener of this.listeners.get(type) ?? []) listener(event); + } + + close(code = 1000, reason = ""): void { + this.closed = true; + this.closeCalls += 1; + if (this.closeMode === "closing") { + this.readyState = WebSocket.CLOSING; + return; + } + if (this.closeMode === "closing-then-close") this.readyState = WebSocket.CLOSING; + this.emit("close", { code, reason }); + } +} + +function fakeSidebandClient( + upstream: FakeUpstreamSocket, + handoff: { + failure(): { status: number; code: string; message: string; closeCode?: number; closeReason?: string } | undefined; + take(): { ok: true; frames: Array } | { + ok: false; + failure: { status: number; code: string; message: string; closeCode?: number; closeReason?: string }; + }; + }, + send: (frame: string | Buffer) => void = () => {}, +) { + let releases = 0; + const ws = { + data: { + kind: "live-sideband" as const, + liveUpstream: upstream as unknown as WebSocket, + liveUpstreamHandoff: handoff, + liveOpened: true, + liveTurnAdmissionLease: { + release: () => { releases += 1; }, + }, + }, + readyState: WebSocket.OPEN, + close: () => {}, + send, + }; + return { ws, releases: () => releases }; +} + +describe("attachLiveSidebandUpstream ownership", () => { + test("transfers the actual captured preamble before subsequent live frames", async () => { + const upstream = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/fixture", {}, () => upstream as unknown as WebSocket); + upstream.emit("open"); + upstream.emit("message", { data: "first" }); + upstream.emit("message", { data: new Uint8Array([2]) }); + const result = await pending; + if (!result.ok) throw new Error("expected open handshake"); + const sent: Array = []; + const client = fakeSidebandClient(upstream, result.handoff, frame => { sent.push(frame); }); + attachLiveSidebandUpstream(client.ws as never); + upstream.emit("message", { data: "third" }); + expect(sent).toEqual(["first", Buffer.from([2]), "third"]); + upstream.emit("close", { code: 1000 }); + expect(client.releases()).toBe(1); + }); + + test("retains admission through a failed takeover until a CLOSING upstream actually closes", async () => { + const upstream = new FakeUpstreamSocket(); + upstream.readyState = WebSocket.OPEN; + upstream.closeMode = "closing"; + const client = fakeSidebandClient(upstream, { + failure: () => undefined, + take: () => ({ + ok: false, + failure: { status: 502, code: "upstream_error", message: "closed", closeCode: 1008 }, + }), + }); + + attachLiveSidebandUpstream(client.ws as never); + + expect(upstream.readyState).toBe(WebSocket.CLOSING); + expect(client.releases()).toBe(0); + await Bun.sleep(1_100); + expect(upstream.closeCalls).toBe(2); + expect(client.releases()).toBe(0); + upstream.emit("close", { code: 1008, reason: "call ended" }); + expect(client.releases()).toBe(1); + upstream.emit("close", { code: 1008, reason: "duplicate close" }); + expect(client.releases()).toBe(1); + }); + + test("registers close ownership before forwarding a pre-opened preamble", () => { + const upstream = new FakeUpstreamSocket(); + upstream.readyState = WebSocket.OPEN; + upstream.closeMode = "closing-then-close"; + const client = fakeSidebandClient( + upstream, + { + failure: () => undefined, + take: () => ({ ok: true, frames: ["session.created"] }), + }, + () => { throw new Error("downstream send failed"); }, + ); + + attachLiveSidebandUpstream(client.ws as never); + + expect(upstream.closeCalls).toBe(1); + expect(upstream.readyState).toBe(WebSocket.CLOSED); + expect(client.releases()).toBe(1); + }); +}); + +describe("openLiveSidebandUpstream", () => { + test("drains the preamble captured before the client socket exists", async () => { + const socket = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + // The session preamble arrives the moment the upstream opens, before the client. + socket.emit("message", { data: "session.created" }); + socket.emit("message", { data: new Uint8Array([1, 2, 3]) }); + socket.emit("open", {}); + + const result = await pending; + expect(result.ok).toBe(true); + if (!result.ok) throw new Error("expected an open upstream"); + expect(result.socket).toBe(socket); + const takeover = result.handoff.take(); + expect(takeover.ok).toBe(true); + if (!takeover.ok) throw new Error("expected a successful handoff"); + const drained = takeover.frames; + expect(drained[0]).toBe("session.created"); + expect(Buffer.isBuffer(drained[1])).toBe(true); + expect(drained[1]).toEqual(Buffer.from([1, 2, 3])); + // Drain is one-shot: the relay owns capture from here on. + expect(result.handoff.take()).toEqual({ ok: true, frames: [] }); + socket.emit("message", { data: "after-drain" }); + expect(result.handoff.take()).toEqual({ ok: true, frames: [] }); + }); + + test("fails explicitly before copying an aggregate preamble overflow", async () => { + const socket = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + const retained = new Uint8Array(1024 * 1024); + socket.emit("message", { data: retained }); + const rejectedView = new Uint8Array(retained.buffer, 0, 1); + socket.emit("message", { data: rejectedView }); + + const result = await pending; + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected an overflow failure"); + expect(result.code).toBe("upstream_overflow"); + expect(socket.closed).toBe(true); + }); + + test("fails explicitly when the preamble frame-count limit is exceeded", async () => { + const socket = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + for (let index = 0; index < 33; index += 1) socket.emit("message", { data: String(index) }); + + const result = await pending; + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected an overflow failure"); + expect(result.code).toBe("upstream_overflow"); + expect(socket.closed).toBe(true); + }); + + test("preserves an open-then-close terminal event until relay handoff", async () => { + const socket = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + socket.emit("open", {}); + socket.emit("close", { code: 1008 }); + + const result = await pending; + expect(result.ok).toBe(true); + if (!result.ok) throw new Error("expected the completed opening handshake"); + const takeover = result.handoff.take(); + expect(takeover.ok).toBe(false); + if (takeover.ok) throw new Error("expected the terminal handoff"); + expect(takeover.failure.closeCode).toBe(1008); + }); + + test("reports failure when the upstream rejects the handshake", async () => { + const socket = new FakeUpstreamSocket(); + socket.closeMode = "closing"; + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + socket.emit("error", {}); + + const result = await pending; + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected a failed handshake"); + expect(result.status).toBe(502); + expect(result.socket).toBe(socket); + expect(socket.readyState).toBe(WebSocket.CLOSING); + }); + + test("reports failure when the upstream closes before opening", async () => { + const socket = new FakeUpstreamSocket(); + const pending = openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 1_000); + socket.emit("close", { code: 1006 }); + + const result = await pending; + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected a failed handshake"); + expect(result.status).toBe(502); + }); + + test("cancels a pending join and closes its upstream socket", async () => { + const socket = new FakeUpstreamSocket(); + const controller = new AbortController(); + const pending = openLiveSidebandUpstream( + "ws://upstream/v1/live/x", + {}, + () => socket as unknown as WebSocket, + 1_000, + controller.signal, + ); + controller.abort(); + + const result = await pending; + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected a cancelled handshake"); + expect(result.code).toBe("request_cancelled"); + expect(socket.closed).toBe(true); + expect(result.socket).toBe(socket); + }); + + test("times out and drops the socket when the upstream never opens", async () => { + const socket = new FakeUpstreamSocket(); + const result = await openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => socket as unknown as WebSocket, 20); + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected a timeout"); + expect(result.status).toBe(504); + expect(socket.closed).toBe(true); + }); + + test("reports failure when the upstream socket cannot be constructed", async () => { + const result = await openLiveSidebandUpstream("ws://upstream/v1/live/x", {}, () => { + throw new Error("connect refused"); + }); + expect(result.ok).toBe(false); + if (result.ok) throw new Error("expected a failed handshake"); + expect(result.status).toBe(502); + expect(result.socket).toBeUndefined(); + }); +}); + +test("a failed pre-upgrade handshake retains admission until its CLOSING upstream closes", async () => { + saveConfig(forwardConfig()); + const upstream = new FakeUpstreamSocket(); + upstream.closeMode = "closing"; + const server = startServer(0, { + liveSidebandWebSocketFactory: () => { + queueMicrotask(() => upstream.emit("error", {})); + return upstream as unknown as WebSocket; + }, + }); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + try { + const wsUrl = new URL("/v1/realtime?call_id=rtc_failed_handshake_closing", server.url); + wsUrl.protocol = "ws:"; + const client = new WebSocket(wsUrl.toString(), { + headers: { + authorization: `Bearer ${DIRECT_CHATGPT_TOKEN}`, + "chatgpt-account-id": "acct-123", + "openai-alpha": "quicksilver=v2", + "x-session-id": "rts_failed_handshake_closing", + }, + } as unknown as string[]); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("client never observed failed upgrade")), 5_000); + const settle = (): void => { + clearTimeout(timer); + resolve(); + }; + client.addEventListener("error", settle, { once: true }); + client.addEventListener("close", settle, { once: true }); + }); + + expect(upstream.readyState).toBe(WebSocket.CLOSING); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore + 1); + upstream.emit("close", { code: 1006, reason: "closed after handshake failure" }); + await Bun.sleep(0); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + upstream.emit("close", { code: 1006, reason: "duplicate close" }); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + } finally { + await server.stop(true); + } +}, { timeout: 10_000 }); + +test("a failed pre-upgrade handoff retains admission until its CLOSING upstream closes", async () => { + saveConfig(forwardConfig()); + const upstream = new FakeUpstreamSocket(); + upstream.closeMode = "closing"; + const server = startServer(0, { + liveSidebandWebSocketFactory: () => { + queueMicrotask(() => { + upstream.emit("open", {}); + upstream.emit("error", {}); + }); + return upstream as unknown as WebSocket; + }, + }); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + try { + const wsUrl = new URL("/v1/realtime?call_id=rtc_failed_handoff_closing", server.url); + wsUrl.protocol = "ws:"; + const client = new WebSocket(wsUrl.toString(), { + headers: { + authorization: `Bearer ${DIRECT_CHATGPT_TOKEN}`, + "chatgpt-account-id": "acct-123", + "openai-alpha": "quicksilver=v2", + "x-session-id": "rts_failed_handoff_closing", + }, + } as unknown as string[]); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("client never observed failed handoff")), 5_000); + const settle = (): void => { + clearTimeout(timer); + resolve(); + }; + client.addEventListener("error", settle, { once: true }); + client.addEventListener("close", settle, { once: true }); + }); + + expect(upstream.readyState).toBe(WebSocket.CLOSING); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore + 1); + upstream.emit("close", { code: 1008, reason: "closed after failed handoff" }); + await Bun.sleep(0); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + upstream.emit("close", { code: 1008, reason: "duplicate close" }); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + } finally { + await server.stop(true); + } +}, { timeout: 10_000 }); + +test("a sideband join whose upstream handshake fails never opens the client socket", async () => { + // An upstream that refuses the upgrade: the shape OpenAI returns for a call id it + // no longer knows (`404 call_id_not_found`). + const upstream = Bun.serve({ + port: 0, + fetch(req) { + if (req.headers.get("upgrade")?.toLowerCase() === "websocket") { + return new Response(JSON.stringify({ error: { code: "call_id_not_found" } }), { + status: 404, + headers: { "content-type": "application/json" }, + }); + } + return new Response("not found", { status: 404 }); + }, + }); + + saveConfig(forwardConfig()); + + const RealWebSocket = globalThis.WebSocket; + const upstreamPort = upstream.port; + globalThis.WebSocket = class extends RealWebSocket { + constructor(url: string | URL, protocols?: string | string[] | Record) { + const parsed = new URL(String(url)); + const target = parsed.hostname === "api.openai.com" + ? `ws://127.0.0.1:${upstreamPort}${parsed.pathname}${parsed.search}` + : String(url); + super(target, protocols as string[]); + } + } as typeof WebSocket; + + const server = startServer(0); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + try { + const wsUrl = new URL(`/v1/realtime?call_id=rtc_dead_call`, server.url); + wsUrl.protocol = "ws:"; + const events: string[] = []; + const client = new RealWebSocket(wsUrl.toString(), { + headers: { + authorization: `Bearer ${DIRECT_CHATGPT_TOKEN}`, + "chatgpt-account-id": "acct-123", + "openai-alpha": "quicksilver=v2", + "x-session-id": "rts_dead", + }, + } as unknown as string[]); + + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("client never settled")), 15_000); + const settle = (): void => { + clearTimeout(timer); + resolve(); + }; + client.addEventListener("open", () => { + events.push("open"); + settle(); + }); + client.addEventListener("error", () => { + events.push("error"); + settle(); + }); + client.addEventListener("close", () => { + events.push("close"); + settle(); + }); + }); + + // The relay never became live, so the client must not have been told it did. + expect(events).not.toContain("open"); + expect(events.length).toBeGreaterThan(0); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + } finally { + globalThis.WebSocket = RealWebSocket; + await server.stop(true); + await upstream.stop(true); + } +}, { timeout: 20_000 }); + +test("an upstream that opens then closes before relay attachment refuses the client and releases admission", async () => { + saveConfig(forwardConfig()); + const upstream = new FakeUpstreamSocket(); + const server = startServer(0, { + liveSidebandWebSocketFactory: () => { + queueMicrotask(() => { + upstream.emit("open", {}); + upstream.emit("close", { code: 1008, reason: "call ended" }); + }); + return upstream as unknown as WebSocket; + }, + }); + const activeTurnsBefore = activeRegistryMetrics().activeTurns.active; + try { + const wsUrl = new URL("/v1/realtime?call_id=rtc_closed_handoff", server.url); + wsUrl.protocol = "ws:"; + const events: string[] = []; + const client = new WebSocket(wsUrl.toString(), { + headers: { + authorization: `Bearer ${DIRECT_CHATGPT_TOKEN}`, + "chatgpt-account-id": "acct-123", + "openai-alpha": "quicksilver=v2", + "x-session-id": "rts_closed_handoff", + }, + } as unknown as string[]); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("client never settled")), 5_000); + const settle = (): void => { + clearTimeout(timer); + resolve(); + }; + client.addEventListener("open", () => { + events.push("open"); + settle(); + }); + client.addEventListener("error", () => { + events.push("error"); + settle(); + }); + client.addEventListener("close", () => { + events.push("close"); + settle(); + }); + }); + + expect(events).not.toContain("open"); + expect(events.length).toBeGreaterThan(0); + expect(activeRegistryMetrics().activeTurns.active).toBe(activeTurnsBefore); + } finally { + await server.stop(true); + } +}, { timeout: 10_000 }); From eca7ce939046fe44c3985220528277cb195747fb Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 16:11:22 +0900 Subject: [PATCH 07/10] fix(cursor): preserve isolated recovery state and active cap retention --- .../content/docs/reference/proxy-formats.md | 5 +- src/adapters/cursor.ts | 1 + src/adapters/cursor/thread-continuity.ts | 8 ++- structure/providers/cursor.md | 2 +- tests/providers/cursor/cursor-adapter.test.ts | 54 +++++++++++++++++++ .../cursor-continuity-retention.test.ts | 44 ++++++++++++++- 6 files changed, 109 insertions(+), 5 deletions(-) diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 865cb63834..a1f045e981 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -29,8 +29,9 @@ Credential-bearing model, image, video, and search requests do not automatically Cursor's first bare context overflow is surfaced to the client. Later eligible requests with a stable client thread may recover with up to three conversation remints per retained scope. The in-memory allowance expires after one idle hour, eviction, or restart. Requests -without a stable thread, tool-result resumes, partial output, compaction and quota errors do -not use this recovery. This does not infer whether a task is making progress. +without a stable thread, isolated helpers, tool-result resumes, partial output, compaction +and quota errors do not use this recovery. Continued eligible overflows keep the existing +allowance active even after it is exhausted; they do not replenish it. This does not infer whether a task is making progress. ## Endpoint overview diff --git a/src/adapters/cursor.ts b/src/adapters/cursor.ts index a240ae1982..91e823be9a 100644 --- a/src/adapters/cursor.ts +++ b/src/adapters/cursor.ts @@ -468,6 +468,7 @@ export function createCursorAdapter(provider: OcxProviderConfig, deps: CursorAda !lastRawIsToolResult && !emittedOutput && !replayUnsafe + && _parsed._cursorIsolateConversation !== true && request.contextUsageStoreCheckpoints !== false && !incoming.abortSignal?.aborted; const overflowScopeKey = cursorOverflowRemintScopeKey( diff --git a/src/adapters/cursor/thread-continuity.ts b/src/adapters/cursor/thread-continuity.ts index aa3c3dac32..fc57099f58 100644 --- a/src/adapters/cursor/thread-continuity.ts +++ b/src/adapters/cursor/thread-continuity.ts @@ -128,8 +128,14 @@ export function markCursorOverflowSurfaced(scopeKey: string): void { } export function shouldSkipCursorOverflowRemint(scopeKey: string): boolean { - pruneOverflowRemints(now()); + const at = now(); + pruneOverflowRemints(at); const entry = overflowRemintByScope.get(scopeKey); + if (entry) { + entry.updatedAt = at; + overflowRemintByScope.delete(scopeKey); + overflowRemintByScope.set(scopeKey, entry); + } return entry?.skip === true || (entry?.remintCount ?? 0) >= CURSOR_OVERFLOW_REMINT_MAX; } diff --git a/structure/providers/cursor.md b/structure/providers/cursor.md index 3d734a33ed..f062ee0d13 100644 --- a/structure/providers/cursor.md +++ b/structure/providers/cursor.md @@ -85,4 +85,4 @@ Namespaced tools do not acquire bare-shell behavior. Regression coverage lives i ## Overflow remint boundary -`src/adapters/cursor.ts` surfaces the first bare context overflow before attempting conversation remint on later eligible requests. `cursorClientThreadOwner` recognizes both client thread aliases; `src/adapters/cursor/thread-continuity.ts` limits recovery to three remints per retained identity-scoped owner, with a one-hour idle TTL and 2,048-entry bound. Conversation-only requests have no stable owner and do not automatically remint. Quota/rate errors, tool-result resumes, partial output, local side effects and compaction remain fail-closed. Retention expiry, eviction or process restart resets the in-memory allowance; this is not a persistent lifetime cap or semantic-progress policy. +`src/adapters/cursor.ts` surfaces the first bare context overflow before attempting conversation remint on later eligible requests. `cursorClientThreadOwner` recognizes both client thread aliases; `src/adapters/cursor/thread-continuity.ts` limits recovery to three remints per retained identity-scoped owner, with a one-hour idle TTL and 2,048-entry bound. Conversation-only requests have no stable owner and do not automatically remint. Quota/rate errors, tool-result resumes, partial output, local side effects, isolated helper/shadow requests and compaction remain fail-closed. Isolated requests neither consume the parent allowance nor invalidate its checkpoint. Eligible overflow checks refresh existing retention timestamps and LRU position even after the cap is exhausted, without allocating absent scopes. Retention expiry, eviction or process restart resets the in-memory allowance; this is not a persistent lifetime cap or semantic-progress policy. diff --git a/tests/providers/cursor/cursor-adapter.test.ts b/tests/providers/cursor/cursor-adapter.test.ts index 1f42b697dc..9a03be37cd 100644 --- a/tests/providers/cursor/cursor-adapter.test.ts +++ b/tests/providers/cursor/cursor-adapter.test.ts @@ -1057,6 +1057,60 @@ describe("Cursor overflow conversation remint", () => { expect(seen).toHaveLength(1); }); + test("isolated non-compaction helpers preserve parent remint allowance and checkpoint", async () => { + clearCursorOverflowRemintForTests(); + clearCursorThreadContinuityForTests(); + clearCursorCheckpointsForTests(); + let attempts = 0; + const adapter = createCursorAdapter({ ...provider, apiKey: "cursor-token" }, { + createTransport: () => ({ + async *run() { + attempts += 1; + throw bareOverflowError(); + }, + writeClient() {}, + }), + }); + try { + const owner = "overflow-isolated-helper"; + await adapter.runTurn?.(overflowTurnBody(owner), { headers: new Headers() }, () => {}); + expect(attempts).toBe(1); + const parentRef = commitCursorCheckpoint({ + conversationId: "cursor_parent_overflow", + identityScope: "acct-overflow-remint", + modelId: "default", + checkpointBytes: toBinary(ConversationStateStructureSchema, create(ConversationStateStructureSchema, { + pendingToolCalls: ["overflow-isolation-fixture"], + })), + coveredMessageCount: 1, + }); + expect(parentRef).toBeDefined(); + const helper = overflowTurnBody(owner); + helper._cursorIsolateConversation = true; + helper._cursorConversationId = "cursor_parent_overflow"; + helper._providerContinuation = { + cursor: { conversationId: "cursor_parent_overflow", checkpointUsable: true, checkpointRef: parentRef }, + }; + expect(helper._compactionRequest).toBeUndefined(); + attempts = 0; + const events: AdapterEvent[] = []; + await adapter.runTurn?.(helper, { headers: new Headers() }, event => events.push(event)); + expect(attempts).toBe(1); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ type: "error", message: expect.stringContaining("Cursor context limit exceeded") }); + expect(getCursorCheckpoint(parentRef)?.ref).toBe(parentRef); + expect(lookupCursorThreadConversation(owner, "acct-overflow-remint")).toBeUndefined(); + + attempts = 0; + await adapter.runTurn?.(overflowTurnBody(owner), { headers: new Headers() }, () => {}); + expect(attempts).toBe(4); + } finally { + clearCursorOverflowRemintForTests(); + clearCursorThreadContinuityForTests(); + clearCursorCheckpointsForTests(); + } + }); + test("does not overflow-remint after non-heartbeat output was emitted", async () => { clearCursorOverflowRemintForTests(); let attempts = 0; diff --git a/tests/providers/cursor/cursor-continuity-retention.test.ts b/tests/providers/cursor/cursor-continuity-retention.test.ts index 2d3f833e8f..ea441e8ad4 100644 --- a/tests/providers/cursor/cursor-continuity-retention.test.ts +++ b/tests/providers/cursor/cursor-continuity-retention.test.ts @@ -1,14 +1,56 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { clearCursorOverflowRemintForTests, CURSOR_OVERFLOW_REMINT_MAX_ENTRIES, + CURSOR_OVERFLOW_REMINT_TTL_MS, cursorOverflowRemintCountForTests, markCursorOverflowSurfaced, + recordCursorOverflowRemint, shouldSkipCursorOverflowRemint, shouldSurfaceCursorOverflowFirst, } from "../../../src/adapters/cursor/thread-continuity"; describe("Cursor overflow remint retention", () => { + test("capped activity refreshes idle expiry without replenishing the allowance", () => { + clearCursorOverflowRemintForTests(); + let at = 1_000; + const clock = spyOn(Date, "now").mockImplementation(() => at); + try { + markCursorOverflowSurfaced("active"); + for (let attempt = 0; attempt < 3; attempt++) expect(recordCursorOverflowRemint("active")).toBe(true); + for (let interval = 0; interval < 8; interval++) { + at += CURSOR_OVERFLOW_REMINT_TTL_MS / 4; + expect(shouldSkipCursorOverflowRemint("active")).toBe(true); + expect(shouldSurfaceCursorOverflowFirst("active")).toBe(false); + } + at += CURSOR_OVERFLOW_REMINT_TTL_MS + 1; + expect(shouldSkipCursorOverflowRemint("active")).toBe(false); + expect(shouldSurfaceCursorOverflowFirst("active")).toBe(true); + expect(cursorOverflowRemintCountForTests()).toBe(0); + } finally { + clock.mockRestore(); + clearCursorOverflowRemintForTests(); + } + }); + + test("capped activity moves an existing scope behind older eviction candidates", () => { + clearCursorOverflowRemintForTests(); + try { + markCursorOverflowSurfaced("active"); + for (let attempt = 0; attempt < 3; attempt++) expect(recordCursorOverflowRemint("active")).toBe(true); + for (let index = 0; index < CURSOR_OVERFLOW_REMINT_MAX_ENTRIES - 1; index++) { + markCursorOverflowSurfaced(`other-${index}`); + } + expect(shouldSkipCursorOverflowRemint("active")).toBe(true); + markCursorOverflowSurfaced("new"); + expect(shouldSurfaceCursorOverflowFirst("other-0")).toBe(true); + expect(shouldSkipCursorOverflowRemint("active")).toBe(true); + expect(cursorOverflowRemintCountForTests()).toBe(CURSOR_OVERFLOW_REMINT_MAX_ENTRIES); + } finally { + clearCursorOverflowRemintForTests(); + } + }); + test("bounds per-scope state", () => { clearCursorOverflowRemintForTests(); for (let index = 0; index < CURSOR_OVERFLOW_REMINT_MAX_ENTRIES + 20; index++) { From 3ad908f16022d6b8464ed157af7c2cf12607448f Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 16:30:26 +0900 Subject: [PATCH 08/10] docs: synchronize live sideband handshake ownership --- structure/adapters/registry.md | 2 ++ structure/catalog.md | 2 ++ structure/clients/claude-desktop.md | 2 ++ structure/data-planes/images.md | 2 ++ structure/data-planes/inbound-compat.md | 2 ++ structure/gui-and-management-api.md | 2 ++ structure/ops/service-and-sidecars.md | 2 ++ structure/providers/xai-grok.md | 2 ++ structure/subagents.md | 2 ++ structure/transports/inventory.md | 2 ++ structure/transports/responses.md | 2 ++ structure/transports/streaming-health.md | 2 ++ 12 files changed, 24 insertions(+) diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index a4dc21adbf..5d9e8322f2 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -66,3 +66,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/catalog.md b/structure/catalog.md index 0ba4acca3e..9326d6f83d 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -278,3 +278,5 @@ see [Combo editor routing quota](gui-and-management-api.md#combo-editor-routing- Claude replay carries [Go conversation affinity](data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 2914823958..115df2ed4d 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -91,3 +91,5 @@ The explicit sync coordinator also accepts Cline CLI as a separate file integrat `claudeCode.stabilizePromptCache` is a default-off operator setting for [translated instruction stabilization](../data-planes/inbound-compat.md#opt-in-claude-instruction-stabilization). Config JSON preserves the boolean; only literal true activates the role-changing transform. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index 25646c7de4..d106b16a23 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -79,3 +79,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 2d17c11875..03e77103d4 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -128,3 +128,5 @@ changes prompt roles, not conversation identity, and cannot guarantee upstream c Instruction notice extraction scans fence ranges once and walks original lines backwards with a decreasing cursor. It accepts exactly one ASCII space inside the token notice, preserves unmatched prefix bytes, and does not repeatedly scan or copy shrinking prompt prefixes. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 73090d646e..bcee0def75 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -543,3 +543,5 @@ integration IO adapter. Its snapshot fingerprint cannot be checked against provi [the integration contract](clients/integrations.md#cline-paired-files) defines recovery. The existing dashboard file-client maps include Cline CLI and reuse its committed color mark. The export panel labels its download as a settings/catalog bundle; all locales explain that Undo restores both original files. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index 39dc9a82da..7dff0ce07b 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -142,3 +142,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index 5b149ac6a2..8e998d00f2 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -65,3 +65,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/subagents.md b/structure/subagents.md index f190aab084..fc793886c9 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -214,3 +214,5 @@ see [Combo editor routing quota](gui-and-management-api.md#combo-editor-routing- Claude replay carries [Go conversation affinity](data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index b2fc3b3fae..2d2f1d36e3 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -70,3 +70,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 2321d78dd6..52adc8f7e5 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -523,3 +523,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index 68093843ea..d216e4a059 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -199,3 +199,5 @@ see [Combo editor routing quota](../gui-and-management-api.md#combo-editor-routi Claude replay carries [Go conversation affinity](../data-planes/inbound-compat.md#claude-affinity-at-final-go-dispatch) privately to final dispatch; preliminary route selection does not inject Go-only headers. + +Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. From efb3936bdad2c172537fe3aa2e93cdda90926b9d Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 16:41:12 +0900 Subject: [PATCH 09/10] fix(search): reject malformed truncated calls before replay --- src/web-search/loop.ts | 3 +- structure/runtime.md | 2 +- tests/web-search/web-search.test.ts | 43 +++++++++++++++++++++++++++-- 3 files changed, 44 insertions(+), 4 deletions(-) diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 8feceb4c27..849eece2da 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -875,7 +875,8 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise event.type === "done"); - if (terminalEvent?.type === "done" && isTruncatedStopReason(terminalEvent.stopReason)) { + if (terminalEvent?.type === "done" && !split.hasMalformedToolCall + && isTruncatedStopReason(terminalEvent.stopReason)) { // A provider refusal or truncation is authoritative, even without text. // Preserve it once; neither an empty-answer retry nor a generic 502 applies. yield* replay(split.passthrough.slice(split.streamedPassthroughCount)); diff --git a/structure/runtime.md b/structure/runtime.md index 60da390f99..69c7c05a66 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -197,7 +197,7 @@ claims stored main, after terminal vision, routed vision and search exclusions. ### Empty forced search answers -`src/web-search/loop.ts` makes at most one extra answer attempt after a clean forced-answer terminal with no visible output or tool call. The recovery has no tools and reuses gathered search results. Malformed calls fail, and recognized refusal/truncation terminals pass through unchanged, including empty or partial answers. The extra generation may incur provider usage. +`src/web-search/loop.ts` makes at most one extra answer attempt after a clean forced-answer terminal with no visible output or tool call. The recovery has no tools and reuses gathered search results. Malformed calls fail before refusal/truncation passthrough, and well-formed recognized refusal/truncation terminals pass through unchanged, including empty or partial answers. The extra generation may incur provider usage. ## Scoped provider quota for Combo selection diff --git a/tests/web-search/web-search.test.ts b/tests/web-search/web-search.test.ts index 9bf7c02444..b2995a9e12 100644 --- a/tests/web-search/web-search.test.ts +++ b/tests/web-search/web-search.test.ts @@ -183,8 +183,23 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = test("the recovery pass asks for text with every tool removed", async () => { const seen: OcxParsedRequest[] = []; + let sidecarCalls = 0; + const evidence = "Distinctive gathered result: fixture-42"; + globalThis.fetch = (async (input, init) => { + sidecarCalls++; + expect(String(input)).toBe("https://chatgpt.test/v1/responses"); + const body = JSON.parse(String(init?.body)); + expect(body.input[0].content[0].text).toBe("recovery fixture query"); + return new Response( + `event: response.output_text.delta\ndata: ${JSON.stringify({ type: "response.output_text.delta", delta: evidence })}\n\n` + + 'event: response.completed\ndata: {"type":"response.completed"}\n\n', + { headers: { "Content-Type": "text/event-stream" } }, + ); + }) as typeof fetch; + const actualSearch: AdapterEvent[] = webSearchFirstPass.map(event => event.type === "tool_call_delta" + ? { ...event, arguments: JSON.stringify({ query: "recovery fixture query" }) } : event); await drivePasses([ - webSearchFirstPass, + actualSearch, [{ type: "done" }], [{ type: "text_delta", text: "recovered answer" }, { type: "done" }], ], seen); @@ -194,7 +209,11 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = expect(recovery.options.toolChoice).toBe("none"); expect(recovery.context.tools).toEqual([]); // The results gathered by the search reach the recovery turn as a tool result ... - expect(recovery.context.messages.filter(message => message.role === "toolResult")).toHaveLength(1); + const results = recovery.context.messages.filter(message => message.role === "toolResult"); + expect(results).toHaveLength(1); + expect(JSON.stringify(results[0])).toContain(evidence); + expect(results).toEqual(seen[1]!.context.messages.filter(message => message.role === "toolResult")); + expect(sidecarCalls).toBe(1); // ... and the recovery turn carries the developer nudge that asks for the missing text. expect(recovery.context.messages.some(message => message.role === "developer" && String(message.content).includes("no tools are available"))) @@ -210,6 +229,26 @@ describe("issue #1001 — forced-answer passes must produce usable output", () = expect(seen[2]!.options.toolChoice).toBe("none"); }); + for (const liveOutput of [false, true]) { + for (const stopReason of ["max_tokens", "content_filter"]) { + test(`malformed calls fail before ${stopReason} passthrough, live=${liveOutput}`, async () => { + const seen: OcxParsedRequest[] = []; + const frames = await drivePasses([webSearchFirstPass, [ + { type: "tool_call_start", id: "partial", name: "fixture" }, + { type: "tool_call_delta", arguments: '{"partial":' }, + { type: "tool_call_start", id: "closed", name: "fixture" }, + { type: "tool_call_delta", arguments: "{}" }, + { type: "tool_call_end" }, + { type: "done", stopReason }, + ]], seen, true, liveOutput); + expect(seen).toHaveLength(2); + expect(frames.some(frame => frame.event === "response.failed")).toBe(true); + expect(frames.some(frame => frame.event === "response.function_call_arguments.done")).toBe(false); + expect(frames.some(frame => frame.event === "response.completed" || frame.event === "response.incomplete")).toBe(false); + }); + } + } + for (const [stopReason, reason] of [["refusal", "content_filter"], ["content_filter", "content_filter"], ["max_tokens", "max_output_tokens"], ["length", "max_output_tokens"]]) { for (const partial of [false, true]) { test(`${stopReason} partial=${partial} stays authoritative without a retry`, async () => { From 57b3057c710a38613eaa4f81e32a17855619fc9a Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 17:31:58 +0900 Subject: [PATCH 10/10] docs: describe cancelled live sideband handshakes --- docs-site/src/content/docs/reference/proxy-formats.md | 2 +- structure/runtime.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index dedd5b83ec..ce15acf90c 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -28,7 +28,7 @@ Credential-bearing model, image, video, and search requests do not automatically The proxy completes the upstream live sideband handshake before accepting the client WebSocket. An upstream rejection fails the upgrade with 502; a ten-second handshake timeout -returns 504. Bun does not expose the exact upstream handshake status, so an upstream 404/410 +returns 504, and client cancellation returns 499. Bun does not expose the exact upstream handshake status, so an upstream 404/410 cannot currently be forwarded precisely. A successful connection preserves the initial session frames in order. This handshake policy is separate from the Responses WebSocket transport. diff --git a/structure/runtime.md b/structure/runtime.md index 3e2401ad0f..68770cb2de 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -215,7 +215,7 @@ see [Combo editor routing quota](gui-and-management-api.md#combo-editor-routing- ## Live sideband handshake -`src/server/index.ts` establishes the authorized upstream live sideband before accepting the client WebSocket upgrade. `openLiveSidebandUpstream` bounds the handshake to ten seconds and retains at most 32 frames and 1 MiB of preamble within the frame limit. `src/server/ws-bridge.ts` defines the runtime handoff carrying captured frames or terminal state. Failed handshakes return 502/504; exact upstream 404/410 status is unavailable from Bun's client WebSocket. Admission ownership lasts until upstream close/CLOSED, including failed upgrades and failed attachment. The ordinary Responses WebSocket exchange remains separate. +`src/server/index.ts` establishes the authorized upstream live sideband before accepting the client WebSocket upgrade. `openLiveSidebandUpstream` bounds the handshake to ten seconds and retains at most 32 frames and 1 MiB of preamble within the frame limit. `src/server/ws-bridge.ts` defines the runtime handoff carrying captured frames or terminal state. Failed handshakes return 502/504 and client cancellation returns 499; exact upstream 404/410 status is unavailable from Bun's client WebSocket. Admission ownership lasts until upstream close/CLOSED, including failed upgrades and failed attachment. The ordinary Responses WebSocket exchange remains separate. ## Paginated history writer boundary