diff --git a/tests/tools/test_memory_tool.py b/tests/tools/test_memory_tool.py index c2903f375c6d..06e082714aa1 100644 --- a/tests/tools/test_memory_tool.py +++ b/tests/tools/test_memory_tool.py @@ -915,3 +915,123 @@ def test_already_blocked_entry_passes_through(self, tmp_path, monkeypatch): # Block marker appears exactly once, not nested assert snapshot.count("[BLOCKED:") == 1 assert "Clean fact" in snapshot + + # --- write-time auto-consolidate tests (added 2026-07-24, fix for the + # "memory full" recurring problem — see skills/hermes-memory-self-management) --- + + def test_add_overflow_triggers_auto_consolidate(self, tmp_path, monkeypatch): + """When add() would push MEMORY.md past cap, _auto_consolidate is + called and the add succeeds if compress freed enough space. + + The compress script is shelled out to. We stub it with a fake that + rewrites the file to be smaller (simulating real compress behavior). + """ + import subprocess + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + # Cap is 2200 by default. Fill MEMORY.md close to cap, then add a + # large entry that would push past. Expect auto-consolidate to fire. + big = "x" * 1800 + (tmp_path / "MEMORY.md").write_text(big + "\n", encoding="utf-8") + s = MemoryStore(memory_char_limit=2200) + s.load_from_disk() + assert s._char_count("memory") > 1500 + + compress_calls = [] + def fake_compress(): + compress_calls.append(1) + # Shrink the file: overwrite with a single small entry + (tmp_path / "MEMORY.md").write_text("§\nshrunken\n", encoding="utf-8") + r = subprocess.CompletedProcess(args=[], returncode=0, stdout="ok", stderr="") + return r + + def fake_run(*args, **kwargs): + return fake_compress() + monkeypatch.setattr("tools.memory_tool.subprocess.run", fake_run) + + result = s.add("memory", "new entry after auto-consolidate") + assert compress_calls, "auto-consolidate was not triggered" + assert result["success"], f"add failed: {result.get('error')}" + assert "auto-consolidated" in result.get("message", "").lower() + + def test_add_overflow_falls_through_when_no_compress_script(self, tmp_path, monkeypatch): + """If memory-auto-compress.py doesn't exist, add() returns the + standard consolidation_failure (back-compat for environments without + the script). + """ + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + big = "x" * 1800 + (tmp_path / "MEMORY.md").write_text(big + "\n", encoding="utf-8") + s = MemoryStore(memory_char_limit=2200) + s.load_from_disk() + + # Simulate the script being missing: subprocess.run returns nonzero. + import subprocess + def fake_run_fail(*args, **kwargs): + return subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="script not found" + ) + monkeypatch.setattr("tools.memory_tool.subprocess.run", fake_run_fail) + + result = s.add("memory", "new entry that won't fit") + # Falls through to standard consolidation_failure + assert not result["success"] + assert "Consolidate" in result.get("error", "") or "exceed" in result.get("error", "") + + + def test_add_overflow_triggers_auto_consolidate(self, tmp_path, monkeypatch): + """When add() would push MEMORY.md past cap, _auto_consolidate is + called and the add succeeds if compress freed enough space. + + The compress script is shelled out to. We stub it with a fake that + rewrites the file to be smaller (simulating real compress behavior). + """ + import subprocess + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + # Cap is 2200 by default. Fill MEMORY.md close to cap, then add a + # large entry that would push past. Expect auto-consolidate to fire. + big = "x" * 1800 + (tmp_path / "MEMORY.md").write_text(big + "\n", encoding="utf-8") + s = MemoryStore(memory_char_limit=2200) + s.load_from_disk() + assert s._char_count("memory") > 1500 + + compress_calls = [] + def fake_compress(): + compress_calls.append(1) + # Shrink the file: overwrite with a single small entry + (tmp_path / "MEMORY.md").write_text("§\nshrunken\n", encoding="utf-8") + r = subprocess.CompletedProcess(args=[], returncode=0, stdout="ok", stderr="") + return r + + def fake_run(*args, **kwargs): + return fake_compress() + monkeypatch.setattr("tools.memory_tool.subprocess.run", fake_run) + + result = s.add("memory", "new entry after auto-consolidate") + assert compress_calls, "auto-consolidate was not triggered" + assert result["success"], f"add failed: {result.get('error')}" + assert "auto-consolidated" in result.get("message", "").lower() + + def test_add_overflow_falls_through_when_no_compress_script(self, tmp_path, monkeypatch): + """If memory-auto-compress.py doesn't exist, add() returns the + standard consolidation_failure (back-compat for environments without + the script). + """ + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + big = "x" * 1800 + (tmp_path / "MEMORY.md").write_text(big + "\n", encoding="utf-8") + s = MemoryStore(memory_char_limit=2200) + s.load_from_disk() + + # Simulate the script being missing: subprocess.run returns nonzero. + import subprocess + def fake_run_fail(*args, **kwargs): + return subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="script not found" + ) + monkeypatch.setattr("tools.memory_tool.subprocess.run", fake_run_fail) + + result = s.add("memory", "new entry that won't fit") + # Falls through to standard consolidation_failure + assert not result["success"] + assert "Consolidate" in result.get("error", "") or "exceed" in result.get("error", "") diff --git a/tools/memory_tool.py b/tools/memory_tool.py index f5b866b80ac7..db399ac6a681 100644 --- a/tools/memory_tool.py +++ b/tools/memory_tool.py @@ -26,14 +26,19 @@ import json import logging import os +import subprocess +import sys import tempfile import time from contextlib import contextmanager from pathlib import Path from hermes_constants import get_hermes_home from typing import Dict, Any, List, Optional - from utils import atomic_replace +from tools.threat_patterns import first_threat_message as _first_threat_message +from tools.registry import registry, tool_error + +log = logging.getLogger(__name__) # fcntl is Unix-only; on Windows use msvcrt for file locking msvcrt = None @@ -175,6 +180,35 @@ def _consolidation_failure(self, response: Dict[str, Any]) -> Dict[str, Any]: ), } + def _auto_consolidate(self, target: str, timeout: int = 5) -> bool: + """Shell out to memory-auto-compress.py to free up space in `target`. + + Called from add() right before the cap-rejection return. If the + compress script runs successfully and removes enough bytes, the + caller re-measures and the add() proceeds. + + Returns True on success (file is now smaller), False otherwise. + Designed to be cheap and safe to call on every overflow — caps at + `timeout` seconds, swallows non-zero exits, logs to memory.log. + """ + from hermes_constants import get_hermes_home + hermes_home = get_hermes_home() + script = hermes_home / "scripts" / "memory-auto-compress.py" + if not script.exists(): + return False + try: + r = subprocess.run( + [sys.executable, str(script)], + capture_output=True, text=True, timeout=timeout, + ) + if r.returncode == 0: + # Re-measure; caller will check whether it freed enough + return self._char_count(target) < int(os.environ.get("HERMES_MEMORY_LIMIT", "6000")) + return False + except (subprocess.TimeoutExpired, OSError) as e: + log.debug("memory auto-consolidate failed: %s", e) + return False + def load_from_disk(self): """Load entries from MEMORY.md and USER.md, capture system prompt snapshot. @@ -375,6 +409,24 @@ def add(self, target: str, content: str) -> Dict[str, Any]: new_total = len(ENTRY_DELIMITER.join(new_entries)) if new_total > limit: + # Write-time auto-consolidate: try to free up space by running + # the same compress script the weekly cron uses, before we + # reject. This is the proper fix for "memory full" — the cron + # alone is reactive, this is proactive. (skill: + # hermes-memory-self-management, fix: write-time auto-consolidate) + if self._auto_consolidate(target): + # Re-measure under the same lock + self._reload_target(target, skip_drift=True) + entries = self._entries_for(target) + new_entries = entries + [content] + new_total = len(ENTRY_DELIMITER.join(new_entries)) + if new_total <= limit: + entries.append(content) + self._set_entries(target, entries) + self.save_to_disk(target) + return self._success_response( + target, "Entry added (auto-consolidated to make room)." + ) current = self._char_count(target) return self._consolidation_failure({ "success": False,