From 3dce01fa201aed397358ecbab7404e0e140b6ea6 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 14:54:16 -0700 Subject: [PATCH] fix(googlechat): retain fallback thread continuity (#117053) --- .../googlechat/src/monitor-reply-delivery.ts | 54 ++++++- .../monitor.reply-delivery-failure.test.ts | 4 +- .../src/monitor.reply-delivery.test.ts | 141 +++++++++++++++++- extensions/googlechat/src/monitor.test.ts | 101 +++++++++++++ extensions/googlechat/src/monitor.ts | 12 +- 5 files changed, 295 insertions(+), 17 deletions(-) diff --git a/extensions/googlechat/src/monitor-reply-delivery.ts b/extensions/googlechat/src/monitor-reply-delivery.ts index e993b648b1e2..d6f72d7eba07 100644 --- a/extensions/googlechat/src/monitor-reply-delivery.ts +++ b/extensions/googlechat/src/monitor-reply-delivery.ts @@ -5,10 +5,35 @@ import type { ResolvedGoogleChatAccount } from "./accounts.js"; import { deleteGoogleChatMessage, sendGoogleChatMessage, updateGoogleChatMessage } from "./api.js"; import type { GoogleChatCoreRuntime, GoogleChatRuntimeEnv } from "./monitor-types.js"; -export type GoogleChatTypingMessage = { - name: string; - thread?: string; -}; +export type GoogleChatTypingMessage = + | { + placement: "top-level"; + name: string; + } + | { + placement: "thread"; + name: string; + requestedThreadName: string; + deliveredThreadName: string; + }; + +export function createGoogleChatTypingMessage(params: { + messageName: string; + requestedThreadName?: string; + deliveredThreadName?: string; +}): GoogleChatTypingMessage { + const name = params.messageName.trim(); + const requestedThreadName = params.requestedThreadName?.trim(); + if (!requestedThreadName) { + return { placement: "top-level", name }; + } + return { + placement: "thread", + name, + requestedThreadName, + deliveredThreadName: params.deliveredThreadName?.trim() || requestedThreadName, + }; +} export async function deliverGoogleChatReply(params: { payload: { @@ -30,12 +55,18 @@ export async function deliverGoogleChatReply(params: { // text delivery can keep retrying a dead message and drop content. let typingMessage = params.typingMessage; const replyThreadName = payload.replyToId?.trim() || undefined; - const typingMessageThreadName = typingMessage?.thread?.trim() || undefined; const reply = resolveSendableOutboundReplyParts(payload); const text = reply.text; let firstTextChunk = true; + let deliveryThreadName = replyThreadName; - if (typingMessage && typingMessageThreadName !== replyThreadName) { + const typingMatchesReply = + typingMessage?.placement === "thread" + ? typingMessage.requestedThreadName === replyThreadName + : typingMessage?.placement === "top-level" + ? replyThreadName === undefined + : false; + if (typingMessage && !typingMatchesReply) { // Typing starts before reply directives are resolved. Never edit a placeholder // from one thread into a final reply targeted at another conversation surface. try { @@ -44,6 +75,10 @@ export async function deliverGoogleChatReply(params: { runtime.error?.(`Google Chat typing cleanup failed: ${String(err)}`); } typingMessage = undefined; + } else if (typingMessage?.placement === "thread") { + // The requested thread decides whether the placeholder still belongs to this reply; + // the provider-returned thread owns every later physical send after fallback. + deliveryThreadName = typingMessage.deliveredThreadName; } if (reply.hasMedia) { @@ -75,12 +110,15 @@ export async function deliverGoogleChatReply(params: { } }; const sendTextMessage = async (chunk: string) => { - await sendGoogleChatMessage({ + const sent = await sendGoogleChatMessage({ account, space: spaceId, text: chunk, - thread: replyThreadName, + thread: deliveryThreadName, }); + if (replyThreadName) { + deliveryThreadName = sent?.threadName?.trim() || deliveryThreadName; + } }; const chunks = core.channel.text.chunkMarkdownTextWithMode(text, chunkLimit, chunkMode); for (const chunk of chunks) { diff --git a/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts b/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts index 83c7067eddea..ad564b5e8e82 100644 --- a/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts +++ b/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts @@ -159,8 +159,10 @@ async function runDelivery(params: { ...(params.withTypingMessage ? { typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, } : {}), diff --git a/extensions/googlechat/src/monitor.reply-delivery.test.ts b/extensions/googlechat/src/monitor.reply-delivery.test.ts index db639881b97b..343c82d4acc9 100644 --- a/extensions/googlechat/src/monitor.reply-delivery.test.ts +++ b/extensions/googlechat/src/monitor.reply-delivery.test.ts @@ -49,11 +49,13 @@ function createRuntime() { } satisfies GoogleChatRuntimeEnv; } +let createGoogleChatTypingMessage: typeof import("./monitor-reply-delivery.js").createGoogleChatTypingMessage; let deliverGoogleChatReply: typeof import("./monitor-reply-delivery.js").deliverGoogleChatReply; beforeEach(async () => { vi.clearAllMocks(); - ({ deliverGoogleChatReply } = await import("./monitor-reply-delivery.js")); + ({ createGoogleChatTypingMessage, deliverGoogleChatReply } = + await import("./monitor-reply-delivery.js")); }); afterAll(() => { @@ -78,8 +80,10 @@ describe("Google Chat reply delivery", () => { config, statusSink, typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, }); @@ -107,6 +111,123 @@ describe("Google Chat reply delivery", () => { ); }); + it("continues later chunks in the provider fallback thread", async () => { + const core = createCore({ chunks: ["first chunk", "second chunk"] }); + const runtime = createRuntime(); + mocks.sendGoogleChatMessage + .mockResolvedValueOnce({ + messageName: "spaces/AAA/messages/first", + threadName: "spaces/AAA/threads/fallback", + }) + .mockResolvedValueOnce({ + messageName: "spaces/AAA/messages/second", + threadName: "spaces/AAA/threads/fallback", + }); + + await deliverGoogleChatReply({ + payload: { text: "two chunks", replyToId: "spaces/AAA/threads/requested" }, + account, + spaceId: "spaces/AAA", + runtime, + core, + config, + }); + + expect(mocks.sendGoogleChatMessage).toHaveBeenNthCalledWith(1, { + account, + space: "spaces/AAA", + text: "first chunk", + thread: "spaces/AAA/threads/requested", + }); + expect(mocks.sendGoogleChatMessage).toHaveBeenNthCalledWith(2, { + account, + space: "spaces/AAA", + text: "second chunk", + thread: "spaces/AAA/threads/fallback", + }); + }); + + it("continues after a fallback typing placeholder in its delivered thread", async () => { + const core = createCore({ chunks: ["first chunk", "second chunk"] }); + const runtime = createRuntime(); + mocks.sendGoogleChatMessage.mockResolvedValueOnce({ + messageName: "spaces/AAA/messages/second", + threadName: "spaces/AAA/threads/fallback", + }); + + await deliverGoogleChatReply({ + payload: { text: "two chunks", replyToId: "spaces/AAA/threads/requested" }, + account, + spaceId: "spaces/AAA", + runtime, + core, + config, + typingMessage: createGoogleChatTypingMessage({ + messageName: "spaces/AAA/messages/typing", + requestedThreadName: "spaces/AAA/threads/requested", + deliveredThreadName: "spaces/AAA/threads/fallback", + }), + }); + + expect(mocks.updateGoogleChatMessage).toHaveBeenCalledWith({ + account, + messageName: "spaces/AAA/messages/typing", + text: "first chunk", + }); + expect(mocks.sendGoogleChatMessage).toHaveBeenCalledOnce(); + expect(mocks.sendGoogleChatMessage).toHaveBeenCalledWith({ + account, + space: "spaces/AAA", + text: "second chunk", + thread: "spaces/AAA/threads/fallback", + }); + }); + + it("keeps the requested thread when the provider omits thread metadata", async () => { + const core = createCore({ chunks: ["first chunk", "second chunk"] }); + const runtime = createRuntime(); + mocks.sendGoogleChatMessage.mockResolvedValue({ + messageName: "spaces/AAA/messages/sent", + }); + + await deliverGoogleChatReply({ + payload: { text: "two chunks", replyToId: "spaces/AAA/threads/requested" }, + account, + spaceId: "spaces/AAA", + runtime, + core, + config, + }); + + expect(mocks.sendGoogleChatMessage).toHaveBeenCalledTimes(2); + for (const call of mocks.sendGoogleChatMessage.mock.calls) { + expect(call[0]?.thread).toBe("spaces/AAA/threads/requested"); + } + }); + + it("keeps top-level chunks top-level when Google returns a thread name", async () => { + const core = createCore({ chunks: ["first chunk", "second chunk"] }); + const runtime = createRuntime(); + mocks.sendGoogleChatMessage.mockResolvedValue({ + messageName: "spaces/AAA/messages/sent", + threadName: "spaces/AAA/threads/provider-created", + }); + + await deliverGoogleChatReply({ + payload: { text: "two top-level chunks" }, + account, + spaceId: "spaces/AAA", + runtime, + core, + config, + }); + + expect(mocks.sendGoogleChatMessage).toHaveBeenCalledTimes(2); + for (const call of mocks.sendGoogleChatMessage.mock.calls) { + expect(call[0]?.thread).toBeUndefined(); + } + }); + it("rejects when a later text chunk send fails instead of dropping it silently", async () => { const core = createCore({ chunks: ["first chunk", "second chunk", "third chunk"] }); const runtime = createRuntime(); @@ -145,8 +266,10 @@ describe("Google Chat reply delivery", () => { core, config, typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, }), ).rejects.toBe(fallbackError); @@ -167,8 +290,10 @@ describe("Google Chat reply delivery", () => { core, config, typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, }); @@ -203,8 +328,10 @@ describe("Google Chat reply delivery", () => { core, config, typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, }); @@ -237,8 +364,10 @@ describe("Google Chat reply delivery", () => { core, config, typingMessage: { + placement: "thread", name: "spaces/AAA/messages/typing", - thread: "spaces/AAA/threads/root", + requestedThreadName: "spaces/AAA/threads/root", + deliveredThreadName: "spaces/AAA/threads/root", }, }), ).rejects.toThrow( diff --git a/extensions/googlechat/src/monitor.test.ts b/extensions/googlechat/src/monitor.test.ts index 7d411c66ec7c..24acc6595d93 100644 --- a/extensions/googlechat/src/monitor.test.ts +++ b/extensions/googlechat/src/monitor.test.ts @@ -8,8 +8,10 @@ import "./monitor.js"; import type { GoogleChatEvent } from "./types.js"; const apiMocks = vi.hoisted(() => ({ + deleteGoogleChatMessage: vi.fn(), downloadGoogleChatMedia: vi.fn(), sendGoogleChatMessage: vi.fn(), + updateGoogleChatMessage: vi.fn(), })); const accessMocks = vi.hoisted(() => ({ @@ -40,8 +42,10 @@ vi.mock("openclaw/plugin-sdk/channel-inbound", async (importOriginal) => { }); vi.mock("./api.js", () => ({ + deleteGoogleChatMessage: apiMocks.deleteGoogleChatMessage, downloadGoogleChatMedia: apiMocks.downloadGoogleChatMedia, sendGoogleChatMessage: apiMocks.sendGoogleChatMessage, + updateGoogleChatMessage: apiMocks.updateGoogleChatMessage, })); vi.mock("./monitor-access.js", () => ({ @@ -64,8 +68,10 @@ vi.mock("./monitor-routing.js", () => ({ })); beforeEach(() => { + apiMocks.deleteGoogleChatMessage.mockReset(); apiMocks.downloadGoogleChatMedia.mockReset(); apiMocks.sendGoogleChatMessage.mockReset(); + apiMocks.updateGoogleChatMessage.mockReset(); accessMocks.applyGoogleChatInboundAccessPolicy.mockReset(); inboundMocks.buildEnvelope.mockReset().mockImplementation(({ body }: { body: string }) => body); inboundMocks.resolveChannelInboundRouteEnvelope @@ -423,6 +429,101 @@ describe("googlechat monitor inbound space classification", () => { thread: expectedThread, }); }); + + it("continues delivery in the typing message's canonical fallback thread", async () => { + const requestedThread = "spaces/CLASSIFY/threads/requested"; + const deliveredThread = "spaces/CLASSIFY/threads/fallback"; + const runTurn = vi.fn( + async (params: { + adapter: { + resolveTurn: () => { + delivery: { + deliver: (payload: { text: string; replyToId: string }) => Promise; + }; + }; + }; + }) => { + const turn = params.adapter.resolveTurn(); + await turn.delivery.deliver({ + text: "two chunks", + replyToId: requestedThread, + }); + }, + ); + const core = { + logging: { shouldLogVerbose: () => false }, + channel: { + inbound: { + buildContext: vi.fn((payload: unknown) => payload), + run: runTurn, + }, + text: { + resolveChunkMode: vi.fn(() => "markdown"), + chunkMarkdownTextWithMode: vi.fn(() => ["first chunk", "second chunk"]), + }, + }, + } as unknown as GoogleChatCoreRuntime; + const account = { + accountId: "work", + config: { replyToMode: "all" }, + credentialSource: "inline", + } as ResolvedGoogleChatAccount; + const event = { + type: "MESSAGE", + space: { name: "spaces/CLASSIFY", spaceType: "SPACE" }, + message: { + name: "spaces/CLASSIFY/messages/1", + text: "hello", + thread: { name: requestedThread }, + sender: { name: "users/alice", displayName: "Alice", type: "HUMAN" }, + }, + } satisfies GoogleChatEvent; + + accessMocks.applyGoogleChatInboundAccessPolicy.mockResolvedValue({ + ok: true, + commandAuthorized: undefined, + effectiveWasMentioned: undefined, + groupBotLoopProtection: undefined, + groupSystemPrompt: undefined, + }); + apiMocks.sendGoogleChatMessage + .mockResolvedValueOnce({ + messageName: "spaces/CLASSIFY/messages/typing", + threadName: deliveredThread, + }) + .mockResolvedValueOnce({ + messageName: "spaces/CLASSIFY/messages/second", + threadName: deliveredThread, + }); + + await processGoogleChatTestEvent({ + event, + account, + config: {}, + runtime: { error: vi.fn(), log: vi.fn() }, + core, + mediaMaxMb: 10, + }); + + expect(apiMocks.sendGoogleChatMessage).toHaveBeenNthCalledWith(1, { + account, + space: "spaces/CLASSIFY", + text: "_OpenClaw is typing..._", + thread: requestedThread, + }); + expect(apiMocks.updateGoogleChatMessage).toHaveBeenCalledWith({ + account, + messageName: "spaces/CLASSIFY/messages/typing", + text: "first chunk", + }); + expect(apiMocks.sendGoogleChatMessage).toHaveBeenCalledTimes(2); + expect(apiMocks.sendGoogleChatMessage).toHaveBeenNthCalledWith(2, { + account, + space: "spaces/CLASSIFY", + text: "second chunk", + thread: deliveredThread, + }); + }); }); describe("googlechat monitor sender bot status", () => { diff --git a/extensions/googlechat/src/monitor.ts b/extensions/googlechat/src/monitor.ts index b137d61a3d31..1b25dbca9d8b 100644 --- a/extensions/googlechat/src/monitor.ts +++ b/extensions/googlechat/src/monitor.ts @@ -20,7 +20,11 @@ import { createGoogleChatIngressMonitor, type GoogleChatIngressLifecycle, } from "./monitor-ingress.js"; -import { deliverGoogleChatReply, type GoogleChatTypingMessage } from "./monitor-reply-delivery.js"; +import { + createGoogleChatTypingMessage, + deliverGoogleChatReply, + type GoogleChatTypingMessage, +} from "./monitor-reply-delivery.js"; import { registerGoogleChatWebhookTarget, setGoogleChatWebhookEventProcessor, @@ -380,7 +384,11 @@ async function processMessageWithPipeline(params: { thread: typingMessageThreadName, }); if (result?.messageName) { - typingMessage = { name: result.messageName, thread: typingMessageThreadName }; + typingMessage = createGoogleChatTypingMessage({ + messageName: result.messageName, + requestedThreadName: typingMessageThreadName, + deliveredThreadName: result.threadName, + }); } } catch (err) { runtime.error?.(`Failed sending typing message: ${String(err)}`);