Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions agent/turn_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -1126,6 +1126,7 @@ def build_turn_context(

# Per-turn file-mutation verifier state.
agent._turn_failed_file_mutations = {}
agent._turn_superseded_file_mutations = {}
agent._turn_file_mutation_paths = set()
agent._verification_stop_nudges = 0
agent._pre_verify_nudges = 0
Expand Down
5 changes: 3 additions & 2 deletions agent/turn_finalizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -474,8 +474,9 @@ def finalize_turn(
if final_response and not interrupted:
try:
_failed = getattr(agent, "_turn_failed_file_mutations", None) or {}
if _failed and agent._file_mutation_verifier_enabled():
footer = agent._format_file_mutation_failure_footer(_failed)
_superseded = getattr(agent, "_turn_superseded_file_mutations", None) or {}
if (_failed or _superseded) and agent._file_mutation_verifier_enabled():
footer = agent._format_file_mutation_failure_footer(_failed, _superseded)
if footer:
final_response = final_response.rstrip() + "\n\n" + footer
except Exception as _ver_err:
Expand Down
141 changes: 126 additions & 15 deletions run_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -3311,11 +3311,21 @@ def _record_file_mutation_result(
state dict hasn't been initialised yet (e.g. a tool dispatched
outside ``run_conversation``).
"""
if tool_name not in _FILE_MUTATING_TOOLS:
return
state = getattr(self, "_turn_failed_file_mutations", None)
if state is None:
return

# A protected config.yaml patch/write_file failure can be recovered in the same
# turn by the sanctioned CLI path (`hermes config set ...`). That is
# not a file-tool mutation, so the verifier should not keep implying
# the turn outcome is unknown when it has direct evidence that the
# approved config command succeeded.
if tool_name == "terminal":
self._record_config_set_superseded_mutation(args, result, is_error)
return

if tool_name not in _FILE_MUTATING_TOOLS:
return
targets = _extract_file_mutation_targets(tool_name, args)
if not targets:
return
Expand All @@ -3339,6 +3349,69 @@ def _record_file_mutation_result(
for path in targets:
state.pop(path, None)

@staticmethod
def _terminal_config_set_path(args: Dict[str, Any], result: Any) -> Optional[str]:
"""Return config.yaml path from a successful sanctioned config write.

This intentionally recognizes only the narrow, approved path that
resolves protected config.yaml patch failures: `hermes config set ...`
with a zero exit code and the CLI's success output naming config.yaml.
It does not try to infer arbitrary file changes from terminal output.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This accepts ;/| command chains and trusts any matching output line. For example, a command can print Set x in ~/.hermes/config.yaml and then run hermes config set OPENROUTER_API_KEY value; that real command writes .env (hermes_cli/config.py:8134-8136), but this matcher would mark the blocked config.yaml edit as recovered. Require a sole parsed hermes config set invocation and add this regression case.

"""
command = str(args.get("command") or "") if isinstance(args, dict) else ""
if not re.search(r"(?:^|[;&|]\s*)hermes\s+config\s+set\b", command):
return None
if not isinstance(result, str):
return None
try:
data = json.loads(result.strip())
except Exception:
return None
if not isinstance(data, dict) or data.get("exit_code") != 0 or data.get("error"):
return None
output = str(data.get("output") or "")
match = re.search(r"(?m)\bSet\b.*?\bin\s+(.+?config\.yaml)\s*$", output)
if not match:
return None
return match.group(1).strip()

@staticmethod
def _same_path(left: str, right: str) -> bool:
def _norm(p: str) -> str:
return os.path.expanduser(str(p)).replace("\\", "/")
return _norm(left) == _norm(right)

def _record_config_set_superseded_mutation(
self,
args: Dict[str, Any],
result: Any,
is_error: bool,
) -> None:
"""Move protected config patch failures to a recovered bucket.

The regular verifier state is keyed by file-tool targets. A later
`hermes config set ...` terminal call can legitimately modify the same
config file through the sanctioned interface, so preserve both facts:
the patch failed, and the approved config command succeeded.
"""
if is_error:
return
config_path = self._terminal_config_set_path(args, result)
if not config_path:
return
failed = getattr(self, "_turn_failed_file_mutations", None)
if not failed:
return
superseded = getattr(self, "_turn_superseded_file_mutations", None)
if superseded is None:
superseded = {}
self._turn_superseded_file_mutations = superseded
for path in list(failed):
if self._same_path(path, config_path):
info = dict(failed.pop(path) or {})
info["superseded_by"] = "terminal: hermes config set"
superseded[path] = info

def _file_mutation_verifier_enabled(self) -> bool:
"""Check whether the per-turn file-mutation verifier footer is on.

Expand Down Expand Up @@ -3396,27 +3469,51 @@ def _neutralize_footer_paths(cls, text: str) -> str:
return cls._FOOTER_PATH_RE.sub(lambda m: f"`{m.group(0)}`", text)

@classmethod
def _format_file_mutation_failure_footer(cls, failed: Dict[str, Dict[str, Any]]) -> str:
"""Render the per-turn failed-mutation dict as a user-facing footer.
def _format_file_mutation_failure_footer(
cls,
failed: Dict[str, Dict[str, Any]],
superseded: Optional[Dict[str, Dict[str, Any]]] = None,
) -> str:
"""Render per-turn file-mutation verifier state as a footer.

Displays up to 10 paths with their first error preview, then a
count of any additional failures. Returns an empty string when
the dict is empty so callers can concatenate unconditionally.
Displays failed file-tool mutations and, when applicable, protected
config edit attempts that failed via patch/write_file but were later
recovered by an approved command such as `hermes config set`.
Returns an empty string when both dicts are empty.

Every file path that reaches the user-facing text — both the bullet
path and any path echoed inside the tool's error preview — is
backtick-wrapped via ``_neutralize_footer_paths`` so the gateway's
bare-path media extractor can never auto-attach a protected file
(e.g. ``~/.hermes/config.yaml``) to a messaging channel (#35584).
"""
if not failed:
failed = failed or {}
superseded = superseded or {}
if not failed and not superseded:
return ""
lines = [
"⚠️ File-mutation verifier: "
f"{len(failed)} file(s) were NOT modified this turn despite any "
"wording above that may suggest otherwise. Run `git status` or "
"`read_file` to confirm."
]
lines: List[str] = []
if failed and superseded:
lines.append(
"⚠️ File-mutation verifier: "
f"{len(failed)} file mutation attempt(s) still failed; "
f"{len(superseded)} protected config edit attempt(s) failed via "
"direct file mutation but were later applied through an approved command. "
"Run `git status` or `read_file` to confirm."
)
elif failed:
lines.append(
"⚠️ File-mutation verifier: "
f"{len(failed)} file(s) were NOT modified this turn despite any "
"wording above that may suggest otherwise. Run `git status` or "
"`read_file` to confirm."
)
else:
lines.append(
"⚠️ File-mutation verifier: direct file mutation was blocked, "
"but the requested protected config change was later applied "
"through the approved `hermes config set` command."
)

shown = 0
for path, info in failed.items():
if shown >= 10:
Expand All @@ -3428,7 +3525,21 @@ def _format_file_mutation_failure_footer(cls, failed: Dict[str, Dict[str, Any]])
else:
lines.append(f" • `{path}` — [{tool}] failed")
shown += 1
remaining = len(failed) - shown

for path, info in superseded.items():
if shown >= 10:
break
preview = (info.get("error_preview") or "").strip()
tool = info.get("tool") or "patch"
superseded_by = info.get("superseded_by") or "terminal: hermes config set"
if preview:
lines.append(f" • `{path}` — [{tool}] blocked: {preview}")
else:
lines.append(f" • `{path}` — [{tool}] blocked")
lines.append(f" ↳ recovered by [{superseded_by}]")
shown += 1

remaining = len(failed) + len(superseded) - shown
if remaining > 0:
lines.append(f" • … and {remaining} more")
# Neutralize any path the preview text echoed (the bullet path is
Expand Down
Loading