core/tools+config: refuse write_file overwrite unless overwrite=true

src/plyngent/config/models.py

@@ -46,6 +46,8 @@ DEFAULT_TOOL_DIRECTIVES = """\
 ### Files
 - Prefer file tools over shell. `edit_replace` fails usually means a bad match — \
 fix `old_string` or `max_replaces`; shell workarounds rarely help.
+- `write_file` refuses to replace an existing file (read it, or change it with \
+`edit_replace` / `edit_lineno`); a full replacement needs `overwrite=true`, which asks the human.
 - `edit_replace` defaults to the first match; if remaining matches are reported, \
 raise `max_replaces` or narrow `old_string`.
 - `edit_lineno` edits only lines you read via `read_file(path, with_lineno=true)`; \

src/plyngent/tools/danger.py

@@ -87,7 +87,13 @@ def _command_reason(argv: Sequence[str], *, via: str) -> str | None:
 
 
 def _write_file_reason(args: Mapping[str, object]) -> str | None:
-    """Confirm only when write_file would replace an existing file."""
+    """Confirm only when write_file was asked to replace an existing file.
+
+    An unset/``false`` ``overwrite`` never reaches this prompt: the tool itself
+    refuses an existing file, so there is nothing to ask about.
+    """
+    if not bool(args.get("overwrite", False)):
+        return None
     path = args.get("path")
     if not isinstance(path, str) or not path:
         return None

src/plyngent/tools/file/write.py

@@ -5,9 +5,15 @@ from plyngent.tools.workspace import AccessMode, resolve_path
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
-async def write_file(path: str, content: str) -> str:
+async def write_file(path: str, content: str, *, overwrite: bool = False) -> str:
-    """Write text content to a file under the workspace (creates parents)."""
+    """Write text content to a file under the workspace (creates parents).
+
+    Refuses to replace an existing file: pass ``overwrite=true`` to replace it
+    (the replacement is then confirmed by the human).
+    """
     target = resolve_path(path, required=AccessMode.WRITE)
+    if target.is_file() and not overwrite:
+        return f"error: file exists: {path} (set overwrite=true to replace it)"
     target.parent.mkdir(parents=True, exist_ok=True)
     _ = target.write_text(content, encoding="utf-8")
     # The write changes the file's mtime, so edit_lineno's freshness check

tests/test_tools/test_danger.py

@@ -24,14 +24,18 @@ def test_classify_write_overwrite_only(tmp_path: Path) -> None:
     instance = InstanceState(workspace_root=tmp_path.resolve())
     with bind_instance(instance):
         _ = set_workspace_root(tmp_path)
-        # New file: no soft-confirm
+        # New file: no soft-confirm, with or without the overwrite flag
         assert classify_danger("write_file", {"path": "new.txt"}) is None
+        assert classify_danger("write_file", {"path": "new.txt", "overwrite": True}) is None
         # Partial edits: never soft-confirm
         assert classify_danger("edit_replace", {"path": "x.txt"}) is None
         assert classify_danger("edit_lineno", {"path": "x.txt", "start_line": 1, "end_line": 2}) is None
-        # Existing file write: confirm total overwrite
+        # Existing file: only an explicit overwrite asks; an unset/false flag is
+        # refused inside the tool, so no prompt may fire for it.
         (tmp_path / "x.txt").write_text("old", encoding="utf-8")
-        reason = classify_danger("write_file", {"path": "x.txt"})
+        assert classify_danger("write_file", {"path": "x.txt"}) is None
+        assert classify_danger("write_file", {"path": "x.txt", "overwrite": False}) is None
+        reason = classify_danger("write_file", {"path": "x.txt", "overwrite": True})
         assert reason is not None and "overwrite" in reason
         (tmp_path / "dst.txt").write_text("d", encoding="utf-8")
         (tmp_path / "src.txt").write_text("s", encoding="utf-8")

tests/test_tools/test_edit_lineno.py

@@ -63,7 +63,7 @@ def test_edit_lineno_error_suggests_the_read_file_call(workspace: object) -> Non
     out = call_sync(edit_lineno, "s.txt", 15, 15, "X\n")
     assert "read_file(path='s.txt', offset=11, limit=7, with_lineno=true)" in out
     # A stale read gets a fresh call for the same range after an external write.
-    _ = call_sync(write_file, "s.txt", "changed\n")
+    _ = call_sync(write_file, "s.txt", "changed\n", overwrite=True)
     out = call_sync(edit_lineno, "s.txt", 1, 1, "X\n")
     assert "changed since it was read" in out
     assert "read_file(path='s.txt', offset=0, limit=1, with_lineno=true)" in out
@@ -149,7 +149,7 @@ def test_edit_lineno_requires_rerun_after_write(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "f.txt", "one\ntwo\nthree\n")
     _ = _read_lineno("f.txt")
-    _ = call_sync(write_file, "f.txt", "changed\n")
+    _ = call_sync(write_file, "f.txt", "changed\n", overwrite=True)
     out = call_sync(edit_lineno, "f.txt", 1, 1, "x\n")
     assert "changed" in out and "re-read" in out
 
@@ -182,7 +182,7 @@ def test_edit_lineno_detects_external_change(workspace: object) -> None:
     _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\n")
     _ = _read_lineno("i.txt")
     # Simulate an external/process-tool mutation: rewrite content + new mtime.
-    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\nCHANGED\n")
+    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\nCHANGED\n", overwrite=True)
     os.utime(workspace / "i.txt", (time.time() + 1, time.time() + 1))
     out = call_sync(edit_lineno, "i.txt", 1, 1, "x\n")
     assert "changed since it was read" in out

tests/test_tools/test_file.py

@@ -28,6 +28,15 @@ def test_write_read_listdir_edit(workspace: object) -> None:
     assert call_sync(read_file, "notes/a.txt") == "L1-1\nhello there"
 
 
+def test_write_file_refuses_existing_file_without_overwrite(workspace: Path) -> None:
+    assert "wrote" in call_sync(write_file, "keep.txt", "first")
+    refused = call_sync(write_file, "keep.txt", "second")
+    assert refused == "error: file exists: keep.txt (set overwrite=true to replace it)"
+    assert (workspace / "keep.txt").read_text(encoding="utf-8") == "first"
+    assert "wrote" in call_sync(write_file, "keep.txt", "second", overwrite=True)
+    assert (workspace / "keep.txt").read_text(encoding="utf-8") == "second"
+
+
 def test_read_file_max_chars_embeds_token(workspace: Path) -> None:
     (workspace / "big.txt").write_text("a" * 1000, encoding="utf-8")
     out = call_sync(read_file, "big.txt", max_chars=200)