fix(matrix): per-room send queue and immediate read receipts

joshjhall · joshjhall · commit 4b29e1995455 · 2026-02-24T15:53:47.000-06:00
Two fixes for the Matrix channel plugin: 1. Per-room send queue (fixes #11614): Introduce send-queue.ts that serializes all sends to the same Matrix room through enqueueSend(), with a 150ms inter-send gap to ensure distinct origin_server_ts values. This prevents message ordering races when auto-reply and message tool sends fire concurrently. 2. Immediate read receipts (fixes #25840): Move sendReadReceiptMatrix from handler.ts (deep in the processing pipeline) to events.ts (on message arrival from SDK). Read receipts now fire before any processing, so Element shows 'read' immediately even when the agent is busy. Self-messages are filtered via lazy getUserId resolution. Both changes are contained within the Matrix extension and have no effect on other channels. AI-assisted: This PR was developed with AI assistance (Claude/OpenClaw). Tested against a live Matrix homeserver (Tuwunel/Conduit fork) with both fixes running in production for ~2 weeks via a local plugin fork.
diff --git a/extensions/matrix/src/matrix/monitor/events.ts b/extensions/matrix/src/matrix/monitor/events.ts
@@ -1,6 +1,7 @@
 import type { MatrixClient } from "@vector-im/matrix-bot-sdk";
 import type { PluginRuntime, RuntimeLogger } from "openclaw/plugin-sdk";
 import type { MatrixAuth } from "../client.js";
+import { sendReadReceiptMatrix } from "../send.js";
 import type { MatrixRawEvent } from "./types.js";
 import { EventType } from "./types.js";
 
@@ -25,7 +26,35 @@ export function registerMatrixMonitorEvents(params: {
     onRoomMessage,
   } = params;
 
-  client.on("room.message", onRoomMessage);
+  // Track our own user ID for filtering (resolved lazily on first message)
+  let selfUserId: string | undefined;
+
+  client.on("room.message", (roomId: string, event: MatrixRawEvent) => {
+    // Fire read receipt immediately — before the handler queues or processes.
+    // This ensures Element shows "read" even when the agent is busy.
+    const eventId = event?.event_id;
+    const senderId = event?.sender;
+    if (eventId && senderId) {
+      const doReceipt = async () => {
+        if (!selfUserId) {
+          try {
+            selfUserId = await client.getUserId();
+          } catch {
+            return;
+          }
+        }
+        if (senderId === selfUserId) return;
+        sendReadReceiptMatrix(roomId, eventId, client).catch((err) => {
+          logVerboseMessage(
+            `matrix: early read receipt failed room=${roomId} id=${eventId}: ${String(err)}`,
+          );
+        });
+      };
+      doReceipt();
+    }
+    // Delegate to the actual message handler
+    onRoomMessage(roomId, event);
+  });
 
   client.on("room.encrypted_event", (roomId: string, event: MatrixRawEvent) => {
     const eventId = event?.event_id ?? "unknown";
diff --git a/extensions/matrix/src/matrix/monitor/handler.ts b/extensions/matrix/src/matrix/monitor/handler.ts
@@ -21,7 +21,6 @@ import {
 import {
   reactMatrixMessage,
   sendMessageMatrix,
-  sendReadReceiptMatrix,
   sendTypingMatrix,
 } from "../send.js";
 import {
@@ -602,13 +601,9 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
         return;
       }
 
-      if (messageId) {
-        sendReadReceiptMatrix(roomId, messageId, client).catch((err) => {
-          logVerboseMessage(
-            `matrix: read receipt failed room=${roomId} id=${messageId}: ${String(err)}`,
-          );
-        });
-      }
+      // Read receipt is now sent immediately in events.ts when the message
+      // arrives from the SDK, before handler processing. This ensures "read"
+      // status shows even when the agent is busy with queued messages.
 
       let didSendReply = false;
       const tableMode = core.channel.text.resolveMarkdownTableMode({
diff --git a/extensions/matrix/src/matrix/send-queue.ts b/extensions/matrix/src/matrix/send-queue.ts
@@ -0,0 +1,65 @@
+/**
+ * Per-room send queue to ensure message ordering.
+ *
+ * Matrix orders messages by origin_server_ts. When multiple code paths
+ * (auto-reply, message tool, media sends) fire concurrently, they race
+ * and messages appear out of order. This queue serializes all sends
+ * per room so they arrive in the order they were enqueued.
+ *
+ * See: https://github.com/openclaw/openclaw/issues/11614
+ */
+
+const queues = new Map<string, Promise<unknown>>();
+
+/**
+ * Enqueue a send operation for a specific room.
+ * Each send waits for the previous one to complete before starting.
+ * The 150ms gap ensures Matrix server assigns distinct timestamps.
+ */
+export async function enqueueSend<T>(roomId: string, fn: () => Promise<T>): Promise<T> {
+  const prev = queues.get(roomId) ?? Promise.resolve();
+  const SEND_GAP_MS = 150;
+
+  const next = prev
+    .catch(() => {}) // don't let previous failures block the queue
+    .then(() => delay(SEND_GAP_MS))
+    .then(() => fn());
+
+  // Store the chain (void version) so next enqueue waits for this one
+  queues.set(
+    roomId,
+    next.then(
+      () => {},
+      () => {},
+    ),
+  );
+
+  // Clean up empty queues to prevent memory leak
+  next.finally(() => {
+    setTimeout(() => {
+      const current = queues.get(roomId);
+      if (current && isSettled(current)) {
+        queues.delete(roomId);
+      }
+    }, 1000);
+  });
+
+  return next;
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise((r) => setTimeout(r, ms));
+}
+
+function isSettled(p: Promise<unknown>): boolean {
+  let settled = false;
+  p.then(
+    () => {
+      settled = true;
+    },
+    () => {
+      settled = true;
+    },
+  );
+  return settled;
+}
diff --git a/extensions/matrix/src/matrix/send.ts b/extensions/matrix/src/matrix/send.ts
@@ -3,6 +3,7 @@ import type { PollInput } from "openclaw/plugin-sdk";
 import { getMatrixRuntime } from "../runtime.js";
 import { buildPollStartContent, M_POLL_START } from "./poll-types.js";
 import { resolveMatrixClient, resolveMediaMaxBytes } from "./send/client.js";
+import { enqueueSend } from "./send-queue.js";
 import {
   buildReplyRelation,
   buildTextContent,
@@ -49,6 +50,7 @@ export async function sendMessageMatrix(
   });
   try {
     const roomId = await resolveMatrixRoomId(client, to);
+    return await enqueueSend(roomId, async () => {
     const cfg = getCore().config.loadConfig();
     const tableMode = getCore().channel.text.resolveMarkdownTableMode({
       cfg,
@@ -146,6 +148,7 @@ export async function sendMessageMatrix(
       messageId: lastMessageId || "unknown",
       roomId,
     };
+    }); // enqueueSend
   } finally {
     if (stopOnDone) {
       client.stop();
