fix(tools): close empty-write races and preserve clears

src/agent_tools/filesystem_tools.py

@@ -345,30 +345,45 @@ class WriteFileTool:
                         old = f.read()
                 except (FileNotFoundError, IsADirectoryError, UnicodeDecodeError, OSError):
                     old = ""
-                if not body.strip() and not declared_clear:
-                    # Why size on disk rather than `old`: the read above answers ""
-                    # for a file it cannot decode, so a non-UTF-8 target holding real
-                    # bytes looks empty through `old` and would still be truncated.
-                    # Why in this position: open(path, "w") truncates on entry, so a
-                    # check after the write has nothing left to protect.
-                    existing_bytes = os.path.getsize(path) if os.path.isfile(path) else 0
-                    if existing_bytes > 0:
-                        raise _EmptyBodyWouldTruncate(path, existing_bytes)
                 d = os.path.dirname(path)
                 if d:
                     os.makedirs(d, exist_ok=True)
+                if not body.strip() and not declared_clear:
+                    # Empty/whitespace-only writes to an already-empty file are no-ops.
+                    # Avoid reopening in truncating mode: another writer may have added
+                    # data since the read above.
+                    if os.path.isfile(path):
+                        existing_bytes = os.path.getsize(path)
+                        if existing_bytes > 0:
+                            raise _EmptyBodyWouldTruncate(path, existing_bytes)
+                        return old, 0
+
+                    # Create a missing target exclusively. If another writer wins
+                    # the race, inspect what appeared rather than truncating it.
+                    try:
+                        with open(path, "x", encoding="utf-8") as f:
+                            f.write(body)
+                    except FileExistsError:
+                        if os.path.isfile(path):
+                            existing_bytes = os.path.getsize(path)
+                            if existing_bytes > 0:
+                                raise _EmptyBodyWouldTruncate(path, existing_bytes)
+                            return old, 0
+                        raise
+                    return old, len(body)
+
                 with open(path, "w", encoding="utf-8") as f:
                     f.write(body)
                 return old, len(body)
             old_content, size = await asyncio.to_thread(_write)
         except _EmptyBodyWouldTruncate as e:
+            clear_call = json.dumps({"path": raw_path, "content": ""})
             return {
                 "error": (
                     f"write_file: refused to write an empty body over {e.path} — it holds "
                     f"{e.existing_bytes} bytes, which the write would have destroyed, so "
                     f"the file is unchanged. To clear it on purpose, resend with an "
-                    f"explicit empty content: "
-                    f'{{"path": "{raw_path}", "content": ""}}'
+                    f"explicit empty content: {clear_call}"
                 ),
                 "exit_code": 1,
             }

src/tool_parsing.py

@@ -664,7 +664,10 @@ def _raw_openai_tool_call_to_block(value) -> Optional[ToolBlock]:
     elif tool_type in ("grep", "glob", "ls", "edit_file"):
         content = json.dumps(args) if args else "{}"
     elif tool_type == "write_file":
-        content = args.get("path", "") + "\n" + args.get("content", "")
+        # Keep raw OpenAI JSON on the canonical path so explicit empty content
+        # remains distinguishable from a missing body.
+        from src.tool_schemas import function_call_to_tool_block
+        return function_call_to_tool_block(name, json.dumps(args))
     elif tool_type == "create_document":
         parts = [args.get("title", "Untitled")]
         if args.get("language"):

src/tool_schemas.py

@@ -1442,7 +1442,13 @@ def function_call_to_tool_block(name: str, arguments: str) -> Optional[ToolBlock
     elif tool_type == "get_workspace":
         content = ""
     elif tool_type == "write_file":
-        content = args.get("path", "") + "\n" + args.get("content", "")
+        body = args.get("content")
+        if isinstance(body, str) and body.strip():
+            content = args.get("path", "") + "\n" + body
+        else:
+            # Preserve missing/empty intent for WriteFileTool instead of folding
+            # all empty shapes into the same path-plus-newline representation.
+            content = json.dumps(args)
     elif tool_type == "edit_file":
         content = json.dumps(args)
     elif tool_type == "apply_patch":

tests/test_write_file_empty_body.py

@@ -17,6 +17,8 @@ import pytest
 from src import tool_execution as te
 from src.agent_tools import ToolBlock
 from src.agent_tools.filesystem_tools import EditFileTool, WriteFileTool
+from src.tool_schemas import function_call_to_tool_block
+from src.tool_parsing import parse_tool_blocks
 
 RECIPE = "# Classic banana cake\n\nMash 3 bananas. Bake 180C for 1 hour.\n"
 
@@ -127,6 +129,22 @@ async def test_refusal_names_the_byte_count_and_the_explicit_form(target):
     assert "output" not in res, res
 
 
+@pytest.mark.asyncio
+async def test_refusal_suggestion_is_valid_json_for_paths_with_quotes(target):
+    quoted_path = os.path.join(os.path.dirname(target), 'recipe"draft.md')
+    _seed(quoted_path)
+    refused = await WriteFileTool().execute(_text_call(quoted_path, ""), {})
+    match = re.search(
+        r"explicit empty content: (\{.*\})$", refused["error"], re.S
+    )
+    assert match, refused
+    suggested_args = json.loads(match.group(1))
+    assert suggested_args == {"path": quoted_path, "content": ""}
+    cleared = await WriteFileTool().execute(match.group(1), {})
+    assert cleared["exit_code"] == 0, cleared
+    assert os.path.getsize(quoted_path) == 0
+
+
 @pytest.mark.asyncio
 async def test_the_resend_the_refusal_prints_actually_clears_the_file(target):
     """The guidance is only useful if a caller can paste it back verbatim. This reads
@@ -158,6 +176,27 @@ async def test_empty_body_on_a_new_path_still_creates_an_empty_file(target):
     assert os.path.isfile(target) and os.path.getsize(target) == 0
 
 
+@pytest.mark.asyncio
+async def test_whitespace_only_body_on_a_new_path_preserves_the_requested_content(
+    target,
+):
+    whitespace = "   \n\t"
+    res = await WriteFileTool().execute(_text_call(target, whitespace), {})
+    assert res["exit_code"] == 0, res
+    assert _read(target) == whitespace
+
+
+@pytest.mark.asyncio
+async def test_explicit_whitespace_only_json_content_preserves_the_requested_content(
+    target,
+):
+    _seed(target)
+    whitespace = " \t "
+    res = await WriteFileTool().execute(_json_call(target, content=whitespace), {})
+    assert res["exit_code"] == 0, res
+    assert _read(target) == whitespace
+
+
 @pytest.mark.asyncio
 async def test_empty_body_over_an_already_empty_file_succeeds(target):
     """Nothing is at risk, so the guard has nothing to refuse."""
@@ -167,6 +206,48 @@ async def test_empty_body_over_an_already_empty_file_succeeds(target):
     assert os.path.getsize(target) == 0
 
 
+@pytest.mark.asyncio
+async def test_empty_body_does_not_truncate_data_written_after_the_size_check(
+    target, monkeypatch
+):
+    """A write racing the size check must survive the empty-body path."""
+    _seed(target, "")
+    real_getsize = os.path.getsize
+
+    def write_after_size_check(path):
+        size = real_getsize(path)
+        if path == target and size == 0:
+            with open(path, "w", encoding="utf-8") as handle:
+                handle.write("concurrent update")
+        return size
+
+    monkeypatch.setattr(os.path, "getsize", write_after_size_check)
+    res = await WriteFileTool().execute(_text_call(target, ""), {})
+    assert res["exit_code"] == 0, res
+    assert _read(target) == "concurrent update"
+
+
+@pytest.mark.asyncio
+async def test_empty_body_does_not_truncate_a_file_created_after_the_absence_check(
+    target, monkeypatch
+):
+    """Exclusive creation must not overwrite a file that appeared during the check."""
+    real_isfile = os.path.isfile
+
+    def create_after_absence_check(path):
+        exists = real_isfile(path)
+        if path == target and not exists:
+            with open(path, "w", encoding="utf-8") as handle:
+                handle.write("concurrent update")
+            return False
+        return exists
+
+    monkeypatch.setattr(os.path, "isfile", create_after_absence_check)
+    res = await WriteFileTool().execute(_text_call(target, ""), {})
+    assert res["exit_code"] == 1, res
+    assert _read(target) == "concurrent update"
+
+
 @pytest.mark.asyncio
 async def test_a_real_body_still_writes_and_reports_a_diff(target):
     _seed(target)
@@ -189,6 +270,38 @@ async def test_edit_file_remains_an_explicit_way_to_clear_a_file(target):
     assert os.path.getsize(target) == 0
 
 
+@pytest.mark.asyncio
+async def test_native_function_call_can_explicitly_clear_a_file(target):
+    """The native schema conversion must preserve the explicit empty-content intent."""
+    _seed(target)
+    block = function_call_to_tool_block(
+        "write_file", json.dumps({"path": target, "content": ""})
+    )
+    assert block is not None
+    res = await WriteFileTool().execute(block.content, {})
+    assert res["exit_code"] == 0, res
+    assert os.path.getsize(target) == 0
+
+
+@pytest.mark.asyncio
+async def test_raw_openai_function_call_can_explicitly_clear_a_file(target):
+    """The raw OpenAI JSON parser must retain explicit empty-content intent too."""
+    _seed(target)
+    arguments = json.dumps({"path": target, "content": ""})
+    raw_call = json.dumps(
+        {
+            "type": "function",
+            "function": {"name": "write_file", "arguments": arguments},
+        }
+    )
+    blocks = parse_tool_blocks(raw_call)
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "write_file"
+    res = await WriteFileTool().execute(blocks[0].content, {})
+    assert res["exit_code"] == 0, res
+    assert os.path.getsize(target) == 0
+
+
 # ── The live dispatch path, not just the handler ──────────────────────────
 @pytest.mark.asyncio
 async def test_execute_tool_block_refuses_a_lost_body_without_touching_the_file(target, monkeypatch):