fix(memory): write-time auto-consolidate on overflow

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", "")

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,