fix: add explicit type hints to elapsed_seconds/dttm locals

Addresses codeant-ai bot review threads asking for explicit type
annotations on the remaining untyped locals touched by the
utcnow() replacement.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset-embedded-sdk/src/guestTokenRefresh.test.ts

@@ -96,6 +96,26 @@ describe("guest token refresh", () => {
     expect(timing).toBe(DEFAULT_TOKEN_EXP_MS - REFRESH_TIMING_BUFFER_MS);
   });
 
+  it("falls back to default timing for a completely malformed token", () => {
+    const timing = getGuestTokenRefreshTiming("not-a-jwt");
+
+    expect(timing).toBe(DEFAULT_TOKEN_EXP_MS - REFRESH_TIMING_BUFFER_MS);
+  });
+
+  it("falls back to default timing for an empty string token", () => {
+    const timing = getGuestTokenRefreshTiming("");
+
+    expect(timing).toBe(DEFAULT_TOKEN_EXP_MS - REFRESH_TIMING_BUFFER_MS);
+  });
+
+  it("falls back to default timing for a token with invalid base64 payload", () => {
+    const timing = getGuestTokenRefreshTiming(
+      "header.!!!invalid-base64!!!.signature",
+    );
+
+    expect(timing).toBe(DEFAULT_TOKEN_EXP_MS - REFRESH_TIMING_BUFFER_MS);
+  });
+
   it("exposes a positive retry delay for failed token refreshes", () => {
     // The refresh loop reschedules itself after this delay when a fetch
     // fails or times out, so it must be a sane positive value.

superset-embedded-sdk/src/guestTokenRefresh.ts

@@ -25,16 +25,20 @@ export const DEFAULT_TOKEN_REFRESH_RETRY_MS = 10000; // wait before retrying a f
 
 // when do we refresh the guest token?
 export function getGuestTokenRefreshTiming(currentGuestToken: string) {
-  const parsedJwt = jwtDecode<Record<string, any>>(currentGuestToken);
-  // if exp is int, it is in seconds, but Date() takes milliseconds
-  const exp = new Date(
-    /[^0-9\.]/g.test(parsedJwt.exp)
-      ? parsedJwt.exp
-      : parseFloat(parsedJwt.exp) * 1000,
-  );
-  const isValidDate = exp.toString() !== "Invalid Date";
-  const ttl = isValidDate
-    ? Math.max(MIN_REFRESH_WAIT_MS, exp.getTime() - Date.now())
-    : DEFAULT_TOKEN_EXP_MS;
-  return ttl - REFRESH_TIMING_BUFFER_MS;
+  try {
+    const parsedJwt = jwtDecode<Record<string, any>>(currentGuestToken);
+    // if exp is int, it is in seconds, but Date() takes milliseconds
+    const exp = new Date(
+      /[^0-9\.]/g.test(parsedJwt.exp)
+        ? parsedJwt.exp
+        : parseFloat(parsedJwt.exp) * 1000,
+    );
+    const isValidDate = exp.toString() !== "Invalid Date";
+    const ttl = isValidDate
+      ? Math.max(MIN_REFRESH_WAIT_MS, exp.getTime() - Date.now())
+      : DEFAULT_TOKEN_EXP_MS;
+    return ttl - REFRESH_TIMING_BUFFER_MS;
+  } catch {
+    return DEFAULT_TOKEN_EXP_MS - REFRESH_TIMING_BUFFER_MS;
+  }
 }

superset/commands/logs/prune.py

@@ -16,7 +16,7 @@
 # under the License.
 import logging
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import sqlalchemy as sa
 
@@ -65,10 +65,12 @@ class LogPruneCommand(BaseCommand):
 
         # Select all IDs that need to be deleted
         # Log.dttm is stored as a naive UTC datetime (no tzinfo), so compute
-        # the cutoff with utcnow() to avoid a naive/aware mismatch that raises
+        # the cutoff as a naive UTC datetime to avoid a naive/aware mismatch that raises
         # on PostgreSQL ("operator does not exist: timestamp without time zone").
         select_stmt = sa.select(Log.id).where(
-            Log.dttm < datetime.utcnow() - timedelta(days=self.retention_period_days)
+            Log.dttm
+            < datetime.now(timezone.utc).replace(tzinfo=None)
+            - timedelta(days=self.retention_period_days)
         )
 
         # Optionally limited by max_rows_per_run

superset/commands/report/execute.py

@@ -19,7 +19,7 @@ import urllib.parse
 import urllib.request
 from collections.abc import Sequence
 from contextlib import closing
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Optional, TYPE_CHECKING, Union
 from urllib.error import URLError
 from uuid import UUID
@@ -136,7 +136,7 @@ class BaseReportState:
     ) -> None:
         self._report_schedule = report_schedule
         self._scheduled_dttm = scheduled_dttm
-        self._start_dttm = datetime.utcnow()
+        self._start_dttm: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         self._execution_id = execution_id
         self._filter_warnings: list[str] = []
 
@@ -166,7 +166,9 @@ class BaseReportState:
             self._report_schedule.last_value_row_json = None
 
         self._report_schedule.last_state = state
-        self._report_schedule.last_eval_dttm = datetime.utcnow()
+        self._report_schedule.last_eval_dttm = datetime.now(timezone.utc).replace(
+            tzinfo=None
+        )
 
     def update_report_schedule_slack_v2(self) -> None:
         """
@@ -232,7 +234,7 @@ class BaseReportState:
             log = ReportExecutionLog(
                 scheduled_dttm=self._scheduled_dttm,
                 start_dttm=self._start_dttm,
-                end_dttm=datetime.utcnow(),
+                end_dttm=datetime.now(timezone.utc).replace(tzinfo=None),
                 value=self._report_schedule.last_value,
                 value_row_json=self._report_schedule.last_value_row_json,
                 state=self._report_schedule.last_state,
@@ -522,7 +524,7 @@ class BaseReportState:
         Get chart or dashboard screenshots
         :raises: ReportScheduleScreenshotFailedError
         """
-        start_time = datetime.utcnow()
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
 
         user, _ = resolve_executor_user(self._report_schedule)
 
@@ -569,14 +571,18 @@ class BaseReportState:
                         "Screenshot failed; aborting to avoid sending a partial report"
                     )
                 imges.append(imge)
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds: float = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.info(
                 "Screenshot capture took %.2fs - execution_id: %s",
                 elapsed_seconds,
                 self._execution_id,
             )
         except SoftTimeLimitExceeded as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.warning(
                 "Screenshot timeout after %.2fs - execution_id: %s",
                 elapsed_seconds,
@@ -584,7 +590,9 @@ class BaseReportState:
             )
             raise ReportScheduleScreenshotTimeout() from ex
         except Exception as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.error(
                 "Screenshot failed after %.2fs - execution_id: %s",
                 elapsed_seconds,
@@ -733,7 +741,7 @@ class BaseReportState:
                 ReportScheduleCsvFailedError,
             )
 
-        start_time = datetime.utcnow()
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         user, username = resolve_executor_user(self._report_schedule)
         auth_cookies = machine_auth_provider_factory.instance.get_auth_cookies(user)
 
@@ -759,7 +767,9 @@ class BaseReportState:
                     request_payload=request_payload,
                     timeout=app.config["ALERT_REPORTS_CSV_REQUEST_TIMEOUT"],
                 )
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds: float = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.info(
                 "%s data generation from %s as user %s took %.2fs - execution_id: %s",
                 label,
@@ -769,7 +779,9 @@ class BaseReportState:
                 self._execution_id,
             )
         except SoftTimeLimitExceeded as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.warning(
                 "%s generation timeout after %.2fs - execution_id: %s",
                 label,
@@ -778,7 +790,9 @@ class BaseReportState:
             )
             raise timeout_error() from ex
         except Exception as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.exception(
                 "%s generation failed after %.2fs - execution_id: %s",
                 label,
@@ -794,7 +808,7 @@ class BaseReportState:
         """
         Return data as a Pandas dataframe, to embed in notifications as a table.
         """
-        start_time = datetime.utcnow()
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
 
         url = self._get_url(result_format=ChartDataResultFormat.JSON)
         user, username = resolve_executor_user(self._report_schedule)
@@ -810,7 +824,9 @@ class BaseReportState:
                 auth_cookies,
                 timeout=app.config["ALERT_REPORTS_CSV_REQUEST_TIMEOUT"],
             )
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds: float = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.info(
                 "DataFrame generation from %s as user %s took %.2fs - execution_id: %s",
                 url,
@@ -819,7 +835,9 @@ class BaseReportState:
                 self._execution_id,
             )
         except SoftTimeLimitExceeded as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.warning(
                 "DataFrame generation timeout after %.2fs - execution_id: %s",
                 elapsed_seconds,
@@ -827,7 +845,9 @@ class BaseReportState:
             )
             raise ReportScheduleDataFrameTimeout() from ex
         except Exception as ex:
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.error(
                 "DataFrame generation failed after %.2fs - execution_id: %s",
                 elapsed_seconds,
@@ -1086,7 +1106,7 @@ class BaseReportState:
         return (
             last_success is not None
             and self._report_schedule.grace_period
-            and datetime.utcnow()
+            and datetime.now(timezone.utc).replace(tzinfo=None)
             - timedelta(seconds=self._report_schedule.grace_period)
             < last_success.end_dttm
         )
@@ -1103,7 +1123,7 @@ class BaseReportState:
         return (
             last_success is not None
             and self._report_schedule.grace_period
-            and datetime.utcnow()
+            and datetime.now(timezone.utc).replace(tzinfo=None)
             - timedelta(seconds=self._report_schedule.grace_period)
             < last_success.end_dttm
         )
@@ -1120,7 +1140,7 @@ class BaseReportState:
         return (
             self._report_schedule.working_timeout is not None
             and self._report_schedule.last_eval_dttm is not None
-            and datetime.utcnow()
+            and datetime.now(timezone.utc).replace(tzinfo=None)
             - timedelta(seconds=self._report_schedule.working_timeout)
             > last_working.end_dttm
         )
@@ -1236,7 +1256,10 @@ class ReportWorkingState(BaseReportState):
                 self._report_schedule
             )
             elapsed_seconds = (
-                (datetime.utcnow() - last_working.end_dttm).total_seconds()
+                (
+                    datetime.now(timezone.utc).replace(tzinfo=None)
+                    - last_working.end_dttm
+                ).total_seconds()
                 if last_working
                 else None
             )
@@ -1434,7 +1457,7 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
             )
             user = security_manager.find_user(username)
 
-            start_time = datetime.utcnow()
+            start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
             with override_user(user):
                 # Pre-commit any permalink rows before the state machine's
                 # @transaction() opens. When called inside a transaction,
@@ -1453,7 +1476,9 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
                     self._execution_id, self._model, self._scheduled_dttm
                 ).run()
 
-            elapsed_seconds = (datetime.utcnow() - start_time).total_seconds()
+            elapsed_seconds: float = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - start_time
+            ).total_seconds()
             logger.info(
                 "Report execution as user %s completed in %.2fs - execution_id: %s",
                 username,

superset/commands/report/log_prune.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy.exc import SQLAlchemyError
 
@@ -41,9 +41,9 @@ class AsyncPruneReportScheduleLogCommand(BaseCommand):
 
         for report_schedule in db.session.query(ReportSchedule).all():
             if report_schedule.log_retention is not None:
-                from_date = datetime.utcnow() - timedelta(
-                    days=report_schedule.log_retention
-                )
+                from_date: datetime = datetime.now(timezone.utc).replace(
+                    tzinfo=None
+                ) - timedelta(days=report_schedule.log_retention)
                 try:
                     row_count = ReportScheduleDAO.bulk_delete_logs(
                         report_schedule,

superset/daos/log.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any
 
 import humanize
@@ -142,7 +142,7 @@ class LogDAO(BaseDAO[Log]):
                     "item_title": item_title,
                     "time": datetime_to_epoch(log.dttm),
                     "time_delta_humanized": humanize.naturaltime(
-                        datetime.utcnow() - log.dttm
+                        datetime.now(timezone.utc).replace(tzinfo=None) - log.dttm
                     ),
                 }
             )

superset/daos/query.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Union
 
 from superset import sql_lab
@@ -48,7 +48,9 @@ class QueryDAO(BaseDAO[Query]):
     @staticmethod
     def get_queries_changed_after(last_updated_ms: Union[float, int]) -> list[Query]:
         # UTC date time, same that is stored in the DB.
-        last_updated_dt = datetime.utcfromtimestamp(last_updated_ms / 1000)
+        last_updated_dt: datetime = datetime.fromtimestamp(
+            last_updated_ms / 1000, timezone.utc
+        ).replace(tzinfo=None)
 
         return (
             db.session.query(Query)

superset/mcp_service/app.py

@@ -125,7 +125,7 @@ necessary access — do NOT attempt to call it.
 Available tools:
 
 Dashboard Management:
-- list_dashboards: List dashboards with advanced filters (1-based pagination)
+- list_dashboards: List dashboards with advanced filters (1-based pagination; deleted_state='only'/'include' surfaces trashed dashboards the caller may restore)
 - get_dashboard_info: Get detailed dashboard information by ID
 - get_dashboard_layout: Get parsed tabs and chart positions for a dashboard (companion to get_dashboard_info when its omitted_fields hint flags position_json)
 - get_dashboard_datasets: List the datasets used by a dashboard's charts, with columns and metrics (context for configuring native filters)
@@ -180,7 +180,7 @@ Dataset Management:
 - query_dataset: Query a dataset using its semantic layer (saved metrics, dimensions, filters) without needing a saved chart
 
 Chart Management:
-- list_charts: List charts with advanced filters (1-based pagination)
+- list_charts: List charts with advanced filters (1-based pagination; deleted_state='only'/'include' surfaces trashed charts the caller may restore)
 - get_chart_info: Get detailed chart information by ID
 - get_chart_preview: Get a visual preview of a chart as formatted content or URL
 - get_chart_data: Get underlying chart data in text-friendly format

superset/mcp_service/chart/schemas.py

@@ -142,6 +142,13 @@ class ChartInfo(BaseModel):
         None, description="Certification details or reason"
     )
     uuid: str | None = Field(None, description="Chart UUID")
+    deleted_at: str | datetime | None = Field(
+        None,
+        description=(
+            "When the chart was moved to trash (soft-deleted); null for live "
+            "charts. Only populated when listing with deleted_state."
+        ),
+    )
     tags: List[TagInfo] = Field(default_factory=list, description="Chart tags")
     editors: List[SubjectInfo] = Field(
         default_factory=list, description="Chart editors"
@@ -626,6 +633,7 @@ def serialize_chart_object(chart: ChartLike | None) -> ChartInfo | None:
             uuid=str(getattr(chart, "uuid", ""))
             if getattr(chart, "uuid", None)
             else None,
+            deleted_at=getattr(chart, "deleted_at", None),
             tags=[
                 TagInfo.model_validate(tag, from_attributes=True)
                 for tag in getattr(chart, "tags", [])
@@ -2161,6 +2169,20 @@ class ListChartsRequest(EditedByMeMixin, CreatedByMeMixin, MetadataCacheControl)
             "used together with 'filters'.",
         ),
     ]
+    deleted_state: Annotated[
+        Literal["include", "only"] | None,
+        Field(
+            default=None,
+            description=(
+                "Surface soft-deleted (trashed) charts: 'only' returns just "
+                "trashed charts, 'include' returns live and trashed charts "
+                "together. Omit for live charts only (default). Trashed rows "
+                "carry a non-null deleted_at and are limited to charts the "
+                "caller owns (admins see all); requires the SOFT_DELETE "
+                "feature flag to have produced trashed rows."
+            ),
+        ),
+    ]
     order_column: Annotated[
         str | None, Field(default=None, description="Column to order results by")
     ]

superset/mcp_service/chart/tool/list_charts.py

@@ -133,6 +133,7 @@ async def list_charts(
         )
     )
 
+    from superset.charts.filters import ChartDeletedStateFilter
     from superset.daos.chart import ChartDAO
     from superset.mcp_service.common.schema_discovery import (
         CHART_SORTABLE_COLUMNS,
@@ -182,6 +183,7 @@ async def list_charts(
         all_columns=all_columns,
         sortable_columns=sortable_columns,
         logger=logger,
+        deleted_state_filter=ChartDeletedStateFilter,
     )
 
     try:
@@ -196,6 +198,7 @@ async def list_charts(
                 page_size=request.page_size,
                 created_by_me=request.created_by_me,
                 edited_by_me=request.edited_by_me,
+                deleted_state=request.deleted_state,
             )
         count = len(result.charts) if hasattr(result, "charts") else 0
         total_pages = getattr(result, "total_pages", None)

superset/mcp_service/dashboard/schemas.py

@@ -253,6 +253,20 @@ class ListDashboardsRequest(EditedByMeMixin, CreatedByMeMixin, MetadataCacheCont
             "Cannot be used together with 'filters'.",
         ),
     ]
+    deleted_state: Annotated[
+        Literal["include", "only"] | None,
+        Field(
+            default=None,
+            description=(
+                "Surface soft-deleted (trashed) dashboards: 'only' returns "
+                "just trashed dashboards, 'include' returns live and trashed "
+                "together. Omit for live dashboards only (default). Trashed "
+                "rows carry a non-null deleted_at and are limited to "
+                "dashboards the caller owns (admins see all); requires the "
+                "SOFT_DELETE feature flag to have produced trashed rows."
+            ),
+        ),
+    ]
     order_column: Annotated[
         str | None, Field(default=None, description="Column to order results by")
     ]
@@ -443,6 +457,13 @@ class DashboardInfo(BaseModel):
     created_on: str | datetime | None = None
     changed_on: str | datetime | None = None
     uuid: str | None = None
+    deleted_at: str | datetime | None = Field(
+        None,
+        description=(
+            "When the dashboard was moved to trash (soft-deleted); null for "
+            "live dashboards. Only populated when listing with deleted_state."
+        ),
+    )
     embedded_uuid: str | None = Field(
         None,
         description=(
@@ -1730,6 +1751,7 @@ def serialize_dashboard_object(dashboard: Any) -> DashboardInfo:
             css=getattr(dashboard, "css", None),
             certified_by=getattr(dashboard, "certified_by", None),
             certification_details=getattr(dashboard, "certification_details", None),
+            deleted_at=getattr(dashboard, "deleted_at", None),
             native_filters=_extract_native_filters(
                 json_metadata_str,
                 include_data_model_metadata=include_data_model_metadata,

superset/mcp_service/dashboard/tool/list_dashboards.py

@@ -130,6 +130,7 @@ async def list_dashboards(
     )
 
     from superset.daos.dashboard import DashboardDAO
+    from superset.dashboards.filters import DashboardDeletedStateFilter
     from superset.mcp_service.common.schema_discovery import (
         DASHBOARD_SORTABLE_COLUMNS,
         get_all_column_names,
@@ -161,6 +162,7 @@ async def list_dashboards(
         all_columns=all_columns,
         sortable_columns=DASHBOARD_SORTABLE_COLUMNS,
         logger=logger,
+        deleted_state_filter=DashboardDeletedStateFilter,
     )
 
     with event_logger.log_context(action="mcp.list_dashboards.query"):
@@ -174,6 +176,7 @@ async def list_dashboards(
             page_size=request.page_size,
             created_by_me=request.created_by_me,
             edited_by_me=request.edited_by_me,
+            deleted_state=request.deleted_state,
         )
     count = len(result.dashboards) if hasattr(result, "dashboards") else 0
     total_pages = getattr(result, "total_pages", None)

superset/mcp_service/mcp_core.py

@@ -43,6 +43,7 @@ from superset.mcp_service.utils.schema_utils import (
     parse_json_or_list,
     parse_json_or_passthrough,
 )
+from superset.models.helpers import skip_visibility_filter
 from superset.utils import json
 
 
@@ -114,6 +115,28 @@ class BaseCore(ABC):
         self.logger.warning(message)
 
 
+class DeletedStateBoundFilter:
+    """Adapt a FAB deleted-state filter for ``BaseDAO.list`` custom_filters.
+
+    ``BaseDAO.list`` invokes custom filters as ``apply(query, None)``, but the
+    ``BaseDeletedStateFilter`` subclasses interpret ``None`` as "live rows
+    only". Binding the value at construction lets the DAO-side invocation
+    reach the FAB filter with the caller's actual ``include``/``only`` choice.
+
+    ``model`` re-exposes the FAB filter's SoftDeleteMixin model class so the
+    caller can scope the session visibility bypass without re-consulting the
+    (Optional) filter-class attribute.
+    """
+
+    def __init__(self, inner: Any, value: str, model: type) -> None:
+        self._inner = inner
+        self._value = value
+        self.model = model
+
+    def apply(self, query: Any, value: Any) -> Any:
+        return self._inner.apply(query, self._value)
+
+
 class ModelListCore(BaseCore, Generic[L]):
     """
     Generic tool for listing model objects with filtering, search, pagination, and
@@ -150,6 +173,7 @@ class ModelListCore(BaseCore, Generic[L]):
         all_columns: List[str] | None = None,
         sortable_columns: List[str] | None = None,
         editor_filter_column: str = "editor",
+        deleted_state_filter: type | None = None,
     ) -> None:
         super().__init__(logger)
         self.dao_class = dao_class
@@ -170,6 +194,12 @@ class ModelListCore(BaseCore, Generic[L]):
             sortable_columns if sortable_columns else []
         )
         self._editor_filter_column = editor_filter_column
+        # A BaseDeletedStateFilter subclass (e.g. ChartDeletedStateFilter).
+        # The FAB filter owns the restore-audience scoping — only owners and
+        # admins may enumerate soft-deleted rows — so reusing it here keeps
+        # that cross-entity contract in one place. None means the resource
+        # does not support deleted_state listing.
+        self._deleted_state_filter = deleted_state_filter
 
     @property
     def all_columns(self) -> List[str]:
@@ -270,6 +300,7 @@ class ModelListCore(BaseCore, Generic[L]):
         page_size: int,
         search: str | None,
         columns_to_load: List[str],
+        custom_filters: Dict[str, Any] | None = None,
     ) -> tuple[List[Any], int]:
         """Call the DAO list method.
 
@@ -284,8 +315,31 @@ class ModelListCore(BaseCore, Generic[L]):
             search=search,
             search_columns=self.search_columns,
             columns=columns_to_load,
+            custom_filters=custom_filters,
         )
 
+    def _build_deleted_state_filter(
+        self, deleted_state: str | None
+    ) -> DeletedStateBoundFilter | None:
+        """Validate deleted_state and bind it to the entity's FAB filter.
+
+        Returns None when trash listing was not requested. Raises for a
+        value other than ``include``/``only`` or when the resource has no
+        deleted-state filter configured.
+        """
+        if deleted_state is None:
+            return None
+        normalized = str(deleted_state).lower().strip()
+        if normalized not in {"include", "only"}:
+            raise ValueError("deleted_state must be 'include' or 'only'")
+        if self._deleted_state_filter is None:
+            raise ValueError("deleted_state is not supported for this resource")
+        # ``model`` is the ClassVar every BaseDeletedStateFilter subclass binds.
+        model = self._deleted_state_filter.model  # type: ignore[attr-defined]
+        datamodel = SQLAInterface(model, db.session)
+        inner = self._deleted_state_filter("id", datamodel)
+        return DeletedStateBoundFilter(inner, normalized, model)
+
     def run_tool(
         self,
         filters: Any | None = None,
@@ -297,6 +351,7 @@ class ModelListCore(BaseCore, Generic[L]):
         page_size: int = 10,
         created_by_me: bool = False,
         edited_by_me: bool = False,
+        deleted_state: str | None = None,
     ) -> L:
         # Clamp page_size to MAX_PAGE_SIZE as defense-in-depth
         page_size = min(page_size, MAX_PAGE_SIZE)
@@ -326,17 +381,40 @@ class ModelListCore(BaseCore, Generic[L]):
 
         self._validate_order_column(order_column)
 
+        deleted_state_bound = self._build_deleted_state_filter(deleted_state)
+        if deleted_state_bound is not None:
+            # Trashed rows must be distinguishable from live ones (matters in
+            # "include" mode), so force deleted_at into the loaded columns
+            # and the serialization allowlist.
+            for column_list in (columns_requested, columns_to_load):
+                if "deleted_at" not in column_list:
+                    column_list.append("deleted_at")
+
         # Query the DAO
         items: List[Any]
-        items, total_count = self._call_dao_list(
-            filters=filters,
-            order_column=order_column or "changed_on",
-            order_direction=str(order_direction or "desc"),
-            page=page,
-            page_size=page_size,
-            search=search,
-            columns_to_load=columns_to_load,
-        )
+        dao_kwargs = {
+            "filters": filters,
+            "order_column": order_column or "changed_on",
+            "order_direction": str(order_direction or "desc"),
+            "page": page,
+            "page_size": page_size,
+            "search": search,
+            "columns_to_load": columns_to_load,
+        }
+        if deleted_state_bound is not None:
+            # The soft-delete ORM listener appends ``deleted_at IS NULL`` at
+            # execution time, so the session-scoped bypass must span both
+            # executions inside DAO.list (count + fetch). The context manager
+            # guarantees release even on exceptions; the FAB filter's
+            # restore-audience scoping (applied via custom_filters) decides
+            # which unhidden rows the caller may actually see.
+            with skip_visibility_filter(db.session, deleted_state_bound.model):
+                items, total_count = self._call_dao_list(
+                    custom_filters={"deleted_state": deleted_state_bound},
+                    **dao_kwargs,
+                )
+        else:
+            items, total_count = self._call_dao_list(**dao_kwargs)
         # Serialize items
         item_objs = []
         for item in items:

superset/utils/cache.py

@@ -18,7 +18,7 @@ from __future__ import annotations
 
 import inspect
 import logging
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from functools import wraps
 from typing import Any, Callable
 
@@ -73,7 +73,9 @@ def set_and_log_cache(
     if timeout == CACHE_DISABLED_TIMEOUT:
         return
     try:
-        dttm = datetime.utcnow().isoformat().split(".")[0]
+        dttm: str = (
+            datetime.now(timezone.utc).replace(tzinfo=None).isoformat().split(".")[0]
+        )
         value = {**cache_value, "dttm": dttm}
         cache_instance.set(cache_key, value, timeout=timeout)
         stats_logger = app.config["STATS_LOGGER"]
@@ -226,7 +228,9 @@ def etag_cache(  # noqa: C901
 
             # Check if the cache is stale. Default the content_changed_time to now
             # if we don't know when it was last modified.
-            content_changed_time = datetime.utcnow()
+            content_changed_time: datetime = datetime.now(timezone.utc).replace(
+                tzinfo=None
+            )
             if get_last_modified:
                 content_changed_time = get_last_modified(*args, **kwargs)
                 if (

superset/utils/dates.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytz
 
@@ -31,4 +31,4 @@ def datetime_to_epoch(dttm: datetime) -> float:
 
 
 def now_as_float() -> float:
-    return datetime_to_epoch(datetime.utcnow())
+    return datetime_to_epoch(datetime.now(timezone.utc).replace(tzinfo=None))

tests/integration_tests/views/test_explore_redirect.py

@@ -28,10 +28,10 @@ failure mode the helper or its callers must close:
 - Loop guard via ``(endpoint, sorted_query_items)`` equality.
 - ``ExploreView.root`` owns the bare ``/explore/`` rule (registration order).
 - The sanctioned callers set is exactly ``{explore.py, core.py}``;
-  a fourth caller fails the static assertion.
+  a third caller fails the static assertion.
 """
 
-import re
+import ast
 from pathlib import Path
 from unittest import mock
 from urllib.parse import quote
@@ -300,21 +300,33 @@ class TestExploreRedirect(SupersetTestCase):
         """Exactly two files call the helper; a third must update this test.
 
         Pins the sanctioned-callers invariant via a source-tree scan so a
-        fourth caller landing without an explicit update fails CI. Mirrors
+        third caller landing without an explicit update fails CI. Mirrors
         the ``applicationRoot()`` and ``DIRECT_DOM_NAV_SANCTIONED`` patterns
         from the frontend L2 scanners.
         """
-        pattern = re.compile(r"\bget_explore_redirect_url\(")
+        # Scan the AST rather than raw text: only executable call sites count,
+        # so a mention of the helper in a comment, docstring, or string literal
+        # is not mistaken for a caller. This also excludes the definition site
+        # by shape instead of by path — `def get_explore_redirect_url` parses to
+        # a FunctionDef, not a Call — which keeps `views/utils.py` itself in
+        # scope. Skipping that module wholesale would hide a genuine new call
+        # added inside it, the exact thing this scan exists to catch.
         callers: set[str] = set()
         for path in (REPO_ROOT / "superset").rglob("*.py"):
-            text = path.read_text(encoding="utf-8")
-            # Skip the helper's own definition site.
-            if path.name == "utils.py" and path.parent.name == "views":
-                # `views/utils.py` contains the `def` and an internal docstring
-                # `f"...({CreateFormDataCommand...".` reference; not a caller.
-                continue
-            if pattern.search(text):
-                callers.add(str(path.relative_to(REPO_ROOT)).replace("\\", "/"))
+            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                func = node.func
+                if isinstance(func, ast.Name):
+                    called = func.id
+                elif isinstance(func, ast.Attribute):
+                    called = func.attr
+                else:
+                    continue
+                if called == "get_explore_redirect_url":
+                    callers.add(str(path.relative_to(REPO_ROOT)).replace("\\", "/"))
+                    break
         assert callers == {
             "superset/views/explore.py",
             "superset/views/core.py",

tests/unit_tests/common/test_query_actions.py

@@ -0,0 +1,88 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from superset.common.chart_data import ChartDataResultType
+from superset.common.query_actions import _get_drill_detail
+from superset.common.query_object import QueryObject
+from superset.utils.core import QueryObjectFilterClause
+
+
+@patch("superset.common.query_actions._get_full")
+def test_get_drill_detail_does_not_strip_filters(
+    mock_get_full: MagicMock,
+) -> None:
+    """
+    Characterization test for ``_get_drill_detail`` (superset/common/query_actions.py),
+    the backend transform behind the "Drill to Detail by" samples query.
+
+    ``_get_drill_detail`` shallow-copies the incoming ``QueryObject`` and rewrites
+    ``is_timeseries``, ``metrics``, ``post_processing``, ``columns`` and ``orderby``;
+    it never reads or reassigns ``QueryObject.filter``. This test pins that
+    behavior down: filters present on the ``QueryObject`` handed to
+    ``_get_drill_detail`` are still present, unmodified, on the object it hands off
+    to ``_get_full``.
+
+    This is deliberately narrow and does NOT reproduce or close #28562 ("Drill to
+    Detail by" ignoring a dashboard's applied filters) -- that report describes
+    filters missing from the query *before* it reaches this function, which points
+    at the payload assembled upstream of ``_get_drill_detail`` (dashboard native
+    filter propagation into the chart's form data / the drill payload built on the
+    frontend), not at this transform. If that assembly is ever changed to drop
+    filters before calling into this code path, this test would not catch it; it
+    only guards against a regression introduced inside ``_get_drill_detail`` itself.
+    """
+    applied_filter: QueryObjectFilterClause = {
+        "col": "region",
+        "op": "==",
+        "val": "USA",
+    }
+
+    query_obj: QueryObject = QueryObject(
+        columns=["region", "sales"],
+        metrics=["count"],
+        filters=[applied_filter],
+    )
+
+    col_region: MagicMock = MagicMock()
+    col_region.column_name = "region"
+    col_sales: MagicMock = MagicMock()
+    col_sales.column_name = "sales"
+
+    datasource = MagicMock()
+    datasource.columns = [col_region, col_sales]
+
+    query_context: MagicMock = MagicMock()
+    query_context.datasource = datasource
+    query_context.result_type = ChartDataResultType.DRILL_DETAIL
+
+    captured: dict[str, QueryObject] = {}
+
+    def _capture(_ctx: MagicMock, obj: QueryObject, _force: bool) -> dict[str, Any]:
+        captured["query_obj"] = obj
+        return {}
+
+    mock_get_full.side_effect = _capture
+
+    _get_drill_detail(query_context, query_obj)
+
+    executed: QueryObject = captured["query_obj"]
+    assert applied_filter in executed.filter, (
+        "_get_drill_detail unexpectedly stripped a filter it never touches; "
+        "this guards against a regression introduced in that function, not #28562."
+    )

tests/unit_tests/mcp_service/chart/tool/test_generate_chart.py

@@ -432,6 +432,7 @@ def _make_mock_chart(chart_id: int = 42) -> Mock:
     chart.created_on = None
     chart.created_on_humanized = "2 days ago"
     chart.uuid = "test-uuid-42"
+    chart.deleted_at = None
     chart.tags = []
     chart.editors = []
     return chart

tests/unit_tests/mcp_service/chart/tool/test_list_charts_deleted_state.py

@@ -0,0 +1,171 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for list_charts ``deleted_state`` trash listing.
+
+The deleted_state param opts the list query into surfacing soft-deleted
+charts: the session-scoped visibility bypass wraps the DAO call, the REST
+``ChartDeletedStateFilter`` (which owns the restore-audience scoping) is
+passed through as a DAO custom filter, and ``deleted_at`` is forced into the
+loaded columns so trashed rows are distinguishable in the response.
+"""
+
+from collections.abc import Iterator
+from datetime import datetime
+from unittest.mock import MagicMock, Mock, patch
+from uuid import UUID
+
+import pytest
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+_DAO_LIST = "superset.daos.chart.ChartDAO.list"
+_BYPASS = "superset.mcp_service.mcp_core.skip_visibility_filter"
+
+
+@pytest.fixture
+def mcp_server() -> object:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Iterator[Mock]:
+    with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _trashed_chart_row() -> Mock:
+    row = Mock(
+        spec=[
+            "id",
+            "slice_name",
+            "viz_type",
+            "changed_on",
+            "created_on",
+            "uuid",
+            "deleted_at",
+        ]
+    )
+    row.id = 10
+    row.slice_name = "Trashed Sales"
+    row.viz_type = "table"
+    row.changed_on = datetime(2026, 6, 1)
+    row.created_on = datetime(2026, 5, 1)
+    row.uuid = UUID("11111111-2222-3333-4444-555555555555")
+    row.deleted_at = datetime(2026, 7, 1)
+    return row
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_deleted_state_only_passes_custom_filter(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_charts", {"request": {"deleted_state": "only"}})
+
+    kwargs = mock_list.call_args.kwargs
+    assert "deleted_state" in (kwargs.get("custom_filters") or {})
+    assert "deleted_at" in kwargs["columns"]
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_default_has_no_deleted_state_filter(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_charts", {})
+
+    kwargs = mock_list.call_args.kwargs
+    assert not kwargs.get("custom_filters")
+
+
+@patch(_BYPASS)
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_deleted_state_wraps_visibility_bypass(
+    mock_list: Mock, mock_bypass: MagicMock, mcp_server: object
+) -> None:
+    from superset.models.slice import Slice
+
+    mock_list.return_value = ([], 0)
+    mock_bypass.return_value.__enter__ = Mock(return_value=None)
+    mock_bypass.return_value.__exit__ = Mock(return_value=False)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_charts", {"request": {"deleted_state": "include"}})
+
+    mock_bypass.assert_called_once()
+    assert Slice in mock_bypass.call_args.args
+
+
+@patch(_BYPASS)
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_no_deleted_state_no_visibility_bypass(
+    mock_list: Mock, mock_bypass: MagicMock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_charts", {})
+
+    mock_bypass.assert_not_called()
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_deleted_state_serializes_deleted_at(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([_trashed_chart_row()], 1)
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "list_charts", {"request": {"deleted_state": "only"}}
+        )
+
+    data = json.loads(result.content[0].text)
+    assert data["charts"][0]["id"] == 10
+    assert data["charts"][0]["deleted_at"] is not None
+    assert "deleted_at" in data["columns_loaded"]
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_charts_deleted_state_invalid_value_rejected(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        with pytest.raises(ToolError):
+            await client.call_tool(
+                "list_charts", {"request": {"deleted_state": "everything"}}
+            )

tests/unit_tests/mcp_service/dashboard/tool/test_dashboard_tools.py

@@ -90,6 +90,7 @@ async def test_list_dashboards_basic(mock_list, mcp_server):
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.is_managed_externally = False
     dashboard.external_url = None
@@ -156,6 +157,7 @@ async def test_list_dashboards_with_filters(mock_list, mcp_server):
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.is_managed_externally = False
     dashboard.external_url = None
@@ -251,6 +253,7 @@ async def test_list_dashboards_with_search(mock_list, mcp_server):
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.is_managed_externally = False
     dashboard.external_url = None
@@ -518,6 +521,7 @@ async def test_get_dashboard_info_permalink_key_includes_filter_state(
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.published = True
     dashboard.is_managed_externally = False
@@ -746,6 +750,7 @@ async def test_get_dashboard_info_does_not_expose_access_list(mock_info, mcp_ser
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.position_json = None
     dashboard.published = True
@@ -802,6 +807,7 @@ async def test_get_dashboard_info_restricted_user_redacts_data_model_metadata(
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = json.dumps(
         {
             "native_filter_configuration": [
@@ -868,6 +874,7 @@ async def test_get_dashboard_info_restricted_user_redacts_permalink_filter_state
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.position_json = None
     dashboard.published = True
@@ -969,6 +976,7 @@ async def test_list_dashboards_omits_requested_user_directory_fields(
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.position_json = None
     dashboard.is_managed_externally = False
@@ -1019,6 +1027,7 @@ async def test_get_dashboard_info_includes_embedded_uuid(mock_find_object, mcp_s
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = "{}"
     dashboard.published = True
     dashboard.is_managed_externally = False
@@ -1062,6 +1071,7 @@ async def test_get_dashboard_info_embedded_uuid_none_when_not_embedded(
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = "{}"
     dashboard.published = True
     dashboard.is_managed_externally = False
@@ -1100,6 +1110,7 @@ async def test_get_dashboard_info_by_uuid(mock_find_object, mcp_server):
     dashboard.css = ""
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = "{}"
     dashboard.published = True
     dashboard.is_managed_externally = False
@@ -1139,6 +1150,7 @@ async def test_get_dashboard_info_by_slug(mock_find_object, mcp_server):
     dashboard.css = ""
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = "{}"
     dashboard.published = True
     dashboard.is_managed_externally = False
@@ -1190,6 +1202,7 @@ async def test_list_dashboards_custom_uuid_slug_columns(mock_list, mcp_server):
     dashboard.css = None
     dashboard.certified_by = None
     dashboard.certification_details = None
+    dashboard.deleted_at = None
     dashboard.json_metadata = None
     dashboard.is_managed_externally = False
     dashboard.external_url = None
@@ -1245,6 +1258,7 @@ async def test_list_dashboards_sanitizes_dashboard_descriptions_and_filter_text(
     dashboard.dashboard_title = "Quarterly Dashboard"
     dashboard.slug = "quarterly-dashboard"
     dashboard.uuid = "uuid-quarterly-3"
+    dashboard.deleted_at = None
     dashboard.url = "/dashboard/3"
     dashboard.published = True
     dashboard.changed_by_name = "admin"
@@ -1410,6 +1424,7 @@ class TestDashboardDefaultColumnFiltering:
         dashboard.css = None
         dashboard.certified_by = None
         dashboard.certification_details = None
+        dashboard.deleted_at = None
         dashboard.json_metadata = None
         dashboard.is_managed_externally = False
         dashboard.external_url = None

tests/unit_tests/mcp_service/dashboard/tool/test_list_dashboards_deleted_state.py

@@ -0,0 +1,176 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for list_dashboards ``deleted_state`` trash listing.
+
+Mirrors the list_charts deleted_state tests: DAO custom filter pass-through,
+session-scoped visibility bypass around the DAO call, and ``deleted_at``
+forced into loaded columns and the serialized response.
+"""
+
+from collections.abc import Iterator
+from datetime import datetime
+from unittest.mock import MagicMock, Mock, patch
+from uuid import UUID
+
+import pytest
+from fastmcp import Client
+
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+_DAO_LIST = "superset.daos.dashboard.DashboardDAO.list"
+_BYPASS = "superset.mcp_service.mcp_core.skip_visibility_filter"
+
+
+@pytest.fixture
+def mcp_server() -> object:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Iterator[Mock]:
+    with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _trashed_dashboard_row() -> Mock:
+    row = Mock(
+        spec=[
+            "id",
+            "dashboard_title",
+            "slug",
+            "published",
+            "changed_on",
+            "created_on",
+            "uuid",
+            "deleted_at",
+        ]
+    )
+    row.id = 1
+    row.dashboard_title = "Trashed Dashboard"
+    row.slug = None
+    row.published = False
+    row.changed_on = datetime(2026, 6, 1)
+    row.created_on = datetime(2026, 5, 1)
+    row.uuid = UUID("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
+    row.deleted_at = datetime(2026, 7, 1)
+    return row
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_deleted_state_only_passes_custom_filter(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "list_dashboards", {"request": {"deleted_state": "only"}}
+        )
+
+    kwargs = mock_list.call_args.kwargs
+    assert "deleted_state" in (kwargs.get("custom_filters") or {})
+    assert "deleted_at" in kwargs["columns"]
+
+
+@patch(_BYPASS)
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_deleted_state_wraps_visibility_bypass(
+    mock_list: Mock, mock_bypass: MagicMock, mcp_server: object
+) -> None:
+    from superset.models.dashboard import Dashboard
+
+    mock_list.return_value = ([], 0)
+    mock_bypass.return_value.__enter__ = Mock(return_value=None)
+    mock_bypass.return_value.__exit__ = Mock(return_value=False)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "list_dashboards", {"request": {"deleted_state": "include"}}
+        )
+
+    mock_bypass.assert_called_once()
+    assert Dashboard in mock_bypass.call_args.args
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_deleted_state_serializes_deleted_at(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([_trashed_dashboard_row()], 1)
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "list_dashboards", {"request": {"deleted_state": "only"}}
+        )
+
+    data = json.loads(result.content[0].text)
+    assert data["dashboards"][0]["id"] == 1
+    assert data["dashboards"][0]["deleted_at"] is not None
+    assert "deleted_at" in data["columns_loaded"]
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_default_has_no_deleted_state_filter(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_dashboards", {})
+
+    kwargs = mock_list.call_args.kwargs
+    assert not kwargs.get("custom_filters")
+
+
+@patch(_BYPASS)
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_no_deleted_state_no_visibility_bypass(
+    mock_list: Mock, mock_bypass: MagicMock, mcp_server: object
+) -> None:
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        await client.call_tool("list_dashboards", {})
+
+    mock_bypass.assert_not_called()
+
+
+@patch(_DAO_LIST)
+@pytest.mark.asyncio
+async def test_list_dashboards_deleted_state_invalid_value_rejected(
+    mock_list: Mock, mcp_server: object
+) -> None:
+    from fastmcp.exceptions import ToolError
+
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        with pytest.raises(ToolError):
+            await client.call_tool(
+                "list_dashboards", {"request": {"deleted_state": "everything"}}
+            )

tests/unit_tests/mcp_service/test_mcp_core.py

@@ -30,7 +30,11 @@ from unittest.mock import MagicMock, patch
 import pytest
 from pydantic import BaseModel
 
-from superset.mcp_service.mcp_core import _slugify, ModelGetInfoCore
+from superset.mcp_service.mcp_core import (
+    _slugify,
+    ModelGetInfoCore,
+    ModelListCore,
+)
 
 
 class _FakeOutput(BaseModel):
@@ -312,3 +316,70 @@ def test_explicit_title_column_overrides_dao_attribute() -> None:
 )
 def test_slugify_handles_edge_cases(identifier: str, expected_slug: str) -> None:
     assert _slugify(identifier) == expected_slug
+
+
+def test_deleted_state_bound_filter_delegates_bound_value() -> None:
+    """The adapter passed to DAO custom_filters must forward the bound
+    deleted_state value — BaseDAO.list invokes custom filters with
+    ``apply(query, None)``, which the FAB filter would treat as 'live only'."""
+    from unittest.mock import Mock
+
+    from superset.mcp_service.mcp_core import DeletedStateBoundFilter
+
+    inner = Mock()
+    inner.apply.return_value = "filtered-query"
+    bound = DeletedStateBoundFilter(inner, "only", model=Mock)
+
+    assert bound.apply("query", None) == "filtered-query"
+    inner.apply.assert_called_once_with("query", "only")
+
+
+def _build_list_core(
+    deleted_state_filter: type | None = None,
+) -> "ModelListCore[Any]":
+    """Minimal ModelListCore for exercising _build_deleted_state_filter."""
+    from unittest.mock import MagicMock
+
+    from pydantic import BaseModel
+
+    class _Out(BaseModel):
+        pass
+
+    return ModelListCore(
+        dao_class=MagicMock(),
+        output_schema=_Out,
+        item_serializer=lambda obj, cols: None,
+        filter_type=_Out,
+        default_columns=["id"],
+        search_columns=[],
+        list_field_name="items",
+        output_list_schema=_Out,
+        deleted_state_filter=deleted_state_filter,
+    )
+
+
+def test_build_deleted_state_filter_none_short_circuits() -> None:
+    """deleted_state=None must engage no filter or bypass machinery."""
+    core = _build_list_core()
+    assert core._build_deleted_state_filter(None) is None
+
+
+def test_build_deleted_state_filter_rejects_invalid_value() -> None:
+    """The core re-validates defensively even though the request schemas
+    bound the value with Literal — a non-schema caller must still be
+    rejected before any visibility bypass engages."""
+    import pytest
+
+    core = _build_list_core()
+    with pytest.raises(ValueError, match="include.*only|'include' or 'only'"):
+        core._build_deleted_state_filter("everything")
+
+
+def test_build_deleted_state_filter_rejects_unconfigured_resource() -> None:
+    """Resources without a configured deleted-state filter reject the
+    parameter instead of silently returning live rows."""
+    import pytest
+
+    core = _build_list_core(deleted_state_filter=None)
+    with pytest.raises(ValueError, match="not supported"):
+        core._build_deleted_state_filter("only")

tests/unit_tests/security/api_test.py

@@ -66,6 +66,61 @@ def test_csrf_exempt_blueprints_with_api_key(app: Any, app_context: None) -> Non
     assert "ApiKeyApi" in {blueprint.name for blueprint in csrf._exempt_blueprints}
 
 
+def test_security_api_trailing_slash_matches_route_ownership(client: Any) -> None:
+    """Regression for #29934: sibling ``/api/v1/security/*`` endpoints respond to
+    a misspelled (wrong trailing-slash) URL differently, and that difference is
+    the *intended* behavior — a Werkzeug routing artifact of who owns each route,
+    not a bug.
+
+    Three routes live under the same ``/api/v1/security/`` prefix but are
+    declared with different slash conventions because they come from different
+    owners:
+
+      * ``login``       -> ``@expose("/login")``        (no trailing slash)
+        Flask-AppBuilder's own route. Superset does not own or register it, so
+        it inherits FAB's no-trailing-slash convention. Werkzeug hard-404s a
+        request that adds a stray trailing slash to a no-slash route (there is
+        no canonical slashed URL to redirect to).
+      * ``csrf_token``  -> ``@expose("/csrf_token/")``  (trailing slash)
+      * ``guest_token`` -> ``@expose("/guest_token/")`` (trailing slash)
+        Superset's own routes, whose trailing-slash URLs are the documented
+        canonical URLs (the Embedded SDK depends on them). Werkzeug 308-redirects
+        a request that omits the trailing slash to the canonical slashed URL.
+
+    Unifying the two would either break the documented ``csrf_token`` /
+    ``guest_token`` URLs the Embedded SDK relies on, or require patching FAB /
+    an app-wide routing change. So the divergence is working-as-designed. This
+    test pins that intended per-route contract so the behavior stays documented
+    and any accidental future change is caught.
+    """
+    # Control: the canonical (no trailing slash) login route is registered and
+    # reachable, so the 404 below is specific to the stray slash rather than
+    # the route being missing entirely.
+    response = client.open("/api/v1/security/login", method="POST")
+    assert response.status_code != 404
+
+    # FAB-owned no-trailing-slash route: adding a stray slash hard-404s because
+    # there is no canonical slashed URL to redirect to.
+    response = client.open(
+        "/api/v1/security/login/", method="POST", follow_redirects=False
+    )
+    assert response.status_code == 404
+
+    # Superset-owned canonical trailing-slash routes: omitting the trailing
+    # slash 308-redirects to the documented canonical URL.
+    response = client.open(
+        "/api/v1/security/csrf_token", method="GET", follow_redirects=False
+    )
+    assert response.status_code == 308
+    assert response.headers["Location"].endswith("/api/v1/security/csrf_token/")
+
+    response = client.open(
+        "/api/v1/security/guest_token", method="POST", follow_redirects=False
+    )
+    assert response.status_code == 308
+    assert response.headers["Location"].endswith("/api/v1/security/guest_token/")
+
+
 def test_rls_rule_schema_accepts_dataset_scoped_rule() -> None:
     """A rule with an integer ``dataset`` and a ``clause`` loads unchanged."""
     result = RlsRuleSchema().load({"dataset": 41, "clause": "tenant_id = 1"})

tests/unit_tests/views/test_redirect_view_subdirectory.py

@@ -46,11 +46,18 @@ from urllib.parse import quote
 import pytest
 from flask import Flask
 
+from superset.views.base import BaseSupersetView
 from superset.views.redirect import RedirectView
 
 INTERNAL_HOST = "http://localhost:8088"
 EXTERNAL_HOST = "https://external.example.com"
 
+#: Stand-in for the React warning page. `RedirectView` delegates the
+#: external-URL branch to `BaseSupersetView.render_app_template`, which needs a
+#: Jinja loader the minimal app below does not have; tests that exercise that
+#: branch patch the method to return this sentinel body.
+_WARNING_PAGE_BODY = "EXTERNAL_LINK_WARNING_PAGE"
+
 
 def _make_app(application_root: str = "/") -> Flask:
     """Build a minimal Flask app that mounts only `RedirectView`.
@@ -70,9 +77,8 @@ def _make_app(application_root: str = "/") -> Flask:
     app.config["WEBDRIVER_BASEURL"] = INTERNAL_HOST
     app.config["WEBDRIVER_BASEURL_USER_FRIENDLY"] = INTERNAL_HOST
     # `BaseSupersetView.render_app_template` is the external-URL branch and
-    # touches Jinja. The subdir contract is the internal-URL 302 branch, so
-    # we only need to mount the handler — the template branch isn't exercised
-    # by these tests.
+    # touches Jinja, which this app has no loader for. Tests that exercise that
+    # branch patch the method out (see `_WARNING_PAGE_BODY`).
     view = RedirectView()
     app.add_url_rule(
         "/redirect/",
@@ -202,20 +208,39 @@ def test_feature_flag_disabled_returns_404_under_subdir() -> None:
     assert response.status_code == 404
 
 
-def test_protocol_relative_url_falls_through_to_warning_page_under_subdir() -> None:
-    """Protocol-relative URLs (`//evil.com`) are not "safe" — the shim must
-    not 302 them. With SCRIPT_NAME mounting the request still reaches the
-    handler; the response falls through to the external-warning branch
-    (which would render the React shell in a full app, but in this minimal
-    setup raises a TemplateNotFound on the bare Flask app — we only assert
-    the 302 path is NOT taken)."""
+@pytest.mark.parametrize(
+    "unsafe_url",
+    [
+        # Protocol-relative — `is_safe_redirect_url` rejects these outright.
+        "//evil.example.com/path",
+        "\\\\evil.example.com/path",
+        # Absolute URL on a host that is not the configured base host.
+        f"{EXTERNAL_HOST}/path",
+    ],
+)
+def test_unsafe_url_renders_warning_page_under_subdir(unsafe_url: str) -> None:
+    """An unsafe target must render the external-link warning page, never a
+    302 to the target itself.
+
+    `render_app_template` (the warning-page branch) is stubbed because the
+    minimal Flask app has no Jinja template loader — unstubbed it raises
+    `TemplateNotFound`, so the branch would 500. Asserting only "not a 302"
+    would pass on that 500, on a 404, and on a genuine warning page alike,
+    which does not distinguish "we showed the user a warning" from "the
+    handler blew up". Pin the status, the rendered body, and the absence of
+    a `Location` header instead.
+    """
     app = _make_app(application_root="/superset")
-    # `//evil.example.com/path` — `is_safe_redirect_url` rejects this branch.
-    response = _get(
-        app,
-        "/redirect/?url=" + quote("//evil.example.com/path", safe=""),
-        script_name="/superset",
-    )
-    # Either the warning page rendered (200) or the template lookup failed
-    # (500) — either way, no 302 to the unsafe target.
-    assert response.status_code != 302
+    with patch.object(
+        BaseSupersetView,
+        "render_app_template",
+        return_value=_WARNING_PAGE_BODY,
+    ):
+        response = _get(
+            app,
+            "/redirect/?url=" + quote(unsafe_url, safe=""),
+            script_name="/superset",
+        )
+    assert response.status_code == 200
+    assert response.get_data(as_text=True) == _WARNING_PAGE_BODY
+    assert "Location" not in response.headers