fix(imessage): repair anchorless payloads before debounce

zhangguiping-xydt · 张贵萍 · commit ee15ec620f4b · 2026-05-25T01:12:40.000+08:00
Move anchor recovery ahead of live debounce classification so malformed group payloads cannot be coalesced as sender DMs before routing.
diff --git a/extensions/imessage/src/monitor.watch-subscribe-retry.test.ts b/extensions/imessage/src/monitor.watch-subscribe-retry.test.ts
@@ -32,13 +32,15 @@ function createRuntime() {
 }
 
 type MockIMessageRpcClient = IMessageRpcClient & {
-  request: ReturnType<typeof vi.fn<(method: string) => Promise<unknown>>>;
+  request: ReturnType<
+    typeof vi.fn<(method: string, params?: Record<string, unknown>) => Promise<unknown>>
+  >;
   waitForClose: ReturnType<typeof vi.fn<() => Promise<void>>>;
   stop: ReturnType<typeof vi.fn<() => Promise<void>>>;
 };
 
 function createRpcClient(overrides?: {
-  request?: (method: string) => Promise<unknown>;
+  request?: (method: string, params?: Record<string, unknown>) => Promise<unknown>;
   waitForClose?: () => Promise<void>;
 }): MockIMessageRpcClient {
   const client = {
@@ -119,6 +121,118 @@ describe("monitorIMessageProvider watch.subscribe startup retry", () => {
     ).toBe(false);
   });
 
+  it("repairs anchorless live payloads before debounce classification", async () => {
+    const runtime = createRuntime();
+    let notify: ((msg: { method: string; params?: unknown }) => void) | undefined;
+    let close!: () => void;
+    const closed = new Promise<void>((resolve) => {
+      close = resolve;
+    });
+    const calls: Array<{ method: string; params?: Record<string, unknown> }> = [];
+    const client = createRpcClient({
+      waitForClose: async () => await closed,
+      request: async (method, params) => {
+        calls.push({ method, params });
+        if (method === "watch.subscribe") {
+          return { subscription: 1 };
+        }
+        if (method === "chats.list") {
+          return { chats: [{ id: 101 }, { id: 202 }] };
+        }
+        if (method === "messages.history") {
+          if (params?.chat_id === 101) {
+            return {
+              messages: [
+                {
+                  guid: "GUID-GROUP-A",
+                  chat_id: 101,
+                  chat_guid: "iMessage;+;group-a",
+                  chat_identifier: "group-a",
+                  is_group: true,
+                  participants: ["+15550001111", "+15550002222"],
+                },
+              ],
+            };
+          }
+          if (params?.chat_id === 202) {
+            return {
+              messages: [
+                {
+                  guid: "GUID-GROUP-B",
+                  chat_id: 202,
+                  chat_guid: "iMessage;+;group-b",
+                  chat_identifier: "group-b",
+                  is_group: true,
+                  participants: ["+15550001111", "+15550003333"],
+                },
+              ],
+            };
+          }
+          return { messages: [] };
+        }
+        return {};
+      },
+    });
+
+    createIMessageRpcClientMock.mockImplementation(async (opts) => {
+      notify = opts?.onNotification;
+      return client;
+    });
+
+    const monitorPromise = monitorIMessageProvider({
+      config: {
+        channels: {
+          imessage: {
+            coalesceSameSenderDms: true,
+            groupPolicy: "open",
+            groups: { "*": { requireMention: true } },
+          },
+        },
+        messages: { groupChat: { mentionPatterns: ["@openclaw"] } },
+      } as never,
+      runtime: runtime as never,
+    });
+
+    await vi.waitFor(() => expect(notify).toBeDefined());
+
+    const baseMessage = {
+      chat_id: 0,
+      sender: "+15550001111",
+      is_from_me: false,
+      attachments: null,
+      chat_identifier: "",
+      chat_guid: "",
+      chat_name: "",
+      participants: null,
+      is_group: false,
+    };
+    notify?.({
+      method: "message",
+      params: { message: { ...baseMessage, guid: "GUID-GROUP-A", text: "first group" } },
+    });
+    notify?.({
+      method: "message",
+      params: { message: { ...baseMessage, guid: "GUID-GROUP-B", text: "second group" } },
+    });
+
+    await vi.waitFor(() =>
+      expect(
+        calls.filter((call) => call.method === "messages.history" && call.params?.chat_id === 202),
+      ).toHaveLength(1),
+    );
+
+    expect(calls.filter((call) => call.method === "chats.list")).toHaveLength(2);
+    expect(
+      calls.filter((call) => call.method === "messages.history" && call.params?.chat_id === 101),
+    ).toHaveLength(2);
+    expect(
+      calls.filter((call) => call.method === "messages.history" && call.params?.chat_id === 202),
+    ).toHaveLength(1);
+
+    close();
+    await monitorPromise;
+  });
+
   it("still fails after bounded startup retries are exhausted", async () => {
     const runtime = createRuntime();
     createIMessageRpcClientMock.mockImplementation(async () =>
diff --git a/extensions/imessage/src/monitor/monitor-provider.ts b/extensions/imessage/src/monitor/monitor-provider.ts
@@ -413,6 +413,14 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P
     }
   }
 
+  async function repairMessageConversationAnchor(message: IMessagePayload) {
+    return await repairIMessageConversationAnchor({
+      message,
+      client: getActiveClient(),
+      runtime,
+    });
+  }
+
   async function handleMessageNow(
     message: IMessagePayload,
     options: { advanceCatchupCursor?: boolean } = {},
@@ -465,11 +473,7 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P
     // (e.g. group link-preview with chat_id=0 and empty chat_guid/
     // chat_identifier). Fail-closed: if recovery cannot determine the real
     // conversation, the message is dropped rather than routed to sender DM.
-    const repaired = await repairIMessageConversationAnchor({
-      message,
-      client: getActiveClient(),
-      runtime,
-    });
+    const repaired = await repairMessageConversationAnchor(message);
     if (!repaired) {
       return;
     }
@@ -865,7 +869,11 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P
       runtime.error?.(`imessage: dropping malformed RPC message payload (keys=${shape})`);
       return;
     }
-    await inboundDebouncer.enqueue({ message });
+    const repaired = await repairMessageConversationAnchor(message);
+    if (!repaired) {
+      return;
+    }
+    await inboundDebouncer.enqueue({ message: repaired });
   };
 
   await waitForTransportReady({
