feat(twitch): preserve accepted chat through local crashes (#110852)

* feat(twitch): add durable ingress queue

* fix(twitch): satisfy ingress type boundaries

extensions/twitch/src/access-control.test.ts

@@ -12,6 +12,7 @@ describe("checkTwitchAccessControl", () => {
   };
 
   const mockMessage: TwitchChatMessage = {
+    id: "message-1",
     username: "testuser",
     userId: "123456",
     message: "hello bot",

extensions/twitch/src/monitor.test.ts

@@ -4,8 +4,11 @@ import type { TwitchChatMessage } from "./types.js";
 
 const mocks = vi.hoisted(() => ({
   checkAccess: vi.fn(async () => ({ allowed: true })),
+  createIngress: vi.fn(),
   getClient: vi.fn(async () => ({})),
   getRuntime: vi.fn(),
+  ingressStart: vi.fn(),
+  ingressStop: vi.fn(async () => undefined),
   onMessage: vi.fn(),
   runInbound: vi.fn(),
   sendMessage: vi.fn(),
@@ -28,6 +31,10 @@ vi.mock("./runtime.js", () => ({
   getTwitchRuntime: mocks.getRuntime,
 }));
 
+vi.mock("./twitch-ingress.js", () => ({
+  createTwitchIngress: mocks.createIngress,
+}));
+
 import { monitorTwitchProvider } from "./monitor.js";
 
 type InboundRunInput = {
@@ -47,6 +54,32 @@ describe("monitorTwitchProvider", () => {
     vi.clearAllMocks();
     mocks.getClient.mockResolvedValue({});
     mocks.sendMessage.mockResolvedValue({ ok: true, messageId: "message-id" });
+    mocks.createIngress.mockImplementation(
+      (options: {
+        deliver: (
+          message: TwitchChatMessage,
+          lifecycle: {
+            admission: "exclusive";
+            abortSignal: AbortSignal;
+            onAdopted: () => Promise<void>;
+            onDeferred: () => void;
+            onAbandoned: () => Promise<void>;
+          },
+        ) => Promise<void>;
+      }) => ({
+        accept: async (message: TwitchChatMessage) => {
+          await options.deliver(message, {
+            admission: "exclusive",
+            abortSignal: new AbortController().signal,
+            onAdopted: async () => undefined,
+            onDeferred: () => undefined,
+            onAbandoned: async () => undefined,
+          });
+        },
+        start: mocks.ingressStart,
+        stop: mocks.ingressStop,
+      }),
+    );
     mocks.runInbound.mockImplementation(async (input: InboundRunInput) => {
       const ingested = input.adapter.ingest(input.raw);
       const turn = await input.adapter.resolveTurn(ingested);
@@ -108,6 +141,7 @@ describe("monitorTwitchProvider", () => {
     });
 
     onMessage?.({
+      id: "message-1",
       username: "viewer",
       userId: "viewer-1",
       message: "hello bot",
@@ -124,7 +158,8 @@ describe("monitorTwitchProvider", () => {
       );
     });
 
-    monitor.stop();
+    await monitor.stop();
     expect(mocks.unregister).toHaveBeenCalledOnce();
+    expect(mocks.ingressStop).toHaveBeenCalledOnce();
   });
 });

extensions/twitch/src/monitor.ts

@@ -13,6 +13,7 @@ import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coer
 import { checkTwitchAccessControl } from "./access-control.js";
 import { getOrCreateClientManager } from "./client-manager-registry.js";
 import { getTwitchRuntime } from "./runtime.js";
+import { createTwitchIngress } from "./twitch-ingress.js";
 import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js";
 import { stripMarkdownForTwitch } from "./utils/markdown.js";
 
@@ -31,10 +32,11 @@ type TwitchMonitorOptions = {
 };
 
 type TwitchMonitorResult = {
-  stop: () => void;
+  stop: () => Promise<void>;
 };
 
 type TwitchCoreRuntime = ReturnType<typeof getTwitchRuntime>;
+type TwitchIngressLifecycle = Parameters<Parameters<typeof createTwitchIngress>[0]["deliver"]>[1];
 
 /**
  * Process an incoming Twitch message and dispatch to agent.
@@ -46,19 +48,22 @@ async function processTwitchMessage(params: {
   config: unknown;
   runtime: TwitchRuntimeEnv;
   core: TwitchCoreRuntime;
+  turnAdoptionLifecycle: TwitchIngressLifecycle;
   statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
 }): Promise<void> {
-  const { message, account, accountId, config, runtime, core, statusSink } = params;
+  const { message, account, accountId, config, runtime, core, turnAdoptionLifecycle, statusSink } =
+    params;
   const cfg = config as OpenClawConfig;
 
   await core.channel.inbound.run({
     channel: "twitch",
     accountId,
     raw: message,
+    turnAdoptionLifecycle,
     adapter: {
       ingest: (incoming) => ({
-        id: incoming.id ?? `${incoming.channel}:${incoming.timestamp?.getTime() ?? Date.now()}`,
-        timestamp: incoming.timestamp?.getTime(),
+        id: incoming.id,
+        timestamp: incoming.timestamp,
         rawText: incoming.message,
         textForAgent: incoming.message,
         textForCommands: incoming.message,
@@ -220,6 +225,7 @@ export async function monitorTwitchProvider(
 
   const core = getTwitchRuntime();
   let stopped = false;
+  let stopTask: Promise<void> | undefined;
 
   const coreLogger = core.logging.getChildLogger({ module: "twitch" });
   const logVerboseMessage = (message: string) => {
@@ -249,12 +255,10 @@ export async function monitorTwitchProvider(
     throw error;
   }
 
-  const unregisterHandler = clientManager.onMessage(account, (message) => {
-    if (stopped) {
-      return;
-    }
-
-    void (async () => {
+  const ingress = createTwitchIngress({
+    accountId,
+    runtime,
+    deliver: async (message, turnAdoptionLifecycle) => {
       const botUsername = normalizeLowercaseStringOrEmpty(account.username);
       if (normalizeLowercaseStringOrEmpty(message.username) === botUsername) {
         return;
@@ -266,7 +270,7 @@ export async function monitorTwitchProvider(
         botUsername,
       });
 
-      if (stopped || !access.allowed) {
+      if (!access.allowed) {
         return;
       }
 
@@ -279,19 +283,41 @@ export async function monitorTwitchProvider(
         config,
         runtime,
         core,
+        turnAdoptionLifecycle,
         statusSink,
       });
-    })().catch((err: unknown) => {
-      runtime.error?.(`Message processing failed: ${String(err)}`);
+    },
+  });
+  ingress.start();
+
+  const unregisterHandler = clientManager.onMessage(account, (message) => {
+    if (stopped) {
+      return;
+    }
+
+    void ingress.accept(message).catch((err: unknown) => {
+      runtime.error?.(`Message durable admission failed: ${String(err)}`);
     });
   });
 
-  const stop = () => {
-    stopped = true;
-    unregisterHandler();
+  const stop = (): Promise<void> => {
+    stopTask ??= (async () => {
+      stopped = true;
+      unregisterHandler();
+      await ingress.stop();
+    })();
+    return stopTask;
   };
 
-  abortSignal.addEventListener("abort", stop, { once: true });
+  abortSignal.addEventListener(
+    "abort",
+    () => {
+      void stop().catch((error: unknown) => {
+        runtime.error?.(`Twitch ingress stop failed: ${String(error)}`);
+      });
+    },
+    { once: true },
+  );
 
   return { stop };
 }

extensions/twitch/src/plugin.live.test.ts

@@ -1,5 +1,5 @@
 /**
- * Live Twitch IRC verification for the runStoppablePassiveMonitor lifecycle
+ * Live Twitch IRC verification for the passive account lifecycle
  * pattern used by the Twitch gateway.
  *
  * This test connects to irc.chat.twitch.tv using the same twurple stack the
@@ -19,7 +19,7 @@
 
 import { StaticAuthProvider } from "@twurple/auth";
 import { ChatClient } from "@twurple/chat";
-import { runStoppablePassiveMonitor } from "openclaw/plugin-sdk/extension-shared";
+import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound";
 import { describe, expect, it } from "vitest";
 
 const LIVE = process.env.TWITCH_LIVE_TEST === "1";
@@ -33,7 +33,7 @@ const HAS_CREDS = Boolean(
 const maybeDescribe = LIVE && HAS_CREDS ? describe : describe.skip;
 
 maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", () => {
-  it("real twurple connection + runStoppablePassiveMonitor stays pending until abort, then stops cleanly", async () => {
+  it("real twurple connection stays pending until abort, then stops cleanly", async () => {
     const accessTokenRaw = process.env.TWITCH_ACCESS_TOKEN!.replace(/^oauth:/, "");
     const clientId = process.env.TWITCH_CLIENT_ID!;
     const channel = process.env.TWITCH_CHANNEL!;
@@ -56,7 +56,7 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", (
     let settled = false;
     let stopCalled = false;
 
-    const task = runStoppablePassiveMonitor({
+    const task = runPassiveAccountLifecycle({
       abortSignal: abort.signal,
       start: async () => {
         const chat = new ChatClient({
@@ -86,6 +86,9 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", (
           },
         };
       },
+      stop: async (monitor) => {
+        monitor.stop();
+      },
     })
       .then(() => {
         settled = true;

extensions/twitch/src/plugin.ts

@@ -12,15 +12,13 @@ import {
   createChatChannelPlugin,
   stripChannelTargetPrefix,
 } from "openclaw/plugin-sdk/channel-core";
+import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound";
 import {
   createLoggedPairingApprovalNotifier,
   createPairingPrefixStripper,
 } from "openclaw/plugin-sdk/channel-pairing";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
-import {
-  buildPassiveProbedChannelStatusSummary,
-  runStoppablePassiveMonitor,
-} from "openclaw/plugin-sdk/extension-shared";
+import { buildPassiveProbedChannelStatusSummary } from "openclaw/plugin-sdk/extension-shared";
 import {
   createComputedAccountStatusAdapter,
   createDefaultChannelRuntimeState,
@@ -221,7 +219,7 @@ export const twitchPlugin: ChannelPlugin<ResolvedTwitchAccount> =
           // supervisor reads the settled task as `channel exited without an
           // error` and triggers a restart loop. See #60071.
           try {
-            await runStoppablePassiveMonitor({
+            await runPassiveAccountLifecycle({
               abortSignal: ctx.abortSignal,
               start: async () => {
                 // Lazy import: the monitor pulls the reply pipeline; avoid ESM init cycles.
@@ -234,6 +232,9 @@ export const twitchPlugin: ChannelPlugin<ResolvedTwitchAccount> =
                   abortSignal: ctx.abortSignal,
                 });
               },
+              stop: async (monitor) => {
+                await monitor.stop();
+              },
             });
           } catch (error) {
             ctx.setStatus?.({

extensions/twitch/src/twitch-client.test.ts

@@ -677,11 +677,11 @@ describe("TwitchClientManager", () => {
       expect(capturedMessage?.displayName).toBe("TestUser");
       expect(capturedMessage?.userId).toBe("12345");
       expect(capturedMessage?.message).toBe("Hello bot!");
-      expect(capturedMessage?.channel).toBe("testchannel");
+      expect(capturedMessage?.channel).toBe("#testchannel");
       expect(capturedMessage?.chatType).toBe("group");
     });
 
-    it("should normalize channel names without # prefix", async () => {
+    it("should preserve channel names without a # prefix", async () => {
       await manager.getClient(testAccount);
 
       const onMessageCallback = expectDefined(messageHandlers[0], "Twitch message handler");

extensions/twitch/src/twitch-client.ts

@@ -274,11 +274,10 @@ export class TwitchClientManager {
     client.onMessage((channelName, _user, messageText, msg) => {
       const handler = this.messageHandlers.get(key);
       if (handler) {
-        const normalizedChannel = channelName.startsWith("#") ? channelName.slice(1) : channelName;
         const from = `twitch:${msg.userInfo.userName}`;
         const preview = sliceUtf16Safe(messageText, 0, 100).replace(/\n/g, "\\n");
         this.logger.debug?.(
-          `twitch inbound: channel=${normalizedChannel} from=${from} len=${messageText.length} preview="${preview}"`,
+          `twitch inbound: channel=${channelName} from=${from} len=${messageText.length} preview="${preview}"`,
         );
 
         handler({
@@ -286,9 +285,10 @@ export class TwitchClientManager {
           displayName: msg.userInfo.displayName,
           userId: msg.userInfo.userId,
           message: messageText,
-          channel: normalizedChannel,
+          // Preserve the raw callback channel; durable dispatch normalizes it.
+          channel: channelName,
           id: msg.id,
-          timestamp: new Date(),
+          timestamp: Date.now(),
           isMod: msg.userInfo.isMod,
           isOwner: msg.userInfo.isBroadcaster,
           isVip: msg.userInfo.isVip,

extensions/twitch/src/twitch-ingress.test-support.ts

@@ -0,0 +1,74 @@
+// Twitch tests share isolated durable-ingress state and raw chat envelopes.
+import fs from "node:fs/promises";
+import path from "node:path";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
+import { expect, vi } from "vitest";
+import { createTwitchIngress } from "./twitch-ingress.js";
+import type { TwitchChatMessage } from "./types.js";
+
+type TwitchIngressTestQueue = NonNullable<Parameters<typeof createTwitchIngress>[0]["queue"]>;
+export type TwitchIngressTestPayload = Parameters<TwitchIngressTestQueue["enqueue"]>[1];
+
+export function createTwitchIngressTestMessage(
+  params: Partial<TwitchChatMessage> = {},
+): TwitchChatMessage {
+  return {
+    id: params.id ?? "message-1",
+    username: params.username ?? "viewer",
+    userId: params.userId ?? "viewer-1",
+    displayName: params.displayName ?? "Viewer",
+    message: params.message ?? "hello bot",
+    channel: params.channel ?? "#TestChannel",
+    timestamp: params.timestamp ?? 1_721_300_000_000,
+    isMod: params.isMod ?? false,
+    isOwner: params.isOwner ?? false,
+    isVip: params.isVip ?? false,
+    isSub: params.isSub ?? false,
+    chatType: "group",
+  };
+}
+
+export async function withTwitchIngressTestQueue<T>(
+  fn: (queue: TwitchIngressTestQueue) => Promise<T>,
+): Promise<T> {
+  const createdDir = await fs.mkdtemp(
+    path.join(resolvePreferredOpenClawTmpDir(), "openclaw-twitch-ingress-"),
+  );
+  const stateDir = await fs.realpath(createdDir);
+  const previousStateDir = process.env.OPENCLAW_STATE_DIR;
+  process.env.OPENCLAW_STATE_DIR = stateDir;
+  const queue = createChannelIngressQueueForTests<TwitchIngressTestPayload>({
+    channelId: "twitch",
+    accountId: "default",
+    stateDir,
+  });
+  try {
+    return await fn(queue);
+  } finally {
+    if (previousStateDir === undefined) {
+      delete process.env.OPENCLAW_STATE_DIR;
+    } else {
+      process.env.OPENCLAW_STATE_DIR = previousStateDir;
+    }
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
+export async function waitForTwitchIngressVerdict(
+  queue: TwitchIngressTestQueue,
+  eventId: string,
+  expected: "completed" | "failed",
+): Promise<void> {
+  await vi.waitFor(
+    async () => {
+      const verdict = await queue.enqueue(eventId, { version: 1, rawEvent: "{}" });
+      expect(verdict.kind).toBe(expected);
+    },
+    { timeout: 5_000 },
+  );
+}

extensions/twitch/src/twitch-ingress.test.ts

@@ -0,0 +1,285 @@
+// Twitch durable ingress tests cover raw admission, recovery, and tombstones.
+import type { ChannelIngressQueue } from "openclaw/plugin-sdk/channel-outbound";
+import { closeOpenClawStateDatabaseForTest } from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createTwitchIngress } from "./twitch-ingress.js";
+import {
+  createTwitchIngressTestMessage,
+  waitForTwitchIngressVerdict,
+  withTwitchIngressTestQueue,
+  type TwitchIngressTestPayload,
+} from "./twitch-ingress.test-support.js";
+
+function runtime() {
+  return { error: vi.fn() };
+}
+
+afterEach(() => {
+  closeOpenClawStateDatabaseForTest();
+  vi.restoreAllMocks();
+});
+
+describe("Twitch durable ingress", () => {
+  it("durably appends before dispatch", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      const realEnqueue = queue.enqueue.bind(queue);
+      let releaseAppend = () => {};
+      const appendGate = new Promise<void>((resolve) => {
+        releaseAppend = resolve;
+      });
+      const enqueue: typeof queue.enqueue = vi.fn(
+        async (...args: Parameters<typeof queue.enqueue>) => {
+          await appendGate;
+          return await realEnqueue(...args);
+        },
+      );
+      const gatedQueue: ChannelIngressQueue<TwitchIngressTestPayload> = { ...queue, enqueue };
+      const deliver = vi.fn(async (_message, lifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue: gatedQueue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      ingress.start();
+      try {
+        const admission = ingress.accept(createTwitchIngressTestMessage({ id: "durable-first" }));
+        await vi.waitFor(() => expect(enqueue).toHaveBeenCalledOnce());
+        expect(deliver).not.toHaveBeenCalled();
+        releaseAppend();
+        await admission;
+        await waitForTwitchIngressVerdict(queue, "durable-first", "completed");
+        expect(deliver).toHaveBeenCalledOnce();
+      } finally {
+        releaseAppend();
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("recovers an uncompleted event with a fresh drain and dispatches exactly once", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      const interrupted = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver: vi.fn(),
+      });
+      await interrupted.accept(createTwitchIngressTestMessage({ id: "restart" }));
+      await interrupted.stop();
+
+      const deliver = vi.fn(async (_message, lifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const recovered = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      recovered.start();
+      try {
+        await waitForTwitchIngressVerdict(queue, "restart", "completed");
+        expect(deliver).toHaveBeenCalledOnce();
+      } finally {
+        await recovered.stop();
+      }
+    });
+  });
+
+  it("keeps a completion tombstone and rejects a post-completion duplicate", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      const deliver = vi.fn(async (_message, lifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      const message = createTwitchIngressTestMessage({ id: "duplicate" });
+      ingress.start();
+      try {
+        await ingress.accept(message);
+        await waitForTwitchIngressVerdict(queue, "duplicate", "completed");
+        await ingress.accept(message);
+        await new Promise<void>((resolve) => {
+          setTimeout(resolve, 30);
+        });
+        expect(deliver).toHaveBeenCalledOnce();
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("stores the raw callback envelope and normalizes its channel only at dispatch", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      const message = createTwitchIngressTestMessage({
+        id: "raw",
+        channel: "#MixedCase",
+        message: "before",
+      });
+      const delivered = vi.fn(async (_message, lifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver: delivered,
+        pollIntervalMs: 5,
+      });
+      await ingress.accept(message);
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({
+          id: "raw",
+          laneKey: "channel:mixedcase",
+          payload: { version: 1, rawEvent: JSON.stringify(message) },
+        }),
+      ]);
+      message.message = "after";
+
+      ingress.start();
+      try {
+        await waitForTwitchIngressVerdict(queue, "raw", "completed");
+        expect(delivered).toHaveBeenCalledWith(
+          expect.objectContaining({ channel: "mixedcase", message: "before" }),
+          expect.any(Object),
+        );
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("dead-letters malformed persisted JSON without dispatch", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      await queue.enqueue(
+        "malformed",
+        { version: 1, rawEvent: "{" },
+        { laneKey: "channel:testchannel" },
+      );
+      const deliver = vi.fn();
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      ingress.start();
+      try {
+        await waitForTwitchIngressVerdict(queue, "malformed", "failed");
+        expect(deliver).not.toHaveBeenCalled();
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("waits for an in-flight durable admission before stop returns", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      const realEnqueue = queue.enqueue.bind(queue);
+      let releaseAppend = () => {};
+      const appendGate = new Promise<void>((resolve) => {
+        releaseAppend = resolve;
+      });
+      const enqueue: typeof queue.enqueue = async (...args: Parameters<typeof queue.enqueue>) => {
+        await appendGate;
+        return await realEnqueue(...args);
+      };
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue: { ...queue, enqueue },
+        deliver: vi.fn(),
+      });
+      const admission = ingress.accept(createTwitchIngressTestMessage({ id: "admitting" }));
+      let stopped = false;
+      const stopping = ingress.stop().then(() => {
+        stopped = true;
+      });
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 30);
+      });
+      expect(stopped).toBe(false);
+      releaseAppend();
+      await admission;
+      await stopping;
+      expect(stopped).toBe(true);
+    });
+  });
+
+  it("waits for an adopted active delivery before stop returns", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      let releaseDelivery = () => {};
+      const deliveryGate = new Promise<void>((resolve) => {
+        releaseDelivery = resolve;
+      });
+      const deliver = vi.fn(async (_message, lifecycle) => {
+        await lifecycle.onAdopted();
+        await deliveryGate;
+      });
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      ingress.start();
+      await ingress.accept(createTwitchIngressTestMessage({ id: "active-stop" }));
+      await vi.waitFor(() => expect(deliver).toHaveBeenCalledOnce());
+
+      let stopped = false;
+      const stopping = ingress.stop().then(() => {
+        stopped = true;
+      });
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 30);
+      });
+      expect(stopped).toBe(false);
+      releaseDelivery();
+      await stopping;
+      expect(stopped).toBe(true);
+    });
+  });
+
+  it("releases a pre-adoption delivery for retry during shutdown", async () => {
+    await withTwitchIngressTestQueue(async (queue) => {
+      let releaseDelivery = () => {};
+      const deliveryGate = new Promise<void>((resolve) => {
+        releaseDelivery = resolve;
+      });
+      const deliver = vi.fn(async () => {
+        await deliveryGate;
+      });
+      const ingress = createTwitchIngress({
+        accountId: "default",
+        runtime: runtime(),
+        queue,
+        deliver,
+        pollIntervalMs: 5,
+      });
+      ingress.start();
+      await ingress.accept(createTwitchIngressTestMessage({ id: "shutdown-retry" }));
+      await vi.waitFor(() => expect(deliver).toHaveBeenCalledOnce());
+
+      const stopping = ingress.stop();
+      releaseDelivery();
+      await stopping;
+
+      expect(await queue.listClaims()).toHaveLength(0);
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({ id: "shutdown-retry", lastError: expect.any(String) }),
+      ]);
+    });
+  });
+});

extensions/twitch/src/twitch-ingress.ts

@@ -0,0 +1,356 @@
+// Twitch plugin owns raw chat-envelope durable admission and replay draining.
+import { HttpStatusCodeError } from "@twurple/api-call";
+import {
+  bindIngressLifecycleToReplyOptions,
+  createChannelIngressDrain,
+  DEFAULT_INGRESS_ADOPTION_STALL_MS,
+  DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+  type ChannelIngressQueue,
+} from "openclaw/plugin-sdk/channel-outbound";
+import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
+import { getTwitchRuntime } from "./runtime.js";
+import type { TwitchChatMessage } from "./types.js";
+import { normalizeTwitchChannel } from "./utils/twitch.js";
+
+const TWITCH_INGRESS_PAYLOAD_VERSION = 1;
+const TWITCH_INGRESS_DRAIN_INTERVAL_MS = 1_000;
+const TWITCH_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000;
+const TWITCH_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
+// Twitch IRC does not replay accepted PRIVMSG lines. These tombstones are near-inert;
+// the durable queue protects the local accept-to-dispatch crash window instead.
+const TWITCH_INGRESS_COMPLETED_MAX_ENTRIES = 1_000;
+const TWITCH_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
+const TWITCH_INGRESS_FAILED_MAX_ENTRIES = 1_000;
+const TWITCH_INGRESS_APPEND_RETRY_DELAYS_MS = [0, 100, 300] as const;
+
+type TwitchIngressPayload = {
+  version: typeof TWITCH_INGRESS_PAYLOAD_VERSION;
+  rawEvent: string;
+};
+
+type TwitchIngressLifecycle = ReturnType<
+  typeof bindIngressLifecycleToReplyOptions
+>["turnAdoptionLifecycle"];
+
+type TwitchIngress = {
+  accept: (message: TwitchChatMessage) => Promise<void>;
+  start: () => void;
+  stop: () => Promise<void>;
+};
+
+class TwitchIngressPermanentError extends Error {
+  constructor(message: string, options?: ErrorOptions) {
+    super(message, options);
+    this.name = "TwitchIngressPermanentError";
+  }
+}
+
+function nonEmptyString(value: unknown): string | null {
+  return typeof value === "string" && value.trim() ? value.trim() : null;
+}
+
+function inspectTwitchIngressEvent(event: unknown): { eventId: string; laneKey: string } {
+  if (!event || typeof event !== "object" || Array.isArray(event)) {
+    throw new TwitchIngressPermanentError("Twitch ingress event must be an object.");
+  }
+  const candidate = event as { id?: unknown; channel?: unknown };
+  const eventId = nonEmptyString(candidate.id);
+  if (!eventId) {
+    throw new TwitchIngressPermanentError("Twitch ingress event is missing its message id.");
+  }
+  const rawChannel = nonEmptyString(candidate.channel);
+  const channel = rawChannel ? normalizeTwitchChannel(rawChannel) : "";
+  if (!channel) {
+    throw new TwitchIngressPermanentError("Twitch ingress event is missing its channel.");
+  }
+  return { eventId, laneKey: `channel:${channel}` };
+}
+
+function parseClaimedTwitchMessage(
+  payload: TwitchIngressPayload,
+  claimedId: string,
+  claimedLaneKey: string | undefined,
+): TwitchChatMessage {
+  if (payload.version !== TWITCH_INGRESS_PAYLOAD_VERSION || typeof payload.rawEvent !== "string") {
+    throw new TwitchIngressPermanentError("Twitch ingress payload is invalid.");
+  }
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(payload.rawEvent);
+  } catch (error) {
+    throw new TwitchIngressPermanentError("Twitch ingress event JSON is invalid.", {
+      cause: error,
+    });
+  }
+  const facts = inspectTwitchIngressEvent(parsed);
+  if (facts.eventId !== claimedId || facts.laneKey !== claimedLaneKey) {
+    throw new TwitchIngressPermanentError(
+      "Twitch ingress event identity changed after durable admission.",
+    );
+  }
+  const candidate = parsed as Partial<TwitchChatMessage>;
+  const username = nonEmptyString(candidate.username);
+  const rawChannel = nonEmptyString(candidate.channel);
+  if (!username || typeof candidate.message !== "string" || !rawChannel) {
+    throw new TwitchIngressPermanentError("Twitch ingress event shape is invalid.");
+  }
+  return {
+    ...candidate,
+    id: claimedId,
+    username,
+    message: candidate.message,
+    channel: normalizeTwitchChannel(rawChannel),
+  } as TwitchChatMessage;
+}
+
+function isTwitchAuthenticationFailure(error: unknown): boolean {
+  let current: unknown = error;
+  for (let depth = 0; depth < 8 && current && typeof current === "object"; depth += 1) {
+    if (
+      current instanceof HttpStatusCodeError &&
+      (current.statusCode === 401 || current.statusCode === 403)
+    ) {
+      return true;
+    }
+    current = (current as { cause?: unknown }).cause;
+  }
+  return false;
+}
+
+function stoppedError(): Error {
+  return new Error("Twitch ingress stopped before dispatch.");
+}
+
+export function createTwitchIngress(options: {
+  accountId: string;
+  runtime: { error?: (message: string) => void };
+  deliver: (message: TwitchChatMessage, lifecycle: TwitchIngressLifecycle) => Promise<void>;
+  queue?: ChannelIngressQueue<TwitchIngressPayload>;
+  pollIntervalMs?: number;
+}): TwitchIngress {
+  const queue =
+    options.queue ??
+    getTwitchRuntime().state.openChannelIngressQueue<TwitchIngressPayload>({
+      accountId: options.accountId,
+    });
+  const shutdown = new AbortController();
+  const activeDeliveries = new Set<Promise<void>>();
+  const deferredClaims = new Map<string, Promise<void>>();
+  let running = false;
+  let stopped = false;
+  let drainRequested = false;
+  let drainTask: Promise<void> | undefined;
+  let drainTimer: ReturnType<typeof setInterval> | undefined;
+  let lastPrunedAt = 0;
+  let admissionTail: Promise<void> = Promise.resolve();
+  let stopTask: Promise<void> | undefined;
+
+  const drain = createChannelIngressDrain<TwitchIngressPayload>({
+    queue,
+    abortSignal: shutdown.signal,
+    adoptionStallTimeoutMs: DEFAULT_INGRESS_ADOPTION_STALL_MS,
+    retryPolicy: {
+      maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+      deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+    },
+    resolveNonRetryableFailure: (error) => {
+      if (error instanceof TwitchIngressPermanentError) {
+        return { reason: "invalid-event", message: error.message };
+      }
+      if (isTwitchAuthenticationFailure(error)) {
+        return { reason: "authentication-failed", message: formatErrorMessage(error) };
+      }
+      return null;
+    },
+    onLog: (message) => options.runtime.error?.(`twitch ingress: ${message}`),
+    dispatchClaimedEvent: async (claimed, lifecycle) => {
+      if (!running || lifecycle.abortSignal.aborted) {
+        return { kind: "failed-retryable", error: stoppedError() };
+      }
+      const message = parseClaimedTwitchMessage(claimed.payload, claimed.id, claimed.laneKey);
+      const bound = bindIngressLifecycleToReplyOptions(lifecycle).turnAdoptionLifecycle;
+      let handedOff = false;
+      let resolveDeferredClaim!: () => void;
+      const deferredClaim = new Promise<void>((resolve) => {
+        resolveDeferredClaim = resolve;
+      });
+      let deferredClaimSettled = false;
+      const settleDeferredClaim = () => {
+        if (deferredClaimSettled) {
+          return;
+        }
+        deferredClaimSettled = true;
+        if (deferredClaims.get(claimed.id) === deferredClaim) {
+          deferredClaims.delete(claimed.id);
+        }
+        resolveDeferredClaim();
+      };
+      const delivery = options.deliver(message, {
+        ...bound,
+        onAdopted: async () => {
+          handedOff = true;
+          try {
+            await bound.onAdopted();
+          } finally {
+            settleDeferredClaim();
+          }
+        },
+        onDeferred: () => {
+          handedOff = true;
+          if (!deferredClaimSettled) {
+            deferredClaims.set(claimed.id, deferredClaim);
+          }
+          bound.onDeferred();
+        },
+        onAbandoned: async () => {
+          handedOff = true;
+          try {
+            await bound.onAbandoned();
+          } finally {
+            settleDeferredClaim();
+          }
+        },
+      });
+      activeDeliveries.add(delivery);
+      try {
+        await delivery;
+      } catch (error) {
+        if (!running || lifecycle.abortSignal.aborted) {
+          return { kind: "failed-retryable", error };
+        }
+        throw error;
+      } finally {
+        activeDeliveries.delete(delivery);
+      }
+      if (!handedOff) {
+        if (!running || lifecycle.abortSignal.aborted) {
+          return { kind: "failed-retryable", error: stoppedError() };
+        }
+        // Echoes and access-gated messages are terminal no-dispatch events.
+        await bound.onAdopted();
+      }
+      return deferredClaims.has(claimed.id) ? { kind: "deferred" } : { kind: "completed" };
+    },
+  });
+
+  const pruneIfDue = async (): Promise<void> => {
+    const now = Date.now();
+    if (now - lastPrunedAt < TWITCH_INGRESS_PRUNE_INTERVAL_MS) {
+      return;
+    }
+    await queue.prune({
+      completedTtlMs: TWITCH_INGRESS_COMPLETED_TTL_MS,
+      completedMaxEntries: TWITCH_INGRESS_COMPLETED_MAX_ENTRIES,
+      failedTtlMs: TWITCH_INGRESS_FAILED_TTL_MS,
+      failedMaxEntries: TWITCH_INGRESS_FAILED_MAX_ENTRIES,
+      now,
+    });
+    lastPrunedAt = now;
+  };
+
+  const requestDrain = (): void => {
+    if (!running || stopped || shutdown.signal.aborted) {
+      return;
+    }
+    drainRequested = true;
+    if (drainTask) {
+      return;
+    }
+    drainTask = (async () => {
+      while (drainRequested) {
+        if (!running) {
+          break;
+        }
+        drainRequested = false;
+        await pruneIfDue();
+        if (!running) {
+          break;
+        }
+        const { started } = await drain.drainOnce({ shouldStop: () => !running });
+        if (!running || (!drainRequested && started === 0)) {
+          break;
+        }
+      }
+    })()
+      .catch((error: unknown) => {
+        options.runtime.error?.(`Twitch ingress drain failed: ${formatErrorMessage(error)}`);
+      })
+      .finally(() => {
+        drainTask = undefined;
+        if (running && drainRequested) {
+          requestDrain();
+        }
+      });
+  };
+
+  const admitOnce = async (message: TwitchChatMessage): Promise<void> => {
+    const facts = inspectTwitchIngressEvent(message);
+    const rawEvent = JSON.stringify(message);
+    const receivedAt = Date.now();
+    let lastError: unknown;
+    for (const delayMs of TWITCH_INGRESS_APPEND_RETRY_DELAYS_MS) {
+      if (delayMs > 0) {
+        await new Promise<void>((resolve) => {
+          setTimeout(resolve, delayMs);
+        });
+      }
+      try {
+        await queue.enqueue(
+          facts.eventId,
+          { version: TWITCH_INGRESS_PAYLOAD_VERSION, rawEvent },
+          { receivedAt, laneKey: facts.laneKey },
+        );
+        requestDrain();
+        return;
+      } catch (error) {
+        lastError = error;
+      }
+    }
+    throw lastError;
+  };
+
+  return {
+    accept: (message) => {
+      if (stopped) {
+        return Promise.reject(stoppedError());
+      }
+      // Preserve socket arrival order across append retry backoff.
+      const admission = admissionTail.then(() => admitOnce(message));
+      admissionTail = admission.catch(() => undefined);
+      return admission;
+    },
+    start: () => {
+      if (running || stopped) {
+        return;
+      }
+      running = true;
+      requestDrain();
+      drainTimer = setInterval(
+        requestDrain,
+        options.pollIntervalMs ?? TWITCH_INGRESS_DRAIN_INTERVAL_MS,
+      );
+      drainTimer.unref?.();
+    },
+    stop: () => {
+      stopTask ??= (async () => {
+        stopped = true;
+        running = false;
+        if (drainTimer) {
+          clearInterval(drainTimer);
+          drainTimer = undefined;
+        }
+        await admissionTail;
+        shutdown.abort(stoppedError());
+        await drainTask;
+        await Promise.allSettled(activeDeliveries);
+        await Promise.allSettled(deferredClaims.values());
+        await drain.waitForIdle();
+        // Stop is idempotent, and drain disposal remains safe if cleanup repeats.
+        drain.dispose();
+        drain.dispose();
+      })();
+      return stopTask;
+    },
+  };
+}

extensions/twitch/src/types.ts

@@ -74,9 +74,9 @@ export interface TwitchChatMessage {
   /** Display name (may include special characters) */
   displayName?: string;
   /** Message ID */
-  id?: string;
-  /** Timestamp */
-  timestamp?: Date;
+  id: string;
+  /** Receive timestamp in milliseconds */
+  timestamp?: number;
   /** Whether the sender is a moderator */
   isMod?: boolean;
   /** Whether the sender is the channel owner/broadcaster */