diff --git a/extensions/slack/src/client-delivery.ts b/extensions/slack/src/client-delivery.ts index 2bafc184f0c6..3c17054206b7 100644 --- a/extensions/slack/src/client-delivery.ts +++ b/extensions/slack/src/client-delivery.ts @@ -13,6 +13,7 @@ import { loadOutboundMediaFromUrl } from "openclaw/plugin-sdk/outbound-media"; import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; import { logVerbose, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { fetchWithSsrFGuard, type SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { formatSlackError } from "./errors.js"; import { postSlackMessageWithIdentityFallback, @@ -41,6 +42,26 @@ const SLACK_DNS_RETRY_CODES = new Set(["EAI_AGAIN", "ENOTFOUND", "UND_ERR_DNS_RE const SLACK_DNS_RETRY_ATTEMPTS = 2; const SLACK_DNS_RETRY_BASE_DELAY_MS = 250; +// Slack reports provider verdicts as `slack_webapi_platform_error` with the code in +// `data.error`; these two mean no recipient can ever see the message, so durable +// recovery must stop retrying. Everything else rethrows by identity — the +// `invalid_blocks` and custom-identity fallbacks match on the original value. +// Pre-dispatch calls only: PlatformMessageNotDispatchedError asserts no send began, +// so a call made after onPlatformSendDispatch must stay ambiguous instead. +export function rethrowSlackPermanentOutboundApiRejection(err: unknown): never { + const rawData = + isRecord(err) && err.code === "slack_webapi_platform_error" ? err.data : undefined; + const data = isRecord(rawData) ? rawData : undefined; + const code = data?.error; + if (data?.ok === false && (code === "messages_tab_disabled" || code === "account_inactive")) { + throw new PlatformMessageNotDispatchedError(`Slack outbound delivery rejected: ${code}`, { + cause: err, + retryable: false, + }); + } + throw err; +} + function readSlackRequestErrorCode(value: unknown): string | undefined { if (!value || typeof value !== "object") { return undefined; @@ -236,7 +257,9 @@ export async function postSlackMessageBestEffort(params: { const basePayload = buildSlackPostMessagePayload(params); const postChatMessage = params.client.chat.postMessage.bind(params.client.chat); const post = async (payload: SlackPostMessagePayload, identity?: SlackPostMessageIdentity) => ({ - response: await withSlackDnsRequestRetry("chat.postMessage", () => postChatMessage(payload)), + response: await withSlackDnsRequestRetry("chat.postMessage", () => + postChatMessage(payload), + ).catch(rethrowSlackPermanentOutboundApiRejection), identity, }); const posted = await postSlackMessageWithIdentityFallback({ @@ -287,7 +310,7 @@ export async function uploadSlackFile(params: { filename: uploadFileName, length: buffer.length, }), - ); + ).catch(rethrowSlackPermanentOutboundApiRejection); if (!uploadUrlResp.ok || !uploadUrlResp.upload_url || !uploadUrlResp.file_id) { throw new Error(`Failed to get upload URL: ${uploadUrlResp.error ?? "unknown error"}`); } @@ -349,6 +372,8 @@ export async function uploadSlackFile(params: { await params.onPlatformSendDispatch?.(); // Slack allows this finalize call only once. Keep only the pre-connect DNS // retry; a timeout or broader retry would create an unknown-send state. + // Dispatch is already recorded above, so this call is the ambiguous send: + // no rejection here may claim non-dispatch, however definitive its code reads. const completionClient = params.completionClient ?? params.client; const completeResp = await withSlackDnsRequestRetry("files.completeUploadExternal", () => completionClient.files.completeUploadExternal({ diff --git a/extensions/slack/src/send.permanent-rejection.test.ts b/extensions/slack/src/send.permanent-rejection.test.ts new file mode 100644 index 000000000000..cf2d92b62cb1 --- /dev/null +++ b/extensions/slack/src/send.permanent-rejection.test.ts @@ -0,0 +1,156 @@ +import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; +import { describe, expect, it, vi } from "vitest"; +import { createSlackSendTestClient } from "./blocks.test-helpers.js"; +import { rethrowSlackPermanentOutboundApiRejection } from "./client-delivery.js"; +import { isSlackInvalidBlocksError } from "./native-data-blocks.js"; + +const { sendMessageSlack } = await import("./send.js"); +const SLACK_TEST_CFG = { channels: { slack: { botToken: "xoxb-test" } } }; +const SLACK_TEXT_LIMIT = 8000; + +function slackPlatformError(code: string): Error { + return Object.assign(new Error(`An API error occurred: ${code}`), { + code: "slack_webapi_platform_error", + data: { ok: false, error: code }, + }); +} + +describe("sendMessageSlack permanent provider rejections", () => { + it.each(["messages_tab_disabled", "account_inactive"])( + "marks Slack %s as a permanent non-dispatch", + async (code) => { + const client = createSlackSendTestClient(); + const rejection = slackPlatformError(code); + const onPlatformSendDispatch = vi.fn(); + client.chat.postMessage.mockRejectedValueOnce(rejection); + + const caught = await sendMessageSlack("channel:C123", "hello", { + token: "xoxb-test", + cfg: SLACK_TEST_CFG, + client, + onPlatformSendDispatch, + }).catch((error: unknown) => error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(caught).toMatchObject({ retryable: false, cause: rejection }); + expect(caught).toMatchObject({ message: expect.stringContaining(code) }); + expect(onPlatformSendDispatch).toHaveBeenCalledOnce(); + expect(client.chat.postMessage).toHaveBeenCalledOnce(); + }, + ); + + it("marks account_inactive from durable DM resolution as a permanent non-dispatch", async () => { + const client = createSlackSendTestClient(); + const rejection = slackPlatformError("account_inactive"); + client.conversations.open.mockRejectedValueOnce(rejection); + + const caught = await sendMessageSlack("user:U123", "hello", { + token: "xoxb-test", + cfg: SLACK_TEST_CFG, + client, + deliveryQueueId: "queue-dm-account-inactive", + }).catch((error: unknown) => error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(caught).toMatchObject({ retryable: false, cause: rejection }); + expect(client.chat.postMessage).not.toHaveBeenCalled(); + }); + + it("retains earlier Slack delivery evidence when a later chunk is permanently rejected", async () => { + const client = createSlackSendTestClient(); + client.chat.postMessage + .mockResolvedValueOnce({ ts: "171234.100", channel: "C123" }) + .mockRejectedValueOnce(slackPlatformError("messages_tab_disabled")); + const delivered: string[] = []; + + const caught = await sendMessageSlack("channel:C123", "a".repeat(SLACK_TEXT_LIMIT + 1), { + token: "xoxb-test", + cfg: SLACK_TEST_CFG, + client, + onDeliveryResult: (result) => { + delivered.push(result.messageId); + }, + }).catch((error: unknown) => error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(delivered).toEqual(["171234.100"]); + expect(client.chat.postMessage).toHaveBeenCalledTimes(2); + }); + + it("does not classify a persistence callback failure as a Slack API rejection", async () => { + const client = createSlackSendTestClient(); + const callbackError = slackPlatformError("messages_tab_disabled"); + client.chat.postMessage.mockResolvedValueOnce({ ts: "171234.100", channel: "C123" }); + + const caught = await sendMessageSlack("channel:C123", "hello", { + token: "xoxb-test", + cfg: SLACK_TEST_CFG, + client, + onDeliveryResult: () => { + throw callbackError; + }, + }).catch((error: unknown) => error); + + expect(caught).toBe(callbackError); + expect(caught).not.toBeInstanceOf(PlatformMessageNotDispatchedError); + }); + + it.each([ + [ + "rate limit", + Object.assign(new Error("A rate limit was exceeded"), { + code: "slack_webapi_rate_limited_error", + retryAfter: 1, + }), + ], + [ + "HTTP 500", + Object.assign(new Error("An HTTP protocol error occurred"), { + code: "slack_webapi_http_error", + statusCode: 500, + }), + ], + [ + "network reset", + Object.assign(new Error("A request error occurred: read ECONNRESET"), { + code: "slack_webapi_request_error", + original: Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }), + }), + ], + ])("does not terminalize a Slack %s", async (_name, error) => { + const client = createSlackSendTestClient(); + client.chat.postMessage.mockRejectedValueOnce(error); + + const caught = await sendMessageSlack("channel:C123", "hello", { + token: "xoxb-test", + cfg: SLACK_TEST_CFG, + client, + }).catch((caughtError: unknown) => caughtError); + + expect(caught).toBe(error); + expect(caught).not.toBeInstanceOf(PlatformMessageNotDispatchedError); + }); + + it.each([ + ["null", null], + ["an unlisted platform error code", slackPlatformError("channel_not_found")], + ])("rethrows %s by identity", (_name, rejection) => { + let caught: unknown = "not thrown"; + try { + rethrowSlackPermanentOutboundApiRejection(rejection); + } catch (error) { + caught = error; + } + expect(caught).toBe(rejection); + }); + + it("keeps a non-Error invalid_blocks rejection matchable after the send boundary", () => { + let caught: unknown = "not thrown"; + try { + rethrowSlackPermanentOutboundApiRejection({ data: { error: "invalid_blocks" } }); + } catch (error) { + caught = error; + } + expect(isSlackInvalidBlocksError(caught)).toBe(true); + }); +}); diff --git a/extensions/slack/src/send.ts b/extensions/slack/src/send.ts index a93b38e558f8..15ed6cd3a32c 100644 --- a/extensions/slack/src/send.ts +++ b/extensions/slack/src/send.ts @@ -35,6 +35,7 @@ import { buildSlackCompleteBlocksFallbackText } from "./blocks-fallback.js"; import { validateSlackBlocksArray } from "./blocks-input.js"; import { postSlackMessageBestEffort, + rethrowSlackPermanentOutboundApiRejection, uploadSlackFile, withSlackDnsRequestRetry, } from "./client-delivery.js"; @@ -608,7 +609,7 @@ async function resolveChannelId( } const response = await withSlackDnsRequestRetry("conversations.open", () => client.conversations.open({ users: recipient.id }), - ); + ).catch(rethrowSlackPermanentOutboundApiRejection); const channelId = response.channel?.id; if (!channelId) { throw new Error("Failed to open Slack DM channel"); diff --git a/extensions/slack/src/send.upload.test.ts b/extensions/slack/src/send.upload.test.ts index 1d5f2dc48448..6c068d6dd904 100644 --- a/extensions/slack/src/send.upload.test.ts +++ b/extensions/slack/src/send.upload.test.ts @@ -214,6 +214,13 @@ function createUploadTestClient(slackApiUrl = "https://slack.com/api/"): UploadT } as unknown as UploadTestClient; } +function slackPlatformError(code: string): Error { + return Object.assign(new Error(`An API error occurred: ${code}`), { + code: "slack_webapi_platform_error", + data: { ok: false, error: code }, + }); +} + type UploadOverrides = Omit[2]>, "cfg" | "client">; type UploadParams = UploadOverrides & { mediaUrl: string; target?: string; message?: string }; @@ -317,6 +324,70 @@ describe("sendMessageSlack file upload with user IDs", () => { }, ); + it("marks account_inactive from files.getUploadURLExternal as a permanent non-dispatch", async () => { + const rejection = slackPlatformError("account_inactive"); + const onPlatformSendDispatch = vi.fn(); + client.files.getUploadURLExternal.mockRejectedValueOnce(rejection); + + const caught = await sendUpload(client, { + mediaUrl: "/tmp/account-inactive.png", + onPlatformSendDispatch, + }).catch((error: unknown) => error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(caught).toMatchObject({ retryable: false, cause: rejection }); + expect(globalThis.fetch).not.toHaveBeenCalled(); + expect(onPlatformSendDispatch).not.toHaveBeenCalled(); + expect(client.files.completeUploadExternal).not.toHaveBeenCalled(); + }); + + it("keeps a definitive completeUploadExternal rejection ambiguous", async () => { + // Dispatch is recorded before this call, so even a code that reads as a final + // verdict cannot prove the file was never shared; it must not become a + // non-dispatch assertion. Pairs with the pre-dispatch getUploadURLExternal case. + const rejection = slackPlatformError("messages_tab_disabled"); + const onPlatformSendDispatch = vi.fn(); + client.files.completeUploadExternal.mockRejectedValueOnce(rejection); + + const caught = await sendUpload(client, { + mediaUrl: "/tmp/messages-tab-disabled.png", + onPlatformSendDispatch, + }).catch((error: unknown) => error); + + expect(onPlatformSendDispatch).toHaveBeenCalledOnce(); + expect(caught).toBe(rejection); + expect(caught).not.toBeInstanceOf(PlatformMessageNotDispatchedError); + }); + + it("keeps getUploadURLExternal network failures ambiguous", async () => { + const rejection = Object.assign(new Error("read ECONNRESET"), { + code: "slack_webapi_request_error", + }); + client.files.getUploadURLExternal.mockRejectedValueOnce(rejection); + + const caught = await sendUpload(client, { + mediaUrl: "/tmp/network-failure.png", + }).catch((error: unknown) => error); + + expect(caught).toBe(rejection); + expect(caught).not.toBeInstanceOf(PlatformMessageNotDispatchedError); + }); + + it("keeps completeUploadExternal HTTP failures ambiguous", async () => { + const rejection = Object.assign(new Error("Slack HTTP 500"), { + code: "slack_webapi_http_error", + statusCode: 500, + }); + client.files.completeUploadExternal.mockRejectedValueOnce(rejection); + + const caught = await sendUpload(client, { + mediaUrl: "/tmp/http-failure.png", + }).catch((error: unknown) => error); + + expect(caught).toBe(rejection); + expect(caught).not.toBeInstanceOf(PlatformMessageNotDispatchedError); + }); + it("disables image optimization for forced-media uploads", async () => { await sendUpload(client, { mediaUrl: "/tmp/original.png", diff --git a/extensions/telegram/src/bot/delivery.test.ts b/extensions/telegram/src/bot/delivery.test.ts index 79a41d0b77d4..0f8bf496b799 100644 --- a/extensions/telegram/src/bot/delivery.test.ts +++ b/extensions/telegram/src/bot/delivery.test.ts @@ -1964,6 +1964,34 @@ describe("deliverReplies", () => { expect(sendMessage).toHaveBeenCalledTimes(3); }); + it("maps a supergroup migration rejection without rewriting the streaming target", async () => { + const chatId = "-123456789"; + const migratedChatId = -1_001_234_567_890; + const terminal = Object.assign( + new Error("400: Bad Request: group chat was upgraded to a supergroup chat"), + { + name: "GrammyError", + error_code: 400, + description: "Bad Request: group chat was upgraded to a supergroup chat", + parameters: { migrate_to_chat_id: migratedChatId }, + }, + ); + const sendMessage = vi.fn().mockRejectedValue(terminal); + + const observed = await sendTelegramText( + createBot({ sendMessage }), + chatId, + "hello", + createRuntime(), + ).catch((error: unknown) => error); + + expect(observed).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(observed).toMatchObject({ cause: terminal, retryable: false }); + expect(observed).toMatchObject({ message: expect.stringContaining(String(migratedChatId)) }); + expect(sendMessage).toHaveBeenCalledOnce(); + expect(firstMockCallArg(sendMessage, 0)).toBe(chatId); + }); + it("keeps broad 421-shaped streaming send errors ambiguous", async () => { const edgeError = Object.assign(new Error("421 Misdirected Request"), { status: 421 }); const sendMessage = vi.fn().mockRejectedValue(edgeError); diff --git a/extensions/telegram/src/network-errors.test.ts b/extensions/telegram/src/network-errors.test.ts index 956859ff32e1..3825cae4cf3a 100644 --- a/extensions/telegram/src/network-errors.test.ts +++ b/extensions/telegram/src/network-errors.test.ts @@ -1,4 +1,5 @@ // Telegram tests cover network errors plugin behavior. +import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; import { describe, expect, it } from "vitest"; import { isRecoverableTelegramNetworkError, @@ -9,6 +10,7 @@ import { isTelegramClientRejection, isTelegramPollingNetworkError, isTelegramServerError, + rethrowTelegramSendError, tagTelegramNetworkError, TelegramRequestNotStartedError, } from "./network-errors.js"; @@ -18,6 +20,15 @@ const errorWithCode = (message: string, code: string) => const errorWithTelegramCode = (message: string, error_code: number) => Object.assign(new Error(message), { error_code }); +function captureTelegramSendError(error: unknown): unknown { + try { + rethrowTelegramSendError(error); + } catch (caught) { + return caught; + } + throw new Error("Expected Telegram send error to be rethrown"); +} + const plainErrorPredicateCases = [ { name: "isTelegramServerError", @@ -333,6 +344,79 @@ describe("isSafeToRetrySendError", () => { }); }); +describe("rethrowTelegramSendError", () => { + const migratedChatId = -1_001_234_567_890; + const migrationError = Object.assign( + new Error("400: Bad Request: group chat was upgraded to a supergroup chat"), + { + name: "GrammyError", + error_code: 400, + description: "Bad Request: group chat was upgraded to a supergroup chat", + parameters: { migrate_to_chat_id: migratedChatId }, + }, + ); + + it.each([ + ["direct grammY rejection", migrationError], + [ + "nested provider rejection", + Object.assign(new Error("Telegram send failed"), { cause: migrationError }), + ], + ])("marks a migrated supergroup as a permanent non-dispatch for %s", (_name, error) => { + const caught = captureTelegramSendError(error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(caught).toMatchObject({ + retryable: false, + cause: error, + }); + expect(caught).toMatchObject({ message: expect.stringContaining(String(migratedChatId)) }); + }); + + it.each([ + ["rate limit", errorWithTelegramCode("Too Many Requests", 429)], + ["server failure", errorWithTelegramCode("Bad Gateway", 502)], + ["unrelated client rejection", errorWithTelegramCode("Bad Request: message is empty", 400)], + ["ambiguous network failure", errorWithCode("read ECONNRESET", "ECONNRESET")], + ...(["status", "statusCode"] as const).map((statusField): [string, Error] => [ + `non-Telegram ${statusField} lookalike`, + Object.assign(new Error("migration-shaped HTTP error"), { + [statusField]: 400, + description: "Bad Request: group chat was upgraded to a supergroup chat", + parameters: { migrate_to_chat_id: migratedChatId }, + }), + ]), + [ + "migration parameter without matching description", + Object.assign(new Error("different bad request"), { + error_code: 400, + description: "Bad Request: chat not found", + parameters: { migrate_to_chat_id: migratedChatId }, + }), + ], + ["plain migration text", new Error("400: group chat was upgraded to a supergroup chat")], + ])("does not terminalize a %s", (_name, error) => { + expect(captureTelegramSendError(error)).toBe(error); + }); + + it.each([ + ["without response parameters", undefined], + ["with an unsafe replacement id", Number.MAX_SAFE_INTEGER + 1], + ])("terminalizes a migration response %s without surfacing a target", (_name, target) => { + const error = Object.assign(new Error("migration"), { + error_code: 400, + description: "Bad Request: group chat was upgraded to a supergroup chat", + ...(target === undefined ? {} : { parameters: { migrate_to_chat_id: target } }), + }); + + const caught = captureTelegramSendError(error); + + expect(caught).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(caught).toMatchObject({ retryable: false, cause: error }); + expect(caught).not.toMatchObject({ message: expect.stringContaining(String(target)) }); + }); +}); + describe("isTelegramServerError", () => { it.each([ ["Internal Server Error", 500, true], diff --git a/extensions/telegram/src/network-errors.ts b/extensions/telegram/src/network-errors.ts index 44f3db3bbc5c..bead80ce1696 100644 --- a/extensions/telegram/src/network-errors.ts +++ b/extensions/telegram/src/network-errors.ts @@ -8,9 +8,14 @@ import { } from "openclaw/plugin-sdk/error-runtime"; import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime"; import { classifyTransientNetworkErrorCode } from "openclaw/plugin-sdk/retry-runtime"; -import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + isRecord, + normalizeLowercaseStringOrEmpty, +} from "openclaw/plugin-sdk/string-coerce-runtime"; const TELEGRAM_NETWORK_ORIGIN = Symbol("openclaw.telegram.network-origin"); +const TELEGRAM_SUPERGROUP_MIGRATION_DESCRIPTION = + "Bad Request: group chat was upgraded to a supergroup chat"; export class TelegramRequestNotStartedError extends Error { constructor(message = "Telegram request did not start") { @@ -28,9 +33,14 @@ function isTelegramRequestNotStartedError(err: unknown): boolean { } export function rethrowTelegramSendError(err: unknown): never { - throw isTelegramRequestNotStartedError(err) - ? new PlatformMessageNotDispatchedError("Telegram request not started", { cause: err }) - : err; + if (isTelegramRequestNotStartedError(err)) { + throw new PlatformMessageNotDispatchedError("Telegram request not started", { cause: err }); + } + const migrationRejection = describeTelegramSupergroupMigration(err); + if (migrationRejection === undefined) { + throw err; + } + throw new PlatformMessageNotDispatchedError(migrationRejection, { cause: err, retryable: false }); } const TELEGRAM_ADDITIONAL_TRANSIENT_ERROR_CODES = new Set([ @@ -139,6 +149,28 @@ function getNumericHttpStatus(err: unknown): number | undefined { return undefined; } +// Once a basic group is upgraded, its old id answers every send with this exact Bot API +// 400 (https://core.telegram.org/bots/api#making-requests). Matching the description +// literally keeps every other 400 retryable; `migrate_to_chat_id` is bounded at 52 +// significant bits, so a non-safe integer is not the documented id and stays unreported. +function describeTelegramSupergroupMigration(err: unknown): string | undefined { + for (const candidate of collectTelegramErrorCandidates(err)) { + if (!isRecord(candidate) || candidate.error_code !== 400) { + continue; + } + if (candidate.description !== TELEGRAM_SUPERGROUP_MIGRATION_DESCRIPTION) { + continue; + } + const migratedChatId = isRecord(candidate.parameters) + ? candidate.parameters.migrate_to_chat_id + : undefined; + return typeof migratedChatId === "number" && Number.isSafeInteger(migratedChatId) + ? `Telegram rejected send: group migrated to supergroup ${migratedChatId}` + : "Telegram rejected send: group migrated to a supergroup"; + } + return undefined; +} + export function isTelegramMisdirectedRequestError(err: unknown): boolean { for (const candidate of collectTelegramErrorCandidates(err)) { const code = normalizeCode(getErrorCode(candidate)); diff --git a/extensions/telegram/src/send.test.ts b/extensions/telegram/src/send.test.ts index 98ddd08e60ac..21e821786462 100644 --- a/extensions/telegram/src/send.test.ts +++ b/extensions/telegram/src/send.test.ts @@ -3489,6 +3489,34 @@ describe("sendMessageTelegram", () => { expect(sendMessage).toHaveBeenCalledTimes(2); }); + it("maps a supergroup migration rejection without rewriting the durable target", async () => { + const chatId = "-123456789"; + const migratedChatId = -1_001_234_567_890; + const terminal = Object.assign( + new Error("400: Bad Request: group chat was upgraded to a supergroup chat"), + { + name: "GrammyError", + error_code: 400, + description: "Bad Request: group chat was upgraded to a supergroup chat", + parameters: { migrate_to_chat_id: migratedChatId }, + }, + ); + const sendMessage = vi.fn().mockRejectedValue(terminal); + const api = makeTelegramApiTestMock({ sendMessage }); + + const observed = await sendMessageTelegram(chatId, "hi", { + cfg: TELEGRAM_TEST_CFG, + token: "tok", + api, + }).catch((error: unknown) => error); + + expect(observed).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(observed).toMatchObject({ cause: terminal, retryable: false }); + expect(observed).toMatchObject({ message: expect.stringContaining(String(migratedChatId)) }); + expect(sendMessage).toHaveBeenCalledOnce(); + expect(firstMockCall(sendMessage, "sendMessage call")[0]).toBe(chatId); + }); + it("keeps broad 421-shaped durable send errors ambiguous", async () => { const chatId = "123"; const edgeError = Object.assign(new Error("421 Misdirected Request"), { status: 421 }); diff --git a/scripts/test-projects.test-support.mts b/scripts/test-projects.test-support.mts index a7313808b233..fe27d3363cdb 100644 --- a/scripts/test-projects.test-support.mts +++ b/scripts/test-projects.test-support.mts @@ -561,6 +561,14 @@ const PRECISE_SOURCE_TEST_TARGETS = new Map([ "extensions/slack/src/monitor/provider.auth-test-token.test.ts", ], ], + [ + "extensions/slack/src/channel-actions.ts", + [ + "extensions/slack/src/actions.reactions-limit.test.ts", + "extensions/slack/src/channel-actions-setup-status.contract.test.ts", + "extensions/slack/src/message-tools.test.ts", + ], + ], [ "src/gateway/worker-environments/worker-turn-launcher.ts", [ @@ -1945,6 +1953,9 @@ function resolveToolingChangedTestTargets(changedPaths: string[], cwd = process. return null; } targets.push(...testTargets); + if (CHANNEL_PLUGIN_SHAPE_PARITY_WIRING_PATHS.has(changedPath)) { + targets.push(CHANNEL_PLUGIN_SHAPE_PARITY_TEST_TARGET); + } } return [...new Set(targets)]; } diff --git a/src/infra/outbound/delivery-queue.recovery.test.ts b/src/infra/outbound/delivery-queue.recovery.test.ts index ac67e3c95f61..fc5cc8db6b09 100644 --- a/src/infra/outbound/delivery-queue.recovery.test.ts +++ b/src/infra/outbound/delivery-queue.recovery.test.ts @@ -763,6 +763,36 @@ describe("delivery-queue recovery", () => { expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); expect(readOutboundQueueStatus(tmpDir(), id)).toBeUndefined(); }); + it("keeps a partially sent batch unknown when a later send has a permanent rejection", async () => { + const id = await enqueueDemoRecoveryDelivery(["first", "second"]); + const rejection = new PlatformMessageNotDispatchedError( + "Slack outbound delivery rejected: messages_tab_disabled", + { cause: new Error("messages_tab_disabled"), retryable: false }, + ); + const partialFailure = new OutboundDeliveryError("second send failed", { + cause: rejection, + results: [{ channel: "demo-channel-c", messageId: "m1" }], + payloadOutcomes: [ + { index: 0, status: "sent", results: [{ channel: "demo-channel-c", messageId: "m1" }] }, + { + index: 1, + status: "failed", + error: rejection, + sentBeforeError: false, + stage: "platform_send", + }, + ], + stage: "platform_send", + }); + + const { result } = await runRecovery({ + deliver: vi.fn().mockRejectedValue(partialFailure), + }); + + expect(result).toMatchObject({ recovered: 0, failed: 1 }); + await expectPendingEntry({ id, recoveryState: "unknown_after_send", retryCount: 0 }); + expect(readOutboundQueueStatus(tmpDir(), id)).toBe("pending"); + }); it("keeps a best-effort recovery failure retryable when no payload was sent", async () => { await enqueueDemoRecoveryDelivery(["first"], { bestEffort: true }); const deliver = vi.fn(async (params: PayloadOutcomeSink) => { @@ -1269,6 +1299,57 @@ describe("delivery-queue recovery", () => { await runIf(typedRejection, () => expect(deliver).toHaveBeenCalledOnce()); await runIf(!typedRejection, () => expectMockMessageContaining(log.warn, "permanent error")); }); + it("persists a nested channel rejection as the only terminal across recovery restart", async () => { + const operationId = "operation-channel-permanent-rejection"; + const scope = await createConversationRecoveryFixture(operationId); + const rejection = new PlatformMessageNotDispatchedError( + "Slack chat.postMessage rejected: messages_tab_disabled", + { + cause: new Error("messages_tab_disabled"), + retryable: false, + }, + ); + const deliver = vi.fn().mockRejectedValue( + new OutboundDeliveryError("Slack delivery failed", { + cause: rejection, + payloadOutcomes: [ + { + index: 0, + status: "failed", + error: rejection, + sentBeforeError: false, + stage: "platform_send", + }, + ], + stage: "platform_send", + }), + ); + + try { + const first = await runRecovery({ deliver }); + expect(first.result).toEqual(RECOVERY_SUMMARY.failed); + expect(deliver).toHaveBeenCalledOnce(); + expect(getConversationDeliveryOperation(scope, operationId)).toMatchObject({ + status: "rejected", + rejectionError: "Slack chat.postMessage rejected: messages_tab_disabled", + }); + expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); + expect(readOutboundQueueStatus(tmpDir(), operationId)).toBe("failed"); + + closeOpenClawAgentDatabasesForTest(); + const replay = vi.fn(); + const second = await runRecovery({ deliver: replay }); + expect(second.result).toEqual(RECOVERY_SUMMARY.empty); + expect(replay).not.toHaveBeenCalled(); + expect(getConversationDeliveryOperation(scope, operationId)).toMatchObject({ + status: "rejected", + rejectionError: "Slack chat.postMessage rejected: messages_tab_disabled", + }); + expect(readOutboundQueueStatus(tmpDir(), operationId)).toBe("failed"); + } finally { + closeOpenClawAgentDatabasesForTest(); + } + }); it("passes skipQueue: true to prevent re-enqueueing during recovery", async () => { await enqueueRecoveryDelivery(); const deliver = vi.fn().mockResolvedValue([]); diff --git a/test/slack-outbound-permanent-rejection-loopback.test.ts b/test/slack-outbound-permanent-rejection-loopback.test.ts new file mode 100644 index 000000000000..a2f3db2bf76c --- /dev/null +++ b/test/slack-outbound-permanent-rejection-loopback.test.ts @@ -0,0 +1,234 @@ +// Root-owned integration may combine the public Slack plugin with the durable queue runtime. +import { createServer, type Server } from "node:http"; +import type { AddressInfo, Socket } from "node:net"; +import { sendDurableMessageBatch } from "openclaw/plugin-sdk/channel-outbound"; +import { + createEmptyPluginRegistry, + createTestRegistry, + resetPluginRuntimeStateForTest, + resetGlobalHookRunner, + setActivePluginRegistry, +} from "openclaw/plugin-sdk/channel-test-helpers"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime"; +import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; +import { + closeOpenClawAgentDatabasesForTest, + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "openclaw/plugin-sdk/sqlite-runtime-testing"; +import { withStateDirEnv } from "openclaw/plugin-sdk/test-env"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { getDeliveryQueueEntryStatus } from "../src/infra/delivery-queue-sqlite.js"; +import { OUTBOUND_DELIVERY_QUEUE_NAME } from "../src/infra/outbound/delivery-queue-media-staging.js"; + +const CLASSIFIED_CODES = ["messages_tab_disabled", "account_inactive"] as const; +const DELIVERY_INTENT_PREFIX = "slack-loopback-permanent-rejection"; + +type SlackPermanentRejectionCode = (typeof CLASSIFIED_CODES)[number]; + +type SlackLoopbackRequest = { + body: string; + code?: SlackPermanentRejectionCode; + method: string | undefined; + text: string; + url: string; +}; + +type SlackLoopback = { + apiUrl: string; + requests: SlackLoopbackRequest[]; + close: () => Promise; +}; + +function readQueueTerminal( + stateDir: string, + intentId: string, +): { retryCount: number; status: string } | undefined { + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + const row = db + // sqlite-allow-raw: The proof reads one exact queue owner after terminalization. + .prepare( + "SELECT status, retry_count FROM delivery_queue_entries WHERE queue_name = ? AND id = ?", + ) + .get(OUTBOUND_DELIVERY_QUEUE_NAME, intentId) as + | { retry_count: number; status: string } + | undefined; + return row ? { retryCount: row.retry_count, status: row.status } : undefined; +} + +async function startSlackPermanentRejectionLoopback(): Promise { + const requests: SlackLoopbackRequest[] = []; + const sockets = new Set(); + const server: Server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + const text = new URLSearchParams(body).get("text") ?? ""; + const code = CLASSIFIED_CODES.find((candidate) => text.includes(candidate)); + requests.push({ + body, + ...(code ? { code } : {}), + method: request.method, + text, + url: request.url ?? "", + }); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: false, error: code ?? "unexpected_test_request" })); + }); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const { port } = server.address() as AddressInfo; + return { + apiUrl: `http://127.0.0.1:${port}/api/`, + requests, + close: async () => { + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }, + }; +} + +describe("Slack permanent rejections over real Web API transport", () => { + afterEach(() => { + vi.unstubAllEnvs(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + resetGlobalHookRunner(); + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it("dead-letters both real platform rejections and never replays them after restart", async () => { + const loopback = await startSlackPermanentRejectionLoopback(); + vi.stubEnv("SLACK_API_URL", loopback.apiUrl); + try { + const { slackPlugin } = await import("../extensions/slack/api.js"); + const cfg = { + channels: { slack: { botToken: "xoxb-loopback" } }, + } satisfies OpenClawConfig; + setActivePluginRegistry( + createTestRegistry([{ pluginId: "slack", plugin: slackPlugin, source: "test" }]), + ); + + await withStateDirEnv("openclaw-slack-permanent-loopback-", async ({ stateDir }) => { + try { + const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + for (const [index, code] of CLASSIFIED_CODES.entries()) { + const intentId = `${DELIVERY_INTENT_PREFIX}-${code}`; + const text = `real transport permanent rejection: ${code}`; + const staged = await sendDurableMessageBatch({ + cfg, + channel: "slack", + to: "channel:C123", + accountId: "default", + durability: "required", + deliveryIntentId: intentId, + completionRetention: { + idPrefix: "slack-loopback-", + maxAgeMs: 60_000, + maxEntries: 10, + }, + maxRetries: 10, + payloads: [{ text }], + deps: { + slack: async () => { + throw new PlatformMessageNotDispatchedError( + "staged before transport for recovery proof", + { cause: new Error("loopback transport not released yet") }, + ); + }, + }, + }); + expect(staged.status).toBe("failed"); + expect(loopback.requests).toHaveLength(index); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, intentId, stateDir), + ).toBe("pending"); + + log.warn.mockClear(); + await drainPendingDeliveries({ + drainKey: `slack:default:${code}`, + logLabel: `Slack loopback ${code} recovery`, + cfg, + stateDir, + log, + selectEntry: (entry) => ({ + match: entry.id === intentId, + bypassBackoff: true, + }), + }); + + expect(loopback.requests).toHaveLength(index + 1); + expect(loopback.requests[index]).toMatchObject({ + code, + method: "POST", + text, + url: "/api/chat.postMessage", + }); + expect(log.warn).toHaveBeenCalledWith(expect.stringContaining(code)); + expect(readQueueTerminal(stateDir, intentId)).toEqual({ + retryCount: 1, + status: "failed", + }); + } + + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + for (const code of CLASSIFIED_CODES) { + expect( + getDeliveryQueueEntryStatus( + OUTBOUND_DELIVERY_QUEUE_NAME, + `${DELIVERY_INTENT_PREFIX}-${code}`, + stateDir, + ), + ).toBe("failed"); + } + + await drainPendingDeliveries({ + drainKey: "slack:default:post-restart", + logLabel: "Slack loopback post-restart recovery", + cfg, + stateDir, + log, + selectEntry: (entry) => ({ + match: entry.channel === "slack", + bypassBackoff: true, + }), + }); + expect(loopback.requests).toHaveLength(CLASSIFIED_CODES.length); + + console.log( + `[slack permanent-rejection proof] ${JSON.stringify({ + queueTerminal: "failed", + restartReplayCount: 0, + providerStatus: 200, + classifiedCodes: CLASSIFIED_CODES, + classification: "typed non-retryable", + transport: "@slack/web-api HTTP to 127.0.0.1:", + })}`, + ); + } finally { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + } + }); + } finally { + await loopback.close(); + } + }); +}); diff --git a/test/telegram-outbound-permanent-rejection-loopback.test.ts b/test/telegram-outbound-permanent-rejection-loopback.test.ts new file mode 100644 index 000000000000..1afa22d79da7 --- /dev/null +++ b/test/telegram-outbound-permanent-rejection-loopback.test.ts @@ -0,0 +1,215 @@ +// Root-owned integration may combine the public Telegram plugin with the durable queue runtime. +import { createServer, type Server } from "node:http"; +import type { AddressInfo, Socket } from "node:net"; +import { sendDurableMessageBatch } from "openclaw/plugin-sdk/channel-outbound"; +import { + createEmptyPluginRegistry, + createTestRegistry, + resetPluginRuntimeStateForTest, + resetGlobalHookRunner, + setActivePluginRegistry, +} from "openclaw/plugin-sdk/channel-test-helpers"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime"; +import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; +import { + closeOpenClawAgentDatabasesForTest, + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "openclaw/plugin-sdk/sqlite-runtime-testing"; +import { withStateDirEnv } from "openclaw/plugin-sdk/test-env"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { getDeliveryQueueEntryStatus } from "../src/infra/delivery-queue-sqlite.js"; +import { OUTBOUND_DELIVERY_QUEUE_NAME } from "../src/infra/outbound/delivery-queue-media-staging.js"; + +const MIGRATION_DESCRIPTION = "Bad Request: group chat was upgraded to a supergroup chat"; +const DELIVERY_INTENT_ID = "telegram-loopback-permanent-rejection"; + +type TelegramLoopback = { + apiRoot: string; + requests: Array<{ body: string; method: string | undefined; url: string }>; + close: () => Promise; +}; + +function readQueueTerminal(stateDir: string): { retryCount: number; status: string } | undefined { + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + const row = db + // sqlite-allow-raw: The proof reads one exact queue owner after terminalization. + .prepare( + "SELECT status, retry_count FROM delivery_queue_entries WHERE queue_name = ? AND id = ?", + ) + .get(OUTBOUND_DELIVERY_QUEUE_NAME, DELIVERY_INTENT_ID) as + | { retry_count: number; status: string } + | undefined; + return row ? { retryCount: row.retry_count, status: row.status } : undefined; +} + +async function startTelegramMigrationLoopback(): Promise { + const requests: TelegramLoopback["requests"] = []; + const sockets = new Set(); + const server: Server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + requests.push({ + body: Buffer.concat(chunks).toString("utf8"), + method: request.method, + url: request.url ?? "", + }); + response.writeHead(400, { "content-type": "application/json" }); + response.end( + JSON.stringify({ + ok: false, + error_code: 400, + description: MIGRATION_DESCRIPTION, + parameters: { migrate_to_chat_id: -1_001_234_567_890 }, + }), + ); + }); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const { port } = server.address() as AddressInfo; + return { + apiRoot: `http://127.0.0.1:${port}`, + requests, + close: async () => { + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }, + }; +} + +describe("Telegram permanent rejection over real Bot API transport", () => { + afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + resetGlobalHookRunner(); + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it("dead-letters one real migration rejection and never replays it after restart", async () => { + const loopback = await startTelegramMigrationLoopback(); + try { + const { telegramPlugin } = await import("../extensions/telegram/api.js"); + const cfg = { + channels: { + telegram: { + botToken: "123456:loopback", + apiRoot: loopback.apiRoot, + }, + }, + } satisfies OpenClawConfig; + setActivePluginRegistry( + createTestRegistry([{ pluginId: "telegram", plugin: telegramPlugin, source: "test" }]), + ); + + await withStateDirEnv("openclaw-telegram-permanent-loopback-", async ({ stateDir }) => { + try { + const staged = await sendDurableMessageBatch({ + cfg, + channel: "telegram", + to: "123", + accountId: "default", + durability: "required", + deliveryIntentId: DELIVERY_INTENT_ID, + completionRetention: { + idPrefix: "telegram-loopback-", + maxAgeMs: 60_000, + maxEntries: 10, + }, + maxRetries: 10, + payloads: [{ text: "real transport permanent rejection" }], + deps: { + telegram: async () => { + throw new PlatformMessageNotDispatchedError( + "staged before transport for recovery proof", + { cause: new Error("loopback transport not released yet") }, + ); + }, + }, + }); + expect(staged.status).toBe("failed"); + expect(loopback.requests).toHaveLength(0); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, DELIVERY_INTENT_ID, stateDir), + ).toBe("pending"); + + const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + await drainPendingDeliveries({ + drainKey: "telegram:default", + logLabel: "Telegram loopback permanent rejection recovery", + cfg, + stateDir, + log, + selectEntry: (entry) => ({ + match: entry.channel === "telegram", + bypassBackoff: true, + }), + }); + + expect(loopback.requests).toHaveLength(1); + expect(log.warn).toHaveBeenCalledWith( + expect.stringContaining( + "Telegram rejected send: group migrated to supergroup -1001234567890", + ), + ); + expect(loopback.requests[0]).toMatchObject({ method: "POST" }); + expect(loopback.requests[0]?.url).toMatch(/\/sendMessage$/u); + expect(loopback.requests[0]?.body).toContain("real transport permanent rejection"); + expect(readQueueTerminal(stateDir)).toEqual({ retryCount: 1, status: "failed" }); + + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, DELIVERY_INTENT_ID, stateDir), + ).toBe("failed"); + + await drainPendingDeliveries({ + drainKey: "telegram:default", + logLabel: "Telegram loopback post-restart recovery", + cfg, + stateDir, + log, + selectEntry: (entry) => ({ + match: entry.channel === "telegram", + bypassBackoff: true, + }), + }); + expect(loopback.requests).toHaveLength(1); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, DELIVERY_INTENT_ID, stateDir), + ).toBe("failed"); + + console.log( + `[telegram permanent-rejection proof] ${JSON.stringify({ + queueTerminal: "failed", + restartReplayCount: 0, + providerStatus: 400, + classification: "typed non-retryable", + transport: "grammY Bot API HTTP to 127.0.0.1:", + })}`, + ); + } finally { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + } + }); + } finally { + await loopback.close(); + } + }); +});