fix(sandbox): resolve the async acquire lock path off the event loop (#5760)

* fix(sandbox): resolve the async acquire lock path off the event loop

_discover_or_create_with_lock_async offloads every step of the acquire —
ensure_thread_dirs, the lock file, cache re-checks, backend discovery and
registration — except the one line that builds the lock path itself, which
resolves through Paths.base_dir and therefore runs a syscall on the loop.

The sibling anchor documented that as "a pre-existing blocking call in this
coroutine" and stubbed the path layer to get past it. With the resolution in
a worker thread the stub is no longer needed, so this removes it and pins the
behaviour with an anchor of its own.

* test(sandbox): word the offload anchor against the state after the stub removal

* test(sandbox): pin that the gate still catches an inline thread_dir call

The lock-path anchor only asserts execution reached the lock file, so it would
stay green if the gate stopped instrumenting the syscalls behind Path.resolve()
and the inline call were restored. Add a meta-check next to the store probe's:
call thread_dir() directly on the event loop under the same isolation and
assert BlockingError.

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py

@@ -2175,7 +2175,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         paths = get_paths()
         effective_user_id = self._effective_acquire_user_id(user_id)
         await asyncio.to_thread(paths.ensure_thread_dirs, thread_id, user_id=effective_user_id)
-        lock_path = paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock"
+
+        def _lock_path():
+            # Worker thread: thread_dir() resolves through Paths.base_dir, which is
+            # a syscall — the same reason ensure_thread_dirs directly above it, and
+            # every later step of this coroutine, is offloaded.
+            return paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock"
+
+        lock_path = await asyncio.to_thread(_lock_path)
 
         lock_file = await asyncio.to_thread(_open_lock_file, lock_path)
         locked = False

backend/tests/blocking_io/test_aio_sandbox_get.py

@@ -133,7 +133,6 @@ async def test_async_acquire_offloads_ownership_publish(tmp_path, monkeypatch):
     these two were called directly, putting a Redis round trip on the event loop
     for every discover/create.
     """
-    import deerflow.community.aio_sandbox.aio_sandbox_provider as aio_mod
     from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
 
     provider = _make_provider(tmp_path)
@@ -145,14 +144,58 @@ async def test_async_acquire_offloads_ownership_publish(tmp_path, monkeypatch):
     )
     provider._backend.discover = MagicMock(return_value=info)
 
-    # Stub the path layer: `get_paths()` resolves the base dir via os.getcwd on
-    # the event loop, which is a pre-existing blocking call in this coroutine and
-    # not what this anchor is about. Scoping it out keeps the test pinned to the
-    # ownership publish this diff added.
-    fake_paths = MagicMock()
-    fake_paths.thread_dir.return_value = tmp_path
-    monkeypatch.setattr(aio_mod, "get_paths", lambda: fake_paths)
+    # The lock path is resolved in a worker thread, so the real `Paths` object can
+    # stay in place; DEER_FLOW_HOME keeps the thread directories and the lock file
+    # inside the test's own tmp_path.
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    monkeypatch.setattr("deerflow.config.paths._paths", None)
 
     sandbox_id = await provider._discover_or_create_with_lock_async("t-async", "sb-async", user_id="u1")
 
     assert sandbox_id == "sb-async"
+
+
+async def test_async_lock_path_resolution_stays_off_the_loop(tmp_path, monkeypatch) -> None:
+    """The async acquire resolves its lock path from ``Paths.base_dir``, which syscalls.
+
+    The sibling anchor used to stub the path layer out and documented why — "a
+    pre-existing blocking call in this coroutine". The resolution now runs in a
+    worker thread, so that stub is gone; this anchor keeps it from coming back.
+    """
+    import deerflow.community.aio_sandbox.aio_sandbox_provider as aio_mod
+
+    provider = _make_provider(tmp_path)
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    monkeypatch.setattr("deerflow.config.paths._paths", None)
+
+    class _ReachedLockFile(Exception):
+        pass
+
+    def _stop_at_lock_file(_lock_path):
+        raise _ReachedLockFile
+
+    monkeypatch.setattr(aio_mod, "_open_lock_file", _stop_at_lock_file)
+
+    # Red before the fix: the gate raises BlockingError from the inline
+    # ``paths.thread_dir(...)`` before the sentinel is ever reached.
+    with pytest.raises(_ReachedLockFile):
+        await provider._discover_or_create_with_lock_async("t-lock", "sb-lock", user_id="u1")
+
+
+async def test_blocking_probe_thread_dir_actually_trips_the_gate(tmp_path, monkeypatch) -> None:
+    """Meta-check: prove the gate still catches an inline ``thread_dir`` call.
+
+    The anchor above only asserts execution reached the lock file, so if the gate
+    went blind to the syscalls behind ``Path.resolve()`` the anchor would pass with
+    the inline call restored. Calling ``thread_dir`` straight from the event loop
+    pins that the same isolation is armed and blocking here.
+    """
+    from blockbuster import BlockingError
+
+    from deerflow.config.paths import get_paths
+
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    monkeypatch.setattr("deerflow.config.paths._paths", None)
+
+    with pytest.raises(BlockingError):
+        get_paths().thread_dir("aio-sandbox-lock-wait")