fix(node): index, serve and write files past Windows' MAX_PATHHEADmain

With the system's default setting (LongPathsEnabled off) every call given
a plain path of 260 characters or more refuses it, so a deep library was
partly invisible: the walk could not list such a directory and its files
never reached the members.

The node now hands each call that touches a group's disk the extended
form (paths.long_path), and keeps the plain path in the index, the hash
cache and the log. A directory is listed under its long form and its
entries named from the plain one (indexer.walk_files, wire._walk_dirs).
Covered: the walk, stats and hashing, reconcile and the watcher, the
folder list members see, serving over WebRTC and QUIC, deletion, uploads
and their recovery, folder creation and removal, audio tags and the
cover beside a track, photos, and a show's episode numbering. ffmpeg
extends a long path itself and keeps the plain one.

long_path now normalises before prefixing (the extended form skips the
system's normalisation) and starts at 248 characters, where creating a
directory is already refused.

test_long_paths.py runs with the setting off: 15 of its 17 tests fail
without the change. Checked by hand in the installed app with files at
312 to 380 characters.

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

3 files changed, 406 insertions, 6 deletions

diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py
index c0ab052..cacf936 100644
--- a/packages/meshbay-node/tests/test_disk_io_off_loop.py
+++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py
@@ -260,7 +260,7 @@ def test_no_handler_touches_the_disk_on_the_loop():
     # Written to block, and reached only through `off_disk`.
     on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt",
                           "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty",
-                          "safe_subdir"}
+                          "_unlink", "_exists", "safe_subdir"}
     # ffmpeg's own output goes through `_read_scratch_capped` and
     # `_discard_scratch` on a worker thread — `asyncio.to_thread` and not
     # `off_disk`, because a temp file is not a group root and has no platter to
diff --git a/packages/meshbay-node/tests/test_indexer_silent_disk.py b/packages/meshbay-node/tests/test_indexer_silent_disk.py
index bb618fc..4073ac5 100644
--- a/packages/meshbay-node/tests/test_indexer_silent_disk.py
+++ b/packages/meshbay-node/tests/test_indexer_silent_disk.py
@@ -133,14 +133,14 @@ async def test_a_file_the_system_refuses_is_listed_with_what_it_said(
         tmp_path, monkeypatch):
     films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"])
     idx = _indexer(RootSet.build([{"path": str(films)}]))
-    real = indexer_mod.long_path
 
-    def refuse(path):
+    def refuse(path, *args, **kw):
+        # What a locked file does: it lists and stats, and refuses to open.
         if Path(path).name == "locked.mkv":
             raise PermissionError(13, "Permission denied")
-        return real(path)
+        return open(path, *args, **kw)
 
-    monkeypatch.setattr(indexer_mod, "long_path", refuse)
+    monkeypatch.setattr(indexer_mod, "open", refuse, raising=False)
     await idx.initial_scan()
 
     assert _names(idx) == {"a.mkv"}
@@ -149,7 +149,7 @@ async def test_a_file_the_system_refuses_is_listed_with_what_it_said(
 
     # A refusal costs nothing to ask again, and permissions change without the
     # file changing: it is not remembered the way a silence is.
-    monkeypatch.setattr(indexer_mod, "long_path", real)
+    monkeypatch.delattr(indexer_mod, "open")
     await idx.reconcile()
     assert "locked.mkv" in _names(idx)
     assert idx.unread == {}
diff --git a/packages/meshbay-node/tests/test_long_paths.py b/packages/meshbay-node/tests/test_long_paths.py
new file mode 100644
index 0000000..9f7876a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_long_paths.py
@@ -0,0 +1,400 @@
+"""
+Files whose full path is longer than Windows' MAX_PATH (260 characters).
+
+A deep library reaches it without trying: "Artist / Album (year, edition) /
+01 - a long track title.flac" below a folder that is already deep. With the
+system's default setting (`LongPathsEnabled` off) every call given a plain path
+that long refuses it, whatever the program declares, and such a file was
+invisible to the members: the walk could not list its directory, and said so
+only in the log. The node now hands the system the extended form (`\\\\?\\`) at
+each call that touches a group's disk, which works whatever the setting, and
+keeps the plain form everywhere else: in the index, the hash cache, the log.
+
+ffmpeg and ffprobe are given the plain form on purpose: they extend a long path
+themselves, which the probe and thumbnail tests below hold.
+
+These run only where the limit exists: with the setting on, a plain path works
+too and they would prove nothing.
+"""
+
+import asyncio
+import os
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.protocol import file_chunk_plaintext
+from meshbay_node import ops
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet, publish_upload, safe_subdir
+from meshbay_node.transport.webrtc.disk import (
+    _append_chunk,
+    _free_bytes,
+    _is_empty_dir,
+    _locate,
+    _mkdir_if_absent,
+    _read_and_encrypt,
+    _rmdir_if_empty,
+)
+
+
+def _limit_in_force() -> bool:
+    if sys.platform != "win32":
+        return False
+    import winreg
+    try:
+        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
+                            r"SYSTEM\CurrentControlSet\Control\FileSystem") as key:
+            return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0
+    except OSError:
+        return True
+
+
+pytestmark = pytest.mark.skipif(
+    not _limit_in_force(), reason="MAX_PATH binds only on Windows with LongPathsEnabled off")
+
+GID = "g" * 32
+SEGMENT = "A folder name as long as the ones a real library has"
+
+
+def _x(path: Path) -> str:
+    """The extended form, for the test's own setup and checks."""
+    return "\\\\?\\" + str(path)
+
+
+def _write(path: Path, data: bytes) -> Path:
+    with open(_x(path), "wb") as f:
+        f.write(data)
+    return path
+
+
+def _read(path: Path) -> bytes:
+    with open(_x(path), "rb") as f:
+        return f.read()
+
+
+@pytest.fixture
+def library(tmp_path):
+    """A root with one short file at the top and a directory past the limit.
+    Removed here: the plain-path cleanup pytest does would stop at it."""
+    root = tmp_path / "Library"
+    deep = root
+    while len(str(deep)) < 280:
+        deep = deep / SEGMENT
+    os.makedirs(_x(deep))
+    _write(root / "short.txt", b"near the top")
+    yield root, deep
+    shutil.rmtree(_x(root), ignore_errors=True)
+
+
+def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
+    return DirectoryIndexer(roots=roots, group_id=GID,
+                            sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)
+
+
+def _names(idx: DirectoryIndexer) -> set[str]:
+    return {e.name for e in idx.index.entries}
+
+
+def _entry(idx: DirectoryIndexer, name: str):
+    (entry,) = [e for e in idx.index.entries if e.name == name]
+    return entry
+
+
+def _ffmpeg() -> str:
+    exe = shutil.which("ffmpeg")
+    if exe is None:
+        pytest.skip("ffmpeg is not on PATH")
+    return exe
+
+
+async def _make_media(args: list[str], out: Path) -> None:
+    proc = await asyncio.create_subprocess_exec(
+        _ffmpeg(), "-v", "error", "-y", *args, str(out),
+        stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE)
+    _, err = await proc.communicate()
+    assert proc.returncode == 0, err.decode(errors="replace")
+
+
+def test_the_fixture_is_past_the_limit(library):
+    """The premise: a plain path this long is refused on this machine."""
+    _, deep = library
+    track = _write(deep / "track.mkv", b"deep down")
+    assert len(str(track)) > 260
+    assert not os.path.exists(str(track))
+    with pytest.raises(OSError):
+        os.scandir(str(deep))
+
+
+# ── Indexing ────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_scan_indexes_a_file_past_the_limit(library):
+    root, deep = library
+    _write(deep / "track.mkv", b"deep down" * 100)
+    idx = _indexer(RootSet.build([{"path": str(root)}]))
+
+    await idx.initial_scan()
+
+    assert _names(idx) == {"short.txt", "track.mkv"}
+    assert _entry(idx, "track.mkv").size == 900
+    assert idx.unread == {}
+
+
+@pytest.mark.asyncio
+async def test_reconcile_finds_a_new_one_and_leaves_a_known_one_alone(library):
+    """Reconcile compares the disk with the index by path. Were the two sides
+    to spell a long path differently, it would rewrite the entry at every
+    pass and push the change to every member."""
+    root, deep = library
+    _write(deep / "track.mkv", b"deep down")
+    idx = _indexer(RootSet.build([{"path": str(root)}]))
+    await idx.initial_scan()
+
+    _write(deep / "later.mkv", b"added while the node ran")
+    await idx.reconcile()
+    assert "later.mkv" in _names(idx)
+
+    version = idx.index.version
+    await idx.reconcile()
+    assert idx.index.version == version, "a pass with nothing new changed the index"
+
+
+@pytest.mark.asyncio
+async def test_a_watcher_event_indexes_it(library):
+    root, deep = library
+    idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.01)
+    await idx.initial_scan()
+    idx._loop = asyncio.get_running_loop()
+
+    _write(deep / "arrived.mkv", b"one more")
+    idx._schedule_update(deep / "arrived.mkv")
+    for _ in range(200):
+        if "arrived.mkv" in _names(idx):
+            break
+        await asyncio.sleep(0.01)
+
+    assert "arrived.mkv" in _names(idx)
+
+
+@pytest.mark.asyncio
+async def test_the_real_watcher_sees_it(library):
+    root, deep = library
+    idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.05)
+    await idx.start()
+    try:
+        _write(deep / "watched.mkv", b"seen by the system's own notification")
+        for _ in range(100):
+            if "watched.mkv" in _names(idx):
+                break
+            await asyncio.sleep(0.05)
+    finally:
+        await idx.stop()
+
+    assert "watched.mkv" in _names(idx)
+
+
+@pytest.mark.asyncio
+async def test_a_root_that_is_itself_past_the_limit(library):
+    _, deep = library
+    _write(deep / "track.mkv", b"deep down")
+    roots = RootSet.build([{"path": str(deep)}])
+    assert roots.roots[0].is_live()
+    idx = _indexer(roots)
+
+    await idx.initial_scan()
+
+    assert _names(idx) == {"track.mkv"}
+
+
+# ── Serving and deleting ────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_it_is_served_to_a_member(library):
+    root, deep = library
+    data = os.urandom(5000)
+    _write(deep / "track.mkv", data)
+    roots = RootSet.build([{"path": str(root)}])
+    idx = _indexer(roots)
+    await idx.initial_scan()
+    entry = _entry(idx, "track.mkv")
+    gek = os.urandom(32)
+
+    path, refusal = _locate(roots, entry)
+    assert refusal is None
+    msg = _read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id)
+
+    assert file_chunk_plaintext(gek, msg) == data
+
+
+@pytest.mark.asyncio
+async def test_it_is_served_over_quic_too(library):
+    from meshbay_node.transport import quic_server
+
+    root, deep = library
+    _write(deep / "track.mkv", b"over the other transport")
+    roots = RootSet.build([{"path": str(root)}])
+    idx = _indexer(roots)
+    await idx.initial_scan()
+    entry = _entry(idx, "track.mkv")
+    gek = os.urandom(32)
+
+    path, refusal = _locate(roots, entry)
+    msg = quic_server._read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id)
+
+    assert file_chunk_plaintext(gek, msg) == b"over the other transport"
+
+
+@pytest.mark.asyncio
+async def test_the_operator_can_delete_it(library):
+    root, deep = library
+    track = _write(deep / "track.mkv", b"to be deleted")
+    roots = RootSet.build([{"path": str(root)}])
+    idx = _indexer(roots)
+    await idx.initial_scan()
+    state = {"groups_ctx": {GID: {"index": idx.index, "roots": roots}}}
+
+    await ops.delete_file(state, GID, _entry(idx, "track.mkv").id)
+
+    assert not os.path.exists(_x(track))
+    assert "track.mkv" not in _names(idx)
+
+
+# ── Uploads and folders ─────────────────────────────────────────────────────
+
+def test_an_upload_lands_in_a_directory_past_the_limit(library):
+    root, deep = library
+    roots = RootSet.build([{"path": str(root), "writable": True}])
+    rel = "/".join(["Library", *deep.relative_to(root).parts])
+    target = safe_subdir(roots, rel)
+    assert target == deep
+
+    assert _free_bytes(target) > 0
+    part = target / "up.bin.0123abcd.part"
+    _append_chunk(part, b"first ", True)
+    _append_chunk(part, b"second", False)
+    assert publish_upload(part, target, "up.bin", "up.bin") == "up.bin"
+    assert _read(target / "up.bin") == b"first second"
+
+    again = target / "up.bin.4567cdef.part"
+    _append_chunk(again, b"same name", True)
+    assert publish_upload(again, target, "up.bin", "up.bin") == "up (2).bin", \
+        "a long path made the existing file look absent, and it was replaced"
+    assert _read(target / "up.bin") == b"first second"
+
+
+def test_an_abandoned_upload_is_found(library):
+    from meshbay_node import uploads
+
+    root, deep = library
+    roots = RootSet.build([{"path": str(root), "writable": True}])
+    part = _write(deep / "film.mkv.89abcdef.part", b"half a film")
+
+    found = [p for p, _ in uploads.find_parts(roots.roots)]
+
+    assert found == [part]
+
+
+def test_members_see_the_folders_past_the_limit(library):
+    """The folder tree a member browses and uploads into."""
+    from meshbay_node.transport.wire import list_dirs
+
+    root, deep = library
+    os.makedirs(_x(deep / "Deeper still"))
+    roots = RootSet.build([{"path": str(root)}])
+
+    dirs = list_dirs(roots)
+
+    rel = "/".join(["Library", *deep.relative_to(root).parts])
+    assert rel in dirs
+    assert f"{rel}/Deeper still" in dirs
+
+
+def test_a_folder_is_created_and_removed_past_the_limit(library):
+    _, deep = library
+    folder = deep / "New folder"
+    assert _mkdir_if_absent(folder) is None
+    assert _mkdir_if_absent(folder) == "Already exists"
+    assert _is_empty_dir(folder)
+    assert _rmdir_if_empty(folder)
+    assert not os.path.exists(_x(folder))
+
+
+# ── Enrichment ──────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+@pytest.mark.daemon_loop      # a subprocess needs the proactor loop
+async def test_audio_tags_and_the_cover_beside_it_are_read(library):
+    from meshbay_node.indexer.enrich_audio import _read_tags_and_cover
+    from PIL import Image
+
+    _, deep = library
+    track = deep / "01 - A Long Track Title.flac"
+    await _make_media(["-f", "lavfi", "-i", "sine=duration=1",
+                       "-metadata", "title=Some Song", "-metadata", "artist=Some Artist",
+                       "-metadata", "album=Some Album"], track)
+    with open(_x(deep / "cover.jpg"), "wb") as f:
+        Image.new("RGB", (8, 8), "red").save(f, "JPEG")
+
+    tags, duration, cover = await asyncio.to_thread(_read_tags_and_cover, track)
+
+    assert tags["title"] == "Some Song" and tags["album"] == "Some Album"
+    assert duration == pytest.approx(1.0, abs=0.1)
+    assert cover == _read(deep / "cover.jpg")
+
+
+@pytest.mark.asyncio
+async def test_a_photo_is_read(library):
+    from meshbay_node.indexer.enrich_photo import _read_image
+    from PIL import Image
+
+    _, deep = library
+    with open(_x(deep / "photo.jpg"), "wb") as f:
+        Image.new("RGB", (64, 48), "blue").save(f, "JPEG")
+
+    thumb, width, height, _, _ = await asyncio.to_thread(_read_image, deep / "photo.jpg")
+
+    assert (width, height) == (64, 48) and thumb
+
+
+@pytest.mark.asyncio
+@pytest.mark.daemon_loop      # a subprocess needs the proactor loop
+async def test_ffmpeg_probes_and_thumbnails_it_from_the_plain_path(library):
+    from meshbay_node.indexer.enrich import _make_thumbnail
+    from meshbay_node.media_probe import probe_video
+
+    _, deep = library
+    clip = deep / "clip.mp4"
+    await _make_media(["-f", "lavfi", "-i", "testsrc=size=64x48:rate=5:duration=2",
+                       "-pix_fmt", "yuv420p"], clip)
+
+    probe = await probe_video(str(clip))
+    thumb = await _make_thumbnail(clip, probe.duration)
+
+    assert probe.duration == pytest.approx(2.0, abs=0.3)
+    assert thumb and thumb[:2] == b"\xff\xd8"
+
+
+@pytest.mark.asyncio
+async def test_a_show_folder_past_the_limit_numbers_its_episodes(library):
+    from meshbay_node.indexer.enrich import (
+        _season_and_show_from_ancestors,
+        _synthetic_episode_number,
+        _title_from_siblings,
+    )
+
+    _, deep = library
+    season = deep / "Some Show" / "Season 1"
+    os.makedirs(_x(season))
+    _write(season / "Some Show S01E01.mkv", b"")
+    _write(season / "Pilot without a number.mkv", b"")
+    _write(season / "Second without a number.mkv", b"")
+    later = season / "Second without a number.mkv"
+
+    found = _season_and_show_from_ancestors(later)
+    assert found is not None
+    number, show = found
+    assert await asyncio.to_thread(_synthetic_episode_number, later, show, number) == 2
+    assert await asyncio.to_thread(_title_from_siblings, later) == "Some Show"