Check the card before writing a print off as internal-storage-only (issue #2856)

    A print's dispatch says where the printer put the sliced file:
    ftp://<name> for external storage, brtc://emmc/<name> for internal. Since
    that there is then no file to find at any path.

    That is where the printer chose to put it, which is not the same as where
    port 990 can read it. The reporter's H2D - firmware 01.03.00.00, card in
    the slot - reports brtc://emmc and keeps the same file under /cache: his
    log has every print from 08-12 downloading from there, 19 MB included,
    until the skip landed and two days of archives came out as a name and
    nothing else. #2780's P2S and H2C really did 550 on every path, so both
    are true and the URL alone cannot tell them apart.

    So ask the printer rather than the model. The dispatch names the exact
    file, which turns the question into one connection walking five
    directories - against the sweep's ~110, which is the cost that made
    skipping worth doing. A hit archives normally and is shared with the
    cover endpoint; a miss keeps #2780's fallback archive and its reason, so
    the archives banner still explains itself. Not probed when the printer
    reports an empty slot, or while its file service is in TLS cool-off:
    both have already answered the question.

    The connection diagnostic asked the same question off the URL and warned
    that the last print was out of reach. On this reporter's printer that
    warning would have sent him to a setting that was already right, so it
    now probes too - by directory listing, since the file it is asking about
    can be tens of megabytes and the answer is a yes or a no. Capped at 6s to
    stay inside the support bundle's per-printer budget, and "could not
    check" leaves the warning standing.

    The probe filename arrives over MQTT and becomes both a remote path and a
    local temp filename, so names carrying separators, traversal or control
    characters are declined rather than cleaned.

backend/app/api/routes/printers.py

@@ -52,7 +52,7 @@ from backend.app.services.bambu_ftp import (
     get_storage_info_async,
     list_files_async,
 )
-from backend.app.services.print_storage import print_file_reachable_over_ftp
+from backend.app.services.print_storage import ftp_probe_paths, print_file_reachable_over_ftp
 from backend.app.services.printer_diagnostic import run_connection_diagnostic
 from backend.app.services.printer_manager import (
     display_temperatures,
@@ -1231,13 +1231,21 @@ async def _produce_cover_image(
     temp_path = settings.archive_dir / "temp" / f"cover_{printer_id}_{temp_filename}"
     temp_path.parent.mkdir(parents=True, exist_ok=True)
 
+    storage = print_file_reachable_over_ftp(printer_manager.get_status(printer_id))
+
     # Cache check (#972): the archive-metadata flow in main.py may have already
     # downloaded this 3MF during the print-start handler. Reusing that file
     # avoids a second 36MB transfer competing with the printer's single FTP
     # socket (which produces the 425 errors that feed the retry storm).
+    #
+    # The dispatch's own filename is a candidate too: it is what the archive
+    # flow's probe cached the file under, and it does not always survive the
+    # trip through subtask_name (#2856).
     downloaded = False
     using_cached = False
-    for candidate_name in possible_filenames:
+    for candidate_name in (*possible_filenames, storage.probe_filename):
+        if not candidate_name:
+            continue
         cached = get_cached_3mf(printer_id, candidate_name)
         if cached:
             logger.info("Cover using cached 3MF from %s (avoided duplicate FTP)", cached)
@@ -1251,21 +1259,36 @@ async def _produce_cover_image(
         # When the printer kept the print on internal storage there is nothing
         # at any of these paths, and walking all sixteen of them just to end on
         # a 404 that reads as "this print has no cover" helps nobody (#2780).
-        storage = print_file_reachable_over_ftp(printer_manager.get_status(printer_id))
+        #
+        # Unless the printer is wrong about that, which an H2D with a card in
+        # routinely is (#2856). The dispatch names the file, so when it does,
+        # trade the immediate 404 for a five-path probe of that one name — same
+        # single connection, and it is the only way this endpoint ever recovers
+        # a cover for a print the archive flow did not see start.
+        max_retries = 2
         if not storage.reachable:
-            _cover_404_cache.setdefault(printer_id, set()).add(cache_key)
-            raise HTTPException(
-                404,
-                f"The print file for '{subtask_name}' is not on storage Bambuddy can read over FTPS "
-                f"({storage.reason}), so it has no cover to extract.",
-            )
+            if not storage.probe_filename:
+                _cover_404_cache.setdefault(printer_id, set()).add(cache_key)
+                raise HTTPException(
+                    404,
+                    f"The print file for '{subtask_name}' is not on storage Bambuddy can read over FTPS "
+                    f"({storage.reason}), so it has no cover to extract.",
+                )
+            remote_paths = ftp_probe_paths(storage.probe_filename)
+            # The dispatch's name is the authoritative one — a print whose
+            # subtask_name has been normalized or truncated would otherwise be
+            # cached under a key the archive flow never looks up.
+            temp_filename = storage.probe_filename
+            temp_path = settings.archive_dir / "temp" / f"cover_{printer_id}_{temp_filename}"
+            # One look, not three: the printer has already said this file is not
+            # here, so a retry storm on top of a hunch is exactly what #2780 was.
+            max_retries = 0
 
         logger.info(
             f"Trying to download cover for '{subtask_name}' from {printer.ip_address} (trying {len(remote_paths)} paths)"
         )
 
         # Retry logic for transient FTP failures
-        max_retries = 2
         last_error = None
 
         for attempt in range(max_retries + 1):
@@ -1304,6 +1327,15 @@ async def _produce_cover_image(
             # Remember this failure so subsequent requests for the same print
             # skip the 8-path FTP fan-out (#1420).
             _cover_404_cache.setdefault(printer_id, set()).add(cache_key)
+            if not storage.reachable:
+                # The probe looked and found nothing, so the printer's own
+                # account of where the file went is the answer after all —
+                # keep saying so rather than reporting a generic miss (#2780).
+                raise HTTPException(
+                    404,
+                    f"The print file for '{subtask_name}' is not on storage Bambuddy can read over FTPS "
+                    f"({storage.reason}), so it has no cover to extract.",
+                )
             raise HTTPException(
                 404,
                 f"Could not download 3MF file for '{subtask_name}' from printer {printer.ip_address}. Tried: {possible_filenames}",

backend/app/main.py

@@ -91,6 +91,7 @@ from backend.app.services.bambu_ftp import (
     cache_3mf_download,
     clear_3mf_cache,
     download_file_async,
+    download_file_try_paths_async,
     ftps_handshake_blocked,
     get_cached_3mf,
     get_ftp_retry_settings,
@@ -109,7 +110,11 @@ from backend.app.services.notification_service import notification_service
 from backend.app.services.obico_detection import obico_detection_service
 from backend.app.services.print_cost_estimate import plate_scoped_run_estimate as _plate_scoped_run_estimate
 from backend.app.services.print_scheduler import scheduler as print_scheduler
-from backend.app.services.print_storage import external_storage_present, print_file_reachable_over_ftp
+from backend.app.services.print_storage import (
+    external_storage_present,
+    ftp_probe_paths,
+    print_file_reachable_over_ftp,
+)
 from backend.app.services.printer_manager import (
     init_printer_connections,
     parse_plate_id,
@@ -3566,17 +3571,63 @@ async def on_print_start(printer_id: int, data: dict):
         # retries, then the directory walk) is ~110 connections that cannot
         # succeed. Skip it and say why (#2780).
         storage = print_file_reachable_over_ftp(printer_manager.get_status(printer_id))
+
+        # Get FTP retry settings
+        ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings()
+
+        # ...but "the printer put it on eMMC" is where it went, not whether we
+        # can read it. An H2D with a card in mirrors the job to /cache and
+        # serves it happily, and skipping on the URL alone cost that reporter
+        # every archive for two days (#2856). So ask the printer instead of
+        # guessing: the dispatch named the exact file, which is one connection
+        # walking five paths rather than the sweep's ~110. Only when the probe
+        # comes back empty does the verdict's reason stand.
+        if not storage.reachable and not downloaded_filename and storage.probe_filename:
+            if ftps_handshake_blocked(printer.ip_address):
+                logger.debug(
+                    "Not probing for %s on printer %s: its file service is not answering over TLS",
+                    storage.probe_filename,
+                    printer_id,
+                )
+            else:
+                probe_path = app_settings.archive_dir / "temp" / storage.probe_filename
+                probe_path.parent.mkdir(parents=True, exist_ok=True)
+                try:
+                    probe_hit = await download_file_try_paths_async(
+                        printer.ip_address,
+                        printer.access_code,
+                        ftp_probe_paths(storage.probe_filename),
+                        probe_path,
+                        socket_timeout=ftp_timeout,
+                        printer_model=printer.model,
+                    )
+                except Exception as e:
+                    logger.debug("3MF probe for %s failed: %s", storage.probe_filename, e)
+                    probe_hit = False
+                if probe_hit:
+                    downloaded_filename = storage.probe_filename
+                    temp_path = probe_path
+                    cache_3mf_download(printer_id, downloaded_filename, probe_path)
+                    logger.info(
+                        "Found %s over FTPS for printer %s even though the printer reported %s",
+                        downloaded_filename,
+                        printer_id,
+                        storage.reason,
+                    )
+
         if not storage.reachable and not downloaded_filename:
+            # Same opening words whether or not a probe ran, because that is
+            # the phrase support asks people to grep for — only the tail says
+            # which of the two happened.
             logger.info(
-                "Skipping the 3MF lookup for printer %s: %s — the print file is not on storage "
-                "Bambuddy can read over FTPS, so no path would find it",
+                "Skipping the 3MF lookup for printer %s: %s — %s",
                 printer_id,
                 storage.reason,
+                "no copy of it on external storage either"
+                if storage.probe_filename
+                else "the print file is not on storage Bambuddy can read over FTPS, so no path would find it",
             )
 
-        # Get FTP retry settings
-        ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings()
-
         for try_filename in possible_names if not downloaded_filename and storage.reachable else []:
             if not try_filename.endswith(".3mf"):
                 continue

backend/app/services/bambu_ftp.py

@@ -1403,6 +1403,54 @@ async def list_files_async(
         return []
 
 
+async def find_remote_file_async(
+    ip_address: str,
+    access_code: str,
+    remote_paths: list[str],
+    timeout: float = 30.0,
+    socket_timeout: float | None = None,
+    printer_model: str | None = None,
+) -> str | None:
+    """First of *remote_paths* the printer actually has, or None.
+
+    Answers "is this file there?" without fetching it, over a single
+    connection: one listing per distinct directory, reused across the
+    candidates that share it, and stops at the first hit. Written for the
+    connection diagnostic (#2856), which needs the answer for a file that can
+    be tens of megabytes and has no use for its contents.
+
+    Listing rather than ``SIZE``: LIST is what every Bambu firmware here is
+    known to answer, and a ``SIZE`` the server simply does not implement would
+    read as "the file is missing".
+    """
+    loop = asyncio.get_event_loop()
+
+    def _find() -> str | None:
+        client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
+        if not client.connect():
+            return None
+        try:
+            listed: dict[str, set[str]] = {}
+            for remote_path in remote_paths:
+                directory, _, name = remote_path.rpartition("/")
+                directory = directory or "/"
+                if directory not in listed:
+                    listed[directory] = {
+                        entry.get("name") for entry in client.list_files(directory) if not entry.get("is_directory")
+                    }
+                if name in listed[directory]:
+                    return remote_path
+            return None
+        finally:
+            client.disconnect()
+
+    try:
+        return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _find), timeout=timeout)
+    except TimeoutError:
+        logger.warning("FTP find_remote_file timed out after %ss on %s", timeout, ip_address)
+        return None
+
+
 async def delete_file_async(
     ip_address: str,
     access_code: str,

backend/app/services/print_storage.py

@@ -7,10 +7,9 @@ FTPS on port 990. On every Bambu model that port serves **external storage only*
 H2-series and P2S firmware default to keeping the sliced file on internal eMMC
 instead, and BambuStudio uploads there over a separate service on port 6000
 (the "BambuTunnelLocal" protocol -- see #2762, which tracks implementing it).
-When that happens there is no file on FTPS to find, at any path, and no TLS
-option, retry or directory guess changes that. The dispatch says so plainly:
-the ``project_file`` command carries ``url``, which is ``ftp://<name>`` for
-external storage and ``brtc://emmc/<name>`` for internal.
+The dispatch says where it went: the ``project_file`` command carries ``url``,
+which is ``ftp://<name>`` for external storage and ``brtc://emmc/<name>`` for
+internal.
 
 Before this module we ignored ``url`` and swept anyway: six filename variants
 across five directories with up to four retries for the 3MF, then sixteen more
@@ -19,6 +18,19 @@ print, every one of them certain to 550. The user-visible result was an archive
 card with nothing on it and no stated reason, which read as a Bambuddy bug and
 was reported as one four times (#1170, #2524, #2762, #2780).
 
+But that URL is not the last word on reachability, and reading it as one was
+itself a regression (#2856). It says where the printer *chose* to put the file,
+not whether port 990 can serve it -- measured on an H2D (firmware 01.03.00.00,
+card in the slot): every ``brtc://emmc/<name>`` print of that reporter's was
+sitting under ``/cache/<name>`` and downloaded fine, 19 MB included, until this
+module started skipping the lookup. On #2780's P2S and H2C the same URL really
+did mean nothing was there. So an internal-storage URL earns a *bounded probe*
+rather than a skip: it names the exact file, which turns the 110-connection
+sweep into one connection walking five paths, and the answer comes from the
+printer instead of from a guess about its model. Only when that probe misses
+does the verdict's ``reason`` stand -- see :func:`probe_filename_from_url` and
+:func:`ftp_probe_paths`, and the callers that run it.
+
 The rule here is deliberately one-sided: **skip only on positive evidence**.
 Silence is not evidence -- a printer that never publishes ``sdcard`` and never
 had a ``project_file`` pass through the request topic (some brokers refuse the
@@ -54,16 +66,33 @@ _INTERNAL_FILE_PREFIXES = ("/userdata/",)
 REASON_INTERNAL_STORAGE = "internal_storage"
 REASON_NO_EXTERNAL_STORAGE = "no_external_storage"
 
+# Where a sliced file has ever been found over FTPS, in the order the sweep in
+# `main.py` tries them -- root first, which is where A1/P1-series uploads land
+# (#972), then `/cache`, which is where the H2D keeps its copy of an eMMC job
+# (#2856).
+_PROBE_DIRECTORIES = ("/", "/cache/", "/model/", "/data/", "/data/Metadata/")
+
+# Longest name worth probing for. Every filesystem the printer could be serving
+# from caps a name at 255 bytes, so anything past this cannot be a file that is
+# actually there -- and it would be written to a local temp path too.
+_MAX_PROBE_FILENAME_LENGTH = 255
+
 
 @dataclass(frozen=True)
 class StorageVerdict:
     """Whether an FTPS sweep for this print's file is worth running.
 
     ``reachable`` False always carries a ``reason``; True never does.
+
+    ``probe_filename`` is the exact name the dispatch gave, present only on an
+    unreachable verdict and only when the URL named a ``.3mf``. It is the
+    caller's chance to check the claim cheaply before acting on ``reason`` --
+    see :func:`ftp_probe_paths`.
     """
 
     reachable: bool
     reason: str | None = None
+    probe_filename: str | None = None
 
 
 _REACHABLE = StorageVerdict(reachable=True)
@@ -102,6 +131,54 @@ def url_is_external_storage(project_url: str | None) -> bool | None:
     return False
 
 
+def probe_filename_from_url(project_url: str | None) -> str | None:
+    """The exact 3MF name *project_url* points at, for a bounded FTPS probe.
+
+    ``brtc://emmc/Cube.gcode.3mf`` -> ``Cube.gcode.3mf``, and likewise for the
+    internal ``file://`` paths. ``None`` when there is no name to probe with,
+    which is the caller's signal to fall back to the sweep it would have run.
+
+    Only ``.3mf`` names come back. A print running from a bare gcode has no 3MF
+    to find at any path, so probing for one would spend connections to learn
+    what the extension already said.
+
+    The value arrives from the network -- whatever the slicer or the printer
+    put in the dispatch -- and callers turn it into both a remote path and a
+    local temp filename, so anything that could steer either is refused rather
+    than sanitized: no separators, no traversal, no control characters.
+    """
+    if not isinstance(project_url, str):
+        return None
+    _scheme, separator, path = project_url.partition("://")
+    if not separator:
+        return None
+    name = path.rpartition("/")[2].strip()
+    if not name or len(name) > _MAX_PROBE_FILENAME_LENGTH:
+        return None
+    # A leading dot is either a traversal segment or a hidden file; neither is
+    # a sliced upload, and both would put an odd path on the wire. A backslash
+    # is a path separator on the host even though it is a legal character in
+    # the printer's own filesystem, which is how a name could reach outside the
+    # temp directory it is written to.
+    if name.startswith(".") or "\\" in name:
+        return None
+    if any(character < " " or character == "\x7f" for character in name):
+        return None
+    if not name.lower().endswith(".3mf"):
+        return None
+    return name
+
+
+def ftp_probe_paths(filename: str) -> list[str]:
+    """Remote paths to try for *filename*, best first.
+
+    One filename across the known directories, because the dispatch already
+    told us the name and only the directory is in question (#2856). Callers
+    walk the list over a single connection, against the sweep's ~110.
+    """
+    return [f"{directory}{filename}" for directory in _PROBE_DIRECTORIES]
+
+
 def external_storage_present(state: object | None) -> bool:
     """Does the printer have external storage for FTPS to serve at all?
 
@@ -153,7 +230,16 @@ def _verdict(project_url: str | None, state: object | None) -> StorageVerdict:
     # named the destination.
     external = url_is_external_storage(project_url)
     if external is False:
-        return StorageVerdict(reachable=False, reason=REASON_INTERNAL_STORAGE)
+        # Worth probing only if there is external storage for the probe to find
+        # anything on. An empty slot answers the question the probe would ask,
+        # and #2780's H2C sat that way for three weeks -- one connection per
+        # print start is small, but it is not worth spending to be told what
+        # the printer already said.
+        return StorageVerdict(
+            reachable=False,
+            reason=REASON_INTERNAL_STORAGE,
+            probe_filename=probe_filename_from_url(project_url) if external_storage_present(state) else None,
+        )
     if external is True:
         # It said external storage, so sweep even if the card flags disagree.
         # Trusting the specific claim over the general one is what keeps a

backend/app/services/printer_diagnostic.py

@@ -17,11 +17,17 @@ import ssl
 
 from backend.app.models.printer import Printer
 from backend.app.schemas.printer import DiagnosticCheck, PrinterDiagnosticResult
+from backend.app.services.bambu_ftp import find_remote_file_async
 from backend.app.services.bambu_mqtt import CONNECT_ERROR_AUTH_REJECTED
 from backend.app.services.camera import get_camera_port
 from backend.app.services.discovery import is_running_in_docker
 from backend.app.services.ftp_profiles import get_ftp_profile
-from backend.app.services.print_storage import REASON_INTERNAL_STORAGE, last_print_storage_verdict
+from backend.app.services.print_storage import (
+    REASON_INTERNAL_STORAGE,
+    StorageVerdict,
+    ftp_probe_paths,
+    last_print_storage_verdict,
+)
 from backend.app.services.printer_manager import printer_manager
 from backend.app.utils.printer_models import has_external_storage, has_remote_storage_toggle
 
@@ -35,6 +41,12 @@ PORT_CHAMBER_IMAGE = 6000  # Chamber image protocol — A1/P1 camera stream; opt
 
 _PORT_PROBE_TIMEOUT = 3.0
 
+# Cap for the storage probe (#2856). One connection and a handful of directory
+# listings, so a healthy printer answers in well under a second. Kept short on
+# purpose: this check sits inside the support bundle's 15s-per-printer budget,
+# and in the interactive run it is spinner time the user is watching.
+_STORAGE_PROBE_TIMEOUT = 6.0
+
 # Default seconds the `printer_publishing` check will wait for the first
 # report-topic message before declaring fail. Bambu printers in idle publish
 # push_status every few seconds; 10s catches healthy bridges with margin while
@@ -116,6 +128,53 @@ async def _check_ftps_tls(ip: str, model: str | None, timeout: float = _PORT_PRO
                 pass
 
 
+async def _last_print_file_is_reachable(
+    printer: Printer | None,
+    verdict: StorageVerdict,
+    *,
+    ftps_ok: bool,
+) -> bool:
+    """Did the last print's file turn up on external storage after all? (#2856)
+
+    ``verdict`` is read off the dispatch URL, which says where the printer
+    *put* the file, not whether port 990 can serve it: an H2D with a card in
+    the slot reports ``brtc://emmc/<name>`` and then hands the same file over
+    from ``/cache`` without complaint. Warning that the file is out of reach
+    while that user's archives are quietly complete would send them chasing a
+    setting that is already right.
+
+    So check before saying it -- one connection, one listing per candidate
+    directory, no transfer, and only for the check that is about to warn.
+    "Could not check" returns False and leaves the warning standing, which is
+    the safe direction: it is a warn, not a fail.
+    """
+    if not ftps_ok or not verdict.probe_filename or printer is None:
+        return False
+    # Attribute reads inside the guard too: this also runs from the support
+    # bundle, where the row may outlive its session, and a detached-instance
+    # error there is "could not check", not a broken diagnostic.
+    ip_address = None
+    try:
+        ip_address = getattr(printer, "ip_address", None)
+        access_code = getattr(printer, "access_code", None)
+        if not ip_address or not access_code:
+            return False
+        found = await find_remote_file_async(
+            ip_address,
+            access_code,
+            ftp_probe_paths(verdict.probe_filename),
+            timeout=_STORAGE_PROBE_TIMEOUT,
+            socket_timeout=_STORAGE_PROBE_TIMEOUT,
+            printer_model=getattr(printer, "model", None),
+        )
+    except Exception as e:
+        logger.debug("Could not probe %s for %s: %s", ip_address, verdict.probe_filename, e)
+        return False
+    if found:
+        logger.debug("Last print file is on external storage at %s despite %s", found, verdict.reason)
+    return found is not None
+
+
 def _auth_reason_params(reason: str | None) -> dict:
     """Map a client's CONNACK-refusal slug onto the check's `params.reason`.
 
@@ -329,11 +388,14 @@ async def run_connection_diagnostic(
                 params={"reason": "no_media"},
             )
         )
-    elif not last_print_storage_verdict(state).reachable:
-        # The toggle is on, a card is in, and the printer still put the last
-        # print on internal storage — which is what H2-series and P2S firmware
-        # does, and no setting here changes it (#2762 tracks reading that
-        # storage). A pass here would be a lie; a fail would be unresolvable.
+    elif not (last_verdict := last_print_storage_verdict(state)).reachable and not await _last_print_file_is_reachable(
+        printer, last_verdict, ftps_ok=ftps_state == "ok"
+    ):
+        # The toggle is on, a card is in, the printer said it put the last print
+        # on internal storage — and a probe confirmed the file really is out of
+        # reach. That is what H2-series and P2S firmware does, and no setting
+        # here changes it (#2762 tracks reading that storage). A pass here would
+        # be a lie; a fail would be unresolvable.
         checks.append(
             DiagnosticCheck(
                 id="external_storage",
@@ -341,6 +403,12 @@ async def run_connection_diagnostic(
                 params={"reason": REASON_INTERNAL_STORAGE},
             )
         )
+    elif not last_verdict.reachable:
+        # Reached only when the probe above found the file: the printer named
+        # internal storage and served it over FTPS anyway. What this check is
+        # for is whether Bambuddy can read the print file, and it demonstrably
+        # can — so pass, whatever the toggle happens to say (#2856).
+        checks.append(DiagnosticCheck(id="external_storage", status="pass"))
     elif store_to_sdcard is True:
         checks.append(DiagnosticCheck(id="external_storage", status="pass"))
     elif store_to_sdcard is False:

backend/tests/integration/test_printers_api.py

@@ -4263,8 +4263,15 @@ class TestCoverWhenThePrintIsOnInternalStorage:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_skips_the_fan_out_and_says_why(self, async_client: AsyncClient, printer_factory, db_session):
-        printer = await printer_factory(name="H2C")
+    async def test_probes_the_named_file_instead_of_fanning_out(
+        self, async_client: AsyncClient, printer_factory, db_session
+    ):
+        """#2856 narrowed this from a skip to a bounded look. The dispatch
+        names the file, and an H2D with a card in serves that exact name from
+        /cache while reporting `brtc://emmc` -- so ask, but ask once: five
+        paths for one name, not sixteen, and no retries on top.
+        """
+        printer = await printer_factory(name="H2D")
         state = MagicMock(
             subtask_name="job",
             state="RUNNING",
@@ -4287,7 +4294,50 @@ class TestCoverWhenThePrintIsOnInternalStorage:
             response = await async_client.get(f"/api/v1/printers/{printer.id}/cover")
 
         assert response.status_code == 404, response.text
+        # A probe that found nothing leaves the printer's own account standing,
+        # so the reason still reaches the caller (#2780).
         assert "internal_storage" in response.json()["detail"]
+        assert mock_download.await_count == 1, "the probe is one look, not the retry loop"
+        assert mock_download.await_args.args[2] == [
+            "/job.gcode.3mf",
+            "/cache/job.gcode.3mf",
+            "/model/job.gcode.3mf",
+            "/data/job.gcode.3mf",
+            "/data/Metadata/job.gcode.3mf",
+        ]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_skips_the_fan_out_and_says_why_with_nothing_to_probe_for(
+        self, async_client: AsyncClient, printer_factory, db_session
+    ):
+        """No dispatch to name a file and an empty slot: nothing to look for
+        and nothing to look on, so the route still short-circuits with the
+        reason the frontend explains (#2780)."""
+        printer = await printer_factory(name="H2C")
+        state = MagicMock(
+            subtask_name="job",
+            state="RUNNING",
+            gcode_file=None,
+            current_project_url=None,
+            sdcard=False,
+            sdcard_reported=True,
+        )
+
+        with (
+            patch("backend.app.api.routes.printers.printer_manager.get_status", return_value=state),
+            patch("backend.app.api.routes.printers.resolve_plate_id", return_value=1),
+            patch("backend.app.api.routes.printers.get_cached_3mf", return_value=None),
+            patch("backend.app.api.routes.printers.ftps_handshake_blocked", return_value=False),
+            patch(
+                "backend.app.api.routes.printers.download_file_try_paths_async",
+                new=AsyncMock(return_value=False),
+            ) as mock_download,
+        ):
+            response = await async_client.get(f"/api/v1/printers/{printer.id}/cover")
+
+        assert response.status_code == 404, response.text
+        assert "no_external_storage" in response.json()["detail"]
         mock_download.assert_not_called()
 
     @pytest.mark.asyncio

backend/tests/unit/services/test_printer_diagnostic.py

@@ -77,6 +77,7 @@ class _Env:
         test_connection_success=True,
         report_messages_since_connect: int | None = 5,
         connect_error: str | None = None,
+        file_found: str | None = None,
     ):
         self.ports = ports or _port_probe()
         # What the FTPS probe reports: "ok", "closed" or "no_tls" (#2780).
@@ -92,6 +93,13 @@ class _Env:
         # CONNACK-refusal slug the live client reports, or None when the last
         # connection attempt was never refused (#2698).
         self.connect_error = connect_error
+        # Remote path the storage probe finds for the last print's file, or
+        # None for "not there" (#2856). Patched unconditionally: without it
+        # the internal-storage branch opens a real FTPS connection to a
+        # printer that does not exist and every such test waits out the
+        # socket timeout.
+        self.file_found = file_found
+        self.find_remote_file = AsyncMock(return_value=file_found)
         self._stack = ExitStack()
 
     def __enter__(self):
@@ -116,6 +124,7 @@ class _Env:
         self._stack.enter_context(patch(f"{MOD}._detect_docker_network_mode", return_value=self.network_mode))
         self._stack.enter_context(patch(f"{MOD}._get_host_ip", return_value=self.host_ip))
         self._stack.enter_context(patch(f"{MOD}.printer_manager", manager))
+        self._stack.enter_context(patch(f"{MOD}.find_remote_file_async", new=self.find_remote_file))
         return self
 
     def __exit__(self, *exc):
@@ -123,8 +132,10 @@ class _Env:
         return False
 
 
-def _printer(ip="192.168.1.50", model=None):
-    return types.SimpleNamespace(id=1, ip_address=ip, model=model)
+def _printer(ip="192.168.1.50", model=None, access_code="12345678"):
+    # access_code is what the storage probe needs to reach port 990 (#2856);
+    # a printer without one can only be reported on, not checked.
+    return types.SimpleNamespace(id=1, ip_address=ip, model=model, access_code=access_code)
 
 
 class TestSameSubnet:
@@ -489,21 +500,86 @@ class TestExternalStorageCheck:
         assert result.overall == "problems"
 
     async def test_warns_when_the_printer_kept_the_print_internally(self):
-        """Toggle on, card in, and the printer still used internal storage --
-        which is what H2-series and P2S firmware does. A pass would be a lie
-        and a fail would be unresolvable, so it warns."""
+        """Toggle on, card in, the printer used internal storage -- and the
+        probe confirmed the file is not on the card either. A pass would be a
+        lie and a fail would be unresolvable, so it warns."""
         state = _state(
             store_to_sdcard=True,
             sdcard=True,
             sdcard_reported=True,
             last_project_url="brtc://emmc/Benchy.gcode.3mf",
         )
-        with _Env(state=state):
+        with _Env(state=state) as env:
             result = await run_connection_diagnostic("192.168.1.50", printer=_printer(model="H2C"))
         check = next(c for c in result.checks if c.id == "external_storage")
         assert check.status == "warn"
         assert check.params == {"reason": "internal_storage"}
         assert result.overall == "warnings"
+        # It asked before it warned, with the name the dispatch gave.
+        paths = env.find_remote_file.await_args.args[2]
+        assert paths[:2] == ["/Benchy.gcode.3mf", "/cache/Benchy.gcode.3mf"]
+
+    async def test_the_internal_storage_warning_yields_to_the_file_being_there(self):
+        """#2856. The URL says where the printer *put* the file, not whether
+        port 990 can serve it: an H2D with a card in reports `brtc://emmc` and
+        then hands the same file over from /cache. Warning that reader's
+        archives are unreachable -- while they are demonstrably complete --
+        sends them chasing a setting that is already right.
+        """
+        state = _state(
+            store_to_sdcard=True,
+            sdcard=True,
+            sdcard_reported=True,
+            last_project_url="brtc://emmc/Benchy.gcode.3mf",
+        )
+        with _Env(state=state, file_found="/cache/Benchy.gcode.3mf"):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer(model="H2D"))
+        assert _statuses(result)["external_storage"] == "pass"
+
+    async def test_the_file_being_there_outranks_the_toggle(self):
+        """Same printer with the toggle off. The check exists to say whether
+        Bambuddy can read the print file, and the probe just proved it can --
+        telling this user to switch something on would be advice for a problem
+        they do not have.
+        """
+        state = _state(
+            store_to_sdcard=False,
+            sdcard=True,
+            sdcard_reported=True,
+            last_project_url="brtc://emmc/Benchy.gcode.3mf",
+        )
+        with _Env(state=state, file_found="/cache/Benchy.gcode.3mf"):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer(model="H2D"))
+        assert _statuses(result)["external_storage"] == "pass"
+
+    async def test_no_probe_when_the_file_service_is_not_answering(self):
+        """The probe is one FTPS connection, and #2780's printer refused those
+        by the hundred. If port 990 is not answering there is nothing to learn
+        and the warning stands on the URL alone."""
+        state = _state(
+            store_to_sdcard=True,
+            sdcard=True,
+            sdcard_reported=True,
+            last_project_url="brtc://emmc/Benchy.gcode.3mf",
+        )
+        with _Env(state=state, ftps="no_tls") as env:
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer(model="H2C"))
+        env.find_remote_file.assert_not_awaited()
+        assert _statuses(result)["external_storage"] == "warn"
+
+    async def test_a_probe_that_cannot_run_leaves_the_warning_in_place(self):
+        """ "Could not check" must not read as "the file is fine" -- the check
+        keeps warning, which is the recoverable direction for a warn."""
+        state = _state(
+            store_to_sdcard=True,
+            sdcard=True,
+            sdcard_reported=True,
+            last_project_url="brtc://emmc/Benchy.gcode.3mf",
+        )
+        with _Env(state=state) as env:
+            env.find_remote_file.side_effect = OSError("connection reset")
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer(model="H2C"))
+        assert _statuses(result)["external_storage"] == "warn"
 
     async def test_an_external_storage_dispatch_still_passes(self):
         """The regression guard: an X1C dispatch says `ftp://`, and that must

backend/tests/unit/test_internal_storage_probe_2856.py

@@ -0,0 +1,395 @@
+"""An eMMC dispatch is not proof the file is out of reach (#2856).
+
+``print_storage`` reads the ``project_file`` URL: ``brtc://emmc/<name>`` says
+the printer put the sliced file on internal storage, and #2780 turned that into
+"skip the FTPS lookup entirely". On an H2D with a card in the slot that is
+wrong. The reporter's log has ``"url": "brtc://emmc/test.gcode.3mf"`` and, a
+second later, ``Downloaded: /cache/test.gcode.3mf`` -- every one of his prints
+back to 08-12, 19 MB included, until the skip landed and two days of archives
+came out as name-only fallbacks.
+
+So the URL earns a *bounded probe* rather than a skip. It names the exact file,
+which is one connection walking five directories instead of the sweep's ~110,
+and the printer answers the question instead of a guess about its model. These
+tests pin both halves: the probe runs and its hit is archived normally, and a
+miss still ends in #2780's cheap fallback with its reason intact.
+"""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.main import (
+    _active_prints,
+    _expected_print_creators,
+    _expected_print_registered_at,
+    _expected_prints,
+    _print_ams_mappings,
+    _timelapse_baselines,
+)
+from backend.app.services.print_storage import (
+    ftp_probe_paths,
+    print_file_reachable_over_ftp,
+    probe_filename_from_url,
+)
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture(autouse=True)
+def _clear_dicts():
+    for d in (
+        _expected_prints,
+        _expected_print_registered_at,
+        _expected_print_creators,
+        _print_ams_mappings,
+        _active_prints,
+        _timelapse_baselines,
+    ):
+        d.clear()
+    yield
+    for d in (
+        _expected_prints,
+        _expected_print_registered_at,
+        _expected_print_creators,
+        _print_ams_mappings,
+        _active_prints,
+        _timelapse_baselines,
+    ):
+        d.clear()
+
+
+class TestProbeFilename:
+    """What the probe asks for. The dispatch's name is the authoritative one --
+    the sweep's guesses are built from ``subtask_name``, which is normalized,
+    truncated and occasionally a plate behind."""
+
+    def test_the_reported_case(self):
+        assert probe_filename_from_url("brtc://emmc/test.gcode.3mf") == "test.gcode.3mf"
+
+    def test_a_name_with_the_punctuation_users_actually_use(self):
+        """Straight from the reporter's log -- ampersands, parentheses, plus."""
+        url = "brtc://emmc/H2D_&_H2S_poop_chute+4_buckets_(no_magnets_&_glue).gcode.3mf"
+        assert probe_filename_from_url(url) == "H2D_&_H2S_poop_chute+4_buckets_(no_magnets_&_glue).gcode.3mf"
+
+    def test_a_non_ascii_name(self):
+        assert probe_filename_from_url("brtc://emmc/小船.gcode.3mf") == "小船.gcode.3mf"
+
+    def test_an_internal_file_path_keeps_only_the_name(self):
+        """The model cache is not reachable at that path, but the same file may
+        well be sitting in /cache under its bare name."""
+        assert probe_filename_from_url("file:///userdata/model/history/Cube.gcode.3mf") == "Cube.gcode.3mf"
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            None,
+            "",
+            "Benchy.gcode.3mf",  # no scheme -- not a dispatch URL at all
+            "brtc://emmc/",
+            "brtc://emmc/Benchy.gcode",  # a gcode job has no 3MF at any path
+            "brtc://emmc/.",
+            "brtc://emmc/..",
+            12345,
+        ],
+    )
+    def test_nothing_to_probe_with(self, url):
+        assert probe_filename_from_url(url) is None
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "brtc://emmc/..\\..\\evil.3mf",  # backslash is a separator on the host
+            "brtc://emmc/sub\\dir\\Cube.3mf",
+            "brtc://emmc/Cube\n.3mf",  # control characters
+            "brtc://emmc/" + "C" * 256 + ".3mf",
+        ],
+    )
+    def test_a_name_that_could_steer_a_path_is_refused_not_cleaned(self, url):
+        """The name comes off the wire and becomes both a remote path and a
+        local temp filename, so a name that could point at either somewhere
+        else is declined outright -- the print falls back to the archive it
+        would have got anyway."""
+        assert probe_filename_from_url(url) is None
+
+
+class TestProbePaths:
+    def test_root_first_then_cache(self):
+        """Order is the sweep's own: root is where A1/P1 uploads land (#972),
+        /cache is where the H2D keeps its copy (#2856)."""
+        assert ftp_probe_paths("Cube.gcode.3mf")[:2] == ["/Cube.gcode.3mf", "/cache/Cube.gcode.3mf"]
+
+    def test_one_filename_five_paths(self):
+        assert len(ftp_probe_paths("Cube.gcode.3mf")) == 5
+
+
+class TestFindRemoteFile:
+    """The lookup the connection diagnostic runs. It must not transfer the
+    file -- the thing it is asking about can be tens of megabytes and the
+    answer is a yes or a no."""
+
+    @staticmethod
+    def _client(listings):
+        client = MagicMock()
+        client.connect.return_value = True
+        client.list_files.side_effect = lambda directory: [
+            {"name": name, "is_directory": False} for name in listings.get(directory, [])
+        ]
+        return client
+
+    @pytest.mark.asyncio
+    async def test_finds_the_file_in_cache_and_stops_there(self):
+        client = self._client({"/": ["other.3mf"], "/cache": ["test.gcode.3mf"]})
+
+        with patch("backend.app.services.bambu_ftp.BambuFTPClient", return_value=client):
+            from backend.app.services.bambu_ftp import find_remote_file_async
+
+            found = await find_remote_file_async("1.2.3.4", "code", ftp_probe_paths("test.gcode.3mf"))
+
+        assert found == "/cache/test.gcode.3mf"
+        # One connection, and the directories after the hit are never listed.
+        client.connect.assert_called_once()
+        assert [call.args[0] for call in client.list_files.call_args_list] == ["/", "/cache"]
+        client.download_to_file.assert_not_called()
+        client.disconnect.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_a_directory_is_listed_once_however_many_candidates_share_it(self):
+        client = self._client({})
+
+        with patch("backend.app.services.bambu_ftp.BambuFTPClient", return_value=client):
+            from backend.app.services.bambu_ftp import find_remote_file_async
+
+            found = await find_remote_file_async("1.2.3.4", "code", ["/a.3mf", "/b.3mf", "/cache/a.3mf"])
+
+        assert found is None
+        assert [call.args[0] for call in client.list_files.call_args_list] == ["/", "/cache"]
+
+    @pytest.mark.asyncio
+    async def test_a_directory_of_the_same_name_is_not_the_file(self):
+        client = MagicMock()
+        client.connect.return_value = True
+        client.list_files.return_value = [{"name": "test.gcode.3mf", "is_directory": True}]
+
+        with patch("backend.app.services.bambu_ftp.BambuFTPClient", return_value=client):
+            from backend.app.services.bambu_ftp import find_remote_file_async
+
+            assert await find_remote_file_async("1.2.3.4", "code", ftp_probe_paths("test.gcode.3mf")) is None
+
+    @pytest.mark.asyncio
+    async def test_a_refused_connection_is_not_an_answer(self):
+        client = MagicMock()
+        client.connect.return_value = False
+
+        with patch("backend.app.services.bambu_ftp.BambuFTPClient", return_value=client):
+            from backend.app.services.bambu_ftp import find_remote_file_async
+
+            assert await find_remote_file_async("1.2.3.4", "code", ftp_probe_paths("test.gcode.3mf")) is None
+
+        client.list_files.assert_not_called()
+
+
+class TestVerdictCarriesTheName:
+    def test_an_internal_dispatch_carries_a_probe_name(self):
+        state = MagicMock(current_project_url="brtc://emmc/test.gcode.3mf", sdcard=True, sdcard_reported=True)
+        verdict = print_file_reachable_over_ftp(state)
+
+        assert verdict.reachable is False
+        assert verdict.reason == "internal_storage"
+        assert verdict.probe_filename == "test.gcode.3mf"
+
+    def test_an_external_dispatch_needs_no_probe(self):
+        """It sweeps as it always did; a probe name here would only invite a
+        caller to shorten a search that is already working."""
+        state = MagicMock(current_project_url="ftp://test.gcode.3mf", sdcard=True, sdcard_reported=True)
+        verdict = print_file_reachable_over_ftp(state)
+
+        assert verdict.reachable is True
+        assert verdict.probe_filename is None
+
+    def test_an_empty_slot_has_nothing_to_probe(self):
+        """No URL and no card: there is no name, and no storage to look on."""
+        state = MagicMock(current_project_url=None, sdcard=False, sdcard_reported=True)
+        verdict = print_file_reachable_over_ftp(state)
+
+        assert verdict.reason == "no_external_storage"
+        assert verdict.probe_filename is None
+
+    def test_an_empty_slot_is_not_probed_even_with_a_name(self):
+        """#2780's H2C ran for three weeks with the toggle on and nothing in
+        the slot. There is no card for a copy to be on, so the name is not
+        worth a connection -- the printer has already answered."""
+        state = MagicMock(current_project_url="brtc://emmc/test.gcode.3mf", sdcard=False, sdcard_reported=True)
+        verdict = print_file_reachable_over_ftp(state)
+
+        assert verdict.reason == "internal_storage"
+        assert verdict.probe_filename is None
+
+    def test_a_printer_that_never_mentions_its_card_is_still_probed(self):
+        """`sdcard` defaults to False, so acting on the default would drop the
+        probe for every printer that simply does not publish the field."""
+        state = MagicMock(current_project_url="brtc://emmc/test.gcode.3mf", sdcard=False, sdcard_reported=False)
+
+        assert print_file_reachable_over_ftp(state).probe_filename == "test.gcode.3mf"
+
+
+def _printer():
+    printer = MagicMock()
+    printer.id = 1
+    printer.auto_archive = True
+    printer.external_camera_enabled = False
+    printer.external_camera_url = None
+    # Every unset MagicMock attribute is truthy, and a truthy one here runs the
+    # plate-detection camera grab against a printer that does not exist.
+    printer.plate_detection_enabled = False
+    printer.name = "H2D"
+    printer.model = "H2D"
+    printer.ip_address = "192.168.1.211"
+    printer.access_code = "12345678"
+    return printer
+
+
+async def _run_print_start(url, *, probe_hit, added, handshake_blocked=False):
+    """Drive on_print_start for an eMMC dispatch, returning the probe mock and
+    the ArchiveService the success path would have used."""
+    printer = _printer()
+    state = MagicMock(current_project_url=url, sdcard=True, sdcard_reported=True)
+
+    def execute_router(stmt, *args, **kwargs):
+        sql = str(stmt).lower()
+        if "from printers" in sql or "from printer " in sql:
+            return MagicMock(
+                scalar_one_or_none=MagicMock(return_value=printer),
+                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[printer]))),
+            )
+        return MagicMock(
+            scalar_one_or_none=MagicMock(return_value=None),
+            scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
+        )
+
+    session = AsyncMock()
+    session.__aenter__ = AsyncMock(return_value=session)
+    session.__aexit__ = AsyncMock()
+    session.execute = AsyncMock(side_effect=execute_router)
+    session.commit = AsyncMock()
+    session.refresh = AsyncMock()
+    session.add = MagicMock(side_effect=added.append)
+
+    probe = AsyncMock(return_value=probe_hit)
+    archive_service = MagicMock()
+    archive_service.archive_print = AsyncMock(return_value=MagicMock(id=20, print_name="test", status="printing"))
+
+    with (
+        patch("backend.app.main.async_session") as session_maker,
+        patch("backend.app.main.notification_service") as notif,
+        patch("backend.app.main.smart_plug_manager") as plug,
+        patch("backend.app.main.ws_manager") as ws,
+        patch("backend.app.main.mqtt_relay") as relay,
+        patch("backend.app.main.printer_manager") as pm,
+        patch("backend.app.main.download_file_try_paths_async", new=probe),
+        patch("backend.app.main.download_file_async", new=AsyncMock(return_value=False)),
+        patch("backend.app.main.with_ftp_retry", new=AsyncMock(return_value=False)),
+        patch("backend.app.main.get_cached_3mf", return_value=None),
+        patch("backend.app.main.cache_3mf_download") as cache,
+        patch("backend.app.services.bambu_ftp.list_files_async", new=AsyncMock(return_value=[])),
+        patch("backend.app.main.ftps_handshake_blocked", return_value=handshake_blocked),
+        patch("backend.app.main.get_ftp_retry_settings", new=AsyncMock(return_value=(False, 3, 2.0, 30))),
+        patch("backend.app.main.ArchiveService", return_value=archive_service),
+        patch("backend.app.main.peek_plate_index_in_3mf", return_value=None),
+        patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
+        patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
+        patch("backend.app.main._maybe_start_layer_timelapse"),
+        patch("backend.app.main._capture_timelapse_baseline_at_start", new_callable=AsyncMock),
+    ):
+        session_maker.return_value = session
+        notif.on_print_start = AsyncMock()
+        plug.on_print_start = AsyncMock()
+        ws.send_print_start = AsyncMock()
+        ws.send_archive_created = AsyncMock()
+        ws.send_archive_updated = AsyncMock()
+        relay.on_print_start = AsyncMock()
+        relay.on_archive_created = AsyncMock()
+        pm.get_status = MagicMock(return_value=state)
+        pm.get_client = MagicMock(return_value=None)
+        pm.get_printer = MagicMock(return_value=MagicMock(serial_number="TEST2856"))
+
+        from backend.app.main import on_print_start
+
+        await on_print_start(1, {"filename": "/data/Metadata/plate_1.gcode", "subtask_name": "test"})
+
+    return probe, archive_service, cache
+
+
+def _fallback(added):
+    for row in added:
+        extra = getattr(row, "extra_data", None)
+        if isinstance(extra, dict) and extra.get("no_3mf_available"):
+            return row
+    return None
+
+
+class TestPrintStart:
+    @pytest.mark.asyncio
+    async def test_a_file_the_printer_serves_anyway_is_archived_in_full(self):
+        """The reported regression, end to end: eMMC dispatch, file present on
+        the card, and the archive gets the real 3MF instead of a name."""
+        added = []
+
+        probe, service, cache = await _run_print_start("brtc://emmc/test.gcode.3mf", probe_hit=True, added=added)
+
+        probe.assert_awaited_once()
+        assert probe.await_args.args[2] == ftp_probe_paths("test.gcode.3mf")
+        service.archive_print.assert_awaited_once()
+        assert Path(service.archive_print.await_args.kwargs["source_file"]).name == "test.gcode.3mf"
+        assert _fallback(added) is None, "a fallback archive here is the bug"
+
+    @pytest.mark.asyncio
+    async def test_the_probed_file_is_shared_with_the_cover_endpoint(self):
+        """Same 3MF, one transfer. The cover endpoint runs seconds later while
+        the frontend opens the card, and re-fetching 19 MB over the printer's
+        single FTP socket is what produced #972's 425 storm."""
+        added = []
+
+        _probe, _service, cache = await _run_print_start("brtc://emmc/test.gcode.3mf", probe_hit=True, added=added)
+
+        cache.assert_called_once()
+        assert cache.call_args.args[1] == "test.gcode.3mf"
+
+    @pytest.mark.asyncio
+    async def test_a_miss_still_ends_in_the_cheap_fallback(self):
+        """#2780's printers are still out there: when the probe finds nothing,
+        the reason has to survive to the archive card, which is what stops the
+        banner telling an H2C owner to switch on a setting that is already on.
+        """
+        added = []
+
+        probe, service, _cache = await _run_print_start("brtc://emmc/test.gcode.3mf", probe_hit=False, added=added)
+
+        probe.assert_awaited_once()
+        service.archive_print.assert_not_awaited()
+        assert _fallback(added).extra_data["no_3mf_reason"] == "internal_storage"
+
+    @pytest.mark.asyncio
+    async def test_no_probe_while_the_file_service_is_in_cool_off(self):
+        """The handshake is failing below the path level, so the probe would
+        only re-run the failure that put the printer in cool-off (#2780)."""
+        added = []
+
+        probe, _service, _cache = await _run_print_start(
+            "brtc://emmc/test.gcode.3mf", probe_hit=True, added=added, handshake_blocked=True
+        )
+
+        probe.assert_not_awaited()
+        assert _fallback(added).extra_data["no_3mf_reason"] == "internal_storage"
+
+    @pytest.mark.asyncio
+    async def test_a_gcode_job_is_not_probed_for(self):
+        """Nothing names a 3MF, so there is no name to ask about and the skip
+        stays exactly as cheap as #2780 made it."""
+        added = []
+
+        probe, _service, _cache = await _run_print_start("brtc://emmc/plate_1.gcode", probe_hit=True, added=added)
+
+        probe.assert_not_awaited()
+        assert _fallback(added) is not None

backend/tests/unit/test_print_start_skips_unreachable_storage_2780.py

@@ -4,10 +4,12 @@ The unit tests around ``print_storage`` pin the decision; this pins that
 ``on_print_start`` actually acts on it. Two things have to hold and neither is
 visible from the helper alone:
 
-* No FTP is attempted. The sweep is six filename variants across five
-  directories with up to four retries, and on the reporter's P2S every one of
-  those failed -- 1813 failures in a day against a printer that, on the leading
-  theory, was refusing precisely because of the connection volume.
+* The sweep does not run. It is six filename variants across five directories
+  with up to four retries, and on the reporter's P2S every one of those failed
+  -- 1813 failures in a day against a printer that, on the leading theory, was
+  refusing precisely because of the connection volume. What replaced it for an
+  internal-storage URL is a single probe of the one name the dispatch gave
+  (#2856); the bound is what mattered here, not the silence.
 * The fallback archive records *which* reason applied. Without it the archives
   banner falls back to its original wording, which tells the user to switch on
   a setting that in this case is already on and would not have helped.
@@ -80,9 +82,12 @@ def _state(current_project_url, sdcard=True, sdcard_reported=True):
     )
 
 
-async def _run_print_start(state, added):
+async def _run_print_start(state, added, probe_hit=False):
     """Drive on_print_start for a print with no matching archive, capturing
     whatever rows it adds and whether it reached the FTP layer.
+
+    ``probe_hit`` is what the bounded internal-storage probe finds (#2856);
+    False is the #2780 case where the file really is out of reach.
     """
     printer = _printer()
 
@@ -109,6 +114,7 @@ async def _run_print_start(state, added):
     session.add = MagicMock(side_effect=added.append)
 
     download = AsyncMock(return_value=False)
+    probe = AsyncMock(return_value=probe_hit)
 
     with (
         patch("backend.app.main.async_session") as session_maker,
@@ -118,6 +124,7 @@ async def _run_print_start(state, added):
         patch("backend.app.main.mqtt_relay") as relay,
         patch("backend.app.main.printer_manager") as pm,
         patch("backend.app.main.download_file_async", new=download),
+        patch("backend.app.main.download_file_try_paths_async", new=probe),
         patch("backend.app.main.with_ftp_retry", new=AsyncMock(return_value=False)) as retry,
         patch("backend.app.main.get_cached_3mf", return_value=None),
         # Imported inside the function, so it has to be patched at its source
@@ -147,7 +154,7 @@ async def _run_print_start(state, added):
             {"filename": "/data/Metadata/plate_1.gcode", "subtask_name": "Halterung"},
         )
 
-    return download, retry
+    return download, retry, probe
 
 
 def _fallback(added):
@@ -160,11 +167,22 @@ def _fallback(added):
 
 
 @pytest.mark.asyncio
-async def test_a_print_on_internal_storage_never_touches_ftp():
+async def test_a_print_on_internal_storage_probes_once_instead_of_sweeping():
+    """The sweep stays off, but the claim gets checked (#2856): one connection
+    walking the five known directories for the one name the dispatch gave,
+    against the sweep's ~110 that cannot succeed."""
     added = []
 
-    download, retry = await _run_print_start(_state("brtc://emmc/Halterung.gcode.3mf"), added)
+    download, retry, probe = await _run_print_start(_state("brtc://emmc/Halterung.gcode.3mf"), added)
 
+    probe.assert_awaited_once()
+    assert probe.await_args.args[2] == [
+        "/Halterung.gcode.3mf",
+        "/cache/Halterung.gcode.3mf",
+        "/model/Halterung.gcode.3mf",
+        "/data/Halterung.gcode.3mf",
+        "/data/Metadata/Halterung.gcode.3mf",
+    ]
     download.assert_not_called()
     retry.assert_not_called()
 
@@ -175,12 +193,15 @@ async def test_a_print_on_internal_storage_never_touches_ftp():
 
 @pytest.mark.asyncio
 async def test_an_empty_slot_never_touches_ftp():
+    """No URL, no card: nothing to probe with and nothing to serve it, so this
+    one stays a pure short-circuit."""
     added = []
 
-    download, retry = await _run_print_start(_state(None, sdcard=False, sdcard_reported=True), added)
+    download, retry, probe = await _run_print_start(_state(None, sdcard=False, sdcard_reported=True), added)
 
     download.assert_not_called()
     retry.assert_not_called()
+    probe.assert_not_awaited()
     assert _fallback(added).extra_data["no_3mf_reason"] == "no_external_storage"
 
 
@@ -190,7 +211,7 @@ async def test_a_print_on_external_storage_still_sweeps():
     the download exactly as it did before the gate existed."""
     added = []
 
-    download, _ = await _run_print_start(_state("ftp://Halterung.gcode.3mf"), added)
+    download, _retry, _probe = await _run_print_start(_state("ftp://Halterung.gcode.3mf"), added)
 
     download.assert_called()
 
@@ -202,7 +223,7 @@ async def test_a_printer_that_said_nothing_still_sweeps():
     either -- that install must behave exactly as before."""
     added = []
 
-    download, _ = await _run_print_start(_state(None, sdcard=False, sdcard_reported=False), added)
+    download, _retry, _probe = await _run_print_start(_state(None, sdcard=False, sdcard_reported=False), added)
 
     download.assert_called()