feat(mem): config-driven allocator trim with telemetry and lifecycle coverage

Add config-driven glibc malloc_trim for long-lived Hermes processes:
- hermes_cli/mem_trim.py: trim_memory() with configurable cooldown,
  RSS snapshot telemetry, and forced-trim INFO logging
- gateway/run.py: periodic trim in gateway housekeeping loop
- tui_gateway/server.py: trim in idle reaper (~every 5 min)
- tui_gateway/slash_worker.py: trim on turn boundary
- run_agent.py: force trim on agent close
- hermes_cli/config.py: context.memory_trim config section
  (enabled, cooldown_seconds, log_every_n, info_log_min_delta_mb)

CSA tier-4 reviewed (4 rounds, 0 HIGH/MEDIUM/CRITICAL remaining).

Supersedes PR #63708 + #64591 with enhanced telemetry and gateway/slash_worker coverage.

gateway/run.py

@@ -25318,6 +25318,7 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop
     PASTE_SWEEP_EVERY = 60   # ticks — once per hour
     CURATOR_EVERY = 60       # ticks — poll hourly (inner gate handles the real cadence)
     AUTO_ARCHIVE_EVERY = 60  # ticks — poll hourly (state_meta gate owns the real cadence)
+    MEMORY_TRIM_EVERY = 1    # shared helper cooldown bounds actual allocator work
 
     # Every platform media cache prunes on the same hourly cadence — one loop
     # over (name, cleanup_fn), not a copy-pasted try/except per cache.
@@ -25426,6 +25427,21 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop
             except Exception as e:
                 logger.debug("Auto-archive tick error: %s", e)
 
+        # This is the long-lived messaging-gateway counterpart to the TUI idle
+        # reaper. The helper is config-gated and rate-limited, so calling it on
+        # the 60s housekeeping cadence does not create a trim storm.
+        if tick_count % MEMORY_TRIM_EVERY == 0:
+            try:
+                from hermes_cli.mem_trim import trim_memory
+
+                trim_memory(reason="messaging gateway housekeeping")
+            except Exception as exc:
+                logger.warning(
+                    "gateway housekeeping memory trim failed: %s: %s",
+                    type(exc).__name__,
+                    exc,
+                )
+
         stop_event.wait(timeout=interval)
     logger.info("Gateway housekeeping stopped")
 

hermes_cli/config_defaults.py

@@ -1587,6 +1587,18 @@ DEFAULT_CONFIG = {
     # 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,
+            # Successful trim calls are INFO logged every Nth periodic call;
+            # force paths always log so process-close behavior is visible.
+            "log_every_n": 1,
+            # Suppress INFO logs only when a readable RSS change is smaller.
+            # 0 reports every successful configured trim.
+            "info_log_min_delta_mb": 0.0,
+        },
     },
 
     # Persistent memory -- bounded curated memory injected into system prompt

hermes_cli/mem_trim.py

@@ -0,0 +1,238 @@
+"""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 pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_COOLDOWN_SECONDS = 60.0
+_DEFAULT_LOG_EVERY_N = 1
+_DEFAULT_INFO_LOG_MIN_DELTA_MB = 0.0
+_trim_lock = threading.Lock()
+_last_trim_monotonic = 0.0
+_probe_done = False
+_malloc_trim: Callable[[int], int] | None = None
+_trim_call_count = 0
+
+
+def _config_settings() -> tuple[bool, float, int, float]:
+    """Return fail-open settings from the normal Hermes config path."""
+    enabled = True
+    cooldown: Any = _DEFAULT_COOLDOWN_SECONDS
+    log_every_n: Any = _DEFAULT_LOG_EVERY_N
+    info_log_min_delta_mb: Any = _DEFAULT_INFO_LOG_MIN_DELTA_MB
+    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)
+            log_every_n = settings.get("log_every_n", _DEFAULT_LOG_EVERY_N)
+            info_log_min_delta_mb = settings.get(
+                "info_log_min_delta_mb", _DEFAULT_INFO_LOG_MIN_DELTA_MB
+            )
+    except Exception:
+        pass
+    return (
+        enabled,
+        _cooldown_seconds(cooldown),
+        _log_every_n(log_every_n),
+        _nonnegative_float(info_log_min_delta_mb, _DEFAULT_INFO_LOG_MIN_DELTA_MB),
+    )
+
+
+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 _log_every_n(value: Any) -> int:
+    if isinstance(value, bool):
+        return _DEFAULT_LOG_EVERY_N
+    try:
+        return max(1, int(value))
+    except (TypeError, ValueError):
+        return _DEFAULT_LOG_EVERY_N
+
+
+def _nonnegative_float(value: Any, default: float) -> float:
+    if isinstance(value, bool):
+        return default
+    try:
+        return max(0.0, float(value))
+    except (TypeError, ValueError):
+        return default
+
+
+def _read_proc_status() -> str | None:
+    """Read Linux process status without making non-Linux callers special-case."""
+    if sys.platform != "linux":
+        return None
+    try:
+        return Path("/proc/self/status").read_text(encoding="utf-8")
+    except OSError:
+        return None
+
+
+def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int | None]:
+    """Return lightweight process-memory telemetry for trim logs and canaries.
+
+    ``VmRSS`` and ``RssAnon`` are Linux-only best effort fields.  The helper is
+    intentionally dependency-free so allocation recovery never requires psutil.
+    """
+    snapshot: dict[str, int | None] = {
+        "rss_kib": None,
+        "rss_anon_kib": None,
+        "thread_count": threading.active_count(),
+    }
+    status = _read_proc_status()
+    if status:
+        for line in status.splitlines():
+            key, separator, raw_value = line.partition(":")
+            if not separator or key not in {"VmRSS", "RssAnon"}:
+                continue
+            value = raw_value.strip().split(maxsplit=1)
+            if value and value[0].isdigit():
+                snapshot["rss_kib" if key == "VmRSS" else "rss_anon_kib"] = int(value[0])
+    if isinstance(history_bytes, int) and history_bytes >= 0:
+        snapshot["history_bytes"] = history_bytes
+    return snapshot
+
+
+def _should_log_trim(
+    *, force: bool, log_every_n: int, call_count: int, before: dict[str, int | None],
+    after: dict[str, int | None], info_log_min_delta_mb: float,
+) -> bool:
+    # trim_memory calls this only after malloc_trim reported success. A forced
+    # successful trim is an explicit observability event, regardless of RSS.
+    if force:
+        return True
+    if not force and call_count % log_every_n:
+        return False
+    before_rss = before.get("rss_kib")
+    after_rss = after.get("rss_kib")
+    if before_rss is None or after_rss is None:
+        return True
+    return abs(after_rss - before_rss) >= info_log_min_delta_mb * 1024
+
+
+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,
+        log_every_n,
+        info_log_min_delta_mb,
+    ) = _config_settings()
+    if not enabled:
+        return False
+
+    global _last_trim_monotonic, _trim_call_count
+    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:
+            before = collect_memory_snapshot()
+            started = time.perf_counter()
+            gc.collect()
+            trim_result = trim(0)
+            released = bool(trim_result)
+            after = collect_memory_snapshot()
+            duration_ms = (time.perf_counter() - started) * 1000
+            _trim_call_count += 1
+            if released and _should_log_trim(
+                force=force,
+                log_every_n=log_every_n,
+                call_count=_trim_call_count,
+                before=before,
+                after=after,
+                info_log_min_delta_mb=info_log_min_delta_mb,
+            ):
+                logger.info(
+                    "memory trim: reason=%s malloc_trim=%s rss_kib=%s->%s "
+                    "rss_anon_kib=%s->%s threads=%s duration_ms=%.1f",
+                    reason or "cleanup",
+                    trim_result,
+                    before.get("rss_kib"),
+                    after.get("rss_kib"),
+                    before.get("rss_anon_kib"),
+                    after.get("rss_anon_kib"),
+                    after.get("thread_count"),
+                    duration_ms,
+                )
+            return released
+        except Exception as exc:
+            logger.warning(
+                "memory trim failed after %s: %s: %s",
+                reason or "cleanup",
+                type(exc).__name__,
+                exc,
+            )
+            return False

run_agent.py

@@ -4131,6 +4131,15 @@ class AIAgent:
         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/gateway/test_memory_trim_housekeeping.py

@@ -0,0 +1,32 @@
+"""Memory-trim coverage for the long-lived messaging gateway housekeeper."""
+
+import gateway.run as gateway_run
+
+
+class _OneTickStopEvent:
+    """Run one housekeeping tick without a sleep or background thread."""
+
+    def __init__(self):
+        self.waited = False
+
+    def is_set(self):
+        return self.waited
+
+    def wait(self, timeout=None):
+        self.waited = True
+        return True
+
+
+def test_gateway_housekeeping_calls_periodic_memory_trim(monkeypatch):
+    import hermes_cli.mem_trim as mem_trim
+
+    calls = []
+    monkeypatch.setattr(
+        mem_trim,
+        "trim_memory",
+        lambda **kwargs: calls.append(kwargs) or True,
+    )
+
+    gateway_run._start_gateway_housekeeping(_OneTickStopEvent(), interval=0)
+
+    assert calls == [{"reason": "messaging gateway housekeeping"}]

tests/hermes_cli/test_mem_trim.py

@@ -0,0 +1,180 @@
+"""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)
+    monkeypatch.setattr(mem_trim, "_trim_call_count", 0)
+
+
+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_from_config_file(monkeypatch, tmp_path):
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir()
+    (hermes_home / "config.yaml").write_text(
+        "context:\n  memory_trim:\n    enabled: false\n",
+        encoding="utf-8",
+    )
+    trim = Mock(return_value=1)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
+    token = set_hermes_home_override(hermes_home)
+
+    try:
+        assert mem_trim.trim_memory(force=True) is False
+        trim.assert_not_called()
+    finally:
+        reset_hermes_home_override(token)
+
+
+def test_default_config_declares_memory_trim_controls():
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    context = DEFAULT_CONFIG["context"]
+    assert isinstance(context, dict)
+    settings = context["memory_trim"]
+    assert isinstance(settings, dict)
+    assert isinstance(settings["enabled"], bool)
+    assert isinstance(settings["cooldown_seconds"], float)
+
+
+def test_collect_memory_snapshot_parses_linux_proc_status(monkeypatch):
+    monkeypatch.setattr(mem_trim.sys, "platform", "linux")
+    monkeypatch.setattr(
+        mem_trim,
+        "_read_proc_status",
+        lambda: "Name:\tpython\nVmRSS:\t1234 kB\nRssAnon:\t567 kB\n",
+    )
+    monkeypatch.setattr(mem_trim.threading, "active_count", lambda: 9)
+
+    assert mem_trim.collect_memory_snapshot(history_bytes=42) == {
+        "rss_kib": 1234,
+        "rss_anon_kib": 567,
+        "thread_count": 9,
+        "history_bytes": 42,
+    }
+
+
+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_success_logs_memory_snapshot_and_trim_result(monkeypatch, caplog):
+    monkeypatch.setattr(mem_trim.gc, "collect", lambda: None)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", lambda _pad: 1)
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+    snapshots = iter(
+        (
+            {"rss_kib": 4096, "rss_anon_kib": 3072, "thread_count": 3},
+            {"rss_kib": 2048, "rss_anon_kib": 1024, "thread_count": 3},
+        )
+    )
+    monkeypatch.setattr(mem_trim, "collect_memory_snapshot", lambda: next(snapshots))
+
+    with caplog.at_level("INFO", logger="hermes_cli.mem_trim"):
+        assert mem_trim.trim_memory(reason="test turn") is True
+
+    assert "reason=test turn" in caplog.text
+    assert "malloc_trim=1" in caplog.text
+    assert "rss_kib=4096->2048" in caplog.text
+
+
+def test_force_logs_even_when_periodic_log_sampling_skips(monkeypatch, caplog):
+    monkeypatch.setattr(mem_trim.gc, "collect", lambda: None)
+    monkeypatch.setattr(mem_trim, "_malloc_trim", lambda _pad: 1)
+    monkeypatch.setattr(mem_trim, "_config_settings", lambda: (True, 0.0, 99, 1.0))
+    monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
+    monkeypatch.setattr(
+        mem_trim,
+        "collect_memory_snapshot",
+        lambda: {"rss_kib": 4096, "rss_anon_kib": 3072, "thread_count": 3},
+    )
+
+    with caplog.at_level("INFO", logger="hermes_cli.mem_trim"):
+        assert mem_trim.trim_memory(reason="periodic") is True
+        assert mem_trim.trim_memory(force=True, reason="close") is True
+
+    messages = [record.getMessage() for record in caplog.records]
+    assert not any("reason=periodic" in message for message in messages)
+    assert any("reason=close" in message for message in messages)
+
+
+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

tests/test_tui_gateway_server.py

@@ -14150,6 +14150,49 @@ def test_reap_idle_sessions_closes_only_evictable(monkeypatch):
         server._sessions.clear()
 
 
+def test_reap_idle_sessions_calls_periodic_trim(monkeypatch):
+    """The idle reaper must call trim_memory every scan, even with no victims."""
+    trim_calls = []
+    monkeypatch.setattr(server, "_session_pending_kind", lambda sid: "")
+    monkeypatch.setattr(server, "_close_session_by_id", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_enforce_session_cap", lambda: None)
+    monkeypatch.setattr(server, "_reclaim_orphaned_leases", lambda: None)
+
+    # Patch the delayed import path: the function does
+    # `from hermes_cli.mem_trim import trim_memory` at call time.
+    import hermes_cli.mem_trim as mem_trim
+
+    monkeypatch.setattr(
+        mem_trim, "trim_memory",
+        lambda **kw: trim_calls.append(kw.get("reason", "")) or True,
+    )
+
+    server._sessions.clear()
+    try:
+        server._reap_idle_sessions()
+        assert len(trim_calls) == 1
+        assert trim_calls[0] == "idle reaper periodic trim"
+    finally:
+        server._sessions.clear()
+
+
+def test_reap_idle_sessions_logs_trim_failure(monkeypatch, caplog):
+    monkeypatch.setattr(server, "_session_pending_kind", lambda sid: "")
+    monkeypatch.setattr(server, "_close_session_by_id", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_enforce_session_cap", lambda: None)
+    monkeypatch.setattr(server, "_reclaim_orphaned_leases", lambda: None)
+    import hermes_cli.mem_trim as mem_trim
+
+    monkeypatch.setattr(mem_trim, "trim_memory", lambda **_kw: (_ for _ in ()).throw(RuntimeError("boom")))
+    server._sessions.clear()
+    try:
+        with caplog.at_level("WARNING", logger="tui_gateway.server"):
+            server._reap_idle_sessions()
+        assert "idle reaper memory trim failed: RuntimeError: boom" in caplog.text
+    finally:
+        server._sessions.clear()
+
+
 def test_session_create_records_ui_model_as_session_override(monkeypatch):
     """The desktop composer owns its model as plain UI state and ships it on
     session.create. The gateway must record it as a PER-SESSION override (built
@@ -15536,3 +15579,70 @@ def test_prompt_submit_passes_persist_user_message_to_agent(monkeypatch):
         assert captured.get("persist_user_message") == "hi"
     finally:
         server._sessions.pop("sid", None)
+
+
+def test_prompt_submit_releases_old_history_before_heap_trim(monkeypatch):
+    """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"] = "/tmp/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

@@ -1054,6 +1054,19 @@ def _reap_idle_sessions() -> None:
         _close_session_by_id(sid, end_reason="idle_timeout")
     _enforce_session_cap()
     _reclaim_orphaned_leases()
+    # Periodic heap release for long-lived gateway processes.  Even when no
+    # session is reaped, Python's generational GC rarely runs gen2 collection
+    # under steady-state allocation, and glibc retains freed pages as RSS.
+    # Calling trim_memory here ensures every reaper scan (default every 5 min)
+    # returns releasable pages, preventing unbounded RSS growth over days/weeks.
+    try:
+        from hermes_cli.mem_trim import trim_memory
+
+        trim_memory(reason="idle reaper periodic trim")
+    except Exception as exc:
+        logger.warning(
+            "idle reaper memory trim failed: %s: %s", type(exc).__name__, exc
+        )
 
 
 def _reclaim_orphaned_leases() -> None:
@@ -9826,6 +9839,23 @@ def _run_prompt_submit(
                 )
                 _emit("error", sid, {"message": str(e)})
         finally:
+            # 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)
+
             if thinking_started:
                 # Kill the ambient thinking sound the moment the turn ends —
                 # error and success paths both land here.

tui_gateway/slash_worker.py

@@ -20,6 +20,7 @@ import argparse
 import contextlib
 import io
 import json
+import logging
 import os
 import sys
 import threading
@@ -48,6 +49,7 @@ def _env_float(name: str, default: float) -> float:
 _WATCHDOG_POLL_S = max(0.05, _env_float("HERMES_SLASH_WATCHDOG_POLL_S", 2.0))
 _ORPHAN_GRACE_S = max(0.0, _env_float("HERMES_SLASH_WATCHDOG_GRACE_S", 5.0))
 _in_flight = threading.Event()  # set while a command is executing
+logger = logging.getLogger(__name__)
 
 
 def _is_orphaned(original_ppid, getppid=os.getppid) -> bool:
@@ -173,6 +175,19 @@ def main():
             sys.stdout.flush()
         finally:
             _in_flight.clear()
+            # Workers persist for the TUI session, so release allocator pages at
+            # the same command boundary as other long-lived gateway processes.
+            # trim_memory's shared cooldown coalesces this with nearby activity.
+            try:
+                from hermes_cli.mem_trim import trim_memory
+
+                trim_memory(reason="slash worker command completion")
+            except Exception as exc:
+                logger.warning(
+                    "slash worker memory trim failed: %s: %s",
+                    type(exc).__name__,
+                    exc,
+                )
 
 
 if __name__ == "__main__":