wl-webrtc/src/state_portal.rs

// 采集门户状态模块 —— 通过 PipeWire/DMA-BUF 进行屏幕采集并编码
// AsRawFd is required by frame.fd.as_raw_fd() in build_drm_descriptor below
// but rustc emits a false "unused_imports" warning because OwnedFd also has
// an inherent as_raw_fd — same quirk as avhw.rs. E0599 if removed → keep it.
use std::os::fd::AsRawFd;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use anyhow::{bail, Result}; // 错误处理工具

use crate::args::Args; // 命令行参数
use crate::avhw::{
    self, BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, ResolutionChange,
    SwEncEncode, SwEncImport, SwEncState,
}; // 软件编码器状态（VAAPI 导入 + H.264 编码）
use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点
use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照）
use crate::webrtc::WebRtcState; // WebRTC 信令与媒体传输

/// 门户采集的阶段状态
/// - WaitingForFormat: 等待接收到第一帧 DMA-BUF 以确定视频格式参数
/// - Streaming: 已完成初始化，正在持续编码流
enum PortalStage {
    WaitingForFormat,
    Streaming,
}

struct EncodeThreadTiming {
    sws_us: u64,
    encode_us: u64,
    output_bytes: usize,
}

struct EncodeThread {
    handle: Option<std::thread::JoinHandle<()>>,
    input_tx: crossbeam_channel::Sender<CpuNv12Frame>,
    timing_rx: crossbeam_channel::Receiver<EncodeThreadTiming>,
    duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
}

struct WebrtcThread {
    handle: Option<std::thread::JoinHandle<()>>,
    sent_gap_rx: crossbeam_channel::Receiver<(f64, Option<f64>)>,
}

/// Static configuration handed to the WebRTC sender thread. Immutable for the
/// thread's lifetime; a resolution tier change rebuilds the whole pipeline
/// (and spawns a new thread) rather than mutating this.
struct WebRtcThreadConfig {
    fps: u32,
    enc_width: u32,
    enc_height: u32,
    max_bitrate: u64,
}

/// Channel endpoints owned exclusively by the WebRTC sender thread after spawn.
/// The reverse endpoints stay with StatePortal (or the encode thread) for
/// inbound/outbound traffic.
struct WebRtcThreadChannels {
    webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
    sent_gap_tx: crossbeam_channel::Sender<(f64, Option<f64>)>,
    bitrate_tx: crossbeam_channel::Sender<BitrateCommand>,
    resolution_tx: crossbeam_channel::Sender<BitrateCommand>,
}

/// 门户模式的主状态机
///
/// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。
/// 工作流程：等待第一帧 → 创建编码器 → 持续编码帧数据。
pub struct StatePortal {
    stage: PortalStage,      // 当前采集阶段（等待首帧 / 流式编码中）
    enc: Option<SwEncState>, // 软件编码器，首帧到达后初始化
    enc_import: Option<SwEncImport>,
    enc_thread: Option<EncodeThread>,
    cap: CapPortal,              // PipeWire 屏幕采集端点
    args: Args,                  // 用户命令行参数
    errored: bool,               // 是否遇到不可恢复的错误
    drm_device: Option<PathBuf>, // DRM 渲染设备路径（可自动检测）
    frames_encoded: u64,         // 已编码帧数（用于 PTS 编号）
    start_time: Option<Instant>, // 编码开始时间
    stats: PipelineStats,        // 管道统计（窗口化帧计时 + 每秒快照）
    pw_dropped_prev: u64,        // 上一窗口的 PipeWire 丢弃帧数（用于增量计算）
    webrtc: Option<WebRtcState>,
    webrtc_thread: Option<WebrtcThread>,
    webrtc_paused: Option<Arc<AtomicBool>>,
    last_capture_arrival: Option<Instant>, // timestamp of last real frame arrival
    idle_log_start: Option<Instant>, // when current idle period began (one-shot DEBUG log guard)
    shutdown_started: bool, // idempotency guard; plain bool because &mut self is exclusive (not AtomicBool)
    // Issue #24: real-capture PTS origin/tracking for WebRTC RTP timestamps.
    first_pts_ns: Option<i128>,
    capture_start: Option<Instant>,
    last_pts_emitted: Option<i64>,
}

impl StatePortal {
    /// 创建门户状态实例
    ///
    /// 初始化 DRM 设备路径和 PipeWire 采集端点，编码器延迟到第一帧到达时创建。
    pub fn new(args: Args) -> Result<Self> {
        let drm_device = resolve_drm_device(&args)?;
        if let Some(ref drm_device) = drm_device {
            tracing::info!("Using DRM device: {}", drm_device.display());
        } else {
            tracing::info!("DRM device auto-detection enabled");
        }

        let cap = CapPortal::new(&args)?;

        let (webrtc, webrtc_paused) = if args.port > 0 {
            let wrtc = WebRtcState::new(args.port, args.fps)?;
            let paused = Arc::new(AtomicBool::new(true));
            (Some(wrtc), Some(paused))
        } else {
            (None, None)
        };

        Ok(Self {
            stage: PortalStage::WaitingForFormat,
            enc: None,
            enc_import: None,
            enc_thread: None,
            cap,
            args,
            errored: false,
            drm_device,
            frames_encoded: 0,
            start_time: None,
            stats: PipelineStats::new(),
            pw_dropped_prev: 0,
            webrtc,
            webrtc_thread: None,
            webrtc_paused,
            last_capture_arrival: None,
            idle_log_start: None,
            shutdown_started: false,
            first_pts_ns: None,
            capture_start: None,
            last_pts_emitted: None,
        })
    }

    /// 轮询 PipeWire 事件并编码帧
    ///
    /// `block=true` 时使用 recv_timeout 阻塞等待帧（最多 2ms），
    /// `block=false` 时使用 try_recv 非阻塞检查。
    /// 返回 `Ok(true)` 表示已处理事件，`Ok(false)` 表示暂无数据。
    pub fn poll_and_encode(&mut self, block: bool) -> Result<bool> {
        // 检查 PipeWire 控制事件（流结束 / 错误）
        if let Ok(ctrl) = self.cap.event_receiver().try_recv() {
            match ctrl {
                PwCtrlEvent::StreamEnded => {
                    tracing::warn!("PipeWire stream ended");
                    self.errored = true;
                    return Ok(true);
                }
                PwCtrlEvent::Error(e) => {
                    tracing::error!("PipeWire error: {e}");
                    self.errored = true;
                    return Ok(true);
                }
                PwCtrlEvent::FormatChanged { width, height } => {
                    tracing::warn!(
                        "PipeWire format renegotiation: new dimensions {}x{} — encoder output remains at original resolution",
                        width,
                        height
                    );
                    // No action yet — VAAPI import/scale handles the conversion.
                    // Full encoder reinit is a future enhancement.
                }
            }
        }

        // 根据阻塞模式选择不同的帧接收策略
        let frame = if block {
            // 阻塞模式：最多等待 2ms 接收帧
            match self
                .cap
                .frame_receiver()
                .recv_timeout(std::time::Duration::from_millis(2))
            {
                Ok(frame) => frame,
                Err(_) => {
                    self.record_capture_timeout();
                    return Ok(false);
                }
            }
        } else {
            // 非阻塞模式：立即尝试接收，无数据则返回
            match self.cap.frame_receiver().try_recv() {
                Ok(frame) => frame,
                Err(_) => {
                    self.record_capture_timeout();
                    return Ok(false);
                }
            }
        };
        self.record_frame_arrival();

        match self.stage {
            PortalStage::WaitingForFormat => {
                tracing::info!(
                    "First DMA-BUF frame: {}x{} format=0x{:08X} stride={} modifier=0x{:X}",
                    frame.width,
                    frame.height,
                    frame.format,
                    frame.stride,
                    frame.modifier
                );

                // 自动检测或确认 DRM 设备是否支持导入该帧
                let drm_path = self.resolve_drm_device_for_frame(&frame)?;
                // 计算编码目标分辨率（不超过 2560x1440）
                let (enc_width, enc_height) = portal_encode_dimensions(frame.width, frame.height);
                tracing::info!(
                    "Portal software encode target: {}x{} -> {}x{} @ {} fps",
                    frame.width,
                    frame.height,
                    enc_width,
                    enc_height,
                    self.args.fps,
                );
                // 码率:WebRTC 模式用保守默认(BWE 连接后立即覆盖),MP4 用公式
                let actual_bitrate = self.args.bitrate.unwrap_or_else(|| {
                    if self.webrtc.is_some() {
                        webrtc_startup_bitrate_bps(enc_width, enc_height)
                    } else {
                        5 * (enc_width as u64) * (enc_height as u64) * (self.args.fps as u64) / 100
                    }
                });
                // GOP 大小：WebRTC 模式使用较大的 GOP（fps*2，最低20），MP4 模式使用 fps
                let actual_gop_size = self.args.gop_size.unwrap_or_else(|| {
                    if self.webrtc.is_some() {
                        (self.args.fps * 2).max(20)
                    } else {
                        self.args.fps
                    }
                });

                // 根据是否启用 WebRTC 选择不同的编码器构造方式
                if self.webrtc.is_some() {
                    let paused = self.webrtc_paused.as_ref()
                        .ok_or_else(|| anyhow::anyhow!("internal invariant broken: webrtc_paused missing while WebRTC mode is active"))?;
                    let (resolution_tx, resolution_rx) =
                        crossbeam_channel::bounded::<BitrateCommand>(4);
                    let (encoder_resolution_tx, encoder_resolution_rx) =
                        crossbeam_channel::bounded::<ResolutionChange>(4);
                    let import = SwEncImport::new_with_resolution_control(
                        &drm_path,
                        frame.width,
                        frame.height,
                        enc_width,
                        enc_height,
                        self.args.fps,
                        resolution_rx,
                        encoder_resolution_tx,
                    )?;
                    let (webrtc_tx, webrtc_rx) = crossbeam_channel::bounded(2);
                    let (input_tx, input_rx) = crossbeam_channel::bounded::<CpuNv12Frame>(1);
                    let (timing_tx, timing_rx) =
                        crossbeam_channel::bounded::<EncodeThreadTiming>(32);
                    let (bitrate_tx, bitrate_rx) = crossbeam_channel::bounded::<BitrateCommand>(4);
                    let encode = SwEncEncode::new_webrtc(
                        enc_width,
                        enc_height,
                        self.args.fps,
                        actual_bitrate,
                        actual_gop_size,
                        webrtc_tx,
                        paused.clone(),
                        bitrate_rx,
                        encoder_resolution_rx,
                    )?;
                    let duplicate_count = std::sync::Arc::new(
                        std::sync::atomic::AtomicU64::new(0),
                    );
                    let duplicate_count_for_thread = duplicate_count.clone();
                    let handle = std::thread::Builder::new()
                        .name("wl-webrtc-encode".into())
                        .spawn(move || {
                            encode_thread_loop(
                                encode,
                                input_rx,
                                timing_tx,
                                duplicate_count_for_thread,
                            )
                        })?;
                    self.enc_import = Some(import);
                    self.enc_thread = Some(EncodeThread {
                        handle: Some(handle),
                        input_tx,
                        timing_rx,
                        duplicate_count,
                    });

                    let wrtc = self.webrtc.take().ok_or_else(|| {
                        anyhow::anyhow!("internal: WebRtcState missing during init")
                    })?;
                    let paused = self
                        .webrtc_paused
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("internal: webrtc_paused missing"))?
                        .clone();
                    let fps = self.args.fps;
                    let max_bitrate = self.args.max_bitrate;
                    let (sent_gap_tx, sent_gap_rx) =
                        crossbeam_channel::bounded::<(f64, Option<f64>)>(64);
                    let webrtc_handle = std::thread::Builder::new()
                        .name("wl-webrtc-webrtc".into())
                        .spawn(move || {
                            webrtc_thread_loop(
                                wrtc,
                                WebRtcThreadConfig {
                                    fps,
                                    enc_width,
                                    enc_height,
                                    max_bitrate,
                                },
                                WebRtcThreadChannels {
                                    webrtc_rx,
                                    sent_gap_tx,
                                    bitrate_tx,
                                    resolution_tx,
                                },
                                paused,
                            )
                        })?;
                    self.webrtc_thread = Some(WebrtcThread {
                        handle: Some(webrtc_handle),
                        sent_gap_rx,
                    });
                } else {
                    // MP4 模式：编码输出写入文件
                    let output_path = self.args.output.as_deref()
                        .ok_or_else(|| anyhow::anyhow!("--output is required in MP4 file output mode; use --port > 0 for WebRTC mode"))?;
                    let enc = avhw::SwEncState::new(
                        &drm_path,
                        std::path::Path::new(output_path),
                        frame.width,
                        frame.height,
                        enc_width,
                        enc_height,
                        self.args.fps,
                        actual_bitrate,
                        actual_gop_size,
                    )?;
                    self.enc = Some(enc);
                };
                self.stage = PortalStage::Streaming; // 切换到流式编码阶段
                self.start_time = Some(Instant::now());
                tracing::info!(
                    "First frame processed, encoder initialized, transitioning to Streaming"
                );
                drop(frame); // 首帧仅用于初始化，不参与编码
            }
            PortalStage::Streaming => {
                // 记录采集帧到达（用于 capture gap 和 capture_fps 统计）
                self.stats.record_capture();
                self.last_capture_arrival = Some(Instant::now());
                // 流式编码阶段：直接处理帧
                self.handle_pw_frame(frame)?;
            }
        }

        // 每秒输出一次结构化管道统计（仅 --stats 启用时记录日志）
        if self.args.stats && self.stats.should_snapshot() {
            // Wire PipeWire drop counter (delta-tracked via pw_dropped_prev) and
            // capture channel depth. Oracle audit 2026-06-28: previously hardcoded
            // (0, 0), which silently zeroed two real diagnostic fields.
            let total_dropped = self.cap.dropped_count();
            self.stats.set_pipewire_dropped(total_dropped, self.pw_dropped_prev);
            self.pw_dropped_prev = total_dropped;
            // capture queue depth is real; encoded side has no exposed depth — the
            // encoder thread publishes timings only, not a frame queue length.
            self.stats.set_queue_depths(self.cap.capture_queue_depth(), 0);
            if let Some(ref enc_thread) = self.enc_thread {
                while let Ok(timing) = enc_thread.timing_rx.try_recv() {
                    self.stats.record_encode_thread(
                        timing.sws_us,
                        timing.encode_us,
                        timing.output_bytes,
                    );
                }
                // Read duplicate counter (delta computed in setter)
                let total = enc_thread
                    .duplicate_count
                    .load(std::sync::atomic::Ordering::Relaxed);
                self.stats.set_duplicate_frames_skipped(total);
            }
            if let Some(ref webrtc_thread) = self.webrtc_thread {
                while let Ok((gap_ms, age_ms)) = webrtc_thread.sent_gap_rx.try_recv() {
                    self.stats.record_send_from_thread(gap_ms, age_ms);
                }
            }
            let snap = self.stats.snapshot_and_reset();
            tracing::info!("stats: {snap}");
        }

        Ok(true)
    }

    fn record_capture_timeout(&mut self) {
        let Some(last_capture_arrival) = self.last_capture_arrival else {
            return;
        };

        let now = Instant::now();
        // Wayland damage-driven delivery: static content means no new frames.
        // This is normal Wayland behavior, not a compositor hang. Only log DEBUG
        // after a meaningful idle period, and only once per idle episode.
        // See issues #15 and #18.
        const CAPTURE_IDLE_LOG_THRESHOLD: Duration = Duration::from_secs(5);
        if now.duration_since(last_capture_arrival) <= CAPTURE_IDLE_LOG_THRESHOLD {
            return;
        }

        if self.idle_log_start.is_none() {
            // Use last_capture_arrival as idle start for accurate elapsed duration.
            self.idle_log_start = Some(last_capture_arrival);
            tracing::debug!(
                elapsed_ms = now.duration_since(last_capture_arrival).as_millis(),
                "portal capture idle; no damage frames received (normal Wayland behavior)"
            );
        }
    }

    fn record_frame_arrival(&mut self) {
        if let Some(idle_start) = self.idle_log_start.take() {
            tracing::debug!(
                idle_ms = idle_start.elapsed().as_millis(),
                "portal capture resumed after idle period"
            );
        }
        self.last_capture_arrival = Some(Instant::now());
    }

    /// 为当前帧解析可用的 DRM 渲染设备
    ///
    /// 如果用户已通过 `--drm-device` 指定设备，直接返回；
    /// 否则遍历系统中所有 DRM render node，逐个尝试导入 DMA-BUF 帧来找到兼容设备。
    fn resolve_drm_device_for_frame(&mut self, frame: &PwDmaBufFrame) -> Result<PathBuf> {
        // 用户已显式指定 DRM 设备，直接使用
        if let Some(ref drm) = self.drm_device {
            return Ok(drm.clone());
        }

        // 查找系统中所有 DRM render node（如 /dev/dri/renderD128）
        let candidates = crate::state::find_drm_render_nodes();
        if candidates.is_empty() {
            bail!("No DRM render device found. Specify --drm-device.");
        }

        // 逐个尝试导入 DMA-BUF 帧，找到第一个兼容的设备
        let mut failures = Vec::new();
        for candidate in &candidates {
            match crate::avhw::test_dma_buf_import(candidate, frame) {
                Ok(()) => {
                    // 成功导入，缓存检测结果并返回
                    tracing::info!(
                        "Auto-detected DRM device: {} (tested {} candidates)",
                        candidate.display(),
                        candidates.len(),
                    );
                    self.drm_device = Some(candidate.clone());
                    return Ok(candidate.clone());
                }
                Err(e) => {
                    // 导入失败，记录原因，继续尝试下一个设备
                    tracing::debug!(
                        "DRM device {} cannot import DMA-BUF: {e}",
                        candidate.display(),
                    );
                    failures.push((candidate, e));
                }
            }
        }

        // 所有候选设备均失败，返回详细错误信息
        bail!(failures
            .into_iter()
            .map(|(p, e)| format!("{} ({e})", p.display()))
            .collect::<Vec<_>>()
            .join(", "));
    }

    /// 处理单帧 DMA-BUF 数据
    ///
    /// 通过 `av_hwframe_map` 零拷贝导入 VAAPI，然后交给 SwEncState 完成：
    /// scale_vaapi GPU 缩放、2K NV12 回读、YUV420P 格式转换、软件 H.264 编码。
    fn handle_pw_frame(&mut self, frame: PwDmaBufFrame) -> Result<()> {
        // #19: When WebRTC mode is paused (no client connected), skip ALL frame
        // processing — DMA-BUF import, VAAPI scale, NV12 clone, channel send, and
        // encode thread wakeup. This eliminates ~60fps of pointless work during
        // the pre-connect idle window. MP4 mode (webrtc_paused == None) is unaffected.
        if let Some(paused) = &self.webrtc_paused {
            if paused.load(Ordering::Relaxed) {
                return Ok(());
            }
        }
        let t_import_start = Instant::now();
        // WebRTC: use real PipeWire capture time so RTP timestamps reflect reality
        // (sequential counter caused client jitter buffers to grow to 2-3s under
        // damage-driven variable fps — issue #24). MP4: keep sequential counter;
        // file output doesn't need real-time PTS and changing it would alter
        // playback speed during static periods.
        let pts = if self.webrtc_thread.is_some() {
            self.compute_capture_pts(frame.pts)
        } else {
            self.frames_encoded as i64
        };

        if let Some(enc) = self.enc.as_mut() {
            // 将 DMA-BUF 帧零拷贝导入 VAAPI 硬件帧池
            // SAFETY: delegates to avhw::import_dma_buf_to_vaapi (itself an unsafe fn);
            // frames_rgb pointer is a valid AVBufferRef owned by enc, and `frame` is the
            // PipeWire-formatted PwDmaBufFrame whose metadata the function reads directly.
            // See that function's own SAFETY contract.
            let mut vaapi_frame = unsafe {
                avhw::import_dma_buf_to_vaapi(enc.frames_rgb().as_ptr(), &frame)
            }?;

            let import_us = t_import_start.elapsed().as_micros() as u64;

            // 设置帧的显示时间戳（PTS），基于已编码帧序号
            // SAFETY: vaapi_frame is the freshly-imported valid AVFrame returned by
            // import_dma_buf_to_vaapi above; pts is a plain i64 field on AVFrame.
            unsafe {
                (*vaapi_frame.as_mut_ptr()).pts = pts;
            }

            // 送入编码器完成：缩放 → 回读 → 格式转换 → H.264 编码
            let stages = enc.encode_frame(&vaapi_frame)?;
            let total_us = t_import_start.elapsed().as_micros() as u64;
            let encode_us = stages.encode_us;

            self.frames_encoded += 1;

            // 记录帧计时到管道统计（scale 来自 filter graph；transfer 在 HW 路径恒为 0）
            let timings = FrameTimings {
                import_us,
                scale_us: stages.scale_us,
                transfer_us: stages.transfer_us,
                encode_us,
                total_us,
                ..Default::default()
            };
            self.stats.record_encode(&timings);
        } else if let Some(import) = self.enc_import.as_mut() {
            // SAFETY: same contract as the enc branch above — frames_rgb owned by
            // import, `frame` carries the PipeWire DMA-BUF metadata.
            let mut vaapi_frame = unsafe {
                avhw::import_dma_buf_to_vaapi(import.frames_rgb().as_ptr(), &frame)
            }?;
            // SAFETY: vaapi_frame is the valid AVFrame returned above; pts is plain i64.
            unsafe {
                (*vaapi_frame.as_mut_ptr()).pts = pts;
            }

            let cpu_nv12 = import.import_and_scale(&vaapi_frame)?;
            let import_us = t_import_start.elapsed().as_micros() as u64;
            self.stats.record_import(import_us);

            let enc_thread = self.enc_thread.as_ref().ok_or_else(|| {
                anyhow::anyhow!(
                    "internal invariant broken: encode thread missing while async import is active"
                )
            })?;
            match enc_thread.input_tx.try_send(cpu_nv12) {
                Ok(()) => {
                    self.frames_encoded += 1;
                }
                Err(crossbeam_channel::TrySendError::Full(_)) => {
                    tracing::debug!("Encode thread input full, dropping portal frame");
                }
                Err(crossbeam_channel::TrySendError::Disconnected(_frame)) => {
                    tracing::error!("Encode thread input disconnected");
                    self.errored = true;
                }
            }
        } else {
            bail!("encoder not initialized");
        }

        Ok(())
    }

    /// Compute PTS in 90kHz media-clock ticks from PipeWire's nanosecond
    /// capture timestamp. Falls back to `Instant`-based elapsed time when PipeWire
    /// does not provide PTS. Maintains strict monotonicity (encoder requirement).
    fn compute_capture_pts(&mut self, pw_pts_ns: i64) -> i64 {
        const NS_PER_SEC: i128 = 1_000_000_000;

        let raw_ns: i128 = if pw_pts_ns > 0 {
            i128::from(pw_pts_ns)
        } else {
            let start = self.capture_start.get_or_insert_with(Instant::now);
            i128::try_from(start.elapsed().as_nanos()).unwrap_or(0)
        };

        if self.first_pts_ns.is_none() && raw_ns > 0 {
            self.first_pts_ns = Some(raw_ns);
        }

        let origin = self.first_pts_ns.unwrap_or(0);
        let relative_ns = if raw_ns >= origin {
            raw_ns - origin
        } else {
            // PipeWire PTS went backwards (stream restart) — reset origin.
            self.first_pts_ns = Some(raw_ns);
            0
        };
        let ticks_i128 =
            (relative_ns.saturating_mul(crate::avhw::WEBRTC_RTP_CLOCK_HZ)) / NS_PER_SEC;
        let computed_pts = i64::try_from(ticks_i128).unwrap_or(i64::MAX);
        let mut pts = computed_pts;

        if let Some(last) = self.last_pts_emitted {
            if pts <= last {
                pts = last.checked_add(1).unwrap_or(last);
            }
        }
        self.last_pts_emitted = Some(pts);
        pts
    }

    /// 关闭状态：刷新编码器并清理资源（幂等）。
    ///
    /// `shutdown_started` 守卫在清理之前置位——防止 panic 时 `Drop` 重入 unwinding。
    pub fn shutdown(&mut self) {
        if self.shutdown_started {
            return;
        }
        self.shutdown_started = true;

        // 1. Stop encode thread (drops webrtc_tx → signals WebRTC thread to exit)
        if let Some(mut enc_thread) = self.enc_thread.take() {
            drop(enc_thread.input_tx);
            if let Some(handle) = enc_thread.handle.take() {
                if handle.join().is_err() {
                    tracing::error!("Encode thread panicked during shutdown");
                }
            }
        }
        self.enc_import = None;
        // 2. Wait for WebRTC thread (exits when webrtc_tx is dropped by encode thread)
        if let Some(mut webrtc_thread) = self.webrtc_thread.take() {
            if let Some(handle) = webrtc_thread.handle.take() {
                if handle.join().is_err() {
                    tracing::error!("WebRTC thread panicked during shutdown");
                }
            }
        }
        // 3. Flush MP4 encoder if present
        if let Some(mut enc) = self.enc.take() {
            if let Err(e) = enc.flush() {
                tracing::error!("Flush error during shutdown: {e}");
            }
        }
        if let Some(start) = self.start_time {
            if self.frames_encoded > 0 {
                let elapsed = start.elapsed().as_secs_f64();
                let fps = self.frames_encoded as f64 / elapsed;
                tracing::info!(
                    "Total: {} frames in {:.1}s, avg {:.1}fps",
                    self.frames_encoded,
                    elapsed,
                    fps,
                );
            }
        }
        tracing::info!("StatePortal shutdown complete");
    }

    /// 返回是否遇到不可恢复的错误
    pub fn is_errored(&self) -> bool {
        self.errored
    }
}

fn encode_thread_loop(
    mut encode: SwEncEncode,
    input_rx: crossbeam_channel::Receiver<CpuNv12Frame>,
    timing_tx: crossbeam_channel::Sender<EncodeThreadTiming>,
    duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
) {
    loop {
        match input_rx.recv() {
            Ok(frame) => {
                match encode.encode_cpu_frame(&frame) {
                    Ok(EncodeOutcome::Encoded) => {
                        let t = encode.take_timing();
                        let _ = timing_tx.try_send(EncodeThreadTiming {
                            sws_us: t.sws_us,
                            encode_us: t.encode_us,
                            output_bytes: t.output_bytes,
                        });
                    }
                    Ok(EncodeOutcome::SkippedDuplicate) => {
                        duplicate_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    }
                    Ok(_) => {
                        // SkippedPaused / SkippedDisconnected — no counter needed
                    }
                    Err(e) => {
                        tracing::error!("Encode thread error: {e}");
                        break;
                    }
                }
            }
            Err(_) => {
                tracing::info!("Encode thread input closed, flushing encoder");
                if let Err(e) = encode.flush() {
                    tracing::error!("Encode thread flush error: {e}");
                }
                break;
            }
        }
    }
    tracing::info!("Encode thread exiting");
}

fn webrtc_thread_loop(
    mut wrtc: WebRtcState,
    config: WebRtcThreadConfig,
    channels: WebRtcThreadChannels,
    paused: Arc<AtomicBool>,
) {
    let WebRtcThreadConfig {
        fps,
        enc_width,
        enc_height,
        max_bitrate,
    } = config;
    let WebRtcThreadChannels {
        webrtc_rx,
        sent_gap_tx,
        bitrate_tx,
        resolution_tx,
    } = channels;
    let mut frames_sent: u64 = 0;
    let mut last_send: Option<std::time::Instant> = None;
    let mut last_sent_bitrate: Option<u64> = None;
    let initial_tier = (enc_width, enc_height);
    let mut current_tier = initial_tier;
    let mut upscale_counter = 0u32;
    let mut last_resolution_eval = Instant::now();
    let timeout = Duration::from_millis(1);

    loop {
        if let Err(e) = wrtc.handle_signaling() {
            tracing::error!("WebRTC signaling error: {e}");
            break;
        }
        if let Err(e) = wrtc.poll_and_feed() {
            tracing::error!("WebRTC poll error: {e}");
            break;
        }

        if wrtc.take_force_keyframe() {
            let _ = bitrate_tx.try_send(BitrateCommand::ForceKeyframe);
        }

        let connected = wrtc.is_connected();
        let was_paused = paused.load(Ordering::Relaxed);
        let now_paused = !connected;
        if was_paused && !now_paused {
            tracing::info!("WebRTC client connected, resuming encoding");
        } else if !was_paused && now_paused {
            tracing::warn!("WebRTC client disconnected, pausing encoding");
        }
        paused.store(now_paused, Ordering::Relaxed);

        if let Some(bwe) = wrtc.get_bwe_estimate() {
            // #23: Cap BWE to prevent runaway bitrate escalation. Without this, BWE
            // estimates can rise to 10+ Mbps, causing IDR bursts and PLI storms.
            let effective_bwe = bwe.min(max_bitrate);
            if effective_bwe != bwe {
                tracing::debug!(
                    bwe,
                    effective_bwe,
                    max_bitrate,
                    "BWE exceeds --max-bitrate cap, clamping"
                );
            }
            let bwe = effective_bwe;

            let should_send = match last_sent_bitrate {
                None => true,
                Some(last) => {
                    let diff = bwe.abs_diff(last);
                    diff * 10 > last
                }
            };
            if should_send {
                let _ = bitrate_tx.try_send(BitrateCommand::UpdateBitrate { target_bps: bwe });
                last_sent_bitrate = Some(bwe);
            }

            if last_resolution_eval.elapsed() >= Duration::from_secs(1) {
                last_resolution_eval = Instant::now();
                let selected = select_resolution(current_tier.0, current_tier.1, bwe, fps);
                if selected != current_tier {
                    current_tier = selected;
                    upscale_counter = 0;
                    let _ = resolution_tx.try_send(BitrateCommand::UpdateResolution {
                        width: current_tier.0,
                        height: current_tier.1,
                    });
                    wrtc.set_need_keyframe();
                } else if let Some(next_tier) = next_upscale_tier(current_tier, initial_tier) {
                    let needed = resolution_bitrate_bps(next_tier.0, next_tier.1, fps);
                    if bwe > needed.saturating_mul(120) / 100 {
                        upscale_counter = upscale_counter.saturating_add(1);
                        if upscale_counter >= 10 {
                            current_tier = next_tier;
                            upscale_counter = 0;
                            let _ = resolution_tx.try_send(BitrateCommand::UpdateResolution {
                                width: current_tier.0,
                                height: current_tier.1,
                            });
                            wrtc.set_need_keyframe();
                        }
                    } else {
                        upscale_counter = 0;
                    }
                } else {
                    upscale_counter = 0;
                }
            }
        }

        if connected {
            while let Ok(enc_frame) = webrtc_rx.try_recv() {
                if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks) {
                    tracing::debug!("WebRTC write frame error: {e}");
                }
                frames_sent = frames_sent.saturating_add(1);
                let gap_ms = last_send
                    .map(|l| l.elapsed().as_secs_f64() * 1000.0)
                    .unwrap_or(0.0);
                // Compute capture-to-send age on the sending thread so the
                // frame_age stat stays accurate when batch-drained later.
                let age_ms =
                    Some(enc_frame.capture_time.elapsed().as_secs_f64() * 1000.0);
                last_send = Some(std::time::Instant::now());
                let _ = sent_gap_tx.try_send((gap_ms, age_ms));
            }
        } else {
            while webrtc_rx.try_recv().is_ok() {}
        }

        match webrtc_rx.recv_timeout(timeout) {
            Ok(enc_frame) => {
                if wrtc.is_connected() {
                    if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks)
                    {
                        tracing::debug!("WebRTC write frame error: {e}");
                    }
                    frames_sent = frames_sent.saturating_add(1);
                    let gap_ms = last_send
                        .map(|l| l.elapsed().as_secs_f64() * 1000.0)
                        .unwrap_or(0.0);
                    let age_ms =
                        Some(enc_frame.capture_time.elapsed().as_secs_f64() * 1000.0);
                    last_send = Some(std::time::Instant::now());
                    let _ = sent_gap_tx.try_send((gap_ms, age_ms));
                }
            }
            Err(crossbeam_channel::RecvTimeoutError::Timeout) => {}
            Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
                tracing::info!("WebRTC channel disconnected, exiting thread");
                return;
            }
        }
    }

    tracing::info!("WebRTC thread exiting");
}

const RESOLUTION_TIERS: &[(u32, u32)] = &[(2560, 1440), (1920, 1080), (1280, 720)];

fn resolution_bitrate_bps(width: u32, height: u32, fps: u32) -> u64 {
    5 * u64::from(width) * u64::from(height) * u64::from(fps) / 100
}

/// Conservative startup bitrate for WebRTC mode, tier-based by total pixel count.
/// BWE estimate arrives within milliseconds of client connect and overrides this;
/// the startup value only affects the first IDR. See issue #21.
fn webrtc_startup_bitrate_bps(width: u32, height: u32) -> u64 {
    let pixels = u64::from(width) * u64::from(height);
    if pixels <= 1_000_000 {
        1_000_000
    } else if pixels <= 2_500_000 {
        2_000_000
    } else if pixels <= 4_500_000 {
        4_000_000
    } else {
        8_000_000
    }
}

/// Select resolution tier based on BWE estimate.
/// Returns (width, height) for the selected tier.
fn select_resolution(current_w: u32, current_h: u32, bwe_bps: u64, fps: u32) -> (u32, u32) {
    let current = (current_w, current_h);
    let current_bitrate = resolution_bitrate_bps(current_w, current_h, fps);
    if bwe_bps >= current_bitrate.saturating_mul(60) / 100 {
        return current;
    }

    let current_index = RESOLUTION_TIERS
        .iter()
        .position(|&tier| tier == current)
        .unwrap_or_else(|| {
            RESOLUTION_TIERS
                .iter()
                .position(|&(w, h)| w <= current_w && h <= current_h)
                .unwrap_or(RESOLUTION_TIERS.len() - 1)
        });
    let next_index = (current_index + 1).min(RESOLUTION_TIERS.len() - 1);
    RESOLUTION_TIERS[next_index]
}

fn next_upscale_tier(current: (u32, u32), ceiling: (u32, u32)) -> Option<(u32, u32)> {
    let current_index = RESOLUTION_TIERS.iter().position(|&tier| tier == current)?;
    if current_index == 0 {
        return None;
    }
    let next = RESOLUTION_TIERS[current_index - 1];
    (next.0 <= ceiling.0 && next.1 <= ceiling.1).then_some(next)
}

impl Drop for StatePortal {
    // 析构时自动调用 shutdown，确保编码器被刷新、资源被释放
    fn drop(&mut self) {
        self.shutdown();
    }
}

/// 计算编码目标分辨率
///
/// 将原始分辨率等比缩放至不超过 2560×1440（2K），并确保宽高为偶数
/// （H.264 编码要求偶数尺寸）。
fn portal_encode_dimensions(width: u32, height: u32) -> (u32, u32) {
    const TARGET_W: u32 = 2560; // 目标最大宽度
    const TARGET_H: u32 = 1440; // 目标最大高度

    // 原始分辨率已在 2K 以内，直接对齐偶数
    if width <= TARGET_W && height <= TARGET_H {
        return (width & !1, height & !1); // & !1 确保为偶数
    }

    // 按宽度限制等比缩放
    let width_limited_h = ((height as u64) * (TARGET_W as u64) / (width as u64)) as u32;
    if width_limited_h <= TARGET_H {
        (TARGET_W & !1, width_limited_h & !1)
    } else {
        // 按高度限制等比缩放
        let height_limited_w = ((width as u64) * (TARGET_H as u64) / (height as u64)) as u32;
        (height_limited_w & !1, TARGET_H & !1)
    }
}

/// 解析 DRM 渲染设备路径
///
/// 仅使用命令行指定的设备路径；未指定则在首帧到达时自动检测。
fn resolve_drm_device(args: &Args) -> Result<Option<PathBuf>> {
    if let Some(ref drm) = args.drm_device {
        return Ok(Some(PathBuf::from(drm)));
    }
    Ok(None)
}

/// 构建测试用的 AVDRMFrameDescriptor（仅测试用途）
///
/// 将 PwDmaBufFrame 转换为 FFmpeg 的 DRM 帧描述符结构体，
/// 用于验证 DMA-BUF 元数据映射的正确性。
#[cfg(test)]
fn build_drm_descriptor(frame: &PwDmaBufFrame) -> ffmpeg_next::ffi::AVDRMFrameDescriptor {
    let mut desc: ffmpeg_next::ffi::AVDRMFrameDescriptor = {
        // SAFETY: AVDRMFrameDescriptor is a POD struct from FFmpeg's C API with no
        // pointers orDrop fields; all-zero is a valid initial state. Every field is
        // explicitly overwritten in the lines below before the descriptor is used.
        unsafe { std::mem::zeroed() }
    };
    desc.nb_objects = 1; // 单个 DMA-BUF 对象
    desc.objects[0].fd = frame.fd.as_raw_fd(); // DMA-BUF 文件描述符
    desc.objects[0].size = 0; // 大小设为 0（内核自动确定）
    desc.objects[0].format_modifier = frame.modifier; // DRM 格式修饰符（如线性、tiled）
    desc.nb_layers = 1; // 单层
    desc.layers[0].format = frame.format; // 像素格式（如 XR24）
    desc.layers[0].nb_planes = 1; // 单平面
    desc.layers[0].planes[0].object_index = 0; // 指向第 0 个对象
    desc.layers[0].planes[0].offset = frame.offset as isize; // 帧数据偏移
    desc.layers[0].planes[0].pitch = frame.stride as isize; // 行跨度（stride）
    desc
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::fd::{FromRawFd, OwnedFd};

    /// 创建测试用的 DMA-BUF 帧数据（使用 stderr fd 的副本作为占位）
    fn make_test_frame() -> PwDmaBufFrame {
        // Create a dummy fd from stderr (always valid fd 2)
        // 使用 stderr（fd 2）的副本作为虚拟文件描述符
        // SAFETY: stderr (fd 2) is always-open in any process; libc::dup(2) returns
        // a fresh fd we solely own. OwnedFd::from_raw_fd takes ownership and closes
        // it on Drop. Test-only; the fd is never actually memory-mapped.
        let fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) };
        PwDmaBufFrame {
            fd,
            offset: 0,
            stride: 1920 * 4, // 每行 1920 像素 × 4 字节（XRGB）
            modifier: 0,      // DRM_FORMAT_MOD_LINEAR（线性布局）
            width: 1920,
            height: 1080,
            format: 0x34325258, // XR24 little-endian（XRGB8888）
            pts: 12345,
        }
    }

    /// 测试 DRM 描述符构建（单平面情况）
    #[test]
    fn build_drm_descriptor_single_plane() {
        let frame = make_test_frame();
        let desc = build_drm_descriptor(&frame);

        assert_eq!(desc.nb_objects, 1);
        assert_eq!(desc.objects[0].format_modifier, 0);
        assert_eq!(desc.nb_layers, 1);
        assert_eq!(desc.layers[0].format, 0x34325258);
        assert_eq!(desc.layers[0].nb_planes, 1);
        assert_eq!(desc.layers[0].planes[0].object_index, 0);
        assert_eq!(desc.layers[0].planes[0].offset, 0);
        assert_eq!(desc.layers[0].planes[0].pitch, 1920 * 4);
    }

    /// 测试显式指定 DRM 设备时的解析
    #[test]
    fn resolve_drm_device_explicit() {
        let args = Args {
            output: Some("test.mp4".to_string()),
            output_name: None,
            fps: 30,
            codec: "h264".to_string(),
            hw_accel: "vaapi".to_string(),
            drm_device: Some("/dev/dri/renderD128".to_string()),
            bitrate: None,
            max_bitrate: 8_000_000,
            gop_size: None,
            verbose: false,
            backend: None,
            port: 0,
            no_persist: false,
            stats: false,
        };
        let result = resolve_drm_device(&args).unwrap();
        assert_eq!(
            result,
            Some(std::path::PathBuf::from("/dev/dri/renderD128"))
        );
    }

    #[test]
    fn resolve_drm_device_none_when_not_specified() {
        let args = Args {
            output: Some("test.mp4".to_string()),
            output_name: None,
            fps: 30,
            codec: "h264".to_string(),
            hw_accel: "vaapi".to_string(),
            drm_device: None,
            bitrate: None,
            max_bitrate: 8_000_000,
            gop_size: None,
            verbose: false,
            backend: None,
            port: 0,
            no_persist: false,
            stats: false,
        };
        let result = resolve_drm_device(&args).unwrap();
        assert_eq!(result, None);
    }

    #[test]
    fn webrtc_startup_bitrate_tiers_by_pixel_count() {
        assert_eq!(webrtc_startup_bitrate_bps(1280, 720), 1_000_000);
        assert_eq!(webrtc_startup_bitrate_bps(1920, 1080), 2_000_000);
        assert_eq!(webrtc_startup_bitrate_bps(2560, 1440), 4_000_000);
        assert_eq!(webrtc_startup_bitrate_bps(3840, 2160), 8_000_000);
    }

    #[test]
    fn select_resolution_downscales_one_tier_below_sixty_percent() {
        let fps = 30;
        let current = resolution_bitrate_bps(1920, 1080, fps);
        assert_eq!(
            select_resolution(1920, 1080, current * 59 / 100, fps),
            (1280, 720)
        );
    }

    #[test]
    fn select_resolution_keeps_tier_at_sixty_percent() {
        let fps = 30;
        let current = resolution_bitrate_bps(1920, 1080, fps);
        assert_eq!(
            select_resolution(1920, 1080, current * 60 / 100, fps),
            (1920, 1080)
        );
    }

    #[test]
    fn select_resolution_never_goes_below_720p() {
        assert_eq!(select_resolution(1280, 720, 1, 30), (1280, 720));
    }

    #[test]
    fn next_upscale_tier_respects_initial_ceiling() {
        assert_eq!(
            next_upscale_tier((1280, 720), (1920, 1080)),
            Some((1920, 1080))
        );
        assert_eq!(next_upscale_tier((1920, 1080), (1920, 1080)), None);
    }

    /// 测试：使用自定义偏移量和 stride 构建 DRM 描述符
    #[test]
    fn build_drm_descriptor_custom_offset_and_stride() {
        // SAFETY: same as make_test_frame — dup of stderr (fd 2), test-only.
        let test_fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) };
        let frame = PwDmaBufFrame {
            fd: test_fd,
            offset: 4096,                 // 4KB 对齐偏移
            stride: 3840 * 4,             // 4K 宽度 × 4 字节
            modifier: 0x0100000000000001, // AMD modifiers
            width: 3840,
            height: 2160,
            format: 0x34325258,
            pts: 0,
        };
        let desc = build_drm_descriptor(&frame);

        assert_eq!(desc.nb_objects, 1);
        assert_eq!(desc.objects[0].format_modifier, 0x0100000000000001);
        assert_eq!(desc.layers[0].planes[0].offset, 4096);
        assert_eq!(desc.layers[0].planes[0].pitch, 3840 * 4);
    }

    // ── issue #8 regression ──

    #[test]
    fn try_send_full_channel_returns_full_not_block() {
        let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
        tx.send(vec![1]).unwrap();
        tx.send(vec![2]).unwrap();

        assert!(matches!(
            tx.try_send(vec![3]),
            Err(crossbeam_channel::TrySendError::Full(_))
        ));
        assert_eq!(rx.len(), 2);
    }

    #[test]
    fn try_send_after_rx_dropped_returns_disconnected() {
        let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
        drop(rx);

        assert!(matches!(
            tx.try_send(vec![1]),
            Err(crossbeam_channel::TrySendError::Disconnected(_))
        ));
    }

    // given: full bounded channel
    // when:  rx is dropped, then try_send
    // expect: Disconnected, not blocking
    #[test]
    fn shutdown_rx_drop_prevents_deadlock_on_full_channel() {
        let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
        tx.send(vec![1]).unwrap();
        tx.send(vec![2]).unwrap();
        drop(rx);

        assert!(matches!(
            tx.try_send(vec![3]),
            Err(crossbeam_channel::TrySendError::Disconnected(_))
        ));
    }

    // ── Task 7: Additional resolution tier edge cases ──

    #[test]
    fn select_resolution_keeps_720p_when_bwe_sufficient() {
        let fps = 30;
        let bitrate_720 = resolution_bitrate_bps(1280, 720, fps);
        assert_eq!(
            select_resolution(1280, 720, bitrate_720, fps),
            (1280, 720)
        );
    }

    #[test]
    fn select_resolution_downscales_1440p_to_1080p() {
        let fps = 30;
        let bitrate_1440 = resolution_bitrate_bps(2560, 1440, fps);
        assert_eq!(
            select_resolution(2560, 1440, bitrate_1440 * 59 / 100, fps),
            (1920, 1080)
        );
    }

    #[test]
    fn select_resolution_1080p_to_720p_at_very_low_bwe() {
        let fps = 30;
        let bitrate_1080 = resolution_bitrate_bps(1920, 1080, fps);
        assert_eq!(
            select_resolution(1920, 1080, bitrate_1080 / 10, fps),
            (1280, 720)
        );
    }

    #[test]
    fn next_upscale_tier_from_720p_to_1080p() {
        assert_eq!(
            next_upscale_tier((1280, 720), (2560, 1440)),
            Some((1920, 1080))
        );
    }

    #[test]
    fn next_upscale_tier_returns_none_at_highest() {
        assert_eq!(next_upscale_tier((2560, 1440), (2560, 1440)), None);
    }
}