From 6194cfdfb2facb595fdb5fe9f5d4563aa0f9b78b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 3 Aug 2026 15:50:30 -0700 Subject: [PATCH] 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 | 187 +++++++++++++++++++++++ extensions/sms/src/media.test.ts | 172 ++++++++++++++++----- extensions/sms/src/media.ts | 45 +++++- 3 files changed, 364 insertions(+), 40 deletions(-) diff --git a/extensions/sms/src/ingress-spool.test.ts b/extensions/sms/src/ingress-spool.test.ts index 5b902cebc820..a76e8b2efe1b 100644 --- a/extensions/sms/src/ingress-spool.test.ts +++ b/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 { await spool.waitForIdle(); } +async function listenSmsTestServer(server: ReturnType): Promise { + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + disposers.push(async () => { + await new Promise((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[0]) => + await saveRemoteMedia({ + ...options, + fetchImpl: async (_url, init) => + await fetch(`${mediaOrigin}/twilio-media`, { + headers: init?.headers, + signal: init?.signal, + }), + }), + }, + inbound: { + buildContext: (input: Parameters[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[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) { + 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({ diff --git a/extensions/sms/src/media.test.ts b/extensions/sms/src/media.test.ts index 4869ba2c7d1e..cc82d2a9d243 100644 --- a/extensions/sms/src/media.test.ts +++ b/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()), 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({ diff --git a/extensions/sms/src/media.ts b/extensions/sms/src/media.ts index cb90e66d8763..8d7374e80bd1 100644 --- a/extensions/sms/src/media.ts +++ b/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 { }; } +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}`,