diff --git a/gateway/agent_cache_pressure.py b/gateway/agent_cache_pressure.py index 7be52e1bc9260..91fa999b64d71 100644 --- a/gateway/agent_cache_pressure.py +++ b/gateway/agent_cache_pressure.py @@ -93,13 +93,34 @@ def _cgroup_limit_bytes() -> Optional[int]: Prefers cgroup v2 ``memory.high`` (the throttling point — passing it is what stalled the reported shutdown) over ``memory.max``, and falls back to cgroup v1. ``max`` / absurd sentinel values mean "unlimited". + + Checks the process's *own* cgroup first (where a systemd unit's + ``MemoryHigh=``/``MemoryMax=`` actually lands — the root files read + ``max`` on those deployments), then walks up to the root for + container-style limits. """ if sys.platform != "linux": return None - candidates = ( - "/sys/fs/cgroup/memory.high", - "/sys/fs/cgroup/memory.max", - "/sys/fs/cgroup/memory/memory.limit_in_bytes", + candidates: list[str] = [] + try: + from gateway.cgroup_cleanup import _own_cgroup_path + + own = _own_cgroup_path() + except Exception: + own = None + if own and own != "/": + candidates.extend( + ( + f"/sys/fs/cgroup{own}/memory.high", + f"/sys/fs/cgroup{own}/memory.max", + ) + ) + candidates.extend( + ( + "/sys/fs/cgroup/memory.high", + "/sys/fs/cgroup/memory.max", + "/sys/fs/cgroup/memory/memory.limit_in_bytes", + ) ) for candidate in candidates: try: @@ -179,7 +200,15 @@ def resolve_agent_cache_bounds(config: Any) -> AgentCacheBounds: max_evictions = _positive_int(section.get("max_evictions_per_pass")) protect_recent = section.get("protect_recent") protect_parsed = _positive_int(protect_recent) - if protect_parsed is None and protect_recent == 0: + if ( + protect_parsed is None + and isinstance(protect_recent, int) + and not isinstance(protect_recent, bool) + and protect_recent == 0 + ): + # 0 means "shed anything" — distinct from unset. The isinstance + # guards keep `protect_recent: false` (a YAML-typo bool, False == 0) + # on the default instead of silently disabling MRU protection. protect_parsed = 0 return AgentCacheBounds( diff --git a/gateway/run.py b/gateway/run.py index 4cc5773e80063..9a9c98cb74cbc 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -23747,6 +23747,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # persisted session JSON on the next turn, so dropping it here is safe. if hasattr(agent, "_session_messages"): agent._session_messages = [] + # _db_flush_scan_prefix is a shallow copy of the flushed transcript + # (run_agent.py, stamped on every successful flush) — it shares every + # message dict, so leaving it pins the multi-MB content strings the + # eviction exists to free. Pressure-evictable agents have flushed by + # definition, so this attribute is always populated on exactly the + # agents the memory valve targets. + if hasattr(agent, "_db_flush_scan_prefix"): + agent._db_flush_scan_prefix = None def _agent_cache_bounds(self): """Operator-configured agent-cache bounds, resolved once per process. @@ -23762,10 +23770,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew try: bounds = resolve_agent_cache_bounds(_load_gateway_config()) except Exception as _e: - from gateway.agent_cache_pressure import AgentCacheBounds - logger.debug("Agent cache bounds config read failed: %s", _e) - bounds = AgentCacheBounds() + # Resolve from an empty config rather than bare + # AgentCacheBounds(): the dataclass default has + # memory_high_mb=None (pressure pass OFF), but an *absent* + # config section means "auto" — a transient config read + # failure must not permanently disable the OOM valve this + # feature exists to provide. + bounds = resolve_agent_cache_bounds({}) self._agent_cache_bounds_cache = bounds return bounds @@ -23850,18 +23862,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not plan: _mid_turn = sum(1 for _, a in ordered if a is not None and id(a) in running_ids) + _unflushed = sum( + 1 + for _, a in ordered + if a is not None + and a is not _AGENT_PENDING_SENTINEL + and id(a) not in running_ids + and not transcript_persistence_caught_up(a) + ) logger.warning( "Agent cache pressure: anon RSS %dMB over budget %dMB but no " - "evictable session (%d cached, %d mid-turn) — memory will keep " - "climbing until those turns finish.", - rss_mb, bounds.memory_high_mb, len(ordered), _mid_turn, + "evictable session (%d cached, %d mid-turn, %d blocked on " + "un-flushed persistence)%s", + rss_mb, bounds.memory_high_mb, len(ordered), _mid_turn, _unflushed, + ( + " — transcripts are not reaching the session DB " + "(session persistence disabled or failing?); the memory " + "valve cannot shed sessions until they persist." + if _unflushed and not _mid_turn + else " — memory will keep climbing until those turns finish." + ), ) return 0 + evicted_count = len(plan) logger.warning( "Agent cache pressure: anon RSS %dMB over budget %dMB — evicting " "%d LRU session(s): %s", - rss_mb, bounds.memory_high_mb, len(plan), + rss_mb, bounds.memory_high_mb, evicted_count, ", ".join(key for key, _ in plan), ) try: @@ -23873,7 +23901,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ).start() except Exception: self._release_pressure_batch(plan) - return len(plan) + # NOTE: _release_pressure_batch drains `plan` in place (so the trim + # runs with no lingering agent references) — len(plan) is 0 by the + # time the daemon thread finishes, hence the pre-captured count. + return evicted_count def _release_pressure_batch(self, plan: List[tuple]) -> None: """Release a pressure-evicted batch, then return the heap to the OS. @@ -23883,12 +23914,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew memory, not to race N teardowns. The trailing ``malloc_trim`` is what turns "Python dropped the transcript" into "RSS actually fell" — without it glibc keeps the freed arenas and the cgroup never notices. + + The plan is drained (``pop`` + ``del``) rather than iterated so that + no local reference pins the evicted agents when ``gc.collect`` + + ``malloc_trim`` run — otherwise the trim frees almost nothing in this + pass, the next tick re-reads a still-high RSS, and the valve + over-evicts an extra batch of warm prompt caches every cycle. """ - for key, agent in plan: + while plan: + key, agent = plan.pop(0) # FIFO — evict LRU-first order preserved try: self._commit_then_release_soft(agent, key) except Exception as _e: logger.debug("Pressure release failed for %s: %s", key, _e) + del agent try: from hermes_cli.mem_trim import trim_memory diff --git a/tests/gateway/test_agent_cache_pressure.py b/tests/gateway/test_agent_cache_pressure.py index 52571a9b0d844..e77acaefa51d7 100644 --- a/tests/gateway/test_agent_cache_pressure.py +++ b/tests/gateway/test_agent_cache_pressure.py @@ -393,3 +393,120 @@ def _wait_for(predicate, timeout: float = 3.0) -> None: return _t.sleep(0.02) assert predicate(), "background release did not complete in time" + + +class TestSalvageFollowups: + """Follow-up behaviors added while salvaging PR #80795.""" + + def test_config_read_failure_still_resolves_auto_budget(self, monkeypatch): + """A transient config-read failure must not permanently disable the + pressure valve — the fallback resolves an empty config, whose absent + section means memory_high_mb='auto', not None.""" + import gateway.run as gw_run + from gateway.run import GatewayRunner + + monkeypatch.setattr( + gw_run, "_load_gateway_config", + lambda: (_ for _ in ()).throw(OSError("transient")), + ) + import gateway.agent_cache_pressure as acp + + monkeypatch.setattr(acp, "_cgroup_limit_bytes", lambda: 8 * 1024**3) + + runner = GatewayRunner.__new__(GatewayRunner) + bounds = runner._agent_cache_bounds() + assert bounds.memory_high_mb is not None, ( + "config-read failure fell back to a disabled valve — " + "the #80764 protection must survive a flaky config read" + ) + + def test_protect_recent_yaml_false_keeps_default(self): + """protect_recent: false (YAML-typo bool; False == 0) must keep the + default MRU protection, not silently disable it.""" + bounds = resolve_agent_cache_bounds( + {"agent": {"agent_cache": {"protect_recent": False}}} + ) + assert bounds.protect_recent > 0 + + def test_release_batch_drains_plan_before_trim(self, monkeypatch): + """The plan list must be empty when trim_memory runs, so no local + reference pins the evicted agents during gc.collect + malloc_trim + (otherwise the in-pass trim frees nothing and the next tick + over-evicts another batch).""" + from collections import OrderedDict as _OD + + from gateway.run import GatewayRunner + + runner = GatewayRunner.__new__(GatewayRunner) + released = [] + runner._commit_then_release_soft = lambda agent, key: released.append(key) + + plan_len_at_trim = {} + + import hermes_cli.mem_trim as mem_trim_mod + + plan = [(f"s{i}", MagicMock()) for i in range(3)] + + def fake_trim(force=False, reason=None): + plan_len_at_trim["len"] = len(plan) + return True + + monkeypatch.setattr(mem_trim_mod, "trim_memory", fake_trim) + + runner._release_pressure_batch(plan) + + assert sorted(released) == ["s0", "s1", "s2"] + assert plan_len_at_trim["len"] == 0, ( + "plan still held agent references when trim_memory ran" + ) + + def test_soft_release_clears_db_flush_scan_prefix(self): + """_db_flush_scan_prefix shallow-copies the flushed transcript and is + populated on exactly the agents the valve targets — leaving it pins + every message dict the eviction claims to free.""" + from gateway.run import GatewayRunner + + runner = GatewayRunner.__new__(GatewayRunner) + agent = MagicMock() + transcript = [{"role": "user", "content": "x" * 1024}] + agent._session_messages = transcript + agent._db_flush_scan_prefix = transcript[:] + + runner._release_evicted_agent_soft(agent) + + assert agent._session_messages == [] + assert agent._db_flush_scan_prefix is None + + def test_no_evictable_warning_distinguishes_unflushed_persistence(self, monkeypatch, caplog): + """When everything is blocked on un-flushed persistence (e.g. the + session DB never initialized), the warning must say so instead of + blaming mid-turn agents.""" + import logging as _logging + + from collections import OrderedDict as _OD + + import gateway.agent_cache_pressure as acp + from gateway.run import GatewayRunner + + runner = GatewayRunner.__new__(GatewayRunner) + runner._agent_cache = _OD() + runner._agent_cache_lock = threading.Lock() + runner._running_agents = {} + runner._agent_cache_bounds_cache = AgentCacheBounds( + memory_high_mb=1000, max_evictions_per_pass=8, protect_recent=0 + ) + monkeypatch.setattr(acp, "read_anon_rss_mb", lambda: 4000) + + for i in range(3): + agent = MagicMock() + agent._session_messages = [{"role": "user", "content": "x"}] + agent._last_flushed_db_idx = 0 # never flushed + runner._agent_cache[f"s{i}"] = (agent, "sig") + + with caplog.at_level(_logging.WARNING, logger="gateway.run"): + evicted = runner._sweep_agent_cache_under_pressure() + + assert evicted == 0 + joined = " ".join(r.getMessage() for r in caplog.records) + assert "blocked on un-flushed persistence" in joined + assert "3 blocked" in joined