feat(mem): add config-driven allocator trim

hermes_cli/config_defaults.py

@@ -1525,6 +1525,12 @@
     # a plugin in plugins/context_engine/<name>/ or ~/.hermes/plugins/.
     "context": {
         "engine": "compressor",
+        # Return freed glibc allocator pages after long-running agent/TUI
+        # cleanup boundaries. Unsupported platforms are safe no-ops.
+        "memory_trim": {
+            "enabled": True,
+            "cooldown_seconds": 60.0,
+        },
     },
 
     # Persistent memory -- bounded curated memory injected into system prompt

hermes_cli/mem_trim.py

@@ -0,0 +1,119 @@
+"""Rate-limited heap release for long-lived Hermes gateway processes.
+
+On Linux/glibc, ``malloc_trim(0)`` can return pages from freed Python/C
+allocations to the OS.  Other platforms and allocators are safe no-ops.
+Behavior is configured under ``context.memory_trim`` in ``config.yaml``.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import gc
+import logging
+import platform
+import sys
+import threading
+import time
+from collections.abc import Callable
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_COOLDOWN_SECONDS = 60.0
+_trim_lock = threading.Lock()
+_last_trim_monotonic = 0.0
+_probe_done = False
+_malloc_trim: Callable[[int], int] | None = None
+
+
+def _config_settings() -> tuple[bool, float]:
+    """Return fail-open settings from the normal Hermes config path."""
+    enabled = True
+    cooldown: Any = _DEFAULT_COOLDOWN_SECONDS
+    try:
+        from hermes_cli.config import load_config
+
+        config = load_config() or {}
+        context = config.get("context") if isinstance(config, dict) else None
+        settings = context.get("memory_trim") if isinstance(context, dict) else None
+        if isinstance(settings, dict):
+            configured_enabled = settings.get("enabled")
+            if isinstance(configured_enabled, bool):
+                enabled = configured_enabled
+            cooldown = settings.get("cooldown_seconds", _DEFAULT_COOLDOWN_SECONDS)
+    except Exception:
+        pass
+    return enabled, _cooldown_seconds(cooldown)
+
+
+def _cooldown_seconds(value: Any) -> float:
+    if isinstance(value, bool):
+        return _DEFAULT_COOLDOWN_SECONDS
+    try:
+        return max(0.0, float(value))
+    except (TypeError, ValueError):
+        return _DEFAULT_COOLDOWN_SECONDS
+
+
+def _probe_glibc_malloc_trim() -> Callable[[int], int] | None:
+    """Resolve glibc's malloc_trim once; return None on unsupported systems."""
+    global _malloc_trim, _probe_done
+    if _probe_done:
+        return _malloc_trim
+    _probe_done = True
+    if sys.platform != "linux":
+        return None
+    try:
+        if platform.libc_ver()[0].lower() != "glibc":
+            return None
+        libc = ctypes.CDLL(None)
+        trim = libc.malloc_trim
+        trim.argtypes = [ctypes.c_size_t]
+        trim.restype = ctypes.c_int
+        _malloc_trim = trim
+    except Exception as exc:
+        logger.debug("malloc_trim unavailable: %s", exc)
+    return _malloc_trim
+
+
+def trim_memory(
+    *,
+    force: bool = False,
+    reason: str = "",
+    cooldown_seconds: float | None = None,
+) -> bool:
+    """Collect cycles and ask glibc to release free heap pages.
+
+    Returns ``True`` only when ``malloc_trim(0)`` ran and reported success.
+    Unsupported allocators, the config kill switch, cooldown suppression, and all
+    runtime errors return ``False`` without affecting the caller.
+    """
+    enabled, configured_cooldown = _config_settings()
+    if not enabled:
+        return False
+
+    global _last_trim_monotonic
+    with _trim_lock:
+        trim = _probe_glibc_malloc_trim()
+        if trim is None:
+            return False
+        now = time.monotonic()
+        cooldown = (
+            configured_cooldown
+            if cooldown_seconds is None
+            else _cooldown_seconds(cooldown_seconds)
+        )
+        if not force and _last_trim_monotonic and now - _last_trim_monotonic < cooldown:
+            return False
+        # Record the attempt before calling into libc so repeated failures do not
+        # turn every turn boundary into an expensive full collection.
+        _last_trim_monotonic = now
+        try:
+            gc.collect()
+            released = bool(trim(0))
+            if reason:
+                logger.debug("malloc_trim(0) after %s: released=%s", reason, released)
+            return released
+        except Exception as exc:
+            logger.debug("malloc_trim failed after %s: %s", reason or "cleanup", exc)
+            return False

run_agent.py

@@ -4037,6 +4037,15 @@ def close(self) -> None:
         except Exception:
             pass
 
+        # The references above are now gone; on Linux/glibc, return their free
+        # heap pages immediately instead of retaining the process RSS high-water
+        # mark until exit.  This helper is a safe no-op on other allocators.
+        try:
+            from hermes_cli.mem_trim import trim_memory
+            trim_memory(force=True, reason="agent close")
+        except Exception:
+            pass
+
         # 8. Finalize the owned SQLite session row unless this agent is only a
         # temporary helper that deliberately handed session ownership forward
         # (manual compression helpers that rotate to a continuation session_id,

tests/hermes_cli/test_mem_trim.py

@@ -0,0 +1,127 @@
+"""Tests for the long-lived gateway heap-trim helper."""
+
+from unittest.mock import Mock
+
+import pytest
+
+import hermes_cli.mem_trim as mem_trim
+
+
+@pytest.fixture(autouse=True)
+def _reset_trim_state(monkeypatch):
+    monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 0.0)
+    monkeypatch.setattr(mem_trim, "_probe_done", True)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", None)
+
+
+def test_unsupported_allocator_is_noop_without_gc(monkeypatch):
+    collect = Mock()
+    monkeypatch.setattr(mem_trim.gc, "collect", collect)
+
+    assert mem_trim.trim_memory(force=True, reason="test") is False
+    collect.assert_not_called()
+
+
+def test_config_kill_switch_overrides_force(monkeypatch):
+    trim = Mock(return_value=1)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"context": {"memory_trim": {"enabled": False}}},
+    )
+
+    assert mem_trim.trim_memory(force=True) is False
+    trim.assert_not_called()
+
+
+def test_default_config_declares_memory_trim_controls():
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    context = DEFAULT_CONFIG["context"]
+    assert isinstance(context, dict)
+    assert context["memory_trim"] == {
+        "enabled": True,
+        "cooldown_seconds": 60.0,
+    }
+
+
+def test_success_collects_then_trims(monkeypatch):
+    calls = []
+    monkeypatch.setattr(mem_trim.gc, "collect", lambda: calls.append("gc"))
+    monkeypatch.setattr(
+        mem_trim, "_malloc_trim", lambda pad: calls.append(("trim", pad)) or 1
+    )
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+
+    assert mem_trim.trim_memory(reason="turn", cooldown_seconds=60) is True
+    assert calls == ["gc", ("trim", 0)]
+    assert mem_trim._last_trim_monotonic == 100.0
+
+
+def test_cooldown_suppresses_repeated_collection(monkeypatch):
+    collect = Mock()
+    trim = Mock(return_value=1)
+    monkeypatch.setattr(mem_trim.gc, "collect", collect)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 95.0)
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+
+    assert mem_trim.trim_memory(cooldown_seconds=60) is False
+    collect.assert_not_called()
+    trim.assert_not_called()
+    assert mem_trim.trim_memory(force=True, cooldown_seconds=60) is True
+
+
+def test_config_cooldown_controls_rate_limit(monkeypatch):
+    trim = Mock(return_value=1)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 1.0)
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "context": {
+                "memory_trim": {"enabled": True, "cooldown_seconds": 120.0}
+            }
+        },
+    )
+
+    assert mem_trim.trim_memory() is False
+    trim.assert_not_called()
+
+
+def test_legacy_environment_switch_does_not_control_behavior(monkeypatch):
+    trim = Mock(return_value=1)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    monkeypatch.setenv("HERMES_DISABLE_MEMORY_TRIM", "1")
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"context": {"memory_trim": {"enabled": True}}},
+    )
+
+    assert mem_trim.trim_memory(force=True) is True
+    trim.assert_called_once_with(0)
+
+
+def test_libc_failure_is_fail_open_and_rate_limited(monkeypatch):
+    trim = Mock(side_effect=RuntimeError("boom"))
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+
+    assert mem_trim.trim_memory(reason="test", cooldown_seconds=60) is False
+    assert mem_trim._last_trim_monotonic == 100.0
+    assert mem_trim.trim_memory(cooldown_seconds=60) is False
+    assert trim.call_count == 1
+
+
+def test_agent_close_forces_memory_trim(monkeypatch):
+    """A hard agent teardown bypasses cooldown after releasing its history."""
+    from run_agent import AIAgent
+
+    trim = Mock()
+    monkeypatch.setattr(mem_trim, "trim_memory", trim)
+    agent = AIAgent.__new__(AIAgent)
+
+    agent.close()
+
+    trim.assert_called_once_with(force=True, reason="agent close")

tests/test_tui_gateway_server.py

@@ -15212,3 +15212,70 @@ def start(self):
         assert captured.get("persist_user_message") == "hi"
     finally:
         server._sessions.pop("sid", None)
+
+
+def test_prompt_submit_releases_old_history_before_heap_trim(monkeypatch, tmp_path):
+    """The trim boundary must not retain the just-pruned history snapshots."""
+    observed = {}
+    cleanup_order = []
+
+    class _Agent:
+        def run_conversation(
+            self, prompt, conversation_history=None, stream_callback=None
+        ):
+            return {
+                "final_response": "reply",
+                "messages": [{"role": "assistant", "content": "reply"}],
+            }
+
+    class _ImmediateThread:
+        def __init__(self, target=None, daemon=None):
+            self._target = target
+
+        def start(self):
+            assert self._target is not None
+            self._target()
+
+    def _inspect_trim_frame(**_kwargs):
+        import inspect
+
+        cleanup_order.append("trim")
+        frame = inspect.currentframe()
+        assert frame is not None and frame.f_back is not None
+        caller_locals = frame.f_back.f_locals
+        observed["history"] = caller_locals.get("history")
+        observed["run_kwargs"] = caller_locals.get("run_kwargs")
+
+    session = _session(agent=_Agent())
+    session["profile_home"] = str(tmp_path / "test-profile")
+    session["history"] = [
+        {"role": "tool", "tool_call_id": "old", "content": "x" * 20_000}
+    ]
+    server._sessions["sid_trim"] = session
+    try:
+        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
+        monkeypatch.setattr(server, "_get_usage", lambda _a: {})
+        monkeypatch.setattr(server, "render_message", lambda _t, _c: "")
+        monkeypatch.setattr(server, "_emit", lambda *a: None)
+        monkeypatch.setattr(server, "set_hermes_home_override", lambda _home: object())
+        monkeypatch.setattr(
+            server,
+            "reset_hermes_home_override",
+            lambda _token: cleanup_order.append("reset_home"),
+        )
+        monkeypatch.setattr("hermes_cli.mem_trim.trim_memory", _inspect_trim_frame)
+
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "prompt.submit",
+                "params": {"session_id": "sid_trim", "text": "hi"},
+            }
+        )
+
+        assert resp is not None and resp.get("result")
+        assert not observed["history"]
+        assert not observed["run_kwargs"]
+        assert cleanup_order == ["trim", "reset_home"]
+    finally:
+        server._sessions.pop("sid_trim", None)

tui_gateway/server.py

@@ -9548,6 +9548,23 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                     _persist_live_session_system_prompt(session)
                 except Exception:
                     logger.debug("TUI one-turn model restore failed", exc_info=True)
+
+            # Drop both local snapshots of the pre-turn history before asking
+            # glibc to return pages. session["history"] already points at the
+            # new/pruned result; retaining either list defeats this trim.
+            history.clear()
+            local_run_kwargs = locals().get("run_kwargs")
+            if isinstance(local_run_kwargs, dict):
+                local_run_kwargs.clear()
+
+            # Run while any profile-specific HERMES_HOME override is still active
+            # so context.memory_trim is resolved from the session's own config.
+            try:
+                from hermes_cli.mem_trim import trim_memory
+
+                trim_memory(reason="tui turn completion")
+            except Exception:
+                logger.debug("post-turn memory trim failed", exc_info=True)
             try:
                 if approval_token is not None:
                     reset_current_session_key(approval_token)