perf(gateway): free an ended session's replay ring and counters

A gateway live session stamps its replay frames under SessionEvents' private key. Nothing
released that key when the session ended: a deleted/pruned session's ring and a closed one's
stayed in tui_gateway.event_replay until the shared 64-session FIFO cap evicted them, and
their seq/watermark counters stayed for the life of the process (#137087 covered only the
tui_gateway teardown path).

- event_replay.release_key purges a private key's ring and counters; a GC finalizer that
  cannot take the lock queues the key for the next lock holder.
- SessionEvents releases its key when the object goes (retired or replaced LiveSession).
- A settled last-viewer close releases the ring at once (the LiveSession stays resumable;
  a later publish/attach starts a new epoch, as after FIFO eviction). A plain detach keeps
  the ring, so reconnect replay is unchanged.

gateway/session_events.py

@@ -3,6 +3,7 @@ from copy import deepcopy
 import logging
 import threading
 import uuid
+import weakref
 
 from tui_gateway import event_replay
 from tui_gateway.transport import FanoutTransport
@@ -15,6 +16,9 @@ class SessionEvents:
         # A private lifetime key prevents two profiles (or legacy TUI owners)
         # with the same persisted ID from reading each other's replay.
         self._key = uuid.uuid4().hex
+        # Nothing but this object can name the key, so its ring and counters go when it does (a
+        # retired or replaced LiveSession), instead of waiting for the shared cache's FIFO cap.
+        weakref.finalize(self, event_replay.release_key, self._key, blocking=False).atexit = False
         self.epoch = uuid.uuid4().hex
         self.sequence = 0
         self._public_id = None
@@ -26,6 +30,12 @@ class SessionEvents:
         # peer whose bounded backlog overflowed, so it cannot linger looking subscribed.
         self.on_overflow = None
 
+    def release(self):
+        """The session ended with no viewer left (it may stay resumable): free its replay frames
+        now. A later publish or attach finds them gone and starts a new epoch, as after eviction."""
+        with self.lock:
+            event_replay.release_key(self._key)
+
     def watermark(self):
         with self.lock:
             if self.sequence and not event_replay.latest_seq(self._key):

gateway/session_viewer_close.py

@@ -316,6 +316,8 @@ async def settle_viewer_close(authority, session_id):
     from gateway.session_vault_prompt import release_unlocks
     release_unlocks(session_id)
     await _run_close_lifecycle(authority, live, target, platform, close.reason)
+    if not live.subscribers:  # ended: no viewer is left to replay to; a resume refetches history
+        live.event_stream.release()
 
 
 async def _resolve_abandoned_unknown(authority, session_id, live, close):

tests/gateway/test_session_events_release.py

@@ -0,0 +1,60 @@
+"""A gateway session's replay frames live only as long as the session does.
+
+Each live session's ``SessionEvents`` stamps its frames into ``tui_gateway.event_replay`` under a private
+lookup key. Before this, nothing released that key when the session ended: a closed or deleted session's
+ring stayed until the shared 64-session cap evicted it, and its counters stayed for the life of the process.
+A session that is only detached (resumable) keeps its ring, since a reconnect replays from it.
+"""
+import gc
+
+import pytest
+
+from tests.gateway.test_viewer_close_lifecycle import _call, _open, _owner
+
+
+def _recorded(key):
+    from tui_gateway import event_replay
+    return (key in event_replay._replay_buffers or key in event_replay._replay_next_seq
+            or key in event_replay._replay_evicted_through)
+
+
+@pytest.mark.asyncio
+async def test_closing_the_last_viewer_frees_the_ring_and_detach_keeps_it(tmp_path, monkeypatch):
+    db, authority, _calls = await _owner(tmp_path, monkeypatch)
+    viewer, subscription = await _open(db, authority, tmp_path)
+    stream = authority.sessions['chat'].event_stream
+    stream.publish('chat', {'text': 'hello'})
+    epoch, sequence = stream.watermark()
+
+    # Detached but resumable: the reconnect contract still replays from the ring.
+    await _call(viewer, 'session.detach', session_id='chat', subscription_id=subscription)
+    again = await _call(viewer, 'session.resume', session_id='chat')
+    replay = await _call(viewer, 'session.events.since', session_id='chat', replay_epoch=epoch, last_sequence=0)
+    assert not replay['snapshot_required'] and replay['count'] == sequence
+
+    await _call(viewer, 'session.detach', session_id='chat', subscription_id=again['subscription_id'],
+                end_reason='cli_close')
+    assert db.get_session('chat')['end_reason'] == 'cli_close'
+    assert not _recorded(stream._key)
+    # A later resume of the ended session refetches history instead of trusting a stale cursor.
+    await _call(viewer, 'session.resume', session_id='chat')
+    replay = await _call(viewer, 'session.events.since', session_id='chat', replay_epoch=epoch,
+                         last_sequence=sequence)
+    assert replay['snapshot_required']
+
+
+def test_a_dropped_session_stream_leaves_nothing_in_the_shared_cache():
+    from gateway.session_events import SessionEvents
+    from tui_gateway import event_replay
+
+    kept, dropped = SessionEvents(), SessionEvents()
+    for stream in (kept, dropped):
+        for n in range(3):
+            stream.publish('s', {'n': n})
+    key = dropped._key
+    del dropped, stream
+    gc.collect()
+    stats = event_replay.replay_stats()  # the next lock holder purges a finalizer's queued key
+    assert not _recorded(key)
+    assert stats['sessions'] >= 1
+    assert [frame['payload'] for frame in kept.since(kept.epoch, 0)['events']] == [{'n': 0}, {'n': 1}, {'n': 2}]

tui_gateway/event_replay.py

@@ -87,6 +87,7 @@ def _stamp_event(obj: dict) -> None:
     size = len(payload)
     with _replay_lock:
         global _replay_total_bytes
+        _drain_released_locked()
         seq = _replay_next_seq.get(sid, 0) + 1
         _replay_next_seq[sid] = seq
         params["seq"] = seq
@@ -147,6 +148,44 @@ def forget_session(sid: str) -> None:
         _drop_ring_locked(sid)
 
 
+# Private keys whose owner is gone (release_key from a GC finalizer that could not take the lock),
+# purged by the next lock holder. list.append/pop are atomic, so a finalizer never needs the lock.
+_released_keys: list[str] = []
+
+
+def _purge_key_locked(key: str) -> None:
+    global _replay_total_bytes
+    if _replay_buffers.pop(key, None) is not None:
+        _replay_total_bytes -= _replay_buffer_bytes.pop(key, 0)
+    _replay_evicted_through.pop(key, None)
+    _replay_next_seq.pop(key, None)
+    _replay_forgotten.discard(key)
+
+
+def _drain_released_locked() -> None:
+    while _released_keys:
+        _purge_key_locked(_released_keys.pop())
+
+
+def release_key(key: str, *, blocking: bool = True) -> None:
+    """Drop everything recorded under a PRIVATE lookup key (ring AND counters) once its session ended.
+
+    Unlike :func:`forget_session` (a public sid that late frames and #100122's monotonic numbering
+    still name), a private key is only ever stamped and read through its owner
+    (``gateway.session_events.SessionEvents``): that owner re-reads :func:`latest_seq` before every
+    publish and starts a new epoch when it finds 0, exactly as after FIFO eviction, so nothing needs
+    the old counters. ``blocking=False`` is for GC finalizers, which can run on a thread that already
+    holds ``_replay_lock``: the key is queued and the next lock holder purges it."""
+    if not key:
+        return
+    _released_keys.append(key)
+    if _replay_lock.acquire(blocking=blocking):
+        try:
+            _drain_released_locked()
+        finally:
+            _replay_lock.release()
+
+
 def events_since(sid: str, last_seen: int) -> list[dict]:
     """Recorded EVENT OBJECTS (each frame's ``params`` dict) with seq > last_seen for *sid*.
 
@@ -182,12 +221,14 @@ def reset_replay_state() -> None:
         _replay_evicted_through.clear()
         _replay_next_seq.clear()
         _replay_forgotten.clear()
+        _released_keys.clear()
         _replay_total_bytes = 0
 
 
 def replay_stats() -> dict:
     """Telemetry: buffer occupancy for the ops/debug surface."""
     with _replay_lock:
+        _drain_released_locked()
         return {
             "sessions": len(_replay_buffers),
             "events": sum(len(buffer) for buffer in _replay_buffers.values()),