From 6d5b25e24bae617871edb98afe073a86287aeabc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 14:02:27 -0700 Subject: [PATCH] refactor(imessage): deduplicate monitor route test fixtures (#117621) --- .../imessage/src/monitor.last-route.test.ts | 1339 +++++------------ 1 file changed, 381 insertions(+), 958 deletions(-) diff --git a/extensions/imessage/src/monitor.last-route.test.ts b/extensions/imessage/src/monitor.last-route.test.ts index 96a3ada3a1d9..d0e7024d9d90 100644 --- a/extensions/imessage/src/monitor.last-route.test.ts +++ b/extensions/imessage/src/monitor.last-route.test.ts @@ -17,6 +17,7 @@ import { loadIMessageRecoveryCursor, resolveIMessageRecoveryCursorDbIdentity, } from "./monitor/recovery-cursor.js"; +import type { IMessagePayload, MonitorIMessageOpts } from "./monitor/types.js"; import { getCachedIMessagePrivateApiStatus, setCachedIMessagePrivateApiStatus, @@ -177,13 +178,120 @@ describe("iMessage monitor last-route updates", () => { } }); - it("keeps native typing alive when tool activity arrives before reply text", async () => { + function setAvailablePrivateApiMethods(rpcMethods: string[]): void { setCachedIMessagePrivateApiStatus("imsg", { available: true, v2Ready: true, selectors: {}, - rpcMethods: ["watch.subscribe", "send", "typing"], + rpcMethods, }); + } + + function createInboundMessage( + message: Pick & + Partial< + Pick + >, + ): IMessagePayload { + return { + id: message.id, + guid: message.guid, + chat_id: message.chat_id ?? 123, + sender: message.sender ?? "+15550001111", + is_from_me: message.is_from_me ?? false, + text: message.text, + is_group: message.is_group ?? false, + created_at: message.created_at ?? new Date().toISOString(), + }; + } + + function createIMessageWatchClient( + params: { + request?: (method: string, params?: Record) => Promise; + message?: () => IMessagePayload; + onClose?: (notify: (message: IMessagePayload) => void) => Promise; + } = {}, + ) { + let onNotification: + | NonNullable[0]>["onNotification"]> + | undefined; + const notify = (message: IMessagePayload) => { + onNotification?.({ method: "message", params: { message } }); + }; + const message = params.message; + const onClose = + params.onClose ?? + (message + ? async (notifyMessage: (message: IMessagePayload) => void) => { + notifyMessage(message()); + await Promise.resolve(); + await Promise.resolve(); + } + : undefined); + const client = { + request: vi.fn(params.request ?? (async () => ({ subscription: 1 }))), + waitForClose: vi.fn(() => onClose?.(notify) ?? Promise.resolve()), + stop: vi.fn(async () => {}), + }; + createIMessageRpcClientMock.mockImplementation(async (clientParams) => { + if (!clientParams?.onNotification) { + throw new Error("expected iMessage notification handler"); + } + onNotification = clientParams.onNotification; + return client as never; + }); + return client; + } + + async function runIMessageMonitor( + params: { + accountId?: string; + imessage?: Record; + session?: Record; + messages?: Record; + agents?: Record; + runtime?: MonitorIMessageOpts["runtime"]; + allowlist?: boolean; + } = {}, + ): Promise { + await monitorIMessageProvider({ + ...(params.accountId ? { accountId: params.accountId } : {}), + config: { + channels: { + imessage: { + ...(params.allowlist === false + ? {} + : { dmPolicy: "allowlist", allowFrom: ["+15550001111"] }), + ...params.imessage, + }, + }, + messages: { inbound: { debounceMs: 0 }, ...params.messages }, + ...(params.agents ? { agents: params.agents } : {}), + session: { mainKey: "main", ...params.session }, + } as never, + runtime: params.runtime ?? { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + }); + } + + function createTestStateDir(prefix: string): string { + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + tempDirs.push(stateDir); + return stateDir; + } + + async function seedChatDb(dbPath: string, label = "boundary"): Promise { + const { DatabaseSync } = await import("node:sqlite"); + const database = new DatabaseSync(dbPath); + try { + database.exec("CREATE TABLE message (text TEXT);"); + database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, label); + } finally { + database.close(); + } + } + + it("keeps native typing alive when tool activity arrives before reply text", async () => { + setAvailablePrivateApiMethods(["watch.subscribe", "send", "typing"]); dispatchReplyWithBufferedBlockDispatcherMock.mockImplementationOnce(async (params) => { expect(params.replyOptions?.suppressDefaultToolProgressMessages).toBe(true); expect(params.replyOptions?.allowProgressCallbacksWhenSourceDeliverySuppressed).toBe(true); @@ -238,9 +346,8 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + const client = createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -248,50 +355,16 @@ describe("iMessage monitor last-route updates", () => { return { ok: true }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 7, - guid: "typing-keepalive-guid-7", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "run a long script", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 7, + guid: "typing-keepalive-guid-7", + text: "run a long script", + }), }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { sendReadReceipts: false } }); await vi.waitFor(() => { expect(client.request).toHaveBeenCalledWith( @@ -310,12 +383,7 @@ describe("iMessage monitor last-route updates", () => { }); it("keeps direct progress options when imsg lacks native typing support", async () => { - setCachedIMessagePrivateApiStatus("imsg", { - available: true, - v2Ready: true, - selectors: {}, - rpcMethods: ["watch.subscribe", "send", "read"], - }); + setAvailablePrivateApiMethods(["watch.subscribe", "send", "read"]); dispatchReplyWithBufferedBlockDispatcherMock.mockImplementationOnce(async (params) => { expect(params.replyOptions?.suppressDefaultToolProgressMessages).toBe(true); expect(params.replyOptions?.allowProgressCallbacksWhenSourceDeliverySuppressed).toBe(true); @@ -329,9 +397,8 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + const client = createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -339,50 +406,16 @@ describe("iMessage monitor last-route updates", () => { throw new Error("typing should not start without native typing support"); } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 13, - guid: "typing-unsupported-guid-13", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "run a long script without native typing", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 13, + guid: "typing-unsupported-guid-13", + text: "run a long script without native typing", + }), }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { sendReadReceipts: false } }); await vi.waitFor(() => { expect(dispatchReplyWithBufferedBlockDispatcherMock).toHaveBeenCalledTimes(1); @@ -396,12 +429,7 @@ describe("iMessage monitor last-route updates", () => { }); it("starts direct typing before dispatching the inbound turn", async () => { - setCachedIMessagePrivateApiStatus("imsg", { - available: true, - v2Ready: true, - selectors: {}, - rpcMethods: ["watch.subscribe", "send", "typing"], - }); + setAvailablePrivateApiMethods(["watch.subscribe", "send", "typing"]); let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; const earlyTypingClient = { @@ -427,16 +455,11 @@ describe("iMessage monitor last-route updates", () => { onNotification?.({ method: "message", params: { - message: { + message: createInboundMessage({ id: 12, guid: "typing-early-guid-12", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, text: "respond after a slow context build", - is_group: false, - created_at: new Date().toISOString(), - }, + }), }, }); await vi.waitFor(() => { @@ -466,20 +489,7 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { sendReadReceipts: false } }); expect(watchClient.request).not.toHaveBeenCalledWith( "typing", @@ -498,12 +508,7 @@ describe("iMessage monitor last-route updates", () => { it.each(["never", "message", "thinking"] as const)( "does not start direct tool typing when typingMode is %s", async (typingMode) => { - setCachedIMessagePrivateApiStatus("imsg", { - available: true, - v2Ready: true, - selectors: {}, - rpcMethods: ["watch.subscribe", "send", "typing"], - }); + setAvailablePrivateApiMethods(["watch.subscribe", "send", "typing"]); dispatchReplyWithBufferedBlockDispatcherMock.mockImplementationOnce(async (params) => { expect(params.replyOptions?.suppressDefaultToolProgressMessages).toBeUndefined(); expect( @@ -513,9 +518,8 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + const client = createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -523,50 +527,18 @@ describe("iMessage monitor last-route updates", () => { throw new Error("typing should not start from tool activity"); } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 8, - guid: `typing-mode-${typingMode}-guid-8`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "run a long script", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 8, + guid: `typing-mode-${typingMode}-guid-8`, + text: "run a long script", + }), }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - agents: { defaults: { typingMode } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { sendReadReceipts: false }, + agents: { defaults: { typingMode } }, }); await vi.waitFor(() => { @@ -581,12 +553,7 @@ describe("iMessage monitor last-route updates", () => { ); it("does not start direct tool typing when sendPolicy denies source delivery", async () => { - setCachedIMessagePrivateApiStatus("imsg", { - available: true, - v2Ready: true, - selectors: {}, - rpcMethods: ["watch.subscribe", "send", "typing"], - }); + setAvailablePrivateApiMethods(["watch.subscribe", "send", "typing"]); dispatchReplyWithBufferedBlockDispatcherMock.mockImplementationOnce(async (params) => { expect(params.replyOptions?.suppressDefaultToolProgressMessages).toBeUndefined(); expect( @@ -596,9 +563,8 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + const client = createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -606,49 +572,18 @@ describe("iMessage monitor last-route updates", () => { throw new Error("typing should not start under sendPolicy deny"); } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 9, - guid: "send-policy-guid-9", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "run a long script", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 9, + guid: "send-policy-guid-9", + text: "run a long script", + }), }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main", sendPolicy: { default: "deny" } }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { sendReadReceipts: false }, + session: { sendPolicy: { default: "deny" } }, }); await vi.waitFor(() => { @@ -662,12 +597,7 @@ describe("iMessage monitor last-route updates", () => { }); it("does not wait for read receipts before dispatching the inbound turn", async () => { - setCachedIMessagePrivateApiStatus("imsg", { - available: true, - v2Ready: true, - selectors: {}, - rpcMethods: ["watch.subscribe", "read"], - }); + setAvailablePrivateApiMethods(["watch.subscribe", "read"]); let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; const readClient = { @@ -690,16 +620,11 @@ describe("iMessage monitor last-route updates", () => { onNotification?.({ method: "message", params: { - message: { + message: createInboundMessage({ id: 11, guid: "read-receipt-guid-11", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, text: "respond without waiting for read receipt", - is_group: false, - created_at: new Date().toISOString(), - }, + }), }, }); await vi.waitFor(() => { @@ -716,19 +641,7 @@ describe("iMessage monitor last-route updates", () => { return readClient as never; }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor(); expect(readClient.request).toHaveBeenCalledWith( "read", @@ -763,58 +676,22 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 10, - guid: `block-streaming-${label}-guid-10`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "stream blocks before the final", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 10, + guid: `block-streaming-${label}-guid-10`, + text: "stream blocks before the final", + }), }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - ...imessagePatch, - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { sendReadReceipts: false, ...imessagePatch } }); await vi.waitFor(() => { expect(dispatchReplyWithBufferedBlockDispatcherMock).toHaveBeenCalledTimes(1); @@ -843,63 +720,32 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 11, - guid: `account-block-streaming-${label}-guid-11`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "stream blocks before the final", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 11, + guid: `account-block-streaming-${label}-guid-11`, + text: "stream blocks before the final", + }), }); - await monitorIMessageProvider({ + await runIMessageMonitor({ accountId: "personal", - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - streaming: { block: { enabled: channelBlockEnabled } }, - accounts: { - personal: { - streaming: { block: { enabled: accountBlockEnabled } }, - }, - }, + imessage: { + sendReadReceipts: false, + streaming: { block: { enabled: channelBlockEnabled } }, + accounts: { + personal: { + streaming: { block: { enabled: accountBlockEnabled } }, }, }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + }, }); await vi.waitFor(() => { @@ -925,63 +771,32 @@ describe("iMessage monitor last-route updates", () => { return { queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } } as const; }); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 11, - guid: `account-streaming-${label}-guid-11`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "stream blocks before the final", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => + createInboundMessage({ + id: 11, + guid: `account-streaming-${label}-guid-11`, + text: "stream blocks before the final", + }), }); - await monitorIMessageProvider({ + await runIMessageMonitor({ accountId: "personal", - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - sendReadReceipts: false, - streaming: { block: { enabled: true } }, - accounts: { - personal: { - streaming: accountStreaming, - }, - }, + imessage: { + sendReadReceipts: false, + streaming: { block: { enabled: true } }, + accounts: { + personal: { + streaming: accountStreaming, }, }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + }, }); await vi.waitFor(() => { @@ -991,50 +806,22 @@ describe("iMessage monitor last-route updates", () => { ); it("keeps per-channel-peer direct-message last-route writes on the isolated session", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-last-route-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-last-route-"); const configuredStore = path.join(stateDir, "sessions.json"); const storePath = resolveStorePath(configuredStore, { agentId: "main" }); const sessionKey = "agent:main:imessage:direct:+15550001111"; const runtimeErrorMock = vi.fn(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 1, - guid: "last-route-guid-1", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "hello from imessage", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + createIMessageWatchClient({ + message: () => + createInboundMessage({ + id: 1, + guid: "last-route-guid-1", + text: "hello from imessage", + }), }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { dmScope: "per-channel-peer", mainKey: "main", store: configuredStore }, - } as never, + await runIMessageMonitor({ + session: { dmScope: "per-channel-peer", store: configuredStore }, runtime: { error: runtimeErrorMock, exit: vi.fn(), log: vi.fn() }, }); @@ -1068,71 +855,38 @@ describe("iMessage monitor last-route updates", () => { const staleCreatedAt = new Date(Date.now() - 60 * 60 * 1000).toISOString(); const freshCreatedAt = new Date().toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { + const client = createIMessageWatchClient({ + onClose: async (notify) => { // Stale backlog row (old send date) Apple delivered after a recovery — // must be suppressed by the age fence. - onNotification?.({ - method: "message", - params: { - message: { - id: 2023, - guid: "OLD-GUID-2023", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "old backlog row", - is_group: false, - created_at: staleCreatedAt, - }, - }, - }); + notify( + createInboundMessage({ + id: 2023, + guid: "OLD-GUID-2023", + text: "old backlog row", + created_at: staleCreatedAt, + }), + ); // Fresh live row — must dispatch. - onNotification?.({ - method: "message", - params: { - message: { - id: 3001, - guid: "LIVE-GUID-2026", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "current row", - is_group: false, - created_at: freshCreatedAt, - }, - }, - }); + notify( + createInboundMessage({ + id: 3001, + guid: "LIVE-GUID-2026", + text: "current row", + created_at: freshCreatedAt, + }), + ); await Promise.resolve(); await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - // Unreadable dbPath => no startup rowid watermark, so this test - // isolates the age-fence behavior on the live path. - dbPath: path.join(os.tmpdir(), `openclaw-missing-chat-${Date.now()}.db`), - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { + // Unreadable dbPath => no startup rowid watermark, so this test + // isolates the age-fence behavior on the live path. + dbPath: path.join(os.tmpdir(), `openclaw-missing-chat-${Date.now()}.db`), + }, }); // No readable db => watch.subscribe carries no since_rowid; the age fence @@ -1152,32 +906,12 @@ describe("iMessage monitor last-route updates", () => { // Regression guard: the watermark is captured before the transport-ready // probe so messages that land during the startup window are not skipped by // imsg's self-fence at subscribe time. - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-startup-rowid-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-startup-rowid-"); const dbPath = path.join(stateDir, "chat.db"); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "watermark"); - } finally { - database.close(); - } - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => {}), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async () => client as never); + await seedChatDb(dbPath, "watermark"); + const client = createIMessageWatchClient(); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); expect(client.request).toHaveBeenCalledWith( "watch.subscribe", @@ -1192,28 +926,14 @@ describe("iMessage monitor last-route updates", () => { resolveIMessageRecoveryCursorDbIdentity({ remoteHost: "user@gateway-host" }), 4990, ); - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => {}), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async () => client as never); + const client = createIMessageWatchClient(); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - // remoteHost set => no local chat.db boundary; recovery must still - // drive since_rowid from the persisted cursor over the RPC client. - remoteHost: "user@gateway-host", - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { + // remoteHost set => no local chat.db boundary; recovery must still + // drive since_rowid from the persisted cursor over the RPC client. + remoteHost: "user@gateway-host", + }, }); expect(client.request).toHaveBeenCalledWith( @@ -1224,31 +944,18 @@ describe("iMessage monitor last-route updates", () => { }); it("routes legacy catchup through durable ingress and rejects a live GUID overlap", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-catchup-window-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-catchup-window-"); const dbPath = path.join(stateDir, "chat.db"); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); const createdAt = new Date().toISOString(); - const historyMessage = { + const historyMessage = createInboundMessage({ id: 4995, guid: "CATCHUP-LIVE-OVERLAP-GUID", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, text: "caught up exactly once", - is_group: false, created_at: createdAt, - }; - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string) => { + }); + const client = createIMessageWatchClient({ + request: async (method: string) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -1259,37 +966,14 @@ describe("iMessage monitor last-route updates", () => { return { messages: [historyMessage] }; } throw new Error(`unexpected request ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { message: { ...historyMessage, id: 5001 } }, - }); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + onClose: async (notify) => { + notify({ ...historyMessage, id: 5001 }); + }, }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dbPath, - catchup: { enabled: true, perRunLimit: 25, maxAgeMinutes: 60 }, - dmPolicy: "allowlist", - allowFrom: ["+15550001111"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { dbPath, catchup: { enabled: true, perRunLimit: 25, maxAgeMinutes: 60 } }, }); expect(client.request).toHaveBeenCalledWith( @@ -1308,84 +992,45 @@ describe("iMessage monitor last-route updates", () => { }); it("recovers downtime messages: replays from the cursor and delivers replay rows older than the live fence", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-recovery-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-recovery-"); const dbPath = path.join(stateDir, "chat.db"); advanceIMessageRecoveryCursor( "default", resolveIMessageRecoveryCursorDbIdentity({ dbPath }), 4990, ); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); // 30 min old: inside the 2h recovery window, outside the 15min live fence. const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { + const client = createIMessageWatchClient({ + onClose: async (notify) => { // Recovery replay row (rowid <= boundary 5000): missed during downtime, // delivered despite being 30min old. - onNotification?.({ - method: "message", - params: { - message: { - id: 4995, - guid: "RECOVERY-GUID-4995", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "missed during downtime", - is_group: false, - created_at: thirtyMinAgo, - }, - }, - }); + notify( + createInboundMessage({ + id: 4995, + guid: "RECOVERY-GUID-4995", + text: "missed during downtime", + created_at: thirtyMinAgo, + }), + ); // Live row (rowid > boundary) with the same old date: this is the // #89237 Push-flush backlog shape, suppressed at the live fence. - onNotification?.({ - method: "message", - params: { - message: { - id: 5001, - guid: "LIVE-OLD-GUID-5001", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "live backlog bomb", - is_group: false, - created_at: thirtyMinAgo, - }, - }, - }); + notify( + createInboundMessage({ + id: 5001, + guid: "LIVE-OLD-GUID-5001", + text: "live backlog bomb", + created_at: thirtyMinAgo, + }), + ); await Promise.resolve(); await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); // since_rowid replays from the persisted cursor, not the boundary. expect(client.request).toHaveBeenCalledWith( @@ -1400,59 +1045,22 @@ describe("iMessage monitor last-route updates", () => { }); it("does not treat startup-boundary rows as recovery replay without a prior cursor", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-first-run-boundary-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-first-run-boundary-"); const dbPath = path.join(stateDir, "chat.db"); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 4995, - guid: "FIRST-RUN-HISTORY-GUID-4995", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "already existed before first monitor start", - is_group: false, - created_at: thirtyMinAgo, - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + const client = createIMessageWatchClient({ + message: () => + createInboundMessage({ + id: 4995, + guid: "FIRST-RUN-HISTORY-GUID-4995", + text: "already existed before first monitor start", + created_at: thirtyMinAgo, + }), }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); expect(client.request).toHaveBeenCalledWith( "watch.subscribe", @@ -1465,78 +1073,38 @@ describe("iMessage monitor last-route updates", () => { }); it("records a suppressed live row so a later replay of the same row is deduped, not delivered", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-suppress-record-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-suppress-record-"); const dbPath = path.join(stateDir, "chat.db"); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { + createIMessageWatchClient({ + onClose: async (notify) => { // Live row (rowid > boundary), 30min old -> suppressed by the live fence // AND recorded in the dedupe. - onNotification?.({ - method: "message", - params: { - message: { - id: 5001, - guid: "SUPPRESSED-GUID", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "stale live backlog", - is_group: false, - created_at: thirtyMinAgo, - }, - }, - }); + notify( + createInboundMessage({ + id: 5001, + guid: "SUPPRESSED-GUID", + text: "stale live backlog", + created_at: thirtyMinAgo, + }), + ); // Same GUID re-emitted fresh (as a restart replay would): must be // dropped as a duplicate, not delivered under the recovery window. - onNotification?.({ - method: "message", - params: { - message: { - id: 5001, - guid: "SUPPRESSED-GUID", - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: "stale live backlog", - is_group: false, - created_at: new Date().toISOString(), - }, - }, - }); + notify( + createInboundMessage({ + id: 5001, + guid: "SUPPRESSED-GUID", + text: "stale live backlog", + }), + ); await Promise.resolve(); await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); await Promise.resolve(); await Promise.resolve(); @@ -1545,64 +1113,34 @@ describe("iMessage monitor last-route updates", () => { it("advances the recovery cursor after durable enqueue before dispatch", async () => { debouncerControl.holdEntries = true; - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-recovery-failed-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-recovery-failed-"); const dbPath = path.join(stateDir, "chat.db"); advanceIMessageRecoveryCursor( "default", resolveIMessageRecoveryCursorDbIdentity({ dbPath }), 4990, ); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { + const client = createIMessageWatchClient({ + onClose: async (notify) => { for (const id of [4995, 4996]) { - onNotification?.({ - method: "message", - params: { - message: { - id, - guid: `FAILED-REPLAY-GUID-${id}`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: `missed during downtime ${id}`, - is_group: false, - created_at: thirtyMinAgo, - }, - }, + notify({ + id, + guid: `FAILED-REPLAY-GUID-${id}`, + chat_id: 123, + sender: "+15550001111", + is_from_me: false, + text: `missed during downtime ${id}`, + is_group: false, + created_at: thirtyMinAgo, }); } - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); expect(client.request).toHaveBeenCalledWith( "watch.subscribe", @@ -1619,64 +1157,34 @@ describe("iMessage monitor last-route updates", () => { it("keeps the durable recovery cursor independent of later dispatch order", async () => { debouncerControl.holdEntries = true; - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-imsg-recovery-ordered-")); - tempDirs.push(stateDir); + const stateDir = createTestStateDir("openclaw-imsg-recovery-ordered-"); const dbPath = path.join(stateDir, "chat.db"); advanceIMessageRecoveryCursor( "default", resolveIMessageRecoveryCursorDbIdentity({ dbPath }), 4990, ); - const { DatabaseSync } = await import("node:sqlite"); - const database = new DatabaseSync(dbPath); - try { - database.exec("CREATE TABLE message (text TEXT);"); - database.prepare("INSERT INTO message(rowid, text) VALUES (?, ?)").run(5000, "boundary"); - } finally { - database.close(); - } + await seedChatDb(dbPath); const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async () => ({ subscription: 1 })), - waitForClose: vi.fn(async () => { + createIMessageWatchClient({ + onClose: async (notify) => { for (const id of [4995, 4996]) { - onNotification?.({ - method: "message", - params: { - message: { - id, - guid: `OUT-OF-ORDER-REPLAY-GUID-${id}`, - chat_id: 123, - sender: "+15550001111", - is_from_me: false, - text: `missed during downtime ${id}`, - is_group: false, - created_at: thirtyMinAgo, - }, - }, + notify({ + id, + guid: `OUT-OF-ORDER-REPLAY-GUID-${id}`, + chat_id: 123, + sender: "+15550001111", + is_from_me: false, + text: `missed during downtime ${id}`, + is_group: false, + created_at: thirtyMinAgo, }); } - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, }); - await monitorIMessageProvider({ - config: { - channels: { imessage: { dbPath, dmPolicy: "allowlist", allowFrom: ["+15550001111"] } }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { dbPath } }); await vi.waitFor(() => { expect(debouncerControl.entries).toHaveLength(2); @@ -1694,9 +1202,8 @@ describe("iMessage monitor last-route updates", () => { }), ); - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string, params?: Record) => { + createIMessageWatchClient({ + request: async (method: string, params?: Record) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -1724,55 +1231,27 @@ describe("iMessage monitor last-route updates", () => { }; } throw new Error(`unexpected imsg method ${method}`); + }, + message: () => ({ + id: 9500, + guid: "ANCHORLESS-GROUP-GUID", + chat_id: 0, + sender: "+15550001111", + is_from_me: false, + text: "@openclaw check this https://example.com", + is_group: false, + chat_guid: "", + chat_identifier: "", + chat_name: "", + participants: null, + created_at: new Date().toISOString(), }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { - message: { - id: 9500, - guid: "ANCHORLESS-GROUP-GUID", - chat_id: 0, - sender: "+15550001111", - is_from_me: false, - text: "@openclaw check this https://example.com", - is_group: false, - chat_guid: "", - chat_identifier: "", - chat_name: "", - participants: null, - created_at: new Date().toISOString(), - }, - }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - groupPolicy: "open", - groups: { "*": { requireMention: true } }, - }, - }, - messages: { - groupChat: { mentionPatterns: ["@openclaw"] }, - inbound: { debounceMs: 0 }, - }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, + await runIMessageMonitor({ + imessage: { groupPolicy: "open", groups: { "*": { requireMention: true } } }, + messages: { groupChat: { mentionPatterns: ["@openclaw"] } }, + allowlist: false, }); await vi.waitFor(() => { @@ -1815,9 +1294,8 @@ describe("iMessage monitor last-route updates", () => { service: "iMessage", }; - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string, params?: Record) => { + createIMessageWatchClient({ + request: async (method: string, params?: Record) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -1829,38 +1307,11 @@ describe("iMessage monitor last-route updates", () => { return { messages: [authoritativeHistory] }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { message: anchorlessNotification }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => anchorlessNotification, }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550000002"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime: { error: vi.fn(), exit: vi.fn(), log: vi.fn() }, - }); + await runIMessageMonitor({ imessage: { allowFrom: ["+15550000002"] } }); await vi.waitFor(() => { expect(dispatchReplyWithBufferedBlockDispatcherMock).toHaveBeenCalledTimes(1); @@ -1909,9 +1360,8 @@ describe("iMessage monitor last-route updates", () => { service: "iMessage", }; - let onNotification: ((message: { method: string; params: unknown }) => void) | undefined; - const client = { - request: vi.fn(async (method: string, params?: Record) => { + createIMessageWatchClient({ + request: async (method: string, params?: Record) => { if (method === "watch.subscribe") { return { subscription: 1 }; } @@ -1923,38 +1373,11 @@ describe("iMessage monitor last-route updates", () => { return { messages: [authoritativeHistory] }; } throw new Error(`unexpected imsg method ${method}`); - }), - waitForClose: vi.fn(async () => { - onNotification?.({ - method: "message", - params: { message: anchorlessNotification }, - }); - await Promise.resolve(); - await Promise.resolve(); - }), - stop: vi.fn(async () => {}), - }; - createIMessageRpcClientMock.mockImplementation(async (params) => { - if (!params?.onNotification) { - throw new Error("expected iMessage notification handler"); - } - onNotification = params.onNotification; - return client as never; + }, + message: () => anchorlessNotification, }); - await monitorIMessageProvider({ - config: { - channels: { - imessage: { - dmPolicy: "allowlist", - allowFrom: ["+15550000002"], - }, - }, - messages: { inbound: { debounceMs: 0 } }, - session: { mainKey: "main" }, - } as never, - runtime, - }); + await runIMessageMonitor({ imessage: { allowFrom: ["+15550000002"] }, runtime }); await vi.waitFor(() => { expect(runtime.error).toHaveBeenCalled();