fix(gateway): keep session tool mirrors under pressure

steipete · web-flow · commit 459e89ada899 · 2026-05-25T15:22:52.000+01:00
Reverts the diagnostic queue-pressure suppression of non-terminal session tool mirrors from PR 84846 while keeping PR 86503 recipient dedupe intact. Session-only Control UI subscribers keep receiving tool lifecycle mirrors; overlapping run and session subscribers still receive one canonical run-scoped frame. Verification: focused gateway and diagnostic tests, diff check, changed check, and autoreview all passed.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- Gateway: keep session-only Control UI tool-start mirrors flowing during diagnostic queue pressure instead of silently dropping non-terminal tool updates.
 - Gateway: avoid sending duplicate tool-event frames to Control UI connections that are subscribed by both run and session.
 - Discord/OpenAI voice: accept longer leading wake-name mistranscripts such as "Open Club" for OpenClaw.
 - Discord/OpenAI voice: accept leading fuzzy wake-name transcripts such as "Monty" or "Moti" for a Molty agent while keeping ambient speech gated.
diff --git a/src/gateway/server-chat.agent-events.test.ts b/src/gateway/server-chat.agent-events.test.ts
@@ -79,7 +79,6 @@ describe("agent event handler", () => {
     resolveSessionKeyForRun?: (runId: string) => string | undefined;
     lifecycleErrorRetryGraceMs?: number;
     isChatSendRunActive?: (runId: string) => boolean;
-    shouldBackoffLowPrioritySessionToolEvents?: () => boolean;
   }) {
     const nowSpy =
       params?.now === undefined ? undefined : vi.spyOn(Date, "now").mockReturnValue(params.now);
@@ -107,7 +106,6 @@ describe("agent event handler", () => {
       loadGatewaySessionRowForSnapshot: loadGatewaySessionRow,
       lifecycleErrorRetryGraceMs: params?.lifecycleErrorRetryGraceMs,
       isChatSendRunActive: params?.isChatSendRunActive,
-      shouldBackoffLowPrioritySessionToolEvents: params?.shouldBackoffLowPrioritySessionToolEvents,
     });
 
     return {
@@ -1564,96 +1562,6 @@ describe("agent event handler", () => {
     );
   });
 
-  it("backs off session-scoped tool mirrors during queued gateway pressure", () => {
-    const { broadcastToConnIds, sessionEventSubscribers, toolEventRecipients, handler } =
-      createHarness({
-        resolveSessionKeyForRun: () => "session-pressure",
-        shouldBackoffLowPrioritySessionToolEvents: () => true,
-      });
-
-    registerAgentRunContext("run-pressure-tool", {
-      sessionKey: "session-pressure",
-      verboseLevel: "off",
-    });
-    toolEventRecipients.add("run-pressure-tool", "conn-run");
-    sessionEventSubscribers.subscribe("conn-session");
-
-    handler({
-      runId: "run-pressure-tool",
-      seq: 1,
-      stream: "tool",
-      ts: 1_234,
-      data: {
-        phase: "start",
-        name: "exec",
-        toolCallId: "tool-pressure-1",
-        args: { command: "echo hi" },
-      },
-    });
-
-    expect(broadcastToConnIds).toHaveBeenCalledTimes(1);
-    expect(requireMockArg(broadcastToConnIds, 0, 0, "run tool event")).toBe("agent");
-    expect(requireMockArg(broadcastToConnIds, 0, 2, "run tool recipients")).toEqual(
-      new Set(["conn-run"]),
-    );
-  });
-
-  it("keeps terminal session-scoped tool mirrors during queued gateway pressure", () => {
-    let backoffActive = false;
-    const { broadcastToConnIds, sessionEventSubscribers, handler } = createHarness({
-      resolveSessionKeyForRun: () => "session-pressure-terminal",
-      shouldBackoffLowPrioritySessionToolEvents: () => backoffActive,
-    });
-
-    registerAgentRunContext("run-pressure-terminal-tool", {
-      sessionKey: "session-pressure-terminal",
-      verboseLevel: "off",
-    });
-    sessionEventSubscribers.subscribe("conn-session");
-
-    handler({
-      runId: "run-pressure-terminal-tool",
-      seq: 1,
-      stream: "tool",
-      ts: 1_234,
-      data: {
-        phase: "start",
-        name: "exec",
-        toolCallId: "tool-pressure-terminal-1",
-        args: { command: "echo hi" },
-      },
-    });
-
-    backoffActive = true;
-    handler({
-      runId: "run-pressure-terminal-tool",
-      seq: 2,
-      stream: "tool",
-      ts: 1_235,
-      data: {
-        phase: "result",
-        name: "exec",
-        toolCallId: "tool-pressure-terminal-1",
-        result: { content: [{ type: "text", text: "done" }] },
-      },
-    });
-
-    expect(broadcastToConnIds).toHaveBeenCalledTimes(2);
-    expect(requireMockArg(broadcastToConnIds, 0, 0, "session tool start event")).toBe(
-      "session.tool",
-    );
-    expect(requireMockArg(broadcastToConnIds, 1, 0, "session tool result event")).toBe(
-      "session.tool",
-    );
-    const resultPayload = requireMockPayload(broadcastToConnIds, 1, 1, "session tool result");
-    expectRecordFields(requireRecord(resultPayload.data, "session tool result data"), {
-      phase: "result",
-      name: "exec",
-      toolCallId: "tool-pressure-terminal-1",
-      result: { content: [{ type: "text", text: "done" }] },
-    });
-  });
-
   it("suppresses heartbeat tool events for Control UI and verbose node subscribers", () => {
     const {
       broadcastToConnIds,
diff --git a/src/gateway/server-chat.ts b/src/gateway/server-chat.ts
@@ -5,7 +5,6 @@ import { getRuntimeConfig } from "../config/io.js";
 import { type AgentEventPayload, getAgentRunContext } from "../infra/agent-events.js";
 import { detectErrorKind, type ErrorKind } from "../infra/errors.js";
 import { resolveHeartbeatVisibility } from "../infra/heartbeat-visibility.js";
-import { isDiagnosticQueuePressureBackoffActive } from "../logging/diagnostic.js";
 import { isAcpSessionKey, isSubagentSessionKey } from "../sessions/session-key-utils.js";
 import { setSafeTimeout } from "../utils/timer-delay.js";
 import {
@@ -177,12 +176,6 @@ function readChatErrorKind(value: unknown): ErrorKind | undefined {
     : undefined;
 }
 
-const TERMINAL_TOOL_EVENT_PHASES = new Set(["end", "error", "result"]);
-
-function isTerminalToolEventPhase(phase: string): boolean {
-  return TERMINAL_TOOL_EVENT_PHASES.has(phase);
-}
-
 function excludeConnIds(
   connIds: ReadonlySet<string>,
   excludedConnIds: ReadonlySet<string> | undefined,
@@ -238,7 +231,6 @@ export type AgentEventHandlerOptions = {
   loadGatewaySessionRowForSnapshot?: typeof loadGatewaySessionRow;
   lifecycleErrorRetryGraceMs?: number;
   isChatSendRunActive?: (runId: string) => boolean;
-  shouldBackoffLowPrioritySessionToolEvents?: () => boolean;
 };
 
 export function createAgentEventHandler({
@@ -255,7 +247,6 @@ export function createAgentEventHandler({
   loadGatewaySessionRowForSnapshot = loadGatewaySessionRow,
   lifecycleErrorRetryGraceMs = AGENT_LIFECYCLE_ERROR_RETRY_GRACE_MS,
   isChatSendRunActive = () => false,
-  shouldBackoffLowPrioritySessionToolEvents = isDiagnosticQueuePressureBackoffActive,
 }: AgentEventHandlerOptions) {
   type TerminalLifecycleOptions = { skipChatErrorFinal?: boolean };
   type PendingTerminalLifecycleError = {
@@ -998,14 +989,7 @@ export function createAgentEventHandler({
       // not know the runId in advance, so they cannot register as run-scoped
       // tool recipients. Mirror tool lifecycle onto a session-scoped event so
       // they can render live pending tool cards without polling history.
-      const shouldMirrorSessionToolEvent =
-        !shouldBackoffLowPrioritySessionToolEvents() || isTerminalToolEventPhase(toolPhase);
-      if (
-        isControlUiVisible &&
-        sessionKey &&
-        !suppressHeartbeatToolEvents &&
-        shouldMirrorSessionToolEvent
-      ) {
+      if (isControlUiVisible && sessionKey && !suppressHeartbeatToolEvents) {
         const sessionSubscribers = excludeConnIds(
           sessionEventSubscribers.getAll(),
           runToolRecipients,
diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts
@@ -34,7 +34,6 @@ import {
   logSessionStateChange,
   logMessageQueued,
   diagnosticLogger,
-  isDiagnosticQueuePressureBackoffActive,
   markDiagnosticSessionProgress,
   resetDiagnosticStateForTest,
   resolveStuckSessionAbortMs,
@@ -1694,35 +1693,6 @@ describe("stuck session diagnostics threshold", () => {
     );
   });
 
-  it("marks queued gateway pressure for low-priority backoff", () => {
-    startDiagnosticHeartbeat(
-      {
-        diagnostics: {
-          enabled: true,
-        },
-      },
-      {
-        emitMemorySample: createEmitMemorySampleMock(),
-        sampleLiveness: () => ({
-          reasons: ["cpu"],
-          intervalMs: 30_000,
-          eventLoopUtilization: 0.99,
-          cpuTotalMs: 30_000,
-          cpuCoreRatio: 1,
-        }),
-      },
-    );
-
-    logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
-    logMessageQueued({ sessionId: "s1", sessionKey: "main", source: "discord" });
-    vi.advanceTimersByTime(30_000);
-
-    expect(isDiagnosticQueuePressureBackoffActive(Date.now())).toBe(true);
-    stopDiagnosticHeartbeat();
-    expect(isDiagnosticQueuePressureBackoffActive(Date.now())).toBe(false);
-    expect(isDiagnosticQueuePressureBackoffActive(Date.now() + 60_001)).toBe(false);
-  });
-
   it("does not let idle liveness samples suppress later active-work warnings", () => {
     const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
 
diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts
@@ -82,7 +82,6 @@ const DEFAULT_LIVENESS_EVENT_LOOP_DELAY_WARN_MS = 1_000;
 const DEFAULT_LIVENESS_EVENT_LOOP_UTILIZATION_WARN = 0.95;
 const DEFAULT_LIVENESS_CPU_CORE_RATIO_WARN = 0.9;
 const DEFAULT_LIVENESS_WARN_COOLDOWN_MS = 120_000;
-const DEFAULT_QUEUE_PRESSURE_BACKOFF_TTL_MS = 60_000;
 let commandPollBackoffRuntimePromise: Promise<
   typeof import("../agents/command-poll-backoff.runtime.js")
 > | null = null;
@@ -1138,32 +1137,6 @@ export function logActiveRuns() {
 }
 
 let heartbeatInterval: NodeJS.Timeout | null = null;
-let diagnosticQueuePressureBackoffUntil = 0;
-
-export function isDiagnosticQueuePressureBackoffActive(now = Date.now()): boolean {
-  if (!heartbeatInterval || !areDiagnosticsEnabledForProcess()) {
-    return false;
-  }
-  return diagnosticQueuePressureBackoffUntil > now;
-}
-
-export function resetDiagnosticQueuePressureBackoffForTest(): void {
-  diagnosticQueuePressureBackoffUntil = 0;
-}
-
-function updateDiagnosticQueuePressureBackoff(
-  now: number,
-  sample: DiagnosticLivenessSample,
-  work: DiagnosticWorkSnapshot,
-): void {
-  if (work.queuedCount <= 0 || sample.reasons.length === 0) {
-    return;
-  }
-  diagnosticQueuePressureBackoffUntil = Math.max(
-    diagnosticQueuePressureBackoffUntil,
-    now + DEFAULT_QUEUE_PRESSURE_BACKOFF_TTL_MS,
-  );
-}
 
 export function startDiagnosticHeartbeat(
   config?: OpenClawConfig,
@@ -1202,9 +1175,6 @@ export function startDiagnosticHeartbeat(
       livenessSample !== null && shouldEmitDiagnosticLivenessEvent(now);
     const shouldEmitLivenessWarning =
       livenessSample !== null && shouldEmitDiagnosticLivenessWarning(now, work);
-    if (livenessSample) {
-      updateDiagnosticQueuePressureBackoff(now, livenessSample, work);
-    }
     const shouldEmitLivenessReport = shouldEmitLivenessEvent || shouldEmitLivenessWarning;
     const shouldRecordMemorySample =
       shouldEmitLivenessReport || hasRecentDiagnosticActivity(now) || hasOpenDiagnosticWork(work);
@@ -1328,7 +1298,6 @@ export function stopDiagnosticHeartbeat() {
   stopDiagnosticLivenessSampler();
   stopDiagnosticStabilityRecorder();
   uninstallDiagnosticStabilityFatalHook();
-  resetDiagnosticQueuePressureBackoffForTest();
 }
 
 export function getDiagnosticSessionStateCountForTest(): number {
@@ -1349,5 +1318,4 @@ export function resetDiagnosticStateForTest(): void {
   resetDiagnosticPhasesForTest();
   resetDiagnosticStabilityRecorderForTest();
   resetDiagnosticStabilityBundleForTest();
-  resetDiagnosticQueuePressureBackoffForTest();
 }
