Skip to content

fix(proxy): finalize completed responses streams after client abort #1251

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ding113 merged 2 commits into from
Jun 11, 2026
+867 −15
Merged

fix(proxy): finalize completed responses streams after client abort #1251

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7efbef0
fix(proxy): finalize complete responses after client abort
Brisbanehuang Jun 6, 2026
0e07d38
fix(proxy): sanitize inert chat chunks in responses streams
Brisbanehuang Jun 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 125 additions & 0 deletions src/app/v1/_lib/proxy/response-fixer/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,55 @@ const DEFAULT_CONFIG: ResponseFixerConfig = {
maxFixSize: 1024 * 1024,
};

const UTF8_DECODER = new TextDecoder();
const UTF8_ENCODER = new TextEncoder();

function nowMs(): number {
if (typeof performance !== "undefined" && typeof performance.now === "function") {
return performance.now();
}
return Date.now();
}

function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}

function hasMeaningfulValue(value: unknown): boolean {
if (value == null) return false;
if (typeof value === "string") return value.length > 0;
if (Array.isArray(value)) return value.length > 0;
if (isRecord(value)) return Object.keys(value).length > 0;
return true;
}

function isInertChatCompletionChoice(choice: unknown): boolean {
if (!isRecord(choice)) return false;
if (choice.finish_reason != null) return false;

const delta = choice.delta;
if (!isRecord(delta)) {
return true;
}

for (const [key, value] of Object.entries(delta)) {
if (key === "role") continue;
if (hasMeaningfulValue(value)) return false;
}

return true;
}
Comment on lines +49 to +64

@gemini-code-assist gemini-code-assist Bot Jun 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The key === "content" check inside the loop is redundant because the fallback hasMeaningfulValue(value) check performs the exact same validation and returns false if the value is meaningful. We can simplify the loop by removing this redundant block.

function isInertChatCompletionChoice(choice: unknown): boolean {
  if (!isRecord(choice)) return false;
  if (choice.finish_reason != null) return false;

  const delta = choice.delta;
  if (!isRecord(delta)) {
    return true;
  }

  for (const [key, value] of Object.entries(delta)) {
    if (key === "role") continue;
    if (hasMeaningfulValue(value)) return false;
  }

  return true;
}


function isInertChatCompletionChunkPayload(payload: unknown): boolean {
if (!isRecord(payload)) return false;
if (payload.object !== "chat.completion.chunk") return false;
if (hasMeaningfulValue(payload.usage)) return false;

const choices = payload.choices;
if (!Array.isArray(choices) || choices.length === 0) return false;
return choices.every(isInertChatCompletionChoice);
}

function toArrayBufferUint8Array(input: Uint8Array): Uint8Array<ArrayBuffer> {
// Response/BodyInit 在 DOM 类型中要求 ArrayBufferView（buffer 为 ArrayBuffer），这里避免 SharedArrayBuffer 类型污染
if (input.buffer instanceof ArrayBuffer) {
Expand Down Expand Up @@ -385,6 +427,13 @@ export class ResponseFixer {
}
}

const filtered = ResponseFixer.filterInertResponsesChatCompletionChunks(session, data);
if (filtered.applied) {
applied.sse.applied = true;
applied.sse.details ??= filtered.details;
data = filtered.data;
}

controller.enqueue(data);
},
flush(controller) {
Expand Down Expand Up @@ -418,6 +467,13 @@ export class ResponseFixer {
}
}

const filtered = ResponseFixer.filterInertResponsesChatCompletionChunks(session, data);
if (filtered.applied) {
applied.sse.applied = true;
applied.sse.details ??= filtered.details;
data = filtered.data;
}

controller.enqueue(data);
}

Expand Down Expand Up @@ -526,6 +582,75 @@ export class ResponseFixer {
return { data: concatUint8Chunks(chunks), applied };
}

private static filterInertResponsesChatCompletionChunks(
session: ProxySession,
data: Uint8Array
): { data: Uint8Array; applied: boolean; details?: string } {
if (session.originalFormat !== "response") {
return { data, applied: false };
}

const text = UTF8_DECODER.decode(data);
if (!text.includes('"chat.completion.chunk"')) {
return { data, applied: false };
}

const lines = text.split("\n");
const out: string[] = [];
let applied = false;
let skipNextBlankLine = false;

for (let i = 0; i < lines.length; i += 1) {
const line = lines[i];
const hasLineBreak = i < lines.length - 1;

if (skipNextBlankLine && ResponseFixer.isBlankSseSeparatorLine(line)) {
skipNextBlankLine = false;
continue;
}
skipNextBlankLine = false;

if (ResponseFixer.isInertChatCompletionDataLine(line)) {
applied = true;
skipNextBlankLine = true;
continue;
}

out.push(line);
if (hasLineBreak) out.push("\n");
}

if (!applied) {
return { data, applied: false };
}

return {
data: UTF8_ENCODER.encode(out.join("")),
applied: true,
details: "filtered_inert_chat_completion_chunk",
};
}

private static isInertChatCompletionDataLine(line: string): boolean {
if (!line.startsWith("data:")) return false;

let payloadText = line.slice(5);
if (payloadText.startsWith(" ")) {
payloadText = payloadText.slice(1);
}
if (!payloadText.startsWith("{")) return false;

try {
return isInertChatCompletionChunkPayload(JSON.parse(payloadText));
} catch {
return false;
}
}

private static isBlankSseSeparatorLine(line: string): boolean {
return line === "" || line === "\r";
}

private static fixMaybeDataJsonLine(
line: Uint8Array,
jsonFixer: JsonFixer
Expand Down
171 changes: 171 additions & 0 deletions src/app/v1/_lib/proxy/response-fixer/response-fixer.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,20 @@ function createSession() {
} as any;
}

function createSseResponse(payloadLines: string[]): Response {
const encoder = new TextEncoder();
const stream = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode(payloadLines.join("\n")));
controller.close();
},
});

return new Response(stream, {
headers: { "content-type": "text/event-stream" },
});
}

describe("ResponseFixer", () => {
beforeEach(() => {
vi.clearAllMocks();
Expand Down Expand Up @@ -205,6 +219,163 @@ describe("ResponseFixer", () => {
expect(session.getSpecialSettings()).toBeNull();
});

test("流式 Responses SSE：应过滤上游混入的空 Chat Completions chunk", async () => {
const { ResponseFixer } = await import("./index");

const session = createSession();
session.originalFormat = "response";
const encoder = new TextEncoder();
const emptyChatChunk = {
id: "chatcmpl-dummy",
object: "chat.completion.chunk",
created: 1780753978,
model: "gpt-5.5",
choices: [{ index: 0, delta: { role: "assistant", content: "" } }],
};
const responseDelta = {
type: "response.output_text.delta",
delta: "Hi",
};
const responseCompleted = {
type: "response.completed",
response: { id: "resp_test", object: "response" },
};

const stream = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(
encoder.encode(
[
`data: ${JSON.stringify(emptyChatChunk)}`,
"",
"event: response.output_text.delta",
`data: ${JSON.stringify(responseDelta)}`,
"",
"event: response.completed",
`data: ${JSON.stringify(responseCompleted)}`,
"",
].join("\n")
)
);
controller.close();
},
});

const response = new Response(stream, {
headers: { "content-type": "text/event-stream" },
});

const fixed = await ResponseFixer.process(session, response);
const text = await fixed.text();

expect(text).not.toContain("chat.completion.chunk");
expect(text).not.toContain("chatcmpl-dummy");
expect(text.startsWith("event: response.output_text.delta")).toBe(true);
expect(text).toContain("response.output_text.delta");
expect(text).toContain("response.completed");
expect(session.getSpecialSettings()).not.toBeNull();
});

test("流式 Responses SSE：包含实际 content 的 Chat Completions chunk 应保留", async () => {
const { ResponseFixer } = await import("./index");

const session = createSession();
session.originalFormat = "response";
const chatChunk = {
id: "chatcmpl-content",
object: "chat.completion.chunk",
created: 1780753978,
model: "gpt-5.5",
choices: [{ index: 0, delta: { role: "assistant", content: "Hi" } }],
};

const fixed = await ResponseFixer.process(
session,
createSseResponse([`data: ${JSON.stringify(chatChunk)}`, ""])
);
const text = await fixed.text();

expect(text).toContain("chat.completion.chunk");
expect(text).toContain("chatcmpl-content");
expect(text).toContain('"content":"Hi"');
expect(session.getSpecialSettings()).toBeNull();
});

test("流式 Responses SSE：带 finish_reason 的 Chat Completions chunk 应保留", async () => {
const { ResponseFixer } = await import("./index");

const session = createSession();
session.originalFormat = "response";
const chatChunk = {
id: "chatcmpl-finish",
object: "chat.completion.chunk",
created: 1780753978,
model: "gpt-5.5",
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
};

const fixed = await ResponseFixer.process(
session,
createSseResponse([`data: ${JSON.stringify(chatChunk)}`, ""])
);
const text = await fixed.text();

expect(text).toContain("chat.completion.chunk");
expect(text).toContain("chatcmpl-finish");
expect(text).toContain("finish_reason");
expect(session.getSpecialSettings()).toBeNull();
});

test("流式 Responses SSE：带 usage 的 Chat Completions chunk 应保留", async () => {
const { ResponseFixer } = await import("./index");

const session = createSession();
session.originalFormat = "response";
const chatChunk = {
id: "chatcmpl-usage",
object: "chat.completion.chunk",
created: 1780753978,
model: "gpt-5.5",
choices: [{ index: 0, delta: { role: "assistant", content: "" } }],
usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
};

const fixed = await ResponseFixer.process(
session,
createSseResponse([`data: ${JSON.stringify(chatChunk)}`, ""])
);
const text = await fixed.text();

expect(text).toContain("chat.completion.chunk");
expect(text).toContain("chatcmpl-usage");
expect(text).toContain("prompt_tokens");
expect(session.getSpecialSettings()).toBeNull();
});

test("流式非 Responses SSE：空 Chat Completions chunk 应保留", async () => {
const { ResponseFixer } = await import("./index");

const session = createSession();
session.originalFormat = "chat";
const emptyChatChunk = {
id: "chatcmpl-chat-format",
object: "chat.completion.chunk",
created: 1780753978,
model: "gpt-5.5",
choices: [{ index: 0, delta: { role: "assistant", content: "" } }],
};

const fixed = await ResponseFixer.process(
session,
createSseResponse([`data: ${JSON.stringify(emptyChatChunk)}`, ""])
);
const text = await fixed.text();

expect(text).toContain("chat.completion.chunk");
expect(text).toContain("chatcmpl-chat-format");
expect(session.getSpecialSettings()).toBeNull();
});

test("流式 SSE：无换行且超过 maxFixSize 时应降级输出，避免无限缓冲", async () => {
const { ResponseFixer } = await import("./index");

Expand Down
Loading