DataStreamingServer

def __init__(self, supervisor: Optional[Any] = None) -> None:
    super().__init__("websockets")
    self.data_ws: Optional[web.WebSocketResponse] = (
        None
    )
    self.clients: set[web.WebSocketResponse] = set()
    self.app = None
    self.cli_args = settings
    self.is_secure_mode = False
    self.input_handler = None
    self._tasks_to_run = []
    self.RECONNECT_DEBOUNCE_MS = 500
    self.RECONNECT_GRACE_S = 3.0
    self._display_teardown_tasks = set()
    self.MAX_RECENT_CLIENTS = 1000
    self.last_connection_times = OrderedDict()
    self._latest_client_render_fps = 0.0
    self._last_time_client_ok = 0.0
    self._client_acknowledged_frame_id = -1
    self._last_client_acknowledged_frame_id_update_time = 0.0
    self._previous_ack_id_for_stall_check = -1
    self._previous_sent_id_for_stall_check = -1
    self._sent_frames_log = deque()
    self.rc_mode = RateControlMode.CRF
    self.config_gate = asyncio.Event()
    self.shutdown_event = asyncio.Event()
    self._shutdown_called = False
    self.supervisor = supervisor

    def get_initial_value(setting_name: str):
        """Get the correct initial integer/bool from a processed setting."""
        processed_value = getattr(self.cli_args, setting_name)
        setting_def = next((s for s in SETTING_DEFINITIONS if s['name'] == setting_name), None)
        if not setting_def: return None

        if setting_def['type'] == 'range':
            min_val, max_val = processed_value
            return min_val if min_val == max_val else setting_def.get('meta', {}).get('default_value')
        elif setting_def['type'] == 'bool':
            return processed_value[0]
        return processed_value

    self._initial_video_crf = get_initial_value('video_crf')
    self.video_crf = self._initial_video_crf
    self._initial_video_fullcolor = get_initial_value('video_fullcolor')
    self.video_fullcolor = self._initial_video_fullcolor
    self._initial_video_streaming_mode = get_initial_value('video_streaming_mode')
    self.video_streaming_mode = self._initial_video_streaming_mode
    self.capture_cursor = False
    self._initial_jpeg_quality = get_initial_value('jpeg_quality')
    self.jpeg_quality = self._initial_jpeg_quality
    self._initial_paint_over_jpeg_quality = get_initial_value('paint_over_jpeg_quality')
    self.paint_over_jpeg_quality = self._initial_paint_over_jpeg_quality
    self._initial_video_paintover_crf = get_initial_value('video_paintover_crf')
    self.video_paintover_crf = self._initial_video_paintover_crf
    self._initial_video_paintover_burst_frames = get_initial_value('video_paintover_burst_frames')
    self.video_paintover_burst_frames = self._initial_video_paintover_burst_frames
    self._initial_use_cpu = get_initial_value('use_cpu')
    self.use_cpu = self._initial_use_cpu
    self._initial_use_paint_over_quality = get_initial_value('use_paint_over_quality')
    self.use_paint_over_quality = self._initial_use_paint_over_quality
    self._initial_video_bitrate = get_initial_value('video_bitrate')
    self.video_bitrate = self._initial_video_bitrate

    self._system_monitor_task_ws = None
    self._gpu_monitor_task_ws = None
    self._network_monitor_task_ws = None
    self._shared_stats_ws = {}
    self._shared_network_stats = {}
    self._gpu_available = None
    self._gpu_probe_lock = asyncio.Lock()
    self.uinput_mouse_socket = UINPUT_MOUSE_SOCKET
    self.js_socket_path = settings.js_socket_path
    self.enable_clipboard = settings.enable_clipboard
    self.enable_binary_clipboard = self.cli_args.enable_binary_clipboard[0]
    self.enable_cursors = ENABLE_CURSORS
    self.cursor_size = CURSOR_SIZE
    self.cursor_scale = 1.0
    self.cursor_debug = DEBUG_CURSORS
    self._last_adjustment_timestamp = 0.0
    self.client_settings_received = asyncio.Event()
    self._reconfigure_lock = asyncio.Lock()
    self._video_capture_lock = asyncio.Lock()
    self._is_reconfiguring = False
    self._reconfigure_pending = False
    self._bytes_sent_in_interval = 0
    self._last_bandwidth_calc_time = time.monotonic()
    self.last_start_video_request_times = {}
    self.last_viewer_keyframe_request_times = {}
    self.video_paused_clients = set()
    self._deferred_viewer_rejoins = {}
    self.allowed_desync_ms = BACKPRESSURE_ALLOWED_DESYNC_MS
    self.latency_threshold_for_adjustment_ms = BACKPRESSURE_LATENCY_THRESHOLD_MS
    self.backpressure_check_interval_s = BACKPRESSURE_CHECK_INTERVAL_S
    self.BACKPRESSURE_QUEUE_SIZE = getattr(settings, 'backpressure_queue_size', 120)
    self._last_client_frame_id_report_time = 0.0
    self.capture_loop = None

    self.display_clients = {}
    self.video_relay_groups = {}
    self.capture_instances = {}
    self.display_layouts = {}
    self._persistent_capture_modules = {}
    self._wayland_ctl_module = None
    self._host_output_capacity = None
    self._last_keyframe_request = {}
    self._last_keyframe_log = {}
    self._keyframe_log_suppressed = {}

    self.audio_device_name = self.cli_args.audio_device_name
    self.pcmflux_module = None
    self.is_pcmflux_capturing = False
    self.pcmflux_settings = None
    self._pcmflux_reported_failure = None
    self._pcmflux_last_restart = 0.0
    self.audio_redundancy_by_ws = {}
    self.audio_redundancy_enabled = bool(settings.audio_redundancy[0])
    self._active_audio_red_distance = 0
    # The vendored WebRTC RedOpusEncoder reads its depth from audio_config,
    # so one control (0 = plain Opus) drives both transports.
    _red_dist = getattr(settings, "audio_redundancy_distance", AUDIO_RED_DISTANCE)
    audio_config.set_red_distance(
        _red_dist if self.audio_redundancy_enabled else 0
    )
    self.pcmflux_callback = None
    self.pcmflux_audio_queue = None
    self.pcmflux_send_task = None
    self.pcmflux_capture_loop = None

    self._last_display_count = 0
    self._wm_swap = MultiMonitorWindowManager()

def initialize(self) -> None:
    """Create the SelkiesStreamingApp and InputHandler and wire their callbacks.

    Must be called before run(). Also resolves secure vs legacy mode (a set
    master token closes the config gate until tokens are provisioned, and
    governs who holds input authority) and installs the WebRTC-dialect live
    verbs (`_arg_fps`, `vb`, `ab`, `_rc`, `_crf`) so both transports honor
    the same per-key tunables. With metrics enabled, the registry-global
    Prometheus gauges are fed server-side (ACK-derived client fps and
    smoothed RTT from the backpressure loop, GPU from the stats collector)
    and by the shared `_f,`/`_l,` verbs when a client reports directly.
    """
    self.is_secure_mode = bool(self.cli_args.master_token)
    if self.is_secure_mode:
        logger.info("Secure Mode ENABLED (SELKIES_MASTER_TOKEN is set).")
    else:
        logger.info("Legacy Mode ENABLED (SELKIES_MASTER_TOKEN is not set).")
        self.config_gate.set()

    global TARGET_FRAMERATE
    processed_framerate = settings.framerate
    min_fr, max_fr = processed_framerate
    if min_fr == max_fr:
        TARGET_FRAMERATE = min_fr
    else:
        fr_def = next((s for s in SETTING_DEFINITIONS if s['name'] == 'framerate'), None)
        TARGET_FRAMERATE = fr_def['meta']['default_value'] if fr_def else 60

    initial_encoder = settings.encoder

    if not settings.debug[0]:
        logging.getLogger("pulsectl_asyncio").setLevel(logging.WARNING)

    logger.info(f"Initializing DataStreamingServer with encoder: {initial_encoder}, Framerate: {TARGET_FRAMERATE}")

    event_loop = asyncio.get_running_loop()
    self.app = SelkiesStreamingApp(
        event_loop,
        framerate=TARGET_FRAMERATE,
        encoder=initial_encoder,
        mode="websockets",
    )
    self.app.server_enable_resize = ENABLE_RESIZE
    self.app.last_resize_success = True
    self.app.data_streaming_server = self
    logger.info(
        f"SelkiesStreamingApp initialized: encoder={self.app.encoder}, display={self.app.display_width}x{self.app.display_height}"
    )

    if settings.enable_rate_control[0]:
        self.rc_mode = RateControlMode(settings.rate_control_mode)

    # The normalized policy string (true/false/in/out); the input handler gates
    # directions off it.
    clipboard_mode = settings.enable_clipboard

    self.input_handler = InputHandler(
        self.app,
        self.uinput_mouse_socket,
        self.js_socket_path,
        clipboard_mode,
        str(settings.enable_binary_clipboard[0]).lower(),
        self.enable_cursors,
        self.cursor_size,
        1.0,
        self.cursor_debug,
        data_server_instance=self,
        is_wayland=IS_WAYLAND,
        wayland_socket_index=settings.wayland_socket_index,
        app_wayland_display=(settings.app_wayland_display
                             or settings.wayland_host_display),
        uinput_gamepad=settings.uinput_gamepad,
    )

    self.input_handler.on_clipboard_read = self.app.send_ws_clipboard_data
    self.input_handler.on_set_fps = self._handle_opcode_fps
    self.input_handler.on_video_encoder_bit_rate = self._handle_opcode_video_bitrate
    self.input_handler.on_audio_encoder_bit_rate = self._handle_opcode_audio_bitrate
    self.input_handler.on_update_rate_control_mode = self._handle_opcode_rate_control
    self.input_handler.on_update_crf = self._handle_opcode_crf
    self.metrics = None
    if settings.enable_metrics_http[0]:
        self.metrics = Metrics()
        self.input_handler.on_client_fps = (
            lambda fps: self.metrics.set_fps(fps) if self.metrics else None
        )
        self.input_handler.on_client_latency = (
            lambda latency: self.metrics.set_latency(latency) if self.metrics else None
        )
    self.input_handler.on_mouse_pointer_visible = self.set_native_cursor_rendering

    if ENABLE_RESIZE:
        self.input_handler.on_resize = lambda res_str, display_id='primary': on_resize_handler(
            res_str, self.app, self, display_id
        )
    else:
        # Only the resolution is frozen: a DPI sync still scales the desktop,
        # applied in this transport's own message loop.
        self.input_handler.on_resize = lambda res_str, display_id='primary': logger.warning("Resize disabled.")
    logger.info("DataStreamingServer initialization complete.")

async def set_native_cursor_rendering(self, enabled: bool) -> None:
    """Compose the cursor into the captured video (vs the client-drawn overlay).

    Applies to every display's capture. Reached both from the
    SET_NATIVE_CURSOR_RENDERING message and the shared input protocol's
    pointer-visibility toggle ("p,N"), which map to the same tunable.
    """
    if self.capture_cursor == enabled:
        data_logger.info(f"Native cursor rendering: value {enabled} is already set.")
        return
    self.capture_cursor = enabled
    if len(self.capture_instances) > 0:
        data_logger.info("Cursor rendering changed, triggering display reconfiguration.")
        await self.reconfigure_displays()

def _opcode_display_module(self, display_id: str) -> Optional[Any]:
    """The display's live ScreenCapture module, or None if not capturing."""
    inst = self.capture_instances.get(display_id)
    return inst.get('module') if inst else None

def _track_capture_settings(self, display_id: str, fresh: Optional[Any] = None,
                            **live_fields: Any) -> None:
    """Record what the display's running capture is actually configured with.

    Pass `fresh` after rebuilding the whole settings object, or individual
    fields after a targeted rate update.

    The tracked object is what _video_relay_budget sizes new relays from and
    what a layout-following reconfigure re-pushes to the module, so a live
    change that skipped it would be applied to the encoder and then silently
    reverted.
    """
    inst = self.capture_instances.get(display_id)
    if inst is None:
        return
    if fresh is not None:
        inst['settings'] = fresh
        return
    cs = inst.get('settings')
    if cs is None:
        return
    for name, value in live_fields.items():
        setattr(cs, name, value)

async def _handle_opcode_fps(self, fps: Any, display_id: str = 'primary') -> None:
    """Live framerate for the shared '_arg_fps' verb (WebRTC-mode parity):
    sanitize against the server range, store, and live-update the display's
    capture; a stopped display applies the new rate at its next START_VIDEO."""
    sanitized = sanitize_client_setting("framerate", fps, self.cli_args, data_logger)
    if sanitized is None:
        return
    if display_id == 'primary':
        # Only the primary controller moves the session default later displays seed from.
        self.app.set_framerate(sanitized)
        data_logger.info(f"Session default framerate updated to {int(sanitized)} for new displays.")
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["framerate"] = sanitized
    module = self._opcode_display_module(display_id)
    if module is not None:
        try:
            module.update_framerate(float(sanitized))
            self._track_capture_settings(display_id, target_fps=float(sanitized))
            data_logger.info(f"Applied framerate live via '_arg_fps': {sanitized} fps for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live framerate update failed for '{display_id}' ({e}).")

async def _handle_opcode_video_bitrate(self, bitrate: Any, display_id: str = "primary") -> None:
    """Live video bitrate (kbps) for the 'vb' verb, sanitized exactly like
    the SETTINGS path so locked server ranges cannot be bypassed."""
    sanitized = sanitize_client_setting("video_bitrate", bitrate, self.cli_args, data_logger)
    if sanitized is None:
        return
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["video_bitrate"] = sanitized
    if display_id == 'primary':
        self._initial_video_bitrate = sanitized
        data_logger.info(f"Session default video_bitrate updated to {int(sanitized)} kbps for new displays.")
    module = self._opcode_display_module(display_id)
    if module is not None:
        kbps = int(round(float(sanitized)))
        try:
            module.update_video_bitrate(kbps)
            self._track_capture_settings(display_id, video_bitrate_kbps=kbps)
            data_logger.info(f"Applied video bitrate live via 'vb': {kbps} kbps for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live bitrate update failed for '{display_id}' ({e}).")

async def _handle_opcode_audio_bitrate(self, bitrate: Any) -> None:
    """Live Opus bitrate (bps) for the 'ab' verb; same live-retarget with
    restart fallback as the SETTINGS path."""
    sanitized = sanitize_client_setting("audio_bitrate", bitrate, self.cli_args, data_logger)
    if sanitized is None:
        return
    self.app.audio_bitrate = sanitized
    for display_state in self.display_clients.values():
        display_state["audio_bitrate"] = self.app.audio_bitrate
    if self.is_pcmflux_capturing and self.pcmflux_module:
        try:
            self.pcmflux_module.update_audio_bitrate(int(self.app.audio_bitrate))
            data_logger.info(f"Applied audio bitrate live: {self.app.audio_bitrate} bps")
        except Exception as e:
            data_logger.warning(f"Live audio bitrate update failed ({e}); restarting audio pipeline.")
            # Under the guard like every audio start/stop, and re-checked there:
            # a concurrent guarded op must not orphan a second AudioCapture.
            async with self._reconfigure_guard():
                if self.is_pcmflux_capturing:
                    await self._stop_pcmflux_pipeline()
                    await self._start_pcmflux_pipeline()

async def _handle_opcode_rate_control(self, mode: Any, display_id: str = 'primary') -> None:
    """Rate-control switch for the '_rc' verb: structural like the SETTINGS
    path (the encoder session must be rebuilt), honoring the server's
    enable_rate_control lock and a stopped display's start gating."""
    enable_rate_control, _ = self.cli_args.enable_rate_control
    if not enable_rate_control:
        data_logger.debug("Server has rate control disabled. Ignoring '_rc' change.")
        return
    # Resolved by value: a duplicate enum class from another module must not
    # leave a stray 'RateControlMode.CBR' repr.
    mode_str = (mode.value if hasattr(mode, "value") else str(mode)).split(".")[-1].lower()
    sanitized = sanitize_client_setting("rate_control_mode", mode_str, self.cli_args, data_logger)
    if sanitized not in ("cbr", "crf"):
        return
    display_state = self.display_clients.get(display_id)
    if display_state is None:
        return
    if display_state.get("rate_control_mode") == sanitized:
        return
    display_state["rate_control_mode"] = sanitized
    if display_id == 'primary':
        self.rc_mode = RateControlMode(sanitized)
        data_logger.info(f"Session default rate_control_mode updated to {sanitized} for new displays.")
    if not display_state.get('video_active', True):
        return
    layout = self.display_layouts.get(display_id)
    if layout is None:
        return
    restart_ok = False
    async with self._reconfigure_guard():
        if display_state.get('video_active', True):
            data_logger.info(f"Applied rate-control via '_rc': {sanitized} for '{display_id}'. Restarting its capture stream.")
            await self._stop_capture_for_display(display_id)
            await self._start_capture_for_display(
                display_id=display_id,
                width=layout['w'], height=layout['h'],
                x_offset=layout['x'], y_offset=layout['y']
            )
            await self._start_backpressure_task_if_needed(display_id)
            self._schedule_idr_for_display(display_id)
            await self._broadcast_live_server_settings(display_id)
            if IS_WAYLAND:
                await self._sync_wayland_realized_geometry(display_id)
            restart_ok = self._opcode_display_module(display_id) is not None
    if not restart_ok:
        data_logger.warning(f"Rate-control restart failed for '{display_id}'; falling back to full reconfiguration.")
        await self.reconfigure_displays()

async def _handle_opcode_crf(self, crf: Any, display_id: str = 'primary') -> None:
    """Live CRF for the '_crf' verb; rides the tunables path like SETTINGS."""
    sanitized = sanitize_client_setting("video_crf", crf, self.cli_args, data_logger)
    if sanitized is None:
        return
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["video_crf"] = sanitized
    module = self._opcode_display_module(display_id)
    layout = self.display_layouts.get(display_id)
    if module is not None and layout is not None:
        try:
            fresh = self._get_capture_settings(
                display_id, layout['w'], layout['h'], layout['x'], layout['y']
            )
            module.update_tunables(fresh)
            self._track_capture_settings(display_id, fresh=fresh)
            data_logger.info(f"Applied CRF live via '_crf': {sanitized} for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live CRF update failed for '{display_id}' ({e}).")

async def broadcast_display_config(self) -> None:
    """Broadcast the current display roster to all clients."""
    if not self.clients:
        return

    connected_displays = list(self.display_clients.keys())
    payload = {
        "type": "display_config_update",
        "displays": connected_displays
    }
    message_str = f"DISPLAY_CONFIG_UPDATE,{json.dumps(payload)}"

    data_logger.info(f"Broadcasting display config update: {message_str}")
    # Bounded: callers hold _reconfigure_lock.
    await _broadcast_to_clients(self.clients, message_str, per_client_timeout=2.0)

def refresh_cursor_cache(self) -> Optional[dict]:
    """Refresh and return the cached cursor payload for late-joining clients."""
    if not self.app:
        return None

    cursor_data = None
    if self.input_handler and hasattr(self.input_handler, "get_current_cursor_data"):
        cursor_data = self.input_handler.get_current_cursor_data()

    if cursor_data is not None:
        self.app.last_cursor_sent = cursor_data

    return self.app.last_cursor_sent

async def send_current_cursor(self, websocket: web.WebSocketResponse, raddr: Any) -> None:
    """Send the current cursor image to one client (used at connect/resume)."""
    cursor_data = self.refresh_cursor_cache()
    if not cursor_data:
        return

    data_logger.info(f"Sending current cursor to client {raddr}")
    try:
        msg_str = json.dumps(cursor_data)
        await websocket.send_str(f"cursor,{msg_str}")
    except Exception as e:
        data_logger.warning(f"Failed to send current cursor to client {raddr}: {e}")

def _pcmflux_audio_callback(self, frame: Any) -> None:
    """Queue one encoded audio frame for broadcast.

    Called from pcmflux's capture thread with an AudioFrame, so it never
    touches asyncio state directly: the enqueue is scheduled onto the loop
    with call_soon_threadsafe, and loop/queue references are snapshotted
    because teardown can null them concurrently.
    """
    if self.is_pcmflux_capturing and frame is not None and self.pcmflux_audio_queue is not None:
        if len(frame) > 0:
            loop = self.pcmflux_capture_loop
            q = self.pcmflux_audio_queue
            if loop is None or q is None or loop.is_closed():
                return
            # Zero-copy: the AudioFrame owns the buffer (header included) and
            # frees it once the queue item drops.
            item = {'data': memoryview(frame), 'owner': frame}
            def _do_put():
                try:
                    q.put_nowait(item)
                except asyncio.QueueFull:
                    pass
            # The loop can close between the check above and here; the
            # RuntimeError would surface in pcmflux's C thread.
            try:
                loop.call_soon_threadsafe(_do_put)
            except RuntimeError:
                pass

def _check_pcmflux_health(self) -> None:
    """Restart the audio pipeline when its pcmflux worker died after start.

    pcmflux's start handshake answers while the worker is still starting,
    so a backend that gives up afterwards (its retry ladder spent, a
    mid-run reconnect budget exhausted) surfaces only through
    `last_error`; the broadcast loop asks here whenever its queue stays
    silent. The failure is logged once per run, and the restart — the same
    stop/start the audio toggles use — runs as its own task (the stop
    cancels the broadcast loop) no more often than the restart floor.
    Older pcmflux builds without the attribute never report one.
    """
    module = self.pcmflux_module
    if module is None or not self.is_pcmflux_capturing:
        return
    error = getattr(module, "last_error", None)
    if not error:
        return
    failure = (id(module), str(error))
    if self._pcmflux_reported_failure != failure:
        self._pcmflux_reported_failure = failure
        data_logger.error(f"pcmflux audio capture failed after start: {error}")
    now = time.monotonic()
    if now - self._pcmflux_last_restart < PCMFLUX_RESTART_FLOOR_SECONDS:
        return
    self._pcmflux_last_restart = now
    _spawn_background_task(self._restart_failed_pcmflux(module), name="pcmflux-restart")

async def _restart_failed_pcmflux(self, failed_module: Any) -> None:
    """Stop and start the audio pipeline under the reconfigure guard, unless
    the failed capture was already replaced or stopped meanwhile."""
    async with self._reconfigure_guard():
        if self.pcmflux_module is not failed_module or not self.is_pcmflux_capturing:
            return
        data_logger.info("Restarting the audio pipeline after its capture failed.")
        await self._stop_pcmflux_pipeline()
        await self._start_pcmflux_pipeline()

async def _pcmflux_send_audio_chunks(self) -> None:
    """Broadcast queued Opus audio chunks to the primary-viewer sockets.

    Runs as a long-lived task. Secondary-display sockets are excluded (they
    render video only; audio rides the primary connection), and sends are
    bounded so one stalled socket cannot freeze the shared stream. A queue
    that stays silent past the health interval is the cue to ask pcmflux
    whether the capture worker died (_check_pcmflux_health).
    """
    data_logger.info("pcmflux audio chunk broadcasting task started.")
    try:
        while True:
            try:
                item = await asyncio.wait_for(
                    self.pcmflux_audio_queue.get(), timeout=PCMFLUX_HEALTH_INTERVAL_SECONDS)
            except asyncio.TimeoutError:
                self._check_pcmflux_health()
                continue

            secondary_websockets = {
                client_info.get('ws')
                for did, client_info in self.display_clients.items()
                if did != 'primary' and client_info.get('ws')
            }
            primary_viewers = self.clients - secondary_websockets

            if not primary_viewers:
                self.pcmflux_audio_queue.task_done()
                continue

            # A zero-copy view over the AudioFrame, header included; sent as-is.
            message_to_send = item['data']
            self._bytes_sent_in_interval += len(message_to_send) * len(primary_viewers)
            dropped = await _broadcast_to_clients(
                primary_viewers, message_to_send,
                per_client_timeout=SHARED_STREAM_SEND_TIMEOUT_SECONDS,
            )
            if dropped:
                # primary_viewers is a per-chunk temporary; the drop must reach the registry.
                self.clients -= dropped

            self.pcmflux_audio_queue.task_done()
    except asyncio.CancelledError:
        data_logger.info("pcmflux audio chunk broadcasting task cancelled.")
    finally:
        data_logger.info("pcmflux audio chunk broadcasting task finished.")

def _compute_audio_red_distance(self) -> int:
    """RED distance for the shared audio broadcast.

    WS is TCP, but the sender still drops frames under backpressure
    (pcmflux delivery ring drop-oldest, and this server's audio queue drops
    on overflow), and RED lets the client recover those within the
    redundancy distance.

    Returns:
        The configured distance only when the server allows it AND there is
        at least one client AND every connected client advertised
        audioRedundancy; otherwise 0 (plain frames, which decode
        everywhere) — a single non-capable or legacy (field-absent) client
        falls the whole stream back.
    """
    if not self.audio_redundancy_enabled:
        return 0
    if not self.clients:
        return 0
    for ws in self.clients:
        if not self.audio_redundancy_by_ws.get(ws):
            return 0
    return getattr(settings, "audio_redundancy_distance", AUDIO_RED_DISTANCE)

async def _regate_audio_redundancy(self) -> None:
    """Recompute the RED gate for the shared audio stream and, if it flipped
    while capturing, restart the pipeline so the new red_distance takes
    effect. Callers hold the reconfigure guard (pipeline start/stop must be
    serialized against reconfigure_displays). A missing app means teardown
    (the last client leaving drops RED to 0, and the disconnect path stops
    the pipeline itself), so no restart is attempted then."""
    desired = self._compute_audio_red_distance()
    if desired == self._active_audio_red_distance:
        return
    if not self.is_pcmflux_capturing:
        return
    if not self.app:
        return
    data_logger.info(
        f"Audio RED gate changed ({self._active_audio_red_distance} -> {desired}); "
        "restarting audio pipeline."
    )
    await self._stop_pcmflux_pipeline()
    await self._start_pcmflux_pipeline()

async def _start_pcmflux_pipeline(self) -> bool:
    """Start the pcmflux audio capture and the shared broadcast task.

    Resolves the RED distance for the current client set at start, so a
    gate change while running requires a restart (see
    _regate_audio_redundancy). Callers serialize via the reconfigure guard.

    Returns:
        True when capturing afterwards (already-running counts); False when
        audio is disabled, pcmflux is unavailable, or the start failed (a
        partial start is cleaned up).
    """
    if not settings.audio_enabled[0]:
        data_logger.info("Audio is disabled by server settings. Not starting pipeline.")
        return False
    if not PCMFLUX_AVAILABLE:
        data_logger.error("Cannot start audio pipeline: pcmflux library not available.")
        return False
    if self.is_pcmflux_capturing:
        data_logger.info("pcmflux audio pipeline is already capturing.")
        return True
    if not self.app:
        data_logger.error("Cannot start pcmflux: self.app (SelkiesStreamingApp instance) is not available.")
        return False

    self.pcmflux_capture_loop = self.capture_loop or asyncio.get_running_loop()
    if not self.pcmflux_capture_loop:
        data_logger.error("Cannot start pcmflux: asyncio event loop not found.")
        return False

    await ensure_capture_sink(self.audio_device_name)
    data_logger.info("Starting pcmflux audio pipeline...")
    try:
        capture_settings = AudioCaptureSettings()
        device_name_bytes = self.audio_device_name.encode('utf-8') if self.audio_device_name else None
        capture_settings.device_name = device_name_bytes
        capture_settings.sample_rate = 48000
        capture_settings.channels = self.app.audio_channels
        capture_settings.opus_bitrate = int(self.app.audio_bitrate)
        # The frame duration is the capture-side latency floor; PulseAudio
        # fragments are kept no larger than one frame.
        frame_ms = float(getattr(settings, 'audio_frame_duration_ms', '20') or 20)
        capture_settings.frame_duration_ms = frame_ms
        capture_settings.use_vbr = True
        capture_settings.use_silence_gate = False
        capture_settings.latency_ms = int(min(10, frame_ms))
        capture_settings.debug_logging = self.cli_args.debug[0]
        # pcmflux's native [0x01,0x00] header goes on the wire; no Python prepend/copy.
        capture_settings.omit_audio_header = False
        red_distance = self._compute_audio_red_distance()
        capture_settings.red_distance = red_distance
        self._active_audio_red_distance = red_distance
        self.pcmflux_settings = capture_settings

        data_logger.info(f"pcmflux settings: device='{self.audio_device_name}', "
                         f"bitrate={capture_settings.opus_bitrate}, channels={capture_settings.channels}, "
                         f"red_distance={red_distance}")

        self.pcmflux_callback = self._pcmflux_audio_callback
        self.pcmflux_module = AudioCapture()
        self.pcmflux_audio_queue = asyncio.Queue(maxsize=getattr(self, 'BACKPRESSURE_QUEUE_SIZE', 120))

        await self.pcmflux_capture_loop.run_in_executor(
            None, self.pcmflux_module.start_capture, self.pcmflux_settings, self.pcmflux_callback
        )

        # The start handshake answers before the worker is up; a run that
        # already died reports through last_error.
        state = getattr(self.pcmflux_module, "state", "running")
        error = getattr(self.pcmflux_module, "last_error", None)
        if state == "failed" or error:
            raise RuntimeError(f"capture {state}: {error}")
        self.is_pcmflux_capturing = True
        if self.pcmflux_send_task is None or self.pcmflux_send_task.done():
            self.pcmflux_send_task = asyncio.create_task(self._pcmflux_send_audio_chunks())

        data_logger.info(f"pcmflux audio capture state: {state}.")
        return True
    except Exception as e:
        data_logger.error(f"Failed to start pcmflux audio pipeline: {e}", exc_info=True)
        await self._stop_pcmflux_pipeline()
        return False

async def _stop_pcmflux_pipeline(self) -> bool:
    """Stop the audio capture and broadcast task; idempotent.

    The capturing flag is cleared first so the capture-thread callback
    stops queueing chunks before the queue is dropped.
    """
    if not self.is_pcmflux_capturing and not self.pcmflux_module:
        return True

    data_logger.info("Stopping pcmflux audio pipeline...")
    self.is_pcmflux_capturing = False

    if self.pcmflux_send_task:
        self.pcmflux_send_task.cancel()
        try:
            await self.pcmflux_send_task
        except asyncio.CancelledError:
            pass
        self.pcmflux_send_task = None

    if self.pcmflux_module:
        try:
            if self.pcmflux_capture_loop:
                await self.pcmflux_capture_loop.run_in_executor(
                    None, self.pcmflux_module.stop_capture
                )
        except Exception as e:
            data_logger.error(f"Error during pcmflux stop_capture: {e}")
        finally:
            del self.pcmflux_module
            self.pcmflux_module = None

    self.pcmflux_audio_queue = None
    data_logger.info("pcmflux audio pipeline stopped.")
    return True

async def shutdown_pipelines(self) -> None:
    """Stop all capture pipelines; the ONLY way pipelines are stopped programmatically.

    Deadlock-proof by construction: reconfigure_displays() self-acquires
    the reconfigure lock, so it runs first and outside the guard; the
    audio/backpressure teardown then runs under the guard (a
    disconnect/connect race could otherwise tear down audio a new client
    just started), and none of the awaited teardowns re-acquire the lock.
    """
    logger.info("Initiating unified pipeline shutdown...")
    await self.reconfigure_displays()
    async with self._reconfigure_guard():
        await self._stop_pcmflux_pipeline()
        if self.display_clients:
            stop_bp_tasks = [
                self._ensure_backpressure_task_is_stopped(disp_id)
                for disp_id in self.display_clients.keys()
            ]
            await asyncio.gather(*stop_bp_tasks, return_exceptions=True)
        if self.pcmflux_send_task and not self.pcmflux_send_task.done():
            self.pcmflux_send_task.cancel()
            try:
                await self.pcmflux_send_task
            except asyncio.CancelledError:
                pass
    logger.info("Unified pipeline shutdown complete.")

async def _ensure_backpressure_task_is_stopped(self, display_id: str, notify: bool = True) -> bool:
    """Cancel and clean up the backpressure task for a specific display.

    Args:
        display_id: The display whose task to stop.
        notify: When True and a task was actually running, reset the frame
            ids and notify the client(s).

    Returns:
        Whether the pipeline-reset notification was sent, so callers that
        must guarantee a reset (capture stop) can send it exactly once
        themselves when no task was running.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        return False

    task_was_running = False
    task = display_state.get('backpressure_task')
    if task and not task.done():
        data_logger.debug(f"Ensuring frame backpressure task for '{display_id}' is stopped.")
        task.cancel()
        try:
            await task
            task_was_running = True
        except asyncio.CancelledError:
            data_logger.debug(f"Backpressure task for '{display_id}' cancelled successfully.")
            task_was_running = True
        except Exception as e_cancel:
            data_logger.error(f"Error awaiting cancellation for '{display_id}' backpressure task: {e_cancel}")
        display_state['backpressure_task'] = None

    display_state['backpressure_enabled'] = True

    if task_was_running and notify:
        data_logger.info(f"Backpressure task for '{display_id}' was stopped. Resetting its frame IDs.")
        await self._reset_frame_ids_and_notify(display_id)
        return True
    return False

async def _reset_frame_ids_and_notify(self, display_id: str) -> None:
    """Reset a display's frame-id state and send PIPELINE_RESETTING.

    For the primary display the reset is broadcast to ALL clients (shared
    viewers decode the same stream); a secondary notifies only its own
    socket. Every id-keyed artifact (send stamps, RTT samples, the fps
    estimator's baseline) resets with the numbering, since a stale stamp
    matched by a NEW id of the same value manufactures a giant RTT sample
    that poisons the smoothed estimate. The notify sends are bounded: this
    runs under _video_capture_lock, so one stalled client must not wedge
    every display's start/stop; a timed-out socket is dropped and closed
    rather than reused, and the id state was already reset either way.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        return

    data_logger.info(f"Resetting frame IDs for display '{display_id}'.")
    display_state['last_sent_frame_id'] = 0
    display_state['has_sent_any_frame'] = False
    display_state['acknowledged_frame_id'] = -1
    sent_ts = display_state.get('sent_timestamps')
    if sent_ts is not None:
        sent_ts.clear()
    rtt_samples = display_state.get('rtt_samples')
    if rtt_samples is not None:
        rtt_samples.clear()
    display_state['smoothed_rtt'] = 0.0
    display_state.pop('_fps_sample_acked', None)
    display_state.pop('_fps_sample_time', None)

    message = f"PIPELINE_RESETTING {display_id}"

    if display_id == 'primary' and self.clients:
        data_logger.info(f"Broadcasting primary pipeline reset to all {len(self.clients)} clients: {message}")
        await _broadcast_to_clients(self.clients, message, per_client_timeout=2.0)
    else:
        websocket = display_state.get('ws')
        if websocket:
            try:
                await asyncio.wait_for(websocket.send_str(message), timeout=2.0)
            except asyncio.TimeoutError:
                data_logger.warning(f"Timed out notifying client for '{display_id}' of reset; dropping socket.")
                self.clients.discard(websocket)
                _close_abandoned_ws(websocket)
            except (ConnectionResetError, OSError, RuntimeError):
                data_logger.warning(f"Could not notify client for '{display_id}' of reset; connection closed.")

    display_state['backpressure_enabled'] = True
    display_state['last_ack_update_time'] = time.monotonic()

async def _start_backpressure_task_if_needed(self, display_id: str) -> None:
    """Start the backpressure task for a specific display if not already running.

    Backend-agnostic: frame ids, ACKs, and RTT flow identically on Wayland.
    A task restart is not a pipeline death, so it never notifies: the
    capture (re)start handles stream freshness, and the client reset
    belongs to the capture-stop path alone.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        data_logger.error(f"Cannot start backpressure task: display '{display_id}' not found.")
        return

    await self._ensure_backpressure_task_is_stopped(display_id, notify=False)

    task = display_state.get('backpressure_task')
    if not task or task.done():
        new_task = asyncio.create_task(self._run_frame_backpressure_logic(display_id))
        display_state['backpressure_task'] = new_task
        data_logger.info(f"New frame backpressure task started for display '{display_id}'.")
    else:
        data_logger.warning(f"Backpressure task for '{display_id}' was already running. Not starting a new one.")

def _active_primary_consumers(self, exclude: Optional[web.WebSocketResponse] = None) -> set:
    """Sockets still consuming the primary broadcast.

    Every client except the secondary displays' owners, minus the paused
    ones. A paused viewer with a deferred rejoin pending counts as active —
    it keeps the capture alive under a waking viewer, while genuinely
    hidden ones let an all-tabs-hidden session stop encoding.

    Args:
        exclude: Drops the socket whose STOP_VIDEO is in flight.
    """
    secondary_ws = {
        info.get('ws')
        for did, info in self.display_clients.items()
        if did != 'primary'
    }
    waking = set(self._deferred_viewer_rejoins)
    consumers = self.clients - secondary_ws - (self.video_paused_clients - waking)
    if exclude is not None:
        consumers.discard(exclude)
    return consumers

def _primary_reconnect_pending(self) -> bool:
    """Whether the primary display entry is being held for a socket that is
    already gone: the reconnect grace keeps the capture warm so a reloading
    page resumes on it instead of paying a full pipeline rebuild."""
    entry = self.display_clients.get('primary')
    return entry is not None and entry.get('ws') not in self.clients

async def _stop_primary_if_unconsumed(self, reason: str) -> None:
    """Stop the primary capture once nothing decodes it — the last unpaused
    consumer hid its tab or disconnected. Hiding and disconnecting take the
    same verdict here; only a pending reconnect grace keeps the capture warm.
    A resume restarts it (START_VIDEO from the display owner, or the viewer
    path's capture ensure)."""
    if 'primary' not in self.capture_instances:
        return
    if self._active_primary_consumers() or self._primary_reconnect_pending():
        return
    data_logger.info(f"{reason} Stopping the 'primary' capture.")
    primary_entry = self.display_clients.get('primary')
    if primary_entry is not None:
        primary_entry['video_active'] = False
    await self._stop_capture_for_display('primary')

def _cancel_deferred_rejoin(self, websocket: web.WebSocketResponse) -> None:
    """Drop a pending deferred rejoin for this socket. A STOP_VIDEO (or a
    disconnect) arriving after a throttled resume supersedes it: the rejoin
    would otherwise un-pause a tab that is hidden again, and the socket would
    keep counting as a live consumer until it fired."""
    rejoin_task = self._deferred_viewer_rejoins.pop(websocket, None)
    if rejoin_task is not None:
        rejoin_task.cancel()

def _schedule_deferred_viewer_rejoin(self, websocket: web.WebSocketResponse, delay: float) -> None:
    """Rejoin a rapid-resume-throttled viewer once the resume floor passes.
    The client already believes it resumed, so a silent discard would leave
    the socket paused until its stall watchdog; at most one deferred rejoin
    is pending per socket."""
    if websocket in self._deferred_viewer_rejoins:
        return

    async def _rejoin():
        try:
            await asyncio.sleep(max(0.05, delay))
            if websocket not in self.clients or websocket not in self.video_paused_clients:
                return
            self.last_start_video_request_times[websocket] = time.monotonic()
            self.video_paused_clients.discard(websocket)
            try:
                await websocket.send_str("PIPELINE_RESETTING primary")
            except (ConnectionResetError, OSError, RuntimeError):
                return
            # A controller tab-hide can tear the capture down mid-resume; an IDR
            # request is then a no-op and the reset decoder would wait forever.
            if 'primary' in self.capture_instances:
                self._schedule_idr_for_display('primary')
            else:
                await self._ensure_viewer_capture()
        finally:
            self._deferred_viewer_rejoins.pop(websocket, None)

    self._deferred_viewer_rejoins[websocket] = asyncio.create_task(_rejoin())

def _video_relay_budget(self, display_id: str, fallback: int) -> int:
    """Skip-ahead byte budget for one client's video relay.

    VIDEO_RELAY_BUDGET_SECONDS of stream at the display's CURRENT
    configured bitrate (1 kbps = 125 B/s), floored so low-bitrate streams
    keep absorbing transport jitter. Read from the live capture settings
    at relay creation so in-place restarts (settings changes that reuse
    the capture callback) are honored.

    Args:
        display_id: The display whose configured bitrate sizes the budget.
        fallback: Covers the start window before capture_instances is
            registered.
    """
    inst = self.capture_instances.get(display_id)
    cs = inst.get('settings') if inst else None
    if cs is None:
        return fallback
    kbps = int(getattr(cs, 'video_bitrate_kbps', 0) or 0)
    return max(VIDEO_RELAY_BUDGET_MIN_BYTES,
               int(kbps * 125 * VIDEO_RELAY_BUDGET_SECONDS))

def _close_video_relays(self, display_id: str) -> None:
    """Stop every per-client video relay for this display. Graceful: each
    relay finishes its in-flight send and its task removes itself."""
    group = self.video_relay_groups.pop(display_id, None)
    if group:
        for relay in list(group.values()):
            relay.stop()

def _schedule_idr_for_display(self, display_id: str) -> None:
    """Ask the encoder for a fresh keyframe on this display.

    request_idr_frame is non-blocking in pixelflux (an atomic flag or a
    channel send) and idempotent, so it runs inline on the event loop.
    """
    instance = self.capture_instances.get(display_id)
    module = instance.get('module') if instance else None
    if module:
        try:
            module.request_idr_frame()
        except Exception:
            pass

def _second_screen_availability(self) -> tuple[bool, str]:
    """Whether this session can actually attach a second display.

    The admin flag gates first; past it, X11 and the self-composited
    Wayland backend mint another output on demand, while host capture is
    bounded by the host compositor's real output count (unknown until the
    first capture start establishes the host session).

    Returns:
        `(available, reason)`; the reason is empty when available.
    """
    enabled, _ = self.cli_args.second_screen
    if not enabled:
        return False, "Second screens are disabled on this server."
    if not IS_WAYLAND or not (self.cli_args.wayland_host_display or '').strip():
        return True, ""
    capacity = self._host_output_capacity
    if capacity is None or capacity < 0:
        return False, "The host compositor's outputs are not known yet."
    if capacity < 2:
        return False, "The host compositor has a single output, so a second display has nothing to capture."
    return True, ""

async def _refresh_second_screen_capacity(self) -> bool:
    """Host-capture mode only: re-read how many outputs the host exposes.

    Returns:
        True when the answer changed, i.e. the second-screen availability
        that clients were told may have flipped.
    """
    if not IS_WAYLAND or not (self.cli_args.wayland_host_display or '').strip():
        return False
    module = self._wayland_control_module()
    if module is None or not hasattr(module, 'output_capacity'):
        return False
    try:
        capacity = int(await asyncio.to_thread(module.output_capacity))
    except Exception as e:
        data_logger.warning(f"Wayland output capacity query failed: {e}")
        return False
    changed = capacity != self._host_output_capacity
    self._host_output_capacity = capacity
    return changed

def _settings_payload_for_display(self, display_id: str) -> dict:
    """Client settings snapshot as it applies to one display.

    build_client_settings_payload() publishes boot config, so the encoder is
    patched to the one this display is actually captured with (its stored
    pick, else the session default a fresh capture would use): clients key
    their wire-format demux off this value and drop every chunk of any other
    format. Also carried: `ws_max_message_bytes` (transport capacity, so
    the client sizes multipart chunks to the whole frame), `app_terminal`
    (the terminal the apps panel launches in, absent when none is
    installed), and `second_screen` and `ui_sidebar_show_apps` as effective
    availability — the admin flag and what the backend can actually do — so
    dashboards never offer a display the server would immediately kill, nor
    an apps panel whose every button would fail.
    """
    payload = build_client_settings_payload()
    live_encoder = (self.display_clients.get(display_id) or {}).get('encoder') or self.app.encoder
    if live_encoder and isinstance(payload.get('encoder'), dict):
        payload['encoder'] = dict(payload['encoder'])
        payload['encoder']['value'] = live_encoder
    payload['ws_max_message_bytes'] = {"value": WS_MAX_MESSAGE_BYTES}
    terminal = self.input_handler.app_terminal() if self.input_handler else None
    if terminal:
        payload['app_terminal'] = {"value": terminal}
    available, _ = self._second_screen_availability()
    entry = payload.get('second_screen')
    if isinstance(entry, dict) and entry.get('value') and not available:
        payload['second_screen'] = dict(entry, value=False)
    apps = payload.get('ui_sidebar_show_apps')
    if (isinstance(apps, dict) and apps.get('value')
            and self.input_handler and not self.input_handler.apps_available()):
        payload['ui_sidebar_show_apps'] = dict(apps, value=False)
    return payload

async def _broadcast_live_server_settings(self, display_id: str) -> None:
    """Re-announce server settings after the given display changed its live encoder.

    The handshake payload holds boot config only, so every connected client —
    shared viewers included — must re-key its wire-format demux or it drops
    the new mode's chunks forever. Routed like broadcast_stream_resolution:
    each display's own socket gets its own encoder and every remaining socket
    gets the primary's, since shared viewers render the primary stream and one
    display's encoder applied on another page keys that page to a format its
    own stream never sends.
    """
    try:
        messages = {}

        def message_for(did):
            if did not in messages:
                messages[did] = json.dumps({
                    "type": "server_settings",
                    "displayId": did,
                    "settings": self._settings_payload_for_display(did),
                })
            return messages[did]

        per_socket = {}
        for did, client in self.display_clients.items():
            ws = client.get('ws')
            if ws is not None:
                per_socket[ws] = message_for(did)
        primary_message = message_for('primary')
    except Exception as e:
        data_logger.warning(f"Could not build live server settings broadcast: {e}")
        return
    groups = {}
    for ws in self.clients:
        groups.setdefault(per_socket.get(ws) or primary_message, set()).add(ws)
    data_logger.info(
        f"Re-announcing live server settings after the '{display_id}' capture restart "
        f"to {len(self.clients)} client(s)."
    )
    for message_str, sockets in groups.items():
        # Bounded: runs under _reconfigure_lock; a frozen client is dropped, not waited on.
        dropped = await _broadcast_to_clients(sockets, message_str, per_client_timeout=2.0)
        # The fan-out ran over a computed set; mirror the drop into the registry.
        for ws in dropped:
            self.clients.discard(ws)

def _set_backpressure_enabled(self, display_id: str, display_state: dict, enabled: bool) -> None:
    """Update the backpressure flag, requesting an IDR when it lifts.

    While backpressure was active, delta frames were dropped, so on the
    False->True (LIFTED) transition the client needs a keyframe to resync;
    otherwise it decodes deltas against a reference it never received.
    """
    prev_enabled = display_state.get('backpressure_enabled', True)
    display_state['backpressure_enabled'] = enabled
    if enabled and not prev_enabled:
        self._schedule_idr_for_display(display_id)

async def _run_frame_backpressure_logic(self, display_id: str) -> None:
    """The core backpressure and latency calculation loop for a single display.

    Every BACKPRESSURE_CHECK_INTERVAL_S it compares the last sent and last
    acked frame ids (uint16 circular distance), sized by the client's
    measured consumption rate and forgiving capped propagation delay, and
    flips the display's backpressure flag: a stalled or lagging client
    stops receiving delta frames, and the lift requests an IDR resync.
    Also feeds the Prometheus fps/latency gauges for the primary display.
    """
    data_logger.info(f"Frame-based backpressure logic task started for display '{display_id}'.")
    display_state = None
    try:
        if self.client_settings_received:
            await self.client_settings_received.wait()
        data_logger.info(f"Client settings received, proceeding with backpressure loop for '{display_id}'.")

        while True:
            await asyncio.sleep(self.backpressure_check_interval_s)

            display_state = self.display_clients.get(display_id)
            if not display_state:
                data_logger.warning(f"Backpressure task for '{display_id}' exiting: display no longer exists.")
                break

            if display_id not in self.capture_instances:
                if not display_state.get('backpressure_enabled', True):
                    data_logger.info(f"Backpressure LIFTED for '{display_id}' (video pipeline is not active).")
                self._set_backpressure_enabled(display_id, display_state, True)
                continue

            current_server_frame_id = display_state.get('last_sent_frame_id', 0)
            last_client_acked_frame_id = display_state.get('acknowledged_frame_id', -1)

            if last_client_acked_frame_id == -1:
                if not display_state.get('backpressure_enabled', True):
                     data_logger.info(f"Backpressure LIFTED for '{display_id}' (client ACK is -1).")
                self._set_backpressure_enabled(display_id, display_state, True)
                display_state['last_ack_update_time'] = time.monotonic()
                continue

            configured_fps = display_state.get('framerate', 60)
            if configured_fps <= 0:
                configured_fps = 60
            client_fps = self._estimate_client_fps(
                display_state, last_client_acked_frame_id, configured_fps, time.monotonic()
            )
            if display_id == 'primary' and getattr(self, 'metrics', None) is not None:
                self.metrics.set_fps(client_fps)
                self.metrics.set_latency(display_state.get('smoothed_rtt', 0.0))

            server_id, client_id = current_server_frame_id, last_client_acked_frame_id

            # Circular distance, so the suspicious-gap test is not tripped at the uint16 wrap.
            wrapped = (server_id - client_id) % (MAX_UINT16_FRAME_ID + 1)

            if wrapped > FRAME_ID_SUSPICIOUS_GAP_THRESHOLD:
                self._set_backpressure_enabled(display_id, display_state, True)
                display_state['last_ack_update_time'] = time.monotonic()
                continue

            # Distinguish 'no frame sent yet' from the counter legitimately wrapping to 0.
            if not display_state.get('has_sent_any_frame', False):
                continue

            frame_desync = wrapped
            allowed_desync_frames = (self.allowed_desync_ms / 1000.0) * client_fps
            # Capped: the RTT estimate rides the queue this loop bounds and must
            # not out-grow the trigger it feeds.
            current_rtt_ms = min(
                display_state.get('smoothed_rtt', 0.0),
                BACKPRESSURE_LATENCY_FORGIVENESS_MAX_MS,
            )
            latency_adjustment_frames = (current_rtt_ms / 1000.0) * client_fps if current_rtt_ms > self.latency_threshold_for_adjustment_ms else 0
            effective_desync_frames = frame_desync - latency_adjustment_frames

            time_since_last_ack = time.monotonic() - display_state.get('last_ack_update_time', time.monotonic())

            if time_since_last_ack > STALLED_CLIENT_TIMEOUT_SECONDS:
                if display_state.get('backpressure_enabled', True):
                    data_logger.warning(f"Client stall for '{display_id}': No ACK in {time_since_last_ack:.1f}s. Forcing backpressure.")
                self._set_backpressure_enabled(display_id, display_state, False)
            elif effective_desync_frames > allowed_desync_frames:
                if display_state.get('backpressure_enabled', True):
                    data_logger.warning(f"Backpressure TRIGGERED for '{display_id}'. S:{server_id}, C:{client_id} (EffDesync:{effective_desync_frames:.1f}f > Allowed:{allowed_desync_frames:.1f}f).")
                self._set_backpressure_enabled(display_id, display_state, False)
            else:
                if not display_state.get('backpressure_enabled', True):
                    data_logger.info(f"Backpressure LIFTED for '{display_id}'. S:{server_id}, C:{client_id} (EffDesync:{effective_desync_frames:.1f}f <= Allowed:{allowed_desync_frames:.1f}f).")
                self._set_backpressure_enabled(display_id, display_state, True)

    except asyncio.CancelledError:
        data_logger.info(f"Backpressure logic task for '{display_id}' cancelled.")
    finally:
        if display_state:
            display_state['backpressure_enabled'] = True
        data_logger.info(f"Backpressure logic task for '{display_id}' finished.")