refactor: unify reply dispatch across providers

src/signal/monitor.tool-result.test.ts

@@ -0,0 +1,96 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { monitorSignalProvider } from "./monitor.js";
+
+const sendMock = vi.fn();
+const replyMock = vi.fn();
+const updateLastRouteMock = vi.fn();
+let config: Record<string, unknown> = {};
+
+vi.mock("../config/config.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../config/config.js")>();
+  return {
+    ...actual,
+    loadConfig: () => config,
+  };
+});
+
+vi.mock("../auto-reply/reply.js", () => ({
+  getReplyFromConfig: (...args: unknown[]) => replyMock(...args),
+}));
+
+vi.mock("./send.js", () => ({
+  sendMessageSignal: (...args: unknown[]) => sendMock(...args),
+}));
+
+vi.mock("../config/sessions.js", () => ({
+  resolveStorePath: vi.fn(() => "/tmp/clawdbot-sessions.json"),
+  updateLastRoute: (...args: unknown[]) => updateLastRouteMock(...args),
+}));
+
+const streamMock = vi.fn();
+const signalCheckMock = vi.fn();
+const signalRpcRequestMock = vi.fn();
+
+vi.mock("./client.js", () => ({
+  streamSignalEvents: (...args: unknown[]) => streamMock(...args),
+  signalCheck: (...args: unknown[]) => signalCheckMock(...args),
+  signalRpcRequest: (...args: unknown[]) => signalRpcRequestMock(...args),
+}));
+
+vi.mock("./daemon.js", () => ({
+  spawnSignalDaemon: vi.fn(() => ({ stop: vi.fn() })),
+}));
+
+const flush = () => new Promise((resolve) => setTimeout(resolve, 0));
+
+beforeEach(() => {
+  config = {
+    messages: { responsePrefix: "PFX" },
+    signal: { autoStart: false },
+    routing: { allowFrom: [] },
+  };
+  sendMock.mockReset().mockResolvedValue(undefined);
+  replyMock.mockReset();
+  updateLastRouteMock.mockReset();
+  streamMock.mockReset();
+  signalCheckMock.mockReset().mockResolvedValue({});
+  signalRpcRequestMock.mockReset().mockResolvedValue({});
+});
+
+describe("monitorSignalProvider tool results", () => {
+  it("sends tool summaries with responsePrefix", async () => {
+    replyMock.mockImplementation(async (_ctx, opts) => {
+      await opts?.onToolResult?.({ text: "tool update" });
+      return { text: "final reply" };
+    });
+
+    streamMock.mockImplementation(async ({ onEvent }) => {
+      const payload = {
+        envelope: {
+          sourceNumber: "+15550001111",
+          sourceName: "Ada",
+          timestamp: 1,
+          dataMessage: {
+            message: "hello",
+          },
+        },
+      };
+      await onEvent({
+        event: "receive",
+        data: JSON.stringify(payload),
+      });
+    });
+
+    await monitorSignalProvider({
+      autoStart: false,
+      baseUrl: "http://127.0.0.1:8080",
+    });
+
+    await flush();
+
+    expect(sendMock).toHaveBeenCalledTimes(2);
+    expect(sendMock.mock.calls[0][1]).toBe("PFX tool update");
+    expect(sendMock.mock.calls[1][1]).toBe("PFX final reply");
+  });
+});

src/signal/monitor.ts

@@ -1,5 +1,6 @@
 import { chunkText, resolveTextChunkLimit } from "../auto-reply/chunk.js";
 import { formatAgentEnvelope } from "../auto-reply/envelope.js";
+import { createReplyDispatcher } from "../auto-reply/reply/reply-dispatcher.js";
 import { getReplyFromConfig } from "../auto-reply/reply.js";
 import type { ReplyPayload } from "../auto-reply/types.js";
 import { loadConfig } from "../config/config.js";
@@ -379,37 +380,36 @@ export async function monitorSignalProvider(
         );
       }
 
-      let blockSendChain: Promise<void> = Promise.resolve();
-      const sendBlockReply = (payload: ReplyPayload) => {
-        if (
-          !payload?.text &&
-          !payload?.mediaUrl &&
-          !(payload?.mediaUrls?.length ?? 0)
-        ) {
-          return;
-        }
-        blockSendChain = blockSendChain
-          .then(async () => {
-            await deliverReplies({
-              replies: [payload],
-              target: ctxPayload.To,
-              baseUrl,
-              account,
-              runtime,
-              maxBytes: mediaMaxBytes,
-              textLimit,
-            });
-          })
-          .catch((err) => {
-            runtime.error?.(
-              danger(`signal block reply failed: ${String(err)}`),
-            );
+      const dispatcher = createReplyDispatcher({
+        responsePrefix: cfg.messages?.responsePrefix,
+        deliver: async (payload) => {
+          await deliverReplies({
+            replies: [payload],
+            target: ctxPayload.To,
+            baseUrl,
+            account,
+            runtime,
+            maxBytes: mediaMaxBytes,
+            textLimit,
           });
-      };
+        },
+        onError: (err, info) => {
+          runtime.error?.(
+            danger(`signal ${info.kind} reply failed: ${String(err)}`),
+          );
+        },
+      });
 
       const replyResult = await getReplyFromConfig(
         ctxPayload,
-        { onBlockReply: sendBlockReply },
+        {
+          onToolResult: (payload) => {
+            dispatcher.sendToolResult(payload);
+          },
+          onBlockReply: (payload) => {
+            dispatcher.sendBlockReply(payload);
+          },
+        },
         cfg,
       );
       const replies = replyResult
@@ -417,18 +417,12 @@ export async function monitorSignalProvider(
           ? replyResult
           : [replyResult]
         : [];
-      await blockSendChain;
-      if (replies.length === 0) return;
-
-      await deliverReplies({
-        replies,
-        target: ctxPayload.To,
-        baseUrl,
-        account,
-        runtime,
-        maxBytes: mediaMaxBytes,
-        textLimit,
-      });
+      let queuedFinal = false;
+      for (const reply of replies) {
+        queuedFinal = dispatcher.sendFinalReply(reply) || queuedFinal;
+      }
+      await dispatcher.waitForIdle();
+      if (!queuedFinal) return;
     };
 
     await streamSignalEvents({