fix(agents): bind read marks to the current tool call (#6363)

* fix(agents): bind read marks to the current tool call

* test(agents): make file and proxy baseline checks portable

* fix(agents): address read-mark review diagnostics and cleanup

* docs(agents): keep read-mark guidance within the chain budget

README.md

@@ -1976,6 +1976,8 @@ DeerFlow doesn't just *talk* about doing things. It has its own computer.
 
 Each task gets its own execution environment with a full filesystem view — skills, workspace, uploads, outputs. The agent reads, writes, and edits files. It can view images and, when configured safely, execute shell commands.
 
+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.
+
 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.
 
 Remote `ls` excludes ignored descendants before applying its 500-entry listing limit, so dependency and build trees do not crowd out visible files. Explicitly listing an ignored directory still lists its contents; normal depth and output limits remain in effect.

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

@@ -45,6 +45,9 @@ so malformed values cannot abort compaction/model calls.
 `tool_calls` update: adapters resend stale `content` tool-call blocks, which
 strict providers reject.
 
+Read marks bind to request `tool_call_id`, including `Command` results.
+No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.py`.
+
 **Shared runtime base** (`build_lead_runtime_middlewares`; subagents reuse most of this via `build_subagent_runtime_middlewares`):
 
 1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages. `additional_kwargs.original_user_content` is server-owned provenance: Gateway strips caller-supplied values for non-internal run requests, trusted IM calls may carry the string they captured before adding transport/file context, and the middleware replaces any non-string value before wrapping. Uploads and sanitization retain first-writer-wins only for validated strings. Caller markers are marked `untrusted_input`, never stripped; scope is every turn.

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

@@ -402,8 +402,12 @@ class ReadBeforeWriteMiddleware(AgentMiddleware):
         path = self._requested_path(request)
         if path is None:
             return
-        message = self._extract_tool_message(result)
-        if message is None or message.status == "error":
+        tool_call_id = str(request.tool_call.get("id") or "")
+        message = self._extract_tool_message(result, tool_call_id=tool_call_id)
+        if message is None:
+            logger.debug("read-before-write mark skipped for %r: no result matching request tool_call_id %r", path, tool_call_id)
+            return
+        if message.status == "error":
             return
         # In-band contract errors (invalid range, start_line exceeds file length, etc.)
         # are returned with status="success" by LangChain even though the model received
@@ -435,13 +439,17 @@ class ReadBeforeWriteMiddleware(AgentMiddleware):
         }
 
     @staticmethod
-    def _extract_tool_message(result: ToolMessage | Command) -> ToolMessage | None:
+    def _extract_tool_message(result: ToolMessage | Command, *, tool_call_id: str) -> ToolMessage | None:
         if isinstance(result, ToolMessage):
-            return result
+            return result if result.tool_call_id == tool_call_id else None
         if isinstance(result, Command) and isinstance(result.update, dict):
-            candidates = [m for m in result.update.get("messages", []) if isinstance(m, ToolMessage)]
-            if candidates:
-                return candidates[-1]
+            messages = result.update.get("messages")
+            if isinstance(messages, ToolMessage):
+                messages = [messages]
+            if isinstance(messages, (list, tuple)):
+                # Command updates may include other calls' results. They are not
+                # evidence that this read succeeded and must never receive its mark.
+                return next((message for message in reversed(messages) if isinstance(message, ToolMessage) and message.tool_call_id == tool_call_id), None)
         return None
 
 

backend/tests/AGENTS.md

@@ -2,6 +2,16 @@
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+Browser egress session-close tests retain the real listener and its sockets,
+verify a SOCKS handshake, then check listener shutdown, closed socket descriptors
+and EOF on the established client. A new connection to the old port is not a
+reliable ownership assertion on hosts with loopback forwarding or port reuse.
+Keep ownership assertions inside the teardown-protected block; always drain
+the client writer and close the saved proxy during teardown.
+
+Read-before-write hash fixtures pin `newline=""` when writing LF/CRLF test
+content, so native Windows cannot translate the bytes before the real read.
+
 Browser-asset confinement tests use `support.symlinks.symlink_or_skip` for real
 file and directory symlinks. Keep missing-file, duplicate-key, and size-limit
 checks separate so they still run when the host cannot create symlinks.

backend/tests/test_browser_egress.py

@@ -9,6 +9,7 @@ skipped when Playwright is not installed.
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import ipaddress
 import socket
 import sys
@@ -360,12 +361,36 @@ async def test_session_launches_chromium_through_its_egress_proxy_and_closes_it(
     assert proxy["server"].startswith("socks5://127.0.0.1:")
     assert proxy["bypass"] == "<-loopback>"
     proxy_port = int(proxy["server"].rsplit(":", 1)[1])
-    _reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
-    writer.close()
+    egress_proxy = session._egress_proxy
+    writer = None
+    try:
+        assert egress_proxy is not None
+        server = egress_proxy._server
+        assert server is not None and server.is_serving()
+        listening_sockets = server.sockets
+        assert listening_sockets
+        reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
+        writer.write(b"\x05\x01\x00")
+        await writer.drain()
+        assert await asyncio.wait_for(reader.readexactly(2), timeout=1.0) == b"\x05\x00"
 
-    await session._close()
-    with pytest.raises(OSError):
-        await asyncio.open_connection("127.0.0.1", proxy_port)
+        await session._close()
+        # A fresh connect can succeed through host loopback forwarding even
+        # after close. Check the listener this session actually owned instead.
+        assert not server.is_serving()
+        assert not server.sockets
+        assert all(sock.fileno() == -1 for sock in listening_sockets)
+        assert egress_proxy._server is None
+        assert session._egress_proxy is None
+        assert await asyncio.wait_for(reader.read(), timeout=1.0) == b""
+    finally:
+        if writer is not None:
+            writer.close()
+            with contextlib.suppress(ConnectionError):
+                await writer.wait_closed()
+        await session._close()
+        if egress_proxy is not None:
+            await egress_proxy.close()
 
 
 @pytest.mark.asyncio

backend/tests/test_read_before_write_middleware.py

@@ -357,7 +357,7 @@ class TestErrorStringSandboxes:
 
     def test_existing_file_read_still_marked_and_write_allowed(self):
         mw = self._error_string_middleware({self.PATH: "v1"})
-        read_request = _make_request("read_file", {"description": "d", "path": self.PATH})
+        read_request = _make_request("read_file", {"description": "d", "path": self.PATH}, tool_call_id="r1")
         read_handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="r1", name="read_file"))
         read_result = mw.wrap_tool_call(read_request, read_handler)
         assert read_result.additional_kwargs["deerflow_read_mark"]["hash"] == _sha("v1")
@@ -423,7 +423,7 @@ class TestSamePathSerialization:
             return ToolMessage(content="OK", tool_call_id="w-call", name="write_file")
 
         async def run():
-            read_task = asyncio.create_task(mw.awrap_tool_call(_make_request("read_file", {"description": "d", "path": self.PATH}), read_handler))
+            read_task = asyncio.create_task(mw.awrap_tool_call(_make_request("read_file", {"description": "d", "path": self.PATH}, tool_call_id="r-call"), read_handler))
             await asyncio.sleep(0.01)
             write_task = asyncio.create_task(
                 mw.awrap_tool_call(
@@ -957,6 +957,9 @@ class TestStampingGateBugFix:
             pytest.param("a\n\nb\n", {"start_line": 2, "end_line": 2}, id="both_bounds"),
             pytest.param("a\n\n", {"start_line": 2}, id="start_only"),
             pytest.param("\nb\n", {"end_line": 1}, id="end_only"),
+            pytest.param("a\r\n\r\nb\r\n", {"start_line": 2, "end_line": 2}, id="crlf_both_bounds"),
+            pytest.param("a\r\n\r\n", {"start_line": 2}, id="crlf_start_only"),
+            pytest.param("\r\nb\r\n", {"end_line": 1}, id="crlf_end_only"),
         ],
     )
     def test_blank_line_range_stamps_full_hash_and_allows_write(self, tmp_path, monkeypatch, content, line_range):
@@ -970,7 +973,7 @@ class TestStampingGateBugFix:
 
         outputs = tmp_path / "outputs"
         outputs.mkdir()
-        (outputs / "report.md").write_text(content, encoding="utf-8")
+        (outputs / "report.md").write_text(content, encoding="utf-8", newline="")
         runtime = SimpleNamespace(
             state={"sandbox": {"sandbox_id": "local:t-test"}, "thread_data": {"outputs_path": str(outputs)}},
             context={"thread_id": "t-test"},

backend/tests/test_read_mark_tool_call_correlation.py

@@ -0,0 +1,124 @@
+"""Read evidence belongs to the current tool call, including Command updates."""
+
+import asyncio
+import copy
+import hashlib
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Annotated
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.tools import InjectedToolCallId, StructuredTool
+from langgraph.prebuilt.tool_node import ToolCallRequest
+from langgraph.types import Command
+
+from deerflow.agents.middlewares.read_before_write_middleware import READ_MARK_KEY, WRITE_BLOCK_KEY, ReadBeforeWriteMiddleware
+from deerflow.sandbox.read_file_contract import READ_FILE_INVALID_RANGE
+
+
+class GateTestModel(GenericFakeChatModel):
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+
+@pytest.mark.parametrize("invocation", ["invoke", "ainvoke"])
+@pytest.mark.parametrize("layout", ["direct", "matching_first", "matching_last"])
+@pytest.mark.parametrize("read_outcome", ["success", "error", "no_content"])
+def test_read_command_cannot_borrow_another_calls_evidence(tmp_path, invocation, layout, read_outcome):
+    target = tmp_path / "report.md"
+    target.write_text("Original report", encoding="utf-8")
+    path = target.as_posix()
+    previous = ToolMessage(content="Earlier audit completed", tool_call_id="audit-old", id="audit-result", name="audit", additional_kwargs={"audit": "keep"})
+    previous_snapshot = copy.deepcopy(previous)
+    executions = []
+    returned = []
+
+    def read_file(path: str, tool_call_id: Annotated[str, InjectedToolCallId]):
+        """Read an existing file and optionally update another historical result."""
+        executions.append("read")
+        content = Path(path).read_text(encoding="utf-8") if read_outcome == "success" else "Error: read failed" if read_outcome == "error" else READ_FILE_INVALID_RANGE
+        own = ToolMessage(content=content, tool_call_id=tool_call_id, status="error" if read_outcome == "error" else "success")
+        if layout == "direct":
+            result = own
+        else:
+            historical = copy.deepcopy(previous)
+            messages = [own, historical] if layout == "matching_first" else [historical, own]
+            result = Command(update={"messages": messages})
+        returned.append(result)
+        return result
+
+    def write_file(path: str, content: str):
+        """Overwrite an existing report after its current version was read."""
+        executions.append("write")
+        Path(path).write_text(content, encoding="utf-8")
+        return "Written."
+
+    async def aread_file(path: str, tool_call_id: Annotated[str, InjectedToolCallId]):
+        return await asyncio.to_thread(read_file, path, tool_call_id)
+
+    async def awrite_file(path: str, content: str):
+        return await asyncio.to_thread(write_file, path, content)
+
+    tools = [StructuredTool.from_function(read_file, coroutine=aread_file), StructuredTool.from_function(write_file, coroutine=awrite_file)]
+    responses = iter(
+        [
+            AIMessage(content="", tool_calls=[{"name": "read_file", "args": {"path": path}, "id": "read-current"}]),
+            AIMessage(content="", tool_calls=[{"name": "write_file", "args": {"path": path, "content": "Updated report"}, "id": "write-current"}]),
+            AIMessage(content="Finished."),
+        ]
+    )
+    gate = ReadBeforeWriteMiddleware(content_reader=lambda _runtime, path: Path(path).read_text(encoding="utf-8"))
+    graph = create_agent(GateTestModel(messages=responses), tools=tools, middleware=[gate])
+    history = [
+        HumanMessage(content="Audit the report.", id="earlier-request"),
+        AIMessage(content="", tool_calls=[{"name": "audit", "args": {}, "id": "audit-old"}], id="earlier-call"),
+        previous,
+        HumanMessage(content="Read the report, then update it.", id="current-request"),
+    ]
+    original_history = copy.deepcopy(history)
+    initial = {"messages": history}
+    result = graph.invoke(initial, context={"thread_id": "gate-test"}) if invocation == "invoke" else asyncio.run(graph.ainvoke(initial, context={"thread_id": "gate-test"}))
+    successful = read_outcome == "success"
+    assert executions == (["read", "write"] if successful else ["read"])
+    assert target.read_text(encoding="utf-8") == ("Updated report" if successful else "Original report")
+    messages = {message.tool_call_id: message for message in result["messages"] if isinstance(message, ToolMessage)}
+    assert messages["audit-old"] == previous_snapshot
+    if successful:
+        assert messages["read-current"].additional_kwargs[READ_MARK_KEY] == {"path": path, "hash": hashlib.sha256(b"Original report").hexdigest()}
+        assert messages["write-current"].status == "success"
+    else:
+        assert READ_MARK_KEY not in messages["read-current"].additional_kwargs
+        assert messages["write-current"].status == "error"
+        assert WRITE_BLOCK_KEY in messages["write-current"].additional_kwargs
+    assert history == original_history
+    if layout != "direct":
+        command_messages = returned[0].update["messages"]
+        historical = next(message for message in command_messages if message.tool_call_id == "audit-old")
+        assert historical == previous_snapshot
+
+
+@pytest.mark.parametrize("container", ["direct", "single", "list", "tuple"])
+@pytest.mark.parametrize("matching", [True, False])
+def test_read_mark_requires_matching_result_before_inspecting_file(container, matching):
+    inspections = []
+
+    def reader(_runtime, path):
+        inspections.append(path)
+        return "Report contents"
+
+    gate = ReadBeforeWriteMiddleware(content_reader=reader)
+    message = ToolMessage(content="Report contents", tool_call_id="read-current" if matching else "other-call", additional_kwargs={"keep": "metadata"})
+    messages = message if container == "single" else [message] if container == "list" else (message,)
+    result = message if container == "direct" else Command(update={"messages": messages, "other_state": "keep"}, goto="next")
+    request = ToolCallRequest(tool_call={"name": "read_file", "args": {"path": "/report.md"}, "id": "read-current"}, tool=None, state={}, runtime=SimpleNamespace(context={"thread_id": "gate-test"}))
+    actual = gate.wrap_tool_call(request, lambda _request: result)
+    assert actual is result
+    assert inspections == (["/report.md"] if matching else [])
+    assert (READ_MARK_KEY in message.additional_kwargs) is matching
+    assert message.additional_kwargs["keep"] == "metadata"
+    if isinstance(result, Command):
+        assert result.goto == "next"
+        assert result.update["other_state"] == "keep"