From da43a8527b47f4cff6a1e6ab93eabb44febd435e Mon Sep 17 00:00:00 2001 From: Ryder Freeman Date: Fri, 17 Jul 2026 07:29:25 -0700 Subject: [PATCH] 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 | 16 ++ hermes_cli/config_defaults.py | 12 + hermes_cli/mem_trim.py | 238 ++++++++++++++++++ run_agent.py | 9 + .../gateway/test_memory_trim_housekeeping.py | 32 +++ tests/hermes_cli/test_mem_trim.py | 180 +++++++++++++ tests/test_tui_gateway_server.py | 110 ++++++++ tui_gateway/server.py | 30 +++ tui_gateway/slash_worker.py | 15 ++ 9 files changed, 642 insertions(+) create mode 100644 hermes_cli/mem_trim.py create mode 100644 tests/gateway/test_memory_trim_housekeeping.py create mode 100644 tests/hermes_cli/test_mem_trim.py diff --git a/gateway/run.py b/gateway/run.py index 2d870b2cd8..1a92da6beb 100644 --- a/gateway/run.py +++ b/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") diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 4545575db2..3181da0a36 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1587,6 +1587,18 @@ DEFAULT_CONFIG = { # a plugin in plugins/context_engine// 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 diff --git a/hermes_cli/mem_trim.py b/hermes_cli/mem_trim.py new file mode 100644 index 0000000000..ff15bea2ec --- /dev/null +++ b/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 diff --git a/run_agent.py b/run_agent.py index 072730f23d..0376c4b16c 100644 --- a/run_agent.py +++ b/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, diff --git a/tests/gateway/test_memory_trim_housekeeping.py b/tests/gateway/test_memory_trim_housekeeping.py new file mode 100644 index 0000000000..c8818f36b3 --- /dev/null +++ b/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"}] diff --git a/tests/hermes_cli/test_mem_trim.py b/tests/hermes_cli/test_mem_trim.py new file mode 100644 index 0000000000..bf6936022f --- /dev/null +++ b/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 diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index d8900b80ba..b2d8e918fa 100644 --- a/tests/test_tui_gateway_server.py +++ b/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) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 029d8ae723..f2e8fb6472 100644 --- a/tui_gateway/server.py +++ b/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. diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py index 5b75dbd165..85bbb5e598 100644 --- a/tui_gateway/slash_worker.py +++ b/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__":