fix(buzz): anchor threaded replies without nesting (#124884)

Co-authored-by: Peter Steinberger <steipete@gmail.com>

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();

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

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