car-server-core 0.51.0

//! The `browser.view.*` JSON-RPC surface — the browser drawer's window onto
//! a CAR browser.
//!
//! Built on the `runs.subscribe` / `runs.trace.event` contract, point for
//! point:
//!
//! - **Snapshot + register is atomic.** [`BrowserView::subscribe`] reads the
//!   presentation AND inserts the subscriber under the same [`ViewFanout`]
//!   lock every emitter holds, so the snapshot covers exactly the events
//!   through `cursor` — no gap, no duplicate, at the boundary.
//! - **Cursor monotonicity is gap detection.** Every event — a presentation
//!   delta or a screencast frame — is stamped with the next cursor. A
//!   subscriber at `n` expects `n+1`; a jump means the daemon dropped
//!   something and the fix is to re-subscribe, which backfills from a fresh
//!   snapshot.
//! - **Bounded channel, drain task per subscriber.** The producer only
//!   `try_send`s ([`BrowserViewSubscriber::push`]); one dedicated task owns
//!   the WS write. A full channel drops the event — frames are large and a
//!   wedged drawer must never stall the browser, the agent, or the daemon.
//! - **Explicit fanout.** Each `(view, connection)` is its own subscriber;
//!   two Command Decks on one browser both get every event.
//! - **Reconnect-durable.** Nothing fails because a subscriber dropped;
//!   disconnect removes only that connection's subscriptions, and the host
//!   re-subscribes on its new connection.
//!
//! ## Which browser a call talks to
//!
//! `conversation_id` is optional on every method. Omit it (or pass `null`)
//! for the **standing session** — the one shared browser every conversation
//! without an agent-attached browser shows (controller ruling R6). Pass it to
//! reach the browser an agent attached for that conversation/agent-session.
//!
//! ## Who produces a view
//!
//! Two kinds of producer, and this surface cannot tell them apart (see
//! [`ViewBrowser`]): a browser the DAEMON owns in this process, and one a
//! SUPERVISED AGENT PROCESS owns, relayed over the WS session that process
//! already holds (see [`crate::browser_relay`]). Every rule below —
//! authorization, cursors, snapshots, the blackout, the control machine —
//! applies identically to both.
//!
//! ## Authorization
//!
//! Every method requires the **host-management client** (`session.auth
//! { host_token }`). This is deliberately stricter than `runs.subscribe`,
//! which also admits the agent that owns the run: frames ARE page
//! screenshots and the input methods drive a logged-in browser, so admitting
//! an agent here would hand it both perception and actuation outside the
//! `full_access` `browse_*` tool tier — and would let it read a page during
//! the privacy blackout that exists to keep it out. Agents browse through
//! their tools; this surface belongs to the person.

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine as _;
use car_browser::{Modifier, ScreencastFrame};
use futures::SinkExt;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;

use crate::assistant::browser_control::{ControlEffect, ControlOwner, Presentation};
use crate::assistant::browser_tools::{AlwaysConnected, BrowserTools, ControlStatus};
use crate::browser_attention::{BrowserSignInSnapshot, SignInAttention};
use crate::browser_relay::{ProducerRegistry, RelayProducer};
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState, WsChannel};

/// Capacity of each drawer subscriber's bounded channel. Much smaller than
/// the run-trace equivalent on purpose: a slot here can hold a full-viewport
/// JPEG, so a generous buffer is megabytes of memory held for a subscriber
/// that has already fallen behind. A drawer that cannot keep up with 32
/// frames is not going to catch up; dropping and letting it re-subscribe is
/// both cheaper and more honest.
pub const BROWSER_VIEW_CHANNEL_CAP: usize = 32;

/// What every gate answers when the agent is driving. One constant because
/// the check happens TWICE on the relay path — once in the daemon against its
/// cached view of the control state, once in the process that owns the browser
/// immediately before it injects — and a person must not be able to tell the
/// two apart.
pub const AGENT_HOLDS_CONTROL: &str =
    "the agent holds control of this browser — call browser.view.take_control first";

/// How long control stays with a connection that dropped while holding it,
/// before it reverts to the agent. Long enough to cover an app restart or a
/// flaky socket; short enough that an agent is never parked forever behind a
/// controller who is not coming back.
pub const CONTROL_GRACE: Duration = Duration::from_secs(30);

/// What a `ControlEffect::StartGracePeriod` coming back from the reducer means
/// at a given [`BrowserView::apply_effects`] call site.
///
/// The reducer runs in the agent process and only ever ASKS for a clock; the
/// daemon owns it. Whether that ask is a fresh clock depends on the caller.
#[derive(Clone, Copy, Debug)]
enum GraceArming {
    /// Start a clock now, with these watcher semantics (see
    /// [`BrowserView::spawn_grace_timer_inner`]).
    Arm { require_unwatched: bool },
    /// The caller already armed this disconnect's clock before relaying, so
    /// the effect is the reducer AGREEING rather than a second clock.
    ///
    /// Arming again would do two wrong things. It would bump
    /// `control.generation` a second time — and on the relay path this runs
    /// DETACHED, up to 2×`RELAY_CALL_TIMEOUT` later, so that second bump can
    /// land after a legitimate `take_control`, capturing the re-taker's
    /// generation and turning the stale-expiry check into a no-op. And it
    /// would re-arm with the holder variant, discarding the `require_unwatched`
    /// semantics the disconnect chose from `was_watching`.
    AlreadyArmed,
}

// ---------------------------------------------------------------------------
// Wire types
// ---------------------------------------------------------------------------

/// One tab, as the drawer's tab strip renders it. `id` is the tab's opaque
/// id rendered as a string (`tab-3`); pass it straight back to
/// `browser.view.tab_close` / `tab_switch`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WireTab {
    pub id: String,
    pub url: String,
    pub title: String,
    pub active: bool,
    pub can_go_back: bool,
    pub can_go_forward: bool,
}

/// Who is driving. Serialized as `"none"` / `"agent"` / `"user"`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WireOwner {
    /// No agent has attached — the user's own browser, no strip, no
    /// ceremony.
    None,
    Agent,
    User,
}

impl From<ControlOwner> for WireOwner {
    fn from(owner: ControlOwner) -> Self {
        // Exhaustive on purpose: a new owner state must not silently
        // serialize as an existing one.
        match owner {
            ControlOwner::NoAgent => WireOwner::None,
            ControlOwner::Agent => WireOwner::Agent,
            ControlOwner::User => WireOwner::User,
        }
    }
}

/// The way back, for a view whose control state arrives over the wire from a
/// supervised agent process rather than from a reducer in this process.
impl From<WireOwner> for ControlOwner {
    fn from(owner: WireOwner) -> Self {
        match owner {
            WireOwner::None => ControlOwner::NoAgent,
            WireOwner::Agent => ControlOwner::Agent,
            WireOwner::User => ControlOwner::User,
        }
    }
}

/// Everything the drawer renders except the picture itself.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WirePresentation {
    /// Advances only when something actually changed (see
    /// `browser_control::PresentationState`). Distinct from the subscription
    /// `cursor`, which counts EVENTS including frames.
    pub revision: u64,
    pub owner: WireOwner,
    /// What the agent is doing right now, in plain words, or `null`.
    pub current_action: Option<String>,
    /// The sign-in strip's plain-words prompt while one is pending.
    pub pending_signin: Option<String>,
    /// While true, no page observation reaches the model and no frame
    /// reaches `browser_record`'s output — the drawer keeps streaming.
    pub blackout_active: bool,
    pub tabs: Vec<WireTab>,
    /// Convenience projections of the active tab, so a client does not have
    /// to scan `tabs` to render the nav bar.
    pub active_tab: Option<String>,
    pub url: Option<String>,
    pub title: Option<String>,
}

impl WirePresentation {
    pub(crate) fn empty() -> Self {
        Self {
            revision: 0,
            owner: WireOwner::None,
            current_action: None,
            pending_signin: None,
            blackout_active: false,
            tabs: Vec::new(),
            active_tab: None,
            url: None,
            title: None,
        }
    }
}

impl From<&Presentation> for WirePresentation {
    fn from(p: &Presentation) -> Self {
        let tabs: Vec<WireTab> = p
            .tabs
            .iter()
            .map(|t| WireTab {
                id: t.id.to_string(),
                url: t.url.clone(),
                title: t.title.clone(),
                active: t.active,
                can_go_back: t.can_go_back,
                can_go_forward: t.can_go_forward,
            })
            .collect();
        let active = tabs.iter().find(|t| t.active);
        Self {
            revision: p.revision,
            owner: p.owner.into(),
            current_action: p.current_action.clone(),
            pending_signin: p.pending_signin.as_ref().map(|s| s.message.clone()),
            blackout_active: p.blackout_active,
            active_tab: active.map(|t| t.id.clone()),
            url: active.map(|t| t.url.clone()),
            title: active.map(|t| t.title.clone()),
            tabs,
        }
    }
}

/// One screencast frame: base64 JPEG plus the viewport it was captured at,
/// so a client can map its own pointer coordinates back onto the page
/// without asking.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WireFrame {
    pub jpeg_base64: String,
    pub width: u32,
    pub height: u32,
    pub device_pixel_ratio: f64,
    /// Wall-clock seconds since this subscription started. Frames are
    /// change-driven, not fixed-rate.
    pub captured_at: f64,
}

impl From<ScreencastFrame> for WireFrame {
    fn from(frame: ScreencastFrame) -> Self {
        Self {
            jpeg_base64: BASE64.encode(&frame.jpeg),
            width: frame.viewport.width,
            height: frame.viewport.height,
            device_pixel_ratio: frame.viewport.device_pixel_ratio,
            captured_at: frame.captured_at,
        }
    }
}

/// What one `browser.view.event` carries.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum BrowserViewPayload {
    Presentation { presentation: WirePresentation },
    Frame { frame: WireFrame },
}

/// One pushed `browser.view.event` notification.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BrowserViewEvent {
    /// The view this belongs to; `null` for the standing session.
    pub conversation_id: Option<String>,
    /// Monotonic per view. A subscriber at `n` expects `n+1`.
    pub cursor: u64,
    #[serde(flatten)]
    pub payload: BrowserViewPayload,
}

// ---------------------------------------------------------------------------
// Subscriber
// ---------------------------------------------------------------------------

/// One live drawer subscriber — the producer side of a bounded channel whose
/// drain task writes `browser.view.event` frames to that connection's
/// WebSocket. Modeled on [`crate::host::RunTraceSubscriber`]; the producer
/// never touches the socket.
/// Hands out [`BrowserViewSubscriber::epoch`] values, PROCESS-WIDE.
///
/// It was per-view, which is the obvious place for it and was wrong for the
/// one path that matters: [`BrowserView::adopt`] moves subscribers ACROSS
/// views, so a per-view counter makes the identity unique only WITHIN the view
/// that issued it. Both counters start at 0, so a subscriber inherited from the
/// previous view and a `subscribe` that landed on the successor inside the
/// register/adopt window collide exactly — `or_insert` drops the inherited one,
/// its drain task exits, and `unsubscribe_epoch` matches the LIVE registration's
/// epoch and removes it. The drawer then holds a successful reply with a cursor
/// that never advances, reporting `.live` over a frozen frame, with no event
/// arriving to trip the cursor-gap recovery. That is the exact failure the epoch
/// was added to prevent, reintroduced one level up.
static NEXT_SUBSCRIBER_EPOCH: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

pub struct BrowserViewSubscriber {
    tx: tokio::sync::mpsc::Sender<BrowserViewEvent>,
    /// The view this subscriber is registered on — SHARED with its drain task,
    /// so `adopt` can re-point it.
    ///
    /// `adopt` moves subscriber structs to a replacement view but cannot move
    /// a spawned task's captured state. With the handle captured by value the
    /// inherited task kept deregistering against the OLD view, whose last
    /// strong reference drops the moment `register` returns — so `upgrade()`
    /// failed and it deregistered from nothing. Its entry stayed in the NEW
    /// view's fanout with a closed channel forever: `subscribers.is_empty()`
    /// never true, so capture never stopped and `release_if_idle` never
    /// evicted the view or its browser — the exact leak that path exists to
    /// close.
    view: Arc<std::sync::Mutex<std::sync::Weak<BrowserView>>>,
    /// Which registration this is, for the fanout's key.
    ///
    /// A re-subscribe on the SAME connection replaces the entry, which drops
    /// the old subscriber, closes its channel, and ends its drain task — and
    /// that task then deregisters. Without an identity check it deregistered
    /// by `client_id` alone, removing the entry the NEW subscribe had just
    /// installed. The trigger is this surface's own documented recovery: a
    /// cursor gap makes the drawer re-subscribe the same key on the same
    /// connection, so ONE dropped frame killed the drawer permanently — the
    /// subscribe returned success with a fresh cursor, and nothing was ever
    /// delivered again.
    ///
    /// Drawn from [`NEXT_SUBSCRIBER_EPOCH`], which is process-wide and not
    /// per-view — the identity has to be unique across the boundary
    /// [`BrowserView::adopt`] moves subscribers over. See that static.
    epoch: u64,
}

impl BrowserViewSubscriber {
    /// How many CONSECUTIVE write stalls a subscriber survives before its
    /// drain task gives up.
    ///
    /// A single stall is a busy socket, not a death — exiting on the first
    /// one killed a subscription permanently. But continuing forever is not
    /// backpressure either, and the previous comment here claimed a drop that
    /// does not happen: `SinkExt::send` is poll_ready + start_send +
    /// poll_flush, and tokio-tungstenite's `start_send` treats `WouldBlock` as
    /// "queued, not an error". By the time the deadline cancels the future it
    /// is cancelling the FLUSH — the frame is already inside tungstenite's
    /// write buffer, which this workspace never configures, so it is
    /// `usize::MAX`. Looping past a permanent stall therefore fed an unbounded
    /// buffer full-viewport JPEGs, and the 32-slot channel in front of it —
    /// sized small precisely because a slot holds one — provided no bound at
    /// all, because the bound it protects sits downstream of the queue.
    ///
    /// So: tolerate a transient stall, and treat a persistent one as the dead
    /// socket it almost certainly is.
    const MAX_CONSECUTIVE_STALLS: u32 = 3;

    /// Re-point this subscriber's drain task at the view that now holds it.
    fn rebind(&self, view: std::sync::Weak<BrowserView>) {
        match self.view.lock() {
            Ok(mut slot) => *slot = view,
            Err(poisoned) => *poisoned.into_inner() = view,
        }
    }

    pub fn spawn(
        view: std::sync::Weak<BrowserView>,
        client_id: String,
        epoch: u64,
        channel: Arc<WsChannel>,
    ) -> Self {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<BrowserViewEvent>(BROWSER_VIEW_CHANNEL_CAP);
        let view = Arc::new(std::sync::Mutex::new(view));
        let task_view = Arc::clone(&view);
        tokio::spawn(async move {
            let mut stalls = 0u32;
            while let Some(event) = rx.recv().await {
                let Ok(json) = serde_json::to_string(&json!({
                    "jsonrpc": "2.0",
                    "method": "browser.view.event",
                    "params": event,
                })) else {
                    continue;
                };
                let mut guard = channel.write.lock().await;
                let send = tokio::time::timeout(
                    Duration::from_secs(10),
                    guard.send(Message::Text(json.into())),
                )
                .await;
                drop(guard);
                match send {
                    Ok(Ok(())) => stalls = 0,
                    // A write ERROR means the socket is gone.
                    Ok(Err(_)) => break,
                    Err(_) => {
                        stalls += 1;
                        if stalls >= Self::MAX_CONSECUTIVE_STALLS {
                            tracing::debug!(
                                client_id,
                                "browser view: subscriber dropped after {} consecutive write stalls",
                                stalls
                            );
                            break;
                        }
                        tracing::debug!(
                            client_id,
                            "browser view: a stalled socket did not accept an event in time"
                        );
                    }
                }
            }
            // Deregister on the way out, whichever way that was. Nothing
            // re-spawns this task, so a subscriber left in the fanout after it
            // ends is a watcher the daemon counts (keeping capture armed) and
            // never serves. Removing it lets `stop_streamer_if_unwatched` stop
            // the screencast, and the host's next `browser.view.*` call
            // re-establishes a working subscription.
            // Read at EXIT, not captured at spawn: `adopt` may have re-pointed
            // this subscriber at a replacement view in the meantime, and the
            // entry to remove is the one in whichever view holds it now.
            let current = match task_view.lock() {
                Ok(view) => view.upgrade(),
                Err(poisoned) => poisoned.into_inner().upgrade(),
            };
            if let Some(view) = current {
                view.unsubscribe_epoch(&client_id, epoch).await;
            }
        });
        Self { tx, epoch, view }
    }

    /// Non-blocking push. `false` when the channel is full (a wedged or slow
    /// drawer) — the event is DROPPED rather than blocking the producer, and
    /// the client detects the resulting cursor gap and re-subscribes.
    pub fn push(&self, event: BrowserViewEvent) -> bool {
        self.tx.try_send(event).is_ok()
    }
}

// ---------------------------------------------------------------------------
// One view
// ---------------------------------------------------------------------------

/// Snapshot + subscriber registry, under one lock so registration and
/// emission are serialized (invariant #1).
struct ViewFanout {
    cursor: u64,
    /// The most recently emitted presentation — what a fresh subscriber's
    /// snapshot is taken from, so the snapshot and the cursor it is paired
    /// with are read under the same lock that stamps events.
    last: WirePresentation,
    subscribers: HashMap<String, BrowserViewSubscriber>,
}

/// Who holds user control, and a generation that invalidates a grace timer
/// whose window was overtaken by a later take/hand-back.
#[derive(Default)]
struct ControlHolder {
    holder: Option<String>,
    generation: u64,
}

/// One control transition, as a view drives it. Deliberately narrower than
/// [`crate::assistant::browser_control::ControlEvent`]: these five are the
/// only transitions the DAEMON originates, and they are the ones that have to
/// cross a process boundary when the browser is a supervised agent's.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ViewControl {
    TakeControl,
    HandBack,
    RunEnded,
    HolderDisconnected,
    GraceExpired,
}

/// One user-driven input, as a view hands it to whatever browser backs it.
///
/// The whole surface a person has: navigate/click/type/keypress/scroll/paste,
/// the three nav-bar history buttons, and the three tab operations. NO perception
/// of any kind — nothing here returns page content, which is what keeps this
/// surface from being a browsing path around the `full_access` `browse_*`
/// tools.
#[derive(Debug, Clone, PartialEq)]
pub enum ViewInput {
    Navigate {
        url: String,
    },
    Click {
        x: f64,
        y: f64,
    },
    Type {
        text: String,
    },
    Keypress {
        key: String,
        modifiers: Vec<Modifier>,
    },
    Scroll {
        delta_y: i32,
    },
    /// Paste at the caret, replacing the selection. Carries the TEXT because
    /// the clipboard belongs to the host's OS, not to the page: CDP's
    /// injected key events cannot reach a clipboard, so a synthesised ⌘V
    /// delivers a key event and nothing arrives. The host reads its own
    /// pasteboard and sends the string.
    Paste {
        text: String,
    },
    /// The nav bar's Back / Forward / Reload buttons. Real history moves
    /// through CDP, not synthesised ⌘←/⌘→ keystrokes — those are browser
    /// chrome the input domain never reaches, so they inject into the PAGE
    /// and the history never moves.
    Back,
    Forward,
    Reload,
    TabOpen,
    TabClose {
        tab_id: String,
    },
    TabSwitch {
        tab_id: String,
    },
}

impl ViewControl {
    /// Drive this transition against a browser in THIS process.
    ///
    /// One definition, two callers: [`ViewBrowser::Local`] here in the daemon,
    /// and the supervised agent process's own `agent.browser.control` handler
    /// (see [`crate::assistant::browser_producer`]). The two paths cannot
    /// diverge because there is only one of them.
    pub async fn apply(self, tools: &BrowserTools) -> (Presentation, Vec<ControlEffect>) {
        match self {
            ViewControl::TakeControl => tools.take_control().await,
            ViewControl::HandBack => tools.hand_back().await,
            ViewControl::RunEnded => tools.note_run_ended().await,
            ViewControl::HolderDisconnected => tools.control_holder_disconnected().await,
            ViewControl::GraceExpired => tools.grace_period_expired().await,
        }
    }
}

impl ViewInput {
    /// Execute this input against a browser in THIS process. `Ok(Some(id))` is
    /// the newly opened tab's id; every other input answers `Ok(None)`.
    ///
    /// Shared by the in-daemon path and the supervised agent process's
    /// `agent.browser.input` handler for the same reason [`ViewControl::apply`]
    /// is: the drawer must behave identically whichever process the browser
    /// happens to live in, and error strings included, that is only guaranteed
    /// if it is literally the same code.
    pub async fn apply(self, tools: &BrowserTools) -> Result<Option<String>, String> {
        // Re-check the control state IN THE PROCESS THAT OWNS THE BROWSER,
        // immediately before injecting. The `browser.view.*` gate ran against
        // the daemon's view of the world; on the relay path that view is a
        // cached presentation and the decision is separated from the injection
        // by a WS round trip, so the agent can legitimately have resumed
        // driving in between. Without this, a user's click could land while
        // the agent is acting, and the blackout would be one hop late — which
        // is the whole property it exists to hold.
        //
        // The predicate matches `BrowserView::require_control`'s exactly
        // (minus the per-connection holder, which only the daemon knows), so
        // the local path re-affirms its own decision rather than changing it.
        let status = tools.control_status().await;
        if status.owner == ControlOwner::Agent && !status.signin_pending {
            return Err(AGENT_HOLDS_CONTROL.to_string());
        }
        // Somebody is demonstrably AT this browser. Recorded here — the one
        // place both the in-daemon and the relayed input paths pass through —
        // because it is what decides whether a sign-in TIMEOUT may end their
        // window: a person who never pressed Take control but is typing into a
        // credential form is exactly the case the ownership flag cannot see.
        tools.note_user_input().await;
        match self {
            ViewInput::Navigate { url } => {
                tools.user_navigate(&url).await?;
                Ok(None)
            }
            ViewInput::Click { x, y } => {
                tools.user_click(x, y).await?;
                Ok(None)
            }
            ViewInput::Type { text } => {
                tools.user_type(&text).await?;
                Ok(None)
            }
            ViewInput::Keypress { key, modifiers } => {
                tools.user_keypress(&key, &modifiers).await?;
                Ok(None)
            }
            ViewInput::Scroll { delta_y } => {
                tools.user_scroll(delta_y).await?;
                Ok(None)
            }
            ViewInput::Paste { text } => {
                tools.user_paste(&text).await?;
                Ok(None)
            }
            ViewInput::Back => {
                tools.user_go_back().await?;
                Ok(None)
            }
            ViewInput::Forward => {
                tools.user_go_forward().await?;
                Ok(None)
            }
            ViewInput::Reload => {
                tools.user_reload().await?;
                Ok(None)
            }
            ViewInput::TabOpen => Ok(Some(tools.user_tab_open().await?.to_string())),
            ViewInput::TabClose { tab_id } => {
                let id = tools.resolve_tab(&tab_id).await?;
                tools.user_tab_close(id).await?;
                Ok(None)
            }
            ViewInput::TabSwitch { tab_id } => {
                let id = tools.resolve_tab(&tab_id).await?;
                tools.user_tab_switch(id).await?;
                Ok(None)
            }
        }
    }
}

/// What a view drives.
///
/// An enum rather than a trait object on purpose: both producers live in this
/// crate, every operation below is matched exhaustively, and a third producer
/// would be a compile error at each site rather than a silently-defaulted arm.
pub enum ViewBrowser {
    /// A browser this process owns: the standing session, or an in-daemon
    /// assistant runtime's own `BrowserTools`.
    Local(Arc<BrowserTools>),
    /// A supervised agent process's browser, relayed over the WS session that
    /// process already holds with the daemon.
    Relay(Arc<RelayProducer>),
}

impl ViewBrowser {
    async fn presentation(&self) -> WirePresentation {
        match self {
            Self::Local(tools) => WirePresentation::from(&tools.presentation().await),
            Self::Relay(producer) => producer.presentation().await,
        }
    }

    /// The cheap "who is driving" read the input path makes on every call —
    /// no CDP locally, no round trip remotely.
    async fn control_status(&self) -> ControlStatus {
        match self {
            Self::Local(tools) => tools.control_status().await,
            Self::Relay(producer) => producer.control_status().await,
        }
    }

    /// Drive the control reducer — which lives WITH the browser — and return
    /// what it asked the caller to do.
    ///
    /// Fallible because a relayed transition can fail to reach the process at
    /// all. A local one never fails: the reducer is a pure function on state
    /// this process owns.
    /// Returns the owner the transition ACTUALLY landed on, alongside the
    /// effects. Both come from the same answer, so no caller has to re-read
    /// state that a concurrent writer can overwrite in between — see
    /// [`BrowserView::take_control`].
    async fn control(
        &self,
        control: ViewControl,
    ) -> Result<(ControlOwner, Vec<ControlEffect>), String> {
        match self {
            Self::Local(tools) => {
                let (presentation, effects) = control.apply(tools).await;
                Ok((presentation.owner, effects))
            }
            Self::Relay(producer) => producer.control(control).await,
        }
    }

    /// [`Self::control`] for the transitions the DAEMON originates on its own
    /// (run end, holder disconnect, grace expiry). There is nobody to report a
    /// failure to and nothing to undo: the transition is the daemon telling
    /// the browser about something that already happened. Logged, not
    /// propagated.
    async fn control_best_effort(&self, control: ViewControl) -> Vec<ControlEffect> {
        match self.control(control).await {
            Ok((_, effects)) => effects,
            Err(error) => {
                tracing::warn!(
                    ?control,
                    %error,
                    "browser view: a daemon-originated control transition did not land"
                );
                Vec::new()
            }
        }
    }

    /// Execute one user input. `Ok(Some(id))` is the newly opened tab's id;
    /// every other input answers `Ok(None)`.
    async fn input(&self, input: ViewInput) -> Result<Option<String>, String> {
        match self {
            Self::Local(tools) => input.apply(tools).await,
            Self::Relay(producer) => producer.input(input).await,
        }
    }

    /// Start delivering presentation changes and frames into `view`.
    ///
    /// A local browser needs a daemon-side pump, so this returns its task
    /// handle. A relayed one pushes on its own — the daemon only has to say
    /// that somebody is watching — so it returns `None`.
    async fn start_capture(
        &self,
        view: std::sync::Weak<BrowserView>,
    ) -> Option<tokio::task::JoinHandle<()>> {
        match self {
            Self::Local(tools) => {
                let tools = Arc::clone(tools);
                Some(tokio::spawn(async move { stream(view, tools).await }))
            }
            Self::Relay(producer) => {
                producer.start_capture();
                None
            }
        }
    }

    /// Nobody is watching any more: stop paying for capture.
    async fn stop_capture(&self) {
        match self {
            Self::Local(tools) => tools.release_frames().await,
            Self::Relay(producer) => producer.stop_capture(),
        }
    }
}

/// Whether capture is running for this view, and the daemon-side pump task
/// when the producer needs one.
#[derive(Default)]
struct CaptureState {
    active: bool,
    task: Option<tokio::task::JoinHandle<()>>,
}

/// One browser, plus everyone watching it.
pub struct BrowserView {
    /// `None` for the standing session.
    key: Option<String>,
    browser: ViewBrowser,
    fanout: Mutex<ViewFanout>,
    control: Mutex<ControlHolder>,
    /// Turns presentation changes and screencast frames into cursor-stamped
    /// events. Runs only while somebody is subscribed.
    capture: Mutex<CaptureState>,
    /// Whether the run that this view was registered for has ended.
    ///
    /// Half of the eviction condition (`run_ended && no subscribers`) —
    /// see [`BrowserViewRegistry::release_if_idle`]. A plain atomic rather
    /// than a read of the control reducer because a relayed view's reducer
    /// lives in another process, and because the eviction decision must be
    /// answerable without a round trip to a process that may be gone.
    run_ended: std::sync::atomic::AtomicBool,
}

impl BrowserView {
    /// The conversation key this view is registered under; `None` is the
    /// standing session. Read by [`crate::browser_relay::RelayProducer`] when
    /// it has to decide whether a retiring view was the last one it served.
    pub(crate) fn key(&self) -> Option<&str> {
        self.key.as_deref()
    }

    fn new(key: Option<String>, tools: Arc<BrowserTools>) -> Self {
        Self::with_browser(key, ViewBrowser::Local(tools))
    }

    fn with_browser(key: Option<String>, browser: ViewBrowser) -> Self {
        Self {
            key,
            browser,
            fanout: Mutex::new(ViewFanout {
                cursor: 0,
                last: WirePresentation::empty(),
                subscribers: HashMap::new(),
            }),
            control: Mutex::new(ControlHolder::default()),
            capture: Mutex::new(CaptureState::default()),
            run_ended: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Is this view served by that supervised process?
    fn is_served_by(&self, producer: &Arc<RelayProducer>) -> bool {
        match &self.browser {
            ViewBrowser::Local(_) => false,
            ViewBrowser::Relay(mine) => Arc::ptr_eq(mine, producer),
        }
    }

    /// The daemon-owned browser behind this view. Test-only sugar: production
    /// code goes through [`ViewBrowser`], which serves both kinds of producer.
    #[cfg(test)]
    pub(crate) fn tools(&self) -> &Arc<BrowserTools> {
        match &self.browser {
            ViewBrowser::Local(tools) => tools,
            ViewBrowser::Relay(_) => panic!("this view is served by an agent process"),
        }
    }

    /// Atomically snapshot the presentation AND register `client_id` as a
    /// subscriber (invariant #1).
    ///
    /// The refresh happens BEFORE the lock so a first subscriber to a
    /// browser that has been running for a while gets the page the agent is
    /// actually on, not the empty state — and so the CDP tab read never
    /// happens under the lock emitters need.
    async fn subscribe(
        self: &Arc<Self>,
        client_id: &str,
        channel: Arc<WsChannel>,
    ) -> (WirePresentation, u64) {
        self.refresh_presentation().await;
        let epoch = NEXT_SUBSCRIBER_EPOCH.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
        let subscriber = BrowserViewSubscriber::spawn(
            Arc::downgrade(self),
            client_id.to_string(),
            epoch,
            channel,
        );
        let mut fanout = self.fanout.lock().await;
        // A re-subscribe replaces the prior entry; dropping the old
        // subscriber drops its sender, ending its drain task.
        fanout.subscribers.insert(client_id.to_string(), subscriber);
        let snapshot = (fanout.last.clone(), fanout.cursor);
        drop(fanout);
        self.ensure_streamer().await;
        snapshot
    }

    /// Take over from the view this one replaces for the same conversation:
    /// stop its capture, release its browser, and inherit its subscribers and
    /// event cursor.
    ///
    /// Inheriting the cursor is not a nicety. A drawer watching this
    /// conversation is mid-stream at some cursor `n`; restarting at 1 on the
    /// new browser would send it a cursor that moves BACKWARDS, which no
    /// gap-detecting client can interpret. Continuing from `n` keeps the
    /// contract intact — the client sees the next event, carrying the new
    /// browser's presentation, exactly as if the conversation's browser had
    /// changed underneath it, which is what happened.
    async fn adopt(self: &Arc<Self>, previous: &Arc<BrowserView>) {
        // Stop the old browser's capture FIRST, so it cannot emit into the
        // subscribers we are about to move (which would race the cursor).
        previous.stop_streamer().await;
        let (cursor, subscribers) = {
            let mut old = previous.fanout.lock().await;
            (old.cursor, std::mem::take(&mut old.subscribers))
        };
        let has_subscribers = {
            let mut fanout = self.fanout.lock().await;
            fanout.cursor = fanout.cursor.max(cursor);
            // MERGE, never assign. `register` inserts this view into the
            // registry and only then calls `adopt`, so a `subscribe` for the
            // same key can land on this view inside that window — and its
            // caller already got a success reply with a cursor. Assigning the
            // previous view's map over the top would deregister that client
            // silently: it would receive nothing forever, and its cursor would
            // never advance, so it could not even detect the gap and recover.
            //
            // `or_insert` rather than `insert`, for the same client landing on
            // both sides: the entry registered on THIS view is the newer one
            // and is what the client's own reply was stamped against.
            for (client_id, subscriber) in subscribers {
                // Re-point BEFORE inserting: from here on this subscriber's
                // drain task must deregister from THIS view, not the one it
                // was born on (which is about to lose its last strong
                // reference).
                subscriber.rebind(Arc::downgrade(self));
                fanout.subscribers.entry(client_id).or_insert(subscriber);
            }
            !fanout.subscribers.is_empty()
        };
        // Keyed on the MERGED map, not on what came across: a client that
        // subscribed inside the window needs the stream started for it just
        // as much as an inherited one does, and it is the case where the
        // previous view had no subscribers at all that would otherwise leave
        // it with a registration and no streamer.
        if has_subscribers {
            // Tell them what they are now looking at, then start streaming
            // the new browser.
            self.refresh_presentation().await;
            self.ensure_streamer().await;
        }
    }

    /// Drop one connection's subscription. Returns whether there was one.
    ///
    /// Unconditional by `client_id`, which is right for the CALLER-driven
    /// paths — an explicit `browser.view.unsubscribe`, the disconnect sweep —
    /// where the intent is "this connection is done with this view, whatever
    /// registration it currently holds".
    async fn unsubscribe(&self, client_id: &str) -> bool {
        self.remove_subscriber(client_id, None).await
    }

    /// Drop a subscription only if it is still THIS registration.
    ///
    /// The drain task's own exit path. A newer `subscribe` from the same
    /// connection replaces the entry and ends the old task, so removing by
    /// `client_id` alone would delete the live registration that replaced it.
    async fn unsubscribe_epoch(&self, client_id: &str, epoch: u64) -> bool {
        self.remove_subscriber(client_id, Some(epoch)).await
    }

    async fn remove_subscriber(&self, client_id: &str, epoch: Option<u64>) -> bool {
        let (removed, empty) = {
            let mut fanout = self.fanout.lock().await;
            let matches = fanout
                .subscribers
                .get(client_id)
                .is_some_and(|s| epoch.is_none_or(|e| s.epoch == e));
            let removed = matches && fanout.subscribers.remove(client_id).is_some();
            (removed, fanout.subscribers.is_empty())
        };
        if empty {
            self.stop_streamer_if_unwatched().await;
        }
        removed
    }

    /// Is anybody watching this view? The producer's frame fan-out asks before
    /// paying for a clone — see [`RelayProducer::push_frame`].
    pub(crate) async fn has_subscribers(&self) -> bool {
        !self.fanout.lock().await.subscribers.is_empty()
    }

    #[cfg(test)]
    async fn subscriber_count(&self) -> usize {
        self.fanout.lock().await.subscribers.len()
    }

    /// [`Self::subscribe`] / [`Self::subscriber_count`], for tests in sibling
    /// modules driving a view without going through the wire handlers.
    /// [`Self::take_control`] / the control-holder record / the grace
    /// generation, for tests in sibling modules driving a RELAY-backed view —
    /// the only path where a control transition can genuinely fail.
    #[cfg(test)]
    pub(crate) async fn take_control_for_test(
        self: &Arc<Self>,
        client_id: &str,
    ) -> Result<(WirePresentation, u64), String> {
        self.take_control(client_id).await
    }

    #[cfg(test)]
    pub(crate) async fn control_holder_for_test(&self) -> Option<String> {
        self.control.lock().await.holder.clone()
    }

    #[cfg(test)]
    pub(crate) async fn grace_generation_for_test(&self) -> u64 {
        self.control.lock().await.generation
    }

    #[cfg(test)]
    pub(crate) async fn subscribe_for_test(
        self: &Arc<Self>,
        client_id: &str,
        channel: Arc<WsChannel>,
    ) -> (WirePresentation, u64) {
        self.subscribe(client_id, channel).await
    }

    #[cfg(test)]
    pub(crate) async fn subscriber_count_for_test(&self) -> usize {
        self.subscriber_count().await
    }

    /// Re-read the presentation and emit a delta if it actually changed.
    ///
    /// `pub(crate)` because a relayed producer drives it from the other side:
    /// the agent process pushes a presentation, the producer caches it, and
    /// this is what turns that into an event for every subscriber.
    pub(crate) async fn refresh_presentation(&self) {
        let wire = self.browser.presentation().await;
        self.publish_presentation(wire).await;
    }

    /// Install a presentation read and emit a delta if it actually changed —
    /// unless it is OLDER than what is already published.
    ///
    /// The read happens OUTSIDE the fanout lock and is not cheap: for a local
    /// view it is a live CDP `list_tabs` round trip. Two concurrent refreshes
    /// therefore have no ordering guarantee, and the colliding callers are
    /// ordinary — the streamer loop, `subscribe`, `snapshot` via
    /// take_control/hand_back, `note_disconnect`, `note_run_ended`, the grace
    /// timer, and a relayed producer's presentation push.
    ///
    /// Without the guard the LOSER of that race wins the lock second and
    /// publishes its older read at a HIGHER cursor: owner, URL, tab strip and
    /// blackout_active all regress on the drawer, `fanout.last` (the snapshot
    /// handed to every later subscriber) goes stale with them, and because
    /// the cursors stay contiguous the client's gap detection cannot fire. On
    /// an idle browser nothing re-emits, so it stays wrong.
    ///
    /// `revision` only advances when something actually changed, so equal
    /// revisions are content-identical and fall through to the equality check.
    /// Safe across `adopt`: a replacement view inherits the cursor and the
    /// subscriber map, never `last`, so it starts at revision 0 against its
    /// own browser's revisions.
    async fn publish_presentation(&self, wire: WirePresentation) {
        {
            let mut fanout = self.fanout.lock().await;
            if wire.revision < fanout.last.revision || fanout.last == wire {
                return;
            }
            fanout.last = wire.clone();
            fanout.cursor += 1;
            let event = BrowserViewEvent {
                conversation_id: self.key.clone(),
                cursor: fanout.cursor,
                payload: BrowserViewPayload::Presentation { presentation: wire },
            };
            fanout_locked(&fanout.subscribers, &event);
        }
    }

    /// End any attention route before this view becomes unreachable.
    async fn resolve_signin_attention_on_teardown(&self) {
        match &self.browser {
            ViewBrowser::Local(tools) => tools.resolve_signin_attention_on_teardown().await,
            ViewBrowser::Relay(producer) => {
                producer.detach_signin_attention(self.key.as_deref()).await
            }
        }
    }

    /// [`Self::publish_presentation`] with the read supplied, so the
    /// ordering guard can be exercised without racing a real CDP round trip.
    #[cfg(test)]
    async fn publish_presentation_for_test(&self, wire: WirePresentation) {
        self.publish_presentation(wire).await;
    }

    async fn emit_frame(&self, frame: ScreencastFrame) {
        self.emit_wire_frame(WireFrame::from(frame)).await;
    }

    /// Stamp and fan out one frame. `pub(crate)` for the same reason
    /// [`Self::refresh_presentation`] is: a relayed producer's frames arrive
    /// already in wire form, straight off the agent's WS session.
    pub(crate) async fn emit_wire_frame(&self, wire: WireFrame) {
        let mut fanout = self.fanout.lock().await;
        fanout.cursor += 1;
        let event = BrowserViewEvent {
            conversation_id: self.key.clone(),
            cursor: fanout.cursor,
            payload: BrowserViewPayload::Frame { frame: wire },
        };
        fanout_locked(&fanout.subscribers, &event);
    }

    /// The current snapshot as a subscriber would see it, without emitting.
    async fn snapshot(&self) -> (WirePresentation, u64) {
        self.refresh_presentation().await;
        let fanout = self.fanout.lock().await;
        (fanout.last.clone(), fanout.cursor)
    }

    /// [`Self::snapshot`], for tests in sibling modules that need to assert
    /// what a subscriber would see without going through the wire handlers.
    #[cfg(test)]
    pub(crate) async fn snapshot_for_test(&self) -> (WirePresentation, u64) {
        self.snapshot().await
    }

    // ---- control ------------------------------------------------------

    /// Is `client_id` allowed to drive right now?
    ///
    /// - **No agent involved** — anyone authorized may drive. That is the
    ///   plan's "zero ceremony": the standing session is just a browser.
    /// - **A sign-in is pending** — the human must be able to type into the
    ///   credential fields; that is the entire point of the request.
    /// - **The user holds control** — only the connection that took it, or
    ///   any authorized connection when no connection holds it (the holder
    ///   disconnected; see [`Self::note_disconnect`]). "A person has control
    ///   but we do not know which connection" must not mean *nobody* may
    ///   drive — that would leave the browser inert for the whole grace
    ///   period, with the drawer's own Hand back refused too.
    /// - **The agent holds control** — nobody, until Take control.
    async fn require_control(&self, client_id: &str) -> Result<(), String> {
        let status = self.browser.control_status().await;
        match status.owner {
            ControlOwner::NoAgent => Ok(()),
            // A pending sign-in opens the AGENT's browser to the person
            // without a Take control press — the orange strip is the
            // affordance, and typing credentials is the whole point. It does
            // NOT open a browser somebody else already took: `TakeControl`
            // deliberately leaves `pending_signin` set, so
            // `owner == User && signin_pending` is exactly the
            // credential-entry window, and a blanket bypass admitted every
            // other authorized connection into it — a second Command Deck's
            // keystrokes interleaving into the password field, or a navigate
            // taking the page away mid-sign-in. Scoped to the arm that needs
            // it; the `User` arm keeps answering on the holder, as it does
            // for every other input.
            ControlOwner::Agent if status.signin_pending => Ok(()),
            ControlOwner::Agent => Err(AGENT_HOLDS_CONTROL.to_string()),
            ControlOwner::User => {
                if self.holder_admits(client_id).await {
                    Ok(())
                } else {
                    Err(
                        "another connection holds control of this browser — input is accepted \
                         only from the control holder"
                            .to_string(),
                    )
                }
            }
        }
    }

    /// The browser is DRIVEN FIRST, and only a transition that actually landed
    /// moves the daemon's record of who holds control.
    ///
    /// The order matters on the relay path and is the whole fix for a wedged
    /// process: if the call never reaches it, the daemon must not end up
    /// believing the host is driving a browser whose own reducer still says
    /// the agent is — that combination refuses every subsequent input, and it
    /// tells the person the blackout is up when the process never entered it.
    /// Failing honestly leaves both sides agreeing, and the drawer can retry.
    /// Gated on the holder, like `hand_back` and every input method.
    ///
    /// Without this the hand-back gate did not fire in the scenario it was
    /// written for: a second host connection could simply `take_control`
    /// first — overwriting the holder — and then `hand_back`, reverting the
    /// browser to the agent and lifting the privacy blackout under whoever
    /// was mid-sign-in. Two calls instead of one is not a defence.
    async fn take_control(
        self: &Arc<Self>,
        client_id: &str,
    ) -> Result<(WirePresentation, u64), String> {
        if !self.holder_admits(client_id).await {
            return Err(
                "another connection holds control of this browser — it must hand back before \
                 another can take control"
                    .to_string(),
            );
        }
        // The owner comes back WITH the effects, from the same answer.
        //
        // The reducer only moves ownership when the AGENT held it: on the
        // standing session (`NoAgent`) `TakeControl` is a documented no-op,
        // and recording a holder there made `holder` mean "the last
        // connection that pressed the button" rather than "who is driving",
        // which the gate above would then enforce against everyone else for a
        // browser nobody had actually taken.
        //
        // Deciding that from a SECOND, independent `control_status()` read
        // was worse than the bug it fixed. On the relay path `control_status`
        // projects the producer's CACHED presentation, which the agent's own
        // presentation pump also writes, unordered against the transition —
        // so a push carrying a pre-take snapshot landing in the window
        // between the two calls made `now_user` false, left the holder
        // unrecorded, and still returned Ok. The daemon then believed nobody
        // held a browser a person had just taken, `holder_admits` admitted
        // every connection, and the gate added directly above was reachable
        // through the recording side instead of the check side.
        let (owner, effects) = self.browser.control(ViewControl::TakeControl).await?;
        let now_user = owner == ControlOwner::User;
        {
            let mut control = self.control.lock().await;
            if now_user {
                control.holder = Some(client_id.to_string());
            }
            control.generation += 1;
        }
        self.apply_effects(
            effects,
            GraceArming::Arm {
                require_unwatched: false,
            },
        )
        .await;
        Ok(self.snapshot().await)
    }

    /// Whether `client_id` may act on the current user-control holder: it IS
    /// the holder, or nothing holds it.
    async fn holder_admits(&self, client_id: &str) -> bool {
        match self.control.lock().await.holder.as_deref() {
            None => true,
            Some(holder) => holder == client_id,
        }
    }

    /// Gated the same way every input method is. `take_control` records the
    /// holder and `require_control` enforces it for input; hand-back enforced
    /// nothing, so a second host connection could revert the browser to the
    /// agent — resolving the holder's pending sign-in and lifting the privacy
    /// blackout — while that person was still typing a password into the
    /// page. Both connections carry the host token, so this is an ownership
    /// defect rather than a boundary crossing, but the asymmetry with the
    /// input path was not intended.
    async fn hand_back(
        self: &Arc<Self>,
        client_id: &str,
    ) -> Result<(WirePresentation, u64), String> {
        if !self.holder_admits(client_id).await {
            return Err(
                "another connection holds control of this browser — only the control holder \
                 may hand it back"
                    .to_string(),
            );
        }
        let (_, effects) = self.browser.control(ViewControl::HandBack).await?;
        {
            let mut control = self.control.lock().await;
            control.holder = None;
            control.generation += 1;
        }
        self.apply_effects(
            effects,
            GraceArming::Arm {
                require_unwatched: false,
            },
        )
        .await;
        Ok(self.snapshot().await)
    }

    /// The connection holding control dropped. Starts the stated grace
    /// period, after which control reverts to the agent — an agent is never
    /// parked forever behind a controller who is not coming back.
    ///
    /// Returns whether this connection actually held control, i.e. whether
    /// this call took ownership of the disconnect clock. The caller uses that
    /// to keep [`Self::note_watcher_disconnect`] off a view whose holder just
    /// left: both arm a grace timer, with DIFFERENT `require_unwatched`
    /// semantics (see [`Self::spawn_grace_timer_inner`]), and each bumps
    /// `control.generation`, retiring the other's timer. Exactly one of the
    /// two owns any given disconnect.
    ///
    /// `was_watching` is why collapsing the two calls does not silently pick
    /// the holder semantics for everybody. The ordinary drawer is the holder
    /// AND the only subscriber, and for it the WATCHER semantics are the
    /// load-bearing ones: a drawer that comes back inside the window is
    /// precisely what says the person's sign-in window is still theirs. So the
    /// single surviving timer takes `require_unwatched` from whether this
    /// connection was the last drawer watching — see the arming site below.
    pub async fn note_disconnect(self: &Arc<Self>, client_id: &str, was_watching: bool) -> bool {
        let held = {
            let mut control = self.control.lock().await;
            let held = control.holder.as_deref() == Some(client_id);
            if held {
                // Cleared HERE, not left to the grace timer. That connection
                // is provably gone, and the timer is not guaranteed to run:
                // `control_best_effort` swallows a failed transition and
                // returns NO effects — an agent process that stopped
                // answering and hit `RELAY_CALL_TIMEOUT` is the ordinary
                // case — so `StartGracePeriod` never arrives, no timer is
                // spawned, and `holder` names a dead `client_id` forever.
                // The timer still owns the other half, reverting ownership.
                control.holder = None;
                control.generation += 1;
            }
            held
        };
        if !held {
            return false;
        }
        // Arm from what the daemon already KNOWS — the holder disconnected —
        // before asking the browser's reducer. Besides covering a failed relay,
        // this makes the timer's generation precede any legitimate re-take.
        //
        // `require_unwatched` — refuse to expire while a drawer is watching —
        // belongs to this disconnect only when this connection was the LAST
        // drawer watching. Then a drawer arriving inside the window is the
        // same person coming back, and their window is still theirs; expiring
        // under them resolves a pending sign-in as `signed_in: false` and
        // hands the page to the agent mid-credential-entry.
        //
        // It does NOT belong when other connections are still subscribed: a
        // DIFFERENT connection watching says nothing about a holder who left,
        // and suppressing the expiry there would park the agent behind them
        // forever. That is the line `note_watcher_disconnect` drew with its
        // own `subscribers.is_empty()` gate before it armed the `true`
        // variant last, and collapsing the two calls must not redraw it.
        //
        // One condition of that old gate is deliberately NOT reproduced:
        // `note_watcher_disconnect` also required the relay to answer with
        // non-empty effects before arming. Arming ahead of the reducer is the
        // point of this call site — a silent or dead relay must still leave a
        // recovery timer behind — so on that path we now arm where `main`
        // would have armed nothing at all. That is a widening, and it is the
        // direction we want: the failure it removes is a holder who never
        // gets their grace period because the agent process went quiet.
        let require_unwatched = was_watching && self.fanout.lock().await.subscribers.is_empty();
        self.spawn_grace_timer_inner(require_unwatched).await;
        if matches!(&self.browser, ViewBrowser::Relay(_)) {
            // A silent supervised process costs up to TWICE
            // `RELAY_CALL_TIMEOUT` here — ~60s, not 30: `call_agent` bounds the
            // write (`browser_relay.rs` :344) and the wait for the reply
            // (:371) with two separate timeouts. Disconnect teardown has
            // already cleared the dead holder and armed recovery, so it must
            // not wait for that process. The view remains alive for this
            // bounded best-effort reconciliation.
            let view = Arc::clone(self);
            tokio::spawn(async move { view.finish_holder_disconnect().await });
        } else {
            // The local reducer has no transport round trip and its effects are
            // deterministic, so preserve synchronous completion for that path.
            Arc::clone(self).finish_holder_disconnect().await;
        }
        true
    }

    async fn finish_holder_disconnect(self: Arc<Self>) {
        let effects = self
            .browser
            .control_best_effort(ViewControl::HolderDisconnected)
            .await;
        // `GraceArming::AlreadyArmed`: `note_disconnect` armed the clock before
        // relaying, so a `StartGracePeriod` coming back here is the reducer
        // AGREEING, not a second clock to start.
        self.apply_effects(effects, GraceArming::AlreadyArmed).await;
        self.refresh_presentation().await;
    }

    /// The run that owned this browser ended: no ceremony, the user's
    /// browser again.
    ///
    /// Deliberately a method on the VIEW, not a registry lookup by key. The
    /// run-end signal is asynchronous (see `mcp_assistant::BrowserViewGuard`),
    /// so re-resolving the key at fire time can land on whatever view holds it
    /// by then — clearing the strip and re-opening input to everyone on a
    /// SUCCESSOR run's browser while its agent is actively driving. Holding
    /// the `Arc` makes the signal act on the identity it was created for, and
    /// a late arrival for a replaced view is then simply inert.
    pub async fn note_run_ended(self: &Arc<Self>) {
        self.run_ended
            .store(true, std::sync::atomic::Ordering::Release);
        let effects = self
            .browser
            .control_best_effort(ViewControl::RunEnded)
            .await;
        self.apply_effects(
            effects,
            GraceArming::Arm {
                require_unwatched: false,
            },
        )
        .await;
        {
            // Only when the run end actually released ownership. A run
            // ending UNDER a person who is driving leaves them driving (the
            // reducer defers it to hand-back), and clearing the holder there
            // would lock that person out of the very page they hold control
            // of — `require_control`'s `User` arm answers on the holder.
            let still_user_driven = self.browser.control_status().await.owner == ControlOwner::User;
            let mut control = self.control.lock().await;
            if !still_user_driven {
                control.holder = None;
                // Bumped in the SAME branch that cleared the holder, not
                // unconditionally. The generation is the only thing the grace
                // timer checks before acting, so bumping it here invalidated
                // an in-flight timer that nothing re-arms — and in the
                // deferred case (the run ends while a person still holds
                // control) that timer is the one thing that would ever return
                // the view from `owner: user` + blackout to `none`. A
                // disconnect followed by a run end stranded it there.
                control.generation += 1;
            }
        }
        self.refresh_presentation().await;
    }

    /// Act on what the control reducer asked for. Exhaustive by design — a
    /// new effect must be a compile error here, not a silently ignored one.
    ///
    /// `grace` says what a [`ControlEffect::StartGracePeriod`] means at THIS
    /// call site; see [`GraceArming`].
    async fn apply_effects(self: &Arc<Self>, effects: Vec<ControlEffect>, grace: GraceArming) {
        for effect in effects {
            match effect {
                ControlEffect::StartGracePeriod => match grace {
                    GraceArming::Arm { require_unwatched } => {
                        self.spawn_grace_timer_inner(require_unwatched).await
                    }
                    GraceArming::AlreadyArmed => {}
                },
                ControlEffect::SignInResolved { signed_in } => {
                    // Nothing to do here: `browser_await_signin` polls its
                    // own pending state and returns this same answer to the
                    // agent (see `BrowserTools::run_await_signin`). Logged
                    // so the honest-resolution path is traceable.
                    tracing::debug!(
                        view = ?self.key,
                        signed_in,
                        "browser view: pending sign-in resolved as a side effect"
                    );
                }
            }
        }
    }

    /// A connection that was WATCHING this view went away.
    ///
    /// Distinct from [`Self::note_disconnect`], which is about the control
    /// HOLDER — and the distinction is the whole point. The ordinary sign-in
    /// flow never involves Take control (the strip IS the affordance, and
    /// `require_control` admits input while a sign-in is pending so the person
    /// can type without pressing anything), so their window has no holder for
    /// `note_disconnect` to match. Since round 8 that window OUTLIVES the run:
    /// an agent-side ending may not clear a strip somebody is standing at. So
    /// this is the only signal left that says they are not coming back, and
    /// without it a person who closed the laptop mid-sign-in would leave the
    /// blackout latched for the daemon's life — wedging every later run's
    /// browse call behind it.
    ///
    /// Armed only when nobody is watching any more: another connection still
    /// subscribed means the drawer is still there. Re-checked at expiry too.
    ///
    /// Never called for a connection that held control: the registry keys this
    /// off `note_disconnect`'s return value, because the ordinary drawer is
    /// BOTH holder and watcher and the two arm grace timers with different
    /// semantics. The `holder.is_some()` guard below covers the other shape —
    /// someone ELSE is driving — and cannot see this connection's own holder
    /// record, which `note_disconnect` clears before returning.
    pub async fn note_watcher_disconnect(self: &Arc<Self>) {
        {
            let control = self.control.lock().await;
            if control.holder.is_some() {
                // Somebody else is driving: their `note_disconnect` owns the
                // clock. Arming here too would race it.
                return;
            }
        }
        if !self.fanout.lock().await.subscribers.is_empty() {
            return;
        }
        if matches!(&self.browser, ViewBrowser::Relay(_)) {
            // Same bound and same reason as `note_disconnect`: a silent
            // supervised process costs up to 2×`RELAY_CALL_TIMEOUT` (~60s),
            // and disconnect teardown must not inherit it. Nothing else arms a
            // timer on this view for this disconnect, so detaching the
            // reconciliation costs no determinism.
            let view = Arc::clone(self);
            tokio::spawn(async move { view.finish_watcher_disconnect().await });
        } else {
            // The local reducer has no transport round trip, so keep the
            // synchronous completion the local tests are written against.
            Arc::clone(self).finish_watcher_disconnect().await;
        }
    }

    async fn finish_watcher_disconnect(self: Arc<Self>) {
        let effects = self
            .browser
            .control_best_effort(ViewControl::HolderDisconnected)
            .await;
        // The reducer arms only for a person-facing window (`owner == User`,
        // or a pending sign-in), so a view with nothing open is untouched.
        if effects.is_empty() {
            return;
        }
        self.spawn_grace_timer_inner(true).await;
    }

    /// `require_unwatched` is the watcher-disconnect variant: it refuses to
    /// fire if a drawer came back inside the window, because the person
    /// returning is precisely what says their sign-in window is still theirs.
    /// The holder variant does NOT take that check — a different connection
    /// watching says nothing about a holder who left, and skipping the expiry
    /// there would park the agent behind them forever.
    async fn spawn_grace_timer_inner(self: &Arc<Self>, require_unwatched: bool) {
        let generation = {
            let mut control = self.control.lock().await;
            control.generation += 1;
            control.generation
        };
        let view = Arc::downgrade(self);
        tokio::spawn(async move {
            tokio::time::sleep(CONTROL_GRACE).await;
            let Some(view) = view.upgrade() else { return };
            {
                let mut control = view.control.lock().await;
                // Somebody took or handed back control inside the window —
                // this expiry is stale and reverting now would yank control
                // from whoever holds it legitimately.
                //
                // The holder check is defense in depth alongside generation:
                // `note_disconnect` clears the old holder before arming, so a
                // holder being present here necessarily took control later.
                //
                // Generation is only load-bearing because a disconnect arms
                // EXACTLY ONCE — see `GraceArming::AlreadyArmed`. A second
                // arming from the detached relay reply would capture the
                // re-taker's generation and leave the holder check as the only
                // thing standing between a legitimate re-take and revocation.
                if control.generation != generation || control.holder.is_some() {
                    return;
                }
                control.holder = None;
            }
            // Control lock released above, deliberately: this is the only
            // place that would nest control inside fanout, and the check
            // needs no atomicity with the generation read — a drawer that
            // arrives after it simply gets the window ended a moment early,
            // which is what the un-engaged contract does anyway.
            if require_unwatched && !view.fanout.lock().await.subscribers.is_empty() {
                return;
            }
            view.browser
                .control_best_effort(ViewControl::GraceExpired)
                .await;
            view.refresh_presentation().await;
        });
    }

    // ---- the streamer -------------------------------------------------

    async fn ensure_streamer(self: &Arc<Self>) {
        let mut capture = self.capture.lock().await;
        // A local producer's pump can die (its browser went away); a relayed
        // one has no daemon-side task at all, so "no task" is not "not
        // running" — `active` is what says whether capture is on.
        let pump_died = match capture.task.as_ref() {
            Some(task) => task.is_finished(),
            None => false,
        };
        if capture.active && !pump_died {
            return;
        }
        if let Some(task) = capture.task.take() {
            task.abort();
        }
        capture.task = self.browser.start_capture(Arc::downgrade(self)).await;
        capture.active = true;
    }

    /// Stop capture unconditionally — the replaced-view teardown
    /// (`adopt`) and the eviction path, both of which are retiring this view
    /// whatever it still holds.
    async fn stop_streamer(&self) {
        self.stop_streamer_inner(false).await;
    }

    /// Stop capture ONLY if nobody is subscribed, re-checked under the
    /// fanout lock rather than on the caller's stale reading.
    ///
    /// `unsubscribe` computes "empty" under the fanout lock and then
    /// RELEASES it before deciding to stop, so a `subscribe` can land in that
    /// window — and `ensure_streamer` returns early for it, seeing a live
    /// pump. Stopping anyway left that subscriber holding a successful reply
    /// with a snapshot and a cursor that never advanced again: zero events
    /// means the cursor-gap contract it would recover through can never fire,
    /// so the drawer freezes on one frame indefinitely.
    async fn stop_streamer_if_unwatched(&self) {
        self.stop_streamer_inner(true).await;
    }

    /// `capture` is taken before `fanout` here. That is the only order any
    /// path takes — `subscribe`, `adopt` and `unsubscribe` all drop `fanout`
    /// before touching `capture` — so the re-check adds no cycle.
    async fn stop_streamer_inner(&self, only_if_unwatched: bool) {
        let mut capture = self.capture.lock().await;
        if !capture.active {
            return;
        }
        if only_if_unwatched && !self.fanout.lock().await.subscribers.is_empty() {
            return;
        }
        capture.active = false;
        if let Some(task) = capture.task.take() {
            task.abort();
            // Await the cancellation so the streamer's frame receiver is
            // actually dropped before we ask the fan-out to prune — without
            // this, CDP capture would keep running until the next frame
            // happened to arrive.
            let _ = task.await;
        }
        drop(capture);
        self.browser.stop_capture().await;
    }
}

/// Push one event to every subscriber of this view. Best-effort: a wedged
/// subscriber's full channel drops the event rather than stalling the
/// producer; the client detects the cursor gap and re-subscribes.
fn fanout_locked(subscribers: &HashMap<String, BrowserViewSubscriber>, event: &BrowserViewEvent) {
    for (client_id, subscriber) in subscribers.iter() {
        if !subscriber.push(event.clone()) {
            tracing::debug!(
                client_id,
                cursor = event.cursor,
                "browser view: dropped event for a slow subscriber (channel full)"
            );
        }
    }
}

/// Turn presentation changes and screencast frames into cursor-stamped
/// events for as long as anyone is subscribed. The DAEMON-owned browser's
/// pump; a relayed producer pushes instead (see [`crate::browser_relay`]).
///
/// Holds only a `Weak` to the view, so a view nobody references any more
/// ends this task instead of keeping the browser alive.
async fn stream(view: std::sync::Weak<BrowserView>, tools: Arc<BrowserTools>) {
    let mut changes = tools.subscribe_changes();
    let (mut frames, _epoch) = tools.subscribe_frames().await;
    let mut tabs = tools.subscribe_tabs().await;

    loop {
        let Some(view) = view.upgrade() else { return };
        tokio::select! {
            changed = changes.changed() => {
                if changed.is_err() {
                    return;
                }
                // A browser may have just launched — bind the tab watch if
                // we could not before.
                if tabs.is_none() {
                    tabs = tools.subscribe_tabs().await;
                }
                view.refresh_presentation().await;
            }
            // Only polled when a browser exists; `tabs` is rebound above
            // when one appears.
            tabs_changed = async {
                match tabs.as_mut() {
                    Some(rx) => rx.changed().await.is_ok(),
                    None => std::future::pending().await,
                }
            } => {
                if !tabs_changed {
                    tabs = None;
                    continue;
                }
                view.refresh_presentation().await;
            }
            frame = frames.recv() => {
                match frame {
                    Some(frame) => view.emit_frame(frame).await,
                    // The fan-out dropped our consumer (a pump generation
                    // ended with nobody listening). Re-register rather than
                    // spinning on a closed channel.
                    None => {
                        let (rx, _epoch) = tools.subscribe_frames().await;
                        frames = rx;
                    }
                }
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Registry
// ---------------------------------------------------------------------------

/// Every browser the drawer can reach, keyed by conversation/agent-session —
/// plus the one standing session shared by every conversation that has no
/// agent-attached browser of its own (controller ruling R6).
pub struct BrowserViewRegistry {
    views: Mutex<HashMap<Option<String>, Arc<BrowserView>>>,
    /// Supervised agent processes currently publishing a browser, keyed by
    /// their WS connection. Lives here because producers and views are torn
    /// down on the same disconnect boundary.
    producers: ProducerRegistry,
    /// Working root for the standing session's `BrowserTools`. Only its
    /// recording output would land here, and the standing session has no
    /// agent to start one — it exists because `BrowserTools` requires one.
    root: PathBuf,
    /// Where a RELAYED browser's pending-sign-in transition becomes an
    /// operator-facing `host.event` — see [`crate::browser_attention`].
    /// Installed once by `ServerState`, which owns the `HostState` this
    /// broadcasts on; absent in every test that stands up a bare registry,
    /// which then behaves exactly as before.
    signin_attention: std::sync::OnceLock<Arc<dyn SignInAttention>>,
}

impl BrowserViewRegistry {
    pub fn new(root: PathBuf) -> Self {
        Self {
            views: Mutex::new(HashMap::new()),
            producers: ProducerRegistry::default(),
            root,
            signin_attention: std::sync::OnceLock::new(),
        }
    }

    /// Install the operator-attention sink relayed views report sign-in waits
    /// through. Called once, by `ServerState::with_config`; a second call is a
    /// silent no-op.
    pub fn set_signin_attention(&self, attention: Arc<dyn SignInAttention>) {
        let _ = self.signin_attention.set(attention);
    }

    /// The installed sink, for a view that is about to be created.
    fn signin_attention(&self) -> Option<Arc<dyn SignInAttention>> {
        self.signin_attention.get().cloned()
    }

    /// The standing user session, created on first reference.
    ///
    /// Creating it does NOT launch Chromium — `BrowserTools` launches
    /// lazily, and for this view the trigger is the user's first navigation
    /// (see `BrowserTools::user_navigate`). Opening the drawer on an empty
    /// standing session therefore costs nothing and shows the empty state,
    /// which is exactly what the design asks for.
    pub async fn standing(&self) -> Arc<BrowserView> {
        let mut views = self.views.lock().await;
        Arc::clone(views.entry(None).or_insert_with(|| {
            // Its OWN persistent profile (`~/.car/browser-profile-user`),
            // not the agent's. Chromium allows exactly one live instance per
            // profile directory, and "an agent browses while the person uses
            // the drawer" is an ordinary situation — sharing one directory
            // made whichever launched second die on SingletonLock. Nothing in
            // the design requires the two to share cookies; the standing
            // session's own sign-ins persist across launches independently.
            let tools = BrowserTools::standing_session(self.root.clone());
            // Task 7's decision rule: this browser is reachable ONLY through
            // browser.view.*, which every method on this surface requires the
            // host-management client for (checked before any view is even
            // resolved — see this module's own authorization doc comment).
            // Whatever call is about to trigger its lazy launch
            // (user_navigate / user_open_tab) is therefore ITSELF coming
            // from a connected host, by construction — no live probe
            // needed, unlike the in-daemon agent path.
            tools.set_host_connectivity(Arc::new(AlwaysConnected));
            // No sign-in attention, deliberately: this browser is reachable
            // only through `browser.view.*` user input, never through an
            // agent's tools, so `browser_await_signin` can never run against
            // it and a pending sign-in can never arise here. Nothing to
            // announce — and announcing the user's own sign-ins would be
            // noise about something they are already looking at.
            Arc::new(BrowserView::new(None, Arc::new(tools)))
        }))
    }

    /// Publish an agent-attached browser under a conversation/agent-session
    /// key, so the drawer can watch that conversation specifically.
    ///
    /// **This is the producer API.** Anything that builds an assistant
    /// runtime hands the runtime's own `BrowserTools`
    /// (`AssistantRuntime::browser`) in here; the registry never creates an
    /// agent's browser itself. Registering costs nothing for a run that never
    /// browses — `BrowserTools` launches Chromium lazily.
    ///
    /// **Replacement is the lifetime bound.** A conversation's view outlives
    /// the run that created it (see [`BrowserView::note_run_ended`]), so the thing that
    /// eventually releases the old browser is a NEW run registering for the
    /// SAME key. That makes the standing cost one idle Chromium per
    /// conversation whose agent actually browsed, not one per run.
    ///
    /// Subscribers and the event cursor are handed over to the new view, so a
    /// drawer watching this conversation follows it to the new browser
    /// without re-subscribing — and, critically, without the cursor going
    /// backwards, which would break gap detection far worse than a gap does.
    pub async fn register(
        &self,
        conversation_id: impl Into<String>,
        tools: Arc<BrowserTools>,
    ) -> Arc<BrowserView> {
        let key = Some(conversation_id.into());
        let view = Arc::new(BrowserView::new(key.clone(), tools));
        let previous = self.views.lock().await.insert(key, Arc::clone(&view));
        if let Some(previous) = previous {
            view.adopt(&previous).await;
            previous.resolve_signin_attention_on_teardown().await;
        }
        view
    }

    /// Publish a SUPERVISED AGENT PROCESS's browser under a conversation key.
    ///
    /// The relay twin of [`Self::register`], with one difference that matters:
    /// **the same process re-claiming its own conversation is a no-op.** A
    /// supervised agent registers on every turn, and churning the view each
    /// time would reset its cursor and drop the drawer's stream for no reason.
    /// A DIFFERENT process claiming the key — the supervisor restarted it —
    /// replaces the view through the ordinary `adopt` path, so a drawer that
    /// never unsubscribed follows the agent across without the cursor moving
    /// backwards.
    pub async fn register_relay(
        &self,
        conversation_id: impl Into<String>,
        producer: Arc<RelayProducer>,
    ) -> Arc<BrowserView> {
        let key = Some(conversation_id.into());
        if let Some(existing) = self.views.lock().await.get(&key) {
            if existing.is_served_by(&producer) {
                return Arc::clone(existing);
            }
        }
        let view = Arc::new(BrowserView::with_browser(
            key.clone(),
            ViewBrowser::Relay(Arc::clone(&producer)),
        ));
        // Attach BEFORE the map insert so a push that lands in the window
        // reaches the view a concurrent subscriber may already hold.
        producer.attach_view(&view).await;
        let previous = self.views.lock().await.insert(key, Arc::clone(&view));
        if let Some(previous) = previous {
            view.adopt(&previous).await;
            previous.resolve_signin_attention_on_teardown().await;
        }
        // The relay path's operator attention belongs to the producer: one
        // process has one browser even though it can back many per-turn views.
        producer
            .set_signin_attention(self.signin_attention(), view.key.clone())
            .await;
        self.retire_views_past_the_cap(&producer).await;
        view
    }

    /// A registration replaces this producer's oldest views — see
    /// [`crate::browser_relay::MAX_VIEWS_PER_PRODUCER`] for why the cap is
    /// where it is and why "retire" is not "delete".
    async fn retire_views_past_the_cap(&self, producer: &Arc<RelayProducer>) {
        for view in producer.views_past_the_cap().await {
            // The atomic only, NOT `note_run_ended`: that relays a run-end
            // transition to the process, which is still serving its other
            // conversations. This is the eviction flag, exactly as
            // `note_producer_disconnected` sets it.
            view.run_ended
                .store(true, std::sync::atomic::Ordering::Release);
            self.release_if_idle(&view).await;
            // The binding goes with it, released or not. It exists to entitle
            // a re-registration between turns, and past this cap there is no
            // re-registration to entitle: the agent's own `known` deque has
            // already forgotten this conversation. Left in place it was the
            // one per-turn entry `note_producer_disconnected` deliberately
            // does NOT reclaim, growing for the daemon's life.
            if let Some(key) = view.key.as_deref() {
                self.producers.forget_binding(key).await;
            }
        }
    }

    /// The producer for an agent connection, created on its first
    /// registration.
    pub async fn producer_for(
        &self,
        client_id: &str,
        agent_id: &str,
        channel: &Arc<WsChannel>,
    ) -> Arc<RelayProducer> {
        self.producers
            .get_or_create(client_id, agent_id, channel)
            .await
    }

    /// Tell every supervised agent process that host connectivity changed.
    ///
    /// They cache the answer — a supervised process has no read of the
    /// daemon's session set — and it decides whether `browser_await_signin`
    /// points the user at the drawer or tells them to open the CAR app.
    pub async fn broadcast_host_connected(&self, connected: bool) {
        self.producers.broadcast_host_connected(connected).await;
    }

    /// The producer a given connection registered, if any. The gate on
    /// inbound pushes: a connection that never registered has none.
    pub async fn producer(&self, client_id: &str) -> Option<Arc<RelayProducer>> {
        self.producers.get(client_id).await
    }

    /// The agent entitled to publish a conversation, and how that gets
    /// recorded. See `browser_relay::authorize_conversation_claim`.
    pub async fn conversation_owner(&self, conversation_id: &str) -> Option<String> {
        self.producers.conversation_owner(conversation_id).await
    }

    pub async fn bind_conversation(&self, conversation_id: &str, agent_id: &str) {
        self.producers
            .bind_conversation(conversation_id, agent_id)
            .await;
    }

    /// An agent connection dropped — its browser went with the process. Its
    /// views stay registered, reporting an empty browser, so a restarted
    /// process can replace them and carry the drawer across.
    pub async fn note_producer_disconnected(&self, client_id: &str) {
        let producer = self.producers.get(client_id).await;
        self.producers.note_disconnected(client_id).await;
        let Some(producer) = producer else { return };

        // The producer's process is gone, so every view it served has, in the
        // only sense that matters here, had its run end. Marking that is what
        // makes those views EVICTABLE at all: `release_if_idle` returns early
        // on `!run_ended`, and `run_ended` was set only by the in-daemon
        // run-end guard — so a relay view could never be released by any path,
        // and each retired conversation permanently retained a map entry, its
        // `RelayProducer`, and that producer's `Arc<WsChannel>` — the write
        // half of a dead socket, whose file descriptor could then never close.
        //
        // Releasing only the UNWATCHED ones preserves the restart-adoption
        // behaviour this deliberately kept: a view a drawer is still
        // subscribed to stays registered, so a restarted process replaces it
        // in place and the drawer follows across without re-subscribing. What
        // goes is a view for a conversation nobody is watching and no process
        // is serving.
        let views: Vec<Arc<BrowserView>> = self
            .views
            .lock()
            .await
            .values()
            .filter(|view| view.is_served_by(&producer))
            .cloned()
            .collect();
        for view in views {
            view.run_ended
                .store(true, std::sync::atomic::Ordering::Release);
            self.release_if_idle(&view).await;
        }
    }

    /// Release a view whose run has ended and that **nobody is subscribed
    /// to**: drop it from the map (and with it the last reference to the
    /// browser it was serving) and stop its capture. Reports whether it went.
    ///
    /// This is the eviction path the lifetime bound needs. Replacement —
    /// "a NEW run registering for the SAME key" — is the documented bound,
    /// and it is genuinely unreachable for a producer that mints a fresh key
    /// per run: the in-daemon `assistant_start` path keys on
    /// `mcp-run-<uuid>`, so before this, every assistant run that browsed
    /// left one idle Chromium registered for the daemon's whole lifetime.
    ///
    /// It does NOT weaken "a view outlives the run that created it". A
    /// subscribed view is never released, so the drawer still keeps showing
    /// the last page exactly as the agent left it, with agent-opened tabs
    /// usable, for as long as anything is actually watching. What goes is a
    /// finished run's browser that no drawer ever attached to — which nobody
    /// can observe, and which is precisely the leak.
    ///
    /// Identity-checked, never key-checked: a successor may already hold the
    /// key by the time an asynchronous run-end signal arrives, and removing
    /// by key would take a LIVE run's browser down. Same reasoning as
    /// [`BrowserView::note_run_ended`] being a method on the view.
    pub async fn release_if_idle(&self, view: &Arc<BrowserView>) -> bool {
        if !view.run_ended.load(std::sync::atomic::Ordering::Acquire) {
            return false;
        }
        // The emptiness decision and the map removal happen under ONE HELD
        // fanout lock. Reading `is_idle()` and then removing would repeat the
        // hazard `stop_streamer_if_unwatched` exists for: `subscribe` does a
        // CDP tab read (`refresh_presentation`) BEFORE it inserts, so the
        // window between "observed empty" and "removed" is round trips wide,
        // and a subscriber landing inside it gets a successful reply, a
        // cursor, and then no event ever again — zero events means the
        // cursor-gap recovery can never fire.
        //
        // `fanout` then `views` is the only nesting this crate has: `get`,
        // `register` and `drop_subscriptions_for_client` all release the
        // registry lock before touching a view's fanout, so there is no cycle.
        let released = {
            let fanout = view.fanout.lock().await;
            if !fanout.subscribers.is_empty() {
                return false;
            }
            let mut views = self.views.lock().await;
            match views.get(&view.key) {
                Some(current) if Arc::ptr_eq(current, view) => views.remove(&view.key).is_some(),
                _ => false,
            }
        };
        if !released {
            return false;
        }
        if !self.stop_or_restore(view).await {
            return false;
        }
        view.resolve_signin_attention_on_teardown().await;
        true
    }

    /// Durable state returned by `host.subscribe`, independent of the bounded
    /// host-event backlog. Relay producers are deduplicated because every view
    /// they back shows the same one process-owned browser.
    pub async fn pending_signins(&self) -> Vec<BrowserSignInSnapshot> {
        let views: Vec<Arc<BrowserView>> = self.views.lock().await.values().cloned().collect();
        let mut seen_producers = HashSet::new();
        let mut pending = Vec::new();
        for view in views {
            match &view.browser {
                ViewBrowser::Local(tools) => {
                    if let Some(message) = tools.pending_signin_message().await {
                        pending.push(BrowserSignInSnapshot::new(view.key.as_deref(), message));
                    }
                }
                ViewBrowser::Relay(producer) => {
                    if seen_producers.insert(producer.client_id().to_string()) {
                        if let Some(signin) = producer.signin_snapshot().await {
                            pending.push(signin);
                        }
                    }
                }
            }
        }
        pending.sort_by(|a, b| a.conversation_id.cmp(&b.conversation_id));
        pending
    }

    /// The second half of [`Self::release_if_idle`]: the view is out of the
    /// registry, so either stop its stream or undo the removal.
    ///
    /// A subscriber can still arrive between the removal and this call — it
    /// took its `Arc` from the registry BEFORE the entry went, which no lock
    /// can undo. The guarded stop refuses in that case, leaving the stream
    /// running, and the view is then put BACK: a live stream on a key that no
    /// longer resolves would render frames while every input and control call
    /// answered "no browser view for conversation".
    async fn stop_or_restore(&self, view: &Arc<BrowserView>) -> bool {
        view.stop_streamer_if_unwatched().await;
        if !view.fanout.lock().await.subscribers.is_empty() {
            let mut views = self.views.lock().await;
            views
                .entry(view.key.clone())
                .or_insert_with(|| Arc::clone(view));
            return false;
        }
        true
    }

    pub async fn get(&self, conversation_id: Option<&str>) -> Option<Arc<BrowserView>> {
        match conversation_id {
            None => Some(self.standing().await),
            Some(id) => self
                .views
                .lock()
                .await
                .get(&Some(id.to_string()))
                .map(Arc::clone),
        }
    }

    /// Disconnect cleanup: drop this connection's subscriptions everywhere,
    /// and start the control grace period on any view it was driving.
    pub async fn drop_subscriptions_for_client(&self, client_id: &str) {
        let views: Vec<Arc<BrowserView>> = self.views.lock().await.values().cloned().collect();
        // Concurrently, not one after another. Both relay-backed cleanups
        // settle local state synchronously and then detach the process
        // reconciliation, so disconnect teardown never inherits the relay's
        // bound — which is up to 2×`RELAY_CALL_TIMEOUT` (~60s), a bounded
        // write plus a bounded wait for the reply, not 30s.
        futures::future::join_all(views.into_iter().map(|view| async move {
            let was_watching = view.unsubscribe(client_id).await;
            let held_control = view.note_disconnect(client_id, was_watching).await;
            // A person mid-sign-in who never pressed Take control has no
            // holder record, so `note_disconnect` above is inert for them —
            // and since round 8 their window survives the run ending. This is
            // what gives it an exit.
            //
            // Skipped when this same connection held control, which is the
            // ordinary drawer shape: the holder is also a subscriber, so
            // `was_watching` alone would fire both. Two timers then retire
            // each other by generation, and which semantics survives is
            // decided by whichever relay reply lands last. Exactly one path
            // owns each disconnect — and `was_watching` is handed to
            // `note_disconnect` above precisely so the surviving timer still
            // carries the WATCHER semantics for that shape.
            if was_watching && !held_control {
                view.note_watcher_disconnect().await;
            }
            // The second eviction trigger (the first is the run ending): a
            // view whose run already ended and whose last watcher just went
            // away has nothing left to show anyone.
            self.release_if_idle(&view).await;
        }))
        .await;
    }
}

impl Default for BrowserViewRegistry {
    fn default() -> Self {
        Self::new(car_home::root().unwrap_or_else(std::env::temp_dir))
    }
}

// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------

#[derive(Debug, Default, Deserialize)]
struct ViewParams {
    #[serde(default)]
    conversation_id: Option<String>,
}

/// Parse the shared `{ conversation_id? }` envelope. Absent params are the
/// standing session, so a bare `{}` — or no params at all — is valid.
fn view_params(req: &JsonRpcMessage) -> Result<ViewParams, String> {
    if req.params.is_null() {
        return Ok(ViewParams::default());
    }
    serde_json::from_value(req.params.clone())
        .map_err(|e| format!("browser.view.* takes an optional {{ conversation_id }}: {e}"))
}

/// Every `browser.view.*` method requires the host-management client. See
/// the module docs for why this is stricter than `runs.subscribe`.
fn authorize(session: &ClientSession) -> Result<(), String> {
    if session.is_host.load(std::sync::atomic::Ordering::Acquire) {
        return Ok(());
    }
    tracing::debug!(
        client_id = %session.client_id,
        "browser.view.* denied: connection is not the host management client"
    );
    Err(
        "not authorized to use browser.view.*: this connection is not the host management \
         client (session.auth { host_token })"
            .to_string(),
    )
}

async fn resolve(
    state: &Arc<ServerState>,
    conversation_id: Option<&str>,
) -> Result<Arc<BrowserView>, String> {
    match state.browser_views.get(conversation_id).await {
        Some(view) => Ok(view),
        None => Err(format!(
            "no browser view for conversation '{}' — that conversation has no agent-attached \
             browser; omit `conversation_id` for the standing session",
            conversation_id.unwrap_or_default()
        )),
    }
}

fn snapshot_value(conversation_id: Option<&str>, snapshot: (WirePresentation, u64)) -> Value {
    let (presentation, cursor) = snapshot;
    json!({
        "conversation_id": conversation_id,
        "standing_session": conversation_id.is_none(),
        "cursor": cursor,
        "presentation": presentation,
    })
}

/// `browser.view.subscribe { conversation_id? }` — snapshot + cursor, then
/// pushed `browser.view.event` notifications.
pub async fn handle_subscribe(
    req: &JsonRpcMessage,
    session: &ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    authorize(session)?;
    let params = view_params(req)?;
    let view = resolve(state, params.conversation_id.as_deref()).await?;
    let snapshot = view
        .subscribe(&session.client_id, session.channel.clone())
        .await;
    Ok(snapshot_value(params.conversation_id.as_deref(), snapshot))
}

/// `browser.view.unsubscribe { conversation_id? }` — idempotent.
pub async fn handle_unsubscribe(
    req: &JsonRpcMessage,
    session: &ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    authorize(session)?;
    let params = view_params(req)?;
    let view = resolve(state, params.conversation_id.as_deref()).await?;
    let removed = view.unsubscribe(&session.client_id).await;
    // The ORDINARY teardown path, and it was the one that never released.
    // `release_if_idle` was wired to the run-end guard and to the disconnect
    // sweep — but a disconnect is the exceptional route; closing the drawer or
    // switching conversations is this plain RPC over a live socket. A view
    // whose run had already ended therefore kept its `Arc<BrowserTools>`, and
    // its Chromium, for the daemon's lifetime — once per watched run, with the
    // documented replacement bound unable to fire because the key carries a
    // fresh uuid.
    //
    // Unconditional is safe: it returns early unless the run has ended, and
    // re-checks emptiness under the fanout lock before removing anything.
    state.browser_views.release_if_idle(&view).await;
    Ok(json!({
        "conversation_id": params.conversation_id,
        "removed": removed,
    }))
}

/// `browser.view.take_control { conversation_id? }`.
pub async fn handle_take_control(
    req: &JsonRpcMessage,
    session: &ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    authorize(session)?;
    let params = view_params(req)?;
    let view = resolve(state, params.conversation_id.as_deref()).await?;
    let snapshot = view.take_control(&session.client_id).await?;
    Ok(snapshot_value(params.conversation_id.as_deref(), snapshot))
}

/// `browser.view.hand_back { conversation_id? }`.
pub async fn handle_hand_back(
    req: &JsonRpcMessage,
    session: &ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    authorize(session)?;
    let params = view_params(req)?;
    let view = resolve(state, params.conversation_id.as_deref()).await?;
    let snapshot = view.hand_back(&session.client_id).await?;
    Ok(snapshot_value(params.conversation_id.as_deref(), snapshot))
}

/// The input methods. One enum + one handler keeps the authorization,
/// view resolution and control check in exactly one place — the thing that
/// must never differ between navigate and click.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputOp {
    Navigate,
    Click,
    Type,
    Keypress,
    Scroll,
    Paste,
    Back,
    Forward,
    Reload,
    TabOpen,
    TabClose,
    TabSwitch,
}

#[derive(Debug, Deserialize)]
struct InputParams {
    #[serde(default)]
    conversation_id: Option<String>,
    #[serde(default)]
    url: Option<String>,
    #[serde(default)]
    x: Option<f64>,
    #[serde(default)]
    y: Option<f64>,
    #[serde(default)]
    text: Option<String>,
    #[serde(default)]
    key: Option<String>,
    #[serde(default)]
    modifiers: Vec<String>,
    #[serde(default)]
    delta_y: Option<i32>,
    #[serde(default)]
    tab_id: Option<String>,
}

/// Map a wire modifier name onto car-browser's own enum. Unknown names are
/// rejected rather than ignored — a typo'd modifier silently dropping is how
/// a paste shortcut turns into a stray keystroke in a page.
pub(crate) fn parse_modifier(name: &str) -> Result<Modifier, String> {
    match name.to_ascii_lowercase().as_str() {
        "alt" | "option" => Ok(Modifier::Alt),
        "control" | "ctrl" => Ok(Modifier::Control),
        "meta" | "command" | "cmd" => Ok(Modifier::Meta),
        "shift" => Ok(Modifier::Shift),
        other => Err(format!(
            "unknown modifier '{other}' — use alt, control, meta, or shift"
        )),
    }
}

/// `browser.view.{navigate,click,type,keypress,scroll,tab_open,tab_close,tab_switch}`.
pub async fn handle_input(
    op: InputOp,
    req: &JsonRpcMessage,
    session: &ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    authorize(session)?;
    let params: InputParams = if req.params.is_null() {
        serde_json::from_value(json!({})).map_err(|e| e.to_string())?
    } else {
        serde_json::from_value(req.params.clone())
            .map_err(|e| format!("invalid browser.view input params: {e}"))?
    };
    let view = resolve(state, params.conversation_id.as_deref()).await?;
    view.require_control(&session.client_id).await?;

    // Params are validated HERE, before the view is touched — so a malformed
    // call is refused identically whether the browser is in this process or
    // in a supervised agent's, and a relayed one never pays a round trip for
    // a request that could never have worked.
    let input = match op {
        InputOp::Navigate => ViewInput::Navigate {
            url: params.url.ok_or("browser.view.navigate requires { url }")?,
        },
        InputOp::Click => match (params.x, params.y) {
            (Some(x), Some(y)) => ViewInput::Click { x, y },
            _ => return Err("browser.view.click requires { x, y }".to_string()),
        },
        InputOp::Type => ViewInput::Type {
            text: params.text.ok_or("browser.view.type requires { text }")?,
        },
        InputOp::Keypress => ViewInput::Keypress {
            key: params.key.ok_or("browser.view.keypress requires { key }")?,
            modifiers: params
                .modifiers
                .iter()
                .map(|m| parse_modifier(m))
                .collect::<Result<Vec<_>, _>>()?,
        },
        InputOp::Scroll => ViewInput::Scroll {
            delta_y: params
                .delta_y
                .ok_or("browser.view.scroll requires { delta_y }")?,
        },
        InputOp::Paste => ViewInput::Paste {
            text: params.text.ok_or("browser.view.paste requires { text }")?,
        },
        // No params of their own: which page Back/Forward/Reload act on is
        // the active tab's history, which the browser already knows.
        InputOp::Back => ViewInput::Back,
        InputOp::Forward => ViewInput::Forward,
        InputOp::Reload => ViewInput::Reload,
        InputOp::TabOpen => ViewInput::TabOpen,
        InputOp::TabClose => ViewInput::TabClose {
            tab_id: params
                .tab_id
                .ok_or("browser.view.tab_close requires { tab_id }")?,
        },
        InputOp::TabSwitch => ViewInput::TabSwitch {
            tab_id: params
                .tab_id
                .ok_or("browser.view.tab_switch requires { tab_id }")?,
        },
    };
    let opened_tab = view.browser.input(input).await?;

    // No explicit refresh here on purpose. Every input that can change what
    // the drawer renders (navigate, click, keypress, the tab operations)
    // signals the change from inside `BrowserTools`, and the streamer
    // coalesces those into one presentation event — whereas refreshing here
    // would cost a CDP sweep per open tab on EVERY call, including the ones
    // that cannot change the tab strip at all (type, scroll).
    let mut out = json!({
        "ok": true,
        "conversation_id": params.conversation_id,
    });
    if let (Some(out), Some(tab_id)) = (out.as_object_mut(), opened_tab) {
        out.insert("tab_id".to_string(), json!(tab_id));
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::assistant::browser_control::ControlEvent;
    use crate::session::{ServerStateConfig, WsSink};
    use futures::StreamExt;

    fn capture_channel() -> (
        Arc<WsChannel>,
        futures::channel::mpsc::UnboundedReceiver<Message>,
    ) {
        use futures::sink::SinkExt as _;
        let (tx, rx) = futures::channel::mpsc::unbounded::<Message>();
        let sink: WsSink =
            Box::pin(tx.sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed));
        let channel = Arc::new(WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(0),
        });
        (channel, rx)
    }

    /// A channel whose sink accepts ONE message and then blocks forever —
    /// the drain task parks on its second write, exactly like a half-open
    /// socket whose TCP buffer is full.
    fn wedged_channel() -> (Arc<WsChannel>, futures::channel::mpsc::Receiver<Message>) {
        use futures::sink::SinkExt as _;
        let (tx, rx) = futures::channel::mpsc::channel::<Message>(1);
        let sink: WsSink =
            Box::pin(tx.sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed));
        let channel = Arc::new(WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(0),
        });
        (channel, rx)
    }

    fn test_view() -> Arc<BrowserView> {
        Arc::new(BrowserView::new(
            None,
            Arc::new(BrowserTools::new(std::env::temp_dir())),
        ))
    }

    async fn next_event(
        rx: &mut futures::channel::mpsc::UnboundedReceiver<Message>,
    ) -> BrowserViewEvent {
        let frame = tokio::time::timeout(Duration::from_secs(2), rx.next())
            .await
            .expect("an event within the deadline")
            .expect("a frame");
        let text = match frame {
            Message::Text(t) => t.to_string(),
            other => panic!("expected a text frame, got {other:?}"),
        };
        let json: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(json["method"], "browser.view.event");
        serde_json::from_value(json["params"].clone()).expect("a browser.view.event payload")
    }

    // ---- snapshot + cursor consistency ---------------------------------

    #[tokio::test]
    async fn subscribe_returns_a_snapshot_and_cursor_and_events_advance_it_by_one() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        let (channel, mut rx) = capture_channel();
        let (snapshot, cursor) = view.subscribe("host-1", channel).await;
        assert_eq!(snapshot.owner, WireOwner::Agent);
        assert!(snapshot.tabs.is_empty());
        assert!(!snapshot.blackout_active);

        // Three state changes → three events, cursors strictly +1.
        view.tools().take_control().await;
        view.refresh_presentation().await;
        let first = next_event(&mut rx).await;
        assert_eq!(first.cursor, cursor + 1);

        view.tools().hand_back().await;
        view.refresh_presentation().await;
        let second = next_event(&mut rx).await;
        assert_eq!(second.cursor, cursor + 2);

        view.emit_frame(test_frame(3)).await;
        let third = next_event(&mut rx).await;
        assert_eq!(third.cursor, cursor + 3);
        match third.payload {
            BrowserViewPayload::Frame { frame } => {
                assert_eq!(BASE64.decode(frame.jpeg_base64).unwrap(), vec![3]);
                assert_eq!(frame.width, 1920);
            }
            BrowserViewPayload::Presentation { .. } => panic!("expected a frame event"),
        }
    }

    /// Take control of WHAT? With no agent attached there is no ceremony to
    /// enter, so `take_control` records nothing and the snapshot still reports
    /// `WireOwner::None`.
    ///
    /// Standalone because the reconnect/resume rewrite of
    /// `re_subscribing_yields_a_fresh_snapshot_at_the_current_cursor` took this
    /// assertion with it, and every other `take_control` call site in this
    /// module attaches an agent first. What it guards: a change that let
    /// ownership be recorded with no agent attached would put the drawer into
    /// a privacy blackout for a browser nobody is driving, with no hand-back
    /// affordance pointing at anyone.
    #[tokio::test]
    async fn take_control_with_no_agent_attached_records_no_owner() {
        let view = test_view();
        view.tools().take_control().await;

        let (channel, _rx) = capture_channel();
        let (snapshot, _) = view.subscribe("host-1", channel).await;
        assert_eq!(snapshot.owner, WireOwner::None);
        assert!(
            !snapshot.blackout_active,
            "and no blackout for a browser nobody is driving"
        );
    }

    #[tokio::test]
    async fn an_unchanged_presentation_emits_nothing_and_does_not_burn_a_cursor() {
        let view = test_view();
        let (channel, mut rx) = capture_channel();
        let (_, cursor) = view.subscribe("host-1", channel).await;
        view.refresh_presentation().await;
        view.refresh_presentation().await;
        assert_eq!(view.fanout.lock().await.cursor, cursor);
        assert!(rx.try_recv().is_err(), "nothing changed, nothing emitted");
    }

    #[tokio::test]
    async fn re_subscribing_yields_a_fresh_snapshot_at_the_current_cursor() {
        let view = test_view();
        let (channel, mut rx) = capture_channel();
        view.subscribe("host-1", channel).await;
        view.emit_frame(test_frame(1)).await;
        assert_eq!(next_event(&mut rx).await.cursor, 1);
        assert!(view.unsubscribe("host-1").await);

        // Advance while disconnected. The reconnect must snapshot this missed
        // cursor, then resume with the immediately following event.
        view.emit_frame(test_frame(2)).await;

        let (channel2, mut rx2) = capture_channel();
        let (snapshot, cursor) = view.subscribe("host-1", channel2).await;
        assert_eq!(snapshot.owner, WireOwner::None);
        assert_eq!(cursor, 2, "the reconnect snapshots the event it missed");
        assert_eq!(cursor, view.fanout.lock().await.cursor);

        view.emit_frame(test_frame(3)).await;
        assert_eq!(next_event(&mut rx2).await.cursor, 3);
    }

    // ---- explicit fanout, and the slow subscriber ----------------------

    #[tokio::test]
    async fn two_subscribers_both_get_every_event_and_dropping_one_leaves_the_other_streaming() {
        let view = test_view();
        let (channel_a, mut rx_a) = capture_channel();
        let (channel_b, mut rx_b) = capture_channel();
        view.subscribe("host-a", channel_a).await;
        view.subscribe("host-b", channel_b).await;

        view.emit_frame(test_frame(1)).await;
        assert_eq!(next_event(&mut rx_a).await.cursor, 1);
        assert_eq!(next_event(&mut rx_b).await.cursor, 1);

        assert!(view.unsubscribe("host-a").await);
        view.emit_frame(test_frame(2)).await;
        assert_eq!(
            next_event(&mut rx_b).await.cursor,
            2,
            "the surviving subscriber keeps streaming"
        );
        assert!(
            rx_a.try_recv().is_err(),
            "the dropped subscriber receives nothing further"
        );
    }

    #[tokio::test]
    async fn unsubscribing_twice_is_idempotent() {
        let view = test_view();
        let (channel, _rx) = capture_channel();
        view.subscribe("host-1", channel).await;
        assert!(view.unsubscribe("host-1").await);
        assert!(!view.unsubscribe("host-1").await);
        assert_eq!(view.subscriber_count().await, 0);
    }

    /// The bounded-fanout property: a subscriber that stops draining is
    /// dropped from the stream rather than blocking the producer. Here the
    /// WS sink is never read, so the drain task parks on its first write and
    /// the channel fills; every push past the cap is DROPPED, and the
    /// producer keeps returning promptly.
    #[tokio::test]
    async fn a_slow_subscriber_loses_events_instead_of_blocking_the_producer() {
        let view = test_view();
        // A wedged socket: a capacity-1 sink nobody reads. The drain task
        // parks on its second write, the bounded channel behind it fills,
        // and every push past the cap is dropped.
        let (channel, _rx) = wedged_channel();
        view.subscribe("host-slow", channel).await;

        let pushes = BROWSER_VIEW_CHANNEL_CAP * 4;
        let start = std::time::Instant::now();
        for i in 0..pushes {
            view.emit_frame(test_frame(i as u8)).await;
        }
        assert!(
            start.elapsed() < Duration::from_secs(2),
            "the producer must never park behind a wedged subscriber"
        );
        assert_eq!(
            view.fanout.lock().await.cursor,
            pushes as u64,
            "every event was stamped; the wedged subscriber simply lost most of them"
        );

        // And the view is still healthy for everyone else.
        let (good, mut good_rx) = capture_channel();
        view.subscribe("host-ok", good).await;
        view.emit_frame(test_frame(0)).await;
        assert_eq!(next_event(&mut good_rx).await.cursor, pushes as u64 + 1);
    }

    // ---- control ownership at the wire level ---------------------------

    #[tokio::test]
    async fn with_no_agent_involved_anyone_may_drive() {
        let view = test_view();
        view.require_control("host-1")
            .await
            .expect("zero ceremony: the standing session is just a browser");
    }

    #[tokio::test]
    async fn while_the_agent_drives_nobody_may_input() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        let err = view.require_control("host-1").await.unwrap_err();
        assert!(err.contains("take_control"), "got: {err}");
    }

    #[tokio::test]
    async fn take_control_moves_input_rights_to_that_connection_and_hand_back_returns_them() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;

        let (snapshot, _) = view
            .take_control("host-1")
            .await
            .expect("a local view never fails");
        assert_eq!(snapshot.owner, WireOwner::User);
        assert!(
            snapshot.blackout_active,
            "user control blacks the model out"
        );
        view.require_control("host-1")
            .await
            .expect("the control holder may drive");
        let err = view.require_control("host-2").await.unwrap_err();
        assert!(
            err.contains("another connection holds control"),
            "got: {err}"
        );

        let (snapshot, _) = view
            .hand_back("host-1")
            .await
            .expect("a local view never fails");
        assert_eq!(snapshot.owner, WireOwner::Agent);
        assert!(!snapshot.blackout_active);
        assert!(view.require_control("host-1").await.is_err());
    }

    /// The sign-in strip hands the page to the human without a Take control
    /// press — the credential fields have to accept their typing.
    #[tokio::test]
    async fn a_pending_signin_lets_the_user_type_without_taking_control() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        assert!(view.require_control("host-1").await.is_err());

        view.tools()
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        view.require_control("host-1")
            .await
            .expect("the human must be able to type their password");

        let (snapshot, _) = view.snapshot().await;
        assert_eq!(snapshot.pending_signin.as_deref(), Some("Sign in at x"));
        assert!(snapshot.blackout_active);
    }

    /// Sign-in lifecycle, hand-back path: the strip clears, control returns
    /// to the agent, and the blackout lifts — all visible on the wire.
    #[tokio::test]
    async fn hand_back_resolves_a_pending_signin_on_the_wire() {
        let view = test_view();
        let (channel, mut rx) = capture_channel();
        view.tools().attach_agent_for_test().await;
        view.subscribe("host-1", channel).await;

        view.tools()
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        view.refresh_presentation().await;
        let pending = next_event(&mut rx).await;
        match pending.payload {
            BrowserViewPayload::Presentation { presentation } => {
                assert_eq!(presentation.pending_signin.as_deref(), Some("Sign in at x"));
                assert!(presentation.blackout_active);
            }
            BrowserViewPayload::Frame { .. } => panic!("expected a presentation event"),
        }

        let (snapshot, _) = view
            .hand_back("host-1")
            .await
            .expect("a local view never fails");
        assert_eq!(snapshot.pending_signin, None, "the strip clears");
        assert_eq!(snapshot.owner, WireOwner::Agent);
        assert!(!snapshot.blackout_active);
    }

    /// Sign-in lifecycle, timeout path: `browser_await_signin`'s own timeout
    /// resolves the strip with nobody handing anything back, and the drawer
    /// sees exactly that.
    #[tokio::test]
    async fn a_timed_out_signin_clears_the_strip_and_leaves_the_agent_driving() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.tools()
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        view.tools()
            .apply_control_for_test(ControlEvent::SignInResolved { signed_in: false })
            .await;

        let (snapshot, _) = view.snapshot().await;
        assert_eq!(snapshot.pending_signin, None);
        assert_eq!(
            snapshot.owner,
            WireOwner::Agent,
            "a timeout returns control to the agent, unchanged from today"
        );
        assert!(!snapshot.blackout_active);
    }

    // ---- disconnect + grace period -------------------------------------

    #[tokio::test(start_paused = true)]
    async fn control_reverts_to_the_agent_after_the_grace_period() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();

        view.note_disconnect("host-1", false).await;
        assert_eq!(
            view.snapshot().await.0.owner,
            WireOwner::User,
            "still the user's during the grace window"
        );

        tokio::time::sleep(CONTROL_GRACE + Duration::from_secs(1)).await;
        assert_eq!(view.snapshot().await.0.owner, WireOwner::Agent);
        assert!(view.control.lock().await.holder.is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn a_disconnect_by_a_connection_that_never_held_control_starts_no_timer() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();

        view.note_disconnect("host-2", false).await;
        tokio::time::sleep(CONTROL_GRACE + Duration::from_secs(1)).await;
        assert_eq!(
            view.snapshot().await.0.owner,
            WireOwner::User,
            "host-1 still holds control; host-2 leaving is irrelevant"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn taking_control_again_inside_the_window_survives_the_stale_expiry() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();
        view.note_disconnect("host-1", false).await;

        // The host reconnects and takes control again before the window
        // closes; the in-flight timer must not yank it away afterwards.
        tokio::time::sleep(Duration::from_secs(1)).await;
        view.take_control("host-2").await.unwrap();
        tokio::time::sleep(CONTROL_GRACE + Duration::from_secs(1)).await;

        assert_eq!(view.snapshot().await.0.owner, WireOwner::User);
        assert_eq!(view.control.lock().await.holder.as_deref(), Some("host-2"));
    }

    // ---- run lifecycle --------------------------------------------------

    #[tokio::test]
    async fn a_run_ending_returns_the_browser_to_the_user_with_no_ceremony() {
        // The agent still held control when its run ended, so there is
        // nothing to hand back: the strip disappears and every control is
        // live immediately, with no Take control click.
        let view = test_view();
        view.tools().attach_agent_for_test().await;

        view.note_run_ended().await;
        let (snapshot, _) = view.snapshot().await;
        assert_eq!(snapshot.owner, WireOwner::None);
        assert_eq!(snapshot.current_action, None);
        assert!(!snapshot.blackout_active);
        view.require_control("host-1")
            .await
            .expect("every control accepts input immediately after a run ends");
    }

    /// The other half of the same bullet pair, on the wire: a run ending
    /// while the USER holds control does not reclaim it, and does not lift
    /// the blackout — the drawer keeps saying the user is driving until they
    /// hand back. (Live verification caught the drawer reporting
    /// `owner=none blackout=false` with the person still at the keyboard.)
    #[tokio::test]
    async fn a_run_ending_while_the_user_drives_leaves_control_and_the_blackout_alone() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();

        view.note_run_ended().await;
        let (snapshot, _) = view.snapshot().await;
        assert_eq!(snapshot.owner, WireOwner::User, "still the user's");
        assert!(
            snapshot.blackout_active,
            "and the model still cannot see the screen"
        );
        assert!(
            view.require_control("host-2").await.is_err(),
            "nor does it re-open input to every connection"
        );
        view.require_control("host-1")
            .await
            .expect("the person who holds control must still be able to type");

        // Hand-back is what ends the ceremony — and with the run gone there
        // is no agent to hand back to, so it lands on the no-agent state.
        let (snapshot, _) = view.hand_back("host-1").await.unwrap();
        assert_eq!(snapshot.owner, WireOwner::None);
        assert!(!snapshot.blackout_active);
        view.require_control("host-1").await.expect("user-drivable");
    }

    /// The interleaving the disconnect sweep produces on an ordinary app
    /// reconnect: `unsubscribe` computes "empty" under the fanout lock and
    /// releases it, the reconnected Command Deck subscribes on a NEW
    /// `client_id` (and `ensure_streamer` returns early, seeing a live pump),
    /// and only then does the sweep's stop land. Stopping there left the new
    /// subscriber with a successful reply, a cursor, and no event ever again
    /// — and because ZERO events arrive, the cursor-gap recovery this module
    /// relies on can never fire, so the drawer freezes on one frame.
    #[tokio::test]
    async fn a_stop_that_lands_after_a_new_subscriber_arrived_leaves_the_stream_running() {
        let view = test_view();
        let (channel_a, _rx_a) = capture_channel();
        view.subscribe("host-a", channel_a).await;
        assert!(
            view.capture.lock().await.active,
            "the first subscriber arms capture"
        );

        // The sweep has already decided "empty" for host-a. Before its stop
        // lands, the reconnected connection subscribes.
        let (channel_b, _rx_b) = capture_channel();
        view.subscribe("host-b", channel_b).await;

        // The late stop.
        view.stop_streamer_if_unwatched().await;
        assert!(
            view.capture.lock().await.active,
            "a subscriber reappeared, so the stop must not land"
        );

        // And the unconditional form still tears a replaced view down.
        view.stop_streamer().await;
        assert!(!view.capture.lock().await.active);
    }

    /// `hand_back` was the one control method that enforced nothing, so a
    /// second host connection could revert the browser to the agent —
    /// resolving the holder's pending sign-in and lifting the privacy
    /// blackout — while that person was mid-sign-in.
    #[tokio::test]
    async fn hand_back_is_refused_from_a_connection_that_does_not_hold_control() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();

        let err = view.hand_back("host-2").await.unwrap_err();
        assert!(
            err.contains("another connection holds control"),
            "got: {err}"
        );
        assert_eq!(
            view.snapshot().await.0.owner,
            WireOwner::User,
            "the holder keeps control"
        );

        view.hand_back("host-1")
            .await
            .expect("the holder may hand back");
        assert_eq!(view.snapshot().await.0.owner, WireOwner::Agent);
    }

    /// The presentation is read OUTSIDE the fanout lock (for a local view
    /// that read is a live CDP round trip), so two concurrent refreshes have
    /// no ordering guarantee — and the loser used to win the lock second and
    /// publish its OLDER state at a HIGHER cursor. Owner, URL, tabs and
    /// blackout all regress on the drawer, `fanout.last` goes stale with
    /// them, and contiguous cursors mean gap detection cannot fire.
    #[tokio::test]
    async fn an_older_concurrent_read_never_overwrites_a_newer_published_one() {
        let view = test_view();
        let (channel, mut rx) = capture_channel();
        view.subscribe("host-1", channel).await;

        // The newer read lands first: the agent attaches, and that is
        // published.
        view.tools().attach_agent_for_test().await;
        view.refresh_presentation().await;
        let newer = next_event(&mut rx).await;
        let BrowserViewPayload::Presentation { presentation } = newer.payload else {
            panic!("expected a presentation event")
        };
        assert_eq!(presentation.owner, WireOwner::Agent);
        let published_revision = presentation.revision;

        // Now the older in-flight read completes. Feeding it directly is the
        // whole point: this is the state a slower `browser.presentation()`
        // call started before the transition above.
        let stale = WirePresentation {
            revision: published_revision - 1,
            owner: WireOwner::None,
            ..presentation.clone()
        };
        view.publish_presentation_for_test(stale).await;

        assert!(
            tokio::time::timeout(Duration::from_millis(50), rx.next())
                .await
                .is_err(),
            "an older read must not be emitted after a newer one"
        );
        assert_eq!(
            view.fanout.lock().await.last.owner,
            WireOwner::Agent,
            "and the cached snapshot every later subscriber gets must not go stale"
        );
    }

    /// The sign-in bypass opens the AGENT's browser to the person without a
    /// Take control press. It must not also open a browser somebody else
    /// already took: `TakeControl` leaves `pending_signin` set, so
    /// `owner == User && signin_pending` is the credential-entry window, and a
    /// blanket bypass admitted every other authorized connection into it.
    #[tokio::test]
    async fn a_pending_signin_does_not_admit_input_from_a_non_holder() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.tools()
            .apply_control_for_test(
                crate::assistant::browser_control::ControlEvent::SignInRequested(
                    "Sign in at x".into(),
                ),
            )
            .await;

        // Before anyone takes control the bypass still does its job: the
        // person types into the agent's browser with no ceremony.
        view.require_control("host-1")
            .await
            .expect("the sign-in strip IS the affordance");

        view.take_control("host-1").await.unwrap();
        view.require_control("host-1")
            .await
            .expect("the holder keeps typing");
        let err = view.require_control("host-2").await.unwrap_err();
        assert!(
            err.contains("another connection holds control"),
            "a second connection must not interleave into the password field: {err}"
        );
    }

    /// The drain task deregisters on exit — and a re-subscribe on the SAME
    /// connection is what ends the old one, by replacing its entry and
    /// closing its channel. Removing by `client_id` alone therefore deleted
    /// the registration the new subscribe had just installed, and the trigger
    /// is this surface's own documented recovery: a cursor gap makes the
    /// drawer re-subscribe the same key on the same connection. One dropped
    /// frame killed the drawer permanently.
    #[tokio::test]
    async fn a_re_subscribe_on_the_same_connection_survives_the_old_drain_task_exiting() {
        let view = test_view();
        let (first, _rx1) = capture_channel();
        view.subscribe("host-1", first).await;
        let (second, mut rx2) = capture_channel();
        view.subscribe("host-1", second).await;

        // Let the replaced subscriber's drain task notice its closed channel
        // and run its exit path.
        for _ in 0..200 {
            tokio::task::yield_now().await;
        }

        assert_eq!(
            view.subscriber_count().await,
            1,
            "the live registration must survive the replaced one's teardown"
        );
        // And it is the SECOND channel that is still being served.
        view.emit_frame(test_frame(7)).await;
        let event = next_event(&mut rx2).await;
        match event.payload {
            BrowserViewPayload::Frame { .. } => {}
            BrowserViewPayload::Presentation { .. } => panic!("expected a frame"),
        }
    }

    /// Round 1 fixed this race on `unsubscribe`'s stop path and left the
    /// twin open on `release_if_idle`, which read `is_idle()`, released the
    /// fanout lock, removed the view, and then took the UNGUARDED stop. The
    /// window is round trips wide: `subscribe` does a CDP tab read before it
    /// inserts.
    #[tokio::test]
    async fn a_release_that_races_a_new_subscriber_keeps_both_the_stream_and_the_view() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register("conv-1", Arc::new(BrowserTools::new(std::env::temp_dir())))
            .await;
        view.note_run_ended().await;

        // The window itself: the release has already removed the entry, and
        // the subscriber lands holding the `Arc` it took from the registry
        // beforehand — which no lock can undo, so this is the state the
        // second half has to cope with rather than prevent.
        registry.views.lock().await.remove(&view.key);
        let (channel, _rx) = capture_channel();
        view.subscribe("host-late", channel).await;

        assert!(
            !registry.stop_or_restore(&view).await,
            "a subscriber arrived, so the release must not complete"
        );
        assert!(
            view.capture.lock().await.active,
            "its stream must not be stopped underneath that subscriber — zero events \
             means the cursor-gap recovery can never fire, so the drawer would freeze"
        );
        assert!(
            registry
                .get(Some("conv-1"))
                .await
                .is_some_and(|v| Arc::ptr_eq(&v, &view)),
            "and the key must resolve again, or every input and control call would \
             answer 'no browser view for conversation' while frames kept arriving"
        );

        // Once it really is unwatched, the release goes through.
        view.unsubscribe("host-late").await;
        assert!(registry.release_if_idle(&view).await);
        assert!(registry.get(Some("conv-1")).await.is_none());
    }

    /// The hand-back gate did not fire in the scenario its own doc comment
    /// describes: `take_control` overwrote the holder unconditionally, so a
    /// second connection reached the same end state in two calls instead of
    /// one.
    #[tokio::test]
    async fn take_control_is_refused_from_a_connection_that_does_not_hold_control() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();

        let err = view.take_control("host-2").await.unwrap_err();
        assert!(
            err.contains("another connection holds control"),
            "got: {err}"
        );
        assert_eq!(
            view.control.lock().await.holder.as_deref(),
            Some("host-1"),
            "the holder is unchanged, so hand_back still refuses host-2 too"
        );
        assert!(view.hand_back("host-2").await.is_err());
    }

    /// `TakeControl` is a documented no-op on the standing session (`NoAgent`
    /// — nobody to take it from), but the holder was recorded anyway. With
    /// the gate above that stale holder would then lock every other
    /// connection out of a browser nobody had actually taken.
    #[tokio::test]
    async fn take_control_records_no_holder_when_the_reducer_did_not_move_ownership() {
        let view = test_view();
        let (snapshot, _) = view.take_control("host-1").await.unwrap();
        assert_eq!(snapshot.owner, WireOwner::None, "documented no-op");
        assert!(
            view.control.lock().await.holder.is_none(),
            "nobody took control, so nobody holds it"
        );
        view.take_control("host-2")
            .await
            .expect("and another connection is not locked out");
    }

    /// Disconnect-then-run-end. `note_run_ended` bumped the control
    /// generation unconditionally, which is the ONLY thing the grace timer
    /// checks — so the timer armed by the disconnect became stale and never
    /// fired, and because the run end is DEFERRED while a person holds
    /// control, nothing else ever cleared `owner: user` or the blackout.
    #[tokio::test(start_paused = true)]
    async fn a_run_ending_after_the_holder_disconnected_still_reverts_when_grace_expires() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();
        view.note_disconnect("host-1", false).await;

        // The run ends inside the grace window, with the user still nominally
        // driving — so the reducer defers, and the timer is the only way out.
        tokio::time::sleep(Duration::from_secs(1)).await;
        view.note_run_ended().await;
        assert_eq!(
            view.snapshot().await.0.owner,
            WireOwner::User,
            "deferred, as designed"
        );

        tokio::time::sleep(CONTROL_GRACE + Duration::from_secs(1)).await;
        let (snapshot, _) = view.snapshot().await;
        assert_eq!(
            snapshot.owner,
            WireOwner::None,
            "the grace timer must still fire — nothing else re-arms it"
        );
        assert!(
            !snapshot.blackout_active,
            "and the blackout must not outlive the vanished controller"
        );
    }

    /// `note_disconnect` used to leave `holder` set and rely on
    /// `ControlEffect::StartGracePeriod` to clear it — but
    /// `control_best_effort` swallows a failed transition and returns NO
    /// effects (a wedged agent process hitting `RELAY_CALL_TIMEOUT` is the
    /// ordinary case), so no timer was spawned and `holder` named a dead
    /// connection forever, with `require_control` refusing every other
    /// connection's input.
    #[tokio::test]
    async fn a_disconnect_clears_the_holder_even_when_the_reducer_asks_for_no_grace_period() {
        let view = test_view();
        view.tools().attach_agent_for_test().await;
        view.take_control("host-1").await.unwrap();
        assert!(
            view.require_control("host-2").await.is_err(),
            "host-1 is driving"
        );

        view.note_disconnect("host-1", false).await;
        assert!(
            view.control.lock().await.holder.is_none(),
            "the connection is provably gone"
        );
        view.require_control("host-2")
            .await
            .expect("a reconnected drawer must be able to drive, and to hand back");
        view.hand_back("host-2")
            .await
            .expect("nobody holds control, so hand-back is admitted");
    }

    // ---- registry + error paths ----------------------------------------

    #[tokio::test]
    async fn the_standing_session_is_one_shared_view_and_launches_nothing() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let a = registry.standing().await;
        let b = registry.get(None).await.expect("always resolvable");
        assert!(
            Arc::ptr_eq(&a, &b),
            "every conversation without an agent browser shares ONE standing session"
        );
        assert!(
            a.snapshot().await.0.tabs.is_empty(),
            "opening the drawer must not launch Chromium"
        );
    }

    #[tokio::test]
    async fn an_unknown_conversation_resolves_to_nothing() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        assert!(registry.get(Some("nope")).await.is_none());
    }

    #[tokio::test]
    async fn a_registered_conversation_is_its_own_view() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let view = registry.register("conv-1", Arc::clone(&tools)).await;
        tools.attach_agent_for_test().await;

        let found = registry.get(Some("conv-1")).await.expect("registered");
        assert!(Arc::ptr_eq(&view, &found));
        assert_eq!(found.snapshot().await.0.owner, WireOwner::Agent);
        assert!(
            !Arc::ptr_eq(&found, &registry.standing().await),
            "an agent's browser is not the standing session"
        );
    }

    /// The browser OUTLIVES its run: when the run ends the strip disappears,
    /// every control accepts input immediately, and the view is still there
    /// to be subscribed and driven — the last page exactly as the agent left
    /// it, agent-opened tabs still usable.
    #[tokio::test]
    async fn a_run_ending_leaves_the_browser_registered_subscribable_and_drivable() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let view = registry.register("conv-1", Arc::clone(&tools)).await;
        tools.attach_agent_for_test().await;
        let (channel, mut rx) = capture_channel();
        view.subscribe("host-1", channel).await;

        view.note_run_ended().await;

        // Still registered, and reachable by the same key.
        let found = registry
            .get(Some("conv-1"))
            .await
            .expect("the browser outlives the run that opened it");
        assert!(Arc::ptr_eq(&view, &found));

        // No ceremony: no strip, no blackout, and input accepted with no
        // Take control click.
        let (snapshot, cursor) = found.snapshot().await;
        assert_eq!(snapshot.owner, WireOwner::None);
        assert_eq!(snapshot.current_action, None);
        assert!(!snapshot.blackout_active);
        found
            .require_control("host-1")
            .await
            .expect("user-drivable");

        // The subscriber saw the transition, and the stream is still live.
        let event = next_event(&mut rx).await;
        match event.payload {
            BrowserViewPayload::Presentation { presentation } => {
                assert_eq!(presentation.owner, WireOwner::None);
            }
            BrowserViewPayload::Frame { .. } => panic!("expected a presentation event"),
        }
        found.emit_frame(test_frame(1)).await;
        assert_eq!(next_event(&mut rx).await.cursor, cursor + 1);
    }

    /// The lifetime bound: a NEW run for the same conversation replaces the
    /// view and RELEASES the previous browser, so the standing cost is one
    /// idle Chromium per conversation, not per run. Subscribers and the
    /// cursor come across, because a cursor that moved backwards would break
    /// gap detection worse than a gap does.
    #[tokio::test]
    async fn a_new_run_for_the_same_conversation_replaces_and_releases_the_previous_browser() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());

        let first_tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let released = Arc::downgrade(&first_tools);
        let first = registry.register("conv-1", Arc::clone(&first_tools)).await;
        first_tools.attach_agent_for_test().await;
        drop(first_tools);

        let (channel, mut rx) = capture_channel();
        first.subscribe("host-1", channel).await;
        first.emit_frame(test_frame(1)).await;
        let before = next_event(&mut rx).await.cursor;

        // A second run starts for the same conversation.
        let second_tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let second = registry.register("conv-1", Arc::clone(&second_tools)).await;
        second_tools.attach_agent_for_test().await;

        let found = registry.get(Some("conv-1")).await.expect("registered");
        assert!(
            Arc::ptr_eq(&found, &second),
            "the key now resolves to the new run's browser"
        );

        // The drawer came across without re-subscribing, and its cursor only
        // ever moved forward.
        assert_eq!(first.subscriber_count().await, 0);
        assert_eq!(second.subscriber_count().await, 1);
        second.emit_frame(test_frame(2)).await;
        let mut seen = next_event(&mut rx).await.cursor;
        // The handover itself may emit a presentation delta first; either
        // way, every cursor the client sees is strictly increasing.
        while seen <= before {
            seen = next_event(&mut rx).await.cursor;
        }
        assert!(
            seen > before,
            "cursor never goes backwards across a handover"
        );

        // And the previous run's browser is released once nothing points at
        // it — the whole reason replacement is the lifetime bound.
        drop(first);
        drop(found);
        for _ in 0..100 {
            if released.upgrade().is_none() {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert!(
            released.upgrade().is_none(),
            "the replaced run's browser must be released, not accumulated"
        );
    }

    #[tokio::test]
    async fn replacing_a_local_view_resolves_its_pending_attention() {
        use crate::assistant::browser_control::ControlEvent;

        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let recorder = Arc::new(crate::browser_attention::RecordingAttention::default());
        let first_tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        first_tools.set_signin_attention(recorder.clone(), Some("conv-1".to_string()));
        registry.register("conv-1", Arc::clone(&first_tools)).await;
        first_tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        assert_eq!(registry.pending_signins().await[0].message, "Sign in");

        registry
            .register("conv-1", Arc::new(BrowserTools::new(std::env::temp_dir())))
            .await;

        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED,
            ],
            "an unreachable predecessor cannot strand its badge"
        );
    }

    /// The local twin of `browser_relay`'s stalled-broadcast test.
    ///
    /// `apply_control` is the SINGLE mutator of the reducer, so every drawer
    /// input, Take control and Hand back runs through the notify path.
    /// `HostState::record_event` awaits each `host.subscribe` socket in turn,
    /// bounded at 10s apiece, so holding `announced` across it made N
    /// backpressured hosts an N x 10s stall on the very next keystroke — a
    /// call with nothing to announce.
    #[tokio::test]
    async fn a_stalled_signin_broadcast_does_not_block_the_next_drawer_input() {
        use crate::assistant::browser_control::ControlEvent;
        use crate::browser_attention::SignInAttention;

        struct BlockingAttention {
            entered: Arc<tokio::sync::Notify>,
            release: Arc<tokio::sync::Notify>,
        }

        #[async_trait::async_trait]
        impl SignInAttention for BlockingAttention {
            async fn signin_needed(&self, _conversation_id: Option<&str>, _message: &str) {
                self.entered.notify_one();
                self.release.notified().await;
            }
            async fn signin_resolved(&self, _conversation_id: Option<&str>) {}
        }

        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        tools.set_signin_attention(
            Arc::new(BlockingAttention {
                entered: Arc::clone(&entered),
                release: Arc::clone(&release),
            }),
            Some("conv-1".to_string()),
        );

        let blocked = tokio::spawn({
            let tools = Arc::clone(&tools);
            async move {
                tools
                    .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
                    .await;
            }
        });
        // The announcement is now in flight and wedged on a host socket.
        entered.notified().await;

        // A click in the drawer. Not a sign-in transition, so it must settle
        // without waiting on that broadcast.
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            tools.apply_control_for_test(ControlEvent::UserInput),
        )
        .await
        .expect("a drawer input must not queue behind a stalled sign-in broadcast");

        release.notify_one();
        blocked.await.unwrap();
    }

    /// `register` inserts the successor and only THEN calls `adopt`, so a
    /// `subscribe` for the same key can land on the successor inside that
    /// window — with a success reply and a cursor already in the client's
    /// hands. `adopt` used to assign the predecessor's map over the top,
    /// silently deregistering that client: it would receive nothing forever,
    /// and its cursor would never advance, so it could not even detect a gap
    /// and re-subscribe.
    #[tokio::test]
    async fn a_subscriber_that_lands_during_the_handover_window_survives_it() {
        let previous = test_view();
        let (channel_old, mut rx_old) = capture_channel();
        previous.subscribe("host-old", channel_old).await;

        // The successor exists (register inserted it) but has not adopted yet.
        let successor = test_view();
        let (channel_new, mut rx_new) = capture_channel();
        successor.subscribe("host-new", channel_new).await;

        successor.adopt(&previous).await;

        assert_eq!(
            successor.subscriber_count().await,
            2,
            "both the inherited and the concurrent subscriber are registered"
        );
        successor.emit_frame(test_frame(1)).await;
        // Drain to the frame: adopt emits a presentation delta first.
        let mut new_kinds = 0;
        loop {
            match next_event(&mut rx_new).await.payload {
                BrowserViewPayload::Frame { .. } => break,
                BrowserViewPayload::Presentation { .. } => {
                    new_kinds += 1;
                    assert!(new_kinds < 4, "expected a frame within a few events");
                }
            }
        }
        let mut old_kinds = 0;
        loop {
            match next_event(&mut rx_old).await.payload {
                BrowserViewPayload::Frame { .. } => break,
                BrowserViewPayload::Presentation { .. } => {
                    old_kinds += 1;
                    assert!(old_kinds < 4, "expected a frame within a few events");
                }
            }
        }
    }

    /// The exit the round-8 engaged-ending rule owes. A run ending under a
    /// person mid-sign-in leaves their strip and blackout up — and they never
    /// pressed Take control, so there is no holder for `note_disconnect` to
    /// match and it is inert for them. Without this the window had exactly one
    /// exit (hand-back), so a person who closed the laptop mid-sign-in left the
    /// blackout latched for the daemon's life, wedging every later run's browse
    /// call behind it.
    #[tokio::test]
    async fn a_watcher_leaving_mid_signin_starts_the_clock_that_ends_their_window() {
        use crate::assistant::browser_control::ControlEvent;

        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let view = registry.register("conv-1", Arc::clone(&tools)).await;
        let (channel, _rx) = capture_channel();
        view.subscribe_for_test("host-1", channel).await;

        tools
            .apply_control_for_test(ControlEvent::AgentAttached)
            .await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        // Typed into the credential form; never pressed Take control.
        tools.apply_control_for_test(ControlEvent::UserInput).await;
        view.note_run_ended().await;
        assert!(
            tools.control_status().await.blackout_active,
            "precondition: the engaged window survived the run ending"
        );
        assert!(
            view.control_holder_for_test().await.is_none(),
            "precondition: nothing holds control, so note_disconnect is inert here"
        );

        let before = view.grace_generation_for_test().await;
        registry.drop_subscriptions_for_client("host-1").await;
        assert!(
            view.grace_generation_for_test().await > before,
            "the watcher going away must arm the clock that settles their sign-in"
        );
    }

    /// The other side: a connection leaving a view with nothing person-facing
    /// open must not arm anything.
    #[tokio::test]
    async fn a_watcher_leaving_an_ordinary_agent_view_arms_nothing() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let view = registry.register("conv-1", Arc::clone(&tools)).await;
        let (channel, _rx) = capture_channel();
        view.subscribe_for_test("host-1", channel).await;
        tools.attach_agent_for_test().await;

        let before = view.grace_generation_for_test().await;
        registry.drop_subscriptions_for_client("host-1").await;
        assert_eq!(view.grace_generation_for_test().await, before);
    }

    /// The SAME client id on both sides of the handover, which the test above
    /// cannot reach: it uses `host-old`/`host-new`, so `or_insert` is always
    /// Vacant, nothing is dropped, and no epoch is ever compared.
    ///
    /// One connection is the real shape — the drawer re-subscribing its key on
    /// the same socket while a restarted process re-registers it. The inherited
    /// subscriber loses the `or_insert` race and is dropped, which ends its
    /// drain task, which deregisters by (client_id, epoch). With a per-VIEW
    /// epoch counter both sides start at 0, so it matched the LIVE
    /// registration and removed it: the drawer held a successful reply with a
    /// cursor that never advanced again, rendering a frozen frame as `.live`
    /// with no event arriving to trip the cursor-gap recovery.
    #[tokio::test]
    async fn one_connection_on_both_sides_of_a_handover_keeps_its_subscription() {
        let previous = test_view();
        let (channel_old, _rx_old) = capture_channel();
        previous.subscribe("host-1", channel_old).await;

        let successor = test_view();
        let (channel_new, mut rx_new) = capture_channel();
        successor.subscribe("host-1", channel_new).await;

        successor.adopt(&previous).await;

        // The inherited subscriber lost the `or_insert` race and was dropped,
        // so its drain task's `recv()` has already returned `None`. Give it
        // room to actually reach its deregistration — the count is 1 either
        // way the instant `adopt` returns (one HashMap key), so asserting
        // before the task runs would pass against the bug too.
        for _ in 0..50 {
            tokio::task::yield_now().await;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(
            successor.subscriber_count().await,
            1,
            "the live registration must survive the inherited one's deregistration"
        );

        // And it is still SERVED, not merely counted.
        successor.emit_frame(test_frame(1)).await;
        let mut seen = 0;
        loop {
            match next_event(&mut rx_new).await.payload {
                BrowserViewPayload::Frame { .. } => break,
                BrowserViewPayload::Presentation { .. } => {
                    seen += 1;
                    assert!(seen < 4, "expected a frame within a few events");
                }
            }
        }
    }

    /// The other half: a concurrent subscriber must get the STREAM started
    /// for it too. Keying that on "did anything come across" left a client
    /// registered with no streamer whenever the predecessor had none.
    #[tokio::test]
    async fn a_handover_from_a_view_nobody_watched_still_starts_the_stream() {
        let previous = test_view();
        let successor = test_view();
        let (channel, _rx) = capture_channel();
        successor.subscribe("host-new", channel).await;

        successor.adopt(&previous).await;

        assert_eq!(successor.subscriber_count().await, 1);
        let capture = successor.capture.lock().await;
        assert!(
            capture.active && capture.task.as_ref().is_some_and(|t| !t.is_finished()),
            "the concurrent subscriber's stream must be running"
        );
    }

    #[tokio::test]
    async fn a_disconnect_drops_only_that_connection_s_subscriptions() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry.standing().await;
        let (channel_a, _rx_a) = capture_channel();
        let (channel_b, mut rx_b) = capture_channel();
        view.subscribe("host-a", channel_a).await;
        view.subscribe("host-b", channel_b).await;

        registry.drop_subscriptions_for_client("host-a").await;
        assert_eq!(view.subscriber_count().await, 1);

        view.emit_frame(test_frame(9)).await;
        assert_eq!(next_event(&mut rx_b).await.cursor, 1);
    }

    #[tokio::test]
    async fn input_where_no_browser_exists_is_a_clean_error() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry.standing().await;
        let err = view.tools().user_click(10.0, 10.0).await.unwrap_err();
        assert!(err.contains("no browser is running"), "got: {err}");
    }

    #[tokio::test]
    async fn unknown_modifiers_are_rejected_rather_than_silently_dropped() {
        assert_eq!(parse_modifier("Shift").unwrap(), Modifier::Shift);
        assert_eq!(parse_modifier("cmd").unwrap(), Modifier::Meta);
        let err = parse_modifier("hyper").unwrap_err();
        assert!(err.contains("unknown modifier"), "got: {err}");
    }

    // ---- wire shapes ----------------------------------------------------

    #[test]
    fn the_event_wire_shape_is_tagged_and_flat() {
        let event = BrowserViewEvent {
            conversation_id: Some("conv-1".into()),
            cursor: 7,
            payload: BrowserViewPayload::Presentation {
                presentation: WirePresentation::empty(),
            },
        };
        let json = serde_json::to_value(&event).unwrap();
        assert_eq!(json["conversation_id"], "conv-1");
        assert_eq!(json["cursor"], 7);
        assert_eq!(json["kind"], "presentation");
        assert_eq!(json["presentation"]["owner"], "none");
        // Round trips, so a binding generated from this shape can decode it.
        let back: BrowserViewEvent = serde_json::from_value(json).unwrap();
        assert_eq!(back, event);
    }

    #[test]
    fn every_owner_state_has_its_own_wire_name() {
        for (owner, name) in [
            (ControlOwner::NoAgent, "none"),
            (ControlOwner::Agent, "agent"),
            (ControlOwner::User, "user"),
        ] {
            let wire: WireOwner = owner.into();
            assert_eq!(serde_json::to_value(wire).unwrap(), name);
        }
    }

    // ---- the dispatcher-facing handlers, end to end --------------------

    async fn host_session(
        state: &Arc<ServerState>,
        client_id: &str,
    ) -> (
        Arc<ClientSession>,
        futures::channel::mpsc::UnboundedReceiver<Message>,
    ) {
        let (channel, rx) = capture_channel();
        let session = state.create_session(client_id, channel).await.unwrap();
        session
            .is_host
            .store(true, std::sync::atomic::Ordering::Release);
        (session, rx)
    }

    fn request(method: &str, params: Value) -> JsonRpcMessage {
        JsonRpcMessage {
            jsonrpc: "2.0".to_string(),
            id: json!(1),
            method: Some(method.to_string()),
            params,
            result: None,
            error: None,
        }
    }

    async fn test_state() -> (Arc<ServerState>, tempfile::TempDir) {
        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::with_config(ServerStateConfig::new(
            temp.path().to_path_buf(),
        )));
        (state, temp)
    }

    #[tokio::test]
    async fn subscribe_answers_the_standing_session_with_a_snapshot_and_cursor() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        let out = handle_subscribe(
            &request("browser.view.subscribe", json!({})),
            &session,
            &state,
        )
        .await
        .expect("the standing session is always subscribable");
        assert_eq!(out["standing_session"], true);
        assert_eq!(out["conversation_id"], Value::Null);
        assert_eq!(out["cursor"], 0);
        assert_eq!(out["presentation"]["owner"], "none");
        assert_eq!(out["presentation"]["tabs"], json!([]));

        let out = handle_unsubscribe(
            &request("browser.view.unsubscribe", json!({})),
            &session,
            &state,
        )
        .await
        .unwrap();
        assert_eq!(out["removed"], true);
    }

    /// The ORDINARY teardown route, driven through the real RPC.
    ///
    /// `release_if_idle` was wired to the run-end guard and the disconnect
    /// sweep — but a disconnect is the exceptional route. Closing the drawer
    /// or switching conversations is this plain `browser.view.unsubscribe`
    /// over a live socket, and it released nothing: a finished run's browser
    /// stayed registered, and alive, for the daemon's lifetime, once per
    /// watched run. The documented replacement bound cannot save it either —
    /// the key carries a fresh uuid every run.
    #[tokio::test]
    async fn unsubscribing_the_last_watcher_releases_a_finished_run_s_browser() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        let weak = Arc::downgrade(&tools);
        let view = state
            .browser_views
            .register("mcp-run-abc", Arc::clone(&tools))
            .await;
        drop(tools);

        handle_subscribe(
            &request(
                "browser.view.subscribe",
                json!({ "conversation_id": "mcp-run-abc" }),
            ),
            &session,
            &state,
        )
        .await
        .expect("subscribed");
        view.note_run_ended().await;
        assert!(
            state.browser_views.get(Some("mcp-run-abc")).await.is_some(),
            "a watched view outlives its run, by ruling"
        );
        drop(view);

        // The user closes the drawer.
        handle_unsubscribe(
            &request(
                "browser.view.unsubscribe",
                json!({ "conversation_id": "mcp-run-abc" }),
            ),
            &session,
            &state,
        )
        .await
        .expect("unsubscribed");

        assert!(
            state.browser_views.get(Some("mcp-run-abc")).await.is_none(),
            "the ordinary close path must release a finished run's view"
        );
        assert!(weak.upgrade().is_none(), "and its browser with it");
    }

    #[tokio::test]
    async fn a_connection_that_is_not_the_host_client_is_refused_everywhere() {
        let (state, _temp) = test_state().await;
        let (channel, _rx) = capture_channel();
        let session = state.create_session("not-host", channel).await.unwrap();
        // No `is_host` — an ordinary authenticated connection.

        // EVERY method, not a sample of three. `authorize(session)` as the
        // first line of each handler is the single control that makes this
        // surface host-only, and a handler added later that forgets it is
        // exactly what this must catch — so the input arm is driven from the
        // `InputOp` enum itself, which cannot gain a variant silently.
        let mut results = vec![
            handle_subscribe(
                &request("browser.view.subscribe", json!({})),
                &session,
                &state,
            )
            .await,
            handle_unsubscribe(
                &request("browser.view.unsubscribe", json!({})),
                &session,
                &state,
            )
            .await,
            handle_take_control(
                &request("browser.view.take_control", json!({})),
                &session,
                &state,
            )
            .await,
            handle_hand_back(
                &request("browser.view.hand_back", json!({})),
                &session,
                &state,
            )
            .await,
        ];
        for op in [
            InputOp::Navigate,
            InputOp::Click,
            InputOp::Type,
            InputOp::Keypress,
            InputOp::Scroll,
            InputOp::Paste,
            InputOp::Back,
            InputOp::Forward,
            InputOp::Reload,
            InputOp::TabOpen,
            InputOp::TabClose,
            InputOp::TabSwitch,
        ] {
            results.push(
                handle_input(
                    op,
                    &request(
                        "browser.view.input",
                        json!({
                            "url": "https://x.test",
                            "x": 1.0, "y": 2.0,
                            "text": "x", "key": "Enter",
                            "delta_y": 1, "tab_id": "1",
                        }),
                    ),
                    &session,
                    &state,
                )
                .await,
            );
        }
        assert_eq!(results.len(), 16, "every dispatched method must be covered");
        for result in results {
            let err = result.unwrap_err();
            assert!(err.contains("not authorized"), "got: {err}");
        }
        // And nothing was created as a side effect of a refused call.
        assert_eq!(
            state
                .browser_views
                .standing()
                .await
                .subscriber_count()
                .await,
            0
        );
    }

    #[tokio::test]
    async fn an_unknown_conversation_is_a_clean_error_not_a_hang_or_an_empty_success() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        let err = handle_subscribe(
            &request(
                "browser.view.subscribe",
                json!({"conversation_id": "ghost"}),
            ),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("no browser view for conversation 'ghost'"),
            "got: {err}"
        );
        assert!(
            err.contains("standing session"),
            "and it says what to do instead"
        );
    }

    #[tokio::test]
    async fn input_is_refused_while_the_agent_drives_and_accepted_after_take_control() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        state
            .browser_views
            .register("conv-1", Arc::clone(&tools))
            .await;
        tools.attach_agent_for_test().await;

        let navigate = request(
            "browser.view.navigate",
            json!({ "conversation_id": "conv-1", "url": "https://x.test" }),
        );
        let err = handle_input(InputOp::Navigate, &navigate, &session, &state)
            .await
            .unwrap_err();
        assert!(err.contains("take_control"), "got: {err}");

        let out = handle_take_control(
            &request(
                "browser.view.take_control",
                json!({ "conversation_id": "conv-1" }),
            ),
            &session,
            &state,
        )
        .await
        .unwrap();
        assert_eq!(out["presentation"]["owner"], "user");

        // Now the control check passes and the call reaches the browser —
        // which does not exist in this test, so it fails there instead of
        // at the gate. That IS the assertion: a different, later error.
        let err = handle_input(InputOp::Navigate, &navigate, &session, &state)
            .await
            .unwrap_err();
        assert!(
            !err.contains("take_control") && !err.contains("holds control"),
            "the control gate must be satisfied now; got: {err}"
        );
    }

    #[tokio::test]
    async fn input_from_a_connection_that_does_not_hold_control_is_refused() {
        let (state, _temp) = test_state().await;
        let (holder, _rx_a) = host_session(&state, "host-holder").await;
        let (other, _rx_b) = host_session(&state, "host-other").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        state
            .browser_views
            .register("conv-1", Arc::clone(&tools))
            .await;
        tools.attach_agent_for_test().await;

        handle_take_control(
            &request(
                "browser.view.take_control",
                json!({ "conversation_id": "conv-1" }),
            ),
            &holder,
            &state,
        )
        .await
        .unwrap();

        let err = handle_input(
            InputOp::Click,
            &request(
                "browser.view.click",
                json!({ "conversation_id": "conv-1", "x": 1.0, "y": 2.0 }),
            ),
            &other,
            &state,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("another connection holds control"),
            "got: {err}"
        );

        // Hand back, and neither connection may drive — the agent has it.
        handle_hand_back(
            &request(
                "browser.view.hand_back",
                json!({ "conversation_id": "conv-1" }),
            ),
            &holder,
            &state,
        )
        .await
        .unwrap();
        let err = handle_input(
            InputOp::Click,
            &request(
                "browser.view.click",
                json!({ "conversation_id": "conv-1", "x": 1.0, "y": 2.0 }),
            ),
            &holder,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("take_control"), "got: {err}");
    }

    #[tokio::test]
    async fn malformed_input_params_are_rejected_before_touching_the_browser() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        for (op, params, needle) in [
            (InputOp::Navigate, json!({}), "requires { url }"),
            (InputOp::Click, json!({ "x": 1.0 }), "requires { x, y }"),
            (InputOp::Type, json!({}), "requires { text }"),
            (InputOp::Keypress, json!({}), "requires { key }"),
            (InputOp::Scroll, json!({}), "requires { delta_y }"),
            (InputOp::Paste, json!({}), "requires { text }"),
            (InputOp::TabClose, json!({}), "requires { tab_id }"),
            (InputOp::TabSwitch, json!({}), "requires { tab_id }"),
        ] {
            let err = handle_input(op, &request("browser.view.x", params), &session, &state)
                .await
                .unwrap_err();
            assert!(err.contains(needle), "{op:?}: got {err}");
        }
    }

    // ---- the nav bar's history buttons ---------------------------------

    /// Back/Forward/Reload are input like any other: refused while the agent
    /// drives, accepted from the control holder. The gate lives in
    /// `ViewInput::apply` BEFORE the op match, so this holds for every op by
    /// construction — this pins that it actually does for the new three.
    #[tokio::test]
    async fn history_ops_are_refused_while_the_agent_drives() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        state
            .browser_views
            .register("conv-1", Arc::clone(&tools))
            .await;
        tools.attach_agent_for_test().await;

        for op in [InputOp::Back, InputOp::Forward, InputOp::Reload] {
            let err = handle_input(
                op,
                &request("browser.view.x", json!({ "conversation_id": "conv-1" })),
                &session,
                &state,
            )
            .await
            .unwrap_err();
            assert!(err.contains("take_control"), "{op:?}: got {err}");
        }
    }

    /// Paste is input like any other — the same control gate, refused while
    /// the agent drives. It carries the TEXT because CDP key events cannot
    /// reach a clipboard, so the host reads its own pasteboard and sends the
    /// string; that means the surface must gate it exactly like typing.
    #[tokio::test]
    async fn paste_is_refused_while_the_agent_drives() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        state
            .browser_views
            .register("conv-1", Arc::clone(&tools))
            .await;
        tools.attach_agent_for_test().await;

        let err = handle_input(
            InputOp::Paste,
            &request(
                "browser.view.paste",
                json!({ "conversation_id": "conv-1", "text": "secret" }),
            ),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("take_control"), "got: {err}");
    }

    #[tokio::test]
    async fn paste_on_an_empty_state_reports_that_there_is_no_browser() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        let err = handle_input(
            InputOp::Paste,
            &request("browser.view.paste", json!({ "text": "hello" })),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("no browser is running"), "got: {err}");
    }

    #[tokio::test]
    async fn history_ops_are_refused_from_a_connection_that_does_not_hold_control() {
        let (state, _temp) = test_state().await;
        let (holder, _rx_a) = host_session(&state, "host-holder").await;
        let (other, _rx_b) = host_session(&state, "host-other").await;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        state
            .browser_views
            .register("conv-1", Arc::clone(&tools))
            .await;
        tools.attach_agent_for_test().await;
        handle_take_control(
            &request(
                "browser.view.take_control",
                json!({ "conversation_id": "conv-1" }),
            ),
            &holder,
            &state,
        )
        .await
        .unwrap();

        for op in [InputOp::Back, InputOp::Forward, InputOp::Reload] {
            let err = handle_input(
                op,
                &request("browser.view.x", json!({ "conversation_id": "conv-1" })),
                &other,
                &state,
            )
            .await
            .unwrap_err();
            assert!(
                err.contains("another connection holds control"),
                "{op:?}: got {err}"
            );
        }
    }

    /// Once the control gate is satisfied they reach the browser — which does
    /// not exist here, so they land on the empty-state error rather than the
    /// gate's. Reload on no page is exactly the outcomes file's "inactive on
    /// the empty state", answered as a clean error rather than a hang.
    #[tokio::test]
    async fn history_ops_on_an_empty_state_report_that_there_is_no_browser() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        for op in [InputOp::Back, InputOp::Forward, InputOp::Reload] {
            let err = handle_input(op, &request("browser.view.x", json!({})), &session, &state)
                .await
                .unwrap_err();
            assert!(err.contains("no browser is running"), "{op:?}: got {err}");
        }
    }

    /// They take no params of their own — which page they act on is the
    /// active tab's own history — so a bare call is valid, unlike every other
    /// input op except `tab_open`.
    #[tokio::test]
    async fn history_ops_need_no_params_of_their_own() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        for op in [InputOp::Back, InputOp::Forward, InputOp::Reload] {
            let err = handle_input(
                op,
                &request("browser.view.x", Value::Null),
                &session,
                &state,
            )
            .await
            .unwrap_err();
            assert!(
                !err.contains("requires"),
                "{op:?} must not demand params; got {err}"
            );
        }
    }

    #[tokio::test]
    async fn input_where_no_browser_exists_reports_that_rather_than_succeeding_emptily() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;

        let err = handle_input(
            InputOp::Click,
            &request("browser.view.click", json!({ "x": 1.0, "y": 2.0 })),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("no browser is running"), "got: {err}");
    }

    #[tokio::test]
    async fn disconnect_cleanup_drops_that_connection_s_subscription() {
        let (state, _temp) = test_state().await;
        let (session, _rx) = host_session(&state, "host-1").await;
        handle_subscribe(
            &request("browser.view.subscribe", json!({})),
            &session,
            &state,
        )
        .await
        .unwrap();
        assert_eq!(
            state
                .browser_views
                .standing()
                .await
                .subscriber_count()
                .await,
            1
        );

        state.remove_session("host-1").await;
        assert_eq!(
            state
                .browser_views
                .standing()
                .await
                .subscriber_count()
                .await,
            0
        );
    }

    fn test_frame(byte: u8) -> ScreencastFrame {
        ScreencastFrame {
            jpeg: vec![byte].into(),
            viewport: car_browser::Viewport {
                width: 1920,
                height: 1080,
                device_pixel_ratio: 1.0,
            },
            captured_at: 0.5,
        }
    }
}