fix: preserve streaming and replay safety after parent review

tests/test_tui_gateway_event_replay.py

@@ -204,3 +204,14 @@ def test_truncation_detection_semantics():
     assert event_replay.is_truncated("s1", 5)
     # Unknown session: nothing evicted, nothing truncated.
     assert not event_replay.is_truncated("nope", 0)
+
+
+def test_oversized_gap_watermark_never_moves_backwards(monkeypatch):
+    monkeypatch.setattr(event_replay, "_REPLAY_BUFFER_MAX", 1)
+    monkeypatch.setattr(event_replay, "_REPLAY_BUFFER_BYTES_MAX", 1000)
+    event_replay._stamp_event(_frame("s"))
+    large = _frame("s")
+    large["params"]["payload"] = {"data": "x" * 2000}
+    event_replay._stamp_event(large)
+    event_replay._stamp_event(_frame("s"))
+    assert event_replay.is_truncated("s", 1)

tui_gateway/event_replay.py

@@ -77,14 +77,14 @@ def _stamp_event(obj: dict) -> None:
             evicted_seq, _event, evicted_size = buf.popleft()
             _replay_buffer_bytes[sid] -= evicted_size
             _replay_total_bytes -= evicted_size
-            _replay_evicted_through[sid] = evicted_seq
+            _replay_evicted_through[sid] = max(_replay_evicted_through.get(sid, 0), evicted_seq)
         while _replay_total_bytes > _REPLAY_PROCESS_BYTES_MAX:
             for evict_sid, evict_buf in _replay_buffers.items():
                 if evict_buf:
                     evicted_seq, _event, evicted_size = evict_buf.popleft()
                     _replay_buffer_bytes[evict_sid] -= evicted_size
                     _replay_total_bytes -= evicted_size
-                    _replay_evicted_through[evict_sid] = evicted_seq
+                    _replay_evicted_through[evict_sid] = max(_replay_evicted_through.get(evict_sid, 0), evicted_seq)
                     break