line:memory-trim-policy: restore RSS low-water and GC cooldown policy

docs/research/memory-runtime-issue-110.md

@@ -0,0 +1,66 @@
+# Memory Runtime reconstruction preflight — issue #110
+
+Date: 2026-08-17
+
+## Authority / pinned refs
+
+- Phase-2 ticket: #110 (`repair:memory-runtime`).
+- Composition authority: PR #108 at `5aa4f4e27ccf2169beb4fc1f1d1eeb655d13b548` — `line:memory-trim-diagnostics EXTENDS line:memory-trim-policy`; diagnostics must fail open.
+- Phase-1 accounting: PR #104 at `f81cd921a89516d855b5b69906ce99e6351bc741`.
+- Frozen fork evidence: current `dev` at preflight time `fa5ed679cc6559c619038f327e6276f4b7e8d735`.
+- Reconstruction substrate: upstream `NousResearch/hermes-agent` `main` at `3b9a963b8e5cdb804a422755bed9a60fcd778273`.
+
+## Upstream prior art / current authority
+
+| Upstream work | State at preflight | Use here |
+|---|---|---|
+| #76905 — config-driven allocator trim with telemetry | **merged** | Current authority. Reuse `hermes_cli.mem_trim.trim_memory`, config loading, glibc probe, lifecycle wiring, cooldown and force-floor seams. |
+| #77356 — post-compression trim | **merged** | Confirms compression should call the same `trim_memory()` seam rather than introduce a second GC path. |
+| #81127 — gateway agent-cache memory-pressure bound | **merged** | Adjacent memory-pressure policy; it calls the current trim seam after releasing cache references. It does not implement #110's low-water/GC-cooldown contract. |
+| #66355 — earlier allocator trim consolidation | closed, unmerged | Superseded by #76905; provenance/design evidence only. |
+| #64591 — periodic idle-reaper trim | closed, unmerged | Absorbed by the later consolidated trim implementation; do not replay its branch shape. |
+| #63708 — earlier config-driven trim | currently open, unmerged | Superseded semantically by merged #76905 even though the old PR is currently open; not current authority. |
+| #46022 — proactive GC above 400 MB | closed, unmerged | Rejected shape for this reconstruction. Review identified its RSS source as a high-water mark, which could retrigger full GC forever after one crossing. |
+| #80974 — GC after large tool results | currently open, unmerged | Design evidence only. It is not current-main authority and is outside the `trim_memory()` housekeeping contract reconstructed here. |
+
+## Fork residual contract
+
+Phase-1 provenance identifies two residual capabilities:
+
+1. `capability:memory-trim-policy`
+   - `threshold_mb` is an RSS low-water gate for non-forced housekeeping work.
+   - `gc.collect()` has an independent `gc_cooldown_seconds` (historical default 300 s).
+   - `malloc_trim(0)` remains eligible on the normal trim cadence even when GC is cooling down.
+   - force and invalid configuration paths remain safe.
+2. `capability:memory-trim-diagnostics`
+   - log GC and allocator-trim cost separately (`gc_ms`, `trim_ms`).
+   - expose `VmSwap` and pre-trim glibc fragmentation evidence where supported.
+   - diagnostics are best-effort and must never be required for recovery.
+
+Historical evidence: fork commits `04f1af72be078cef69de538f1519f93a73088b0d` and `a9d2b9af4f800fef23fa7ecaf2ea270b43e326eb`.
+
+## Reconstruction decisions
+
+This is a conscious port onto current upstream, not a cherry-pick of the old fork implementation.
+
+- Keep the merged #76905 `trim_memory()` seam and all current lifecycle callers unchanged.
+- Use current `/proc/self/status` `VmRSS` as the low-water signal. If current RSS is unavailable, fail open and keep recovery eligible; do **not** treat missing RSS as zero.
+- A low-water skip is not a trim attempt and therefore must not consume the normal cooldown or the 5-second forced-close floor.
+- Keep GC state separate from allocator-trim state. A cooling-down GC does not suppress `malloc_trim(0)`.
+- Coerce `gc_cooldown_seconds` with its own 300-second fallback rather than accidentally falling back to the allocator trim's 60-second default.
+- Preserve `force=True` as a bypass of the RSS gate and GC cooldown while retaining upstream's burst-close force floor.
+- Rebuild fragmentation collection as best-effort glibc instrumentation without the historical predictable `/tmp/hermes_malloc_info_<pid>.xml` pathname. Diagnostic failure returns no diagnostic data and cannot veto trim policy.
+- Read `malloc_info` pre-trim so fragmentation describes the state that motivated recovery.
+
+## Acceptance mapping
+
+Tests for this feature line must cover:
+
+- below-threshold skip and unavailable-RSS fail-open behavior;
+- low-water skips not suppressing a following forced trim;
+- independent GC cooldown with allocator trim still running;
+- force bypass of the low-water and GC-cooldown gates;
+- invalid policy values falling back safely;
+- `VmSwap` parsing;
+- GC-vs-trim timing in the operator-visible log;
+- fragmentation parsing/collection and diagnostic failure not blocking trim.

hermes_cli/mem_trim.py

@@ -1,8 +1,13 @@
 """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.
+allocations to the OS. Other platforms and allocators are safe no-ops.
 Behavior is configured under ``context.memory_trim`` in ``config.yaml``.
+
+The allocator trim and Python cyclic GC intentionally have independent cadence:
+``malloc_trim(0)`` is cheap and may run at the normal trim cadence, while
+``gc.collect()`` is separately rate-limited because a full collection can be
+orders of magnitude more expensive in a large or swapped process.
 """
 
 from __future__ import annotations
@@ -21,21 +26,26 @@
 logger = logging.getLogger(__name__)
 
 _DEFAULT_COOLDOWN_SECONDS = 60.0
+_DEFAULT_GC_COOLDOWN_SECONDS = 300.0
 _DEFAULT_LOG_EVERY_N = 1
 _DEFAULT_INFO_LOG_MIN_DELTA_MB = 0.0
+_DEFAULT_THRESHOLD_MB = None
 _trim_lock = threading.Lock()
 _last_trim_monotonic = 0.0
+_last_gc_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]:
+def _config_settings() -> tuple[bool, float, float, int, float, float | None]:
     """Return fail-open settings from the normal Hermes config path."""
     enabled = True
     cooldown: Any = _DEFAULT_COOLDOWN_SECONDS
+    gc_cooldown: Any = _DEFAULT_GC_COOLDOWN_SECONDS
     log_every_n: Any = _DEFAULT_LOG_EVERY_N
     info_log_min_delta_mb: Any = _DEFAULT_INFO_LOG_MIN_DELTA_MB
+    threshold_mb: Any = _DEFAULT_THRESHOLD_MB
     try:
         # Read-only access: settings are only .get()ed and coerced, never
         # mutated — use the no-deepcopy variant. This runs on EVERY trim
@@ -52,27 +62,46 @@ def _config_settings() -> tuple[bool, float, int, float]:
             if isinstance(configured_enabled, bool):
                 enabled = configured_enabled
             cooldown = settings.get("cooldown_seconds", _DEFAULT_COOLDOWN_SECONDS)
+            gc_cooldown = settings.get(
+                "gc_cooldown_seconds", _DEFAULT_GC_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
             )
+            threshold_mb = settings.get("threshold_mb", _DEFAULT_THRESHOLD_MB)
     except Exception:
         pass
     return (
         enabled,
         _cooldown_seconds(cooldown),
+        _cooldown_seconds(gc_cooldown, default=_DEFAULT_GC_COOLDOWN_SECONDS),
         _log_every_n(log_every_n),
         _nonnegative_float(info_log_min_delta_mb, _DEFAULT_INFO_LOG_MIN_DELTA_MB),
+        _threshold_mb(threshold_mb),
     )
 
 
-def _cooldown_seconds(value: Any) -> float:
+def _cooldown_seconds(
+    value: Any, *, default: float = _DEFAULT_COOLDOWN_SECONDS
+) -> float:
     if isinstance(value, bool):
-        return _DEFAULT_COOLDOWN_SECONDS
+        return default
     try:
         return max(0.0, float(value))
     except (TypeError, ValueError):
-        return _DEFAULT_COOLDOWN_SECONDS
+        return default
+
+
+def _threshold_mb(value: Any) -> float | None:
+    """Coerce the RSS low-water mark; invalid/non-positive values disable it."""
+    if isinstance(value, bool) or value is None:
+        return None
+    try:
+        threshold = float(value)
+    except (TypeError, ValueError):
+        return None
+    return threshold if threshold > 0 else None
 
 
 def _log_every_n(value: Any) -> int:
@@ -106,7 +135,7 @@ def _read_proc_status() -> str | 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
+    ``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] = {
@@ -122,15 +151,22 @@ def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int |
                 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])
+                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,
+    *,
+    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.
@@ -172,53 +208,85 @@ def trim_memory(
     reason: str = "",
     cooldown_seconds: float | None = None,
 ) -> bool:
-    """Collect cycles and ask glibc to release free heap pages.
+    """Collect cycles when eligible 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.
+    Unsupported allocators, the config kill switch, cooldown suppression, the
+    RSS low-water gate, and all runtime errors return ``False`` without
+    affecting the caller.
     """
     (
         enabled,
         configured_cooldown,
+        gc_cooldown,
         log_every_n,
         info_log_min_delta_mb,
+        threshold_mb,
     ) = _config_settings()
     if not enabled:
         return False
 
-    global _last_trim_monotonic, _trim_call_count
+    global _last_trim_monotonic, _last_gc_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:
+        if (
+            not force
+            and _last_trim_monotonic
+            and now - _last_trim_monotonic < cooldown
+        ):
             return False
+
         # Even forced trims honor a short floor: AIAgent.close() forces a trim,
         # and delegate batches close N child subagents back-to-back in the SAME
-        # process — without a floor that stacks N+1 uncooled full gc.collect()
-        # passes (50-500ms each in a large gateway process). 5s coalesces the
-        # burst while keeping the parent's final close-trim effective.
+        # process. The floor coalesces that burst while keeping a later parent
+        # close effective.
         _FORCE_FLOOR_SECONDS = 5.0
         if (
             force
             and _last_trim_monotonic
             and now - _last_trim_monotonic < _FORCE_FLOOR_SECONDS
         ):
             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()
+
+            # A missing RSS sample cannot prove that the process is below the
+            # low-water mark. Fail open so platform/telemetry unavailability
+            # never disables memory recovery.
+            current_rss_kib = before.get("rss_kib")
+            if (
+                not force
+                and threshold_mb is not None
+                and current_rss_kib is not None
+                and current_rss_kib < threshold_mb * 1024
+            ):
+                # This was only a cheap eligibility check, not a trim attempt.
+                # Do not consume the normal cooldown or the forced-close floor.
+                return False
+
+            # From here on an actual recovery attempt is eligible. Record it
+            # before expensive work so malloc_trim failures remain rate-limited.
+            _last_trim_monotonic = now
+
+            should_gc = (
+                force
+                or not _last_gc_monotonic
+                or now - _last_gc_monotonic >= gc_cooldown
+            )
             started = time.perf_counter()
-            gc.collect()
+            if should_gc:
+                gc.collect()
+                _last_gc_monotonic = time.monotonic()
             trim_result = trim(0)
             released = bool(trim_result)
             after = collect_memory_snapshot()