fix(node): creating and removing a folder are syscalls too

The last two handlers on the loop. Both were synchronous, so a member creating a
folder on a root that had spun down held the node for the spin-up, exactly as a
chunk read did.

Where a check and an act belong together they are now one call rather than two
awaits, and the single disk thread is what makes that atomic: `_mkdir_if_absent`
so two members creating the same name cannot both find nothing there and have
the second `mkdir` raise where a refusal was meant, and `_rmdir_if_empty` for
the reason the caller already re-tested emptiness — the first test happened
before a round trip to the operator's browser, and a file can land in between.
Two awaits would reopen that window one size smaller.

The guard is now the whole class rather than the calls that were fixed. It walks
the module's syntax tree and fails on any filesystem call outside the handful of
functions written to be run through `off_disk` — a new handler that stats a root
inline would pass every measured test, because those exercise the handlers that
exist today. Checked by putting a call back: it names the function and the line.

It leaves ffmpeg's own scratch files out, listed rather than silently allowed:
they are under `tempfile.mkstemp` on the system disk, not on a group root, so
they are not what spins down — but they do read a whole transcode into memory
from the loop, and the day that matters it is a different measurement from this
one.

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

5 files changed, 122 insertions, 53 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 92f2951..3bb0df7 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -646,9 +646,9 @@ class WebRTCPeerSession:
             elif mtype == MNP.FILE_UPLOAD:
                 self._spawn(self._do_file_upload(msg))
             elif mtype == MNP.DIR_CREATE:
-                self._do_dir_create(msg)
+                self._spawn(self._do_dir_create(msg))
             elif mtype == MNP.DIR_DELETE:
-                self._do_dir_delete(msg)
+                self._spawn(self._do_dir_delete(msg))
             elif mtype == MNP.FILE_DELETE:
                 self._do_file_delete(msg)
             elif mtype == MNP.ADMIN_RESPONSE:
@@ -2034,7 +2034,7 @@ class WebRTCPeerSession:
         self._send(reply)
         self._audit_join("gek_wrapped", f"group={group_id[:8]}")
 
-    def _do_dir_create(self, msg: dict) -> None:
+    async def _do_dir_create(self, msg: dict) -> None:
         """
         Create a directory, for any member of the group.
 
@@ -2085,20 +2085,19 @@ class WebRTCPeerSession:
                         "code": "root_unavailable"})
             return
 
-        parent = safe_subdir(roots, parent_rel)
-        if parent is None or not parent.is_dir():
+        parent = await off_disk(roots, safe_subdir, roots, parent_rel)
+        if parent is None or not await off_disk(roots, parent.is_dir):
             self._send({"type": "error", "detail": "Invalid directory"})
             return
 
-        target = safe_subdir(roots, f"{parent_rel}/{name}")
+        target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}")
         if target is None:
             self._send({"type": "error", "detail": "Invalid directory"})
             return
-        if target.exists():
-            self._send({"type": "error", "detail": "Already exists"})
+        refusal = await off_disk(roots, _mkdir_if_absent, target)
+        if refusal is not None:
+            self._send({"type": "error", "detail": refusal})
             return
-
-        target.mkdir(parents=False)
         virtual = roots.virtual_of(target) or f"{parent_rel}/{name}"
         log.info("Directory created by %s: %s", self._user_id[:8], virtual)
         self._audit("dir_create", virtual)
@@ -2113,7 +2112,7 @@ class WebRTCPeerSession:
         found = roots.split(rel or "")
         return found is not None and not found[1]
 
-    def _do_dir_delete(self, msg: dict) -> None:
+    async def _do_dir_delete(self, msg: dict) -> None:
         """
         Remove an empty directory, for the node operator.
 
@@ -2131,17 +2130,17 @@ class WebRTCPeerSession:
             return
 
         rel = (msg.get("dir") or "").strip("/")
-        target = safe_subdir(roots, rel)
+        target = await off_disk(roots, safe_subdir, roots, rel)
         # A root itself is not deletable here: removing one is a configuration
         # change, and doing it through a file operation would leave the group
         # config naming a directory nobody can reach.
         if target is None or self._names_a_root(roots, rel):
             self._send({"type": "error", "detail": "Invalid directory"})
             return
-        if not target.is_dir():
+        if not await off_disk(roots, target.is_dir):
             self._send({"type": "error", "detail": "Not a directory"})
             return
-        if any(target.iterdir()):
+        if not await off_disk(roots, _is_empty_dir, target):
             self._send({"type": "error", "detail": "Directory is not empty"})
             return
         if not self._has_admin_authority():
@@ -2157,9 +2156,9 @@ class WebRTCPeerSession:
         rel = pending["subject"]
         ctx = self._group_ctx()
         roots: RootSet | None = ctx.get("roots")
-        target = safe_subdir(roots, rel) if roots else None
+        target = await off_disk(roots, safe_subdir, roots, rel) if roots else None
         if (target is None or self._names_a_root(roots, rel)
-                or not target.is_dir()):
+                or not await off_disk(roots, target.is_dir)):
             self._send({"type": "error", "detail": "Not a directory"})
             return
 
@@ -2173,11 +2172,9 @@ class WebRTCPeerSession:
         # Checked again after the signature: the emptiness test that let this
         # through happened before a round trip to the operator's browser, and a
         # file could have landed in the meantime.
-        if any(target.iterdir()):
+        if not await off_disk(roots, _rmdir_if_empty, target):
             self._send({"type": "error", "detail": "Directory is not empty"})
             return
-
-        target.rmdir()
         log.info("Directory removed by %s: %s", self._user_id[:8], rel)
         self._audit("dir_delete", rel)
         self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel})
@@ -6607,6 +6604,42 @@ class WebRTCPeerSession:
         await self._pc.close()
 
 
+def _mkdir_if_absent(target: Path) -> str | None:
+    """
+    Create a directory unless it is already there, or say why not.
+
+    Both in one call, not a check awaited and then an act: the disk thread is
+    one worker, so nothing can slip between them. Split across two awaits, two
+    members creating the same name would both find nothing there and the second
+    `mkdir` would raise where a refusal was meant. Blocking; called through
+    `off_disk`.
+    """
+    if target.exists():
+        return "Already exists"
+    target.mkdir(parents=False)
+    return None
+
+
+def _is_empty_dir(target: Path) -> bool:
+    """Blocking; called through `off_disk`."""
+    return not any(target.iterdir())
+
+
+def _rmdir_if_empty(target: Path) -> bool:
+    """
+    Remove a directory if nothing is in it. False if something is.
+
+    The emptiness test and the removal are one call for the reason the caller
+    re-tests at all: the first test happened before a round trip to the
+    operator's browser, and a file can land in between. Two awaits here would
+    reopen the same window one size smaller. Blocking; called through `off_disk`.
+    """
+    if any(target.iterdir()):
+        return False
+    target.rmdir()
+    return True
+
+
 def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
     """
     Where an entry is, and whether it is readable — or the refusal to send.
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 2811836..2179e66 100644
--- a/packages/meshbay-node/tests/test_disk_io_off_loop.py
+++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py
@@ -19,8 +19,8 @@ means two reads of the same file can never be inside it at once, and that is
 what makes the single `f.seek()`/`f.read()` pair safe without a lock.
 """
 
+import ast
 import asyncio
-import re
 import threading
 import time
 from pathlib import Path
@@ -204,29 +204,6 @@ async def test_one_root_set_reads_one_chunk_at_a_time(tmp_path):
     assert peak == 1, f"{peak} reads of one root set were inside the disk at once"
 
 
-def test_no_handler_resolves_a_path_on_the_loop():
-    """
-    `entry_abs_path` is `Path.resolve()`, which is syscalls — it belongs on the
-    disk thread with everything else.
-
-    This reads the source because there is nothing else to read: a handler added
-    later that resolves an entry inline would pass every test above, since those
-    only exercise the handlers that exist today. `_locate` is the one place
-    allowed to call it, and `off_disk` is how `_locate` is reached.
-    """
-    src = Path(webrtc_server.__file__).read_text()
-    # Every call site, not the first one: a guard that stops at the first
-    # occurrence stops guarding the moment a new call is inserted above it.
-    calls = [m.start() for m in re.finditer(r"\bentry_abs_path\(", src)]
-    body = re.search(r"\ndef _locate\(.*?\n(?=\n\ndef |\n\nclass )", src, re.S)
-    assert body, "_locate is gone or has been renamed — this guard needs rewriting"
-    allowed = range(body.start(), body.end())
-    stray = [c for c in calls if c not in allowed]
-    assert not stray, (
-        f"{len(stray)} call(s) to entry_abs_path outside _locate: "
-        f"resolve a path through `off_disk(roots, _locate, ...)` instead")
-
-
 async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatch):
     """
     The poll is the one that runs whether anybody asked for anything.
@@ -255,6 +232,65 @@ async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatc
         f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s poll")
 
 
+def test_no_handler_touches_the_disk_on_the_loop():
+    """
+    The whole class, not the calls that were fixed.
+
+    Every measured test above exercises a handler that exists today; a new one
+    that stats a root inline would pass all of them. So this walks the module's
+    syntax tree instead and fails on any filesystem call outside the few
+    functions written to be run through `off_disk`.
+
+    `entry_abs_path` and `safe_subdir` are in the list because both are
+    `Path.resolve()` underneath, and a resolve is syscalls whatever it is
+    called.
+    """
+    blocking = {"is_dir", "exists", "mkdir", "unlink", "rename", "rmdir",
+                "iterdir", "read_bytes", "write_bytes", "stat",
+                "resolve", "entry_abs_path"}
+    # 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"}
+    # ffmpeg's own output, under `tempfile.mkstemp` on the system disk — not a
+    # group root, so not what spins down. Listed rather than silently allowed:
+    # these still read a whole transcode into memory from the loop, and the day
+    # that matters it is a different measurement from this one.
+    ffmpeg_scratch = {"_transcode_audio_to_aac", "_seek_lands_at",
+                      "_extract_subtitle_to_webvtt"}
+    allowed = on_the_disk_thread | ffmpeg_scratch
+
+    found = []
+
+    def visit(node, owner):
+        for child in ast.iter_child_nodes(node):
+            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                visit(child, child.name)
+                continue
+            if isinstance(child, ast.Call):
+                fn = child.func
+                name = (fn.attr if isinstance(fn, ast.Attribute)
+                        else getattr(fn, "id", ""))
+                if name in blocking and owner not in allowed:
+                    found.append(f"{owner} calls {name}() at line {child.lineno}")
+            visit(child, owner)
+
+    tree = ast.parse(Path(webrtc_server.__file__).read_text())
+    for node in tree.body:
+        if isinstance(node, ast.ClassDef):
+            for member in node.body:
+                if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                    visit(member, member.name)
+        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            visit(node, node.name)
+
+    assert not found, (
+        "filesystem calls made from the event loop:\n  "
+        + "\n  ".join(found)
+        + "\nRun them through `off_disk(roots, ...)`, or put the call in a "
+          "helper that is only reached that way.")
+
+
 def _upload_session(tmp_path):
     """One connection into a group with one writable root, as a node has."""
     shared = tmp_path / "shared"
diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
index 23b55cb..3c8a837 100644
--- a/packages/meshbay-node/tests/test_root_writable_policy.py
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -121,7 +121,7 @@ async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path):
     requires the same root to be writable that adding the file would have.
     """
     session = _session(tmp_path, "member-1", writable=False)
-    session._do_dir_create({"dir": "shared", "name": "New folder"})
+    await session._do_dir_create({"dir": "shared", "name": "New folder"})
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
     assert refusal and refusal[0].get("code") == "root_read_only"
@@ -131,7 +131,7 @@ async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path):
 async def test_a_member_can_create_a_folder_in_a_writable_root(tmp_path):
     """The counter-property: it must stay unprivileged where it is allowed."""
     session = _session(tmp_path, "member-1", writable=True)
-    session._do_dir_create({"dir": "shared", "name": "New folder"})
+    await session._do_dir_create({"dir": "shared", "name": "New folder"})
 
     assert not [m for m in session.sent if m.get("type") == "error"]
     assert (tmp_path / "shared" / "New folder").is_dir()
@@ -144,7 +144,7 @@ async def test_an_ejected_root_refuses_a_new_folder(tmp_path):
     roots.roots[0].ejected = True
     roots.roots[0].available = False
 
-    session._do_dir_create({"dir": "shared", "name": "New folder"})
+    await session._do_dir_create({"dir": "shared", "name": "New folder"})
     refusal = [m for m in session.sent if m.get("type") == "error"]
     assert refusal and refusal[0].get("code") == "root_unavailable"
     assert not (tmp_path / "shared" / "New folder").exists()
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py
index eb13c61..14285e7 100644
--- a/packages/meshbay-node/tests/test_roster_pairing.py
+++ b/packages/meshbay-node/tests/test_roster_pairing.py
@@ -984,7 +984,7 @@ async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster):
     full.mkdir()
     (full / "keep.txt").write_text("still here")
 
-    session._do_dir_delete({"dir": "shared/full"})
+    await session._do_dir_delete({"dir": "shared/full"})
 
     assert _last(session).get("detail") == "Directory is not empty"
     assert full.exists() and (full / "keep.txt").exists()
@@ -996,7 +996,7 @@ async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster):
     session._ctx["has_admin_authority"] = False
     (tmp_path / "shared" / "empty").mkdir()
 
-    session._do_dir_delete({"dir": "shared/empty"})
+    await session._do_dir_delete({"dir": "shared/empty"})
 
     assert _last(session).get("detail") == "No authorized key for deletion"
     assert (tmp_path / "shared" / "empty").exists()
@@ -1013,7 +1013,7 @@ async def test_a_root_itself_is_not_a_target(tmp_path, roster):
     """
     session = await _dir_session(tmp_path, roster)
     for attempt in ("", ".", "/", "../shared", "shared", "shared/", "SHARED"):
-        session._do_dir_delete({"dir": attempt})
+        await session._do_dir_delete({"dir": attempt})
         assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
     assert (tmp_path / "shared").is_dir()
 
@@ -1028,7 +1028,7 @@ async def test_escaping_the_shared_root_is_refused(tmp_path, roster):
     for attempt in ("../outside", "../../outside", "sub/../../outside",
                     "shared/../outside", "shared/../../outside",
                     "shared/sub/../../outside"):
-        session._do_dir_delete({"dir": attempt})
+        await session._do_dir_delete({"dir": attempt})
         assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
     assert outside.is_dir(), "a path leaving the shared root removed a directory"
 
@@ -1044,7 +1044,7 @@ async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, rost
     session = await _dir_session(tmp_path, roster)
     (tmp_path / "shared" / "gone").mkdir()
 
-    session._do_dir_delete({"dir": "shared/gone"})
+    await session._do_dir_delete({"dir": "shared/gone"})
     challenge = _last(session)
     assert challenge["type"] == "admin_challenge"
     assert challenge["op"] == "dir_delete"
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index e203811..a229153 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -225,12 +225,12 @@ async def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path)
     {"dir": "/etc", "name": "evil"},
     {"dir": "", "name": ".hidden"},
 ])
-def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
+async def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
     """Creating a directory is not privileged, but it still writes to a disk."""
     session = _session(tmp_path, "user-1")
     before = set(tmp_path.rglob("*"))
 
-    session._do_dir_create(bad)
+    await session._do_dir_create(bad)
 
     assert any(m.get("type") == "error" for m in session.sent), bad
     assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}"