fix(reports): harden Slack fallback delivery

UPDATING.md

@@ -467,7 +467,9 @@ With the flag enabled: `DELETE /api/v1/chart/<id>` no longer hard-deletes the ch
 
 **Importer behavior:** importing a chart YAML whose UUID matches an existing **soft-deleted** chart is treated as an implicit restore-with-update — **and this happens even when `overwrite` is not set**. This is a deliberate asymmetry with active rows: an active chart imported without `overwrite=true` is returned unchanged, but a soft-deleted UUID match is restored *and* has the upload's contents applied regardless of the `overwrite` argument, on the reasoning that re-importing a deleted chart's exact UUID is an explicit request to bring it back. The restore preserves the original PK and all out-of-archive references (`dashboard_slices` junctions, `report.chart_id`, tag rows). The operation is permission-gated: non-editors get `ImportFailedError`, and callers without `can_write` get `ImportFailedError` instead of silently receiving the soft-deleted row.
 
-- [39914](https://github.com/apache/superset/pull/39914) `ALERT_REPORT_SLACK_V2` now defaults to `True` and the legacy Slack v1 integration (`Slack` recipient type, `files.upload` API) is deprecated for removal in the next major. Slack blocked new apps from `files.upload` in May 2024 and fully retired the method for all apps on November 12, 2025; because the v1 path sends files through `files.upload`, v1 file-bearing sends now fail at the API level — only text-only `chat_postMessage` still works via the legacy path. Grant your Slack bot the `channels:read` and `groups:read` scopes so existing `Slack` recipients can be auto-upgraded to `SlackV2` on next send. Operators who explicitly override the flag to `False`, or whose Slack bot is missing those scopes, will see deprecation warnings while text-only sends continue through the legacy path.
+### Slack v1 deprecation and automatic Slack v2 upgrades
+
+- [39914](https://github.com/apache/superset/pull/39914) makes `ALERT_REPORT_SLACK_V2` default to `True` and deprecates the legacy Slack v1 integration (`Slack` recipient type and retired `files.upload` API) for removal in the next major; [42089](https://github.com/apache/superset/pull/42089) makes v1 file-bearing sends fail before attempting the retired v1 upload with actionable Slack v2 scope guidance, preserves text-only delivery through v1 when saved channel names cannot be upgraded, and limits application retries to transient transport/server failures plus envelope-only `ratelimited` responses (Slack SDK configuration owns HTTP 429 retries, while permanent 4xx errors fail immediately). Grant the Slack bot both `channels:read` and `groups:read` so existing `Slack` recipients can be auto-upgraded to `SlackV2` on their next send. Operators who disable the flag or whose bot lacks those scopes receive deprecation warnings while text-only legacy delivery remains available. The application retry budget applies to each Slack API operation, so high-fan-out reports can occupy a worker for roughly 150 seconds per persistently failing channel/file operation; size report worker pools and time limits accordingly.
 
 ### Soft delete and restore for dashboards
 

superset/commands/report/execute.py

@@ -81,16 +81,25 @@ from superset.reports.notifications.exceptions import (
     NotificationParamException,
     SlackV1NotificationError,
 )
+from superset.reports.notifications.slack import (
+    SLACK_V1_FILE_UPLOAD_MESSAGE,
+    SlackNotification,
+)
 from superset.subjects.types import SubjectType
 from superset.tasks.utils import get_executor
 from superset.utils import json
-from superset.utils.core import HeaderDataType, override_user, recipients_string_to_list
+from superset.utils.core import HeaderDataType, override_user
 from superset.utils.csv import get_chart_csv_data, get_chart_dataframe
 from superset.utils.decorators import logs_context, transaction
 from superset.utils.file import sanitize_title
 from superset.utils.pdf import build_pdf_from_screenshots
 from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
-from superset.utils.slack import get_channels_with_search, SlackChannelTypes
+from superset.utils.slack import (
+    get_channels_with_search,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+    SlackChannelTypes,
+)
 from superset.utils.urls import get_url_path
 
 if TYPE_CHECKING:
@@ -99,6 +108,48 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _get_slack_channels_by_target(
+    search_string: str,
+    *,
+    force: bool = False,
+) -> dict[str, Any]:
+    """Fetch Slack channels and index them by case-insensitive name and id."""
+    force_kwargs = {"force": True} if force else {}
+    channels = get_channels_with_search(
+        search_string=search_string,
+        types=[
+            SlackChannelTypes.PRIVATE,
+            SlackChannelTypes.PUBLIC,
+        ],
+        exact_match=True,
+        **force_kwargs,
+    )
+    return {
+        target.casefold(): channel
+        for channel in channels
+        for target in (channel["name"], channel["id"])
+    }
+
+
+def _resolve_slack_channel_targets(targets: list[str]) -> dict[str, Any]:
+    """Resolve configured Slack names or ids after one forced cache refresh."""
+    search_string = ",".join(targets)
+    channels_by_target = _get_slack_channels_by_target(search_string)
+    missing_channels = [
+        target for target in targets if target.casefold() not in channels_by_target
+    ]
+    if missing_channels:
+        channels_by_target = _get_slack_channels_by_target(search_string, force=True)
+        missing_channels = [
+            target for target in targets if target.casefold() not in channels_by_target
+        ]
+    if missing_channels:
+        raise NotificationParamException(
+            f"Could not find the following channels: {', '.join(missing_channels)}"
+        )
+    return channels_by_target
+
+
 def resolve_executor_user(model: ReportSchedule) -> tuple["User", str]:
     """
     Resolve the executor user for a report schedule.
@@ -139,6 +190,9 @@ class BaseReportState:
         self._start_dttm = datetime.utcnow()
         self._execution_id = execution_id
         self._filter_warnings: list[str] = []
+        self._slack_v2_upgrade_error: (
+            NotificationParamException | UpdateFailedError | None
+        ) = None
 
     def update_report_schedule_and_log(
         self,
@@ -184,31 +238,35 @@ class BaseReportState:
             for recipient in self._report_schedule.recipients:
                 if recipient.type != ReportRecipientType.SLACK:
                     continue
-                slack_recipients = json.loads(recipient.recipient_config_json)
+                try:
+                    slack_recipients = json.loads(recipient.recipient_config_json)
+                except (TypeError, ValueError) as ex:
+                    raise NotificationParamException(
+                        "Invalid Slack recipient configuration"
+                    ) from ex
+                target = (
+                    slack_recipients.get("target")
+                    if isinstance(slack_recipients, dict)
+                    else None
+                )
+                if not isinstance(target, str):
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
                 # V1 method allowed to use leading `#` in the channel name
-                channel_names = (slack_recipients["target"] or "").replace("#", "")
+                channels_list = parse_slack_recipient_targets(target.replace("#", ""))
+                if not channels_list:
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
                 # we need to ensure that existing reports can also fetch
                 # ids from private channels
-                channels = get_channels_with_search(
-                    search_string=channel_names,
-                    types=[
-                        SlackChannelTypes.PRIVATE,
-                        SlackChannelTypes.PUBLIC,
-                    ],
-                    exact_match=True,
+                channels_by_target = _resolve_slack_channel_targets(channels_list)
+                channel_ids = ",".join(
+                    channels_by_target[channel.casefold()]["id"]
+                    for channel in channels_list
                 )
-                channels_list = recipients_string_to_list(channel_names)
-                if len(channels_list) != len(channels):
-                    missing_channels = set(channels_list) - {
-                        channel["name"] for channel in channels
-                    }
-                    msg = (
-                        "Could not find the following channels: "
-                        f"{', '.join(missing_channels)}"
-                    )
-                    raise UpdateFailedError(msg)
-                channel_ids = ",".join(channel["id"] for channel in channels)
                 resolved.append((recipient, json.dumps({"target": channel_ids})))
+        except NotificationParamException as ex:
+            msg = f"Failed to update slack recipients to v2: {str(ex)}"
+            logger.warning(msg)
+            raise NotificationParamException(msg) from ex
         except Exception as ex:
             # No recipient has been mutated yet, so there is no partial upgrade
             # to revert; surface the failure so the configuration can be fixed
@@ -990,6 +1048,99 @@ class BaseReportState:
             header_data=header_data,
         )
 
+    def _send_slack_v1_fallback(
+        self,
+        notification: SlackNotification,
+        notification_content: NotificationContent,
+        update_error: NotificationParamException | UpdateFailedError,
+        *,
+        record_upgrade_failure: bool,
+    ) -> None:
+        """Deliver a text-only notification after one failed atomic v2 upgrade."""
+        if notification_content.has_attachments:
+            if isinstance(update_error, UpdateFailedError):
+                raise UpdateFailedError(
+                    f"{SLACK_V1_FILE_UPLOAD_MESSAGE} "
+                    f"Slack v2 upgrade failed: {update_error}"
+                ) from update_error
+            raise NotificationParamException(
+                f"{SLACK_V1_FILE_UPLOAD_MESSAGE} "
+                f"Slack v2 upgrade failed: {update_error}"
+            ) from update_error
+
+        if record_upgrade_failure:
+            app.config["STATS_LOGGER"].incr("reports.slack.v1_fallback")
+            if isinstance(update_error, UpdateFailedError):
+                app.config["STATS_LOGGER"].incr(
+                    "reports.slack.v1_fallback.system_error"
+                )
+                logger.error(
+                    "Slack v2 upgrade failed with a system error; delivering the "
+                    "text-only report through Slack v1 for this execution: %s",
+                    update_error,
+                    extra={
+                        "execution_id": self._execution_id,
+                        "report_schedule_id": self._report_schedule.id,
+                    },
+                )
+            else:
+                logger.warning(
+                    "Slack v2 upgrade unavailable; attempting a text-only Slack v1 "
+                    "fallback for this execution: %s",
+                    update_error,
+                )
+        notification.send_legacy_text()
+
+    def _send_notification(
+        self,
+        notification_content: NotificationContent,
+        recipient: ReportRecipients,
+    ) -> None:
+        """Send one notification, upgrading Slack v1 recipients when required."""
+        notification = create_notification(recipient, notification_content)
+        if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
+            logger.info(
+                "Would send notification for alert %s, to %s. "
+                "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
+                "set it to False to send notifications.",
+                self._report_schedule.name,
+                recipient.recipient_config_json,
+            )
+            return
+
+        if self._slack_v2_upgrade_error is not None and isinstance(
+            notification, SlackNotification
+        ):
+            self._send_slack_v1_fallback(
+                notification,
+                notification_content,
+                self._slack_v2_upgrade_error,
+                record_upgrade_failure=False,
+            )
+            return
+
+        try:
+            notification.send()
+        except SlackV1NotificationError as ex:
+            if not isinstance(notification, SlackNotification):
+                raise
+            logger.info("Attempting to upgrade the report to Slackv2: %s", str(ex))
+            try:
+                self.update_report_schedule_slack_v2()
+            except (
+                NotificationParamException,
+                UpdateFailedError,
+            ) as update_error:
+                self._slack_v2_upgrade_error = update_error
+                self._send_slack_v1_fallback(
+                    notification,
+                    notification_content,
+                    update_error,
+                    record_upgrade_failure=True,
+                )
+            else:
+                create_notification(recipient, notification_content).send()
+
     def _send(
         self,
         notification_content: NotificationContent,
@@ -1001,29 +1152,10 @@ class BaseReportState:
         :raises: CommandException
         """
         notification_errors: list[SupersetError] = []
+        self._slack_v2_upgrade_error = None
         for recipient in recipients:
-            notification = create_notification(recipient, notification_content)
             try:
-                try:
-                    if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
-                        logger.info(
-                            "Would send notification for alert %s, to %s. "
-                            "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
-                            "set it to False to send notifications.",
-                            self._report_schedule.name,
-                            recipient.recipient_config_json,
-                        )
-                    else:
-                        notification.send()
-                except SlackV1NotificationError as ex:
-                    # The slack notification should be sent with the v2 api
-                    logger.info(
-                        "Attempting to upgrade the report to Slackv2: %s", str(ex)
-                    )
-                    self.update_report_schedule_slack_v2()
-                    recipient.type = ReportRecipientType.SLACKV2
-                    notification = create_notification(recipient, notification_content)
-                    notification.send()
+                self._send_notification(notification_content, recipient)
             except (
                 UpdateFailedError,
                 NotificationParamException,

superset/reports/notifications/base.py

@@ -36,6 +36,11 @@ class NotificationContent:
     url: Optional[str] = None  # url to chart/dashboard for this screenshot
     embedded_data: Optional[pd.DataFrame] = None
 
+    @property
+    def has_attachments(self) -> bool:
+        """Return whether the notification contains any file attachment."""
+        return bool(self.csv or self.xlsx or self.pdf or self.screenshots)
+
 
 class BaseNotification:  # pylint: disable=too-few-public-methods
     """

superset/reports/notifications/slack.py

@@ -15,18 +15,13 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Sequence
-from io import IOBase
-from typing import Union
 
-import backoff
 from flask import g
+from slack_sdk import WebClient
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
@@ -41,24 +36,36 @@ from superset.reports.notifications.exceptions import (
     NotificationUnprocessableException,
     SlackV1NotificationError,
 )
-from superset.reports.notifications.slack_mixin import SlackMixin
+from superset.reports.notifications.slack_mixin import (
+    _call_slack_api,
+    _send_to_slack_channels,
+    SlackMixin,
+)
 from superset.utils import json
-from superset.utils.core import recipients_string_to_list
 from superset.utils.decorators import statsd_gauge
 from superset.utils.slack import (
     get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
     should_use_v2_api,
 )
 
 logger = logging.getLogger(__name__)
 
+SLACK_V1_FILE_UPLOAD_MESSAGE = (
+    "Slack v1 file uploads are no longer supported because Slack retired "
+    "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+    "both the `channels:read` and `groups:read` scopes so the recipient can "
+    "be upgraded to Slack v2."
+)
+
 
 # Deprecated: Slack v1 will be removed in the next major release. The Slack
 # `files.upload` endpoint was retired in 2025, so file-bearing sends already
-# fail at the API level; only text-only `chat_postMessage` sends still work
-# here. When the Slack bot has the `channels:read` and `groups:read` scopes,
-# existing v1 recipients are auto-upgraded to SlackV2 on first send via
-# `update_report_schedule_slack_v2`.
+# fail before attempting the retired v1 upload; only text-only
+# `chat_postMessage` sends still work here. When the Slack bot has the
+# `channels:read` and `groups:read` scopes, existing v1 recipients are
+# auto-upgraded to SlackV2 on first send via `update_report_schedule_slack_v2`.
 class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
     """
     Sends a slack notification for a report recipient
@@ -66,58 +73,46 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
 
     type = ReportRecipientType.SLACK
 
-    def _get_channel(self) -> str:
+    def _get_channels(self) -> list[str]:
         """
-        Get the recipient's channel(s).
-        Note Slack SDK uses "channel" to refer to one or more
-        channels. Multiple channels are demarcated by a comma.
-        :returns: The comma separated list of channel(s)
+        Get the recipient's normalized channel list.
+
+        :returns: channel names with duplicates removed in configured order
         """
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        try:
+            recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE) from ex
 
-        return ",".join(recipients_string_to_list(recipient_str))
+        if not isinstance(recipient_str, str):
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
 
-    def _get_inline_files(
-        self,
-    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
-        if self._content.csv:
-            return ("csv", [self._content.csv])
-        if self._content.xlsx:
-            return ("xlsx", [self._content.xlsx])
-        if self._content.screenshots:
-            return ("png", self._content.screenshots)
-        if self._content.pdf:
-            return ("pdf", [self._content.pdf])
-        return (None, [])
+        return parse_slack_recipient_targets(recipient_str)
+
+    @staticmethod
+    def _send_text(client: WebClient, channels: list[str], body: str) -> None:
+        """Send a text notification once to each configured channel."""
+        if not channels:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+        _send_to_slack_channels(
+            channels,
+            lambda target: _call_slack_api(
+                client.chat_postMessage,
+                channel=target,
+                text=body,
+            ),
+        )
+
+    def _send_legacy_text(self) -> None:
+        if self._content.has_attachments:
+            raise NotificationParamException(SLACK_V1_FILE_UPLOAD_MESSAGE)
 
-    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
-    @statsd_gauge("reports.slack.send")
-    def send(self) -> None:
-        file_type, files = self._get_inline_files()
-        title = self._content.name
         body = self._get_body(content=self._content)
         global_logs_context = getattr(g, "logs_context", {}) or {}
-
-        # see if the v2 api will work
-        if should_use_v2_api():
-            # if we can fetch channels, then raise an error and use the v2 api
-            raise SlackV1NotificationError
-
         try:
             client = get_slack_client()
-            channel = self._get_channel()
-            # files_upload returns SlackResponse as we run it in sync mode.
-            if files:
-                for file in files:
-                    client.files_upload(
-                        channels=channel,
-                        file=file,
-                        initial_comment=body,
-                        title=title,
-                        filetype=file_type,
-                    )
-            else:
-                client.chat_postMessage(channel=channel, text=body)
+            channels = self._get_channels()
+            self._send_text(client, channels, body)
             logger.info(
                 "Report sent to slack",
                 extra={
@@ -134,9 +129,18 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
             # this is the base class for all slack client errors
             # keep it last so that it doesn't interfere with @backoff
             raise NotificationUnprocessableException(str(ex)) from ex
+
+    @statsd_gauge("reports.slack.send")
+    def send_legacy_text(self) -> None:
+        """Send through Slack v1 without repeating the v2 availability probe."""
+        self._send_legacy_text()
+
+    @statsd_gauge("reports.slack.send")
+    def send(self) -> None:
+        if should_use_v2_api(raise_on_error=self._content.has_attachments):
+            raise SlackV1NotificationError
+        self._send_legacy_text()

superset/reports/notifications/slack_mixin.py

@@ -15,15 +15,90 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from collections.abc import Callable
+
+import backoff
 import pandas as pd
 from flask_babel import gettext as __
+from slack_sdk.errors import SlackApiError, SlackClientNotConnectedError
 
 from superset.reports.notifications.base import NotificationContent
+from superset.reports.notifications.exceptions import (
+    NotificationError,
+    NotificationUnprocessableException,
+)
+from superset.utils.slack import (
+    _get_slack_api_error_code,
+    _get_slack_api_status_code,
+    _is_transient_slack_api_error,
+)
 
 # Slack only allows Markdown messages up to 4k chars
 MAXIMUM_MESSAGE_SIZE = 4000
 
 
+def _give_up_slack_api_retry(ex: Exception) -> bool:
+    if not isinstance(ex, SlackApiError):
+        return False
+
+    status_code = _get_slack_api_status_code(ex)
+    # WebClient's RateLimitErrorRetryHandler owns the operator-configured 429
+    # retry budget. Retrying an exhausted 429 here would multiply that budget
+    # by this helper's max_tries.
+    error_code = _get_slack_api_error_code(ex)
+    if status_code == 429:
+        return True
+    if _is_transient_slack_api_error(ex, error_code):
+        return False
+    if status_code is not None and 400 <= status_code < 500:
+        return True
+    return bool(error_code)
+
+
+@backoff.on_exception(
+    backoff.expo,
+    (SlackApiError, SlackClientNotConnectedError),
+    factor=10,
+    base=2,
+    max_tries=5,
+    giveup=_give_up_slack_api_retry,
+)
+def _call_slack_api(method: Callable[..., object], **kwargs: object) -> None:
+    method(**kwargs)
+
+
+def _send_to_slack_channels(
+    channels: list[str],
+    send_to_channel: Callable[[str], None],
+) -> None:
+    """Send to every channel and raise one channel-aware aggregate error."""
+    failures: list[tuple[str, SlackApiError | SlackClientNotConnectedError]] = []
+    for channel in channels:
+        try:
+            send_to_channel(channel)
+        except (SlackApiError, SlackClientNotConnectedError) as ex:
+            failures.append((channel, ex))
+
+    if not failures:
+        return
+
+    details = "; ".join(f"{channel}: {error}" for channel, error in failures)
+    message = f"Slack delivery failed for the following channels: {details}"
+    if any(
+        isinstance(error, SlackClientNotConnectedError)
+        or (
+            isinstance(error, SlackApiError)
+            and _is_transient_slack_api_error(
+                error,
+                _get_slack_api_error_code(error),
+            )
+        )
+        for _, error in failures
+    ):
+        raise NotificationError(message) from failures[0][1]
+    raise NotificationUnprocessableException(message) from failures[0][1]
+
+
 # pylint: disable=too-few-public-methods
 class SlackMixin:
     def _message_template(

superset/reports/notifications/slackv2.py

@@ -15,18 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Callable, Sequence
+from collections.abc import Sequence
 from io import IOBase
 from typing import List, Union
 
-import backoff
 from flask import g
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
@@ -40,63 +37,21 @@ from superset.reports.notifications.exceptions import (
     NotificationParamException,
     NotificationUnprocessableException,
 )
-from superset.reports.notifications.slack_mixin import SlackMixin
+from superset.reports.notifications.slack_mixin import (
+    _call_slack_api,
+    _send_to_slack_channels,
+    SlackMixin,
+)
 from superset.utils import json
-from superset.utils.core import recipients_string_to_list
 from superset.utils.decorators import statsd_gauge
-from superset.utils.slack import get_slack_client
+from superset.utils.slack import (
+    get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+)
 
 logger = logging.getLogger(__name__)
 
-_TRANSIENT_SLACK_API_ERROR_CODES = frozenset(
-    {
-        "fatal_error",
-        "internal_error",
-        "ratelimited",
-        "request_timeout",
-        "rollup_error",
-        "service_unavailable",
-        "timeout",
-    }
-)
-
-
-def _get_slack_api_error_code(ex: SlackApiError) -> str:
-    response = getattr(ex, "response", None)
-    data = getattr(response, "data", None)
-    if not isinstance(data, dict):
-        data = response if isinstance(response, dict) else {}
-    return str(data.get("error") or "")
-
-
-def _get_slack_api_status_code(ex: SlackApiError) -> int | None:
-    response = getattr(ex, "response", None)
-    return getattr(response, "status_code", None)
-
-
-def _give_up_slack_api_retry(ex: Exception) -> bool:
-    if not isinstance(ex, SlackApiError):
-        return False
-
-    status_code = _get_slack_api_status_code(ex)
-    if status_code == 429 or (status_code is not None and 500 <= status_code < 600):
-        return False
-
-    error_code = _get_slack_api_error_code(ex)
-    return bool(error_code and error_code not in _TRANSIENT_SLACK_API_ERROR_CODES)
-
-
-@backoff.on_exception(
-    backoff.expo,
-    (SlackApiError, SlackClientNotConnectedError),
-    factor=10,
-    base=2,
-    max_tries=5,
-    giveup=_give_up_slack_api_retry,
-)
-def _call_slack_api(method: Callable[..., object], **kwargs: object) -> None:
-    method(**kwargs)
-
 
 class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
     """
@@ -111,9 +66,15 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
         :returns: A list of channel ids: "EID676L"
         :raises NotificationParamException or SlackApiError: If the recipient is not found
         """  # noqa: E501
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        try:
+            recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE) from ex
 
-        return recipients_string_to_list(recipient_str)
+        if not isinstance(recipient_str, str):
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+
+        return parse_slack_recipient_targets(recipient_str)
 
     def _get_inline_files(
         self,
@@ -139,13 +100,12 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
             channels = self._get_channels()
 
             if not channels:
-                raise NotificationParamException("No recipients saved in the report")
+                raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
 
             file_type, files = self._get_inline_files()
             file_name = f"{title}.{file_type}"
 
-            # files_upload returns SlackResponse as we run it in sync mode.
-            for channel in channels:
+            def send_to_channel(channel: str) -> None:
                 if len(files) > 0:
                     for file in files:
                         _call_slack_api(
@@ -159,6 +119,9 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
                 else:
                     _call_slack_api(client.chat_postMessage, channel=channel, text=body)
 
+            # files_upload returns SlackResponse as we run it in sync mode.
+            _send_to_slack_channels(channels, send_to_channel)
+
             logger.info(
                 "Report sent to slack",
                 extra={
@@ -175,8 +138,6 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
             # this is the base class for all slack client errors
             # keep it last so that it doesn't interfere with @backoff

superset/utils/slack.py

@@ -23,7 +23,11 @@ from typing import Any, Callable, Optional
 
 from flask import current_app as app
 from slack_sdk import WebClient
-from slack_sdk.errors import SlackApiError, SlackClientError as SlackSDKClientError
+from slack_sdk.errors import (
+    SlackApiError,
+    SlackClientError as SlackSDKClientError,
+    SlackClientNotConnectedError,
+)
 from slack_sdk.http_retry.builtin_handlers import RateLimitErrorRetryHandler
 
 from superset import feature_flag_manager
@@ -76,6 +80,62 @@ class SlackClientError(Exception):
     pass
 
 
+class SlackV2ProbeError(SupersetException):
+    """Slack v2 availability probe failed."""
+
+
+class SlackV2ProbeClientError(SlackV2ProbeError):
+    """Slack v2 availability failed due to permanent token or client setup."""
+
+    status = 422
+
+
+_TRANSIENT_SLACK_API_ERROR_CODES = frozenset(
+    {
+        "fatal_error",
+        "internal_error",
+        "ratelimited",
+        "request_timeout",
+        "rollup_error",
+        "service_unavailable",
+        "timeout",
+    }
+)
+
+
+NO_SLACK_RECIPIENTS_MESSAGE = "No recipients saved in the report"
+
+
+def parse_slack_recipient_targets(target: str) -> list[str]:
+    """Parse Slack targets, removing duplicates while preserving their order."""
+    return list(dict.fromkeys(recipients_string_to_list(target)))
+
+
+def _get_slack_api_error_data(ex: SlackApiError) -> dict[str, Any]:
+    response = getattr(ex, "response", None)
+    data = getattr(response, "data", None)
+    if not isinstance(data, dict):
+        data = response if isinstance(response, dict) else {}
+    return data
+
+
+def _get_slack_api_error_code(ex: SlackApiError) -> str:
+    return str(_get_slack_api_error_data(ex).get("error") or "")
+
+
+def _get_slack_api_status_code(ex: SlackApiError) -> int | None:
+    return getattr(getattr(ex, "response", None), "status_code", None)
+
+
+def _is_transient_slack_api_error(ex: SlackApiError, error_code: str) -> bool:
+    status_code = _get_slack_api_status_code(ex)
+    return bool(
+        status_code in {408, 429}
+        or (status_code is not None and 500 <= status_code < 600)
+        or error_code in _TRANSIENT_SLACK_API_ERROR_CODES
+    )
+
+
 def get_slack_client() -> WebClient:
     token: str = app.config["SLACK_API_TOKEN"]
     if callable(token):
@@ -261,7 +321,7 @@ _SCOPE_MISSING_ERROR_CODES = frozenset(
 )
 
 
-def should_use_v2_api() -> bool:
+def should_use_v2_api(*, raise_on_error: bool = False) -> bool:
     if not feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):
         _emit_v1_flag_off_deprecation()
         return False
@@ -282,17 +342,11 @@ def should_use_v2_api() -> bool:
         # Only the scope-missing branch is a v1-deprecation signal; other
         # SlackApiError codes (invalid_auth, ratelimited, server errors, etc.)
         # are unrelated probe failures and should not be reported as a missing
-        # scope. We still fall back to v1 in both cases so a transient probe
-        # failure doesn't break sends — operators get an actionable log either
-        # way.
-        # `response` is normally a SlackResponse whose payload lives in `.data`,
-        # but the SDK (and our tests) can also hand back a plain dict. Read the
-        # error code in either shape so the scope-missing branch isn't missed.
-        response = getattr(ex, "response", None)
-        data = getattr(response, "data", None)
-        if not isinstance(data, dict):
-            data = response if isinstance(response, dict) else {}
-        error_code = data.get("error", "")
+        # scope. Scope errors continue through v1 for compatibility. Other
+        # failures also fall back for text-only reports, while file-bearing
+        # reports request an exception so monitoring retains the system/client
+        # classification.
+        error_code = _get_slack_api_error_code(ex)
         if error_code in _SCOPE_MISSING_ERROR_CODES:
             # The DeprecationWarning fires once per process, but the actionable
             # log line fires every send so operators see it in their report logs.
@@ -305,11 +359,19 @@ def should_use_v2_api() -> bool:
             )
         else:
             logger.warning(
-                "Slack v2 probe failed with error %r; falling back to the "
-                "deprecated v1 API for this send. Investigate the underlying "
-                "Slack API error — this is not a missing-scope problem.",
+                "Slack v2 probe failed with error %r. Investigate the underlying "
+                "Slack API error; this is not a missing-scope problem.",
                 error_code or str(ex),
             )
+            if raise_on_error:
+                error_class = (
+                    SlackV2ProbeError
+                    if _is_transient_slack_api_error(ex, error_code)
+                    else SlackV2ProbeClientError
+                )
+                raise error_class(
+                    f"Slack v2 availability probe failed: {error_code or str(ex)}"
+                ) from ex
         return False
     except SlackSDKClientError as ex:
         # Non-API SDK failures (e.g. SlackClientNotConnectedError,
@@ -318,14 +380,22 @@ def should_use_v2_api() -> bool:
         # raw. The caller runs this probe *before* the mapped Slack send `try`,
         # so an un-caught probe error aborts the entire recipient loop instead
         # of failing a single recipient. Treat any probe connection/transport
-        # failure as "v2 unavailable" and fall back to the deprecated v1 API,
-        # matching the SlackApiError behavior above.
+        # failure as "v2 unavailable" for text reports. File-bearing reports
+        # request an exception so the command can retain error classification.
         logger.warning(
-            "Slack v2 probe failed to connect (%s: %s); falling back to the "
-            "deprecated v1 API for this send.",
+            "Slack v2 probe failed (%s: %s).",
             type(ex).__name__,
             ex,
         )
+        if raise_on_error:
+            error_class = (
+                SlackV2ProbeError
+                if isinstance(ex, SlackClientNotConnectedError)
+                else SlackV2ProbeClientError
+            )
+            raise error_class(
+                f"Slack v2 availability probe failed: {type(ex).__name__}: {ex}"
+            ) from ex
         return False
 
 

tests/integration_tests/reports/commands_tests.py

@@ -21,6 +21,7 @@ from typing import Optional
 from unittest.mock import call, Mock, patch
 from uuid import uuid4
 
+import pandas as pd
 import pytest
 from flask.ctx import AppContext
 from flask_appbuilder.security.sqla.models import User
@@ -73,6 +74,7 @@ from superset.models.slice import Slice
 from superset.reports.models import (
     ReportDataFormat,
     ReportExecutionLog,
+    ReportRecipients,
     ReportRecipientType,
     ReportSchedule,
     ReportScheduleType,
@@ -1541,15 +1543,13 @@ def test_slack_chart_report_schedule_converts_to_v2_channel_with_hash(
 @patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-def test_slack_chart_report_schedule_fails_to_converts_to_v2(
+def test_slack_chart_report_schedule_failed_v2_conversion_rejects_v1_file_upload(
     screenshot_mock,
     slack_client_mock,
     slack_should_use_v2_api_mock,
     get_channels_with_search_mock,
 ):
-    """
-    ExecuteReport Command: Test converting a Slack report to v2 fails.
-    """
+    """A failed Slack v2 conversion rejects unsupported Slack v1 file uploads."""
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
     channel_id = "slack_channel_id"
@@ -1566,25 +1566,30 @@ def test_slack_chart_report_schedule_fails_to_converts_to_v2(
         },
     ]
 
-    with pytest.raises(ReportScheduleSystemErrorsException):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, report_schedule.id, datetime.utcnow()
-        ).run()
+    try:
+        with pytest.raises(ReportScheduleClientErrorsException):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, report_schedule.id, datetime.utcnow()
+            ).run()
 
-    # Assert failuer with proper log
-    expected_message = (
-        "Failed to update slack recipients to v2: "
-        "Could not find the following channels: my_member_ID"
-    )
-    assert_log(ReportState.ERROR, error_message=expected_message)
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2. Slack v2 upgrade failed: Failed to update "
+            "slack recipients to v2: Could not find the following channels: "
+            "my_member_ID"
+        )
+        assert_log(ReportState.ERROR, error_message=expected_message)
 
-    # Assert that previous configuration was kept for manual correction
-    assert report_schedule.recipients[0].recipient_config_json == json.dumps(
-        {"target": "#slack_channel,my_member_ID"}
-    )
-    assert report_schedule.recipients[0].type == ReportRecipientType.SLACK
-
-    cleanup_report_schedule(report_schedule)
+        # Keep the previous configuration for manual correction.
+        assert report_schedule.recipients[0].recipient_config_json == json.dumps(
+            {"target": "#slack_channel,my_member_ID"}
+        )
+        assert report_schedule.recipients[0].type == ReportRecipientType.SLACK
+        slack_client_mock.assert_not_called()
+    finally:
+        cleanup_report_schedule(report_schedule)
 
 
 @pytest.mark.usefixtures("create_report_slack_chartv2")
@@ -1695,9 +1700,7 @@ def test_slack_chart_report_schedule_with_csv(
     slack_should_use_v2_api_mock,
     create_report_slack_chart_with_csv,
 ):
-    """
-    ExecuteReport Command: Test chart slack report V1 schedule with CSV
-    """
+    """A v1 CSV report fails before calling Slack's retired upload API."""
     # setup csv mock
     response = Mock()
     mock_open.return_value = response
@@ -1705,28 +1708,20 @@ def test_slack_chart_report_schedule_with_csv(
     mock_urlopen.return_value.getcode.return_value = 200
     response.read.return_value = CSV_FILE
 
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_csv
-    )
-
-    channel_name = notification_targets[0]
-
     with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
-        ).run()
+        with pytest.raises(ReportScheduleClientErrorsException):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
+            ).run()
 
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["channels"]
-            == channel_name
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2."
         )
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["file"]
-            == CSV_FILE
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
+        assert_log(ReportState.ERROR, error_message=expected_message)
+        slack_client_mock_class.assert_not_called()
 
 
 @pytest.mark.usefixtures(
@@ -1745,9 +1740,7 @@ def test_slack_chart_report_schedule_with_xlsx(
     slack_should_use_v2_api_mock: Mock,
     create_report_slack_chart_with_xlsx: ReportSchedule,
 ) -> None:
-    """
-    ExecuteReport Command: Test chart slack report V1 schedule with Excel
-    """
+    """A v1 XLSX report fails before calling Slack's retired upload API."""
     # setup xlsx mock
     response = Mock()
     mock_open.return_value = response
@@ -1755,28 +1748,20 @@ def test_slack_chart_report_schedule_with_xlsx(
     mock_urlopen.return_value.getcode.return_value = 200
     response.read.return_value = XLSX_FILE
 
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_xlsx
-    )
-
-    channel_name = notification_targets[0]
-
     with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_xlsx.id, datetime.utcnow()
-        ).run()
+        with pytest.raises(ReportScheduleClientErrorsException):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, create_report_slack_chart_with_xlsx.id, datetime.utcnow()
+            ).run()
 
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["channels"]
-            == channel_name
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2."
         )
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["file"]
-            == XLSX_FILE
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
+        assert_log(ReportState.ERROR, error_message=expected_message)
+        slack_client_mock_class.assert_not_called()
 
 
 @pytest.mark.usefixtures(
@@ -1846,6 +1831,153 @@ def test_slack_chart_report_schedule_with_text(
         assert_log(ReportState.SUCCESS)
 
 
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
+)
+@patch("superset.commands.report.execute.get_channels_with_search", return_value=[])
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_chart_dataframe")
+def test_slack_text_fallback_persists_success_for_multiple_recipient_rows(
+    dataframe_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
+    create_report_slack_chart_with_text,
+):
+    """Failed migration sends every text recipient and persists v1 success."""
+    dataframe_mock.return_value = pd.DataFrame({"value": [1]})
+    original_configs = [
+        json.dumps({"target": "private-a"}),
+        json.dumps({"target": "private-b"}),
+    ]
+    create_report_slack_chart_with_text.recipients[
+        0
+    ].recipient_config_json = original_configs[0]
+    create_report_slack_chart_with_text.recipients.append(
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=original_configs[1],
+        )
+    )
+    db.session.commit()
+    report_schedule_id = create_report_slack_chart_with_text.id
+
+    AsyncExecuteReportScheduleCommand(
+        TEST_ID,
+        report_schedule_id,
+        datetime.utcnow(),
+    ).run()
+
+    db.session.expire_all()
+    persisted_schedule = db.session.get(ReportSchedule, report_schedule_id)
+    assert persisted_schedule is not None
+    assert persisted_schedule.last_state == ReportState.SUCCESS
+    assert all(
+        recipient.type == ReportRecipientType.SLACK
+        for recipient in persisted_schedule.recipients
+    )
+    assert {
+        recipient.recipient_config_json for recipient in persisted_schedule.recipients
+    } == set(original_configs)
+    assert {
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    } == {"private-a", "private-b"}
+    assert slack_client_mock.return_value.chat_postMessage.call_count == 2
+    assert slack_should_use_v2_api_mock.call_count == 1
+    assert get_channels_with_search_mock.call_count == 2
+
+
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
+)
+@patch("superset.commands.report.execute.get_channels_with_search", return_value=[])
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_chart_dataframe")
+@patch("superset.reports.notifications.email.send_email_smtp")
+def test_slack_text_fallback_persists_later_recipient_retry_exhaustion(
+    email_mock,
+    dataframe_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
+    create_report_slack_chart_with_text,
+):
+    """A later failed recipient does not duplicate an earlier successful send."""
+    dataframe_mock.return_value = pd.DataFrame({"value": [1]})
+    original_configs = [
+        json.dumps({"target": "private-a"}),
+        json.dumps({"target": "private-b"}),
+    ]
+    create_report_slack_chart_with_text.recipients[
+        0
+    ].recipient_config_json = original_configs[0]
+    create_report_slack_chart_with_text.recipients.append(
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=original_configs[1],
+        )
+    )
+    db.session.commit()
+    report_schedule_id = create_report_slack_chart_with_text.id
+
+    successful_channels: list[str] = []
+
+    def chat_side_effect(channel, text):
+        if not successful_channels:
+            successful_channels.append(channel)
+            return {"ok": True}
+        if channel != successful_channels[0]:
+            raise SlackApiError(
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
+            )
+        return {"ok": True}
+
+    slack_client_mock.return_value.chat_postMessage.side_effect = chat_side_effect
+
+    with (
+        patch("time.sleep"),
+        pytest.raises(ReportScheduleSystemErrorsException),
+    ):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID,
+            report_schedule_id,
+            datetime.utcnow(),
+        ).run()
+
+    db.session.expire_all()
+    persisted_schedule = db.session.get(ReportSchedule, report_schedule_id)
+    assert persisted_schedule is not None
+    assert persisted_schedule.last_state == ReportState.ERROR
+    assert all(
+        recipient.type == ReportRecipientType.SLACK
+        for recipient in persisted_schedule.recipients
+    )
+    assert {
+        recipient.recipient_config_json for recipient in persisted_schedule.recipients
+    } == set(original_configs)
+    call_channels = [
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    ]
+    successful_channel = call_channels[0]
+    failed_channel = ({"private-a", "private-b"} - {successful_channel}).pop()
+    assert call_channels == [successful_channel, *(failed_channel for _ in range(5))]
+    log_states = {
+        log.state
+        for log in db.session.query(ReportExecutionLog)
+        .filter(ReportExecutionLog.report_schedule_id == report_schedule_id)
+        .all()
+    }
+    assert {ReportState.WORKING, ReportState.ERROR} <= log_states
+    assert slack_should_use_v2_api_mock.call_count == 1
+    assert get_channels_with_search_mock.call_count == 2
+    email_mock.assert_called_once()
+
+
 @pytest.mark.usefixtures("create_report_slack_chart")
 def test_report_schedule_not_found(create_report_slack_chart):
     """
@@ -1922,18 +2054,18 @@ def test_report_schedule_success_grace(create_alert_slack_chart_success):
 
 
 @pytest.mark.usefixtures("create_alert_slack_chart_grace")
-@patch("superset.utils.slack.WebClient.files_upload")
+@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-@patch("superset.reports.notifications.slack.get_slack_client")
 def test_report_schedule_success_grace_end(
-    slack_client_mock_class,
     screenshot_mock,
-    file_upload_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
     create_alert_slack_chart_grace,
 ):
-    """
-    ExecuteReport Command: Test report schedule on grace to noop
-    """
+    """A Slack alert leaving grace upgrades to v2 and sends successfully."""
 
     screenshot_mock.return_value = SCREENSHOT_FILE
 
@@ -1949,9 +2081,14 @@ def test_report_schedule_success_grace_end(
     channel_name = notification_targets[0]
     channel_id = "channel_id"
 
-    slack_client_mock_class.return_value.conversations_list.return_value = {
-        "channels": [{"id": channel_id, "name": channel_name}]
-    }
+    get_channels_with_search_mock.return_value = [
+        {
+            "id": channel_id,
+            "name": channel_name,
+            "is_member": True,
+            "is_private": True,
+        }
+    ]
 
     with freeze_time(current_time):
         AsyncExecuteReportScheduleCommand(
@@ -1960,6 +2097,13 @@ def test_report_schedule_success_grace_end(
 
     db.session.commit()
     assert create_alert_slack_chart_grace.last_state == ReportState.SUCCESS
+    recipient = create_alert_slack_chart_grace.recipients[0]
+    assert recipient.type == ReportRecipientType.SLACKV2
+    assert json.loads(recipient.recipient_config_json) == {"target": channel_id}
+    slack_should_use_v2_api_mock.assert_called_once_with(raise_on_error=True)
+    upload_call = slack_client_mock.return_value.files_upload_v2.call_args
+    assert upload_call.kwargs["channel"] == channel_id
+    assert upload_call.kwargs["file"] == SCREENSHOT_FILE
 
 
 @pytest.mark.usefixtures("create_alert_email_chart")
@@ -2564,9 +2708,10 @@ def test_prune_log_soft_time_out(bulk_delete_logs, create_report_email_dashboard
 def test__send_with_client_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = NotificationParamException()
     with pytest.raises(ReportScheduleClientErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     logger_mock.warning.assert_called_with(
@@ -2579,13 +2724,14 @@ def test__send_with_client_errors(notification_mock, logger_mock):
 def test__send_with_multiple_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com", "test2@bar.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = [
         NotificationParamException(),
         NotificationError(),
     ]
     # it raises the error with a 500 status if present
     with pytest.raises(ReportScheduleSystemErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     # it logs both errors as warnings
@@ -2606,9 +2752,10 @@ def test__send_with_multiple_errors(notification_mock, logger_mock):
 def test__send_with_server_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = NotificationError()
     with pytest.raises(ReportScheduleSystemErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     # it logs the error

tests/unit_tests/commands/report/execute_test.py

@@ -18,6 +18,7 @@
 import json  # noqa: TID251
 from datetime import datetime, timedelta
 from typing import Any
+from unittest import mock
 from unittest.mock import MagicMock, Mock, patch
 from urllib.error import URLError
 from uuid import UUID, uuid4
@@ -30,6 +31,7 @@ from superset.app import SupersetApp
 from superset.commands.exceptions import UpdateFailedError
 from superset.commands.report.exceptions import (
     ReportScheduleAlertGracePeriodError,
+    ReportScheduleClientErrorsException,
     ReportScheduleCsvFailedError,
     ReportScheduleExecuteUnexpectedError,
     ReportScheduleExecutorNotFoundError,
@@ -37,6 +39,7 @@ from superset.commands.report.exceptions import (
     ReportScheduleScreenshotFailedError,
     ReportScheduleScreenshotTimeout,
     ReportScheduleStateNotFoundError,
+    ReportScheduleSystemErrorsException,
     ReportScheduleUnexpectedError,
     ReportScheduleWorkingTimeoutError,
     ReportScheduleXlsxFailedError,
@@ -52,6 +55,7 @@ from superset.commands.report.execute import (
 from superset.common.chart_data import ChartDataResultFormat, ChartDataResultType
 from superset.daos.report import REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER
 from superset.dashboards.permalink.types import DashboardPermalinkState
+from superset.exceptions import SupersetException
 from superset.reports.models import (
     ReportDataFormat,
     ReportRecipients,
@@ -61,9 +65,12 @@ from superset.reports.models import (
     ReportSourceFormat,
     ReportState,
 )
+from superset.reports.notifications.base import NotificationContent
+from superset.reports.notifications.exceptions import NotificationParamException
 from superset.subjects.types import SubjectType
 from superset.utils.core import HeaderDataType
 from superset.utils.screenshots import ChartScreenshot
+from superset.utils.slack import SlackV2ProbeClientError, SlackV2ProbeError
 from tests.integration_tests.conftest import with_feature_flags
 
 
@@ -1815,16 +1822,90 @@ def test_update_recipient_to_slack_v2_missing_channels(mocker: MockerFixture):
     mock_cmmd: BaseReportState = BaseReportState(
         mock_report_schedule, "January 1, 2021", "execution_id_example"
     )
-    with pytest.raises(UpdateFailedError):
+    with pytest.raises(NotificationParamException):
         mock_cmmd.update_report_schedule_slack_v2()
 
 
+def test_update_recipient_to_slack_v2_refreshes_stale_channel_cache(
+    mocker: MockerFixture,
+) -> None:
+    """A cache miss gets one fresh lookup before the upgrade falls back."""
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        side_effect=[
+            [],
+            [
+                {"id": "C2", "name": "second", "is_private": True},
+                {"id": "C1", "name": "channel-1", "is_private": False},
+            ],
+        ],
+    )
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps({"target": "Channel-1,C2"}),
+    )
+    state = BaseReportState(
+        ReportSchedule(recipients=[recipient]),
+        "January 1, 2021",
+        "execution_id_example",
+    )
+
+    state.update_report_schedule_slack_v2()
+
+    assert channel_search.call_args_list == [
+        mock.call(
+            search_string="Channel-1,C2",
+            types=mocker.ANY,
+            exact_match=True,
+        ),
+        mock.call(
+            search_string="Channel-1,C2",
+            types=mocker.ANY,
+            exact_match=True,
+            force=True,
+        ),
+    ]
+    assert recipient.type == ReportRecipientType.SLACKV2
+    assert recipient.recipient_config_json == '{"target": "C1,C2"}'
+
+
+def test_update_recipient_to_slack_v2_reports_only_unresolved_channels(
+    mocker: MockerFixture,
+) -> None:
+    """Diagnostics use the same case-insensitive name-or-id match as resolution."""
+    mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        return_value=[
+            {"id": "C123", "name": "private-channel", "is_private": True},
+            {"id": "C999", "name": "other", "is_private": False},
+        ],
+    )
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps(
+            {"target": "PRIVATE-CHANNEL,c999,missing-channel"}
+        ),
+    )
+    state = BaseReportState(
+        ReportSchedule(recipients=[recipient]),
+        "January 1, 2021",
+        "execution_id_example",
+    )
+
+    with pytest.raises(NotificationParamException) as exc_info:
+        state.update_report_schedule_slack_v2()
+
+    assert "missing-channel" in str(exc_info.value)
+    assert "PRIVATE-CHANNEL" not in str(exc_info.value)
+    assert "c999" not in str(exc_info.value)
+
+
 def test_update_recipient_to_slack_v2_multiple_recipients(
     mocker: MockerFixture,
 ) -> None:
     """All Slack recipients are upgraded atomically when every channel resolves."""
 
-    def fake_get_channels(search_string, types, exact_match):
+    def fake_get_channels(search_string, types, exact_match, force=False):
         return {
             "channel-1": [{"id": "C1", "name": "channel-1", "is_private": False}],
             "channel-2": [{"id": "C2", "name": "channel-2", "is_private": False}],
@@ -1874,10 +1955,10 @@ def test_update_recipient_to_slack_v2_partial_failure_is_atomic(
     persist as a half-upgraded schedule.
     """
 
-    def fake_get_channels(search_string, types, exact_match):
+    def fake_get_channels(search_string, types, exact_match, force=False):
         if search_string == "channel-1":
             return [{"id": "C1", "name": "channel-1", "is_private": False}]
-        # "missing-channel" resolves to nothing -> triggers UpdateFailedError
+        # "missing-channel" resolves to nothing -> invalid recipient parameters
         return []
 
     mocker.patch(
@@ -1902,7 +1983,7 @@ def test_update_recipient_to_slack_v2_partial_failure_is_atomic(
     mock_cmmd: BaseReportState = BaseReportState(
         mock_report_schedule, "January 1, 2021", "execution_id_example"
     )
-    with pytest.raises(UpdateFailedError):
+    with pytest.raises(NotificationParamException):
         mock_cmmd.update_report_schedule_slack_v2()
 
     recipients = mock_cmmd._report_schedule.recipients
@@ -1971,6 +2052,608 @@ def test_update_recipient_to_slack_v2_no_slack_recipients_is_noop(
     mock_search.assert_not_called()
 
 
+@pytest.mark.parametrize(
+    "recipient_config_json",
+    [
+        "{not-json",
+        json.dumps({"target": ["private-channel"]}),
+    ],
+    ids=["malformed-json", "non-string-target"],
+)
+def test_update_recipient_to_slack_v2_rejects_invalid_config_without_traceback(
+    mocker: MockerFixture,
+    recipient_config_json: str,
+) -> None:
+    """Operator-fixable recipient configuration logs no exception traceback."""
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=recipient_config_json,
+    )
+    state = BaseReportState(
+        ReportSchedule(recipients=[recipient]),
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    logger = mocker.patch("superset.commands.report.execute.logger")
+
+    with pytest.raises(NotificationParamException):
+        state.update_report_schedule_slack_v2()
+
+    logger.warning.assert_called_once()
+    logger.exception.assert_not_called()
+    assert recipient.type == ReportRecipientType.SLACK
+    assert recipient.recipient_config_json == recipient_config_json
+
+
+def test_update_recipient_to_slack_v2_deduplicates_channels(
+    mocker: MockerFixture,
+) -> None:
+    """Repeated channel names resolve once and persist one channel id."""
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        return_value=[
+            {
+                "id": "C1",
+                "name": "private-channel",
+                "is_private": True,
+            }
+        ],
+    )
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps(
+            {"target": "private-channel, private-channel"}
+        ),
+    )
+    state = BaseReportState(
+        ReportSchedule(recipients=[recipient]),
+        "January 1, 2021",
+        "execution_id_example",
+    )
+
+    state.update_report_schedule_slack_v2()
+
+    channel_search.assert_called_once_with(
+        search_string="private-channel",
+        types=mocker.ANY,
+        exact_match=True,
+    )
+    assert recipient.type == ReportRecipientType.SLACKV2
+    assert recipient.recipient_config_json == '{"target": "C1"}'
+
+
+def test_send_falls_back_to_slack_v1_when_private_channels_upgrade_fails(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """A failed migration must fall back for every Slack recipient row."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    original_configs = [
+        json.dumps({"target": "private-a"}),
+        json.dumps({"target": "private-b"}),
+    ]
+    recipients = [
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=config,
+        )
+        for config in original_configs
+    ]
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=recipients,
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "TEXT",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-a", "private-b"],
+            "execution_id": "execution_id_example",
+        },
+        description="Text-only report",
+        url="https://superset.example/report",
+    )
+    v2_probe = mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        return_value=[],
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    stats_logger = mocker.Mock()
+    mocker.patch.dict(app.config, {"STATS_LOGGER": stats_logger})
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    report_state._send(notification_content, report_schedule.recipients)
+
+    assert v2_probe.call_count == 1
+    assert channel_search.call_count == 2
+    stats_logger.incr.assert_called_once_with("reports.slack.v1_fallback")
+    assert slack_client.return_value.chat_postMessage.call_count == 2
+    assert [
+        call.kwargs["channel"]
+        for call in slack_client.return_value.chat_postMessage.call_args_list
+    ] == ["private-a", "private-b"]
+    assert [recipient.type for recipient in recipients] == [
+        ReportRecipientType.SLACK,
+        ReportRecipientType.SLACK,
+    ]
+    assert [recipient.recipient_config_json for recipient in recipients] == (
+        original_configs
+    )
+
+
+def test_send_records_system_upgrade_failure_when_text_fallback_succeeds(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """Delivery continuity retains an observable system-failure signal."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    stats_logger = mocker.Mock()
+    mocker.patch.dict(app.config, {"STATS_LOGGER": stats_logger})
+    recipients = [
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=json.dumps({"target": channel}),
+        )
+        for channel in ("private-a", "private-b")
+    ]
+    report_schedule = ReportSchedule(
+        id=42,
+        name="Private channel report",
+        recipients=recipients,
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "TEXT",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-channel"],
+            "execution_id": "execution_id_example",
+        },
+        description="Text-only report",
+        url="https://superset.example/report",
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        side_effect=SupersetException("Slack channel listing unavailable"),
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    logger = mocker.patch("superset.commands.report.execute.logger")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    report_state._send(notification_content, recipients)
+
+    assert [
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client.return_value.chat_postMessage.call_args_list
+    ] == ["private-a", "private-b"]
+    assert stats_logger.incr.call_args_list == [
+        mock.call("reports.slack.v1_fallback"),
+        mock.call("reports.slack.v1_fallback.system_error"),
+    ]
+    logger.error.assert_called_once()
+    assert logger.error.call_args.kwargs["extra"] == {
+        "execution_id": "execution_id_example",
+        "report_schedule_id": 42,
+    }
+
+
+def test_failed_slack_upgrade_fallback_does_not_affect_other_recipient_types(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """Only the legacy Slack recipient uses fallback in a mixed schedule."""
+    from superset.reports.notifications.slack import SlackNotification
+
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipients = [
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=json.dumps({"target": "private-channel"}),
+        ),
+        ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json=json.dumps({"target": "C123"}),
+        ),
+        ReportRecipients(
+            type=ReportRecipientType.EMAIL,
+            recipient_config_json=json.dumps({"target": "user@example.com"}),
+        ),
+    ]
+    report_schedule = ReportSchedule(
+        name="Mixed recipient report",
+        recipients=recipients,
+    )
+    state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    content = NotificationContent(
+        name="Mixed recipient report",
+        header_data={
+            "notification_format": "TEXT",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-channel"],
+            "execution_id": "execution_id_example",
+        },
+        description="Text-only report",
+        url="https://superset.example/report",
+    )
+    legacy_notification = SlackNotification(recipients[0], content)
+    v2_notification = mocker.Mock()
+    email_notification = mocker.Mock()
+    mocker.patch(
+        "superset.commands.report.execute.create_notification",
+        side_effect=[legacy_notification, v2_notification, email_notification],
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        return_value=[],
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    state._send(content, recipients)
+
+    assert channel_search.call_count == 2
+    slack_client.return_value.chat_postMessage.assert_called_once_with(
+        channel="private-channel",
+        text=mocker.ANY,
+    )
+    v2_notification.send.assert_called_once_with()
+    email_notification.send.assert_called_once_with()
+
+
+def test_send_malformed_slack_recipient_does_not_suppress_later_recipient(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """A malformed recipient is aggregated while later recipients still send."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipients = [
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=json.dumps({}),
+        ),
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=json.dumps({"target": "private-b"}),
+        ),
+    ]
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=recipients,
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "TEXT",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-b"],
+            "execution_id": "execution_id_example",
+        },
+        description="Text-only report",
+        url="https://superset.example/report",
+    )
+    v2_probe = mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    with pytest.raises(ReportScheduleClientErrorsException) as exc_info:
+        report_state._send(notification_content, recipients)
+
+    assert exc_info.value.errors[0].message == "No recipients saved in the report"
+    slack_client.return_value.chat_postMessage.assert_called_once_with(
+        channel="private-b",
+        text=mocker.ANY,
+    )
+    assert v2_probe.call_count == 1
+    channel_search.assert_not_called()
+    assert [recipient.type for recipient in recipients] == [
+        ReportRecipientType.SLACK,
+        ReportRecipientType.SLACK,
+    ]
+
+
+@pytest.mark.parametrize(
+    "attachment",
+    [
+        {"screenshots": [b"screenshot"]},
+        {"xlsx": b"xlsx_content"},
+    ],
+    ids=["screenshot", "xlsx"],
+)
+def test_send_preserves_transient_upgrade_failure_for_file_reports(
+    app: SupersetApp,
+    mocker: MockerFixture,
+    attachment: dict[str, Any],
+) -> None:
+    """A transient v2 migration failure remains a system error for files."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps({"target": "private-channel"}),
+    )
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=[recipient],
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-channel"],
+            "execution_id": "execution_id_example",
+        },
+        description="File-bearing report",
+        url="https://superset.example/report",
+        **attachment,
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        side_effect=SupersetException("Slack channel listing unavailable"),
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    with pytest.raises(ReportScheduleSystemErrorsException) as exc_info:
+        report_state._send(notification_content, [recipient])
+
+    error_message = exc_info.value.errors[0].message
+    assert "Slack v1 file uploads are no longer supported" in error_message
+    assert "channels:read" in error_message
+    assert "groups:read" in error_message
+    assert "Slack channel listing unavailable" in error_message
+    slack_client.assert_not_called()
+    assert recipient.type == ReportRecipientType.SLACK
+    assert recipient.recipient_config_json == '{"target": "private-channel"}'
+
+
+@pytest.mark.parametrize(
+    "probe_error,expected_exception",
+    [
+        (
+            SlackV2ProbeError(
+                "Slack v2 availability probe failed: service_unavailable"
+            ),
+            ReportScheduleSystemErrorsException,
+        ),
+        (
+            SlackV2ProbeClientError("Slack v2 availability probe failed: invalid_auth"),
+            ReportScheduleClientErrorsException,
+        ),
+    ],
+    ids=["system", "client"],
+)
+def test_send_classifies_probe_failure_for_file_reports(
+    app: SupersetApp,
+    mocker: MockerFixture,
+    probe_error: SlackV2ProbeError,
+    expected_exception: type[Exception],
+) -> None:
+    """Slack capability probe failures retain system/client classification."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps({"target": "private-channel"}),
+    )
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=[recipient],
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-channel"],
+            "execution_id": "execution_id_example",
+        },
+        screenshots=[b"screenshot"],
+        description="File-bearing report",
+        url="https://superset.example/report",
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        side_effect=probe_error,
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    with pytest.raises(expected_exception) as exc_info:
+        report_state._send(notification_content, [recipient])
+
+    assert str(probe_error) in exc_info.value.errors[0].message
+    slack_client.assert_not_called()
+
+
+def test_send_classifies_malformed_file_recipient_as_client_error(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """Malformed file recipients retain actionable client classification."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps({}),
+    )
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=[recipient],
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": [],
+            "execution_id": "execution_id_example",
+        },
+        screenshots=[b"screenshot"],
+        description="File-bearing report",
+        url="https://superset.example/report",
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    channel_search = mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    with pytest.raises(ReportScheduleClientErrorsException) as exc_info:
+        report_state._send(notification_content, [recipient])
+
+    assert "No recipients saved in the report" in exc_info.value.errors[0].message
+    slack_client.assert_not_called()
+    channel_search.assert_not_called()
+    assert recipient.type == ReportRecipientType.SLACK
+    assert recipient.recipient_config_json == "{}"
+
+
+def test_send_does_not_fall_back_to_slack_v1_for_file_uploads(
+    app: SupersetApp,
+    mocker: MockerFixture,
+) -> None:
+    """A failed v2 migration must not retry a retired v1 file upload."""
+    app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"] = False
+    recipient = ReportRecipients(
+        type=ReportRecipientType.SLACK,
+        recipient_config_json=json.dumps({"target": "private-channel"}),
+    )
+    report_schedule = ReportSchedule(
+        name="Private channel report",
+        recipients=[recipient],
+    )
+    report_state = BaseReportState(
+        report_schedule,
+        "January 1, 2021",
+        "execution_id_example",
+    )
+    notification_content = NotificationContent(
+        name="Private channel report",
+        header_data={
+            "notification_format": "PNG",
+            "notification_type": "Report",
+            "editors": [],
+            "notification_source": None,
+            "chart_id": None,
+            "dashboard_id": None,
+            "slack_channels": ["private-channel"],
+            "execution_id": "execution_id_example",
+        },
+        screenshots=[b"screenshot"],
+        description="File-bearing report",
+        url="https://superset.example/report",
+    )
+    mocker.patch(
+        "superset.reports.notifications.slack.should_use_v2_api",
+        return_value=True,
+    )
+    mocker.patch(
+        "superset.commands.report.execute.get_channels_with_search",
+        return_value=[],
+    )
+    slack_client = mocker.patch("superset.reports.notifications.slack.get_slack_client")
+    mocker.patch("superset.reports.notifications.slack.g", logs_context={})
+
+    with pytest.raises(ReportScheduleClientErrorsException) as exc_info:
+        report_state._send(notification_content, report_schedule.recipients)
+
+    error_message = str(exc_info.value.errors[0].message)
+    assert "Slack v1 file uploads are no longer supported" in error_message
+    assert "`channels:read` and `groups:read`" in error_message
+    assert "Could not find the following channels: private-channel" in error_message
+    slack_client.return_value.files_upload.assert_not_called()
+    slack_client.return_value.chat_postMessage.assert_not_called()
+    assert recipient.type == ReportRecipientType.SLACK
+    assert recipient.recipient_config_json == '{"target": "private-channel"}'
+
+
 # ---------------------------------------------------------------------------
 # Tier 1: _update_query_context + create_log
 # ---------------------------------------------------------------------------

tests/unit_tests/reports/notifications/slack_tests.py

@@ -16,8 +16,8 @@
 # under the License.
 
 import uuid
-from typing import Any
-from unittest.mock import call, MagicMock, patch
+from typing import Any, TYPE_CHECKING
+from unittest.mock import ANY, call, MagicMock, patch
 
 import pandas as pd
 import pytest
@@ -35,15 +35,21 @@ from slack_sdk.web.slack_response import SlackResponse
 
 from superset.reports.notifications.exceptions import (
     NotificationAuthorizationException,
+    NotificationError,
     NotificationMalformedException,
     NotificationParamException,
     NotificationUnprocessableException,
 )
-from superset.reports.notifications.slackv2 import (
+from superset.reports.notifications.slack_mixin import (
+    _call_slack_api,
     _give_up_slack_api_retry,
-    SlackV2Notification,
 )
+from superset.reports.notifications.slackv2 import SlackV2Notification
 from superset.utils.core import HeaderDataType
+from superset.utils.slack import SlackV2ProbeError
+
+if TYPE_CHECKING:
+    from superset.reports.notifications.slack import SlackNotification
 
 
 @pytest.fixture(autouse=True)
@@ -56,7 +62,7 @@ def _skip_backoff_sleep():
     inside backoff's sync runner keeps the assertion semantics (call_count,
     raised exception type) without the wait.
     """
-    with patch("backoff._sync.time.sleep"):
+    with patch("time.sleep"):
         yield
 
 
@@ -74,10 +80,34 @@ def mock_header_data() -> HeaderDataType:
     }
 
 
-def test_get_channel_with_multi_recipients(mock_header_data) -> None:
+def _make_v1_notification(
+    mock_header_data: HeaderDataType,
+    *,
+    target: str,
+    **content_overrides: Any,
+) -> "SlackNotification":
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    content_values: dict[str, Any] = {
+        "name": "test alert",
+        "header_data": mock_header_data,
+        "description": "desc",
+    }
+    content_values.update(content_overrides)
+    return SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=f'{{"target": "{target}"}}',
+        ),
+        content=NotificationContent(**content_values),
+    )
+
+
+def test_get_channels_with_multi_recipients(mock_header_data) -> None:
     """
-    Test the _get_channel function to ensure it will return a string
-    with recipients separated by commas without interstitial spacing
+    Test _get_channels returns normalized recipients without duplicates.
     """
     from superset.reports.models import ReportRecipients, ReportRecipientType
     from superset.reports.notifications.base import NotificationContent
@@ -98,16 +128,69 @@ def test_get_channel_with_multi_recipients(mock_header_data) -> None:
     slack_notification = SlackNotification(
         recipient=ReportRecipients(
             type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel; second_channel, third_channel"}',  # noqa: E501
+            recipient_config_json='{"target": "some_channel; second_channel, third_channel, some_channel"}',  # noqa: E501
         ),
         content=content,
     )
 
-    result = slack_notification._get_channel()
+    result = slack_notification._get_channels()
 
-    assert result == "some_channel,second_channel,third_channel"
+    assert result == ["some_channel", "second_channel", "third_channel"]
 
-    # Test if the recipient configuration JSON is valid when using a SlackV2 recipient type  # noqa: E501
+
+@pytest.mark.parametrize(
+    "recipient_config_json",
+    ["{not-json", '{"target": ["private-channel"]}'],
+    ids=["malformed-json", "non-string-target"],
+)
+def test_get_channels_rejects_invalid_recipient_config(
+    mock_header_data: HeaderDataType,
+    recipient_config_json: str,
+) -> None:
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+    from superset.reports.notifications.slack import SlackNotification
+
+    notification = SlackNotification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=recipient_config_json,
+        ),
+        content=NotificationContent(
+            name="test alert",
+            header_data=mock_header_data,
+        ),
+    )
+
+    with pytest.raises(NotificationParamException, match="No recipients"):
+        notification._get_channels()
+
+
+@pytest.mark.parametrize(
+    "recipient_config_json",
+    ["{not-json", '{"target": ["C12345"]}'],
+    ids=["malformed-json", "non-string-target"],
+)
+def test_slackv2_get_channels_rejects_invalid_recipient_config(
+    mock_header_data: HeaderDataType,
+    recipient_config_json: str,
+) -> None:
+    from superset.reports.models import ReportRecipients, ReportRecipientType
+    from superset.reports.notifications.base import NotificationContent
+
+    notification = SlackV2Notification(
+        recipient=ReportRecipients(
+            type=ReportRecipientType.SLACKV2,
+            recipient_config_json=recipient_config_json,
+        ),
+        content=NotificationContent(
+            name="test alert",
+            header_data=mock_header_data,
+        ),
+    )
+
+    with pytest.raises(NotificationParamException, match="No recipients"):
+        notification._get_channels()
 
 
 def test_valid_recipient_config_json_slackv2(mock_header_data) -> None:
@@ -145,93 +228,31 @@ def test_valid_recipient_config_json_slackv2(mock_header_data) -> None:
     # Ensure _get_inline_files function returns the correct tuple when content has screenshots  # noqa: E501
 
 
-def test_get_inline_files_with_screenshots(mock_header_data) -> None:
-    """
-    Test the _get_inline_files function to ensure it will return the correct tuple
-    when content has screenshots
-    """
-    from superset.reports.models import ReportRecipients, ReportRecipientType
+@pytest.mark.parametrize(
+    "content_overrides,expected",
+    [
+        ({"screenshots": [b"screenshot"]}, True),
+        ({"csv": b"csv_content"}, True),
+        ({"xlsx": b"xlsx_content"}, True),
+        ({"pdf": b"pdf_content"}, True),
+        ({}, False),
+    ],
+    ids=["screenshots", "csv", "xlsx", "pdf", "none"],
+)
+def test_notification_content_has_attachments(
+    mock_header_data: HeaderDataType,
+    content_overrides: dict[str, Any],
+    expected: bool,
+) -> None:
     from superset.reports.notifications.base import NotificationContent
-    from superset.reports.notifications.slack import SlackNotification
 
     content = NotificationContent(
         name="test alert",
         header_data=mock_header_data,
-        embedded_data=pd.DataFrame(
-            {
-                "A": [1, 2, 3],
-                "B": [4, 5, 6],
-                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
-            }
-        ),
-        description='<p>This is <a href="#">a test</a> alert</p><br />',
-        screenshots=[b"screenshot1", b"screenshot2"],
-    )
-    slack_notification = SlackNotification(
-        recipient=ReportRecipients(
-            type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel"}',
-        ),
-        content=content,
+        **content_overrides,
     )
 
-    result = slack_notification._get_inline_files()
-
-    assert result == ("png", [b"screenshot1", b"screenshot2"])
-
-
-def test_get_inline_files_with_csv(mock_header_data: HeaderDataType) -> None:
-    """
-    Test the _get_inline_files function to ensure it returns the correct tuple
-    when content has a CSV attachment
-    """
-    from superset.reports.models import ReportRecipients, ReportRecipientType
-    from superset.reports.notifications.base import NotificationContent
-    from superset.reports.notifications.slack import SlackNotification
-
-    content = NotificationContent(
-        name="test alert",
-        header_data=mock_header_data,
-        csv=b"csv_content",
-    )
-    slack_notification = SlackNotification(
-        recipient=ReportRecipients(
-            type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel"}',
-        ),
-        content=content,
-    )
-
-    result = slack_notification._get_inline_files()
-
-    assert result == ("csv", [b"csv_content"])
-
-
-def test_get_inline_files_with_xlsx(mock_header_data: HeaderDataType) -> None:
-    """
-    Test the _get_inline_files function to ensure it returns the correct tuple
-    when content has an Excel attachment
-    """
-    from superset.reports.models import ReportRecipients, ReportRecipientType
-    from superset.reports.notifications.base import NotificationContent
-    from superset.reports.notifications.slack import SlackNotification
-
-    content = NotificationContent(
-        name="test alert",
-        header_data=mock_header_data,
-        xlsx=b"xlsx_content",
-    )
-    slack_notification = SlackNotification(
-        recipient=ReportRecipients(
-            type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel"}',
-        ),
-        content=content,
-    )
-
-    result = slack_notification._get_inline_files()
-
-    assert result == ("xlsx", [b"xlsx_content"])
+    assert content.has_attachments is expected
 
 
 def test_get_inline_files_with_xlsx_slackv2(mock_header_data: HeaderDataType) -> None:
@@ -260,38 +281,218 @@ def test_get_inline_files_with_xlsx_slackv2(mock_header_data: HeaderDataType) ->
     # Ensure _get_inline_files function returns None when content has no screenshots or csv  # noqa: E501
 
 
-def test_get_inline_files_with_no_screenshots_or_csv(mock_header_data) -> None:
-    """
-    Test the _get_inline_files function to ensure it will return None
-    when content has no screenshots or csv
-    """
-    from superset.reports.models import ReportRecipients, ReportRecipientType
-    from superset.reports.notifications.base import NotificationContent
-    from superset.reports.notifications.slack import SlackNotification
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_send_without_channels_raises(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    notification = _make_v1_notification(mock_header_data, target="")
 
-    content = NotificationContent(
-        name="test alert",
-        header_data=mock_header_data,
-        embedded_data=pd.DataFrame(
-            {
-                "A": [1, 2, 3],
-                "B": [4, 5, 6],
-                "C": ["111", "222", '<a href="http://www.example.com">333</a>'],
-            }
-        ),
-        description='<p>This is <a href="#">a test</a> alert</p><br />',
-    )
-    slack_notification = SlackNotification(
-        recipient=ReportRecipients(
-            type=ReportRecipientType.SLACK,
-            recipient_config_json='{"target": "some_channel"}',
-        ),
-        content=content,
+    with pytest.raises(NotificationParamException, match="No recipients"):
+        notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=False)
+    slack_client_mock.return_value.chat_postMessage.assert_not_called()
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api")
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_send_legacy_text_skips_v2_probe(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-channel",
     )
 
-    result = slack_notification._get_inline_files()
+    notification.send_legacy_text()
 
-    assert result == (None, [])
+    should_use_v2_api_mock.assert_not_called()
+    slack_client_mock.return_value.chat_postMessage.assert_called_once_with(
+        channel="private-channel",
+        text=ANY,
+    )
+
+
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_send_legacy_text_rejects_attachments(
+    slack_client_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-channel",
+        screenshots=[b"screenshot"],
+    )
+
+    with pytest.raises(
+        NotificationParamException,
+        match="Slack v1 file uploads are no longer supported",
+    ):
+        notification.send_legacy_text()
+
+    slack_client_mock.assert_not_called()
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_send_rejects_files_when_v2_probe_is_unavailable(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-channel",
+        screenshots=[b"screenshot"],
+    )
+
+    with pytest.raises(
+        NotificationParamException,
+        match="Slack v1 file uploads are no longer supported",
+    ):
+        notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=True)
+    slack_client_mock.return_value.files_upload.assert_not_called()
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api")
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_file_send_propagates_transient_v2_probe_failure(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    should_use_v2_api_mock.side_effect = SlackV2ProbeError(
+        "Slack v2 availability probe failed: service_unavailable"
+    )
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-channel",
+        screenshots=[b"screenshot"],
+    )
+
+    with pytest.raises(SlackV2ProbeError, match="service_unavailable"):
+        notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=True)
+    slack_client_mock.assert_not_called()
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_send_retries_only_the_failed_channel(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    failed_attempts = 0
+
+    def chat_side_effect(channel: str, text: str) -> dict[str, bool]:
+        nonlocal failed_attempts
+        if channel == "private-b" and failed_attempts < 2:
+            failed_attempts += 1
+            raise SlackApiError(
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
+            )
+        return {"ok": True}
+
+    slack_client_mock.return_value.chat_postMessage.side_effect = chat_side_effect
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-a,private-b",
+    )
+
+    notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=False)
+    assert [
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    ] == ["private-a", "private-b", "private-b", "private-b"]
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_send_reports_failed_channel_and_continues_later_channels(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+
+    def chat_side_effect(channel: str, text: str) -> dict[str, bool]:
+        if channel == "private-b":
+            raise SlackApiError(
+                message="channel not found",
+                response={"ok": False, "error": "channel_not_found"},
+            )
+        return {"ok": True}
+
+    slack_client_mock.return_value.chat_postMessage.side_effect = chat_side_effect
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-a,private-b,private-c",
+    )
+
+    with pytest.raises(
+        NotificationUnprocessableException,
+        match="private-b",
+    ):
+        notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=False)
+    assert [
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    ] == ["private-a", "private-b", "private-c"]
+
+
+@patch("superset.reports.notifications.slack.g")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=False)
+@patch("superset.reports.notifications.slack.get_slack_client")
+def test_v1_send_deduplicates_channels(
+    slack_client_mock: MagicMock,
+    should_use_v2_api_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    notification = _make_v1_notification(
+        mock_header_data,
+        target="private-a, private-a",
+    )
+
+    notification.send()
+
+    should_use_v2_api_mock.assert_called_once_with(raise_on_error=False)
+    slack_client_mock.return_value.chat_postMessage.assert_called_once_with(
+        channel="private-a",
+        text=ANY,
+    )
 
 
 @patch("superset.reports.notifications.slackv2.g")
@@ -704,6 +905,27 @@ def test_v2_send_to_multiple_channels_uploads_per_channel(
     }
 
 
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_deduplicates_channels(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+    notification = _make_v2_notification(
+        _make_content(mock_header_data),
+        target="C12345,C12345",
+    )
+
+    notification.send()
+
+    slack_client_mock.return_value.chat_postMessage.assert_called_once_with(
+        channel="C12345",
+        text=ANY,
+    )
+
+
 @patch("superset.reports.notifications.slackv2.g")
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 def test_v2_send_text_only_uses_chat_post_message(
@@ -782,7 +1004,7 @@ def test_v2_inline_files_precedence(mock_header_data) -> None:
         ),
         (
             lambda: SlackClientNotConnectedError("offline"),
-            NotificationUnprocessableException,
+            NotificationError,
         ),
         (
             # Fallback: any other SlackClientError becomes Unprocessable.
@@ -817,23 +1039,17 @@ def test_v2_send_retries_on_transient_slack_api_error(
     flask_global_mock: MagicMock,
     mock_header_data,
 ) -> None:
-    """`@backoff.on_exception(NotificationUnprocessableException, max_tries=5)`
-    retries the wrapped exception that send() actually raises.
-
-    A persistent Slack rate-limit (or any other transient failure that maps to
-    NotificationUnprocessableException) results in exactly max_tries=5 send
-    attempts before the final exception propagates. This mirrors the existing
-    pattern in webhook.py.
-    """
+    """A persistent 5xx-style Slack error uses five application attempts."""
     flask_global_mock.logs_context = {}
     slack_client_mock.return_value.chat_postMessage.side_effect = SlackApiError(
-        message="rate limited", response={"ok": False, "error": "ratelimited"}
+        message="service unavailable",
+        response={"ok": False, "error": "service_unavailable"},
     )
 
     content = _make_content(mock_header_data)
     notification = _make_v2_notification(content, target="C12345")
 
-    with pytest.raises(NotificationUnprocessableException):
+    with pytest.raises(NotificationError, match="C12345"):
         notification.send()
 
     assert slack_client_mock.return_value.chat_postMessage.call_count == 5
@@ -846,19 +1062,16 @@ def test_v2_send_retries_then_succeeds_on_transient_failure(
     flask_global_mock: MagicMock,
     mock_header_data,
 ) -> None:
-    """The point of switching backoff to NotificationUnprocessableException is
-    that a *transient* failure now retries and the send ultimately succeeds —
-    behavior the old (dead) SlackApiError decorator never delivered. Fail twice,
-    then succeed: send() must return normally after exactly 3 attempts and still
-    record the success gauge.
-    """
+    """A transient server failure can recover within the application budget."""
     flask_global_mock.logs_context = {}
     slack_client_mock.return_value.chat_postMessage.side_effect = [
         SlackApiError(
-            message="rate limited", response={"ok": False, "error": "ratelimited"}
+            message="service unavailable",
+            response={"ok": False, "error": "service_unavailable"},
         ),
         SlackApiError(
-            message="rate limited", response={"ok": False, "error": "ratelimited"}
+            message="service unavailable",
+            response={"ok": False, "error": "service_unavailable"},
         ),
         {"ok": True},
     ]
@@ -890,7 +1103,8 @@ def test_v2_send_retries_only_failed_channel(
         if channel == "C67890" and failed_attempts < 2:
             failed_attempts += 1
             raise SlackApiError(
-                message="rate limited", response={"ok": False, "error": "ratelimited"}
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
             )
         return {"ok": True}
 
@@ -929,6 +1143,50 @@ def test_v2_send_does_not_retry_permanent_slack_api_error(
     assert slack_client_mock.return_value.chat_postMessage.call_count == 1
 
 
+@pytest.mark.parametrize(
+    ("content_overrides", "method_name"),
+    [
+        ({}, "chat_postMessage"),
+        ({"screenshots": [b"screenshot"]}, "files_upload_v2"),
+    ],
+    ids=["text", "file"],
+)
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_reports_failed_channel_and_continues_later_channels(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    content_overrides: dict[str, Any],
+    method_name: str,
+    mock_header_data: HeaderDataType,
+) -> None:
+    flask_global_mock.logs_context = {}
+
+    def send_side_effect(channel: str, **kwargs: object) -> dict[str, bool]:
+        if channel == "C2":
+            raise SlackApiError(
+                message="channel not found",
+                response={"ok": False, "error": "channel_not_found"},
+            )
+        return {"ok": True}
+
+    method = getattr(slack_client_mock.return_value, method_name)
+    method.side_effect = send_side_effect
+    notification = _make_v2_notification(
+        _make_content(mock_header_data, **content_overrides),
+        target="C1,C2,C3",
+    )
+
+    with pytest.raises(NotificationUnprocessableException, match="C2"):
+        notification.send()
+
+    assert [slack_call.kwargs["channel"] for slack_call in method.call_args_list] == [
+        "C1",
+        "C2",
+        "C3",
+    ]
+
+
 @patch("superset.reports.notifications.slackv2.g")
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 def test_v2_send_does_not_retry_param_errors(
@@ -984,8 +1242,6 @@ def _make_slack_response(status_code: int, data: dict[str, Any]) -> SlackRespons
         # the case the dict-based tests silently lost.
         (503, {"ok": False}),
         (502, {}),
-        # Rate limiting is retryable by status code as well as error code.
-        (429, {"ok": False, "error": "ratelimited"}),
     ],
 )
 def test_give_up_slack_api_retry_retries_on_status_code(
@@ -998,6 +1254,70 @@ def test_give_up_slack_api_retry_retries_on_status_code(
     assert _give_up_slack_api_retry(ex) is False
 
 
+def test_give_up_slack_api_retry_retries_http_408() -> None:
+    """Request timeout stays transient even though it is a 4xx response."""
+    response = _make_slack_response(
+        408,
+        {"ok": False, "error": "request_timeout"},
+    )
+    ex = SlackApiError(message="request timed out", response=response)
+
+    assert _give_up_slack_api_retry(ex) is False
+
+
+def test_give_up_slack_api_retry_delegates_rate_limits_to_sdk() -> None:
+    """Do not multiply WebClient's configured rate-limit retry budget."""
+    response = _make_slack_response(
+        429,
+        {"ok": False, "error": "ratelimited"},
+    )
+    ex = SlackApiError(message="rate limited", response=response)
+
+    assert _give_up_slack_api_retry(ex) is True
+
+
+def test_call_slack_api_retries_ratelimited_code_without_http_429() -> None:
+    """A ratelimited response without HTTP 429 uses application backoff."""
+    ex = SlackApiError(
+        message="rate limited",
+        response={"ok": False, "error": "ratelimited"},
+    )
+    method = MagicMock(side_effect=ex)
+
+    with pytest.raises(SlackApiError):
+        _call_slack_api(method)
+
+    assert method.call_count == 5
+
+
+@patch("superset.reports.notifications.slackv2.g")
+@patch("superset.reports.notifications.slackv2.get_slack_client")
+def test_v2_send_does_not_stack_retries_after_sdk_rate_limit_exhaustion(
+    slack_client_mock: MagicMock,
+    flask_global_mock: MagicMock,
+    mock_header_data: HeaderDataType,
+) -> None:
+    """An exhausted SDK 429 budget is not retried by application backoff."""
+    flask_global_mock.logs_context = {}
+    response = _make_slack_response(
+        429,
+        {"ok": False, "error": "ratelimited"},
+    )
+    slack_client_mock.return_value.chat_postMessage.side_effect = SlackApiError(
+        message="rate limited",
+        response=response,
+    )
+    notification = _make_v2_notification(
+        _make_content(mock_header_data),
+        target="C12345",
+    )
+
+    with pytest.raises(NotificationError, match="C12345"):
+        notification.send()
+
+    slack_client_mock.return_value.chat_postMessage.assert_called_once()
+
+
 def test_give_up_slack_api_retry_gives_up_on_permanent_status_code() -> None:
     """Control: a 4xx (non-429) with a non-transient error code is not retried,
     even when carried by a faithful SlackResponse — so the new helper above is
@@ -1009,6 +1329,18 @@ def test_give_up_slack_api_retry_gives_up_on_permanent_status_code() -> None:
     assert _give_up_slack_api_retry(ex) is True
 
 
+def test_call_slack_api_does_not_retry_empty_client_error_response() -> None:
+    """A non-429 4xx is permanent even when Slack omits an error code."""
+    response = _make_slack_response(400, {})
+    ex = SlackApiError(message="bad request", response=response)
+    method = MagicMock(side_effect=ex)
+
+    with pytest.raises(SlackApiError):
+        _call_slack_api(method)
+
+    method.assert_called_once_with()
+
+
 @patch("superset.reports.notifications.slackv2.g")
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 def test_v2_send_records_statsd_gauge_on_success(

tests/unit_tests/utils/slack_test.py

@@ -20,23 +20,29 @@ import warnings
 import pytest
 from slack_sdk.errors import (
     SlackApiError,
+    SlackClientConfigurationError,
     SlackClientNotConnectedError,
     SlackRequestError,
+    SlackTokenRotationError,
 )
 
 from superset.utils.slack import (
     _emit_v1_flag_off_deprecation,
     _emit_v1_scope_missing_deprecation,
+    _is_transient_slack_api_error,
     _SLACK_V1_DEPRECATION_MESSAGE,
     get_channels_with_search,
     should_use_v2_api,
     SlackChannelTypes,
+    SlackV2ProbeClientError,
+    SlackV2ProbeError,
 )
 
 
 class MockResponse:
-    def __init__(self, data):
+    def __init__(self, data, status_code: int | None = None):
         self._data = data
+        self.status_code = status_code
 
     @property
     def data(self):
@@ -384,7 +390,7 @@ class TestShouldUseV2Api:
 
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter("always")
-            assert should_use_v2_api() is False
+            assert should_use_v2_api(raise_on_error=True) is False
             assert should_use_v2_api() is False
             assert should_use_v2_api() is False
 
@@ -433,6 +439,19 @@ class TestShouldUseV2Api:
         assert logger_mock.warning.call_count == 1
         assert "channels:read" in logger_mock.warning.call_args.args[0]
 
+    @pytest.mark.parametrize("status_code", [429, 503])
+    def test_transient_probe_error_detected_from_http_status(
+        self,
+        status_code: int,
+    ) -> None:
+        """Rate-limit and server HTTP statuses remain system failures."""
+        error = SlackApiError(
+            message="probe failed",
+            response=MockResponse({"ok": False}, status_code=status_code),
+        )
+
+        assert _is_transient_slack_api_error(error, "") is True
+
     @pytest.mark.parametrize(
         "error_code",
         ["invalid_auth", "ratelimited", "fatal_error", "account_inactive", ""],
@@ -502,7 +521,72 @@ class TestShouldUseV2Api:
 
         assert should_use_v2_api() is False
         assert logger_mock.warning.call_count == 1
-        assert "probe failed to connect" in logger_mock.warning.call_args.args[0]
+        assert "probe failed" in logger_mock.warning.call_args.args[0]
+
+    @pytest.mark.parametrize(
+        "exception",
+        [
+            SlackApiError(
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
+            ),
+            SlackApiError(
+                message="rate limited",
+                response={"ok": False, "error": "ratelimited"},
+            ),
+            SlackClientNotConnectedError("transport closed"),
+        ],
+    )
+    def test_raises_system_error_for_transient_probe_when_requested(
+        self,
+        exception: Exception,
+        mocker,
+    ) -> None:
+        """File sends preserve transient probe failures for system handling."""
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = exception
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SlackV2ProbeError, match="probe failed"):
+            should_use_v2_api(raise_on_error=True)
+
+    @pytest.mark.parametrize(
+        "exception",
+        [
+            SlackApiError(
+                message="invalid auth",
+                response={"ok": False, "error": "invalid_auth"},
+            ),
+            SlackRequestError("bad request args"),
+            SlackClientConfigurationError("invalid client configuration"),
+            SlackTokenRotationError("token rotation failed"),
+        ],
+    )
+    def test_raises_client_error_for_permanent_probe_failure_when_requested(
+        self,
+        exception: Exception,
+        mocker,
+    ) -> None:
+        """Permanent probe failures remain operator-fixable client errors."""
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = exception
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SlackV2ProbeClientError, match="probe failed") as exc_info:
+            should_use_v2_api(raise_on_error=True)
+
+        assert exc_info.value.status == 422
+
+    def test_client_probe_error_has_common_probe_error_parent(self) -> None:
+        assert issubclass(SlackV2ProbeClientError, SlackV2ProbeError)
 
     def test_propagates_non_sdk_errors_from_probe(self, mocker):
         """A truly unexpected, non-SDK exception (e.g. a programming error) still