From 2cf25ce07844bab3b3daf64af8cf603a64e2d33a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:47:05 -0800 Subject: [PATCH] fix(mem): a finished turn's lease and timers free by refcount, not only by a full GC Each turn's DurableTurnLease kept owner -> scheduler handle -> bound method -> owner and lease <-> liveness-watchdog cycles (delegate heartbeats too), and those cycles held the agent. Measured on a scratch gateway after 18 delegation turns and a 240 s idle: 17 closed AIAgents (each with its OpenAI client and compressor) were reachable only through these cycles. With the timer trims no longer forcing a full collection, Python 3.14's incremental collector (which only advances on allocation) left them resident on a quiet gateway. A cancelled or stopped periodic handle now drops its callback and context, and join_threads drops the lease's watchdog and handles. --- agent/periodic_scheduler.py | 22 ++++-- agent/turn_facade_lease.py | 2 + .../test_turn_lease_frees_by_refcount.py | 75 +++++++++++++++++++ .../gateway/test_housekeeping_idle_collect.py | 75 +++++++++++++++++++ 4 files changed, 167 insertions(+), 7 deletions(-) create mode 100644 tests/agent/test_turn_lease_frees_by_refcount.py create mode 100644 tests/gateway/test_housekeeping_idle_collect.py diff --git a/agent/periodic_scheduler.py b/agent/periodic_scheduler.py index 421f93055f..db07d30583 100644 --- a/agent/periodic_scheduler.py +++ b/agent/periodic_scheduler.py @@ -21,7 +21,7 @@ failure never retires the handle: it is re-queued and logged at warning. from __future__ import annotations -from contextvars import copy_context +from contextvars import Context, copy_context import heapq import itertools import logging @@ -79,7 +79,7 @@ class PeriodicScheduler: def _cancel(self, handle: ScheduledHandle, wait: Optional[float]) -> None: with self._cond: - handle._cancelled = True + self._retire(handle) self._cond.notify() runner = handle._runner if wait and runner is not None and threading.current_thread() is not runner: @@ -90,7 +90,7 @@ class PeriodicScheduler: so ``cancel`` can never observe a half-set runner.""" runner = threading.Thread( target=self._run_callback, - args=(handle,), + args=(handle, handle._fn, handle._context), name=f"{_CALLBACK_THREAD_PREFIX}-{id(handle):x}", daemon=True, ) @@ -109,21 +109,29 @@ class PeriodicScheduler: self._requeue(handle) self._cond.notify() + @staticmethod + def _retire(handle: ScheduledHandle) -> None: + """Mark cancelled and drop the callback (``_cond`` held). The callback is usually a bound method of + the handle's owner, which keeps the handle: an owner -> handle -> method -> owner cycle that outlives + the turn and, on a quiet gateway, pins the whole agent until a full GC.""" + handle._cancelled = True + handle._fn = handle._context = None + def _requeue(self, handle: ScheduledHandle) -> None: """Push ``handle``'s next due time (``_cond`` held).""" heapq.heappush(self._heap, (time.monotonic() + handle._interval, next(self._seq), handle)) - def _run_callback(self, handle: ScheduledHandle) -> None: + def _run_callback(self, handle: ScheduledHandle, fn: Callable[[], object], context: Context) -> None: stop = False try: - stop = handle._context.run(handle._fn) is False + stop = context.run(fn) is False except Exception: - logger.debug("periodic callback %r raised", handle._fn, exc_info=True) + logger.debug("periodic callback %r raised", fn, exc_info=True) finally: with self._cond: handle._runner = None if stop: - handle._cancelled = True + self._retire(handle) elif not handle._cancelled: self._requeue(handle) self._cond.notify() diff --git a/agent/turn_facade_lease.py b/agent/turn_facade_lease.py index 82b3673834..093add37c8 100644 --- a/agent/turn_facade_lease.py +++ b/agent/turn_facade_lease.py @@ -109,6 +109,8 @@ class DurableTurnLease: in-flight tick finishes before ``clear_interrupt`` runs.""" for handle in self.timer_handles: handle.cancel(wait=timeout) + # The watchdog holds this lease's bound methods: drop it so the finished turn frees by refcount. + self.timer_handles, self.watchdog = [], None def release(self) -> None: """Release the row and drop the agent's holder attrs (only if they still name this lease).""" diff --git a/tests/agent/test_turn_lease_frees_by_refcount.py b/tests/agent/test_turn_lease_frees_by_refcount.py new file mode 100644 index 0000000000..48b651264f --- /dev/null +++ b/tests/agent/test_turn_lease_frees_by_refcount.py @@ -0,0 +1,75 @@ +"""A finished turn's lease and timers free by reference counting, not only by a full GC. + +The gateway's timer trims no longer force a full ``gc.collect()`` (it froze every session for +250-750 ms). Python 3.14's incremental collector advances only on new allocations, so on a quiet +gateway anything held by a reference cycle stays resident: each turn's lease <-> watchdog and +owner -> scheduler handle -> bound method -> owner cycles used to pin the whole agent (history, +client, compressor) after it closed. +""" + +import gc +import threading +import weakref + +from agent.periodic_scheduler import schedule +from agent.turn_facade_lease import DurableTurnLease + + +class _Db: + def refresh_session_turn_lease(self, session_id, holder, **kwargs): + return True + + def release_session_turn_lease(self, session_id, holder): + pass + + +class _Agent: + def __init__(self): + self.session_id = "s1" + self._session_turn_lease_refresh_interval = 60.0 + self._active_session_turn_lease_holder = None + self._lock = threading.Lock() + + def _touch_activity(self, *args, **kwargs): + pass + + def _liveness_activity_lock(self): + return self._lock + + +def test_finished_turn_lease_does_not_keep_the_agent_alive_without_gc(): + gc.disable() + try: + agent = _Agent() + lease = DurableTurnLease(agent, _Db(), "s1", "h") + lease.build_threads() + assert lease.watchdog is not None # the liveness watchdog is on by default + lease.start() + lease.stop_refresher() + lease.join_threads() + lease.clear_interrupt() + lease.release() + agent_ref = weakref.ref(agent) + del lease, agent + assert agent_ref() is None + finally: + gc.enable() + + +def test_cancelled_handle_releases_its_owner_without_gc(): + class Owner: + handle = None + + def tick(self): + return None + + gc.disable() + try: + owner = Owner() + owner.handle = schedule(owner.tick, 3600) + owner.handle.cancel() + owner_ref = weakref.ref(owner) + del owner + assert owner_ref() is None + finally: + gc.enable() diff --git a/tests/gateway/test_housekeeping_idle_collect.py b/tests/gateway/test_housekeeping_idle_collect.py new file mode 100644 index 0000000000..8c5af788d8 --- /dev/null +++ b/tests/gateway/test_housekeeping_idle_collect.py @@ -0,0 +1,75 @@ +"""Timer-lane trims run the full collection only while nobody is waiting (#139763). + +The gateway housekeeping tick dropped its forced ``gc.collect()`` because a full pass over a +multi-million-object heap holds the GIL for 250-750 ms and froze every live session once a minute. +Python 3.14's incremental collector only advances on allocation, so a quiet gateway would then keep +closed agents' reference cycles (and their pages) forever. The tick collects once the process has +been turn-free for ``IDLE_COLLECT_AFTER_SECONDS``: once per quiet period, again after a later +agent close, and never while a turn runs. +""" + +import pytest + +from agent import review_idle_queue +from hermes_cli import mem_trim + + +@pytest.fixture +def clock(monkeypatch): + queue = review_idle_queue.ReviewIdleQueue() + now = [1000.0] + queue._now = lambda: now[0] + monkeypatch.setattr(review_idle_queue, "QUEUE", queue) + collects = [] + monkeypatch.setattr(mem_trim.gc, "collect", lambda *a: collects.append(now[0]) or 0) + monkeypatch.setattr(mem_trim, "_malloc_trim", lambda pad: 1) + monkeypatch.setattr(mem_trim, "_probe_done", True) + monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 0.0) + monkeypatch.setattr(mem_trim, "_config_settings", lambda: (True, 0.0, 1, 0.0)) + monkeypatch.setattr(mem_trim, "_idle_collected_for", None) + monkeypatch.setattr(mem_trim, "_agent_closed_since_idle_collect", False) + return queue, now, collects + + +def _tick(): + from gateway import run as gateway_run + gateway_run._housekeeping_memory_trim() + + +def test_housekeeping_collects_once_per_quiet_period_and_never_during_a_turn(clock): + queue, now, collects = clock + queue.note_turn_started() + now[0] += mem_trim.IDLE_COLLECT_AFTER_SECONDS * 3 + _tick() + assert collects == [] # a turn is live: no freeze, however long it runs + + queue.note_turn_finished() + now[0] += mem_trim.IDLE_COLLECT_AFTER_SECONDS - 1 + _tick() + assert collects == [] # quiet, but not for long enough + + now[0] += 2 + _tick() + _tick() + assert len(collects) == 1 # the quiet period's one collection + + mem_trim.note_agent_closed() # idle-cache eviction later in the same quiet period + _tick() + _tick() + assert len(collects) == 2 + + +def test_a_new_turn_starts_a_new_quiet_period(clock): + queue, now, collects = clock + queue.note_turn_started() + queue.note_turn_finished() + now[0] += mem_trim.IDLE_COLLECT_AFTER_SECONDS + _tick() + queue.note_turn_started() + queue.note_turn_finished() + now[0] += 1 + _tick() + assert len(collects) == 1 # just talked: the next collection waits for a full quiet window + now[0] += mem_trim.IDLE_COLLECT_AFTER_SECONDS + _tick() + assert len(collects) == 2