fix: decouple Discord inbound worker timeout from listener timeout (#36602) (thanks @dutifulbob) (#36602)

Co-authored-by: Onur Solmaz <2453968+osolmaz@users.noreply.github.com>

src/discord/monitor/inbound-job.test.ts

@@ -0,0 +1,148 @@
+import { Message } from "@buape/carbon";
+import { describe, expect, it } from "vitest";
+import { buildDiscordInboundJob, materializeDiscordInboundJob } from "./inbound-job.js";
+import { createBaseDiscordMessageContext } from "./message-handler.test-harness.js";
+
+describe("buildDiscordInboundJob", () => {
+  it("keeps live runtime references out of the payload", async () => {
+    const ctx = await createBaseDiscordMessageContext({
+      message: {
+        id: "m1",
+        channelId: "thread-1",
+        timestamp: new Date().toISOString(),
+        attachments: [],
+        channel: {
+          id: "thread-1",
+          isThread: () => true,
+        },
+      },
+      data: {
+        guild: { id: "g1", name: "Guild" },
+        message: {
+          id: "m1",
+          channelId: "thread-1",
+          timestamp: new Date().toISOString(),
+          attachments: [],
+          channel: {
+            id: "thread-1",
+            isThread: () => true,
+          },
+        },
+      },
+      threadChannel: {
+        id: "thread-1",
+        name: "codex",
+        parentId: "forum-1",
+        parent: {
+          id: "forum-1",
+          name: "Forum",
+        },
+        ownerId: "user-1",
+      },
+    });
+
+    const job = buildDiscordInboundJob(ctx);
+
+    expect("runtime" in job.payload).toBe(false);
+    expect("client" in job.payload).toBe(false);
+    expect("threadBindings" in job.payload).toBe(false);
+    expect("discordRestFetch" in job.payload).toBe(false);
+    expect("channel" in job.payload.message).toBe(false);
+    expect("channel" in job.payload.data.message).toBe(false);
+    expect(job.runtime.client).toBe(ctx.client);
+    expect(job.runtime.threadBindings).toBe(ctx.threadBindings);
+    expect(job.payload.threadChannel).toEqual({
+      id: "thread-1",
+      name: "codex",
+      parentId: "forum-1",
+      parent: {
+        id: "forum-1",
+        name: "Forum",
+      },
+      ownerId: "user-1",
+    });
+    expect(() => JSON.stringify(job.payload)).not.toThrow();
+  });
+
+  it("re-materializes the process context with an overridden abort signal", async () => {
+    const ctx = await createBaseDiscordMessageContext();
+    const job = buildDiscordInboundJob(ctx);
+    const overrideAbortController = new AbortController();
+
+    const rematerialized = materializeDiscordInboundJob(job, overrideAbortController.signal);
+
+    expect(rematerialized.runtime).toBe(ctx.runtime);
+    expect(rematerialized.client).toBe(ctx.client);
+    expect(rematerialized.threadBindings).toBe(ctx.threadBindings);
+    expect(rematerialized.abortSignal).toBe(overrideAbortController.signal);
+    expect(rematerialized.message).toEqual(job.payload.message);
+    expect(rematerialized.data).toEqual(job.payload.data);
+  });
+
+  it("preserves Carbon message getters across queued jobs", async () => {
+    const ctx = await createBaseDiscordMessageContext();
+    const message = new Message(
+      ctx.client as never,
+      {
+        id: "m1",
+        channel_id: "c1",
+        content: "hello",
+        attachments: [{ id: "a1", filename: "note.txt" }],
+        timestamp: new Date().toISOString(),
+        author: {
+          id: "u1",
+          username: "alice",
+          discriminator: "0",
+          avatar: null,
+        },
+        referenced_message: {
+          id: "m0",
+          channel_id: "c1",
+          content: "earlier",
+          attachments: [],
+          timestamp: new Date().toISOString(),
+          author: {
+            id: "u2",
+            username: "bob",
+            discriminator: "0",
+            avatar: null,
+          },
+          type: 0,
+          tts: false,
+          mention_everyone: false,
+          pinned: false,
+          flags: 0,
+        },
+        type: 0,
+        tts: false,
+        mention_everyone: false,
+        pinned: false,
+        flags: 0,
+      } as ConstructorParameters<typeof Message>[1],
+    );
+    const runtimeChannel = { id: "c1", isThread: () => false };
+    Object.defineProperty(message, "channel", {
+      value: runtimeChannel,
+      configurable: true,
+      enumerable: true,
+      writable: true,
+    });
+
+    const job = buildDiscordInboundJob({
+      ...ctx,
+      message,
+      data: {
+        ...ctx.data,
+        message,
+      },
+    });
+    const rematerialized = materializeDiscordInboundJob(job);
+
+    expect(job.payload.message).toBeInstanceOf(Message);
+    expect("channel" in job.payload.message).toBe(false);
+    expect(rematerialized.message.content).toBe("hello");
+    expect(rematerialized.message.attachments).toHaveLength(1);
+    expect(rematerialized.message.timestamp).toBe(message.timestamp);
+    expect(rematerialized.message.referencedMessage?.content).toBe("earlier");
+  });
+});

src/discord/monitor/inbound-job.ts

@@ -0,0 +1,111 @@
+import type { DiscordMessagePreflightContext } from "./message-handler.preflight.types.js";
+
+type DiscordInboundJobRuntimeField =
+  | "runtime"
+  | "abortSignal"
+  | "guildHistories"
+  | "client"
+  | "threadBindings"
+  | "discordRestFetch";
+
+export type DiscordInboundJobRuntime = Pick<
+  DiscordMessagePreflightContext,
+  DiscordInboundJobRuntimeField
+>;
+
+export type DiscordInboundJobPayload = Omit<
+  DiscordMessagePreflightContext,
+  DiscordInboundJobRuntimeField
+>;
+
+export type DiscordInboundJob = {
+  queueKey: string;
+  payload: DiscordInboundJobPayload;
+  runtime: DiscordInboundJobRuntime;
+};
+
+export function resolveDiscordInboundJobQueueKey(ctx: DiscordMessagePreflightContext): string {
+  const sessionKey = ctx.route.sessionKey?.trim();
+  if (sessionKey) {
+    return sessionKey;
+  }
+  const baseSessionKey = ctx.baseSessionKey?.trim();
+  if (baseSessionKey) {
+    return baseSessionKey;
+  }
+  return ctx.messageChannelId;
+}
+
+export function buildDiscordInboundJob(ctx: DiscordMessagePreflightContext): DiscordInboundJob {
+  const {
+    runtime,
+    abortSignal,
+    guildHistories,
+    client,
+    threadBindings,
+    discordRestFetch,
+    message,
+    data,
+    threadChannel,
+    ...payload
+  } = ctx;
+
+  const sanitizedMessage = sanitizeDiscordInboundMessage(message);
+  return {
+    queueKey: resolveDiscordInboundJobQueueKey(ctx),
+    payload: {
+      ...payload,
+      message: sanitizedMessage,
+      data: {
+        ...data,
+        message: sanitizedMessage,
+      },
+      threadChannel: normalizeDiscordThreadChannel(threadChannel),
+    },
+    runtime: {
+      runtime,
+      abortSignal,
+      guildHistories,
+      client,
+      threadBindings,
+      discordRestFetch,
+    },
+  };
+}
+
+export function materializeDiscordInboundJob(
+  job: DiscordInboundJob,
+  abortSignal?: AbortSignal,
+): DiscordMessagePreflightContext {
+  return {
+    ...job.payload,
+    ...job.runtime,
+    abortSignal: abortSignal ?? job.runtime.abortSignal,
+  };
+}
+
+function sanitizeDiscordInboundMessage<T extends object>(message: T): T {
+  const descriptors = Object.getOwnPropertyDescriptors(message);
+  delete descriptors.channel;
+  return Object.create(Object.getPrototypeOf(message), descriptors) as T;
+}
+
+function normalizeDiscordThreadChannel(
+  threadChannel: DiscordMessagePreflightContext["threadChannel"],
+): DiscordMessagePreflightContext["threadChannel"] {
+  if (!threadChannel) {
+    return null;
+  }
+  return {
+    id: threadChannel.id,
+    name: threadChannel.name,
+    parentId: threadChannel.parentId,
+    parent: threadChannel.parent
+      ? {
+          id: threadChannel.parent.id,
+          name: threadChannel.parent.name,
+        }
+      : undefined,
+    ownerId: threadChannel.ownerId,
+  };
+}

src/discord/monitor/inbound-worker.ts

@@ -0,0 +1,105 @@
+import { createRunStateMachine } from "../../channels/run-state-machine.js";
+import { danger } from "../../globals.js";
+import { formatDurationSeconds } from "../../infra/format-time/format-duration.ts";
+import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js";
+import { materializeDiscordInboundJob, type DiscordInboundJob } from "./inbound-job.js";
+import type { RuntimeEnv } from "./message-handler.preflight.types.js";
+import { processDiscordMessage } from "./message-handler.process.js";
+import type { DiscordMonitorStatusSink } from "./status.js";
+import { normalizeDiscordInboundWorkerTimeoutMs, runDiscordTaskWithTimeout } from "./timeouts.js";
+
+type DiscordInboundWorkerParams = {
+  runtime: RuntimeEnv;
+  setStatus?: DiscordMonitorStatusSink;
+  abortSignal?: AbortSignal;
+  runTimeoutMs?: number;
+};
+
+export type DiscordInboundWorker = {
+  enqueue: (job: DiscordInboundJob) => void;
+  deactivate: () => void;
+};
+
+function formatDiscordRunContextSuffix(job: DiscordInboundJob): string {
+  const channelId = job.payload.messageChannelId?.trim();
+  const messageId = job.payload.data?.message?.id?.trim();
+  const details = [
+    channelId ? `channelId=${channelId}` : null,
+    messageId ? `messageId=${messageId}` : null,
+  ].filter((entry): entry is string => Boolean(entry));
+  if (details.length === 0) {
+    return "";
+  }
+  return ` (${details.join(", ")})`;
+}
+
+async function processDiscordInboundJob(params: {
+  job: DiscordInboundJob;
+  runtime: RuntimeEnv;
+  lifecycleSignal?: AbortSignal;
+  runTimeoutMs?: number;
+}) {
+  const timeoutMs = normalizeDiscordInboundWorkerTimeoutMs(params.runTimeoutMs);
+  const contextSuffix = formatDiscordRunContextSuffix(params.job);
+  await runDiscordTaskWithTimeout({
+    run: async (abortSignal) => {
+      await processDiscordMessage(materializeDiscordInboundJob(params.job, abortSignal));
+    },
+    timeoutMs,
+    abortSignals: [params.job.runtime.abortSignal, params.lifecycleSignal],
+    onTimeout: (resolvedTimeoutMs) => {
+      params.runtime.error?.(
+        danger(
+          `discord inbound worker timed out after ${formatDurationSeconds(resolvedTimeoutMs, {
+            decimals: 1,
+            unit: "seconds",
+          })}${contextSuffix}`,
+        ),
+      );
+    },
+    onErrorAfterTimeout: (error) => {
+      params.runtime.error?.(
+        danger(`discord inbound worker failed after timeout: ${String(error)}${contextSuffix}`),
+      );
+    },
+  });
+}
+
+export function createDiscordInboundWorker(
+  params: DiscordInboundWorkerParams,
+): DiscordInboundWorker {
+  const runQueue = new KeyedAsyncQueue();
+  const runState = createRunStateMachine({
+    setStatus: params.setStatus,
+    abortSignal: params.abortSignal,
+  });
+
+  return {
+    enqueue(job) {
+      void runQueue
+        .enqueue(job.queueKey, async () => {
+          if (!runState.isActive()) {
+            return;
+          }
+          runState.onRunStart();
+          try {
+            if (!runState.isActive()) {
+              return;
+            }
+            await processDiscordInboundJob({
+              job,
+              runtime: params.runtime,
+              lifecycleSignal: params.abortSignal,
+              runTimeoutMs: params.runTimeoutMs,
+            });
+          } finally {
+            runState.onRunEnd();
+          }
+        })
+        .catch((error) => {
+          params.runtime.error?.(danger(`discord inbound worker failed: ${String(error)}`));
+        });
+    },
+    deactivate: runState.deactivate,
+  };
+}

src/discord/monitor/listeners.ts

@@ -34,6 +34,7 @@ import { resolveDiscordChannelInfo } from "./message-utils.js";
 import { setPresence } from "./presence-cache.js";
 import { isThreadArchived } from "./thread-bindings.discord-api.js";
 import { closeDiscordThreadSessions } from "./thread-session-close.js";
+import { normalizeDiscordListenerTimeoutMs, runDiscordTaskWithTimeout } from "./timeouts.js";
 
 type LoadedConfig = ReturnType<typeof import("../../config/config.js").loadConfig>;
 type RuntimeEnv = import("../../runtime.js").RuntimeEnv;
@@ -70,16 +71,8 @@ type DiscordReactionRoutingParams = {
 };
 
 const DISCORD_SLOW_LISTENER_THRESHOLD_MS = 30_000;
-const DISCORD_DEFAULT_LISTENER_TIMEOUT_MS = 120_000;
 const discordEventQueueLog = createSubsystemLogger("discord/event-queue");
 
-function normalizeDiscordListenerTimeoutMs(raw: number | undefined): number {
-  if (!Number.isFinite(raw) || (raw ?? 0) <= 0) {
-    return DISCORD_DEFAULT_LISTENER_TIMEOUT_MS;
-  }
-  return Math.max(1_000, Math.floor(raw!));
-}
-
 function formatListenerContextValue(value: unknown): string | null {
   if (value === undefined || value === null) {
     return null;
@@ -138,57 +131,44 @@ async function runDiscordListenerWithSlowLog(params: {
   logger: Logger | undefined;
   listener: string;
   event: string;
-  run: (abortSignal: AbortSignal) => Promise<void>;
+  run: (abortSignal: AbortSignal | undefined) => Promise<void>;
   timeoutMs?: number;
   context?: Record<string, unknown>;
   onError?: (err: unknown) => void;
 }) {
   const startedAt = Date.now();
   const timeoutMs = normalizeDiscordListenerTimeoutMs(params.timeoutMs);
-  let timedOut = false;
-  let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
   const logger = params.logger ?? discordEventQueueLog;
-  const abortController = new AbortController();
-  const runPromise = params.run(abortController.signal).catch((err) => {
-    if (timedOut) {
-      const errorName =
-        err && typeof err === "object" && "name" in err ? String(err.name) : undefined;
-      if (abortController.signal.aborted && errorName === "AbortError") {
+  let timedOut = false;
+
+  try {
+    timedOut = await runDiscordTaskWithTimeout({
+      run: params.run,
+      timeoutMs,
+      onTimeout: (resolvedTimeoutMs) => {
+        logger.error(
+          danger(
+            `discord handler timed out after ${formatDurationSeconds(resolvedTimeoutMs, {
+              decimals: 1,
+              unit: "seconds",
+            })}${formatListenerContextSuffix(params.context)}`,
+          ),
+        );
+      },
+      onAbortAfterTimeout: () => {
         logger.warn(
           `discord handler canceled after timeout${formatListenerContextSuffix(params.context)}`,
         );
-        return;
-      }
-      logger.error(
-        danger(
-          `discord handler failed after timeout: ${String(err)}${formatListenerContextSuffix(params.context)}`,
-        ),
-      );
-      return;
-    }
-    throw err;
-  });
-
-  try {
-    const timeoutPromise = new Promise<"timeout">((resolve) => {
-      timeoutHandle = setTimeout(() => resolve("timeout"), timeoutMs);
-      timeoutHandle.unref?.();
+      },
+      onErrorAfterTimeout: (err) => {
+        logger.error(
+          danger(
+            `discord handler failed after timeout: ${String(err)}${formatListenerContextSuffix(params.context)}`,
+          ),
+        );
+      },
     });
-    const result = await Promise.race([
-      runPromise.then(() => "completed" as const),
-      timeoutPromise,
-    ]);
-    if (result === "timeout") {
-      timedOut = true;
-      abortController.abort();
-      logger.error(
-        danger(
-          `discord handler timed out after ${formatDurationSeconds(timeoutMs, {
-            decimals: 1,
-            unit: "seconds",
-          })}${formatListenerContextSuffix(params.context)}`,
-        ),
-      );
+    if (timedOut) {
       return;
     }
   } catch (err) {
@@ -198,9 +178,6 @@ async function runDiscordListenerWithSlowLog(params: {
     }
     throw err;
   } finally {
-    if (timeoutHandle) {
-      clearTimeout(timeoutHandle);
-    }
     if (!timedOut) {
       logSlowDiscordListener({
         logger: params.logger,

src/discord/monitor/message-handler.process.ts

@@ -1,4 +1,4 @@
-import { ChannelType } from "@buape/carbon";
+import { ChannelType, type RequestClient } from "@buape/carbon";
 import { resolveAckReaction, resolveHumanDelayConfig } from "../../agents/identity.js";
 import { EmbeddedBlockChunker } from "../../agents/pi-embedded-block-chunker.js";
 import { resolveChunkMode } from "../../auto-reply/chunk.js";
@@ -161,15 +161,17 @@ export async function processDiscordMessage(ctx: DiscordMessagePreflightContext)
       }),
     );
   const statusReactionsEnabled = shouldAckReaction();
+  // Discord outbound helpers expect Carbon's request client shape explicitly.
+  const discordRest = client.rest as unknown as RequestClient;
   const discordAdapter: StatusReactionAdapter = {
     setReaction: async (emoji) => {
       await reactMessageDiscord(messageChannelId, message.id, emoji, {
-        rest: client.rest as never,
+        rest: discordRest,
       });
     },
     removeReaction: async (emoji) => {
       await removeReactionDiscord(messageChannelId, message.id, emoji, {
-        rest: client.rest as never,
+        rest: discordRest,
       });
     },
   };

src/discord/monitor/message-handler.queue.test.ts

@@ -4,6 +4,7 @@ import { createNoopThreadBindingManager } from "./thread-bindings.js";
 
 const preflightDiscordMessageMock = vi.hoisted(() => vi.fn());
 const processDiscordMessageMock = vi.hoisted(() => vi.fn());
+const eventualReplyDeliveredMock = vi.hoisted(() => vi.fn());
 
 vi.mock("./message-handler.preflight.js", () => ({
   preflightDiscordMessage: preflightDiscordMessageMock,
@@ -26,7 +27,7 @@ function createDeferred<T = void>() {
 function createHandlerParams(overrides?: {
   setStatus?: (patch: Record<string, unknown>) => void;
   abortSignal?: AbortSignal;
-  listenerTimeoutMs?: number;
+  workerRunTimeoutMs?: number;
 }) {
   const cfg: OpenClawConfig = {
     channels: {
@@ -65,7 +66,7 @@ function createHandlerParams(overrides?: {
     threadBindings: createNoopThreadBindingManager("default"),
     setStatus: overrides?.setStatus,
     abortSignal: overrides?.abortSignal,
-    listenerTimeoutMs: overrides?.listenerTimeoutMs,
+    workerRunTimeoutMs: overrides?.workerRunTimeoutMs,
   };
 }
 
@@ -85,6 +86,19 @@ function createMessageData(messageId: string, channelId = "ch-1") {
 
 function createPreflightContext(channelId = "ch-1") {
   return {
+    data: {
+      channel_id: channelId,
+      message: {
+        id: `msg-${channelId}`,
+        channel_id: channelId,
+        attachments: [],
+      },
+    },
+    message: {
+      id: `msg-${channelId}`,
+      channel_id: channelId,
+      attachments: [],
+    },
     route: {
       sessionKey: `agent:main:discord:channel:${channelId}`,
     },
@@ -169,7 +183,7 @@ describe("createDiscordMessageHandler queue behavior", () => {
     });
   });
 
-  it("applies listener timeout to queued runs so stalled runs do not block the queue", async () => {
+  it("applies explicit inbound worker timeout to queued runs so stalled runs do not block the queue", async () => {
     vi.useFakeTimers();
     try {
       preflightDiscordMessageMock.mockReset();
@@ -191,7 +205,7 @@ describe("createDiscordMessageHandler queue behavior", () => {
           createPreflightContext(params.data.channel_id),
       );
 
-      const params = createHandlerParams({ listenerTimeoutMs: 50 });
+      const params = createHandlerParams({ workerRunTimeoutMs: 50 });
       const handler = createDiscordMessageHandler(params);
 
       await expect(
@@ -211,7 +225,50 @@ describe("createDiscordMessageHandler queue behavior", () => {
         | undefined;
       expect(firstCtx?.abortSignal?.aborted).toBe(true);
       expect(params.runtime.error).toHaveBeenCalledWith(
-        expect.stringContaining("discord queued run timed out after"),
+        expect.stringContaining("discord inbound worker timed out after"),
+      );
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not time out queued runs when the inbound worker timeout is disabled", async () => {
+    vi.useFakeTimers();
+    try {
+      preflightDiscordMessageMock.mockReset();
+      processDiscordMessageMock.mockReset();
+      eventualReplyDeliveredMock.mockReset();
+
+      processDiscordMessageMock.mockImplementationOnce(
+        async (ctx: { abortSignal?: AbortSignal }) => {
+          await new Promise<void>((resolve) => {
+            setTimeout(() => {
+              if (!ctx.abortSignal?.aborted) {
+                eventualReplyDeliveredMock();
+              }
+              resolve();
+            }, 80);
+          });
+        },
+      );
+      preflightDiscordMessageMock.mockImplementation(
+        async (params: { data: { channel_id: string } }) =>
+          createPreflightContext(params.data.channel_id),
+      );
+
+      const params = createHandlerParams({ workerRunTimeoutMs: 0 });
+      const handler = createDiscordMessageHandler(params);
+
+      await expect(
+        handler(createMessageData("m-1") as never, {} as never),
+      ).resolves.toBeUndefined();
+
+      await vi.advanceTimersByTimeAsync(80);
+      await Promise.resolve();
+
+      expect(eventualReplyDeliveredMock).toHaveBeenCalledTimes(1);
+      expect(params.runtime.error).not.toHaveBeenCalledWith(
+        expect.stringContaining("discord inbound worker timed out after"),
       );
     } finally {
       vi.useRealTimers();

src/discord/monitor/message-handler.ts

@@ -3,18 +3,13 @@ import {
   createChannelInboundDebouncer,
   shouldDebounceTextInbound,
 } from "../../channels/inbound-debounce-policy.js";
-import { createRunStateMachine } from "../../channels/run-state-machine.js";
 import { resolveOpenProviderRuntimeGroupPolicy } from "../../config/runtime-group-policy.js";
 import { danger } from "../../globals.js";
-import { formatDurationSeconds } from "../../infra/format-time/format-duration.ts";
-import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js";
+import { buildDiscordInboundJob } from "./inbound-job.js";
+import { createDiscordInboundWorker } from "./inbound-worker.js";
 import type { DiscordMessageEvent, DiscordMessageHandler } from "./listeners.js";
 import { preflightDiscordMessage } from "./message-handler.preflight.js";
-import type {
-  DiscordMessagePreflightContext,
-  DiscordMessagePreflightParams,
-} from "./message-handler.preflight.types.js";
-import { processDiscordMessage } from "./message-handler.process.js";
+import type { DiscordMessagePreflightParams } from "./message-handler.preflight.types.js";
 import {
   hasDiscordMessageStickers,
   resolveDiscordMessageChannelId,
@@ -28,154 +23,13 @@ type DiscordMessageHandlerParams = Omit<
 > & {
   setStatus?: DiscordMonitorStatusSink;
   abortSignal?: AbortSignal;
-  listenerTimeoutMs?: number;
+  workerRunTimeoutMs?: number;
 };
 
 export type DiscordMessageHandlerWithLifecycle = DiscordMessageHandler & {
   deactivate: () => void;
 };
 
-const DEFAULT_DISCORD_RUN_TIMEOUT_MS = 120_000;
-const MAX_DISCORD_TIMEOUT_MS = 2_147_483_647;
-
-function normalizeDiscordRunTimeoutMs(timeoutMs?: number): number {
-  if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs) || timeoutMs <= 0) {
-    return DEFAULT_DISCORD_RUN_TIMEOUT_MS;
-  }
-  return Math.max(1, Math.min(Math.floor(timeoutMs), MAX_DISCORD_TIMEOUT_MS));
-}
-
-function isAbortError(error: unknown): boolean {
-  if (typeof error !== "object" || error === null) {
-    return false;
-  }
-  return "name" in error && String((error as { name?: unknown }).name) === "AbortError";
-}
-
-function formatDiscordRunContextSuffix(ctx: DiscordMessagePreflightContext): string {
-  const eventData = ctx as {
-    data?: {
-      channel_id?: string;
-      message?: {
-        id?: string;
-      };
-    };
-  };
-  const channelId = ctx.messageChannelId?.trim() || eventData.data?.channel_id?.trim();
-  const messageId = eventData.data?.message?.id?.trim();
-  const details = [
-    channelId ? `channelId=${channelId}` : null,
-    messageId ? `messageId=${messageId}` : null,
-  ].filter((entry): entry is string => Boolean(entry));
-  if (details.length === 0) {
-    return "";
-  }
-  return ` (${details.join(", ")})`;
-}
-
-function mergeAbortSignals(signals: Array<AbortSignal | undefined>): AbortSignal | undefined {
-  const activeSignals = signals.filter((signal): signal is AbortSignal => Boolean(signal));
-  if (activeSignals.length === 0) {
-    return undefined;
-  }
-  if (activeSignals.length === 1) {
-    return activeSignals[0];
-  }
-  if (typeof AbortSignal.any === "function") {
-    return AbortSignal.any(activeSignals);
-  }
-  const fallbackController = new AbortController();
-  for (const signal of activeSignals) {
-    if (signal.aborted) {
-      fallbackController.abort();
-      return fallbackController.signal;
-    }
-  }
-  const abortFallback = () => {
-    fallbackController.abort();
-    for (const signal of activeSignals) {
-      signal.removeEventListener("abort", abortFallback);
-    }
-  };
-  for (const signal of activeSignals) {
-    signal.addEventListener("abort", abortFallback, { once: true });
-  }
-  return fallbackController.signal;
-}
-
-async function processDiscordRunWithTimeout(params: {
-  ctx: DiscordMessagePreflightContext;
-  runtime: DiscordMessagePreflightParams["runtime"];
-  lifecycleSignal?: AbortSignal;
-  timeoutMs?: number;
-}) {
-  const timeoutMs = normalizeDiscordRunTimeoutMs(params.timeoutMs);
-  const timeoutAbortController = new AbortController();
-  const combinedSignal = mergeAbortSignals([
-    params.ctx.abortSignal,
-    params.lifecycleSignal,
-    timeoutAbortController.signal,
-  ]);
-  const processCtx =
-    combinedSignal && combinedSignal !== params.ctx.abortSignal
-      ? { ...params.ctx, abortSignal: combinedSignal }
-      : params.ctx;
-  const contextSuffix = formatDiscordRunContextSuffix(params.ctx);
-  let timedOut = false;
-  let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
-  const processPromise = processDiscordMessage(processCtx).catch((error) => {
-    if (timedOut) {
-      if (timeoutAbortController.signal.aborted && isAbortError(error)) {
-        return;
-      }
-      params.runtime.error?.(
-        danger(`discord queued run failed after timeout: ${String(error)}${contextSuffix}`),
-      );
-      return;
-    }
-    throw error;
-  });
-
-  try {
-    const timeoutPromise = new Promise<"timeout">((resolve) => {
-      timeoutHandle = setTimeout(() => resolve("timeout"), timeoutMs);
-      timeoutHandle.unref?.();
-    });
-    const result = await Promise.race([
-      processPromise.then(() => "completed" as const),
-      timeoutPromise,
-    ]);
-    if (result === "timeout") {
-      timedOut = true;
-      timeoutAbortController.abort();
-      params.runtime.error?.(
-        danger(
-          `discord queued run timed out after ${formatDurationSeconds(timeoutMs, {
-            decimals: 1,
-            unit: "seconds",
-          })}${contextSuffix}`,
-        ),
-      );
-    }
-  } finally {
-    if (timeoutHandle) {
-      clearTimeout(timeoutHandle);
-    }
-  }
-}
-
-function resolveDiscordRunQueueKey(ctx: DiscordMessagePreflightContext): string {
-  const sessionKey = ctx.route.sessionKey?.trim();
-  if (sessionKey) {
-    return sessionKey;
-  }
-  const baseSessionKey = ctx.baseSessionKey?.trim();
-  if (baseSessionKey) {
-    return baseSessionKey;
-  }
-  return ctx.messageChannelId;
-}
-
 export function createDiscordMessageHandler(
   params: DiscordMessageHandlerParams,
 ): DiscordMessageHandlerWithLifecycle {
@@ -188,39 +42,13 @@ export function createDiscordMessageHandler(
     params.discordConfig?.ackReactionScope ??
     params.cfg.messages?.ackReactionScope ??
     "group-mentions";
-  const runQueue = new KeyedAsyncQueue();
-  const runState = createRunStateMachine({
+  const inboundWorker = createDiscordInboundWorker({
+    runtime: params.runtime,
     setStatus: params.setStatus,
     abortSignal: params.abortSignal,
+    runTimeoutMs: params.workerRunTimeoutMs,
   });
 
-  const enqueueDiscordRun = (ctx: DiscordMessagePreflightContext) => {
-    const queueKey = resolveDiscordRunQueueKey(ctx);
-    void runQueue
-      .enqueue(queueKey, async () => {
-        if (!runState.isActive()) {
-          return;
-        }
-        runState.onRunStart();
-        try {
-          if (!runState.isActive()) {
-            return;
-          }
-          await processDiscordRunWithTimeout({
-            ctx,
-            runtime: params.runtime,
-            lifecycleSignal: params.abortSignal,
-            timeoutMs: params.listenerTimeoutMs,
-          });
-        } finally {
-          runState.onRunEnd();
-        }
-      })
-      .catch((err) => {
-        params.runtime.error?.(danger(`discord process failed: ${String(err)}`));
-      });
-  };
-
   const { debouncer } = createChannelInboundDebouncer<{
     data: DiscordMessageEvent;
     client: Client;
@@ -279,7 +107,7 @@ export function createDiscordMessageHandler(
         if (!ctx) {
           return;
         }
-        enqueueDiscordRun(ctx);
+        inboundWorker.enqueue(buildDiscordInboundJob(ctx));
         return;
       }
       const combinedBaseText = entries
@@ -324,7 +152,7 @@ export function createDiscordMessageHandler(
           ctxBatch.MessageSidLast = ids[ids.length - 1];
         }
       }
-      enqueueDiscordRun(ctx);
+      inboundWorker.enqueue(buildDiscordInboundJob(ctx));
     },
     onError: (err) => {
       params.runtime.error?.(danger(`discord debounce flush failed: ${String(err)}`));
@@ -352,7 +180,7 @@ export function createDiscordMessageHandler(
     }
   };
 
-  handler.deactivate = runState.deactivate;
+  handler.deactivate = inboundWorker.deactivate;
 
   return handler;
 }

src/discord/monitor/provider.test.ts

@@ -22,6 +22,7 @@ const {
   clientConstructorOptionsMock,
   createDiscordAutoPresenceControllerMock,
   createDiscordNativeCommandMock,
+  createDiscordMessageHandlerMock,
   createNoopThreadBindingManagerMock,
   createThreadBindingManagerMock,
   reconcileAcpThreadBindingsOnStartupMock,
@@ -49,6 +50,14 @@ const {
     clientFetchUserMock: vi.fn(async (_target: string) => ({ id: "bot-1" })),
     clientGetPluginMock: vi.fn<(_name: string) => unknown>(() => undefined),
     createDiscordNativeCommandMock: vi.fn(() => ({ name: "mock-command" })),
+    createDiscordMessageHandlerMock: vi.fn(() =>
+      Object.assign(
+        vi.fn(async () => undefined),
+        {
+          deactivate: vi.fn(),
+        },
+      ),
+    ),
     createNoopThreadBindingManagerMock: vi.fn(() => {
       const manager = { stop: vi.fn() };
       createdBindingManagers.push(manager);
@@ -248,7 +257,7 @@ vi.mock("./listeners.js", () => ({
 }));
 
 vi.mock("./message-handler.js", () => ({
-  createDiscordMessageHandler: () => ({ handle: vi.fn() }),
+  createDiscordMessageHandler: createDiscordMessageHandlerMock,
 }));
 
 vi.mock("./native-command.js", () => ({
@@ -346,6 +355,14 @@ describe("monitorDiscordProvider", () => {
       refresh: vi.fn(),
       runNow: vi.fn(),
     }));
+    createDiscordMessageHandlerMock.mockClear().mockImplementation(() =>
+      Object.assign(
+        vi.fn(async () => undefined),
+        {
+          deactivate: vi.fn(),
+        },
+      ),
+    );
     clientFetchUserMock.mockClear().mockResolvedValue({ id: "bot-1" });
     clientGetPluginMock.mockClear().mockReturnValue(undefined);
     createDiscordNativeCommandMock.mockClear().mockReturnValue({ name: "mock-command" });
@@ -629,6 +646,63 @@ describe("monitorDiscordProvider", () => {
     expect(eventQueue?.listenerTimeout).toBe(300_000);
   });
 
+  it("does not reuse eventQueue.listenerTimeout as the queued inbound worker timeout", async () => {
+    const { monitorDiscordProvider } = await import("./provider.js");
+
+    resolveDiscordAccountMock.mockImplementation(() => ({
+      accountId: "default",
+      token: "cfg-token",
+      config: {
+        commands: { native: true, nativeSkills: false },
+        voice: { enabled: false },
+        agentComponents: { enabled: false },
+        execApprovals: { enabled: false },
+        eventQueue: { listenerTimeout: 50_000 },
+      },
+    }));
+
+    await monitorDiscordProvider({
+      config: baseConfig(),
+      runtime: baseRuntime(),
+    });
+
+    expect(createDiscordMessageHandlerMock).toHaveBeenCalledTimes(1);
+    const firstCall = createDiscordMessageHandlerMock.mock.calls.at(0) as
+      | [{ workerRunTimeoutMs?: number; listenerTimeoutMs?: number }]
+      | undefined;
+    const params = firstCall?.[0];
+    expect(params?.workerRunTimeoutMs).toBeUndefined();
+    expect("listenerTimeoutMs" in (params ?? {})).toBe(false);
+  });
+
+  it("forwards inbound worker timeout config to the Discord message handler", async () => {
+    const { monitorDiscordProvider } = await import("./provider.js");
+
+    resolveDiscordAccountMock.mockImplementation(() => ({
+      accountId: "default",
+      token: "cfg-token",
+      config: {
+        commands: { native: true, nativeSkills: false },
+        voice: { enabled: false },
+        agentComponents: { enabled: false },
+        execApprovals: { enabled: false },
+        inboundWorker: { runTimeoutMs: 300_000 },
+      },
+    }));
+
+    await monitorDiscordProvider({
+      config: baseConfig(),
+      runtime: baseRuntime(),
+    });
+
+    expect(createDiscordMessageHandlerMock).toHaveBeenCalledTimes(1);
+    const firstCall = createDiscordMessageHandlerMock.mock.calls.at(0) as
+      | [{ workerRunTimeoutMs?: number }]
+      | undefined;
+    const params = firstCall?.[0];
+    expect(params?.workerRunTimeoutMs).toBe(300_000);
+  });
+
   it("registers plugin commands as native Discord commands", async () => {
     const { monitorDiscordProvider } = await import("./provider.js");
     listNativeCommandSpecsForConfigMock.mockReturnValue([

src/discord/monitor/provider.ts

@@ -600,8 +600,9 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
     if (voiceEnabled) {
       clientPlugins.push(new VoicePlugin());
     }
-    // Pass eventQueue config to Carbon so the listener timeout can be tuned.
-    // Default listenerTimeout is 120s (Carbon defaults to 30s which is too short for LLM calls).
+    // Pass eventQueue config to Carbon so the gateway listener budget can be tuned.
+    // Default listenerTimeout is 120s (Carbon defaults to 30s, which is too short for some
+    // Discord normalization/enqueue work).
     const eventQueueOpts = {
       listenerTimeout: 120_000,
       ...discordCfg.eventQueue,
@@ -683,7 +684,7 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
       runtime,
       setStatus: opts.setStatus,
       abortSignal: opts.abortSignal,
-      listenerTimeoutMs: eventQueueOpts.listenerTimeout,
+      workerRunTimeoutMs: discordCfg.inboundWorker?.runTimeoutMs,
       botUserId,
       guildHistories,
       historyLimit,

src/discord/monitor/timeouts.ts

@@ -0,0 +1,120 @@
+const MAX_DISCORD_TIMEOUT_MS = 2_147_483_647;
+
+export const DISCORD_DEFAULT_LISTENER_TIMEOUT_MS = 120_000;
+export const DISCORD_DEFAULT_INBOUND_WORKER_TIMEOUT_MS = 30 * 60_000;
+
+function clampDiscordTimeoutMs(timeoutMs: number, minimumMs: number): number {
+  return Math.max(minimumMs, Math.min(Math.floor(timeoutMs), MAX_DISCORD_TIMEOUT_MS));
+}
+
+export function normalizeDiscordListenerTimeoutMs(raw: number | undefined): number {
+  if (!Number.isFinite(raw) || (raw ?? 0) <= 0) {
+    return DISCORD_DEFAULT_LISTENER_TIMEOUT_MS;
+  }
+  return clampDiscordTimeoutMs(raw!, 1_000);
+}
+
+export function normalizeDiscordInboundWorkerTimeoutMs(
+  raw: number | undefined,
+): number | undefined {
+  if (raw === 0) {
+    return undefined;
+  }
+  if (typeof raw !== "number" || !Number.isFinite(raw) || raw < 0) {
+    return DISCORD_DEFAULT_INBOUND_WORKER_TIMEOUT_MS;
+  }
+  return clampDiscordTimeoutMs(raw, 1);
+}
+
+export function isAbortError(error: unknown): boolean {
+  if (typeof error !== "object" || error === null) {
+    return false;
+  }
+  return "name" in error && String((error as { name?: unknown }).name) === "AbortError";
+}
+
+export function mergeAbortSignals(
+  signals: Array<AbortSignal | undefined>,
+): AbortSignal | undefined {
+  const activeSignals = signals.filter((signal): signal is AbortSignal => Boolean(signal));
+  if (activeSignals.length === 0) {
+    return undefined;
+  }
+  if (activeSignals.length === 1) {
+    return activeSignals[0];
+  }
+  if (typeof AbortSignal.any === "function") {
+    return AbortSignal.any(activeSignals);
+  }
+  const fallbackController = new AbortController();
+  for (const signal of activeSignals) {
+    if (signal.aborted) {
+      fallbackController.abort();
+      return fallbackController.signal;
+    }
+  }
+  const abortFallback = () => {
+    fallbackController.abort();
+    for (const signal of activeSignals) {
+      signal.removeEventListener("abort", abortFallback);
+    }
+  };
+  for (const signal of activeSignals) {
+    signal.addEventListener("abort", abortFallback, { once: true });
+  }
+  return fallbackController.signal;
+}
+
+export async function runDiscordTaskWithTimeout(params: {
+  run: (abortSignal: AbortSignal | undefined) => Promise<void>;
+  timeoutMs?: number;
+  abortSignals?: Array<AbortSignal | undefined>;
+  onTimeout: (timeoutMs: number) => void;
+  onAbortAfterTimeout?: () => void;
+  onErrorAfterTimeout?: (error: unknown) => void;
+}): Promise<boolean> {
+  const timeoutAbortController = params.timeoutMs ? new AbortController() : undefined;
+  const mergedAbortSignal = mergeAbortSignals([
+    ...(params.abortSignals ?? []),
+    timeoutAbortController?.signal,
+  ]);
+
+  let timedOut = false;
+  let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
+  const runPromise = params.run(mergedAbortSignal).catch((error) => {
+    if (!timedOut) {
+      throw error;
+    }
+    if (timeoutAbortController?.signal.aborted && isAbortError(error)) {
+      params.onAbortAfterTimeout?.();
+      return;
+    }
+    params.onErrorAfterTimeout?.(error);
+  });
+
+  try {
+    if (!params.timeoutMs) {
+      await runPromise;
+      return false;
+    }
+    const timeoutPromise = new Promise<"timeout">((resolve) => {
+      timeoutHandle = setTimeout(() => resolve("timeout"), params.timeoutMs);
+      timeoutHandle.unref?.();
+    });
+    const result = await Promise.race([
+      runPromise.then(() => "completed" as const),
+      timeoutPromise,
+    ]);
+    if (result === "timeout") {
+      timedOut = true;
+      timeoutAbortController?.abort();
+      params.onTimeout(params.timeoutMs);
+      return true;
+    }
+    return false;
+  } finally {
+    if (timeoutHandle) {
+      clearTimeout(timeoutHandle);
+    }
+  }
+}