From 889a15317182ec75d08a15c8e41fdaa65c940686 Mon Sep 17 00:00:00 2001 From: Leah Armstrong Date: Tue, 25 Aug 2026 20:12:26 -0400 Subject: [PATCH] fix(buzz): anchor threaded replies without nesting (#124884) Co-authored-by: Peter Steinberger --- .../buzz/src/buzz-bus.lifecycle.test.ts | 81 ++++++++++++++++++- extensions/buzz/src/inbound.test.ts | 6 +- extensions/buzz/src/inbound.ts | 18 ++--- 3 files changed, 90 insertions(+), 15 deletions(-) diff --git a/extensions/buzz/src/buzz-bus.lifecycle.test.ts b/extensions/buzz/src/buzz-bus.lifecycle.test.ts index 5fed6f14c157..66d446f340c7 100644 --- a/extensions/buzz/src/buzz-bus.lifecycle.test.ts +++ b/extensions/buzz/src/buzz-bus.lifecycle.test.ts @@ -1,7 +1,8 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; -import { finalizeEvent, getPublicKey, type Event, type Filter } from "nostr-tools"; +import { finalizeEvent, getPublicKey, verifyEvent, type Event, type Filter } from "nostr-tools"; +import { createPluginRuntimeMock } from "openclaw/plugin-sdk/channel-test-helpers"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const relayMocks = vi.hoisted(() => ({ @@ -111,11 +112,15 @@ vi.mock("nostr-tools", async (importOriginal) => { }); import { sendBuzzTextOneShot, startBuzzBus, type BuzzBus } from "./buzz-bus.js"; +import { handleBuzzInbound } from "./inbound.js"; import { BUZZ_DIFF_MESSAGE_KIND, BUZZ_INBOUND_MESSAGE_KINDS, + BUZZ_TYPING_INDICATOR_KIND, type BuzzInboundMessage, } from "./message-event.js"; +import { setBuzzRuntime } from "./runtime.js"; +import type { ResolvedBuzzAccount } from "./types.js"; const BUZZ_RICH_MESSAGE_KIND = 40_002; const PRIVATE_KEY = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f"; @@ -398,6 +403,80 @@ describe("Buzz bus lifecycle", () => { expect(relayMocks.send).not.toHaveBeenCalled(); }); + it("anchors signed threaded replies and typing while preserving top-level replies", async () => { + relayMocks.auth.mockResolvedValue("ok"); + const runtime = createPluginRuntimeMock(); + setBuzzRuntime(runtime); + const account: ResolvedBuzzAccount = { + accountId: ACCOUNT_ID, + name: "OpenClaw", + enabled: true, + configured: true, + relayUrl: "wss://buzz.example.com", + privateKey: PRIVATE_KEY, + authTag: "", + publicKey: BOT_PUBLIC_KEY, + config: { + groupPolicy: "open", + groups: { [CHANNEL_ID]: { requireMention: false } }, + }, + }; + const bus = await startTestBus({ + onMessage: async (message, activeBus, signal) => + await handleBuzzInbound({ account, cfg: {}, bus: activeBus, message, signal }), + }); + + try { + const rootId = "a".repeat(64); + const messageSubscription = relayMocks.subscriptions.find((entry) => + subscriptionIncludesKind(entry, 9), + ); + for (const [index, parentId] of ["b".repeat(64), "c".repeat(64), undefined].entries()) { + const inbound = signSenderEvent({ + kind: 9, + created_at: 1_700_000_000 + index, + content: `follow-up ${index + 1}`, + tags: [ + ["h", CHANNEL_ID], + ...(parentId + ? [ + ["e", rootId, "", "root"], + ["e", parentId, "", "reply"], + ] + : []), + ], + }); + messageSubscription?.handlers.onevent(inbound); + await vi.waitFor(() => + expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(index + 1), + ); + const dispatch = vi.mocked(runtime.channel.inbound.dispatch).mock.calls[index]?.[0]; + await dispatch?.delivery.deliver({ text: `reply ${index + 1}` }, { kind: "final" }); + await dispatch?.replyPipeline?.typing?.start(); + + const published = relayMocks.publish.mock.calls.find( + ([event]) => event.kind === 9 && event.content === `reply ${index + 1}`, + )?.[0]; + const typing = relayMocks.send.mock.calls + .map(([frame]) => JSON.parse(frame) as [string, Event]) + .filter( + ([frameType, event]) => + frameType === "EVENT" && event.kind === BUZZ_TYPING_INDICATOR_KIND, + )[index]?.[1]; + const expectedTags = [ + ["h", CHANNEL_ID], + ["e", parentId ? rootId : inbound.id, "", "reply"], + ]; + expect(published?.tags).toEqual(expectedTags); + expect(typing?.tags).toEqual(expectedTags); + expect(published && verifyEvent(published)).toBe(true); + expect(typing && verifyEvent(typing)).toBe(true); + } + } finally { + await bus.close(); + } + }); + it("drops typing while the active relay is disconnected", async () => { relayMocks.auth.mockResolvedValue("ok"); const bus = await startTestBus(); diff --git a/extensions/buzz/src/inbound.test.ts b/extensions/buzz/src/inbound.test.ts index bcac887c9615..a74a91bce1bc 100644 --- a/extensions/buzz/src/inbound.test.ts +++ b/extensions/buzz/src/inbound.test.ts @@ -365,7 +365,7 @@ describe("handleBuzzInbound", () => { expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled(); }); - it("preserves Buzz thread and reply identifiers for agent replies", async () => { + it("anchors Buzz agent replies and typing to the original thread root", async () => { const runtime = createPluginRuntimeMock(); setBuzzRuntime(runtime); const bus = createBus(); @@ -398,7 +398,7 @@ describe("handleBuzzInbound", () => { channelId: ROOM_ID, text: "threaded reply to @Alice", threadId: "event-root", - replyToId: "event-reply", + replyToId: "event-root", }); const typing = dispatch.replyPipeline?.typing; @@ -407,7 +407,7 @@ describe("handleBuzzInbound", () => { expect(bus.sendTyping).toHaveBeenCalledWith({ channelId: ROOM_ID, threadId: "event-root", - replyToId: "event-reply", + replyToId: "event-root", }); }); diff --git a/extensions/buzz/src/inbound.ts b/extensions/buzz/src/inbound.ts index d84287fc356f..03801f236ba8 100644 --- a/extensions/buzz/src/inbound.ts +++ b/extensions/buzz/src/inbound.ts @@ -138,6 +138,11 @@ export async function handleBuzzInbound(params: { BuzzEventKind: message.kind, }, }); + const replyTarget = { + channelId, + threadId: message.threadId, + replyToId: message.threadId ?? message.id, + }; await runtime.channel.inbound.dispatch({ cfg, @@ -158,12 +163,7 @@ export async function handleBuzzInbound(params: { if (!text.trim()) { return; } - await bus.sendText({ - channelId, - text, - threadId: message.threadId, - replyToId: message.id, - }); + await bus.sendText({ ...replyTarget, text }); }, onError: (error) => { throw error instanceof Error ? error : new Error(String(error)); @@ -175,11 +175,7 @@ export async function handleBuzzInbound(params: { replyPipeline: { typing: { start: async () => { - await bus.sendTyping({ - channelId, - threadId: message.threadId, - replyToId: message.id, - }); + await bus.sendTyping(replyTarget); }, keepaliveIntervalMs: 3_000, onStartError: (error: unknown) => {