fix(slack): stop repeated outage notices in active threads (#122782)

This commit is contained in:
Kimi Yu
2026-08-12 14:41:41 -07:00
committed by GitHub
parent 431b8ec4a4
commit f9316c4697
5 changed files with 648 additions and 2 deletions
@@ -0,0 +1,328 @@
import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
import { createPluginStateKeyedStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
import { setReplyPayloadMetadata } from "openclaw/plugin-sdk/reply-payload-testing";
import { resetInboundDedupe } from "openclaw/plugin-sdk/reply-runtime";
import { beforeEach, describe, expect, it } from "vitest";
import {
getSlackTestState,
resetSlackTestState,
runSlackMessageOnce,
} from "./monitor.test-helpers.js";
import { getSlackRuntime, setSlackRuntime } from "./runtime.js";
import {
clearSlackThreadParticipationCache,
hasSlackThreadParticipation,
} from "./sent-thread-cache.js";
const { monitorSlackProvider } = await import("./monitor/provider.js");
const slackTestState = getSlackTestState();
const AUTH_FAILURE = "⚠️ Model login expired on the gateway.";
const BACKEND_FAILURE = "⚠️ Codex app-server is unavailable.";
type SlackFailureTestEvent = {
type: "message";
user: string;
text: string;
ts: string;
channel: string;
channel_type: "im" | "mpim" | "channel";
thread_ts?: string;
parent_user_id?: string;
};
function makeEvent(overrides: Partial<SlackFailureTestEvent>): SlackFailureTestEvent {
return {
type: "message",
user: "U1",
text: "ordinary follow-up",
ts: "100.000001",
channel: "C1",
channel_type: "channel",
...overrides,
};
}
async function dispatchEvent(overrides: Partial<SlackFailureTestEvent>): Promise<void> {
await runSlackMessageOnce(
monitorSlackProvider,
{ event: makeEvent(overrides) },
{ awaitDispatch: true },
);
}
function mockReplySequence(...payloads: Array<{ text: string; isError?: boolean }>): void {
let runIndex = 0;
slackTestState.replyMock.mockImplementation(async (...args: unknown[]) => {
const options = args[1] as { onAgentRunStart?: (runId: string) => void } | undefined;
options?.onAgentRunStart?.(`slack-failure-notice-test-${runIndex}`);
const payload = payloads[Math.min(runIndex, payloads.length - 1)];
runIndex += 1;
return payload;
});
}
function enableAmbientChannelReplies(replyToMode: "all" | "off" = "all"): void {
slackTestState.config = {
messages: { groupChat: { visibleReplies: "automatic" } },
channels: {
slack: {
dm: { enabled: true },
dmPolicy: "open",
allowFrom: ["*"],
groupPolicy: "open",
requireMention: false,
replyToMode,
channels: { C1: { allow: true, requireMention: false } },
},
},
};
}
describe("Slack thread failure notices", () => {
beforeEach(() => {
resetInboundDedupe();
clearSlackThreadParticipationCache();
resetSlackTestState({
messages: { groupChat: { visibleReplies: "automatic" } },
channels: {
slack: {
dm: { enabled: true },
dmPolicy: "open",
allowFrom: ["*"],
groupPolicy: "open",
requireMention: true,
replyToMode: "all",
channels: { C1: { allow: true, requireMention: true } },
},
},
});
});
it("shows an explicit mention's failure and suppresses matching passive follow-ups", async () => {
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ text: "<@bot-user> please help", ts: "100.000000" });
await dispatchEvent({ ts: "100.000001", thread_ts: "100.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "100.000002", thread_ts: "100.000000", parent_user_id: "U1" });
expect(slackTestState.replyMock).toHaveBeenCalledTimes(3);
expect(slackTestState.sendMock).toHaveBeenCalledTimes(1);
});
it("announces the first failure after an established thread was working", async () => {
mockReplySequence({ text: "Working normally" }, { text: AUTH_FAILURE, isError: true });
await dispatchEvent({ text: "<@bot-user> please help", ts: "101.000000" });
await dispatchEvent({ ts: "101.000001", thread_ts: "101.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "101.000002", thread_ts: "101.000000", parent_user_id: "U1" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock.mock.calls[1]?.[1]).toBe(AUTH_FAILURE);
});
it("announces the first failure for participation restored after a restart", async () => {
const threadTs = "101.100000";
const openKeyedStore = <T>(options: OpenKeyedStoreOptions) =>
createPluginStateKeyedStoreForTests<T>("slack", options);
const persistedStore = openKeyedStore<{ repliedAt: number }>({
namespace: "slack.thread-participation",
maxEntries: 1000,
});
await persistedStore.register(
`default:C1:${threadTs}`,
{ repliedAt: Date.now() },
{
ttlMs: 60_000,
},
);
const runtime = getSlackRuntime();
setSlackRuntime({
...runtime,
state: {
...runtime.state,
openKeyedStore,
},
});
expect(hasSlackThreadParticipation("default", "C1", threadTs)).toBe(false);
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ ts: "101.100001", thread_ts: threadTs, parent_user_id: "U1" });
await dispatchEvent({ ts: "101.100002", thread_ts: threadTs, parent_user_id: "U1" });
expect(slackTestState.replyMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock).toHaveBeenCalledTimes(1);
expect(slackTestState.sendMock.mock.calls[0]?.[1]).toBe(AUTH_FAILURE);
});
it("announces a different failure after suppressing repeated copies of the first", async () => {
mockReplySequence(
{ text: "Working normally" },
{ text: AUTH_FAILURE, isError: true },
{ text: AUTH_FAILURE, isError: true },
{ text: BACKEND_FAILURE, isError: true },
);
await dispatchEvent({ text: "<@bot-user> please help", ts: "102.000000" });
await dispatchEvent({ ts: "102.000001", thread_ts: "102.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "102.000002", thread_ts: "102.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "102.000003", thread_ts: "102.000000", parent_user_id: "U1" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(3);
expect(slackTestState.sendMock.mock.calls[2]?.[1]).toBe(BACKEND_FAILURE);
});
it("announces the same failure again after a successful reply", async () => {
mockReplySequence(
{ text: "Working normally" },
{ text: AUTH_FAILURE, isError: true },
{ text: "Recovered" },
{ text: AUTH_FAILURE, isError: true },
);
await dispatchEvent({ text: "<@bot-user> please help", ts: "103.000000" });
await dispatchEvent({ ts: "103.000001", thread_ts: "103.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "103.000002", thread_ts: "103.000000", parent_user_id: "U1" });
await dispatchEvent({ ts: "103.000003", thread_ts: "103.000000", parent_user_id: "U1" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(4);
expect(slackTestState.sendMock.mock.calls[3]?.[1]).toBe(AUTH_FAILURE);
});
it("always explains the current failure when the user explicitly mentions the bot", async () => {
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ text: "<@bot-user> please help", ts: "104.000000" });
await dispatchEvent({ ts: "104.000001", thread_ts: "104.000000", parent_user_id: "U1" });
await dispatchEvent({
text: "<@bot-user> are you working now?",
ts: "104.000002",
thread_ts: "104.000000",
parent_user_id: "U1",
});
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
});
it.each(["all", "off"] as const)(
"announces one failure for unmentioned channel messages with reply mode %s",
async (replyToMode) => {
enableAmbientChannelReplies(replyToMode);
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ ts: "105.000000" });
await dispatchEvent({ ts: "105.000001" });
expect(slackTestState.replyMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock).toHaveBeenCalledTimes(1);
expect(slackTestState.sendMock.mock.calls[0]?.[1]).toBe(AUTH_FAILURE);
},
);
it("announces a changed failure for unmentioned channel messages", async () => {
enableAmbientChannelReplies();
mockReplySequence(
{ text: AUTH_FAILURE, isError: true },
{ text: AUTH_FAILURE, isError: true },
{ text: BACKEND_FAILURE, isError: true },
);
await dispatchEvent({ ts: "105.010000" });
await dispatchEvent({ ts: "105.010001" });
await dispatchEvent({ ts: "105.010002" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock.mock.calls[1]?.[1]).toBe(BACKEND_FAILURE);
});
it("announces an unmentioned channel failure again after a successful reply", async () => {
enableAmbientChannelReplies();
mockReplySequence(
{ text: AUTH_FAILURE, isError: true },
{ text: AUTH_FAILURE, isError: true },
{ text: "Recovered" },
{ text: AUTH_FAILURE, isError: true },
);
await dispatchEvent({ ts: "105.020000" });
await dispatchEvent({ ts: "105.020001" });
await dispatchEvent({ ts: "105.020002" });
await dispatchEvent({ ts: "105.020003" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(3);
expect(slackTestState.sendMock.mock.calls[2]?.[1]).toBe(AUTH_FAILURE);
});
it("always answers an explicit mention after an unmentioned channel failure", async () => {
enableAmbientChannelReplies();
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ ts: "105.030000" });
await dispatchEvent({ ts: "105.030001" });
await dispatchEvent({ text: "<@bot-user> are you working now?", ts: "105.030002" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock.mock.calls[1]?.[1]).toBe(AUTH_FAILURE);
});
it("retries the same thread failure when its first Slack delivery fails", async () => {
mockReplySequence(
{ text: "Working normally" },
{ text: AUTH_FAILURE, isError: true },
{ text: AUTH_FAILURE, isError: true },
);
await dispatchEvent({ text: "<@bot-user> please help", ts: "105.040000" });
slackTestState.sendMock.mockRejectedValueOnce(new Error("Slack delivery unavailable"));
await dispatchEvent({ ts: "105.040001", thread_ts: "105.040000", parent_user_id: "U1" });
await dispatchEvent({ ts: "105.040002", thread_ts: "105.040000", parent_user_id: "U1" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(3);
expect(slackTestState.sendMock.mock.calls[2]?.[1]).toBe(AUTH_FAILURE);
});
it("does not suppress warnings for non-terminal tool failures", async () => {
const warning = setReplyPayloadMetadata(
{ text: "A tool failed, but the run completed.", isError: true },
{ nonTerminalToolErrorWarning: true },
);
mockReplySequence({ text: "Working normally" }, warning, warning);
await dispatchEvent({ text: "<@bot-user> please help", ts: "105.100000" });
await dispatchEvent({ ts: "105.100001", thread_ts: "105.100000", parent_user_id: "U1" });
await dispatchEvent({ ts: "105.100002", thread_ts: "105.100000", parent_user_id: "U1" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(3);
});
it("keeps failures visible in direct messages", async () => {
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ channel: "D1", channel_type: "im", ts: "106.000000" });
await dispatchEvent({ channel: "D1", channel_type: "im", ts: "106.000001" });
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
});
it("keeps failures visible in Slack group direct messages", async () => {
slackTestState.config = {
messages: { groupChat: { visibleReplies: "automatic" } },
channels: {
slack: {
dm: { enabled: true, groupEnabled: true },
dmPolicy: "open",
allowFrom: ["U1"],
groupPolicy: "open",
replyToMode: "off",
},
},
};
mockReplySequence({ text: AUTH_FAILURE, isError: true });
await dispatchEvent({ channel: "G1", channel_type: "mpim", ts: "107.000000" });
await dispatchEvent({ channel: "G1", channel_type: "mpim", ts: "107.000001" });
expect(slackTestState.replyMock).toHaveBeenCalledTimes(2);
expect(slackTestState.sendMock).toHaveBeenCalledTimes(2);
});
});
@@ -810,6 +810,7 @@ vi.mock("openclaw/plugin-sdk/reply-history", () => ({
}));
vi.mock("openclaw/plugin-sdk/reply-payload", () => ({
isReplyPayloadNonTerminalToolErrorWarning: () => false,
buildTtsSupplementMediaPayload: (payload: {
text?: string;
mediaUrl?: string;
@@ -900,6 +901,9 @@ vi.mock("../../limits.js", () => ({
}));
vi.mock("../../sent-thread-cache.js", () => ({
clearSlackThreadFailureNotice: () => {},
hasSlackThreadParticipation: () => false,
recordSlackThreadFailureNotice: () => true,
recordSlackThreadParticipation: recordSlackThreadParticipationMock,
}));
@@ -14,6 +14,7 @@ import { toErrorObject } from "openclaw/plugin-sdk/error-runtime";
import {
buildTtsSupplementMediaPayload,
getReplyPayloadTtsSupplement,
isReplyPayloadNonTerminalToolErrorWarning,
resolveSendableOutboundReplyParts,
} from "openclaw/plugin-sdk/reply-payload";
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
@@ -24,7 +25,13 @@ import { normalizeSlackOutboundText } from "../../format.js";
import { SLACK_EDIT_TEXT_MAX_BYTES } from "../../limits.js";
import { emitSlackMessageSentHooks } from "../../message-sent-hook.js";
import { resolveSlackReplyRenderPlan } from "../../reply-blocks.js";
import { recordSlackThreadParticipation } from "../../sent-thread-cache.js";
import {
clearSlackThreadFailureNotice,
hasSlackThreadFailureNotice,
hasSlackThreadParticipation,
recordSlackThreadFailureNotice,
recordSlackThreadParticipation,
} from "../../sent-thread-cache.js";
import {
SlackStreamNotDeliveredError,
stopSlackStream,
@@ -78,6 +85,73 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
},
});
const draftStream = progress.draftStream;
const failureNoticeThreadTs = message.thread_ts;
const failureNoticeTeamId = prepared.eventScope?.teamId;
let sawTerminalFailurePayload = false;
let pendingFailureNotice:
| {
accountId: string;
channelId: string;
threadTs?: string;
failureText: string;
teamId?: string;
}
| undefined;
const filterPassiveThreadFailure = (payload: ReplyPayload): ReplyPayload | null => {
if (
payload.isError !== true ||
prepared.ctxPayload.ChatType !== "channel" ||
isReplyPayloadNonTerminalToolErrorWarning(payload)
) {
return payload;
}
sawTerminalFailurePayload = true;
if (delivery.observedReplyDelivery || draftPreviewCommitted.value) {
return payload;
}
const explicitlyAddressed =
prepared.ctxPayload.ExplicitlyMentionedBot === true ||
prepared.ctxPayload.MentionSource === "explicit_bot" ||
prepared.ctxPayload.MentionSource === "subteam" ||
prepared.ctxPayload.MentionSource === "mention_pattern" ||
prepared.ctxPayload.MentionSource === "command_bypass" ||
(prepared.ctxPayload.CommandTurn?.kind !== undefined &&
prepared.ctxPayload.CommandTurn.kind !== "normal" &&
prepared.ctxPayload.CommandTurn.authorized);
const noticeThreadTs =
failureNoticeThreadTs ?? (explicitlyAddressed ? statusThreadTs : undefined);
const notice = {
accountId: account.accountId,
channelId: message.channel,
...(noticeThreadTs ? { threadTs: noticeThreadTs } : {}),
failureText: payload.text ?? "",
...(failureNoticeTeamId ? { teamId: failureNoticeTeamId } : {}),
};
if (
failureNoticeThreadTs &&
!explicitlyAddressed &&
prepared.ctxPayload.MentionSource !== "implicit_thread" &&
!hasSlackThreadParticipation(
notice.accountId,
notice.channelId,
failureNoticeThreadTs,
failureNoticeTeamId,
)
) {
logVerbose("slack: suppressed passive failure before thread participation");
return null;
}
if (!explicitlyAddressed && hasSlackThreadFailureNotice(notice)) {
logVerbose("slack: suppressed repeated passive channel or thread failure");
return null;
}
pendingFailureNotice = notice;
return payload;
};
const deliverSlackPayload = async (
payload: ReplyPayload,
@@ -385,6 +459,13 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
ctxPayload: prepared.ctxPayload,
dispatcherOptions: {
...replyPipeline,
// A channel transform marks intentional silence before core can synthesize an empty-reply error.
transformReplyPayload: (payload) => {
const transformed = replyPipeline.transformReplyPayload
? replyPipeline.transformReplyPayload(payload)
: payload;
return transformed ? filterPassiveThreadFailure(transformed) : null;
},
humanDelay: resolveHumanDelayConfig(cfg, route.agentId),
},
delivery: {
@@ -511,7 +592,20 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
const result = turnResult.dispatchResult;
queuedFinal = result.queuedFinal;
counts = result.counts;
agentRunFailed = readAgentRunTerminalOutcome(result) === "failed";
const agentRunOutcome = readAgentRunTerminalOutcome(result);
agentRunFailed = agentRunOutcome === "failed";
if (
agentRunOutcome === "completed" &&
!sawTerminalFailurePayload &&
prepared.ctxPayload.ChatType === "channel"
) {
clearSlackThreadFailureNotice({
accountId: account.accountId,
channelId: message.channel,
...(failureNoticeThreadTs ? { threadTs: failureNoticeThreadTs } : {}),
...(failureNoticeTeamId ? { teamId: failureNoticeTeamId } : {}),
});
}
}
} catch (err) {
dispatchError = err;
@@ -590,6 +684,10 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
},
);
if (pendingFailureNotice && anyReplyDelivered) {
recordSlackThreadFailureNotice(pendingFailureNotice);
}
if (dispatchError || agentRunFailed) {
await progress.finalizeDraftProgressCard("error");
}
@@ -9,9 +9,12 @@ import { withOpenClawTestState } from "openclaw/plugin-sdk/test-state";
import { afterEach, describe, expect, it, vi } from "vitest";
import { setSlackRuntime } from "./runtime.js";
import {
clearSlackThreadFailureNotice,
clearSlackThreadParticipationCache,
hasSlackThreadFailureNotice,
hasSlackThreadParticipation,
hasSlackThreadParticipationWithPersistence,
recordSlackThreadFailureNotice,
recordSlackThreadParticipation,
} from "./sent-thread-cache.js";
@@ -52,6 +55,142 @@ describe("slack sent-thread-cache", () => {
expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(false);
});
it("announces a repeated thread failure only once until its message changes", () => {
const notice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
};
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
expect(recordSlackThreadFailureNotice(notice)).toBe(false);
expect(
recordSlackThreadFailureNotice({ ...notice, failureText: "Model login\nexpired" }),
).toBe(false);
expect(
recordSlackThreadFailureNotice({ ...notice, failureText: "App server unavailable" }),
).toBe(true);
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
});
it("checks a failure without marking it delivered", () => {
const notice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
};
expect(hasSlackThreadFailureNotice(notice)).toBe(false);
expect(hasSlackThreadFailureNotice(notice)).toBe(false);
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
expect(hasSlackThreadFailureNotice(notice)).toBe(true);
expect(hasSlackThreadFailureNotice({ ...notice, failureText: "Model login\nexpired" })).toBe(
true,
);
expect(hasSlackThreadFailureNotice({ ...notice, failureText: "App server unavailable" })).toBe(
false,
);
});
it("deduplicates top-level failures per channel without mixing them with threads", () => {
const channelNotice = {
accountId: "A1",
channelId: "C123",
failureText: "Model login expired",
teamId: "T1",
};
expect(hasSlackThreadFailureNotice(channelNotice)).toBe(false);
expect(recordSlackThreadFailureNotice(channelNotice)).toBe(true);
expect(hasSlackThreadFailureNotice(channelNotice)).toBe(true);
expect(recordSlackThreadFailureNotice(channelNotice)).toBe(false);
expect(hasSlackThreadFailureNotice({ ...channelNotice, channelId: "C456" })).toBe(false);
expect(hasSlackThreadFailureNotice({ ...channelNotice, accountId: "A2" })).toBe(false);
expect(hasSlackThreadFailureNotice({ ...channelNotice, teamId: "T2" })).toBe(false);
expect(hasSlackThreadFailureNotice({ ...channelNotice, threadTs: "1700000000.000001" })).toBe(
false,
);
clearSlackThreadFailureNotice(channelNotice);
expect(hasSlackThreadFailureNotice(channelNotice)).toBe(false);
expect(recordSlackThreadFailureNotice(channelNotice)).toBe(true);
});
it("does not deduplicate failures with empty text", () => {
const notice = {
accountId: "A1",
channelId: "C123",
failureText: " ",
};
expect(hasSlackThreadFailureNotice(notice)).toBe(false);
expect(recordSlackThreadFailureNotice(notice)).toBe(false);
});
it("isolates thread failures by account, channel, thread, and enterprise workspace", () => {
const notice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
teamId: "T1",
};
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
expect(recordSlackThreadFailureNotice({ ...notice, accountId: "A2" })).toBe(true);
expect(recordSlackThreadFailureNotice({ ...notice, channelId: "C456" })).toBe(true);
expect(recordSlackThreadFailureNotice({ ...notice, threadTs: "1700000000.000002" })).toBe(true);
expect(recordSlackThreadFailureNotice({ ...notice, teamId: "T2" })).toBe(true);
expect(recordSlackThreadFailureNotice(notice)).toBe(false);
});
it("allows the same thread failure again after a successful turn clears its notice", () => {
const notice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
};
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
clearSlackThreadFailureNotice(notice);
expect(recordSlackThreadFailureNotice(notice)).toBe(true);
});
it("does not treat failure notices as thread participation", () => {
recordSlackThreadFailureNotice({
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
});
expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(false);
});
it("bounds failure notices and evicts the oldest thread", () => {
const firstNotice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000000",
failureText: "Model login expired",
};
expect(recordSlackThreadFailureNotice(firstNotice)).toBe(true);
for (let index = 1; index <= 1000; index += 1) {
expect(
recordSlackThreadFailureNotice({
...firstNotice,
threadTs: `1700000000.${String(index).padStart(6, "0")}`,
}),
).toBe(true);
}
expect(recordSlackThreadFailureNotice(firstNotice)).toBe(true);
});
it("ignores empty accountId, channelId, or threadTs", () => {
recordSlackThreadParticipation("", "C123", "1700000000.000001");
recordSlackThreadParticipation("A1", "", "1700000000.000001");
@@ -84,9 +223,18 @@ describe("slack sent-thread-cache", () => {
try {
cacheA.recordSlackThreadParticipation("A1", "C123", "1700000000.000001");
expect(cacheB.hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(true);
const failureNotice = {
accountId: "A1",
channelId: "C123",
threadTs: "1700000000.000001",
failureText: "Model login expired",
};
expect(cacheA.recordSlackThreadFailureNotice(failureNotice)).toBe(true);
expect(cacheB.recordSlackThreadFailureNotice(failureNotice)).toBe(false);
cacheB.clearSlackThreadParticipationCache();
expect(cacheA.hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(false);
expect(cacheA.recordSlackThreadFailureNotice(failureNotice)).toBe(true);
} finally {
cacheA.clearSlackThreadParticipationCache();
}
+68
View File
@@ -1,5 +1,6 @@
// Slack plugin module implements sent thread cache behavior.
import { createPersistentDedupeCache } from "openclaw/plugin-sdk/dedupe-runtime";
import { resolveGlobalSingleton } from "openclaw/plugin-sdk/global-singleton";
import { createPluginStateErrorReporter } from "openclaw/plugin-sdk/plugin-state-runtime";
import { getOptionalSlackRuntime } from "./runtime.js";
@@ -10,6 +11,7 @@ import { getOptionalSlackRuntime } from "./runtime.js";
const MAX_ENTRIES = 5000;
const PERSISTENT_MAX_ENTRIES = 1000;
const MAX_FAILURE_NOTICES = 1000;
const PERSISTENT_NAMESPACE = "slack.thread-participation";
type SlackThreadParticipationRecord = {
@@ -22,6 +24,7 @@ type SlackThreadParticipationRecord = {
* auto-reply gating does not diverge between prepare/dispatch call paths.
*/
const SLACK_THREAD_PARTICIPATION_KEY = Symbol.for("openclaw.slackThreadParticipation");
const SLACK_THREAD_FAILURE_NOTICES_KEY = Symbol.for("openclaw.slackThreadFailureNotices");
const threadParticipation = createPersistentDedupeCache<SlackThreadParticipationRecord>({
globalKey: SLACK_THREAD_PARTICIPATION_KEY,
// Participation remains valid until bounded oldest-entry eviction removes it.
@@ -39,6 +42,11 @@ const threadParticipation = createPersistentDedupeCache<SlackThreadParticipation
),
},
});
const threadFailureNotices = resolveGlobalSingleton(
SLACK_THREAD_FAILURE_NOTICES_KEY,
() => new Map<string, string>(),
(notices) => notices.clear(),
);
function makeKey(accountId: string, channelId: string, threadTs: string, teamId?: string): string {
return `${accountId}:${teamId ? `${teamId}:` : ""}${channelId}:${threadTs}`;
@@ -86,6 +94,66 @@ export async function hasSlackThreadParticipationWithPersistence(params: {
);
}
type SlackFailureNotice = {
accountId: string;
channelId: string;
threadTs?: string;
failureText: string;
teamId?: string;
};
function makeFailureNoticeKey(params: Omit<SlackFailureNotice, "failureText">): string {
const scope = params.threadTs ? `thread:${params.threadTs}` : "channel";
return makeKey(params.accountId, params.channelId, scope, params.teamId);
}
/** Returns whether this failure was already delivered in the thread or channel. */
export function hasSlackThreadFailureNotice(params: SlackFailureNotice): boolean {
const { accountId, channelId, failureText } = params;
const fingerprint = failureText.trim().replace(/\s+/gu, " ");
if (!accountId || !channelId || !fingerprint) {
return false;
}
return threadFailureNotices.get(makeFailureNoticeKey(params)) === fingerprint;
}
/** Records a failure after it was delivered in the thread or channel. */
export function recordSlackThreadFailureNotice(params: SlackFailureNotice): boolean {
const { accountId, channelId, failureText } = params;
const fingerprint = failureText.trim().replace(/\s+/gu, " ");
if (!accountId || !channelId || !fingerprint) {
return false;
}
const key = makeFailureNoticeKey(params);
if (threadFailureNotices.get(key) === fingerprint) {
return false;
}
threadFailureNotices.delete(key);
threadFailureNotices.set(key, fingerprint);
if (threadFailureNotices.size > MAX_FAILURE_NOTICES) {
const oldestKey = threadFailureNotices.keys().next().value;
if (oldestKey !== undefined) {
threadFailureNotices.delete(oldestKey);
}
}
return true;
}
/** Clears a thread or channel outage notice after a healthy model turn completes. */
export function clearSlackThreadFailureNotice(params: {
accountId: string;
channelId: string;
threadTs?: string;
teamId?: string;
}): void {
const { accountId, channelId } = params;
if (!accountId || !channelId) {
return;
}
threadFailureNotices.delete(makeFailureNoticeKey(params));
}
export function clearSlackThreadParticipationCache(): void {
threadParticipation.clearForTest();
threadFailureNotices.clear();
}