From 343850d0df81e4b4dda0369d85e7278aa4579974 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 12 Aug 2026 11:34:59 -0700 Subject: [PATCH] fix(googlechat): preserve partial delivery evidence (#122760) --- .../googlechat/src/monitor-reply-delivery.ts | 45 +++++++++++------- .../monitor.reply-delivery-failure.test.ts | 47 +++++++++++++++++-- .../src/monitor.reply-delivery.test.ts | 24 +--------- extensions/googlechat/src/monitor.test.ts | 4 +- 4 files changed, 76 insertions(+), 44 deletions(-) diff --git a/extensions/googlechat/src/monitor-reply-delivery.ts b/extensions/googlechat/src/monitor-reply-delivery.ts index 12ded3e7583e..16330672aa6c 100644 --- a/extensions/googlechat/src/monitor-reply-delivery.ts +++ b/extensions/googlechat/src/monitor-reply-delivery.ts @@ -1,4 +1,5 @@ // Googlechat plugin module implements monitor reply delivery behavior. +import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound"; import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload"; import type { OpenClawConfig } from "../runtime-api.js"; @@ -62,9 +63,19 @@ export async function deliverGoogleChatReply(params: { let typingMessage = params.typingMessage; const replyThreadName = payload.replyToId?.trim() || undefined; const reply = resolveSendableOutboundReplyParts(payload); - const text = reply.text; - let firstTextChunk = true; let deliveryThreadName = replyThreadName; + const acceptedText: Array<{ id?: string; text: string }> = []; + const runTextOperation = async (operation: Promise): Promise => + await operation.catch((error: unknown) => { + if (acceptedText.length === 0) { + throw error; + } + throw createChannelPartialDeliveryError(error, { + messageIds: acceptedText.flatMap(({ id }) => (id ? [id] : [])), + content: acceptedText.map(({ text }) => text).join("\n"), + visibleReplySent: true, + }); + }); const typingMatchesReply = typingMessage?.placement === "thread" @@ -119,44 +130,46 @@ export async function deliverGoogleChatReply(params: { } }; const sendTextMessage = async (chunk: string) => { - const sent = await sendGoogleChatMessage({ - account, - space: spaceId, - text: chunk, - thread: deliveryThreadName, - }); + const sent = await runTextOperation( + sendGoogleChatMessage({ + account, + space: spaceId, + text: chunk, + thread: deliveryThreadName, + }), + ); + if (sent) { + acceptedText.push({ id: sent.messageName?.trim() || undefined, text: chunk }); + } if (replyThreadName) { deliveryThreadName = sent?.threadName?.trim() || deliveryThreadName; } }; - const chunks = core.channel.text.chunkMarkdownTextWithMode(text, chunkLimit, chunkMode); + const chunks = core.channel.text.chunkMarkdownTextWithMode(reply.text, chunkLimit, chunkMode); for (const chunk of chunks) { if (!chunk) { continue; } - if (firstTextChunk && typingMessage) { + if (typingMessage) { try { - await updateGoogleChatMessage({ + const updated = await updateGoogleChatMessage({ account, messageName: typingMessage.name, text: chunk, }); + acceptedText.push({ id: updated.messageName?.trim() || typingMessage.name, text: chunk }); } catch (error) { if (!(error instanceof GoogleChatApiError) || error.status !== 404) { throw error; } runtime.error?.(`Google Chat typing update failed: ${String(error)}`); - typingMessage = undefined; await sendTextMessage(chunk); } - firstTextChunk = false; + typingMessage = undefined; recordOutboundStatus(); continue; } - // Core delivery contract: a failed send must reject so the reply dispatcher - // routes to onError instead of recording a dropped chunk as delivered. await sendTextMessage(chunk); - firstTextChunk = false; recordOutboundStatus(); } } diff --git a/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts b/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts index ad564b5e8e82..4df00af15593 100644 --- a/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts +++ b/extensions/googlechat/src/monitor.reply-delivery-failure.test.ts @@ -7,6 +7,7 @@ // The service account key is a throwaway RSA key generated in-process; no real // credentials or network access are involved. import { generateKeyPairSync } from "node:crypto"; +import { isChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound"; import { withServer } from "openclaw/plugin-sdk/test-env"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../runtime-api.js"; @@ -39,7 +40,7 @@ const CHUNKS = [ "First chunk of the assistant reply.", "Second chunk of the assistant reply.", "Third chunk of the assistant reply.", -]; +] as const; const core = { channel: { @@ -92,9 +93,13 @@ function createStubHandler(params: { failCreateIndexes: Set; patchStatus const messageMatch = url.pathname.match(/^\/v1\/(spaces\/[^/]+\/messages\/[^/]+)$/); if (req.method === "PATCH" && messageMatch?.[1]) { patchAttempts.push(messageMatch[1]); - json(params.patchStatus ?? 200, { - error: { code: 404, message: "stub: message not found", status: "NOT_FOUND" }, - }); + const status = params.patchStatus ?? 200; + json( + status, + status === 200 + ? { name: messageMatch[1] } + : { error: { code: status, message: "stub: message not found", status: "NOT_FOUND" } }, + ); return; } json(400, { error: { code: 400, message: "stub: unhandled request" } }); @@ -174,6 +179,17 @@ async function runDelivery(params: { } } +function expectPartialDelivery( + error: unknown, + deliveryResult: { messageIds: string[]; content: string; visibleReplySent: true }, +) { + expect(isChannelPartialDeliveryError(error)).toBe(true); + if (!isChannelPartialDeliveryError(error)) { + throw new Error("expected partial delivery error"); + } + expect(error.deliveryResult).toEqual(deliveryResult); +} + describe("Google Chat reply delivery failure propagation (integration)", () => { let fetchControl: ReturnType; @@ -195,6 +211,11 @@ describe("Google Chat reply delivery failure propagation (integration)", () => { expect(result.deliverError).toBeInstanceOf(Error); expect((result.deliverError as Error).message).toContain("Google Chat API 500"); expect((result.deliverError as Error).message).toContain("stub: backend unavailable"); + expectPartialDelivery(result.deliverError, { + messageIds: ["spaces/AAA/messages/stub-m1"], + content: CHUNKS[0], + visibleReplySent: true, + }); expect(result.onErrorCalls).toHaveLength(1); // The failing create rejects the whole delivery: the third chunk is never attempted. expect(stub.createAttempts.map((attempt) => attempt.status)).toEqual([200, 500]); @@ -213,6 +234,23 @@ describe("Google Chat reply delivery failure propagation (integration)", () => { }); }); + it("preserves a successful typing-placeholder update when a later create fails", async () => { + const stub = createStubHandler({ failCreateIndexes: new Set([1]) }); + await withServer(stub.handler, async (baseUrl) => { + fetchControl.pointAtStub(baseUrl); + const result = await runDelivery({ withTypingMessage: true }); + + expect(result.outcome).toBe("failed-deliver"); + expectPartialDelivery(result.deliverError, { + messageIds: ["spaces/AAA/messages/typing"], + content: CHUNKS[0], + visibleReplySent: true, + }); + expect(stub.patchAttempts).toEqual(["spaces/AAA/messages/typing"]); + expect(stub.createAttempts.map((attempt) => attempt.status)).toEqual([500]); + }); + }); + it("rejects when the resend after a typing-placeholder update failure also fails", async () => { const stub = createStubHandler({ failCreateIndexes: new Set([1]), patchStatus: 404 }); await withServer(stub.handler, async (baseUrl) => { @@ -220,6 +258,7 @@ describe("Google Chat reply delivery failure propagation (integration)", () => { const result = await runDelivery({ withTypingMessage: true }); expect(result.outcome).toBe("failed-deliver"); + expect(isChannelPartialDeliveryError(result.deliverError)).toBe(false); expect((result.deliverError as Error).message).toContain("Google Chat API 500"); expect(result.onErrorCalls).toHaveLength(1); expect(stub.patchAttempts).toEqual(["spaces/AAA/messages/typing"]); diff --git a/extensions/googlechat/src/monitor.reply-delivery.test.ts b/extensions/googlechat/src/monitor.reply-delivery.test.ts index b4dc2fffd265..cacd24802956 100644 --- a/extensions/googlechat/src/monitor.reply-delivery.test.ts +++ b/extensions/googlechat/src/monitor.reply-delivery.test.ts @@ -57,6 +57,8 @@ let deliverGoogleChatReply: typeof import("./monitor-reply-delivery.js").deliver beforeEach(async () => { vi.clearAllMocks(); + mocks.sendGoogleChatMessage.mockResolvedValue(null); + mocks.updateGoogleChatMessage.mockResolvedValue({}); ({ createGoogleChatTypingMessage, deliverGoogleChatReply } = await import("./monitor-reply-delivery.js")); }); @@ -244,28 +246,6 @@ describe("Google Chat reply delivery", () => { } }); - 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(); - const sendError = new Error("API 500"); - mocks.sendGoogleChatMessage - .mockResolvedValueOnce({ messageName: "spaces/AAA/messages/one" }) - .mockRejectedValueOnce(sendError); - - await expect( - deliverGoogleChatReply({ - payload: { text: "three chunks", replyToId: "spaces/AAA/threads/root" }, - account, - spaceId: "spaces/AAA", - runtime, - core, - config, - }), - ).rejects.toBe(sendError); - - expect(mocks.sendGoogleChatMessage).toHaveBeenCalledTimes(2); - }); - it("replaces a typing message when the final reply target changed", async () => { const core = createCore(); const runtime = createRuntime(); diff --git a/extensions/googlechat/src/monitor.test.ts b/extensions/googlechat/src/monitor.test.ts index ab42ebd55a68..62b5d7a6ef01 100644 --- a/extensions/googlechat/src/monitor.test.ts +++ b/extensions/googlechat/src/monitor.test.ts @@ -81,8 +81,8 @@ vi.mock("./monitor-routing.js", () => ({ beforeEach(() => { apiMocks.deleteGoogleChatMessage.mockReset(); apiMocks.downloadGoogleChatMedia.mockReset(); - apiMocks.sendGoogleChatMessage.mockReset(); - apiMocks.updateGoogleChatMessage.mockReset(); + apiMocks.sendGoogleChatMessage.mockReset().mockResolvedValue(null); + apiMocks.updateGoogleChatMessage.mockReset().mockResolvedValue({}); accessMocks.applyGoogleChatInboundAccessPolicy.mockReset(); inboundMocks.buildEnvelope.mockReset().mockImplementation(({ body }: { body: string }) => body); inboundMocks.resolveChannelInboundRouteEnvelope