fix(sms): retry transient MMS failures before durable adoption (#118994)

* fix(sms): retry transient MMS downloads before durable adoption

* test(sms): align durable ingress harness with typed runtime contracts

extensions/sms/src/ingress-spool.test.ts

@@ -1,12 +1,16 @@
 // Sms tests cover durable Twilio webhook admission and replay.
+import { createHmac } from "node:crypto";
 import { mkdtemp, realpath, rm } from "node:fs/promises";
+import { createServer } from "node:http";
 import os from "node:os";
 import path from "node:path";
+import { saveRemoteMedia } from "openclaw/plugin-sdk/media-runtime";
 import { createChannelIngressQueueForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { SmsChannelRuntime } from "./inbound.js";
 import { createSmsIngressSpool } from "./ingress-spool.js";
 import type { ResolvedSmsAccount } from "./types.js";
+import { createSmsWebhookHandler } from "./webhook.js";
 
 type SmsIngressPayload = {
   version: 1;
@@ -64,6 +68,22 @@ async function drainSpool(spool: SmsIngressSpool): Promise<void> {
   await spool.waitForIdle();
 }
 
+async function listenSmsTestServer(server: ReturnType<typeof createServer>): Promise<string> {
+  await new Promise<void>((resolve) => {
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  disposers.push(async () => {
+    await new Promise<void>((resolve) => {
+      server.close(() => resolve());
+    });
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("expected a loopback HTTP server address");
+  }
+  return `http://127.0.0.1:${address.port}`;
+}
+
 afterEach(async () => {
   for (const dispose of disposers.splice(0).toReversed()) {
     await dispose();
@@ -74,6 +94,173 @@ afterEach(async () => {
 });
 
 describe("createSmsIngressSpool", () => {
+  it("retries acknowledged MMS provider outages before adopting later same-sender messages", async () => {
+    const stateDir = await createStateDir();
+    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
+    disposers.push(() => {
+      vi.unstubAllEnvs();
+    });
+    const mediaBytes = Buffer.from(
+      "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/a0cAAAAASUVORK5CYII=",
+      "base64",
+    );
+    let mediaRequests = 0;
+    let authenticatedMediaRequests = 0;
+    const mediaOrigin = await listenSmsTestServer(
+      createServer((req, res) => {
+        mediaRequests += 1;
+        if (req.headers.authorization?.startsWith("Basic ")) {
+          authenticatedMediaRequests += 1;
+        }
+        if (mediaRequests === 1) {
+          res.statusCode = 429;
+          res.end("Twilio rate limited");
+          return;
+        }
+        res.setHeader("content-type", "image/png");
+        res.end(mediaBytes);
+      }),
+    );
+    const sender = "+15551234567";
+    const messageSid = `MM${"a".repeat(32)}`;
+    const testAccount = {
+      ...account,
+      accountSid: `AC${"c".repeat(32)}`,
+      dmPolicy: "allowlist" as const,
+      allowFrom: [sender],
+    };
+    const deliveries: Array<{ id: string; body: string; attachments: number }> = [];
+    const channelRuntime = {
+      commands: {
+        shouldComputeCommandAuthorized: () => false,
+        isControlCommandMessage: () => false,
+      },
+      pairing: { readAllowFromStore: async () => [] },
+      routing: {
+        resolveAgentRoute: () => ({
+          agentId: "main",
+          accountId: account.accountId,
+          sessionKey: `agent:main:sms:direct:${sender}`,
+        }),
+      },
+      media: {
+        saveRemoteMedia: async (options: Parameters<typeof saveRemoteMedia>[0]) =>
+          await saveRemoteMedia({
+            ...options,
+            fetchImpl: async (_url, init) =>
+              await fetch(`${mediaOrigin}/twilio-media`, {
+                headers: init?.headers,
+                signal: init?.signal,
+              }),
+          }),
+      },
+      inbound: {
+        buildContext: (input: Parameters<SmsChannelRuntime["inbound"]["buildContext"]>[0]) => {
+          deliveries.push({
+            id: String(input.extra?.MessageSid),
+            body: input.message.bodyForAgent ?? input.message.rawBody,
+            attachments: input.media?.length ?? 0,
+          });
+          return {};
+        },
+        run: async (input: Parameters<SmsChannelRuntime["inbound"]["run"]>[0]) => {
+          const turnInput = await input.adapter.ingest(input.raw);
+          if (!turnInput) {
+            throw new Error("expected normalized SMS turn");
+          }
+          await input.adapter.resolveTurn(
+            turnInput,
+            { kind: "message", canStartAgentTurn: true },
+            {},
+          );
+          await input.turnAdoptionLifecycle?.onAdopted();
+        },
+      },
+      reply: {},
+      session: {},
+    } as unknown as SmsChannelRuntime;
+    const queue = createQueue(stateDir);
+    const spool = createSmsIngressSpool({
+      cfg: {},
+      account: testAccount,
+      channelRuntime,
+      queue,
+      log: { warn: () => undefined },
+    });
+    disposers.push(spool.stop);
+    const webhookHandler = createSmsWebhookHandler({
+      cfg: {},
+      account: testAccount,
+      ingress: spool,
+    });
+    const webhookOrigin = await listenSmsTestServer(
+      createServer((req, res) => {
+        void webhookHandler(req, res).catch((error: unknown) => {
+          res.statusCode = 500;
+          res.end(error instanceof Error ? error.message : String(error));
+        });
+      }),
+    );
+    testAccount.publicWebhookUrl = `${webhookOrigin}/webhooks/sms`;
+    spool.start();
+
+    async function postSignedCallback(callback: Record<string, string>) {
+      const signatureData =
+        testAccount.publicWebhookUrl +
+        Object.keys(callback)
+          .toSorted()
+          .map((key) => `${key}${callback[key] ?? ""}`)
+          .join("");
+      const signature = createHmac("sha1", testAccount.authToken)
+        .update(signatureData)
+        .digest("base64");
+      return await fetch(testAccount.publicWebhookUrl, {
+        method: "POST",
+        headers: { "x-twilio-signature": signature },
+        body: new URLSearchParams(callback),
+      });
+    }
+
+    const mediaCallback = {
+      ...form(messageSid),
+      AccountSid: testAccount.accountSid,
+      Body: "keep this attachment",
+      NumMedia: "1",
+      MediaUrl0: `https://api.twilio.com/2010-04-01/Accounts/${testAccount.accountSid}/Messages/${messageSid}/Media/ME${"b".repeat(32)}`,
+      MediaContentType0: "image/png",
+    };
+    const firstResponse = await postSignedCallback(mediaCallback);
+    expect(firstResponse.status).toBe(200);
+    expect(firstResponse.headers.get("x-openclaw-delivery-accepted")).toBe("durable");
+    await vi.waitFor(async () => {
+      expect(mediaRequests).toBe(1);
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({ id: messageSid, lastError: expect.stringContaining("HTTP 429") }),
+      ]);
+    });
+    expect(deliveries).toEqual([]);
+
+    const secondSid = `SM${"d".repeat(32)}`;
+    const secondResponse = await postSignedCallback({
+      ...form(secondSid),
+      AccountSid: testAccount.accountSid,
+      Body: "the later message",
+    });
+    expect(secondResponse.status).toBe(200);
+    await vi.waitFor(() => expect(deliveries).toHaveLength(2), { timeout: 5_000 });
+    expect(deliveries).toEqual([
+      { id: messageSid, body: "keep this attachment", attachments: 1 },
+      { id: secondSid, body: "the later message", attachments: 0 },
+    ]);
+    expect(mediaRequests).toBe(2);
+    expect(authenticatedMediaRequests).toBe(2);
+    expect(await spool.enqueue(mediaCallback)).toMatchObject({
+      kind: "completed",
+      duplicate: true,
+    });
+    expect(await queue.listPending()).toEqual([]);
+  });
+
   it("recovers an uncompleted message with a fresh drain instance", async () => {
     const stateDir = await createStateDir();
     const first = createSmsIngressSpool({

extensions/sms/src/media.test.ts

@@ -1,13 +1,17 @@
 // Sms tests cover outbound MMS media hosting behavior.
 import fs from "node:fs";
 import path from "node:path";
-import type { unlinkIfExists as unlinkIfExistsType } from "openclaw/plugin-sdk/media-runtime";
+import {
+  MediaFetchError,
+  type unlinkIfExists as unlinkIfExistsType,
+} from "openclaw/plugin-sdk/media-runtime";
 import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime";
 import type {
   OpenKeyedStoreOptions,
   PluginStateKeyedStore,
 } from "openclaw/plugin-sdk/plugin-state-runtime";
 import { createPluginStateKeyedStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { SsrFBlockedError } from "openclaw/plugin-sdk/security-runtime";
 import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
 import type { loadWebMedia as loadWebMediaType } from "openclaw/plugin-sdk/web-media";
 import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -80,7 +84,8 @@ const TWILIO_MMS_FILENAME_CASES = [
 vi.mock("openclaw/plugin-sdk/web-media", () => ({
   loadWebMedia: loadWebMediaMock,
 }));
-vi.mock("openclaw/plugin-sdk/media-runtime", () => ({
+vi.mock("openclaw/plugin-sdk/media-runtime", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("openclaw/plugin-sdk/media-runtime")>()),
   unlinkIfExists: unlinkIfExistsMock,
 }));
 
@@ -594,6 +599,91 @@ describe("SMS inbound MMS materialization", () => {
     unlinkIfExistsMock.mockClear();
   });
 
+  async function expectInboundMediaFailure(error: MediaFetchError, retryable: boolean) {
+    const pending = materializeSmsInboundMedia({
+      account: createAccount(),
+      msg: {
+        accountSid: ACCOUNT_SID,
+        from: "+15551234567",
+        to: "+15557654321",
+        body: "keep this caption",
+        messageSid: MESSAGE_SID,
+        media: [{ url: twilioMediaUrl(), contentType: "image/jpeg" }],
+      },
+      mediaRuntime: {
+        media: {
+          saveRemoteMedia: async () => {
+            throw error;
+          },
+        },
+      } as never,
+    });
+
+    if (retryable) {
+      await expect(pending).rejects.toBe(error);
+    } else {
+      await expect(pending).resolves.toMatchObject({
+        body: "keep this caption\n\n[1 Twilio MMS attachment unavailable]",
+        media: [],
+      });
+    }
+  }
+
+  it.each([
+    [408, true],
+    [429, true],
+    [500, true],
+    [502, true],
+    [503, true],
+    [504, true],
+    [400, false],
+    [401, false],
+    [403, false],
+    [404, false],
+    [410, false],
+  ] as const)("classifies Twilio HTTP %i before durable adoption", async (status, retryable) => {
+    await expectInboundMediaFailure(
+      new MediaFetchError("http_error", `Twilio returned ${status}`, { status }),
+      retryable,
+    );
+  });
+
+  it.each([
+    {
+      name: "nested connection reset",
+      cause: new Error("fetch failed", {
+        cause: Object.assign(new Error("reset"), { code: "ECONNRESET" }),
+      }),
+      retryable: true,
+    },
+    {
+      name: "media request deadline",
+      cause: new DOMException("timed out", "TimeoutError"),
+      retryable: true,
+    },
+    {
+      name: "blocked SSRF with nested transient error",
+      cause: Object.assign(new SsrFBlockedError("blocked private address"), {
+        cause: Object.assign(new Error("reset"), { code: "ECONNRESET" }),
+      }),
+      retryable: false,
+    },
+    {
+      name: "local storage permission",
+      cause: Object.assign(new Error("permission denied"), { code: "EACCES" }),
+      retryable: false,
+    },
+  ])("classifies $name without poisoning a sender lane", async ({ cause, retryable }) => {
+    await expectInboundMediaFailure(
+      new MediaFetchError("fetch_failed", "download failed", { cause }),
+      retryable,
+    );
+  });
+
+  it("keeps oversized MMS visible without retrying its sender lane", async () => {
+    await expectInboundMediaFailure(new MediaFetchError("max_bytes", "too large"), false);
+  });
+
   it("keeps the message visible when declared attachments exceed the download bound", async () => {
     const saveRemoteMedia = vi.fn();
 
@@ -786,43 +876,51 @@ describe("SMS inbound MMS materialization", () => {
     expect(saveRemoteMedia).not.toHaveBeenCalled();
   });
 
-  it("cleans already-saved files when a later attachment aborts the batch", async () => {
-    const abortController = new AbortController();
-    const abortReason = new Error("SMS ingress claim superseded");
-    const saveRemoteMedia = vi
-      .fn()
-      .mockResolvedValueOnce({
-        path: "/tmp/first.jpg",
-        size: 128,
-        contentType: "image/jpeg",
-      })
-      .mockImplementationOnce(async () => {
-        abortController.abort(abortReason);
-        throw abortReason;
-      });
+  it.each(["claim cancellation", "retryable provider failure"])(
+    "cleans already-saved files when a later attachment ends with %s",
+    async (failureKind) => {
+      const abortController = new AbortController();
+      const abortReason =
+        failureKind === "claim cancellation"
+          ? new Error("SMS ingress claim superseded")
+          : new MediaFetchError("http_error", "Twilio temporarily unavailable", { status: 503 });
+      const saveRemoteMedia = vi
+        .fn()
+        .mockResolvedValueOnce({
+          path: "/tmp/first.jpg",
+          size: 128,
+          contentType: "image/jpeg",
+        })
+        .mockImplementationOnce(async () => {
+          if (failureKind === "claim cancellation") {
+            abortController.abort(abortReason);
+          }
+          throw abortReason;
+        });
 
-    await expect(
-      materializeSmsInboundMedia({
-        account: createAccount(),
-        msg: {
-          accountSid: ACCOUNT_SID,
-          from: "+15551234567",
-          to: "+15557654321",
-          body: "photos",
-          messageSid: MESSAGE_SID,
-          media: [
-            { url: twilioMediaUrl(), contentType: "image/jpeg" },
-            { url: twilioMediaUrl({ mediaSid: OTHER_MEDIA_SID }), contentType: "image/jpeg" },
-          ],
-        },
-        mediaRuntime: { media: { saveRemoteMedia } } as never,
-        abortSignal: abortController.signal,
-      }),
-    ).rejects.toBe(abortReason);
+      await expect(
+        materializeSmsInboundMedia({
+          account: createAccount(),
+          msg: {
+            accountSid: ACCOUNT_SID,
+            from: "+15551234567",
+            to: "+15557654321",
+            body: "photos",
+            messageSid: MESSAGE_SID,
+            media: [
+              { url: twilioMediaUrl(), contentType: "image/jpeg" },
+              { url: twilioMediaUrl({ mediaSid: OTHER_MEDIA_SID }), contentType: "image/jpeg" },
+            ],
+          },
+          mediaRuntime: { media: { saveRemoteMedia } } as never,
+          abortSignal: abortController.signal,
+        }),
+      ).rejects.toBe(abortReason);
 
-    expect(unlinkIfExistsMock).toHaveBeenCalledOnce();
-    expect(unlinkIfExistsMock).toHaveBeenCalledWith("/tmp/first.jpg");
-  });
+      expect(unlinkIfExistsMock).toHaveBeenCalledOnce();
+      expect(unlinkIfExistsMock).toHaveBeenCalledWith("/tmp/first.jpg");
+    },
+  );
 
   it("exposes idempotent cleanup for successfully materialized files", async () => {
     const result = await materializeSmsInboundMedia({

extensions/sms/src/media.ts

@@ -6,8 +6,13 @@ import {
   toInboundMediaFactsWithMetadata,
   type InboundMediaFacts,
 } from "openclaw/plugin-sdk/channel-inbound";
+import {
+  collectErrorGraphCandidates,
+  extractErrorCode,
+  readErrorName,
+} from "openclaw/plugin-sdk/error-runtime";
 import { extensionForMime } from "openclaw/plugin-sdk/media-mime";
-import { unlinkIfExists } from "openclaw/plugin-sdk/media-runtime";
+import { MediaFetchError, unlinkIfExists } from "openclaw/plugin-sdk/media-runtime";
 import { resolveExpiresAtMsFromDurationMs } from "openclaw/plugin-sdk/number-runtime";
 import {
   createHostedOutboundMediaStore,
@@ -17,7 +22,8 @@ import {
   type OutboundMediaLoadOptions,
 } from "openclaw/plugin-sdk/outbound-media";
 import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime";
-import { safeEqualSecret } from "openclaw/plugin-sdk/security-runtime";
+import { classifyTransientNetworkErrorCode } from "openclaw/plugin-sdk/retry-runtime";
+import { safeEqualSecret, SsrFBlockedError } from "openclaw/plugin-sdk/security-runtime";
 import { getSmsRuntime } from "./runtime.js";
 import { TWILIO_MMS_MAX_BYTES } from "./twilio.js";
 import type { ResolvedSmsAccount, SmsInboundMessage } from "./types.js";
@@ -340,6 +346,34 @@ function createInboundMediaCleanup(paths: string[]): () => Promise<void> {
   };
 }
 
+function isRetryableSmsInboundMediaError(error: unknown): boolean {
+  if (!(error instanceof MediaFetchError)) {
+    return false;
+  }
+  if (error.code === "http_error") {
+    return (
+      error.status === 408 ||
+      error.status === 429 ||
+      (typeof error.status === "number" && error.status >= 500)
+    );
+  }
+  if (error.code !== "fetch_failed") {
+    return false;
+  }
+  const causes = collectErrorGraphCandidates(error.cause, (candidate) => [candidate.cause]);
+  if (causes.some((candidate) => candidate instanceof SsrFBlockedError)) {
+    return false;
+  }
+  return causes.some((candidate) => {
+    const name = readErrorName(candidate);
+    return (
+      classifyTransientNetworkErrorCode(extractErrorCode(candidate)) !== undefined ||
+      name === "AbortError" ||
+      name.endsWith("TimeoutError")
+    );
+  });
+}
+
 export async function materializeSmsInboundMedia(params: {
   account: ResolvedSmsAccount;
   msg: SmsInboundMessage;
@@ -419,8 +453,13 @@ export async function materializeSmsInboundMedia(params: {
           contentType: saved.contentType ?? media.contentType,
           messageId: params.msg.messageSid,
         });
-      } catch {
+      } catch (error) {
         abortSignal.throwIfAborted();
+        // Adoption tombstones the callback, so only a pre-adoption throw lets
+        // the durable sender lane retry a transient Twilio media failure.
+        if (isRetryableSmsInboundMediaError(error)) {
+          throw error;
+        }
         unavailableCount += 1;
         params.log?.warn?.(
           `Failed to download Twilio MMS attachment ${index + 1} for ${params.msg.messageSid}`,