diff --git a/extensions/beam/src/mirror.test.ts b/extensions/beam/src/mirror.test.ts index 3baedabec967..a693e6783081 100644 --- a/extensions/beam/src/mirror.test.ts +++ b/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, +): typeof fetch { return (async (url: unknown, init?: RequestInit) => { const headers = (init?.headers ?? {}) as Record; 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({ + 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 () => { diff --git a/extensions/beam/src/mirror.ts b/extensions/beam/src/mirror.ts index a7ee6cddf4a1..36f3986fadd2 100644 --- a/extensions/beam/src/mirror.ts +++ b/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 ( diff --git a/extensions/buzz/src/gateway.lifecycle.test.ts b/extensions/buzz/src/gateway.lifecycle.test.ts index fcd7226317b5..66be69a657b1 100644 --- a/extensions/buzz/src/gateway.lifecycle.test.ts +++ b/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 () => {}, }, ); diff --git a/extensions/buzz/src/inbound.ts b/extensions/buzz/src/inbound.ts index 2fb2413705db..f2b4d7ce3f84 100644 --- a/extensions/buzz/src/inbound.ts +++ b/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)}`); }, }, },