fix(mattermost): carry thread context to non-inbound reply paths (#44283)

Merged via squash.

Prepared head SHA: 2846a6cfa9
Co-authored-by: teconomix <6959299+teconomix@users.noreply.github.com>
Co-authored-by: mukhtharcm <56378562+mukhtharcm@users.noreply.github.com>
Reviewed-by: @mukhtharcm
This commit is contained in:
Teconomix 2026-03-14 07:53:23 +01:00 committed by GitHub
parent 17cb60080a
commit 0c926a2c5e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 244 additions and 25 deletions

View File

@ -75,6 +75,7 @@ Docs: https://docs.openclaw.ai
- Config/discovery: accept `discovery.wideArea.domain` in strict config validation so unicast DNS-SD gateway configs no longer fail with an unrecognized-key error. (#35615) Thanks @ingyukoh. - Config/discovery: accept `discovery.wideArea.domain` in strict config validation so unicast DNS-SD gateway configs no longer fail with an unrecognized-key error. (#35615) Thanks @ingyukoh.
- Telegram/media errors: redact Telegram file URLs before building media fetch errors so failed inbound downloads do not leak bot tokens into logs. Thanks @space08. - Telegram/media errors: redact Telegram file URLs before building media fetch errors so failed inbound downloads do not leak bot tokens into logs. Thanks @space08.
- Agents/failover: normalize abort-wrapped `429 RESOURCE_EXHAUSTED` provider failures before abort short-circuiting so wrapped Google/Vertex rate limits continue across configured fallback models, including the embedded runner prompt-error path. (#39820) Thanks @lupuletic. - Agents/failover: normalize abort-wrapped `429 RESOURCE_EXHAUSTED` provider failures before abort short-circuiting so wrapped Google/Vertex rate limits continue across configured fallback models, including the embedded runner prompt-error path. (#39820) Thanks @lupuletic.
- Mattermost/thread routing: non-inbound reply paths (TUI/WebUI turns, tool-call callbacks, subagent responses) now correctly route to the originating Mattermost thread when `replyToMode: "all"` is active; also prevents stale `origin.threadId` metadata from resurrecting cleared thread routes. (#44283) thanks @teconomix
## 2026.3.12 ## 2026.3.12

View File

@ -355,6 +355,53 @@ describe("mattermostPlugin", () => {
}), }),
); );
}); });
it("uses threadId as fallback when replyToId is absent (sendText)", async () => {
const sendText = mattermostPlugin.outbound?.sendText;
if (!sendText) {
return;
}
await sendText({
to: "channel:CHAN1",
text: "hello",
accountId: "default",
threadId: "post-root",
} as any);
expect(sendMessageMattermostMock).toHaveBeenCalledWith(
"channel:CHAN1",
"hello",
expect.objectContaining({
accountId: "default",
replyToId: "post-root",
}),
);
});
it("uses threadId as fallback when replyToId is absent (sendMedia)", async () => {
const sendMedia = mattermostPlugin.outbound?.sendMedia;
if (!sendMedia) {
return;
}
await sendMedia({
to: "channel:CHAN1",
text: "caption",
mediaUrl: "https://example.com/image.png",
accountId: "default",
threadId: "post-root",
} as any);
expect(sendMessageMattermostMock).toHaveBeenCalledWith(
"channel:CHAN1",
"caption",
expect.objectContaining({
accountId: "default",
replyToId: "post-root",
}),
);
});
}); });
describe("config", () => { describe("config", () => {

View File

@ -390,21 +390,30 @@ export const mattermostPlugin: ChannelPlugin<ResolvedMattermostAccount> = {
} }
return { ok: true, to: trimmed }; return { ok: true, to: trimmed };
}, },
sendText: async ({ cfg, to, text, accountId, replyToId }) => { sendText: async ({ cfg, to, text, accountId, replyToId, threadId }) => {
const result = await sendMessageMattermost(to, text, { const result = await sendMessageMattermost(to, text, {
cfg, cfg,
accountId: accountId ?? undefined, accountId: accountId ?? undefined,
replyToId: replyToId ?? undefined, replyToId: replyToId ?? (threadId != null ? String(threadId) : undefined),
}); });
return { channel: "mattermost", ...result }; return { channel: "mattermost", ...result };
}, },
sendMedia: async ({ cfg, to, text, mediaUrl, mediaLocalRoots, accountId, replyToId }) => { sendMedia: async ({
cfg,
to,
text,
mediaUrl,
mediaLocalRoots,
accountId,
replyToId,
threadId,
}) => {
const result = await sendMessageMattermost(to, text, { const result = await sendMessageMattermost(to, text, {
cfg, cfg,
accountId: accountId ?? undefined, accountId: accountId ?? undefined,
mediaUrl, mediaUrl,
mediaLocalRoots, mediaLocalRoots,
replyToId: replyToId ?? undefined, replyToId: replyToId ?? (threadId != null ? String(threadId) : undefined),
}); });
return { channel: "mattermost", ...result }; return { channel: "mattermost", ...result };
}, },

View File

@ -41,6 +41,12 @@ const acpMocks = vi.hoisted(() => ({
const sessionBindingMocks = vi.hoisted(() => ({ const sessionBindingMocks = vi.hoisted(() => ({
listBySession: vi.fn<(targetSessionKey: string) => SessionBindingRecord[]>(() => []), listBySession: vi.fn<(targetSessionKey: string) => SessionBindingRecord[]>(() => []),
})); }));
const sessionStoreMocks = vi.hoisted(() => ({
currentEntry: undefined as Record<string, unknown> | undefined,
loadSessionStore: vi.fn(() => ({})),
resolveStorePath: vi.fn(() => "/tmp/mock-sessions.json"),
resolveSessionStoreEntry: vi.fn(() => ({ existing: sessionStoreMocks.currentEntry })),
}));
const ttsMocks = vi.hoisted(() => { const ttsMocks = vi.hoisted(() => {
const state = { const state = {
synthesizeFinalAudio: false, synthesizeFinalAudio: false,
@ -77,9 +83,16 @@ vi.mock("./route-reply.js", () => ({
isRoutableChannel: (channel: string | undefined) => isRoutableChannel: (channel: string | undefined) =>
Boolean( Boolean(
channel && channel &&
["telegram", "slack", "discord", "signal", "imessage", "whatsapp", "feishu"].includes( [
channel, "telegram",
), "slack",
"discord",
"signal",
"imessage",
"whatsapp",
"feishu",
"mattermost",
].includes(channel),
), ),
routeReply: mocks.routeReply, routeReply: mocks.routeReply,
})); }));
@ -100,6 +113,15 @@ vi.mock("../../logging/diagnostic.js", () => ({
logMessageProcessed: diagnosticMocks.logMessageProcessed, logMessageProcessed: diagnosticMocks.logMessageProcessed,
logSessionStateChange: diagnosticMocks.logSessionStateChange, logSessionStateChange: diagnosticMocks.logSessionStateChange,
})); }));
vi.mock("../../config/sessions.js", async (importOriginal) => {
const actual = await importOriginal<typeof import("../../config/sessions.js")>();
return {
...actual,
loadSessionStore: sessionStoreMocks.loadSessionStore,
resolveStorePath: sessionStoreMocks.resolveStorePath,
resolveSessionStoreEntry: sessionStoreMocks.resolveSessionStoreEntry,
};
});
vi.mock("../../plugins/hook-runner-global.js", () => ({ vi.mock("../../plugins/hook-runner-global.js", () => ({
getGlobalHookRunner: () => hookMocks.runner, getGlobalHookRunner: () => hookMocks.runner,
@ -228,6 +250,10 @@ describe("dispatchReplyFromConfig", () => {
acpMocks.requireAcpRuntimeBackend.mockReset(); acpMocks.requireAcpRuntimeBackend.mockReset();
sessionBindingMocks.listBySession.mockReset(); sessionBindingMocks.listBySession.mockReset();
sessionBindingMocks.listBySession.mockReturnValue([]); sessionBindingMocks.listBySession.mockReturnValue([]);
sessionStoreMocks.currentEntry = undefined;
sessionStoreMocks.loadSessionStore.mockClear();
sessionStoreMocks.resolveStorePath.mockClear();
sessionStoreMocks.resolveSessionStoreEntry.mockClear();
ttsMocks.state.synthesizeFinalAudio = false; ttsMocks.state.synthesizeFinalAudio = false;
ttsMocks.maybeApplyTtsToPayload.mockClear(); ttsMocks.maybeApplyTtsToPayload.mockClear();
ttsMocks.normalizeTtsAutoMode.mockClear(); ttsMocks.normalizeTtsAutoMode.mockClear();
@ -293,6 +319,88 @@ describe("dispatchReplyFromConfig", () => {
); );
}); });
it("falls back to thread-scoped session key when current ctx has no MessageThreadId", async () => {
setNoAbort();
mocks.routeReply.mockClear();
sessionStoreMocks.currentEntry = {
deliveryContext: {
channel: "mattermost",
to: "channel:CHAN1",
accountId: "default",
},
origin: {
threadId: "stale-origin-root",
},
lastThreadId: "stale-origin-root",
};
const cfg = emptyConfig;
const dispatcher = createDispatcher();
const ctx = buildTestCtx({
Provider: "webchat",
Surface: "webchat",
SessionKey: "agent:main:mattermost:channel:CHAN1:thread:post-root",
AccountId: "default",
MessageThreadId: undefined,
OriginatingChannel: "mattermost",
OriginatingTo: "channel:CHAN1",
ExplicitDeliverRoute: true,
});
const replyResolver = async () => ({ text: "hi" }) satisfies ReplyPayload;
await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver });
expect(mocks.routeReply).toHaveBeenCalledWith(
expect.objectContaining({
channel: "mattermost",
to: "channel:CHAN1",
threadId: "post-root",
}),
);
});
it("does not resurrect a cleared route thread from origin metadata", async () => {
setNoAbort();
mocks.routeReply.mockClear();
// Simulate the real store: lastThreadId and deliveryContext.threadId may be normalised from
// origin.threadId on read, but a non-thread session key must still route to channel root.
sessionStoreMocks.currentEntry = {
deliveryContext: {
channel: "mattermost",
to: "channel:CHAN1",
accountId: "default",
threadId: "stale-root",
},
lastThreadId: "stale-root",
origin: {
threadId: "stale-root",
},
};
const cfg = emptyConfig;
const dispatcher = createDispatcher();
const ctx = buildTestCtx({
Provider: "webchat",
Surface: "webchat",
SessionKey: "agent:main:mattermost:channel:CHAN1",
AccountId: "default",
MessageThreadId: undefined,
OriginatingChannel: "mattermost",
OriginatingTo: "channel:CHAN1",
ExplicitDeliverRoute: true,
});
const replyResolver = async () => ({ text: "hi" }) satisfies ReplyPayload;
await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver });
const routeCall = mocks.routeReply.mock.calls[0]?.[0] as
| { channel?: string; to?: string; threadId?: string | number }
| undefined;
expect(routeCall).toMatchObject({
channel: "mattermost",
to: "channel:CHAN1",
});
expect(routeCall?.threadId).toBeUndefined();
});
it("forces suppressTyping when routing to a different originating channel", async () => { it("forces suppressTyping when routing to a different originating channel", async () => {
setNoAbort(); setNoAbort();
const cfg = emptyConfig; const cfg = emptyConfig;

View File

@ -2,6 +2,7 @@ import { resolveSessionAgentId } from "../../agents/agent-scope.js";
import type { OpenClawConfig } from "../../config/config.js"; import type { OpenClawConfig } from "../../config/config.js";
import { import {
loadSessionStore, loadSessionStore,
parseSessionThreadInfo,
resolveSessionStoreEntry, resolveSessionStoreEntry,
resolveStorePath, resolveStorePath,
type SessionEntry, type SessionEntry,
@ -172,6 +173,12 @@ export async function dispatchReplyFromConfig(params: {
const sessionStoreEntry = resolveSessionStoreLookup(ctx, cfg); const sessionStoreEntry = resolveSessionStoreLookup(ctx, cfg);
const acpDispatchSessionKey = sessionStoreEntry.sessionKey ?? sessionKey; const acpDispatchSessionKey = sessionStoreEntry.sessionKey ?? sessionKey;
// Restore route thread context only from the active turn or the thread-scoped session key.
// Do not read thread ids from the normalised session store here: `origin.threadId` can be
// folded back into lastThreadId/deliveryContext during store normalisation and resurrect a
// stale route after thread delivery was intentionally cleared.
const routeThreadId =
ctx.MessageThreadId ?? parseSessionThreadInfo(acpDispatchSessionKey).threadId;
const inboundAudio = isInboundAudioContext(ctx); const inboundAudio = isInboundAudioContext(ctx);
const sessionTtsAuto = normalizeTtsAutoMode(sessionStoreEntry.entry?.ttsAuto); const sessionTtsAuto = normalizeTtsAutoMode(sessionStoreEntry.entry?.ttsAuto);
const hookRunner = getGlobalHookRunner(); const hookRunner = getGlobalHookRunner();
@ -260,7 +267,7 @@ export async function dispatchReplyFromConfig(params: {
to: originatingTo, to: originatingTo,
sessionKey: ctx.SessionKey, sessionKey: ctx.SessionKey,
accountId: ctx.AccountId, accountId: ctx.AccountId,
threadId: ctx.MessageThreadId, threadId: routeThreadId,
cfg, cfg,
abortSignal, abortSignal,
mirror, mirror,
@ -289,7 +296,7 @@ export async function dispatchReplyFromConfig(params: {
to: originatingTo, to: originatingTo,
sessionKey: ctx.SessionKey, sessionKey: ctx.SessionKey,
accountId: ctx.AccountId, accountId: ctx.AccountId,
threadId: ctx.MessageThreadId, threadId: routeThreadId,
cfg, cfg,
isGroup, isGroup,
groupId, groupId,
@ -519,7 +526,7 @@ export async function dispatchReplyFromConfig(params: {
to: originatingTo, to: originatingTo,
sessionKey: ctx.SessionKey, sessionKey: ctx.SessionKey,
accountId: ctx.AccountId, accountId: ctx.AccountId,
threadId: ctx.MessageThreadId, threadId: routeThreadId,
cfg, cfg,
isGroup, isGroup,
groupId, groupId,
@ -571,7 +578,7 @@ export async function dispatchReplyFromConfig(params: {
to: originatingTo, to: originatingTo,
sessionKey: ctx.SessionKey, sessionKey: ctx.SessionKey,
accountId: ctx.AccountId, accountId: ctx.AccountId,
threadId: ctx.MessageThreadId, threadId: routeThreadId,
cfg, cfg,
isGroup, isGroup,
groupId, groupId,

View File

@ -1,4 +1,5 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { mattermostPlugin } from "../../../extensions/mattermost/src/channel.js";
import { discordOutbound } from "../../channels/plugins/outbound/discord.js"; import { discordOutbound } from "../../channels/plugins/outbound/discord.js";
import { imessageOutbound } from "../../channels/plugins/outbound/imessage.js"; import { imessageOutbound } from "../../channels/plugins/outbound/imessage.js";
import { signalOutbound } from "../../channels/plugins/outbound/signal.js"; import { signalOutbound } from "../../channels/plugins/outbound/signal.js";
@ -24,6 +25,7 @@ const mocks = vi.hoisted(() => ({
sendMessageSlack: vi.fn(async () => ({ messageId: "m1", channelId: "c1" })), sendMessageSlack: vi.fn(async () => ({ messageId: "m1", channelId: "c1" })),
sendMessageTelegram: vi.fn(async () => ({ messageId: "m1", chatId: "c1" })), sendMessageTelegram: vi.fn(async () => ({ messageId: "m1", chatId: "c1" })),
sendMessageWhatsApp: vi.fn(async () => ({ messageId: "m1", toJid: "jid" })), sendMessageWhatsApp: vi.fn(async () => ({ messageId: "m1", toJid: "jid" })),
sendMessageMattermost: vi.fn(async () => ({ messageId: "m1", channelId: "c1" })),
deliverOutboundPayloads: vi.fn(), deliverOutboundPayloads: vi.fn(),
})); }));
@ -46,6 +48,9 @@ vi.mock("../../web/outbound.js", () => ({
sendMessageWhatsApp: mocks.sendMessageWhatsApp, sendMessageWhatsApp: mocks.sendMessageWhatsApp,
sendPollWhatsApp: mocks.sendMessageWhatsApp, sendPollWhatsApp: mocks.sendMessageWhatsApp,
})); }));
vi.mock("../../../extensions/mattermost/src/mattermost/send.js", () => ({
sendMessageMattermost: mocks.sendMessageMattermost,
}));
vi.mock("../../infra/outbound/deliver.js", async () => { vi.mock("../../infra/outbound/deliver.js", async () => {
const actual = await vi.importActual<typeof import("../../infra/outbound/deliver.js")>( const actual = await vi.importActual<typeof import("../../infra/outbound/deliver.js")>(
"../../infra/outbound/deliver.js", "../../infra/outbound/deliver.js",
@ -335,6 +340,33 @@ describe("routeReply", () => {
); );
}); });
it("uses threadId as replyToId for Mattermost when replyToId is missing", async () => {
mocks.deliverOutboundPayloads.mockResolvedValue([]);
await routeReply({
payload: { text: "hi" },
channel: "mattermost",
to: "channel:CHAN1",
threadId: "post-root",
cfg: {
channels: {
mattermost: {
enabled: true,
botToken: "test-token",
baseUrl: "https://chat.example.com",
},
},
} as unknown as OpenClawConfig,
});
expect(mocks.deliverOutboundPayloads).toHaveBeenCalledWith(
expect.objectContaining({
channel: "mattermost",
to: "channel:CHAN1",
replyToId: "post-root",
threadId: "post-root",
}),
);
});
it("sends multiple mediaUrls (caption only on first)", async () => { it("sends multiple mediaUrls (caption only on first)", async () => {
mocks.sendMessageSlack.mockClear(); mocks.sendMessageSlack.mockClear();
await routeReply({ await routeReply({
@ -501,4 +533,9 @@ const defaultRegistry = createTestRegistry([
}), }),
source: "test", source: "test",
}, },
{
pluginId: "mattermost",
plugin: mattermostPlugin,
source: "test",
},
]); ]);

View File

@ -149,7 +149,9 @@ export async function routeReply(params: RouteReplyParams): Promise<RouteReplyRe
const resolvedReplyToId = const resolvedReplyToId =
replyToId ?? replyToId ??
(channelId === "slack" && threadId != null && threadId !== "" ? String(threadId) : undefined); ((channelId === "slack" || channelId === "mattermost") && threadId != null && threadId !== ""
? String(threadId)
: undefined);
const resolvedThreadId = channelId === "slack" ? null : (threadId ?? null); const resolvedThreadId = channelId === "slack" ? null : (threadId ?? null);
try { try {

View File

@ -91,12 +91,16 @@ vi.mock("../pairing/pairing-store.js", () => ({
upsertChannelPairingRequest: (...args: unknown[]) => upsertPairingRequestMock(...args), upsertChannelPairingRequest: (...args: unknown[]) => upsertPairingRequestMock(...args),
})); }));
vi.mock("../config/sessions.js", () => ({ vi.mock("../config/sessions.js", async (importOriginal) => {
resolveStorePath: vi.fn(() => "/tmp/openclaw-sessions.json"), const actual = await importOriginal<typeof import("../config/sessions.js")>();
updateLastRoute: (...args: unknown[]) => updateLastRouteMock(...args), return {
readSessionUpdatedAt: vi.fn(() => undefined), ...actual,
recordSessionMetaFromInbound: vi.fn().mockResolvedValue(undefined), resolveStorePath: vi.fn(() => "/tmp/openclaw-sessions.json"),
})); updateLastRoute: (...args: unknown[]) => updateLastRouteMock(...args),
readSessionUpdatedAt: vi.fn(() => undefined),
recordSessionMetaFromInbound: vi.fn().mockResolvedValue(undefined),
};
});
vi.mock("./client.js", () => ({ vi.mock("./client.js", () => ({
streamSignalEvents: (...args: unknown[]) => streamMock(...args), streamSignalEvents: (...args: unknown[]) => streamMock(...args),

View File

@ -180,13 +180,17 @@ vi.mock("../pairing/pairing-store.js", () => ({
slackTestState.upsertPairingRequestMock(...args), slackTestState.upsertPairingRequestMock(...args),
})); }));
vi.mock("../config/sessions.js", () => ({ vi.mock("../config/sessions.js", async (importOriginal) => {
resolveStorePath: vi.fn(() => "/tmp/openclaw-sessions.json"), const actual = await importOriginal<typeof import("../config/sessions.js")>();
updateLastRoute: (...args: unknown[]) => slackTestState.updateLastRouteMock(...args), return {
resolveSessionKey: vi.fn(), ...actual,
readSessionUpdatedAt: vi.fn(() => undefined), resolveStorePath: vi.fn(() => "/tmp/openclaw-sessions.json"),
recordSessionMetaFromInbound: vi.fn().mockResolvedValue(undefined), updateLastRoute: (...args: unknown[]) => slackTestState.updateLastRouteMock(...args),
})); resolveSessionKey: vi.fn(),
readSessionUpdatedAt: vi.fn(() => undefined),
recordSessionMetaFromInbound: vi.fn().mockResolvedValue(undefined),
};
});
vi.mock("@slack/bolt", () => { vi.mock("@slack/bolt", () => {
const handlers = new Map<string, SlackHandler>(); const handlers = new Map<string, SlackHandler>();