From 1edbbc9dd72c6aeb73d9fd2f2288fb6700ffa248 Mon Sep 17 00:00:00 2001 From: xingzhou Date: Tue, 25 Aug 2026 07:41:36 +0800 Subject: [PATCH] fix(ui): prevent oversized WebRTC SDP answers from hanging Talk setup (#119295) * fix(ui): bound WebRTC SDP answer reads and release failed bodies * fix(ui): scope WebRTC SDP limits to OpenAI * fix(ui): release bounded response readers * test(ui): split WebRTC SDP browser coverage * fix(ui): reject unsafe declared response lengths * test(ui): reuse pending WebRTC SDP response fixture * test(ui): use native response for pending SDP body * style(ui): align pending SDP fixture formatting * test(ui): capture WebRTC SDP alert proof --------- Co-authored-by: Patrick Erichsen --- .../realtime-quicksilver-session.test.ts | 1 + .../openai/realtime-quicksilver-session.ts | 1 + .../realtime-voice-browser-auth.test.ts | 1 + extensions/openai/realtime-voice-provider.ts | 1 + src/talk/provider-types.ts | 1 + ui/src/app/custom-theme.test.ts | 21 +++ ui/src/app/custom-theme.ts | 50 +----- .../e2e/browser-talk-start-stop.fixtures.ts | 115 +++++++++++++ ...ser-talk-start-stop.webrtc-sdp.e2e.test.ts | 114 +++++++++++++ ui/src/lib/response-body.ts | 63 +++++++ ui/src/pages/chat/realtime-talk-shared.ts | 1 + .../chat/realtime-talk-webrtc-support.test.ts | 158 ++++++++++++++++++ .../chat/realtime-talk-webrtc-support.ts | 12 +- .../pages/chat/realtime-talk-webrtc.test.ts | 58 ++++--- 14 files changed, 522 insertions(+), 75 deletions(-) create mode 100644 ui/src/e2e/browser-talk-start-stop.webrtc-sdp.e2e.test.ts create mode 100644 ui/src/lib/response-body.ts diff --git a/extensions/openai/realtime-quicksilver-session.test.ts b/extensions/openai/realtime-quicksilver-session.test.ts index cfd38b83a402..adad83595e8b 100644 --- a/extensions/openai/realtime-quicksilver-session.test.ts +++ b/extensions/openai/realtime-quicksilver-session.test.ts @@ -169,6 +169,7 @@ describe("GPT-Live offer broker", () => { } expect(reservation).not.toHaveProperty("model"); expect(reservation).not.toHaveProperty("voice"); + expect(reservation.offerResponseMaxBytes).toBe(256 * 1024); const response = createResponseHarness(); const handling = realtime.handler( createRequest({ diff --git a/extensions/openai/realtime-quicksilver-session.ts b/extensions/openai/realtime-quicksilver-session.ts index 052c93a1ef10..676875c3b8b4 100644 --- a/extensions/openai/realtime-quicksilver-session.ts +++ b/extensions/openai/realtime-quicksilver-session.ts @@ -347,6 +347,7 @@ export function createOpenAIQuicksilverBrowserSessionBroker(params: { transport: "webrtc", clientSecret: token, offerUrl: OPENAI_QUICKSILVER_OFFER_PATH, + offerResponseMaxBytes: 256 * 1024, ...(request.gaSideband ? {} : { model, voice }), expiresAt, }; diff --git a/extensions/openai/realtime-voice-browser-auth.test.ts b/extensions/openai/realtime-voice-browser-auth.test.ts index 95de657acfdd..21e5d9cb7b9b 100644 --- a/extensions/openai/realtime-voice-browser-auth.test.ts +++ b/extensions/openai/realtime-voice-browser-auth.test.ts @@ -307,6 +307,7 @@ describe("OpenAI realtime voice browser authentication", () => { transport: "webrtc", clientSecret: "client-secret-123", offerUrl: "https://api.openai.com/v1/realtime/calls", + offerResponseMaxBytes: 256 * 1024, model: "gpt-realtime-2.1", expiresAt: 1_765_000_000_000, }); diff --git a/extensions/openai/realtime-voice-provider.ts b/extensions/openai/realtime-voice-provider.ts index c61988d5a3e5..df08c897948c 100644 --- a/extensions/openai/realtime-voice-provider.ts +++ b/extensions/openai/realtime-voice-provider.ts @@ -320,6 +320,7 @@ async function createOpenAIRealtimeBrowserSession( transport: "webrtc", clientSecret: clientSecret.value, offerUrl: "https://api.openai.com/v1/realtime/calls", + offerResponseMaxBytes: 256 * 1024, ...(offerHeaders ? { offerHeaders } : {}), model, voice, diff --git a/src/talk/provider-types.ts b/src/talk/provider-types.ts index ed44a7545d3a..7905f81193c4 100644 --- a/src/talk/provider-types.ts +++ b/src/talk/provider-types.ts @@ -271,6 +271,7 @@ type RealtimeVoiceBrowserWebRtcSdpSession = { clientSecret: string; offerUrl?: string; offerHeaders?: Record; + offerResponseMaxBytes?: number; model?: string; voice?: string; expiresAt?: number; diff --git a/ui/src/app/custom-theme.test.ts b/ui/src/app/custom-theme.test.ts index 1b552c7ec2d0..bcd982dbf489 100644 --- a/ui/src/app/custom-theme.test.ts +++ b/ui/src/app/custom-theme.test.ts @@ -123,6 +123,27 @@ describe("custom theme import helpers", () => { ).rejects.toThrow("too large"); }); + it("rejects unsafe tweakcn content-length values before acquiring the body reader", async () => { + const cancel = vi.fn(() => Promise.resolve()); + const getReader = vi.fn(() => { + throw new Error("reader should not be acquired"); + }); + const response = { + ok: true, + status: 200, + headers: new Headers({ "content-length": "9007199254740993" }), + body: { cancel, getReader }, + url: "", + } as unknown as Response; + const fetchImpl = vi.fn(async () => response) as unknown as typeof fetch; + + await expect( + importCustomThemeFromUrl("https://tweakcn.com/themes/cmlhfpjhw000004l4f4ax3m7z", fetchImpl), + ).rejects.toThrow("too large"); + expect(getReader).not.toHaveBeenCalled(); + expect(cancel).toHaveBeenCalledOnce(); + }); + it("rejects tweakcn theme responses without a bounded body stream", async () => { const response = createResponse(JSON.stringify(createTweakcnPayload()), { body: null }); const fetchImpl = vi.fn(async () => response) as unknown as typeof fetch; diff --git a/ui/src/app/custom-theme.ts b/ui/src/app/custom-theme.ts index 04378ae15f73..cfd6a71a3ced 100644 --- a/ui/src/app/custom-theme.ts +++ b/ui/src/app/custom-theme.ts @@ -2,6 +2,7 @@ import { asNullableRecord as readThemeRecord } from "@openclaw/normalization-cor // Control UI module implements custom theme behavior. import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import { readResponseTextWithLimit } from "../lib/response-body.ts"; const TWEAKCN_HOSTS = new Set(["tweakcn.com", "www.tweakcn.com"]); const THEME_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/; @@ -473,51 +474,12 @@ function assertTweakcnResponseUrl(value: string | undefined) { } } -function parseContentLength(headers: Headers): number | null { - const raw = headers.get("content-length"); - if (!raw) { - return null; - } - const parsed = Number(raw); - return Number.isFinite(parsed) && parsed >= 0 ? parsed : null; -} - -async function readResponseTextWithLimit(response: Response): Promise { - const contentLength = parseContentLength(response.headers); - if (contentLength != null && contentLength > MAX_TWEAKCN_THEME_BYTES) { - throw new Error("tweakcn theme payload is too large."); - } - - if (!response.body) { - throw new Error("tweakcn returned an unreadable theme payload."); - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let bytes = 0; - let text = ""; - try { - while (true) { - const chunk = await reader.read(); - if (chunk.done) { - break; - } - bytes += chunk.value.byteLength; - if (bytes > MAX_TWEAKCN_THEME_BYTES) { - await reader.cancel().catch(() => undefined); - throw new Error("tweakcn theme payload is too large."); - } - text += decoder.decode(chunk.value, { stream: true }); - } - text += decoder.decode(); - return text; - } finally { - reader.releaseLock(); - } -} - async function readJsonResponseWithLimit(response: Response): Promise { - const text = await readResponseTextWithLimit(response); + const text = await readResponseTextWithLimit(response, { + maxBytes: MAX_TWEAKCN_THEME_BYTES, + tooLargeMessage: "tweakcn theme payload is too large.", + missingBodyMessage: "tweakcn returned an unreadable theme payload.", + }); try { return JSON.parse(text) as unknown; } catch { diff --git a/ui/src/e2e/browser-talk-start-stop.fixtures.ts b/ui/src/e2e/browser-talk-start-stop.fixtures.ts index 60a8baa3fbe0..db126d52ee95 100644 --- a/ui/src/e2e/browser-talk-start-stop.fixtures.ts +++ b/ui/src/e2e/browser-talk-start-stop.fixtures.ts @@ -2,6 +2,19 @@ import { mkdir } from "node:fs/promises"; import path from "node:path"; import type { Page } from "playwright"; +export type WebRtcSdpE2eProof = { + bodyCancelCount: number; + bodyCancelResolvedCount: number; + fetchCount: number; + remoteDescriptionCount: number; + statuses: number[]; +}; + +type WebRtcSdpResponseFixture = { + body: string; + status: number; +}; + export function videoTalkCatalog(activeProvider: "google" | "openai") { return { realtime: { @@ -91,6 +104,100 @@ export async function installTalkBrowserFixtures(page: Page) { }); } +async function installWebRtcSdpResponseFixture(page: Page, fixture: WebRtcSdpResponseFixture) { + await page.addInitScript(() => { + const proofWindow = window as Window & { openclawWebRtcSdpE2e?: WebRtcSdpE2eProof }; + proofWindow.openclawWebRtcSdpE2e = { + bodyCancelCount: 0, + bodyCancelResolvedCount: 0, + fetchCount: 0, + remoteDescriptionCount: 0, + statuses: [], + }; + const originalFetch = window.fetch.bind(window); + window.fetch = async (input, init) => { + const response = await originalFetch(input, init); + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (!url.includes("api.openai.com/v1/realtime/calls")) { + return response; + } + const proof = proofWindow.openclawWebRtcSdpE2e; + if (!proof || !response.body) { + return response; + } + proof.fetchCount += 1; + proof.statuses.push(response.status); + const originalCancel = response.body.cancel.bind(response.body); + response.body.cancel = async (reason) => { + proof.bodyCancelCount += 1; + try { + return await originalCancel(reason); + } finally { + proof.bodyCancelResolvedCount += 1; + } + }; + return response; + }; + + class FakeDataChannel extends EventTarget { + readyState = "open"; + send() {} + close() { + this.readyState = "closed"; + } + } + + class FakePeerConnection extends EventTarget { + connectionState = "new"; + sctp = { maxMessageSize: 256 * 1024 }; + channel = new FakeDataChannel(); + addTrack() {} + createDataChannel() { + return this.channel; + } + async createOffer() { + return { type: "offer" as const, sdp: "offer-sdp" }; + } + async setLocalDescription() {} + async setRemoteDescription() { + const proof = proofWindow.openclawWebRtcSdpE2e; + if (proof) { + proof.remoteDescriptionCount += 1; + } + } + close() { + this.connectionState = "closed"; + } + } + + Object.defineProperty(window, "RTCPeerConnection", { + configurable: true, + value: FakePeerConnection, + }); + }); + await page.route("https://api.openai.com/v1/realtime/calls", async (route) => { + await route.fulfill({ + status: fixture.status, + contentType: "application/sdp", + body: fixture.body, + }); + }); +} + +export async function installWebRtcSdpFailureFixture(page: Page) { + await installWebRtcSdpResponseFixture(page, { + status: 502, + body: "provider failure", + }); +} + +export async function installOversizedWebRtcSdpFixture(page: Page) { + await installWebRtcSdpResponseFixture(page, { + status: 200, + body: "x".repeat(256 * 1024 + 1), + }); +} + export async function captureComposerProof(page: Page, fileName: string) { const artifactDir = path.join(process.cwd(), ".artifacts", "control-ui-e2e", "voice-controls"); await mkdir(artifactDir, { recursive: true }); @@ -107,6 +214,14 @@ export async function captureVideoTalkProof(page: Page, fileName: string) { .screenshot({ path: path.join(artifactDir, fileName) }); } +export async function captureWebRtcSdpAlertProof(page: Page, fileName: string) { + const artifactDir = path.join(process.cwd(), ".artifacts", "control-ui-e2e", "webrtc-sdp"); + await mkdir(artifactDir, { recursive: true }); + await page + .locator('.agent-chat__talk-status[role="alert"]') + .screenshot({ path: path.join(artifactDir, fileName) }); +} + export async function installBlockedMicrophoneFixture(page: Page) { await page.addInitScript(() => { Object.defineProperty(navigator, "mediaDevices", { diff --git a/ui/src/e2e/browser-talk-start-stop.webrtc-sdp.e2e.test.ts b/ui/src/e2e/browser-talk-start-stop.webrtc-sdp.e2e.test.ts new file mode 100644 index 000000000000..ee6c25f01698 --- /dev/null +++ b/ui/src/e2e/browser-talk-start-stop.webrtc-sdp.e2e.test.ts @@ -0,0 +1,114 @@ +// Control UI E2E tests cover WebRTC SDP response handling through a real page. +import { expect, it } from "vitest"; +import { installMockGateway } from "../test-helpers/control-ui-e2e.ts"; +import { + captureWebRtcSdpAlertProof, + installOversizedWebRtcSdpFixture, + installWebRtcSdpFailureFixture, + type WebRtcSdpE2eProof, + videoTalkCatalog, +} from "./browser-talk-start-stop.fixtures.ts"; +import { createControlUiE2eSuite } from "./control-ui-e2e-suite.test-support.ts"; + +const suite = createControlUiE2eSuite({ + name: "Control UI browser Talk WebRTC SDP responses", + browserLaunchOptions: { + args: ["--use-fake-device-for-media-stream", "--use-fake-ui-for-media-stream"], + }, +}); + +suite.define(() => { + it("cancels a failed OpenAI WebRTC SDP response body in the live Control UI", async () => { + await suite.withPage({ permissions: ["microphone"] }, async ({ page }) => { + const gateway = await installMockGateway(page, { + methodResponses: { + "talk.catalog": videoTalkCatalog("openai"), + "talk.client.create": { + provider: "openai", + voiceSessionId: "voice-openai-sdp-error-e2e", + transport: "webrtc", + clientSecret: "test-client-secret", + offerUrl: "https://api.openai.com/v1/realtime/calls", + offerResponseMaxBytes: 256 * 1024, + }, + }, + }); + await installWebRtcSdpFailureFixture(page); + + await page.goto(`${suite.server.baseUrl}chat`); + await page.getByRole("button", { name: "Start voice input" }).click(); + await gateway.waitForRequest("talk.client.create"); + + const alert = page.locator('.agent-chat__talk-status[role="alert"]'); + await expect.poll(() => alert.textContent()).toContain("Realtime WebRTC setup failed (502)"); + await captureWebRtcSdpAlertProof(page, "01-http-failure-alert.png"); + await expect + .poll(() => + page.evaluate( + () => + (window as Window & { openclawWebRtcSdpE2e?: WebRtcSdpE2eProof }) + .openclawWebRtcSdpE2e, + ), + ) + .toEqual({ + bodyCancelCount: 1, + bodyCancelResolvedCount: 1, + fetchCount: 1, + remoteDescriptionCount: 0, + statuses: [502], + }); + console.info( + `[webrtc-sdp-e2e] trigger=OpenAI WebRTC offer; transition=status:error+502; ` + + `body.cancel=1/resolved; outcome=visible setup failure`, + ); + }); + }); + + it("rejects and cancels an oversized OpenAI SDP answer before peer setup", async () => { + await suite.withPage({ permissions: ["microphone"] }, async ({ page }) => { + const gateway = await installMockGateway(page, { + methodResponses: { + "talk.catalog": videoTalkCatalog("openai"), + "talk.client.create": { + provider: "openai", + voiceSessionId: "voice-openai-sdp-oversized-e2e", + transport: "webrtc", + clientSecret: "test-client-secret", + offerUrl: "https://api.openai.com/v1/realtime/calls", + offerResponseMaxBytes: 256 * 1024, + }, + }, + }); + await installOversizedWebRtcSdpFixture(page); + + await page.goto(`${suite.server.baseUrl}chat`); + await page.getByRole("button", { name: "Start voice input" }).click(); + await gateway.waitForRequest("talk.client.create"); + + const alert = page.locator('.agent-chat__talk-status[role="alert"]'); + await expect + .poll(() => alert.textContent()) + .toContain("Realtime WebRTC SDP answer: text response exceeds 262144 bytes"); + await captureWebRtcSdpAlertProof(page, "02-oversized-answer-alert.png"); + await expect + .poll(() => + page.evaluate( + () => + (window as Window & { openclawWebRtcSdpE2e?: WebRtcSdpE2eProof }) + .openclawWebRtcSdpE2e, + ), + ) + .toEqual({ + bodyCancelCount: 1, + bodyCancelResolvedCount: 1, + fetchCount: 1, + remoteDescriptionCount: 0, + statuses: [200], + }); + console.info( + `[webrtc-sdp-e2e] trigger=oversized OpenAI SDP answer; ` + + `body.cancel=1/resolved; remote-description=0; outcome=visible size failure`, + ); + }); + }); +}); diff --git a/ui/src/lib/response-body.ts b/ui/src/lib/response-body.ts new file mode 100644 index 000000000000..429e913570f8 --- /dev/null +++ b/ui/src/lib/response-body.ts @@ -0,0 +1,63 @@ +// Control UI response helpers own bounded browser body consumption. + +type ResponseTextLimitOptions = { + maxBytes: number; + tooLargeMessage: string; + missingBodyMessage?: string; +}; + +function parseContentLength(headers: Headers): number | null { + const raw = headers.get("content-length"); + if (!raw) { + return null; + } + if (!/^\d+$/u.test(raw)) { + return null; + } + const parsed = Number(raw); + return Number.isSafeInteger(parsed) ? parsed : Number.POSITIVE_INFINITY; +} + +export async function readResponseTextWithLimit( + response: Response, + options: ResponseTextLimitOptions, +): Promise { + const contentLength = parseContentLength(response.headers); + if (contentLength !== null && contentLength > options.maxBytes) { + void response.body?.cancel().catch(() => undefined); + throw new Error(options.tooLargeMessage); + } + if (!response.body) { + if (options.missingBodyMessage) { + throw new Error(options.missingBodyMessage); + } + return ""; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const chunks: string[] = []; + let totalBytes = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) { + const tail = decoder.decode(); + if (tail) { + chunks.push(tail); + } + break; + } + totalBytes += value.byteLength; + if (totalBytes > options.maxBytes) { + // Cancellation is best-effort; finally always releases this reader's lock. + void reader.cancel().catch(() => undefined); + throw new Error(options.tooLargeMessage); + } + chunks.push(decoder.decode(value, { stream: true })); + } + } finally { + reader.releaseLock(); + } + return chunks.join(""); +} diff --git a/ui/src/pages/chat/realtime-talk-shared.ts b/ui/src/pages/chat/realtime-talk-shared.ts index b77909759f33..566a20cf3878 100644 --- a/ui/src/pages/chat/realtime-talk-shared.ts +++ b/ui/src/pages/chat/realtime-talk-shared.ts @@ -50,6 +50,7 @@ export type RealtimeTalkWebRtcSdpSessionResult = { clientSecret: string; offerUrl?: string; offerHeaders?: Record; + offerResponseMaxBytes?: number; model?: string; voice?: string; expiresAt?: number; diff --git a/ui/src/pages/chat/realtime-talk-webrtc-support.test.ts b/ui/src/pages/chat/realtime-talk-webrtc-support.test.ts index 30ae19d98131..22f8ea1ea516 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc-support.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc-support.test.ts @@ -2,6 +2,28 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { RealtimeTalkWebRtcOfferExchange } from "./realtime-talk-webrtc-support.ts"; +const OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES = 256 * 1024; + +function readAnswer( + exchange: RealtimeTalkWebRtcOfferExchange, + isCurrent = () => true, + offerResponseMaxBytes: number | null = OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES, + provider = "openai", +) { + return exchange.readAnswer({ + session: { + provider, + transport: "webrtc", + clientSecret: "reservation-token", + offerUrl: "https://gateway.example.test/realtime/calls", + ...(offerResponseMaxBytes === null ? {} : { offerResponseMaxBytes }), + }, + offer: { type: "offer", sdp: "offer-sdp" }, + gatewayUrl: "wss://gateway.example.test/control", + isCurrent, + }); +} + describe("RealtimeTalkWebRtcOfferExchange", () => { afterEach(() => { vi.unstubAllGlobals(); @@ -34,4 +56,140 @@ describe("RealtimeTalkWebRtcOfferExchange", () => { }), ); }); + + it("accepts an SDP answer at the 256 KiB boundary", async () => { + const answer = "x".repeat(OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES); + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(answer)), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange)).resolves.toBe(answer); + }); + + it("preserves oversized SDP answers when the provider declares no limit", async () => { + const answer = "x".repeat(OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES + 1); + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(answer)), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange, () => true, null, "openai")).resolves.toBe(answer); + }); + + it("honors a response limit declared by a custom provider", async () => { + const answer = "x".repeat(OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES + 1); + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(answer)), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect( + readAnswer(exchange, () => true, OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES, "custom-provider"), + ).rejects.toThrow("Realtime WebRTC SDP answer: text response exceeds 262144 bytes"); + }); + + it("rejects and cancels a streamed SDP answer over the 256 KiB boundary", async () => { + const cancel = vi.fn(() => Promise.resolve()); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES)); + controller.enqueue(new Uint8Array(1)); + }, + cancel, + }); + vi.stubGlobal( + "fetch", + vi.fn( + async () => + ({ + ok: true, + status: 200, + headers: new Headers(), + body, + text: vi.fn(), + }) as unknown as Response, + ), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange)).rejects.toThrow( + "Realtime WebRTC SDP answer: text response exceeds 262144 bytes", + ); + expect(cancel).toHaveBeenCalledOnce(); + expect(body.locked).toBe(false); + }); + + it.each([String(OPENAI_REALTIME_SDP_ANSWER_MAX_BYTES + 1), "9007199254740993"])( + "rejects a declared oversized SDP answer of %s before acquiring its body reader", + async (contentLength) => { + const cancel = vi.fn(() => Promise.resolve()); + const getReader = vi.fn(() => { + throw new Error("reader should not be acquired"); + }); + vi.stubGlobal( + "fetch", + vi.fn( + async () => + ({ + ok: true, + status: 200, + headers: new Headers({ + "content-length": contentLength, + }), + body: { cancel, getReader }, + text: vi.fn(), + }) as unknown as Response, + ), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange)).rejects.toThrow( + "Realtime WebRTC SDP answer: text response exceeds 262144 bytes", + ); + expect(getReader).not.toHaveBeenCalled(); + expect(cancel).toHaveBeenCalledOnce(); + }, + ); + + it("cancels a non-2xx SDP response body without waiting for cancellation", async () => { + const cancel = vi.fn(() => new Promise(() => {})); + vi.stubGlobal( + "fetch", + vi.fn( + async () => + ({ + ok: false, + status: 502, + body: { cancel }, + }) as unknown as Response, + ), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange)).rejects.toThrow("Realtime WebRTC setup failed (502)"); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it("cancels a stale successful SDP response body", async () => { + const cancel = vi.fn(() => Promise.resolve()); + vi.stubGlobal( + "fetch", + vi.fn( + async () => + ({ + ok: true, + status: 200, + body: { cancel }, + }) as unknown as Response, + ), + ); + const exchange = new RealtimeTalkWebRtcOfferExchange(); + + await expect(readAnswer(exchange, () => false)).resolves.toBeUndefined(); + expect(cancel).toHaveBeenCalledOnce(); + }); }); diff --git a/ui/src/pages/chat/realtime-talk-webrtc-support.ts b/ui/src/pages/chat/realtime-talk-webrtc-support.ts index 298f91fe1163..c71296158801 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc-support.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc-support.ts @@ -1,5 +1,6 @@ // Control UI chat module owns low-level WebRTC offer and media-message helpers. import { normalizeRealtimeVoiceResponseOutcome } from "../../../../src/talk/provider-types.js"; +import { readResponseTextWithLimit } from "../../lib/response-body.ts"; import type { RealtimeTalkWebRtcSdpSessionResult } from "./realtime-talk-shared.ts"; import type { RealtimeTalkVideoFrame } from "./realtime-talk-video.ts"; @@ -143,14 +144,23 @@ export class RealtimeTalkWebRtcOfferExchange { throw error; } if (!params.isCurrent()) { + void response.body?.cancel().catch(() => undefined); return undefined; } if (!response.ok) { + void response.body?.cancel().catch(() => undefined); throw new Error(`Realtime WebRTC setup failed (${response.status})`); } let answer: string; try { - answer = await response.text(); + const maxBytes = params.session.offerResponseMaxBytes; + answer = + maxBytes === undefined + ? await response.text() + : await readResponseTextWithLimit(response, { + maxBytes, + tooLargeMessage: `Realtime WebRTC SDP answer: text response exceeds ${maxBytes} bytes`, + }); } catch (error) { if (!params.isCurrent()) { return undefined; diff --git a/ui/src/pages/chat/realtime-talk-webrtc.test.ts b/ui/src/pages/chat/realtime-talk-webrtc.test.ts index 12cabe58852b..2d6b69bc1b71 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.test.ts @@ -78,6 +78,23 @@ function stubAnswerSdpFetch(): void { vi.stubGlobal("fetch", vi.fn(async () => new Response("answer-sdp")) as unknown as typeof fetch); } +function createPendingSdpResponse(signal: AbortSignal | undefined): Response { + return new Response( + new ReadableStream({ + start(controller) { + signal?.addEventListener( + "abort", + () => { + const reason = signal?.reason; + controller.error(reason instanceof Error ? reason : new Error("offer request aborted")); + }, + { once: true }, + ); + }, + }), + ); +} + function createOpenAiTransport( client: Record = {}, callbacks: Record = {}, @@ -88,6 +105,7 @@ function createOpenAiTransport( provider: "openai", transport: "webrtc", clientSecret: "client-secret-123", + offerResponseMaxBytes: 256 * 1024, }, { client: client as never, @@ -368,23 +386,11 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { it("aborts stalled WebRTC SDP answer body reads after the offer timeout", async () => { vi.useFakeTimers(); let offerSignal: AbortSignal | undefined; + let response: Response | undefined; const fetchMock = vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { offerSignal = init?.signal ?? undefined; - return { - ok: true, - status: 200, - text: () => - new Promise((_, reject) => { - offerSignal?.addEventListener( - "abort", - () => { - const reason = offerSignal?.reason; - reject(reason instanceof Error ? reason : new Error("offer request aborted")); - }, - { once: true }, - ); - }), - } as Response; + response = createPendingSdpResponse(offerSignal); + return response; }); vi.stubGlobal("fetch", fetchMock as unknown as typeof fetch); const transport = createOpenAiTransport(); @@ -396,6 +402,7 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { await waitForFast(() => expect(fetchMock).toHaveBeenCalledTimes(1)); expect(offerSignal?.aborted).toBe(false); + expect(response?.body?.locked).toBe(true); await vi.runAllTimersAsync(); @@ -403,38 +410,29 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { new Error("Realtime WebRTC offer request timed out after 30000ms"), ); expect(offerSignal?.aborted).toBe(true); + expect(response?.body?.locked).toBe(false); }); it("aborts a pending WebRTC SDP answer body read when stopped", async () => { let offerSignal: AbortSignal | undefined; + let response: Response | undefined; const fetchMock = vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { offerSignal = init?.signal ?? undefined; - return { - ok: true, - status: 200, - text: () => - new Promise((_, reject) => { - offerSignal?.addEventListener( - "abort", - () => { - const reason = offerSignal?.reason; - reject(reason instanceof Error ? reason : new Error("offer request aborted")); - }, - { once: true }, - ); - }), - } as Response; + response = createPendingSdpResponse(offerSignal); + return response; }); vi.stubGlobal("fetch", fetchMock as unknown as typeof fetch); const transport = createOpenAiTransport(); const startResult = transport.start(); await waitForFast(() => expect(fetchMock).toHaveBeenCalledTimes(1)); + expect(response?.body?.locked).toBe(true); transport.stop(); await expect(startResult).resolves.toBe("cancelled"); expect(offerSignal?.aborted).toBe(true); + expect(response?.body?.locked).toBe(false); }); it("reports a closed candidate when the peer fails during final setup", async () => {