hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix(disk-cleanup): address review — key snapshots per session, fail closed on unsnapshotted paths

Browse Source 
The pre/post snapshot handshake had two ways to mark a user's existing
file as created, which queues it for deletion when the turn ends:

- Snapshots were keyed by tool_call_id alone. llama.cpp sends one
  constant id for every call. When session B's pre hook ran between
  session A's pre and post, A consumed B's snapshot. A's
  already-existing test_user.py then looked "not in existing", and A
  tracked it.
- Every pre_tool_call hook sees the original args. Another plugin's
  `modify` directive is merged afterwards, and post_tool_call sees the
  final args. A path swapped in by that hook was never snapshotted, so
  "not in existing" held for it too. The file was tracked even when the
  write was refused.

Snapshots are now keyed by (task_id or session_id, tool_call_id), the
identity both hooks receive from the dispatcher. They store the paths the
call named that did not exist yet, instead of the ones that did. The post
hook tracks a path only if its own snapshot saw it absent and it exists
now. Any path the snapshot did not cover fails closed.

I reproduced both end to end through model_tools.handle_function_call
with a second plugin's modify hook and two interleaved sessions. The
user's file was deleted before this fix and survives after it, and a
file the call really created is still cleaned up. The README hook
contract is updated to match.

(cherry picked from commit 8d0b853981f1ab0775549f4831a3d4f733a46106)
This commit is contained in:
Froraut
2026-09-29 11:04:46 +00:00
committed by kshitij
parent 17d55a2032
commit c467053f52
3 changed files with 71 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/disk-cleanup/README.md
View File

@@ -14,8 +14,8 @@ never needs to remember to call a tool.
| Hook | Behaviour |
|---|---|
| `pre_tool_call` | Snapshot which paths named in the call's arguments already exist. |
| `post_tool_call` | When `write_file` / `terminal` / `patch` creates a file matching `test_*`, `tmp_*`, or `*.test.*` inside `HERMES_HOME` — a path named in the call that did not exist before it — track it silently as `test` / `temp` / `cron-output`. Existing files the call edits, runs or lists are never tracked. |
| `pre_tool_call` | Snapshot which paths named in the call's arguments do not exist yet, keyed by the owning task/session plus `tool_call_id` (the id alone is not unique: llama.cpp reuses one constant id). |
| `post_tool_call` | When `write_file` / `terminal` / `patch` creates a file matching `test_*`, `tmp_*`, or `*.test.*` inside `HERMES_HOME` — a path named in the call that did not exist before it — track it silently as `test` / `temp` / `cron-output`. Existing files the call edits, runs or lists are never tracked, and neither is any path the call's own snapshot did not cover (e.g. one another plugin's `modify` hook swapped in). |
| `on_session_end` | If any test files were auto-tracked during this turn, run `quick` cleanup (no prompts). |
Deletion rules (same as the original PR):

51
plugins/disk-cleanup/__init__.py
View File

@@ -14,7 +14,7 @@ import shlex
import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Set
from typing import Any, Callable, Dict, FrozenSet, List, Optional, Set, Tuple
from . import disk_cleanup as dg
@@ -60,50 +60,57 @@ _PATH_EXTRACTORS: Dict[str, Callable[[Dict[str, Any]], Set[str]]] = {
"terminal": _extract_paths_from_terminal}
# tool_call_id -> (snapshot time, the call's argument paths that already existed, or None if unknown).
# Tracking is for what a call CREATED: 'test' items are deleted at age 0 when the turn ends (rmtree
# for dirs), so a user's test_* script the agent patched, ran or merely listed must never be queued.
_pre_call: Dict[str, tuple] = {}
# (task_id or session_id, tool_call_id) -> (snapshot time, the call's argument paths that did NOT
# exist before it ran). Tracking is for what a call CREATED: 'test' items are deleted at age 0 when
# the turn ends (rmtree for dirs), so a user's test_* script the agent patched, ran or merely listed
# must never be queued. The owning task/session is part of the key because tool_call_id is not
# unique (llama.cpp sends one constant id for every call). Post tracks only paths this snapshot saw
# absent, so a path it never covered — e.g. one another plugin's ``modify`` directive swapped in
# after every pre hook ran on the original args — fails closed.
_pre_call: Dict[Tuple[str, str], Tuple[float, FrozenSet[str]]] = {}
_PRE_CALL_TTL_S = 3600.0 # a call whose post hook never fires (blocked) must not leak its entry
def _on_pre_tool_call(tool_name: str = "", args: Optional[Dict[str, Any]] = None,
tool_call_id: str = "", **_: Any) -> None:
"""Snapshot which argument paths exist before the call. Never raises: pre_tool_call hooks
fail CLOSED (a raise would block the tool), so an error just records "unknown"."""
def _pre_call_key(task_id: str, session_id: str, tool_call_id: str) -> Tuple[str, str]:
return (task_id or session_id or "default", tool_call_id)
def _on_pre_tool_call(tool_name: str = "", args: Optional[Dict[str, Any]] = None, task_id: str = "",
session_id: str = "", tool_call_id: str = "", **_: Any) -> None:
"""Snapshot which argument paths are absent before the call. Never raises: pre_tool_call hooks
fail CLOSED (a raise would block the tool), so an error records no absent path."""
extractor = _PATH_EXTRACTORS.get(tool_name)
if not tool_call_id or extractor is None or not isinstance(args, dict):
return None
now = time.time()
try:
existing: Optional[Set[str]] = {
str(Path(p).expanduser()) for p in extractor(args) if Path(p).expanduser().exists()}
absent = frozenset(str(p) for p in (Path(s).expanduser() for s in extractor(args)) if not p.exists())
except Exception:
existing = None
absent = frozenset()
with _lock:
for key in [k for k, (taken, _e) in _pre_call.items() if now - taken > _PRE_CALL_TTL_S]:
for key in [k for k, (taken, _a) in _pre_call.items() if now - taken > _PRE_CALL_TTL_S]:
del _pre_call[key]
_pre_call[tool_call_id] = (now, existing)
_pre_call[_pre_call_key(task_id, session_id, tool_call_id)] = (now, absent)
return None
def _on_post_tool_call(tool_name: str = "", args: Optional[Dict[str, Any]] = None, result: Any = None,
task_id: str = "", session_id: str = "", tool_call_id: str = "", **_: Any) -> None:
"""Auto-track ephemeral files THIS call created: a path from the call's own arguments that was
absent in the pre-call snapshot and exists now. Paths seen only in terminal OUTPUT (a `find` or
`ls` listing) carry no such proof and are never tracked; no snapshot means no proof either.
Best-effort, never raises."""
"""Auto-track ephemeral files THIS call created: a path from the call's final arguments that
this call's own pre-call snapshot saw absent and that exists now. Paths seen only in terminal
OUTPUT (a `find` or `ls` listing), or absent from the snapshot, carry no such proof and are
never tracked. Best-effort, never raises."""
extractor = _PATH_EXTRACTORS.get(tool_name)
if not isinstance(args, dict) or extractor is None:
if not isinstance(args, dict) or extractor is None or not tool_call_id:
return
with _lock:
_taken, existing = _pre_call.pop(tool_call_id, (0.0, None)) if tool_call_id else (0.0, None)
if existing is None:
_taken, absent = _pre_call.pop(_pre_call_key(task_id, session_id, tool_call_id), (0.0, frozenset()))
if not absent:
return
for path_str in extractor(args):
try:
p = Path(path_str).expanduser()
created = str(p) not in existing and p.exists()
created = str(p) in absent and p.exists()
except Exception:
continue
category = dg.guess_category(p) if created else None

42
tests/plugins/test_disk_cleanup_plugin.py
View File

@@ -51,9 +51,10 @@ _CALL_IDS = iter(range(1, 10**6))
def _run_tool(pi, tool_name, args, create=None, result="OK", **ids):
"""The real hook order: pre_tool_call snapshot, the call's side effect, post_tool_call."""
"""The real hook order: pre_tool_call snapshot, the call's side effect, post_tool_call — both
hooks carry the same identity fields, as the dispatcher sends them."""
call_id = f"call-{next(_CALL_IDS)}"
pi._on_pre_tool_call(tool_name=tool_name, args=args, tool_call_id=call_id)
pi._on_pre_tool_call(tool_name=tool_name, args=args, tool_call_id=call_id, **ids)
if create is not None:
create()
pi._on_post_tool_call(tool_name=tool_name, args=args, result=result, tool_call_id=call_id, **ids)
@@ -542,6 +543,43 @@ class TestPostToolCallHook:
assert script.exists() and hook_dir.is_dir()
def test_a_call_never_uses_another_sessions_snapshot(self, _isolate_env):
"""tool_call_id is not unique: llama.cpp sends one constant id for every call. Session A
patching the user's test_user.py must not consume session B's snapshot of a new file under
that id, or the user's file reads as created and is deleted when the turn ends; B's own
created file is still tracked and cleaned up."""
pi = _load_plugin_init()
user_file = _isolate_env / "test_user.py"
user_file.write_text("keep")
new_file = _isolate_env / "test_new.py"
a = {"tool_call_id": "call_0", "task_id": "sA", "session_id": "sA"}
b = {"tool_call_id": "call_0", "task_id": "sB", "session_id": "sB"}
pi._on_pre_tool_call(tool_name="patch", args={"path": str(user_file)}, **a)
pi._on_pre_tool_call(tool_name="write_file", args={"path": str(new_file)}, **b)
new_file.write_text("x")
pi._on_post_tool_call(tool_name="patch", args={"path": str(user_file)}, result="OK", **a)
pi._on_post_tool_call(tool_name="write_file", args={"path": str(new_file)}, result="OK", **b)
pi._on_session_end(session_id="sA", completed=True, interrupted=False)
assert user_file.read_text() == "keep"
assert not new_file.exists()
def test_a_path_rewritten_after_the_snapshot_is_never_tracked(self, _isolate_env):
"""Every pre_tool_call hook sees the ORIGINAL args; another plugin's ``modify`` directive is
merged afterwards and post_tool_call sees the final args. A path the snapshot never covered
has no proof the call created it, so the user's existing file must survive the turn."""
pi = _load_plugin_init()
user_file = _isolate_env / "test_user.py"
user_file.write_text("keep")
ids = {"tool_call_id": "call_1", "task_id": "s_mod", "session_id": "s_mod"}
pi._on_pre_tool_call(tool_name="write_file", args={"path": str(_isolate_env / "test_absent.py")}, **ids)
pi._on_post_tool_call(tool_name="write_file", args={"path": str(user_file)}, result="OK", **ids)
pi._on_session_end(session_id="s_mod", completed=True, interrupted=False)
assert user_file.read_text() == "keep"
class TestOnSessionEndHook:
def test_runs_quick_when_test_files_tracked(self, _isolate_env):