fix(beam): prevent mirror retries from stalling on slow responses (#116174)

* fix(beam): release mirror upload responses

* test(beam): cover response cancellation failures

* fix(buzz): repair typing indicator contracts

---------

Co-authored-by: Vincent Koc <vincentkoc@ieee.org>

extensions/beam/src/mirror.test.ts

@@ -5,7 +5,7 @@ import type {
   SessionsCatalogReadResult,
 } from "openclaw/plugin-sdk/session-catalog";
 import type { ActiveSessionCatalog } from "openclaw/plugin-sdk/session-catalog-runtime";
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
   beamMirrorId,
   buildBeamMirrorItems,
@@ -87,7 +87,11 @@ function fakeRuntime(config: unknown): PluginRuntime {
 
 type SentRequest = { url: string; auth?: string; payload: BeamMirrorUpload };
 
-function captureFetch(sent: SentRequest[], status = 200): typeof fetch {
+function captureFetch(
+  sent: SentRequest[],
+  status = 200,
+  onCancel?: () => void | Promise<void>,
+): typeof fetch {
   return (async (url: unknown, init?: RequestInit) => {
     const headers = (init?.headers ?? {}) as Record<string, string>;
     sent.push({
@@ -95,7 +99,12 @@ function captureFetch(sent: SentRequest[], status = 200): typeof fetch {
       ...(headers.Authorization ? { auth: headers.Authorization } : {}),
       payload: JSON.parse(init?.body as string) as BeamMirrorUpload,
     });
-    return new Response("{}", { status });
+    const body = onCancel
+      ? new ReadableStream<Uint8Array>({
+          cancel: onCancel,
+        })
+      : "{}";
+    return new Response(body, { status });
   }) as typeof fetch;
 }
 
@@ -210,6 +219,7 @@ describe("createBeamMirrorRunner", () => {
   it("uploads active local sessions and skips unchanged ones", async () => {
     const sent: SentRequest[] = [];
     const reads: string[] = [];
+    const cancel = vi.fn();
     const catalog = fakeCatalog({
       id: "claude",
       sessions: [{ threadId: "t1", name: "Fix flow", recencyAt: NOW - 60_000 }],
@@ -218,7 +228,7 @@ describe("createBeamMirrorRunner", () => {
     const runner = createBeamMirrorRunner({
       runtime: fakeRuntime(mirrorConfig({ token: "scratch-token" })),
       logger: silentLogger,
-      fetchFn: captureFetch(sent),
+      fetchFn: captureFetch(sent, 200, cancel),
       now: () => NOW,
       listCatalogs: () => [catalog],
     });
@@ -235,6 +245,33 @@ describe("createBeamMirrorRunner", () => {
       completed: false,
     });
     expect(parseBeamUpload(structuredClone(sent[0]?.payload)).ok).toBe(true);
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it("keeps successful uploads successful when response cancellation rejects", async () => {
+    const sent: SentRequest[] = [];
+    const warnings: string[] = [];
+    const cancel = vi.fn(async () => {
+      throw new Error("cancel failed");
+    });
+    const catalog = fakeCatalog({
+      id: "claude",
+      sessions: [{ threadId: "t1", recencyAt: NOW - 60_000 }],
+    });
+    const runner = createBeamMirrorRunner({
+      runtime: fakeRuntime(mirrorConfig()),
+      logger: { warn: (message) => warnings.push(message), info: () => {} },
+      fetchFn: captureFetch(sent, 200, cancel),
+      now: () => NOW,
+      listCatalogs: () => [catalog],
+    });
+
+    await runner.tick();
+    await runner.tick();
+
+    expect(sent).toHaveLength(1);
+    expect(cancel).toHaveBeenCalledOnce();
+    expect(warnings).toEqual([]);
   });
 
   it("ignores idle sessions, node hosts, the beam catalog, and unlisted catalogs", async () => {
@@ -318,6 +355,7 @@ describe("createBeamMirrorRunner", () => {
   it("keeps tracking for retry when the receiver rejects an upload", async () => {
     const sent: SentRequest[] = [];
     const warnings: string[] = [];
+    const cancel = vi.fn();
     const catalog = fakeCatalog({
       id: "claude",
       sessions: [{ threadId: "t1", recencyAt: NOW - 60_000 }],
@@ -325,7 +363,7 @@ describe("createBeamMirrorRunner", () => {
     const runner = createBeamMirrorRunner({
       runtime: fakeRuntime(mirrorConfig()),
       logger: { warn: (message) => warnings.push(message), info: () => {} },
-      fetchFn: captureFetch(sent, 503),
+      fetchFn: captureFetch(sent, 503, cancel),
       now: () => NOW,
       listCatalogs: () => [catalog],
     });
@@ -334,6 +372,7 @@ describe("createBeamMirrorRunner", () => {
     // Both ticks retry because the failed upload was never fingerprinted.
     expect(sent).toHaveLength(2);
     expect(warnings.length).toBeGreaterThan(0);
+    expect(cancel).toHaveBeenCalledTimes(2);
   });
 
   it("skips ticks when a configured token cannot be resolved", async () => {

extensions/beam/src/mirror.ts

@@ -293,11 +293,17 @@ export function createBeamMirrorRunner(params: {
       },
       body: JSON.stringify(payload),
     });
-    if (!response.ok) {
-      warnThrottled(`beam mirror upload failed (${response.status}) for ${payload.source}`);
-      return false;
+    try {
+      if (!response.ok) {
+        warnThrottled(`beam mirror upload failed (${response.status}) for ${payload.source}`);
+        return false;
+      }
+      return true;
+    } finally {
+      // The mirror uses only the status; cancel the ignored payload so slow
+      // receiver responses cannot retain connection slots across poll retries.
+      await response.body?.cancel().catch(() => undefined);
     }
-    return true;
   };
 
   const buildUpload = async (

extensions/buzz/src/gateway.lifecycle.test.ts

@@ -399,6 +399,7 @@ describe("Buzz gateway lifecycle", () => {
       {
         publicKey: "a".repeat(64),
         sendText: async () => "event-id",
+        sendTyping: async () => {},
         close: async () => {},
       },
     );

extensions/buzz/src/inbound.ts

@@ -4,6 +4,7 @@ import {
 } from "openclaw/plugin-sdk/channel-inbound";
 import { resolveStableChannelMessageIngress } from "openclaw/plugin-sdk/channel-ingress-runtime";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
+import { createSubsystemLogger } from "openclaw/plugin-sdk/logging-core";
 import type { BuzzBus } from "./buzz-bus.js";
 import {
   BUZZ_DIFF_MESSAGE_KIND,
@@ -14,6 +15,8 @@ import { getBuzzRuntime } from "./runtime.js";
 import { buildBuzzTarget, parseBuzzTarget } from "./target.js";
 import type { ResolvedBuzzAccount } from "./types.js";
 
+const log = createSubsystemLogger("buzz/inbound");
+
 function senderLabel(pubkey: string): string {
   return `${pubkey.slice(0, 8)}...${pubkey.slice(-6)}`;
 }
@@ -172,9 +175,7 @@ export async function handleBuzzInbound(params: {
         },
         keepaliveIntervalMs: 3_000,
         onStartError: (error: unknown) => {
-          runtime.error(
-            `[${account.accountId}] Buzz typing failed for ${channelId}: ${String(error)}`,
-          );
+          log.error(`[${account.accountId}] Buzz typing failed for ${channelId}: ${String(error)}`);
         },
       },
     },