From 8369eb949fa6f01bfbe846caf80d400f0f8af951 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 6 Aug 2026 09:15:55 +0200 Subject: [PATCH 1/2] fix(providers): normalize GitHub Copilot Responses streams Adopted from PR #1111 for issue #1110. --- src/server/github-copilot-responses-repair.ts | 331 ++++++++++++++++++ src/server/relay.ts | 28 ++ src/server/responses/core.ts | 36 +- src/server/sse-payload-rewrite.ts | 39 ++- tests/github-copilot-sse-rewrite.test.ts | 254 ++++++++++++++ tests/github-copilot-stream-contract.test.ts | 248 +++++++++++++ tests/passthrough-abort.test.ts | 11 +- tests/sse-inspector-bounds.test.ts | 19 + tests/sse-payload-rewrite.test.ts | 34 ++ 9 files changed, 966 insertions(+), 34 deletions(-) create mode 100644 src/server/github-copilot-responses-repair.ts create mode 100644 tests/github-copilot-sse-rewrite.test.ts create mode 100644 tests/github-copilot-stream-contract.test.ts diff --git a/src/server/github-copilot-responses-repair.ts b/src/server/github-copilot-responses-repair.ts new file mode 100644 index 0000000000..ec1088ee6a --- /dev/null +++ b/src/server/github-copilot-responses-repair.ts @@ -0,0 +1,331 @@ +/** + * Client-facing repair for GitHub Copilot's `vscode-chat` Responses stream. + * + * Copilot can rotate opaque response/item ids between lifecycle events, expose + * provider-private reasoning ciphertext, and attach per-delta obfuscation + * padding while the authoritative complete tool input arrives on `.done`. + * Stock Responses clients cannot reconcile that dialect. This stateful block + * rewrite emits one coherent Responses SSE stream while the inspection branch + * keeps the raw upstream snapshot for provider replay. + */ +import { + TranslatorBudgetExceededError, + type TranslatorBudget, +} from "../lib/translator-budget"; +import { + replaceSseDataPayload, + sseDataPayload, + type SseBlockRewrite, +} from "./sse-payload-rewrite"; + +const MAX_TRACKED_ITEMS = 256; +const MAX_TRACKED_DEFERRED_CALLS = 256; +const RETAINED_STATE_LIMIT_BYTES = 256 * 1024; + +type JsonRecord = Record; + +type TrackedItem = { + id: string; + type?: string; + encryptedReasoning: boolean; +}; + +type DeferredCallKind = "function" | "custom"; + +function isRecord(value: unknown): value is JsonRecord { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function outputIndexOf(value: unknown): number | undefined { + return typeof value === "number" && Number.isInteger(value) && value >= 0 + ? value + : undefined; +} + +function byteLength(value: unknown): number { + return Buffer.byteLength(JSON.stringify(value), "utf8"); +} + +function blockNewline(block: string): "\r\n" | "\n" { + return block.includes("\r\n") ? "\r\n" : "\n"; +} + +/** Keep a named SSE block's `event:` field aligned with its JSON `type`. */ +function withEventName(block: string, eventName: string): string { + const newline = blockNewline(block); + const lines = block.split(/\r?\n/); + let replaced = false; + const rewritten = lines.flatMap((line) => { + if (!line.startsWith("event:")) return [line]; + if (replaced) return []; + replaced = true; + return [`event: ${eventName}`]; + }); + if (!replaced) { + const dataIndex = rewritten.findIndex(line => line.startsWith("data:")); + rewritten.splice(dataIndex >= 0 ? dataIndex : 0, 0, `event: ${eventName}`); + } + return rewritten.join(newline); +} + +function rewriteJsonBlock(block: string, event: JsonRecord): string { + const type = typeof event.type === "string" ? event.type : "message"; + return withEventName(replaceSseDataPayload(block, JSON.stringify(event)), type); +} + +function syntheticJsonBlock(template: string, event: JsonRecord): string { + const newline = blockNewline(template); + const type = typeof event.type === "string" ? event.type : "message"; + return `event: ${type}${newline}data: ${JSON.stringify(event)}`; +} + +function callKindForDelta(type: unknown): DeferredCallKind | undefined { + if (type === "response.function_call_arguments.delta") return "function"; + if (type === "response.custom_tool_call_input.delta") return "custom"; + return undefined; +} + +function callKindForDone(type: unknown): DeferredCallKind | undefined { + if (type === "response.function_call_arguments.done") return "function"; + if (type === "response.custom_tool_call_input.done") return "custom"; + return undefined; +} + +function callKey(kind: DeferredCallKind, outputIndex: number): string { + return `${kind}:${outputIndex}`; +} + +function deltaTypeFor(kind: DeferredCallKind): string { + return kind === "function" + ? "response.function_call_arguments.delta" + : "response.custom_tool_call_input.delta"; +} + +function completeValueFor(kind: DeferredCallKind, event: JsonRecord): string { + const value = kind === "function" ? event.arguments : event.input; + return typeof value === "string" ? value : ""; +} + +/** + * Create one provider-scoped block rewrite per upstream response. + * + * Retained ids and pending-call markers are charged to the request translator + * budget and hard-capped. `dispose` releases them on terminal, EOF, cancel, or + * relay failure through the shared block-rewrite lifecycle. + */ +export function createGithubCopilotResponsesBlockRewrite( + budget?: TranslatorBudget, +): SseBlockRewrite { + const items = new Map(); + const deferredCalls = new Map(); + let responseId: string | undefined; + let itemStateBytes = 0; + let callStateBytes = 0; + let nextSequenceNumber = 0; + let disposed = false; + + const ensureStateLimit = (nextBytes: number): void => { + if (itemStateBytes + callStateBytes + nextBytes > RETAINED_STATE_LIMIT_BYTES) { + throw new TranslatorBudgetExceededError("item_ids", RETAINED_STATE_LIMIT_BYTES); + } + }; + + const chargeItemState = (bytes: number): void => { + ensureStateLimit(bytes); + budget?.chargeRetained(bytes, { kind: "item_ids" }); + itemStateBytes += bytes; + }; + + const chargeCallState = (bytes: number): void => { + ensureStateLimit(bytes); + budget?.chargeRetained(bytes, { kind: "retained_collectors" }); + callStateBytes += bytes; + }; + + const releaseDeferredCall = (key: string): boolean => { + const bytes = deferredCalls.get(key); + if (bytes === undefined) return false; + deferredCalls.delete(key); + callStateBytes -= bytes; + budget?.releaseRetained(bytes, { kind: "retained_collectors" }); + return true; + }; + + const rememberResponseId = (candidate: string): string => { + if (responseId !== undefined) return responseId; + const bytes = byteLength(candidate); + chargeItemState(bytes); + responseId = candidate; + return candidate; + }; + + const rememberItem = ( + outputIndex: number, + candidateId: string, + itemType?: string, + encryptedReasoning = false, + ): TrackedItem => { + const existing = items.get(outputIndex); + if (existing) { + if (existing.type === undefined && itemType !== undefined) { + const extra = byteLength(itemType); + chargeItemState(extra); + existing.type = itemType; + } + existing.encryptedReasoning ||= encryptedReasoning; + return existing; + } + if (items.size >= MAX_TRACKED_ITEMS) { + throw new TranslatorBudgetExceededError("item_ids", RETAINED_STATE_LIMIT_BYTES); + } + chargeItemState(byteLength([outputIndex, candidateId, itemType ?? null])); + const tracked = { id: candidateId, type: itemType, encryptedReasoning }; + items.set(outputIndex, tracked); + return tracked; + }; + + const rememberDeferredCall = (key: string): void => { + if (deferredCalls.has(key)) return; + if (deferredCalls.size >= MAX_TRACKED_DEFERRED_CALLS) { + throw new TranslatorBudgetExceededError("retained_collectors", RETAINED_STATE_LIMIT_BYTES); + } + const bytes = byteLength(key); + chargeCallState(bytes); + deferredCalls.set(key, bytes); + }; + + const sanitizeReasoningItem = ( + item: JsonRecord, + tracked: TrackedItem, + ): void => { + if (item.type !== "reasoning") return; + const encrypted = typeof item.encrypted_content === "string"; + tracked.encryptedReasoning ||= encrypted; + if (encrypted) delete item.encrypted_content; + if (tracked.encryptedReasoning) { + const summary = Array.isArray(item.summary) ? item.summary : []; + const hasPlaintextSummary = summary.some(part => + isRecord(part) + && part.type === "summary_text" + && typeof part.text === "string" + && part.text.length > 0); + if (!hasPlaintextSummary) item.summary = [{ type: "summary_text", text: "" }]; + if (!Array.isArray(item.content)) item.content = []; + } + }; + + const normalizeNestedItem = (item: JsonRecord, outputIndex: number): void => { + const candidateId = typeof item.id === "string" ? item.id : undefined; + if (candidateId === undefined) return; + const itemType = typeof item.type === "string" ? item.type : undefined; + const tracked = rememberItem( + outputIndex, + candidateId, + itemType, + itemType === "reasoning" && typeof item.encrypted_content === "string", + ); + item.id = tracked.id; + sanitizeReasoningItem(item, tracked); + }; + + const normalizeSequence = (event: JsonRecord): void => { + if (typeof event.sequence_number !== "number" || !Number.isFinite(event.sequence_number)) return; + event.sequence_number = nextSequenceNumber++; + }; + + const dispose = (): void => { + if (disposed) return; + disposed = true; + if (itemStateBytes > 0) budget?.releaseRetained(itemStateBytes, { kind: "item_ids" }); + if (callStateBytes > 0) budget?.releaseRetained(callStateBytes, { kind: "retained_collectors" }); + items.clear(); + deferredCalls.clear(); + responseId = undefined; + itemStateBytes = 0; + callStateBytes = 0; + }; + + const rewrite: SseBlockRewrite = (block) => { + const payload = sseDataPayload(block); + if (payload === null || payload === "[DONE]") return [block]; + + let event: unknown; + try { + event = JSON.parse(payload); + } catch { + return [block]; + } + if (!isRecord(event)) return [block]; + + if (typeof event.obfuscation === "string") delete event.obfuscation; + + const type = event.type; + const outputIndex = outputIndexOf(event.output_index); + const deltaKind = callKindForDelta(type); + const deferToolDelta = deltaKind !== undefined + && outputIndex !== undefined; + + if (isRecord(event.response) && typeof event.response.id === "string") { + event.response.id = rememberResponseId(event.response.id); + } + + if (outputIndex !== undefined && isRecord(event.item)) { + normalizeNestedItem(event.item, outputIndex); + } + // A deferred delta is not client-visible, so it must not establish the + // public item id if a malformed provider sends it before output_item.added. + if (!deferToolDelta && outputIndex !== undefined && typeof event.item_id === "string") { + const tracked = rememberItem(outputIndex, event.item_id); + event.item_id = tracked.id; + } + + if (type === "response.completed" && isRecord(event.response) && Array.isArray(event.response.output)) { + event.response.output.forEach((item, index) => { + if (isRecord(item)) normalizeNestedItem(item, index); + }); + } + + if (typeof type === "string" && type.startsWith("response.reasoning")) { + if (typeof event.encrypted_content === "string") delete event.encrypted_content; + // `encrypted_content` is opaque replay state; summary delta/text fields + // are a separate, potentially readable part of the public Responses + // contract and must survive when Copilot provides them. + } + + if (deferToolDelta) { + rememberDeferredCall(callKey(deltaKind, outputIndex)); + // The OpenAI `obfuscation` field is padding, not a decryption key. Defer + // every Copilot tool fragment anyway: mixed padded/unpadded fragments and + // rotating ids cannot then produce duplicate or irreconcilable input. + // `.done` carries the authoritative complete value, which is emitted as + // one canonical delta immediately before the original `.done` event. + return []; + } + + const emitted: Array<{ event: JsonRecord; synthetic: boolean }> = []; + const doneKind = callKindForDone(type); + if (doneKind !== undefined && outputIndex !== undefined) { + const key = callKey(doneKind, outputIndex); + if (releaseDeferredCall(key)) { + emitted.push({ + synthetic: true, + event: { + type: deltaTypeFor(doneKind), + output_index: outputIndex, + ...(typeof event.item_id === "string" ? { item_id: event.item_id } : {}), + ...(typeof event.sequence_number === "number" ? { sequence_number: event.sequence_number } : {}), + delta: completeValueFor(doneKind, event), + }, + }); + } + } + emitted.push({ event, synthetic: false }); + + for (const entry of emitted) normalizeSequence(entry.event); + return emitted.map(entry => entry.synthetic + ? syntheticJsonBlock(block, entry.event) + : rewriteJsonBlock(block, entry.event)); + }; + rewrite.dispose = dispose; + return rewrite; +} diff --git a/src/server/relay.ts b/src/server/relay.ts index c1d8c6b6a6..2378c02c67 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -587,6 +587,12 @@ export type SseInspectorHandlers = { logCtx?: RequestLogContext; onCompletedResponse?: (response: { id?: unknown; output?: unknown; status?: unknown }) => void; onFirstOutput?: () => void; + /** + * Provider-scoped compatibility: persist the completed snapshot under the + * first response id exposed to the client when an upstream changes ids + * between `response.created` and `response.completed`. + */ + pinCompletedResponseIdToFirstSeen?: boolean; }; type CompletedOutputItem = { item: unknown; sourceBytes: number }; @@ -658,6 +664,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector : null; let aggregateItemBytes = 0; let reconstructionTainted = false; + let firstResponseId: string | undefined; const clearFrameState = (): void => { delimiterTail = new Uint8Array(0); @@ -678,6 +685,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector decoder = null; clearFrameState(); clearCompletedItems(); + firstResponseId = undefined; }; const retainCandidateSlice = (slice: Uint8Array): void => { @@ -772,6 +780,17 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector const parsedEvent = parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as ParsedSseEvent : null; + const responseRecord = parsedEvent + && typeof parsedEvent.response === "object" + && parsedEvent.response !== null + && !Array.isArray(parsedEvent.response) + ? parsedEvent.response as { id?: unknown } + : null; + if (handlers.pinCompletedResponseIdToFirstSeen + && responseRecord + && typeof responseRecord.id === "string") { + firstResponseId ??= responseRecord.id; + } const doneItem = parsedEvent?.type === "response.output_item.done" ? parsedEvent.item : undefined; if (parsedEvent && doneItem !== undefined @@ -786,6 +805,11 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector let response = completedResponseFromParsedEvent(parsedEvent); if (response) { + if (handlers.pinCompletedResponseIdToFirstSeen + && firstResponseId !== undefined + && response.id !== firstResponseId) { + response = { ...response, id: firstResponseId }; + } // Authoritative output is a NON-EMPTY ARRAY only. Anything else // (missing, null, scalar, object) keeps the historical backfill // behavior so a malformed terminal cannot reach rememberResponseState @@ -901,6 +925,8 @@ export type InspectionConsumerOptions = { drainBounds?: Partial; upstream?: AbortController; now?: () => number; + /** Forward provider-scoped response-id pinning to the owned inspector. */ + pinCompletedResponseIdToFirstSeen?: boolean; /** Test seam for proving both public consumers dispose their owned inspector. */ inspectorFactory?: (handlers: SseInspectorHandlers) => SseInspector; }; @@ -1057,6 +1083,7 @@ export function consumeForInspection( logCtx, onCompletedResponse, onFirstOutput, + pinCompletedResponseIdToFirstSeen: options?.pinCompletedResponseIdToFirstSeen, }); startBoundedInspectionPump({ ...options, @@ -1102,6 +1129,7 @@ export function consumeForResponseLogMetadata( logCtx, onCompletedResponse, onFirstOutput, + pinCompletedResponseIdToFirstSeen: options?.pinCompletedResponseIdToFirstSeen, }); startBoundedInspectionPump({ ...options, reader, inspector, signal, onDone }); } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 25bf843194..e8bc1e7e84 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -171,7 +171,6 @@ import { imageGenToolCallAliases, restoreImageGenCallsInJson, } from "../responses-image-gen-repair"; -import { composeSsePayloadRewrites, relaySseWithPayloadRewrite } from "../sse-payload-rewrite"; import { createResponsesModelPayloadRewrite, rewriteResponsesModelJson } from "../responses-model-rewrite"; import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; @@ -187,9 +186,11 @@ import { } from "../responses-snapshot-repair"; import { composeSseBlockRewrites, + composeSsePayloadRewrites, payloadRewriteAsBlockRewrite, relaySseWithBlockRewrite, } from "../sse-payload-rewrite"; +import { createGithubCopilotResponsesBlockRewrite } from "../github-copilot-responses-repair"; import { responsesJsonToSseBody } from "../responses-json-events"; import { guardTerminalEventStream } from "./terminal-guard"; @@ -2086,14 +2087,11 @@ async function handleResponsesInner( if (isEventStream && upstreamResponse.body) { const repairConfig = route.provider.responsesItemIdRepair; const snapshotRepairEnabled = hasResponsesSnapshotRepair(route.provider.responsesSnapshotRepair); + const githubCopilotRepairEnabled = route.providerName === "github-copilot"; const responseModelRewrite = parsed._responseModelId !== undefined && parsed._responseModelId !== parsed.modelId ? createResponsesModelPayloadRewrite(parsed._responseModelId) : undefined; - const needsClientRewrite = imageGenCallAliases.size > 0 - || hasResponsesItemIdRepair(repairConfig) - || snapshotRepairEnabled - || responseModelRewrite !== undefined; // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first). const payloadRewrites = [ createImageGenCallRestoreRewrite(imageGenCallAliases), @@ -2113,12 +2111,21 @@ async function handleResponsesInner( return undefined; } })(); - const clientBlockRewrite = snapshotRepairEnabled - ? composeSseBlockRewrites( - payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)), - createResponsesSnapshotBlockRewrite(snapshotDefaultsRequest, translatorBudget), - ) + const blockRewrites = [ + payloadRewrites.length > 0 + ? payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)) + : undefined, + githubCopilotRepairEnabled + ? createGithubCopilotResponsesBlockRewrite(translatorBudget) + : undefined, + snapshotRepairEnabled + ? createResponsesSnapshotBlockRewrite(snapshotDefaultsRequest, translatorBudget) + : undefined, + ].filter((rewrite): rewrite is NonNullable => rewrite !== undefined); + const clientBlockRewrite = blockRewrites.length > 0 + ? composeSseBlockRewrites(...blockRewrites) : undefined; + const needsClientRewrite = clientBlockRewrite !== undefined; // #864: win32 rewrite traffic must never enter the tee()+JS-pull chain // (Bun#32111 JS-sink segfault — text frames pass, the terminal block is // lost). The eager single reader applies the same rewrites inline. @@ -2159,6 +2166,7 @@ async function handleResponsesInner( logCtx, onCompletedResponse: rememberPassthroughResponse, onFirstOutput: options.onFirstOutput, + pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled, }); const eagerBody = relaySseEagerBounded(upstreamResponse.body, turnAc, { inspectChunk: chunk => inspector.feed(chunk), @@ -2167,9 +2175,6 @@ async function handleResponsesInner( // Stream lifetime follows the protocol terminal even when this request // has no outcome callback configured (reported() would stay false). sawTerminal: () => inspector.terminalSeen(), - ...(win32EagerRewrite - ? { rewritePayload: composeSsePayloadRewrites(...payloadRewrites) } - : {}), ...(clientBlockRewrite ? { rewriteBlocks: clientBlockRewrite } : {}), @@ -2208,6 +2213,7 @@ async function handleResponsesInner( clientGoneSignal: clientGone.signal, drainBounds: { ms: 15_000, bytes: 32 * 1024 * 1024 }, upstream, + pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled, }; if (recordTerminalOutcomes) { // A real terminal was parsed from the (teed) inspection stream — record it as the outcome @@ -2260,8 +2266,8 @@ async function handleResponsesInner( // Windows was handled by the eager terminal-aware branch above. Remaining // tee traffic can use the JS relay to close on a protocol terminal and to // convert a mid-stream reset into a clean response.failed event. - const rewrittenBody = clientBlockRewrite !== undefined || payloadRewrites.length > 0 - ? relaySseWithBlockRewrite(nativeBody, clientBlockRewrite ?? payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)), translatorBudget) + const rewrittenBody = clientBlockRewrite !== undefined + ? relaySseWithBlockRewrite(nativeBody, clientBlockRewrite, translatorBudget) : nativeBody; const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); return markNativePassthroughSseResponse(new Response(clientBody, { diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index a791515324..3c6d825e6c 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -198,12 +198,14 @@ export function relaySseWithBlockRewrite( const emitProcessedBlocks = ( controller: ReadableStreamDefaultController, flushFinal = false, - ): void => { + ): number => { + let emitted = 0; let next: { block: string; delimiter: string; rest: string } | null; while ((next = nextSseBlock(buffer))) { replaceBuffer(next.rest); for (const outBlock of rewrite(next.block)) { enqueueText(controller, outBlock + next.delimiter); + emitted += 1; } } if (flushFinal && buffer.length > 0) { @@ -213,28 +215,37 @@ export function relaySseWithBlockRewrite( const tailDelimiter = buffer.includes("\r\n") ? "\r\n\r\n" : "\n\n"; for (let i = 0; i < tailBlocks.length; i++) { enqueueText(controller, tailBlocks[i]! + (i < tailBlocks.length - 1 ? tailDelimiter : "")); + emitted += 1; } releaseBuffer(); } + return emitted; }; return new ReadableStream({ async pull(controller) { try { - const { done, value } = await reader.read(); - // A cancel raced this pending read: never feed the rewriter again - // after its disposal (#893 review). - if (cancelled) return; - if (done) { - appendBuffer(decoder.decode()); - emitProcessedBlocks(controller, true); - releaseBuffer(); - disposeRewrite(); - controller.close(); - return; + // A network chunk is not an SSE-event boundary. Bun may not issue a + // second pull after a fulfilled pull enqueues nothing, so keep reading + // until at least one complete rewritten block is available or EOF is + // reached. This also handles block rewrites that intentionally drop an + // event without parking the client stream. + for (;;) { + const { done, value } = await reader.read(); + // A cancel raced this pending read: never feed the rewriter again + // after its disposal (#893 review). + if (cancelled) return; + if (done) { + appendBuffer(decoder.decode()); + emitProcessedBlocks(controller, true); + releaseBuffer(); + disposeRewrite(); + controller.close(); + return; + } + appendBuffer(decoder.decode(value, { stream: true })); + if (emitProcessedBlocks(controller) > 0) return; } - appendBuffer(decoder.decode(value, { stream: true })); - emitProcessedBlocks(controller); } catch (error) { releaseBuffer(); disposeRewrite(); diff --git a/tests/github-copilot-sse-rewrite.test.ts b/tests/github-copilot-sse-rewrite.test.ts new file mode 100644 index 0000000000..4fecade68e --- /dev/null +++ b/tests/github-copilot-sse-rewrite.test.ts @@ -0,0 +1,254 @@ +import { describe, expect, test } from "bun:test"; +import { createGithubCopilotResponsesBlockRewrite } from "../src/server/github-copilot-responses-repair"; +import { relaySseWithBlockRewrite } from "../src/server/sse-payload-rewrite"; +import { createTestTranslatorBudget } from "./helpers/translator-budget"; + +function streamFromChunks(...chunks: string[]): ReadableStream { + const encoder = new TextEncoder(); + let index = 0; + return new ReadableStream({ + pull(controller) { + const chunk = chunks[index++]; + if (chunk === undefined) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(chunk)); + }, + }); +} + +async function readAll(stream: ReadableStream): Promise { + const decoder = new TextDecoder(); + const reader = stream.getReader(); + let text = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return text + decoder.decode(); + text += decoder.decode(value, { stream: true }); + } +} + +function frame(type: string, payload: Record, newline = "\n"): string { + return `event: ${type}${newline}data: ${JSON.stringify(payload)}${newline}${newline}`; +} + +function parsedFrames(text: string): Array<{ event?: string; payload: Record }> { + return text + .trim() + .split(/\r?\n\r?\n/) + .map((block) => { + const lines = block.split(/\r?\n/); + const event = lines.find(line => line.startsWith("event:"))?.slice(6).trim(); + const data = lines.find(line => line.startsWith("data:"))?.slice(5).trim(); + if (!data || data === "[DONE]") return null; + return { event, payload: JSON.parse(data) as Record }; + }) + .filter((entry): entry is { event?: string; payload: Record } => entry !== null); +} + +describe("GitHub Copilot Responses SSE repair", () => { + test("reconstructs a custom-tool input as separately named plaintext frames", async () => { + const upstream = [ + frame("response.output_item.added", { + type: "response.output_item.added", + output_index: 0, + sequence_number: 40, + item: { type: "custom_tool_call", id: "ctc-first", call_id: "call-1", name: "apply_patch", input: "" }, + }), + frame("response.custom_tool_call_input.delta", { + type: "response.custom_tool_call_input.delta", + output_index: 0, + item_id: "ctc-cipher-delta", + sequence_number: 41, + delta: "cipher-tool-input", + obfuscation: "padding-or-key", + }), + frame("response.custom_tool_call_input.done", { + type: "response.custom_tool_call_input.done", + output_index: 0, + item_id: "ctc-cipher-input-done", + sequence_number: 42, + input: "*** Begin Patch\n*** End Patch", + }), + frame("response.output_item.done", { + type: "response.output_item.done", + output_index: 0, + sequence_number: 43, + item: { + type: "custom_tool_call", + id: "ctc-cipher-item-done", + call_id: "call-1", + name: "apply_patch", + input: "*** Begin Patch\n*** End Patch", + }, + }), + ].join(""); + + const budget = createTestTranslatorBudget(); + const rewritten = await readAll(relaySseWithBlockRewrite( + streamFromChunks(upstream.slice(0, 137), upstream.slice(137)), + createGithubCopilotResponsesBlockRewrite(budget), + budget, + )); + const events = parsedFrames(rewritten); + + expect(events.map(event => event.payload.type)).toEqual([ + "response.output_item.added", + "response.custom_tool_call_input.delta", + "response.custom_tool_call_input.done", + "response.output_item.done", + ]); + expect(events.map(event => event.event)).toEqual(events.map(event => event.payload.type)); + expect(events.map(event => event.payload.sequence_number)).toEqual([0, 1, 2, 3]); + expect(events.slice(1, 3).map(event => event.payload.item_id)).toEqual([ + "ctc-first", + "ctc-first", + ]); + expect(events.slice(1, 3).map(event => event.payload.delta ?? event.payload.input)).toEqual([ + "*** Begin Patch\n*** End Patch", + "*** Begin Patch\n*** End Patch", + ]); + expect((events[3]!.payload.item as { id: string }).id).toBe("ctc-first"); + expect(rewritten).not.toContain("obfuscation"); + expect(rewritten).not.toContain("cipher-tool-input"); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("preserves readable delta content when OpenAI-style padding is present", async () => { + const upstream = frame("response.output_text.delta", { + type: "response.output_text.delta", + output_index: 0, + item_id: "msg-readable", + sequence_number: 7, + delta: "bonjour", + obfuscation: "random-padding", + }); + const budget = createTestTranslatorBudget(); + const rewritten = await readAll(relaySseWithBlockRewrite( + streamFromChunks(upstream), + createGithubCopilotResponsesBlockRewrite(budget), + budget, + )); + + expect(rewritten).toContain('"delta":"bonjour"'); + expect(rewritten).not.toContain("obfuscation"); + }); + + test("removes encrypted reasoning state, preserves readable summaries, and pins ids", async () => { + const upstream = [ + frame("response.created", { + type: "response.created", + sequence_number: 10, + response: { id: "resp-first", status: "in_progress", output: [] }, + }), + frame("response.output_item.added", { + type: "response.output_item.added", + output_index: 0, + sequence_number: 11, + item: { type: "reasoning", id: "rs-first", summary: [], encrypted_content: "reasoning-cipher-1" }, + }), + frame("response.reasoning_summary_text.delta", { + type: "response.reasoning_summary_text.delta", + output_index: 0, + item_id: "rs-cipher-delta", + summary_index: 0, + sequence_number: 12, + delta: "plan lisible", + obfuscation: "padding-or-key", + }), + frame("response.output_item.done", { + type: "response.output_item.done", + output_index: 0, + sequence_number: 13, + item: { + type: "reasoning", + id: "rs-cipher-done", + summary: [{ type: "summary_text", text: "plan lisible" }], + encrypted_content: "reasoning-cipher-2", + }, + }), + frame("response.completed", { + type: "response.completed", + sequence_number: 14, + response: { + id: "resp-cipher-completed", + status: "completed", + output: [{ + type: "reasoning", + id: "rs-cipher-completed", + summary: [{ type: "summary_text", text: "plan lisible" }], + encrypted_content: "reasoning-cipher-3", + }], + }, + }), + ].join(""); + const budget = createTestTranslatorBudget(); + const rewritten = await readAll(relaySseWithBlockRewrite( + streamFromChunks(upstream), + createGithubCopilotResponsesBlockRewrite(budget), + budget, + )); + const events = parsedFrames(rewritten); + + expect(rewritten).not.toContain("encrypted_content"); + expect(rewritten).not.toContain("reasoning-cipher"); + expect(events[2]!.payload.delta).toBe("plan lisible"); + expect(events[2]!.payload.item_id).toBe("rs-first"); + expect((events[3]!.payload.item as { id: string }).id).toBe("rs-first"); + const completed = events[4]!.payload.response as { id: string; output: Array<{ id: string; summary: unknown[] }> }; + expect(completed.id).toBe("resp-first"); + expect(completed.output[0]!.id).toBe("rs-first"); + expect(completed.output[0]!.summary).toEqual([{ type: "summary_text", text: "plan lisible" }]); + }); + + test("consolidates unpadded function arguments and preserves CRLF framing", async () => { + const upstream = [ + frame("response.output_item.added", { + type: "response.output_item.added", + output_index: 0, + sequence_number: 0, + item: { type: "function_call", id: "fc-first", call_id: "call-fn", name: "shell", arguments: "" }, + }, "\r\n"), + frame("response.function_call_arguments.delta", { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: "fc-cipher", + sequence_number: 1, + delta: "function-cipher", + }, "\r\n"), + frame("response.function_call_arguments.done", { + type: "response.function_call_arguments.done", + output_index: 0, + item_id: "fc-done-cipher", + sequence_number: 2, + arguments: "{\"command\":\"pwd\"}", + }, "\r\n"), + ].join(""); + const budget = createTestTranslatorBudget(); + const rewritten = await readAll(relaySseWithBlockRewrite( + streamFromChunks(upstream), + createGithubCopilotResponsesBlockRewrite(budget), + budget, + )); + const events = parsedFrames(rewritten); + + expect(events.map(event => event.payload.type)).toEqual([ + "response.output_item.added", + "response.function_call_arguments.delta", + "response.function_call_arguments.done", + ]); + expect(events[1]!.payload.delta).toBe('{"command":"pwd"}'); + expect(events.slice(1).every(event => event.payload.item_id === "fc-first")).toBe(true); + expect(rewritten).not.toContain("function-cipher"); + expect(rewritten).toContain("\r\n\r\n"); + expect(rewritten.replaceAll("\r\n", "")).not.toContain("\n"); + }); + + test("passes malformed and sentinel blocks through without throwing", () => { + const rewrite = createGithubCopilotResponsesBlockRewrite(); + expect(rewrite("event: message\ndata: {not-json")).toEqual(["event: message\ndata: {not-json"]); + expect(rewrite("data: [DONE]")).toEqual(["data: [DONE]"]); + rewrite.dispose?.(); + }); +}); diff --git a/tests/github-copilot-stream-contract.test.ts b/tests/github-copilot-stream-contract.test.ts new file mode 100644 index 0000000000..7843f914db --- /dev/null +++ b/tests/github-copilot-stream-contract.test.ts @@ -0,0 +1,248 @@ +/** + * Regression contract for GitHub Copilot's Responses SSE dialect. + * + * Copilot rotates opaque response/item ids between events, exposes encrypted + * reasoning, and pads streamed tool deltas. Those fields are valid for the + * vscode-chat client, but not for a stock Responses client such as codex-rs. + * The proxy must expose one internally consistent, plaintext Responses stream. + */ +import { afterEach, describe, expect, test } from "bun:test"; +import { providerConfigSeed } from "../src/providers/derive"; +import { getProviderRegistryEntry } from "../src/providers/registry"; +import { handleResponses } from "../src/server/responses/core"; +import type { OcxConfig, OcxProviderConfig } from "../src/types"; + +interface SseEvent { + event?: string; + data: string; + payload?: Record; +} + +function copilotProvider(): OcxProviderConfig { + return { + ...providerConfigSeed(getProviderRegistryEntry("github-copilot")!), + authMode: "key", + apiKey: "sk-test", + }; +} + +function sse(event: string, payload: Record): string { + return `event: ${event}\ndata: ${JSON.stringify(payload)}\n\n`; +} + +function parseSse(text: string): SseEvent[] { + return text + .split(/\r?\n\r?\n/) + .map((block) => { + const event = block.split(/\r?\n/).find(line => line.startsWith("event:"))?.slice(6).trim(); + const data = block.split(/\r?\n/).find(line => line.startsWith("data:"))?.slice(5).trim(); + if (data === undefined) return null; + if (data === "[DONE]") return { event, data }; + try { + return { event, data, payload: JSON.parse(data) as Record }; + } catch { + return { event, data }; + } + }) + .filter((event): event is SseEvent => event !== null); +} + +const COPILOT_OBFUSCATED_STREAM = [ + sse("response.created", { + type: "response.created", + sequence_number: 20, + response: { id: "resp-created", status: "in_progress", output: [] }, + }), + sse("response.in_progress", { + type: "response.in_progress", + sequence_number: 21, + response: { id: "resp-in-progress", status: "in_progress", output: [] }, + }), + sse("response.output_item.added", { + type: "response.output_item.added", + sequence_number: 22, + output_index: 0, + item: { + type: "reasoning", + id: "rs-added", + summary: [], + content: [], + encrypted_content: "cipher-reasoning-added", + }, + }), + sse("response.reasoning_summary_text.delta", { + type: "response.reasoning_summary_text.delta", + sequence_number: 23, + output_index: 0, + item_id: "rs-delta", + summary_index: 0, + delta: "résumé lisible", + obfuscation: "copilot-vscode-chat", + }), + sse("response.output_item.done", { + type: "response.output_item.done", + sequence_number: 24, + output_index: 0, + item: { + type: "reasoning", + id: "rs-done", + summary: [], + content: [], + encrypted_content: "cipher-reasoning-done", + }, + }), + sse("response.output_item.added", { + type: "response.output_item.added", + sequence_number: 25, + output_index: 1, + item: { + type: "custom_tool_call", + id: "ctc-added", + call_id: "call-patch", + name: "apply_patch", + input: "", + status: "in_progress", + }, + }), + sse("response.custom_tool_call_input.delta", { + type: "response.custom_tool_call_input.delta", + sequence_number: 26, + output_index: 1, + item_id: "ctc-delta", + delta: "cipher-tool-delta", + obfuscation: "copilot-vscode-chat", + }), + sse("response.custom_tool_call_input.done", { + type: "response.custom_tool_call_input.done", + sequence_number: 27, + output_index: 1, + item_id: "ctc-input-done", + input: "*** Begin Patch\n*** End Patch", + }), + sse("response.output_item.done", { + type: "response.output_item.done", + sequence_number: 28, + output_index: 1, + item: { + type: "custom_tool_call", + id: "ctc-done", + call_id: "call-patch", + name: "apply_patch", + input: "*** Begin Patch\n*** End Patch", + status: "completed", + }, + }), + sse("response.completed", { + type: "response.completed", + sequence_number: 29, + response: { + id: "resp-completed", + status: "completed", + output: [ + { + type: "reasoning", + id: "rs-completed", + summary: [], + content: [], + encrypted_content: "cipher-reasoning-completed", + }, + { + type: "custom_tool_call", + id: "ctc-completed", + call_id: "call-patch", + name: "apply_patch", + input: "*** Begin Patch\n*** End Patch", + status: "completed", + }, + ], + }, + }), + "data: [DONE]\n\n", +].join(""); + +describe("GitHub Copilot Responses client stream contract", () => { + const originalFetch = globalThis.fetch; + + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + test("normalizes obfuscated Copilot SSE before it reaches Codex", async () => { + globalThis.fetch = (async () => new Response(COPILOT_OBFUSCATED_STREAM, { + status: 200, + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch; + + const config = { + providers: { "github-copilot": copilotProvider() }, + } as unknown as OcxConfig; + + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "github-copilot/gpt-5.6-luna", + input: "Use apply_patch", + stream: true, + }), + }), + config, + { model: "", provider: "" }, + ); + + expect(response.status).toBe(200); + const body = await response.text(); + const events = parseSse(body); + const payloads = events.flatMap(event => event.payload ? [event.payload] : []); + + // Copilot-only ciphertext/metadata must never leak onto the stock Responses wire. + expect(body).not.toContain("obfuscation"); + expect(body).not.toContain("encrypted_content"); + expect(body).not.toContain("cipher-"); + expect(body).toContain("résumé lisible"); + + // The first-seen response id is the public id for the whole streamed response. + const created = payloads.find(event => event.type === "response.created")!; + const inProgress = payloads.find(event => event.type === "response.in_progress")!; + const completed = payloads.find(event => event.type === "response.completed")!; + expect((inProgress.response as { id: string }).id) + .toBe((created.response as { id: string }).id); + expect((completed.response as { id: string }).id) + .toBe((created.response as { id: string }).id); + + // All events for an output item must agree on the id Codex first observed. + const reasoningIds = payloads.flatMap((event) => { + if (event.output_index !== 0) return []; + if (typeof event.item_id === "string") return [event.item_id]; + const item = event.item as { id?: unknown } | undefined; + return typeof item?.id === "string" ? [item.id] : []; + }); + expect(new Set(reasoningIds)).toEqual(new Set(["rs-added"])); + + const customIds = payloads.flatMap((event) => { + if (event.output_index !== 1) return []; + if (typeof event.item_id === "string") return [event.item_id]; + const item = event.item as { id?: unknown } | undefined; + return typeof item?.id === "string" ? [item.id] : []; + }); + expect(new Set(customIds)).toEqual(new Set(["ctc-added"])); + + // Obfuscated deltas are replaced by one plaintext delta before the authoritative done. + const customEvents = payloads.filter(event => + event.type === "response.custom_tool_call_input.delta" + || event.type === "response.custom_tool_call_input.done"); + expect(customEvents.map(event => event.type)).toEqual([ + "response.custom_tool_call_input.delta", + "response.custom_tool_call_input.done", + ]); + expect(customEvents.map(event => event.delta ?? event.input)).toEqual([ + "*** Begin Patch\n*** End Patch", + "*** Begin Patch\n*** End Patch", + ]); + + // The client sees a coherent ordered stream and a real terminal marker. + expect(payloads.map(event => event.sequence_number)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]); + expect(events.at(-1)?.data).toBe("[DONE]"); + }); +}); diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index 5c782b6072..c38e29f73f 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -47,19 +47,20 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { ); expect(sseBranch).toContain("upstreamResponse.body.tee()"); - // Windows no-rewrite traffic must honor the stream-mode/runtime gate so - // legacy-tee remains a safety escape hatch for Bun#32111. + // Rewrite traffic is derived from the finalized block chain so every + // provider-specific transform participates in the platform gate. expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;"); - expect(sseBranch).toContain("const needsClientRewrite = imageGenCallAliases.size > 0"); + expect(sseBranch).toContain('const githubCopilotRepairEnabled = route.providerName === "github-copilot";'); + expect(sseBranch).toContain("const needsClientRewrite = clientBlockRewrite !== undefined;"); expect(sseBranch).toContain("new Response(eagerBody"); - expect(sseBranch).toContain("const rewrittenBody = clientBlockRewrite !== undefined || payloadRewrites.length > 0"); + expect(sseBranch).toContain("const rewrittenBody = clientBlockRewrite !== undefined"); expect(sseBranch).toContain("eagerPath?.useEagerRelay || win32EagerRewrite"); expect(sseBranch).not.toContain("win32TerminalRelay"); // #864: win32 traffic that DOES need a client rewrite takes the eager single // reader with the payload rewrite applied inline — never the tee()+JS-pull // chain that loses the terminal block on Windows (Bun#32111). expect(sseBranch).toContain("win32EagerRewrite"); - expect(sseBranch).toContain("rewritePayload: composeSsePayloadRewrites(...payloadRewrites)"); + expect(sseBranch).toContain("rewriteBlocks: clientBlockRewrite"); // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed. expect(sseBranch).toContain("relaySseWithFailedTail(rewrittenBody, upstream"); expect(sseBranch).toContain("new Response(clientBody"); diff --git a/tests/sse-inspector-bounds.test.ts b/tests/sse-inspector-bounds.test.ts index 8ecc9621cf..4bc959460b 100644 --- a/tests/sse-inspector-bounds.test.ts +++ b/tests/sse-inspector-bounds.test.ts @@ -60,6 +60,25 @@ async function readAllBytes(stream: ReadableStream): Promise resetInspectionCountersForTest()); describe("createSseInspector frame bounds", () => { + test("optionally persists a completed snapshot under the first client-visible response id", () => { + const completed: Array<{ id?: unknown }> = []; + const inspector = createSseInspector({ + pinCompletedResponseIdToFirstSeen: true, + onCompletedResponse: response => completed.push(response), + }); + + inspector.feed(frame({ + type: "response.created", + response: { id: "resp-client-visible", status: "in_progress", output: [] }, + })); + inspector.feed(frame(completedEvent("resp-upstream-terminal", [{ type: "message", id: "msg-1" }]))); + + expect(completed).toEqual([ + expect.objectContaining({ id: "resp-client-visible" }), + ]); + inspector.dispose(); + }); + test("oversized candidate is inspection-only: tee client bytes stay exact and the next event resynchronizes", async () => { const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x\n\n`); const terminal = frame(completedEvent("after-cap")); diff --git a/tests/sse-payload-rewrite.test.ts b/tests/sse-payload-rewrite.test.ts index c6c72f615b..64d342a8d2 100644 --- a/tests/sse-payload-rewrite.test.ts +++ b/tests/sse-payload-rewrite.test.ts @@ -6,6 +6,7 @@ import { createImageGenCallRestoreRewrite } from "../src/server/responses-image- import { createResponsesItemIdPayloadRewrite } from "../src/server/responses-item-id-repair"; import { composeSsePayloadRewrites, + relaySseWithBlockRewrite, relaySseWithPayloadRewrite, } from "../src/server/sse-payload-rewrite"; import { createTestTranslatorBudget } from "./helpers/translator-budget"; @@ -26,6 +27,21 @@ function streamFromText(text: string): ReadableStream { }); } +function streamFromTexts(texts: string[]): ReadableStream { + const encoder = new TextEncoder(); + let index = 0; + return new ReadableStream({ + pull(controller) { + const text = texts[index++]; + if (text === undefined) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(text)); + }, + }); +} + async function readAll(stream: ReadableStream): Promise { const reader = stream.getReader(); const decoder = new TextDecoder(); @@ -103,6 +119,24 @@ describe("SSE payload rewrite composition", () => { expect(composeSsePayloadRewrites()('{"a":1}')).toBe('{"a":1}'); }); + test("keeps pulling after a partial or intentionally dropped block", async () => { + const budget = createTestTranslatorBudget(); + const upstream = streamFromTexts([ + 'event: drop\ndata: {"type":"drop"', + '}\n\nevent: keep\ndata: {"type":"keep","delta":"ok"}\n\n', + ]); + const rewritten = relaySseWithBlockRewrite( + upstream, + (block) => block.includes('"type":"drop"') ? [] : [block], + budget, + ); + + expect(await readAll(rewritten)).toBe( + 'event: keep\ndata: {"type":"keep","delta":"ok"}\n\n', + ); + expect(budget.snapshot().currentBytes).toBe(0); + }); + test("unterminated rewrite accumulation closes through a typed failed tail", async () => { const budget = createTestTranslatorBudget({ maxTurnBytes: 64 }); const upstream = new AbortController(); From 58500d27724911e016f59326226b93058da21993 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 6 Aug 2026 11:18:38 +0200 Subject: [PATCH 2/2] fix(copilot): keep oversized response streams flowing Adopted from PR #1111 for issue #1110. --- src/server/github-copilot-responses-repair.ts | 11 +++++-- tests/github-copilot-sse-rewrite.test.ts | 29 ++++++++++++++++++- tests/sse-inspector-bounds.test.ts | 18 ++++++++++++ 3 files changed, 55 insertions(+), 3 deletions(-) diff --git a/src/server/github-copilot-responses-repair.ts b/src/server/github-copilot-responses-repair.ts index ec1088ee6a..449c208068 100644 --- a/src/server/github-copilot-responses-repair.ts +++ b/src/server/github-copilot-responses-repair.ts @@ -122,6 +122,7 @@ export function createGithubCopilotResponsesBlockRewrite( let itemStateBytes = 0; let callStateBytes = 0; let nextSequenceNumber = 0; + let itemTrackingTainted = false; let disposed = false; const ensureStateLimit = (nextBytes: number): void => { @@ -175,8 +176,13 @@ export function createGithubCopilotResponsesBlockRewrite( existing.encryptedReasoning ||= encryptedReasoning; return existing; } - if (items.size >= MAX_TRACKED_ITEMS) { - throw new TranslatorBudgetExceededError("item_ids", RETAINED_STATE_LIMIT_BYTES); + if (itemTrackingTainted || items.size >= MAX_TRACKED_ITEMS) { + // Keep relaying client-visible events after the count cap, but stop + // retaining new output-index mappings. The current event still receives + // the provider id, while earlier retained mappings remain available for + // stable ids. Byte-budget failures stay hard failures. + itemTrackingTainted = true; + return { id: candidateId, type: itemType, encryptedReasoning }; } chargeItemState(byteLength([outputIndex, candidateId, itemType ?? null])); const tracked = { id: candidateId, type: itemType, encryptedReasoning }; @@ -243,6 +249,7 @@ export function createGithubCopilotResponsesBlockRewrite( responseId = undefined; itemStateBytes = 0; callStateBytes = 0; + itemTrackingTainted = false; }; const rewrite: SseBlockRewrite = (block) => { diff --git a/tests/github-copilot-sse-rewrite.test.ts b/tests/github-copilot-sse-rewrite.test.ts index 4fecade68e..80b7d3e6b7 100644 --- a/tests/github-copilot-sse-rewrite.test.ts +++ b/tests/github-copilot-sse-rewrite.test.ts @@ -86,8 +86,10 @@ describe("GitHub Copilot Responses SSE repair", () => { ].join(""); const budget = createTestTranslatorBudget(); + const splitAt = upstream.indexOf("cipher-tool-input") + 5; + expect(splitAt).toBeGreaterThan(5); const rewritten = await readAll(relaySseWithBlockRewrite( - streamFromChunks(upstream.slice(0, 137), upstream.slice(137)), + streamFromChunks(upstream.slice(0, splitAt), upstream.slice(splitAt)), createGithubCopilotResponsesBlockRewrite(budget), budget, )); @@ -115,6 +117,31 @@ describe("GitHub Copilot Responses SSE repair", () => { expect(budget.snapshot().currentBytes).toBe(0); }); + test("continues relaying when output-index retention reaches its count cap", async () => { + const upstream = Array.from({ length: 257 }, (_, outputIndex) => frame( + "response.output_item.added", + { + type: "response.output_item.added", + output_index: outputIndex, + item: { type: "message", id: `message-${outputIndex}`, role: "assistant", content: [] }, + }, + )).join(""); + const budget = createTestTranslatorBudget(); + + const rewritten = await readAll(relaySseWithBlockRewrite( + streamFromChunks(upstream), + createGithubCopilotResponsesBlockRewrite(budget), + budget, + )); + const events = parsedFrames(rewritten); + + expect(events).toHaveLength(257); + expect((events[0]!.payload.item as { id: string }).id).toBe("message-0"); + expect((events[255]!.payload.item as { id: string }).id).toBe("message-255"); + expect((events[256]!.payload.item as { id: string }).id).toBe("message-256"); + expect(budget.snapshot().currentBytes).toBe(0); + }); + test("preserves readable delta content when OpenAI-style padding is present", async () => { const upstream = frame("response.output_text.delta", { type: "response.output_text.delta", diff --git a/tests/sse-inspector-bounds.test.ts b/tests/sse-inspector-bounds.test.ts index 4bc959460b..b036b06005 100644 --- a/tests/sse-inspector-bounds.test.ts +++ b/tests/sse-inspector-bounds.test.ts @@ -79,6 +79,24 @@ describe("createSseInspector frame bounds", () => { inspector.dispose(); }); + test("keeps the upstream completed response id when pinning is not enabled", () => { + const completed: Array<{ id?: unknown }> = []; + const inspector = createSseInspector({ + onCompletedResponse: response => completed.push(response), + }); + + inspector.feed(frame({ + type: "response.created", + response: { id: "resp-client-visible", status: "in_progress", output: [] }, + })); + inspector.feed(frame(completedEvent("resp-upstream-terminal", [{ type: "message", id: "msg-1" }]))); + + expect(completed).toEqual([ + expect.objectContaining({ id: "resp-upstream-terminal" }), + ]); + inspector.dispose(); + }); + test("oversized candidate is inspection-only: tee client bytes stay exact and the next event resynchronizes", async () => { const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x\n\n`); const terminal = frame(completedEvent("after-cap"));