fix(reports): derive tile-wait budget from the running Celery task's own time limit

The fixed 1440s budget was derived only from the reports task's 1740s
Celery time_limit. A concurrent change (superset-shell#4389) gives
thumbnail tasks a much shorter per-task limit (120s hard / 90s soft) via
task_annotations; inside a 120s-limited task the fixed budget would never
bind before Celery SIGKILLs the worker.

Derive the budget from the currently running task's own effective limit
via celery.current_task.request.timelimit (soft preferred, else hard),
minus a safety margin that scales with the limit -- min(300s, 20% of the
limit) -- so a large report-task limit keeps the existing 300s margin
while a small thumbnail-task limit doesn't have its whole budget eaten by
a fixed margin. The result is floored at 30s so a pathologically small
task limit still leaves the tile loop a chance to capture something.

Falls back to the existing fixed TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
constant (1440s) when there's no Celery task context (e.g. synchronous
thumbnail generation), the task exposes no usable limit, or anything goes
wrong while inspecting it -- derivation failures are caught, logged at
DEBUG, and never propagate.

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

superset/commands/report/execute.py

@@ -563,7 +563,10 @@ class BaseReportState:
         try:
             imges = []
             for screenshot in screenshots:
-                imge = screenshot.get_screenshot(user=user)
+                imge = screenshot.get_screenshot(
+                    user=user,
+                    log_context=f"execution_id={self._execution_id}",
+                )
                 if imge is None:
                     raise ReportScheduleScreenshotFailedError(
                         "Screenshot failed; aborting to avoid sending a partial report"

superset/utils/screenshot_utils.py

@@ -19,8 +19,10 @@ from __future__ import annotations
 
 import io
 import logging
+import time
 from typing import TYPE_CHECKING
 
+from celery import current_task
 from PIL import Image
 
 logger = logging.getLogger(__name__)
@@ -28,6 +30,109 @@ logger = logging.getLogger(__name__)
 # Time to wait after scrolling for content to settle and load (in milliseconds)
 SCROLL_SETTLE_TIMEOUT_MS = 1000
 
+# Fallback wall-clock budget, in seconds, for the entire tiled-screenshot
+# operation (element lookup plus all per-tile spinner/animation waits
+# combined), used when the budget can't be derived from the currently
+# running Celery task's own time limit (see _resolve_wait_budget_seconds).
+# Each tile's wait is capped at whatever remains of the budget so a slow
+# dashboard degrades gracefully instead of running past the task's time
+# limit and getting SIGKILLed mid-capture.
+#
+# A static config value isn't a reliable substitute for runtime derivation:
+# superset/tasks/scheduler.py sets a report's limit per-schedule via
+# `apply_async(time_limit=..., soft_time_limit=...)`, and per-task
+# `task_annotations` (e.g. an operator giving thumbnail tasks a much
+# shorter limit than reports) are Celery worker configuration invisible to
+# any Superset config key -- only the running task itself knows its
+# effective limit.
+#
+# Production has observed a Celery hard task_time_limit of 1740s (29 min)
+# for report execution (2026-07-13 incident: a tiled screenshot was killed
+# mid-capture with SoftTimeLimitExceeded). This fallback leaves a 300s
+# margin under that ceiling for the rest of the pipeline that runs after
+# tiling completes: combining tiles into one image, building the PDF, and
+# uploading/delivering the notification. It applies when there's no Celery
+# task context at all (e.g. synchronous thumbnail generation) or the task
+# exposes no usable limit.
+TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS = 1440  # 1740s limit - 300s margin
+
+# Safety margin taken off a runtime-derived budget: min(this cap, this
+# fraction of the task's own limit). The 300s cap matches the fallback
+# budget's margin for large limits (e.g. the 1740s report limit); the
+# fraction scales the margin down for small limits (e.g. a 120s thumbnail
+# task limit) so it doesn't eat the whole budget or push it negative.
+TILED_SCREENSHOT_BUDGET_MARGIN_FRACTION = 0.2
+TILED_SCREENSHOT_BUDGET_MAX_MARGIN_SECONDS = 300
+
+# Floor for a runtime-derived budget. Guards against a pathologically small
+# task limit (well under a minute) yielding a near-zero or negative budget
+# that would abort before capturing a single tile; a small positive budget
+# is still capped by -- and will still be killed by -- the task's actual
+# limit if it's smaller than this floor, but at least gives the tile loop a
+# chance to capture what it can before that happens.
+TILED_SCREENSHOT_MIN_BUDGET_SECONDS = 30
+
+
+class TiledScreenshotBudgetExceededError(RuntimeError):
+    """Raised when the tiled-screenshot time budget runs out mid-capture."""
+
+
+def _resolve_wait_budget_seconds(log_context: str | None = None) -> float:
+    """
+    Derive the tiled-screenshot time budget from the currently running
+    Celery task's own soft/hard time limit, if one is running and exposes
+    one. This reflects per-task overrides (e.g. task_annotations giving
+    thumbnail tasks a shorter limit than reports) that no static config
+    value can see, since those overrides only exist at the Celery worker/
+    runtime level.
+
+    Falls back to TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS if there's no
+    task context, the task exposes no usable limit, or anything goes wrong
+    while inspecting it -- this must never be able to break a screenshot.
+    """
+    context_suffix = f" [{log_context}]" if log_context else ""
+    try:
+        if current_task:
+            soft_limit, hard_limit = current_task.request.timelimit or (
+                None,
+                None,
+            )
+            limit = soft_limit or hard_limit
+            if limit:
+                margin = min(
+                    TILED_SCREENSHOT_BUDGET_MAX_MARGIN_SECONDS,
+                    limit * TILED_SCREENSHOT_BUDGET_MARGIN_FRACTION,
+                )
+                budget = max(TILED_SCREENSHOT_MIN_BUDGET_SECONDS, limit - margin)
+                logger.info(
+                    "Tiled screenshot budget derived from Celery task %s=%.1fs: "
+                    "%.1fs (margin=%.1fs).%s",
+                    "soft_time_limit" if soft_limit else "time_limit",
+                    limit,
+                    budget,
+                    margin,
+                    context_suffix,
+                )
+                return budget
+    except Exception:
+        logger.debug(
+            "Failed to derive tiled screenshot budget from the Celery task "
+            "context; using fallback budget of %ss.%s",
+            TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS,
+            context_suffix,
+            exc_info=True,
+        )
+        return TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
+
+    logger.debug(
+        "No usable Celery task time limit found; using fallback tiled "
+        "screenshot budget of %ss.%s",
+        TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS,
+        context_suffix,
+    )
+    return TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
+
+
 try:
     from playwright.sync_api import TimeoutError as PlaywrightTimeout
 except ImportError:
@@ -40,12 +145,16 @@ if TYPE_CHECKING:
         Page = None
 
 
-def combine_screenshot_tiles(screenshot_tiles: list[bytes]) -> bytes:
+def combine_screenshot_tiles(
+    screenshot_tiles: list[bytes], log_context: str | None = None
+) -> bytes:
     """
     Combine multiple screenshot tiles into a single vertical image.
 
     Args:
         screenshot_tiles: List of screenshot bytes in PNG format
+        log_context: Optional identifier (e.g. report execution id, or a
+            thumbnail cache key) appended to log lines for tracing.
 
     Returns:
         Combined screenshot as bytes
@@ -56,6 +165,7 @@ def combine_screenshot_tiles(screenshot_tiles: list[bytes]) -> bytes:
     if len(screenshot_tiles) == 1:
         return screenshot_tiles[0]
 
+    context_suffix = f" [{log_context}]" if log_context else ""
     try:
         # Open all images
         images = [Image.open(io.BytesIO(tile)) for tile in screenshot_tiles]
@@ -79,7 +189,7 @@ def combine_screenshot_tiles(screenshot_tiles: list[bytes]) -> bytes:
         return output.getvalue()
 
     except Exception as e:
-        logger.exception("Failed to combine screenshot tiles: %s", e)
+        logger.exception("Failed to combine screenshot tiles: %s%s", e, context_suffix)
         # Return the first tile as fallback
         return screenshot_tiles[0]
 
@@ -90,6 +200,7 @@ def take_tiled_screenshot(
     tile_height: int,
     load_wait: int = 60,
     animation_wait: int = 0,
+    log_context: str | None = None,
 ) -> bytes | None:
     """
     Take a tiled screenshot of a large dashboard by scrolling and capturing sections.
@@ -100,10 +211,22 @@ def take_tiled_screenshot(
         tile_height: Height of each tile in pixels
         load_wait: Seconds to wait for charts to load per tile (default 60)
         animation_wait: Seconds to wait for chart animations per tile (default 0)
+        log_context: Optional identifier (e.g. report execution id, or a
+            thumbnail cache key) appended to log lines so a slow/timed-out
+            capture can be traced back to the run that produced it.
 
     Returns:
         Combined screenshot bytes or None if failed
+
+    Raises:
+        TiledScreenshotBudgetExceededError: If the total time budget for the
+            tiled-screenshot operation runs out before every tile has been
+            verifiably captured. Callers must treat this as a hard failure
+            rather than fall back to an unchecked/partial screenshot.
     """
+    context_suffix = f" [{log_context}]" if log_context else ""
+    wait_budget_seconds = _resolve_wait_budget_seconds(log_context)
+    start_time = time.monotonic()
     try:
         # Get the target element
         element = page.locator(f".{element_name}")
@@ -127,32 +250,72 @@ def take_tiled_screenshot(
         dashboard_top = element_info["top"]
 
         logger.info(
-            "Dashboard: %sx%spx at (%s, %s)",
+            "Dashboard: %sx%spx at (%s, %s)%s",
             dashboard_width,
             dashboard_height,
             dashboard_left,
             dashboard_top,
+            context_suffix,
         )
 
         # Calculate number of tiles needed
         num_tiles = max(1, (dashboard_height + tile_height - 1) // tile_height)
-        logger.info("Taking %s screenshot tiles", num_tiles)
+        logger.info("Taking %s screenshot tiles%s", num_tiles, context_suffix)
 
-        screenshot_tiles = []
+        screenshot_tiles: list[bytes] = []
 
         for i in range(num_tiles):
+            # Check the time budget before starting this tile's readiness wait.
+            # If it's already exhausted, we can no longer verify this (or any
+            # later) tile is actually ready to capture -- fail loudly instead
+            # of silently snapshotting a spinner or blank chart, or running
+            # past the Celery task time limit and getting SIGKILLed.
+            tile_start = time.monotonic()
+            elapsed = tile_start - start_time
+            remaining_budget = wait_budget_seconds - elapsed
+            if remaining_budget <= 0:
+                # A customer-side chart-loading issue (a slow/hung dashboard),
+                # not a Superset system fault, so this is a WARNING rather
+                # than an ERROR -- consistent with #38130/#38441, which
+                # deliberately downgraded screenshot timeout logs the same way.
+                logger.warning(
+                    "Tiled screenshot time budget exhausted on tile %s/%s: "
+                    "%s/%s tiles captured so far, %.1fs elapsed of a %.1fs "
+                    "budget. Aborting instead of capturing remaining tiles "
+                    "unchecked.%s",
+                    i + 1,
+                    num_tiles,
+                    len(screenshot_tiles),
+                    num_tiles,
+                    elapsed,
+                    wait_budget_seconds,
+                    context_suffix,
+                )
+                raise TiledScreenshotBudgetExceededError(
+                    f"Tiled screenshot budget of "
+                    f"{wait_budget_seconds:.1f}s exhausted "
+                    f"after {len(screenshot_tiles)}/{num_tiles} tiles"
+                )
+
             # Calculate scroll position to show this tile's content
             scroll_y = dashboard_top + (i * tile_height)
 
             page.evaluate(f"window.scrollTo(0, {scroll_y})")
             logger.debug(
-                "Scrolled window to %s for tile %s/%s", scroll_y, i + 1, num_tiles
+                "Scrolled window to %s for tile %s/%s%s",
+                scroll_y,
+                i + 1,
+                num_tiles,
+                context_suffix,
             )
             # Wait for scroll to settle and content to load
             page.wait_for_timeout(SCROLL_SETTLE_TIMEOUT_MS)
-            # Wait for any loading spinners visible in the current viewport to clear.
+            # Wait for any loading spinners visible in the current viewport to clear,
+            # capped at whatever remains of the total time budget so a slow
+            # dashboard degrades gracefully instead of exceeding it.
             # Only check viewport-visible spinners to avoid blocking on
             # virtualization placeholders rendered for off-screen charts.
+            tile_load_wait = min(load_wait, remaining_budget)
             try:
                 page.wait_for_function(
                     """() => {
@@ -165,22 +328,56 @@ def take_tiled_screenshot(
                         }
                         return true;
                     }""",
-                    timeout=load_wait * 1000,
+                    timeout=tile_load_wait * 1000,
                 )
             except PlaywrightTimeout:
+                # Customer chart-loading timeout, not a system fault -- WARNING,
+                # matching #38130/#38441 (see the budget-exhaustion log above).
                 logger.warning(
-                    "Timed out waiting for visible spinners to clear on tile %s/%s "
-                    "(load_wait=%ss)",
+                    "Timed out waiting for visible spinners to clear on tile "
+                    "%s/%s (waited %.1fs of a %ss requested load_wait; %.1fs "
+                    "elapsed of a %.1fs total budget; %s/%s tiles captured so "
+                    "far).%s",
                     i + 1,
                     num_tiles,
+                    tile_load_wait,
                     load_wait,
+                    elapsed,
+                    wait_budget_seconds,
+                    len(screenshot_tiles),
+                    num_tiles,
+                    context_suffix,
                 )
+            spinner_wait_elapsed = time.monotonic() - tile_start
 
             # Wait for chart animations (e.g. ECharts) to finish after spinner clears.
             # The global animation wait before tiling only covers the first tile;
-            # subsequent tiles need their own wait after data loads.
+            # subsequent tiles need their own wait after data loads. Capped at
+            # whatever remains of the budget; unlike the spinner wait above this
+            # is cosmetic settling, not a readiness check, so we simply skip it
+            # (rather than raise) once the budget runs out.
+            animation_wait_elapsed = 0.0
             if animation_wait > 0:
-                page.wait_for_timeout(animation_wait * 1000)
+                elapsed = time.monotonic() - start_time
+                remaining_budget = wait_budget_seconds - elapsed
+                tile_animation_wait = max(0, min(animation_wait, remaining_budget))
+                if tile_animation_wait > 0:
+                    animation_wait_start = time.monotonic()
+                    page.wait_for_timeout(tile_animation_wait * 1000)
+                    animation_wait_elapsed = time.monotonic() - animation_wait_start
+
+            # Per-tile timing breakdown so slow dashboards can be profiled from
+            # logs alone. DEBUG rather than INFO: this fires once per tile, and
+            # large dashboards can have dozens of tiles per report run.
+            logger.debug(
+                "Tile %s/%s timing: %.2fs waiting for spinners to clear, "
+                "%.2fs waiting for animations.%s",
+                i + 1,
+                num_tiles,
+                spinner_wait_elapsed,
+                animation_wait_elapsed,
+                context_suffix,
+            )
 
             # Calculate what portion of the element we want to capture for this tile
             tile_start_in_element = i * tile_height
@@ -199,13 +396,14 @@ def take_tiled_screenshot(
                 logger.warning(
                     "Skipping tile %s/%s due to invalid clip dimensions: "
                     "x=%s, y=%s, width=%s, height=%s "
-                    "(element may be scrolled out of viewport)",
+                    "(element may be scrolled out of viewport).%s",
                     i + 1,
                     num_tiles,
                     clip_x,
                     clip_y,
                     dashboard_width,
                     clip_height,
+                    context_suffix,
                 )
                 continue
 
@@ -221,14 +419,27 @@ def take_tiled_screenshot(
             tile_screenshot = page.screenshot(type="png", clip=clip)
             screenshot_tiles.append(tile_screenshot)
 
-            logger.debug("Captured tile %s/%s with clip %s", i + 1, num_tiles, clip)
+            logger.debug(
+                "Captured tile %s/%s with clip %s%s",
+                i + 1,
+                num_tiles,
+                clip,
+                context_suffix,
+            )
 
         # Combine all tiles
-        logger.info("Combining screenshot tiles...")
-        combined_screenshot = combine_screenshot_tiles(screenshot_tiles)
+        logger.info("Combining screenshot tiles...%s", context_suffix)
+        combined_screenshot = combine_screenshot_tiles(
+            screenshot_tiles, log_context=log_context
+        )
 
         return combined_screenshot
 
+    except TiledScreenshotBudgetExceededError:
+        # Budget exhaustion must fail cleanly, not be swallowed into a
+        # `return None` (which upstream callers treat as "fall back to a
+        # standard, unchecked screenshot").
+        raise
     except Exception as e:
-        logger.exception("Tiled screenshot failed: %s", e)
+        logger.exception("Tiled screenshot failed: %s%s", e, context_suffix)
         return None

superset/utils/screenshots.py

@@ -190,7 +190,10 @@ class BaseScreenshot:
         self.screenshot = None
 
     def driver(
-        self, window_size: WindowSize | None = None, user: User | None = None
+        self,
+        window_size: WindowSize | None = None,
+        user: User | None = None,
+        log_context: str | None = None,
     ) -> WebDriverProxy:
         window_size = window_size or self.window_size
         if feature_flag_manager.is_feature_enabled("PLAYWRIGHT_REPORTS_AND_THUMBNAILS"):
@@ -199,22 +202,29 @@ class BaseScreenshot:
                 return WebDriverPlaywright(self.driver_type, window_size)
 
             # Playwright not available, falling back to Selenium
+            context_suffix = f" [{log_context}]" if log_context else ""
             logger.info(
                 "PLAYWRIGHT_REPORTS_AND_THUMBNAILS enabled but Playwright not "
                 "installed. Falling back to Selenium (WebGL/Canvas charts may "
-                "not render correctly). %s",
+                "not render correctly). %s%s",
                 PLAYWRIGHT_INSTALL_MESSAGE,
+                context_suffix,
             )
 
         # Use Selenium as default/fallback
         return WebDriverSelenium(self.driver_type, window_size, user)
 
     def get_screenshot(
-        self, user: User, window_size: WindowSize | None = None
+        self,
+        user: User,
+        window_size: WindowSize | None = None,
+        log_context: str | None = None,
     ) -> bytes | None:
-        driver = self.driver(window_size, user)
+        driver = self.driver(window_size, user, log_context=log_context)
         try:
-            self.screenshot = driver.get_screenshot(self.url, self.element, user)
+            self.screenshot = driver.get_screenshot(
+                self.url, self.element, user, log_context=log_context
+            )
         finally:
             if isinstance(driver, WebDriverSelenium):
                 driver.destroy()
@@ -305,22 +315,38 @@ class BaseScreenshot:
                 image = None
                 # Assuming all sorts of things can go wrong with Selenium
                 try:
-                    logger.info("trying to generate screenshot")
+                    logger.info(
+                        "trying to generate screenshot for cache_key=%s", cache_key
+                    )
                     with event_logger.log_context(
                         f"screenshot.compute.{self.thumbnail_type}"
                     ):
-                        image = self.get_screenshot(user=user, window_size=window_size)
+                        image = self.get_screenshot(
+                            user=user,
+                            window_size=window_size,
+                            log_context=f"cache_key={cache_key}",
+                        )
                 except Exception as ex:  # pylint: disable=broad-except
                     logger.warning(
-                        "Failed at generating thumbnail %s", ex, exc_info=True
+                        "Failed at generating thumbnail for cache_key=%s: %s",
+                        cache_key,
+                        ex,
+                        exc_info=True,
                     )
                     cache_payload.error()
                 if image and window_size != thumb_size:
                     try:
-                        image = self.resize_image(image, thumb_size=thumb_size)
+                        image = self.resize_image(
+                            image,
+                            thumb_size=thumb_size,
+                            log_context=f"cache_key={cache_key}",
+                        )
                     except Exception as ex:  # pylint: disable=broad-except
                         logger.warning(
-                            "Failed at resizing thumbnail %s", ex, exc_info=True
+                            "Failed at resizing thumbnail for cache_key=%s: %s",
+                            cache_key,
+                            ex,
+                            exc_info=True,
                         )
                         cache_payload.error()
                         image = None
@@ -339,7 +365,9 @@ class BaseScreenshot:
                 logger.info("Caching thumbnail: %s", cache_key)
                 self.cache.set(cache_key, cache_payload.to_dict())
                 logger.info(
-                    "Updated thumbnail cache; Status: %s", cache_payload.get_status()
+                    "Updated thumbnail cache for %s; Status: %s",
+                    cache_key,
+                    cache_payload.get_status(),
                 )
         except LockAlreadyHeldException:
             logger.info(
@@ -354,16 +382,23 @@ class BaseScreenshot:
         output: str = "png",
         thumb_size: WindowSize | None = None,
         crop: bool = True,
+        log_context: str | None = None,
     ) -> bytes:
+        context_suffix = f" [{log_context}]" if log_context else ""
         thumb_size = thumb_size or cls.thumb_size
         img = Image.open(BytesIO(img_bytes))
-        logger.debug("Selenium image size: %s", str(img.size))
+        logger.debug("Selenium image size: %s%s", str(img.size), context_suffix)
         if crop and img.size[1] != cls.window_size[1]:
             desired_ratio = float(cls.window_size[1]) / cls.window_size[0]
             desired_width = int(img.size[0] * desired_ratio)
-            logger.debug("Cropping to: %s*%s", str(img.size[0]), str(desired_width))
+            logger.debug(
+                "Cropping to: %s*%s%s",
+                str(img.size[0]),
+                str(desired_width),
+                context_suffix,
+            )
             img = img.crop((0, 0, img.size[0], desired_width))
-        logger.debug("Resizing to %s", str(thumb_size))
+        logger.debug("Resizing to %s%s", str(thumb_size), context_suffix)
         img = img.resize(thumb_size, Image.Resampling.LANCZOS)
         new_img = BytesIO()
         if output != "png":

superset/utils/webdriver.py

@@ -203,10 +203,17 @@ class WebDriverProxy(ABC):
 
     @abstractmethod
     def get_screenshot(
-        self, url: str, element_name: str, user: User | None = None
+        self,
+        url: str,
+        element_name: str,
+        user: User | None = None,
+        log_context: str | None = None,
     ) -> bytes | None:
         """
         Run webdriver and return a screenshot
+
+        :param log_context: Optional identifier (e.g. report execution id, or
+            a thumbnail cache key) included in log lines for tracing.
         """
 
 
@@ -218,14 +225,17 @@ class WebDriverPlaywright(WebDriverProxy):
         )
 
     @staticmethod
-    def find_unexpected_errors(page: Page) -> list[str]:
+    def find_unexpected_errors(page: Page, log_context: str | None = None) -> list[str]:
         error_messages = []
+        context_suffix = f" [{log_context}]" if log_context else ""
 
         try:
             alert_divs = page.get_by_role("alert").all()
 
             logger.debug(
-                "%i alert elements have been found in the screenshot", len(alert_divs)
+                "%i alert elements have been found in the screenshot%s",
+                len(alert_divs),
+                context_suffix,
             )
 
             for alert_div in alert_divs:
@@ -255,9 +265,12 @@ class WebDriverPlaywright(WebDriverProxy):
                         [error_as_html],
                     )
                 except PlaywrightError:
-                    logger.exception("Failed to update error messages using alert_div")
+                    logger.exception(
+                        "Failed to update error messages using alert_div%s",
+                        context_suffix,
+                    )
         except PlaywrightError:
-            logger.exception("Failed to capture unexpected errors")
+            logger.exception("Failed to capture unexpected errors%s", context_suffix)
 
         return error_messages
 
@@ -269,14 +282,20 @@ class WebDriverPlaywright(WebDriverProxy):
             return element.screenshot()
 
     def get_screenshot(  # pylint: disable=too-many-locals, too-many-statements  # noqa: C901
-        self, url: str, element_name: str, user: User | None = None
+        self,
+        url: str,
+        element_name: str,
+        user: User | None = None,
+        log_context: str | None = None,
     ) -> bytes | None:
+        context_suffix = f" [{log_context}]" if log_context else ""
         if not PLAYWRIGHT_AVAILABLE:
             logger.info(
                 "Playwright not available - falling back to Selenium. "
                 "Note: WebGL/Canvas charts may not render correctly with Selenium. "
-                "%s",
+                "%s%s",
                 PLAYWRIGHT_INSTALL_MESSAGE,
+                context_suffix,
             )
             return None
 
@@ -306,50 +325,66 @@ class WebDriverPlaywright(WebDriverProxy):
                 )
             except PlaywrightTimeout:
                 logger.exception(
-                    "Web event %s not detected. Page %s might not have been fully loaded",  # noqa: E501
+                    "Web event %s not detected. Page %s might not have been fully loaded%s",  # noqa: E501
                     app.config["SCREENSHOT_PLAYWRIGHT_WAIT_EVENT"],
                     url,
+                    context_suffix,
                 )
 
             selenium_headstart = app.config["SCREENSHOT_SELENIUM_HEADSTART"]
-            logger.debug("Sleeping for %i seconds", selenium_headstart)
+            logger.debug(
+                "Sleeping for %i seconds%s", selenium_headstart, context_suffix
+            )
             page.wait_for_timeout(selenium_headstart * 1000)
             element: Locator
             try:
                 try:
                     # page didn't load
                     logger.debug(
-                        "Wait for the presence of %s at url: %s", element_name, url
+                        "Wait for the presence of %s at url: %s%s",
+                        element_name,
+                        url,
+                        context_suffix,
                     )
                     element = page.locator(f".{element_name}")
                     element.wait_for()
                 except PlaywrightTimeout:
-                    logger.exception("Timed out requesting url %s", url)
+                    logger.exception(
+                        "Timed out requesting url %s%s", url, context_suffix
+                    )
                     raise
 
                 try:
                     # chart containers didn't render
-                    logger.debug("Wait for chart containers to draw at url: %s", url)
+                    logger.debug(
+                        "Wait for chart containers to draw at url: %s%s",
+                        url,
+                        context_suffix,
+                    )
                     slice_container_locator = page.locator(".chart-container")
                     for slice_container_elem in slice_container_locator.all():
                         slice_container_elem.wait_for()
                 except PlaywrightTimeout:
                     logger.exception(
-                        "Timed out waiting for chart containers to draw at url %s",
+                        "Timed out waiting for chart containers to draw at url %s%s",
                         url,
+                        context_suffix,
                     )
                     raise
                 selenium_animation_wait = app.config[
                     "SCREENSHOT_SELENIUM_ANIMATION_WAIT"
                 ]
                 if app.config["SCREENSHOT_REPLACE_UNEXPECTED_ERRORS"]:
-                    unexpected_errors = WebDriverPlaywright.find_unexpected_errors(page)
+                    unexpected_errors = WebDriverPlaywright.find_unexpected_errors(
+                        page, log_context=log_context
+                    )
                     if unexpected_errors:
                         logger.warning(
-                            "%i errors found in the screenshot. URL: %s. Errors are: %s",  # noqa: E501
+                            "%i errors found in the screenshot. URL: %s. Errors are: %s%s",  # noqa: E501
                             len(unexpected_errors),
                             url,
                             unexpected_errors,
+                            context_suffix,
                         )
                 # Detect large dashboards and use tiled screenshots if enabled
                 tiled_enabled = app.config.get("SCREENSHOT_TILED_ENABLED", False)
@@ -377,9 +412,11 @@ class WebDriverPlaywright(WebDriverProxy):
                     if dashboard_height == 0:
                         logger.warning(
                             "Could not determine dashboard height for element %s "
-                            "at url %s; falling back to standard screenshot behavior",
+                            "at url %s; falling back to standard screenshot "
+                            "behavior.%s",
                             element_name,
                             url,
+                            context_suffix,
                         )
 
                     # Use tiled screenshots for large dashboards
@@ -391,9 +428,10 @@ class WebDriverPlaywright(WebDriverProxy):
                     if use_tiled:
                         logger.info(
                             "Large dashboard detected: %s charts, %spx height. "
-                            "Using tiled screenshots.",
+                            "Using tiled screenshots.%s",
                             chart_count,
                             dashboard_height,
+                            context_suffix,
                         )
                         # set viewport height to tile height for easier calculations
                         page.set_viewport_size(
@@ -405,30 +443,32 @@ class WebDriverPlaywright(WebDriverProxy):
                             tile_height,
                             load_wait=self._screenshot_load_wait,
                             animation_wait=selenium_animation_wait,
+                            log_context=log_context,
                         )
                         if not img:
                             logger.warning(
-                                (
-                                    "Tiled screenshot failed, "
-                                    "falling back to standard screenshot"
-                                )
+                                "Tiled screenshot failed, falling back to "
+                                "standard screenshot.%s",
+                                context_suffix,
                             )
                             img = WebDriverPlaywright._get_screenshot(
                                 page, element, element_name
                             )
                         logger.debug(
-                            "Tiled screenshot result: %d bytes for url: %s",
+                            "Tiled screenshot result: %d bytes for url: %s%s",
                             len(img) if img else 0,
                             url,
+                            context_suffix,
                         )
                     else:
                         logger.debug(
                             "Dashboard below tiling threshold "
                             "(%s charts, %spx height); using standard screenshot "
-                            "for url: %s",
+                            "for url: %s%s",
                             chart_count,
                             dashboard_height,
                             url,
+                            context_suffix,
                         )
                         # Standard screenshot captures the full element including
                         # below-the-fold content, so wait for all spinners globally.
@@ -455,28 +495,32 @@ class WebDriverPlaywright(WebDriverProxy):
                         logger.debug("All spinners cleared for url: %s", url)
                         if selenium_animation_wait > 0:
                             logger.debug(
-                                "Wait %i seconds for chart animation",
+                                "Wait %i seconds for chart animation%s",
                                 selenium_animation_wait,
+                                context_suffix,
                             )
                             page.wait_for_timeout(selenium_animation_wait * 1000)
                         logger.debug(
-                            "Taking screenshot of url %s as user %s",
+                            "Taking screenshot of url %s as user %s%s",
                             url,
                             user.username if user else "None",
+                            context_suffix,
                         )
                         img = WebDriverPlaywright._get_screenshot(
                             page, element, element_name
                         )
                         logger.debug(
-                            "Screenshot result: %d bytes for url: %s",
+                            "Screenshot result: %d bytes for url: %s%s",
                             len(img) if img else 0,
                             url,
+                            context_suffix,
                         )
                 else:
                     logger.debug(
                         "Tiled screenshots disabled; using standard screenshot "
-                        "for url: %s",
+                        "for url: %s%s",
                         url,
+                        context_suffix,
                     )
                     # Standard screenshot captures the full element including
                     # below-the-fold content, so wait for all spinners globally.
@@ -502,29 +546,34 @@ class WebDriverPlaywright(WebDriverProxy):
                     logger.debug("All spinners cleared for url: %s", url)
                     if selenium_animation_wait > 0:
                         logger.debug(
-                            "Wait %i seconds for chart animation",
+                            "Wait %i seconds for chart animation%s",
                             selenium_animation_wait,
+                            context_suffix,
                         )
                         page.wait_for_timeout(selenium_animation_wait * 1000)
                     logger.debug(
-                        "Taking screenshot of url %s as user %s",
+                        "Taking screenshot of url %s as user %s%s",
                         url,
                         user.username if user else "None",
+                        context_suffix,
                     )
                     img = WebDriverPlaywright._get_screenshot(
                         page, element, element_name
                     )
                     logger.debug(
-                        "Screenshot result: %d bytes for url: %s",
+                        "Screenshot result: %d bytes for url: %s%s",
                         len(img) if img else 0,
                         url,
+                        context_suffix,
                     )
 
             except PlaywrightTimeout:
                 raise
             except PlaywrightError:
                 logger.exception(
-                    "Encountered an unexpected error when requesting url %s", url
+                    "Encountered an unexpected error when requesting url %s%s",
+                    url,
+                    context_suffix,
                 )
         finally:
             context.close()
@@ -728,13 +777,18 @@ class WebDriverSelenium(WebDriverProxy):
         self._driver = None
 
     @staticmethod
-    def find_unexpected_errors(driver: WebDriver) -> list[str]:
+    def find_unexpected_errors(
+        driver: WebDriver, log_context: str | None = None
+    ) -> list[str]:
         error_messages = []
+        context_suffix = f" [{log_context}]" if log_context else ""
 
         try:
             alert_divs = driver.find_elements(By.XPATH, "//div[@role = 'alert']")
             logger.debug(
-                "%i alert elements have been found in the screenshot", len(alert_divs)
+                "%i alert elements have been found in the screenshot%s",
+                len(alert_divs),
+                context_suffix,
             )
 
             for alert_div in alert_divs:
@@ -777,15 +831,26 @@ class WebDriverSelenium(WebDriverProxy):
                         f"arguments[0].innerHTML = '{error_as_html}'", alert_div
                     )
                 except WebDriverException:
-                    logger.exception("Failed to update error messages using alert_div")
+                    logger.exception(
+                        "Failed to update error messages using alert_div%s",
+                        context_suffix,
+                    )
         except WebDriverException:
-            logger.exception("Failed to capture unexpected errors")
+            logger.exception("Failed to capture unexpected errors%s", context_suffix)
 
         return error_messages
 
     def get_screenshot(  # noqa: C901
-        self, url: str, element_name: str, user: User | None = None
+        self,
+        url: str,
+        element_name: str,
+        user: User | None = None,
+        log_context: str | None = None,
     ) -> bytes | None:
+        # Selenium doesn't take tiled screenshots (only WebDriverPlaywright
+        # does), but log_context is still used below to trace the rest of
+        # this method's logging back to the run that triggered it.
+        context_suffix = f" [{log_context}]" if log_context else ""
         # If a user is passed explicitly and differs from the stored user,
         # update and re-authenticate
         if user and user != self._user:
@@ -796,7 +861,7 @@ class WebDriverSelenium(WebDriverProxy):
         driver.get(url)
         img: bytes | None = None
         selenium_headstart = app.config["SCREENSHOT_SELENIUM_HEADSTART"]
-        logger.debug("Sleeping for %i seconds", selenium_headstart)
+        logger.debug("Sleeping for %i seconds%s", selenium_headstart, context_suffix)
         sleep(selenium_headstart)
 
         # WebDriver cleanup is intentionally not performed in this method. When the
@@ -807,27 +872,37 @@ class WebDriverSelenium(WebDriverProxy):
             try:
                 # page didn't load
                 logger.debug(
-                    "Wait for the presence of %s at url: %s", element_name, url
+                    "Wait for the presence of %s at url: %s%s",
+                    element_name,
+                    url,
+                    context_suffix,
                 )
                 element = WebDriverWait(driver, self._screenshot_locate_wait).until(
                     EC.presence_of_element_located((By.CLASS_NAME, element_name))
                 )
             except TimeoutException:
                 logger.warning(
-                    "Selenium timed out requesting url %s", url, exc_info=True
+                    "Selenium timed out requesting url %s%s",
+                    url,
+                    context_suffix,
+                    exc_info=True,
                 )
                 raise
 
             try:
                 # chart containers didn't render
-                logger.debug("Wait for chart containers to draw at url: %s", url)
+                logger.debug(
+                    "Wait for chart containers to draw at url: %s%s",
+                    url,
+                    context_suffix,
+                )
                 WebDriverWait(driver, self._screenshot_locate_wait).until(
                     EC.visibility_of_all_elements_located(
                         (By.CLASS_NAME, "chart-container")
                     )
                 )
             except TimeoutException:
-                logger.info("Timeout Exception caught")
+                logger.info("Timeout Exception caught%s", context_suffix)
                 # Fallback to allow a screenshot of an empty dashboard
                 try:
                     WebDriverWait(driver, 0).until(
@@ -837,8 +912,9 @@ class WebDriverSelenium(WebDriverProxy):
                     )
                 except Exception:
                     logger.warning(
-                        "Selenium timed out waiting for dashboard to draw at url %s",
+                        "Selenium timed out waiting for dashboard to draw at url %s%s",
                         url,
+                        context_suffix,
                         exc_info=True,
                     )
                     raise
@@ -860,22 +936,30 @@ class WebDriverSelenium(WebDriverProxy):
                 raise
 
             selenium_animation_wait = app.config["SCREENSHOT_SELENIUM_ANIMATION_WAIT"]
-            logger.debug("Wait %i seconds for chart animation", selenium_animation_wait)
+            logger.debug(
+                "Wait %i seconds for chart animation%s",
+                selenium_animation_wait,
+                context_suffix,
+            )
             sleep(selenium_animation_wait)
             logger.debug(
-                "Taking a PNG screenshot of url %s as user %s",
+                "Taking a PNG screenshot of url %s as user %s%s",
                 url,
                 self._user.username if self._user else "None",
+                context_suffix,
             )
 
             if app.config["SCREENSHOT_REPLACE_UNEXPECTED_ERRORS"]:
-                unexpected_errors = WebDriverSelenium.find_unexpected_errors(driver)
+                unexpected_errors = WebDriverSelenium.find_unexpected_errors(
+                    driver, log_context=log_context
+                )
                 if unexpected_errors:
                     logger.warning(
-                        "%i errors found in the screenshot. URL: %s. Errors are: %s",
+                        "%i errors found in the screenshot. URL: %s. Errors are: %s%s",
                         len(unexpected_errors),
                         url,
                         unexpected_errors,
+                        context_suffix,
                     )
 
             img = element.screenshot_as_png
@@ -884,19 +968,21 @@ class WebDriverSelenium(WebDriverProxy):
             raise
         except StaleElementReferenceException:
             logger.warning(
-                "Selenium got a stale element while requesting url %s",
+                "Selenium got a stale element while requesting url %s%s",
                 url,
+                context_suffix,
                 exc_info=True,
             )
             raise
         except WebDriverException:
             logger.warning(
-                "Encountered an unexpected error when requesting url %s",
+                "Encountered an unexpected error when requesting url %s%s",
                 url,
+                context_suffix,
                 exc_info=True,
             )
             raise
         except Exception as ex:
-            logger.warning("exception in webdriver", exc_info=ex)
+            logger.warning("exception in webdriver%s", context_suffix, exc_info=ex)
             raise
         return img

tests/unit_tests/utils/test_screenshot_utils.py

@@ -22,9 +22,12 @@ import pytest
 from PIL import Image
 
 from superset.utils.screenshot_utils import (
+    _resolve_wait_budget_seconds,
     combine_screenshot_tiles,
     SCROLL_SETTLE_TIMEOUT_MS,
     take_tiled_screenshot,
+    TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS,
+    TiledScreenshotBudgetExceededError,
 )
 
 
@@ -183,10 +186,10 @@ class TestTakeTiledScreenshot:
 
                 # Should log dashboard dimensions with lazy logging format
                 mock_logger.info.assert_any_call(
-                    "Dashboard: %sx%spx at (%s, %s)", 800, 5000, 50, 100
+                    "Dashboard: %sx%spx at (%s, %s)%s", 800, 5000, 50, 100, ""
                 )
                 # Should log number of tiles with lazy logging format
-                mock_logger.info.assert_any_call("Taking %s screenshot tiles", 3)
+                mock_logger.info.assert_any_call("Taking %s screenshot tiles%s", 3, "")
 
     def test_exception_handling_returns_none(self):
         """Test that exceptions are handled and None is returned."""
@@ -199,8 +202,9 @@ class TestTakeTiledScreenshot:
             assert result is None
             # The exception object is passed, not the string
             call_args = mock_logger.exception.call_args
-            assert call_args[0][0] == "Tiled screenshot failed: %s"
+            assert call_args[0][0] == "Tiled screenshot failed: %s%s"
             assert str(call_args[0][1]) == "Unexpected error"
+            assert call_args[0][2] == ""
 
     def test_screenshot_clip_parameters(self, mock_page):
         """Test that screenshot clipping parameters are correct."""
@@ -342,25 +346,36 @@ class TestTakeTiledScreenshot:
         """A per-tile spinner timeout logs a warning but still takes the screenshot."""
         from superset.utils.screenshot_utils import PlaywrightTimeout
 
-        timeout = PlaywrightTimeout()
+        timeout = PlaywrightTimeout("mocked timeout")
         mock_page.wait_for_function.side_effect = timeout
 
-        with patch("superset.utils.screenshot_utils.logger") as mock_logger:
-            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
-                result = take_tiled_screenshot(
-                    mock_page, "dashboard", tile_height=2000, load_wait=30
-                )
+        with patch("superset.utils.screenshot_utils.time.monotonic", return_value=0):
+            with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    result = take_tiled_screenshot(
+                        mock_page, "dashboard", tile_height=2000, load_wait=30
+                    )
 
         # Screenshot should still proceed (non-fatal)
         assert result is not None
-        # Warning logged for each tile that timed out
+        # Warning (not error/exception) logged for each tile that timed out --
+        # this is a customer chart-loading issue, not a Superset system fault.
         assert mock_logger.warning.call_count == 3
+        assert mock_logger.error.call_count == 0
         mock_logger.warning.assert_any_call(
-            "Timed out waiting for visible spinners to clear on tile %s/%s "
-            "(load_wait=%ss)",
+            "Timed out waiting for visible spinners to clear on tile "
+            "%s/%s (waited %.1fs of a %ss requested load_wait; %.1fs "
+            "elapsed of a %.1fs total budget; %s/%s tiles captured so "
+            "far).%s",
             1,
             3,
             30,
+            30,
+            0.0,
+            1440,
+            0,
+            3,
+            "",
         )
 
     def test_load_wait_default_is_sixty_seconds(self):
@@ -397,3 +412,259 @@ class TestTakeTiledScreenshot:
 
         sig = inspect.signature(take_tiled_screenshot)
         assert sig.parameters["animation_wait"].default == 0
+
+
+class TestTileWaitBudget:
+    @pytest.fixture
+    def mock_page(self):
+        """Create a mock Playwright page object for a 3-tile (5000px) dashboard."""
+        page = MagicMock()
+        element = MagicMock()
+        page.locator.return_value = element
+        page.evaluate.return_value = {
+            "height": 5000,
+            "top": 100,
+            "left": 50,
+            "width": 800,
+        }
+        page.screenshot.return_value = b"fake_screenshot_data"
+        return page
+
+    def test_per_tile_wait_shrinks_as_budget_depletes(self, mock_page, monkeypatch):
+        """Each tile's spinner-wait timeout is capped at the remaining budget."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        # monotonic() is called: once for start_time, then per tile once to
+        # compute elapsed/remaining budget (before the spinner wait) and once
+        # more right after the spinner wait (for the per-tile timing line).
+        monotonic_values = iter([0, 0, 0, 950, 950, 990, 990])
+        with patch(
+            "superset.utils.screenshot_utils.time.monotonic",
+            side_effect=lambda: next(monotonic_values),
+        ):
+            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                result = take_tiled_screenshot(
+                    mock_page, "dashboard", tile_height=2000, load_wait=100
+                )
+
+        assert result is not None
+        timeouts = [
+            call[1]["timeout"] for call in mock_page.wait_for_function.call_args_list
+        ]
+        # remaining budget: 1000, 50, 10 seconds -> capped timeouts shrink
+        assert timeouts == [100 * 1000, 50 * 1000, 10 * 1000]
+        assert timeouts == sorted(timeouts, reverse=True)
+
+    def test_budget_exhausted_raises_and_stops_capturing(self, mock_page, monkeypatch):
+        """Exhausting the budget aborts cleanly instead of capturing unchecked."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        # start_time=0, tile 0: elapsed=0 (proceeds, captures, then a
+        # post-spinner-wait timestamp for the timing line), tile 1 check:
+        # elapsed=1000 -> remaining=0 -> raise before capturing.
+        monotonic_values = iter([0, 0, 0, 1000])
+        with patch(
+            "superset.utils.screenshot_utils.time.monotonic",
+            side_effect=lambda: next(monotonic_values),
+        ):
+            with patch(
+                "superset.utils.screenshot_utils.combine_screenshot_tiles"
+            ) as mock_combine:
+                with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                    with pytest.raises(TiledScreenshotBudgetExceededError):
+                        take_tiled_screenshot(
+                            mock_page, "dashboard", tile_height=2000, load_wait=100
+                        )
+
+        # Only the first tile was captured before the budget ran out.
+        assert mock_page.screenshot.call_count == 1
+        # Tiles were never combined -- the function raised before that point.
+        mock_combine.assert_not_called()
+
+        # Budget exhaustion is a customer chart-loading issue, not a Superset
+        # system fault, so it must log at WARNING (not ERROR) -- consistent
+        # with the #38130/#38441 precedent for screenshot timeout logging.
+        assert mock_logger.error.call_count == 0
+        mock_logger.warning.assert_called_once()
+        warning_args = mock_logger.warning.call_args[0]
+        assert "budget exhausted" in warning_args[0]
+        # tile index, tiles total, tiles captured, tiles total,
+        # elapsed seconds, budget seconds, log-context suffix
+        assert warning_args[1] == 2
+        assert warning_args[2] == 3
+        assert warning_args[3] == 1
+        assert warning_args[4] == 3
+        assert warning_args[5] == 1000
+        assert warning_args[6] == 1000
+        assert warning_args[7] == ""
+
+    def test_budget_exhausted_warning_includes_log_context(
+        self, mock_page, monkeypatch
+    ):
+        """log_context (e.g. report execution id) is appended to the warning."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        monotonic_values = iter([0, 0, 0, 1000])
+        with patch(
+            "superset.utils.screenshot_utils.time.monotonic",
+            side_effect=lambda: next(monotonic_values),
+        ):
+            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                    with pytest.raises(TiledScreenshotBudgetExceededError):
+                        take_tiled_screenshot(
+                            mock_page,
+                            "dashboard",
+                            tile_height=2000,
+                            load_wait=100,
+                            log_context="execution_id=abc-123",
+                        )
+
+        warning_args = mock_logger.warning.call_args[0]
+        assert warning_args[-1] == " [execution_id=abc-123]"
+
+    def test_per_tile_timing_debug_line_logged(self, mock_page):
+        """Each tile logs a DEBUG timing breakdown (spinner wait, animation wait)."""
+        with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                take_tiled_screenshot(
+                    mock_page,
+                    "dashboard",
+                    tile_height=2000,
+                    log_context="cache_key=xyz",
+                )
+
+        timing_calls = [
+            call for call in mock_logger.debug.call_args_list if "timing" in call[0][0]
+        ]
+        assert len(timing_calls) == 3
+        for i, call in enumerate(timing_calls):
+            args = call[0]
+            assert args[1] == i + 1  # tile index
+            assert args[2] == 3  # total tiles
+            assert args[-1] == " [cache_key=xyz]"
+
+    def test_fast_dashboard_matches_default_behavior(self, mock_page):
+        """Well under budget, waits are not capped and behavior is unchanged."""
+        with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+            result = take_tiled_screenshot(
+                mock_page,
+                "dashboard",
+                tile_height=2000,
+                load_wait=30,
+                animation_wait=5,
+            )
+
+        assert result is not None
+        assert mock_page.screenshot.call_count == 3
+
+        for call in mock_page.wait_for_function.call_args_list:
+            assert call[1]["timeout"] == 30 * 1000
+
+        animation_calls = [
+            call
+            for call in mock_page.wait_for_timeout.call_args_list
+            if call[0][0] == 5 * 1000
+        ]
+        assert len(animation_calls) == 3
+
+
+class TestResolveWaitBudgetSeconds:
+    """The budget is derived from the running Celery task's own time limit
+    when available, and falls back to the fixed constant otherwise."""
+
+    def _mock_task(self, soft=None, hard=None):
+        task = MagicMock()
+        task.request.timelimit = (soft, hard)
+        return task
+
+    def test_derives_budget_from_soft_time_limit(self):
+        """soft_time_limit is preferred over the hard time_limit when both are set."""
+        task = self._mock_task(soft=90, hard=120)
+        with patch("superset.utils.screenshot_utils.current_task", task):
+            budget = _resolve_wait_budget_seconds()
+
+        # margin = min(300, 90 * 0.2) = 18; budget = 90 - 18 = 72
+        assert budget == 72
+
+    def test_small_task_limit_yields_positive_scaled_margin_budget(self):
+        """A 120s thumbnail-task limit (superset-shell#4389) still gets a
+        usable, positive budget via the scaled-down margin, not the fixed
+        300s margin that would otherwise wipe it out."""
+        task = self._mock_task(soft=None, hard=120)
+        with patch("superset.utils.screenshot_utils.current_task", task):
+            budget = _resolve_wait_budget_seconds()
+
+        # margin = min(300, 120 * 0.2) = 24; budget = 120 - 24 = 96
+        assert budget == 96
+        assert budget > 0
+        assert budget < 120
+
+    def test_no_task_context_falls_back_to_constant(self):
+        """Outside of a Celery task, the fixed fallback budget is used."""
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            budget = _resolve_wait_budget_seconds()
+
+        assert budget == TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
+
+    def test_task_with_no_timelimit_falls_back_to_constant(self):
+        """A task with no soft or hard limit set falls back to the constant."""
+        task = self._mock_task(soft=None, hard=None)
+        with patch("superset.utils.screenshot_utils.current_task", task):
+            budget = _resolve_wait_budget_seconds()
+
+        assert budget == TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
+
+    def test_derivation_exception_falls_back_to_constant(self):
+        """Any failure while inspecting the task context must never break a
+        screenshot -- fall back to the constant and log at DEBUG."""
+
+        class _BrokenTask:
+            """Simulates a task-like object whose .request raises."""
+
+            @property
+            def request(self):
+                raise RuntimeError("boom")
+
+        with patch("superset.utils.screenshot_utils.current_task", _BrokenTask()):
+            with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                budget = _resolve_wait_budget_seconds(log_context="execution_id=abc")
+
+        assert budget == TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS
+        mock_logger.debug.assert_called_once()
+        debug_args = mock_logger.debug.call_args
+        assert "Failed to derive" in debug_args[0][0]
+        assert debug_args[1]["exc_info"] is True
+
+    def test_take_tiled_screenshot_uses_derived_budget_from_task_limit(self):
+        """take_tiled_screenshot caps waits using the task-derived budget."""
+        mock_page = MagicMock()
+        mock_page.locator.return_value = MagicMock()
+        mock_page.evaluate.return_value = {
+            "height": 5000,
+            "top": 100,
+            "left": 50,
+            "width": 800,
+        }
+        mock_page.screenshot.return_value = b"fake_screenshot_data"
+
+        task = self._mock_task(soft=None, hard=120)
+        with patch("superset.utils.screenshot_utils.current_task", task):
+            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                take_tiled_screenshot(
+                    mock_page, "dashboard", tile_height=2000, load_wait=200
+                )
+
+        # load_wait=200s requested, but the derived 96s budget caps the very
+        # first tile's wait well below that (allow a small tolerance for the
+        # real wall-clock time elapsed between deriving the budget and
+        # capping the first tile's wait).
+        first_timeout = mock_page.wait_for_function.call_args_list[0][1]["timeout"]
+        assert first_timeout == pytest.approx(96 * 1000, abs=1000)
+        assert first_timeout < 200 * 1000

tests/unit_tests/utils/webdriver_test.py

@@ -627,7 +627,7 @@ class TestWebDriverPlaywrightErrorHandling:
 
         assert result == []
         mock_logger.exception.assert_called_once_with(
-            "Failed to capture unexpected errors"
+            "Failed to capture unexpected errors%s", ""
         )
 
     @patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@@ -820,7 +820,7 @@ class TestWebDriverPlaywrightErrorHandling:
         # not installed) accepting unrelated exceptions.
         assert exc_info.value is timeout
         mock_logger.exception.assert_any_call(
-            "Timed out requesting url %s", "http://example.com"
+            "Timed out requesting url %s%s", "http://example.com", ""
         )
 
     @patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@@ -901,9 +901,10 @@ class TestWebDriverPlaywrightErrorHandling:
         assert result == b"fake_screenshot"
         mock_logger.warning.assert_any_call(
             "Could not determine dashboard height for element %s at url %s; "
-            "falling back to standard screenshot behavior",
+            "falling back to standard screenshot behavior.%s",
             "dashboard",
             "http://example.com",
+            "",
         )
 
     @patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@@ -970,7 +971,8 @@ class TestWebDriverPlaywrightErrorHandling:
         assert result == b"fallback_screenshot"
         mock_take_tiled.assert_called_once()
         mock_logger.warning.assert_any_call(
-            ("Tiled screenshot failed, falling back to standard screenshot"),
+            "Tiled screenshot failed, falling back to standard screenshot.%s",
+            "",
         )
 
 
@@ -1038,9 +1040,9 @@ class TestWebDriverPlaywrightAnimationWaitOrder:
         assert "animation_wait" in call_order
         spinner_idx = call_order.index("spinner_wait")
         anim_idx = call_order.index("animation_wait")
-        assert spinner_idx < anim_idx, (
-            "spinner wait must precede animation wait in non-tiled path"
-        )
+        assert (
+            spinner_idx < anim_idx
+        ), "spinner wait must precede animation wait in non-tiled path"
 
     @patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
     @patch("superset.utils.webdriver._browser_manager")
@@ -1121,6 +1123,7 @@ class TestWebDriverPlaywrightAnimationWaitOrder:
             600,
             load_wait=30,
             animation_wait=2,
+            log_context=None,
         )
         # The only wait_for_timeout call should be the 0ms headstart; no global
         # animation wait should be issued (handled per-tile by take_tiled_screenshot)
@@ -1129,9 +1132,9 @@ class TestWebDriverPlaywrightAnimationWaitOrder:
             for call in mock_page.wait_for_timeout.call_args_list
             if call[0][0] == 2 * 1000
         ]
-        assert animation_waits == [], (
-            "No global 2s animation wait_for_timeout should fire on the tiled path"
-        )
+        assert (
+            animation_waits == []
+        ), "No global 2s animation wait_for_timeout should fire on the tiled path"
 
     @patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
     @patch("superset.utils.webdriver._browser_manager")
@@ -1194,6 +1197,6 @@ class TestWebDriverPlaywrightAnimationWaitOrder:
         timeout_values = [
             call[0][0] for call in mock_page.wait_for_timeout.call_args_list
         ]
-        assert timeout_values == [0], (
-            f"Expected only [0] (headstart), got {timeout_values}"
-        )
+        assert timeout_values == [
+            0
+        ], f"Expected only [0] (headstart), got {timeout_values}"