diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 1aba52051..9c437bbdf 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -214,27 +214,28 @@ Lead-agent middlewares are assembled in strict order across three functions: the 7. **LLMErrorHandlingMiddleware** - Normalizes provider/model invocation failures into recoverable assistant-facing errors before later stages run 8. **GuardrailMiddleware** - *(optional, if `guardrails.enabled`)* Pre-tool-call authorization via pluggable `GuardrailProvider`; returns an error ToolMessage on deny. Providers: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) 9. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution -10. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting +10. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Version gate on file writes (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped) +11. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting **Lead-only middlewares** (`build_middlewares`, appended after the base): -11. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse -12. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event -13. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. -14. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits -15. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool -16. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position -17. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint. -18. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses) -19. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call -20. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) -21. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives -22. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit -23. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer -24. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits -25. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail -26. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first -27. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) +12. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse +13. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event +14. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. +15. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits +16. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool +17. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position +18. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint. +19. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses) +20. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call +21. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) +22. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives +23. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit +24. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer +25. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits +26. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail +27. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first +28. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) ### Configuration System @@ -328,8 +329,8 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti - `bash` - Execute commands with path translation and error handling. For `LocalSandbox` (host bash), POSIX output is captured through bounded pipe-drain threads and stdin is `/dev/null`, so a backgrounded long-lived process (`server &`) returns immediately instead of blocking the turn on an inherited pipe, while unredirected background output is drained without growing anonymous temp files. Commands that read stdin get immediate EOF. The command runs in its own process group with a wall-clock timeout (`sandbox.bash_command_timeout`, default 600s); on timeout the whole group is killed and the agent gets a notice telling it to background long-lived processes. The bash tool description itself also instructs the model to background long-lived processes (e.g. servers) up front so it doesn't waste the turn waiting on a foreground server. See `LocalSandbox.execute_command` / `_run_posix_command` and `bash_tool`'s docstring. - `ls` - Directory listing (tree format, max 2 levels) - `read_file` - Read file contents with optional line range -- `write_file` - Write/append to files, creates directories; overwrites by default and exposes the `append` argument in the model-facing schema for end-of-file writes -- `str_replace` - Substring replacement (single or all occurrences); same-path serialization is scoped to `(sandbox.id, path)` so isolated sandboxes do not contend on identical virtual paths inside one process +- `write_file` - Write/append to files, creates directories; overwrites by default and exposes the `append` argument in the model-facing schema for end-of-file writes; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain) +- `str_replace` - Substring replacement (single or all occurrences); same-path serialization is scoped to `(sandbox.id, path)` so isolated sandboxes do not contend on identical virtual paths inside one process; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain) ### Subagent System (`packages/harness/deerflow/subagents/`) diff --git a/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py new file mode 100644 index 000000000..ea75acdcd --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py @@ -0,0 +1,265 @@ +"""Deterministic read-before-write gate for file-modifying tools (issue #3857). + +The lead agent's duplicate-output failure mode (the same report section +appended five times) came from "append-only, never read back" writes. This +middleware enforces a version gate: modifying an existing file requires a +``read_file`` of the file's *current* version earlier in the conversation. + +Design invariants: +- Tools stay stateless. The read mark (``sha256`` of the full file content) + is stamped on the ``read_file`` ToolMessage's ``additional_kwargs``, so the + gate's state lives in ``state["messages"]``. +- Summarization deleting the read result deletes the mark with it — the gate + can never pass while the read content is gone from context. +- Writes never refresh marks: any successful write changes the file hash and + therefore invalidates every earlier read, forcing a re-read between + consecutive modifications. +- Gate check and tool execution are serialized per (scope, path): LangGraph + runs the tool calls of one AIMessage concurrently, so without a critical + section two same-turn writes could both pass on one stale mark before + either mutation lands. The same lock covers ``read_file`` + mark stamping, + so a mark always hashes the version the model was actually shown. +- Fail-open: if the gate itself cannot inspect the file (sandbox hiccup, + binary content, or sandboxes like AIO/E2B that report read failures as + ``"Error: ..."`` strings instead of raising), it lets the tool run and + produce its own error. +""" + +import asyncio +import hashlib +import logging +import posixpath +import threading +import weakref +from collections.abc import Awaitable, Callable +from typing import Any, override + +from langchain.agents.middleware import AgentMiddleware +from langchain_core.messages import ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest +from langgraph.types import Command + +from deerflow.sandbox.tools import read_current_file_content + +logger = logging.getLogger(__name__) + +READ_MARK_KEY = "deerflow_read_mark" + +_READ_TOOLS = frozenset({"read_file"}) +_GATED_WRITE_TOOLS = frozenset({"write_file", "str_replace"}) + +# AIO/E2B-style sandboxes convert read failures (including missing files) +# into "Error: ..." strings instead of raising. Content with this prefix is +# treated as "cannot inspect" — the gate fails open and no mark is stamped. +_UNINSPECTABLE_CONTENT_PREFIX = "Error:" + +_BLOCK_MESSAGE = ( + "Error: {tool_name} blocked — {path} already exists and you have not read its current version. " + "Any write invalidates earlier reads, so re-read before every modification. " + "Call read_file on it (a ranged read of the relevant section is enough, e.g. the last ~30 lines " + "before an append), check what is already there, then retry." +) + +# 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() +_GATE_LOCKS_GUARD = threading.Lock() + + +def _get_gate_lock(scope: str, norm_path: str) -> threading.Lock: + key = (scope, norm_path) + with _GATE_LOCKS_GUARD: + lock = _GATE_LOCKS.get(key) + if lock is None: + lock = threading.Lock() + _GATE_LOCKS[key] = lock + return lock + + +def _normalize_mark_path(path: str) -> str: + return posixpath.normpath(path) + + +def _content_hash(content: str) -> str: + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + +class ReadBeforeWriteMiddleware(AgentMiddleware): + """Version gate: block writes to existing files not read at their current version.""" + + def __init__(self, content_reader: Callable[[Any, str], str] | None = None) -> None: + super().__init__() + self._content_reader = content_reader or read_current_file_content + + @override + def wrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], ToolMessage | Command], + ) -> ToolMessage | Command: + name = request.tool_call.get("name") + if name in _GATED_WRITE_TOOLS: + path = self._requested_path(request) + if path is None: + return handler(request) + with self._lock_for(request, path): + blocked = self._check_write_gate(request) + if blocked is not None: + return blocked + return handler(request) + if name in _READ_TOOLS: + path = self._requested_path(request) + if path is None: + return handler(request) + with self._lock_for(request, path): + result = handler(request) + self._attach_read_mark(request, result) + return result + return handler(request) + + @override + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]], + ) -> ToolMessage | Command: + name = request.tool_call.get("name") + if name in _GATED_WRITE_TOOLS: + path = self._requested_path(request) + if path is None: + return await handler(request) + # threading.Lock may be released from a different thread than the + # acquiring one, so acquiring in a worker thread and releasing on + # the event-loop thread is safe. + lock = self._lock_for(request, path) + await asyncio.to_thread(lock.acquire) + try: + blocked = await asyncio.to_thread(self._check_write_gate, request) + if blocked is not None: + return blocked + return await handler(request) + finally: + lock.release() + if name in _READ_TOOLS: + path = self._requested_path(request) + if path is None: + return await handler(request) + lock = self._lock_for(request, path) + await asyncio.to_thread(lock.acquire) + try: + result = await handler(request) + await asyncio.to_thread(self._attach_read_mark, request, result) + return result + finally: + lock.release() + return await handler(request) + + # -- locking --------------------------------------------------------- + + def _lock_for(self, request: ToolCallRequest, path: str) -> threading.Lock: + return _get_gate_lock(self._lock_scope(request), _normalize_mark_path(path)) + + @staticmethod + def _lock_scope(request: ToolCallRequest) -> str: + """Scope locks per thread (or sandbox) so unrelated agents never contend.""" + context = getattr(request.runtime, "context", None) + if isinstance(context, dict): + thread_id = context.get("thread_id") + if isinstance(thread_id, str) and thread_id: + return thread_id + state = request.state + if isinstance(state, dict): + sandbox_state = state.get("sandbox") + if isinstance(sandbox_state, dict): + sandbox_id = sandbox_state.get("sandbox_id") + if isinstance(sandbox_id, str) and sandbox_id: + return sandbox_id + return "global" + + # -- gate ---------------------------------------------------------- + + def _check_write_gate(self, request: ToolCallRequest) -> ToolMessage | None: + tool_call = request.tool_call + path = self._requested_path(request) + if path is None: + return None + try: + current = self._content_reader(request.runtime, path) + except FileNotFoundError: + # write_file creates the file; str_replace surfaces its own error. + return None + except Exception: + logger.warning("read-before-write gate could not inspect %r; allowing the write (fail-open)", path, exc_info=True) + return None + if current.startswith(_UNINSPECTABLE_CONTENT_PREFIX): + # Error-string sandbox read channel (AIO/E2B): "missing" and + # "unreadable" are indistinguishable here, so fail open — creation + # proceeds and genuine failures surface from the tool itself. + logger.debug("read-before-write gate got an error-string read for %r; allowing the write (fail-open)", path) + return None + norm_path = _normalize_mark_path(path) + if self._latest_mark_hash(request.state, norm_path) == _content_hash(current): + return None + tool_name = str(tool_call.get("name", "write")) + return ToolMessage( + content=_BLOCK_MESSAGE.format(tool_name=tool_name, path=path), + tool_call_id=str(tool_call.get("id", "")), + name=tool_name, + status="error", + ) + + @staticmethod + def _requested_path(request: ToolCallRequest) -> str | None: + args = request.tool_call.get("args") or {} + if not isinstance(args, dict): + return None + path = args.get("path") + return path if isinstance(path, str) and path else None + + @staticmethod + def _latest_mark_hash(state: Any, norm_path: str) -> str | None: + messages = state.get("messages") if isinstance(state, dict) else getattr(state, "messages", None) + if not messages: + return None + for message in reversed(messages): + if not isinstance(message, ToolMessage): + continue + mark = (message.additional_kwargs or {}).get(READ_MARK_KEY) + if isinstance(mark, dict) and mark.get("path") == norm_path: + mark_hash = mark.get("hash") + return mark_hash if isinstance(mark_hash, str) else None + return None + + # -- mark stamping --------------------------------------------------- + + def _attach_read_mark(self, request: ToolCallRequest, result: ToolMessage | Command) -> None: + path = self._requested_path(request) + if path is None: + return + message = self._extract_tool_message(result) + if message is None or message.status == "error": + return + try: + content = self._content_reader(request.runtime, path) + except Exception: + logger.debug("read-before-write mark skipped for %r: file not hashable", path, exc_info=True) + return + if content.startswith(_UNINSPECTABLE_CONTENT_PREFIX): + logger.debug("read-before-write mark skipped for %r: error-string read channel", path) + return + message.additional_kwargs[READ_MARK_KEY] = { + "path": _normalize_mark_path(path), + "hash": _content_hash(content), + } + + @staticmethod + def _extract_tool_message(result: ToolMessage | Command) -> ToolMessage | None: + if isinstance(result, ToolMessage): + return result + 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] + return None diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index 206fd50bf..6cc37acf1 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -192,6 +192,12 @@ def _build_runtime_middlewares( from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware tail.append(SandboxAuditMiddleware()) + + if app_config.read_before_write.enabled: + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + + tail.append(ReadBeforeWriteMiddleware()) + tail.append(ToolErrorHandlingMiddleware()) return [*outer_wrappers, *thread_hooks, *tail] diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index 6ef27fdcf..e7ebfeb5d 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -21,6 +21,7 @@ from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_ from deerflow.config.loop_detection_config import LoopDetectionConfig from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_dict from deerflow.config.model_config import ModelConfig +from deerflow.config.read_before_write_config import ReadBeforeWriteConfig from deerflow.config.reload_boundary import format_field_description from deerflow.config.run_events_config import RunEventsConfig from deerflow.config.runtime_paths import existing_project_file @@ -172,6 +173,7 @@ class AppConfig(BaseModel): ), ) loop_detection: LoopDetectionConfig = Field(default_factory=LoopDetectionConfig, description="Loop detection middleware configuration") + read_before_write: ReadBeforeWriteConfig = Field(default_factory=ReadBeforeWriteConfig, description="Read-before-write file gate middleware configuration") safety_finish_reason: SafetyFinishReasonConfig = Field(default_factory=SafetyFinishReasonConfig, description="Provider safety-filter finish_reason interception middleware configuration") auth: AuthAppConfig = Field(default_factory=AuthAppConfig, description="Authentication configuration (local + OIDC SSO)") model_config = ConfigDict(extra="allow") diff --git a/backend/packages/harness/deerflow/config/read_before_write_config.py b/backend/packages/harness/deerflow/config/read_before_write_config.py new file mode 100644 index 000000000..e67ebaf06 --- /dev/null +++ b/backend/packages/harness/deerflow/config/read_before_write_config.py @@ -0,0 +1,18 @@ +"""Configuration for the read-before-write file gate middleware (issue #3857).""" + +from pydantic import BaseModel, Field + + +class ReadBeforeWriteConfig(BaseModel): + """Deterministic version gate on file-modifying tools. + + When enabled, ``write_file`` (append or overwrite of an existing file) and + ``str_replace`` are blocked unless the file was read (``read_file``) after + its last modification, forcing the agent to see the file's current state + before changing it. + """ + + enabled: bool = Field( + default=True, + description="Whether to block writes to existing files that were not read at their current version", + ) diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index 5217d80d7..d5b62c334 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -1707,6 +1707,29 @@ async def _grep_tool_async( grep_tool.coroutine = _grep_tool_async +def read_current_file_content(runtime: Runtime | None, path: str) -> str: + """Read the full current content of ``path`` using read_file's resolution rules. + + Shared by ``read_file_tool`` and ``ReadBeforeWriteMiddleware`` (issue #3857) + so the gate hashes exactly the bytes the read tool would see. Raises + ``FileNotFoundError`` when the file does not exist; other sandbox errors + propagate to the caller. + """ + sandbox = ensure_sandbox_initialized(runtime) + ensure_thread_directories_exist(runtime) + if is_local_sandbox(runtime): + thread_data = get_thread_data(runtime) + validate_local_tool_path(path, thread_data, read_only=True) + if _is_skills_path(path): + path = _resolve_skills_path(path) + elif _is_acp_workspace_path(path): + path = _resolve_acp_workspace_path(path, _extract_thread_id_from_thread_data(thread_data)) + elif not _is_custom_mount_path(path): + path = _resolve_and_validate_user_data_path(path, thread_data) + # Custom mount paths are resolved by LocalSandbox._resolve_path() + return sandbox.read_file(path) + + @tool("read_file", parse_docstring=True) def read_file_tool( runtime: Runtime, @@ -1724,20 +1747,8 @@ def read_file_tool( end_line: Optional ending line number (1-indexed, inclusive). Use with start_line to read a specific range. """ try: - sandbox = ensure_sandbox_initialized(runtime) - ensure_thread_directories_exist(runtime) requested_path = path - if is_local_sandbox(runtime): - thread_data = get_thread_data(runtime) - validate_local_tool_path(path, thread_data, read_only=True) - if _is_skills_path(path): - path = _resolve_skills_path(path) - elif _is_acp_workspace_path(path): - path = _resolve_acp_workspace_path(path, _extract_thread_id_from_thread_data(thread_data)) - elif not _is_custom_mount_path(path): - path = _resolve_and_validate_user_data_path(path, thread_data) - # Custom mount paths are resolved by LocalSandbox._resolve_path() - content = sandbox.read_file(path) + content = read_current_file_content(runtime, path) if not content: return "(empty)" if start_line is not None and end_line is not None: @@ -1808,6 +1819,12 @@ def write_file_tool( ) -> str: """Write text content to a file. By default this overwrites the target file; set append=True to add content to the end without replacing existing content. + READ-BEFORE-WRITE (issue #3857): if the target file already exists (including + append=True), you must have read its CURRENT version with read_file first. + Any write invalidates earlier reads, so re-read between consecutive + modifications — a ranged read of the relevant section is enough. Writes + that fail this check are rejected with an error. + SIZE POLICY (issue #3189): A single non-append write_file call must not exceed 80 KB of UTF-8 content. Oversized single-shot writes correlate with LLM streaming chunk-gap @@ -1903,6 +1920,9 @@ def str_replace_tool( """Replace a substring in a file with another substring. If `replace_all` is False (default), the substring to replace must appear **exactly once** in the file. + READ-BEFORE-WRITE (issue #3857): you must have read the file's CURRENT + version with read_file first; any write invalidates earlier reads. + Args: description: Explain why you are replacing the substring in short words. ALWAYS PROVIDE THIS PARAMETER FIRST. path: The **absolute** path to the file to replace the substring in. ALWAYS PROVIDE THIS PARAMETER SECOND. diff --git a/backend/tests/test_read_before_write_middleware.py b/backend/tests/test_read_before_write_middleware.py new file mode 100644 index 000000000..c07a0b8d8 --- /dev/null +++ b/backend/tests/test_read_before_write_middleware.py @@ -0,0 +1,413 @@ +"""Tests for the read-before-write gate (issue #3857, output layer).""" + +import hashlib +import posixpath +from unittest.mock import MagicMock, patch + +import pytest +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _make_request(name, args, messages=()): + runtime = MagicMock() + runtime.context = {"thread_id": "t-test"} + return ToolCallRequest( + tool_call={"name": name, "args": args, "id": "call-1"}, + tool=None, + state={"messages": list(messages)}, + runtime=runtime, + ) + + +def _read_marked_message(path, content, tool_call_id="r1"): + msg = ToolMessage(content=content[:20], tool_call_id=tool_call_id, name="read_file") + msg.additional_kwargs["deerflow_read_mark"] = {"path": path, "hash": _sha(content)} + return msg + + +def _middleware(files: dict[str, str]): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + + def reader(_runtime, path): + normalized = posixpath.normpath(path) + if normalized not in files: + raise FileNotFoundError(path) + value = files[normalized] + if isinstance(value, Exception): + raise value + return value + + return ReadBeforeWriteMiddleware(content_reader=reader) + + +class TestReadCurrentFileContent: + def test_reads_via_sandbox_with_resolution(self): + from deerflow.sandbox import tools as sandbox_tools + + sandbox = MagicMock() + sandbox.read_file.return_value = "hello" + runtime = MagicMock() + with ( + patch.object(sandbox_tools, "ensure_sandbox_initialized", return_value=sandbox), + patch.object(sandbox_tools, "ensure_thread_directories_exist"), + patch.object(sandbox_tools, "is_local_sandbox", return_value=False), + ): + assert sandbox_tools.read_current_file_content(runtime, "/mnt/user-data/outputs/report.md") == "hello" + sandbox.read_file.assert_called_once_with("/mnt/user-data/outputs/report.md") + + def test_propagates_file_not_found(self): + from deerflow.sandbox import tools as sandbox_tools + + sandbox = MagicMock() + sandbox.read_file.side_effect = FileNotFoundError() + with ( + patch.object(sandbox_tools, "ensure_sandbox_initialized", return_value=sandbox), + patch.object(sandbox_tools, "ensure_thread_directories_exist"), + patch.object(sandbox_tools, "is_local_sandbox", return_value=False), + ): + with pytest.raises(FileNotFoundError): + sandbox_tools.read_current_file_content(MagicMock(), "/mnt/user-data/outputs/missing.md") + + +class TestReadMarkStamping: + def test_read_file_success_stamps_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "v1"}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + mark = result.additional_kwargs["deerflow_read_mark"] + assert mark == {"path": "/mnt/user-data/outputs/report.md", "hash": _sha("v1")} + + def test_ranged_read_stamps_full_file_hash(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "line1\nline2\nline3"}) + request = _make_request( + "read_file", + {"description": "d", "path": "/mnt/user-data/outputs/report.md", "start_line": 3, "end_line": 3}, + ) + handler = MagicMock(return_value=ToolMessage(content="line3", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + assert result.additional_kwargs["deerflow_read_mark"]["hash"] == _sha("line1\nline2\nline3") + + def test_error_tool_message_gets_no_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "v1"}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="boom", tool_call_id="call-1", name="read_file", status="error")) + result = mw.wrap_tool_call(request, handler) + assert "deerflow_read_mark" not in result.additional_kwargs + + def test_reader_failure_means_no_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": RuntimeError("sandbox down")}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + assert "deerflow_read_mark" not in result.additional_kwargs + + def test_non_file_tools_untouched(self): + mw = _middleware({}) + request = _make_request("bash", {"description": "d", "command": "ls"}) + sentinel = ToolMessage(content="ok", tool_call_id="call-1", name="bash") + handler = MagicMock(return_value=sentinel) + assert mw.wrap_tool_call(request, handler) is sentinel + + +class TestWriteGate: + PATH = "/mnt/user-data/outputs/report.md" + + def test_new_file_write_allowed(self): + mw = _middleware({}) # file does not exist + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v1"}) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_overwrite_existing_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert isinstance(result, ToolMessage) + assert result.status == "error" + assert result.tool_call_id == "call-1" + assert "read" in result.content.lower() + + def test_append_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "more", "append": True}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_str_replace_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("str_replace", {"description": "d", "path": self.PATH, "old_str": "v1", "new_str": "v2"}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_str_replace_missing_file_passes_through(self): + mw = _middleware({}) + request = _make_request("str_replace", {"description": "d", "path": self.PATH, "old_str": "a", "new_str": "b"}) + native_error = ToolMessage(content="Error: File not found", tool_call_id="call-1", name="str_replace", status="error") + handler = MagicMock(return_value=native_error) + assert mw.wrap_tool_call(request, handler) is native_error + + def test_fresh_mark_allows_write(self): + mw = _middleware({self.PATH: "v1"}) + messages = [HumanMessage("hi"), AIMessage(""), _read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_stale_mark_after_modification_blocked(self): + mw = _middleware({self.PATH: "v2"}) # file changed since the read of v1 + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v3", "append": True}, messages) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_newest_mark_wins(self): + mw = _middleware({self.PATH: "v2"}) + messages = [_read_marked_message(self.PATH, "v1", "r1"), _read_marked_message(self.PATH, "v2", "r2")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v3"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_mark_removed_by_summarization_blocks(self): + mw = _middleware({self.PATH: "v1"}) + messages = [HumanMessage("Here is a summary of the conversation to date: ...", name="summary")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_gate_read_failure_fails_open(self): + mw = _middleware({self.PATH: RuntimeError("sandbox hiccup")}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_normalized_path_matching(self): + mw = _middleware({self.PATH: "v1"}) + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": "/mnt/user-data/outputs/../outputs/report.md", "content": "v2"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + +class TestAsyncPaths: + PATH = "/mnt/user-data/outputs/report.md" + + def test_async_block(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + + async def handler(_request): + raise AssertionError("handler must not run when blocked") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.status == "error" + + def test_async_read_stamps_mark(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("read_file", {"description": "d", "path": self.PATH}) + + async def handler(_request): + return ToolMessage(content="v1", tool_call_id="call-1", name="read_file") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.additional_kwargs["deerflow_read_mark"]["hash"] == _sha("v1") + + def test_async_allowed_write_calls_handler(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + + async def handler(_request): + return ToolMessage(content="OK", tool_call_id="call-1", name="write_file") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.status != "error" + + +def _wiring_app_config(**overrides): + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + + return AppConfig(sandbox=SandboxConfig(use="test"), **overrides) + + +class TestChainWiring: + def test_enabled_by_default_in_runtime_chain(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware, build_lead_runtime_middlewares + + middlewares = build_lead_runtime_middlewares(app_config=_wiring_app_config()) + types = [type(m) for m in middlewares] + assert ReadBeforeWriteMiddleware in types + assert types.index(SandboxAuditMiddleware) < types.index(ReadBeforeWriteMiddleware) < types.index(ToolErrorHandlingMiddleware) + + def test_disabled_removes_middleware(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares + from deerflow.config.read_before_write_config import ReadBeforeWriteConfig + + app_config = _wiring_app_config(read_before_write=ReadBeforeWriteConfig(enabled=False)) + middlewares = build_lead_runtime_middlewares(app_config=app_config) + assert ReadBeforeWriteMiddleware not in [type(m) for m in middlewares] + + def test_subagents_get_the_gate_too(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares + + middlewares = build_subagent_runtime_middlewares(app_config=_wiring_app_config()) + assert ReadBeforeWriteMiddleware in [type(m) for m in middlewares] + + +class TestErrorStringSandboxes: + """AIO/E2B sandboxes report read failures as "Error: ..." strings, not exceptions.""" + + PATH = "/mnt/user-data/outputs/report.md" + + def _error_string_middleware(self, files): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + + def reader(_runtime, path): + normalized = posixpath.normpath(path) + if normalized not in files: + return f"Error: can't read file {path}: file not found" + return files[normalized] + + return ReadBeforeWriteMiddleware(content_reader=reader) + + def test_new_file_creation_not_blocked(self): + mw = self._error_string_middleware({}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v1"}) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_no_mark_when_reread_returns_error_string(self): + mw = self._error_string_middleware({}) + request = _make_request("read_file", {"description": "d", "path": self.PATH}) + handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + assert "deerflow_read_mark" not in result.additional_kwargs + + def test_existing_file_still_gated(self): + mw = self._error_string_middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + 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_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") + + write_request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, [read_result]) + write_handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(write_request, write_handler) + write_handler.assert_called_once() + assert result.status != "error" + + +class TestSamePathSerialization: + """LangGraph runs one AIMessage's tool calls concurrently; the gate must not + let two same-turn writes pass on one stale mark (issue #3912 review).""" + + PATH = "/mnt/user-data/outputs/report.md" + + def test_parallel_appends_exactly_one_lands(self): + import asyncio + + files = {self.PATH: "v1"} + mw = _middleware(files) + messages = [_read_marked_message(self.PATH, "v1")] + + def make_handler(suffix): + async def handler(_request): + await asyncio.sleep(0.02) + files[self.PATH] = files[self.PATH] + suffix + return ToolMessage(content="OK", tool_call_id="call-1", name="write_file") + + return handler + + async def run(): + return await asyncio.gather( + mw.awrap_tool_call( + _make_request("write_file", {"description": "d", "path": self.PATH, "content": "A", "append": True}, messages), + make_handler("A"), + ), + mw.awrap_tool_call( + _make_request("write_file", {"description": "d", "path": self.PATH, "content": "B", "append": True}, messages), + make_handler("B"), + ), + ) + + results = asyncio.run(run()) + assert sorted(r.status for r in results) == ["error", "success"] + assert files[self.PATH] in ("v1A", "v1B") + + def test_read_mark_matches_content_shown_to_model(self): + import asyncio + + files = {self.PATH: "v1"} + mw = _middleware(files) + write_messages = [_read_marked_message(self.PATH, "v1")] + + async def read_handler(_request): + snapshot = files[self.PATH] + await asyncio.sleep(0.03) + return ToolMessage(content=snapshot, tool_call_id="r-call", name="read_file") + + async def write_handler(_request): + files[self.PATH] = "v2" + 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)) + await asyncio.sleep(0.01) + write_task = asyncio.create_task( + mw.awrap_tool_call( + _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, write_messages), + write_handler, + ) + ) + return await asyncio.gather(read_task, write_task) + + read_result, _write_result = asyncio.run(run()) + mark = read_result.additional_kwargs.get("deerflow_read_mark") + assert mark is not None + assert mark["hash"] == _sha(read_result.content) diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index 8c2f63007..e30a47c97 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/backend/tests/test_tool_error_handling_middleware.py @@ -142,11 +142,12 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware assert captured["app_config"] is app_config # 8 baseline (InputSanitization, ToolOutputBudget, ThreadData, Sandbox, # DanglingToolCall, LLMErrorHandling, SandboxAudit, ToolErrorHandling) - # + 1 SafetyFinishReasonMiddleware (enabled by default). + # + 1 ReadBeforeWriteMiddleware + 1 SafetyFinishReasonMiddleware (both + # enabled by default). from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware - assert len(middlewares) == 9 + assert len(middlewares) == 10 assert isinstance(middlewares[0], FakeMiddleware) # InputSanitizationMiddleware stub assert isinstance(middlewares[1], ToolOutputBudgetMiddleware) assert any(isinstance(m, ToolErrorHandlingMiddleware) for m in middlewares) @@ -187,6 +188,7 @@ def test_build_lead_runtime_middlewares_chain_order_matches_agents_md(): from deerflow.agents.middlewares.dangling_tool_call_middleware import DanglingToolCallMiddleware from deerflow.agents.middlewares.input_sanitization_middleware import InputSanitizationMiddleware from deerflow.agents.middlewares.llm_error_handling_middleware import LLMErrorHandlingMiddleware + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware @@ -212,6 +214,7 @@ def test_build_lead_runtime_middlewares_chain_order_matches_agents_md(): ("DanglingToolCallMiddleware", DanglingToolCallMiddleware), ("LLMErrorHandlingMiddleware", LLMErrorHandlingMiddleware), ("SandboxAuditMiddleware", SandboxAuditMiddleware), + ("ReadBeforeWriteMiddleware", ReadBeforeWriteMiddleware), ("ToolErrorHandlingMiddleware", ToolErrorHandlingMiddleware), ] actual = [(label, idx_of(cls, label=label)) for label, cls in expected_order] diff --git a/config.example.yaml b/config.example.yaml index ffedd6d7c..e4115677b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -920,6 +920,17 @@ loop_detection: # warn: 150 # hard_limit: 300 +# ============================================================================ +# Read-Before-Write File Gate (issue #3857) +# ============================================================================ +# Blocks write_file (append / overwrite of an existing file) and str_replace +# unless the agent has read the file's current version first; any write +# invalidates earlier reads, forcing a re-read between consecutive edits. +# Deterministic guardrail against blind duplicate appends in long tasks. + +read_before_write: + enabled: true + # ============================================================================ # Provider Safety Termination Configuration # ============================================================================ diff --git a/docs/superpowers/plans/2026-07-02-read-before-write-gate.md b/docs/superpowers/plans/2026-07-02-read-before-write-gate.md new file mode 100644 index 000000000..523544d31 --- /dev/null +++ b/docs/superpowers/plans/2026-07-02-read-before-write-gate.md @@ -0,0 +1,786 @@ +# Read-Before-Write Gate (ReadBeforeWriteMiddleware) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Deterministically block `write_file` (append / overwrite-existing) and `str_replace` unless the agent has read the file's *current* version, fixing issue #3857's output-layer duplicate-append failure. + +**Architecture:** A new `ReadBeforeWriteMiddleware` intercepts file tools via `wrap_tool_call`/`awrap_tool_call`. On a successful `read_file` it stamps `sha256(full file content)` into the returned `ToolMessage.additional_kwargs["deerflow_read_mark"]`. Before a gated write it re-hashes the file and requires the newest mark for that path in `state["messages"]` to match. Marks live on messages, so summarization deleting the read result automatically invalidates the mark (the issue's "mark tied to context presence" requirement) — no reserved region needed. Writes never refresh marks, so consecutive modifications force a re-read. + +**Tech Stack:** Python 3.12, LangChain `AgentMiddleware`, LangGraph `ToolCallRequest`, pytest. + +## Global Constraints + +- Spec: `docs/superpowers/specs/2026-07-02-read-before-write-gate-design.md`. +- Tools stay stateless — all gate state derives from messages (issue #3857 requirement). +- Fail-open on unexpected gate errors (only a missing/stale mark blocks); blocked-tool errors must not leak backend config keys/paths. +- Async hooks must not run blocking IO on the event loop (`asyncio.to_thread` / `ensure_sandbox_initialized_async` pattern, see `_run_sync_tool_after_async_sandbox_init` in `sandbox/tools.py`). +- Default enabled (`read_before_write.enabled: true`); config schema change ⇒ bump `config_version` in `config.example.yaml` (15 → 16 at planning time; landed as 16 → 17 after merging upstream/main, where 16 was taken by `max_recursion_limit`). +- Backend TDD is mandatory; run `cd backend && make format` before finishing. +- All backend commands run from the repo's `backend/` directory with `PYTHONPATH=. uv run pytest ...`. + +--- + +### Task 1: Extract `read_current_file_content` helper in sandbox tools + +**Files:** +- Modify: `backend/packages/harness/deerflow/sandbox/tools.py` (read_file_tool at ~1666; place helper right above `@tool("read_file", ...)`) +- Test: `backend/tests/test_read_before_write_middleware.py` (new file, helper test only in this task) + +**Interfaces:** +- Produces: `deerflow.sandbox.tools.read_current_file_content(runtime, path: str) -> str` — full current content using `read_file`'s path-resolution rules; raises `FileNotFoundError` when missing, propagates other errors. + +- [ ] **Step 1: Write the failing test** + +```python +"""Tests for the read-before-write gate (issue #3857, output layer).""" + +from unittest.mock import MagicMock, patch + + +class TestReadCurrentFileContent: + def test_reads_via_sandbox_with_resolution(self): + from deerflow.sandbox import tools as sandbox_tools + + sandbox = MagicMock() + sandbox.read_file.return_value = "hello" + runtime = MagicMock() + with ( + patch.object(sandbox_tools, "ensure_sandbox_initialized", return_value=sandbox), + patch.object(sandbox_tools, "ensure_thread_directories_exist"), + patch.object(sandbox_tools, "is_local_sandbox", return_value=False), + ): + assert sandbox_tools.read_current_file_content(runtime, "/mnt/user-data/outputs/report.md") == "hello" + sandbox.read_file.assert_called_once_with("/mnt/user-data/outputs/report.md") + + def test_propagates_file_not_found(self): + import pytest + + from deerflow.sandbox import tools as sandbox_tools + + sandbox = MagicMock() + sandbox.read_file.side_effect = FileNotFoundError() + with ( + patch.object(sandbox_tools, "ensure_sandbox_initialized", return_value=sandbox), + patch.object(sandbox_tools, "ensure_thread_directories_exist"), + patch.object(sandbox_tools, "is_local_sandbox", return_value=False), + ): + with pytest.raises(FileNotFoundError): + sandbox_tools.read_current_file_content(MagicMock(), "/mnt/user-data/outputs/missing.md") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py -v` +Expected: FAIL — `AttributeError: ... has no attribute 'read_current_file_content'` + +- [ ] **Step 3: Implement the helper and reuse it in `read_file_tool`** + +Add above `@tool("read_file", parse_docstring=True)` in `sandbox/tools.py`: + +```python +def read_current_file_content(runtime: Runtime | None, path: str) -> str: + """Read the full current content of ``path`` using read_file's resolution rules. + + Shared by ``read_file_tool`` and ``ReadBeforeWriteMiddleware`` (issue #3857) + so the gate hashes exactly the bytes the read tool would see. Raises + ``FileNotFoundError`` when the file does not exist; other sandbox errors + propagate to the caller. + """ + sandbox = ensure_sandbox_initialized(runtime) + ensure_thread_directories_exist(runtime) + if is_local_sandbox(runtime): + thread_data = get_thread_data(runtime) + validate_local_tool_path(path, thread_data, read_only=True) + if _is_skills_path(path): + path = _resolve_skills_path(path) + elif _is_acp_workspace_path(path): + path = _resolve_acp_workspace_path(path, _extract_thread_id_from_thread_data(thread_data)) + elif not _is_custom_mount_path(path): + path = _resolve_and_validate_user_data_path(path, thread_data) + # Custom mount paths are resolved by LocalSandbox._resolve_path() + return sandbox.read_file(path) +``` + +Then replace the duplicated body at the top of `read_file_tool` (keep behavior identical): + +```python + try: + requested_path = path + content = read_current_file_content(runtime, path) + if not content: + return "(empty)" +``` + +(delete the now-redundant `sandbox = ensure_sandbox_initialized(...)` / `ensure_thread_directories_exist(...)` / local-resolution block / `content = sandbox.read_file(path)` lines from `read_file_tool`; note `requested_path = path` must be assigned **before** the helper call so error messages keep the original path.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py tests/test_sandbox_tools.py -v` (second file: existing read_file coverage if present; otherwise `uv run pytest tests -k read_file -v`) +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add backend/packages/harness/deerflow/sandbox/tools.py backend/tests/test_read_before_write_middleware.py +git commit -m "refactor(sandbox): extract read_current_file_content helper (#3857)" +``` + +--- + +### Task 2: `ReadBeforeWriteMiddleware` — mark stamping + sync write gate + +**Files:** +- Create: `backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py` +- Test: `backend/tests/test_read_before_write_middleware.py` + +**Interfaces:** +- Consumes: `read_current_file_content(runtime, path)` from Task 1. +- Produces: `ReadBeforeWriteMiddleware(content_reader=None)` with `wrap_tool_call`; module constants `READ_MARK_KEY = "deerflow_read_mark"`. `content_reader: Callable[[Any, str], str]` defaults to `read_current_file_content` (injectable for tests). + +- [ ] **Step 1: Write failing tests** + +Append to `backend/tests/test_read_before_write_middleware.py`: + +```python +import hashlib + +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _make_request(name, args, messages=()): + return ToolCallRequest( + tool_call={"name": name, "args": args, "id": "call-1"}, + tool=None, + state={"messages": list(messages)}, + runtime=MagicMock(), + ) + + +def _read_marked_message(path, content, tool_call_id="r1"): + msg = ToolMessage(content=content[:20], tool_call_id=tool_call_id, name="read_file") + msg.additional_kwargs["deerflow_read_mark"] = {"path": path, "hash": _sha(content)} + return msg + + +def _middleware(files: dict[str, str]): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + + def reader(_runtime, path): + import posixpath + + normalized = posixpath.normpath(path) + if normalized not in files: + raise FileNotFoundError(path) + value = files[normalized] + if isinstance(value, Exception): + raise value + return value + + return ReadBeforeWriteMiddleware(content_reader=reader) + + +class TestReadMarkStamping: + def test_read_file_success_stamps_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "v1"}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + mark = result.additional_kwargs["deerflow_read_mark"] + assert mark == {"path": "/mnt/user-data/outputs/report.md", "hash": _sha("v1")} + + def test_ranged_read_stamps_full_file_hash(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "line1\nline2\nline3"}) + request = _make_request( + "read_file", + {"description": "d", "path": "/mnt/user-data/outputs/report.md", "start_line": 3, "end_line": 3}, + ) + handler = MagicMock(return_value=ToolMessage(content="line3", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + assert result.additional_kwargs["deerflow_read_mark"]["hash"] == _sha("line1\nline2\nline3") + + def test_error_tool_message_gets_no_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": "v1"}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="boom", tool_call_id="call-1", name="read_file", status="error")) + result = mw.wrap_tool_call(request, handler) + assert "deerflow_read_mark" not in result.additional_kwargs + + def test_reader_failure_means_no_mark(self): + mw = _middleware({"/mnt/user-data/outputs/report.md": RuntimeError("sandbox down")}) + request = _make_request("read_file", {"description": "d", "path": "/mnt/user-data/outputs/report.md"}) + handler = MagicMock(return_value=ToolMessage(content="v1", tool_call_id="call-1", name="read_file")) + result = mw.wrap_tool_call(request, handler) + assert "deerflow_read_mark" not in result.additional_kwargs + + def test_non_file_tools_untouched(self): + mw = _middleware({}) + request = _make_request("bash", {"description": "d", "command": "ls"}) + sentinel = ToolMessage(content="ok", tool_call_id="call-1", name="bash") + handler = MagicMock(return_value=sentinel) + assert mw.wrap_tool_call(request, handler) is sentinel + + +class TestWriteGate: + PATH = "/mnt/user-data/outputs/report.md" + + def test_new_file_write_allowed(self): + mw = _middleware({}) # file does not exist + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v1"}) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_overwrite_existing_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert isinstance(result, ToolMessage) + assert result.status == "error" + assert result.tool_call_id == "call-1" + assert "read" in result.content.lower() + + def test_append_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "more", "append": True}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_str_replace_without_read_blocked(self): + mw = _middleware({self.PATH: "v1"}) + request = _make_request("str_replace", {"description": "d", "path": self.PATH, "old_str": "v1", "new_str": "v2"}) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_str_replace_missing_file_passes_through(self): + mw = _middleware({}) + request = _make_request("str_replace", {"description": "d", "path": self.PATH, "old_str": "a", "new_str": "b"}) + native_error = ToolMessage(content="Error: File not found", tool_call_id="call-1", name="str_replace", status="error") + handler = MagicMock(return_value=native_error) + assert mw.wrap_tool_call(request, handler) is native_error + + def test_fresh_mark_allows_write(self): + mw = _middleware({self.PATH: "v1"}) + messages = [HumanMessage("hi"), AIMessage(""), _read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_stale_mark_after_modification_blocked(self): + mw = _middleware({self.PATH: "v2"}) # file changed since the read of v1 + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v3", "append": True}, messages) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_newest_mark_wins(self): + mw = _middleware({self.PATH: "v2"}) + messages = [_read_marked_message(self.PATH, "v1", "r1"), _read_marked_message(self.PATH, "v2", "r2")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v3"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_mark_removed_by_summarization_blocks(self): + mw = _middleware({self.PATH: "v1"}) + messages = [HumanMessage("Here is a summary of the conversation to date: ...", name="summary")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + handler = MagicMock() + result = mw.wrap_tool_call(request, handler) + handler.assert_not_called() + assert result.status == "error" + + def test_gate_read_failure_fails_open(self): + mw = _middleware({self.PATH: RuntimeError("sandbox hiccup")}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" + + def test_normalized_path_matching(self): + mw = _middleware({self.PATH: "v1"}) + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": "/mnt/user-data/outputs/../outputs/report.md", "content": "v2"}, messages) + handler = MagicMock(return_value=ToolMessage(content="OK", tool_call_id="call-1", name="write_file")) + result = mw.wrap_tool_call(request, handler) + handler.assert_called_once() + assert result.status != "error" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py -v` +Expected: FAIL — `ModuleNotFoundError: ... read_before_write_middleware` + +- [ ] **Step 3: Implement the middleware (sync paths)** + +Create `backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py`: + +```python +"""Deterministic read-before-write gate for file-modifying tools (issue #3857). + +The lead agent's duplicate-output failure mode (the same report section +appended five times) came from "append-only, never read back" writes. This +middleware enforces a version gate: modifying an existing file requires a +``read_file`` of the file's *current* version earlier in the conversation. + +Design invariants: +- Tools stay stateless. The read mark (``sha256`` of the full file content) + is stamped on the ``read_file`` ToolMessage's ``additional_kwargs``, so the + gate's state lives in ``state["messages"]``. +- Summarization deleting the read result deletes the mark with it — the gate + can never pass while the read content is gone from context. +- Writes never refresh marks: any successful write changes the file hash and + therefore invalidates every earlier read, forcing a re-read between + consecutive modifications. +- Fail-open: if the gate itself cannot inspect the file (sandbox hiccup, + binary content), it lets the tool run and produce its own error. +""" + +import asyncio +import hashlib +import logging +import posixpath +from collections.abc import Awaitable, Callable +from typing import Any, override + +from langchain.agents.middleware import AgentMiddleware +from langchain_core.messages import ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest +from langgraph.types import Command + +from deerflow.sandbox.tools import read_current_file_content + +logger = logging.getLogger(__name__) + +READ_MARK_KEY = "deerflow_read_mark" + +_READ_TOOLS = frozenset({"read_file"}) +_GATED_WRITE_TOOLS = frozenset({"write_file", "str_replace"}) + +_BLOCK_MESSAGE = ( + "Error: {tool_name} blocked — {path} already exists and you have not read its current version. " + "Any write invalidates earlier reads, so re-read before every modification. " + "Call read_file on it (a ranged read of the relevant section is enough, e.g. the last ~30 lines " + "before an append), check what is already there, then retry." +) + + +def _normalize_mark_path(path: str) -> str: + return posixpath.normpath(path) + + +def _content_hash(content: str) -> str: + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + +class ReadBeforeWriteMiddleware(AgentMiddleware): + """Version gate: block writes to existing files not read at their current version.""" + + def __init__(self, content_reader: Callable[[Any, str], str] | None = None) -> None: + super().__init__() + self._content_reader = content_reader or read_current_file_content + + # -- wrap_tool_call ------------------------------------------------ + + @override + def wrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], ToolMessage | Command], + ) -> ToolMessage | Command: + name = request.tool_call.get("name") + if name in _GATED_WRITE_TOOLS: + blocked = self._check_write_gate(request) + if blocked is not None: + return blocked + return handler(request) + result = handler(request) + if name in _READ_TOOLS: + self._attach_read_mark(request, result) + return result + + @override + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]], + ) -> ToolMessage | Command: + name = request.tool_call.get("name") + if name in _GATED_WRITE_TOOLS: + blocked = await asyncio.to_thread(self._check_write_gate, request) + if blocked is not None: + return blocked + return await handler(request) + result = await handler(request) + if name in _READ_TOOLS: + await asyncio.to_thread(self._attach_read_mark, request, result) + return result + + # -- gate ---------------------------------------------------------- + + def _check_write_gate(self, request: ToolCallRequest) -> ToolMessage | None: + tool_call = request.tool_call + path = self._requested_path(request) + if path is None: + return None + try: + current = self._content_reader(request.runtime, path) + except FileNotFoundError: + # write_file creates the file; str_replace surfaces its own error. + return None + except Exception: + logger.warning("read-before-write gate could not inspect %r; allowing the write (fail-open)", path, exc_info=True) + return None + norm_path = _normalize_mark_path(path) + if self._latest_mark_hash(request.state, norm_path) == _content_hash(current): + return None + tool_name = str(tool_call.get("name", "write")) + return ToolMessage( + content=_BLOCK_MESSAGE.format(tool_name=tool_name, path=path), + tool_call_id=str(tool_call.get("id", "")), + name=tool_name, + status="error", + ) + + @staticmethod + def _requested_path(request: ToolCallRequest) -> str | None: + args = request.tool_call.get("args") or {} + if not isinstance(args, dict): + return None + path = args.get("path") + return path if isinstance(path, str) and path else None + + @staticmethod + def _latest_mark_hash(state: Any, norm_path: str) -> str | None: + messages = state.get("messages") if isinstance(state, dict) else getattr(state, "messages", None) + if not messages: + return None + for message in reversed(messages): + if not isinstance(message, ToolMessage): + continue + mark = (message.additional_kwargs or {}).get(READ_MARK_KEY) + if isinstance(mark, dict) and mark.get("path") == norm_path: + mark_hash = mark.get("hash") + return mark_hash if isinstance(mark_hash, str) else None + return None + + # -- mark stamping --------------------------------------------------- + + def _attach_read_mark(self, request: ToolCallRequest, result: ToolMessage | Command) -> None: + path = self._requested_path(request) + if path is None: + return + message = self._extract_tool_message(result) + if message is None or message.status == "error": + return + try: + content = self._content_reader(request.runtime, path) + except Exception: + logger.debug("read-before-write mark skipped for %r: file not hashable", path, exc_info=True) + return + message.additional_kwargs[READ_MARK_KEY] = { + "path": _normalize_mark_path(path), + "hash": _content_hash(content), + } + + @staticmethod + def _extract_tool_message(result: ToolMessage | Command) -> ToolMessage | None: + if isinstance(result, ToolMessage): + return result + 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] + return None +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py -v` +Expected: PASS (all TestReadMarkStamping + TestWriteGate) + +- [ ] **Step 5: Commit** + +```bash +git add backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py backend/tests/test_read_before_write_middleware.py +git commit -m "feat(middlewares): read-before-write version gate for file tools (#3857)" +``` + +--- + +### Task 3: Async `awrap_tool_call` coverage + +**Files:** +- Modify: `backend/tests/test_read_before_write_middleware.py` (implementation from Task 2 already includes `awrap_tool_call`; this task pins it) + +**Interfaces:** +- Consumes: `ReadBeforeWriteMiddleware` from Task 2. + +- [ ] **Step 1: Write the failing/verifying tests** + +```python +class TestAsyncPaths: + PATH = "/mnt/user-data/outputs/report.md" + + def test_async_block(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + + async def handler(_request): + raise AssertionError("handler must not run when blocked") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.status == "error" + + def test_async_read_stamps_mark(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("read_file", {"description": "d", "path": self.PATH}) + + async def handler(_request): + return ToolMessage(content="v1", tool_call_id="call-1", name="read_file") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.additional_kwargs["deerflow_read_mark"]["hash"] == _sha("v1") + + def test_async_allowed_write_calls_handler(self): + import asyncio + + mw = _middleware({self.PATH: "v1"}) + messages = [_read_marked_message(self.PATH, "v1")] + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}, messages) + + async def handler(_request): + return ToolMessage(content="OK", tool_call_id="call-1", name="write_file") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + assert result.status != "error" +``` + +- [ ] **Step 2: Run tests** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py -v` +Expected: PASS (Task 2 already implemented `awrap_tool_call`; if any fail, fix the middleware, not the tests) + +- [ ] **Step 3: Commit** + +```bash +git add backend/tests/test_read_before_write_middleware.py +git commit -m "test(middlewares): pin async read-before-write gate paths (#3857)" +``` + +--- + +### Task 4: Config + chain wiring + +**Files:** +- Create: `backend/packages/harness/deerflow/config/read_before_write_config.py` +- Modify: `backend/packages/harness/deerflow/config/app_config.py` (imports at top; field after `loop_detection` at ~line 133) +- Modify: `backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py:192-197` (tail layer) +- Modify: `config.example.yaml` (bump `config_version` — 16 → 17 as landed; new section near `loop_detection:` at ~line 836) +- Modify: `backend/tests/test_tool_error_handling_middleware.py:178-218` (chain-order pin test) +- Test: `backend/tests/test_read_before_write_middleware.py` + +**Interfaces:** +- Produces: `AppConfig.read_before_write: ReadBeforeWriteConfig` with `enabled: bool = True`; `ReadBeforeWriteMiddleware` present in `_build_runtime_middlewares` tail between `SandboxAuditMiddleware` and `ToolErrorHandlingMiddleware` when enabled. + +- [ ] **Step 1: Write failing tests** + +Append to `backend/tests/test_read_before_write_middleware.py`: + +```python +class TestChainWiring: + def test_enabled_by_default_in_runtime_chain(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware, build_lead_runtime_middlewares + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + + app_config = AppConfig(sandbox=SandboxConfig(use="deerflow.sandbox.local.local_sandbox_provider:LocalSandboxProvider")) + middlewares = build_lead_runtime_middlewares(app_config=app_config) + types = [type(m) for m in middlewares] + assert ReadBeforeWriteMiddleware in types + assert types.index(SandboxAuditMiddleware) < types.index(ReadBeforeWriteMiddleware) < types.index(ToolErrorHandlingMiddleware) + + def test_disabled_removes_middleware(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares + from deerflow.config.app_config import AppConfig + from deerflow.config.read_before_write_config import ReadBeforeWriteConfig + from deerflow.config.sandbox_config import SandboxConfig + + app_config = AppConfig( + sandbox=SandboxConfig(use="deerflow.sandbox.local.local_sandbox_provider:LocalSandboxProvider"), + read_before_write=ReadBeforeWriteConfig(enabled=False), + ) + middlewares = build_lead_runtime_middlewares(app_config=app_config) + assert ReadBeforeWriteMiddleware not in [type(m) for m in middlewares] + + def test_subagents_get_the_gate_too(self): + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + + app_config = AppConfig(sandbox=SandboxConfig(use="deerflow.sandbox.local.local_sandbox_provider:LocalSandboxProvider")) + middlewares = build_subagent_runtime_middlewares(app_config=app_config) + assert ReadBeforeWriteMiddleware in [type(m) for m in middlewares] +``` + +(If `_make_app_config` in `tests/test_tool_error_handling_middleware.py` builds AppConfig differently, mirror that fixture instead of the inline construction above.) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py::TestChainWiring -v` +Expected: FAIL — no `read_before_write_config` module / middleware missing from chain + +- [ ] **Step 3: Implement config + wiring** + +`backend/packages/harness/deerflow/config/read_before_write_config.py`: + +```python +"""Configuration for the read-before-write file gate middleware (issue #3857).""" + +from pydantic import BaseModel, Field + + +class ReadBeforeWriteConfig(BaseModel): + """Deterministic version gate on file-modifying tools. + + When enabled, ``write_file`` (append or overwrite of an existing file) and + ``str_replace`` are blocked unless the file was read (``read_file``) after + its last modification, forcing the agent to see the file's current state + before changing it. + """ + + enabled: bool = Field( + default=True, + description="Whether to block writes to existing files that were not read at their current version", + ) +``` + +`app_config.py` — add import + field (place after `loop_detection`): + +```python +from deerflow.config.read_before_write_config import ReadBeforeWriteConfig +... + read_before_write: ReadBeforeWriteConfig = Field(default_factory=ReadBeforeWriteConfig, description="Read-before-write file gate middleware configuration") +``` + +`tool_error_handling_middleware.py` — in `_build_runtime_middlewares`, between `tail.append(SandboxAuditMiddleware())` and `tail.append(ToolErrorHandlingMiddleware())`: + +```python + tail.append(SandboxAuditMiddleware()) + + if app_config.read_before_write.enabled: + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + + tail.append(ReadBeforeWriteMiddleware()) + + tail.append(ToolErrorHandlingMiddleware()) +``` + +`config.example.yaml` — bump `config_version` by one (landed as 17); add next to the `loop_detection:` section: + +```yaml +# Read-before-write file gate (issue #3857). +# Blocks write_file (append / overwrite of an existing file) and str_replace +# unless the agent has read the file's current version first; any write +# invalidates earlier reads, forcing a re-read between consecutive edits. +read_before_write: + enabled: true +``` + +`tests/test_tool_error_handling_middleware.py::test_build_lead_runtime_middlewares_chain_order_matches_agents_md` — add to imports and `expected_order` between `SandboxAuditMiddleware` and `ToolErrorHandlingMiddleware`: + +```python + from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware + ... + ("SandboxAuditMiddleware", SandboxAuditMiddleware), + ("ReadBeforeWriteMiddleware", ReadBeforeWriteMiddleware), + ("ToolErrorHandlingMiddleware", ToolErrorHandlingMiddleware), +``` + +- [ ] **Step 4: Run tests** + +Run: `cd backend && PYTHONPATH=. uv run pytest tests/test_read_before_write_middleware.py tests/test_tool_error_handling_middleware.py tests/test_app_config.py -v` (drop `test_app_config.py` if it doesn't exist) +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add backend/packages/harness/deerflow/config/read_before_write_config.py backend/packages/harness/deerflow/config/app_config.py backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py config.example.yaml backend/tests/test_tool_error_handling_middleware.py backend/tests/test_read_before_write_middleware.py +git commit -m "feat(config): wire ReadBeforeWriteMiddleware into runtime chain, default on (#3857)" +``` + +--- + +### Task 5: Tool docstrings, docs, full verification + +**Files:** +- Modify: `backend/packages/harness/deerflow/sandbox/tools.py` (`write_file_tool` docstring ~1765; `str_replace_tool` docstring ~1859) +- Modify: `backend/AGENTS.md` ("Middleware Chain" → Shared runtime base list at ~line 202; "Sandbox Tools" bullet list) +- Modify: `docs/superpowers/specs/2026-07-02-read-before-write-gate-design.md` only if implementation deviated + +- [ ] **Step 1: Update tool docstrings** + +`write_file_tool` docstring — after the first line, add: + +``` + READ-BEFORE-WRITE (issue #3857): if the target file already exists (including + append=True), you must have read its CURRENT version with read_file first. + Any write invalidates earlier reads, so re-read between consecutive + modifications — a ranged read of the relevant section is enough. Writes + that fail this check are rejected with an error. +``` + +`str_replace_tool` docstring — after the first paragraph, add: + +``` + READ-BEFORE-WRITE (issue #3857): you must have read the file's CURRENT + version with read_file first; any write invalidates earlier reads. +``` + +- [ ] **Step 2: Update backend/AGENTS.md** + +In "Shared runtime base" list, insert after **SandboxAuditMiddleware** (renumber the tail): + +``` +11. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Version gate on file writes: `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Marks live on messages, so summarization dropping the read result invalidates the gate automatically (issue #3857) +``` + +In "Sandbox Tools" section, extend the `write_file` / `str_replace` bullets with: "subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain)". + +- [ ] **Step 3: Full backend verification** + +Run: `cd backend && make format && make lint && make test` +Expected: format clean, lint clean, full suite PASS + +- [ ] **Step 4: Commit** + +```bash +git add backend/packages/harness/deerflow/sandbox/tools.py backend/AGENTS.md +git commit -m "docs(sandbox): document read-before-write gate in tool docstrings and AGENTS.md (#3857)" +``` diff --git a/docs/superpowers/specs/2026-07-02-read-before-write-gate-design.md b/docs/superpowers/specs/2026-07-02-read-before-write-gate-design.md new file mode 100644 index 000000000..3b2d8fa2d --- /dev/null +++ b/docs/superpowers/specs/2026-07-02-read-before-write-gate-design.md @@ -0,0 +1,51 @@ +# 产物层：修改已存在文件前强制先读（ReadBeforeWriteMiddleware）— 设计 + +对应 issue #3857（parent）产物层子项。目标：用确定性的"版本闸"逼 agent 在修改已存在文件前看见文件现状，治"只追加、不回读"导致的重复产出（Case 2），并作为通用的文件写入新鲜度护栏。 + +定位（采纳 issue 评论区共识）：本机制是**新鲜度护栏**——保证"agent 读过当前版本"，不保证"最终产物无语义重复"。后者（结构化产物状态、终稿去重校验）是独立后续项。 + +## 行为规格 + +**闸规则**（与 issue 描述一致）： + +1. `read_file` 成功 → 系统记录该路径的 read-mark：`sha256(当前完整文件内容)`。带行号范围/被截断的读同样记 mark（hash 始终基于完整文件内容）。 +2. `write_file(append=True)`、`write_file` 覆盖**已存在**文件、`str_replace` 执行前：校验"该路径最新 read-mark 的 hash == 当前文件 hash"。不通过 → 拦截，不执行写入，返回引导性错误（"请先读该文件"；对 append 场景提示读尾部若干行即可，控制上下文成本）。 +3. 写入**永不**刷新 mark：任何成功写入都会改变文件 hash → 上一次读立即过期 → 连续修改之间被逼重读。这是治 Case 2 的关键强制点。 +4. `write_file` 写不存在的文件（新建）→ 放行，不记 mark（新建后的第一次 append/str_replace 也要求先读）。 +5. `str_replace` 目标文件不存在 → 放行，由工具自身返回 not-found 错误。 + +**read-mark 与上下文存活的绑定**（issue 第三小项）： + +- mark 不落 ThreadState，而是附着在 `read_file` 返回的 `ToolMessage.additional_kwargs["deerflow_read_mark"]` 上（`{path, hash}`）。 +- 闸校验时从 `state["messages"]` 由新到旧扫描该路径的 mark。总结（summarization）删掉该 ToolMessage ⇒ mark 自然消失 ⇒ 闸拦截。"闸通过但内容已被总结删掉"被结构性排除，无需保留区，也无需 summarization hook。 +- 推论：同一轮并行 read+write 同一文件会被拦——此时模型尚未看到读结果，拦截语义正确。 + +**失败语义**： + +- 闸自身读文件/求 hash 出现意外错误（非 FileNotFoundError，如二进制 UnicodeDecodeError、沙箱瞬时故障）→ fail-open 放行并记日志。护栏不应把 agent 砖死。 +- 非本地沙箱（AIO/E2B）的 `read_file` 把读失败（含文件不存在）吞成 `"Error: ..."` 字符串而不抛异常：闸读到以 `Error:` 开头的内容按"无法检视"处理 → fail-open 放行、不打标记。新建文件因此在这类沙箱上正常放行；已存在文件的正常读写不受影响（#3912 review 修复）。 +- 拦截返回 `ToolMessage(status="error")`，措辞引导恢复路径，不暴露后端配置细节。 + +**并发语义（#3912 review 修复）**： + +- LangGraph 会并发执行同一条 AIMessage 里的多个 tool_calls。中间件对每个 (thread, 规范化路径) 持有独立锁，把"闸校验 + 写入执行"以及"读取执行 + 打标"分别放进同一临界区：同轮第二个同路径写必须等第一个完成后再校验（hash 已变 → 确定性拦截）；读的标记保证 hash 的是模型实际看到的那个版本。该锁与工具内部的 `file_operation_lock` 分属不同命名空间，无嵌套获取，不会死锁。 + +## 实现结构 + +- 新文件 `packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py`：`ReadBeforeWriteMiddleware(AgentMiddleware)`，实现 `wrap_tool_call` / `awrap_tool_call`。 + - 读状态由中间件逻辑持有/解释，工具零改动、不持状态（issue 要求）。 + - 当前文件内容读取复用 `deerflow.sandbox.tools` 的路径解析与沙箱读取逻辑（提取一个共享 helper，避免复制 `_resolve_*` 细节）；对 local 与 AIO 沙箱一致生效。 + - mark 的路径键：对 agent 提供的虚拟路径做 `posixpath.normpath` 规范化。 +- 装配位置：`tool_error_handling_middleware.py::_build_runtime_middlewares` 的 `tail` 层、`SandboxAuditMiddleware` 之后 / `ToolErrorHandlingMiddleware` 之前 → lead 与 subagent 共同生效（issue：通用机制）。#3809 的链序 pin 测试同步更新。 +- 配置：新增 `deerflow/config/read_before_write_config.py`（`enabled: bool = True`），挂到 `AppConfig.read_before_write`；`config.example.yaml` 增段并 bump `config_version`。默认开启（issue 第 4 点：护栏要真正生效）。 +- 工具描述：`write_file` / `str_replace` docstring 增补"目标文件已存在时须先读到当前版本，过期写入会被拒绝"的提示，使模型可自解释错误。 + +## 已知边界（记录，不在本项处理） + +- 语义重复仍可能发生（读了现状仍决定再追加同一节）——ShenAC-SAC 评论指出的产物状态/终稿校验属后续项。 +- `bash` 修改文件不走闸；但它改变 hash，会使后续 `write_file`/`str_replace` 被逼重读，方向一致。 +- ~~闸校验与实际写入之间存在极窄的 TOCTOU 窗口（并行工具调用），作为护栏可接受。~~ 该窗口已按 #3912 review 意见通过 per-path 临界区消除（见"并发语义"）——同轮并行重复写正是 Case 2 的变体，不应写掉。 + +## 测试（TDD，`backend/tests/test_read_before_write_middleware.py`） + +覆盖：新建放行；未读改已存在文件（overwrite/append/str_replace）拦截；读后放行；写后 mark 过期、再写拦截、重读后放行；mark 随消息被删（模拟总结）后拦截；范围读也记 mark；hash 失败 fail-open；str_replace 不存在文件放行；`enabled=False` 全透传；同步与异步路径；链序 pin 测试更新。