fix(gateway): dedupe session tool fanout

steipete · web-flow · commit 89a21db627f5 · 2026-05-25T14:40:07.000+01:00
Dedupe gateway tool-event fanout so connections subscribed by both run and session receive the canonical run-scoped agent event only, while session-only subscribers keep the compatibility session.tool mirror.\n\nVerification:\n- node scripts/run-vitest.mjs src/gateway/server-chat.agent-events.test.ts\n- git diff --check\n- env -u OPENCLAW_TESTBOX pnpm check:changed\n- .agents/skills/autoreview/scripts/autoreview --mode local
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- Gateway: avoid sending duplicate tool-event frames to Control UI connections that are subscribed by both run and session.
 - Discord/OpenAI voice: accept leading fuzzy wake-name transcripts such as "Monty" or "Moti" for a Molty agent while keeping ambient speech gated.
 - Media understanding: convert HEIC and HEIF images to JPEG before image description providers run so iPhone photos work in direct and configured image-description flows. (#86037)
 - Discord/OpenAI voice: rotate Realtime sessions at provider max duration without logging the expected session-expiry event as an error.
diff --git a/src/gateway/server-chat.agent-events.test.ts b/src/gateway/server-chat.agent-events.test.ts
@@ -107,8 +107,7 @@ describe("agent event handler", () => {
       loadGatewaySessionRowForSnapshot: loadGatewaySessionRow,
       lifecycleErrorRetryGraceMs: params?.lifecycleErrorRetryGraceMs,
       isChatSendRunActive: params?.isChatSendRunActive,
-      shouldBackoffLowPrioritySessionToolEvents:
-        params?.shouldBackoffLowPrioritySessionToolEvents,
+      shouldBackoffLowPrioritySessionToolEvents: params?.shouldBackoffLowPrioritySessionToolEvents,
     });
 
     return {
@@ -1526,6 +1525,45 @@ describe("agent event handler", () => {
     resetAgentRunContextForTest();
   });
 
+  it("does not duplicate tool events to clients subscribed by run and session", () => {
+    const { broadcastToConnIds, sessionEventSubscribers, toolEventRecipients, handler } =
+      createHarness({
+        resolveSessionKeyForRun: () => "session-dedupe",
+      });
+
+    registerAgentRunContext("run-session-dedupe-tool", {
+      sessionKey: "session-dedupe",
+      verboseLevel: "off",
+    });
+    toolEventRecipients.add("run-session-dedupe-tool", "conn-overlap");
+    toolEventRecipients.add("run-session-dedupe-tool", "conn-run-only");
+    sessionEventSubscribers.subscribe("conn-overlap");
+    sessionEventSubscribers.subscribe("conn-session-only");
+
+    handler({
+      runId: "run-session-dedupe-tool",
+      seq: 1,
+      stream: "tool",
+      ts: 1_234,
+      data: {
+        phase: "start",
+        name: "exec",
+        toolCallId: "tool-session-dedupe-1",
+        args: { command: "echo hi" },
+      },
+    });
+
+    expect(broadcastToConnIds).toHaveBeenCalledTimes(2);
+    expect(requireMockArg(broadcastToConnIds, 0, 0, "run tool event")).toBe("agent");
+    expect(requireMockArg(broadcastToConnIds, 0, 2, "run tool recipients")).toEqual(
+      new Set(["conn-overlap", "conn-run-only"]),
+    );
+    expect(requireMockArg(broadcastToConnIds, 1, 0, "session tool event")).toBe("session.tool");
+    expect(requireMockArg(broadcastToConnIds, 1, 2, "session tool recipients")).toEqual(
+      new Set(["conn-session-only"]),
+    );
+  });
+
   it("backs off session-scoped tool mirrors during queued gateway pressure", () => {
     const { broadcastToConnIds, sessionEventSubscribers, toolEventRecipients, handler } =
       createHarness({
diff --git a/src/gateway/server-chat.ts b/src/gateway/server-chat.ts
@@ -183,6 +183,22 @@ function isTerminalToolEventPhase(phase: string): boolean {
   return TERMINAL_TOOL_EVENT_PHASES.has(phase);
 }
 
+function excludeConnIds(
+  connIds: ReadonlySet<string>,
+  excludedConnIds: ReadonlySet<string> | undefined,
+): ReadonlySet<string> {
+  if (!excludedConnIds || excludedConnIds.size === 0 || connIds.size === 0) {
+    return connIds;
+  }
+  const filtered = new Set<string>();
+  for (const connId of connIds) {
+    if (!excludedConnIds.has(connId)) {
+      filtered.add(connId);
+    }
+  }
+  return filtered;
+}
+
 type BroadcastDelta = { deltaText: string; replace?: true };
 
 function resolveBroadcastDelta(params: {
@@ -957,12 +973,17 @@ export function createAgentEventHandler({
       // tool-events capability, regardless of verboseLevel. The verbose
       // setting only controls whether tool details are sent as channel
       // messages to messaging surfaces (Telegram, Discord, etc.).
-      const recipients = toolEventRecipients.get(evt.runId);
-      if (isControlUiVisible && !suppressHeartbeatToolEvents && recipients && recipients.size > 0) {
+      const runToolRecipients = toolEventRecipients.get(evt.runId);
+      if (
+        isControlUiVisible &&
+        !suppressHeartbeatToolEvents &&
+        runToolRecipients &&
+        runToolRecipients.size > 0
+      ) {
         broadcastToConnIds(
           "agent",
           sessionKey ? { ...agentPayload, ...buildSessionEventSnapshot(sessionKey) } : agentPayload,
-          recipients,
+          runToolRecipients,
         );
       }
       if (!isControlUiVisible && sessionKey && !suppressHeartbeatToolEvents) {
@@ -985,7 +1006,10 @@ export function createAgentEventHandler({
         !suppressHeartbeatToolEvents &&
         shouldMirrorSessionToolEvent
       ) {
-        const sessionSubscribers = sessionEventSubscribers.getAll();
+        const sessionSubscribers = excludeConnIds(
+          sessionEventSubscribers.getAll(),
+          runToolRecipients,
+        );
         if (sessionSubscribers.size > 0) {
           broadcastToConnIds(
             "session.tool",
