fix(googlechat): retain fallback thread continuity (#117053)

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) {

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",
             },
           }
         : {}),

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(

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<void>;
+            };
+          };
+        };
+      }) => {
+        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", () => {

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)}`);