fix(agents): avoid executor starvation in file-gate waits (#6511)

* fix(agents): avoid executor starvation in file-gate waits

Keep async file-gate waiters out of the executor needed by the holder.
Preserve sync and cross-loop exclusion, and drain started file work on cancellation.

* test(agents): cover mixed file-gate callers

Document why release wakes every async waiter.
Cover a synchronous tool holder and asynchronous waiter through the middleware entry points.

* test(agents): cover sync waiters released by async file tools

Signed-off-by: LingYi <lingyi01525@gmail.com>

---------

Signed-off-by: LingYi <lingyi01525@gmail.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2086,6 +2086,8 @@ Each task gets its own execution environment with a full filesystem view — ski
 
 The read-before-write gate ties each read mark to that `read_file` call's result, including custom tools returning multi-message `Command` updates. An unrelated result cannot authorize a write after a failed read or hide a successful read.
 
+Concurrent reads and writes to the same file share a gate across synchronous and asynchronous tool calls. Async callers waiting for that gate do not occupy worker threads needed to finish the current read or write. Cancelling a waiting call leaves the current operation running; a call that already started file inspection still waits for that work to finish before releasing its gate.
+
 The built-in `grep` tool searches either one text file or all matching text files below a directory, so an agent can search an uploaded document directly without first broadening the request to the entire uploads directory.
 E2B's `glob` filter preserves spaces, quotes, and dollar signs in filename patterns, while wildcard matching and root-relative directory scoping remain unchanged.
 

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -1,5 +1,9 @@
 ### Middleware Chain
 
+Async gate waiters own separate cross-loop futures. Cancel only their signal;
+notify outside the state lock. `_await_off_thread` drains started I/O across
+repeated cancellation.
+
 Table synopses count nonblank CSV/TSV logical records, excluding the header.
 Preserve the recognition sample and 5,000,000 UTF-8-byte guard. Count incrementally;
 on parsing or field-limit errors, report an undetermined count rather than a
@@ -85,7 +89,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
    forgeries). Consumers pop it; the publisher and the consumer share only that
    contract module.
 12. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations before tool execution; command classification is **defense-in-depth and audit, not a security boundary** (the sandbox is the isolation boundary). Command substitution is judged by *position*, not the presence of `$(`: **command position** (`$(curl url)`, `` `curl url` ``, the word after `|`/`&&`/`;`, an `eval`/`source` argument) executes fetched content and is blocked; **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). So `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is matched anchored against each sub-command from `_split_compound_command(split_pipes=True)`, never the whole string; pipe-spanning rules (`| sh`, `base64 -d | ...`) still use `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`); its assignment branch requires whitespace before the substitution, which keeps `x=$(curl url)` in value position. Two contexts are deliberately **position-blind** (matched whole-command in Pass 1, since they execute their input anywhere, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument, and an interpreter **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) reaching the same place via stdin. All three substitution spellings (`$(`, `<(`, `` ` ``) share one `_RISKY_SUBSTITUTION` opener. An unquoted newline splits like `;` (else `echo hi\n$(curl url)` evades the anchored rules). Heredoc bodies are data: `_split_compound_command` records headers (`<<EOF`, `<<-EOF`, `<<'EOF'`) and consumes their bodies verbatim, so a body line starting `$(curl url)` isn't promoted to a command position; `<<<` (here-string, needs look-ahead + look-behind) and a `<<` inside `$(( ))`/`(( ))` (bit shift, arithmetic depth tracked with the quote flags) must not open one. This is a heuristic, not shell parsing — an unterminated body consumes the rest of the string, an unclosed `((` only disables heredoc detection, and the failure direction is always toward *more* command positions, not fewer. Known gaps: process substitution outside `eval`/`source` (`. <(curl u)`) is undetected, and two-step forms (`x=$(curl u); eval "$x"`) need dataflow analysis. Medium-risk warnings precede a trailing exit marker and keep `deerflow_tool_meta`. No config gate — appended unconditionally in `_build_runtime_middlewares`, for both lead and subagents.
-13. **ReadBeforeWriteMiddleware** - *(optional, `read_before_write.enabled`, default on)* Outermost write gate (#3857): `read_file` stamps a content hash on its ToolMessage; `write_file` (existing file, incl. append) and `str_replace` are blocked unless the newest mark for the path matches its current hash. Sits outside ToolProgress/ToolErrorHandling (a block consumes no ToolProgress slot); blocked results self-stamp `deerflow_tool_meta` and carry `deerflow_write_block` (`{path, tool}`). Marks live on messages, so summarization dropping the read invalidates the gate; writes never refresh marks. Gate check + execution are serialized per (thread, path); `"Error: ..."`-string sandboxes (AIO/E2B) fail open. It owns the composed call's sandbox authorization scope; `SandboxAuthorizationError` becomes an error ToolMessage. Its `wrap_model_call` swaps blocked calls' dead payload (`content`, `old_str`/`new_str`) for a deterministic placeholder in the model-bound request only (`elide_blocked_payloads`, `elide_min_chars`); state, receipts, and the journal keep the originals. Policy stays in the gate; the shared `tool_call_args` helper rewrites every arg surface together and every model-bound arg rewrite must use it.
+13. **ReadBeforeWriteMiddleware** - *(`read_before_write.enabled`, default on; #3857)* Existing-file `write_file` (including append) and `str_replace` require the newest matching `read_file` content hash. Marks live on ToolMessages: dropping a read loses its mark; writes never refresh it. Sync/async read/check/execution/mark steps share a per-(scope, path) gate. Uninspectable content fails open. It owns sandbox authorization; `SandboxAuthorizationError` becomes an error ToolMessage. Outside ToolProgress/ToolErrorHandling, blocks consume no slot and stamp `deerflow_tool_meta` plus `deerflow_write_block` (`{path, tool}`). `wrap_model_call` elides blocked payloads with deterministic placeholders (`elide_blocked_payloads`, `elide_min_chars`), preserving state, receipts and journal. All model-bound argument rewrites use `tool_call_args`; policy stays in this gate.
 14. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its tool wrapper receives results already stamped with `deerflow_tool_meta`. Recoverable problems stay WARNED with hints; retryable non-recoverable problems escalate to BLOCKED; stop-category failures block immediately. It is independent of LoopDetection's turn-level call-pattern guard. Effective phase changes emit `middleware:tool_progress`, including `reset` when a later agent invocation (for example, goal continuation) clears WARNED/BLOCKED state. The state machine supplies the action and count threshold (`null` for category rules, recovery, and reset); recorder calls happen after releasing the state lock. Events include the actual sync/async hook plus bounded status, error, and recovery fields. Server-owned recorder keys determine subagent attribution. Arguments, content, prompts, and hashes are never persisted; recorder failures are fail-open. See [event semantics](../../../../../docs/RUN_EVENT_STREAM.md).
 15. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of authorization, audit, write and progress gates — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit annotates medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `ToolErrorHandlingMiddleware` receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
 16. **ArtifactCaptureMiddleware** - See [tool artifacts](TOOL_ARTIFACTS.md).

backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py

@@ -41,6 +41,7 @@ import posixpath
 import threading
 import weakref
 from collections.abc import Awaitable, Callable
+from concurrent.futures import Future, InvalidStateError
 from typing import Any, override
 
 from langchain.agents.middleware import AgentMiddleware
@@ -85,20 +86,81 @@ _UNINSPECTABLE_CONTENT_PREFIX = "Error:"
 
 _BLOCK_MESSAGE = "Error: {tool_name} blocked — {path} already exists ({line_desc}) and you have not read its current version. Any write invalidates earlier reads, so re-read before every modification. {read_hint}"
 
+
 # Per-(scope, path) locks serializing gate check + tool execution. Same
 # WeakValueDictionary pattern as sandbox/file_operation_lock.py, but a
 # separate namespace: the tool-internal file lock only guards the mutation,
 # while this one also spans the authorization that precedes it.
-_GATE_LOCKS: weakref.WeakValueDictionary[tuple[str, str], threading.Lock] = weakref.WeakValueDictionary()
+class _GateLock:
+    """Cross-thread gate whose async waiters never occupy executor workers."""
+
+    def __init__(self) -> None:
+        self._condition = threading.Condition(threading.Lock())
+        self._locked = False
+        self._waiters: set[Future[None]] = set()
+
+    def acquire(self) -> bool:
+        with self._condition:
+            self._condition.wait_for(lambda: not self._locked)
+            self._locked = True
+        return True
+
+    async def acquire_async(self) -> None:
+        while True:
+            with self._condition:
+                if not self._locked:
+                    self._locked = True
+                    return
+                waiter: Future[None] = Future()
+                self._waiters.add(waiter)
+            try:
+                # Each caller owns its signal, so cancellation cannot cancel
+                # another loop's waiter or leave an eventual lock acquisition.
+                await asyncio.wrap_future(waiter)
+            finally:
+                with self._condition:
+                    self._waiters.discard(waiter)
+
+    def release(self) -> None:
+        with self._condition:
+            if not self._locked:
+                raise RuntimeError("release unlocked gate")
+            self._locked = False
+            waiters = tuple(self._waiters)
+            self._waiters.clear()
+            self._condition.notify_all()
+        # Future callbacks can schedule work on other event loops. Notify after
+        # releasing the state lock, and tolerate a concurrent waiter cancellation.
+        # Wake everyone: a caller cancelled before re-acquiring must not leave
+        # the remaining waiters asleep on an unlocked gate.
+        for waiter in waiters:
+            try:
+                waiter.set_result(None)
+            except InvalidStateError:
+                if not waiter.cancelled():
+                    raise
+
+    def locked(self) -> bool:
+        with self._condition:
+            return self._locked
+
+    def __enter__(self) -> bool:
+        return self.acquire()
+
+    def __exit__(self, *_exc: object) -> None:
+        self.release()
+
+
+_GATE_LOCKS: weakref.WeakValueDictionary[tuple[str, str], _GateLock] = weakref.WeakValueDictionary()
 _GATE_LOCKS_GUARD = threading.Lock()
 
 
-def _get_gate_lock(scope: str, norm_path: str) -> threading.Lock:
+def _get_gate_lock(scope: str, norm_path: str) -> _GateLock:
     key = (scope, norm_path)
     with _GATE_LOCKS_GUARD:
         lock = _GATE_LOCKS.get(key)
         if lock is None:
-            lock = threading.Lock()
+            lock = _GateLock()
             _GATE_LOCKS[key] = lock
         return lock
 
@@ -139,15 +201,9 @@ async def _await_off_thread(task: asyncio.Task[Any]) -> Any:
             raise first_cancel
 
 
-async def _acquire_gate_lock(lock: threading.Lock) -> None:
-    """Acquire off-loop safely; threading.Lock permits cross-thread release."""
-    acquire_task = asyncio.create_task(asyncio.to_thread(lock.acquire))
-    try:
-        await _await_off_thread(acquire_task)
-    except asyncio.CancelledError:
-        if acquire_task.done() and not acquire_task.cancelled() and acquire_task.exception() is None:
-            lock.release()
-        raise
+async def _acquire_gate_lock(lock: _GateLock) -> None:
+    """Wait without reserving the executor needed by the current gate holder."""
+    await lock.acquire_async()
 
 
 class ReadBeforeWriteMiddleware(AgentMiddleware):
@@ -262,7 +318,7 @@ class ReadBeforeWriteMiddleware(AgentMiddleware):
 
     # -- locking ---------------------------------------------------------
 
-    def _lock_for(self, request: ToolCallRequest, path: str) -> threading.Lock:
+    def _lock_for(self, request: ToolCallRequest, path: str) -> _GateLock:
         return _get_gate_lock(self._lock_scope(request), _normalize_mark_path(path))
 
     @staticmethod

backend/tests/test_read_before_write_cancellation.py

@@ -9,6 +9,7 @@ import pytest
 from langchain_core.messages import ToolMessage
 from langgraph.prebuilt.tool_node import ToolCallRequest
 
+from deerflow.agents.middlewares import read_before_write_middleware as gate
 from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware, _await_off_thread
 from deerflow.sandbox.exceptions import SandboxAuthorizationError
 
@@ -29,28 +30,6 @@ def _request(tool_name: str) -> ToolCallRequest:
     )
 
 
-class _BlockingAcquireLock:
-    def __init__(self) -> None:
-        self.acquire_started = threading.Event()
-        self.allow_acquire = threading.Event()
-        self.acquired = threading.Event()
-        self.release_calls = 0
-        self._guard = threading.Lock()
-
-    def acquire(self) -> bool:
-        self.acquire_started.set()
-        assert self.allow_acquire.wait(timeout=5), "test did not unblock gate-lock acquisition"
-        with self._guard:
-            self.acquired.set()
-        return True
-
-    def release(self) -> None:
-        with self._guard:
-            assert self.acquired.is_set(), "gate lock released without ownership"
-            self.acquired.clear()
-            self.release_calls += 1
-
-
 class _TrackingLock:
     def __init__(self) -> None:
         self.acquired = False
@@ -63,6 +42,9 @@ class _TrackingLock:
             self.acquired = True
         return True
 
+    async def acquire_async(self) -> None:
+        self.acquire()
+
     def release(self) -> None:
         with self._guard:
             assert self.acquired
@@ -71,12 +53,21 @@ class _TrackingLock:
 
 
 @pytest.mark.parametrize("tool_name", ["read_file", "write_file"])
-def test_async_gate_cancellation_drains_queued_lock_acquisition(
+def test_async_gate_cancellation_does_not_wait_for_another_holder(
     monkeypatch: pytest.MonkeyPatch,
     tool_name: str,
 ) -> None:
     async def scenario() -> None:
-        gate_lock = _BlockingAcquireLock()
+        gate_lock = gate._get_gate_lock(f"cancel-waiter-{tool_name}", _PATH)
+        gate_lock.acquire()
+        acquire_started = asyncio.Event()
+        acquire = gate._acquire_gate_lock
+
+        async def observed_acquire(lock):
+            acquire_started.set()
+            await acquire(lock)
+
+        monkeypatch.setattr(gate, "_acquire_gate_lock", observed_acquire)
         middleware = ReadBeforeWriteMiddleware(content_reader=lambda _runtime, _path: "v1")
         monkeypatch.setattr(middleware, "_lock_for", lambda _request, _path: gate_lock)
         handler_called = False
@@ -88,43 +79,25 @@ def test_async_gate_cancellation_drains_queued_lock_acquisition(
 
         task = asyncio.create_task(middleware.awrap_tool_call(_request(tool_name), handler))
         try:
-            assert await asyncio.to_thread(gate_lock.acquire_started.wait, 2), "gate-lock acquisition did not start"
-
+            await asyncio.wait_for(acquire_started.wait(), 5)
             task.cancel("first cancellation")
-            await asyncio.sleep(0)
-
-            # threading.Lock.acquire() is already running in a worker thread and
-            # cannot be cancelled. The middleware must retain ownership of that
-            # acquisition until it lands, then release it before propagating the
-            # original cancellation. Returning cancellation here would orphan a
-            # future successful acquire with nobody left to release it.
-            assert not task.done()
-            assert gate_lock.release_calls == 0
-
-            task.cancel("second cancellation")
-            await asyncio.sleep(0)
-
-            assert not task.done()
-            assert gate_lock.release_calls == 0
-            assert not handler_called
-
-            gate_lock.allow_acquire.set()
+            done, _ = await asyncio.wait((task,), timeout=5)
+            assert task in done, "a cancelled waiter still depends on another holder"
             with pytest.raises(asyncio.CancelledError) as exc_info:
                 await task
-
             assert exc_info.value.args == ("first cancellation",)
-            assert gate_lock.release_calls == 1
-            assert not gate_lock.acquired.is_set()
+            # No synchronous acquisition was dispatched. Cancellation must not
+            # release somebody else's gate or leave a later acquisition behind.
+            assert gate_lock.locked()
             assert not handler_called
         finally:
-            gate_lock.allow_acquire.set()
-            await asyncio.to_thread(gate_lock.acquired.wait, 0.1)
-            if gate_lock.acquired.is_set() and gate_lock.release_calls == 0:
-                gate_lock.release()
+            gate_lock.release()
             if not task.done():
                 task.cancel()
                 with suppress(asyncio.CancelledError):
                     await task
+        await asyncio.wait_for(acquire(gate_lock), 5)
+        gate_lock.release()
 
     asyncio.run(scenario())
 

backend/tests/test_read_before_write_coordination.py

@@ -0,0 +1,339 @@
+"""Gate waiters must not consume the executor needed by the lock holder."""
+
+import asyncio
+import hashlib
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
+from langchain_core.messages import ToolMessage
+from langgraph.prebuilt.tool_node import ToolCallRequest
+
+from deerflow.agents.middlewares import read_before_write_middleware as gate
+
+
+def _request(tag: str) -> ToolCallRequest:
+    return ToolCallRequest(
+        tool_call={"name": "read_file", "args": {"path": "/report.md", "description": "read"}, "id": tag},
+        tool=None,
+        state={"messages": []},
+        runtime=SimpleNamespace(context={"thread_id": "gate-coordination"}, state={}),
+    )
+
+
+class _ObservedLock:
+    """Observe arrivals without replacing the production locking behavior."""
+
+    def __init__(self, lock, arrived):
+        self.lock = lock
+        self.arrived = arrived
+
+    def acquire(self):
+        self.arrived()
+        return self.lock.acquire()
+
+    async def acquire_async(self):
+        self.arrived()
+        return await self.lock.acquire_async()
+
+    def release(self):
+        self.lock.release()
+
+    def __enter__(self):
+        return self.acquire()
+
+    def __exit__(self, *_exc):
+        self.release()
+
+
+@pytest.mark.parametrize("workers", [2, 32])
+def test_same_path_waiters_do_not_starve_the_holder_or_unrelated_work(tmp_path, monkeypatch, workers):
+    path = tmp_path / "report.md"
+    path.write_text("actual file bytes", encoding="utf-8")
+
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        pool = ThreadPoolExecutor(max_workers=workers)
+        loop.set_default_executor(pool)
+        holder_entered = asyncio.Event()
+        return_read_result = asyncio.Event()
+        waiters_entered = asyncio.Event()
+        calls = 0
+        count_guard = threading.Lock()
+
+        def arrived():
+            nonlocal calls
+            with count_guard:
+                calls += 1
+                if calls == workers + 1:
+                    loop.call_soon_threadsafe(waiters_entered.set)
+
+        middleware = gate.ReadBeforeWriteMiddleware(content_reader=lambda _runtime, _path: path.read_text(encoding="utf-8"))
+        observed = _ObservedLock(middleware._lock_for(_request("holder"), "/report.md"), arrived)
+        monkeypatch.setattr(middleware, "_lock_for", lambda _request, _path: observed)
+
+        async def handler(request):
+            if request.tool_call["id"] == "holder":
+                holder_entered.set()
+                await return_read_result.wait()
+            return ToolMessage(content="actual file bytes", tool_call_id=request.tool_call["id"], name="read_file")
+
+        tasks = [asyncio.create_task(middleware.awrap_tool_call(_request("holder"), handler))]
+        probe = None
+        completed = set()
+        try:
+            await asyncio.wait_for(holder_entered.wait(), 5)
+            tasks.extend(asyncio.create_task(middleware.awrap_tool_call(_request(f"waiter-{i}"), handler)) for i in range(workers))
+            await asyncio.wait_for(waiters_entered.wait(), 5)
+            return_read_result.set()
+            probe = asyncio.create_task(asyncio.to_thread(lambda: "unrelated work"))
+            completed, _ = await asyncio.wait((tasks[0], probe), timeout=5)
+        finally:
+            return_read_result.set()
+            if tasks[0] not in completed:
+                # Only a failing implementation needs this escape hatch. A
+                # spare worker drains the cycle so pytest teardown cannot hang.
+                pool._max_workers += 1
+                pool._adjust_thread_count()
+            results = await asyncio.wait_for(asyncio.gather(*tasks), 5)
+            if probe is not None:
+                await asyncio.wait_for(probe, 5)
+
+        assert tasks[0] in completed, "gate holder cannot dispatch its read-mark worker"
+        assert probe in completed, "contended file lock starved unrelated executor work"
+        assert all(result.additional_kwargs[gate.READ_MARK_KEY]["hash"] == hashlib.sha256(b"actual file bytes").hexdigest() for result in results)
+
+    asyncio.run(scenario())
+
+
+def test_cancel_after_unlock_does_not_orphan_the_gate():
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        arrived = asyncio.Event()
+        lock = gate._get_gate_lock("gate-cancel-after-unlock", "/report.md")
+        lock.acquire()
+        observed = _ObservedLock(lock, lambda: loop.call_soon_threadsafe(arrived.set))
+        task = asyncio.create_task(gate._acquire_gate_lock(observed))
+        try:
+            await asyncio.wait_for(arrived.wait(), 5)
+            lock.release()
+            task.cancel("cancel before taking ownership")
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, 5)
+            await asyncio.wait_for(gate._acquire_gate_lock(lock), 5)
+            lock.release()
+        finally:
+            if not task.done():
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+
+    asyncio.run(scenario())
+
+
+def test_async_tool_waits_for_sync_tool_on_the_same_path(tmp_path, monkeypatch):
+    path = tmp_path / "report.md"
+    path.write_text("shared file", encoding="utf-8")
+
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        holder_entered = threading.Event()
+        release_holder = threading.Event()
+        waiter_arrived = asyncio.Event()
+        waiter_ran = False
+        sync_results = []
+        sync_errors = []
+        arrivals = 0
+        guard = threading.Lock()
+        middleware = gate.ReadBeforeWriteMiddleware(content_reader=lambda _runtime, _path: path.read_text(encoding="utf-8"))
+
+        def arrived():
+            nonlocal arrivals
+            with guard:
+                arrivals += 1
+                if arrivals == 2:
+                    loop.call_soon_threadsafe(waiter_arrived.set)
+
+        observed = _ObservedLock(middleware._lock_for(_request("sync-holder"), "/report.md"), arrived)
+        monkeypatch.setattr(middleware, "_lock_for", lambda _request, _path: observed)
+
+        def sync_handler(request):
+            holder_entered.set()
+            assert release_holder.wait(5), "test did not release the synchronous holder"
+            return ToolMessage(content="shared file", tool_call_id=request.tool_call["id"], name="read_file")
+
+        def sync_call():
+            try:
+                sync_results.append(middleware.wrap_tool_call(_request("sync-holder"), sync_handler))
+            except BaseException as exc:
+                sync_errors.append(exc)
+
+        async def async_handler(request):
+            nonlocal waiter_ran
+            waiter_ran = True
+            return ToolMessage(content="shared file", tool_call_id=request.tool_call["id"], name="read_file")
+
+        thread = threading.Thread(target=sync_call)
+        task = None
+        thread.start()
+        try:
+            assert await asyncio.to_thread(holder_entered.wait, 5)
+            task = asyncio.create_task(middleware.awrap_tool_call(_request("async-waiter"), async_handler))
+            await asyncio.wait_for(waiter_arrived.wait(), 5)
+            assert not waiter_ran
+            release_holder.set()
+            result = await asyncio.wait_for(task, 5)
+            assert waiter_ran
+        finally:
+            release_holder.set()
+            if task is not None and not task.done():
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            await asyncio.to_thread(thread.join, 5)
+            assert not thread.is_alive()
+
+        assert not sync_errors
+        expected = hashlib.sha256(b"shared file").hexdigest()
+        assert len(sync_results) == 1
+        assert sync_results[0].additional_kwargs[gate.READ_MARK_KEY]["hash"] == expected
+        assert result.additional_kwargs[gate.READ_MARK_KEY]["hash"] == expected
+
+    asyncio.run(scenario())
+
+
+def test_sync_tool_waits_for_async_tool_on_the_same_path(tmp_path, monkeypatch):
+    path = tmp_path / "report.md"
+    path.write_text("shared file", encoding="utf-8")
+
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        holder_entered = asyncio.Event()
+        release_holder = asyncio.Event()
+        waiter_parked = asyncio.Event()
+        waiter_done = threading.Event()
+        waiter_ran = False
+        sync_results = []
+        sync_errors = []
+        middleware = gate.ReadBeforeWriteMiddleware(content_reader=lambda _runtime, _path: path.read_text(encoding="utf-8"))
+        lock = middleware._lock_for(_request("async-holder"), "/report.md")
+        condition = lock._condition
+        original_wait = condition.wait
+        cleanup_notify = condition.notify_all
+
+        def observed_wait(timeout=None):
+            # Called with the condition held; taking it below confirms that
+            # the real wait has queued this thread and released the mutex.
+            loop.call_soon_threadsafe(waiter_parked.set)
+            return original_wait(timeout)
+
+        monkeypatch.setattr(condition, "wait", observed_wait)
+
+        async def async_handler(request):
+            holder_entered.set()
+            await release_holder.wait()
+            return ToolMessage(content="shared file", tool_call_id=request.tool_call["id"], name="read_file")
+
+        def sync_handler(request):
+            nonlocal waiter_ran
+            waiter_ran = True
+            return ToolMessage(content="shared file", tool_call_id=request.tool_call["id"], name="read_file")
+
+        def sync_call():
+            try:
+                sync_results.append(middleware.wrap_tool_call(_request("sync-waiter"), sync_handler))
+            except BaseException as exc:
+                sync_errors.append(exc)
+            finally:
+                waiter_done.set()
+
+        task = asyncio.create_task(middleware.awrap_tool_call(_request("async-holder"), async_handler))
+        thread = threading.Thread(target=sync_call)
+        started = False
+        woke_after_release = False
+        try:
+            await asyncio.wait_for(holder_entered.wait(), 5)
+            thread.start()
+            started = True
+            await asyncio.wait_for(waiter_parked.wait(), 5)
+            with condition:
+                assert lock._locked
+                assert not waiter_ran
+                assert not waiter_done.is_set()
+            release_holder.set()
+            result = await asyncio.wait_for(task, 5)
+            woke_after_release = await asyncio.to_thread(waiter_done.wait, 5)
+        finally:
+            release_holder.set()
+            if not task.done():
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            # A missing production notify must fail the assertion, not leave
+            # a sleeping non-daemon thread hanging pytest at shutdown.
+            with condition:
+                cleanup_notify()
+            if started:
+                await asyncio.to_thread(thread.join, 5)
+                assert not thread.is_alive()
+
+        assert woke_after_release, "async holder did not wake the synchronous gate waiter"
+        assert not sync_errors
+        assert waiter_ran
+        expected = hashlib.sha256(b"shared file").hexdigest()
+        assert len(sync_results) == 1
+        assert sync_results[0].additional_kwargs[gate.READ_MARK_KEY]["hash"] == expected
+        assert result.additional_kwargs[gate.READ_MARK_KEY]["hash"] == expected
+
+    asyncio.run(scenario())
+
+
+def test_cancelled_waiter_does_not_cancel_another_event_loops_waiter():
+    lock = gate._get_gate_lock("gate-cross-loop-cancellation", "/report.md")
+    lock.acquire()
+    owner_holds_lock = True
+    arrived = [threading.Event(), threading.Event()]
+    cancelled = threading.Event()
+    tasks = {}
+    errors = []
+    acquired = []
+
+    def worker(index):
+        async def scenario():
+            observed = _ObservedLock(lock, arrived[index].set)
+            task = asyncio.create_task(gate._acquire_gate_lock(observed))
+            tasks[index] = (asyncio.get_running_loop(), task)
+            try:
+                await task
+            except asyncio.CancelledError:
+                assert index == 0
+                cancelled.set()
+            else:
+                try:
+                    acquired.append(index)
+                finally:
+                    lock.release()
+
+        try:
+            asyncio.run(scenario())
+        except BaseException as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(index,)) for index in range(2)]
+    for thread in threads:
+        thread.start()
+    try:
+        assert all(event.wait(5) for event in arrived)
+        loop, task = tasks[0]
+        loop.call_soon_threadsafe(task.cancel)
+        assert cancelled.wait(5), "cancellation waited for the synchronous holder"
+        assert acquired == []
+        owner_holds_lock = False
+        lock.release()
+    finally:
+        if owner_holds_lock:
+            lock.release()
+        for thread in threads:
+            thread.join(5)
+        assert not any(thread.is_alive() for thread in threads)
+    assert not errors
+    assert acquired == [1]
+    assert not lock.locked()