vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-10-03 09:39:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 30

fix(channels): bind group-thread participant delivery to the participant session (#145735)

Browse source
This commit is contained in:
Peter Steinberger 2026-09-12 00:26:26 -07:00 • committed by GitHub
parent 70ad5a4093
commit 0eea7170b3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 371 additions and 226 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/channels/broadcast-groups.md
View file

@ -244,6 +244,11 @@ Each agent in a broadcast group maintains completely separate:
- **Tool access** (different allow/deny lists)
- **Memory/context** (separate `IDENTITY.md`, `SOUL.md`, etc.)
On Discord, Slack, and Telegram, reply delivery and completion hooks use the
responding participant's session, and local media resolves with that participant's
media roots. This also applies to qualified entries with one participant, whose
replies do not have a participant name label.
On WhatsApp, one input is shared on purpose: the **group context buffer** (recent group messages used for context) is shared per peer, so all broadcast agents see the same context when triggered. It is cleared once after the fan-out completes.
This allows each agent to have different personalities, models, skills, and tool access (for example read-only vs. read-write).

4
extensions/discord/src/monitor/message-handler.process.ack.test.ts
View file

@ -384,7 +384,7 @@ describe("processDiscordMessage ack reactions", () => {
it("falls back to plain ack when status reactions are disabled", async () => {
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
await params?.replyOptions?.onReasoningStream?.();
await params?.replyOptions?.onReasoningStream?.({});
return createNoQueuedDispatchResult();
});
@ -406,7 +406,7 @@ describe("processDiscordMessage ack reactions", () => {
it("keeps one acknowledgement through reasoning, tools, compaction, silence, and success", async () => {
vi.useFakeTimers();
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
await params?.replyOptions?.onReasoningStream?.();
await params?.replyOptions?.onReasoningStream?.({});
await vi.advanceTimersByTimeAsync(DEFAULT_TIMING.debounceMs);
await params?.replyOptions?.onToolStart?.({ name: "exec", phase: "start" });
await vi.advanceTimersByTimeAsync(DEFAULT_TIMING.debounceMs);

109
extensions/discord/src/monitor/message-handler.process.group-thread.test.ts Normal file
View file

@ -0,0 +1,109 @@
import { resolveGroupThreadMentionFacts } from "openclaw/plugin-sdk/channel-inbound";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { setReplyPayloadMetadata } from "openclaw/plugin-sdk/reply-payload-testing";
import * as replyRuntime from "openclaw/plugin-sdk/reply-runtime";
import { describe, expect, it, vi } from "vitest";
import {
BASE_CHANNEL_ROUTE,
createAutomaticSourceDeliveryContext,
createDiscordDraftStream,
deliverDiscordReply,
dispatchBufferedReplyForTest,
registerDiscordProcessTestLifecycle,
runProcessDiscordMessage,
} from "./message-handler.process.test-harness.js";
registerDiscordProcessTestLifecycle();
describe("Discord group-thread participant delivery", () => {
it.each([
{ name: "ordinary route", agents: undefined },
{ name: "unlabeled participant", agents: ["alice"] },
{ name: "parallel participants", agents: ["alice", "bob"] },
{ name: "deferred warning", agents: ["alice"], warning: true },
])("binds delivery to the $name", async ({ agents, warning }) => {
const cfg: OpenClawConfig = {
agents: {
ownership: "explicit",
entries: {
main: { workspace: "/tmp/.openclaw/workspace-main" },
alice: { workspace: "/tmp/.openclaw/workspace-alice" },
bob: { workspace: "/tmp/.openclaw/workspace-bob" },
},
},
broadcast: agents ? { "discord:c1": agents } : undefined,
};
const ctx = await createAutomaticSourceDeliveryContext({
cfg,
route: BASE_CHANNEL_ROUTE,
baseSessionKey: BASE_CHANNEL_ROUTE.sessionKey,
discordConfig: { streaming: { mode: "partial" } },
groupThread: resolveGroupThreadMentionFacts({
cfg,
channel: "discord",
peerId: "c1",
text: "Review this attachment.",
}),
});
const actual = await vi.importActual<typeof replyRuntime>("openclaw/plugin-sdk/reply-runtime");
const errors = vi.spyOn(ctx.runtime, "error");
const participantRuns: string[] = [];
dispatchBufferedReplyForTest.mockImplementationOnce((params) =>
actual.dispatchReplyWithBufferedBlockDispatcher({
...params,
dispatchReplyFromConfig: async ({ ctx: participant, dispatcher }) => {
const agentId = participant.AgentId ?? "main";
participantRuns.push(agentId);
dispatcher.sendBlockReply({
text: `Reasoning from ${agentId}`,
isReasoning: true,
mediaUrl: `/tmp/.openclaw/workspace-${agentId}/reasoning.txt`,
});
const queuedFinal = dispatcher.sendFinalReply(
warning
? setReplyPayloadMetadata(
{ text: "The attachment could not be processed.", isError: true },
{ nonTerminalToolErrorWarning: true },
)
: {
text: `Answer from ${agentId}`,
mediaUrl: `/tmp/.openclaw/workspace-${agentId}/answer.txt`,
},
);
return { queuedFinal, counts: dispatcher.getQueuedCounts() };
},
}),
);
await runProcessDiscordMessage(ctx);
const responders = agents ?? ["main"];
expect(errors.mock.calls).toEqual([]);
expect(participantRuns).toEqual(responders);
expect(deliverDiscordReply).toHaveBeenCalledTimes(responders.length * 2);
for (const agentId of responders) {
for (const kind of ["block", "final"]) {
expect(deliverDiscordReply).toHaveBeenCalledWith(
expect.objectContaining({
target: "channel:c1",
accountId: "default",
sessionKey: `agent:${agentId}:discord:channel:c1`,
mediaLocalRoots: expect.arrayContaining([`/tmp/.openclaw/workspace-${agentId}`]),
kind,
replies: [
expect.objectContaining(
warning && kind === "final"
? { text: "The attachment could not be processed." }
: {
mediaUrl: `/tmp/.openclaw/workspace-${agentId}/${kind === "block" ? "reasoning" : "answer"}.txt`,
},
),
],
}),
);
}
}
if (agents) {
expect(createDiscordDraftStream).not.toHaveBeenCalled();
}
});
});

4
extensions/discord/src/monitor/message-handler.process.room-events.test.ts
View file

@ -29,7 +29,7 @@ describe("processDiscordMessage session routing and room events", () => {
it("suppresses Discord reactions for room events when ack scope does not force all messages", async () => {
vi.useFakeTimers();
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
await params?.replyOptions?.onReasoningStream?.();
await params?.replyOptions?.onReasoningStream?.({});
await new Promise((resolve) => {
setTimeout(resolve, 1_000);
});
@ -67,7 +67,7 @@ describe("processDiscordMessage session routing and room events", () => {
it("sends Discord ack reactions for room events when ack scope is all", async () => {
vi.useFakeTimers();
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
await params?.replyOptions?.onReasoningStream?.();
await params?.replyOptions?.onReasoningStream?.({});
await new Promise((resolve) => {
setTimeout(resolve, 1_000);
});

2
extensions/discord/src/monitor/message-handler.process.session-routing.test.ts
View file

@ -467,7 +467,7 @@ describe("processDiscordMessage session routing", () => {
it("honors explicit status reactions for always-on guild replies", async () => {
vi.useFakeTimers();
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
await params?.replyOptions?.onReasoningStream?.();
await params?.replyOptions?.onReasoningStream?.({});
await new Promise((resolve) => {
setTimeout(resolve, 1_000);
});

209
extensions/discord/src/monitor/message-handler.process.test-harness.ts
View file

@ -159,80 +159,7 @@ export type DispatchInboundParams = {
sendFinalReply: (payload: ReplyPayload) => boolean | Promise<boolean>;
waitForIdle: () => Promise<void>;
};
replyOptions?: {
onReasoningStream?: (payload?: {
text?: string;
isReasoningSnapshot?: boolean;
requiresReasoningProgressOptIn?: boolean;
}) => Promise<void> | void;
onReasoningEnd?: () => Promise<void> | void;
onToolStart?: (payload: {
itemId?: string;
toolCallId?: string;
name?: string;
phase?: string;
args?: Record<string, unknown>;
detailMode?: "explain" | "raw";
}) => Promise<void> | void;
onItemEvent?: (payload: {
itemId?: string;
kind?: string;
phase?: string;
status?: string;
progressText?: string;
summary?: string;
title?: string;
name?: string;
}) => Promise<boolean | void> | boolean | void;
onNarrationUpdate?: (payload: { text: string }) => Promise<void> | void;
onProgressNarratorLifecycle?: (lifecycle: {
beginTurn: () => void;
stopTurn: () => void;
}) => void;
isProgressDraftVisible?: () => boolean;
progressPreambleEnabled?: boolean;
narrationHideCommandText?: boolean;
commentaryPayloadsEnabled?: boolean;
shouldDeliverCommentaryPayloads?: () => boolean;
onVerboseProgressVisibility?: (isActive: () => boolean) => void;
onPlanUpdate?: (payload: {
phase?: string;
explanation?: string;
steps?: Array<{ step: string; status: "pending" | "in_progress" | "completed" }>;
}) => Promise<void> | void;
onApprovalEvent?: (payload: { phase?: string; command?: string }) => Promise<void> | void;
onCommandOutput?: (payload: {
toolCallId?: string;
phase?: string;
name?: string;
title?: string;
status?: string;
exitCode?: number | null;
}) => Promise<false | void> | false | void;
onPatchSummary?: (payload: {
phase?: string;
summary?: string;
title?: string;
name?: string;
added?: string[];
modified?: string[];
deleted?: string[];
}) => Promise<void> | void;
onReplyStart?: () => Promise<void> | void;
sourceReplyDeliveryMode?: "automatic" | "message_tool_only";
typingKeepalive?: boolean;
disableBlockStreaming?: boolean;
suppressDefaultToolProgressMessages?: boolean;
queuedDeliveryCorrelations?: Array<{ begin: () => () => void }>;
suppressTyping?: boolean;
onCompactionStart?: () => Promise<void> | void;
onCompactionEnd?: () => Promise<void> | void;
onPartialReply?: (payload: { text?: string }) => Promise<void> | void;
onAssistantMessageStart?: () => Promise<void> | void;
onQueuedFollowupAdmitted?: () => Promise<void> | void;
allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean;
onTypingCleanup?: () => Promise<void> | void;
};
replyOptions?: import("openclaw/plugin-sdk/reply-runtime").GetReplyOptions;
};
const dispatchInboundMessage = vi.hoisted(() =>
vi.fn<
@ -305,82 +232,70 @@ let processDiscordMessage: typeof import("./message-handler.process.js").process
export let formatDiscordReplySkip: typeof import("./message-handler.process.js").formatDiscordReplySkip;
export let discordInboundEventDelivery: typeof import("../inbound-event-delivery.js").discordInboundEventDelivery;
const dispatchBufferedReply = vi.hoisted(() =>
vi.fn<
typeof import("openclaw/plugin-sdk/reply-runtime").dispatchReplyWithBufferedBlockDispatcher
>(),
);
export const dispatchBufferedReplyForTest = dispatchBufferedReply;
vi.mock("openclaw/plugin-sdk/reply-runtime", () => ({
dispatchReplyWithBufferedBlockDispatcher: async (params: {
dispatcherOptions: {
beforeDeliver?: (
payload: ReplyPayload,
info: { kind: "block" | "final" },
) => Promise<ReplyPayload | null> | ReplyPayload | null;
deliver: (payload: unknown, info: { kind: "block" | "final" }) => Promise<void> | void;
onError?: (err: unknown, info: { kind: "block" | "final" }) => void;
transformReplyPayload?: (payload: ReplyPayload) => ReplyPayload | null;
typingCallbacks?: {
onReplyStart?: () => Promise<void> | void;
onIdle?: () => void;
onCleanup?: () => void;
dispatchReplyWithBufferedBlockDispatcher: dispatchBufferedReply.mockImplementation(
async (params) => {
const pendingDeliveries: Promise<void>[] = [];
const deliver = async (payload: ReplyPayload, info: { kind: "block" | "final" }) => {
const transformed = params.dispatcherOptions.transformReplyPayload
? params.dispatcherOptions.transformReplyPayload(payload)
: payload;
if (!transformed) {
return;
}
const deliverPayload = params.dispatcherOptions.beforeDeliver
? await params.dispatcherOptions.beforeDeliver(transformed, info)
: transformed;
if (!deliverPayload) {
return;
}
await params.dispatcherOptions.deliver(deliverPayload, info);
};
onReplyStart?: () => Promise<void> | void;
onIdle?: () => void;
onCleanup?: () => void;
onSettled?: () => unknown;
onFreshSettledDelivery?: () => unknown;
};
ctx?: Record<string, unknown>;
replyOptions?: DispatchInboundParams["replyOptions"];
}) => {
const pendingDeliveries: Promise<void>[] = [];
const deliver = async (payload: ReplyPayload, info: { kind: "block" | "final" }) => {
const transformed = params.dispatcherOptions.transformReplyPayload
? params.dispatcherOptions.transformReplyPayload(payload)
: payload;
if (!transformed) {
return;
const queueDelivery = (payload: ReplyPayload, info: { kind: "block" | "final" }) => {
const delivery = Promise.resolve(deliver(payload, info)).catch(async (err: unknown) => {
await params.dispatcherOptions.onError?.(err, info);
});
pendingDeliveries.push(delivery);
return true;
};
const typingCallbacks = params.dispatcherOptions.typingCallbacks;
const replyOptions = {
...params.replyOptions,
onReplyStart: params.dispatcherOptions.onReplyStart ?? typingCallbacks?.onReplyStart,
onTypingCleanup: params.dispatcherOptions.onCleanup ?? typingCallbacks?.onCleanup,
};
try {
return await dispatchInboundMessage({
ctx: params.ctx,
replyOptions,
dispatcher: {
sendBlockReply: vi.fn((payload: ReplyPayload) =>
queueDelivery(payload, { kind: "block" }),
),
sendFinalReply: vi.fn((payload: ReplyPayload) =>
queueDelivery(payload, { kind: "final" }),
),
waitForIdle: vi.fn(async () => {
await Promise.all(pendingDeliveries);
}),
},
});
} finally {
await params.dispatcherOptions.onSettled?.();
await params.dispatcherOptions.onFreshSettledDelivery?.();
await params.dispatcherOptions.onIdle?.();
typingCallbacks?.onIdle?.();
}
const deliverPayload = params.dispatcherOptions.beforeDeliver
? await params.dispatcherOptions.beforeDeliver(transformed, info)
: transformed;
if (!deliverPayload) {
return;
}
await params.dispatcherOptions.deliver(deliverPayload, info);
};
const queueDelivery = (payload: ReplyPayload, info: { kind: "block" | "final" }) => {
const delivery = Promise.resolve(deliver(payload, info)).catch((err: unknown) => {
params.dispatcherOptions.onError?.(err, info);
});
pendingDeliveries.push(delivery);
return true;
};
const typingCallbacks = params.dispatcherOptions.typingCallbacks;
const replyOptions = {
...params.replyOptions,
onReplyStart: params.dispatcherOptions.onReplyStart ?? typingCallbacks?.onReplyStart,
onTypingCleanup: params.dispatcherOptions.onCleanup ?? typingCallbacks?.onCleanup,
};
try {
return await dispatchInboundMessage({
ctx: params.ctx,
replyOptions,
dispatcher: {
sendBlockReply: vi.fn((payload: ReplyPayload) =>
queueDelivery(payload, { kind: "block" }),
),
sendFinalReply: vi.fn((payload: ReplyPayload) =>
queueDelivery(payload, { kind: "final" }),
),
waitForIdle: vi.fn(async () => {
await Promise.all(pendingDeliveries);
}),
},
});
} finally {
await params.dispatcherOptions.onSettled?.();
await params.dispatcherOptions.onFreshSettledDelivery?.();
params.dispatcherOptions.onIdle?.();
typingCallbacks?.onIdle?.();
}
},
},
),
dispatchInboundMessage: (params: DispatchInboundParams) => dispatchInboundMessage(params),
settleReplyDispatcher: async (params: {
dispatcher: { markComplete: () => void; waitForIdle: () => Promise<void> };

84
extensions/discord/src/monitor/message-handler.process.ts
View file

@ -3,6 +3,7 @@ import type { APIAllowedMentions } from "discord-api-types/v10";
import { resolveAgentConfig, resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime";
import {
dispatchChannelInboundTurn,
getGroupThreadDeliverySession,
hasFinalInboundReplyDispatch,
readAgentRunTerminalOutcome,
} from "openclaw/plugin-sdk/channel-inbound";
@ -212,7 +213,11 @@ export async function processDiscordMessage(
let userFacingFinalDelivered = false;
let userFacingFinalDeliveryFailed = false;
let pendingToolWarningFinal:
| { payload: ReplyPayload; info: DiscordProviderDeliveryInfo }
| {
payload: ReplyPayload;
info: DiscordProviderDeliveryInfo;
deliverySession: ReturnType<typeof getGroupThreadDeliverySession>;
}
| undefined;
const markFinalReplyDelivered = (isError = false) => {
draftPreview.markFinalReplyDelivered(isError);
@ -264,6 +269,7 @@ export async function processDiscordMessage(
options?: {
allowFallbackOnlyToolWarning?: boolean;
allowProgressBlock?: boolean;
deliverySession?: ReturnType<typeof getGroupThreadDeliverySession>;
},
) => {
if (abortSignal?.aborted) {
@ -279,6 +285,27 @@ export async function processDiscordMessage(
);
return { visibleReplySent: false };
}
const deliverySession = options?.deliverySession ?? getGroupThreadDeliverySession();
const deliveryOptions = {
cfg,
token,
accountId,
rest: reactions.deliveryRest,
runtime,
replyToMode,
textLimit,
maxLinesPerMessage,
tableMode,
chunkMode,
sessionKey: deliverySession?.sessionKey ?? ctxPayload.SessionKey,
threadBindings,
mediaLocalRoots: deliverySession
? getAgentScopedMediaLocalRoots(cfg, deliverySession.agentId)
: mediaLocalRoots,
bindPendingFinalDelivery: info.bindPendingFinalDelivery,
onPlatformSendDispatch: info.onPlatformSendDispatch,
assertPlatformSendAuthorized: info.assertPlatformSendAuthorized,
};
let payload = incomingPayload;
if (info.participant && (payload.text || payload.mediaUrl || payload.mediaUrls?.length)) {
payload = {
@ -307,26 +334,11 @@ export async function processDiscordMessage(
return { visibleReplySent: false };
}
const result = await deliverDiscordReply({
cfg,
...deliveryOptions,
replies,
target: deliverTarget,
token,
accountId,
rest: reactions.deliveryRest,
runtime,
replyToId: replyReference.use(),
replyToMode,
textLimit,
maxLinesPerMessage,
tableMode,
chunkMode,
sessionKey: ctxPayload.SessionKey,
threadBindings,
mediaLocalRoots,
kind: "block",
bindPendingFinalDelivery: info.bindPendingFinalDelivery,
onPlatformSendDispatch: info.onPlatformSendDispatch,
assertPlatformSendAuthorized: info.assertPlatformSendAuthorized,
});
if (result.visibleReplySent) {
replyReference.markSent();
@ -342,7 +354,8 @@ export async function processDiscordMessage(
!userFacingFinalDelivered &&
(!finalReplyStartNotified || userFacingFinalDeliveryFailed)
) {
pendingToolWarningFinal = { payload, info };
// Root settlement can outlive this participant's dispatch scope.
pendingToolWarningFinal = { payload, info, deliverySession };
}
return { visibleReplySent: false };
}
@ -464,27 +477,12 @@ export async function processDiscordMessage(
const replyToId = replyReference.use();
notifyFinalReplyStart();
const deliveryResult = await deliverDiscordReply({
cfg,
...deliveryOptions,
replies: [fallbackPayload],
target: deliverTarget,
token,
accountId,
rest: reactions.deliveryRest,
runtime,
replyToId,
replyToMode,
textLimit,
maxLinesPerMessage,
tableMode,
chunkMode,
sessionKey: ctxPayload.SessionKey,
threadBindings,
mediaLocalRoots,
allowedMentions,
kind: info.kind,
bindPendingFinalDelivery: info.bindPendingFinalDelivery,
onPlatformSendDispatch: info.onPlatformSendDispatch,
assertPlatformSendAuthorized: info.assertPlatformSendAuthorized,
});
return deliveryResult.visibleReplySent;
},
@ -516,26 +514,11 @@ export async function processDiscordMessage(
notifyFinalReplyStart();
}
const result = await deliverDiscordReply({
cfg,
...deliveryOptions,
replies: [deliverablePayload],
target: deliverTarget,
token,
accountId,
rest: reactions.deliveryRest,
runtime,
replyToId,
replyToMode,
textLimit,
maxLinesPerMessage,
tableMode,
chunkMode,
sessionKey: ctxPayload.SessionKey,
threadBindings,
mediaLocalRoots,
kind: info.kind,
bindPendingFinalDelivery: info.bindPendingFinalDelivery,
onPlatformSendDispatch: info.onPlatformSendDispatch,
assertPlatformSendAuthorized: info.assertPlatformSendAuthorized,
});
if (!result.visibleReplySent) {
return result;
@ -587,6 +570,7 @@ export async function processDiscordMessage(
try {
return await deliverDiscordPayload(pending.payload, pending.info, {
allowFallbackOnlyToolWarning: true,
deliverySession: pending.deliverySession,
});
} catch (err) {
dispatchError = true;

135
extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts
View file

@ -6,6 +6,7 @@ import {
} from "openclaw/plugin-sdk/channel-test-helpers";
import {
createReplyDispatcher,
finalizeInboundContext,
type GetReplyOptions,
type ReplyPayload,
} from "openclaw/plugin-sdk/reply-runtime";
@ -21,10 +22,17 @@ const SAME_TEXT = "same reply";
const getGlobalHookRunnerMock = vi.hoisted(() => vi.fn());
const createSlackDraftStreamMock = vi.fn();
type DeliveryParams = Omit<
Parameters<typeof import("../replies.js").deliverReplies>[0],
"replies"
> & {
replies: ReplyPayload[];
};
const deliverRepliesMock = vi.fn(
async (_params: { replies: ReplyPayload[] }) =>
async (_params: DeliveryParams) =>
undefined as { messageId?: string; channelId?: string } | undefined,
);
const sendMessageSlackMock = vi.fn<typeof import("../send.runtime.js").sendMessageSlack>();
const finalizeSlackPreviewEditMock = vi.fn(async (_input: { blocks?: unknown }) => {});
const normalizeSlackOutboundTextMock = vi.fn((value: string) => value.trim());
const postMessageMock = vi.fn(async () => ({ ok: true, ts: "171234.999" }));
@ -109,6 +117,7 @@ let mockedQueuedDispatchCounts: TestDispatchCounts = { tool: 0, block: 0, final:
let mockedAgentRunTerminalOutcome: "completed" | "failed" | undefined;
let mockedSourceReplyDelivered = false;
let mockedDispatchError: Error | undefined;
let useRealChannelInboundTurn = false;
let mockedProgressEvents: string[] = [];
let mockedEmptyProgressToolName: string | undefined;
@ -839,7 +848,8 @@ vi.mock("openclaw/plugin-sdk/reply-history", () => ({
}),
}));
vi.mock("openclaw/plugin-sdk/reply-payload", () => ({
vi.mock("openclaw/plugin-sdk/reply-payload", async (importOriginal) => ({
...(await importOriginal<typeof import("openclaw/plugin-sdk/reply-payload")>()),
resolveAskUserQuestionOptionIndices: () => undefined,
isReplyPayloadNonTerminalToolErrorWarning: () => false,
buildTtsSupplementMediaPayload: (payload: {
@ -1005,6 +1015,8 @@ vi.mock("../replies.js", async (importOriginal) => ({
resolveSlackThreadTs: () => mockedReplyThreadTs,
}));
vi.mock("../send.runtime.js", () => ({ sendMessageSlack: sendMessageSlackMock }));
vi.mock("openclaw/plugin-sdk/channel-inbound", async (importOriginal) => {
const actual = await importOriginal<typeof import("openclaw/plugin-sdk/channel-inbound")>();
type DispatchParams = Parameters<typeof actual.dispatchChannelInboundTurn>[0];
@ -1012,6 +1024,9 @@ vi.mock("openclaw/plugin-sdk/channel-inbound", async (importOriginal) => {
...actual,
readAgentRunTerminalOutcome: () => mockedAgentRunTerminalOutcome,
dispatchChannelInboundTurn: async (params: DispatchParams) => {
if (useRealChannelInboundTurn) {
return actual.dispatchChannelInboundTurn(params);
}
capturedReplyOptions = params.replyOptions as typeof capturedReplyOptions;
capturedDispatchReplyFromConfig = params.dispatchReplyFromConfig;
if (mockedReplyOptionEvents.length > 0) {
@ -1167,6 +1182,8 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
);
createSlackDraftStreamMock.mockReset();
deliverRepliesMock.mockReset();
sendMessageSlackMock.mockReset();
useRealChannelInboundTurn = false;
finalizeSlackPreviewEditMock.mockReset();
normalizeSlackOutboundTextMock.mockClear();
postMessageMock.mockClear();
@ -1222,6 +1239,120 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
afterEach(() => resetPluginRuntimeStateForTest());
it.each([
{ agents: ["alice"], withMedia: true },
{ agents: ["alice", "bob"], withMedia: true },
{ agents: ["alice", "bob"], withMedia: false },
])(
"binds group-thread completion hooks and media to the participant: $agents (media: $withMedia)",
async ({ agents, withMedia }) => {
useRealChannelInboundTurn = true;
mockedNativeStreaming = true;
const { resolveGroupThreadMentionFacts } =
await import("openclaw/plugin-sdk/channel-inbound");
const { createMessageReceiptFromOutboundResults } =
await import("openclaw/plugin-sdk/channel-outbound");
const cfg = {
agents: {
entries: {
root: { workspace: "/tmp/.openclaw/workspace-root" },
alice: { workspace: "/tmp/.openclaw/workspace-alice" },
bob: { workspace: "/tmp/.openclaw/workspace-bob" },
},
},
broadcast: { "slack:C123": agents },
};
const rootSessionKey = `agent:root:slack:channel:c123:thread:${THREAD_TS}`;
const actualReplies = await vi.importActual<typeof import("../replies.js")>("../replies.js");
const { prepareSlackReply } = await import("../../reply-blocks.js");
deliverRepliesMock.mockImplementation(async (params) =>
actualReplies.deliverReplies({ ...params, replies: params.replies.map(prepareSlackReply) }),
);
sendMessageSlackMock.mockResolvedValue({
messageId: "sent-1",
channelId: "C123",
receipt: createMessageReceiptFromOutboundResults({
results: [{ channel: "slack", messageId: "sent-1", channelId: "C123" }],
kind: withMedia ? "media" : "text",
}),
});
const participantRuns: string[] = [];
const dispatchReplyFromConfig: NonNullable<
Parameters<typeof dispatchPreparedSlackMessage>[0]["ctx"]["dispatchReplyFromConfig"]
> = async ({ ctx, dispatcher }) => {
if (!ctx.AgentId) {
throw new Error("Expected participant agent identity");
}
participantRuns.push(ctx.AgentId);
return {
queuedFinal: dispatcher.sendFinalReply({
text: `Reply from ${ctx.AgentId}`,
...(withMedia
? { mediaUrl: `/tmp/.openclaw/workspace-${ctx.AgentId}/attachment.txt` }
: {}),
}),
counts: dispatcher.getQueuedCounts(),
};
};
await dispatchPreparedSlackMessage(
createPreparedSlackMessage({
cfg,
route: { agentId: "root", sessionKey: rootSessionKey },
ctxPayload: finalizeInboundContext({
AgentId: "root",
SessionKey: rootSessionKey,
ChatType: "channel",
Provider: "slack",
Surface: "slack",
OriginatingChannel: "slack",
OriginatingTo: "channel:C123",
NativeChannelId: "C123",
AccountId: "default",
From: "slack:C123",
To: "channel:C123",
SenderId: "U123",
MessageSid: "171234.111",
MessageThreadId: THREAD_TS,
Body: "Review the attachment.",
GroupThread: resolveGroupThreadMentionFacts({
cfg,
channel: "slack",
peerId: "C123",
text: "Review the attachment.",
sessionKey: rootSessionKey,
}),
}),
dispatchReplyFromConfig,
}),
);
expect(participantRuns.toSorted()).toEqual(agents.toSorted());
expect(emitSlackMessageSentHooksMock).toHaveBeenCalledTimes(agents.length);
expect(sendMessageSlackMock).toHaveBeenCalledTimes(agents.length);
for (const agentId of agents) {
expect(emitSlackMessageSentHooksMock).toHaveBeenCalledWith(
expect.objectContaining({
sessionKeyForInternalHooks: `agent:${agentId}:slack:channel:c123:thread:${THREAD_TS}`,
success: true,
}),
);
expect(sendMessageSlackMock).toHaveBeenCalledWith(
"channel:C123",
expect.stringContaining(`Reply from ${agentId}`),
expect.objectContaining({
...(withMedia
? { mediaUrl: `/tmp/.openclaw/workspace-${agentId}/attachment.txt` }
: {}),
mediaLocalRoots: expect.arrayContaining([`/tmp/.openclaw/workspace-${agentId}`]),
}),
);
}
expect(startSlackStreamMock).not.toHaveBeenCalled();
expect(createSlackDraftStreamMock).not.toHaveBeenCalled();
},
);
it.each([false, true])(
"delivers literal authored fallback normally with native streaming %s",
async (nativeStreaming) => {

45
extensions/slack/src/monitor/replies.ts
View file

@ -1,10 +1,14 @@
// Slack plugin module implements replies behavior.
import type { MessageMetadata } from "@slack/types";
import type { Block, KnownBlock } from "@slack/web-api";
import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound";
import {
createChannelPartialDeliveryError,
getGroupThreadDeliverySession,
} from "openclaw/plugin-sdk/channel-inbound";
import { createMessageReceiptFromOutboundResults } from "openclaw/plugin-sdk/channel-outbound";
import type { MarkdownTableMode, OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
import { getAgentScopedMediaLocalRoots } from "openclaw/plugin-sdk/media-runtime";
import {
chunkMarkdownTextWithMode,
isSilentReplyText,
@ -154,6 +158,12 @@ export async function deliverReplies(params: {
/** Validated non-serializable client scope for an enterprise listener turn. */
eventScope?: SlackEventScope;
}) {
const deliverySession = getGroupThreadDeliverySession();
const sessionKeyForInternalHooks =
deliverySession?.sessionKey ?? params.sessionKeyForInternalHooks;
const mediaLocalRoots = deliverySession
? getAgentScopedMediaLocalRoots(params.cfg, deliverySession.agentId)
: undefined;
let latestResult: SlackSendResult | undefined;
for (const prepared of params.replies) {
const { payload } = prepared;
@ -196,6 +206,7 @@ export async function deliverReplies(params: {
acceptedResults.push(result);
},
...(input.mediaUrl ? { mediaUrl: input.mediaUrl } : {}),
...(mediaLocalRoots ? { mediaLocalRoots } : {}),
...(input.blocks ? { blocks: input.blocks } : {}),
...(input.authoredTextPlacement
? { authoredTextPlacement: input.authoredTextPlacement }
@ -223,32 +234,19 @@ export async function deliverReplies(params: {
// `emitMessageSentHooks` in `extensions/telegram/src/bot/delivery.replies.ts`.
// `emitSlackMessageSentHooks` self-gates on registered listeners, so this is
// a no-op when no plugin observes `message_sent`.
const emitSent = (content: string, result?: SlackSendResult) => {
const emitDelivery = (
content: string,
result: { success: true; messageId?: string } | { success: false; error: string },
) => {
if (params.deferMessageSentHooks) {
return;
}
emitSlackMessageSentHooks({
sessionKeyForInternalHooks: params.sessionKeyForInternalHooks,
sessionKeyForInternalHooks,
to: params.messageSentHookTarget ?? params.target,
accountId: params.accountId,
content,
success: true,
messageId: result?.messageId,
isGroup: params.isGroup,
groupId: params.groupId,
});
};
const emitFailed = (content: string, error: unknown) => {
if (params.deferMessageSentHooks) {
return;
}
emitSlackMessageSentHooks({
sessionKeyForInternalHooks: params.sessionKeyForInternalHooks,
to: params.messageSentHookTarget ?? params.target,
accountId: params.accountId,
content,
success: false,
error: formatErrorMessage(error),
...result,
isGroup: params.isGroup,
groupId: params.groupId,
});
@ -304,7 +302,7 @@ export async function deliverReplies(params: {
}
} catch (error) {
const hookContent = hookParts.join("\n\n") || textRaw || spokenText || "";
emitFailed(hookContent, error);
emitDelivery(hookContent, { success: false, error: formatErrorMessage(error) });
if (acceptedResults.length === 0) {
throw error;
}
@ -319,7 +317,10 @@ export async function deliverReplies(params: {
const hookContent = hookParts.join("\n\n") || textRaw || spokenText || "";
// Preserve the media hook contract even when a trailing block send has a
// message `ts`; the logical payload still spans multiple Slack objects.
emitSent(hookContent, reply.hasMedia ? undefined : lastResult);
emitDelivery(hookContent, {
success: true,
messageId: reply.hasMedia ? undefined : lastResult?.messageId,
});
latestResult = lastResult;
params.runtime.log?.(`delivered reply to ${params.target}`);
}