fix(gateway): dedupe inflight outbound requests (#68341)

Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-07-21 10:16:44 +00:00 · 2026-05-11 15:10:21 +01:00
parent 0343a2e689
commit 83ffc1f47a
3 changed files with 407 additions and 161 deletions
@@ -61,6 +61,7 @@ Docs: https://docs.openclaw.ai

 - Cron: keep long manual cron runs active in the task registry until completion, preventing transient `lost` markers before durable recovery reconciles. Fixes #78233. (#78243) Thanks @Feelw00.
 - Doctor/GitHub CLI: surface a `GH_CONFIG_DIR` hint when the GitHub skill is usable but `gh` auth lives under a different operator HOME than the agent process, without warning for disabled or filtered skills. Fixes #78063. (#78095) Thanks @tmimmanuel.
+- Gateway: dedupe concurrent `send`, `poll`, and `message.action` requests while delivery is still in flight, preventing duplicate outbound work for the same idempotency key. (#68341) Thanks @thesomewhatyou.
 - Gateway: clear speculative node wake state when APNs registration is missing, preventing unregistered or mistyped node IDs from retaining wake throttle entries. Fixes #68847. (#68848) Thanks @Feelw00.
 - Auto-reply: keep late follow-up queue drain finalizers from deleting a replacement queue registered after `/stop`, preventing immediate follow-up messages from being orphaned. Fixes #68838. (#68839) Thanks @Feelw00.
 - Feishu: make manual App ID/App Secret setup the default channel-binding path while keeping QR scan-to-create as an optional best-effort flow, and document the manual fallback for domestic Feishu mobile clients that do not react to the QR code. Fixes #80591. Thanks @wei-wei-zhao.
@@ -15,6 +15,7 @@ const mocks = vi.hoisted(() => ({
  resolveOutboundSessionRoute: vi.fn(),
  ensureOutboundSessionEntry: vi.fn(async () => undefined),
  resolveMessageChannelSelection: vi.fn(),
+  dispatchChannelMessageAction: vi.fn(),
  sendPoll: vi.fn<
    () => Promise<{
      messageId: string;
@@ -44,6 +45,10 @@ vi.mock("../../channels/plugins/index.js", () => ({
  normalizeChannelId: (value: string) => (value === "webchat" ? null : value),
 }));

+vi.mock("../../channels/plugins/message-action-dispatch.js", () => ({
+  dispatchChannelMessageAction: mocks.dispatchChannelMessageAction,
+}));
+
 const TEST_AGENT_WORKSPACE = "/tmp/openclaw-test-workspace";
 let sendHandlers: typeof import("./send.js").sendHandlers;

@@ -163,6 +168,16 @@ async function runPollWithClient(
  return { respond };
 }

+function createDeferred<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
 async function runMessageActionRequest(
  params: Record<string, unknown>,
  client?: { connect?: { scopes?: string[] } } | null,
@@ -256,8 +271,193 @@ describe("gateway send mirroring", () => {
      channel: "slack",
      configured: ["slack"],
    });
+    mocks.dispatchChannelMessageAction.mockResolvedValue({
+      details: { action: "handled" },
+    });
    mocks.sendPoll.mockResolvedValue({ messageId: "poll-1" });
-    mocks.getChannelPlugin.mockReturnValue({ outbound: { sendPoll: mocks.sendPoll } });
+    mocks.getChannelPlugin.mockReturnValue({
+      actions: { handleAction: true },
+      outbound: { sendPoll: mocks.sendPoll },
+    });
+  });
+
+  it("dedupes concurrent message.action requests while inflight", async () => {
+    const context = makeContext();
+    const firstRespond = vi.fn();
+    const secondRespond = vi.fn();
+    const actionDeferred = createDeferred<{ details: { action: string } }>();
+    mocks.dispatchChannelMessageAction.mockReturnValueOnce(actionDeferred.promise);
+
+    const firstRequest = sendHandlers["message.action"]({
+      params: {
+        channel: "slack",
+        action: "poll",
+        params: { question: "Q?" },
+        idempotencyKey: "idem-action-concurrent",
+      } as never,
+      respond: firstRespond,
+      context,
+      req: { type: "req", id: "1", method: "message.action" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    const secondRequest = sendHandlers["message.action"]({
+      params: {
+        channel: "slack",
+        action: "poll",
+        params: { question: "Q?" },
+        idempotencyKey: "idem-action-concurrent",
+      } as never,
+      respond: secondRespond,
+      context,
+      req: { type: "req", id: "2", method: "message.action" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    await Promise.resolve();
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(1);
+
+    actionDeferred.resolve({ details: { action: "handled" } });
+    await Promise.all([firstRequest, secondRequest]);
+
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(1);
+    expect(firstRespond).toHaveBeenCalledWith(
+      true,
+      { action: "handled" },
+      undefined,
+      expect.objectContaining({ channel: "slack" }),
+    );
+    expect(secondRespond).toHaveBeenCalledWith(
+      true,
+      { action: "handled" },
+      undefined,
+      expect.objectContaining({ channel: "slack", cached: true }),
+    );
+  });
+
+  it("dedupes concurrent send requests while inflight", async () => {
+    const context = makeContext();
+    const firstRespond = vi.fn();
+    const secondRespond = vi.fn();
+    const deliveryDeferred = createDeferred<Array<{ messageId: string; channel: string }>>();
+    mocks.deliverOutboundPayloads.mockReturnValueOnce(deliveryDeferred.promise);
+
+    const firstRequest = sendHandlers.send({
+      params: {
+        to: "channel:C1",
+        message: "hi",
+        channel: "slack",
+        idempotencyKey: "idem-send-concurrent",
+      } as never,
+      respond: firstRespond,
+      context,
+      req: { type: "req", id: "1", method: "send" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    const secondRequest = sendHandlers.send({
+      params: {
+        to: "channel:C1",
+        message: "hi",
+        channel: "slack",
+        idempotencyKey: "idem-send-concurrent",
+      } as never,
+      respond: secondRespond,
+      context,
+      req: { type: "req", id: "2", method: "send" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    await vi.waitFor(() => {
+      expect(mocks.deliverOutboundPayloads).toHaveBeenCalledTimes(1);
+    });
+
+    deliveryDeferred.resolve([{ messageId: "m-concurrent", channel: "slack" }]);
+    await Promise.all([firstRequest, secondRequest]);
+
+    expect(mocks.deliverOutboundPayloads).toHaveBeenCalledTimes(1);
+    expect(firstRespond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ messageId: "m-concurrent", runId: "idem-send-concurrent" }),
+      undefined,
+      expect.objectContaining({ channel: "slack" }),
+    );
+    expect(secondRespond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ messageId: "m-concurrent", runId: "idem-send-concurrent" }),
+      undefined,
+      expect.objectContaining({ channel: "slack", cached: true }),
+    );
+  });
+
+  it("dedupes concurrent poll requests while inflight", async () => {
+    const context = makeContext();
+    const firstRespond = vi.fn();
+    const secondRespond = vi.fn();
+    const pollDeferred = createDeferred<{ messageId: string; pollId: string }>();
+    mocks.sendPoll.mockReturnValueOnce(pollDeferred.promise);
+
+    const firstRequest = sendHandlers.poll({
+      params: {
+        to: "channel:C1",
+        question: "Q?",
+        options: ["A", "B"],
+        channel: "slack",
+        idempotencyKey: "idem-poll-concurrent",
+      } as never,
+      respond: firstRespond,
+      context,
+      req: { type: "req", id: "1", method: "poll" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    const secondRequest = sendHandlers.poll({
+      params: {
+        to: "channel:C1",
+        question: "Q?",
+        options: ["A", "B"],
+        channel: "slack",
+        idempotencyKey: "idem-poll-concurrent",
+      } as never,
+      respond: secondRespond,
+      context,
+      req: { type: "req", id: "2", method: "poll" },
+      client: null as never,
+      isWebchatConnect: () => false,
+    });
+
+    await Promise.resolve();
+    expect(mocks.sendPoll).toHaveBeenCalledTimes(1);
+
+    pollDeferred.resolve({ messageId: "poll-concurrent", pollId: "poll-1" });
+    await Promise.all([firstRequest, secondRequest]);
+
+    expect(mocks.sendPoll).toHaveBeenCalledTimes(1);
+    expect(firstRespond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({
+        messageId: "poll-concurrent",
+        pollId: "poll-1",
+        runId: "idem-poll-concurrent",
+      }),
+      undefined,
+      expect.objectContaining({ channel: "slack" }),
+    );
+    expect(secondRespond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({
+        messageId: "poll-concurrent",
+        pollId: "poll-1",
+        runId: "idem-poll-concurrent",
+      }),
+      undefined,
+      expect.objectContaining({ channel: "slack", cached: true }),
+    );
  });

  it("accepts media-only sends without message", async () => {
@@ -1016,6 +1216,18 @@ describe("gateway send mirroring", () => {
      ]),
      "send-test-message-action",
    );
+    mocks.dispatchChannelMessageAction.mockResolvedValueOnce(
+      jsonResult({
+        ok: true,
+        messageId: "wamid.1",
+        requesterSenderId: "trusted-user",
+        currentMessageId: "wamid.1",
+        currentGraphChannelId: "graph:team/chan",
+        replyToMode: "first",
+        hasRepliedRef: true,
+        skipCrossContextDecoration: true,
+      }),
+    );

    const { respond } = await runMessageActionRequest({
      channel: "whatsapp",
@@ -1055,7 +1267,6 @@ describe("gateway send mirroring", () => {
  });

  it("passes agent-scoped media roots to gateway message actions", async () => {
-    let capturedMediaLocalRoots: readonly string[] | undefined;
    const mediaActionPlugin: ChannelPlugin = {
      id: "telegram",
      meta: {
@@ -1074,10 +1285,7 @@ describe("gateway send mirroring", () => {
      actions: {
        describeMessageTool: () => ({ actions: ["sendAttachment"] }),
        supportsAction: ({ action }) => action === "sendAttachment",
-        handleAction: async ({ mediaLocalRoots }) => {
-          capturedMediaLocalRoots = mediaLocalRoots;
-          return jsonResult({ ok: true });
-        },
+        handleAction: async () => jsonResult({ ok: true }),
      },
    };
    mocks.getChannelPlugin.mockReturnValue(mediaActionPlugin);
@@ -1095,7 +1303,11 @@ describe("gateway send mirroring", () => {
    });

    expect(firstRespondCall(respond)?.[0]).toBe(true);
-    expect(capturedMediaLocalRoots).toContain(TEST_AGENT_WORKSPACE);
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenLastCalledWith(
+      expect.objectContaining({
+        mediaLocalRoots: expect.arrayContaining([TEST_AGENT_WORKSPACE]),
+      }),
+    );
  });

  it("forces senderIsOwner=false for narrowly-scoped callers but honors it for full operators", async () => {
@@ -1144,7 +1356,9 @@ describe("gateway send mirroring", () => {
      },
      { connect: { scopes: ["operator.write"] } },
    );
-    expect(capture.senderIsOwner).toBe(false);
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenLastCalledWith(
+      expect.objectContaining({ senderIsOwner: false }),
+    );

    // Full operator (admin-scoped): the trusted runtime is allowed to
    // forward the real channel-sender ownership bit. Wire true → true.
@@ -1162,7 +1376,9 @@ describe("gateway send mirroring", () => {
      },
      { connect: { scopes: ["operator.admin"] } },
    );
-    expect(capture.senderIsOwner).toBe(true);
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenLastCalledWith(
+      expect.objectContaining({ senderIsOwner: true }),
+    );

    // Full operator forwarding a non-owner sender: wire false → false
    // (admin scope does not inflate ownership on its own).
@@ -1180,6 +1396,8 @@ describe("gateway send mirroring", () => {
      },
      { connect: { scopes: ["operator.admin"] } },
    );
-    expect(capture.senderIsOwner).toBe(false);
+    expect(mocks.dispatchChannelMessageAction).toHaveBeenLastCalledWith(
+      expect.objectContaining({ senderIsOwner: false }),
+    );
  });
 });
@@ -60,27 +60,29 @@ const getInflightMap = (context: GatewayRequestContext) => {
  return inflight;
 };

-async function resolveGatewayInflightMap(params: {
-  context: GatewayRequestContext;
-  dedupeKey: string;
-  respond: RespondFn;
-}): Promise<Map<string, Promise<InflightResult>> | undefined> {
+function resolveGatewayInflightMap(params: { context: GatewayRequestContext; dedupeKey: string }):
+  | {
+      kind: "cached";
+      cached: NonNullable<ReturnType<GatewayRequestContext["dedupe"]["get"]>>;
+    }
+  | {
+      kind: "inflight";
+      inflight: Promise<InflightResult>;
+    }
+  | {
+      kind: "ready";
+      inflightMap: Map<string, Promise<InflightResult>>;
+    } {
  const cached = params.context.dedupe.get(params.dedupeKey);
  if (cached) {
-    params.respond(cached.ok, cached.payload, cached.error, {
-      cached: true,
-    });
-    return undefined;
+    return { kind: "cached", cached };
  }
  const inflightMap = getInflightMap(params.context);
  const inflight = inflightMap.get(params.dedupeKey);
  if (inflight) {
-    const result = await inflight;
-    const meta = result.meta ? { ...result.meta, cached: true } : { cached: true };
-    params.respond(result.ok, result.payload, result.error, meta);
-    return undefined;
+    return { kind: "inflight", inflight };
  }
-  return inflightMap;
+  return { kind: "ready", inflightMap };
 }

 async function runGatewayInflightWork(params: {
@@ -312,35 +314,45 @@ export const sendHandlers: GatewayRequestHandlers = {
    const senderIsOwner = callerIsFullOperator && request.senderIsOwner === true;
    const idem = request.idempotencyKey;
    const dedupeKey = `message.action:${idem}`;
-    const inflightMap = await resolveGatewayInflightMap({ context, dedupeKey, respond });
-    if (!inflightMap) {
+    const inflight = resolveGatewayInflightMap({ context, dedupeKey });
+    if (inflight.kind === "cached") {
+      respond(inflight.cached.ok, inflight.cached.payload, inflight.cached.error, {
+        cached: true,
+      });
      return;
    }
-    const resolvedChannel = await resolveRequestedChannel({
-      requestChannel: request.channel,
-      unsupportedMessage: (input) => `unsupported channel: ${input}`,
-      context,
-      rejectWebchatAsInternalOnly: true,
-    });
-    if ("error" in resolvedChannel) {
-      respond(false, undefined, resolvedChannel.error);
+    if (inflight.kind === "inflight") {
+      const result = await inflight.inflight;
+      const meta = result.meta ? { ...result.meta, cached: true } : { cached: true };
+      respond(result.ok, result.payload, result.error, meta);
      return;
    }
-    const { cfg, channel } = resolvedChannel;
-    const plugin = resolveOutboundChannelPlugin({ channel, cfg });
-    if (!plugin?.actions?.handleAction) {
-      respond(
-        false,
-        undefined,
-        errorShape(
-          ErrorCodes.INVALID_REQUEST,
-          `Channel ${channel} does not support action ${request.action}.`,
-        ),
-      );
+    if (inflight.kind !== "ready") {
      return;
    }
-
+    const inflightMap = inflight.inflightMap;
    const work = (async (): Promise<InflightResult> => {
+      const resolvedChannel = await resolveRequestedChannel({
+        requestChannel: request.channel,
+        unsupportedMessage: (input) => `unsupported channel: ${input}`,
+        context,
+        rejectWebchatAsInternalOnly: true,
+      });
+      if ("error" in resolvedChannel) {
+        return { ok: false, error: resolvedChannel.error };
+      }
+      const { cfg, channel } = resolvedChannel;
+      const plugin = resolveOutboundChannelPlugin({ channel, cfg });
+      if (!plugin?.actions?.handleAction) {
+        return {
+          ok: false,
+          error: errorShape(
+            ErrorCodes.INVALID_REQUEST,
+            `Channel ${channel} does not support action ${request.action}.`,
+          ),
+        };
+      }
+
      try {
        const handled = await dispatchChannelMessageAction({
          channel,
@@ -410,10 +422,20 @@ export const sendHandlers: GatewayRequestHandlers = {
    };
    const idem = request.idempotencyKey;
    const dedupeKey = `send:${idem}`;
-    const inflightMap = await resolveGatewayInflightMap({ context, dedupeKey, respond });
-    if (!inflightMap) {
+    const inflight = resolveGatewayInflightMap({ context, dedupeKey });
+    if (inflight.kind === "cached") {
+      respond(inflight.cached.ok, inflight.cached.payload, inflight.cached.error, {
+        cached: true,
+      });
      return;
    }
+    if (inflight.kind === "inflight") {
+      const result = await inflight.inflight;
+      const meta = result.meta ? { ...result.meta, cached: true } : { cached: true };
+      respond(result.ok, result.payload, result.error, meta);
+      return;
+    }
+    const inflightMap = inflight.inflightMap;
    const to = normalizeOptionalString(request.to) ?? "";
    const message = normalizeOptionalString(request.message) ?? "";
    const mediaUrl = normalizeOptionalString(request.mediaUrl);
@@ -430,32 +452,30 @@ export const sendHandlers: GatewayRequestHandlers = {
      );
      return;
    }
-    const resolvedChannel = await resolveRequestedChannel({
-      requestChannel: request.channel,
-      unsupportedMessage: (input) => `unsupported channel: ${input}`,
-      context,
-      rejectWebchatAsInternalOnly: true,
-    });
-    if ("error" in resolvedChannel) {
-      respond(false, undefined, resolvedChannel.error);
-      return;
-    }
-    const { cfg, channel } = resolvedChannel;
    const accountId = normalizeOptionalString(request.accountId);
    const replyToId = normalizeOptionalString(request.replyToId);
    const threadId = normalizeOptionalString(request.threadId);
-    const outboundChannel = channel;
-    const plugin = resolveOutboundChannelPlugin({ channel, cfg });
-    if (!plugin) {
-      respond(
-        false,
-        undefined,
-        errorShape(ErrorCodes.INVALID_REQUEST, `unsupported channel: ${channel}`),
-      );
-      return;
-    }

    const work = (async (): Promise<InflightResult> => {
+      const resolvedChannel = await resolveRequestedChannel({
+        requestChannel: request.channel,
+        unsupportedMessage: (input) => `unsupported channel: ${input}`,
+        context,
+        rejectWebchatAsInternalOnly: true,
+      });
+      if ("error" in resolvedChannel) {
+        return { ok: false, error: resolvedChannel.error };
+      }
+      const { cfg, channel } = resolvedChannel;
+      const outboundChannel = channel;
+      const plugin = resolveOutboundChannelPlugin({ channel, cfg });
+      if (!plugin) {
+        return {
+          ok: false,
+          error: errorShape(ErrorCodes.INVALID_REQUEST, `unsupported channel: ${channel}`),
+        };
+      }
+
      try {
        const resolvedTarget = resolveGatewayOutboundTarget({
          channel: outboundChannel,
@@ -617,107 +637,114 @@ export const sendHandlers: GatewayRequestHandlers = {
      idempotencyKey: string;
    };
    const idem = request.idempotencyKey;
-    const cached = context.dedupe.get(`poll:${idem}`);
-    if (cached) {
-      respond(cached.ok, cached.payload, cached.error, {
+    const dedupeKey = `poll:${idem}`;
+    const inflight = resolveGatewayInflightMap({ context, dedupeKey });
+    if (inflight.kind === "cached") {
+      respond(inflight.cached.ok, inflight.cached.payload, inflight.cached.error, {
        cached: true,
      });
      return;
    }
-    const to = request.to.trim();
-    const resolvedChannel = await resolveRequestedChannel({
-      requestChannel: request.channel,
-      unsupportedMessage: (input) => `unsupported poll channel: ${input}`,
-      context,
-    });
-    if ("error" in resolvedChannel) {
-      respond(false, undefined, resolvedChannel.error);
+    if (inflight.kind === "inflight") {
+      const result = await inflight.inflight;
+      const meta = result.meta ? { ...result.meta, cached: true } : { cached: true };
+      respond(result.ok, result.payload, result.error, meta);
      return;
    }
-    const { cfg, channel } = resolvedChannel;
-    const plugin = resolveOutboundChannelPlugin({ channel, cfg });
-    const outbound = plugin?.outbound;
-    if (
-      typeof request.durationSeconds === "number" &&
-      outbound?.supportsPollDurationSeconds !== true
-    ) {
-      respond(
-        false,
-        undefined,
-        errorShape(
-          ErrorCodes.INVALID_REQUEST,
-          `durationSeconds is not supported for ${channel} polls`,
-        ),
-      );
+    if (inflight.kind !== "ready") {
      return;
    }
-    if (typeof request.isAnonymous === "boolean" && outbound?.supportsAnonymousPolls !== true) {
-      respond(
-        false,
-        undefined,
-        errorShape(ErrorCodes.INVALID_REQUEST, `isAnonymous is not supported for ${channel} polls`),
-      );
-      return;
-    }
-    const poll = {
-      question: request.question,
-      options: request.options,
-      maxSelections: request.maxSelections,
-      durationSeconds: request.durationSeconds,
-      durationHours: request.durationHours,
-    };
-    const threadId = normalizeOptionalString(request.threadId);
-    const accountId = normalizeOptionalString(request.accountId);
-    try {
-      if (!outbound?.sendPoll) {
-        respond(
-          false,
-          undefined,
-          errorShape(ErrorCodes.INVALID_REQUEST, `unsupported poll channel: ${channel}`),
-        );
-        return;
-      }
-      const resolvedTarget = resolveGatewayOutboundTarget({
-        channel: channel,
-        to,
-        cfg,
-        accountId,
-      });
-      if (!resolvedTarget.ok) {
-        respond(false, undefined, resolvedTarget.error);
-        return;
-      }
-      const normalized = outbound.pollMaxOptions
-        ? normalizePollInput(poll, { maxOptions: outbound.pollMaxOptions })
-        : normalizePollInput(poll);
-      const result = await outbound.sendPoll({
-        cfg,
-        to: resolvedTarget.to,
-        poll: normalized,
-        accountId,
-        threadId,
-        silent: request.silent,
-        isAnonymous: request.isAnonymous,
-        gatewayClientScopes: client?.connect?.scopes ?? [],
-      });
-      const payload = buildGatewayDeliveryPayload({ runId: idem, channel, result });
-      cacheGatewayDedupeSuccess({
+    const inflightMap = inflight.inflightMap;
+    const work = (async (): Promise<InflightResult> => {
+      const resolvedChannel = await resolveRequestedChannel({
+        requestChannel: request.channel,
+        unsupportedMessage: (input) => `unsupported poll channel: ${input}`,
        context,
-        dedupeKey: `poll:${idem}`,
-        payload,
      });
-      respond(true, payload, undefined, { channel });
-    } catch (err) {
-      const error = errorShape(ErrorCodes.UNAVAILABLE, String(err));
-      cacheGatewayDedupeFailure({
-        context,
-        dedupeKey: `poll:${idem}`,
-        error,
-      });
-      respond(false, undefined, error, {
-        channel,
-        error: formatForLog(err),
-      });
-    }
+      if ("error" in resolvedChannel) {
+        return { ok: false, error: resolvedChannel.error };
+      }
+      const { cfg, channel } = resolvedChannel;
+      const plugin = resolveOutboundChannelPlugin({ channel, cfg });
+      const outbound = plugin?.outbound;
+      if (
+        typeof request.durationSeconds === "number" &&
+        outbound?.supportsPollDurationSeconds !== true
+      ) {
+        return {
+          ok: false,
+          error: errorShape(
+            ErrorCodes.INVALID_REQUEST,
+            `durationSeconds is not supported for ${channel} polls`,
+          ),
+        };
+      }
+      if (typeof request.isAnonymous === "boolean" && outbound?.supportsAnonymousPolls !== true) {
+        return {
+          ok: false,
+          error: errorShape(
+            ErrorCodes.INVALID_REQUEST,
+            `isAnonymous is not supported for ${channel} polls`,
+          ),
+        };
+      }
+      const poll = {
+        question: request.question,
+        options: request.options,
+        maxSelections: request.maxSelections,
+        durationSeconds: request.durationSeconds,
+        durationHours: request.durationHours,
+      };
+      const threadId = normalizeOptionalString(request.threadId);
+      const accountId = normalizeOptionalString(request.accountId);
+      try {
+        if (!outbound?.sendPoll) {
+          const error = errorShape(
+            ErrorCodes.INVALID_REQUEST,
+            `unsupported poll channel: ${channel}`,
+          );
+          return { ok: false, error };
+        }
+        const resolvedTarget = resolveGatewayOutboundTarget({
+          channel: channel,
+          to: request.to.trim(),
+          cfg,
+          accountId,
+        });
+        if (!resolvedTarget.ok) {
+          return { ok: false, error: resolvedTarget.error };
+        }
+        const normalized = outbound.pollMaxOptions
+          ? normalizePollInput(poll, { maxOptions: outbound.pollMaxOptions })
+          : normalizePollInput(poll);
+        const result = await outbound.sendPoll({
+          cfg,
+          to: resolvedTarget.to,
+          poll: normalized,
+          accountId,
+          threadId,
+          silent: request.silent,
+          isAnonymous: request.isAnonymous,
+          gatewayClientScopes: client?.connect?.scopes ?? [],
+        });
+        const payload = buildGatewayDeliveryPayload({ runId: idem, channel, result });
+        cacheGatewayDedupeSuccess({
+          context,
+          dedupeKey,
+          payload,
+        });
+        return { ok: true, payload, meta: { channel } };
+      } catch (err) {
+        const error = errorShape(ErrorCodes.UNAVAILABLE, String(err));
+        cacheGatewayDedupeFailure({
+          context,
+          dedupeKey,
+          error,
+        });
+        return { ok: false, error, meta: { channel, error: formatForLog(err) } };
+      }
+    })();
+
+    await runGatewayInflightWork({ inflightMap, dedupeKey, work, respond });
  },
 };