ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity
Files
main
hermes-agent/tests/agent/test_file_mutation_verifier.py
T
teknium1 ee49b7d25d fix(agent): file-mutation footer states failed edits, not "files were NOT modified" 
The turn-end file-mutation verifier only sees write_file/patch receipts. It
asserted "N file(s) were NOT modified this turn" whenever a call had failed,
which is wrong when the file was in fact changed afterwards through a path
that leaves no receipt (terminal redirect, execute_code) or when the
successful retry used another spelling of the same path (relative vs
absolute, separator/case variants on Windows): the state dict was keyed on
the model's raw `path` argument, so the pop never matched.

- Header now says what the recorder knows: "N file edit(s) FAILED this turn",
  and asks the user to confirm what actually landed.
- Failure entries carry the task-resolved, normcase'd on-disk identity plus a
  (mtime_ns, size) snapshot; a later success clears every entry with the same
  identity regardless of spelling.
- At turn end `_file_mutations_still_failed` re-stats each target and drops
  entries whose file changed since the failed call, so a receipt-less
  mutation no longer produces a false footer.
- `tool_executor` passes the effective task id so relative paths resolve the
  way the file tools resolved them.

Kept the deliberate first-error-per-path semantics (the pinned test says why);
did not add an "unverified" bucket for receipt-less non-error results, since
the built-in tools always return a receipt on success and it would only add
noise.

Co-authored-by: KoNit. <124019182+KoNit-K@users.noreply.github.com>
2026-09-15 18:22:12 -07:00

424 lines
17 KiB
Python
Raw Permalink Blame History

"""Tests for the per-turn file-mutation verifier footer.
Covers the three moving pieces:
1. ``_extract_file_mutation_targets`` — pulls file paths from write_file /
patch (replace + V4A) tool-call argument dicts.
2. ``AIAgent._record_file_mutation_result`` — builds the per-turn state
dict, removing entries when a later success supersedes an earlier
failure for the same path.
3. ``AIAgent._format_file_mutation_failure_footer`` — renders the dict
as a user-visible advisory.
Regression target: the "Ben Eng llm-wiki" session where grok-4.1-fast
batched parallel patches, half failed, and the model summarised the
turn claiming every file was edited. This verifier makes over-claiming
structurally impossible past the model: the user always sees the real
list of files that did NOT change.
"""
from __future__ import annotations
import json
import pytest
from agent.tool_dispatch_helpers import (
_FILE_MUTATING_TOOLS,
_extract_error_preview,
_extract_file_mutation_targets,
_extract_landed_file_mutation_paths,
)
from run_agent import AIAgent
# ---------------------------------------------------------------------------
# _extract_file_mutation_targets
# ---------------------------------------------------------------------------
class TestExtractFileMutationTargets:
def test_non_mutating_tool_returns_empty(self):
assert _extract_file_mutation_targets("read_file", {"path": "/x"}) == []
assert _extract_file_mutation_targets("terminal", {"command": "ls"}) == []
def test_patch_replace_mode_returns_path(self):
args = {"mode": "replace", "path": "/tmp/a.md", "old_string": "x", "new_string": "y"}
assert _extract_file_mutation_targets("patch", args) == ["/tmp/a.md"]
def test_patch_v4a_multi_file(self):
body = (
"*** Begin Patch\n"
"*** Update File: /tmp/a.md\n"
"@@ @@\n-a\n+b\n"
"*** Add File: /tmp/new.md\n"
"+fresh\n"
"*** Delete File: /tmp/old.md\n"
"*** End Patch\n"
)
args = {"mode": "patch", "patch": body}
paths = _extract_file_mutation_targets("patch", args)
assert paths == ["/tmp/a.md", "/tmp/new.md", "/tmp/old.md"]
def test_patch_v4a_accepts_no_space_after_asterisks(self):
"""Match patch_parser / file_tools: ``***Update File:`` (no space)."""
body = "***Update File: nospace.py\n"
assert _extract_file_mutation_targets(
"patch", {"mode": "patch", "patch": body}
) == ["nospace.py"]
# ---------------------------------------------------------------------------
# _extract_error_preview
# ---------------------------------------------------------------------------
class TestExtractErrorPreview:
def test_json_error_field_preferred(self):
raw = json.dumps({"success": False, "error": "Could not find old_string in /tmp/x"})
assert _extract_error_preview(raw) == "Could not find old_string in /tmp/x"
def test_plain_string_falls_through(self):
assert _extract_error_preview("Error executing tool: boom") == "Error executing tool: boom"
def test_long_preview_truncated(self):
long = "x" * 500
out = _extract_error_preview(long, max_len=50)
assert len(out) <= 50
assert out.endswith("…")
# ---------------------------------------------------------------------------
# _record_file_mutation_result — state transitions
# ---------------------------------------------------------------------------
def _bare_agent() -> AIAgent:
"""Skip __init__ and only attach the per-turn state dict.
AIAgent.__init__ takes ~60 parameters and touches network, auth, and
the filesystem. For these tests we only need the two methods —
``_record_file_mutation_result`` and ``_format_file_mutation_failure_footer``.
Using ``object.__new__`` mirrors the gateway-test pattern documented in
the agent pitfalls list.
"""
agent = object.__new__(AIAgent)
agent._turn_failed_file_mutations = {}
agent._turn_file_mutation_paths = set()
return agent
class TestRecordFileMutationResult:
def test_non_mutating_tool_ignored(self):
agent = _bare_agent()
agent._record_file_mutation_result(
"read_file", {"path": "/tmp/x"}, "{}", is_error=True,
)
assert agent._turn_failed_file_mutations == {}
def test_failure_recorded(self):
agent = _bare_agent()
result = json.dumps({"success": False, "error": "Could not find old_string"})
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "/tmp/a.md", "old_string": "x", "new_string": "y"},
result, is_error=True,
)
state = agent._turn_failed_file_mutations
assert "/tmp/a.md" in state
assert state["/tmp/a.md"]["tool"] == "patch"
assert "Could not find old_string" in state["/tmp/a.md"]["error_preview"]
def test_success_removes_prior_failure(self):
agent = _bare_agent()
# First attempt fails
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "/tmp/a.md", "old_string": "x", "new_string": "y"},
json.dumps({"error": "not found"}), is_error=True,
)
assert "/tmp/a.md" in agent._turn_failed_file_mutations
# Second attempt with corrected old_string succeeds
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "/tmp/a.md", "old_string": "real", "new_string": "fixed"},
json.dumps({"success": True, "diff": "..."}), is_error=False,
)
assert agent._turn_failed_file_mutations == {}
assert agent._turn_file_mutation_paths == {"/tmp/a.md"}
def test_landed_paths_prefer_resolved_tool_result(self):
paths = _extract_landed_file_mutation_paths(
"patch",
{"mode": "replace", "path": "src/app.py"},
json.dumps({
"success": True,
"files_modified": ["/tmp/project/src/app.py"],
}),
)
assert paths == ["/tmp/project/src/app.py"]
def test_write_file_with_lint_error_counts_as_landed(self):
agent = _bare_agent()
agent._record_file_mutation_result(
"write_file",
{"path": "/tmp/a.py", "content": "bad"},
json.dumps({"error": "write failed"}),
is_error=True,
)
assert "/tmp/a.py" in agent._turn_failed_file_mutations
result = json.dumps({
"bytes_written": 24,
"lint": {"status": "error", "output": "SyntaxError: invalid syntax"},
})
agent._record_file_mutation_result(
"write_file",
{"path": "/tmp/a.py", "content": "def nope(:\n"},
result,
is_error=True,
)
assert agent._turn_failed_file_mutations == {}
def test_patch_with_lsp_diagnostics_counts_as_landed(self):
agent = _bare_agent()
agent._record_file_mutation_result(
"patch",
{"mode": "replace", "path": "/tmp/a.py", "old_string": "x", "new_string": "y"},
json.dumps({"error": "Could not find old_string"}),
is_error=True,
)
assert "/tmp/a.py" in agent._turn_failed_file_mutations
result = json.dumps({
"success": True,
"diff": "--- a/tmp.py\n+++ b/tmp.py\n",
"files_modified": ["/tmp/a.py"],
"lsp_diagnostics": "<diagnostics>ERROR [1:1] type mismatch</diagnostics>",
})
agent._record_file_mutation_result(
"patch",
{"mode": "replace", "path": "/tmp/a.py", "old_string": "x", "new_string": "y"},
result,
is_error=True,
)
assert agent._turn_failed_file_mutations == {}
def test_repeated_failure_keeps_first_error(self):
agent = _bare_agent()
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "/tmp/a.md", "old_string": "v1", "new_string": "y"},
json.dumps({"error": "first error"}), is_error=True,
)
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "/tmp/a.md", "old_string": "v2", "new_string": "y"},
json.dumps({"error": "second error"}), is_error=True,
)
# Keep the original error — swapping to the latest would obscure
# the initial root cause.
assert "first error" in agent._turn_failed_file_mutations["/tmp/a.md"]["error_preview"]
def test_success_under_another_spelling_clears_failure(self, tmp_path, monkeypatch):
"""A failure recorded as a relative path is cleared by the write that lands on the
resolved absolute path: entries are matched on the on-disk target, not the model's
spelling (#111771)."""
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
task = "fmv-spelling-task"
agent = _bare_agent()
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": "notes.md", "old_string": "x", "new_string": "y"},
json.dumps({"error": "Could not find old_string"}), is_error=True, task_id=task,
)
assert "notes.md" in agent._turn_failed_file_mutations
landed = str(tmp_path / "notes.md")
agent._record_file_mutation_result(
"write_file", {"path": landed, "content": "y"},
json.dumps({"bytes_written": 1, "files_modified": [landed]}), is_error=False, task_id=task,
)
assert agent._turn_failed_file_mutations == {}
def test_footer_omits_file_changed_after_failed_call(self, tmp_path):
"""A file modified after the failed call (terminal redirect, execute_code — no receipt)
is not reported at turn end; an untouched one still is (#111771)."""
changed, untouched = tmp_path / "changed.txt", tmp_path / "untouched.txt"
changed.write_text("v1")
untouched.write_text("c")
agent = _bare_agent()
for target in (changed, untouched):
agent._record_file_mutation_result(
"patch", {"mode": "replace", "path": str(target), "old_string": "x", "new_string": "y"},
json.dumps({"error": "Could not find old_string"}), is_error=True,
)
import os
st = changed.stat()
changed.write_text("v2 rewritten")
os.utime(changed, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000_000))
still_failed = agent._file_mutations_still_failed(agent._turn_failed_file_mutations)
assert list(still_failed) == [str(untouched)]
# ---------------------------------------------------------------------------
# _format_file_mutation_failure_footer
# ---------------------------------------------------------------------------
class TestFormatFooter:
def test_empty_returns_empty_string(self):
assert AIAgent._format_file_mutation_failure_footer({}) == ""
def test_single_failure(self):
out = AIAgent._format_file_mutation_failure_footer(
{"/tmp/a.md": {"tool": "patch", "error_preview": "Could not find old_string"}},
)
# The recorder only sees tool receipts, so the header states what it knows (the
# call failed), never that no bytes changed (#111771).
assert "1 file edit(s) FAILED" in out
assert "NOT modified" not in out
assert "/tmp/a.md" in out
assert "Could not find old_string" in out
assert "git status" in out # user-actionable hint
def test_truncation_at_10_entries(self):
failed = {
f"/tmp/f{i}.md": {"tool": "patch", "error_preview": "err"}
for i in range(15)
}
out = AIAgent._format_file_mutation_failure_footer(failed)
assert "15 file edit(s) FAILED" in out
assert "… and 5 more" in out
# Ten file bullets + header + "and X more" line
lines = out.split("\n")
bullet_lines = [ln for ln in lines if ln.lstrip().startswith("•")]
assert len(bullet_lines) == 11 # 10 shown + 1 summary
def test_footer_path_not_extracted_by_gateway(self):
"""End-to-end: the gateway's extract_local_files must NOT pull a
config.yaml path out of the rendered footer (#35584)."""
import os
import tempfile
from gateway.platforms.base import BasePlatformAdapter
tmp = tempfile.mkdtemp(prefix="hermes_footer_")
try:
cfg = os.path.join(tmp, "config.yaml")
with open(cfg, "w") as fh:
fh.write("openrouter_api_key: sk-LEAK\n")
footer = AIAgent._format_file_mutation_failure_footer(
{cfg: {
"tool": "patch",
"error_preview": (
f"Write denied: '{cfg}' is a protected "
"system/credential file."
),
}},
)
response = "I updated your config.\n\n" + footer
paths, _ = BasePlatformAdapter.extract_local_files(response)
assert paths == [], f"footer leaked deliverable path(s): {paths}"
finally:
import shutil
shutil.rmtree(tmp, ignore_errors=True)
# ---------------------------------------------------------------------------
# _file_mutation_verifier_enabled — env + config precedence
# ---------------------------------------------------------------------------
class TestVerifierEnabled:
def test_default_is_enabled(self, monkeypatch):
monkeypatch.delenv("HERMES_FILE_MUTATION_VERIFIER", raising=False)
agent = _bare_agent()
# With no env and no config present, safe default is True.
# load_config may surface a user config.yaml in some envs — stub it.
import hermes_cli.config as _cfg_mod
monkeypatch.setattr(_cfg_mod, "load_config", lambda: {})
assert agent._file_mutation_verifier_enabled() is True
@pytest.mark.parametrize("value", ["0", "false", "FALSE", "no", "off"])
def test_env_disables(self, monkeypatch, value):
monkeypatch.setenv("HERMES_FILE_MUTATION_VERIFIER", value)
agent = _bare_agent()
assert agent._file_mutation_verifier_enabled() is False
def test_config_read_once_then_cached(self, monkeypatch):
"""Measured-work pin: the config lookup happens once per agent.
The footer gate runs at the end of every turn, so a fresh
``load_config()`` per call is wasted work (measured ~0.9 ms/call on
a warm mtime-cache on this host; the sibling per-turn-config kill in
#74211 removed exactly this class of read). The config read must be
cached after the first call; the env-var override must still win on
every call, cached or not.
"""
monkeypatch.delenv("HERMES_FILE_MUTATION_VERIFIER", raising=False)
agent = _bare_agent()
calls = {"n": 0}
import hermes_cli.config as _cfg_mod
def counting_load():
calls["n"] += 1
return {"display": {"file_mutation_verifier": True}}
monkeypatch.setattr(_cfg_mod, "load_config", counting_load)
# First call reads config and caches the result.
assert agent._file_mutation_verifier_enabled() is True
assert calls["n"] == 1
# Subsequent calls must not re-read config.
assert agent._file_mutation_verifier_enabled() is True
assert agent._file_mutation_verifier_enabled() is True
assert calls["n"] == 1
# Env override stays authoritative even after the cache is warm.
monkeypatch.setenv("HERMES_FILE_MUTATION_VERIFIER", "0")
assert agent._file_mutation_verifier_enabled() is False
assert calls["n"] == 1 # env path never touches config
def test_cache_respects_config_value(self, monkeypatch):
"""A disabled config value is cached as False, not re-read."""
monkeypatch.delenv("HERMES_FILE_MUTATION_VERIFIER", raising=False)
agent = _bare_agent()
import hermes_cli.config as _cfg_mod
monkeypatch.setattr(
_cfg_mod, "load_config", lambda: {"display": {"file_mutation_verifier": False}}
)
assert agent._file_mutation_verifier_enabled() is False
# Warm cache: flip the underlying config; the agent still reports the
# cached value (same next-session semantics as _credits_notices_enabled).
monkeypatch.setattr(
_cfg_mod, "load_config", lambda: {"display": {"file_mutation_verifier": True}}
)
assert agent._file_mutation_verifier_enabled() is False
# ---------------------------------------------------------------------------
# Module-level invariants
# ---------------------------------------------------------------------------
def test_file_mutating_tools_set_shape():
"""write_file + patch are the only tools the verifier tracks.
Guard rail: if someone adds a third file-mutating tool (e.g. a new
``append_file``), they should also audit whether the verifier should
track it. This test fails loudly on unilateral additions.
"""
assert _FILE_MUTATING_TOOLS == frozenset({"write_file", "patch"})
Reference in New Issue View Git Blame Copy Permalink