refactor(channels): store inbound queues in SQLite

2026-08-05 06:01:41 +00:00 · 2026-05-31 21:14:44 +01:00
parent 80b7f56603
commit b0679d1f13
24 changed files with 2058 additions and 456 deletions
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -74,6 +74,7 @@ Skills own workflows; root owns hard policy and routing.
 - Core runtime consumes only current canonical shapes/config/data. Legacy or retired shapes normalize only in doctor/migration code before runtime; no runtime shims, aliases, or fallback readers.
 - State/storage migrations are database-first. Runtime reads/writes the canonical store only. Old file stores, sidecars, aliases, and fallback readers belong in `openclaw doctor --fix` migration code only, never steady-state runtime.
 - Storage default: SQLite only. Do not add JSON/JSONL/TXT/sidecar files for OpenClaw-owned runtime state, caches, queues, registries, indexes, cursors, checkpoints, or plugin scratch data.
+- SQLite runtime access uses Kysely helpers, not raw SQL statement strings, except schema DDL, migrations, low-level DB bootstrap, or narrowly justified SQLite primitives.
 - Use the shared state DB (`state/openclaw.sqlite`) for global runtime state and plugin KV data. Use the per-agent DB (`agents/<agentId>/agent/openclaw-agent.sqlite`) for agent-scoped state/cache. Use a dedicated SQLite DB only when schema, volume, or lifecycle clearly does not fit those stores.
 - Legacy state/cache files are migration debt. When touching code that reads/writes them, prefer moving the data into SQLite or calling out the refactor follow-up; do not add parallel file paths.
 - File storage must be a named product artifact: import/export, user attachment, log, backup, or external tool contract. If it is app state or cache, it belongs in SQLite.
--- a/docs/.generated/plugin-sdk-api-baseline.sha256
+++ b/docs/.generated/plugin-sdk-api-baseline.sha256
@@ -1,2 +1,2 @@
-eadfc9b897a05664735f8e2abcb70cb3f33c19427c20802bf8b035520b7a2ea1  plugin-sdk-api-baseline.json
-8e10e093068d73b9ac50d3f265bf7d892652b0392c677be4e332248499cf7ed0  plugin-sdk-api-baseline.jsonl
+19bdf1196ec771a00777a16fd1e9c3662b8fd788a81034e705c41a74ee79c7ec  plugin-sdk-api-baseline.json
+43feff80c90adad0f821d1f1e184a9bff1e93d81e6d53a26a26fd9e2972be759  plugin-sdk-api-baseline.jsonl
--- a/extensions/telegram/src/polling-session.test.ts
+++ b/extensions/telegram/src/polling-session.test.ts
@@ -3,7 +3,16 @@ import os from "node:os";
 import path from "node:path";
 import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract";
 import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
-import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { createChannelIngressQueue } from "../../../src/channels/message/ingress-queue.js";
+import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../../src/infra/kysely-sync.js";
+import type { DB as OpenClawStateKyselyDatabase } from "../../../src/state/openclaw-state-db.generated.js";
+import {
+  closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
+} from "../../../src/state/openclaw-state-db.js";
+import { clearTelegramRuntime, setTelegramRuntime } from "./runtime.js";
+import type { TelegramRuntime } from "./runtime.types.js";
 import type { TelegramIngressWorkerMessage } from "./telegram-ingress-worker.js";

 const runMock = vi.hoisted(() => vi.fn());
@@ -96,9 +105,21 @@ type WorkerPollErrorListener = (message: {
 type WorkerMessageListener = (message: TelegramIngressWorkerMessage) => void;
 type AsyncVoidFn = () => Promise<void>;
 type MockCallSource = { mock: { calls: Array<Array<unknown>> } };
+type TelegramPollingTestDatabase = Pick<OpenClawStateKyselyDatabase, "channel_ingress_events">;

 const POLLING_TEST_WATCHDOG_INTERVAL_MS = 30_000;

+function installTelegramIngressQueueRuntime(resolveStateDir: () => string): void {
+  setTelegramRuntime({
+    state: {
+      resolveStateDir,
+      openChannelIngressQueue: (
+        options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
+      ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "telegram" }),
+    },
+  } as TelegramRuntime);
+}
+
 function mockObjectArg(
  source: MockCallSource,
  label: string,
@@ -411,17 +432,59 @@ async function pendingUpdateIds(spoolDir: string, limit: number | "all" = 100):
  return (await listTelegramSpooledUpdates({ spoolDir, limit })).map((update) => update.updateId);
 }

-async function failedUpdateIds(spoolDir: string): Promise<number[]> {
-  const entries = await fs.readdir(spoolDir).catch((err) => {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return [];
-    }
-    throw err;
+function normalizeTelegramTestAccountId(spoolDir: string): string {
+  const trimmed = path.basename(spoolDir).trim();
+  return trimmed ? trimmed.replace(/[^a-z0-9._-]+/gi, "_") : "default";
+}
+
+function telegramTestQueueName(spoolDir: string): string {
+  return JSON.stringify(["telegram", normalizeTelegramTestAccountId(spoolDir)]);
+}
+
+function openTelegramSpoolTestKysely(spoolDir: string) {
+  const database = openOpenClawStateDatabase({
+    env: { ...process.env, OPENCLAW_STATE_DIR: spoolDir },
  });
-  return entries
-    .filter((entry) => entry.endsWith(".json.failed"))
-    .map((entry) => Number(entry.slice(0, 16)))
-    .toSorted((a, b) => a - b);
+  return {
+    database,
+    kysely: getNodeSqliteKysely<TelegramPollingTestDatabase>(database.db),
+  };
+}
+
+async function failedUpdateIds(spoolDir: string): Promise<number[]> {
+  const { database, kysely } = openTelegramSpoolTestKysely(spoolDir);
+  const rows = executeSqliteQuerySync(
+    database.db,
+    kysely
+      .selectFrom("channel_ingress_events")
+      .select("event_id")
+      .where("queue_name", "=", telegramTestQueueName(spoolDir))
+      .where("status", "=", "failed")
+      .orderBy("event_id", "asc"),
+  ).rows;
+  return rows.map((row) => Number(row.event_id));
+}
+
+async function adoptClaimOwner(params: {
+  spoolDir: string;
+  updateId: number;
+  ownerId: string;
+  claimedAt: number;
+}): Promise<void> {
+  const { database, kysely } = openTelegramSpoolTestKysely(params.spoolDir);
+  executeSqliteQuerySync(
+    database.db,
+    kysely
+      .updateTable("channel_ingress_events")
+      .set({
+        claim_owner: params.ownerId,
+        claimed_at: params.claimedAt,
+        updated_at: params.claimedAt,
+      })
+      .where("queue_name", "=", telegramTestQueueName(params.spoolDir))
+      .where("event_id", "=", String(params.updateId).padStart(16, "0"))
+      .where("status", "=", "claimed"),
+  );
 }

 async function withTempSpool<T>(fn: (spoolDir: string) => Promise<T>): Promise<T> {
@@ -429,6 +492,7 @@ async function withTempSpool<T>(fn: (spoolDir: string) => Promise<T>): Promise<T
  try {
    return await fn(spoolDir);
  } finally {
+    closeOpenClawStateDatabaseForTest();
    await fs.rm(spoolDir, { recursive: true, force: true });
  }
 }
@@ -526,6 +590,14 @@ describe("TelegramPollingSession", () => {
    sleepWithAbortMock.mockReset().mockResolvedValue(undefined);
    drainPendingDeliveriesMock.mockReset().mockResolvedValue(undefined);
    resetTelegramReplyFenceForTests();
+    installTelegramIngressQueueRuntime(() =>
+      path.join(os.tmpdir(), "openclaw-telegram-test-state"),
+    );
+  });
+
+  afterEach(() => {
+    clearTelegramRuntime();
+    closeOpenClawStateDatabaseForTest();
  });

  it("uses backoff helpers for recoverable polling retries", async () => {
@@ -667,7 +739,14 @@ describe("TelegramPollingSession", () => {

      const runPromise = session.runUntilAbort();
      await vi.waitFor(() => expect(handleUpdate).toHaveBeenCalledTimes(1));
-      await vi.waitFor(async () => expect(await fs.readdir(tempDir)).toEqual([]));
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
+      await vi.waitFor(async () =>
+        expect(
+          await listTelegramSpooledUpdateClaims({
+            spoolDir: tempDir,
+          }),
+        ).toEqual([]),
+      );
      abort.abort();
      await runPromise;

@@ -686,6 +765,76 @@ describe("TelegramPollingSession", () => {
      expect(init).toHaveBeenCalledBefore(handleUpdate);
      expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } });
    } finally {
+      abort.abort();
+      await fs.rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
+  it("writes isolated worker updates through the main runtime queue", async () => {
+    const abort = new AbortController();
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+    const handleUpdate = vi.fn(async () => undefined);
+    const bot = {
+      api: {
+        deleteWebhook: vi.fn(async () => true),
+        config: { use: vi.fn() },
+      },
+      init: vi.fn(async () => undefined),
+      handleUpdate,
+      stop: vi.fn(async () => undefined),
+    };
+    createTelegramBotMock.mockReturnValueOnce(bot);
+    let onMessage: WorkerMessageListener | undefined;
+    let stopWorker: (() => void) | undefined;
+    const workerDone = new Promise<void>((resolve) => {
+      stopWorker = resolve;
+    });
+    const ackSpooledUpdate = vi.fn();
+    const createWorker = vi.fn(() => ({
+      onMessage: vi.fn((listener: WorkerMessageListener) => {
+        onMessage = listener;
+        return () => undefined;
+      }),
+      ackSpooledUpdate,
+      stop: vi.fn(async () => {
+        stopWorker?.();
+      }),
+      task: vi.fn(async () => {
+        await workerDone;
+      }),
+    }));
+
+    try {
+      const session = createPollingSession({
+        abortSignal: abort.signal,
+        isolatedIngress: {
+          enabled: true,
+          spoolDir: tempDir,
+          createWorker,
+          drainIntervalMs: 10,
+        },
+      });
+
+      const runPromise = session.runUntilAbort();
+      await vi.waitFor(() => expect(onMessage).toBeDefined());
+      onMessage?.({
+        type: "update",
+        requestId: "write-1",
+        update: { update_id: 42, message: { text: "hello" } },
+        queued: 1,
+      });
+
+      await vi.waitFor(() =>
+        expect(ackSpooledUpdate).toHaveBeenCalledWith("write-1", { ok: true, updateId: 42 }),
+      );
+      await vi.waitFor(() =>
+        expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }),
+      );
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
+      abort.abort();
+      await runPromise;
+    } finally {
+      abort.abort();
      await fs.rm(tempDir, { recursive: true, force: true });
    }
  });
@@ -735,7 +884,14 @@ describe("TelegramPollingSession", () => {

      const runPromise = session.runUntilAbort();
      await vi.waitFor(() => expect(handleUpdate).toHaveBeenCalledTimes(1));
-      await vi.waitFor(async () => expect(await fs.readdir(tempDir)).toEqual([]));
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
+      await vi.waitFor(async () =>
+        expect(
+          await listTelegramSpooledUpdateClaims({
+            spoolDir: tempDir,
+          }),
+        ).toEqual([]),
+      );
      abort.abort();
      await runPromise;

@@ -1128,7 +1284,7 @@ describe("TelegramPollingSession", () => {
      await runPromise;
      expect(events).toEqual(["handled:42", "handled:44"]);
      expect(await pendingUpdateIds(tempDir)).toEqual([43]);
-      expect((await fs.readdir(tempDir)).toSorted()).toEqual(["0000000000000043.json"]);
+      expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]);
      stopWorker();
    });
  });
@@ -1189,21 +1345,12 @@ describe("TelegramPollingSession", () => {
      if (!claimed) {
        throw new Error("Expected claimed update");
      }
-      await fs.writeFile(
-        claimed.path,
-        `${JSON.stringify({
-          version: 1,
-          updateId: 42,
-          receivedAt: interrupted.receivedAt,
-          update: interruptedUpdate,
-          claim: {
-            processId: "other-process",
-            processPid: process.pid,
-            claimedAt: Date.now(),
-          },
-        })}\n`,
-        { mode: 0o600 },
-      );
+      await adoptClaimOwner({
+        spoolDir: tempDir,
+        updateId: 42,
+        ownerId: `${process.pid}:other-process`,
+        claimedAt: Date.now(),
+      });

      const recovered = await recoverStaleTelegramSpooledUpdateClaims({
        spoolDir: tempDir,
@@ -1213,10 +1360,11 @@ describe("TelegramPollingSession", () => {

      expect(recovered).toBe(0);
      expect(await pendingUpdateIds(tempDir)).toEqual([43]);
-      expect((await fs.readdir(tempDir)).toSorted()).toEqual([
-        "0000000000000042.json.processing",
-        "0000000000000043.json",
-      ]);
+      expect(
+        (await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).map(
+          (claim) => claim.updateId,
+        ),
+      ).toEqual([42]);
    });
  });

@@ -2360,7 +2508,7 @@ describe("TelegramPollingSession", () => {
    }
  });

-  it("keeps a timed-out lane guarded when its failed tombstone cannot be written", async () => {
+  it("keeps a timed-out lane guarded when its failed state cannot be written", async () => {
    vi.useFakeTimers({ shouldAdvanceTime: true });
    const abort = new AbortController();
    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -2371,15 +2519,10 @@ describe("TelegramPollingSession", () => {
    const regularTurnDone = new Promise<void>((resolve) => {
      releaseRegularTurn = resolve;
    });
-    const originalWriteFile = fs.writeFile.bind(fs);
-    const writeFileSpy = vi
-      .spyOn(fs, "writeFile")
-      .mockImplementation(async (...args: Parameters<typeof fs.writeFile>) => {
-        if (typeof args[0] === "string" && args[0].includes(".json.failed.")) {
-          throw new Error("disk full");
-        }
-        return await originalWriteFile(...args);
-      });
+    const spoolModule = await import("./telegram-ingress-spool.js");
+    const failSpy = vi
+      .spyOn(spoolModule, "failTelegramSpooledUpdateClaim")
+      .mockRejectedValueOnce(new Error("disk full"));
    createTelegramBotMock.mockReturnValueOnce({
      api: {
        deleteWebhook: vi.fn(async () => true),
@@ -2460,7 +2603,7 @@ describe("TelegramPollingSession", () => {
      await vi.advanceTimersByTimeAsync(20_000);
      await runPromise;
    } finally {
-      writeFileSpy.mockRestore();
+      failSpy.mockRestore();
      releaseRegularTurn?.();
      abort.abort();
      stopWorker?.();
--- a/extensions/telegram/src/polling-session.ts
+++ b/extensions/telegram/src/polling-session.ts
@@ -36,6 +36,7 @@ import {
  recoverStaleTelegramSpooledUpdateClaims,
  releaseTelegramSpooledUpdateClaim,
  resolveTelegramIngressSpoolDir,
+  writeTelegramSpooledUpdate,
  type ClaimedTelegramSpooledUpdate,
  type TelegramSpooledUpdate,
 } from "./telegram-ingress-spool.js";
@@ -767,6 +768,23 @@ export class TelegramPollingSession {
    });
    const stalledBacklogKeys = new Set<string>();
    const unsubscribe = worker.onMessage((message) => {
+      const ackSpooledUpdate = (
+        requestId: string,
+        result:
+          | { ok: true; updateId: number }
+          | {
+              ok: false;
+              message: string;
+            },
+      ): void => {
+        try {
+          worker.ackSpooledUpdate?.(requestId, result);
+        } catch (err) {
+          this.opts.log(
+            `[telegram][diag] isolated polling worker ack failed: ${formatErrorMessage(err)}`,
+          );
+        }
+      };
      if (message.type === "poll-start") {
        liveness.noteGetUpdatesStarted({ offset: message.offset }, message.startedAt);
        pollState.startedAt = message.startedAt;
@@ -792,6 +810,23 @@ export class TelegramPollingSession {
        pollState.error = message.message;
        return;
      }
+      if (message.type === "update") {
+        void writeTelegramSpooledUpdate({
+          spoolDir,
+          update: message.update,
+        }).then(
+          (updateId) => {
+            ackSpooledUpdate(message.requestId, { ok: true, updateId });
+          },
+          (err: unknown) => {
+            ackSpooledUpdate(message.requestId, {
+              ok: false,
+              message: formatErrorMessage(err),
+            });
+          },
+        );
+        return;
+      }
      if (message.type === "spooled") {
        liveness.noteGetUpdatesActivity();
      }
--- a/extensions/telegram/src/telegram-ingress-spool.test.ts
+++ b/extensions/telegram/src/telegram-ingress-spool.test.ts
@@ -1,7 +1,11 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it } from "vitest";
+import { createChannelIngressQueue } from "../../../src/channels/message/ingress-queue.js";
+import { closeOpenClawStateDatabaseForTest } from "../../../src/state/openclaw-state-db.js";
+import { clearTelegramRuntime, setTelegramRuntime } from "./runtime.js";
+import type { TelegramRuntime } from "./runtime.types.js";
 import {
  claimTelegramSpooledUpdate,
  deleteTelegramSpooledUpdate,
@@ -15,16 +19,37 @@ import {
  writeTelegramSpooledUpdate,
 } from "./telegram-ingress-spool.js";

+function installTelegramIngressQueueRuntime(resolveStateDir: () => string): void {
+  setTelegramRuntime({
+    state: {
+      resolveStateDir,
+      openChannelIngressQueue: (
+        options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
+      ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "telegram" }),
+    },
+  } as TelegramRuntime);
+}
+
 async function withTempSpool<T>(fn: (spoolDir: string) => Promise<T>): Promise<T> {
-  const spoolDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+  const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+  const spoolDir = path.join(stateDir, "telegram", "ingress-spool-test");
+  await fs.mkdir(spoolDir, { recursive: true });
+  installTelegramIngressQueueRuntime(() => stateDir);
  try {
    return await fn(spoolDir);
  } finally {
-    await fs.rm(spoolDir, { recursive: true, force: true });
+    clearTelegramRuntime();
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
  }
 }

 describe("Telegram ingress spool", () => {
+  afterEach(() => {
+    clearTelegramRuntime();
+    closeOpenClawStateDatabaseForTest();
+  });
+
  it("persists updates durably in update_id order and deletes handled entries", async () => {
    await withTempSpool(async (spoolDir) => {
      await writeTelegramSpooledUpdate({
@@ -84,9 +109,8 @@ describe("Telegram ingress spool", () => {
      if (!claimed) {
        throw new Error("Expected a claimed update");
      }
-      await fs.writeFile(claimed.pendingPath, "duplicate pending race", { mode: 0o600 });
      await deleteTelegramSpooledUpdate(claimed);
-      expect(await fs.readdir(spoolDir)).toEqual([]);
+      expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]);
    });
  });

@@ -139,48 +163,17 @@ describe("Telegram ingress spool", () => {

      expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]);
      expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]);
-      const entries = await fs.readdir(spoolDir);
-      expect(entries).toEqual(["0000000000000032.json.failed"]);
-      const failed = JSON.parse(
-        await fs.readFile(path.join(spoolDir, "0000000000000032.json.failed"), "utf8"),
-      ) as {
-        update?: unknown;
-        claim?: unknown;
-        failure?: { reason?: string; message?: string; failedAt?: number };
-      };
-      expect(failed.update).toBeUndefined();
-      expect(failed.claim).toBeUndefined();
-      expect(failed.failure).toEqual({
-        reason: "handler-timeout",
-        message: "timed out",
-        failedAt: 123,
-      });

      await writeTelegramSpooledUpdate({
        spoolDir,
        update: { update_id: 32, message: { text: "redelivered poison" } },
+        now: 124,
      });
      expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]);
-      expect(await fs.readdir(spoolDir)).toEqual(["0000000000000032.json.failed"]);
-
-      const leakedProcessingPath = path.join(spoolDir, "0000000000000032.json.processing");
-      await fs.writeFile(
-        leakedProcessingPath,
-        `${JSON.stringify({
-          version: 1,
-          updateId: 32,
-          receivedAt: 100,
-          update: { update_id: 32, message: { text: "crashed poison claim" } },
-        })}\n`,
-        { mode: 0o600 },
-      );
-      const staleTime = new Date(Date.now() - TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS - 1);
-      await fs.utimes(leakedProcessingPath, staleTime, staleTime);

      await expect(recoverStaleTelegramSpooledUpdateClaims({ spoolDir })).resolves.toBe(0);
      expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]);
      expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]);
-      expect(await fs.readdir(spoolDir)).toEqual(["0000000000000032.json.failed"]);
    });
  });

@@ -197,56 +190,44 @@ describe("Telegram ingress spool", () => {
      await deleteTelegramSpooledUpdate(update);

      await expect(claimTelegramSpooledUpdate(update)).resolves.toBeNull();
-      expect(await fs.readdir(spoolDir)).toEqual([]);
+      expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]);
    });
  });

-  it("recovers stale processing claims without replaying fresh claims", async () => {
+  it("recovers stale processing claims selected by the caller", async () => {
    await withTempSpool(async (spoolDir) => {
-      await writeTelegramSpooledUpdate({
-        spoolDir,
-        update: { update_id: 40, message: { text: "fresh" } },
-      });
      await writeTelegramSpooledUpdate({
        spoolDir,
        update: { update_id: 41, message: { text: "stale" } },
      });
      const updates = await listTelegramSpooledUpdates({ spoolDir });
-      const fresh = updates.find((update) => update.updateId === 40);
      const stale = updates.find((update) => update.updateId === 41);
-      if (!fresh || !stale) {
+      if (!stale) {
        throw new Error("Expected spooled updates");
      }
-      const claimedFresh = await claimTelegramSpooledUpdate(fresh);
      const claimedStale = await claimTelegramSpooledUpdate(stale);
-      if (!claimedFresh || !claimedStale) {
+      if (!claimedStale) {
        throw new Error("Expected claimed updates");
      }
      const now = Date.now();
-      const oldClaimTime = new Date(now - TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS - 1);
-      await fs.utimes(claimedStale.path, oldClaimTime, oldClaimTime);

      const recovered = await recoverStaleTelegramSpooledUpdateClaims({
        spoolDir,
-        now,
+        now: now + TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS + 1,
      });

      expect(recovered).toBe(1);
      expect(
        (await listTelegramSpooledUpdates({ spoolDir })).map((update) => update.updateId),
      ).toEqual([41]);
-      expect((await fs.readdir(spoolDir)).toSorted()).toEqual([
-        "0000000000000040.json.processing",
-        "0000000000000041.json",
-      ]);
    });
  });

-  it("handles ENOENT race when processing file is removed before recovery rename", async () => {
+  it("lets recovery callers keep a claim in processing", async () => {
    await withTempSpool(async (spoolDir) => {
      await writeTelegramSpooledUpdate({
        spoolDir,
-        update: { update_id: 45, message: { text: "vanishes" } },
+        update: { update_id: 45, message: { text: "busy" } },
      });
      const update = (await listTelegramSpooledUpdates({ spoolDir }))[0];
      if (!update) {
@@ -260,16 +241,17 @@ describe("Telegram ingress spool", () => {
      const recovered = await recoverStaleTelegramSpooledUpdateClaims({
        spoolDir,
        staleMs: 0,
-        shouldRecover: async () => {
+        shouldRecover: () => {
          shouldRecoverCalls += 1;
-          await fs.unlink(claimed.path);
-          return true;
+          return false;
        },
      });

      expect(recovered).toBe(0);
      expect(shouldRecoverCalls).toBe(1);
-      expect(await fs.readdir(spoolDir)).toEqual([]);
+      expect(
+        (await listTelegramSpooledUpdateClaims({ spoolDir })).map((claim) => claim.updateId),
+      ).toEqual([45]);
    });
  });

--- a/extensions/telegram/src/telegram-ingress-spool.ts
+++ b/extensions/telegram/src/telegram-ingress-spool.ts
@@ -1,18 +1,27 @@
 import { randomUUID } from "node:crypto";
-import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store";
+import {
+  type ChannelIngressQueue,
+  type ChannelIngressQueueClaim,
+  type ChannelIngressQueueClaimRef,
+  type ChannelIngressQueueRecord,
+} from "openclaw/plugin-sdk/channel-outbound";
 import { resolveStateDir } from "openclaw/plugin-sdk/state-paths";
+import { getTelegramRuntime } from "./runtime.js";

 const SPOOL_VERSION = 1;
+const TELEGRAM_INGRESS_SPOOL_PREFIX = "ingress-spool-";
 export const TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS = 6 * 60 * 60 * 1000;
+const TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000;
+const TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES = 1000;
 const TELEGRAM_SPOOLED_UPDATE_PROCESS_ID = `${process.pid}:${randomUUID()}`;

 type TelegramSpooledUpdateClaimOwner = {
  processId: string;
  processPid: number;
  claimedAt: number;
+  claimToken?: string;
 };

 type TelegramSpooledUpdatePayload = {
@@ -20,16 +29,6 @@ type TelegramSpooledUpdatePayload = {
  updateId: number;
  receivedAt: number;
  update: unknown;
-  claim?: TelegramSpooledUpdateClaimOwner;
-  failure?: {
-    reason: string;
-    message: string;
-    failedAt: number;
-  };
-};
-
-type TelegramFailedSpooledUpdatePayload = Omit<TelegramSpooledUpdatePayload, "claim" | "update"> & {
-  failure: NonNullable<TelegramSpooledUpdatePayload["failure"]>;
 };

 export type TelegramSpooledUpdate = {
@@ -61,7 +60,11 @@ export function resolveTelegramIngressSpoolDir(params: {
  env?: NodeJS.ProcessEnv;
 }): string {
  const stateDir = resolveStateDir(params.env, os.homedir);
-  return path.join(stateDir, "telegram", `ingress-spool-${normalizeAccountId(params.accountId)}`);
+  return path.join(
+    stateDir,
+    "telegram",
+    `${TELEGRAM_INGRESS_SPOOL_PREFIX}${normalizeAccountId(params.accountId)}`,
+  );
 }

 export function resolveTelegramUpdateId(update: unknown): number | null {
@@ -80,94 +83,75 @@ function processingFileName(updateId: number): string {
  return `${spoolFileName(updateId)}.processing`;
 }

-function failedFileName(updateId: number): string {
-  return `${spoolFileName(updateId)}.failed`;
+function queueEventId(updateId: number): string {
+  return String(updateId).padStart(16, "0");
 }

-function isProcessingFileName(fileName: string): boolean {
-  return fileName.endsWith(".json.processing");
-}
-
-function pendingFileNameFromProcessing(fileName: string): string {
-  return fileName.slice(0, -".processing".length);
+function pendingPath(spoolDir: string, updateId: number): string {
+  return path.join(spoolDir, spoolFileName(updateId));
 }

 function processingPath(spoolDir: string, updateId: number): string {
  return path.join(spoolDir, processingFileName(updateId));
 }

-function failedPath(spoolDir: string, updateId: number): string {
-  return path.join(spoolDir, failedFileName(updateId));
-}
-
-async function pathExists(filePath: string): Promise<boolean> {
-  try {
-    await fs.access(filePath);
-    return true;
-  } catch (err) {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return false;
-    }
-    throw err;
-  }
-}
-
-async function unlinkIfPresent(filePath: string): Promise<void> {
-  try {
-    await fs.unlink(filePath);
-  } catch (err) {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return;
-    }
-    throw err;
-  }
-}
-
-function parseSpooledUpdate(value: unknown, filePath: string): TelegramSpooledUpdate | null {
-  if (!value || typeof value !== "object") {
-    return null;
-  }
-  const payload = value as Partial<TelegramSpooledUpdatePayload>;
-  if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) {
-    return null;
-  }
-  const update: TelegramSpooledUpdate = {
-    updateId: payload.updateId,
-    path: filePath,
-    update: payload.update,
-    receivedAt: typeof payload.receivedAt === "number" ? payload.receivedAt : 0,
-  };
-  if (
-    payload.claim &&
-    typeof payload.claim.processId === "string" &&
-    isValidUpdateId(payload.claim.processPid) &&
-    typeof payload.claim.claimedAt === "number"
-  ) {
-    update.claim = payload.claim;
-  }
-  return update;
-}
-
-function buildClaimedPayload(update: TelegramSpooledUpdate): TelegramSpooledUpdatePayload {
+function resolveQueueParts(spoolDir: string): {
+  accountId: string;
+  stateDir: string;
+} {
+  const basename = path.basename(spoolDir);
+  const accountId = normalizeAccountId(
+    basename.startsWith(TELEGRAM_INGRESS_SPOOL_PREFIX)
+      ? basename.slice(TELEGRAM_INGRESS_SPOOL_PREFIX.length)
+      : basename,
+  );
+  const stateDir =
+    basename.startsWith(TELEGRAM_INGRESS_SPOOL_PREFIX) &&
+    path.basename(path.dirname(spoolDir)) === "telegram"
+      ? path.dirname(path.dirname(spoolDir))
+      : spoolDir;
  return {
-    version: SPOOL_VERSION,
-    updateId: update.updateId,
-    receivedAt: update.receivedAt,
-    update: update.update,
-    claim: {
-      processId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID,
-      processPid: process.pid,
-      claimedAt: Date.now(),
-    },
+    accountId,
+    stateDir,
  };
 }

+function createTelegramIngressQueue(
+  spoolDir: string,
+): ChannelIngressQueue<TelegramSpooledUpdatePayload> {
+  const parts = resolveQueueParts(spoolDir);
+  return getTelegramRuntime().state.openChannelIngressQueue<TelegramSpooledUpdatePayload>({
+    accountId: parts.accountId,
+    stateDir: parts.stateDir,
+  });
+}
+
+async function pruneTelegramIngressQueue(
+  queue: ChannelIngressQueue<TelegramSpooledUpdatePayload>,
+  now?: number,
+): Promise<void> {
+  await queue.prune({
+    failedTtlMs: TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS,
+    failedMaxEntries: TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES,
+    ...(now === undefined ? {} : { now }),
+  });
+}
+
+function processPidFromOwnerId(ownerId: string): number {
+  const pid = Number.parseInt(ownerId.split(":", 1)[0] ?? "", 10);
+  return Number.isSafeInteger(pid) && pid > 0 ? pid : -1;
+}
+
 function processExists(pid: number): boolean {
+  if (!Number.isSafeInteger(pid) || pid <= 0) {
+    return false;
+  }
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
-    return (err as { code?: string }).code !== "ESRCH";
+    const code = (err as { code?: string }).code;
+    return code !== "ESRCH" && code !== "EINVAL";
  }
 }

@@ -175,6 +159,52 @@ function isFreshClaimOwner(claim: TelegramSpooledUpdateClaimOwner): boolean {
  return Date.now() - claim.claimedAt < TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS;
 }

+function parseQueueRecord(
+  spoolDir: string,
+  record: ChannelIngressQueueRecord<TelegramSpooledUpdatePayload>,
+): TelegramSpooledUpdate | null {
+  const payload = record.payload;
+  if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) {
+    return null;
+  }
+  return {
+    updateId: payload.updateId,
+    path: pendingPath(spoolDir, payload.updateId),
+    update: payload.update,
+    receivedAt: payload.receivedAt,
+  };
+}
+
+function parseQueueClaim(
+  spoolDir: string,
+  record: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>,
+): ClaimedTelegramSpooledUpdate | null {
+  const update = parseQueueRecord(spoolDir, record);
+  if (!update) {
+    return null;
+  }
+  return {
+    ...update,
+    path: processingPath(spoolDir, update.updateId),
+    pendingPath: pendingPath(spoolDir, update.updateId),
+    claim: {
+      processId: record.claim.ownerId,
+      processPid: processPidFromOwnerId(record.claim.ownerId),
+      claimedAt: record.claim.claimedAt,
+      claimToken: record.claim.token,
+    },
+  };
+}
+
+function sortTelegramUpdates<T extends TelegramSpooledUpdate>(updates: T[]): T[] {
+  return updates.toSorted((a, b) => a.updateId - b.updateId);
+}
+
+function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef {
+  const id = queueEventId(update.updateId);
+  return update.claim?.claimToken ? { id, claim: { token: update.claim.claimToken } } : id;
+}
+
 export function isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(
  claim: ClaimedTelegramSpooledUpdate,
 ): boolean {
@@ -195,26 +225,19 @@ export async function writeTelegramSpooledUpdate(params: {
  if (updateId === null) {
    throw new Error("Telegram update missing numeric update_id.");
  }
-  await fs.mkdir(params.spoolDir, { recursive: true });
-  const targetPath = path.join(params.spoolDir, spoolFileName(updateId));
-  const claimedPath = processingPath(params.spoolDir, updateId);
-  const tombstonePath = failedPath(params.spoolDir, updateId);
-  if ((await pathExists(claimedPath)) || (await pathExists(tombstonePath))) {
-    return updateId;
-  }
-  const tempPath = path.join(params.spoolDir, `${spoolFileName(updateId)}.${randomUUID()}.tmp`);
-  const payload: TelegramSpooledUpdatePayload = {
-    version: SPOOL_VERSION,
-    updateId,
-    receivedAt: params.now ?? Date.now(),
-    update: params.update,
-  };
-  await fs.writeFile(tempPath, `${JSON.stringify(payload)}\n`, { mode: 0o600 });
-  if ((await pathExists(claimedPath)) || (await pathExists(tombstonePath))) {
-    await unlinkIfPresent(tempPath);
-    return updateId;
-  }
-  await fs.rename(tempPath, targetPath);
+  const receivedAt = params.now ?? Date.now();
+  const queue = createTelegramIngressQueue(params.spoolDir);
+  await pruneTelegramIngressQueue(queue, params.now);
+  await queue.enqueue(
+    queueEventId(updateId),
+    {
+      version: SPOOL_VERSION,
+      updateId,
+      receivedAt,
+      update: params.update,
+    },
+    { receivedAt },
+  );
  return updateId;
 }

@@ -222,95 +245,38 @@ export async function listTelegramSpooledUpdates(params: {
  spoolDir: string;
  limit?: number | "all";
 }): Promise<TelegramSpooledUpdate[]> {
-  let entries: string[];
-  try {
-    entries = await fs.readdir(params.spoolDir);
-  } catch (err) {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return [];
-    }
-    throw err;
-  }
-  const entrySet = new Set(entries);
-  const files = entries
-    .filter((entry) => entry.endsWith(".json") && !entrySet.has(`${entry}.failed`))
-    .toSorted();
-  const limitedFiles =
-    params.limit === "all" ? files : files.slice(0, Math.max(1, params.limit ?? 100));
-  const updates: TelegramSpooledUpdate[] = [];
-  for (const file of limitedFiles) {
-    const filePath = path.join(params.spoolDir, file);
-    const { value } = await readJsonFileWithFallback<unknown>(filePath, null);
-    const parsed = parseSpooledUpdate(value, filePath);
-    if (parsed) {
-      updates.push(parsed);
-    }
-  }
-  return updates;
+  const records = await createTelegramIngressQueue(params.spoolDir).listPending({
+    limit: params.limit ?? 100,
+    orderBy: "id",
+  });
+  return sortTelegramUpdates(
+    records.flatMap((record) => {
+      const update = parseQueueRecord(params.spoolDir, record);
+      return update ? [update] : [];
+    }),
+  );
 }

 export async function deleteTelegramSpooledUpdate(update: TelegramSpooledUpdate): Promise<void> {
-  await unlinkIfPresent(update.path);
-  if ("pendingPath" in update && typeof update.pendingPath === "string") {
-    await unlinkIfPresent(update.pendingPath);
-  }
+  await createTelegramIngressQueue(path.dirname(update.path)).delete(queueMutationTarget(update));
 }

 export async function claimTelegramSpooledUpdate(
  update: TelegramSpooledUpdate,
 ): Promise<ClaimedTelegramSpooledUpdate | null> {
-  const claimedPath = processingPath(path.dirname(update.path), update.updateId);
-  const holdPath = path.join(
-    path.dirname(update.path),
-    `${spoolFileName(update.updateId)}.${randomUUID()}.claim`,
-  );
-  const tempPath = path.join(
-    path.dirname(update.path),
-    `${processingFileName(update.updateId)}.${randomUUID()}.tmp`,
-  );
-  try {
-    const claimedAt = new Date();
-    await fs.writeFile(tempPath, `${JSON.stringify(buildClaimedPayload(update))}\n`, {
-      mode: 0o600,
-    });
-    await fs.link(update.path, holdPath);
-    await fs.link(tempPath, claimedPath);
-    await unlinkIfPresent(tempPath);
-    await unlinkIfPresent(holdPath);
-    await fs.utimes(claimedPath, claimedAt, claimedAt);
-    await unlinkIfPresent(update.path);
-  } catch (err) {
-    const code = (err as { code?: string }).code;
-    await unlinkIfPresent(tempPath);
-    await unlinkIfPresent(holdPath);
-    if (code === "ENOENT" || code === "EEXIST") {
-      return null;
-    }
-    throw err;
-  }
-  return {
-    ...update,
-    path: claimedPath,
-    pendingPath: update.path,
-  };
+  const spoolDir = path.dirname(update.path);
+  const claimed = await createTelegramIngressQueue(spoolDir).claim(queueEventId(update.updateId), {
+    ownerId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID,
+  });
+  return claimed ? parseQueueClaim(spoolDir, claimed) : null;
 }

 export async function releaseTelegramSpooledUpdateClaim(
  update: ClaimedTelegramSpooledUpdate,
 ): Promise<void> {
-  try {
-    await fs.rename(update.path, update.pendingPath);
-  } catch (err) {
-    const code = (err as { code?: string }).code;
-    if (code === "ENOENT") {
-      return;
-    }
-    if (code === "EEXIST") {
-      await unlinkIfPresent(update.path);
-      return;
-    }
-    throw err;
-  }
+  await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
+    queueMutationTarget(update),
+  );
 }

 export async function failTelegramSpooledUpdateClaim(params: {
@@ -319,70 +285,26 @@ export async function failTelegramSpooledUpdateClaim(params: {
  message: string;
  now?: number;
 }): Promise<boolean> {
-  const tombstonePath = failedPath(path.dirname(params.update.path), params.update.updateId);
-  const tempPath = path.join(
-    path.dirname(params.update.path),
-    `${failedFileName(params.update.updateId)}.${randomUUID()}.tmp`,
-  );
-  try {
-    const { value } = await readJsonFileWithFallback<unknown>(params.update.path, null);
-    const parsed = parseSpooledUpdate(value, params.update.path);
-    if (!parsed) {
-      return false;
-    }
-    const payload: TelegramFailedSpooledUpdatePayload = {
-      version: SPOOL_VERSION,
-      updateId: parsed.updateId,
-      receivedAt: parsed.receivedAt,
-      failure: {
-        reason: params.reason,
-        message: params.message,
-        failedAt: params.now ?? Date.now(),
-      },
-    };
-    await fs.writeFile(tempPath, `${JSON.stringify(payload)}\n`, { mode: 0o600 });
-    await fs.rename(tempPath, tombstonePath);
-    await unlinkIfPresent(params.update.path);
-    await unlinkIfPresent(params.update.pendingPath);
-    return true;
-  } catch (err) {
-    await unlinkIfPresent(tempPath);
-    if ((err as { code?: string }).code === "ENOENT") {
-      return false;
-    }
-    throw err;
-  }
+  const queue = createTelegramIngressQueue(path.dirname(params.update.pendingPath));
+  const failed = await queue.fail(queueMutationTarget(params.update), {
+    reason: params.reason,
+    message: params.message,
+    ...(params.now === undefined ? {} : { failedAt: params.now }),
+  });
+  await pruneTelegramIngressQueue(queue, params.now);
+  return failed;
 }

 export async function listTelegramSpooledUpdateClaims(params: {
  spoolDir: string;
 }): Promise<ClaimedTelegramSpooledUpdate[]> {
-  let entries: string[];
-  try {
-    entries = await fs.readdir(params.spoolDir);
-  } catch (err) {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return [];
-    }
-    throw err;
-  }
-  const claims: ClaimedTelegramSpooledUpdate[] = [];
-  const entrySet = new Set(entries);
-  for (const file of entries.filter(isProcessingFileName).toSorted()) {
-    if (entrySet.has(`${pendingFileNameFromProcessing(file)}.failed`)) {
-      continue;
-    }
-    const filePath = path.join(params.spoolDir, file);
-    const { value } = await readJsonFileWithFallback<unknown>(filePath, null);
-    const parsed = parseSpooledUpdate(value, filePath);
-    if (parsed) {
-      claims.push({
-        ...parsed,
-        pendingPath: path.join(params.spoolDir, pendingFileNameFromProcessing(file)),
-      });
-    }
-  }
-  return claims;
+  const claims = await createTelegramIngressQueue(params.spoolDir).listClaims();
+  return sortTelegramUpdates(
+    claims.flatMap((claim) => {
+      const update = parseQueueClaim(params.spoolDir, claim);
+      return update ? [update] : [];
+    }),
+  );
 }

 export async function recoverStaleTelegramSpooledUpdateClaims(params: {
@@ -391,73 +313,17 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: {
  now?: number;
  shouldRecover?: (claim: ClaimedTelegramSpooledUpdate) => boolean | Promise<boolean>;
 }): Promise<number> {
-  let entries: string[];
-  try {
-    entries = await fs.readdir(params.spoolDir);
-  } catch (err) {
-    if ((err as { code?: string }).code === "ENOENT") {
-      return 0;
-    }
-    throw err;
-  }
-  const staleMs = Math.max(
-    0,
-    Math.floor(params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS),
-  );
-  const now = params.now ?? Date.now();
-  let recovered = 0;
-  const entrySet = new Set(entries);
-  for (const entry of entries.filter(isProcessingFileName).toSorted()) {
-    const claimedPath = path.join(params.spoolDir, entry);
-    const pendingPath = path.join(params.spoolDir, pendingFileNameFromProcessing(entry));
-    if (entrySet.has(`${pendingFileNameFromProcessing(entry)}.failed`)) {
-      await unlinkIfPresent(claimedPath);
-      await unlinkIfPresent(pendingPath);
-      continue;
-    }
-    let stat;
-    try {
-      stat = await fs.stat(claimedPath);
-    } catch (err) {
-      if ((err as { code?: string }).code === "ENOENT") {
-        continue;
-      }
-      throw err;
-    }
-    if (now - stat.mtimeMs < staleMs) {
-      continue;
-    }
-    if (params.shouldRecover) {
-      const { value } = await readJsonFileWithFallback<unknown>(claimedPath, null);
-      const parsed = parseSpooledUpdate(value, claimedPath);
-      if (
-        parsed &&
-        !(await params.shouldRecover({
-          ...parsed,
-          pendingPath,
-        }))
-      ) {
-        continue;
-      }
-    }
-    if (await pathExists(pendingPath)) {
-      await unlinkIfPresent(claimedPath);
-    } else {
-      try {
-        await fs.rename(claimedPath, pendingPath);
-      } catch (err) {
-        const code = (err as { code?: string }).code;
-        if (code === "ENOENT") {
-          continue;
+  const shouldRecover = params.shouldRecover;
+  return await createTelegramIngressQueue(params.spoolDir).recoverStaleClaims({
+    staleMs: params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS,
+    ...(params.now === undefined ? {} : { now: params.now }),
+    ...(shouldRecover
+      ? {
+          shouldRecover: async (claim: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>) => {
+            const update = parseQueueClaim(params.spoolDir, claim);
+            return update ? await shouldRecover(update) : false;
+          },
        }
-        if (code === "EEXIST") {
-          await unlinkIfPresent(claimedPath);
-        } else {
-          throw err;
-        }
-      }
-    }
-    recovered += 1;
-  }
-  return recovered;
+      : {}),
+  });
 }
--- a/extensions/telegram/src/telegram-ingress-worker.runtime.ts
+++ b/extensions/telegram/src/telegram-ingress-worker.runtime.ts
@@ -8,8 +8,8 @@ import {
  TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS,
  resolveTelegramLongPollTimeoutSeconds,
 } from "./request-timeouts.js";
-import { writeTelegramSpooledUpdate } from "./telegram-ingress-spool.js";
 import type {
+  TelegramIngressWorkerCommand,
  TelegramIngressWorkerMessage,
  TelegramIngressWorkerOptions,
 } from "./telegram-ingress-worker.js";
@@ -20,6 +20,14 @@ const retryInitialMs = 1000;
 const retryMaxMs = 30_000;
 let stopped = false;
 let activeController: AbortController | undefined;
+let nextSpoolRequestId = 0;
+const pendingSpoolRequests = new Map<
+  string,
+  {
+    resolve(updateId: number): void;
+    reject(err: Error): void;
+  }
+>();

 function post(message: TelegramIngressWorkerMessage): void {
  if (parentPort) {
@@ -44,14 +52,53 @@ function resolveBackoff(attempt: number): number {
  return Math.min(retryMaxMs, retryInitialMs * 2 ** Math.max(0, attempt - 1));
 }

-parentPort?.on("message", (message: { type?: string }) => {
-  if (message?.type !== "stop") {
+function rejectPendingSpoolRequests(err: Error): void {
+  for (const pending of pendingSpoolRequests.values()) {
+    pending.reject(err);
+  }
+  pendingSpoolRequests.clear();
+}
+
+parentPort?.on("message", (message: TelegramIngressWorkerCommand) => {
+  if (message?.type === "stop") {
+    stopped = true;
+    const err = new Error("telegram ingress worker stopped");
+    activeController?.abort(err);
+    rejectPendingSpoolRequests(err);
    return;
  }
-  stopped = true;
-  activeController?.abort(new Error("telegram ingress worker stopped"));
+  if (message?.type !== "spool-ack") {
+    return;
+  }
+  const pending = pendingSpoolRequests.get(message.requestId);
+  if (!pending) {
+    return;
+  }
+  pendingSpoolRequests.delete(message.requestId);
+  if (message.result.ok) {
+    pending.resolve(message.result.updateId);
+    return;
+  }
+  pending.reject(new Error(message.result.message));
 });

+async function requestSpoolUpdate(params: { update: unknown; queued: number }): Promise<number> {
+  if (!parentPort) {
+    throw new Error("Telegram ingress worker missing parent port.");
+  }
+  const requestId = String(++nextSpoolRequestId);
+  const updateId = await new Promise<number>((resolve, reject) => {
+    pendingSpoolRequests.set(requestId, { resolve, reject });
+    post({
+      type: "update",
+      requestId,
+      update: params.update,
+      queued: params.queued,
+    });
+  });
+  return updateId;
+}
+
 async function fetchJson(params: {
  fetch: typeof fetch;
  url: string;
@@ -127,10 +174,7 @@ async function main(): Promise<void> {
          if (stopped) {
            break;
          }
-          const updateId = await writeTelegramSpooledUpdate({
-            spoolDir: options.spoolDir,
-            update,
-          });
+          const updateId = await requestSpoolUpdate({ update, queued: result.length });
          if (lastUpdateId === null || updateId > lastUpdateId) {
            lastUpdateId = updateId;
          }
--- a/extensions/telegram/src/telegram-ingress-worker.ts
+++ b/extensions/telegram/src/telegram-ingress-worker.ts
@@ -22,6 +22,30 @@ export type TelegramIngressWorkerMessage =
      type: "spooled";
      updateId: number;
      queued: number;
+    }
+  | {
+      type: "update";
+      requestId: string;
+      update: unknown;
+      queued: number;
+    };
+
+export type TelegramIngressWorkerCommand =
+  | {
+      type: "stop";
+    }
+  | {
+      type: "spool-ack";
+      requestId: string;
+      result:
+        | {
+            ok: true;
+            updateId: number;
+          }
+        | {
+            ok: false;
+            message: string;
+          };
    };

 export type TelegramIngressWorkerOptions = {
@@ -37,6 +61,18 @@ export type TelegramIngressWorkerOptions = {

 export type TelegramIngressWorkerHandle = {
  onMessage(listener: (message: TelegramIngressWorkerMessage) => void): () => void;
+  ackSpooledUpdate?(
+    requestId: string,
+    result:
+      | {
+          ok: true;
+          updateId: number;
+        }
+      | {
+          ok: false;
+          message: string;
+        },
+  ): void;
  stop(): Promise<void>;
  task(): Promise<void>;
 };
@@ -73,9 +109,18 @@ export const createTelegramIngressWorker: TelegramIngressWorkerFactory = (option
        listeners.delete(listener);
      };
    },
+    ackSpooledUpdate(requestId, result) {
+      try {
+        Reflect.apply(Reflect.get(worker, "postMessage") as (value: unknown) => void, worker, [
+          { type: "spool-ack", requestId, result } satisfies TelegramIngressWorkerCommand,
+        ]);
+      } catch {
+        // Worker may have exited after the parent committed the queue write.
+      }
+    },
    async stop() {
      Reflect.apply(Reflect.get(worker, "postMessage") as (value: unknown) => void, worker, [
-        { type: "stop" },
+        { type: "stop" } satisfies TelegramIngressWorkerCommand,
      ]);
      const timeout = setTimeout(() => {
        void worker.terminate();
--- a/extensions/whatsapp/src/inbound.media.test.ts
+++ b/extensions/whatsapp/src/inbound.media.test.ts
@@ -140,15 +140,25 @@ vi.mock("openclaw/plugin-sdk/media-store", async () => {
  };
 });

-vi.mock("./runtime.js", () => ({
-  getOptionalWhatsAppRuntime: () => undefined,
-  getWhatsAppRuntime: () => ({
-    state: {
-      openKeyedStore: () => createInMemoryKeyedStore(),
-    },
-  }),
-  setWhatsAppRuntime: vi.fn(),
-}));
+vi.mock("./runtime.js", async () => {
+  const { createChannelIngressQueue } = await vi.importActual<
+    typeof import("../../../src/channels/message/ingress-queue.js")
+  >("../../../src/channels/message/ingress-queue.js");
+  const stateDir = `/tmp/openclaw-whatsapp-inbound-media-${Date.now()}-${Math.random()}`;
+  return {
+    getOptionalWhatsAppRuntime: () => undefined,
+    getWhatsAppRuntime: () => ({
+      state: {
+        resolveStateDir: () => stateDir,
+        openKeyedStore: () => createInMemoryKeyedStore(),
+        openChannelIngressQueue: (
+          options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
+        ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "whatsapp" }),
+      },
+    }),
+    setWhatsAppRuntime: vi.fn(),
+  };
+});

 const HOME = path.join(os.tmpdir(), `openclaw-inbound-media-${crypto.randomUUID()}`);
 const ORIGINAL_HOME = process.env.HOME;
--- a/extensions/whatsapp/src/inbound/durable-receive.ts
+++ b/extensions/whatsapp/src/inbound/durable-receive.ts
@@ -1,6 +1,6 @@
 import { createHash } from "node:crypto";
 import type { WAMessage } from "baileys";
-import { createDurableInboundReceiveJournal } from "openclaw/plugin-sdk/channel-outbound";
+import { createDurableInboundReceiveJournalFromQueue } from "openclaw/plugin-sdk/channel-outbound";
 import type { PluginJsonValue } from "openclaw/plugin-sdk/plugin-entry";
 import { getWhatsAppRuntime } from "../runtime.js";
 import { BufferJSON } from "../session.runtime.js";
@@ -56,24 +56,25 @@ export function deserializeWhatsAppDurableInboundMessage(
 }

 export function createWhatsAppDurableInboundReceiveJournal(accountId: string) {
-  const runtime = getWhatsAppRuntime();
  const accountPart = hashNamespacePart(accountId);
-  return createDurableInboundReceiveJournal<
+  const runtime = getWhatsAppRuntime();
+  const queue = runtime.state.openChannelIngressQueue<
    WhatsAppDurableInboundPayload,
    WhatsAppDurableInboundMetadata,
    WhatsAppDurableInboundCompletedMetadata
  >({
-    pendingStore: runtime.state.openKeyedStore({
-      namespace: `inbound.v1.pending.${accountPart}`,
-      maxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES,
-      defaultTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS,
-    }),
-    completedStore: runtime.state.openKeyedStore({
-      namespace: `inbound.v1.completed.${accountPart}`,
-      maxEntries: WHATSAPP_DURABLE_INBOUND_COMPLETED_MAX_ENTRIES,
-      defaultTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS,
-    }),
-    pendingTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS,
-    completedTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS,
+    accountId: accountPart,
+    stateDir: runtime.state.resolveStateDir(),
+  });
+  return createDurableInboundReceiveJournalFromQueue({
+    queue,
+    retention: {
+      pendingTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS,
+      completedTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS,
+      failedTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS,
+      pendingMaxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES,
+      completedMaxEntries: WHATSAPP_DURABLE_INBOUND_COMPLETED_MAX_ENTRIES,
+      failedMaxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES,
+    },
  });
 }
--- a/extensions/whatsapp/src/monitor-inbox.test-harness.ts
+++ b/extensions/whatsapp/src/monitor-inbox.test-harness.ts
@@ -63,6 +63,7 @@ const pluginRuntimeMocks = vi.hoisted(() => {
  type StoreEntry = { key: string; value: unknown; createdAt: number };
  const stores = new Map<string, Map<string, StoreEntry>>();
  let nextRegisterIfAbsentError: Error | undefined;
+  let stateDir = `/tmp/openclaw-whatsapp-ingress-${Date.now()}-${Math.random()}`;

  const openKeyedStore = vi.fn((options: { namespace: string }) => {
    let store = stores.get(options.namespace);
@@ -102,6 +103,7 @@ const pluginRuntimeMocks = vi.hoisted(() => {

  return {
    openKeyedStore,
+    stateDir: () => stateDir,
    failNextRegisterIfAbsent: (error: Error) => {
      nextRegisterIfAbsentError = error;
    },
@@ -109,6 +111,7 @@ const pluginRuntimeMocks = vi.hoisted(() => {
      stores.clear();
      nextRegisterIfAbsentError = undefined;
      openKeyedStore.mockClear();
+      stateDir = `/tmp/openclaw-whatsapp-ingress-${Date.now()}-${Math.random()}`;
    },
  };
 });
@@ -132,14 +135,23 @@ vi.mock("openclaw/plugin-sdk/channel-activity-runtime", async () => {
  };
 });

-vi.mock("./runtime.js", () => ({
-  getWhatsAppRuntime: () => ({
-    state: {
-      openKeyedStore: pluginRuntimeMocks.openKeyedStore,
-    },
-  }),
-  setWhatsAppRuntime: vi.fn(),
-}));
+vi.mock("./runtime.js", async () => {
+  const { createChannelIngressQueue } = await vi.importActual<
+    typeof import("../../../src/channels/message/ingress-queue.js")
+  >("../../../src/channels/message/ingress-queue.js");
+  return {
+    getWhatsAppRuntime: () => ({
+      state: {
+        resolveStateDir: pluginRuntimeMocks.stateDir,
+        openKeyedStore: pluginRuntimeMocks.openKeyedStore,
+        openChannelIngressQueue: (
+          options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
+        ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "whatsapp" }),
+      },
+    }),
+    setWhatsAppRuntime: vi.fn(),
+  };
+});

 const inboundRuntimeMocks = vi.hoisted(() => {
  const wrapperKeys = [
--- a/src/channels/message/durable-receive.test.ts
+++ b/src/channels/message/durable-receive.test.ts
@@ -1,9 +1,15 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
 import { describe, expect, it } from "vitest";
 import type {
  PluginStateEntry,
  PluginStateKeyedStore,
 } from "../../plugin-state/plugin-state-store.types.js";
+import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js";
+import { createDurableInboundReceiveJournalFromQueue } from "./durable-receive.js";
 import { createDurableInboundReceiveJournal } from "./durable-receive.js";
+import { createChannelIngressQueue } from "./ingress-queue.js";

 type TestPayload = { body: string };
 type TestMetadata = { source: string };
@@ -56,6 +62,16 @@ function createMemoryStore<T>(): PluginStateKeyedStore<T> {
  };
 }

+async function withTempState<T>(fn: (stateDir: string) => Promise<T>): Promise<T> {
+  const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-durable-receive-"));
+  try {
+    return await fn(stateDir);
+  } finally {
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
 describe("createDurableInboundReceiveJournal", () => {
  it("accepts pending records once and reports duplicate pending deliveries", async () => {
    const journal = createDurableInboundReceiveJournal<
@@ -298,4 +314,70 @@ describe("createDurableInboundReceiveJournal", () => {
      },
    ]);
  });
+
+  it("can use the shared channel ingress queue as durable storage", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueue<TestPayload, TestMetadata, TestCompletedMetadata>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => 10,
+      });
+      const journal = createDurableInboundReceiveJournalFromQueue({
+        queue,
+        retention: { completedMaxEntries: 1 },
+      });
+
+      await expect(
+        journal.accept("message-1", { body: "hello" }, { metadata: { source: "live" } }),
+      ).resolves.toMatchObject({
+        kind: "accepted",
+        duplicate: false,
+        record: {
+          id: "message-1",
+          payload: { body: "hello" },
+          metadata: { source: "live" },
+          receivedAt: 10,
+        },
+      });
+
+      await expect(journal.pending()).resolves.toMatchObject([
+        {
+          id: "message-1",
+          payload: { body: "hello" },
+          metadata: { source: "live" },
+        },
+      ]);
+
+      await expect(journal.release("message-1", { lastError: "retry" })).resolves.toBe(true);
+      await expect(journal.pending()).resolves.toMatchObject([
+        {
+          id: "message-1",
+          attempts: 1,
+          lastError: "retry",
+        },
+      ]);
+
+      await journal.complete("message-1", {
+        metadata: { delivered: true },
+        completedAt: 20,
+      });
+      await expect(journal.accept("message-1", { body: "again" })).resolves.toMatchObject({
+        kind: "completed",
+        duplicate: true,
+        record: {
+          id: "message-1",
+          completedAt: 20,
+          metadata: { delivered: true },
+        },
+      });
+
+      await journal.accept("message-2", { body: "new" });
+      await journal.complete("message-2", { completedAt: 21 });
+      await expect(journal.accept("message-1", { body: "past retention" })).resolves.toMatchObject({
+        kind: "accepted",
+        duplicate: false,
+      });
+    });
+  });
 });
--- a/src/channels/message/durable-receive.ts
+++ b/src/channels/message/durable-receive.ts
@@ -1,4 +1,5 @@
 import type { PluginStateKeyedStore } from "../../plugin-state/plugin-state-store.types.js";
+import type { ChannelIngressQueue, ChannelIngressQueuePruneOptions } from "./ingress-queue.js";

 export type DurableInboundReceivePendingRecord<TPayload, TMetadata = unknown> = {
  id: string;
@@ -72,6 +73,11 @@ export type DurableInboundReceiveJournal<TPayload, TMetadata, TCompletedMetadata
  deletePending(id: string): Promise<boolean>;
 };

+export type DurableInboundReceiveQueueJournalOptions<TPayload, TMetadata, TCompletedMetadata> = {
+  queue: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>;
+  retention?: ChannelIngressQueuePruneOptions;
+};
+
 function normalizeDurableInboundReceiveId(id: string): string {
  const normalized = id.trim();
  if (!normalized) {
@@ -222,3 +228,80 @@ export function createDurableInboundReceiveJournal<
    deletePending: (id) => options.pendingStore.delete(normalizeDurableInboundReceiveId(id)),
  };
 }
+
+export function createDurableInboundReceiveJournalFromQueue<
+  TPayload,
+  TMetadata = unknown,
+  TCompletedMetadata = unknown,
+>(
+  options: DurableInboundReceiveQueueJournalOptions<TPayload, TMetadata, TCompletedMetadata>,
+): DurableInboundReceiveJournal<TPayload, TMetadata, TCompletedMetadata> {
+  const prune = async (protectId?: string) => {
+    if (options.retention) {
+      await options.queue.prune({
+        ...options.retention,
+        ...(protectId === undefined ? {} : { protectIds: [protectId] }),
+      });
+    }
+  };
+  return {
+    accept: async (id, payload, acceptOptions) => {
+      await prune();
+      const result = await options.queue.enqueue(normalizeDurableInboundReceiveId(id), payload, {
+        ...(acceptOptions?.metadata === undefined ? {} : { metadata: acceptOptions.metadata }),
+        ...(acceptOptions?.receivedAt === undefined
+          ? {}
+          : { receivedAt: acceptOptions.receivedAt }),
+      });
+      await prune(normalizeDurableInboundReceiveId(id));
+      if (result.kind === "accepted") {
+        return { kind: "accepted", duplicate: false, record: result.record };
+      }
+      if (result.kind === "completed") {
+        return { kind: "completed", duplicate: true, record: result.record };
+      }
+      if (result.kind === "pending" || result.kind === "claimed") {
+        return { kind: "pending", duplicate: true, record: result.record };
+      }
+      return {
+        kind: "pending",
+        duplicate: true,
+        record: {
+          id: result.record.id,
+          payload,
+          receivedAt: result.record.failedAt,
+          updatedAt: result.record.failedAt,
+          attempts: 0,
+        },
+      };
+    },
+    pending: async () => {
+      await prune();
+      return await options.queue.listPending({ limit: "all" });
+    },
+    complete: async (id, completeOptions) => {
+      await options.queue.complete(normalizeDurableInboundReceiveId(id), {
+        ...(completeOptions?.metadata === undefined ? {} : { metadata: completeOptions.metadata }),
+        ...(completeOptions?.completedAt === undefined
+          ? {}
+          : { completedAt: completeOptions.completedAt }),
+      });
+      await prune(normalizeDurableInboundReceiveId(id));
+    },
+    release: async (id, releaseOptions) => {
+      const released = await options.queue.release(normalizeDurableInboundReceiveId(id), {
+        ...(releaseOptions?.lastError === undefined ? {} : { lastError: releaseOptions.lastError }),
+        ...(releaseOptions?.releasedAt === undefined
+          ? {}
+          : { releasedAt: releaseOptions.releasedAt }),
+      });
+      await prune(normalizeDurableInboundReceiveId(id));
+      return released;
+    },
+    deletePending: async (id) => {
+      const deleted = await options.queue.delete(normalizeDurableInboundReceiveId(id));
+      await prune();
+      return deleted;
+    },
+  };
+}
--- a/src/channels/message/index.ts
+++ b/src/channels/message/index.ts
@@ -1,7 +1,11 @@
 export { deriveDurableFinalDeliveryRequirements } from "./capabilities.js";
 export { defineChannelMessageAdapter } from "./adapter.js";
 export { createChannelMessageAdapterFromOutbound } from "./outbound-bridge.js";
-export { createDurableInboundReceiveJournal } from "./durable-receive.js";
+export {
+  createDurableInboundReceiveJournal,
+  createDurableInboundReceiveJournalFromQueue,
+} from "./durable-receive.js";
+export { createChannelIngressQueue } from "./ingress-queue.js";
 export {
  listDeclaredChannelMessageLiveCapabilities,
  listDeclaredDurableFinalCapabilities,
@@ -48,8 +52,20 @@ export type {
  DurableInboundReceiveJournal,
  DurableInboundReceiveJournalOptions,
  DurableInboundReceivePendingRecord,
+  DurableInboundReceiveQueueJournalOptions,
  DurableInboundReceiveReleaseOptions,
 } from "./durable-receive.js";
+export type {
+  ChannelIngressQueue,
+  ChannelIngressQueueClaim,
+  ChannelIngressQueueClaimRef,
+  ChannelIngressQueueCompletedRecord,
+  ChannelIngressQueueEnqueueResult,
+  ChannelIngressQueueFailedRecord,
+  ChannelIngressQueuePruneOptions,
+  ChannelIngressQueueRecord,
+  CreateChannelIngressQueueOptions,
+} from "./ingress-queue.js";
 export type {
  ChannelMessageOutboundBridgeAdapter,
  ChannelMessageOutboundBridgeResult,
--- a/src/channels/message/ingress-queue.test.ts
+++ b/src/channels/message/ingress-queue.test.ts
@@ -0,0 +1,323 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js";
+import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js";
+import {
+  closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
+} from "../../state/openclaw-state-db.js";
+import { createChannelIngressQueue } from "./ingress-queue.js";
+
+type ChannelIngressTestDatabase = Pick<OpenClawStateKyselyDatabase, "channel_ingress_events">;
+
+async function withTempState<T>(fn: (stateDir: string) => Promise<T>): Promise<T> {
+  const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-queue-"));
+  try {
+    return await fn(stateDir);
+  } finally {
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
+describe("channel ingress queue", () => {
+  afterEach(() => {
+    closeOpenClawStateDatabaseForTest();
+  });
+
+  it("deduplicates pending and completed ingress events", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueue<
+        { text: string },
+        { source: string },
+        { handledBy: string }
+      >({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => 100,
+      });
+
+      const accepted = await queue.enqueue(
+        "event-1",
+        { text: "first" },
+        { metadata: { source: "fixture" }, receivedAt: 50 },
+      );
+      const pending = await queue.enqueue("event-1", { text: "duplicate" });
+      await queue.complete("event-1", { metadata: { handledBy: "worker" }, completedAt: 150 });
+      const completed = await queue.enqueue("event-1", { text: "late duplicate" });
+
+      expect(accepted.kind).toBe("accepted");
+      expect(pending.kind).toBe("pending");
+      expect(pending.record.payload).toEqual({ text: "first" });
+      expect(completed).toEqual({
+        kind: "completed",
+        duplicate: true,
+        record: {
+          id: "event-1",
+          channelId: "test",
+          accountId: "account",
+          queueName: JSON.stringify(["test", "account"]),
+          completedAt: 150,
+          metadata: { handledBy: "worker" },
+        },
+      });
+      expect(await queue.listPending()).toEqual([]);
+
+      expect(
+        await queue.complete("missing-event", {
+          metadata: { handledBy: "late-worker" },
+          completedAt: 200,
+        }),
+      ).toBe(true);
+      expect(await queue.enqueue("missing-event", { text: "late duplicate" })).toMatchObject({
+        kind: "completed",
+        duplicate: true,
+        record: {
+          id: "missing-event",
+          completedAt: 200,
+          metadata: { handledBy: "late-worker" },
+        },
+      });
+
+      await queue.enqueue(" spaced-event ", { text: "spaced" });
+      expect(await queue.complete(" spaced-event ", { completedAt: 250 })).toBe(true);
+      expect(await queue.enqueue("spaced-event", { text: "duplicate" })).toMatchObject({
+        kind: "completed",
+        duplicate: true,
+        record: { id: "spaced-event", completedAt: 250 },
+      });
+    });
+  });
+
+  it("keeps channel and account queue identities unambiguous", async () => {
+    await withTempState(async (stateDir) => {
+      const first = createChannelIngressQueue<{ text: string }>({
+        channelId: "a",
+        accountId: "b:c",
+        stateDir,
+      });
+      const second = createChannelIngressQueue<{ text: string }>({
+        channelId: "a:b",
+        accountId: "c",
+        stateDir,
+      });
+
+      expect(await first.enqueue("same-id", { text: "first" })).toMatchObject({
+        kind: "accepted",
+      });
+      expect(await second.enqueue("same-id", { text: "second" })).toMatchObject({
+        kind: "accepted",
+      });
+
+      await first.complete("same-id");
+
+      expect(await first.enqueue("same-id", { text: "first duplicate" })).toMatchObject({
+        kind: "completed",
+      });
+      expect(await second.enqueue("same-id", { text: "second duplicate" })).toMatchObject({
+        kind: "pending",
+        record: { payload: { text: "second" } },
+      });
+    });
+  });
+
+  it("can bound pending scans and prune stale pending rows", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 1;
+      const queue = createChannelIngressQueue<{ index: number }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => clock++,
+      });
+
+      await queue.enqueue("0002", { index: 2 });
+      await queue.enqueue("0001", { index: 1 });
+      await queue.enqueue("0003", { index: 3 });
+
+      expect(
+        (await queue.listPending({ limit: 2, orderBy: "id" })).map((record) => record.id),
+      ).toEqual(["0001", "0002"]);
+      expect(await queue.prune({ pendingTtlMs: 3, pendingMaxEntries: 1, now: 7 })).toBe(2);
+      expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([
+        "0003",
+      ]);
+    });
+  });
+
+  it("does not prune protected rows while enforcing max-entry limits", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueue<{ index: number }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => 10,
+      });
+
+      await queue.enqueue("z", { index: 1 });
+      await queue.enqueue("a", { index: 2 });
+
+      expect(await queue.prune({ pendingMaxEntries: 1, protectIds: ["a"] })).toBe(0);
+      expect(
+        (await queue.listPending({ limit: "all", orderBy: "id" })).map((row) => row.id),
+      ).toEqual(["a", "z"]);
+    });
+  });
+
+  it("prunes max-entry overflow across bounded batches", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 1;
+      const queue = createChannelIngressQueue<{ index: number }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => clock++,
+      });
+
+      for (let index = 0; index < 520; index += 1) {
+        await queue.enqueue(String(index).padStart(4, "0"), { index });
+      }
+
+      expect(await queue.prune({ pendingMaxEntries: 2 })).toBe(518);
+      expect((await queue.listPending({ limit: "all" })).map((row) => row.id)).toEqual([
+        "0518",
+        "0519",
+      ]);
+    });
+  });
+
+  it("claims, releases, and skips blocked lanes", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 1;
+      const queue = createChannelIngressQueue<{ text: string }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => clock++,
+      });
+
+      await queue.enqueue("a", { text: "blocked" }, { laneKey: "chat-1", receivedAt: 1 });
+      await queue.enqueue("b", { text: "open" }, { laneKey: "chat-2", receivedAt: 2 });
+
+      const claimed = await queue.claimNext({
+        ownerId: "worker",
+        blockedLaneKeys: ["chat-1"],
+      });
+
+      expect(claimed?.id).toBe("b");
+      if (!claimed) {
+        throw new Error("Expected a claimed ingress event");
+      }
+      expect(await queue.release(claimed, { lastError: "retry", releasedAt: 20 })).toBe(true);
+      expect((await queue.listPending()).find((record) => record.id === "b")).toMatchObject({
+        attempts: 1,
+        lastAttemptAt: 20,
+        lastError: "retry",
+      });
+    });
+  });
+
+  it("requires claim tokens before mutating claimed rows", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueue<{ text: string }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => 10,
+      });
+
+      await queue.enqueue("event-1", { text: "claimed" });
+      const claimed = await queue.claim("event-1", { ownerId: "worker" });
+      if (!claimed) {
+        throw new Error("Expected a claimed ingress event");
+      }
+
+      expect(await queue.complete("event-1")).toBe(false);
+      expect(await queue.release("event-1")).toBe(false);
+      expect(await queue.fail("event-1", { reason: "stale-handler" })).toBe(false);
+      expect(await queue.delete("event-1")).toBe(false);
+
+      expect(await queue.complete(claimed, { completedAt: 20 })).toBe(true);
+      const duplicate = await queue.enqueue("event-1", { text: "duplicate" });
+      expect(duplicate.kind).toBe("completed");
+    });
+  });
+
+  it("recovers stale claims and prunes completed or failed rows", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueue<{ text: string }>({
+        channelId: "test",
+        accountId: "account",
+        stateDir,
+        now: () => 10,
+      });
+
+      await queue.enqueue("old", { text: "old" });
+      await queue.enqueue("keep", { text: "keep" });
+      const old = await queue.claim("old", { ownerId: "worker" });
+      const keep = await queue.claim("keep", { ownerId: "worker" });
+      if (!keep) {
+        throw new Error("Expected a claimed ingress event");
+      }
+
+      expect(
+        await queue.recoverStaleClaims({
+          staleMs: 5,
+          now: 20,
+          shouldRecover: (claim) => claim.id === old?.id,
+        }),
+      ).toBe(1);
+      expect((await queue.listPending()).map((record) => record.id)).toEqual(["old"]);
+      expect((await queue.listClaims()).map((record) => record.id)).toEqual(["keep"]);
+
+      await queue.complete("old", { completedAt: 25 });
+      await queue.fail(keep, { reason: "poison", message: "bad", failedAt: 25 });
+      await queue.enqueue("retry", { text: "retry" });
+      await queue.release("retry", { lastError: "stale retry text", releasedAt: 26 });
+      await queue.complete("retry", { completedAt: 27 });
+
+      const database = openOpenClawStateDatabase({
+        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
+      });
+      const kysely = getNodeSqliteKysely<ChannelIngressTestDatabase>(database.db);
+      const rows = executeSqliteQuerySync(
+        database.db,
+        kysely
+          .selectFrom("channel_ingress_events")
+          .select(["event_id", "payload_json", "metadata_json", "last_attempt_at", "last_error"])
+          .where("event_id", "in", ["old", "keep", "retry"])
+          .orderBy("event_id", "asc"),
+      ).rows;
+      expect(rows).toEqual([
+        {
+          event_id: "keep",
+          last_attempt_at: null,
+          last_error: "bad",
+          metadata_json: null,
+          payload_json: "null",
+        },
+        {
+          event_id: "old",
+          last_attempt_at: null,
+          last_error: null,
+          metadata_json: null,
+          payload_json: "null",
+        },
+        {
+          event_id: "retry",
+          last_attempt_at: null,
+          last_error: null,
+          metadata_json: null,
+          payload_json: "null",
+        },
+      ]);
+
+      expect(await queue.prune({ completedTtlMs: 10, failedTtlMs: 10, now: 40 })).toBe(3);
+      expect(await queue.listPending()).toEqual([]);
+      expect(await queue.listClaims()).toEqual([]);
+    });
+  });
+});
--- a/src/channels/message/ingress-queue.ts
+++ b/src/channels/message/ingress-queue.ts
@@ -0,0 +1,829 @@
+import { randomUUID } from "node:crypto";
+import type { DatabaseSync } from "node:sqlite";
+import type { Selectable } from "kysely";
+import {
+  executeSqliteQuerySync,
+  executeSqliteQueryTakeFirstSync,
+  getNodeSqliteKysely,
+} from "../../infra/kysely-sync.js";
+import type {
+  ChannelIngressEvents,
+  DB as OpenClawStateKyselyDatabase,
+} from "../../state/openclaw-state-db.generated.js";
+import {
+  openOpenClawStateDatabase,
+  runOpenClawStateWriteTransaction,
+} from "../../state/openclaw-state-db.js";
+
+export type ChannelIngressQueueRecord<TPayload, TMetadata = unknown> = {
+  id: string;
+  channelId: string;
+  accountId: string;
+  queueName: string;
+  payload: TPayload;
+  metadata?: TMetadata;
+  receivedAt: number;
+  updatedAt: number;
+  laneKey?: string;
+  attempts: number;
+  lastAttemptAt?: number;
+  lastError?: string;
+};
+
+export type ChannelIngressQueueClaim<TPayload, TMetadata = unknown> = ChannelIngressQueueRecord<
+  TPayload,
+  TMetadata
+> & {
+  claim: {
+    token: string;
+    ownerId: string;
+    claimedAt: number;
+  };
+};
+
+export type ChannelIngressQueueClaimRef = {
+  id: string;
+  claim: {
+    token: string;
+  };
+};
+
+export type ChannelIngressQueueCompletedRecord<TCompletedMetadata = unknown> = {
+  id: string;
+  channelId: string;
+  accountId: string;
+  queueName: string;
+  completedAt: number;
+  metadata?: TCompletedMetadata;
+};
+
+export type ChannelIngressQueueFailedRecord = {
+  id: string;
+  channelId: string;
+  accountId: string;
+  queueName: string;
+  failedAt: number;
+  reason: string;
+  message?: string;
+};
+
+export type ChannelIngressQueuePruneOptions = {
+  pendingTtlMs?: number;
+  completedTtlMs?: number;
+  failedTtlMs?: number;
+  pendingMaxEntries?: number;
+  completedMaxEntries?: number;
+  failedMaxEntries?: number;
+  protectIds?: Iterable<string>;
+  now?: number;
+};
+
+export type ChannelIngressQueueEnqueueResult<TPayload, TMetadata, TCompletedMetadata> =
+  | {
+      kind: "accepted";
+      duplicate: false;
+      record: ChannelIngressQueueRecord<TPayload, TMetadata>;
+    }
+  | {
+      kind: "pending";
+      duplicate: true;
+      record: ChannelIngressQueueRecord<TPayload, TMetadata>;
+    }
+  | {
+      kind: "claimed";
+      duplicate: true;
+      record: ChannelIngressQueueClaim<TPayload, TMetadata>;
+    }
+  | {
+      kind: "completed";
+      duplicate: true;
+      record: ChannelIngressQueueCompletedRecord<TCompletedMetadata>;
+    }
+  | {
+      kind: "failed";
+      duplicate: true;
+      record: ChannelIngressQueueFailedRecord;
+    };
+
+export type ChannelIngressQueue<TPayload, TMetadata = unknown, TCompletedMetadata = unknown> = {
+  enqueue(
+    id: string,
+    payload: TPayload,
+    options?: {
+      metadata?: TMetadata;
+      receivedAt?: number;
+      laneKey?: string;
+    },
+  ): Promise<ChannelIngressQueueEnqueueResult<TPayload, TMetadata, TCompletedMetadata>>;
+  listPending(options?: {
+    limit?: number | "all";
+    orderBy?: "received" | "id";
+  }): Promise<Array<ChannelIngressQueueRecord<TPayload, TMetadata>>>;
+  listClaims(): Promise<Array<ChannelIngressQueueClaim<TPayload, TMetadata>>>;
+  claimNext(options?: {
+    ownerId?: string;
+    blockedLaneKeys?: Iterable<string>;
+    staleMs?: number;
+  }): Promise<ChannelIngressQueueClaim<TPayload, TMetadata> | null>;
+  claim(
+    id: string,
+    options?: { ownerId?: string },
+  ): Promise<ChannelIngressQueueClaim<TPayload, TMetadata> | null>;
+  complete(
+    idOrClaim: string | ChannelIngressQueueClaimRef,
+    options?: { metadata?: TCompletedMetadata; completedAt?: number },
+  ): Promise<boolean>;
+  release(
+    idOrClaim: string | ChannelIngressQueueClaimRef,
+    options?: { lastError?: string; releasedAt?: number },
+  ): Promise<boolean>;
+  fail(
+    idOrClaim: string | ChannelIngressQueueClaimRef,
+    options: { reason: string; message?: string; failedAt?: number },
+  ): Promise<boolean>;
+  delete(
+    idOrClaim:
+      | string
+      | ChannelIngressQueueRecord<TPayload, TMetadata>
+      | ChannelIngressQueueClaimRef,
+  ): Promise<boolean>;
+  recoverStaleClaims(options?: {
+    staleMs?: number;
+    now?: number;
+    shouldRecover?: (
+      claim: ChannelIngressQueueClaim<TPayload, TMetadata>,
+    ) => boolean | Promise<boolean>;
+  }): Promise<number>;
+  prune(options?: ChannelIngressQueuePruneOptions): Promise<number>;
+};
+
+export type CreateChannelIngressQueueOptions = {
+  channelId: string;
+  accountId?: string;
+  stateDir?: string;
+  now?: () => number;
+};
+
+type ChannelIngressDatabase = Pick<OpenClawStateKyselyDatabase, "channel_ingress_events">;
+type ChannelIngressRow = Selectable<ChannelIngressEvents>;
+
+function normalizePart(value: string | undefined, fallback: string): string {
+  const normalized = value?.trim();
+  return normalized ? normalized : fallback;
+}
+
+function openStateDatabase(stateDir?: string) {
+  return openOpenClawStateDatabase({
+    env: stateDir ? { ...process.env, OPENCLAW_STATE_DIR: stateDir } : process.env,
+  });
+}
+
+function getChannelIngressKysely(db: DatabaseSync) {
+  return getNodeSqliteKysely<ChannelIngressDatabase>(db);
+}
+
+function affectedRows(result: { numAffectedRows?: bigint }): number {
+  return Number(result.numAffectedRows ?? 0n);
+}
+
+function parseJson<T>(value: string): T {
+  return JSON.parse(value) as T;
+}
+
+function baseRecord<TPayload, TMetadata>(
+  row: ChannelIngressRow,
+): ChannelIngressQueueRecord<TPayload, TMetadata> {
+  return {
+    id: row.event_id,
+    channelId: row.channel_id,
+    accountId: row.account_id,
+    queueName: row.queue_name,
+    payload: parseJson<TPayload>(row.payload_json),
+    ...(row.metadata_json === null ? {} : { metadata: parseJson<TMetadata>(row.metadata_json) }),
+    receivedAt: Number(row.received_at),
+    updatedAt: Number(row.updated_at),
+    ...(row.lane_key === null ? {} : { laneKey: row.lane_key }),
+    attempts: Number(row.attempts),
+    ...(row.last_attempt_at === null ? {} : { lastAttemptAt: Number(row.last_attempt_at) }),
+    ...(row.last_error === null ? {} : { lastError: row.last_error }),
+  };
+}
+
+function claimedRecord<TPayload, TMetadata>(
+  row: ChannelIngressRow,
+): ChannelIngressQueueClaim<TPayload, TMetadata> {
+  return {
+    ...baseRecord<TPayload, TMetadata>(row),
+    claim: {
+      token: row.claim_token ?? "",
+      ownerId: row.claim_owner ?? "",
+      claimedAt: Number(row.claimed_at ?? 0),
+    },
+  };
+}
+
+function completedRecord<TCompletedMetadata>(
+  row: ChannelIngressRow,
+): ChannelIngressQueueCompletedRecord<TCompletedMetadata> {
+  return {
+    id: row.event_id,
+    channelId: row.channel_id,
+    accountId: row.account_id,
+    queueName: row.queue_name,
+    completedAt: Number(row.completed_at ?? row.updated_at),
+    ...(row.completed_metadata_json === null
+      ? {}
+      : { metadata: parseJson<TCompletedMetadata>(row.completed_metadata_json) }),
+  };
+}
+
+function failedRecord(row: ChannelIngressRow): ChannelIngressQueueFailedRecord {
+  return {
+    id: row.event_id,
+    channelId: row.channel_id,
+    accountId: row.account_id,
+    queueName: row.queue_name,
+    failedAt: Number(row.failed_at ?? row.updated_at),
+    reason: row.failed_reason ?? "failed",
+    ...(row.last_error === null ? {} : { message: row.last_error }),
+  };
+}
+
+function selectRow(db: DatabaseSync, queueName: string, id: string) {
+  const kysely = getChannelIngressKysely(db);
+  return executeSqliteQueryTakeFirstSync(
+    db,
+    kysely
+      .selectFrom("channel_ingress_events")
+      .selectAll()
+      .where("queue_name", "=", queueName)
+      .where("event_id", "=", id),
+  );
+}
+
+function idFrom(idOrRecord: string | { id: string }): string {
+  const id = normalizePart(typeof idOrRecord === "string" ? idOrRecord : idOrRecord.id, "");
+  if (!id) {
+    throw new Error("Channel ingress event id cannot be empty");
+  }
+  return id;
+}
+
+function claimTokenFrom(
+  idOrClaim: string | { id: string; claim?: { token: string } },
+): string | null {
+  return typeof idOrClaim === "string" ? null : (idOrClaim.claim?.token ?? null);
+}
+
+function rowToEnqueueResult<TPayload, TMetadata, TCompletedMetadata>(
+  row: ChannelIngressRow,
+): ChannelIngressQueueEnqueueResult<TPayload, TMetadata, TCompletedMetadata> {
+  if (row.status === "completed") {
+    return { kind: "completed", duplicate: true, record: completedRecord(row) };
+  }
+  if (row.status === "failed") {
+    return { kind: "failed", duplicate: true, record: failedRecord(row) };
+  }
+  if (row.status === "claimed") {
+    return { kind: "claimed", duplicate: true, record: claimedRecord(row) };
+  }
+  return { kind: "pending", duplicate: true, record: baseRecord(row) };
+}
+
+function normalizeLimit(limit: number | "all" | undefined): number {
+  return limit === "all" ? Number.MAX_SAFE_INTEGER : Math.max(1, Math.floor(limit ?? 100));
+}
+
+function normalizeMaxEntries(value: number | undefined): number | null {
+  return value === undefined ? null : Math.max(0, Math.floor(value));
+}
+
+function normalizedProtectedIds(ids: Iterable<string> | undefined): string[] {
+  return [...(ids ?? [])].map((id) => id.trim()).filter(Boolean);
+}
+
+function queueNameForParts(channelId: string, accountId: string): string {
+  return JSON.stringify([channelId, accountId]);
+}
+
+export function createChannelIngressQueue<
+  TPayload,
+  TMetadata = unknown,
+  TCompletedMetadata = unknown,
+>(
+  options: CreateChannelIngressQueueOptions,
+): ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata> {
+  const channelId = normalizePart(options.channelId, "unknown");
+  const accountId = normalizePart(options.accountId, "default");
+  const queueName = queueNameForParts(channelId, accountId);
+  const now = options.now ?? Date.now;
+
+  const enqueue: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["enqueue"] = async (
+    id,
+    payload,
+    enqueueOptions,
+  ) => {
+    const eventId = normalizePart(id, "");
+    if (!eventId) {
+      throw new Error("Channel ingress event id cannot be empty");
+    }
+    const receivedAt = enqueueOptions?.receivedAt ?? now();
+    const updatedAt = now();
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const insert = executeSqliteQuerySync(
+          tx.db,
+          kysely
+            .insertInto("channel_ingress_events")
+            .values({
+              queue_name: queueName,
+              event_id: eventId,
+              channel_id: channelId,
+              account_id: accountId,
+              status: "pending",
+              lane_key: enqueueOptions?.laneKey ?? null,
+              payload_json: JSON.stringify(payload),
+              metadata_json:
+                enqueueOptions?.metadata === undefined
+                  ? null
+                  : JSON.stringify(enqueueOptions.metadata),
+              received_at: receivedAt,
+              updated_at: updatedAt,
+              attempts: 0,
+            })
+            .onConflict((conflict) => conflict.columns(["queue_name", "event_id"]).doNothing()),
+        );
+        const row = selectRow(tx.db, queueName, eventId);
+        if (!row) {
+          throw new Error(`Failed to read channel ingress event ${queueName}/${eventId}`);
+        }
+        if (affectedRows(insert) > 0) {
+          return {
+            kind: "accepted",
+            duplicate: false,
+            record: baseRecord<TPayload, TMetadata>(row),
+          };
+        }
+        return rowToEnqueueResult<TPayload, TMetadata, TCompletedMetadata>(row);
+      },
+      { path: database.path },
+    );
+  };
+
+  const listPending: ChannelIngressQueue<
+    TPayload,
+    TMetadata,
+    TCompletedMetadata
+  >["listPending"] = async (listOptions) => {
+    const { db } = openStateDatabase(options.stateDir);
+    const kysely = getChannelIngressKysely(db);
+    const baseQuery = kysely
+      .selectFrom("channel_ingress_events")
+      .selectAll()
+      .where("queue_name", "=", queueName)
+      .where("status", "=", "pending")
+      .limit(normalizeLimit(listOptions?.limit));
+    const query =
+      listOptions?.orderBy === "id"
+        ? baseQuery.orderBy("event_id", "asc")
+        : baseQuery.orderBy("received_at", "asc").orderBy("event_id", "asc");
+    const rows = executeSqliteQuerySync(db, query).rows;
+    return rows.map((row) => baseRecord<TPayload, TMetadata>(row));
+  };
+
+  const listClaims: ChannelIngressQueue<
+    TPayload,
+    TMetadata,
+    TCompletedMetadata
+  >["listClaims"] = async () => {
+    const { db } = openStateDatabase(options.stateDir);
+    const kysely = getChannelIngressKysely(db);
+    const rows = executeSqliteQuerySync(
+      db,
+      kysely
+        .selectFrom("channel_ingress_events")
+        .selectAll()
+        .where("queue_name", "=", queueName)
+        .where("status", "=", "claimed")
+        .orderBy("claimed_at", "asc")
+        .orderBy("received_at", "asc")
+        .orderBy("event_id", "asc"),
+    ).rows;
+    return rows.map((row) => claimedRecord<TPayload, TMetadata>(row));
+  };
+
+  const recoverStaleClaims: ChannelIngressQueue<
+    TPayload,
+    TMetadata,
+    TCompletedMetadata
+  >["recoverStaleClaims"] = async (recoverOptions) => {
+    const staleMs = Math.max(0, Math.floor(recoverOptions?.staleMs ?? 0));
+    const cutoff = (recoverOptions?.now ?? now()) - staleMs;
+    const claims = (await listClaims()).filter((claim) => claim.claim.claimedAt <= cutoff);
+    let recovered = 0;
+    for (const claim of claims) {
+      if (recoverOptions?.shouldRecover && !(await recoverOptions.shouldRecover(claim))) {
+        continue;
+      }
+      if (await release(claim, { releasedAt: recoverOptions?.now ?? now() })) {
+        recovered += 1;
+      }
+    }
+    return recovered;
+  };
+
+  const claimNext: ChannelIngressQueue<
+    TPayload,
+    TMetadata,
+    TCompletedMetadata
+  >["claimNext"] = async (claimOptions) => {
+    if (claimOptions?.staleMs !== undefined) {
+      await recoverStaleClaims({ staleMs: claimOptions.staleMs });
+    }
+    const blocked = new Set(
+      [...(claimOptions?.blockedLaneKeys ?? [])].map((key) => key.trim()).filter(Boolean),
+    );
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const baseSelect = kysely
+          .selectFrom("channel_ingress_events")
+          .select(["event_id", "lane_key"])
+          .where("queue_name", "=", queueName)
+          .where("status", "=", "pending");
+        const select =
+          blocked.size === 0
+            ? baseSelect
+            : baseSelect.where((eb) =>
+                eb.or([eb("lane_key", "is", null), eb("lane_key", "not in", [...blocked])]),
+              );
+        const selected = executeSqliteQueryTakeFirstSync(
+          tx.db,
+          select.orderBy("received_at", "asc").orderBy("event_id", "asc").limit(1),
+        );
+        if (!selected) {
+          return null;
+        }
+        const token = randomUUID();
+        const claimedAt = now();
+        const ownerId = normalizePart(claimOptions?.ownerId, `${process.pid}`);
+        const result = executeSqliteQuerySync(
+          tx.db,
+          kysely
+            .updateTable("channel_ingress_events")
+            .set({
+              status: "claimed",
+              claim_token: token,
+              claim_owner: ownerId,
+              claimed_at: claimedAt,
+              updated_at: claimedAt,
+            })
+            .where("queue_name", "=", queueName)
+            .where("event_id", "=", selected.event_id)
+            .where("status", "=", "pending"),
+        );
+        if (affectedRows(result) === 0) {
+          return null;
+        }
+        const row = selectRow(tx.db, queueName, selected.event_id);
+        return row ? claimedRecord<TPayload, TMetadata>(row) : null;
+      },
+      { path: database.path },
+    );
+  };
+
+  const claim: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["claim"] = async (
+    id,
+    claimOptions,
+  ) => {
+    const eventId = normalizePart(id, "");
+    if (!eventId) {
+      throw new Error("Channel ingress event id cannot be empty");
+    }
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const token = randomUUID();
+        const claimedAt = now();
+        const ownerId = normalizePart(claimOptions?.ownerId, `${process.pid}`);
+        const result = executeSqliteQuerySync(
+          tx.db,
+          kysely
+            .updateTable("channel_ingress_events")
+            .set({
+              status: "claimed",
+              claim_token: token,
+              claim_owner: ownerId,
+              claimed_at: claimedAt,
+              updated_at: claimedAt,
+            })
+            .where("queue_name", "=", queueName)
+            .where("event_id", "=", eventId)
+            .where("status", "=", "pending"),
+        );
+        if (affectedRows(result) === 0) {
+          return null;
+        }
+        const row = selectRow(tx.db, queueName, eventId);
+        return row ? claimedRecord<TPayload, TMetadata>(row) : null;
+      },
+      { path: database.path },
+    );
+  };
+
+  const complete: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["complete"] = async (
+    idOrClaim,
+    completeOptions,
+  ) => {
+    const eventId = idFrom(idOrClaim);
+    const token = claimTokenFrom(idOrClaim);
+    const completedAt = completeOptions?.completedAt ?? now();
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const baseUpdate = kysely
+          .updateTable("channel_ingress_events")
+          .set({
+            status: "completed",
+            completed_at: completedAt,
+            completed_metadata_json:
+              completeOptions?.metadata === undefined
+                ? null
+                : JSON.stringify(completeOptions.metadata),
+            payload_json: "null",
+            metadata_json: null,
+            claim_token: null,
+            claim_owner: null,
+            claimed_at: null,
+            last_attempt_at: null,
+            last_error: null,
+            updated_at: completedAt,
+          })
+          .where("queue_name", "=", queueName)
+          .where("event_id", "=", eventId);
+        const update =
+          token === null
+            ? baseUpdate.where("status", "=", "pending")
+            : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token);
+        const result = executeSqliteQuerySync(tx.db, update);
+        if (affectedRows(result) > 0) {
+          return true;
+        }
+        if (token !== null) {
+          return false;
+        }
+        const insert = executeSqliteQuerySync(
+          tx.db,
+          kysely
+            .insertInto("channel_ingress_events")
+            .values({
+              queue_name: queueName,
+              event_id: eventId,
+              channel_id: channelId,
+              account_id: accountId,
+              status: "completed",
+              lane_key: null,
+              payload_json: "null",
+              metadata_json: null,
+              received_at: completedAt,
+              updated_at: completedAt,
+              attempts: 0,
+              completed_at: completedAt,
+              completed_metadata_json:
+                completeOptions?.metadata === undefined
+                  ? null
+                  : JSON.stringify(completeOptions.metadata),
+            })
+            .onConflict((conflict) => conflict.columns(["queue_name", "event_id"]).doNothing()),
+        );
+        return affectedRows(insert) > 0;
+      },
+      { path: database.path },
+    );
+  };
+
+  const release: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["release"] = async (
+    idOrClaim,
+    releaseOptions,
+  ) => {
+    const eventId = idFrom(idOrClaim);
+    const token = claimTokenFrom(idOrClaim);
+    const releasedAt = releaseOptions?.releasedAt ?? now();
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const baseUpdate = kysely
+          .updateTable("channel_ingress_events")
+          .set((eb) => ({
+            status: "pending",
+            claim_token: null,
+            claim_owner: null,
+            claimed_at: null,
+            attempts: eb("attempts", "+", 1),
+            last_attempt_at: releasedAt,
+            ...(releaseOptions?.lastError === undefined
+              ? {}
+              : { last_error: releaseOptions.lastError }),
+            updated_at: releasedAt,
+          }))
+          .where("queue_name", "=", queueName)
+          .where("event_id", "=", eventId);
+        const update =
+          token === null
+            ? baseUpdate.where("status", "=", "pending")
+            : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token);
+        return affectedRows(executeSqliteQuerySync(tx.db, update)) > 0;
+      },
+      { path: database.path },
+    );
+  };
+
+  const fail: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["fail"] = async (
+    idOrClaim,
+    failOptions,
+  ) => {
+    const eventId = idFrom(idOrClaim);
+    const token = claimTokenFrom(idOrClaim);
+    const failedAt = failOptions.failedAt ?? now();
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const baseUpdate = kysely
+          .updateTable("channel_ingress_events")
+          .set({
+            status: "failed",
+            failed_at: failedAt,
+            failed_reason: failOptions.reason,
+            last_error: failOptions.message ?? null,
+            payload_json: "null",
+            metadata_json: null,
+            claim_token: null,
+            claim_owner: null,
+            claimed_at: null,
+            updated_at: failedAt,
+          })
+          .where("queue_name", "=", queueName)
+          .where("event_id", "=", eventId);
+        const update =
+          token === null
+            ? baseUpdate.where("status", "=", "pending")
+            : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token);
+        return affectedRows(executeSqliteQuerySync(tx.db, update)) > 0;
+      },
+      { path: database.path },
+    );
+  };
+
+  const deleteEntry: ChannelIngressQueue<
+    TPayload,
+    TMetadata,
+    TCompletedMetadata
+  >["delete"] = async (idOrRecord) => {
+    const eventId = idFrom(idOrRecord);
+    const token = claimTokenFrom(idOrRecord);
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        const baseDelete = kysely
+          .deleteFrom("channel_ingress_events")
+          .where("queue_name", "=", queueName)
+          .where("event_id", "=", eventId);
+        const deleteQuery =
+          token === null
+            ? baseDelete.where("status", "=", "pending")
+            : baseDelete.where("status", "=", "claimed").where("claim_token", "=", token);
+        return affectedRows(executeSqliteQuerySync(tx.db, deleteQuery)) > 0;
+      },
+      { path: database.path },
+    );
+  };
+
+  const prune: ChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>["prune"] = async (
+    pruneOptions,
+  ) => {
+    const current = pruneOptions?.now ?? now();
+    const pendingCutoff =
+      pruneOptions?.pendingTtlMs === undefined ? null : current - pruneOptions.pendingTtlMs;
+    const completedCutoff =
+      pruneOptions?.completedTtlMs === undefined ? null : current - pruneOptions.completedTtlMs;
+    const failedCutoff =
+      pruneOptions?.failedTtlMs === undefined ? null : current - pruneOptions.failedTtlMs;
+    const pendingMaxEntries = normalizeMaxEntries(pruneOptions?.pendingMaxEntries);
+    const completedMaxEntries = normalizeMaxEntries(pruneOptions?.completedMaxEntries);
+    const failedMaxEntries = normalizeMaxEntries(pruneOptions?.failedMaxEntries);
+    const protectIds = normalizedProtectedIds(pruneOptions?.protectIds);
+    if (
+      pendingCutoff === null &&
+      completedCutoff === null &&
+      failedCutoff === null &&
+      pendingMaxEntries === null &&
+      completedMaxEntries === null &&
+      failedMaxEntries === null
+    ) {
+      return 0;
+    }
+    const database = openStateDatabase(options.stateDir);
+    return runOpenClawStateWriteTransaction(
+      (tx) => {
+        const kysely = getChannelIngressKysely(tx.db);
+        let deleted = 0;
+        if (pendingCutoff !== null) {
+          let deleteQuery = kysely
+            .deleteFrom("channel_ingress_events")
+            .where("queue_name", "=", queueName)
+            .where("status", "=", "pending")
+            .where("updated_at", "<", pendingCutoff);
+          if (protectIds.length > 0) {
+            deleteQuery = deleteQuery.where("event_id", "not in", protectIds);
+          }
+          deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery));
+        }
+        if (completedCutoff !== null) {
+          let deleteQuery = kysely
+            .deleteFrom("channel_ingress_events")
+            .where("queue_name", "=", queueName)
+            .where("status", "=", "completed")
+            .where("completed_at", "<", completedCutoff);
+          if (protectIds.length > 0) {
+            deleteQuery = deleteQuery.where("event_id", "not in", protectIds);
+          }
+          deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery));
+        }
+        if (failedCutoff !== null) {
+          let deleteQuery = kysely
+            .deleteFrom("channel_ingress_events")
+            .where("queue_name", "=", queueName)
+            .where("status", "=", "failed")
+            .where("failed_at", "<", failedCutoff);
+          if (protectIds.length > 0) {
+            deleteQuery = deleteQuery.where("event_id", "not in", protectIds);
+          }
+          deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery));
+        }
+        const pruneMaxEntries = (status: string, maxEntries: number | null) => {
+          if (maxEntries === null) {
+            return;
+          }
+          const batchSize = 500;
+          const protectedSet = new Set(protectIds);
+          while (true) {
+            const rowsToDelete = executeSqliteQuerySync(
+              tx.db,
+              kysely
+                .selectFrom("channel_ingress_events")
+                .select("event_id")
+                .where("queue_name", "=", queueName)
+                .where("status", "=", status)
+                .orderBy("updated_at", "desc")
+                .orderBy("event_id", "desc")
+                .limit(maxEntries + batchSize),
+            ).rows.slice(maxEntries);
+            const ids = rowsToDelete
+              .map((row) => row.event_id)
+              .filter((id) => !protectedSet.has(id));
+            if (ids.length === 0) {
+              return;
+            }
+            deleted += affectedRows(
+              executeSqliteQuerySync(
+                tx.db,
+                kysely
+                  .deleteFrom("channel_ingress_events")
+                  .where("queue_name", "=", queueName)
+                  .where("status", "=", status)
+                  .where("event_id", "in", ids),
+              ),
+            );
+          }
+        };
+        pruneMaxEntries("pending", pendingMaxEntries);
+        pruneMaxEntries("completed", completedMaxEntries);
+        pruneMaxEntries("failed", failedMaxEntries);
+        return deleted;
+      },
+      { path: database.path },
+    );
+  };
+
+  return {
+    enqueue,
+    listPending,
+    listClaims,
+    claimNext,
+    claim,
+    complete,
+    release,
+    fail,
+    delete: deleteEntry,
+    recoverStaleClaims,
+    prune,
+  };
+}
--- a/src/plugin-sdk/channel-outbound.ts
+++ b/src/plugin-sdk/channel-outbound.ts
@@ -78,6 +78,7 @@ export {
  classifyDurableSendRecoveryState,
  createChannelMessageAdapterFromOutbound,
  createDurableInboundReceiveJournal,
+  createDurableInboundReceiveJournalFromQueue,
  createMessageReceiptFromOutboundResults,
  listMessageReceiptPlatformIds,
  createMessageReceiveContext,
@@ -134,6 +135,14 @@ export type {
  ChannelMessageUnknownSendReconciliationResult,
  CreateChannelReplyPipelineParams,
  CreateChannelMessageAdapterFromOutboundParams,
+  ChannelIngressQueue,
+  ChannelIngressQueueClaim,
+  ChannelIngressQueueClaimRef,
+  ChannelIngressQueueCompletedRecord,
+  ChannelIngressQueueEnqueueResult,
+  ChannelIngressQueueFailedRecord,
+  ChannelIngressQueuePruneOptions,
+  ChannelIngressQueueRecord,
  DeriveDurableFinalDeliveryRequirementsParams,
  ChannelMessageLiveCapabilityProof,
  ChannelMessageLiveCapabilityProofMap,
@@ -155,6 +164,7 @@ export type {
  DurableInboundReceiveJournal,
  DurableInboundReceiveJournalOptions,
  DurableInboundReceivePendingRecord,
+  DurableInboundReceiveQueueJournalOptions,
  DurableInboundReceiveReleaseOptions,
  DurableMessageSendIntent,
  DurableMessageSendState,
--- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
+++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
@@ -767,6 +767,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> =
      openSyncKeyedStore: vi.fn(() => {
        throw new Error("openSyncKeyedStore mock is not configured");
      }) as unknown as PluginRuntime["state"]["openSyncKeyedStore"],
+      openChannelIngressQueue: vi.fn(() => {
+        throw new Error("openChannelIngressQueue mock is not configured");
+      }) as unknown as PluginRuntime["state"]["openChannelIngressQueue"],
    },
    tasks: {
      runs: {
--- a/src/plugins/registry.ts
+++ b/src/plugins/registry.ts
@@ -15,6 +15,7 @@ import {
 } from "../agents/harness/registry.js";
 import type { AgentHarness } from "../agents/harness/types.js";
 import type { AnyAgentTool } from "../agents/tools/common.js";
+import { createChannelIngressQueue } from "../channels/message/ingress-queue.js";
 import type { ChannelPlugin } from "../channels/plugins/types.plugin.js";
 import {
  normalizeCommandDescriptorName,
@@ -2629,6 +2630,17 @@ export function createPluginRegistry(registryParams: PluginRegistryParams) {
              assertPluginStateAllowed();
              return createPluginStateSyncKeyedStore<T>(pluginId, options);
            },
+            openChannelIngressQueue: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
+              options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
+            ) => {
+              assertPluginStateAllowed();
+              const stateDir = options?.stateDir ?? baseState.resolveStateDir();
+              return createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>({
+                ...(options ?? {}),
+                channelId: pluginId,
+                stateDir,
+              });
+            },
          } satisfies PluginRuntime["state"];
        }
        if (prop === "config") {
--- a/src/plugins/runtime/index.ts
+++ b/src/plugins/runtime/index.ts
@@ -256,6 +256,11 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}):
      openSyncKeyedStore: () => {
        throw new Error("openSyncKeyedStore is only available through the plugin runtime proxy.");
      },
+      openChannelIngressQueue: () => {
+        throw new Error(
+          "openChannelIngressQueue is only available through the plugin runtime proxy.",
+        );
+      },
    },
    tasks,
    taskFlow,
--- a/src/plugins/runtime/types-core.ts
+++ b/src/plugins/runtime/types-core.ts
@@ -314,6 +314,16 @@ export type PluginRuntimeCore = {
    openSyncKeyedStore: <T>(
      options: import("../../plugin-state/plugin-state-store.types.js").OpenKeyedStoreOptions,
    ) => import("../../plugin-state/plugin-state-store.types.js").PluginStateSyncKeyedStore<T>;
+    openChannelIngressQueue: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
+      options?: Omit<
+        import("../../channels/message/ingress-queue.js").CreateChannelIngressQueueOptions,
+        "channelId"
+      >,
+    ) => import("../../channels/message/ingress-queue.js").ChannelIngressQueue<
+      TPayload,
+      TMetadata,
+      TCompletedMetadata
+    >;
  };
  tasks: {
    runs: PluginRuntimeTaskRuns;
--- a/src/state/openclaw-state-db.generated.d.ts
+++ b/src/state/openclaw-state-db.generated.d.ts
@@ -144,6 +144,29 @@ export interface CaptureSessions {
  started_at: number;
 }

+export interface ChannelIngressEvents {
+  account_id: string;
+  attempts: Generated<number>;
+  channel_id: string;
+  claim_owner: string | null;
+  claim_token: string | null;
+  claimed_at: number | null;
+  completed_at: number | null;
+  completed_metadata_json: string | null;
+  event_id: string;
+  failed_at: number | null;
+  failed_reason: string | null;
+  lane_key: string | null;
+  last_attempt_at: number | null;
+  last_error: string | null;
+  metadata_json: string | null;
+  payload_json: string;
+  queue_name: string;
+  received_at: number;
+  status: string;
+  updated_at: number;
+}
+
 export interface ChannelPairingAllowEntries {
  account_id: string;
  channel_key: string;
@@ -931,6 +954,7 @@ export interface DB {
  capture_blobs: CaptureBlobs;
  capture_events: CaptureEvents;
  capture_sessions: CaptureSessions;
+  channel_ingress_events: ChannelIngressEvents;
  channel_pairing_allow_entries: ChannelPairingAllowEntries;
  channel_pairing_requests: ChannelPairingRequests;
  command_log_entries: CommandLogEntries;
--- a/src/state/openclaw-state-schema.generated.ts
+++ b/src/state/openclaw-state-schema.generated.ts
@@ -607,6 +607,39 @@ CREATE INDEX IF NOT EXISTS idx_plugin_state_expiry
 CREATE INDEX IF NOT EXISTS idx_plugin_state_listing
  ON plugin_state_entries(plugin_id, namespace, created_at, entry_key);

+CREATE TABLE IF NOT EXISTS channel_ingress_events (
+  queue_name TEXT NOT NULL,
+  event_id TEXT NOT NULL,
+  channel_id TEXT NOT NULL,
+  account_id TEXT NOT NULL,
+  status TEXT NOT NULL,
+  lane_key TEXT,
+  payload_json TEXT NOT NULL,
+  metadata_json TEXT,
+  received_at INTEGER NOT NULL,
+  updated_at INTEGER NOT NULL,
+  claim_token TEXT,
+  claim_owner TEXT,
+  claimed_at INTEGER,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at INTEGER,
+  last_error TEXT,
+  failed_reason TEXT,
+  failed_at INTEGER,
+  completed_at INTEGER,
+  completed_metadata_json TEXT,
+  PRIMARY KEY (queue_name, event_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_pending
+  ON channel_ingress_events(queue_name, status, received_at, event_id);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_claims
+  ON channel_ingress_events(queue_name, status, claimed_at);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_lane
+  ON channel_ingress_events(queue_name, status, lane_key);
+
 CREATE TABLE IF NOT EXISTS plugin_blob_entries (
  plugin_id TEXT NOT NULL,
  namespace TEXT NOT NULL,
--- a/src/state/openclaw-state-schema.sql
+++ b/src/state/openclaw-state-schema.sql
@@ -602,6 +602,39 @@ CREATE INDEX IF NOT EXISTS idx_plugin_state_expiry
 CREATE INDEX IF NOT EXISTS idx_plugin_state_listing
  ON plugin_state_entries(plugin_id, namespace, created_at, entry_key);

+CREATE TABLE IF NOT EXISTS channel_ingress_events (
+  queue_name TEXT NOT NULL,
+  event_id TEXT NOT NULL,
+  channel_id TEXT NOT NULL,
+  account_id TEXT NOT NULL,
+  status TEXT NOT NULL,
+  lane_key TEXT,
+  payload_json TEXT NOT NULL,
+  metadata_json TEXT,
+  received_at INTEGER NOT NULL,
+  updated_at INTEGER NOT NULL,
+  claim_token TEXT,
+  claim_owner TEXT,
+  claimed_at INTEGER,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at INTEGER,
+  last_error TEXT,
+  failed_reason TEXT,
+  failed_at INTEGER,
+  completed_at INTEGER,
+  completed_metadata_json TEXT,
+  PRIMARY KEY (queue_name, event_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_pending
+  ON channel_ingress_events(queue_name, status, received_at, event_id);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_claims
+  ON channel_ingress_events(queue_name, status, claimed_at);
+
+CREATE INDEX IF NOT EXISTS idx_channel_ingress_lane
+  ON channel_ingress_events(queue_name, status, lane_key);
+
 CREATE TABLE IF NOT EXISTS plugin_blob_entries (
  plugin_id TEXT NOT NULL,
  namespace TEXT NOT NULL,