fix(zalo): bound stalled inbound media header waits (#104578)

* fix(zalo): bound stalled inbound media header waits

* test(zalo): type inbound media mock to runtime contract

* test(zalo): return complete saved media metadata

---------

Co-authored-by: NIO <nocodet@mail.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/zalo/src/monitor.image.polling.test.ts

@@ -152,4 +152,88 @@ describe("Zalo polling image handling", () => {
     abort.abort();
     await run;
   });
+
+  it("times out inbound image downloads when photo_url headers never arrive", async () => {
+    const { createServer } = await import("node:http");
+    const { saveRemoteMedia } = await import("openclaw/plugin-sdk/media-runtime");
+    const { ZALO_MEDIA_READ_IDLE_TIMEOUT_MS, ZALO_MEDIA_RESPONSE_HEADER_TIMEOUT_MS } =
+      await import("./monitor.js");
+
+    const server = createServer((_req, _res) => {
+      // Accept the connection but never write status/headers.
+    });
+    await new Promise<void>((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(0, "127.0.0.1", () => {
+        server.off("error", reject);
+        resolve();
+      });
+    });
+    const address = server.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected loopback TCP address");
+    }
+    const stallUrl = `http://127.0.0.1:${address.port}/stall.jpg`;
+    const headerTimeoutMs = 250;
+
+    // Production monitor passes the full timeout budget; the harness shortens
+    // only the actual fetch so the stalled-header case stays fast.
+    const saveRemoteMediaWithHeaderTimeout: typeof saveRemoteMedia = async (params) => {
+      expect(params).toEqual({
+        url: stallUrl,
+        maxBytes: 5 * 1024 * 1024,
+        responseHeaderTimeoutMs: ZALO_MEDIA_RESPONSE_HEADER_TIMEOUT_MS,
+        readIdleTimeoutMs: ZALO_MEDIA_READ_IDLE_TIMEOUT_MS,
+      });
+      return await saveRemoteMedia({
+        ...params,
+        responseHeaderTimeoutMs: headerTimeoutMs,
+        ssrfPolicy: { ...params.ssrfPolicy, dangerouslyAllowPrivateNetwork: true },
+      });
+    };
+    saveRemoteMediaMock.mockImplementation(saveRemoteMediaWithHeaderTimeout);
+
+    getUpdatesMock
+      .mockResolvedValueOnce({
+        ok: true,
+        result: createImageUpdate({
+          caption: "stalled photo",
+          photoUrl: stallUrl,
+        }),
+      })
+      .mockImplementation(() => new Promise(() => {}));
+
+    const { monitorZaloProvider } = await loadCachedLifecycleMonitorModule("zalo-image-polling");
+    const abort = new AbortController();
+    const runtime = createRuntimeEnv();
+    const { account, config } = createLifecycleMonitorSetup({
+      accountId: "default",
+      dmPolicy: "open",
+    });
+    const started = Date.now();
+    const run = monitorZaloProvider({
+      token: "zalo-token", // pragma: allowlist secret
+      account,
+      config,
+      runtime,
+      abortSignal: abort.signal,
+    });
+
+    await vi.waitFor(() => expect(finalizeInboundContextMock).toHaveBeenCalledTimes(1));
+    const elapsedMs = Date.now() - started;
+    expect(elapsedMs).toBeGreaterThanOrEqual(headerTimeoutMs - 50);
+    expect(elapsedMs).toBeLessThan(headerTimeoutMs + 5_000);
+    expect(finalizeInboundContextMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        BodyForAgent: "stalled photo\n\n[zalo image attachment unavailable]",
+        MediaPath: undefined,
+      }),
+    );
+
+    abort.abort();
+    await run;
+    await new Promise<void>((resolve) => {
+      server.close(() => resolve());
+    });
+  });
 });

extensions/zalo/src/monitor.ts

@@ -49,6 +49,11 @@ import {
   tryHandleHostedZaloMediaRequest,
 } from "./outbound-media.js";
 
+/** Default idle timeout for Zalo inbound photo downloads (30 seconds). */
+export const ZALO_MEDIA_READ_IDLE_TIMEOUT_MS = 30_000;
+/** Maximum wait for Zalo inbound photo response headers (120 seconds). */
+export const ZALO_MEDIA_RESPONSE_HEADER_TIMEOUT_MS = 120_000;
+
 type ZaloMonitorOptions = {
   token: string;
   account: ResolvedZaloAccount;
@@ -378,7 +383,14 @@ async function handleImageMessage(params: ZaloImageMessageParams): Promise<void>
   if (photo_url) {
     try {
       const maxBytes = mediaMaxMb * 1024 * 1024;
-      const saved = await core.channel.media.saveRemoteMedia({ url: photo_url, maxBytes });
+      // Without header/idle deadlines, a stalled photo_url host can block inbound
+      // image preprocessing indefinitely (idle timeout never starts).
+      const saved = await core.channel.media.saveRemoteMedia({
+        url: photo_url,
+        maxBytes,
+        responseHeaderTimeoutMs: ZALO_MEDIA_RESPONSE_HEADER_TIMEOUT_MS,
+        readIdleTimeoutMs: ZALO_MEDIA_READ_IDLE_TIMEOUT_MS,
+      });
       mediaPath = saved.path;
       mediaType = saved.contentType;
     } catch (err) {

extensions/zalo/src/test-support/lifecycle-test-support.ts

@@ -174,10 +174,16 @@ export function createImageLifecycleCore() {
     buffer: Buffer.from("image-bytes"),
     contentType: "image/jpeg",
   }));
-  const saveRemoteMediaMock = vi.fn(async () => ({
-    path: "/tmp/zalo-photo.jpg",
-    contentType: "image/jpeg",
-  }));
+  // Keep the mock arity aligned with PluginRuntime.saveRemoteMedia so
+  // mockImplementation callbacks that inspect timeout options typecheck.
+  const saveRemoteMediaMock = vi.fn<PluginRuntime["channel"]["media"]["saveRemoteMedia"]>(
+    async (_params) => ({
+      id: "zalo-photo.jpg",
+      path: "/tmp/zalo-photo.jpg",
+      size: Buffer.byteLength("image-bytes"),
+      contentType: "image/jpeg",
+    }),
+  );
   const saveMediaBufferMock = vi.fn(async () => ({
     path: "/tmp/zalo-photo.jpg",
     contentType: "image/jpeg",
@@ -377,6 +383,8 @@ export function expectImageLifecycleDelivery(params: {
   expect(saveRemoteMediaMock).toHaveBeenCalledWith({
     url: photoUrl,
     maxBytes: 5 * 1024 * 1024,
+    responseHeaderTimeoutMs: 120_000,
+    readIdleTimeoutMs: 30_000,
   });
   expect(params.saveMediaBufferMock).not.toHaveBeenCalled();
   expect(params.finalizeInboundContextMock).toHaveBeenCalledWith(