diff --git a/extensions/slack/src/monitor.failure-notices.test.ts b/extensions/slack/src/monitor.failure-notices.test.ts new file mode 100644 index 000000000000..2cee9bff9122 --- /dev/null +++ b/extensions/slack/src/monitor.failure-notices.test.ts @@ -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 { + return { + type: "message", + user: "U1", + text: "ordinary follow-up", + ts: "100.000001", + channel: "C1", + channel_type: "channel", + ...overrides, + }; +} + +async function dispatchEvent(overrides: Partial): Promise { + 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 = (options: OpenKeyedStoreOptions) => + createPluginStateKeyedStoreForTests("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); + }); +}); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts index 246d1a156fbd..efe5af1b829c 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts @@ -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, })); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index 39aa90b06e0e..f23f9ade010a 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -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"); } diff --git a/extensions/slack/src/sent-thread-cache.test.ts b/extensions/slack/src/sent-thread-cache.test.ts index 808972bcc599..56e6144a575a 100644 --- a/extensions/slack/src/sent-thread-cache.test.ts +++ b/extensions/slack/src/sent-thread-cache.test.ts @@ -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(); } diff --git a/extensions/slack/src/sent-thread-cache.ts b/extensions/slack/src/sent-thread-cache.ts index 96da39293459..90fb07eb539d 100644 --- a/extensions/slack/src/sent-thread-cache.ts +++ b/extensions/slack/src/sent-thread-cache.ts @@ -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({ globalKey: SLACK_THREAD_PARTICIPATION_KEY, // Participation remains valid until bounded oldest-entry eviction removes it. @@ -39,6 +42,11 @@ const threadParticipation = createPersistentDedupeCache new Map(), + (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): 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(); }