feat(zalouser): recover accepted messages after local crashes (#110916)

* feat(zalouser): add durable ingress recovery

* fix(zalouser): settle aborted ingress claims

* fix(zalouser): preserve concurrent ingress lanes

* docs: refresh generated channel map

* refactor(zalouser): simplify ingress abort listener

docs/channels/zalouser.md

@@ -69,6 +69,13 @@ openclaw directory groups list --channel zalouser --query "work"
 
 - Outbound text is chunked to 2000 characters (Zalo client limit).
 - Streaming is not supported.
+- Completed inbound message ids are retained for 30 days, bounded to the 1000 most recent entries per account.
+
+## Inbound durability
+
+OpenClaw stores each raw `zca-js` message callback before processing it. Pending messages resume from the account queue after a Gateway restart, and processing stays serialized per direct chat or group.
+
+The `zca-js` socket listener does not expose a delivery acknowledgement or automatically replay old messages after reconnect. The durable queue therefore protects the local crash window after a callback reaches OpenClaw; it cannot recover a message the socket never delivered. Replay tombstones are mostly a safeguard against a repeated callback with the same Zalo message id.
 
 ## Access control (DMs)
 

docs/docs_map.md

@@ -1143,6 +1143,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
   - H2: Naming
   - H2: Finding IDs (directory)
   - H2: Limits
+  - H2: Inbound durability
   - H2: Access control (DMs)
   - H2: Group access (optional)
   - H3: Group mention gating

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

@@ -0,0 +1,104 @@
+// Zalouser tests share isolated durable-ingress state and raw zca-js envelopes.
+import fs from "node:fs/promises";
+import path from "node:path";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
+import { expect, vi } from "vitest";
+import type { createZalouserIngressMonitor } from "./ingress.js";
+import type { ZaloInboundMessage } from "./types.js";
+import type { Message } from "./zca-client.js";
+import { ThreadType } from "./zca-constants.js";
+
+type CreateZalouserIngressMonitor = typeof createZalouserIngressMonitor;
+type ZalouserTestQueue = NonNullable<Parameters<CreateZalouserIngressMonitor>[0]["queue"]>;
+export type ZalouserTestIngressPayload = Parameters<ZalouserTestQueue["enqueue"]>[1];
+
+export function createRawZalouserMessage(params?: {
+  msgId?: string;
+  cliMsgId?: string;
+  senderId?: string;
+  threadId?: string;
+  content?: string;
+  timestamp?: string;
+  isGroup?: boolean;
+}): Message {
+  const isGroup = params?.isGroup ?? false;
+  const senderId = params?.senderId ?? "sender-1";
+  const threadId = params?.threadId ?? (isGroup ? "group-1" : senderId);
+  return {
+    type: isGroup ? ThreadType.Group : ThreadType.User,
+    threadId,
+    isSelf: false,
+    data: {
+      msgId: params?.msgId ?? "message-1",
+      cliMsgId: params?.cliMsgId ?? "client-1",
+      uidFrom: senderId,
+      idTo: isGroup ? threadId : "owner-1",
+      dName: "Test Sender",
+      content: params?.content ?? "hello",
+      ts: params?.timestamp ?? "1764000000000",
+    },
+  };
+}
+
+export function createRawZalouserMessageFromNormalized(message: ZaloInboundMessage): Message {
+  const raw = createRawZalouserMessage({
+    msgId: message.msgId,
+    cliMsgId: message.cliMsgId,
+    senderId: message.senderId,
+    threadId: message.threadId,
+    content: message.content,
+    timestamp: String(message.timestampMs),
+    isGroup: message.isGroup,
+  });
+  raw.data.testNormalizedMessage = message;
+  return raw;
+}
+
+export async function withZalouserIngressTestQueue<T>(
+  fn: (queue: ZalouserTestQueue) => Promise<T>,
+): Promise<T> {
+  const createdDir = await fs.mkdtemp(
+    path.join(resolvePreferredOpenClawTmpDir(), "openclaw-zalouser-ingress-"),
+  );
+  const stateDir = await fs.realpath(createdDir);
+  const previousStateDir = process.env.OPENCLAW_STATE_DIR;
+  process.env.OPENCLAW_STATE_DIR = stateDir;
+  const queue = createChannelIngressQueueForTests<ZalouserTestIngressPayload>({
+    channelId: "zalouser",
+    accountId: "default",
+    stateDir,
+  });
+  try {
+    return await fn(queue);
+  } finally {
+    if (previousStateDir === undefined) {
+      delete process.env.OPENCLAW_STATE_DIR;
+    } else {
+      process.env.OPENCLAW_STATE_DIR = previousStateDir;
+    }
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
+export async function waitForZalouserIngressVerdict(
+  queue: ZalouserTestQueue,
+  eventId: string,
+  expected: "completed" | "failed",
+): Promise<void> {
+  await vi.waitFor(
+    async () => {
+      const verdict = await queue.enqueue(eventId, {
+        version: 1,
+        receivedAt: 0,
+        rawMessage: "{}",
+      });
+      expect(verdict.kind).toBe(expected);
+    },
+    { timeout: 5_000 },
+  );
+}

extensions/zalouser/src/ingress.test.ts

@@ -0,0 +1,478 @@
+// Zalouser tests cover durable socket admission, recovery, and replay semantics.
+import type { ChannelIngressQueue } from "openclaw/plugin-sdk/channel-outbound";
+import { closeOpenClawStateDatabaseForTest } from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createZalouserIngressMonitor, type ZalouserIngressLifecycle } from "./ingress.js";
+import {
+  createRawZalouserMessage,
+  waitForZalouserIngressVerdict,
+  withZalouserIngressTestQueue,
+  type ZalouserTestIngressPayload,
+} from "./ingress.test-support.js";
+
+function runtime() {
+  return { error: vi.fn(), log: vi.fn() };
+}
+
+afterEach(() => {
+  closeOpenClawStateDatabaseForTest();
+  vi.restoreAllMocks();
+});
+
+describe("Zalouser durable ingress", () => {
+  it("finishes the durable append before dispatching", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let releaseAppend = () => {};
+      const appendGate = new Promise<void>((resolve) => {
+        releaseAppend = resolve;
+      });
+      const realEnqueue = queue.enqueue.bind(queue);
+      const enqueue: typeof queue.enqueue = async (...args) => {
+        await appendGate;
+        return await realEnqueue(...args);
+      };
+      const gatedQueue: ChannelIngressQueue<ZalouserTestIngressPayload> = {
+        ...queue,
+        enqueue,
+      };
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue: gatedQueue,
+        dispatch,
+      });
+
+      const admission = ingress.receive(createRawZalouserMessage({ msgId: "durable-first" }));
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 20);
+      });
+      expect(dispatch).not.toHaveBeenCalled();
+
+      releaseAppend();
+      await admission;
+      await waitForZalouserIngressVerdict(queue, "durable-first", "completed");
+      expect(dispatch).toHaveBeenCalledOnce();
+      await ingress.stop();
+    });
+  });
+
+  it("recovers a pending event with a fresh drain and dispatches exactly once", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      const rawMessage = createRawZalouserMessage({ msgId: "restart" });
+      await queue.enqueue(
+        "restart",
+        {
+          version: 1,
+          receivedAt: 1,
+          rawMessage: JSON.stringify(rawMessage),
+        },
+        { receivedAt: 1, laneKey: "direct:sender-1" },
+      );
+
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const recovered = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      try {
+        await waitForZalouserIngressVerdict(queue, "restart", "completed");
+        expect(dispatch).toHaveBeenCalledOnce();
+      } finally {
+        await recovered.stop();
+      }
+    });
+  });
+
+  it("rejects a post-completion duplicate by platform msgId", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        await lifecycle.onAdopted();
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      try {
+        await ingress.receive(
+          createRawZalouserMessage({ msgId: "duplicate", content: "original" }),
+        );
+        await waitForZalouserIngressVerdict(queue, "duplicate", "completed");
+        await ingress.receive(
+          createRawZalouserMessage({ msgId: "duplicate", content: "changed redelivery" }),
+        );
+        await ingress.waitForIdle();
+        expect(dispatch).toHaveBeenCalledOnce();
+        expect(dispatch.mock.calls[0]?.[0].content).toBe("original");
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("stores the raw callback envelope and derives a conversation lane before normalization", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      const enqueues: Parameters<typeof queue.enqueue>[] = [];
+      const realEnqueue = queue.enqueue.bind(queue);
+      const observedQueue: ChannelIngressQueue<ZalouserTestIngressPayload> = {
+        ...queue,
+        enqueue: async (...args) => {
+          enqueues.push(args);
+          return await realEnqueue(...args);
+        },
+      };
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue: observedQueue,
+        dispatch: async (_message, lifecycle) => {
+          await lifecycle.onAdopted();
+        },
+      });
+      const raw = createRawZalouserMessage({
+        msgId: "raw",
+        threadId: "group-42",
+        content: "before",
+        isGroup: true,
+      });
+      await ingress.receive(raw);
+      raw.data.content = "after";
+
+      expect(enqueues[0]?.[0]).toBe("raw");
+      expect(enqueues[0]?.[1]).toMatchObject({
+        version: 1,
+        rawMessage: expect.stringContaining('"content":"before"'),
+      });
+      expect(enqueues[0]?.[2]).toMatchObject({ laneKey: "group:group-42" });
+      await ingress.stop();
+    });
+  });
+
+  it("serializes same-conversation claims until adoption", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let firstLifecycle: ZalouserIngressLifecycle | undefined;
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        if (!firstLifecycle) {
+          firstLifecycle = lifecycle;
+          lifecycle.onDeferred();
+          return;
+        }
+        await lifecycle.onAdopted();
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+        pollIntervalMs: 60_000,
+      });
+      await ingress.receive(createRawZalouserMessage({ msgId: "lane-1" }));
+      await ingress.receive(createRawZalouserMessage({ msgId: "lane-2" }));
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(1));
+
+      await firstLifecycle?.onAdopted();
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(2));
+      await ingress.stop();
+    });
+  });
+
+  it("dispatches another conversation while a deferred delivery is still active", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let firstLifecycle: ZalouserIngressLifecycle | undefined;
+      let releaseFirst = () => {};
+      const firstGate = new Promise<void>((resolve) => {
+        releaseFirst = resolve;
+      });
+      const dispatch = vi.fn(
+        async (message, lifecycle: ZalouserIngressLifecycle): Promise<void> => {
+          if (message.msgId === "lane-active") {
+            firstLifecycle = lifecycle;
+            lifecycle.onDeferred();
+            await firstGate;
+            return;
+          }
+          await lifecycle.onAdopted();
+        },
+      );
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      await ingress.receive(
+        createRawZalouserMessage({ msgId: "lane-active", senderId: "sender-1" }),
+      );
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce());
+
+      await ingress.receive(
+        createRawZalouserMessage({ msgId: "lane-independent", senderId: "sender-2" }),
+      );
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(2));
+      await waitForZalouserIngressVerdict(queue, "lane-independent", "completed");
+
+      releaseFirst();
+      await firstLifecycle?.onAdopted();
+      await ingress.stop();
+    });
+  });
+
+  it("settles deferred bookkeeping when the adoption watchdog aborts a claim", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let deferredLifecycle: ZalouserIngressLifecycle | undefined;
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        deferredLifecycle = lifecycle;
+        lifecycle.onDeferred();
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+        adoptionStallTimeoutMs: 10,
+      });
+      await ingress.receive(createRawZalouserMessage({ msgId: "deferred-timeout" }));
+      await waitForZalouserIngressVerdict(queue, "deferred-timeout", "failed");
+      expect(deferredLifecycle?.abortSignal.aborted).toBe(true);
+
+      await ingress.stop();
+    });
+  });
+
+  it("aborts deferred bookkeeping during shutdown without waiting for adoption", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let deferredLifecycle: ZalouserIngressLifecycle | undefined;
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        deferredLifecycle = lifecycle;
+        lifecycle.onDeferred();
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      await ingress.receive(createRawZalouserMessage({ msgId: "deferred-stop" }));
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce());
+      expect(await queue.listClaims()).toHaveLength(1);
+
+      await ingress.stop();
+      expect(deferredLifecycle?.abortSignal.aborted).toBe(true);
+      expect(await queue.listClaims()).toHaveLength(1);
+    });
+  });
+
+  it("dead-letters malformed persisted envelopes without dispatch", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      await queue.enqueue(
+        "malformed",
+        { version: 1, receivedAt: 1, rawMessage: "{" },
+        { receivedAt: 1, laneKey: "direct:sender-1" },
+      );
+      const dispatch = vi.fn();
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      try {
+        await waitForZalouserIngressVerdict(queue, "malformed", "failed");
+        expect(dispatch).not.toHaveBeenCalled();
+        const verdict = await queue.enqueue("malformed", {
+          version: 1,
+          receivedAt: 2,
+          rawMessage: "{}",
+        });
+        expect(verdict.kind).toBe("failed");
+        if (verdict.kind === "failed") {
+          expect(verdict.record.reason).toBe("invalid-event");
+        }
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("dead-letters authentication failures without retry", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      const dispatch = vi.fn(async () => {
+        throw Object.assign(new Error("expired session"), { code: 401 });
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      try {
+        await ingress.receive(createRawZalouserMessage({ msgId: "auth-failure" }));
+        await waitForZalouserIngressVerdict(queue, "auth-failure", "failed");
+        expect(dispatch).toHaveBeenCalledOnce();
+      } finally {
+        await ingress.stop();
+      }
+    });
+  });
+
+  it("waits for in-flight admission and remains safe to stop twice", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let releaseAppend = () => {};
+      const appendGate = new Promise<void>((resolve) => {
+        releaseAppend = resolve;
+      });
+      const realEnqueue = queue.enqueue.bind(queue);
+      const gatedQueue: ChannelIngressQueue<ZalouserTestIngressPayload> = {
+        ...queue,
+        enqueue: async (...args) => {
+          await appendGate;
+          return await realEnqueue(...args);
+        },
+      };
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue: gatedQueue,
+        dispatch: vi.fn(),
+      });
+      const admission = ingress.receive(createRawZalouserMessage({ msgId: "stop-admission" }));
+      let stopped = false;
+      const stopping = ingress.stop().then(() => {
+        stopped = true;
+      });
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 20);
+      });
+      expect(stopped).toBe(false);
+
+      releaseAppend();
+      await admission;
+      await stopping;
+      await expect(ingress.stop()).resolves.toBeUndefined();
+    });
+  });
+
+  it("waits for an adopted active delivery before stop returns", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let releaseDelivery = () => {};
+      const deliveryGate = new Promise<void>((resolve) => {
+        releaseDelivery = resolve;
+      });
+      const dispatch = vi.fn(async (_message, lifecycle: ZalouserIngressLifecycle) => {
+        await lifecycle.onAdopted();
+        await deliveryGate;
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      await ingress.receive(createRawZalouserMessage({ msgId: "active-stop" }));
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce());
+
+      let stopped = false;
+      const stopping = ingress.stop().then(() => {
+        stopped = true;
+      });
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 20);
+      });
+      expect(stopped).toBe(false);
+
+      releaseDelivery();
+      await stopping;
+      expect(stopped).toBe(true);
+    });
+  });
+
+  it("releases a pre-adoption delivery for retry during shutdown", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      let releaseDelivery = () => {};
+      const deliveryGate = new Promise<void>((resolve) => {
+        releaseDelivery = resolve;
+      });
+      const dispatch = vi.fn(async () => {
+        await deliveryGate;
+      });
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      await ingress.receive(createRawZalouserMessage({ msgId: "shutdown-retry" }));
+      await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce());
+
+      const stopping = ingress.stop();
+      releaseDelivery();
+      await stopping;
+
+      expect(await queue.listClaims()).toHaveLength(0);
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({ id: "shutdown-retry", lastError: expect.any(String) }),
+      ]);
+    });
+  });
+
+  it("does not start a claim after stop wins an async prune", async () => {
+    await withZalouserIngressTestQueue(async (queue) => {
+      await queue.enqueue(
+        "shutdown",
+        {
+          version: 1,
+          receivedAt: 1,
+          rawMessage: JSON.stringify(createRawZalouserMessage({ msgId: "shutdown" })),
+        },
+        { receivedAt: 1, laneKey: "direct:sender-1" },
+      );
+      let releasePrune = () => {};
+      const pruneGate = new Promise<void>((resolve) => {
+        releasePrune = resolve;
+      });
+      const realPrune = queue.prune.bind(queue);
+      const gatedQueue: ChannelIngressQueue<ZalouserTestIngressPayload> = {
+        ...queue,
+        prune: async (...args) => {
+          await pruneGate;
+          return await realPrune(...args);
+        },
+      };
+      const dispatch = vi.fn();
+      const ingress = createZalouserIngressMonitor({
+        accountId: "default",
+        ownUserId: "owner-1",
+        runtime: runtime(),
+        queue: gatedQueue,
+        dispatch,
+      });
+      const stopping = ingress.stop();
+      releasePrune();
+      await stopping;
+
+      expect(dispatch).not.toHaveBeenCalled();
+      expect(await queue.listPending()).toHaveLength(1);
+    });
+  });
+});

extensions/zalouser/src/ingress.ts

@@ -0,0 +1,432 @@
+// Zalouser plugin owns raw zca-js message admission and replay draining.
+import {
+  bindIngressLifecycleToReplyOptions,
+  createChannelIngressDrain,
+  DEFAULT_INGRESS_ADOPTION_STALL_MS,
+  DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+  type ChannelIngressDrain,
+  type ChannelIngressQueue,
+} from "openclaw/plugin-sdk/channel-outbound";
+import { collectErrorGraphCandidates, extractErrorCode } from "openclaw/plugin-sdk/error-runtime";
+import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime";
+import { getZalouserRuntime } from "./runtime.js";
+import type { ZaloInboundMessage } from "./types.js";
+import { normalizeZaloInboundMessage } from "./zalo-js.js";
+import type { Message } from "./zca-client.js";
+import { ThreadType } from "./zca-constants.js";
+
+const ZALOUSER_INGRESS_PAYLOAD_VERSION = 1;
+const ZALOUSER_INGRESS_POLL_INTERVAL_MS = 1_000;
+const ZALOUSER_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000;
+const ZALOUSER_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
+const ZALOUSER_INGRESS_COMPLETED_MAX_ENTRIES = 1_000;
+const ZALOUSER_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
+const ZALOUSER_INGRESS_FAILED_MAX_ENTRIES = 1_000;
+const ZALOUSER_INGRESS_APPEND_RETRY_DELAYS_MS = [0, 100, 300] as const;
+
+type ZalouserIngressPayload = {
+  version: 1;
+  receivedAt: number;
+  rawMessage: string;
+};
+
+export type ZalouserIngressLifecycle = ReturnType<
+  typeof bindIngressLifecycleToReplyOptions
+>["turnAdoptionLifecycle"];
+
+type ZalouserIngressDispatch = (
+  message: ZaloInboundMessage,
+  lifecycle: ZalouserIngressLifecycle,
+) => Promise<void> | void;
+
+type ZalouserIngressMonitor = {
+  receive: (message: Message) => Promise<void>;
+  stop: () => Promise<void>;
+  waitForIdle: () => Promise<void>;
+};
+
+class ZalouserIngressPayloadError extends Error {
+  constructor(message: string, options?: ErrorOptions) {
+    super(message, options);
+    this.name = "ZalouserIngressPayloadError";
+  }
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function nonEmptyString(value: unknown): string | null {
+  return typeof value === "string" && value.trim() ? value.trim() : null;
+}
+
+function inspectZalouserIngressMessage(message: unknown): {
+  eventId: string;
+  laneKey: string;
+} {
+  if (!isRecord(message) || !isRecord(message.data)) {
+    throw new ZalouserIngressPayloadError("zca-js message envelope must contain data.");
+  }
+  const eventId = nonEmptyString(message.data.msgId);
+  if (!eventId) {
+    throw new ZalouserIngressPayloadError("zca-js message envelope is missing data.msgId.");
+  }
+  if (message.type === ThreadType.Group) {
+    const groupId = nonEmptyString(message.data.idTo);
+    if (!groupId) {
+      throw new ZalouserIngressPayloadError("zca-js group message is missing data.idTo.");
+    }
+    return { eventId, laneKey: `group:${groupId}` };
+  }
+  if (message.type !== ThreadType.User) {
+    throw new ZalouserIngressPayloadError("zca-js message has an unsupported thread type.");
+  }
+  const senderId = nonEmptyString(message.data.uidFrom);
+  if (!senderId) {
+    throw new ZalouserIngressPayloadError("zca-js direct message is missing data.uidFrom.");
+  }
+  return { eventId, laneKey: `direct:${senderId}` };
+}
+
+function serializeZalouserIngressMessage(message: Message): string {
+  try {
+    const serialized = JSON.stringify(message);
+    if (typeof serialized !== "string") {
+      throw new ZalouserIngressPayloadError("zca-js message envelope is not serializable.");
+    }
+    return serialized;
+  } catch (error) {
+    if (error instanceof ZalouserIngressPayloadError) {
+      throw error;
+    }
+    throw new ZalouserIngressPayloadError("zca-js message envelope is not serializable.", {
+      cause: error,
+    });
+  }
+}
+
+function parseClaimedMessage(
+  payload: unknown,
+  claimedId: string,
+  claimedLaneKey: string | undefined,
+  ownUserId: string,
+): ZaloInboundMessage {
+  if (
+    !isRecord(payload) ||
+    payload.version !== ZALOUSER_INGRESS_PAYLOAD_VERSION ||
+    typeof payload.rawMessage !== "string"
+  ) {
+    throw new ZalouserIngressPayloadError("Zalouser ingress payload is invalid.");
+  }
+  let rawMessage: unknown;
+  try {
+    rawMessage = JSON.parse(payload.rawMessage);
+  } catch (error) {
+    throw new ZalouserIngressPayloadError("Zalouser ingress message JSON is invalid.", {
+      cause: error,
+    });
+  }
+  const facts = inspectZalouserIngressMessage(rawMessage);
+  if (facts.eventId !== claimedId || facts.laneKey !== claimedLaneKey) {
+    throw new ZalouserIngressPayloadError(
+      "Zalouser message identity changed after durable admission.",
+    );
+  }
+  const normalized = normalizeZaloInboundMessage(rawMessage as Message, ownUserId);
+  if (!normalized) {
+    throw new ZalouserIngressPayloadError("Zalouser message could not be normalized.");
+  }
+  return normalized;
+}
+
+function isZalouserAuthenticationFailure(error: unknown): boolean {
+  for (const candidate of collectErrorGraphCandidates(error, (current) => [current.cause])) {
+    const code = extractErrorCode(candidate);
+    const record = candidate as { status?: unknown; statusCode?: unknown };
+    if (
+      code === "401" ||
+      code === "403" ||
+      record.status === 401 ||
+      record.status === 403 ||
+      record.statusCode === 401 ||
+      record.statusCode === 403
+    ) {
+      return true;
+    }
+  }
+  return false;
+}
+
+function errorText(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
+
+export function createZalouserIngressMonitor(options: {
+  accountId: string;
+  ownUserId: string;
+  runtime: Pick<RuntimeEnv, "error" | "log">;
+  dispatch: ZalouserIngressDispatch;
+  queue?: ChannelIngressQueue<ZalouserIngressPayload>;
+  pollIntervalMs?: number;
+  adoptionStallTimeoutMs?: number;
+}): ZalouserIngressMonitor {
+  let queue = options.queue;
+  let drain: ChannelIngressDrain | undefined;
+  let running = true;
+  let requested = false;
+  let pumping: Promise<void> | undefined;
+  let lastPrunedAt = 0;
+  let admissionTail: Promise<void> = Promise.resolve();
+  let stopTask: Promise<void> | undefined;
+  const shutdown = new AbortController();
+  const activeDeliveries = new Set<Promise<void>>();
+  const deferredClaims = new Map<string, Promise<void>>();
+
+  const getQueue = (): ChannelIngressQueue<ZalouserIngressPayload> => {
+    queue ??= getZalouserRuntime().state.openChannelIngressQueue<ZalouserIngressPayload>({
+      accountId: options.accountId,
+    });
+    return queue;
+  };
+
+  const getDrain = (): ChannelIngressDrain => {
+    drain ??= createChannelIngressDrain<ZalouserIngressPayload>({
+      queue: getQueue(),
+      orderBy: "received",
+      adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS,
+      retryPolicy: {
+        maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+        deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+      },
+      abortSignal: shutdown.signal,
+      resolveNonRetryableFailure: (error) => {
+        if (error instanceof ZalouserIngressPayloadError) {
+          return { reason: "invalid-event", message: error.message };
+        }
+        if (isZalouserAuthenticationFailure(error)) {
+          return { reason: "authentication-failed", message: errorText(error) };
+        }
+        return null;
+      },
+      onLog: (message) => options.runtime.error?.(`zalouser ingress: ${message}`),
+      dispatchClaimedEvent: async (claimed, lifecycle) => {
+        if (!running || lifecycle.abortSignal.aborted) {
+          return {
+            kind: "failed-retryable",
+            error: new Error("Zalouser ingress stopped before dispatch."),
+          };
+        }
+        const message = parseClaimedMessage(
+          claimed.payload,
+          claimed.id,
+          claimed.laneKey,
+          options.ownUserId,
+        );
+        const bound = bindIngressLifecycleToReplyOptions(lifecycle).turnAdoptionLifecycle;
+        let handedOff = false;
+        let resolveDeferredClaim!: () => void;
+        const deferredClaim = new Promise<void>((resolve) => {
+          resolveDeferredClaim = resolve;
+        });
+        let deferredClaimSettled = false;
+        const settleDeferredClaim = () => {
+          if (deferredClaimSettled) {
+            return;
+          }
+          deferredClaimSettled = true;
+          lifecycle.abortSignal.removeEventListener("abort", settleDeferredClaim);
+          if (deferredClaims.get(claimed.id) === deferredClaim) {
+            deferredClaims.delete(claimed.id);
+          }
+          resolveDeferredClaim();
+          requestDrain();
+        };
+        // The drain can guillotine or dispose a deferred claim without invoking
+        // the reply lifecycle again. Release local bookkeeping on that abort.
+        lifecycle.abortSignal.addEventListener("abort", settleDeferredClaim, { once: true });
+        if (lifecycle.abortSignal.aborted) {
+          settleDeferredClaim();
+        }
+        const delivery = Promise.resolve(
+          options.dispatch(message, {
+            ...bound,
+            onAdopted: async () => {
+              handedOff = true;
+              try {
+                await bound.onAdopted();
+              } finally {
+                settleDeferredClaim();
+              }
+            },
+            onDeferred: () => {
+              handedOff = true;
+              if (!deferredClaimSettled) {
+                deferredClaims.set(claimed.id, deferredClaim);
+              }
+              bound.onDeferred();
+            },
+            onAbandoned: async () => {
+              handedOff = true;
+              try {
+                await bound.onAbandoned();
+              } finally {
+                settleDeferredClaim();
+              }
+            },
+          }),
+        );
+        activeDeliveries.add(delivery);
+        try {
+          await delivery;
+        } catch (error) {
+          if (!running || lifecycle.abortSignal.aborted) {
+            return { kind: "failed-retryable", error };
+          }
+          throw error;
+        } finally {
+          activeDeliveries.delete(delivery);
+        }
+        if (!handedOff) {
+          if (!running || lifecycle.abortSignal.aborted) {
+            return {
+              kind: "failed-retryable",
+              error: new Error("Zalouser ingress stopped before adoption."),
+            };
+          }
+          // Policy gates and deliberate no-dispatch turns are terminal.
+          await bound.onAdopted();
+        }
+        return deferredClaims.has(claimed.id) ? { kind: "deferred" } : { kind: "completed" };
+      },
+    });
+    return drain;
+  };
+
+  const pruneIfDue = async (): Promise<void> => {
+    const now = Date.now();
+    if (now - lastPrunedAt < ZALOUSER_INGRESS_PRUNE_INTERVAL_MS) {
+      return;
+    }
+    await getQueue().prune({
+      completedTtlMs: ZALOUSER_INGRESS_COMPLETED_TTL_MS,
+      completedMaxEntries: ZALOUSER_INGRESS_COMPLETED_MAX_ENTRIES,
+      failedTtlMs: ZALOUSER_INGRESS_FAILED_TTL_MS,
+      failedMaxEntries: ZALOUSER_INGRESS_FAILED_MAX_ENTRIES,
+      now,
+    });
+    lastPrunedAt = now;
+  };
+
+  const runPump = async (): Promise<void> => {
+    try {
+      for (;;) {
+        requested = false;
+        await pruneIfDue();
+        // stop() can run during the async prune; never start a claim afterwards.
+        if (!running) {
+          break;
+        }
+        const activeDrain = getDrain();
+        const { started } = await activeDrain.drainOnce();
+        if (!running || (!requested && started === 0)) {
+          break;
+        }
+      }
+    } catch (error) {
+      options.runtime.error?.(`zalouser ingress drain failed: ${errorText(error)}`);
+    } finally {
+      pumping = undefined;
+      if (running && requested) {
+        requestDrain();
+      }
+    }
+  };
+
+  function requestDrain(): void {
+    requested = true;
+    if (!running || pumping) {
+      return;
+    }
+    pumping = runPump();
+  }
+
+  const timer = setInterval(
+    requestDrain,
+    options.pollIntervalMs ?? ZALOUSER_INGRESS_POLL_INTERVAL_MS,
+  );
+  timer.unref?.();
+  requestDrain();
+
+  const admitOnce = async (message: Message): Promise<void> => {
+    const facts = inspectZalouserIngressMessage(message);
+    const rawMessage = serializeZalouserIngressMessage(message);
+    const receivedAt = Date.now();
+    let lastError: unknown;
+    for (const delayMs of ZALOUSER_INGRESS_APPEND_RETRY_DELAYS_MS) {
+      if (delayMs > 0) {
+        await new Promise<void>((resolve) => {
+          setTimeout(resolve, delayMs);
+        });
+      }
+      try {
+        await getQueue().enqueue(
+          facts.eventId,
+          {
+            version: ZALOUSER_INGRESS_PAYLOAD_VERSION,
+            receivedAt,
+            rawMessage,
+          },
+          { receivedAt, laneKey: facts.laneKey },
+        );
+        requestDrain();
+        return;
+      } catch (error) {
+        lastError = error;
+      }
+    }
+    throw new Error("Zalouser durable ingress append failed.", { cause: lastError });
+  };
+
+  return {
+    receive: (message) => {
+      if (!running) {
+        return Promise.reject(new Error("Zalouser ingress monitor is stopped."));
+      }
+      // zca-js callbacks can overlap. Preserve arrival order through append backoff.
+      const admission = admissionTail.then(() => admitOnce(message));
+      admissionTail = admission.catch(() => undefined);
+      return admission;
+    },
+    stop: () => {
+      stopTask ??= (async () => {
+        running = false;
+        clearInterval(timer);
+        await admissionTail;
+        shutdown.abort(new Error("Zalouser ingress stopped."));
+        await pumping;
+        await Promise.allSettled(activeDeliveries);
+        // Abort deferred per-claim lifecycles so their durable rows stay available
+        // for recovery instead of holding shutdown open indefinitely.
+        drain?.dispose();
+        await Promise.allSettled(deferredClaims.values());
+        await drain?.waitForIdle();
+        // Dispose remains safe if monitor cleanup repeats.
+        drain?.dispose();
+        drain?.dispose();
+      })();
+      return stopTask;
+    },
+    waitForIdle: async () => {
+      await admissionTail;
+      for (;;) {
+        const activePump = pumping;
+        if (!activePump) {
+          break;
+        }
+        await activePump;
+      }
+      await drain?.waitForIdle();
+    },
+  };
+}

extensions/zalouser/src/monitor.account-scope.test.ts

@@ -1,9 +1,13 @@
-import { KeyedAsyncQueue } from "openclaw/plugin-sdk/core";
 // Zalouser tests cover monitor.account scope plugin behavior.
 import { describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig, PluginRuntime } from "../runtime-api.js";
 import "./monitor.send.test-mocks.js";
 import "./zalo-js.test-mocks.js";
+import {
+  createRawZalouserMessageFromNormalized,
+  waitForZalouserIngressVerdict,
+  withZalouserIngressTestQueue,
+} from "./ingress.test-support.js";
 import { monitorZalouserProvider } from "./monitor.js";
 import { sendMessageZalouserMock } from "./monitor.send.test-mocks.js";
 import { setZalouserRuntime } from "./runtime.js";
@@ -100,39 +104,32 @@ describe("zalouser monitor pairing account scoping", () => {
       raw: { source: "test" },
     };
 
-    const enqueueSpy = vi.spyOn(KeyedAsyncQueue.prototype, "enqueue");
-    const abortController = new AbortController();
-    let resolveListener: ((params: ListenerParams) => void) | undefined;
-    const listenerReady = new Promise<ListenerParams>((resolve) => {
-      resolveListener = resolve;
-    });
-    startZaloListenerMock.mockImplementationOnce(async (listenerParams) => {
-      resolveListener?.(listenerParams);
-      return { stop: vi.fn() };
-    });
-    const run = monitorZalouserProvider({
-      account,
-      config,
-      runtime: createZalouserRuntimeEnv(),
-      abortSignal: abortController.signal,
-    });
-    try {
-      const listenerParams = await listenerReady;
-      const resultIndex = enqueueSpy.mock.results.length;
-      listenerParams.onMessage(message);
-      const queued = enqueueSpy.mock.results[resultIndex]?.value;
-      if (!(queued instanceof Promise)) {
-        throw new Error("Zalouser monitor did not enqueue the inbound message");
-      }
-      await queued;
-    } finally {
-      abortController.abort();
+    await withZalouserIngressTestQueue(async (ingressQueue) => {
+      const abortController = new AbortController();
+      let resolveListener: ((params: ListenerParams) => void) | undefined;
+      const listenerReady = new Promise<ListenerParams>((resolve) => {
+        resolveListener = resolve;
+      });
+      startZaloListenerMock.mockImplementationOnce(async (listenerParams) => {
+        resolveListener?.(listenerParams);
+        return { stop: vi.fn() };
+      });
+      const run = monitorZalouserProvider({
+        account,
+        config,
+        runtime: createZalouserRuntimeEnv(),
+        abortSignal: abortController.signal,
+        ingressQueue,
+      });
       try {
-        await run;
+        const listenerParams = await listenerReady;
+        await listenerParams.onMessage(createRawZalouserMessageFromNormalized(message));
+        await waitForZalouserIngressVerdict(ingressQueue, "msg-1", "completed");
       } finally {
-        enqueueSpy.mockRestore();
+        abortController.abort();
+        await run;
       }
-    }
+    });
 
     expect(readAllowFromStore).toHaveBeenCalledOnce();
     const allowStoreParams = requireRecord(

extensions/zalouser/src/monitor.group-gating.test.ts

@@ -1,11 +1,15 @@
 // Zalouser tests cover monitor.group gating plugin behavior.
 import { createChannelMessageReplyPipeline } from "openclaw/plugin-sdk/channel-outbound";
-import { KeyedAsyncQueue } from "openclaw/plugin-sdk/core";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig, PluginRuntime } from "../runtime-api.js";
 import "./monitor.send.test-mocks.js";
 import "./zalo-js.test-mocks.js";
 import { resolveZalouserAccountSync } from "./accounts.js";
+import {
+  createRawZalouserMessageFromNormalized,
+  waitForZalouserIngressVerdict,
+  withZalouserIngressTestQueue,
+} from "./ingress.test-support.js";
 import { monitorZalouserProvider } from "./monitor.js";
 import {
   sendDeliveredZalouserMock,
@@ -295,7 +299,6 @@ async function processMessageThroughMonitor(params: {
   historyState?: { historyLimit?: number };
   statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
 }): Promise<void> {
-  const enqueueSpy = vi.spyOn(KeyedAsyncQueue.prototype, "enqueue");
   const messages = params.messages ?? (params.message ? [params.message] : []);
   const account = params.historyState?.historyLimit
     ? {
@@ -303,38 +306,35 @@ async function processMessageThroughMonitor(params: {
         config: { ...params.account.config, historyLimit: params.historyState.historyLimit },
       }
     : params.account;
-  const abortController = new AbortController();
-  let resolveProcessed: (() => void) | undefined;
-  const processed = new Promise<void>((resolve) => {
-    resolveProcessed = resolve;
-  });
-  startZaloListenerMock.mockImplementationOnce(async (listenerParams) => {
-    for (const message of messages) {
-      const resultIndex = enqueueSpy.mock.results.length;
-      listenerParams.onMessage(message);
-      const queued = enqueueSpy.mock.results[resultIndex]?.value;
-      if (!(queued instanceof Promise)) {
-        throw new Error("Zalouser monitor did not enqueue the inbound message");
+  await withZalouserIngressTestQueue(async (ingressQueue) => {
+    const abortController = new AbortController();
+    let resolveProcessed: (() => void) | undefined;
+    const processed = new Promise<void>((resolve) => {
+      resolveProcessed = resolve;
+    });
+    startZaloListenerMock.mockImplementationOnce(async (listenerParams) => {
+      for (const message of messages) {
+        await listenerParams.onMessage(createRawZalouserMessageFromNormalized(message));
+        if (!message.msgId) {
+          throw new Error("Zalouser monitor test message requires msgId");
+        }
+        await waitForZalouserIngressVerdict(ingressQueue, message.msgId, "completed");
       }
-      await queued;
-    }
-    resolveProcessed?.();
-    return { stop: vi.fn() };
-  });
-  try {
+      resolveProcessed?.();
+      return { stop: vi.fn() };
+    });
     const run = monitorZalouserProvider({
       account,
       config: params.config,
       runtime: params.runtime,
       abortSignal: abortController.signal,
       statusSink: params.statusSink,
+      ingressQueue,
     });
     await processed;
     abortController.abort();
     await run;
-  } finally {
-    enqueueSpy.mockRestore();
-  }
+  });
 }
 
 async function processGroupControlCommand(params: {
@@ -440,14 +440,17 @@ describe("zalouser monitor group mention gating", () => {
     installRuntime({ commandAuthorized: false });
     const abortController = new AbortController();
     abortController.abort();
-    await monitorZalouserProvider({
-      account: {
-        ...createAccount(),
-        config: accountConfig,
-      },
-      config: createConfig(),
-      runtime: createRuntimeEnv(),
-      abortSignal: abortController.signal,
+    await withZalouserIngressTestQueue(async (ingressQueue) => {
+      await monitorZalouserProvider({
+        account: {
+          ...createAccount(),
+          config: accountConfig,
+        },
+        config: createConfig(),
+        runtime: createRuntimeEnv(),
+        abortSignal: abortController.signal,
+        ingressQueue,
+      });
     });
   }
 
@@ -930,11 +933,13 @@ describe("zalouser monitor group mention gating", () => {
         }),
         createGroupMessage({
           content: "second line @bot",
+          msgId: "history-2",
           hasAnyMention: true,
           wasExplicitlyMentioned: true,
         }),
         createGroupMessage({
           content: "third line @bot",
+          msgId: "history-3",
           hasAnyMention: true,
           wasExplicitlyMentioned: true,
         }),

extensions/zalouser/src/monitor.ts

@@ -7,7 +7,6 @@ import {
 import { resolveStableChannelMessageIngress } from "openclaw/plugin-sdk/channel-ingress-runtime";
 import { createChannelPairingController } from "openclaw/plugin-sdk/channel-pairing";
 import type { MarkdownTableMode, OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
-import { KeyedAsyncQueue } from "openclaw/plugin-sdk/core";
 import { isDangerousNameMatchingEnabled } from "openclaw/plugin-sdk/dangerous-name-runtime";
 // Zalouser plugin module implements monitor behavior.
 import { expectDefined } from "openclaw/plugin-sdk/expect-runtime";
@@ -38,6 +37,7 @@ import {
   findZalouserGroupEntry,
   isZalouserGroupEntryAllowed,
 } from "./group-policy.js";
+import { createZalouserIngressMonitor, type ZalouserIngressLifecycle } from "./ingress.js";
 import { formatZalouserMessageSidFull, resolveZalouserMessageSid } from "./message-sid.js";
 import { getZalouserRuntime } from "./runtime.js";
 import {
@@ -51,6 +51,7 @@ import type { ResolvedZalouserAccount, ZaloInboundMessage } from "./types.js";
 import {
   listZaloFriends,
   listZaloGroups,
+  resolveZaloOwnUserId,
   resolveZaloGroupContext,
   startZaloListener,
 } from "./zalo-js.js";
@@ -61,10 +62,11 @@ type ZalouserMonitorOptions = {
   runtime: RuntimeEnv;
   abortSignal: AbortSignal;
   statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
+  ingressQueue?: Parameters<typeof createZalouserIngressMonitor>[0]["queue"];
 };
 
 type ZalouserMonitorResult = {
-  stop: () => void;
+  stop: () => Promise<void>;
 };
 
 const ZALOUSER_TEXT_LIMIT = 2000;
@@ -127,15 +129,6 @@ function normalizeZalouserSender(value: string): string | null {
   return normalizeOptionalLowercaseString(normalizeZalouserAllowEntry(value)) || null;
 }
 
-function resolveInboundQueueKey(message: ZaloInboundMessage): string {
-  const threadId = message.threadId?.trim() || "unknown";
-  if (message.isGroup) {
-    return `group:${threadId}`;
-  }
-  const senderId = message.senderId?.trim();
-  return `direct:${senderId || threadId}`;
-}
-
 function resolveZalouserRouteAccess(params: {
   groupPolicy: "open" | "disabled" | "allowlist";
   configured: boolean;
@@ -224,6 +217,7 @@ async function processMessage(
   runtime: RuntimeEnv,
   historyState: ZalouserGroupHistoryState,
   statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void,
+  turnAdoptionLifecycle?: ZalouserIngressLifecycle,
 ): Promise<void> {
   const pairing = createChannelPairingController({
     core,
@@ -677,6 +671,7 @@ async function processMessage(
         runtime.error?.(`zalouser: failed updating session meta: ${String(err)}`);
       },
     },
+    replyOptions: turnAdoptionLifecycle ? { turnAdoptionLifecycle } : undefined,
   });
   if (isGroup && historyKey) {
     channelHistory.clear({
@@ -755,7 +750,6 @@ export async function monitorZalouserProvider(
   const { abortSignal, statusSink, runtime } = options;
 
   const core = getZalouserRuntime();
-  const inboundQueue = new KeyedAsyncQueue();
   const historyLimit = Math.max(
     0,
     account.config.historyLimit ??
@@ -857,16 +851,38 @@ export async function monitorZalouserProvider(
     runtime.log?.(`zalouser resolve failed; using config entries. ${String(err)}`);
   }
 
+  const ownUserId = await resolveZaloOwnUserId(account.profile);
+  const ingress = createZalouserIngressMonitor({
+    accountId: account.accountId,
+    ownUserId,
+    runtime,
+    ...(options.ingressQueue ? { queue: options.ingressQueue } : {}),
+    dispatch: async (message, lifecycle) => {
+      await processMessage(
+        message,
+        account,
+        config,
+        core,
+        runtime,
+        { historyLimit, groupHistories },
+        statusSink,
+        lifecycle,
+      );
+    },
+  });
+
   let listenerStop: (() => void) | null = null;
   let stopped = false;
+  let stopTask: Promise<void> | undefined;
 
-  const stop = () => {
-    if (stopped) {
-      return;
-    }
-    stopped = true;
-    listenerStop?.();
-    listenerStop = null;
+  const stop = (): Promise<void> => {
+    stopTask ??= (async () => {
+      stopped = true;
+      listenerStop?.();
+      listenerStop = null;
+      await ingress.stop();
+    })();
+    return stopTask;
   };
 
   let settled = false;
@@ -877,8 +893,7 @@ export async function monitorZalouserProvider(
       return;
     }
     settled = true;
-    stop();
-    resolveRun();
+    void stop().then(resolveRun, rejectRun);
   };
 
   const settleFailure = (error: unknown) => {
@@ -886,8 +901,12 @@ export async function monitorZalouserProvider(
       return;
     }
     settled = true;
-    stop();
-    rejectRun(error instanceof Error ? error : new Error(String(error)));
+    const failure = error instanceof Error ? error : new Error(String(error));
+    void stop().then(
+      () => rejectRun(failure),
+      (stopError: unknown) =>
+        rejectRun(stopError instanceof Error ? stopError : new Error(String(stopError))),
+    );
   };
 
   const onAbort = () => {
@@ -901,31 +920,13 @@ export async function monitorZalouserProvider(
       accountId: account.accountId,
       profile: account.profile,
       abortSignal,
-      onMessage: (msg) => {
+      onMessage: async (msg) => {
         if (stopped) {
           return;
         }
         logVerbose(core, runtime, `[${account.accountId}] inbound message`);
         statusSink?.({ lastInboundAt: Date.now() });
-        const queueKey = resolveInboundQueueKey(msg);
-        void inboundQueue
-          .enqueue(queueKey, async () => {
-            if (stopped || abortSignal.aborted) {
-              return;
-            }
-            await processMessage(
-              msg,
-              account,
-              config,
-              core,
-              runtime,
-              { historyLimit, groupHistories },
-              statusSink,
-            );
-          })
-          .catch((err: unknown) => {
-            runtime.error(`[${account.accountId}] Failed to process message: ${String(err)}`);
-          });
+        await ingress.receive(msg);
       },
       onError: (err) => {
         if (stopped || abortSignal.aborted) {
@@ -937,6 +938,7 @@ export async function monitorZalouserProvider(
     });
   } catch (error) {
     abortSignal.removeEventListener("abort", onAbort);
+    await ingress.stop();
     throw error;
   }
 

extensions/zalouser/src/zalo-js.test-mocks.ts

@@ -11,9 +11,11 @@ type ZaloJsMocks = {
   listZaloGroupsMock: Mock<ZaloJsModule["listZaloGroups"]>;
   listZaloGroupsMatchingMock: Mock<ZaloJsModule["listZaloGroupsMatching"]>;
   logoutZaloProfileMock: Mock<ZaloJsModule["logoutZaloProfile"]>;
+  normalizeZaloInboundMessageMock: Mock<ZaloJsModule["normalizeZaloInboundMessage"]>;
   resolveZaloAllowFromEntriesMock: Mock<ZaloJsModule["resolveZaloAllowFromEntries"]>;
   resolveZaloGroupContextMock: Mock<ZaloJsModule["resolveZaloGroupContext"]>;
   resolveZaloGroupsByEntriesMock: Mock<ZaloJsModule["resolveZaloGroupsByEntries"]>;
+  resolveZaloOwnUserIdMock: Mock<ZaloJsModule["resolveZaloOwnUserId"]>;
   startZaloListenerMock: Mock<ZaloJsModule["startZaloListener"]>;
   startZaloQrLoginMock: Mock<ZaloJsModule["startZaloQrLogin"]>;
   waitForZaloQrLoginMock: Mock<ZaloJsModule["waitForZaloQrLogin"]>;
@@ -33,6 +35,12 @@ const zaloJsMocks = vi.hoisted(
       loggedOut: true,
       message: "Logged out and cleared local session.",
     })),
+    normalizeZaloInboundMessageMock: vi.fn((message) => {
+      const normalized = message.data.testNormalizedMessage;
+      return normalized && typeof normalized === "object"
+        ? (normalized as ReturnType<ZaloJsModule["normalizeZaloInboundMessage"]>)
+        : null;
+    }),
     resolveZaloAllowFromEntriesMock: vi.fn(async ({ entries }: { entries: string[] }) =>
       entries.map((entry) => ({ input: entry, resolved: true, id: entry, note: undefined })),
     ),
@@ -44,6 +52,7 @@ const zaloJsMocks = vi.hoisted(
     resolveZaloGroupsByEntriesMock: vi.fn(async ({ entries }: { entries: string[] }) =>
       entries.map((entry) => ({ input: entry, resolved: true, id: entry, note: undefined })),
     ),
+    resolveZaloOwnUserIdMock: vi.fn(async () => "owner-1"),
     startZaloListenerMock: vi.fn(async () => ({ stop: vi.fn() })),
     startZaloQrLoginMock: vi.fn(async () => ({
       message: "qr pending",
@@ -78,9 +87,11 @@ vi.mock("./zalo-js.js", () => ({
   listZaloGroups: listZaloGroupsMock,
   listZaloGroupsMatching: zaloJsMocks.listZaloGroupsMatchingMock,
   logoutZaloProfile: zaloJsMocks.logoutZaloProfileMock,
+  normalizeZaloInboundMessage: zaloJsMocks.normalizeZaloInboundMessageMock,
   resolveZaloAllowFromEntries: zaloJsMocks.resolveZaloAllowFromEntriesMock,
   resolveZaloGroupContext: zaloJsMocks.resolveZaloGroupContextMock,
   resolveZaloGroupsByEntries: zaloJsMocks.resolveZaloGroupsByEntriesMock,
+  resolveZaloOwnUserId: zaloJsMocks.resolveZaloOwnUserIdMock,
   startZaloListener: startZaloListenerMock,
   startZaloQrLogin: startZaloQrLoginMock,
   waitForZaloQrLogin: waitForZaloQrLoginMock,

extensions/zalouser/src/zalo-js.ts

@@ -860,7 +860,10 @@ function extractGroupMembersFromInfo(
   return members;
 }
 
-function toInboundMessage(message: Message, ownUserId?: string): ZaloInboundMessage | null {
+export function normalizeZaloInboundMessage(
+  message: Message,
+  ownUserId?: string,
+): ZaloInboundMessage | null {
   const data = message.data;
   const isGroup = message.type === ThreadType.Group;
   const senderId = toNumberId(data.uidFrom);
@@ -1332,6 +1335,10 @@ async function resolveOwnUserId(api: API): Promise<string> {
   return "";
 }
 
+export async function resolveZaloOwnUserId(profileInput?: string | null): Promise<string> {
+  return await withZaloApi(profileInput, resolveOwnUserId);
+}
+
 export async function sendZaloReaction(params: {
   profile?: string | null;
   threadId: string;
@@ -1713,7 +1720,7 @@ export async function startZaloListener(params: {
   accountId: string;
   profile?: string | null;
   abortSignal: AbortSignal;
-  onMessage: (message: ZaloInboundMessage) => void;
+  onMessage: (message: Message) => void | Promise<void>;
   onError: (error: Error) => void;
 }): Promise<{ stop: () => void }> {
   const profile = normalizeProfile(params.profile);
@@ -1725,10 +1732,7 @@ export async function startZaloListener(params: {
     );
   }
 
-  const { api, ownUserId } = await withZaloApi(profile, async (apiLocal) => ({
-    api: apiLocal,
-    ownUserId: await resolveOwnUserId(apiLocal),
-  }));
+  const api = await withZaloApi(profile, async (apiLocal) => apiLocal);
   let stopped = false;
   let watchdogTimer: ReturnType<typeof setInterval> | null = null;
   let lastWatchdogTickAt = Date.now();
@@ -1761,11 +1765,9 @@ export async function startZaloListener(params: {
     if (incoming.isSelf) {
       return;
     }
-    const normalized = toInboundMessage(incoming, ownUserId);
-    if (!normalized) {
-      return;
-    }
-    params.onMessage(normalized);
+    void Promise.resolve(params.onMessage(incoming)).catch((error: unknown) => {
+      failListener(error instanceof Error ? error : new Error(String(error)));
+    });
   };
 
   const failListener = (error: Error) => {

extensions/zalouser/src/zalo-quote-metadata.test.ts

@@ -22,7 +22,12 @@ vi.mock("./zca-client.js", () => ({
 
 import { setZalouserRuntime } from "./runtime.js";
 import { saveStoredZaloCredentials } from "./session-state.js";
-import { resolveZaloGroupContext, sendZaloTextMessage, startZaloListener } from "./zalo-js.js";
+import {
+  normalizeZaloInboundMessage,
+  resolveZaloGroupContext,
+  sendZaloTextMessage,
+  startZaloListener,
+} from "./zalo-js.js";
 
 type ListenerOn = ReturnType<typeof vi.fn>;
 
@@ -140,7 +145,7 @@ describe("Zalo inbound normalization", () => {
         stop: vi.fn(),
       },
     } as Partial<API>);
-    const received: Array<Record<string, unknown>> = [];
+    const received: Message[] = [];
 
     await withStoredSession({
       profile,
@@ -151,7 +156,9 @@ describe("Zalo inbound normalization", () => {
           accountId: "default",
           profile,
           abortSignal: abortController.signal,
-          onMessage: (message) => received.push(message as unknown as Record<string, unknown>),
+          onMessage: (message) => {
+            received.push(message);
+          },
           onError: vi.fn(),
         });
         const onMessage = findListener(listenerOn, "message");
@@ -162,7 +169,10 @@ describe("Zalo inbound normalization", () => {
       },
     });
 
-    return received;
+    return received
+      .map((message) => normalizeZaloInboundMessage(message, "555444333"))
+      .filter((message): message is NonNullable<typeof message> => message !== null)
+      .map((message) => message as unknown as Record<string, unknown>);
   }
 
   it("extracts quote metadata and implicit mentions", async () => {