feat: add video_buffered option to control video frame buffering

Pass video_buffered=False through ConnectionManager → PeerConnectionManager
→ SubscriberPeerConnection to relay.subscribe(buffered=False) for video
tracks. Default True preserves current behavior.

Without buffering, only the latest video frame is kept in memory instead
of an unbounded queue. This prevents OOM on voice-only agents that
subscribe to video but never consume frames (~400 MiB/10sec leak).

Author: aliev

getstream/video/rtc/connection_manager.py

@@ -58,6 +58,7 @@ def __init__(
         create: bool = True,
         subscription_config: Optional[SubscriptionConfig] = None,
         max_join_retries: int = 3,
+        video_buffered: bool = True,
         **kwargs: Any,
     ):
         super().__init__()
@@ -90,7 +91,9 @@ def __init__(
         self._subscription_manager: SubscriptionManager = SubscriptionManager(
             self, subscription_config
         )
-        self._peer_manager: PeerConnectionManager = PeerConnectionManager(self)
+        self._peer_manager: PeerConnectionManager = PeerConnectionManager(
+            self, video_buffered=video_buffered
+        )
 
         self.recording_manager = self._recording_manager
         self.participants_state = self._participants_state

getstream/video/rtc/pc.py

@@ -131,12 +131,14 @@ def __init__(
         self,
         connection,
         configuration: aiortc.RTCConfiguration,
+        video_buffered: bool = True,
     ) -> None:
         logger.info(
             f"creating subscriber peer connection with configuration: {configuration}"
         )
         super().__init__(configuration)
         self.connection = connection
+        self._video_buffered = video_buffered
 
         self.track_map = {}  # track_id -> (MediaRelay, original_track)
         self.video_frame_trackers = {}  # track_id -> VideoFrameTracker
@@ -177,7 +179,8 @@ def _emit_pcm(pcm: PcmData):
                 handler = AudioTrackHandler(relay.subscribe(tracked_track), _emit_pcm)
                 asyncio.create_task(handler.start())
 
-            self.emit("track_added", relay.subscribe(tracked_track), user)
+            buffered = self._video_buffered if track.kind == "video" else True
+            self.emit("track_added", relay.subscribe(tracked_track, buffered=buffered), user)
 
         @self.on("icegatheringstatechange")
         def on_icegatheringstatechange():

getstream/video/rtc/peer_connection.py

@@ -28,8 +28,9 @@
 class PeerConnectionManager:
     """Manages WebRTC peer connections for publishing and subscribing."""
 
-    def __init__(self, connection_manager):
+    def __init__(self, connection_manager, video_buffered: bool = True):
         self.connection_manager = connection_manager
+        self._video_buffered = video_buffered
         self.publisher_pc: Optional[PublisherPeerConnection] = None
         self.subscriber_pc: Optional[SubscriberPeerConnection] = None
         self.publisher_negotiation_lock = asyncio.Lock()
@@ -47,6 +48,7 @@ async def setup_subscriber(self):
             self.subscriber_pc = SubscriberPeerConnection(
                 connection=self.connection_manager,
                 configuration=self._build_rtc_configuration(),
+                video_buffered=self._video_buffered,
             )
 
             # Trace create event