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 <masonxhuang@proton.me>

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<void>,
+): Promise<void> {
+  let requestCount = 0;
+  await withServer(
+    (_req, _res) => {
+      requestCount += 1;
+    },
+    async (baseUrl) => {
+      await run(`${baseUrl}/v1`, () => requestCount);
+    },
+  );
+}
+
 async function withMockedSpeechFetch(
   run: (fetchMock: ReturnType<typeof vi.fn>) => Promise<void>,
   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<string, unknown>;
+  timeoutMs: number;
+}): Promise<void> {
+  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<string, unknown>,
   providerId: string,
@@ -267,22 +307,17 @@ function buildTestOpenAISpeechProvider(): SpeechProviderPlugin {
     isConfigured: ({ providerConfig }) =>
       typeof (providerConfig as Record<string, unknown> | 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<string, unknown> | 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<string, unknown> | 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<void>((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<typeof setTimeout> | undefined;
+
+          try {
+            const result = await Promise.race([
+              testCase.run(cfg, timeoutMs),
+              new Promise<never>((_, 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);
+            }
+          }
+        });
+      },
+    );
   });
 }