From 851860339207ce8436b33b8f5a8375bfd73345bf Mon Sep 17 00:00:00 2001 From: xingzhou Date: Mon, 20 Jul 2026 21:02:39 +0800 Subject: [PATCH] fix(tts): contract speech requests hang on stalled endpoints (#110725) * fix(tts): time out stalled contract speech requests * fix(tts): release discarded contract responses * test(tts): reuse canonical HTTP server helper * test(tts): mark fixture credential synthetic --------- Co-authored-by: Mason Huang --- src/plugins/contracts/tts-contract-suites.ts | 227 +++++++++++++------ 1 file changed, 162 insertions(+), 65 deletions(-) diff --git a/src/plugins/contracts/tts-contract-suites.ts b/src/plugins/contracts/tts-contract-suites.ts index 97177d6f7e0..5cd940f367e 100644 --- a/src/plugins/contracts/tts-contract-suites.ts +++ b/src/plugins/contracts/tts-contract-suites.ts @@ -1,5 +1,4 @@ // TTS contract suites provide reusable text-to-speech plugin contract assertions. -import http from "node:http"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createEmptyPluginRegistry, @@ -7,7 +6,11 @@ import { setActivePluginRegistry, } from "openclaw/plugin-sdk/plugin-test-runtime"; import type { ResolvedTtsConfig, SpeechProviderPlugin } from "openclaw/plugin-sdk/speech-core"; -import { withEnv, withEnvAsync } from "openclaw/plugin-sdk/test-env"; +import { + fetchWithSsrFGuard, + ssrfPolicyFromHttpBaseUrlAllowedHostname, +} from "openclaw/plugin-sdk/ssrf-runtime"; +import { withEnv, withEnvAsync, withServer } from "openclaw/plugin-sdk/test-env"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import type { AssistantMessage, Model } from "../../llm/types.js"; import { resolveWorkspacePackagePublicModuleUrl } from "../../plugin-sdk/test-helpers/public-surface-loader.js"; @@ -168,6 +171,20 @@ function createAudioBuffer(length = 2): Buffer { return Buffer.from(new Uint8Array(length).fill(1)); } +async function withHangingSpeechServer( + run: (baseUrl: string, getRequestCount: () => number) => Promise, +): Promise { + let requestCount = 0; + await withServer( + (_req, _res) => { + requestCount += 1; + }, + async (baseUrl) => { + await run(`${baseUrl}/v1`, () => requestCount); + }, + ); +} + async function withMockedSpeechFetch( run: (fetchMock: ReturnType) => Promise, audioLength: number, @@ -189,6 +206,29 @@ function resolveBaseUrl(rawValue: unknown, fallback: string): string { return typeof rawValue === "string" && rawValue.trim() ? rawValue.replace(/\/+$/u, "") : fallback; } +async function requestTestOpenAISpeech(params: { + baseUrl: string; + body: Record; + timeoutMs: number; +}): Promise { + const requestUrl = `${params.baseUrl}/audio/speech`; + const { response, release } = await fetchWithSsrFGuard({ + url: requestUrl, + init: { + method: "POST", + body: JSON.stringify(params.body), + }, + timeoutMs: params.timeoutMs, + policy: ssrfPolicyFromHttpBaseUrlAllowedHostname(params.baseUrl), + auditContext: "tts-contract-openai", + }); + try { + await response.body?.cancel().catch(() => {}); + } finally { + await release(); + } +} + function resolveTestProviderConfig( rawConfig: Record, providerId: string, @@ -267,22 +307,17 @@ function buildTestOpenAISpeechProvider(): SpeechProviderPlugin { isConfigured: ({ providerConfig }) => typeof (providerConfig as Record | undefined)?.apiKey === "string" || typeof process.env.OPENAI_API_KEY === "string", - synthesize: async ({ text, providerConfig, providerOverrides }) => { + synthesize: async ({ text, providerConfig, providerOverrides, timeoutMs }) => { const config = providerConfig as Record | undefined; - const res = await fetch( - `${resolveBaseUrl(config?.baseUrl, "https://api.openai.com/v1")}/audio/speech`, - { - method: "POST", - body: JSON.stringify({ - input: text, - model: providerOverrides?.model ?? config?.model ?? "gpt-4o-mini-tts", - voice: providerOverrides?.voice ?? config?.voice ?? "alloy", - }), + await requestTestOpenAISpeech({ + baseUrl: resolveBaseUrl(config?.baseUrl, "https://api.openai.com/v1"), + body: { + input: text, + model: providerOverrides?.model ?? config?.model ?? "gpt-4o-mini-tts", + voice: providerOverrides?.voice ?? config?.voice ?? "alloy", }, - ); - // The contract only asserts the request, but an unread audio body pins the - // connection-pool socket — release it instead of leaking fds across calls. - await res.body?.cancel().catch(() => {}); + timeoutMs, + }); return { audioBuffer: createAudioBuffer(1), outputFormat: "mp3", @@ -290,7 +325,7 @@ function buildTestOpenAISpeechProvider(): SpeechProviderPlugin { voiceCompatible: true, }; }, - synthesizeTelephony: async ({ text, providerConfig }) => { + synthesizeTelephony: async ({ text, providerConfig, timeoutMs }) => { const config = providerConfig as Record | undefined; const configuredModel = typeof config?.model === "string" ? config.model : undefined; const model = configuredModel ?? "tts-1"; @@ -298,19 +333,16 @@ function buildTestOpenAISpeechProvider(): SpeechProviderPlugin { typeof config?.instructions === "string" ? config.instructions : undefined; const instructions = model === "gpt-4o-mini-tts" ? configuredInstructions || undefined : undefined; - const res = await fetch( - `${resolveBaseUrl(config?.baseUrl, "https://api.openai.com/v1")}/audio/speech`, - { - method: "POST", - body: JSON.stringify({ - input: text, - model, - voice: config?.voice ?? "alloy", - instructions, - }), + await requestTestOpenAISpeech({ + baseUrl: resolveBaseUrl(config?.baseUrl, "https://api.openai.com/v1"), + body: { + input: text, + model, + voice: config?.voice ?? "alloy", + instructions, }, - ); - await res.body?.cancel().catch(() => {}); + timeoutMs, + }); return { audioBuffer: createAudioBuffer(2), outputFormat: "mp3", @@ -1142,44 +1174,36 @@ export function describeTtsProviderRuntimeContract() { it("cancels the discarded speech response body after synthesize", async () => { await withIsolatedSpeechProviderEnvAsync({}, async () => { let sawConnectionClose = false; - const server = http.createServer((_req, res) => { - res.writeHead(200, { "content-type": "audio/mpeg" }); - res.write(Buffer.alloc(16)); - res.on("close", () => { - sawConnectionClose = true; - }); - // Intentionally never res.end(): an unread body must still be - // released by the caller, not left pinning the connection. - }); - await new Promise((resolve) => { - server.listen(0, "127.0.0.1", resolve); - }); - const address = server.address(); - const port = typeof address === "object" && address ? address.port : 0; - try { - const result = await ttsRuntime.synthesizeSpeech({ - text: "hello cancel", - cfg: asLegacyOpenClawConfig({ - agents: { defaults: { model: { primary: "openai/gpt-4o-mini" } } }, - messages: { - tts: { - provider: "openai", - openai: { - baseUrl: `http://127.0.0.1:${port}/v1`, - apiKey: "fixture-api-key", + await withServer( + (_req, res) => { + res.writeHead(200, { "content-type": "audio/mpeg" }); + res.write(Buffer.alloc(16)); + res.on("close", () => { + sawConnectionClose = true; + }); + // Intentionally never res.end(): an unread body must still be + // released by the caller, not left pinning the connection. + }, + async (baseUrl) => { + const result = await ttsRuntime.synthesizeSpeech({ + text: "hello cancel", + cfg: asLegacyOpenClawConfig({ + agents: { defaults: { model: { primary: "openai/gpt-4o-mini" } } }, + messages: { + tts: { + provider: "openai", + openai: { + baseUrl: `${baseUrl}/v1`, + apiKey: "fixture-api-key", + }, }, }, - }, - }), - }); - expect(result.success).toBe(true); - await vi.waitFor(() => expect(sawConnectionClose).toBe(true), { timeout: 5_000 }); - } finally { - server.closeAllConnections(); - await new Promise((resolve) => { - server.close(resolve); - }); - } + }), + }); + expect(result.success).toBe(true); + await vi.waitFor(() => expect(sawConnectionClose).toBe(true), { timeout: 5_000 }); + }, + ); }); }); @@ -1260,6 +1284,79 @@ export function describeTtsProviderRuntimeContract() { }, ); }); + + it.each([ + { + name: "ordinary synthesis", + run: async (cfg: OpenClawConfig, timeoutMs: number) => + await ttsRuntime.textToSpeech({ + text: "Hello from the timeout contract.", + cfg, + disableFallback: true, + timeoutMs, + }), + }, + { + name: "telephony synthesis", + run: async (cfg: OpenClawConfig, timeoutMs: number) => + await ttsRuntime.textToSpeechTelephony({ + text: "Hello from the telephony timeout contract.", + cfg, + timeoutMs, + }), + }, + ] as const)( + "aborts stalled OpenAI $name within the caller timeout", + { timeout: 2_000 }, + async (testCase) => { + await withHangingSpeechServer(async (baseUrl, getRequestCount) => { + const registry = createEmptyPluginRegistry(); + registry.speechProviders = [ + { pluginId: "openai", provider: buildTestOpenAISpeechProvider(), source: "test" }, + ]; + setActivePluginRegistry(registry); + const cfg = asLegacyTtsConfig({ + messages: { + tts: { + provider: "openai", + providers: { + openai: { + apiKey: "test-api-key", + baseUrl, + model: "gpt-4o-mini-tts", + voice: "alloy", + }, + }, + }, + }, + }); + const timeoutMs = 100; + const startedAt = Date.now(); + let watchdog: ReturnType | undefined; + + try { + const result = await Promise.race([ + testCase.run(cfg, timeoutMs), + new Promise((_, reject) => { + watchdog = setTimeout( + () => reject(new Error(`${testCase.name} did not time out`)), + 1_000, + ); + }), + ]); + + expect(result.success).toBe(false); + expect(result.error).toMatch(/aborted|timeout|timed out/i); + expect(Date.now() - startedAt).toBeLessThan(1_000); + expect(getRequestCount()).toBe(1); + } finally { + if (watchdog) { + clearTimeout(watchdog); + } + } + }); + }, + ); }); }