fix(viz): evict idle LogStream entries without requiring a later release

LogStreamRegistry only ran _sweep_idle_streams_locked() from inside
release(), so a stream whose refcount dropped to zero without EOF
stayed resident for the process lifetime whenever no subsequent
release() fired after the idle TTL. Long-lived dashboards with
low-churn traffic leaked every LogStream that ever had a one-off
disconnect, together with its retained event buffer.

Piggyback the sweep onto acquire() and streams_in_cache_dir() too.
New SSE connections anywhere on the dashboard reclaim stale
retention deques from unrelated sessions, and the cache watcher
callback (which calls streams_in_cache_dir on every observed write)
keeps eviction driven by ongoing activity instead of only by
balanced refcount transitions. Add a streaming regression that
rewinds _idle_since and verifies both call-sites evict idle streams.

Author: SihaoLiu

tests/test-streaming.sh

@@ -471,6 +471,54 @@ else
     _fail "out-of-window reconnect wrong: $OUTPUT"
 fi
 
+# ─── Idle stream eviction without follow-up release ───
+# Regression: a stream whose refcount drops to zero without EOF should
+# not survive forever when no subsequent release() ever fires. Sweeps
+# must also run on other registry interactions (acquire,
+# streams_in_cache_dir) so idle retention deques are reclaimed under
+# low-churn traffic.
+SWEEPLOG="$CACHE_DIR/round-9-codex-run.log"
+: > "$SWEEPLOG"
+
+OUTPUT="$(_run_py "
+import time
+from log_streamer import LogStreamRegistry
+# Use a non-zero TTL and rewind the recorded idle timestamp so the
+# next sweep observes the TTL as elapsed without real waiting. Reaching
+# into a private dict is acceptable in a white-box regression test:
+# the point is to verify which call-sites sweep, not real-time timing.
+reg = LogStreamRegistry(idle_ttl_seconds=60.0)
+# Stream A: one-off disconnect, no EOF, no further release on the same key.
+reg.acquire('$CACHE_DIR', 'sid-sweep-A', 'round-9-codex-run.log')
+reg.release('sid-sweep-A', 'round-9-codex-run.log')
+print('A_PRESENT_BEFORE_SWEEP:', ('sid-sweep-A', 'round-9-codex-run.log') in reg)
+# Force A's idle_since far in the past so any subsequent sweep evicts it.
+reg._idle_since[('sid-sweep-A', 'round-9-codex-run.log')] = time.monotonic() - 1e6
+# New acquire on a different session must trigger the sweep.
+reg.acquire('$CACHE_DIR', 'sid-sweep-B', 'round-9-codex-run.log')
+print('A_EVICTED_BY_ACQUIRE:', ('sid-sweep-A', 'round-9-codex-run.log') not in reg)
+print('B_PRESENT:', ('sid-sweep-B', 'round-9-codex-run.log') in reg)
+
+# Independent registry: verify streams_in_cache_dir() (invoked by the
+# cache watcher callback on every observed write) also evicts idle
+# streams even when no release() follows.
+reg2 = LogStreamRegistry(idle_ttl_seconds=60.0)
+reg2.acquire('$CACHE_DIR', 'sid-sweep-C', 'round-9-codex-run.log')
+reg2.release('sid-sweep-C', 'round-9-codex-run.log')
+reg2._idle_since[('sid-sweep-C', 'round-9-codex-run.log')] = time.monotonic() - 1e6
+_ = reg2.streams_in_cache_dir('$CACHE_DIR', 'round-9-codex-run.log')
+print('C_EVICTED_BY_STREAMS_LOOKUP:', ('sid-sweep-C', 'round-9-codex-run.log') not in reg2)
+")"
+
+if grep -q '^A_PRESENT_BEFORE_SWEEP: True$' <<<"$OUTPUT" && \
+   grep -q '^A_EVICTED_BY_ACQUIRE: True$' <<<"$OUTPUT" && \
+   grep -q '^B_PRESENT: True$' <<<"$OUTPUT" && \
+   grep -q '^C_EVICTED_BY_STREAMS_LOOKUP: True$' <<<"$OUTPUT"; then
+    _pass "idle streams are evicted by acquire() and streams_in_cache_dir(), not only by a follow-up release()"
+else
+    _fail "idle-stream sweep regression: $OUTPUT"
+fi
+
 # ─── Summary ───
 echo
 echo "========================================"

viz/server/log_streamer.py

@@ -398,6 +398,13 @@ def acquire(self, cache_dir: str, session_id: str, basename: str) -> LogStream:
         """
         key = (session_id, basename)
         with self._lock:
+            # Every new acquire is also a chance to drop OTHER entries
+            # whose idle TTL has elapsed without a follow-up release.
+            # Without this, a refcount=0 stream that is never released
+            # again (one-off disconnect on a long-lived session) would
+            # stay resident for the process lifetime and leak its
+            # retention deque.
+            self._sweep_idle_streams_locked()
             stream = self._streams.get(key)
             if stream is None:
                 stream = LogStream(cache_dir, basename)
@@ -475,6 +482,12 @@ def get(self, session_id: str, basename: str) -> Optional[LogStream]:
     def streams_in_cache_dir(self, cache_dir: str, basename: str) -> List[LogStream]:
         """Return all streams that observe a specific cache file."""
         with self._lock:
+            # Piggyback a sweep: this method is invoked from the cache
+            # watcher callback on every observed write, so leveraging
+            # it keeps idle eviction driven by ongoing activity rather
+            # than only by the next ``release()`` call, which may
+            # never happen on long-lived dashboards with low churn.
+            self._sweep_idle_streams_locked()
             return [
                 s for s in self._streams.values()
                 if s.cache_dir == cache_dir and s.basename == basename