Something went wrong. Try again.
The agent that grows with you
Something went wrong. Try again.
11 kB · 316 lines
Python
at bb/theme
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317"""disk-cleanup plugin — auto-cleanup of ephemeral Hermes session files.
Wires three behaviours:
1. ``post_tool_call`` hook — inspects ``write_file`` and ``terminal`` tool results for newly-created paths matching test/temp patterns under ``HERMES_HOME`` and tracks them silently. Zero agent compliance required.
2. ``on_session_end`` hook — when any test files were auto-tracked during the just-finished turn, runs :func:`disk_cleanup.quick` and logs a single line to ``$HERMES_HOME/disk-cleanup/cleanup.log``.
3. ``/disk-cleanup`` slash command — manual ``status``, ``dry-run``, ``quick``, ``deep``, ``track``, ``forget``.
Replaces PR #12212's skill-plus-script design: the agent no longerneeds to remember to run commands."""
from __future__ import annotations
import loggingimport reimport shleximport threadingfrom pathlib import Pathfrom typing import Any, Dict, Optional, Set
from . import disk_cleanup as dg
logger = logging.getLogger(__name__)
# Per-task set of "test files newly tracked this turn". Keyed by task_id# (or session_id as fallback) so on_session_end can decide whether to run# cleanup. Guarded by a lock — post_tool_call can fire concurrently on# parallel tool calls._recent_test_tracks: Dict[str, Set[str]] = {}_lock = threading.Lock()
# Tool-call result shapes we can parse_WRITE_FILE_PATH_KEY = "path"_TERMINAL_PATH_REGEX = re.compile(r"(?:^|\s)(/[^\s'\"`]+|\~/[^\s'\"`]+)")
# ---------------------------------------------------------------------------# Helpers# ---------------------------------------------------------------------------
def _tracker_key(task_id: str, session_id: str) -> str: return task_id or session_id or "default"
def _record_track(task_id: str, session_id: str, path: Path, category: str) -> None: """Record that we tracked *path* as *category* during this turn.""" if category != "test": return key = _tracker_key(task_id, session_id) with _lock: _recent_test_tracks.setdefault(key, set()).add(str(path))
def _drain(task_id: str, session_id: str) -> Set[str]: """Pop the set of test paths tracked during this turn.""" key = _tracker_key(task_id, session_id) with _lock: return _recent_test_tracks.pop(key, set())
def _attempt_track(path_str: str, task_id: str, session_id: str) -> None: """Best-effort auto-track. Never raises.""" try: p = Path(path_str).expanduser() except Exception: return if not p.exists(): return category = dg.guess_category(p) if category is None: return newly = dg.track(str(p), category, silent=True) if newly: _record_track(task_id, session_id, p, category)
def _extract_paths_from_write_file(args: Dict[str, Any]) -> Set[str]: path = args.get(_WRITE_FILE_PATH_KEY) return {path} if isinstance(path, str) and path else set()
def _extract_paths_from_patch(args: Dict[str, Any]) -> Set[str]: # The patch tool creates new files via the `mode="patch"` path too, but # most of its use is editing existing files — we only care about new # ephemeral creations, so treat patch conservatively and only pick up # the single-file `path` arg. Track-then-cleanup is idempotent, so # re-tracking an already-tracked file is a no-op (dedup in track()). path = args.get("path") return {path} if isinstance(path, str) and path else set()
def _extract_paths_from_terminal(args: Dict[str, Any], result: str) -> Set[str]: """Best-effort: pull candidate filesystem paths from a terminal command and its output, then let ``guess_category`` / ``is_safe_path`` filter. """ paths: Set[str] = set() cmd = args.get("command") or "" if isinstance(cmd, str) and cmd: # Tokenise the command — catches `touch /tmp/hermes-x/test_foo.py` try: for tok in shlex.split(cmd, posix=True): if tok.startswith(("/", "~")): paths.add(tok) except ValueError: pass # Only scan the result text if it's a reasonable size (avoid 50KB dumps). if isinstance(result, str) and len(result) < 4096: for match in _TERMINAL_PATH_REGEX.findall(result): paths.add(match) return paths
# ---------------------------------------------------------------------------# Hooks# ---------------------------------------------------------------------------
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 created by recent tool calls.""" if not isinstance(args, dict): return
candidates: Set[str] = set() if tool_name == "write_file": candidates = _extract_paths_from_write_file(args) elif tool_name == "patch": candidates = _extract_paths_from_patch(args) elif tool_name == "terminal": candidates = _extract_paths_from_terminal(args, result if isinstance(result, str) else "") else: return
for path_str in candidates: _attempt_track(path_str, task_id, session_id)
def _on_session_end( session_id: str = "", completed: bool = True, interrupted: bool = False, **_: Any,) -> None: """Run quick cleanup if any test files were tracked during this turn.""" # Drain both task-level and session-level buckets. In practice only one # is populated per turn; the other is empty. drained_session = _drain("", session_id) # Also drain any task-scoped buckets that happen to exist. This is a # cheap sweep: if an agent spawned subagents (each with their own # task_id) they'll have recorded into separate buckets; we want to # cleanup them all at session end. with _lock: task_buckets = list(_recent_test_tracks.keys()) for key in task_buckets: if key and key != session_id: _recent_test_tracks.pop(key, None)
if not drained_session and not task_buckets: return
try: summary = dg.quick() except Exception as exc: logger.debug("disk-cleanup quick cleanup failed: %s", exc) return
if summary["deleted"] or summary["empty_dirs"]: dg._log( f"AUTO_QUICK (session_end): deleted={summary['deleted']} " f"dirs={summary['empty_dirs']} freed={dg.fmt_size(summary['freed'])}" )
# ---------------------------------------------------------------------------# Slash command# ---------------------------------------------------------------------------
_HELP_TEXT = """\/disk-cleanup — ephemeral-file cleanup
Subcommands: status Per-category breakdown + top-10 largest dry-run Preview what quick/deep would delete quick Run safe cleanup now (no prompts) deep Run quick, then list items that need prompts track <path> <category> Manually add a path to tracking forget <path> Stop tracking a path (does not delete)
Categories: temp | test | research | download | chrome-profile | cron-output | other
All operations are scoped to HERMES_HOME and /tmp/hermes-*.Test files are auto-tracked on write_file / terminal and auto-cleaned at session end."""
def _fmt_summary(summary: Dict[str, Any]) -> str: base = ( f"[disk-cleanup] Cleaned {summary['deleted']} files + " f"{summary['empty_dirs']} empty dirs, freed {dg.fmt_size(summary['freed'])}." ) if summary.get("errors"): base += f"\n {len(summary['errors'])} error(s); see cleanup.log." return base
def _handle_slash(raw_args: str) -> Optional[str]: argv = raw_args.strip().split() if not argv or argv[0] in ("help", "-h", "--help"): return _HELP_TEXT
sub = argv[0]
if sub == "status": return dg.format_status(dg.status())
if sub == "dry-run": auto, prompt = dg.dry_run() auto_size = sum(i["size"] for i in auto) prompt_size = sum(i["size"] for i in prompt) lines = [ "Dry-run preview (nothing deleted):", f" Auto-delete : {len(auto)} files ({dg.fmt_size(auto_size)})", ] for item in auto: lines.append(f" [{item['category']}] {item['path']}") lines.append( f" Needs prompt: {len(prompt)} files ({dg.fmt_size(prompt_size)})" ) for item in prompt: lines.append(f" [{item['category']}] {item['path']}") lines.append( f"\n Total potential: {dg.fmt_size(auto_size + prompt_size)}" ) return "\n".join(lines)
if sub == "quick": return _fmt_summary(dg.quick())
if sub == "deep": # In-session deep can't prompt the user interactively — show what # quick cleaned plus the items that WOULD need confirmation. quick_summary = dg.quick() _auto, prompt_items = dg.dry_run() lines = [_fmt_summary(quick_summary)] if prompt_items: size = sum(i["size"] for i in prompt_items) lines.append( f"\n{len(prompt_items)} item(s) need confirmation " f"({dg.fmt_size(size)}):" ) for item in prompt_items: lines.append(f" [{item['category']}] {item['path']}") lines.append( "\nRun `/disk-cleanup forget <path>` to skip, or delete " "manually via terminal." ) return "\n".join(lines)
if sub == "track": if len(argv) < 3: return "Usage: /disk-cleanup track <path> <category>" path_arg = argv[1] category = argv[2] if category not in dg.ALLOWED_CATEGORIES: return ( f"Unknown category '{category}'. " f"Allowed: {sorted(dg.ALLOWED_CATEGORIES)}" ) if dg.track(path_arg, category, silent=True): return f"Tracked {path_arg} as '{category}'." return ( f"Not tracked (already present, missing, or outside HERMES_HOME): " f"{path_arg}" )
if sub == "forget": if len(argv) < 2: return "Usage: /disk-cleanup forget <path>" n = dg.forget(argv[1]) return ( f"Removed {n} tracking entr{'y' if n == 1 else 'ies'} for {argv[1]}." if n else f"Not found in tracking: {argv[1]}" )
return f"Unknown subcommand: {sub}\n\n{_HELP_TEXT}"
# ---------------------------------------------------------------------------# Plugin registration# ---------------------------------------------------------------------------
def register(ctx) -> None: ctx.register_hook("post_tool_call", _on_post_tool_call) ctx.register_hook("on_session_end", _on_session_end) ctx.register_command( "disk-cleanup", handler=_handle_slash, description="Track and clean up ephemeral Hermes session files.", )