car-server-core 0.52.1

//! One CDP screencast per browser, fanned out to the drawer AND to disk
//! recording — with the privacy blackout applied per consumer.
//!
//! `car_browser::ScreencastPump` already fans one capture out to N
//! consumers, and that is the layer this builds on. Two things it
//! deliberately does not do, both of which the drawer needs, live here:
//!
//! - **Per-consumer gating.** Controller ruling R3 says the blackout
//!   (user control, or a pending sign-in) must suspend model-facing
//!   observation AND keep frames out of `browser_record`'s disk output,
//!   while the drawer keeps streaming throughout. That is an asymmetric
//!   gate, so it cannot live in the pump's own fan-out loop (which gates
//!   every consumer at once, by design). [`FrameFanout::subscribe`] takes
//!   a [`FrameAudience`]; [`publish`] skips [`FrameAudience::Model`]
//!   consumers while the shared blackout flag is set.
//! - **Following the active tab.** A pump is attached to ONE page. The
//!   drawer must follow the agent across tab opens/closes/switches, so the
//!   supervisor watches `ChromiumBackend::subscribe_tabs()` and re-attaches
//!   the pump when the active tab changes. Consumers keep their
//!   subscription across a re-attach — the fan-out list lives here, above
//!   the pump, and outlives any single pump generation.
//!
//! Two properties carried over from the pump and preserved here:
//!
//! - **Zero consumers ⇒ zero frames.** The supervisor is spawned on the
//!   first [`FrameFanout::subscribe`] and exits once the consumer list is
//!   empty, stopping CDP capture on its way out. A browser nobody is
//!   watching and nobody is recording pays nothing.
//! - **Frames are change-driven**, so every frame carries a real wall-clock
//!   `captured_at`. It is re-stamped PER CONSUMER against the instant that
//!   consumer subscribed, because consumers arrive at different times and a
//!   recording's manifest measures durations from its own start — not from
//!   whenever the drawer happened to open.

use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use car_browser::{
    ChromiumBackend, FrameReceiver, ScreencastFrame, ScreencastPump, FRAME_CHANNEL_CAP,
};
use tokio::sync::{mpsc, watch, Mutex};

/// How long [`FrameFanout::stop_supervisor`] waits for a supervisor to tear
/// its own pump down before falling back to an abort. Long enough for the
/// `Page.stopScreencast` round trip it is waiting on, short enough that a
/// wedged CDP connection cannot stall `browser_record_start`.
const SUPERVISOR_STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5);

/// Who a frame consumer is, which is what decides whether the privacy
/// blackout applies to it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameAudience {
    /// A human watching the drawer. Never gated — the whole point of the
    /// blackout is that the person driving keeps seeing the page while the
    /// model does not.
    Viewer,
    /// Anything that outlives the moment or can reach the model — today
    /// that is `browser_record`'s disk output. Gated: receives nothing
    /// while a blackout is active (R3).
    Model,
}

/// One registered frame consumer.
struct Consumer {
    tx: mpsc::Sender<ScreencastFrame>,
    audience: FrameAudience,
    /// When this consumer subscribed — the epoch its `captured_at` stamps
    /// are measured against.
    epoch: Instant,
}

/// Fan one frame out to every live consumer, skipping [`FrameAudience::Model`]
/// consumers while `blackout` is set and pruning consumers whose receiver has
/// gone away. Returns how many consumers actually received it.
///
/// This is the whole R3 recording gate: a dropped frame is never written to
/// disk, so the produced video contains no frame captured inside the blackout
/// window. `Viewer` consumers are unaffected.
fn publish(consumers: &mut Vec<Consumer>, frame: &ScreencastFrame, blackout: bool) -> usize {
    let mut delivered = 0usize;
    consumers.retain(|c| {
        if blackout && c.audience == FrameAudience::Model {
            // Still a live consumer — just not one that may see this
            // frame. Keeping it registered is what makes the recording
            // resume by itself when the blackout lifts.
            return true;
        }
        let stamped = ScreencastFrame {
            jpeg: frame.jpeg.clone(),
            viewport: frame.viewport,
            captured_at: c.epoch.elapsed().as_secs_f64(),
        };
        // Bounded and dropping, per `car_browser::FRAME_CHANNEL_CAP`: this is
        // the hop in front of the recorder's writer task, whose per-frame
        // `std::fs::write` is the one consumer here that can genuinely fall
        // behind Chrome. Unbounded, it grew a full-viewport JPEG per frame
        // for as long as the disk stayed slow. A consumer that is merely
        // behind stays registered — only a closed channel deregisters it.
        match c.tx.try_send(stamped) {
            Ok(()) => {
                delivered += 1;
                true
            }
            Err(mpsc::error::TrySendError::Full(_)) => true,
            Err(mpsc::error::TrySendError::Closed(_)) => false,
        }
    });
    delivered
}

/// The live-frame fan-out for one browser.
pub struct FrameFanout {
    consumers: Arc<Mutex<Vec<Consumer>>>,
    /// Set by [`BrowserTools`](super::browser_tools::BrowserTools) on every
    /// control-state change. Read per frame by the supervisor — an atomic
    /// rather than a callback so the gate cannot deadlock against the
    /// presentation lock the state change is already holding.
    blackout: Arc<AtomicBool>,
    /// The browser to capture, once one has launched. `None` until then —
    /// subscribing before a browser exists is legal (the drawer opens on an
    /// empty standing session) and simply yields no frames yet.
    backend: Mutex<Option<Arc<ChromiumBackend>>>,
    /// The running supervisor, if any. Ends on its own when the consumer
    /// list empties — and CLEARS THIS SLOT as it goes, under this mutex.
    ///
    /// `Arc` because the supervisor task holds it too: liveness cannot be
    /// read off `JoinHandle::is_finished()`. Every exit path observes "no
    /// consumers" and then AWAITS (`pump.stop()` is a real CDP round trip)
    /// before returning, so a consumer subscribing inside that window pushed
    /// itself onto `consumers`, called `ensure_supervisor`, saw a task that
    /// had not finished YET, and returned — leaving a registered consumer
    /// with no supervisor, no capture, and nothing that polls. The slot is
    /// the liveness signal precisely because the exiting task publishes into
    /// it while holding this lock, which linearizes "decide to exit" against
    /// "check the slot and spawn".
    supervisor: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
    /// Which supervisor the slot holds. Bumped by every spawn.
    ///
    /// The slot alone cannot answer "is this still the one I asked to stop":
    /// a `subscribe` landing after the cancel legitimately spawns a
    /// REPLACEMENT, and a live replacement is indistinguishable from a
    /// still-running original — both are `Some(task)` with `is_finished()`
    /// false. `stop_supervisor` then waited out its grace period on the wrong
    /// task and ABORTED it, which is the abort-instead-of-clean-exit path the
    /// cancel signal exists to avoid, applied to a supervisor that had done
    /// nothing wrong: its pump's `Drop` stops the screencast on a detached
    /// task, racing whatever starts next. Same family as `browser_view`'s
    /// subscriber epoch — an identity that is not unique across the handover.
    supervisor_generation: AtomicU64,
    /// JPEG quality the pump captures at. `browser_record_start` can ask
    /// for a different one; see [`Self::set_quality`].
    quality: AtomicI64,
    /// Bumped when the last consumer goes away, so the supervisor can stop
    /// capture immediately instead of at the next frame.
    ///
    /// A `watch` rather than a `Notify`, because `Notify::notify_waiters()`
    /// stores no permit — it wakes only tasks already parked at the instant
    /// of the call. The supervisor spends several CDP round trips
    /// (`page_handle`, `ScreencastPump::attach`, `pump.subscribe`) between
    /// its emptiness check and the first poll of this signal, and a `prune`
    /// landing anywhere in that window was dropped on the floor: the
    /// supervisor then parked on a still page holding an armed screencast
    /// with zero consumers, which is exactly what "zero consumers ⇒ zero
    /// frames" exists to prevent. A watch retains the change for a receiver
    /// created BEFORE it, so the edge cannot be missed.
    idle: watch::Sender<u64>,
    /// Bumped to ask the running supervisor to tear its pump down and exit
    /// CLEANLY — see [`Self::stop_supervisor`]. Aborting the task instead
    /// left `ScreencastPump::drop` to stop the screencast on a DETACHED
    /// task, which then races the replacement supervisor's
    /// `Page.startScreencast` on the same page.
    cancel: watch::Sender<u64>,
    /// Serializes [`Self::stop_supervisor`] against itself. Held for the whole
    /// teardown, unlike the supervisor slot — which must stay populated so a
    /// concurrent `ensure_supervisor` does not spawn a replacement into the
    /// window where the old pump is still stopping.
    teardown: Mutex<()>,
    width: u32,
    height: u32,
}

impl FrameFanout {
    pub fn new(width: u32, height: u32, default_quality: i64) -> Self {
        Self {
            consumers: Arc::new(Mutex::new(Vec::new())),
            blackout: Arc::new(AtomicBool::new(false)),
            backend: Mutex::new(None),
            supervisor: Arc::new(Mutex::new(None)),
            supervisor_generation: AtomicU64::new(0),
            quality: AtomicI64::new(default_quality.clamp(1, 100)),
            idle: watch::channel(0).0,
            cancel: watch::channel(0).0,
            teardown: Mutex::new(()),
            width,
            height,
        }
    }

    /// Mirror of the control state's blackout predicate. Called on every
    /// control-state change.
    pub fn set_blackout(&self, active: bool) {
        self.blackout.store(active, Ordering::SeqCst);
    }

    pub fn blackout_active(&self) -> bool {
        self.blackout.load(Ordering::SeqCst)
    }

    /// Point the fan-out at a launched browser. Called once the assistant's
    /// Chromium actually exists; starts capture immediately if anyone is
    /// already subscribed (the drawer opened first, the browser launched
    /// second — the ordinary case for the standing session).
    pub async fn bind(&self, backend: Arc<ChromiumBackend>) {
        *self.backend.lock().await = Some(backend);
        self.ensure_supervisor().await;
    }

    /// Register a consumer. The first one starts CDP capture (via the
    /// supervisor); dropping the returned receiver deregisters it at the
    /// next frame. Returns the epoch its frames are timestamped against.
    pub async fn subscribe(&self, audience: FrameAudience) -> (FrameReceiver, Instant) {
        let (tx, rx) = mpsc::channel(FRAME_CHANNEL_CAP);
        let epoch = Instant::now();
        self.consumers.lock().await.push(Consumer {
            tx,
            audience,
            epoch,
        });
        self.ensure_supervisor().await;
        (rx, epoch)
    }

    /// The quality the pump is currently capturing at.
    pub fn quality(&self) -> i64 {
        self.quality.load(Ordering::SeqCst)
    }

    /// Re-capture at `quality`. A no-op when it already matches; otherwise
    /// the current pump generation is stopped so the supervisor re-attaches
    /// at the new quality. Consumers keep their subscriptions across it.
    ///
    /// This exists because ONE pump now serves both the drawer and
    /// `browser_record_start`, whose documented `quality` parameter would
    /// otherwise silently stop having an effect.
    pub async fn set_quality(&self, quality: i64) {
        let quality = quality.clamp(1, 100);
        if self.quality.swap(quality, Ordering::SeqCst) == quality {
            return;
        }
        // Ending the supervisor is what forces a fresh attach: the next
        // `subscribe`/`bind` (the caller's own, immediately after) spawns a
        // new one, which reads the quality atomic on its way in.
        self.stop_supervisor().await;
        self.ensure_supervisor().await;
    }

    /// Drop consumers whose receiver has gone away and, if that was the
    /// last one, wake the supervisor so it stops CDP capture now rather than
    /// at whatever future moment the page next happens to change.
    pub async fn prune(&self) {
        let empty = {
            let mut consumers = self.consumers.lock().await;
            consumers.retain(|c| !c.tx.is_closed());
            consumers.is_empty()
        };
        if empty {
            self.idle.send_modify(|n| *n = n.wrapping_add(1));
        }
    }

    /// The receiver a supervisor would hold. Test-only window into the
    /// idle signal, so the "a prune landing mid-attach is not lost" property
    /// can be asserted without a live Chromium to attach to.
    #[cfg(test)]
    pub(crate) fn idle_watch_for_test(&self) -> watch::Receiver<u64> {
        self.idle.subscribe()
    }

    /// How many consumers are currently registered. Test-only window into
    /// otherwise-private state, so a caller-level test (`browser_tools`'s
    /// own suite, which drives `run_record_stop_with`) can assert a dead
    /// consumer actually got pruned rather than merely trusting that it did.
    #[cfg(test)]
    pub async fn consumer_count_for_test(&self) -> usize {
        self.consumers.lock().await.len()
    }

    /// Spawn the pump supervisor if a browser is bound, somebody is
    /// subscribed, and one isn't already running.
    async fn ensure_supervisor(&self) {
        let mut running = self.supervisor.lock().await;
        if running.as_ref().is_some_and(|t| !t.is_finished()) {
            return;
        }
        let Some(backend) = self.backend.lock().await.clone() else {
            return;
        };
        if self.consumers.lock().await.is_empty() {
            return;
        }
        let consumers = Arc::clone(&self.consumers);
        let blackout = Arc::clone(&self.blackout);
        // Subscribed HERE, before the spawn — so every `prune` from this
        // moment on is retained for the supervisor, including the ones that
        // land while it is still attaching.
        let idle = self.idle.subscribe();
        let cancel = self.cancel.subscribe();
        let quality = self.quality.load(Ordering::SeqCst);
        let (width, height) = (self.width, self.height);
        // Cloned before the spawn, and `running` is held across it — so the
        // task cannot reach `publish_exit` (which wants this same lock)
        // before the handle below has been stored.
        let slot = Arc::clone(&self.supervisor);
        // Bumped under the supervisor lock `running` holds, so a
        // `stop_supervisor` that snapshots it and then observes a different
        // value has provably had a replacement spawned underneath it.
        self.supervisor_generation.fetch_add(1, Ordering::AcqRel);
        *running = Some(tokio::spawn(async move {
            supervise(
                backend, consumers, blackout, idle, cancel, quality, width, height, slot,
            )
            .await;
        }));
    }

    /// Tear the running supervisor down and WAIT for it to be gone.
    ///
    /// `abort()` alone was not enough, and awaiting the aborted handle would
    /// not have been either. Aborting drops the supervisor's `pump` local,
    /// and `ScreencastPump::drop` can only send `Page.stopScreencast` from a
    /// spawned, detached task — which is scheduled independently of the
    /// replacement supervisor's own CDP round trips, so the stop could land
    /// AFTER the new pump's `startScreencast`. Chrome then stops emitting
    /// while the new pump's `running` is `Some` with a live task, so
    /// `reusable_start` reports the capture as healthy and nothing notices:
    /// on `browser_record_start { quality }` — the one-pump-serves-both case
    /// this design exists for — the drawer freezes AND the recording captures
    /// nothing, with no error anywhere.
    ///
    /// So the supervisor is asked to stop instead: it runs `pump.stop().await`
    /// itself, which leaves `running` empty so its `Drop` is inert, and only
    /// then does this return. Bounded, because a supervisor wedged in a CDP
    /// call must not wedge `set_quality` — past the deadline the abort is
    /// still strictly better than nothing.
    async fn stop_supervisor(&self) {
        // Serialized against another teardown, and — critically — the handle
        // stays IN the slot for the whole wait.
        //
        // Taking it out first vacated the slot immediately, so a concurrent
        // `subscribe` → `ensure_supervisor` read "no supervisor" and spawned a
        // second one, which sent `Page.startScreencast` on the same page while
        // the old supervisor was still on its way to `pump.stop()` —
        // `Page.stopScreencast`. Chrome then stops emitting with the new
        // supervisor's slot Some and its task alive, so `reusable_start`
        // reports the capture healthy and nothing notices. That is the exact
        // failure the cancel signal was added to close, reached through the
        // subscribe door instead of the abort door: the cancel arm's "provably
        // done before the caller spawns a replacement" is true for the caller
        // (`set_quality`) and was never true for anyone else.
        //
        // A separate mutex because `publish_exit` wants the supervisor lock,
        // so holding that one across the join would deadlock the very task
        // being waited on.
        let _teardown = self.teardown.lock().await;
        if self
            .supervisor
            .lock()
            .await
            .as_ref()
            .is_none_or(|t| t.is_finished())
        {
            return;
        }
        // The identity of the supervisor this teardown is FOR, read before the
        // cancel goes out. Everything below refuses to act on any other one.
        let generation = self.supervisor_generation.load(Ordering::Acquire);
        self.cancel.send_modify(|n| *n = n.wrapping_add(1));
        let deadline = Instant::now() + SUPERVISOR_STOP_GRACE;
        loop {
            {
                let mut slot = self.supervisor.lock().await;
                if self.supervisor_generation.load(Ordering::Acquire) != generation {
                    // A subscribe spawned a replacement after our cancel
                    // landed. Ours is therefore gone (the spawn happens under
                    // this same lock, and only after the slot read empty or
                    // finished), and the live task is not ours to wait on —
                    // still less to abort.
                    return;
                }
                match slot.as_ref() {
                    // Gone on its own — the slot is not ours to clear.
                    None => return,
                    Some(task) if task.is_finished() => {
                        *slot = None;
                        return;
                    }
                    Some(_) => {}
                }
            }
            if Instant::now() >= deadline {
                tracing::debug!(
                    "browser stream: supervisor did not stop within the grace period; aborting"
                );
                let mut slot = self.supervisor.lock().await;
                // Re-checked under the lock: the replacement could have landed
                // between the loop's check and this deadline branch, and
                // aborting it here is the whole defect.
                if self.supervisor_generation.load(Ordering::Acquire) != generation {
                    return;
                }
                if let Some(task) = slot.take() {
                    drop(slot);
                    task.abort();
                    let _ = task.await;
                }
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }
}

/// Publish this supervisor's exit — or refuse to exit, because a consumer
/// arrived while it was tearing down. `true` means "exit now".
///
/// Every one of `supervise`'s exit paths observes "no consumers" and then
/// AWAITS before returning (`pump.stop()` is a CDP round trip), and
/// `ensure_supervisor` cannot see that coming: a `JoinHandle` that has not
/// finished YET looks alive, so a consumer subscribing inside the teardown
/// window was left registered with no supervisor and no capture, and nothing
/// polls for that condition.
///
/// The fix is to make the exit decision and the liveness signal the same
/// event. A subscriber always pushes onto `consumers` BEFORE it calls
/// `ensure_supervisor`, and both that call and this one take the supervisor
/// mutex — so under the lock, "consumers is empty" means the subscriber has
/// not pushed yet, and clearing the slot here means its `ensure_supervisor`
/// will spawn a fresh supervisor. Neither order loses.
async fn publish_exit(
    slot: &Mutex<Option<tokio::task::JoinHandle<()>>>,
    consumers: &Mutex<Vec<Consumer>>,
) -> bool {
    let mut slot = slot.lock().await;
    if !consumers.lock().await.is_empty() {
        return false;
    }
    *slot = None;
    true
}

/// Park until it is worth trying to attach again — the tab registry changed
/// (a page appeared) — or until the last consumer left, which is the one
/// answer that means "stop". `false` ⇒ the supervisor should exit.
///
/// Without the `idle` arm, a supervisor parked here on a browser with no page
/// would never learn that everybody unsubscribed, and would hold the browser
/// and its consumer list until the whole fan-out was dropped.
/// Why [`wait_for_retry`] stopped waiting.
///
/// Cancellation and idleness are NOT the same answer, and collapsing them into
/// one `bool` was the bug: the caller then asked `publish_exit`, which is a
/// question about consumer EMPTINESS. On an explicit teardown with consumers
/// still subscribed that answers "do not exit", so the supervisor looped and
/// ignored the request — `set_quality` burned its whole grace period waiting
/// for a task that was never going to leave, and then aborted a pump that was
/// by then live again, which is the very abort the cancel signal was added to
/// avoid.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RetryWake {
    /// A page may exist now — try attaching again.
    Retry,
    /// The consumer list emptied.
    Idle,
    /// An explicit teardown request. The caller already owns the slot.
    Cancelled,
}

async fn wait_for_retry(
    tabs: &mut watch::Receiver<car_browser::TabsSnapshot>,
    idle: &mut watch::Receiver<u64>,
    cancel: &mut watch::Receiver<u64>,
) -> RetryWake {
    tokio::select! {
        changed = tabs.changed() => {
            if changed.is_ok() { RetryWake::Retry } else { RetryWake::Idle }
        }
        // Only bumped when the consumer list has actually emptied.
        _ = idle.changed() => RetryWake::Idle,
        // An explicit teardown. No pump is attached on this path, so there is
        // nothing to stop — just leave, without consulting a predicate about
        // consumers.
        _ = cancel.changed() => RetryWake::Cancelled,
    }
}

/// Own one pump generation at a time: attach to whatever tab is active,
/// forward its frames to the fan-out list, and re-attach when the active tab
/// changes. Exits — stopping capture — as soon as no consumers remain.
///
/// CDP-bound, so it is exercised by `cargo check` and by live use rather than
/// by a unit test; the parts that carry policy ([`publish`], and the
/// active-tab comparison) are pulled out so they can be tested without Chrome.
async fn supervise(
    backend: Arc<ChromiumBackend>,
    consumers: Arc<Mutex<Vec<Consumer>>>,
    blackout: Arc<AtomicBool>,
    mut idle: watch::Receiver<u64>,
    mut cancel: watch::Receiver<u64>,
    quality: i64,
    width: u32,
    height: u32,
    slot: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
) {
    let mut tabs = backend.subscribe_tabs();
    loop {
        if publish_exit(&slot, &consumers).await {
            return;
        }
        let Ok(page) = backend.page_handle().await else {
            // No page yet (or the browser is shutting down). Wait for the
            // tab registry to say otherwise rather than spinning.
            match wait_for_retry(&mut tabs, &mut idle, &mut cancel).await {
                RetryWake::Retry => continue,
                RetryWake::Cancelled => return,
                RetryWake::Idle => {
                    if publish_exit(&slot, &consumers).await {
                        return;
                    }
                    continue;
                }
            }
        };
        let attached_to = backend.active_tab_id();
        let Ok(pump) = ScreencastPump::attach(&page, quality, 1, width, height).await else {
            match wait_for_retry(&mut tabs, &mut idle, &mut cancel).await {
                RetryWake::Retry => continue,
                RetryWake::Cancelled => return,
                RetryWake::Idle => {
                    if publish_exit(&slot, &consumers).await {
                        return;
                    }
                    continue;
                }
            }
        };
        let Ok((mut incoming, _started)) = pump.subscribe().await else {
            match wait_for_retry(&mut tabs, &mut idle, &mut cancel).await {
                RetryWake::Retry => continue,
                RetryWake::Cancelled => return,
                RetryWake::Idle => {
                    if publish_exit(&slot, &consumers).await {
                        return;
                    }
                    continue;
                }
            }
        };
        // Attaching costs several CDP round trips, and the last consumer can
        // leave inside them. Re-checked here rather than trusted from the
        // outer loop, so a drawer opened and closed inside that window does
        // not leave an armed screencast on a still page.
        if consumers.lock().await.is_empty() {
            pump.stop().await;
            if publish_exit(&slot, &consumers).await {
                return;
            }
            continue;
        }

        // Pump this generation until the active tab moves, the stream ends
        // (navigation, tab close, Chrome exit), or everyone unsubscribes.
        loop {
            tokio::select! {
                frame = incoming.recv() => {
                    let Some(frame) = frame else { break };
                    let blacked = blackout.load(Ordering::SeqCst);
                    let mut list = consumers.lock().await;
                    publish(&mut list, &frame, blacked);
                    if list.is_empty() {
                        drop(list);
                        pump.stop().await;
                        if publish_exit(&slot, &consumers).await {
                            return;
                        }
                        break;
                    }
                }
                changed = tabs.changed() => {
                    if changed.is_err() {
                        break;
                    }
                    if backend.active_tab_id() != attached_to {
                        break;
                    }
                }
                _ = idle.changed() => {
                    if consumers.lock().await.is_empty() {
                        pump.stop().await;
                        if publish_exit(&slot, &consumers).await {
                            return;
                        }
                        break;
                    }
                }
                // An explicit teardown request (`set_quality`). Stop the pump
                // HERE, awaited, so `Page.stopScreencast` is provably done
                // before the caller spawns a replacement supervisor that
                // sends `startScreencast` on the same page — and so the
                // pump's own `Drop` has nothing left to do on a detached
                // task. The slot is left for the caller, which already took
                // it.
                _ = cancel.changed() => {
                    pump.stop().await;
                    return;
                }
            }
        }
        pump.stop().await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_browser::models::Viewport;

    fn frame(byte: u8) -> ScreencastFrame {
        ScreencastFrame {
            jpeg: vec![byte].into(),
            viewport: Viewport {
                width: 4,
                height: 3,
                device_pixel_ratio: 1.0,
            },
            captured_at: 0.0,
        }
    }

    fn consumer(audience: FrameAudience) -> (Consumer, mpsc::Receiver<ScreencastFrame>) {
        let (tx, rx) = mpsc::channel(FRAME_CHANNEL_CAP);
        (
            Consumer {
                tx,
                audience,
                epoch: Instant::now(),
            },
            rx,
        )
    }

    /// The lost-wakeup this signal was a `Notify` for. The supervisor checks
    /// emptiness in its outer loop, then spends several CDP round trips
    /// attaching before it first polls the idle signal — and
    /// `Notify::notify_waiters()` stores no permit, so a `prune()` landing in
    /// that window woke nobody and was gone. The supervisor then parked on a
    /// still page holding an armed screencast with zero consumers.
    ///
    /// Stated the way the supervisor sees it: the receiver exists (it is
    /// subscribed at spawn time), the prune lands, and only afterwards does
    /// anything await it.
    #[tokio::test]
    async fn a_prune_that_lands_while_the_supervisor_is_attaching_is_not_lost() {
        let fanout = FrameFanout::new(4, 3, 80);
        let mut idle = fanout.idle_watch_for_test();

        // A drawer opens and closes again inside the attach window.
        let (rx, _epoch) = fanout.subscribe(FrameAudience::Viewer).await;
        drop(rx);
        fanout.prune().await;

        // The supervisor reaches its `idle` arm only now.
        tokio::time::timeout(std::time::Duration::from_millis(100), idle.changed())
            .await
            .expect("the idle signal must still be there when the supervisor finally parks on it")
            .expect("the fan-out is still alive");
        assert_eq!(fanout.consumer_count_for_test().await, 0);
    }

    /// The teardown race `JoinHandle::is_finished()` cannot see. Every exit
    /// path awaits (`pump.stop()` is a CDP round trip) after observing "no
    /// consumers", so a consumer that subscribes inside that window found a
    /// not-yet-finished handle, returned without spawning, and was left
    /// registered with no supervisor and no capture.
    ///
    /// Stated as the two orders that matter, against the real helper.
    #[tokio::test]
    async fn a_supervisor_tearing_down_does_not_exit_once_a_consumer_has_arrived() {
        let slot: Mutex<Option<tokio::task::JoinHandle<()>>> =
            Mutex::new(Some(tokio::spawn(async {
                std::future::pending::<()>().await
            })));
        let consumers: Mutex<Vec<Consumer>> = Mutex::new(Vec::new());

        // The consumer pushes itself BEFORE calling `ensure_supervisor`, so
        // under the supervisor lock a non-empty list means it is already
        // registered — the supervisor must keep running for it.
        let (c, _rx) = consumer(FrameAudience::Viewer);
        consumers.lock().await.push(c);
        assert!(
            !publish_exit(&slot, &consumers).await,
            "a registered consumer must keep the supervisor alive"
        );
        assert!(
            slot.lock().await.is_some(),
            "and the slot must still read as live, or nothing would restart it"
        );

        // The other order: nobody has arrived, so the exit is published and
        // the slot reads as vacant — the next `ensure_supervisor` spawns.
        consumers.lock().await.clear();
        assert!(publish_exit(&slot, &consumers).await);
        assert!(
            slot.lock().await.is_none(),
            "the exiting task must publish its own absence, not leave a stale handle"
        );
    }

    #[test]
    fn every_consumer_receives_a_frame_when_no_blackout_is_active() {
        let (viewer, mut viewer_rx) = consumer(FrameAudience::Viewer);
        let (model, mut model_rx) = consumer(FrameAudience::Model);
        let mut list = vec![viewer, model];

        assert_eq!(publish(&mut list, &frame(7), false), 2);
        assert_eq!(viewer_rx.try_recv().unwrap().jpeg.as_ref(), &[7]);
        assert_eq!(model_rx.try_recv().unwrap().jpeg.as_ref(), &[7]);
    }

    /// Controller ruling R3, at the exact line that enforces it: while the
    /// blackout is active the drawer keeps streaming and the recording (the
    /// disk artifact) receives nothing.
    #[test]
    fn blackout_suppresses_the_recording_consumer_and_only_that_one() {
        let (viewer, mut viewer_rx) = consumer(FrameAudience::Viewer);
        let (model, mut model_rx) = consumer(FrameAudience::Model);
        let mut list = vec![viewer, model];

        assert_eq!(publish(&mut list, &frame(1), true), 1);
        assert_eq!(
            viewer_rx.try_recv().unwrap().jpeg.as_ref(),
            &[1],
            "the human driving must keep seeing the page"
        );
        assert!(
            model_rx.try_recv().is_err(),
            "no frame captured during a blackout may reach disk"
        );
    }

    /// The gated consumer is kept registered through the blackout, so the
    /// recording resumes by itself instead of silently ending at the first
    /// blacked-out frame.
    #[test]
    fn a_gated_consumer_resumes_once_the_blackout_lifts() {
        let (viewer, _viewer_rx) = consumer(FrameAudience::Viewer);
        let (model, mut model_rx) = consumer(FrameAudience::Model);
        let mut list = vec![viewer, model];

        publish(&mut list, &frame(1), true);
        publish(&mut list, &frame(2), true);
        assert_eq!(list.len(), 2, "the gated consumer stays registered");
        publish(&mut list, &frame(3), false);

        assert_eq!(model_rx.try_recv().unwrap().jpeg.as_ref(), &[3]);
        assert!(
            model_rx.try_recv().is_err(),
            "only the post-blackout frame lands"
        );
    }

    #[test]
    fn a_dropped_consumer_is_pruned_and_the_survivor_keeps_streaming() {
        let (viewer, viewer_rx) = consumer(FrameAudience::Viewer);
        let (other, mut other_rx) = consumer(FrameAudience::Viewer);
        let mut list = vec![viewer, other];

        drop(viewer_rx);
        assert_eq!(publish(&mut list, &frame(5), false), 1);
        assert_eq!(list.len(), 1);
        assert_eq!(other_rx.try_recv().unwrap().jpeg.as_ref(), &[5]);
    }

    /// Each consumer's `captured_at` is measured from ITS OWN subscription,
    /// not the pump's — a recording that starts long after the drawer opened
    /// must still produce a manifest whose first frame is at ~0s.
    ///
    /// Asserted as the **difference** between the two stamps, never as an
    /// absolute bound on the late one. Both are stamped inside the same
    /// `publish` call, so `early_at - late_at` is exactly the wall gap between
    /// the two subscriptions — the sleep — and any scheduling delay between
    /// `list.push` and `publish` lands in both stamps and cancels. An earlier
    /// version asserted `late_at < 0.030` instead, which is the same 30 ms the
    /// test itself sleeps: a loaded runner that took longer than that to reach
    /// `publish` failed a test whose subject had not moved. It went red on a
    /// mobile-only PR that touches no Rust (car#1115) and blocked the merge.
    #[tokio::test]
    async fn captured_at_is_stamped_per_consumer_epoch() {
        const GAP: std::time::Duration = std::time::Duration::from_millis(30);

        let (early, mut early_rx) = consumer(FrameAudience::Viewer);
        let mut list = vec![early];
        tokio::time::sleep(GAP).await;
        let (late, mut late_rx) = consumer(FrameAudience::Viewer);
        list.push(late);

        publish(&mut list, &frame(1), false);
        let early_at = early_rx.try_recv().unwrap().captured_at;
        let late_at = late_rx.try_recv().unwrap().captured_at;
        assert!(
            early_at > late_at,
            "the earlier subscriber sees a larger elapsed time ({early_at} vs {late_at})"
        );
        // The tolerance is on the sleep's *lower* bound only: `sleep` may
        // overshoot without bound, but it never returns early, so the gap can
        // only be larger than 30 ms. A single shared epoch would put this
        // difference at ~0 and fail — which is the property under test.
        let gap = early_at - late_at;
        assert!(
            gap >= GAP.as_secs_f64() * 0.9,
            "the two stamps must differ by the subscription gap, so the late \
             subscriber is measuring its own epoch and not a shared one \
             (early {early_at}, late {late_at}, difference {gap})"
        );
    }

    #[tokio::test]
    async fn subscribing_without_a_browser_yields_no_frames_and_no_supervisor() {
        let fanout = FrameFanout::new(1920, 1080, 80);
        let (mut rx, _epoch) = fanout.subscribe(FrameAudience::Viewer).await;
        assert!(
            fanout.supervisor.lock().await.is_none(),
            "no browser bound yet, so nothing to capture"
        );
        assert!(rx.try_recv().is_err());
    }

    #[tokio::test]
    async fn the_blackout_flag_round_trips() {
        let fanout = FrameFanout::new(1920, 1080, 80);
        assert!(!fanout.blackout_active());
        fanout.set_blackout(true);
        assert!(fanout.blackout_active());
        fanout.set_blackout(false);
        assert!(!fanout.blackout_active());
    }
}