From 40a2d9de33f796962e2c3e808ddb2024dd1c3dc6 Mon Sep 17 00:00:00 2001
From: ROOOO <liam.oh.work@gmail.com>
Date: Fri, 14 Aug 2026 08:08:36 -0400
Subject: [PATCH] fix(proxy): decode ArrayBuffer bodies for Responses WebSocket

Symptom:
- Remote Compaction v2 requests with previous_response_id skipped the upstream WebSocket path and fell back to HTTP, where they were rejected.

Cause:
- Raw passthrough request bodies are stored as ArrayBuffer values.
- Responses WebSocket body decoding did not support ArrayBuffer, so the upstream WebSocket frame was never sent.

Solution:
- Decode ArrayBuffer request bodies before constructing the upstream WebSocket frame.
- Add regression coverage ensuring Remote Compaction v2 continues over WebSocket.
---
 src/app/v1/_lib/proxy/forwarder.ts            |  2 +
 ...rwarder-raw-passthrough-regression.test.ts | 78 +++++++++++++++++++
 2 files changed, 80 insertions(+)

diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts
index 84bc26942..5be712d44 100644
--- a/src/app/v1/_lib/proxy/forwarder.ts
+++ b/src/app/v1/_lib/proxy/forwarder.ts
@@ -163,6 +163,8 @@ function decodeRequestBodyAsJson(body: BodyInit | undefined): Record<string, unk
     text = body;
   } else if (Buffer.isBuffer(body)) {
     text = body.toString("utf8");
+  } else if (body instanceof ArrayBuffer) {
+    text = Buffer.from(body).toString("utf8");
   } else if (body instanceof Uint8Array) {
     text = Buffer.from(body).toString("utf8");
   }
diff --git a/tests/unit/proxy/proxy-forwarder-raw-passthrough-regression.test.ts b/tests/unit/proxy/proxy-forwarder-raw-passthrough-regression.test.ts
index 67495901f..8bea59b9f 100644
--- a/tests/unit/proxy/proxy-forwarder-raw-passthrough-regression.test.ts
+++ b/tests/unit/proxy/proxy-forwarder-raw-passthrough-regression.test.ts
@@ -11,6 +11,11 @@ const mocks = vi.hoisted(() => ({
     getAgent: vi.fn(),
     markOriginUnhealthy: vi.fn(),
   })),
+  evaluateResponsesWsEligibility: vi.fn(async () => ({
+    isWebsocketClient: false,
+    eligible: false,
+  })),
+  tryResponsesWebsocketUpstream: vi.fn(),
 }));
 
 vi.mock("@/lib/config", async (importOriginal) => {
@@ -27,6 +32,24 @@ vi.mock("@/lib/proxy-agent", () => ({
   getGlobalAgentPool: mocks.getGlobalAgentPool,
 }));
 
+vi.mock("@/app/v1/_lib/responses-ws/eligibility", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/app/v1/_lib/responses-ws/eligibility")>();
+  return {
+    ...actual,
+    evaluateResponsesWsEligibility: mocks.evaluateResponsesWsEligibility,
+    getResponsesWsSessionId: vi.fn(() => "client-ws-session"),
+  };
+});
+
+vi.mock("@/app/v1/_lib/responses-ws/upstream-adapter", async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import("@/app/v1/_lib/responses-ws/upstream-adapter")>();
+  return {
+    ...actual,
+    tryResponsesWebsocketUpstream: mocks.tryResponsesWebsocketUpstream,
+  };
+});
+
 import { resolveEndpointPolicy } from "@/app/v1/_lib/proxy/endpoint-policy";
 import { ProxyForwarder } from "@/app/v1/_lib/proxy/forwarder";
 import { rectifyResponseInput } from "@/app/v1/_lib/proxy/response-input-rectifier";
@@ -124,6 +147,11 @@ function readBodyText(body: BodyInit | undefined): string | null {
 describe("ProxyForwarder raw passthrough regression", () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    mocks.evaluateResponsesWsEligibility.mockResolvedValue({
+      isWebsocketClient: false,
+      eligible: false,
+    });
+    mocks.tryResponsesWebsocketUpstream.mockReset();
   });
 
   it("raw passthrough 应优先保留原始请求体字节，而不是重新 JSON.stringify", async () => {
@@ -189,6 +217,56 @@ describe("ProxyForwarder raw passthrough regression", () => {
     expect(await response.text()).toBe(upstreamSse);
   });
 
+  it("remote compaction v2 ArrayBuffer 请求体仍通过上游 Responses WebSocket", async () => {
+    const requestBody = {
+      model: "gpt-5.5",
+      stream: true,
+      previous_response_id: "resp_previous",
+      input: [{ type: "compaction_trigger" }],
+    };
+    const originalBody = JSON.stringify(requestBody);
+    const upstreamSse =
+      'event: response.completed\ndata: {"type":"response.completed","response":{"id":"resp_compact","status":"completed"}}\n\n';
+    const session = createRawPassthroughSession(originalBody, {
+      "x-codex-beta-features": "remote_compaction_v2",
+    });
+    session.requestUrl = new URL("https://proxy.example.com/v1/responses");
+    const provider = createProvider();
+
+    mocks.evaluateResponsesWsEligibility.mockResolvedValue({
+      isWebsocketClient: true,
+      eligible: true,
+      endpointId: null,
+    });
+    mocks.tryResponsesWebsocketUpstream.mockResolvedValue({
+      response: new Response(upstreamSse, {
+        status: 200,
+        headers: { "content-type": "text/event-stream" },
+      }),
+      connected: true,
+      reused: true,
+    });
+
+    const fetchWithoutAutoDecode = vi.spyOn(ProxyForwarder as any, "fetchWithoutAutoDecode");
+    fetchWithoutAutoDecode.mockImplementationOnce(
+      async () => new Response("unexpected HTTP fallback", { status: 500 })
+    );
+    const { doForward } = ProxyForwarder as unknown as {
+      doForward: (session: ProxySession, provider: Provider, baseUrl: string) => Promise<Response>;
+    };
+
+    const response = await doForward(session, provider, provider.url);
+
+    expect(mocks.tryResponsesWebsocketUpstream).toHaveBeenCalledWith(
+      expect.objectContaining({
+        body: requestBody,
+        sessionId: "client-ws-session",
+      })
+    );
+    expect(fetchWithoutAutoDecode).not.toHaveBeenCalled();
+    expect(await response.text()).toBe(upstreamSse);
+  });
+
   it("remote compaction v2 将单对象 input 规范化后再透传", async () => {
     const originalBody = '{"model":"gpt-5.5","stream":true,"input":{"type":"compaction_trigger"}}';
     const session = createRawPassthroughSession(originalBody, {