From e24aa41e256f06bc04bdf473acbca6b5b2c498a2 Mon Sep 17 00:00:00 2001
From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com>
Date: Fri, 21 Aug 2026 09:17:06 -0500
Subject: [PATCH] Verify UF2 flashes instead of assuming the copy worked

`firmware flash` already copied a .uf2 onto a bootloader drive for rp2 and
samd, but it reported success whenever shutil.copyfile did not raise. That
is not evidence of anything. A UF2 flash has two failure modes that look
identical to success from the host side:

  - a copy that reports fine and writes nothing (PowerShell's Copy-Item
    fails non-terminatingly, exit 0, no file), and
  - a bootloader that silently skips every block whose family ID it does
    not own, leaving a perfectly successful copy sitting on the drive.

So the copy is no longer the proof. The proof is that the volume unmounts:
a UF2 bootloader reboots into the new firmware once it has accepted a
complete image, and the mount goes with it. That signal comes from the
board rather than from the host filesystem. A volume still mounted after
--uf2-timeout (default 30s) is now a failure that names the image's family,
since a family mismatch is the usual cause.

Before copying, the new uf2 module validates the file: block magic, payload
bounds, and the block count against the header -- a file that disagrees
with its own numBlocks makes a bootloader wait forever rather than flash.
Family IDs are reported, not enforced: the upstream registry and
CircuitPython's espressif Makefile disagree about 0x540ddf62 (ESP32-C6 vs
esp32p4), so a hard check would reject working images.

--uf2 forces this path for any port, which is what reaches a board whose
UF2 bootloader is not implied by its MicroPython port -- an ESP32-S3
carrying tinyuf2 being the case that matters.

Volume discovery moves to INFO_UF2.TXT rather than labels, which differ per
family, and asks Windows for drive letters so it works under WSL, where a
removable drive is usually not mounted under /mnt at all. Two mounted
volumes now stop the command instead of picking one, because guessing
overwrites the firmware on a board the caller did not name.

test:python needed PYTHONPATH to import the engine at all; three test files
were erroring out before this.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 docs/agent-guide.md            |  30 +++
 package.json                   |   2 +-
 python/firmware_engine.py      | 230 +++++++++++++-----
 python/mpftp_cli.py            |   8 +
 python/tests/test_uf2.py       | 221 +++++++++++++++++
 python/tests/test_uf2_flash.py | 201 ++++++++++++++++
 python/uf2.py                  | 428 +++++++++++++++++++++++++++++++++
 7 files changed, 1061 insertions(+), 59 deletions(-)
 create mode 100644 python/tests/test_uf2.py
 create mode 100644 python/tests/test_uf2_flash.py
 create mode 100644 python/uf2.py

diff --git a/docs/agent-guide.md b/docs/agent-guide.md
index 7ec361e..b7dd2bd 100644
--- a/docs/agent-guide.md
+++ b/docs/agent-guide.md
@@ -296,6 +296,36 @@ Flash without rebuild to the next board:
 
 Supported flashers: `esp32` (esptool), `rp2` / `samd` (UF2; BOOTSEL first).
 
+### Flashing over UF2
+
+`rp2` and `samd` take the UF2 path automatically. `--uf2` forces it for any
+port, which is what reaches a board whose UF2 bootloader is not implied by its
+MicroPython port — an ESP32-S3 carrying tinyuf2, most usefully:
+
+```bash
+./scripts/mpftp bootloader -d COM7          # or double-tap reset / hold BOOTSEL
+./scripts/mpftp firmware flash --port esp32 --uf2 --artifact build/firmware.uf2
+```
+
+The artifact must be a `.uf2`; the command refuses a `.bin` rather than copying
+something the bootloader will ignore.
+
+**The copy is not the proof.** A UF2 flash has two failure modes that both look
+exactly like success from the host: a copy that reports fine while writing
+nothing, and a bootloader that silently skips every block whose family ID it
+does not own. So the engine validates the file first (magic, block count against
+the header, family IDs), verifies the byte count it wrote, and then waits for
+the bootloader volume to **unmount** — which only happens once the board has
+accepted a complete image and rebooted into it. A volume still mounted after
+`--uf2-timeout` seconds (default 30) is reported as a failure naming the image's
+family, because a family mismatch is the usual cause.
+
+Volumes are found by `INFO_UF2.TXT`, not by label, since labels differ per family
+(`RPI-RP2`, `FTHRS3BOOT`, …). If more than one is mounted the command stops and
+asks for `--device` rather than guessing which board to overwrite. On WSL a
+removable drive is usually *not* mounted under `/mnt`, so discovery also asks
+Windows for drive letters; `--device 'D:'` works there too.
+
 ### ESP32 partition autosize
 
 If an esp32 build fails because the app image is larger than the `factory` (or
diff --git a/package.json b/package.json
index 66d625d..8a3e70c 100644
--- a/package.json
+++ b/package.json
@@ -296,7 +296,7 @@
     "watch": "tsc -watch -p ./",
     "lint": "tsc -p ./ --noEmit",
     "package": "npx --yes @vscode/vsce package --no-dependencies",
-    "test:python": "python3 -m unittest discover -s python/tests -v"
+    "test:python": "PYTHONPATH=python python3 -m unittest discover -s python/tests -v"
   },
   "repository": {
     "type": "git",
diff --git a/python/firmware_engine.py b/python/firmware_engine.py
index 0284b9e..81bc167 100644
--- a/python/firmware_engine.py
+++ b/python/firmware_engine.py
@@ -39,6 +39,8 @@
 import sys
 import time
 
+import uf2
+
 
 def _no_window_kwargs() -> dict:
     """Avoid flashing a blank console on Windows when spawning esptool/make."""
@@ -577,6 +579,11 @@ def resolve_build_toolchains(
 # Tree / port model
 # --------------------------------------------------------------------------- #
 
+# How long to wait for a UF2 bootloader volume to unmount after the copy.
+# Generous because the board erases and writes flash before it reboots, and a
+# false timeout here reports failure on a flash that worked.
+UF2_REBOOT_TIMEOUT = 30.0
+
 # Ports we can flash (others are build-only in the UI).
 FLASHERS = {
     "esp32": "esptool",
@@ -1429,82 +1436,176 @@ def flash_esp32(ns: argparse.Namespace, mp: Optional[Path], artifact: Path) -> N
     log_activity("firmware_flash", f"esp32 {ns.board} -> {ns.device}", {"offset": offset})
 
 
-def _find_uf2_drive() -> Optional[str]:
-    """Find a mounted UF2 bootloader drive (RPI-RP2, etc.)."""
-    roots: list[Path] = []
-    if HOST in ("wsl", "linux"):
-        roots += [Path("/media"), Path("/mnt"), Path("/run/media")]
-    if HOST == "windows":
-        for letter in "DEFGHIJKLMNOP":
-            roots.append(Path(f"{letter}:/"))
-    checked: list[Path] = []
-    for root in roots:
-        try:
-            if root.name.endswith(":") or str(root).endswith(":/"):
-                checked.append(root)
-            elif root.is_dir():
-                for sub in root.iterdir():
-                    if sub.is_dir():
-                        # one more level (/media/<user>/<LABEL>)
-                        checked.append(sub)
-                        for sub2 in sub.iterdir() if sub.is_dir() else []:
-                            checked.append(sub2)
-        except Exception:
-            continue
-    for d in checked:
-        try:
-            if (d / "INFO_UF2.TXT").is_file():
-                return str(d)
-        except Exception:
-            continue
-    return None
+def flash_uf2(ns: argparse.Namespace, artifact: Path) -> None:
+    """Flash by copying a .uf2 onto a mounted bootloader volume.
 
+    The copy is not the proof -- see ``uf2.wait_for_volume_gone``. Every exit
+    path here reports what was actually observed, because the two ways this
+    goes wrong (a copy that writes nothing, a bootloader that ignores an image
+    it does not own) both look exactly like success from the host side.
+    """
+    try:
+        meta = uf2.parse_uf2(artifact)
+    except uf2.Uf2Error as e:
+        emit_result(False, error=str(e), method="uf2")
+        return
+    except OSError as e:
+        emit_result(False, error=f"Cannot read {artifact}: {e}", method="uf2")
+        return
 
-def flash_uf2(ns: argparse.Namespace, artifact: Path) -> None:
-    """rp2 / samd: copy .uf2 to the bootloader drive; rp2 falls back to picotool."""
-    target = ns.device if ns.device and _looks_like_mount(ns.device) else _find_uf2_drive()
-    if target:
-        dest = Path(target) / artifact.name
-        emit_log(f"[mpftp] copying {artifact.name} -> {dest}")
-        try:
-            shutil.copyfile(str(artifact), str(dest))
-            try:
-                # Flush; some FS need it before the board reboots.
-                with open(dest, "rb+") as f:
-                    os.fsync(f.fileno())
-            except Exception:
-                pass
-            emit_result(True, device=target, artifact=str(artifact), method="uf2")
-            log_activity("firmware_flash", f"uf2 -> {target}", {"artifact": str(artifact)})
+    families = "/".join(meta["family_names"]) or "none"
+    emit_log(
+        f"[mpftp] {artifact.name}: {meta['blocks']} blocks, "
+        f"{meta['payload_bytes']} bytes, family {families}"
+    )
+    for warning in meta["warnings"]:
+        emit_log(f"[mpftp] warning: {warning}")
+
+    volume, ambiguous = _select_uf2_volume(ns)
+    if ambiguous:
+        emit_result(False, method="uf2", error=ambiguous)
+        return
+    if volume is None:
+        _uf2_no_volume(ns, artifact)
+        return
+
+    root = Path(volume["path"])
+    board_id = volume.get("board_id") or "unknown"
+    emit_log(f"[mpftp] bootloader volume {root} (Board-ID: {board_id})")
+
+    dest = root / artifact.name
+    emit_phase("flashing", f"copying {artifact.name} -> {root}")
+    try:
+        written = uf2.copy_uf2(artifact, dest)
+    except OSError as e:
+        # A write error part-way through can also mean the board rebooted early.
+        # Check before blaming the copy.
+        if uf2.wait_for_volume_gone(root, timeout=2.0):
+            emit_log(f"[mpftp] write ended with {e}, but the volume went away")
+            _uf2_success(ns, artifact, root, board_id, meta, written=-1)
             return
-        except Exception as e:
-            emit_log(f"[mpftp] UF2 copy failed: {e}")
+        emit_result(
+            False,
+            method="uf2",
+            device=str(root),
+            artifact=str(artifact),
+            error=f"Copy to {dest} failed: {e}",
+        )
+        return
 
-    if ns.port == "rp2" and shutil.which("picotool"):
-        emit_log("[mpftp] no UF2 drive; trying picotool load")
+    expected = artifact.stat().st_size
+    if written != expected:
+        emit_result(
+            False,
+            method="uf2",
+            device=str(root),
+            artifact=str(artifact),
+            error=f"Short write: {written} of {expected} bytes reached {dest}.",
+        )
+        return
+    emit_log(f"[mpftp] wrote {written} bytes; waiting for the board to reboot")
+
+    timeout = float(getattr(ns, "uf2_timeout", 0) or UF2_REBOOT_TIMEOUT)
+    if uf2.wait_for_volume_gone(root, timeout=timeout):
+        _uf2_success(ns, artifact, root, board_id, meta, written)
+        return
+
+    # The volume is still mounted, so the bootloader did not accept the image.
+    # A family the board does not own is by far the most common cause: those
+    # blocks are skipped in silence, leaving a perfectly successful copy behind.
+    hint = (
+        f"The image is {families}; check it matches this board."
+        if meta["families"]
+        else "This UF2 carries no family ID, so it cannot be matched to the board."
+    )
+    emit_result(
+        False,
+        method="uf2",
+        device=str(root),
+        artifact=str(artifact),
+        family=meta["family_names"],
+        board_id=board_id,
+        error=(
+            f"Copied {written} bytes to {dest}, but {root} is still mounted after "
+            f"{timeout:.0f}s -- the bootloader did not accept the image. {hint}"
+        ),
+    )
+
+
+def _uf2_success(ns: argparse.Namespace, artifact: Path, root: Path, board_id: str,
+                 meta: dict, written: int) -> None:
+    emit_log(f"[mpftp] {root} unmounted -- board rebooted into the new firmware")
+    emit_result(
+        True,
+        method="uf2",
+        device=str(root),
+        artifact=str(artifact),
+        board_id=board_id,
+        family=meta["family_names"],
+        bytes_written=written,
+        blocks=meta["blocks"],
+    )
+    log_activity("firmware_flash", f"uf2 -> {root}", {"artifact": str(artifact)})
+
+
+def _select_uf2_volume(ns: argparse.Namespace) -> tuple[Optional[dict], Optional[str]]:
+    """Resolve which bootloader volume to write to.
+
+    Returns ``(volume, error)``. Reporting is left to the caller so that exactly
+    one result line is ever emitted -- the streaming protocol requires the
+    result to be both unique and last.
+
+    ``(None, None)`` means no volume was found, which is recoverable (picotool);
+    ``(None, message)`` means the choice was ambiguous, which is not.
+    """
+    device = getattr(ns, "device", "") or ""
+    if device and uf2.looks_like_volume(device):
+        root = Path(device)
+        return {"path": device,
+                "board_id": uf2.read_volume_info(root).get("Board-ID", "")}, None
+
+    volumes = uf2.find_uf2_volumes(HOST)
+    if len(volumes) == 1:
+        return volumes[0], None
+    if len(volumes) > 1:
+        # Never guess: the wrong choice silently overwrites the firmware on a
+        # board the caller did not name.
+        listing = ", ".join(f"{v['path']} ({v['board_id'] or 'unknown'})" for v in volumes)
+        return None, (f"Multiple UF2 bootloader volumes mounted: {listing}. "
+                      "Pass --device with the one to flash.")
+    return None, None
+
+
+def _uf2_no_volume(ns: argparse.Namespace, artifact: Path) -> None:
+    """No bootloader volume: try picotool for rp2, else explain how to get one."""
+    if getattr(ns, "port", "") == "rp2" and shutil.which("picotool"):
+        emit_log("[mpftp] no UF2 volume; trying picotool load")
         rc = stream_process(
             ["picotool", "load", "-f", "-x", str(artifact)], Path.cwd(), dict(os.environ)
         )
         emit_result(rc == 0, method="picotool", artifact=str(artifact),
                     error=None if rc == 0 else f"picotool failed (exit {rc})")
         return
-
     emit_result(
         False,
-        error="No UF2 bootloader drive found. Put the board in BOOTSEL/bootloader "
-        "mode (double-tap reset) and select its drive, or install picotool (rp2).",
+        method="uf2",
+        error="No UF2 bootloader volume found. Put the board in bootloader mode "
+              "(double-tap reset, hold BOOTSEL, or `mpftp bootloader`) and retry, "
+              "or pass --device with the volume path. On WSL a removable drive is "
+              "often not mounted under /mnt -- the Windows path (e.g. D:\\) works.",
     )
 
 
-def _looks_like_mount(dev: str) -> bool:
-    return "/" in dev or dev.endswith(":") or dev.endswith(":/")
-
-
 def do_flash(ns: argparse.Namespace) -> None:
     port = ns.port
     board = ns.board or ""
     variant = ns.variant or ""
-    if port not in FLASHERS:
+    # --uf2 is what makes this reachable for ports with no entry in FLASHERS:
+    # a board is UF2-flashable because a bootloader is running on it, which is a
+    # property of the board's provisioning rather than of its MicroPython port.
+    # An esp32 carrying tinyuf2 is the case that matters.
+    force_uf2 = bool(getattr(ns, "uf2", False))
+    if port not in FLASHERS and not force_uf2:
         emit_result(False, error=f"Flashing not supported for port '{port}'.")
         return
     mp: Optional[Path] = None
@@ -1525,7 +1626,16 @@ def do_flash(ns: argparse.Namespace) -> None:
         emit_result(False, error=f"Artifact not found: {artifact}")
         return
 
-    if port == "esp32":
+    if force_uf2:
+        if artifact.suffix.lower() != ".uf2":
+            emit_result(
+                False,
+                error=f"--uf2 needs a .uf2 artifact, got {artifact.name}. "
+                      "Build a UF2-capable target, or drop --uf2 to flash over serial.",
+            )
+            return
+        flash_uf2(ns, artifact)
+    elif port == "esp32":
         if not ns.device:
             emit_result(False, error="No device selected.")
             return
@@ -2571,6 +2681,10 @@ def add_mp(sp: argparse.ArgumentParser, required: bool = False) -> None:
     f.add_argument("--offset", default="",
                    help="esp32 flash offset override (default: board.json / chip family)")
     f.add_argument("--erase", action="store_true")
+    f.add_argument("--uf2", action="store_true",
+                   help="force the UF2 copy path (any port with a bootloader volume)")
+    f.add_argument("--uf2-timeout", dest="uf2_timeout", type=float, default=0.0,
+                   help=f"seconds to wait for the volume to unmount (default {UF2_REBOOT_TIMEOUT:.0f})")
     f.add_argument(
         "--before",
         default="default-reset",
diff --git a/python/mpftp_cli.py b/python/mpftp_cli.py
index 5767f9a..3b3b1bd 100755
--- a/python/mpftp_cli.py
+++ b/python/mpftp_cli.py
@@ -948,6 +948,10 @@ def cmd_firmware(ns: argparse.Namespace) -> None:
             extra += ["--family", ns.family]
         if getattr(ns, "erase", False):
             extra.append("--erase")
+        if getattr(ns, "uf2", False):
+            extra.append("--uf2")
+        if getattr(ns, "uf2_timeout", 0):
+            extra += ["--uf2-timeout", str(ns.uf2_timeout)]
         res = _engine_stream("flash", extra)
         out(res)
         if not res.get("ok"):
@@ -1281,6 +1285,10 @@ def build_parser() -> argparse.ArgumentParser:
     fwf.add_argument("--artifact", help="Explicit firmware file (else last build)")
     fwf.add_argument("--family", default="", help="MCU family for flash offset (download mode)")
     fwf.add_argument("--erase", action="store_true", help="esp32: erase flash first")
+    fwf.add_argument("--uf2", action="store_true",
+                     help="Copy a .uf2 to a bootloader volume instead of flashing over serial")
+    fwf.add_argument("--uf2-timeout", dest="uf2_timeout", type=float, default=0.0,
+                     help="Seconds to wait for the volume to unmount (default 30)")
     fwf.set_defaults(func=cmd_firmware)
 
     fwdt = fwsub.add_parser(
diff --git a/python/tests/test_uf2.py b/python/tests/test_uf2.py
new file mode 100644
index 0000000..f85ce3c
--- /dev/null
+++ b/python/tests/test_uf2.py
@@ -0,0 +1,221 @@
+"""UF2 file validation, volume discovery, and the copy/verify flash path."""
+
+from __future__ import annotations
+
+import struct
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import uf2
+
+
+def _block(addr: int, payload: bytes, blk_no: int, num_blocks: int,
+           family: int = 0xE48BFF56, flags: int = uf2.UF2_FLAG_FAMILY_ID,
+           start0: int = uf2.UF2_MAGIC_START0, end: int = uf2.UF2_MAGIC_END) -> bytes:
+    header = struct.pack(
+        "<8I", start0, uf2.UF2_MAGIC_START1, flags, addr,
+        len(payload), blk_no, num_blocks, family,
+    )
+    body = payload + b"\x00" * (476 - len(payload))
+    return header + body + struct.pack("<I", end)
+
+
+def _write_uf2(path: Path, blocks: list[bytes]) -> Path:
+    path.write_bytes(b"".join(blocks))
+    return path
+
+
+class ParseUf2Tests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def test_parses_a_well_formed_file(self) -> None:
+        blocks = [_block(0x10000 + i * 256, b"\xaa" * 256, i, 3) for i in range(3)]
+        meta = uf2.parse_uf2(_write_uf2(self.dir / "fw.uf2", blocks))
+        self.assertEqual(meta["blocks"], 3)
+        self.assertEqual(meta["family_names"], ["RP2040"])
+        self.assertEqual(meta["payload_bytes"], 768)
+        self.assertEqual(meta["target_start"], 0x10000)
+        self.assertEqual(meta["warnings"], [])
+
+    def test_rejects_a_truncated_file(self) -> None:
+        path = self.dir / "cut.uf2"
+        path.write_bytes(_block(0, b"\x01" * 256, 0, 1)[:400])
+        with self.assertRaises(uf2.Uf2Error) as cm:
+            uf2.parse_uf2(path)
+        self.assertIn("not a multiple", str(cm.exception))
+
+    def test_rejects_bad_magic(self) -> None:
+        path = _write_uf2(self.dir / "bad.uf2", [_block(0, b"\x01", 0, 1, start0=0xDEADBEEF)])
+        with self.assertRaises(uf2.Uf2Error):
+            uf2.parse_uf2(path)
+
+    def test_rejects_bad_end_magic(self) -> None:
+        path = _write_uf2(self.dir / "bad2.uf2", [_block(0, b"\x01", 0, 1, end=0)])
+        with self.assertRaises(uf2.Uf2Error):
+            uf2.parse_uf2(path)
+
+    def test_rejects_block_count_disagreeing_with_header(self) -> None:
+        # A bootloader waits for numBlocks blocks, so this hangs rather than flashes.
+        blocks = [_block(0, b"\x01" * 16, i, 9) for i in range(2)]
+        with self.assertRaises(uf2.Uf2Error) as cm:
+            uf2.parse_uf2(_write_uf2(self.dir / "short.uf2", blocks))
+        self.assertIn("declares 9 blocks", str(cm.exception))
+
+    def test_not_main_flash_blocks_are_not_counted(self) -> None:
+        blocks = [
+            _block(0, b"\x01" * 16, 0, 1, flags=uf2.UF2_FLAG_NOT_MAIN_FLASH),
+            _block(0, b"\x02" * 32, 0, 1),
+        ]
+        meta = uf2.parse_uf2(_write_uf2(self.dir / "meta.uf2", blocks))
+        self.assertEqual(meta["blocks"], 1)
+        self.assertEqual(meta["payload_bytes"], 32)
+
+    def test_missing_family_id_warns_but_parses(self) -> None:
+        blocks = [_block(0, b"\x01" * 16, 0, 1, flags=0)]
+        meta = uf2.parse_uf2(_write_uf2(self.dir / "nofam.uf2", blocks))
+        self.assertEqual(meta["families"], [])
+        self.assertTrue(any("no family ID" in w for w in meta["warnings"]))
+
+    def test_empty_file_is_rejected(self) -> None:
+        path = self.dir / "empty.uf2"
+        path.write_bytes(b"")
+        with self.assertRaises(uf2.Uf2Error):
+            uf2.parse_uf2(path)
+
+
+class FamilyNameTests(unittest.TestCase):
+    def test_known_families(self) -> None:
+        self.assertEqual(uf2.family_name(0xE48BFF56), "RP2040")
+        self.assertEqual(uf2.family_name(0xC47E5767), "ESP32S3")
+        self.assertEqual(uf2.family_name(0x68ED2B88), "SAMD21")
+
+    def test_unknown_family_degrades_to_hex(self) -> None:
+        self.assertEqual(uf2.family_name(0x12345678), "0x12345678")
+
+
+class InfoUf2Tests(unittest.TestCase):
+    def test_parses_board_id(self) -> None:
+        info = uf2.parse_info_uf2(
+            "UF2 Bootloader v3.0\r\nModel: Raspberry Pi RP2\r\nBoard-ID: RPI-RP2\r\n"
+        )
+        self.assertEqual(info["banner"], "UF2 Bootloader v3.0")
+        self.assertEqual(info["Board-ID"], "RPI-RP2")
+        self.assertEqual(info["Model"], "Raspberry Pi RP2")
+
+    def test_ignores_lines_without_a_colon(self) -> None:
+        info = uf2.parse_info_uf2("Banner\njunk\nBoard-ID: X\n")
+        self.assertEqual(info["Board-ID"], "X")
+        self.assertNotIn("junk", info)
+
+    def test_missing_file_gives_empty_info(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self.assertEqual(uf2.read_volume_info(Path(d)), {})
+
+
+class VolumeDiscoveryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def _volume(self, name: str, board_id: str = "RPI-RP2") -> Path:
+        root = self.dir / name
+        root.mkdir()
+        (root / "INFO_UF2.TXT").write_text(
+            f"UF2 Bootloader v3.0\nBoard-ID: {board_id}\n"
+        )
+        return root
+
+    def test_finds_volumes_by_info_file(self) -> None:
+        root = self._volume("RPI-RP2")
+        self.dir.joinpath("NotABootloader").mkdir()
+        with mock.patch.object(uf2, "_candidate_roots",
+                               return_value=[root, self.dir / "NotABootloader"]):
+            found = uf2.find_uf2_volumes("linux")
+        self.assertEqual(len(found), 1)
+        self.assertEqual(found[0]["board_id"], "RPI-RP2")
+
+    def test_no_volumes_when_no_info_file(self) -> None:
+        with mock.patch.object(uf2, "_candidate_roots", return_value=[self.dir]):
+            self.assertEqual(uf2.find_uf2_volumes("linux"), [])
+
+    def test_windows_letters_are_probed_on_wsl(self) -> None:
+        # WSL does not auto-mount removable drives, so the Windows path is the
+        # only way to reach a bootloader volume there.
+        with mock.patch.object(uf2, "_windows_volume_letters", return_value=["D:"]):
+            roots = [str(r) for r in uf2._candidate_roots("wsl")]
+        self.assertTrue(any(r.startswith("D:") or r == "/mnt/d" for r in roots))
+
+    def test_looks_like_volume(self) -> None:
+        self.assertTrue(uf2.looks_like_volume("D:"))
+        self.assertTrue(uf2.looks_like_volume("/media/brad/RPI-RP2"))
+        self.assertFalse(uf2.looks_like_volume("COM7"))
+
+
+class WaitForVolumeGoneTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+        (self.dir / "INFO_UF2.TXT").write_text("UF2 Bootloader\nBoard-ID: X\n")
+
+    def test_returns_true_once_the_volume_unmounts(self) -> None:
+        clock = {"t": 0.0}
+        info = self.dir / "INFO_UF2.TXT"
+
+        def sleep(dt: float) -> None:
+            clock["t"] += dt
+            if clock["t"] >= 1.0 and info.exists():
+                info.unlink()  # the board rebooted
+
+        self.assertTrue(
+            uf2.wait_for_volume_gone(self.dir, timeout=10.0, poll=0.25,
+                                     sleep=sleep, now=lambda: clock["t"])
+        )
+
+    def test_returns_false_when_the_volume_stays(self) -> None:
+        clock = {"t": 0.0}
+
+        def sleep(dt: float) -> None:
+            clock["t"] += dt
+
+        self.assertFalse(
+            uf2.wait_for_volume_gone(self.dir, timeout=2.0, poll=0.25,
+                                     sleep=sleep, now=lambda: clock["t"])
+        )
+
+    def test_volume_present_survives_an_oserror(self) -> None:
+        with mock.patch.object(Path, "is_file", side_effect=OSError("gone")):
+            self.assertFalse(uf2.volume_present(self.dir))
+
+
+class CopyUf2Tests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def test_copies_and_reports_bytes_written(self) -> None:
+        src = self.dir / "fw.uf2"
+        src.write_bytes(b"\xa5" * (512 * 5))
+        dest = self.dir / "out.uf2"
+        self.assertEqual(uf2.copy_uf2(src, dest), 512 * 5)
+        self.assertEqual(dest.read_bytes(), src.read_bytes())
+
+    def test_fsync_failure_is_not_an_error(self) -> None:
+        # The board reboots the moment it has the last block, so fsync on a
+        # vanished volume is expected rather than a failure.
+        src = self.dir / "fw.uf2"
+        src.write_bytes(b"\x01" * 512)
+        dest = self.dir / "out.uf2"
+        with mock.patch.object(uf2.os, "fsync", side_effect=OSError("device gone")):
+            self.assertEqual(uf2.copy_uf2(src, dest), 512)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/python/tests/test_uf2_flash.py b/python/tests/test_uf2_flash.py
new file mode 100644
index 0000000..99d7759
--- /dev/null
+++ b/python/tests/test_uf2_flash.py
@@ -0,0 +1,201 @@
+"""The flash_uf2 path in the firmware engine.
+
+These cover the failure modes that motivated the feature: a copy that reports
+success while writing nothing, and a bootloader that ignores an image whose
+family it does not own. Both look identical to a naive implementation, so each
+one has to produce a distinguishable, non-ok result.
+"""
+
+from __future__ import annotations
+
+import argparse
+import struct
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import firmware_engine as fe
+import uf2
+
+
+def _uf2_bytes(nblocks: int = 2, family: int = 0xE48BFF56) -> bytes:
+    out = []
+    for i in range(nblocks):
+        header = struct.pack(
+            "<8I", uf2.UF2_MAGIC_START0, uf2.UF2_MAGIC_START1,
+            uf2.UF2_FLAG_FAMILY_ID, 0x10000 + i * 256, 256, i, nblocks, family,
+        )
+        out.append(header + b"\xaa" * 256 + b"\x00" * 220
+                   + struct.pack("<I", uf2.UF2_MAGIC_END))
+    return b"".join(out)
+
+
+class FlashUf2Tests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+        self.artifact = self.dir / "firmware.uf2"
+        self.artifact.write_bytes(_uf2_bytes())
+
+        self.volume = self.dir / "RPI-RP2"
+        self.volume.mkdir()
+        (self.volume / "INFO_UF2.TXT").write_text(
+            "UF2 Bootloader v3.0\nBoard-ID: RPI-RP2\n"
+        )
+
+        self.results: list[dict] = []
+        self.logs: list[str] = []
+        patches = [
+            mock.patch.object(fe, "emit_result",
+                              side_effect=lambda ok, **kw: self.results.append(dict(ok=ok, **kw))),
+            mock.patch.object(fe, "emit_log", side_effect=self.logs.append),
+            mock.patch.object(fe, "emit_phase", lambda *a, **k: None),
+            mock.patch.object(fe, "log_activity", lambda *a, **k: None),
+        ]
+        for patch in patches:
+            patch.start()
+            self.addCleanup(patch.stop)
+
+    def _ns(self, **kw) -> argparse.Namespace:
+        base = dict(device=str(self.volume), port="rp2", uf2_timeout=1.0)
+        base.update(kw)
+        return argparse.Namespace(**base)
+
+    @property
+    def result(self) -> dict:
+        self.assertEqual(len(self.results), 1, f"expected one result, got {self.results}")
+        return self.results[0]
+
+    def test_success_when_the_volume_unmounts(self) -> None:
+        with mock.patch.object(uf2, "wait_for_volume_gone", return_value=True):
+            fe.flash_uf2(self._ns(), self.artifact)
+        self.assertTrue(self.result["ok"])
+        self.assertEqual(self.result["method"], "uf2")
+        self.assertEqual(self.result["board_id"], "RPI-RP2")
+        self.assertEqual(self.result["family"], ["RP2040"])
+        self.assertEqual(self.result["bytes_written"], self.artifact.stat().st_size)
+        self.assertEqual((self.volume / "firmware.uf2").read_bytes(),
+                         self.artifact.read_bytes())
+
+    def test_volume_still_mounted_is_a_failure_naming_the_family(self) -> None:
+        # The copy succeeded; the bootloader ignored the image. Reporting ok
+        # here is the exact bug this path exists to prevent.
+        with mock.patch.object(uf2, "wait_for_volume_gone", return_value=False):
+            fe.flash_uf2(self._ns(), self.artifact)
+        self.assertFalse(self.result["ok"])
+        self.assertIn("still mounted", self.result["error"])
+        self.assertIn("RP2040", self.result["error"])
+
+    def test_short_write_is_reported_rather_than_claimed_as_success(self) -> None:
+        with mock.patch.object(uf2, "copy_uf2", return_value=16):
+            fe.flash_uf2(self._ns(), self.artifact)
+        self.assertFalse(self.result["ok"])
+        self.assertIn("Short write", self.result["error"])
+
+    def test_invalid_uf2_is_rejected_before_any_copy(self) -> None:
+        bad = self.dir / "bad.uf2"
+        bad.write_bytes(b"not a uf2 at all")
+        fe.flash_uf2(self._ns(), bad)
+        self.assertFalse(self.result["ok"])
+        self.assertFalse((self.volume / "bad.uf2").exists())
+
+    def test_copy_error_after_an_early_reboot_is_a_success(self) -> None:
+        with mock.patch.object(uf2, "copy_uf2", side_effect=OSError("device gone")), \
+             mock.patch.object(uf2, "wait_for_volume_gone", return_value=True):
+            fe.flash_uf2(self._ns(), self.artifact)
+        self.assertTrue(self.result["ok"])
+
+    def test_copy_error_with_the_volume_still_there_is_a_failure(self) -> None:
+        with mock.patch.object(uf2, "copy_uf2", side_effect=OSError("no space")), \
+             mock.patch.object(uf2, "wait_for_volume_gone", return_value=False):
+            fe.flash_uf2(self._ns(), self.artifact)
+        self.assertFalse(self.result["ok"])
+        self.assertIn("no space", self.result["error"])
+
+    def test_no_volume_explains_how_to_get_one(self) -> None:
+        with mock.patch.object(uf2, "find_uf2_volumes", return_value=[]), \
+             mock.patch.object(fe.shutil, "which", return_value=None):
+            fe.flash_uf2(self._ns(device=""), self.artifact)
+        self.assertFalse(self.result["ok"])
+        self.assertIn("bootloader mode", self.result["error"])
+
+    def test_multiple_volumes_refuses_to_guess(self) -> None:
+        volumes = [
+            {"path": "/media/a", "board_id": "RPI-RP2", "info": {}},
+            {"path": "/media/b", "board_id": "FTHRS3BOOT", "info": {}},
+        ]
+        with mock.patch.object(uf2, "find_uf2_volumes", return_value=volumes):
+            fe.flash_uf2(self._ns(device=""), self.artifact)
+        self.assertFalse(self.result["ok"])
+        self.assertIn("Multiple UF2 bootloader volumes", self.result["error"])
+        self.assertIn("RPI-RP2", self.result["error"])
+        self.assertFalse((self.volume / "firmware.uf2").exists())
+
+    def test_single_discovered_volume_is_used(self) -> None:
+        found = [{"path": str(self.volume), "board_id": "RPI-RP2", "info": {}}]
+        with mock.patch.object(uf2, "find_uf2_volumes", return_value=found), \
+             mock.patch.object(uf2, "wait_for_volume_gone", return_value=True):
+            fe.flash_uf2(self._ns(device=""), self.artifact)
+        self.assertTrue(self.result["ok"])
+
+
+class DoFlashUf2RoutingTests(unittest.TestCase):
+    """--uf2 must reach ports that have no serial flasher (esp32 + tinyuf2)."""
+
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+        self.results: list[dict] = []
+        patch = mock.patch.object(
+            fe, "emit_result",
+            side_effect=lambda ok, **kw: self.results.append(dict(ok=ok, **kw)),
+        )
+        patch.start()
+        self.addCleanup(patch.stop)
+        patch2 = mock.patch.object(fe, "save_state", lambda *a, **k: None)
+        patch2.start()
+        self.addCleanup(patch2.stop)
+
+    def _ns(self, artifact: Path, **kw) -> argparse.Namespace:
+        base = dict(port="esp32", board="", variant="", mp=None, device="",
+                    artifact=str(artifact), uf2=True, uf2_timeout=1.0)
+        base.update(kw)
+        return argparse.Namespace(**base)
+
+    def test_uf2_flag_routes_esp32_to_the_uf2_path(self) -> None:
+        art = self.dir / "firmware.uf2"
+        art.write_bytes(_uf2_bytes(family=0xC47E5767))
+        with mock.patch.object(fe, "flash_uf2") as flash:
+            fe.do_flash(self._ns(art))
+        flash.assert_called_once()
+
+    def test_uf2_flag_rejects_a_non_uf2_artifact(self) -> None:
+        art = self.dir / "firmware.bin"
+        art.write_bytes(b"\x00" * 64)
+        with mock.patch.object(fe, "flash_uf2") as flash:
+            fe.do_flash(self._ns(art))
+        flash.assert_not_called()
+        self.assertFalse(self.results[0]["ok"])
+        self.assertIn("needs a .uf2 artifact", self.results[0]["error"])
+
+    def test_unflashable_port_still_rejected_without_the_flag(self) -> None:
+        art = self.dir / "firmware.uf2"
+        art.write_bytes(_uf2_bytes())
+        fe.do_flash(self._ns(art, port="stm32", uf2=False))
+        self.assertFalse(self.results[0]["ok"])
+        self.assertIn("not supported", self.results[0]["error"])
+
+    def test_unflashable_port_is_allowed_with_the_flag(self) -> None:
+        art = self.dir / "firmware.uf2"
+        art.write_bytes(_uf2_bytes())
+        with mock.patch.object(fe, "flash_uf2") as flash:
+            fe.do_flash(self._ns(art, port="stm32"))
+        flash.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/python/uf2.py b/python/uf2.py
new file mode 100644
index 0000000..a75cd4b
--- /dev/null
+++ b/python/uf2.py
@@ -0,0 +1,428 @@
+#!/usr/bin/env python3
+"""
+UF2 support for the mpftp firmware engine: file validation, bootloader-volume
+discovery, and the copy-and-verify flash itself.
+
+Flashing over UF2 is a file copy, which makes the *failure* modes the hard part
+rather than the success path. Two in particular drove this module:
+
+* A copy can report success and write nothing. PowerShell's ``Copy-Item`` fails
+  non-terminatingly (exit 0, no file), and for a flash operation "succeeded" and
+  "wrote nothing" being indistinguishable is the worst possible ambiguity.
+* A bootloader silently ignores UF2 blocks whose family ID it does not own. The
+  copy then genuinely succeeds, the drive stays mounted, and nothing is flashed.
+
+So the copy is never the proof. The proof is that the volume *disappears*: a UF2
+bootloader reboots into the new firmware once it has accepted a full image, and
+the mount goes with it. ``flash_uf2_file`` therefore waits for the volume to go
+away and treats a volume that is still there as a failure -- see the module's
+one real assumption, documented on ``wait_for_volume_gone``.
+
+Stdlib only, like the rest of the engine.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import struct
+import subprocess
+import time
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+# UF2 block format: 512 bytes, 32-byte header, 476-byte payload area, 4-byte end
+# magic. https://github.com/microsoft/uf2
+UF2_MAGIC_START0 = 0x0A324655  # "UF2\n"
+UF2_MAGIC_START1 = 0x9E5D5157
+UF2_MAGIC_END = 0x0AB16F30
+UF2_BLOCK_SIZE = 512
+
+# Header flags we care about.
+UF2_FLAG_NOT_MAIN_FLASH = 0x00000001
+UF2_FLAG_FILE_CONTAINER = 0x00001000
+UF2_FLAG_FAMILY_ID = 0x00002000
+UF2_FLAG_MD5_CHECKSUM = 0x00004000
+
+# Family IDs from the upstream registry (microsoft/uf2 utils/uf2families.json).
+# Vendored rather than read from a sibling checkout so the engine stays
+# self-contained; unknown IDs degrade to hex, never to an error.
+#
+# Note that this registry is not universally honoured: CircuitPython's espressif
+# Makefile builds esp32p4 images with 0x540ddf62, which the registry assigns to
+# ESP32-C6. That disagreement is exactly why a family mismatch is reported as
+# information rather than enforced as a hard failure.
+UF2_FAMILIES: dict[int, str] = {
+    0x00ff6919: "STM32L4",
+    0x04240bdf: "STM32L5",
+    0x06d1097b: "STM32F411xC",
+    0x11de784a: "M0SENSE",
+    0x16573617: "ATMEGA32",
+    0x1851780a: "SAML21",
+    0x1b57745f: "NRF52",
+    0x1c5f21b0: "ESP32",
+    0x1e1f432d: "STM32L1",
+    0x202e3a91: "STM32L0",
+    0x21460ff0: "STM32WL",
+    0x22e0d6fc: "RTL8710B",
+    0x2abc77ec: "LPC55",
+    0x2b88d29c: "ESP32C2",
+    0x2dc309c5: "STM32F411xE",
+    0x300f5633: "STM32G0",
+    0x3101f7c1: "ESP32S31",
+    0x31d228c6: "GD32F350",
+    0x332726f6: "ESP32H2",
+    0x3379CFE2: "RTL8720D",
+    0x3d308e94: "ESP32P4",
+    0x4b684d71: "MaixPlay-U4",
+    0x4c71240a: "STM32G4",
+    0x4e8f1c5d: "STM32H5",
+    0x4f6ace52: "CSK4",
+    0x4fb2d5bd: "MIMXRT10XX",
+    0x51e903a8: "XR809",
+    0x53b80f00: "STM32F7",
+    0x540ddf62: "ESP32C6",
+    0x55114460: "SAMD51",
+    0x57755a57: "STM32F4",
+    0x5a18069b: "FX2",
+    0x5d1a0a2e: "STM32F2",
+    0x5ee21072: "STM32F1",
+    0x621e937a: "NRF52833",
+    0x647824b6: "STM32F0",
+    0x675a40b0: "BK7231U",
+    0x68ed2b88: "SAMD21",
+    0x699b62ec: "CH32V",
+    0x6a82cc42: "BK7251",
+    0x6b846188: "STM32F3",
+    0x6d0922fa: "STM32F407",
+    0x6db66082: "STM32H7",
+    0x6e7348a8: "CSK6",
+    0x6f752678: "NRF52832xxAB",
+    0x70d16653: "STM32WB",
+    0x72721d4e: "NRF52832xxAA",
+    0x7410520a: "MAX32690",
+    0x77d850c4: "ESP32C61",
+    0x7b3ef230: "BK7231N",
+    0x7be8976d: "RA4M1",
+    0x7d7a66ef: "PY32F071-UVK5-V3",
+    0x7eab61ed: "ESP8266",
+    0x7f83e793: "KL32L2",
+    0x820d9a5f: "NRF52820",
+    0x8fb060fe: "STM32F407VG",
+    0x91d3fd18: "MAX78002",
+    0x9517422f: "RZA1LU",
+    0x9af03e33: "GD32VF103",
+    0x9e0baa8a: "ESP32H4",
+    0x9fffd543: "RTL8710A",
+    0xa0c97b8e: "AT32F415",
+    0xada52840: "NRF52840",
+    0xb6dd00af: "ESP32H21",
+    0xbfdd4eee: "ESP32S2",
+    0xc47e5767: "ESP32S3",
+    0xd42ba06c: "ESP32C3",
+    0xd63f8632: "MAX32650",
+    0xde1270b7: "BL602",
+    0xe08f7564: "RTL8720C",
+    0xe48bff56: "RP2040",
+    0xe48bff57: "RP2XXX_ABSOLUTE",
+    0xe48bff58: "RP2XXX_DATA",
+    0xe48bff59: "RP2350_ARM_S",
+    0xe48bff5a: "RP2350_RISCV",
+    0xe48bff5b: "RP2350_ARM_NS",
+    0xf0c30d71: "MAX32666",
+    0xf71c0343: "ESP32C5",
+}
+
+
+def family_name(fid: int) -> str:
+    """Human name for a family ID, falling back to hex for unknown ones."""
+    return UF2_FAMILIES.get(fid, "0x%08x" % fid)
+
+
+# --------------------------------------------------------------------------- #
+# UF2 file validation
+# --------------------------------------------------------------------------- #
+
+class Uf2Error(Exception):
+    """A .uf2 file that no bootloader would accept."""
+
+
+def parse_uf2(path: Path, max_blocks: int = 0) -> dict:
+    """Validate a .uf2 and summarise what a bootloader would make of it.
+
+    Returns ``{"blocks", "families", "family_names", "payload_bytes",
+    "target_start", "target_end", "warnings"}``.
+
+    Raises :class:`Uf2Error` for anything structurally wrong -- a truncated
+    file, bad magic, or a block count that disagrees with the header. Checking
+    this up front matters because the bootloader's response to a bad image is
+    to ignore it in silence, which is indistinguishable from a failed copy.
+    """
+    size = path.stat().st_size
+    if size == 0:
+        raise Uf2Error(f"{path.name} is empty")
+    if size % UF2_BLOCK_SIZE:
+        raise Uf2Error(
+            f"{path.name} is {size} bytes, not a multiple of {UF2_BLOCK_SIZE} "
+            "-- truncated or not a UF2 file"
+        )
+
+    total = size // UF2_BLOCK_SIZE
+    families: list[int] = []
+    warnings: list[str] = []
+    payload_bytes = 0
+    addr_lo: Optional[int] = None
+    addr_hi: Optional[int] = None
+    declared_total: Optional[int] = None
+    counted = 0
+
+    with path.open("rb") as fh:
+        for index in range(total):
+            block = fh.read(UF2_BLOCK_SIZE)
+            if len(block) != UF2_BLOCK_SIZE:
+                raise Uf2Error(f"{path.name}: short read at block {index}")
+            (start0, start1, flags, addr, payload_len,
+             blk_no, num_blocks, family) = struct.unpack("<8I", block[:32])
+            if start0 != UF2_MAGIC_START0 or start1 != UF2_MAGIC_START1:
+                raise Uf2Error(f"{path.name}: bad start magic at block {index}")
+            if struct.unpack("<I", block[-4:])[0] != UF2_MAGIC_END:
+                raise Uf2Error(f"{path.name}: bad end magic at block {index}")
+            if payload_len > 476:
+                raise Uf2Error(
+                    f"{path.name}: block {index} claims {payload_len} payload bytes (max 476)"
+                )
+
+            # Blocks flagged NOT_MAIN_FLASH are metadata the bootloader skips;
+            # they must not count toward what actually gets written.
+            if flags & UF2_FLAG_NOT_MAIN_FLASH:
+                continue
+            counted += 1
+            payload_bytes += payload_len
+            if declared_total is None:
+                declared_total = num_blocks
+            if flags & UF2_FLAG_FAMILY_ID:
+                if family not in families:
+                    families.append(family)
+            addr_lo = addr if addr_lo is None else min(addr_lo, addr)
+            addr_hi = addr + payload_len if addr_hi is None else max(addr_hi, addr + payload_len)
+
+            if max_blocks and index + 1 >= max_blocks:
+                break
+
+    if counted == 0:
+        raise Uf2Error(f"{path.name}: no flashable blocks")
+    if not max_blocks and declared_total is not None and declared_total != counted:
+        # A bootloader waits for numBlocks blocks before committing, so a file
+        # that disagrees with its own header will hang rather than flash.
+        raise Uf2Error(
+            f"{path.name}: header declares {declared_total} blocks but the file has {counted}"
+        )
+    if not families:
+        # Legal, but it means the file targets whatever it is copied onto.
+        warnings.append(
+            "no family ID in this UF2 -- it cannot be checked against the board"
+        )
+    elif len(families) > 1:
+        warnings.append(
+            "multiple family IDs: " + ", ".join(family_name(f) for f in families)
+        )
+
+    return {
+        "blocks": counted,
+        "families": families,
+        "family_names": [family_name(f) for f in families],
+        "payload_bytes": payload_bytes,
+        "target_start": addr_lo,
+        "target_end": addr_hi,
+        "warnings": warnings,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Bootloader volume discovery
+# --------------------------------------------------------------------------- #
+
+def parse_info_uf2(text: str) -> dict:
+    """Parse INFO_UF2.TXT into a dict.
+
+    The first line is a free-form banner ("UF2 Bootloader v3.0"); the rest are
+    ``Key: Value``. Only ``Board-ID`` is reliably present across bootloaders.
+    """
+    info: dict[str, str] = {}
+    lines = text.splitlines()
+    if lines:
+        info["banner"] = lines[0].strip()
+    for line in lines[1:]:
+        if ":" not in line:
+            continue
+        key, _, value = line.partition(":")
+        key = key.strip()
+        if key:
+            info[key] = value.strip()
+    return info
+
+
+def read_volume_info(root: Path) -> dict:
+    """Read INFO_UF2.TXT from a mounted bootloader volume, or {} if unreadable."""
+    try:
+        return parse_info_uf2((root / "INFO_UF2.TXT").read_text(errors="replace"))
+    except Exception:
+        return {}
+
+
+def _windows_volume_letters(timeout: int = 15) -> list[str]:
+    """Drive letters of removable/fixed volumes, via PowerShell.
+
+    Used from WSL as well as Windows: WSL does not auto-mount removable drives,
+    so ``/mnt/d`` is typically absent for exactly the volume we are looking for
+    and a filesystem scan alone would miss every bootloader drive.
+    """
+    ps = (
+        "Get-CimInstance Win32_LogicalDisk | "
+        "Where-Object { $_.DriveType -eq 2 -or $_.DriveType -eq 3 } | "
+        "ForEach-Object { $_.DeviceID }"
+    )
+    try:
+        out = subprocess.check_output(
+            ["powershell.exe", "-NoProfile", "-Command", ps],
+            text=True,
+            timeout=timeout,
+            stderr=subprocess.DEVNULL,
+        )
+    except Exception:
+        return []
+    letters = []
+    for line in out.splitlines():
+        letter = line.strip().rstrip("\\/")
+        if len(letter) >= 2 and letter[1] == ":":
+            letters.append(letter)
+    return letters
+
+
+def _candidate_roots(host: str) -> list[Path]:
+    """Every path that might be a mounted UF2 volume, most specific first."""
+    roots: list[Path] = []
+    if host in ("wsl", "windows"):
+        for letter in _windows_volume_letters():
+            # Prefer a WSL mount when one exists; a Windows-side path still
+            # works for a Windows python and is the only option under WSL when
+            # the removable drive was never mounted.
+            wsl = Path(f"/mnt/{letter[0].lower()}")
+            roots.append(wsl if wsl.is_dir() else Path(letter + "\\"))
+    if host in ("wsl", "linux", "macos"):
+        for base in ("/media", "/run/media", "/Volumes", str(Path.home() / "media")):
+            base_path = Path(base)
+            try:
+                if not base_path.is_dir():
+                    continue
+                for child in base_path.iterdir():
+                    if not child.is_dir():
+                        continue
+                    roots.append(child)
+                    # /media/<user>/<LABEL>
+                    try:
+                        roots.extend(g for g in child.iterdir() if g.is_dir())
+                    except OSError:
+                        pass
+            except OSError:
+                continue
+    seen: set[str] = set()
+    uniq: list[Path] = []
+    for r in roots:
+        key = str(r)
+        if key not in seen:
+            seen.add(key)
+            uniq.append(r)
+    return uniq
+
+
+def find_uf2_volumes(host: str) -> list[dict]:
+    """Mounted UF2 bootloader volumes: ``[{"path", "info", "board_id"}]``.
+
+    Identified by INFO_UF2.TXT, which every UF2 bootloader exposes and nothing
+    else does -- more reliable than matching volume labels, which differ per
+    board family (RPI-RP2, FTHRS3BOOT, ...).
+    """
+    found: list[dict] = []
+    for root in _candidate_roots(host):
+        try:
+            if not (root / "INFO_UF2.TXT").is_file():
+                continue
+        except OSError:
+            continue
+        info = read_volume_info(root)
+        found.append({
+            "path": str(root),
+            "info": info,
+            "board_id": info.get("Board-ID", ""),
+        })
+    return found
+
+
+def looks_like_volume(dev: str) -> bool:
+    """True when a --device value names a mount point rather than a serial port."""
+    return "/" in dev or dev.endswith(":") or dev.endswith(":\\")
+
+
+# --------------------------------------------------------------------------- #
+# Copy + verify
+# --------------------------------------------------------------------------- #
+
+def volume_present(root: Path) -> bool:
+    """True while the bootloader volume is still mounted."""
+    try:
+        return (root / "INFO_UF2.TXT").is_file()
+    except OSError:
+        # A vanishing removable volume can raise rather than return False.
+        return False
+
+
+def wait_for_volume_gone(root: Path, timeout: float, poll: float = 0.25,
+                         sleep: Callable[[float], None] = time.sleep,
+                         now: Callable[[], float] = time.monotonic) -> bool:
+    """Wait for the bootloader volume to unmount, which is what proves the flash.
+
+    A UF2 bootloader reboots into the new firmware once it has received a
+    complete, acceptable image, and the mount goes away with it. That makes
+    disappearance a far stronger signal than a successful copy: it is reported
+    by the board, not by the host filesystem.
+
+    The assumption -- and the one thing that would make this wrong -- is a
+    bootloader that accepts an image without rebooting. None of the bootloaders
+    in scope (RP2040/RP2350, SAMD, nRF, tinyuf2) behave that way, but a false
+    negative here reports failure after a flash that actually worked, so the
+    caller must say so rather than claim the image was rejected.
+    """
+    deadline = now() + timeout
+    while now() < deadline:
+        if not volume_present(root):
+            return True
+        sleep(poll)
+    return not volume_present(root)
+
+
+def copy_uf2(src: Path, dest: Path) -> int:
+    """Copy a UF2 onto a bootloader volume, returning the bytes written.
+
+    Written with an explicit loop and flushed to the device rather than via
+    ``shutil.copyfile`` so that a short write is an error here instead of a
+    mystery later. The board can reboot mid-copy once it has the last block, so
+    errors *after* the final write are expected and are left for the caller's
+    disappearance check to adjudicate.
+    """
+    written = 0
+    with src.open("rb") as fin, dest.open("wb") as fout:
+        while True:
+            chunk = fin.read(UF2_BLOCK_SIZE * 64)
+            if not chunk:
+                break
+            fout.write(chunk)
+            written += len(chunk)
+        try:
+            fout.flush()
+            os.fsync(fout.fileno())
+        except OSError:
+            # The board rebooted as soon as it had the last block. Not an
+            # error by itself -- wait_for_volume_gone decides.
+            pass
+    return written