fix(zai): keep probe deadline through stalled error-body reads (#109026)

* fix(zai): keep probe deadline through stalled error-body reads

Fixes an issue where Z.AI endpoint detection could hang indefinitely
when a probe returned error headers and then stalled on the error
response body. The previous helper cleared the AbortSignal as soon as
fetch resolved headers, so a never-chunking error body bypassed the
probe deadline.

Keep one AbortController deadline across the probe request and the
bounded error-body read, and pass a clamped chunkTimeoutMs into
readResponseWithLimit so a stalled error body fails closed inside the
remaining budget. The chunkTimeoutMs is clamped to the remaining
deadline after fetch returns headers, so the idle timeout does not
outlast the overall probe deadline.

Tighten the stalled-body test timing assertion from 5s to 2x timeoutMs
to verify the deadline is actually respected.

* fix(zai): enforce probe body deadline

Co-authored-by: NIO <noreply@github.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
Co-authored-by: NIO <noreply@github.com>

extensions/zai/detect.test.ts

@@ -335,4 +335,73 @@ describe("detectZaiEndpoint", () => {
       }),
     ).rejects.toThrow(/exceeded size limit/);
   });
+
+  it("fails closed when a probe error body stalls without chunks", async () => {
+    // Headers return 400, but the error body never enqueues. Without
+    // the whole-body deadline the probe would hang indefinitely.
+    const fetchFn = (async (url: string) => {
+      if (url !== "https://api.z.ai/api/paas/v4/chat/completions") {
+        throw new Error(`unexpected url: ${url}`);
+      }
+      const body = new ReadableStream<Uint8Array>({
+        start() {
+          // Intentionally never enqueue or close — idle timeout must fire.
+        },
+      });
+      return new Response(body, {
+        status: 400,
+        headers: { "content-type": "application/json" },
+      });
+    }) as typeof fetch;
+
+    const timeoutMs = 80;
+    const startedAt = Date.now();
+    const detected = await detectZaiEndpoint({
+      apiKey: "sk-test", // pragma: allowlist secret
+      endpoint: "global",
+      timeoutMs,
+      fetchFn,
+    });
+    const elapsedMs = Date.now() - startedAt;
+
+    expect(detected).toBeNull();
+    // The probe must fail within the deadline budget, not hang indefinitely.
+    // Allow 2× the timeout for scheduling overhead; a hang would take seconds.
+    expect(elapsedMs).toBeLessThan(2 * timeoutMs);
+  });
+
+  it("keeps one probe deadline through a slow-drip error body", async () => {
+    const state = { cancelled: false };
+    let interval: ReturnType<typeof setInterval> | undefined;
+    const fetchFn = (async () => {
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          interval = setInterval(() => controller.enqueue(new Uint8Array([123])), 10);
+        },
+        cancel() {
+          state.cancelled = true;
+          if (interval) {
+            clearInterval(interval);
+          }
+        },
+      });
+      return new Response(body, {
+        status: 400,
+        headers: { "content-type": "application/json" },
+      });
+    }) as typeof fetch;
+
+    const timeoutMs = 80;
+    const startedAt = Date.now();
+    const detected = await detectZaiEndpoint({
+      apiKey: "sk-test", // pragma: allowlist secret
+      endpoint: "global",
+      timeoutMs,
+      fetchFn,
+    });
+
+    expect(detected).toBeNull();
+    expect(Date.now() - startedAt).toBeLessThan(3 * timeoutMs);
+    expect(state.cancelled).toBe(true);
+  });
 });

extensions/zai/detect.ts

@@ -1,5 +1,9 @@
 // Zai plugin module implements detect behavior.
 import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
+import {
+  createProviderOperationDeadline,
+  createProviderOperationTimeoutResolver,
+} from "openclaw/plugin-sdk/provider-http";
 import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
 import {
   ZAI_CN_BASE_URL,
@@ -60,21 +64,6 @@ function isUnsupportedModelResult(result: ProbeResult): boolean {
   );
 }
 
-async function fetchWithTimeoutLocal(
-  fetchFn: typeof fetch,
-  url: string,
-  init: RequestInit,
-  timeoutMs: number,
-): Promise<Response> {
-  const controller = new AbortController();
-  const timeout = setTimeout(() => controller.abort(), timeoutMs);
-  try {
-    return await fetchFn(url, { ...init, signal: controller.signal });
-  } finally {
-    clearTimeout(timeout);
-  }
-}
-
 async function probeZaiChatCompletions(params: {
   baseUrl: string;
   apiKey: string;
@@ -82,26 +71,34 @@ async function probeZaiChatCompletions(params: {
   timeoutMs: number;
   fetchFn?: typeof fetch;
 }): Promise<ProbeResult> {
+  const deadline = createProviderOperationDeadline({
+    timeoutMs: params.timeoutMs,
+    label: "Z.AI endpoint probe",
+  });
+  const resolveTimeoutMs = createProviderOperationTimeoutResolver({
+    deadline,
+    defaultTimeoutMs: params.timeoutMs,
+  });
+  const controller = new AbortController();
+  const timeout = setTimeout(() => controller.abort(), params.timeoutMs);
+  timeout.unref?.();
+  let res: Response | undefined;
   try {
     const fetchFn = params.fetchFn ?? globalThis.fetch;
-    const res = await fetchWithTimeoutLocal(
-      fetchFn,
-      `${params.baseUrl}/chat/completions`,
-      {
-        method: "POST",
-        headers: {
-          authorization: `Bearer ${params.apiKey}`,
-          "content-type": "application/json",
-        },
-        body: JSON.stringify({
-          model: params.modelId,
-          stream: false,
-          max_tokens: 1,
-          messages: [{ role: "user", content: "ping" }],
-        }),
+    res = await fetchFn(`${params.baseUrl}/chat/completions`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${params.apiKey}`,
+        "content-type": "application/json",
       },
-      params.timeoutMs,
-    );
+      body: JSON.stringify({
+        model: params.modelId,
+        stream: false,
+        max_tokens: 1,
+        messages: [{ role: "user", content: "ping" }],
+      }),
+      signal: controller.signal,
+    });
 
     if (res.ok) {
       return { ok: true };
@@ -111,6 +108,11 @@ async function probeZaiChatCompletions(params: {
     let errorMessage: string | undefined;
     try {
       const bytes = await readResponseWithLimit(res, ZAI_DETECT_ERROR_BODY_MAX_BYTES, {
+        // Resolve immediately before body consumption so headers and every
+        // body shape share one operation budget, including slow-drip streams.
+        timeoutMs: resolveTimeoutMs,
+        onTimeout: ({ timeoutMs }) =>
+          new Error(`Z.AI probe error body timed out after ${timeoutMs}ms`),
         onOverflow: ({ maxBytes }) =>
           new Error(`Z.AI probe error body exceeded size limit (${maxBytes} bytes)`),
       });
@@ -131,12 +133,17 @@ async function probeZaiChatCompletions(params: {
         errorMessage = msg;
       }
     } catch {
-      // ignore malformed error bodies
+      // ignore malformed / stalled / oversized error bodies
     }
 
     return { ok: false, status: res.status, errorCode, errorMessage };
   } catch {
     return { ok: false };
+  } finally {
+    clearTimeout(timeout);
+    if (res?.bodyUsed !== true) {
+      await res?.body?.cancel().catch(() => undefined);
+    }
   }
 }