feat(memory): add auto_archive opt-in to escape oversize drift (#26045)

tests/tools/test_memory_tool.py

@@ -575,6 +575,116 @@ def test_drift_backup_filename_is_unique_per_invocation(self, store):
 # =========================================================================
 
 
+# =========================================================================
+# Auto-archive oversize entries (issue #26045 follow-up)
+#
+# When drift signal #2 fires because of an externally-written oversize entry,
+# the default behavior is to refuse the mutation. With ``auto_archive=True``,
+# the oversize entry is moved to a sidecar archive file (``MEMORY-archive-<ts>.md``)
+# and the primary file is repaired in place. This is the documented escape
+# hatch for the case where the operator KNOWS the entry is fine and just needs
+# to get back to a working state without losing the oversize content.
+# =========================================================================
+
+
+class TestAutoArchiveOversize:
+    """``add(..., auto_archive=True)`` repairs signal-#2 drift by archiving the oversized entry."""
+
+    def _plant_oversize_entry(self, store, target="memory"):
+        """Write a single entry larger than the test fixture's char_limit (500)."""
+        path = store._path_for(target)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        # One entry that's clearly over the 500-char test limit
+        giant_entry = "x" * 800
+        path.write_text(giant_entry, encoding="utf-8")
+        return path, giant_entry
+
+    def test_add_without_auto_archive_refuses_oversize_drift(self, store):
+        """Default behaviour is unchanged: oversize drift still refuses the add."""
+        self._plant_oversize_entry(store)
+        result = store.add("memory", "Small entry under drift.")
+        assert result["success"] is False
+        assert "drift_backup" in result
+        assert "26045" in result["error"]
+
+    def test_add_with_auto_archive_repairs_signal_2_drift(self, store, tmp_path):
+        """auto_archive=True moves the oversize entry to a sidecar file."""
+        path, giant_entry = self._plant_oversize_entry(store)
+
+        result = store.add("memory", "Small entry after repair.", auto_archive=True)
+
+        assert result["success"] is True, f"expected success, got: {result}"
+        # The sidecar archive exists with the oversize entry body
+        archives = list(tmp_path.glob("MEMORY-archive-*.md"))
+        assert len(archives) == 1, f"expected 1 archive, got {len(archives)}: {archives}"
+        archive_body = archives[0].read_text(encoding="utf-8")
+        assert giant_entry in archive_body
+        assert "§§ archived" in archive_body  # envelope header
+        # The primary file no longer contains the giant entry
+        primary = path.read_text(encoding="utf-8")
+        assert giant_entry not in primary
+        # The original primary was preserved as a .bak
+        baks = list(tmp_path.glob("MEMORY.md.bak.*"))
+        assert len(baks) >= 1
+
+    def test_add_with_auto_archive_response_includes_archive_path(self, store, tmp_path):
+        """The success message tells the operator where the archive landed."""
+        self._plant_oversize_entry(store)
+        result = store.add("memory", "anything", auto_archive=True)
+        assert result["success"] is True
+        assert "Auto-archived" in result["message"]
+        assert "MEMORY-archive-" in result["message"]
+
+    def test_add_with_auto_archive_refuses_round_trip_mismatch(self, store):
+        """auto_archive only repairs signal #2 (oversize), NOT signal #1 (round-trip).
+
+        A round-trip mismatch means the file is structurally corrupted —
+        automatic re-parse could lose data. We refuse even with auto_archive=True.
+        """
+        path = store._path_for("memory")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        # Trigger signal #1 (round-trip mismatch) WITHOUT triggering signal #2.
+        # The standard parser splits on "\n§\n" (3 chars). If we write
+        # "entry\n§\n\nentry" with an EXTRA newline after the delimiter,
+        # the round-trip join won't match because the parser strips each
+        # entry but the raw still has the orphan blank line.
+        path.write_text("small entry 1\n§\n\nsmall entry 2", encoding="utf-8")
+        result = store.add("memory", "another", auto_archive=True)
+        assert result["success"] is False
+        assert "drift_backup" in result
+
+    def test_add_with_auto_archive_works_for_user_target(self, store, tmp_path):
+        """Same logic applies to USER.md — auto_archive uses the right archive filename."""
+        path, giant = self._plant_oversize_entry(store, target="user")
+        result = store.add("user", "ok", auto_archive=True)
+        assert result["success"] is True
+        archives = list(tmp_path.glob("USER-archive-*.md"))
+        assert len(archives) == 1
+        assert giant in archives[0].read_text(encoding="utf-8")
+
+    def test_tool_dispatcher_passes_auto_archive_flag(self, store, tmp_path):
+        """The top-level memory_tool() entry point forwards auto_archive correctly."""
+        import json
+        self._plant_oversize_entry(store)
+        result_json = memory_tool(
+            action="add",
+            target="memory",
+            content="via dispatcher",
+            auto_archive=True,
+            store=store,
+        )
+        result = json.loads(result_json)
+        assert result["success"] is True
+        assert "Auto-archived" in result["message"]
+
+    def test_schema_documents_auto_archive_parameter(self):
+        """LLM-visible schema lists auto_archive so the model knows it's available."""
+        props = MEMORY_SCHEMA["parameters"]["properties"]
+        assert "auto_archive" in props
+        assert props["auto_archive"]["type"] == "boolean"
+        assert props["auto_archive"]["default"] is False
+
+
 class TestLoadTimeSnapshotSanitization:
     def test_clean_entries_pass_through_snapshot(self, tmp_path, monkeypatch):
         monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path)

tools/memory_tool.py

@@ -31,7 +31,7 @@
 from contextlib import contextmanager
 from pathlib import Path
 from hermes_constants import get_hermes_home
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any, List, Optional, Tuple
 
 from utils import atomic_replace
 
@@ -294,8 +294,19 @@ def _char_limit(self, target: str) -> int:
             return self.user_char_limit
         return self.memory_char_limit
 
-    def add(self, target: str, content: str) -> Dict[str, Any]:
-        """Append a new entry. Returns error if it would exceed the char limit."""
+    def add(self, target: str, content: str, auto_archive: bool = False) -> Dict[str, Any]:
+        """Append a new entry. Returns error if it would exceed the char limit.
+
+        If ``auto_archive=True`` and the on-disk file contains one or more
+        entries larger than the store's char limit (drift signal #2),
+        the oversized entries are moved to a sidecar archive file and the
+        primary file is rewritten without them so the new entry can be
+        added without hitting the drift guard. The original content is
+        never destroyed — it lands in ``MEMORY-archive-<ts>.md`` (or
+        ``USER-archive-<ts>.md``) with an envelope header. This is the
+        documented escape hatch for issue #26045 follow-up #4 (oversize
+        drift) and is opt-in so default behaviour is unchanged.
+        """
         content = content.strip()
         if not content:
             return {"success": False, "error": "Content cannot be empty."}
@@ -312,6 +323,36 @@ def add(self, target: str, content: str) -> Dict[str, Any]:
             # content the patch tool / shell append / sister session wrote.
             bak = self._reload_target(target)
             if bak:
+                # Drift guard fired. Round-trip mismatches are NOT auto-fixable
+                # (structural corruption, needs human eyes). Oversize entries
+                # CAN be auto-archived if the operator opts in.
+                if not auto_archive:
+                    return _drift_error(self._path_for(target), bak)
+
+                # Drift is likely from oversize entries — try auto-archive.
+                # _reload_target already populated self.<target>_entries from
+                # the parsed disk file. Check if signal #2 (oversize) is the
+                # actual cause.
+                entries = self._entries_for(target)
+                limit = self._char_limit(target)
+                if any(len(e) > limit for e in entries):
+                    clean, archive_path = self._archive_oversize_entries(
+                        target, entries, limit,
+                    )
+                    self._set_entries(target, clean)
+                    self.save_to_disk(target)
+                    # NOTE: do NOT delete the .bak — keep both as audit trail.
+                    return self._success_response(
+                        target,
+                        message=(
+                            f"Auto-archived oversize entries to {archive_path}. "
+                            f"Primary file now has {len(clean)} entries "
+                            f"({self._char_count(target):,}/{limit:,} chars). "
+                            f"Original file preserved at {bak}. Retry your add."
+                        ),
+                    )
+                # Round-trip mismatch without oversize — still structural
+                # corruption, refuse even with auto_archive=True.
                 return _drift_error(self._path_for(target), bak)
 
             entries = self._entries_for(target)
@@ -574,6 +615,46 @@ def _detect_external_drift(self, target: str) -> Optional[str]:
             return str(bak_path) + " (BACKUP FAILED — file unchanged on disk)"
         return str(bak_path)
 
+    def _archive_oversize_entries(
+        self, target: str, entries: List[str], char_limit: int,
+    ) -> Tuple[List[str], Optional[str]]:
+        """Move entries that exceed ``char_limit`` to a sidecar archive file.
+
+        Used as the auto-repair path when drift signal #2 fires on the
+        ``add`` mutation: the operator passes ``auto_archive=True`` to
+        accept the side effect of moving the oversized entry out of
+        MEMORY.md into ``MEMORY-archive-<ts>.md`` so the primary file
+        remains within budget and the round-trip guard stops firing.
+
+        Round-trip mismatch (signal #1) is NOT repaired here — that's a
+        structural corruption that needs human eyes. This function only
+        handles the case where the file parses cleanly but one entry is
+        larger than the per-store budget, which is the most common
+        user-visible drift pattern in practice (issue #26045 follow-up).
+
+        Returns ``(cleaned_entries, archive_path)``. ``archive_path`` is
+        ``None`` if nothing was archived.
+        """
+        clean = [e for e in entries if len(e) <= char_limit]
+        oversize = [e for e in entries if len(e) > char_limit]
+        if not oversize:
+            return entries, None
+
+        # Sidecar file in the same dir so it shows up in the same backups.
+        mem_dir = get_memory_dir()
+        ts = time.strftime("%Y%m%d-%H%M%S")
+        archive_path = mem_dir / f"{self._path_for(target).name.replace('.md', '')}-archive-{ts}.md"
+
+        # Append each oversized entry with an envelope so it's recoverable
+        # as a standalone record (timestamp + char count + body).
+        with open(archive_path, "a", encoding="utf-8") as f:
+            for entry in oversize:
+                f.write(f"§§ archived {ts} ({len(entry)} chars) §§\n")
+                f.write(entry.rstrip() + "\n")
+                f.write(f"§§ end archived ({len(entry)} chars) §§\n\n")
+
+        return clean, str(archive_path)
+
     @staticmethod
     def _write_file(path: Path, entries: List[str]):
         """Write entries to a memory file using atomic temp-file + rename.
@@ -611,12 +692,18 @@ def memory_tool(
     target: str = "memory",
     content: str = None,
     old_text: str = None,
+    auto_archive: bool = False,
     store: Optional[MemoryStore] = None,
 ) -> str:
     """
     Single entry point for the memory tool. Dispatches to MemoryStore methods.
 
     Returns JSON string with results.
+
+    ``auto_archive`` only takes effect on the ``add`` action. When set,
+    if drift signal #2 fires because of an oversized entry on disk, the
+    entry is moved to ``MEMORY-archive-<ts>.md`` (or ``USER-archive-<ts>.md``)
+    and the primary file is repaired so the add can succeed on retry.
     """
     if store is None:
         return tool_error("Memory is not available. It may be disabled in config or this environment.", success=False)
@@ -627,7 +714,7 @@ def memory_tool(
     if action == "add":
         if not content:
             return tool_error("Content is required for 'add' action.", success=False)
-        result = store.add(target, content)
+        result = store.add(target, content, auto_archive=auto_archive)
 
     elif action == "replace":
         if not old_text:
@@ -702,6 +789,17 @@ def check_memory_requirements() -> bool:
                 "type": "string",
                 "description": "Short unique substring identifying the entry to replace or remove."
             },
+            "auto_archive": {
+                "type": "boolean",
+                "default": False,
+                "description": (
+                    "add-only: when True, if drift signal #2 fires on the on-disk "
+                    "file (oversized entry), automatically move the oversized entry "
+                    "to a MEMORY-archive-<ts>.md sidecar file and rewrite the primary "
+                    "file so the add can succeed on retry. Default False (refuse "
+                    "mutation, surface drift error). See issue #26045 follow-up."
+                ),
+            },
         },
         "required": ["action", "target"],
     },
@@ -720,6 +818,7 @@ def check_memory_requirements() -> bool:
         target=args.get("target", "memory"),
         content=args.get("content"),
         old_text=args.get("old_text"),
+        auto_archive=args.get("auto_archive", False),
         store=kw.get("store")),
     check_fn=check_memory_requirements,
     emoji="🧠",