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

@@ -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()

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)."""

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()

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