mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-04-25 00:51:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

docs(plugins): rename disk-guardian to disk-cleanup + bundled-plugins docs

Browse source 
The original name was cute but non-obvious; disk-cleanup says what it
does. Plugin directory, script, state path, log lines, slash command,
and test module all renamed. No user-visible state exists yet, so no
migration path is needed.

New website page "Built-in Plugins" documents the <repo>/plugins/<name>/
source, how discovery interacts with user/project plugins, the
HERMES_DISABLE_BUNDLED_PLUGINS escape hatch, disk-cleanup's hook
behaviour and deletion rules, and guidance on when a plugin belongs
bundled vs. user-installable. Added to the Features → Core sidebar next
to the main Plugins page, with a cross-reference from plugins.md.
This commit is contained in:
Teknium 2026-04-20 03:06:17 -07:00 • committed by Teknium
parent 1386e277e5
commit a25c8c6a56
8 changed files with 184 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

51
plugins/disk-cleanup/README.md Normal file
View file

@ -0,0 +1,51 @@
# disk-cleanup
Auto-tracks and cleans up ephemeral files created during Hermes Agent
sessions — test scripts, temp outputs, cron logs, stale chrome profiles.
Scoped strictly to `$HERMES_HOME` and `/tmp/hermes-*`.
Originally contributed by [@LVT382009](https://github.com/LVT382009) as a
skill in PR #12212. Ported to the plugin system so the behaviour runs
automatically via `post_tool_call` and `on_session_end` hooks — the agent
never needs to remember to call a tool.
## How it works
| Hook | Behaviour |
|---|---|
| `post_tool_call` | When `write_file` / `terminal` / `patch` creates a file matching `test_*`, `tmp_*`, or `*.test.*` inside `HERMES_HOME`, track it silently as `test` / `temp` / `cron-output`. |
| `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):
| Category | Threshold | Confirmation |
|---|---|---|
| `test` | every session end | Never |
| `temp` | >7 days since tracked | Never |
| `cron-output` | >14 days since tracked | Never |
| empty dirs under HERMES_HOME | always | Never |
| `research` | >30 days, beyond 10 newest | Always (deep only) |
| `chrome-profile` | >14 days since tracked | Always (deep only) |
| files >500 MB | never auto | Always (deep only) |
## Slash command
```
/disk-cleanup status # breakdown + top-10 largest
/disk-cleanup dry-run # preview without deleting
/disk-cleanup quick # run safe cleanup now
/disk-cleanup deep # quick + list items needing prompt
/disk-cleanup track <path> <category> # manual tracking
/disk-cleanup forget <path> # stop tracking
```
## Safety
- `is_safe_path()` rejects anything outside `HERMES_HOME` or `/tmp/hermes-*`
- Windows mounts (`/mnt/c` etc.) are rejected
- The state directory `$HERMES_HOME/disk-cleanup/` is itself excluded
- `$HERMES_HOME/logs/`, `memories/`, `sessions/`, `skills/`, `plugins/`,
and config files are never tracked
- Backup/restore is scoped to `tracked.json` — the plugin never touches
agent logs
- Atomic writes: `.tmp` → backup → rename

316
plugins/disk-cleanup/__init__.py Normal file
View file

@ -0,0 +1,316 @@
"""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 longer
needs to remember to run commands.
"""
from __future__ import annotations
import logging
import re
import shlex
import threading
from pathlib import Path
from 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.",
)

496
plugins/disk-cleanup/disk_cleanup.py Executable file
View file

@ -0,0 +1,496 @@
"""disk_cleanup — ephemeral file cleanup for Hermes Agent.
Library module wrapping the deterministic cleanup rules written by
@LVT382009 in PR #12212. The plugin ``__init__.py`` wires these
functions into ``post_tool_call`` and ``on_session_end`` hooks so
tracking and cleanup happen automatically — the agent never needs to
call a tool or remember a skill.
Rules:
- test files → delete immediately at task end (age >= 0)
- temp files → delete after 7 days
- cron-output → delete after 14 days
- empty dirs → always delete (under HERMES_HOME)
- research → keep 10 newest, prompt for older (deep only)
- chrome-profile→ prompt after 14 days (deep only)
- >500 MB files → prompt always (deep only)
Scope: strictly HERMES_HOME and /tmp/hermes-*
Never touches: ~/.hermes/logs/ or any system directory.
"""
from __future__ import annotations
import json
import logging
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
try:
from hermes_constants import get_hermes_home
except Exception: # pragma: no cover — plugin may load before constants resolves
import os
def get_hermes_home() -> Path: # type: ignore[no-redef]
val = (os.environ.get("HERMES_HOME") or "").strip()
return Path(val).resolve() if val else (Path.home() / ".hermes").resolve()
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
def get_state_dir() -> Path:
"""State dir — separate from ``$HERMES_HOME/logs/``."""
return get_hermes_home() / "disk-cleanup"
def get_tracked_file() -> Path:
return get_state_dir() / "tracked.json"
def get_log_file() -> Path:
"""Audit log — intentionally NOT under ``$HERMES_HOME/logs/``."""
return get_state_dir() / "cleanup.log"
# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------
def is_safe_path(path: Path) -> bool:
"""Accept only paths under HERMES_HOME or ``/tmp/hermes-*``.
Rejects Windows mounts (``/mnt/c`` etc.) and any system directory.
"""
hermes_home = get_hermes_home()
try:
path.resolve().relative_to(hermes_home)
return True
except (ValueError, OSError):
pass
# Allow /tmp/hermes-* explicitly
parts = path.parts
if len(parts) >= 3 and parts[1] == "tmp" and parts[2].startswith("hermes-"):
return True
return False
# ---------------------------------------------------------------------------
# Audit log
# ---------------------------------------------------------------------------
def _log(message: str) -> None:
try:
log_file = get_log_file()
log_file.parent.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
with open(log_file, "a") as f:
f.write(f"[{ts}] {message}\n")
except OSError:
# Never let the audit log break the agent loop.
pass
# ---------------------------------------------------------------------------
# tracked.json — atomic read/write, backup scoped to tracked.json only
# ---------------------------------------------------------------------------
def load_tracked() -> List[Dict[str, Any]]:
"""Load tracked.json. Restores from ``.bak`` on corruption."""
tf = get_tracked_file()
tf.parent.mkdir(parents=True, exist_ok=True)
if not tf.exists():
return []
try:
return json.loads(tf.read_text())
except (json.JSONDecodeError, ValueError):
bak = tf.with_suffix(".json.bak")
if bak.exists():
try:
data = json.loads(bak.read_text())
_log("WARN: tracked.json corrupted — restored from .bak")
return data
except Exception:
pass
_log("WARN: tracked.json corrupted, no backup — starting fresh")
return []
def save_tracked(tracked: List[Dict[str, Any]]) -> None:
"""Atomic write: ``.tmp`` → backup old → rename."""
tf = get_tracked_file()
tf.parent.mkdir(parents=True, exist_ok=True)
tmp = tf.with_suffix(".json.tmp")
tmp.write_text(json.dumps(tracked, indent=2))
if tf.exists():
shutil.copy2(tf, tf.with_suffix(".json.bak"))
tmp.replace(tf)
# ---------------------------------------------------------------------------
# Categories
# ---------------------------------------------------------------------------
ALLOWED_CATEGORIES = {
"temp", "test", "research", "download",
"chrome-profile", "cron-output", "other",
}
def fmt_size(n: float) -> str:
for unit in ("B", "KB", "MB", "GB", "TB"):
if n < 1024:
return f"{n:.1f} {unit}"
n /= 1024
return f"{n:.1f} PB"
# ---------------------------------------------------------------------------
# Track / forget
# ---------------------------------------------------------------------------
def track(path_str: str, category: str, silent: bool = False) -> bool:
"""Register a file for tracking. Returns True if newly tracked."""
if category not in ALLOWED_CATEGORIES:
_log(f"WARN: unknown category '{category}', using 'other'")
category = "other"
path = Path(path_str).resolve()
if not path.exists():
_log(f"SKIP: {path} (does not exist)")
return False
if not is_safe_path(path):
_log(f"REJECT: {path} (outside HERMES_HOME)")
return False
size = path.stat().st_size if path.is_file() else 0
tracked = load_tracked()
# Deduplicate
if any(item["path"] == str(path) for item in tracked):
return False
tracked.append({
"path": str(path),
"timestamp": datetime.now(timezone.utc).isoformat(),
"category": category,
"size": size,
})
save_tracked(tracked)
_log(f"TRACKED: {path} ({category}, {fmt_size(size)})")
if not silent:
print(f"Tracked: {path} ({category}, {fmt_size(size)})")
return True
def forget(path_str: str) -> int:
"""Remove a path from tracking without deleting the file."""
p = Path(path_str).resolve()
tracked = load_tracked()
before = len(tracked)
tracked = [i for i in tracked if Path(i["path"]).resolve() != p]
removed = before - len(tracked)
if removed:
save_tracked(tracked)
_log(f"FORGOT: {p} ({removed} entries)")
return removed
# ---------------------------------------------------------------------------
# Dry run
# ---------------------------------------------------------------------------
def dry_run() -> Tuple[List[Dict], List[Dict]]:
"""Return (auto_delete_list, needs_prompt_list) without touching files."""
tracked = load_tracked()
now = datetime.now(timezone.utc)
auto: List[Dict] = []
prompt: List[Dict] = []
for item in tracked:
p = Path(item["path"])
if not p.exists():
continue
age = (now - datetime.fromisoformat(item["timestamp"])).days
cat = item["category"]
size = item["size"]
if cat == "test":
auto.append(item)
elif cat == "temp" and age > 7:
auto.append(item)
elif cat == "cron-output" and age > 14:
auto.append(item)
elif cat == "research" and age > 30:
prompt.append(item)
elif cat == "chrome-profile" and age > 14:
prompt.append(item)
elif size > 500 * 1024 * 1024:
prompt.append(item)
return auto, prompt
# ---------------------------------------------------------------------------
# Quick cleanup
# ---------------------------------------------------------------------------
def quick() -> Dict[str, Any]:
"""Safe deterministic cleanup — no prompts.
Returns: ``{"deleted": N, "empty_dirs": N, "freed": bytes,
"errors": [str, ...]}``.
"""
tracked = load_tracked()
now = datetime.now(timezone.utc)
deleted = 0
freed = 0
new_tracked: List[Dict] = []
errors: List[str] = []
for item in tracked:
p = Path(item["path"])
cat = item["category"]
if not p.exists():
_log(f"STALE: {p} (removed from tracking)")
continue
age = (now - datetime.fromisoformat(item["timestamp"])).days
should_delete = (
cat == "test"
or (cat == "temp" and age > 7)
or (cat == "cron-output" and age > 14)
)
if should_delete:
try:
if p.is_file():
p.unlink()
elif p.is_dir():
shutil.rmtree(p)
freed += item["size"]
deleted += 1
_log(f"DELETED: {p} ({cat}, {fmt_size(item['size'])})")
except OSError as e:
_log(f"ERROR deleting {p}: {e}")
errors.append(f"{p}: {e}")
new_tracked.append(item)
else:
new_tracked.append(item)
# Remove empty dirs under HERMES_HOME (but leave HERMES_HOME itself and
# a short list of well-known top-level state dirs alone — a fresh install
# has these empty, and deleting them would surprise the user).
hermes_home = get_hermes_home()
_PROTECTED_TOP_LEVEL = {
"logs", "memories", "sessions", "cron", "cronjobs",
"cache", "skills", "plugins", "disk-cleanup", "optional-skills",
"hermes-agent", "backups", "profiles", ".worktrees",
}
empty_removed = 0
try:
for dirpath in sorted(hermes_home.rglob("*"), reverse=True):
if not dirpath.is_dir() or dirpath == hermes_home:
continue
try:
rel_parts = dirpath.relative_to(hermes_home).parts
except ValueError:
continue
# Skip the well-known top-level state dirs themselves.
if len(rel_parts) == 1 and rel_parts[0] in _PROTECTED_TOP_LEVEL:
continue
try:
if not any(dirpath.iterdir()):
dirpath.rmdir()
empty_removed += 1
_log(f"DELETED: {dirpath} (empty dir)")
except OSError:
pass
except OSError:
pass
save_tracked(new_tracked)
_log(
f"QUICK_SUMMARY: {deleted} files, {empty_removed} dirs, "
f"{fmt_size(freed)}"
)
return {
"deleted": deleted,
"empty_dirs": empty_removed,
"freed": freed,
"errors": errors,
}
# ---------------------------------------------------------------------------
# Deep cleanup (interactive — not called from plugin hooks)
# ---------------------------------------------------------------------------
def deep(
confirm: Optional[callable] = None,
) -> Dict[str, Any]:
"""Deep cleanup.
Runs :func:`quick` first, then asks the *confirm* callable for each
risky item (research > 30d beyond 10 newest, chrome-profile > 14d,
any file > 500 MB). *confirm(item)* must return True to delete.
Returns: ``{"quick": {...}, "deep_deleted": N, "deep_freed": bytes}``.
"""
quick_result = quick()
if confirm is None:
# No interactive confirmer — deep stops after the quick pass.
return {"quick": quick_result, "deep_deleted": 0, "deep_freed": 0}
tracked = load_tracked()
now = datetime.now(timezone.utc)
research, chrome, large = [], [], []
for item in tracked:
p = Path(item["path"])
if not p.exists():
continue
age = (now - datetime.fromisoformat(item["timestamp"])).days
cat = item["category"]
if cat == "research" and age > 30:
research.append(item)
elif cat == "chrome-profile" and age > 14:
chrome.append(item)
elif item["size"] > 500 * 1024 * 1024:
large.append(item)
research.sort(key=lambda x: x["timestamp"], reverse=True)
old_research = research[10:]
freed, count = 0, 0
to_remove: List[Dict] = []
for group in (old_research, chrome, large):
for item in group:
if confirm(item):
try:
p = Path(item["path"])
if p.is_file():
p.unlink()
elif p.is_dir():
shutil.rmtree(p)
to_remove.append(item)
freed += item["size"]
count += 1
_log(
f"DELETED: {p} ({item['category']}, "
f"{fmt_size(item['size'])})"
)
except OSError as e:
_log(f"ERROR deleting {item['path']}: {e}")
if to_remove:
remove_paths = {i["path"] for i in to_remove}
save_tracked([i for i in tracked if i["path"] not in remove_paths])
return {"quick": quick_result, "deep_deleted": count, "deep_freed": freed}
# ---------------------------------------------------------------------------
# Status
# ---------------------------------------------------------------------------
def status() -> Dict[str, Any]:
"""Return per-category breakdown and top 10 largest tracked files."""
tracked = load_tracked()
cats: Dict[str, Dict] = {}
for item in tracked:
c = item["category"]
cats.setdefault(c, {"count": 0, "size": 0})
cats[c]["count"] += 1
cats[c]["size"] += item["size"]
existing = [
(i["path"], i["size"], i["category"])
for i in tracked if Path(i["path"]).exists()
]
existing.sort(key=lambda x: x[1], reverse=True)
return {
"categories": cats,
"top10": existing[:10],
"total_tracked": len(tracked),
}
def format_status(s: Dict[str, Any]) -> str:
"""Human-readable status string (for slash command output)."""
lines = [f"{'Category':<20} {'Files':>6} {'Size':>10}", "-" * 40]
cats = s["categories"]
for cat, d in sorted(cats.items(), key=lambda x: x[1]["size"], reverse=True):
lines.append(f"{cat:<20} {d['count']:>6} {fmt_size(d['size']):>10}")
if not cats:
lines.append("(nothing tracked yet)")
lines.append("")
lines.append("Top 10 largest tracked files:")
if not s["top10"]:
lines.append(" (none)")
else:
for rank, (path, size, cat) in enumerate(s["top10"], 1):
lines.append(f" {rank:>2}. {fmt_size(size):>8} [{cat}] {path}")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Auto-categorisation from tool-call inspection
# ---------------------------------------------------------------------------
_TEST_PATTERNS = ("test_", "tmp_")
_TEST_SUFFIXES = (".test.py", ".test.js", ".test.ts", ".test.md")
def guess_category(path: Path) -> Optional[str]:
"""Return a category label for *path*, or None if we shouldn't track it.
Used by the ``post_tool_call`` hook to auto-track ephemeral files.
"""
if not is_safe_path(path):
return None
# Skip the state dir itself, logs, memory files, sessions, config.
hermes_home = get_hermes_home()
try:
rel = path.resolve().relative_to(hermes_home)
top = rel.parts[0] if rel.parts else ""
if top in {
"disk-cleanup", "logs", "memories", "sessions", "config.yaml",
"skills", "plugins", ".env", "USER.md", "MEMORY.md", "SOUL.md",
"auth.json", "hermes-agent",
}:
return None
if top == "cron" or top == "cronjobs":
return "cron-output"
if top == "cache":
return "temp"
except ValueError:
# Path isn't under HERMES_HOME (e.g. /tmp/hermes-*) — fall through.
pass
name = path.name
if name.startswith(_TEST_PATTERNS):
return "test"
if any(name.endswith(sfx) for sfx in _TEST_SUFFIXES):
return "test"
return None

7
plugins/disk-cleanup/plugin.yaml Normal file
View file

@ -0,0 +1,7 @@
name: disk-cleanup
version: 2.0.0
description: "Auto-track and clean up ephemeral files (test scripts, temp outputs, cron logs) created during Hermes sessions. Runs via plugin hooks — no agent action required."
author: "@LVT382009 (original), NousResearch (plugin port)"
hooks:
- post_tool_call
- on_session_end