fix(sandbox): retain AIO flock across cancellation (#5820)

* fix(sandbox): retain AIO flock across cancellation

* fix(sandbox): drain async create before flock release

* test(sandbox): cover cancellation during AIO create

* docs(sandbox): pin create drain ownership

* fix(sandbox): keep flock waits cancellable

* test(sandbox): pin AIO flock lifecycle drains

* docs(sandbox): document cancellable flock wait

* test(sandbox): follow nonblocking flock helper

* test(sandbox): update create cancellation lock mock

* style(sandbox): sort lifecycle imports

* test(sandbox): assert cancelled flock wait closes file

* docs(sandbox): trim cancellation guidance below chain limit

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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

@@ -13,6 +13,7 @@ The provider itself handles:
 import asyncio
 import atexit
 import contextlib
+import errno
 import hashlib
 import logging
 import math
@@ -46,9 +47,11 @@ from deerflow.integrations.lark_cli import LARK_CLI_SANDBOX_CONFIG_DIR, LARK_CLI
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.acquire_serialization import AcquireSerializer
 from deerflow.sandbox.identity import derive_sandbox_scope_token
+from deerflow.sandbox.lease import run_sync_lifecycle_operation
 from deerflow.sandbox.sandbox import Sandbox
 from deerflow.sandbox.sandbox_provider import SandboxProvider
 from deerflow.skills.types import SkillCategory
+from deerflow.utils.file_io import await_drained
 
 from .aio_sandbox import AioSandbox
 from .backend import SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT, SandboxBackend, wait_for_sandbox_ready, wait_for_sandbox_ready_async
@@ -123,6 +126,21 @@ def _lock_file_exclusive(lock_file) -> None:
     msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
 
 
+def _try_lock_file_exclusive(lock_file) -> bool:
+    """Attempt the cross-process lock once without blocking the worker thread."""
+    try:
+        if fcntl is not None:
+            fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        else:  # pragma: no cover - Windows
+            lock_file.seek(0)
+            msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
+    except OSError as error:
+        if error.errno in (errno.EACCES, errno.EAGAIN, errno.EDEADLK):
+            return False
+        raise
+    return True
+
+
 def _unlock_file(lock_file) -> None:
     if fcntl is not None:
         fcntl.flock(lock_file, fcntl.LOCK_UN)
@@ -2187,20 +2205,22 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         lock_file = await asyncio.to_thread(_open_lock_file, lock_path)
         locked = False
         try:
-            await asyncio.to_thread(_lock_file_exclusive, lock_file)
-            locked = True
+            while not locked:
+                locked = await run_sync_lifecycle_operation(_try_lock_file_exclusive, lock_file)
+                if not locked:
+                    await asyncio.sleep(0.02)
             # Re-check in-process caches under the file lock in case another
             # thread in this process won the race while we were waiting.
-            cached_id = await asyncio.to_thread(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id)
+            cached_id = await run_sync_lifecycle_operation(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id)
             if cached_id is not None:
                 return cached_id
 
             # Backend discovery is sync because local discovery may inspect
             # Docker and perform a health check; keep it off the event loop.
-            discovered = await asyncio.to_thread(self._backend.discover, sandbox_id)
+            discovered = await run_sync_lifecycle_operation(self._backend.discover, sandbox_id)
             if discovered is not None:
                 if discovered.requires_replacement:
-                    replaced = await asyncio.to_thread(
+                    replaced = await run_sync_lifecycle_operation(
                         self._replace_incompatible_sandbox,
                         discovered,
                         time.time(),
@@ -2211,13 +2231,20 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                     # Registration publishes ownership, which is blocking store
                     # IO (filesystem or network depending on the backend) — same
                     # reason every other step in this coroutine is offloaded.
-                    return await asyncio.to_thread(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id)
+                    return await run_sync_lifecycle_operation(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id)
 
-            return await self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id)
+            # Keep the entire async create lifecycle under the cross-process flock.
+            # Shielding only individual to_thread workers is insufficient: cancellation
+            # after backend.create() would otherwise discard SandboxInfo before readiness
+            # and registration/cleanup run, then finally release the flock over an
+            # unregistered deterministic container.
+            return await await_drained(self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id))
         finally:
-            if locked:
-                await asyncio.to_thread(_unlock_file, lock_file)
-            await asyncio.to_thread(lock_file.close)
+            try:
+                if locked:
+                    await run_sync_lifecycle_operation(_unlock_file, lock_file)
+            finally:
+                await run_sync_lifecycle_operation(lock_file.close)
 
     def _destroy_unready_sandbox(self, sandbox_id: str, info: SandboxInfo) -> None:
         """Tear down a freshly-created container whose readiness check failed.
@@ -2333,10 +2360,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
     async def _create_sandbox_async(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
         """Async counterpart to ``_create_sandbox``."""
         effective_user_id = self._effective_acquire_user_id(user_id)
-        extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id, user_id=effective_user_id)
-        provision_lark_cli_runtime = await asyncio.to_thread(self._lark_integration_active, effective_user_id)
-        provision_lark_cli_broker = await asyncio.to_thread(self._lark_broker_active, effective_user_id)
-        config_mount_exclusion_root = await asyncio.to_thread(
+        extra_mounts = await run_sync_lifecycle_operation(self._get_extra_mounts, thread_id, user_id=effective_user_id)
+        provision_lark_cli_runtime = await run_sync_lifecycle_operation(self._lark_integration_active, effective_user_id)
+        provision_lark_cli_broker = await run_sync_lifecycle_operation(self._lark_broker_active, effective_user_id)
+        config_mount_exclusion_root = await run_sync_lifecycle_operation(
             self._local_config_mount_exclusion_root,
             thread_id,
             user_id=effective_user_id,
@@ -2346,7 +2373,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         # Active sandboxes are in use by live threads and must not be forcibly stopped.
         replicas, total = self._replica_count()
         if total >= replicas:
-            evicted = await asyncio.to_thread(self._evict_oldest_warm)
+            evicted = await run_sync_lifecycle_operation(self._evict_oldest_warm)
             self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
 
         create_kwargs = {}
@@ -2354,7 +2381,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             create_kwargs["config_mount_exclusion_root"] = config_mount_exclusion_root
         if isinstance(self._backend, RemoteSandboxBackend):
             create_kwargs["skills_container_path"] = self._configured_skills_container_path()
-        info = await asyncio.to_thread(
+        info = await run_sync_lifecycle_operation(
             self._backend.create,
             thread_id,
             sandbox_id,
@@ -2376,12 +2403,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             # ``_register_created_sandbox`` after this gate. Claim the teardown
             # lease before stopping it so a peer cannot adopt the not-yet-ready
             # Pod in the meantime (#4248).
-            await asyncio.to_thread(self._destroy_unready_sandbox, sandbox_id, info)
+            await run_sync_lifecycle_operation(self._destroy_unready_sandbox, sandbox_id, info)
             raise RuntimeError(f"Sandbox {sandbox_id} failed to become ready within timeout at {info.sandbox_url}")
 
         # Registration publishes ownership (blocking store IO), so it is offloaded
         # like every other blocking step on this path.
-        return await asyncio.to_thread(self._register_created_sandbox, thread_id, sandbox_id, info, user_id=effective_user_id)
+        return await run_sync_lifecycle_operation(self._register_created_sandbox, thread_id, sandbox_id, info, user_id=effective_user_id)
 
     def get(self, sandbox_id: str) -> Sandbox | None:
         """Get a sandbox by ID. Updates last activity timestamp.

backend/packages/harness/deerflow/sandbox/AGENTS.md

@@ -19,7 +19,7 @@ to respond.
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-user/thread `LocalSandbox` (id `local:{user_id}:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Shared runs use category mappings; a policy-scoped run replaces them with one `/mnt/skills` root mapping to the coherent thread view, so structured file tools resolve through one managed boundary. This is not a host filesystem security boundary: an enabled host `bash` subprocess can use canonical paths without `PathMapping`, so `supports_agent_skill_isolation` is dynamic and explicit Agent policies fail closed while host bash is enabled. Host-to-virtual output masking scans dynamic per-user/per-thread roots directly instead of compiling path-specific regexes, so evicted thread IDs do not remain in Python's global regex caches; a separate 256-entry root cache prevents repeated `realpath()` walks for every glob/grep match while bounding dynamic-path retention, and only the small process-stable skill/integration source set uses a bounded compiled cache. The shared `path_patterns.py` tail stops at `:`, so `$PATH`-style lists mask every entry; a mount symlink resolving outside all mounts keeps its mount path. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes; do not restore a blanket conversion disable, because host-native CLI launchers need normal MSYS path conversion for their own installation paths.
   PowerShell fallback pairs guarded console UTF-8 setters with explicit UTF-8 pipe decoding; cmd/MSYS retain locale decoding. Encoding regressions include real-pipe probes and Windows-only PowerShell roundtrips with and without a console.
-- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker isolation. Reuse enforces minimum persisted shell capacity even without image-env overrides. Upgrades require ownership-fenced teardown; provisioner create returns 409 for undersized Pods without deleting them. Acquire/reuse checks active-cache and warm-pool entries with the backend, dropping definitively dead containers from all in-process maps before rediscovery/creation. Health-check failures mean unknown, not dead; local discovery skips unverifiable containers and tries creation. Acquire/reclaim publishes ownership; `_renew_owned_leases` refreshes it on a dedicated thread. Text appends use native AIO append, never read-modify-write that could overwrite after a failed read. `reset()` closes the acquire serializer and its workers; `shutdown()` performs full remote teardown. `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner=False); `sandbox.thread_data_mounts` overrides this for deployments sharing thread user-data directories. `true` skips upload acquire/sync; a false positive makes uploads unavailable. Explicit Agent policies use four thread projection category mounts and a distinct deterministic identity, excluding older shared-mount containers. The startup snapshot of `skills.container_path` drives mounts, identity, remote requests, and provisioner validation; root changes prevent container/Pod reuse. Gateway and provisioner independently require a canonical absolute root outside reserved platform mounts and derive four category allowlist entries from it. All four category overrides suppress the provisioner's default hostPath/skills-PVC mount; with `USERDATA_PVC_NAME`, categories use subpaths on that shared PVC. Readiness probes and `agent_sandbox` clients set `trust_env=False` for direct control-plane destinations (loopback/private IPs, single-label cluster hosts, Docker/Podman internal hostnames); external FQDNs and public IPs retain environment proxies.
+- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker isolation. Reuse enforces minimum persisted shell capacity even without image-env overrides. Upgrades require ownership-fenced teardown; provisioner create returns 409 for undersized Pods without deleting them. Acquire/reuse checks active-cache and warm-pool entries with the backend, dropping definitively dead containers from all in-process maps before rediscovery/creation. Health-check failures mean unknown, not dead; local discovery skips unverifiable containers and tries creation. Acquire/reclaim publishes ownership; `_renew_owned_leases` refreshes it on a dedicated thread. Text appends use native AIO append, never read-modify-write that could overwrite after a failed read. `reset()` closes the acquire serializer and its workers; `shutdown()` performs full remote teardown. Async create polls flock acquisition so cancelled waiters leave promptly. Once locked, cancellation drains admitted workers and create readiness/registration/cleanup before unlock/close. `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner=False); `sandbox.thread_data_mounts` overrides this for deployments sharing thread user-data directories. `true` skips upload acquire/sync; a false positive makes uploads unavailable. Explicit Agent policies use four thread projection category mounts and a distinct deterministic identity, excluding older shared-mount containers. The startup snapshot of `skills.container_path` drives mounts, identity, remote requests, and provisioner validation; root changes prevent container/Pod reuse. Gateway and provisioner independently require a canonical absolute root outside reserved platform mounts and derive four category allowlist entries from it. All four category overrides suppress the provisioner's default hostPath/skills-PVC mount; with `USERDATA_PVC_NAME`, categories use subpaths on that shared PVC. Readiness probes and `agent_sandbox` clients set `trust_env=False` for direct control-plane destinations (loopback/private IPs, single-label cluster hosts, Docker/Podman internal hostnames); external FQDNs and public IPs retain environment proxies.
 - `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) provides E2B remote isolation.
   New unrestricted sandboxes receive a one-shot upload from the enabled-only
   public, custom, legacy, and managed integration projections. For a thread

backend/tests/test_aio_sandbox_provider.py

@@ -937,6 +937,332 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl
     assert any(getattr(func, "__name__", "") == "close" for func in to_thread_calls)
 
 
+@pytest.mark.anyio
+async def test_discover_or_create_with_lock_async_cancellation_aborts_flock_wait(tmp_path, monkeypatch):
+    """A cancelled waiter must not park until a peer releases the cross-process flock."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__(
+        provider,
+        aio_mod.AioSandboxProvider,
+    )
+    provider._backend = SimpleNamespace(discover=MagicMock(return_value=None))
+
+    peer_lock = threading.Lock()
+    peer_lock.acquire()
+    attempt_seen = threading.Event()
+    closed = threading.Event()
+
+    class TrackedLockFile:
+        def __init__(self, inner):
+            self._inner = inner
+
+        def __getattr__(self, name):
+            return getattr(self._inner, name)
+
+        def close(self):
+            self._inner.close()
+            closed.set()
+
+    def open_lock_file(lock_path):
+        return TrackedLockFile(open(lock_path, "a", encoding="utf-8"))
+
+    def try_lock(_lock_file) -> bool:
+        attempt_seen.set()
+        return peer_lock.acquire(blocking=False)
+
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr(aio_mod, "_open_lock_file", open_lock_file)
+    monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", try_lock)
+
+    owner = asyncio.create_task(
+        provider._discover_or_create_with_lock_async(
+            "thread-cancel-flock-wait",
+            "sandbox-cancel-flock-wait",
+            user_id="default",
+        )
+    )
+    try:
+        assert await asyncio.to_thread(attempt_seen.wait, 2)
+        owner.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(owner, timeout=0.5)
+        assert closed.is_set()
+        provider._backend.discover.assert_not_called()
+    finally:
+        peer_lock.release()
+        if not owner.done():
+            owner.cancel()
+            await asyncio.gather(owner, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_discover_or_create_with_lock_async_cancellation_keeps_file_lock_until_worker_finishes(tmp_path, monkeypatch):
+    """Caller cancellation must not release the cross-process lock over an admitted worker."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__(
+        provider,
+        aio_mod.AioSandboxProvider,
+    )
+    provider._warm_pool = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._sandboxes = {}
+    provider._last_activity = {}
+    provider._lock = aio_mod.threading.Lock()
+
+    discover_started = threading.Event()
+    allow_discover = threading.Event()
+    contender_acquired = threading.Event()
+    release_contender = threading.Event()
+    cross_process_lock = threading.Lock()
+
+    def blocking_discover(_sandbox_id: str):
+        discover_started.set()
+        assert allow_discover.wait(timeout=2)
+        return None
+
+    create_calls: list[tuple[str | None, str, str | None]] = []
+
+    async def fake_create(thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
+        create_calls.append((thread_id, sandbox_id, user_id))
+        return sandbox_id
+
+    def contend_for_lock() -> None:
+        cross_process_lock.acquire()
+        try:
+            contender_acquired.set()
+            assert release_contender.wait(timeout=2)
+        finally:
+            cross_process_lock.release()
+
+    provider._backend = SimpleNamespace(discover=blocking_discover)
+    monkeypatch.setattr(provider, "_create_sandbox_async", fake_create)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False))
+    monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release())
+
+    owner = asyncio.create_task(
+        provider._discover_or_create_with_lock_async(
+            "thread-cancel-flock",
+            "sandbox-cancel-flock",
+            user_id="default",
+        )
+    )
+    contender = None
+    try:
+        assert await asyncio.to_thread(discover_started.wait, 2)
+        owner.cancel()
+        contender = asyncio.create_task(asyncio.to_thread(contend_for_lock))
+
+        await asyncio.sleep(0.05)
+        assert not contender_acquired.is_set(), "cross-process lock released while discover worker was still running"
+
+        allow_discover.set()
+        with pytest.raises(asyncio.CancelledError):
+            await owner
+
+        assert await asyncio.to_thread(contender_acquired.wait, 2)
+        assert create_calls == []
+    finally:
+        allow_discover.set()
+        release_contender.set()
+        if not owner.done():
+            owner.cancel()
+            await asyncio.gather(owner, return_exceptions=True)
+        if contender is not None:
+            await contender
+
+
+@pytest.mark.anyio
+async def test_discover_or_create_with_lock_async_cancellation_drains_create_before_unlock(tmp_path, monkeypatch):
+    """Cancellation during create must finish the async lifecycle before releasing the flock."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__(
+        provider,
+        aio_mod.AioSandboxProvider,
+    )
+    provider._warm_pool = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._sandboxes = {}
+    provider._last_activity = {}
+    provider._lock = aio_mod.threading.Lock()
+
+    create_started = asyncio.Event()
+    allow_create = asyncio.Event()
+    create_finished = asyncio.Event()
+    contender_acquired = threading.Event()
+    release_contender = threading.Event()
+    cross_process_lock = threading.Lock()
+
+    async def fake_create(thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
+        assert thread_id == "thread-cancel-create"
+        assert sandbox_id == "sandbox-cancel-create"
+        assert user_id == "default"
+        create_started.set()
+        await allow_create.wait()
+        create_finished.set()
+        return sandbox_id
+
+    def contend_for_lock() -> None:
+        cross_process_lock.acquire()
+        try:
+            contender_acquired.set()
+            assert release_contender.wait(timeout=2)
+        finally:
+            cross_process_lock.release()
+
+    provider._backend = SimpleNamespace(discover=MagicMock(return_value=None))
+    monkeypatch.setattr(provider, "_create_sandbox_async", fake_create)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False))
+    monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release())
+
+    owner = asyncio.create_task(
+        provider._discover_or_create_with_lock_async(
+            "thread-cancel-create",
+            "sandbox-cancel-create",
+            user_id="default",
+        )
+    )
+    contender = None
+    try:
+        await asyncio.wait_for(create_started.wait(), timeout=2)
+        owner.cancel()
+        contender = asyncio.create_task(asyncio.to_thread(contend_for_lock))
+
+        await asyncio.sleep(0.05)
+        assert not contender_acquired.is_set(), "flock released before the async create lifecycle finished"
+
+        allow_create.set()
+        with pytest.raises(asyncio.CancelledError):
+            await owner
+
+        assert create_finished.is_set()
+        assert await asyncio.to_thread(contender_acquired.wait, 2)
+    finally:
+        allow_create.set()
+        release_contender.set()
+        if not owner.done():
+            owner.cancel()
+            await asyncio.gather(owner, return_exceptions=True)
+        if contender is not None:
+            await contender
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("blocked_step", ["create", "register"])
+async def test_discover_or_create_with_lock_async_cancellation_drains_create_lifecycle_steps(
+    blocked_step,
+    tmp_path,
+    monkeypatch,
+):
+    """Create/register workers must settle before cancellation releases the flock."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__(
+        provider,
+        aio_mod.AioSandboxProvider,
+    )
+    provider._warm_pool = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._sandboxes = {}
+    provider._last_activity = {}
+    provider._lock = aio_mod.threading.Lock()
+    provider._config = {"command_timeout": 600.0, "idle_timeout": 600, "replicas": 3}
+
+    step_started = threading.Event()
+    allow_step = threading.Event()
+    step_finished = threading.Event()
+    register_called = threading.Event()
+    contender_acquired = threading.Event()
+    release_contender = threading.Event()
+    cross_process_lock = threading.Lock()
+
+    info = aio_mod.SandboxInfo(
+        sandbox_id="sandbox-cancel-lifecycle",
+        sandbox_url="http://sandbox",
+    )
+
+    def create(*_args, **_kwargs):
+        if blocked_step == "create":
+            step_started.set()
+            assert allow_step.wait(timeout=2)
+            step_finished.set()
+        return info
+
+    def register(*_args, **_kwargs):
+        register_called.set()
+        if blocked_step == "register":
+            step_started.set()
+            assert allow_step.wait(timeout=2)
+            step_finished.set()
+        return info.sandbox_id
+
+    async def ready(*_args, **_kwargs):
+        return True
+
+    def contend_for_lock() -> None:
+        cross_process_lock.acquire()
+        try:
+            contender_acquired.set()
+            assert release_contender.wait(timeout=2)
+        finally:
+            cross_process_lock.release()
+
+    provider._backend = SimpleNamespace(
+        create=create,
+        destroy=MagicMock(),
+        discover=MagicMock(return_value=None),
+    )
+    monkeypatch.setattr(provider, "_get_extra_mounts", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(provider, "_lark_integration_active", lambda *_args, **_kwargs: False)
+    monkeypatch.setattr(provider, "_lark_broker_active", lambda *_args, **_kwargs: False)
+    monkeypatch.setattr(provider, "_local_config_mount_exclusion_root", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(provider, "_replica_count", lambda: (3, 0))
+    monkeypatch.setattr(provider, "_register_created_sandbox", register)
+    monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready_async", ready)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False))
+    monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release())
+
+    owner = asyncio.create_task(
+        provider._discover_or_create_with_lock_async(
+            "thread-cancel-lifecycle",
+            info.sandbox_id,
+            user_id="default",
+        )
+    )
+    contender = None
+    try:
+        assert await asyncio.to_thread(step_started.wait, 2)
+        owner.cancel()
+        contender = asyncio.create_task(asyncio.to_thread(contend_for_lock))
+
+        await asyncio.sleep(0.05)
+        assert not contender_acquired.is_set()
+
+        allow_step.set()
+        with pytest.raises(asyncio.CancelledError):
+            await owner
+
+        assert step_finished.is_set()
+        assert register_called.is_set()
+        assert await asyncio.to_thread(contender_acquired.wait, 2)
+    finally:
+        allow_step.set()
+        release_contender.set()
+        if not owner.done():
+            owner.cancel()
+            await asyncio.gather(owner, return_exceptions=True)
+        if contender is not None:
+            await contender
+
+
 @pytest.mark.anyio
 async def test_acquire_async_lock_wait_uses_dedicated_executor(tmp_path, monkeypatch):
     """Per-thread lock waits should not consume the default asyncio.to_thread pool."""