feat(file): verify write_file content on disk and say so (verified: true)
write_file confirmed only SIZE (wc -c) after writing — never content.
Models compensated by re-reading files immediately after writing them
(154 verify-reads in a 400k-msg production window), and a corrupted
write (truncated pipe, backend FS oddity) could silently pass.
The write path now compares the on-disk sha256 against the intended
content (one shell call). Three outcomes:
- match -> result carries verified: true; the schema tells the model
an explicit contract: do NOT re-read to check the write landed.
- mismatch -> hard error ('The write did not persist correctly'),
mirroring patch_replace's existing post-write verification.
- backend can't hash (no sha256sum) -> flag omitted, write unaffected.
Hashes the shim-adjusted content (after CRLF/BOM preservation) so
Windows-line-ending and BOM round-trips verify correctly; surrogatepass
encoding matches the rest of the codebase's hashing of model text.
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
"""Tests for write_file post-write content verification (verified flag)."""
|
||||
|
||||
import json
|
||||
from unittest.mock import patch as mock_patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.file_tools import write_file_tool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def workdir(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
return tmp_path
|
||||
|
||||
|
||||
class TestWriteVerification:
|
||||
def test_successful_write_reports_verified(self, workdir):
|
||||
f = workdir / "out.txt"
|
||||
r = json.loads(write_file_tool(str(f), "hello verified world\n", task_id="t-wv"))
|
||||
assert r.get("bytes_written") == len("hello verified world\n")
|
||||
assert r.get("verified") is True
|
||||
|
||||
def test_unicode_content_verified(self, workdir):
|
||||
f = workdir / "uni.txt"
|
||||
content = "línea → uno · ✓\n"
|
||||
r = json.loads(write_file_tool(str(f), content, task_id="t-wv"))
|
||||
assert r.get("verified") is True
|
||||
|
||||
def test_crlf_preservation_still_verifies(self, workdir):
|
||||
# Existing CRLF file: write_file converts LF content to CRLF before
|
||||
# writing; verification hashes the shim-adjusted content, so it must
|
||||
# still report verified.
|
||||
f = workdir / "win.txt"
|
||||
f.write_bytes(b"old line\r\n")
|
||||
r = json.loads(write_file_tool(str(f), "new line\nsecond\n", task_id="t-wv"))
|
||||
assert "error" not in r
|
||||
assert r.get("verified") is True
|
||||
assert b"\r\n" in f.read_bytes()
|
||||
|
||||
def test_hash_mismatch_is_hard_error(self, workdir):
|
||||
f = workdir / "bad.txt"
|
||||
import tools.file_operations as fo
|
||||
real_sha = fo.hashlib.sha256
|
||||
|
||||
class _WrongHash:
|
||||
def __init__(self, *a, **k):
|
||||
self._h = real_sha(b"different content entirely")
|
||||
def hexdigest(self):
|
||||
return self._h.hexdigest()
|
||||
|
||||
with mock_patch.object(fo.hashlib, "sha256", _WrongHash):
|
||||
r = json.loads(write_file_tool(str(f), "actual content\n", task_id="t-wv"))
|
||||
assert "error" in r
|
||||
assert "did not persist" in r["error"]
|
||||
|
||||
def test_verification_failure_never_breaks_write(self, workdir):
|
||||
# sha256sum unavailable/failing -> verified omitted, write still ok.
|
||||
f = workdir / "ok.txt"
|
||||
import tools.file_operations as fo
|
||||
|
||||
real_exec = fo.ShellFileOperations._exec
|
||||
|
||||
def flaky_exec(self, cmd, **kw):
|
||||
if "sha256sum" in cmd:
|
||||
raise RuntimeError("no hash binary")
|
||||
return real_exec(self, cmd, **kw)
|
||||
|
||||
with mock_patch.object(fo.ShellFileOperations, "_exec", flaky_exec):
|
||||
r = json.loads(write_file_tool(str(f), "content lands anyway\n", task_id="t-wv2"))
|
||||
assert "error" not in r
|
||||
assert f.read_text() == "content lands anyway\n"
|
||||
assert "verified" not in r or r.get("verified") is None
|
||||
@@ -28,6 +28,7 @@ Usage:
|
||||
import os
|
||||
import re
|
||||
import difflib
|
||||
import hashlib
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional, List, Dict, Any, ClassVar
|
||||
@@ -178,6 +179,11 @@ class WriteResult:
|
||||
"""Result from writing a file."""
|
||||
bytes_written: int = 0
|
||||
dirs_created: bool = False
|
||||
# True when the on-disk sha256 matched the intended content after the
|
||||
# write (post-write verification). None when the backend couldn't
|
||||
# verify (no sha256sum). A mismatch never reaches the caller as a
|
||||
# flag — it becomes a hard error.
|
||||
verified: Optional[bool] = None
|
||||
lint: Optional[Dict[str, Any]] = None
|
||||
# Semantic diagnostics from the LSP layer, when applicable. Kept in
|
||||
# its own field (not folded into ``lint``) so the model and any
|
||||
@@ -1567,6 +1573,33 @@ class ShellFileOperations(FileOperations):
|
||||
except ValueError:
|
||||
bytes_written = len(content.encode('utf-8'))
|
||||
|
||||
# Post-write content verification (cheap, one shell call): compare
|
||||
# the on-disk sha256 to the intended content's hash. Production
|
||||
# mining shows models re-reading files right after writing them to
|
||||
# confirm persistence (154 verify-reads in a 400k-msg window) —
|
||||
# an explicit verified flag makes that turn unnecessary, and a
|
||||
# mismatch is surfaced as a hard error instead of silent corruption
|
||||
# (mirrors patch_replace's post-write verification).
|
||||
content_verified: Optional[bool] = None
|
||||
try:
|
||||
hash_cmd = f"sha256sum {self._escape_shell_arg(path)} 2>/dev/null"
|
||||
hash_result = self._exec(hash_cmd)
|
||||
if hash_result.exit_code == 0 and hash_result.stdout.strip():
|
||||
disk_sha = hash_result.stdout.strip().split()[0]
|
||||
expected_sha = hashlib.sha256(content.encode("utf-8", "surrogatepass")).hexdigest()
|
||||
content_verified = disk_sha == expected_sha
|
||||
if not content_verified:
|
||||
return WriteResult(
|
||||
error=(
|
||||
f"Post-write verification failed for {path}: on-disk "
|
||||
"content hash differs from the intended write. The "
|
||||
"write did not persist correctly — re-read the file "
|
||||
"and retry."
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
content_verified = None
|
||||
|
||||
# Post-write lint with delta refinement.
|
||||
lint_result = self._check_lint_delta(path, pre_content=pre_content, post_content=content)
|
||||
|
||||
@@ -1587,6 +1620,7 @@ class ShellFileOperations(FileOperations):
|
||||
return WriteResult(
|
||||
bytes_written=bytes_written,
|
||||
dirs_created=dirs_created,
|
||||
verified=content_verified,
|
||||
lint=lint_result.to_dict() if lint_result else None,
|
||||
lsp_diagnostics=lsp_diagnostics,
|
||||
)
|
||||
|
||||
+1
-1
@@ -2172,7 +2172,7 @@ READ_FILE_SCHEMA = {
|
||||
|
||||
WRITE_FILE_SCHEMA = {
|
||||
"name": "write_file",
|
||||
"description": "Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by this write are surfaced (pre-existing errors are filtered out).",
|
||||
"description": "Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by this write are surfaced (pre-existing errors are filtered out). The result's verified:true means the on-disk content hash was confirmed — do NOT re-read the file to check the write landed.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
||||
Reference in New Issue
Block a user