fix(sms): prevent silent loss of acknowledged Twilio messages (#109866)

* fix(sms): adopt durable ingress drain with ack gated on MessageSid enqueue

Twilio inbound webhooks acked 200 then dispatched detached behind a 10-minute
in-memory replay guard; Twilio does not reliably retry inbound messages, so a
crash between ack and dispatch silently lost the SMS. The raw Twilio form now
enqueues durably (event_id = MessageSid, lane per sender) before the TwiML
200; dispatch, retry, dead-letter, and tombstones run through the core drain
with a startup + interval pump for restart recovery. Payload/account
validation moved to parse-at-dispatch and dead-letters as non-retryable. The
replay guard and its saturation 429 path are deleted; 24h/20k tombstones
strictly cover the old 10min/10k window.

Autoreview P1 (ambiguous non-adopted settlement) rejected with evidence:
bindIngressLifecycleToReplyOptions threads onAdopted/onDeferred/onAbandoned
into reply admission, so deferral and abandonment settle via core lifecycle
callbacks; normal non-adopted return = terminal local handling, matching the
Telegram reference mapping (skipped -> completed). Part of #109657 wave 1.

* fix(sms): keep ingress payload type internal

* fix(sms): preserve Twilio message SID aliases

* fix(sms): preserve ingress ordering and timestamps

* test(sms): satisfy type and lint gates

extensions/sms/src/gateway.test.ts

@@ -4,6 +4,16 @@ import { startSmsGatewayAccount } from "./gateway.js";
 import type { SmsChannelRuntime } from "./inbound.js";
 import type { ResolvedSmsAccount } from "./types.js";
 
+const drainSmsIngress = vi.hoisted(() => vi.fn(async () => undefined));
+const disposeSmsIngress = vi.hoisted(() => vi.fn());
+const createSmsIngressSpool = vi.hoisted(() =>
+  vi.fn(() => ({
+    enqueue: vi.fn(),
+    drainOnce: drainSmsIngress,
+    dispose: disposeSmsIngress,
+  })),
+);
+
 const { registeredRoutes, registerPluginHttpRoute, waitUntilAbort } = vi.hoisted(() => {
   const routeCleanups: Array<() => void> = [];
   return {
@@ -19,6 +29,8 @@ const { registeredRoutes, registerPluginHttpRoute, waitUntilAbort } = vi.hoisted
 
 vi.mock("openclaw/plugin-sdk/channel-outbound", () => ({ waitUntilAbort }));
 
+vi.mock("./ingress-spool.js", () => ({ createSmsIngressSpool }));
+
 vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({
   createFixedWindowRateLimiter: () => ({
     clear: vi.fn(),
@@ -51,6 +63,9 @@ describe("startSmsGatewayAccount", () => {
   beforeEach(() => {
     registerPluginHttpRoute.mockClear();
     waitUntilAbort.mockClear();
+    createSmsIngressSpool.mockClear();
+    drainSmsIngress.mockClear();
+    disposeSmsIngress.mockClear();
   });
 
   afterEach(() => {

extensions/sms/src/gateway.ts

@@ -1,10 +1,12 @@
 // Sms plugin module implements gateway behavior.
 import { waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound";
 import { registerPluginHttpRoute } from "openclaw/plugin-sdk/webhook-ingress";
+import { createSmsIngressSpool } from "./ingress-spool.js";
 import type { ResolvedSmsAccount } from "./types.js";
 import { createSmsWebhookHandler, type SmsWebhookHandlerParams } from "./webhook.js";
 
 const CHANNEL_ID = "sms";
+const SMS_INGRESS_DRAIN_INTERVAL_MS = 500;
 
 const activeRoutes = new Map<string, () => void>();
 const activeRoutePaths = new Map<string, string>();
@@ -52,7 +54,8 @@ export function collectSmsStartupWarnings(account: ResolvedSmsAccount): string[]
 function registerSmsWebhookRoute(params: {
   cfg: SmsWebhookHandlerParams["cfg"];
   account: ResolvedSmsAccount;
-  channelRuntime: SmsWebhookHandlerParams["channelRuntime"];
+  channelRuntime: Parameters<typeof createSmsIngressSpool>[0]["channelRuntime"];
+  abortSignal: AbortSignal;
   log?: SmsGatewayLog;
 }): () => void {
   const key = routeKey(params.account);
@@ -65,29 +68,49 @@ function registerSmsWebhookRoute(params: {
   }
   activeRoutes.get(key)?.();
   activeRoutePaths.delete(webhookPath);
-  const unregister = registerPluginHttpRoute({
-    path: webhookPath,
-    auth: "plugin",
-    pluginId: CHANNEL_ID,
-    accountId: params.account.accountId,
-    log: (msg) => params.log?.info?.(msg),
-    handler: createSmsWebhookHandler(params),
-  });
-  activeRoutes.set(key, unregister);
-  activeRoutePaths.set(webhookPath, params.account.accountId);
-  return () => {
-    unregister();
-    activeRoutes.delete(key);
-    if (activeRoutePaths.get(webhookPath) === params.account.accountId) {
-      activeRoutePaths.delete(webhookPath);
-    }
+  const ingress = createSmsIngressSpool(params);
+  const requestDrain = () => {
+    void ingress.drainOnce().catch((error: unknown) => {
+      params.log?.error?.(
+        `SMS ingress drain failed: ${error instanceof Error ? error.message : String(error)}`,
+      );
+    });
   };
+  const drainTimer = setInterval(requestDrain, SMS_INGRESS_DRAIN_INTERVAL_MS);
+  drainTimer.unref?.();
+  try {
+    const unregisterRoute = registerPluginHttpRoute({
+      path: webhookPath,
+      auth: "plugin",
+      pluginId: CHANNEL_ID,
+      accountId: params.account.accountId,
+      log: (msg) => params.log?.info?.(msg),
+      handler: createSmsWebhookHandler({ ...params, ingress }),
+    });
+    const unregister = () => {
+      clearInterval(drainTimer);
+      ingress.dispose();
+      unregisterRoute();
+      activeRoutes.delete(key);
+      if (activeRoutePaths.get(webhookPath) === params.account.accountId) {
+        activeRoutePaths.delete(webhookPath);
+      }
+    };
+    activeRoutes.set(key, unregister);
+    activeRoutePaths.set(webhookPath, params.account.accountId);
+    requestDrain();
+    return unregister;
+  } catch (error) {
+    clearInterval(drainTimer);
+    ingress.dispose();
+    throw error;
+  }
 }
 
 export async function startSmsGatewayAccount(params: {
   cfg: SmsWebhookHandlerParams["cfg"];
   account: ResolvedSmsAccount;
-  channelRuntime: SmsWebhookHandlerParams["channelRuntime"];
+  channelRuntime: Parameters<typeof createSmsIngressSpool>[0]["channelRuntime"];
   abortSignal: AbortSignal;
   log?: SmsGatewayLog;
 }) {

extensions/sms/src/inbound.test.ts

@@ -38,6 +38,7 @@ function createRuntime() {
   const resolveAgentRoute = vi.fn();
   const run = vi.fn<
     (params: {
+      turnAdoptionLifecycle?: { onAdopted: () => void | Promise<void> };
       adapter: {
         ingest: (msg: {
           from: string;
@@ -93,6 +94,7 @@ describe("dispatchSmsInboundEvent", () => {
       cfg: {},
       account: createAccount(),
       channelRuntime: runtime,
+      receivedAt: 1_700_000_000_000,
       msg: {
         from: "+15551234567",
         to: "+15557654321",
@@ -130,6 +132,7 @@ describe("dispatchSmsInboundEvent", () => {
     });
     buildContext.mockReturnValue({ SessionKey: "agent:main:sms:direct:+15551234567" });
     resolveStorePath.mockReturnValue("/tmp/openclaw-sessions");
+    const turnAdoptionLifecycle = { onAdopted: vi.fn(async () => undefined) };
 
     await dispatchSmsInboundEvent({
       cfg: {},
@@ -138,6 +141,8 @@ describe("dispatchSmsInboundEvent", () => {
         allowFrom: ["+15551234567"],
       }),
       channelRuntime: runtime,
+      receivedAt: 1_700_000_000_123,
+      turnAdoptionLifecycle,
       msg: {
         from: "+15551234567",
         to: "+15557654321",
@@ -148,6 +153,7 @@ describe("dispatchSmsInboundEvent", () => {
     });
 
     const runParams = expectDefined(run.mock.calls[0]?.[0], "SMS inbound run parameters");
+    expect(runParams.turnAdoptionLifecycle).toBe(turnAdoptionLifecycle);
     const ingested = runParams.adapter.ingest({
       from: "+15551234567",
       to: "+15557654321",
@@ -164,6 +170,7 @@ describe("dispatchSmsInboundEvent", () => {
     );
     expect(buildContext).toHaveBeenCalledWith(
       expect.objectContaining({
+        timestamp: 1_700_000_000_123,
         from: "sms:+15551234567",
         sender: expect.objectContaining({ id: "+15551234567" }),
         conversation: expect.objectContaining({ id: "+15551234567" }),

extensions/sms/src/inbound.ts

@@ -89,6 +89,10 @@ export async function dispatchSmsInboundEvent(params: {
   account: ResolvedSmsAccount;
   msg: SmsInboundMessage;
   channelRuntime: SmsChannelRuntime;
+  receivedAt: number;
+  turnAdoptionLifecycle?: NonNullable<
+    Parameters<SmsChannelRuntime["inbound"]["run"]>[0]["turnAdoptionLifecycle"]
+  >;
   log?: SmsLog;
 }): Promise<void> {
   const from = normalizeSmsPhoneNumber(params.msg.from);
@@ -127,10 +131,13 @@ export async function dispatchSmsInboundEvent(params: {
     channel: CHANNEL_ID,
     accountId: params.account.accountId,
     raw: params.msg,
+    ...(params.turnAdoptionLifecycle
+      ? { turnAdoptionLifecycle: params.turnAdoptionLifecycle }
+      : {}),
     adapter: {
       ingest: (msg) => ({
         id: msg.messageSid,
-        timestamp: Date.now(),
+        timestamp: params.receivedAt,
         rawText: msg.body,
         textForAgent: msg.body,
         textForCommands: msg.body,

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

@@ -0,0 +1,272 @@
+// Sms tests cover durable Twilio webhook admission and replay.
+import { mkdtemp, realpath, rm } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+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";
+
+type SmsIngressPayload = {
+  version: 1;
+  form: Record<string, string>;
+};
+
+const account: ResolvedSmsAccount = {
+  accountId: "default",
+  enabled: true,
+  accountSid: "AC123",
+  authToken: "secret",
+  fromNumber: "+15557654321",
+  messagingServiceSid: "",
+  defaultTo: "",
+  webhookPath: "/webhooks/sms",
+  publicWebhookUrl: "https://gateway.example.com/webhooks/sms",
+  dangerouslyDisableSignatureValidation: false,
+  dmPolicy: "pairing",
+  allowFrom: [],
+  textChunkLimit: 1500,
+};
+
+const stateDirs: string[] = [];
+const disposers: Array<() => void> = [];
+type SmsIngressDeliver = NonNullable<Parameters<typeof createSmsIngressSpool>[0]["deliver"]>;
+type SmsIngressSpool = ReturnType<typeof createSmsIngressSpool>;
+
+async function createStateDir(): Promise<string> {
+  const created = await mkdtemp(path.join(os.tmpdir(), "openclaw-sms-ingress-"));
+  const resolved = await realpath(created);
+  stateDirs.push(resolved);
+  return resolved;
+}
+
+function createQueue(stateDir: string) {
+  return createChannelIngressQueueForTests<SmsIngressPayload>({
+    channelId: "sms",
+    accountId: account.accountId,
+    stateDir,
+  });
+}
+
+function form(messageSid: string): Record<string, string> {
+  return {
+    AccountSid: account.accountSid,
+    From: "+15551234567",
+    To: "+15557654321",
+    Body: "hello",
+    MessageSid: messageSid,
+  };
+}
+
+async function drainSpool(spool: SmsIngressSpool): Promise<void> {
+  await spool.drainOnce();
+  await spool.waitForIdle();
+}
+
+afterEach(async () => {
+  for (const dispose of disposers.splice(0).toReversed()) {
+    dispose();
+  }
+  for (const stateDir of stateDirs.splice(0).toReversed()) {
+    await rm(stateDir, { recursive: true, force: true });
+  }
+});
+
+describe("createSmsIngressSpool", () => {
+  it("recovers an uncompleted message with a fresh drain instance", async () => {
+    const stateDir = await createStateDir();
+    const first = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver: vi.fn<SmsIngressDeliver>(async () => undefined),
+    });
+    disposers.push(first.dispose);
+    await first.enqueue(form("SM-restart"));
+    first.dispose();
+
+    const deliver = vi.fn<SmsIngressDeliver>(async (_message, lifecycle) => {
+      await lifecycle.onAdopted();
+    });
+    const recovered = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver,
+    });
+    disposers.push(recovered.dispose);
+    await drainSpool(recovered);
+
+    expect(deliver).toHaveBeenCalledOnce();
+  });
+
+  it("keeps a completed MessageSid tombstone from dispatching twice", async () => {
+    const stateDir = await createStateDir();
+    const deliver = vi.fn<SmsIngressDeliver>(async (_message, lifecycle) => {
+      await lifecycle.onAdopted();
+    });
+    const spool = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver,
+    });
+    disposers.push(spool.dispose);
+
+    expect(await spool.enqueue(form("SM-completed"))).toMatchObject({
+      kind: "accepted",
+      duplicate: false,
+    });
+    await drainSpool(spool);
+    expect(await spool.enqueue(form("SM-completed"))).toMatchObject({
+      kind: "completed",
+      duplicate: true,
+    });
+    await drainSpool(spool);
+
+    expect(deliver).toHaveBeenCalledOnce();
+  });
+
+  it.each(["SmsSid", "SmsMessageSid"])("accepts the legacy %s event id alias", async (key) => {
+    const stateDir = await createStateDir();
+    const deliver = vi.fn<SmsIngressDeliver>(async (_message, lifecycle) => {
+      await lifecycle.onAdopted();
+    });
+    const spool = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver,
+    });
+    disposers.push(spool.dispose);
+    const rawForm = form("SM-alias");
+    delete rawForm.MessageSid;
+    rawForm[key] = "SM-alias";
+
+    expect(await spool.enqueue(rawForm)).toMatchObject({ kind: "accepted", duplicate: false });
+    await drainSpool(spool);
+
+    expect(deliver).toHaveBeenCalledWith(
+      expect.objectContaining({ messageSid: "SM-alias" }),
+      expect.any(Object),
+      expect.any(Number),
+    );
+  });
+
+  it("uses the canonical sender as the durable lane", async () => {
+    const stateDir = await createStateDir();
+    const queue = createQueue(stateDir);
+    const spool = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue,
+      deliver: vi.fn<SmsIngressDeliver>(async () => undefined),
+    });
+    disposers.push(spool.dispose);
+
+    await spool.enqueue({ ...form("SM-canonical-lane"), From: "RcS:+1 (555) 123-4567" });
+
+    expect(await queue.listPending()).toEqual([
+      expect.objectContaining({ laneKey: "sender:+15551234567" }),
+    ]);
+  });
+
+  it("replays with the original webhook receipt timestamp", async () => {
+    const stateDir = await createStateDir();
+    const receivedAt = 1_700_000_000_456;
+    const now = vi
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(receivedAt)
+      .mockReturnValue(receivedAt + 60_000);
+    const first = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver: vi.fn<SmsIngressDeliver>(async () => undefined),
+    });
+    disposers.push(first.dispose);
+    await first.enqueue(form("SM-received-at"));
+    first.dispose();
+    now.mockRestore();
+
+    const deliver = vi.fn<SmsIngressDeliver>(async (_message, lifecycle) => {
+      await lifecycle.onAdopted();
+    });
+    const recovered = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver,
+    });
+    disposers.push(recovered.dispose);
+    await drainSpool(recovered);
+
+    expect(deliver).toHaveBeenCalledWith(expect.any(Object), expect.any(Object), receivedAt);
+  });
+
+  it("preserves the old handler-reload replay guard scenario with a tombstone", async () => {
+    const stateDir = await createStateDir();
+    const firstDeliver = vi.fn<SmsIngressDeliver>(async (_message, lifecycle) => {
+      await lifecycle.onAdopted();
+    });
+    const first = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver: firstDeliver,
+    });
+    disposers.push(first.dispose);
+    await first.enqueue(form("SM-handler-reload"));
+    await drainSpool(first);
+    first.dispose();
+
+    const reloadedDeliver = vi.fn<SmsIngressDeliver>(async () => undefined);
+    const reloaded = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver: reloadedDeliver,
+    });
+    disposers.push(reloaded.dispose);
+    expect(await reloaded.enqueue(form("SM-handler-reload"))).toMatchObject({
+      kind: "completed",
+      duplicate: true,
+    });
+    await drainSpool(reloaded);
+
+    expect(firstDeliver).toHaveBeenCalledOnce();
+    expect(reloadedDeliver).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    ["invalid payload", { MessageSid: "SM-invalid", From: "+15551234567" }],
+    ["account mismatch", { ...form("SM-account"), AccountSid: "AC-other" }],
+  ])("dead-letters a permanent %s failure", async (_label, rawForm) => {
+    const stateDir = await createStateDir();
+    const deliver = vi.fn<SmsIngressDeliver>(async () => undefined);
+    const spool = createSmsIngressSpool({
+      cfg: {},
+      account,
+      channelRuntime: {} as SmsChannelRuntime,
+      queue: createQueue(stateDir),
+      deliver,
+    });
+    disposers.push(spool.dispose);
+
+    await spool.enqueue(rawForm);
+    await drainSpool(spool);
+
+    expect(await spool.enqueue(rawForm)).toMatchObject({ kind: "failed", duplicate: true });
+    expect(deliver).not.toHaveBeenCalled();
+  });
+});

extensions/sms/src/ingress-spool.ts

@@ -0,0 +1,129 @@
+// Sms plugin module owns durable Twilio webhook admission and replay.
+import {
+  bindIngressLifecycleToReplyOptions,
+  createChannelIngressDrain,
+  type ChannelIngressQueue,
+} from "openclaw/plugin-sdk/channel-outbound";
+import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
+import { dispatchSmsInboundEvent, type SmsChannelRuntime } from "./inbound.js";
+import { getSmsRuntime } from "./runtime.js";
+import {
+  buildTwilioInboundMessage,
+  resolveTwilioInboundSender,
+  resolveTwilioMessageSid,
+} from "./twilio.js";
+import type { ResolvedSmsAccount, SmsInboundMessage } from "./types.js";
+
+const SMS_INGRESS_PAYLOAD_VERSION = 1;
+// Tombstones dominate the retired 10-minute / 10,000-key replay cache.
+const SMS_COMPLETED_TTL_MS = 24 * 60 * 60 * 1000;
+const SMS_COMPLETED_MAX_ENTRIES = 20_000;
+const SMS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000;
+const SMS_FAILED_MAX_ENTRIES = 1_000;
+
+type SmsIngressPayload = {
+  version: typeof SMS_INGRESS_PAYLOAD_VERSION;
+  form: Record<string, string>;
+};
+
+type SmsIngressLifecycle = ReturnType<
+  typeof bindIngressLifecycleToReplyOptions
+>["turnAdoptionLifecycle"];
+
+class SmsIngressPermanentError extends Error {}
+
+function parseSmsIngressPayload(
+  payload: SmsIngressPayload,
+  account: ResolvedSmsAccount,
+): SmsInboundMessage {
+  if (payload.version !== SMS_INGRESS_PAYLOAD_VERSION) {
+    throw new SmsIngressPermanentError("SMS ingress payload version is invalid.");
+  }
+  const message = buildTwilioInboundMessage(payload.form);
+  if (!message) {
+    throw new SmsIngressPermanentError("SMS ingress payload is invalid.");
+  }
+  if (message.accountSid && message.accountSid !== account.accountSid) {
+    throw new SmsIngressPermanentError("SMS ingress payload has an invalid Twilio account.");
+  }
+  return message;
+}
+
+export function createSmsIngressSpool(params: {
+  cfg: OpenClawConfig;
+  account: ResolvedSmsAccount;
+  channelRuntime: SmsChannelRuntime;
+  queue?: ChannelIngressQueue<SmsIngressPayload>;
+  abortSignal?: AbortSignal;
+  log?: { info?: (message: string) => void; warn?: (message: string) => void };
+  deliver?: (
+    message: SmsInboundMessage,
+    lifecycle: SmsIngressLifecycle,
+    receivedAt: number,
+  ) => Promise<void>;
+}) {
+  const queue =
+    params.queue ??
+    getSmsRuntime().state.openChannelIngressQueue<SmsIngressPayload>({
+      accountId: params.account.accountId,
+    });
+  const deliver =
+    params.deliver ??
+    (async (message: SmsInboundMessage, lifecycle: SmsIngressLifecycle, receivedAt: number) => {
+      await dispatchSmsInboundEvent({
+        cfg: params.cfg,
+        account: params.account,
+        channelRuntime: params.channelRuntime,
+        msg: message,
+        receivedAt,
+        turnAdoptionLifecycle: lifecycle,
+        log: params.log,
+      });
+    });
+  const drain = createChannelIngressDrain<SmsIngressPayload>({
+    queue,
+    ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}),
+    ...(params.log?.warn ? { onLog: (message: string) => params.log?.warn?.(message) } : {}),
+    resolveNonRetryableFailure: (error) =>
+      error instanceof SmsIngressPermanentError
+        ? { reason: "invalid-payload", message: error.message }
+        : null,
+    dispatchClaimedEvent: async (event, lifecycle) => {
+      await deliver(
+        parseSmsIngressPayload(event.payload, params.account),
+        bindIngressLifecycleToReplyOptions(lifecycle).turnAdoptionLifecycle,
+        event.receivedAt,
+      );
+    },
+  });
+  return {
+    enqueue: async (form: Record<string, string>) => {
+      const receivedAt = Date.now();
+      const eventId = resolveTwilioMessageSid(form);
+      if (!eventId) {
+        throw new Error("SMS webhook is missing MessageSid.");
+      }
+      const sender = resolveTwilioInboundSender(form);
+      await queue.prune({
+        completedTtlMs: SMS_COMPLETED_TTL_MS,
+        completedMaxEntries: SMS_COMPLETED_MAX_ENTRIES,
+        failedTtlMs: SMS_FAILED_TTL_MS,
+        failedMaxEntries: SMS_FAILED_MAX_ENTRIES,
+      });
+      const result = await queue.enqueue(
+        eventId,
+        { version: SMS_INGRESS_PAYLOAD_VERSION, form },
+        {
+          receivedAt,
+          laneKey: sender ? `sender:${sender}` : `event:${eventId}`,
+        },
+      );
+      return { kind: result.kind, duplicate: result.duplicate };
+    },
+    drainOnce: async () => {
+      await drain.drainOnce();
+    },
+    waitForIdle: drain.waitForIdle,
+    dispose: () => drain.dispose(),
+  };
+}

extensions/sms/src/twilio.ts

@@ -234,23 +234,32 @@ function parseTwilioInboundFrom(raw: string): string | null {
   return phoneNumber;
 }
 
+export function resolveTwilioInboundSender(form: Record<string, string>): string {
+  return parseTwilioInboundFrom(firstTrimmedString(form.From)) ?? "";
+}
+
 export function buildTwilioInboundMessage(form: Record<string, string>): SmsInboundMessage | null {
   // Signature verification owns the untouched form. Canonicalize only after
   // that boundary so Twilio channel prefixes never change its signed input.
-  const from = parseTwilioInboundFrom(firstTrimmedString(form.From));
+  const from = resolveTwilioInboundSender(form);
   const to = firstTrimmedString(form.To);
   const body = firstString(form.Body);
   const accountSid = firstTrimmedString(form.AccountSid);
-  const messageSid =
-    firstTrimmedString(form.MessageSid) ||
-    firstTrimmedString(form.SmsSid) ||
-    firstTrimmedString(form.SmsMessageSid);
+  const messageSid = resolveTwilioMessageSid(form);
   if (!from || !to || !body || !messageSid) {
     return null;
   }
   return { accountSid, from, to, body, messageSid };
 }
 
+export function resolveTwilioMessageSid(form: Record<string, string>): string {
+  return (
+    firstTrimmedString(form.MessageSid) ||
+    firstTrimmedString(form.SmsSid) ||
+    firstTrimmedString(form.SmsMessageSid)
+  );
+}
+
 export async function readTwilioWebhookForm(req: IncomingMessage): Promise<Record<string, string>> {
   const body = await readRequestBodyWithLimit(req, {
     maxBytes: WEBHOOK_BODY_LIMIT_BYTES,

extensions/sms/src/webhook-replay-guard.test.ts

@@ -1,51 +0,0 @@
-import { describe, expect, it } from "vitest";
-import { createSmsWebhookReplayGuard } from "./webhook-replay-guard.js";
-
-describe("createSmsWebhookReplayGuard", () => {
-  it("prunes only the expired insertion prefix without refreshing replays", () => {
-    let nowMs = 0;
-    const replayGuard = createSmsWebhookReplayGuard({
-      ttlMs: 10,
-      maxKeys: 2,
-      now: () => nowMs,
-    });
-
-    expect(replayGuard.remember("first")).toEqual({ kind: "accepted" });
-    nowMs = 2;
-    expect(replayGuard.remember("second")).toEqual({ kind: "accepted" });
-    nowMs = 5;
-    expect(replayGuard.remember("first")).toEqual({ kind: "replayed" });
-    expect(replayGuard.remember("overflow")).toEqual({
-      kind: "saturated",
-      retryAfterMs: 5,
-    });
-
-    nowMs = 10;
-    expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" });
-    expect(replayGuard.remember("second")).toEqual({ kind: "replayed" });
-  });
-
-  it("keeps live replay keys and fails closed until capacity expires", () => {
-    let nowMs = 1_000;
-    const replayGuard = createSmsWebhookReplayGuard({
-      ttlMs: 10_000,
-      maxKeys: 2,
-      now: () => nowMs,
-    });
-
-    expect(replayGuard.remember("first")).toEqual({ kind: "accepted" });
-    expect(replayGuard.remember("second")).toEqual({ kind: "accepted" });
-    expect(replayGuard.remember("overflow")).toEqual({
-      kind: "saturated",
-      retryAfterMs: 10_000,
-    });
-    expect(replayGuard.remember("overflow")).toEqual({
-      kind: "saturated",
-      retryAfterMs: 10_000,
-    });
-    expect(replayGuard.remember("first")).toEqual({ kind: "replayed" });
-
-    nowMs += 10_000;
-    expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" });
-  });
-});

extensions/sms/src/webhook-replay-guard.ts

@@ -1,56 +0,0 @@
-import { performance } from "node:perf_hooks";
-
-const REPLAY_CACHE_TTL_MS = 10 * 60_000;
-const REPLAY_CACHE_MAX_KEYS = 10_000;
-
-type ReplayCacheDecision =
-  | { kind: "accepted" }
-  | { kind: "replayed" }
-  | { kind: "saturated"; retryAfterMs: number };
-
-export type SmsWebhookReplayGuard = {
-  remember: (messageSid: string) => ReplayCacheDecision;
-};
-
-export function createSmsWebhookReplayGuard(
-  options: {
-    ttlMs?: number;
-    maxKeys?: number;
-    now?: () => number;
-  } = {},
-): SmsWebhookReplayGuard {
-  const ttlMs = options.ttlMs ?? REPLAY_CACHE_TTL_MS;
-  const maxKeys = options.maxKeys ?? REPLAY_CACHE_MAX_KEYS;
-  const now = options.now ?? (() => performance.now());
-  const entries = new Map<string, number>();
-
-  const pruneExpired = (nowMs: number) => {
-    // Fixed TTLs on a monotonic clock expire in insertion order, so only inspect
-    // the expired prefix. Full live caches stay O(1) instead of rescanning 10k keys.
-    for (const [key, expiresAt] of entries) {
-      if (expiresAt > nowMs) {
-        break;
-      }
-      entries.delete(key);
-    }
-  };
-
-  return {
-    remember: (messageSid) => {
-      const nowMs = now();
-      pruneExpired(nowMs);
-      if (entries.has(messageSid)) {
-        return { kind: "replayed" };
-      }
-      if (entries.size >= maxKeys) {
-        const oldestExpiresAt = entries.values().next().value ?? nowMs;
-        return {
-          kind: "saturated",
-          retryAfterMs: Math.max(0, oldestExpiresAt - nowMs),
-        };
-      }
-      entries.set(messageSid, nowMs + ttlMs);
-      return { kind: "accepted" };
-    },
-  };
-}

extensions/sms/src/webhook.test.ts

@@ -3,11 +3,13 @@ import { createHmac } from "node:crypto";
 import type { IncomingMessage, ServerResponse } from "node:http";
 import { Readable } from "node:stream";
 import { beforeEach, describe, expect, it, vi } from "vitest";
-import type { SmsChannelRuntime } from "./inbound.js";
 import type { ResolvedSmsAccount } from "./types.js";
 import { createSmsWebhookHandler } from "./webhook.js";
 
-const dispatchSmsInboundEvent = vi.hoisted(() => vi.fn(async () => undefined));
+const enqueueSmsIngress = vi.hoisted(() =>
+  vi.fn(async () => ({ kind: "accepted" as const, duplicate: false })),
+);
+const drainSmsIngress = vi.hoisted(() => vi.fn(async () => undefined));
 const runDetachedWebhookWork = vi.hoisted(() => vi.fn((run: () => Promise<void>) => run()));
 
 vi.mock("openclaw/plugin-sdk/webhook-request-guards", async (importOriginal) => {
@@ -16,13 +18,16 @@ vi.mock("openclaw/plugin-sdk/webhook-request-guards", async (importOriginal) =>
   return { ...actual, runDetachedWebhookWork };
 });
 
-vi.mock("./inbound.js", () => ({
-  dispatchSmsInboundEvent,
-}));
-
 let testAccountSequence = 0;
 let activeAccountId = "test-0";
 
+function createIngress() {
+  return {
+    enqueue: enqueueSmsIngress,
+    drainOnce: drainSmsIngress,
+  };
+}
+
 function parseTestTwilioForm(body: string): Record<string, string> {
   return Object.fromEntries(new URLSearchParams(body));
 }
@@ -100,18 +105,21 @@ function createRequest(
 type TestResponse = ServerResponse & {
   body?: string;
   setHeaderMock: ReturnType<typeof vi.fn>;
+  endMock: ReturnType<typeof vi.fn>;
 };
 
 function createResponse(): TestResponse {
   const setHeaderMock = vi.fn();
+  const endMock = vi.fn(function (this: ServerResponse & { body?: string }, body?: string) {
+    this.body = body;
+    return this;
+  });
   return {
     statusCode: 200,
     setHeader: setHeaderMock,
     setHeaderMock,
-    end: vi.fn(function (this: ServerResponse & { body?: string }, body?: string) {
-      this.body = body;
-      return this;
-    }),
+    end: endMock,
+    endMock,
   } as unknown as TestResponse;
 }
 
@@ -142,39 +150,92 @@ function createMessageSid(index: number): string {
 
 describe("createSmsWebhookHandler", () => {
   beforeEach(() => {
-    dispatchSmsInboundEvent.mockClear();
+    enqueueSmsIngress.mockReset();
+    enqueueSmsIngress.mockResolvedValue({ kind: "accepted", duplicate: false });
+    drainSmsIngress.mockClear();
     runDetachedWebhookWork.mockClear();
     activeAccountId = `test-${++testAccountSequence}`;
   });
 
-  it("validates a fragmentless signature and preserves dedupe across handler reloads", async () => {
+  it("validates a fragmentless signature before enqueuing the raw Twilio form", async () => {
     const { body, signature } = createSignedSmsPayload(createMessageSid(1));
     const handler = createSmsWebhookHandler({
       cfg: {},
       account: createAccount({
         publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx",
       }),
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
 
-    const firstRes = createResponse();
-    await handler(createRequest(body, signature), firstRes);
-    const replayRes = createResponse();
-    const reloadedHandler = createSmsWebhookHandler({
-      cfg: {},
-      account: createAccount({
-        publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx",
-      }),
-      channelRuntime: {} as SmsChannelRuntime,
-    });
-    await reloadedHandler(createRequest(body, signature), replayRes);
+    const res = createResponse();
+    await handler(createRequest(body, signature), res);
 
-    expect(firstRes.statusCode).toBe(200);
-    expect(replayRes.statusCode).toBe(200);
-    expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(1);
+    expect(res.statusCode).toBe(200);
+    expect(enqueueSmsIngress).toHaveBeenCalledWith(parseTestTwilioForm(body));
     expect(runDetachedWebhookWork).toHaveBeenCalledTimes(1);
   });
 
+  it("does not acknowledge when the durable enqueue fails", async () => {
+    const { body, signature } = createSignedSmsPayload(createMessageSid(2));
+    enqueueSmsIngress.mockRejectedValueOnce(new Error("sqlite unavailable"));
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+    });
+    const res = createResponse();
+
+    await expect(handler(createRequest(body, signature), res)).rejects.toThrow(
+      "sqlite unavailable",
+    );
+
+    expect(res.endMock).not.toHaveBeenCalled();
+    expect(runDetachedWebhookWork).not.toHaveBeenCalled();
+  });
+
+  it("rejects a signed webhook without a stable MessageSid", async () => {
+    const body = "AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello";
+    const signature = computeTestTwilioSignature({
+      url: "https://gateway.example.com/webhooks/sms",
+      authToken: "secret",
+      form: parseTestTwilioForm(body),
+    });
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+    });
+    const res = createResponse();
+
+    await handler(createRequest(body, signature), res);
+
+    expect(res.statusCode).toBe(400);
+    expect(enqueueSmsIngress).not.toHaveBeenCalled();
+  });
+
+  it("accepts the legacy SmsMessageSid event id alias", async () => {
+    const body =
+      "AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=SM-alias";
+    const signature = computeTestTwilioSignature({
+      url: "https://gateway.example.com/webhooks/sms",
+      authToken: "secret",
+      form: parseTestTwilioForm(body),
+    });
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+    });
+    const res = createResponse();
+
+    await handler(createRequest(body, signature), res);
+
+    expect(res.statusCode).toBe(200);
+    expect(enqueueSmsIngress).toHaveBeenCalledWith(
+      expect.objectContaining({ SmsMessageSid: "SM-alias" }),
+    );
+  });
+
   it("validates the raw RCS form before canonicalizing its sender", async () => {
     const messageSid = createMessageSid(9);
     const { body, signature } = createSignedSmsPayload(messageSid, {
@@ -184,7 +245,7 @@ describe("createSmsWebhookHandler", () => {
     const handler = createSmsWebhookHandler({
       cfg: {},
       account: createAccount(),
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
 
     expect(parseTestTwilioForm(body).From).toBe("RcS:+1 (555) 123-4567");
@@ -193,21 +254,19 @@ describe("createSmsWebhookHandler", () => {
     await handler(createRequest(body, signature), res);
 
     expect(res.statusCode).toBe(200);
-    expect(dispatchSmsInboundEvent).toHaveBeenCalledWith(
+    expect(enqueueSmsIngress).toHaveBeenCalledWith(
       expect.objectContaining({
-        msg: {
-          accountSid: "AC123",
-          from: "+15551234567",
-          to: "rcs:example-agent",
-          body: "hello",
-          messageSid,
-        },
+        AccountSid: "AC123",
+        From: "RcS:+1 (555) 123-4567",
+        To: "rcs:example-agent",
+        Body: "hello",
+        MessageSid: messageSid,
       }),
     );
   });
 
-  it("rejects signed webhooks for a different Twilio account", async () => {
-    const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=${createMessageSid(8)}`;
+  it("durably accepts a signed account mismatch for non-retryable drain classification", async () => {
+    const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&MessageSid=${createMessageSid(8)}`;
     const signature = computeTestTwilioSignature({
       url: "https://gateway.example.com/webhooks/sms",
       authToken: "secret",
@@ -216,14 +275,16 @@ describe("createSmsWebhookHandler", () => {
     const handler = createSmsWebhookHandler({
       cfg: {},
       account: createAccount(),
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
 
     const res = createResponse();
     await handler(createRequest(body, signature), res);
 
-    expect(res.statusCode).toBe(403);
-    expect(dispatchSmsInboundEvent).not.toHaveBeenCalled();
+    expect(res.statusCode).toBe(200);
+    expect(enqueueSmsIngress).toHaveBeenCalledWith(
+      expect.objectContaining({ AccountSid: "AC-other" }),
+    );
   });
 
   it("does not let unsigned proxy traffic consume the same client's signed webhook rate limit", async () => {
@@ -231,7 +292,7 @@ describe("createSmsWebhookHandler", () => {
     const handler = createSmsWebhookHandler({
       cfg: { gateway: { trustedProxies: ["127.0.0.1"] } },
       account,
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
     const unsignedBody =
       "AccountSid=AC123&From=%2B15550000000&To=%2B15557654321&Body=bad&MessageSid=SM-bad";
@@ -264,7 +325,7 @@ describe("createSmsWebhookHandler", () => {
     );
 
     expect(accepted.statusCode).toBe(200);
-    expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(1);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(1);
   });
 
   it("scopes signed webhook rate limits to one SMS account and route", async () => {
@@ -278,12 +339,12 @@ describe("createSmsWebhookHandler", () => {
     const supportHandler = createSmsWebhookHandler({
       cfg: {},
       account: supportAccount,
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
     const defaultHandler = createSmsWebhookHandler({
       cfg: {},
       account: defaultAccount,
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
 
     for (let i = 0; i < 30; i += 1) {
@@ -318,7 +379,7 @@ describe("createSmsWebhookHandler", () => {
     const handler = createSmsWebhookHandler({
       cfg: { gateway: { trustedProxies: ["127.0.0.1"] } },
       account,
-      channelRuntime: {} as SmsChannelRuntime,
+      ingress: createIngress(),
     });
 
     for (let i = 0; i < 30; i += 1) {
@@ -349,6 +410,6 @@ describe("createSmsWebhookHandler", () => {
     );
 
     expect(overBudgetRes.statusCode).toBe(429);
-    expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(30);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(30);
   });
 });

extensions/sms/src/webhook.ts

@@ -6,16 +6,14 @@ import {
   resolveRequestClientIp,
 } from "openclaw/plugin-sdk/webhook-ingress";
 import { runDetachedWebhookWork } from "openclaw/plugin-sdk/webhook-request-guards";
-import { dispatchSmsInboundEvent, type SmsChannelRuntime } from "./inbound.js";
 import {
-  buildTwilioInboundMessage,
   readTwilioWebhookForm,
   respondTwiml,
+  resolveTwilioMessageSid,
   resolveTwilioWebhookSignatureUrl,
   verifyTwilioSignature,
 } from "./twilio.js";
 import type { ResolvedSmsAccount } from "./types.js";
-import { createSmsWebhookReplayGuard, type SmsWebhookReplayGuard } from "./webhook-replay-guard.js";
 
 const INVALID_REQUEST_MAX_REQUESTS = 300;
 const CALLBACK_DISPATCH_MAX_REQUESTS = 30;
@@ -33,20 +31,6 @@ const callbackDispatchRateLimiter = createFixedWindowRateLimiter({
   windowMs: 60_000,
   maxTrackedKeys: 5_000,
 });
-const replayGuardsByAccount = new Map<string, SmsWebhookReplayGuard>();
-
-function resolveSmsWebhookReplayGuard(account: ResolvedSmsAccount): SmsWebhookReplayGuard {
-  // Config reloads replace route handlers. Keep the guard with the Twilio account
-  // identity so retries cannot cross that lifecycle boundary or block sibling accounts.
-  const key = `${account.accountId}\0${account.accountSid}`;
-  const existing = replayGuardsByAccount.get(key);
-  if (existing) {
-    return existing;
-  }
-  const created = createSmsWebhookReplayGuard();
-  replayGuardsByAccount.set(key, created);
-  return created;
-}
 
 type SmsWebhookLog = {
   info?: (message: string) => void;
@@ -57,7 +41,10 @@ type SmsWebhookLog = {
 export type SmsWebhookHandlerParams = {
   cfg: OpenClawConfig;
   account: ResolvedSmsAccount;
-  channelRuntime: SmsChannelRuntime;
+  ingress: {
+    enqueue: (form: Record<string, string>) => Promise<{ duplicate: boolean }>;
+    drainOnce: () => Promise<void>;
+  };
   log?: SmsWebhookLog;
 };
 
@@ -94,9 +81,8 @@ function rejectInvalidRequestRateLimit(params: {
   return true;
 }
 
-// Each account route owns its guard so one saturated account cannot block sibling accounts.
+// Each account route owns one durable ingress adapter.
 export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
-  const webhookReplayGuard = resolveSmsWebhookReplayGuard(params.account);
   return async (req: IncomingMessage, res: ServerResponse) => {
     if (req.method !== "POST") {
       respondTwiml(res, 405, "Method not allowed");
@@ -138,22 +124,6 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
       }
     }
 
-    const msg = buildTwilioInboundMessage(form);
-    if (!msg) {
-      if (invalidRequestRateLimited) {
-        return rejectInvalidRequestRateLimit({ key, log: params.log, res });
-      }
-      respondTwiml(res, 400, "Missing SMS payload");
-      return true;
-    }
-    if (msg.accountSid && msg.accountSid !== params.account.accountSid) {
-      if (invalidRequestRateLimited) {
-        return rejectInvalidRequestRateLimit({ key, log: params.log, res });
-      }
-      params.log?.warn?.("SMS webhook rejected mismatched Twilio AccountSid");
-      respondTwiml(res, 403, "Invalid account");
-      return true;
-    }
     if (invalidRequestRateLimited && params.account.dangerouslyDisableSignatureValidation) {
       return rejectInvalidRequestRateLimit({ key, log: params.log, res });
     }
@@ -162,36 +132,23 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
       respondTwiml(res, 429, "Rate limit exceeded");
       return true;
     }
-    const replayDecision = webhookReplayGuard.remember(msg.messageSid);
-    if (replayDecision.kind === "replayed") {
-      params.log?.warn?.(`SMS webhook ignored replayed message ${msg.messageSid}`);
-      respondTwiml(res, 200);
+    const messageSid = resolveTwilioMessageSid(form);
+    if (!messageSid) {
+      respondTwiml(res, 400, "Missing MessageSid");
       return true;
     }
-    if (replayDecision.kind === "saturated") {
-      const retryAfterSeconds = Math.max(1, Math.ceil(replayDecision.retryAfterMs / 1000));
-      params.log?.warn?.("SMS webhook replay cache is full of unexpired message SIDs");
-      res.setHeader("Retry-After", String(retryAfterSeconds));
-      respondTwiml(res, 429, "Replay cache saturated");
-      return true;
+    // Signature validation owns the parsed-but-otherwise-raw Twilio form.
+    // A 200 is impossible until SQLite commits this exact transport envelope.
+    const verdict = await params.ingress.enqueue(form);
+    if (verdict.duplicate) {
+      params.log?.warn?.(`SMS webhook ignored replayed message ${messageSid}`);
     }
-
-    // Reserve the detached task before the HTTP admission is released;
-    // otherwise later queue work inherits a released admission root.
-    void runDetachedWebhookWork(() =>
-      dispatchSmsInboundEvent({
-        cfg: params.cfg,
-        account: params.account,
-        msg,
-        channelRuntime: params.channelRuntime,
-        log: params.log,
-      }),
-    ).catch((err: unknown) => {
+    // Reserve detached work under HTTP admission; it only pumps the durable drain.
+    void runDetachedWebhookWork(() => params.ingress.drainOnce()).catch((err: unknown) => {
       params.log?.error?.(
-        `SMS webhook dispatch failed: ${err instanceof Error ? err.message : String(err)}`,
+        `SMS ingress drain failed: ${err instanceof Error ? err.message : String(err)}`,
       );
     });
-
     respondTwiml(res, 200);
     return true;
   };