From a3155ca99a8945150113f6b46a63feda3f752f07 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 19 Jun 2026 13:38:37 +0100 Subject: [PATCH] fix(slack): retain partial stream participation --- extensions/discord/src/pluralkit.test.ts | 48 +++++++++++++++++++ extensions/discord/src/pluralkit.ts | 6 ++- .../message-handler.authz.test.ts | 12 +++-- .../src/monitor-handler/message-handler.ts | 2 +- .../monitor/message-handler/prepare.test.ts | 9 ++-- .../src/monitor/message-handler/prepare.ts | 2 +- src/plugins/memory-state.test.ts | 3 +- src/plugins/memory-state.ts | 2 - .../registry.dual-kind-memory-gate.test.ts | 4 +- 9 files changed, 72 insertions(+), 16 deletions(-) diff --git a/extensions/discord/src/pluralkit.test.ts b/extensions/discord/src/pluralkit.test.ts index 3b5e64fd74cc..b5141b8ba42f 100644 --- a/extensions/discord/src/pluralkit.test.ts +++ b/extensions/discord/src/pluralkit.test.ts @@ -20,6 +20,28 @@ const buildResponse = (params: { status: number; body?: unknown }): MockResponse }; }; +function cancelTrackedResponse( + text: string, + init: ResponseInit, +): { + response: Response; + wasCanceled: () => boolean; +} { + let canceled = false; + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + cancel() { + canceled = true; + }, + }); + return { + response: new Response(stream, init), + wasCanceled: () => canceled, + }; +} + describe("fetchPluralKitMessageInfo", () => { it("returns null when disabled", async () => { const fetcher = vi.fn(); @@ -65,4 +87,30 @@ describe("fetchPluralKitMessageInfo", () => { expect(result?.member?.id).toBe("mem_1"); expect(receivedHeaders?.Authorization).toBe("pk_test"); }); + + it("bounds PluralKit API error bodies without using response.text()", async () => { + const tracked = cancelTrackedResponse(`${"plural failure ".repeat(1024)}tail`, { + status: 500, + headers: { "content-type": "text/plain" }, + }); + const textSpy = vi.spyOn(tracked.response, "text").mockRejectedValue(new Error("unbounded")); + const fetcher = vi.fn(async () => tracked.response); + + let caught: Error | undefined; + try { + await fetchPluralKitMessageInfo({ + messageId: "boom", + config: { enabled: true }, + fetcher: fetcher as unknown as typeof fetch, + }); + } catch (error) { + caught = error as Error; + } + + expect(caught?.message).toContain("PluralKit API failed (500): plural failure"); + expect(caught?.message).not.toContain("tail"); + expect(caught?.message.length).toBeLessThan(8_400); + expect(tracked.wasCanceled()).toBe(true); + expect(textSpy).not.toHaveBeenCalled(); + }); }); diff --git a/extensions/discord/src/pluralkit.ts b/extensions/discord/src/pluralkit.ts index ba746cbc3dc8..1e24a4fef672 100644 --- a/extensions/discord/src/pluralkit.ts +++ b/extensions/discord/src/pluralkit.ts @@ -1,7 +1,9 @@ // Discord plugin module implements pluralkit behavior. import { resolveFetch } from "openclaw/plugin-sdk/fetch-runtime"; +import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http"; const PLURALKIT_API_BASE = "https://api.pluralkit.me/v2"; +const PLURALKIT_ERROR_BODY_LIMIT_BYTES = 8 * 1024; export type DiscordPluralKitConfig = { enabled?: boolean; @@ -51,7 +53,9 @@ export async function fetchPluralKitMessageInfo(params: { return null; } if (!res.ok) { - const text = await res.text().catch(() => ""); + const text = await readResponseTextLimited(res, PLURALKIT_ERROR_BODY_LIMIT_BYTES).catch( + () => "", + ); const detail = text.trim() ? `: ${text.trim()}` : ""; throw new Error(`PluralKit API failed (${res.status})${detail}`); } diff --git a/extensions/msteams/src/monitor-handler/message-handler.authz.test.ts b/extensions/msteams/src/monitor-handler/message-handler.authz.test.ts index 9bc7f48793c8..ad85f9852749 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.authz.test.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.authz.test.ts @@ -734,7 +734,7 @@ describe("msteams monitor handler authz", () => { expect(ctxPayload.CommandAuthorized).toBe(true); }); - it("marks skipped channel message system events as non-owner", async () => { + it("marks skipped channel message system events as non-owner without duplicating body text", async () => { resetThreadMocks(); const { deps, enqueueSystemEvent } = createDeps({ channels: { @@ -768,15 +768,16 @@ describe("msteams monitor handler authz", () => { expect(runtimeApiMockState.dispatchReplyFromConfigWithSettledDispatcher).not.toHaveBeenCalled(); const systemEventCall = enqueueSystemEvent.mock.calls.find( - ([text]) => typeof text === "string" && text.includes("please run the deployment"), + ([text]) => text === "Teams message in channel from Member", ); if (!systemEventCall) { throw new Error("expected skipped Teams message system event"); } expect(systemEventCall[1]).toMatchObject({}); + expect(systemEventCall[0]).not.toContain("please run the deployment"); }); - it("keeps dispatched primary message system events owner-neutral", async () => { + it("keeps dispatched primary message system events owner-neutral without duplicating body text", async () => { resetThreadMocks(); const { deps, enqueueSystemEvent } = createDeps({ channels: { @@ -810,11 +811,14 @@ describe("msteams monitor handler authz", () => { expect(runtimeApiMockState.dispatchReplyFromConfigWithSettledDispatcher).toHaveBeenCalled(); const systemEventCall = enqueueSystemEvent.mock.calls.find( - ([text]) => typeof text === "string" && text.includes("please check the build"), + ([text]) => text === "Teams message in channel from Member", ); if (!systemEventCall) { throw new Error("expected active Teams message system event"); } + expect(systemEventCall[0]).not.toContain("please check the build"); + const dispatched = firstSettledDispatch(); + expect(recordFromMockCall(dispatched.ctxPayload).BodyForAgent).toBe("please check the build"); }); it("authorizes text control commands from static access groups", async () => { diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index 66c02a3b2e95..60bdd80644fa 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -508,7 +508,7 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { : `Teams message in ${conversationType} from ${senderName}`; const enqueuePrimaryMessageSystemEvent = () => - core.system.enqueueSystemEvent(`${inboundLabel}: ${preview}`, { + core.system.enqueueSystemEvent(inboundLabel, { sessionKey: route.sessionKey, contextKey: `msteams:message:${conversationId}:${activity.id ?? "unknown"}`, }); diff --git a/extensions/slack/src/monitor/message-handler/prepare.test.ts b/extensions/slack/src/monitor/message-handler/prepare.test.ts index 8e931b033ef5..60728041e117 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.test.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.test.ts @@ -158,14 +158,17 @@ describe("slack prepareSlackMessage inbound contract", () => { }); } - it("queues inbound message system events as untrusted", async () => { - const prepared = await prepareWithDefaultCtx(createSlackMessage({})); + it("queues inbound message system events without duplicating body text", async () => { + const body = + "please summarize the deployment, rollback checks, health checks, and follow-up items"; + const prepared = await prepareWithDefaultCtx(createSlackMessage({ text: body })); assertPrepared(prepared); - expect(enqueueSystemEventMock).toHaveBeenCalledWith("Slack DM from Alice: hi", { + expect(enqueueSystemEventMock).toHaveBeenCalledWith("Slack DM from Alice", { sessionKey: prepared.ctxPayload.SessionKey, contextKey: "slack:message:D123:1.000", }); + expect(prepared.ctxPayload.BodyForAgent).toContain(body); }); it("prepares wildcard open-policy account DMs", async () => { diff --git a/extensions/slack/src/monitor/message-handler/prepare.ts b/extensions/slack/src/monitor/message-handler/prepare.ts index 2c4edbd256d5..2be05dccf9f2 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.ts @@ -1131,7 +1131,7 @@ export async function prepareSlackMessage(params: { ? `slack:channel:${message.channel}` : `slack:group:${message.channel}`; - enqueueSystemEvent(`${inboundLabel}: ${preview}`, { + enqueueSystemEvent(inboundLabel, { sessionKey, contextKey: `slack:message:${message.channel}:${message.ts ?? "unknown"}`, }); diff --git a/src/plugins/memory-state.test.ts b/src/plugins/memory-state.test.ts index 758b3ff5e5c4..5382aa53b9c0 100644 --- a/src/plugins/memory-state.test.ts +++ b/src/plugins/memory-state.test.ts @@ -1,7 +1,6 @@ // Covers plugin-backed memory state registration and reset behavior. import { afterEach, describe, expect, it } from "vitest"; import { - resetMemoryPluginState, buildMemoryPromptSection, clearMemoryPluginState, getMemoryCapabilityRegistration, @@ -363,7 +362,7 @@ describe("memory plugin state", () => { }); const snapshot = createMemoryStateSnapshot(); - resetMemoryPluginState(); + clearMemoryPluginState(); expectClearedMemoryState(); restoreMemoryPluginState(snapshot); diff --git a/src/plugins/memory-state.ts b/src/plugins/memory-state.ts index 974a8c3366f4..5701d472ce88 100644 --- a/src/plugins/memory-state.ts +++ b/src/plugins/memory-state.ts @@ -348,5 +348,3 @@ export function clearMemoryPluginState(): void { memoryPluginState.corpusSupplements = []; memoryPluginState.promptSupplements = []; } - -export const resetMemoryPluginState = clearMemoryPluginState; diff --git a/src/plugins/registry.dual-kind-memory-gate.test.ts b/src/plugins/registry.dual-kind-memory-gate.test.ts index 83b38d774ed7..4c07d52adb08 100644 --- a/src/plugins/registry.dual-kind-memory-gate.test.ts +++ b/src/plugins/registry.dual-kind-memory-gate.test.ts @@ -7,14 +7,14 @@ import { import { afterEach, describe, expect, it } from "vitest"; import { clearMemoryEmbeddingProviders } from "./memory-embedding-providers.js"; import { - resetMemoryPluginState, + clearMemoryPluginState, getMemoryCapabilityRegistration, getMemoryRuntime, } from "./memory-state.js"; import { createPluginRecord } from "./status.test-helpers.js"; afterEach(() => { - resetMemoryPluginState(); + clearMemoryPluginState(); clearMemoryEmbeddingProviders(); });