car_server_core/

browser_relay.rs

//! The agent→daemon browser hop: making a SUPERVISED AGENT PROCESS's browser a
//! first-class producer for the daemon's [`crate::browser_view`] surface.
//!
//! ## Why this exists
//!
//! `browser.view.*` serves browsers the daemon can touch. The Command Deck's
//! CAR Chat does not go through one: `agents.chat` hands the turn to a
//! supervised `car do --serve` process, which builds its OWN
//! [`crate::assistant::browser_tools::BrowserTools`] in that process. Without
//! this module the drawer cannot show the browser the flagship agent actually
//! uses.
//!
//! ## The shape
//!
//! Nothing new is invented for transport. The supervised process already holds
//! a persistent WebSocket session with the daemon (`session.auth { token,
//! agent_id }`), and that session is already bidirectional — it is how
//! `agents.chat` reaches the process (`agent.chat` reverse request → ack →
//! `agent.chat.event` notifications back). This module reuses exactly that
//! machinery:
//!
//! ```text
//!   agent process                              daemon
//!   ─────────────                              ──────
//!   browser.producer.register  ──── call ────▶  RelayProducer + BrowserView
//!   browser.producer.presentation ─ notify ──▶  view.refresh_presentation()
//!   browser.producer.frame ──────── notify ──▶  view.emit_wire_frame()
//!   BrowserTools (+ Task 3 reducer) ◀── call ── agent.browser.input
//!                                  ◀── call ── agent.browser.control
//!                                  ◀── call ── agent.browser.capture
//! ```
//!
//! The daemon is a RELAY, not a second source of truth: the reducer and the
//! browser both live in the agent process, and Task 4's fanout (bounded
//! channels, cursors, snapshots, host-only authorization) stays the single
//! host-facing surface. A [`RelayProducer`] is indistinguishable from a
//! daemon-owned browser above [`crate::browser_view::ViewBrowser`].
//!
//! ## Authorization
//!
//! `browser.producer.*` is the AGENT side and requires an agent session — the
//! `session.auth { token, agent_id }` binding the supervisor's token
//! establishes. It grants an agent exactly one thing: publishing ITS OWN
//! browser. It cannot subscribe, it cannot read another view, and nothing it
//! pushes ever comes back to it — so no agent gains perception or actuation it
//! did not already have through its `full_access` `browse_*` tools. The
//! host-only rule on `browser.view.*` is untouched.
//!
//! A conversation key may only be claimed by the agent that is actually
//! serving that chat session (checked against `ServerState::chat_sessions`), so
//! one agent cannot hijack another's drawer.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Weak};
use std::time::Duration;

use car_browser::Modifier;
use futures::SinkExt;
use serde_json::{json, Value};
use tokio::sync::{watch, Mutex, MutexGuard};
use tokio_tungstenite::tungstenite::Message;

use crate::assistant::browser_control::{ControlEffect, ControlOwner};
use crate::assistant::browser_tools::ControlStatus;
use crate::browser_attention::{notify_signin_transition, BrowserSignInSnapshot, SignInAttention};
use crate::browser_view::{BrowserView, ViewControl, ViewInput, WireFrame, WirePresentation};
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState, WsChannel};

/// How long the daemon waits for a supervised process to answer a relayed
/// call. Generous enough for a navigation that has to launch Chromium, short
/// enough that a wedged process surfaces as an error instead of a hung drawer.
pub const RELAY_CALL_TIMEOUT: Duration = Duration::from_secs(30);

/// How long the capture pump waits before re-sending a request that did not
/// reach the agent process. Short: the drawer is frameless until it lands, and
/// the call itself is already bounded by [`RELAY_CALL_TIMEOUT`].
const CAPTURE_RETRY_BACKOFF: Duration = Duration::from_secs(2);

/// How many conversation views one supervised process keeps.
///
/// `register_relay`'s doc calls a view "one per conversation key", and it is —
/// but `agents.chat`'s `session_id` is minted FRESH PER TURN (`ChatTabView.send`,
/// and `ChatBrowserBindingTracker`'s own header says so), so in practice that
/// was one view per turn, forever: a long-lived `car do --serve` process
/// accumulated a `BrowserView`, a registry entry and a binding for every turn it
/// had ever served, and every screencast frame was cloned into all of them.
/// `note_producer_disconnected` closed exactly this leak for the process-DEATH
/// path; the per-turn key means the live-process path is the one that fires.
/// A registration now retires the older ones — replacement, at the recency the
/// two ends already agree on.
///
/// The number is [`MAX_KNOWN_CONVERSATIONS`](crate::assistant::browser_producer::MAX_KNOWN_CONVERSATIONS), not 1, and that is load-bearing:
/// it is exactly the set the agent side re-publishes (`BrowserProducer::known`,
/// same cap, same reasoning), so retiring anything newer would be churn the
/// 10-second republish sweep immediately undoes. Anything older it will never
/// re-register and the drawer's candidate chain — which reaches back exactly one
/// turn — can never ask for.
///
/// Retiring is not deleting: `release_if_idle` refuses to drop a view anybody is
/// subscribed to, so a drawer still watching an older key keeps it, and the
/// eviction happens when that drawer unsubscribes.
pub const MAX_VIEWS_PER_PRODUCER: usize =
    crate::assistant::browser_producer::MAX_KNOWN_CONVERSATIONS;

/// Largest `browser.producer.frame` payload the daemon will fan out, in
/// base64 characters.
///
/// A registered producer already runs arbitrary code on this box, so this is
/// not a privilege boundary — but the daemon is the SHARED component and must
/// not be OOM-able by one agent's bug, and the only bound underneath is
/// tungstenite's 64 MiB message cap. A 1920x1080 quality-60 JPEG is a few
/// hundred KB and base64 adds a third, so 8 MiB is generous by an order of
/// magnitude for anything the pump can legitimately produce.
const MAX_PRODUCER_FRAME_BYTES: usize = 8 * 1024 * 1024;

// ---------------------------------------------------------------------------
// The codec — ONE definition, used by both ends
// ---------------------------------------------------------------------------
//
// The daemon encodes and the agent process decodes (and vice versa for the
// results), and both ends live in this crate. Writing the codec once, here,
// is what keeps them from drifting: a new `ViewInput` variant or a new
// `ControlEffect` is a compile error in the exhaustive matches below rather
// than a shape one side silently fails to parse.

/// Modifier → its wire name. Exhaustive: a new modifier must not silently
/// serialize as an existing one.
pub fn modifier_name(modifier: Modifier) -> &'static str {
    match modifier {
        Modifier::Shift => "shift",
        Modifier::Control => "control",
        Modifier::Alt => "alt",
        Modifier::Meta => "meta",
    }
}

/// One user input, as `agent.browser.input` carries it. The shape deliberately
/// mirrors the `browser.view.*` input params a host sends, so the same field
/// names mean the same thing end to end.
pub fn input_to_wire(input: &ViewInput) -> Value {
    match input {
        ViewInput::Navigate { url } => json!({ "op": "navigate", "url": url }),
        ViewInput::Click { x, y } => json!({ "op": "click", "x": x, "y": y }),
        ViewInput::Type { text } => json!({ "op": "type", "text": text }),
        ViewInput::Keypress { key, modifiers } => json!({
            "op": "keypress",
            "key": key,
            "modifiers": modifiers.iter().map(|m| modifier_name(*m)).collect::<Vec<_>>(),
        }),
        ViewInput::Scroll { delta_y } => json!({ "op": "scroll", "delta_y": delta_y }),
        ViewInput::Paste { text } => json!({ "op": "paste", "text": text }),
        ViewInput::Back => json!({ "op": "back" }),
        ViewInput::Forward => json!({ "op": "forward" }),
        ViewInput::Reload => json!({ "op": "reload" }),
        ViewInput::TabOpen => json!({ "op": "tab_open" }),
        ViewInput::TabClose { tab_id } => json!({ "op": "tab_close", "tab_id": tab_id }),
        ViewInput::TabSwitch { tab_id } => json!({ "op": "tab_switch", "tab_id": tab_id }),
    }
}

/// The agent process's half of [`input_to_wire`].
pub fn input_from_wire(params: &Value) -> Result<ViewInput, String> {
    let op = params
        .get("op")
        .and_then(Value::as_str)
        .ok_or("agent.browser.input requires { op }")?;
    let string = |field: &str| -> Result<String, String> {
        params
            .get(field)
            .and_then(Value::as_str)
            .map(str::to_string)
            .ok_or_else(|| format!("agent.browser.input `{op}` requires {{ {field} }}"))
    };
    match op {
        "navigate" => Ok(ViewInput::Navigate {
            url: string("url")?,
        }),
        "click" => {
            let (x, y) = match (
                params.get("x").and_then(Value::as_f64),
                params.get("y").and_then(Value::as_f64),
            ) {
                (Some(x), Some(y)) => (x, y),
                _ => return Err("agent.browser.input `click` requires { x, y }".to_string()),
            };
            Ok(ViewInput::Click { x, y })
        }
        "type" => Ok(ViewInput::Type {
            text: string("text")?,
        }),
        "keypress" => {
            let mut modifiers = Vec::new();
            for name in params
                .get("modifiers")
                .and_then(Value::as_array)
                .unwrap_or(&Vec::new())
            {
                let name = name
                    .as_str()
                    .ok_or("agent.browser.input `keypress` modifiers must be strings")?;
                modifiers.push(crate::browser_view::parse_modifier(name)?);
            }
            Ok(ViewInput::Keypress {
                key: string("key")?,
                modifiers,
            })
        }
        "scroll" => Ok(ViewInput::Scroll {
            delta_y: params
                .get("delta_y")
                .and_then(Value::as_i64)
                .and_then(|n| i32::try_from(n).ok())
                .ok_or("agent.browser.input `scroll` requires { delta_y }")?,
        }),
        "paste" => Ok(ViewInput::Paste {
            text: string("text")?,
        }),
        "back" => Ok(ViewInput::Back),
        "forward" => Ok(ViewInput::Forward),
        "reload" => Ok(ViewInput::Reload),
        "tab_open" => Ok(ViewInput::TabOpen),
        "tab_close" => Ok(ViewInput::TabClose {
            tab_id: string("tab_id")?,
        }),
        "tab_switch" => Ok(ViewInput::TabSwitch {
            tab_id: string("tab_id")?,
        }),
        other => Err(format!("unknown agent.browser.input op '{other}'")),
    }
}

/// One control transition, as `agent.browser.control` carries it.
pub fn control_to_wire(control: ViewControl) -> &'static str {
    match control {
        ViewControl::TakeControl => "take_control",
        ViewControl::HandBack => "hand_back",
        ViewControl::RunEnded => "run_ended",
        ViewControl::HolderDisconnected => "holder_disconnected",
        ViewControl::GraceExpired => "grace_expired",
    }
}

/// The agent process's half of [`control_to_wire`].
pub fn control_from_wire(action: &str) -> Result<ViewControl, String> {
    match action {
        "take_control" => Ok(ViewControl::TakeControl),
        "hand_back" => Ok(ViewControl::HandBack),
        "run_ended" => Ok(ViewControl::RunEnded),
        "holder_disconnected" => Ok(ViewControl::HolderDisconnected),
        "grace_expired" => Ok(ViewControl::GraceExpired),
        other => Err(format!(
            "unknown agent.browser.control action '{other}' — use take_control, hand_back, \
             run_ended, holder_disconnected or grace_expired"
        )),
    }
}

/// What the reducer asked the CALLER to do. The reducer runs in the agent
/// process; the effects have to cross back because the daemon owns the clock
/// the grace period runs on.
pub fn effects_to_wire(effects: &[ControlEffect]) -> Value {
    Value::Array(
        effects
            .iter()
            .map(|effect| match effect {
                ControlEffect::StartGracePeriod => json!({ "effect": "start_grace_period" }),
                ControlEffect::SignInResolved { signed_in } => json!({
                    "effect": "sign_in_resolved",
                    "signed_in": signed_in,
                }),
            })
            .collect(),
    )
}

/// The daemon's half of [`effects_to_wire`]. An effect it does not recognize is
/// dropped rather than failing the call: a newer agent process talking to an
/// older daemon must still be able to hand control back.
pub fn effects_from_wire(value: &Value) -> Vec<ControlEffect> {
    let Some(items) = value.as_array() else {
        return Vec::new();
    };
    items
        .iter()
        .filter_map(|item| match item.get("effect").and_then(Value::as_str) {
            Some("start_grace_period") => Some(ControlEffect::StartGracePeriod),
            Some("sign_in_resolved") => Some(ControlEffect::SignInResolved {
                signed_in: item
                    .get("signed_in")
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
            }),
            _ => {
                tracing::debug!(effect = ?item, "browser relay: ignoring an unknown control effect");
                None
            }
        })
        .collect()
}

// ---------------------------------------------------------------------------
// The reverse call
// ---------------------------------------------------------------------------

/// Send one JSON-RPC request DOWN the agent's own session and await its reply.
///
/// Same machinery `agents.chat` uses for `agent.chat`: a string request id, a
/// oneshot parked in [`WsChannel::pending`], and the dispatcher's response
/// demuxer routing the agent's reply back. Nothing about the transport is new.
async fn call_agent(
    channel: &Arc<WsChannel>,
    method: &str,
    params: Value,
) -> Result<Value, String> {
    let request_id = channel.next_request_id();
    let (tx, rx) = tokio::sync::oneshot::channel();
    channel.pending.lock().await.insert(request_id.clone(), tx);

    let frame = json!({
        "jsonrpc": "2.0",
        "method": method,
        "params": params,
        "id": request_id,
    });
    let text = match serde_json::to_string(&frame) {
        Ok(text) => text,
        Err(e) => {
            channel.pending.lock().await.remove(&request_id);
            return Err(format!("serialize {method}: {e}"));
        }
    };
    // Bounded write, and the timeout wraps the LOCK as well as the send.
    //
    // `channel.write` is shared with `agents.chat`'s own reverse call and with
    // every response that connection sends, so parking here on a full TCP
    // buffer parks all of them — up to `handler_default_deadline_secs` (1800s)
    // of wedged chat, and the detached `broadcast_host_connected` spawns
    // outlive even that. There is no transport-level timeout underneath:
    // `accept_async` installs no `WebSocketConfig` and tokio-tungstenite adds
    // none. Same shape, and the same reason, as `handler.rs`'s keepalive ping
    // and its `tools.stream.event` forwarder; the agent side of this very
    // feature already bounds its half (`FRAME_PUSH_TIMEOUT`).
    let sent = tokio::time::timeout(RELAY_CALL_TIMEOUT, async {
        channel
            .write
            .lock()
            .await
            .send(Message::Text(text.into()))
            .await
    })
    .await;
    match sent {
        Ok(Ok(())) => {}
        Ok(Err(e)) => {
            channel.pending.lock().await.remove(&request_id);
            return Err(format!(
                "the agent process serving this browser is unreachable: {e}"
            ));
        }
        Err(_) => {
            channel.pending.lock().await.remove(&request_id);
            return Err(format!(
                "the agent process serving this browser is unreachable: its connection did not \
                 accept `{method}` within {}s",
                RELAY_CALL_TIMEOUT.as_secs()
            ));
        }
    }

    match tokio::time::timeout(RELAY_CALL_TIMEOUT, rx).await {
        Ok(Ok(response)) => match (response.error, response.output) {
            (Some(error), _) => Err(error),
            (None, Some(output)) => Ok(output),
            (None, None) => Ok(Value::Null),
        },
        Ok(Err(_)) => {
            Err("the agent process serving this browser disconnected before answering".to_string())
        }
        Err(_) => {
            channel.pending.lock().await.remove(&request_id);
            Err(format!(
                "the agent process serving this browser did not answer `{method}` within {}s",
                RELAY_CALL_TIMEOUT.as_secs()
            ))
        }
    }
}

/// What every call on a producer whose process has gone away answers.
pub const PRODUCER_GONE: &str =
    "the agent process that owns this browser has disconnected — its browser is gone";

// ---------------------------------------------------------------------------
// The producer
// ---------------------------------------------------------------------------

/// One supervised agent process's browser, as the daemon sees it.
///
/// A producer is per PROCESS (per agent WS connection), not per conversation:
/// `car do --serve` builds one `AssistantRuntime` and multiplexes every chat
/// session through it, so the process has exactly one browser. It may
/// therefore back several views — one per conversation key it registered —
/// and every push fans out to all of them.
pub struct RelayProducer {
    /// The agent connection's client id. This is the producer's IDENTITY: a
    /// reconnecting process is a different producer, which is what makes
    /// re-registration a clean replacement rather than a resurrection.
    client_id: String,
    /// The agent this process serves, from `session.auth { agent_id }`.
    agent_id: String,
    channel: Arc<WsChannel>,
    /// The last presentation the process pushed. Served to every view with no
    /// round trip, exactly like `BrowserTools::control_status` is the cheap
    /// read on the local path.
    last: Mutex<WirePresentation>,
    /// Operator attention belongs to the PROCESS, just like `last`.
    ///
    /// One supervised process owns one browser but can back up to eight
    /// per-turn views. Keeping this state on a view made one presentation
    /// push announce once per view. The conversation id is the newest view
    /// key; moving it while a wait is pending resolves the old key before
    /// announcing the new one so host state never contains both.
    signin_attention: Mutex<RelaySignInAttention>,
    /// The views this producer backs, one per conversation key.
    views: Mutex<Vec<Weak<BrowserView>>>,
    alive: AtomicBool,
    /// How many views currently want frames, so a browser nobody is watching
    /// never pays for a screencast or the WS traffic.
    ///
    /// A `std::sync::Mutex` rather than an atomic because the count and the
    /// `capture` signal it drives MUST move together — see
    /// [`Self::set_watchers`] for the interleaving that a separate atomic and
    /// send allowed.
    watchers: std::sync::Mutex<usize>,
    capture: watch::Sender<bool>,
    /// The newest host-connectivity transition, and the serializer that keeps
    /// the wire order equal to the transition order.
    ///
    /// `broadcast_host_connected` fires one detached task per producer per
    /// transition, and each one is a `RELAY_CALL_TIMEOUT`-bounded round trip —
    /// so a host flap (disconnect, immediate reconnect) had two tasks racing
    /// and whichever finished last decided what the process believed. The
    /// process caches that answer, and it decides whether a browser launches
    /// headless and whether `browser_await_signin` points the person at the
    /// drawer or at an app that is not running.
    ///
    /// `host_desired` is read UNDER `host_push`, so a waiter always sends the
    /// newest value, and `host_push`'s payload is the last value actually
    /// delivered so a superseded transition collapses instead of being re-sent.
    host_desired: AtomicBool,
    host_push: Mutex<Option<bool>>,

    /// Serializes the sign-in announcements themselves, so `signin_attention`
    /// never has to be.
    ///
    /// Taken while the state lock is STILL held and released only after the
    /// broadcast, which is what keeps two concurrent transitions — a hand-back
    /// racing the sign-in tool's own request — reaching the host in the order
    /// their decisions landed. Without that order a resolution can overtake
    /// the request that preceded it and leave a badge asserting a wait that
    /// already ended. What it buys is that every NON-transition caller (every
    /// relayed drawer input, every presentation republish) settles its
    /// compare-and-return under `signin_attention` and never waits on a host
    /// socket at all.
    ///
    /// **A known residual, deliberately kept.** A second concurrent transition
    /// still holds the state lock while it queues here, so a third caller can
    /// block on that state lock for the length of the first broadcast. Every
    /// way of removing that — take a ticket under the state lock, wait for
    /// your turn after releasing it — trades a bounded stall for an unbounded
    /// hazard, because a task cancelled between taking the ticket and taking
    /// its turn either wedges every later announcement for this producer (if
    /// the queue only advances in turn) or breaks the ordering the queue
    /// exists to provide (if it always advances on drop), and a WS session
    /// task being dropped is exactly the cancellation this code lives with.
    /// The one cheap ordered variant — first-polling the `lock()` future under
    /// the state lock — depends on tokio enqueuing a semaphore waiter on first
    /// poll, which is an implementation detail and not a documented contract.
    /// A FIFO mutex is correct under cancellation by construction: the guard
    /// drops, the next waiter proceeds, order holds. Do not "fix" this back.
    announce_order: Mutex<()>,
}

/// One decided sign-in transition, waiting to be told to the operator.
///
/// The point of the type is the split it forces: everything needed to make
/// the announcement is COPIED OUT under the state lock, so the broadcast that
/// follows — `HostState::record_event`, which awaits every `host.subscribe`
/// socket in turn at up to 10s each — happens with that lock released. Held
/// across the broadcast, it stalled `push_presentation` and every relayed
/// drawer input behind N backpressured host sockets.
struct PendingSignInAnnouncement {
    attention: Arc<dyn SignInAttention>,
    conversation_id: Option<String>,
    before: Option<String>,
    after: Option<String>,
}

#[derive(Default)]
struct RelaySignInAttention {
    attention: Option<Arc<dyn SignInAttention>>,
    /// Route that owns the currently-announced wait.
    conversation_id: Option<String>,
    /// Most recently registered route, adopted after the current wait ends.
    latest_conversation_id: Option<String>,
    announced: Option<String>,
}

impl RelayProducer {
    pub fn new(client_id: String, agent_id: String, channel: Arc<WsChannel>) -> Arc<Self> {
        let (capture, rx) = watch::channel(false);
        let producer = Arc::new(Self {
            client_id,
            agent_id,
            channel,
            last: Mutex::new(WirePresentation::empty()),
            signin_attention: Mutex::new(RelaySignInAttention::default()),
            views: Mutex::new(Vec::new()),
            alive: AtomicBool::new(true),
            watchers: std::sync::Mutex::new(0),
            capture,
            host_desired: AtomicBool::new(false),
            host_push: Mutex::new(None),
            announce_order: Mutex::new(()),
        });
        producer.spawn_capture_pump(rx);
        producer
    }

    pub fn client_id(&self) -> &str {
        &self.client_id
    }

    pub fn agent_id(&self) -> &str {
        &self.agent_id
    }

    pub fn is_alive(&self) -> bool {
        self.alive.load(Ordering::Acquire)
    }

    /// The cached presentation — what `browser.view.subscribe` snapshots and
    /// what `require_control` reads.
    pub async fn presentation(&self) -> WirePresentation {
        self.last.lock().await.clone()
    }

    /// Install (or move) the one process-level attention route.
    ///
    /// A wait stays pinned to the conversation that raised it. Another turn
    /// registering while the process is blocked must not move the badge to
    /// that unrelated chat; its route becomes eligible only after resolution.
    pub async fn set_signin_attention(
        &self,
        attention: Option<Arc<dyn SignInAttention>>,
        conversation_id: Option<String>,
    ) {
        let mut binding = self.signin_attention.lock().await;
        binding.latest_conversation_id = conversation_id.clone();
        if binding.announced.is_none() {
            binding.conversation_id = conversation_id;
        }
        binding.attention = attention;
        let pending = self.decide_signin_transition(&mut binding).await;
        self.announce(binding, Vec::from_iter(pending)).await;
    }

    /// Resolve the route `conversation_id` owns, and drop the sink only if
    /// this view was the LAST one this producer serves.
    ///
    /// Three things are being separated here.
    ///
    /// Retiring an older view after the route already moved to a newer one
    /// must not clear the newer badge — that is the key check at the top.
    ///
    /// One supervised process backs up to [`MAX_VIEWS_PER_PRODUCER`] views,
    /// so nulling `attention` because the view that happened to own the route
    /// retired left the producer with no sink while it was still serving the
    /// others: a sign-in raised on a surviving view then fell straight through
    /// [`Self::decide_signin_transition`]'s `attention` guard and told nobody.
    /// When a survivor exists the route MOVES instead — to the newest one,
    /// because that is the turn the operator is actually looking at. `views`
    /// is in registration order (see [`Self::views_past_the_cap`]), so the
    /// newest is the LAST match; taking the first handed the banner and the
    /// badge to the oldest surviving turn and opened a stale conversation.
    ///
    /// And the wait itself does not end just because the view reporting it
    /// retired. The resolve below is true of the ROUTE, not of the browser,
    /// so when the route moves the wait is immediately re-raised against its
    /// new owner — otherwise `announced` sits at `None` against a
    /// `pending_signin` that is still `Some`, and since a browser parked on a
    /// static login page publishes no further presentation, nothing would
    /// re-announce it. The operator sees the badge move rather than vanish.
    pub async fn detach_signin_attention(&self, conversation_id: Option<&str>) {
        let mut binding = self.signin_attention.lock().await;
        if binding.conversation_id.as_deref() != conversation_id {
            return;
        }
        // Read the survivors UNDER this lock. Read before taking it, a
        // `register_relay` landing in the window was invisible here — it
        // attaches its view and then leaves `conversation_id` pinned to the
        // announced wait — and the sink was nulled with that new view live,
        // which is the exact state this method exists to prevent. The nesting
        // is safe and one-directional: `attach_view`, `views_past_the_cap`
        // and `live_views` are the only holders of `views`, and none of them
        // takes `signin_attention`.
        let live = self.live_view_keys().await;
        let retiring_was_latest = binding.latest_conversation_id.as_deref() == conversation_id;
        let latest_is_live = live
            .iter()
            .any(|key| key.as_deref() == binding.latest_conversation_id.as_deref());
        let successor = if !retiring_was_latest && latest_is_live {
            // The newest registration is where the operator should be sent,
            // and it is still live. It is not necessarily the last element of
            // `live`: a wait pins `conversation_id` while later turns move
            // `latest_conversation_id` on ahead of it.
            Some(binding.latest_conversation_id.clone())
        } else {
            live.into_iter()
                .rev()
                .find(|key| key.as_deref() != conversation_id)
        };

        let mut pending = Vec::new();
        if let Some(before) = binding.announced.take() {
            if let Some(attention) = binding.attention.as_ref() {
                pending.push(PendingSignInAnnouncement {
                    attention: Arc::clone(attention),
                    conversation_id: conversation_id.map(str::to_string),
                    before: Some(before),
                    after: None,
                });
            }
        }

        match successor {
            // The process is still serving somebody, so the browser — and the
            // sink that reports it — outlive this view.
            Some(key) => {
                // `latest_conversation_id` is the NEWEST registration, which
                // is still correct unless the view retiring is that
                // registration. Overwriting it with the successor discarded
                // the newest route and sent the next wait to a stale turn.
                // It is ALSO wrong to keep when it names a view that is no
                // longer live: another process can claim that key, and the
                // adopt path's detach early-returns while the route is pinned
                // by an announced wait, so nothing else rewrites it. Left
                // stale, `decide_signin_transition` adopts it as the route on
                // resolve and the next wait announces a conversation the
                // operator cannot reach.
                if retiring_was_latest || !latest_is_live {
                    binding.latest_conversation_id = key.clone();
                }
                binding.conversation_id = key;
                pending.extend(self.decide_signin_transition(&mut binding).await);
            }
            None => {
                binding.attention = None;
                binding.conversation_id = None;
                binding.latest_conversation_id = None;
            }
        }
        self.announce(binding, pending).await;
    }

    pub async fn signin_snapshot(&self) -> Option<BrowserSignInSnapshot> {
        let binding = self.signin_attention.lock().await;
        binding.announced.as_ref().map(|message| {
            BrowserSignInSnapshot::new(binding.conversation_id.as_deref(), message.clone())
        })
    }

    async fn sync_signin_attention(&self) {
        let mut binding = self.signin_attention.lock().await;
        let pending = self.decide_signin_transition(&mut binding).await;
        self.announce(binding, Vec::from_iter(pending)).await;
    }

    /// Compare the cached presentation against what the operator was last
    /// told and record the answer, WITHOUT telling anyone.
    ///
    /// The decision and the state write stay atomic under the caller's held
    /// lock — that is what makes "one browser, one notification" hold when
    /// several per-turn views push the same presentation. Only the broadcast
    /// moves out, to [`Self::announce`].
    async fn decide_signin_transition(
        &self,
        binding: &mut RelaySignInAttention,
    ) -> Option<PendingSignInAnnouncement> {
        let attention = Arc::clone(binding.attention.as_ref()?);
        let current = self.last.lock().await.pending_signin.clone();
        if binding.announced == current {
            return None;
        }
        let before = std::mem::replace(&mut binding.announced, current.clone());
        let conversation_id = binding.conversation_id.clone();
        if current.is_none() {
            binding.conversation_id = binding.latest_conversation_id.clone();
        }
        Some(PendingSignInAnnouncement {
            attention,
            conversation_id,
            before,
            after: current,
        })
    }

    /// Broadcast decided transitions with the state lock released — see
    /// [`Self::announce_order`] for why the order lock is taken first and the
    /// state lock dropped second, and never the other way round.
    ///
    /// Takes a sequence rather than one announcement because a route move
    /// resolves the old owner and re-raises against the new one, and those two
    /// must reach the operator in that order with nothing interleaved between
    /// them — so they share a single hold of the order lock.
    async fn announce(
        &self,
        binding: MutexGuard<'_, RelaySignInAttention>,
        pending: Vec<PendingSignInAnnouncement>,
    ) {
        if pending.is_empty() {
            return;
        }
        let _order = self.announce_order.lock().await;
        drop(binding);
        for announcement in pending {
            notify_signin_transition(
                &announcement.attention,
                announcement.conversation_id.as_deref(),
                announcement.before.as_deref(),
                announcement.after.as_deref(),
            )
            .await;
        }
    }

    /// The conversation keys this producer still backs, in REGISTRATION
    /// ORDER — oldest first, so the newest is the last element.
    async fn live_view_keys(&self) -> Vec<Option<String>> {
        self.live_views()
            .await
            .into_iter()
            .map(|view| view.key().map(str::to_string))
            .collect()
    }

    /// Who is driving, derived from the cache. No round trip, by design: this
    /// is consulted on every input call.
    pub async fn control_status(&self) -> ControlStatus {
        let last = self.last.lock().await;
        ControlStatus {
            owner: last.owner.into(),
            signin_pending: last.pending_signin.is_some(),
            blackout_active: last.blackout_active,
        }
    }

    /// Drive the reducer that lives WITH the browser, in the agent process,
    /// and bring its effects back — the daemon owns the grace-period clock.
    ///
    /// **Fallible, and the caller must respect that.** A transition that never
    /// reached the process did not happen: the process's reducer still says
    /// whatever it said before. Reporting success would leave the two sides
    /// disagreeing about who is driving — the daemon admitting input the
    /// process then refuses, and, worse, a person told the blackout is up when
    /// the process never entered it.
    /// Returns the owner the agent's reducer landed on, taken from THIS
    /// response rather than from the cache: the presentation pump writes that
    /// cache too, unordered against this call, so a re-read can answer with a
    /// pre-transition snapshot (see `BrowserView::take_control`).
    pub async fn control(
        &self,
        control: ViewControl,
    ) -> Result<(ControlOwner, Vec<ControlEffect>), String> {
        if !self.is_alive() {
            return Err(PRODUCER_GONE.to_string());
        }
        let value = call_agent(
            &self.channel,
            "agent.browser.control",
            json!({ "action": control_to_wire(control) }),
        )
        .await
        .map_err(|error| {
            tracing::warn!(
                agent_id = %self.agent_id,
                action = control_to_wire(control),
                %error,
                "browser relay: control transition did not reach the agent process"
            );
            error
        })?;
        // Read the owner off THIS response before caching it, and fall back
        // to the cache only when the agent sent no presentation at all.
        let mut owner: Option<ControlOwner> = None;
        if let Some(presentation) = value.get("presentation") {
            match serde_json::from_value::<WirePresentation>(presentation.clone()) {
                Ok(presentation) => {
                    // The owner is read off THIS response (the reply is
                    // authoritative about the transition it just performed),
                    // but the CACHE write goes through the same monotonicity
                    // guard every other writer uses. This response can be
                    // older than a push that overtook it — the agent's
                    // presentation pump fires on the same transition and can
                    // land first — and the cache is what the input gate reads.
                    owner = Some(presentation.owner.into());
                    self.cache(presentation).await;
                }
                Err(e) => tracing::warn!(
                    error = %e,
                    "browser relay: agent returned an unparseable presentation"
                ),
            }
        }
        let owner = match owner {
            Some(owner) => owner,
            None => self.last.lock().await.owner.into(),
        };
        Ok((
            owner,
            effects_from_wire(value.get("effects").unwrap_or(&Value::Null)),
        ))
    }

    /// Relay one user input to the process, which executes it against its own
    /// `BrowserTools` exactly as the in-daemon path does. Errors come back
    /// verbatim — the same code produces them on both sides.
    pub async fn input(&self, input: ViewInput) -> Result<Option<String>, String> {
        if !self.is_alive() {
            return Err(PRODUCER_GONE.to_string());
        }
        let out = call_agent(&self.channel, "agent.browser.input", input_to_wire(&input)).await?;
        Ok(out
            .get("tab_id")
            .and_then(Value::as_str)
            .map(str::to_string))
    }

    /// The capture state this producer currently wants, for the registration
    /// ack to carry.
    ///
    /// The daemon's signal is per-producer and edge-published while the agent
    /// process's `capture` watch is per-PROCESS and survives the connection —
    /// so a process that was capturing when its session dropped
    /// (`note_disconnected`'s `send(false)` never reaches the wire; the pump
    /// returns at `!is_alive()`) reconnects under a NEW producer whose count is
    /// 0 and whose `send_if_modified` emits nothing. It kept screencasting and
    /// pushing JPEGs with nobody watching. Answering it on the ack costs no
    /// extra round trip and resynchronises exactly when the process reappears.
    pub fn desired_capture(&self) -> bool {
        match self.watchers.lock() {
            Ok(watchers) => *watchers > 0,
            Err(poisoned) => *poisoned.into_inner() > 0,
        }
    }

    /// A view started streaming. The first one turns capture on in the process.
    pub fn start_capture(&self) {
        self.set_watchers(|n| n + 1);
    }

    /// A view stopped streaming. The last one turns capture off.
    pub fn stop_capture(&self) {
        self.set_watchers(|n| n.saturating_sub(1));
    }

    /// Move the watcher count and publish the state it implies as ONE step,
    /// under the count's own mutex.
    ///
    /// The count and the signal used to be two unsynchronized operations —
    /// an atomic, then a `send` on the 0→1 / 1→0 edge — and nothing above
    /// serialized them either: `stop_streamer_inner` deliberately drops the
    /// view's capture mutex BEFORE calling `stop_capture`. One host closing
    /// and reopening the drawer was enough to interleave them as
    /// stop(count 1→0) … start(0→1, send true) … stop's send(false), leaving
    /// the agent process not capturing while the view believed it was — a
    /// drawer frozen on its snapshot frame that `ensure_streamer` would never
    /// restart, because `capture.active` was true.
    ///
    /// Publishing the DERIVED state (`count > 0`) rather than only the edges
    /// is what makes the result independent of arrival order: whichever call
    /// takes the lock last publishes the state that matches the final count.
    /// `send_if_modified` keeps that from costing wire traffic — an unchanged
    /// value notifies nobody, so a second watcher still does not re-ask the
    /// process to start capturing. `std::sync::Mutex` because both callers are
    /// synchronous and nothing awaits inside.
    fn set_watchers(&self, f: impl Fn(usize) -> usize) {
        let mut watchers = match self.watchers.lock() {
            Ok(watchers) => watchers,
            Err(poisoned) => poisoned.into_inner(),
        };
        *watchers = f(*watchers);
        let desired = *watchers > 0;
        self.capture.send_if_modified(|current| {
            let changed = *current != desired;
            *current = desired;
            changed
        });
    }

    /// Bind a view to this producer so pushes reach it.
    pub async fn attach_view(&self, view: &Arc<BrowserView>) {
        let mut views = self.views.lock().await;
        views.retain(|existing| existing.strong_count() > 0);
        views.push(Arc::downgrade(view));
    }

    /// The views this producer backs beyond the newest
    /// [`MAX_VIEWS_PER_PRODUCER`], oldest first — the ones a fresh
    /// registration retires. `views` is in registration order, which is what
    /// makes "oldest" answerable here at all: the daemon sees only opaque
    /// conversation ids and cannot tell one turn's key from another
    /// conversation's.
    pub async fn views_past_the_cap(&self) -> Vec<Arc<BrowserView>> {
        let mut views = self.views.lock().await;
        views.retain(|view| view.strong_count() > 0);
        if views.len() <= MAX_VIEWS_PER_PRODUCER {
            return Vec::new();
        }
        let stale = views.len() - MAX_VIEWS_PER_PRODUCER;
        views.iter().take(stale).filter_map(Weak::upgrade).collect()
    }

    /// Seed the cache before the first push — the register call carries the
    /// process's current presentation so a view is never born empty when its
    /// browser is not.
    pub async fn set_presentation(&self, presentation: WirePresentation) {
        self.cache(presentation).await;
        self.sync_signin_attention().await;
    }

    /// The process pushed a presentation delta. Cache it, then let every view
    /// re-read: the view's own dedup decides whether that is an event.
    pub async fn push_presentation(&self, presentation: WirePresentation) {
        self.cache(presentation).await;
        self.sync_signin_attention().await;
        for view in self.live_views().await {
            view.refresh_presentation().await;
        }
    }

    /// Install a presentation, refusing one OLDER than what is cached.
    ///
    /// This cache is not a display detail — `control_status` projects it, and
    /// that is what the input gate answers from. Both writers ran
    /// unconditionally, so a `browser.producer.register` carrying a snapshot
    /// taken before its round trip could rewind a newer push that landed
    /// during it: a user who had just taken control would have `owner` read
    /// back as `Agent` and every click refused with "the agent holds control
    /// of this browser". Its two siblings already respect monotonicity —
    /// `note_disconnected` bumps the revision so the empty state never moves
    /// backwards, and `BrowserView::publish_presentation` refuses an older
    /// read for the same reason — so this closes the last unguarded writer.
    ///
    /// Equal revisions are content-identical by construction (the reducer
    /// bumps only on real change), so `<` rather than `<=` keeps a re-push of
    /// the current state working.
    async fn cache(&self, presentation: WirePresentation) {
        let mut last = self.last.lock().await;
        if presentation.revision < last.revision {
            return;
        }
        *last = presentation;
    }

    /// The process pushed a screencast frame.
    ///
    /// Only views somebody is actually WATCHING get one, and the last of them
    /// gets the frame by move. A `WireFrame` is a base64 full-viewport JPEG, so
    /// each clone is a memcpy of a few hundred KB — and this fanned out to
    /// every view the producer had ever registered, watched or not. The
    /// ordinary case (one drawer, on one conversation) is now zero clones.
    ///
    /// A skipped view's cursor does not advance, which is exactly right:
    /// `subscribe` hands out the CURRENT cursor, so a drawer arriving later
    /// starts from wherever the view is and detects gaps from there.
    pub async fn push_frame(&self, frame: WireFrame) {
        let mut watched = Vec::new();
        for view in self.live_views().await {
            if view.has_subscribers().await {
                watched.push(view);
            }
        }
        let Some(last) = watched.pop() else { return };
        for view in watched {
            view.emit_wire_frame(frame.clone()).await;
        }
        last.emit_wire_frame(frame).await;
    }

    /// Tell this process that a host-client connected or disconnected.
    ///
    /// A reverse call like the other three `agent.browser.*` methods, so it
    /// travels the same path and the process answers the same way. Its result
    /// is dropped: the daemon has nothing to do about a process that cannot
    /// be told, and the caller — a disconnect sweep or an auth handshake —
    /// must not wait on it. [`ProducerRegistry::broadcast_host_connected`] is
    /// what keeps that non-blocking.
    pub async fn push_host_connected(&self, connected: bool) {
        if !self.is_alive() {
            return;
        }
        self.host_desired.store(connected, Ordering::Release);
        // Serialized: the wire order is the lock order, so two transitions can
        // no longer land out of order. See `host_push`.
        let mut last = self.host_push.lock().await;
        // Re-read under the lock — a newer transition arriving while this task
        // waited supersedes the value it was spawned with, and sending the
        // stale one would tell the process the older truth.
        let connected = self.host_desired.load(Ordering::Acquire);
        if *last == Some(connected) {
            return;
        }
        match call_agent(
            &self.channel,
            "agent.browser.host_connected",
            json!({ "connected": connected }),
        )
        .await
        {
            Ok(_) => *last = Some(connected),
            Err(error) => tracing::debug!(
                agent_id = %self.agent_id,
                %error,
                "browser relay: could not tell the agent process about a host transition"
            ),
        }
    }

    /// The process's connection dropped. Every call from here on is a clean
    /// error, and the views report an empty browser — which is the truth: the
    /// process is gone and its Chromium went with it.
    ///
    /// The views stay REGISTERED. When the supervisor restarts the process and
    /// it registers the same conversation again, that registration replaces
    /// this view through the ordinary `adopt` path, so a drawer that never
    /// unsubscribed follows the agent to its new process without the cursor
    /// moving backwards.
    pub async fn note_disconnected(&self) {
        self.alive.store(false, Ordering::Release);
        let _ = self.capture.send(false);
        {
            let mut last = self.last.lock().await;
            let revision = last.revision.saturating_add(1);
            *last = WirePresentation::empty();
            // Never let the revision move backwards: a client that treats it
            // as monotonic would read the reset as a change it already saw.
            last.revision = revision;
        }
        self.sync_signin_attention().await;
        for view in self.live_views().await {
            view.refresh_presentation().await;
        }
    }

    async fn live_views(&self) -> Vec<Arc<BrowserView>> {
        let mut views = self.views.lock().await;
        views.retain(|view| view.strong_count() > 0);
        views.iter().filter_map(Weak::upgrade).collect()
    }

    fn spawn_capture_pump(self: &Arc<Self>, mut rx: watch::Receiver<bool>) {
        let producer = Arc::downgrade(self);
        tokio::spawn(async move {
            // A `watch` collapses intermediate values, so a rapid
            // open/close/open settles on the LAST desired state rather than
            // racing two calls into the process out of order.
            // `undelivered` is what turns a logged failure into a retry: it
            // means "a desired state has not reached the process", so the loop
            // re-sends instead of parking on the next change.
            let mut undelivered = false;
            loop {
                if !undelivered && rx.changed().await.is_err() {
                    return;
                }
                let enabled = *rx.borrow_and_update();
                let Some(producer) = producer.upgrade() else {
                    return;
                };
                if !producer.is_alive() {
                    return;
                }
                let sent = call_agent(
                    &producer.channel,
                    "agent.browser.capture",
                    json!({ "enabled": enabled }),
                )
                .await;
                undelivered = match sent {
                    Ok(_) => false,
                    Err(e) => {
                        // RETRIED, not just logged. Nothing else re-arms this:
                        // for a relay view `start_capture` returns no task, so
                        // `ensure_streamer` sees `capture.active == true` (set
                        // the instant the count moved, regardless of whether
                        // the wire call landed) and returns early forever; and
                        // `set_watchers` publishes only on a CHANGE, so a
                        // second host subscribing sends nothing new. One
                        // `RELAY_CALL_TIMEOUT` against a busy agent therefore
                        // left the drawer on its subscribe-time snapshot with
                        // zero frames until the user closed and reopened it.
                        tracing::debug!(
                            agent_id = %producer.agent_id,
                            enabled,
                            error = %e,
                            "browser relay: capture request did not reach the agent process; retrying"
                        );
                        true
                    }
                };
                drop(producer);
                if undelivered {
                    // A newer desired state supersedes the retry; otherwise
                    // back off and send the same one again. Either way the
                    // value is re-read at the top, so a retry never delivers
                    // something the watch has already replaced.
                    tokio::select! {
                        changed = rx.changed() => {
                            if changed.is_err() {
                                return;
                            }
                        }
                        _ = tokio::time::sleep(CAPTURE_RETRY_BACKOFF) => {}
                    }
                }
            }
        });
    }
}

// ---------------------------------------------------------------------------
// Handlers — the agent-facing wire surface
// ---------------------------------------------------------------------------

/// The agent identity bound to this connection by `session.auth`, or a clean
/// refusal. This is the whole authorization rule for `browser.producer.*`: an
/// agent may publish its own browser and nothing else.
async fn authorize_producer(session: &ClientSession) -> Result<String, String> {
    session.agent_id.lock().await.clone().ok_or_else(|| {
        "not authorized to use browser.producer.*: this connection is not a supervised agent \
         (session.auth { token, agent_id })"
            .to_string()
    })
}

/// May this agent publish a browser for this conversation?
///
/// Two ways to be entitled, and both validate the SAME binding — an agent may
/// only ever claim a conversation it serves:
///
/// - **A live chat session for it**, which is the first registration: the turn
///   is running, and `chat_sessions` says which agent the daemon dispatched it
///   to. This is the only way a binding is ESTABLISHED.
/// - **A binding this agent already established**, which is re-registration.
///   Needed because `chat_sessions` is per TURN (dropped on the terminal
///   event) while a published browser outlives its run: without this, a
///   process whose daemon session dropped between turns could not restore its
///   own drawer until the user happened to send another message, and the
///   drawer would sit on "its browser is gone" pointing at a live browser.
///
/// A conversation nobody has ever served, and one served by somebody else, are
/// both refused — the second case identically to before. This is a liveness
/// relaxation, not an authorization one: the agent_id ↔ conversation binding
/// is still validated against the daemon's own record every time.
pub fn authorize_conversation_claim(
    conversation_id: &str,
    agent_id: &str,
    live_owner: Option<&str>,
    bound_owner: Option<&str>,
) -> Result<(), String> {
    match live_owner.or(bound_owner) {
        Some(owner) if owner == agent_id => Ok(()),
        Some(owner) => Err(format!(
            "conversation '{conversation_id}' is served by agent '{owner}', not '{agent_id}'"
        )),
        None => Err(format!(
            "conversation '{conversation_id}' is not an active chat session for agent \
             '{agent_id}' — register from inside the turn that serves it"
        )),
    }
}

/// `browser.producer.register { conversation_id, presentation? }` — the
/// supervised process publishes its browser for one chat session.
///
/// Idempotent: the same process registering the same conversation again (every
/// turn does) keeps the existing view, its subscribers and its cursor. A
/// DIFFERENT process claiming the key replaces the view through `adopt`, which
/// carries subscribers and the cursor across.
pub async fn handle_producer_register(
    req: &JsonRpcMessage,
    session: &Arc<ClientSession>,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let agent_id = authorize_producer(session).await?;
    let conversation_id = req
        .params
        .get("conversation_id")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|id| !id.is_empty())
        .ok_or("browser.producer.register requires a non-empty { conversation_id }")?
        .to_string();

    // A conversation may only be claimed by the agent actually serving it.
    // `chat_sessions` is populated BEFORE the daemon sends `agent.chat`, so a
    // registration from inside the turn always finds its entry; a
    // re-registration between turns is entitled by the binding that first
    // registration established. See `authorize_conversation_claim`.
    let live_owner = state
        .chat_sessions
        .lock()
        .await
        .get(&conversation_id)
        .map(|chat| chat.agent_id.clone());
    let bound_owner = state
        .browser_views
        .conversation_owner(&conversation_id)
        .await;
    authorize_conversation_claim(
        &conversation_id,
        &agent_id,
        live_owner.as_deref(),
        bound_owner.as_deref(),
    )?;
    state
        .browser_views
        .bind_conversation(&conversation_id, &agent_id)
        .await;

    let producer = state
        .browser_views
        .producer_for(&session.client_id, &agent_id, &session.channel)
        .await;
    if let Some(presentation) = req.params.get("presentation") {
        match serde_json::from_value::<WirePresentation>(presentation.clone()) {
            Ok(presentation) => producer.set_presentation(presentation).await,
            Err(e) => {
                return Err(format!(
                    "browser.producer.register `presentation` is not a presentation object: {e}"
                ))
            }
        }
    }
    state
        .browser_views
        .register_relay(conversation_id.clone(), Arc::clone(&producer))
        .await;

    // Task 7: the supervised process has no direct read of the daemon's
    // session set, so it learns "is a CarHost host-client connected right
    // now" from this acknowledgment — the natural existing round trip,
    // rather than a dedicated method. See
    // `assistant::browser_producer::BrowserProducer` for how it's applied
    // (and that module's own doc comment for the freshness bound this
    // implies: refreshed on every registration that reaches the daemon, not
    // continuously).
    Ok(json!({
        "ok": true,
        "conversation_id": conversation_id,
        "host_connected": state.any_host_connected().await,
        // And whether anything is actually watching this producer's browser.
        // The process's capture watch is per-PROCESS and survives a
        // reconnect, while this signal is per-producer and edge-published, so
        // without an authoritative answer here a process that was capturing
        // when its session dropped keeps screencasting under a fresh producer
        // that will never tell it otherwise. See `RelayProducer::desired_capture`.
        "capture": producer.desired_capture(),
    }))
}

/// Intercept the two producer NOTIFICATIONS (`browser.producer.presentation`,
/// `browser.producer.frame`). Returns `true` when the frame was ours, so the
/// dispatcher skips a method-not-found reply to something that has no id —
/// the same shape as `try_forward_agent_chat_event`.
pub(crate) async fn try_handle_producer_push(
    parsed: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> bool {
    let Some(method) = parsed.method.as_deref() else {
        return false;
    };
    if method != "browser.producer.presentation" && method != "browser.producer.frame" {
        return false;
    }
    if !parsed.id.is_null() {
        // Has an id → a request, not a notification. Let the dispatcher
        // answer method-not-found rather than swallowing it.
        return false;
    }
    // Fall THROUGH to the auth gate on an unauthenticated connection, rather
    // than consuming the frame here.
    //
    // Nothing leaks either way — the lookup below is by `session.client_id`,
    // which an unauthenticated connection cannot have registered a producer
    // under, so this handler is already fail-closed. What consuming the frame
    // costs is the reject-AND-CLOSE property: an unauthenticated peer could
    // hold the socket open indefinitely, forcing a full JSON parse (tungstenite's
    // 64 MiB default is the only bound) and a registry lock acquisition per
    // frame. Returning false hands it to the gate, which answers and closes.
    if state.auth_token.get().is_some()
        && !session
            .authenticated
            .load(std::sync::atomic::Ordering::Acquire)
    {
        return false;
    }
    // Only a connection that has actually registered a producer has one; an
    // unregistered session's push finds nothing and is dropped.
    let Some(producer) = state.browser_views.producer(&session.client_id).await else {
        tracing::debug!(
            client_id = %session.client_id,
            method,
            "browser relay: push from a connection with no registered producer"
        );
        return true;
    };

    if method == "browser.producer.presentation" {
        match serde_json::from_value::<WirePresentation>(
            parsed
                .params
                .get("presentation")
                .cloned()
                .unwrap_or(Value::Null),
        ) {
            Ok(presentation) => producer.push_presentation(presentation).await,
            Err(e) => tracing::debug!(
                error = %e,
                "browser relay: unparseable presentation push"
            ),
        }
    } else {
        match serde_json::from_value::<WireFrame>(
            parsed.params.get("frame").cloned().unwrap_or(Value::Null),
        ) {
            Ok(frame) if frame.jpeg_base64.len() > MAX_PRODUCER_FRAME_BYTES => {
                tracing::debug!(
                    client_id = %session.client_id,
                    bytes = frame.jpeg_base64.len(),
                    "browser relay: dropped an oversized producer frame"
                );
            }
            Ok(frame) => producer.push_frame(frame).await,
            Err(e) => tracing::debug!(error = %e, "browser relay: unparseable frame push"),
        }
    }
    true
}

/// The producers currently attached, keyed by the agent connection's client id.
/// Lives beside the view registry because both are torn down on the same
/// disconnect boundary.
#[derive(Default)]
pub struct ProducerRegistry {
    producers: Mutex<HashMap<String, Arc<RelayProducer>>>,
    /// conversation → the agent that established the claim on it. Outlives
    /// both the chat session (per turn) and the producer (per connection),
    /// because it is what entitles that agent — and only that agent — to
    /// republish its browser after either goes away.
    bindings: Mutex<HashMap<String, String>>,
}

impl ProducerRegistry {
    pub async fn get(&self, client_id: &str) -> Option<Arc<RelayProducer>> {
        self.producers.lock().await.get(client_id).cloned()
    }

    /// The agent entitled to publish this conversation, if one established a
    /// claim on it.
    pub async fn conversation_owner(&self, conversation_id: &str) -> Option<String> {
        self.bindings.lock().await.get(conversation_id).cloned()
    }

    /// Record a validated claim. Only ever called AFTER
    /// [`authorize_conversation_claim`] passed, so this can never widen who is
    /// entitled to a conversation — it only remembers what the live chat
    /// session already said.
    pub async fn bind_conversation(&self, conversation_id: &str, agent_id: &str) {
        self.bindings
            .lock()
            .await
            .insert(conversation_id.to_string(), agent_id.to_string());
    }

    /// Drop a claim whose view has been retired past
    /// [`MAX_VIEWS_PER_PRODUCER`]. Never widens anything: an agent that wants
    /// this conversation back has to be serving a LIVE chat session for it,
    /// which is the same check that established the binding in the first place.
    pub async fn forget_binding(&self, conversation_id: &str) {
        self.bindings.lock().await.remove(conversation_id);
    }

    /// The producer for this connection, created on first registration.
    pub async fn get_or_create(
        &self,
        client_id: &str,
        agent_id: &str,
        channel: &Arc<WsChannel>,
    ) -> Arc<RelayProducer> {
        let mut producers = self.producers.lock().await;
        Arc::clone(producers.entry(client_id.to_string()).or_insert_with(|| {
            RelayProducer::new(
                client_id.to_string(),
                agent_id.to_string(),
                Arc::clone(channel),
            )
        }))
    }

    /// Tell every live producer that host connectivity changed.
    ///
    /// Called on the two transitions the daemon actually observes: a
    /// connection authenticating as the host client, and a host connection
    /// dropping. Producers cache the answer, so a push is what keeps that
    /// cache honest between registrations.
    pub async fn broadcast_host_connected(&self, connected: bool) {
        let producers: Vec<Arc<RelayProducer>> =
            self.producers.lock().await.values().cloned().collect();
        // Fire-and-forget, one task each. The callers are a disconnect sweep
        // and an auth handshake — neither may be held up for the relay
        // timeout by a process that has stopped answering, and one wedged
        // producer must not delay telling the others.
        for producer in producers {
            tokio::spawn(async move { producer.push_host_connected(connected).await });
        }
    }

    /// The connection dropped: the producer is gone. Its views stay registered
    /// and report an empty browser until a replacement claims the key.
    pub async fn note_disconnected(&self, client_id: &str) {
        let producer = self.producers.lock().await.remove(client_id);
        if let Some(producer) = producer {
            producer.note_disconnected().await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::browser_view::{BrowserViewRegistry, WireOwner};

    use crate::browser_view::BrowserViewEvent;
    use crate::session::WsSink;
    use futures::StreamExt;

    fn agent_channel() -> (
        Arc<WsChannel>,
        std::sync::Arc<std::sync::Mutex<Vec<String>>>,
    ) {
        let (channel, frames) = WsChannel::test_capture();
        (Arc::new(channel), frames)
    }

    /// A host connection whose pushed `browser.view.event` frames a test can
    /// read, mirroring `browser_view::tests::capture_channel`.
    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()),
            active_actions: Mutex::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(0),
        });
        (channel, rx)
    }

    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(text) => text.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")
    }

    /// Answer the next reverse call the daemon parked on `channel`, exactly as
    /// the supervised process's `DaemonClient` would: match the request id,
    /// resolve the oneshot. Returns the request it answered.
    async fn answer_next_call(
        channel: &Arc<WsChannel>,
        frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>,
        result: Value,
    ) -> Value {
        for _ in 0..200 {
            let request = frames
                .lock()
                .unwrap()
                .iter()
                .filter_map(|text| serde_json::from_str::<Value>(text).ok())
                .find(|value| value.get("id").and_then(Value::as_str).is_some());
            if let Some(request) = request {
                let id = request["id"].as_str().unwrap().to_string();
                let waiter = channel.pending.lock().await.remove(&id);
                if let Some(waiter) = waiter {
                    let _ = waiter.send(car_proto::ToolExecuteResponse {
                        action_id: id,
                        output: Some(result),
                        error: None,
                        terminal: false,
                    });
                    frames.lock().unwrap().clear();
                    return request;
                }
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        panic!("no reverse call arrived on the agent channel");
    }

    /// Wait for a reverse call to be written, then forget the frame WITHOUT
    /// answering it — leaving that request pending so it times out. Used to
    /// park one call while another is issued and answered.
    async fn park_next_call(frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
        for _ in 0..200 {
            let seen = frames
                .lock()
                .unwrap()
                .iter()
                .filter_map(|text| serde_json::from_str::<Value>(text).ok())
                .any(|value| value.get("id").and_then(Value::as_str).is_some());
            if seen {
                frames.lock().unwrap().clear();
                return;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        panic!("no reverse call arrived on the agent channel");
    }

    fn presentation(owner: WireOwner, url: &str) -> WirePresentation {
        let mut wire = WirePresentation::empty();
        wire.revision = 3;
        wire.owner = owner;
        wire.url = Some(url.to_string());
        wire
    }

    // -----------------------------------------------------------------
    // Sign-in attention: the relay twin (Parslee-ai/car#1040)
    // -----------------------------------------------------------------

    /// A presentation at `revision`, optionally with a sign-in pending.
    fn presentation_at(revision: u64, pending_signin: Option<&str>) -> WirePresentation {
        let mut wire = WirePresentation::empty();
        wire.revision = revision;
        wire.owner = WireOwner::Agent;
        wire.pending_signin = pending_signin.map(str::to_string);
        wire.blackout_active = pending_signin.is_some();
        wire
    }

    /// A registry whose relayed views report to a recorder, plus a live
    /// producer registered under `conv-1` — the whole supervised-agent path
    /// minus a real agent process.
    async fn relayed_view_watching_signin() -> (
        Arc<RelayProducer>,
        Arc<crate::browser_view::BrowserViewRegistry>,
        Arc<crate::browser_attention::RecordingAttention>,
    ) {
        let registry = Arc::new(crate::browser_view::BrowserViewRegistry::default());
        let recorder = Arc::new(crate::browser_attention::RecordingAttention::default());
        registry.set_signin_attention(recorder.clone());
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        (producer, registry, recorder)
    }

    /// The relay half of the headline. The reducer lives in the agent
    /// process, so the daemon has to DERIVE the transition from the
    /// presentation push it already receives — and that push is not gated on
    /// anybody watching the drawer, which is the entire reason this works
    /// with the drawer closed.
    #[tokio::test]
    async fn a_relayed_sign_in_notifies_from_the_presentation_push_alone() {
        let (producer, _registry, recorder) = relayed_view_watching_signin().await;

        producer.push_presentation(presentation_at(1, None)).await;
        assert!(
            recorder.kinds().is_empty(),
            "an ordinary presentation is not news"
        );

        producer
            .push_presentation(presentation_at(
                2,
                Some("Sign in at https://example.com/login"),
            ))
            .await;
        assert_eq!(
            recorder.calls(),
            vec![(
                crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                Some("conv-1".to_string()),
                Some("Sign in at https://example.com/login".to_string()),
            )],
            "the view key and the agent's own prompt both travel"
        );

        producer.push_presentation(presentation_at(3, None)).await;
        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED
            ],
            "and the agent process resolving it clears the badge"
        );
    }

    /// `presentation_pump` republishes on every change and the 10-second
    /// sweep re-registers, so the SAME state arriving again must say nothing.
    /// This is the case that would otherwise banner an operator every few
    /// seconds for one sign-in.
    #[tokio::test]
    async fn republishing_the_same_pending_sign_in_says_nothing() {
        let (producer, _registry, recorder) = relayed_view_watching_signin().await;

        let pending = presentation_at(2, Some("Sign in at https://example.com/login"));
        for _ in 0..4 {
            producer.push_presentation(pending.clone()).await;
        }
        assert_eq!(
            recorder.kinds(),
            vec![crate::browser_attention::BROWSER_SIGNIN_NEEDED],
            "one wait is one notification, however many times it is republished"
        );

        // A later revision that still has the sign-in up — the tab list moved
        // under it — is still not a transition.
        let mut moved = presentation_at(3, Some("Sign in at https://example.com/login"));
        moved.url = Some("https://example.com/login?step=2".into());
        producer.push_presentation(moved).await;
        assert_eq!(
            recorder.kinds(),
            vec![crate::browser_attention::BROWSER_SIGNIN_NEEDED]
        );
    }

    #[tokio::test]
    async fn one_process_with_two_views_emits_one_needed_event() {
        let (producer, registry, recorder) = relayed_view_watching_signin().await;
        registry
            .register_relay("conv-2", Arc::clone(&producer))
            .await;

        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;

        assert_eq!(
            recorder.calls(),
            vec![(
                crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                Some("conv-2".to_string()),
                Some("Sign in".to_string()),
            )],
            "attention is process-owned and routed through the newest view"
        );
        assert_eq!(registry.pending_signins().await.len(), 1);
    }

    #[tokio::test]
    async fn a_new_conversation_does_not_steal_an_announced_wait() {
        let (producer, registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in for chat one")))
            .await;

        registry
            .register_relay("conv-2", Arc::clone(&producer))
            .await;
        assert_eq!(
            recorder.calls(),
            vec![(
                crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                Some("conv-1".to_string()),
                Some("Sign in for chat one".to_string()),
            )],
            "a later turn cannot move an active badge off the blocked chat"
        );
        assert_eq!(
            registry.pending_signins().await[0].conversation_id,
            "conv-1"
        );

        producer.push_presentation(presentation_at(3, None)).await;
        producer
            .push_presentation(presentation_at(4, Some("Sign in for chat two")))
            .await;
        assert_eq!(
            recorder.calls().last().unwrap().1.as_deref(),
            Some("conv-2"),
            "after resolution the newest registered chat owns the next wait"
        );
    }

    #[tokio::test]
    async fn changing_the_pending_prompt_refreshes_operator_attention() {
        let (producer, registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in at A")))
            .await;
        producer
            .push_presentation(presentation_at(3, Some("Sign in at B")))
            .await;

        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
            ]
        );
        assert_eq!(registry.pending_signins().await[0].message, "Sign in at B");
    }

    /// The supervised process dying is a real ending: its views publish the
    /// empty presentation, which is a `pending -> none` transition. Without
    /// this the badge would outlive the process that raised it, with nothing
    /// left anywhere that could ever clear it.
    #[tokio::test]
    async fn the_agent_process_going_away_resolves_its_pending_sign_in() {
        let (producer, _registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;
        producer.note_disconnected().await;
        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED
            ]
        );
    }

    /// A registry with no sink installed — every other test in this crate,
    /// and every embedder with no daemon behind it — relays the presentation
    /// exactly as it did before, and announces nothing because there is
    /// nowhere to announce to.
    #[tokio::test]
    async fn a_registry_with_no_attention_sink_relays_but_announces_nothing() {
        let registry = Arc::new(crate::browser_view::BrowserViewRegistry::default());
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;
        // The drawer's own orange strip is unaffected: it reads the relayed
        // presentation, not the attention route.
        assert_eq!(
            view.snapshot_for_test().await.0.pending_signin.as_deref(),
            Some("Sign in")
        );
        // But nothing was ANNOUNCED. `announced` is the record of what an
        // operator was actually told, and with no sink it never advances —
        // which is the only handle this test has on "no attention call
        // happened", and is what `pending_signins` projects.
        assert!(
            producer.signin_snapshot().await.is_none(),
            "no sink means nothing was ever announced"
        );
        assert!(registry.pending_signins().await.is_empty());
    }

    /// One supervised process backs up to [`MAX_VIEWS_PER_PRODUCER`] views.
    /// Retiring the view that happens to own the attention route must not
    /// leave the process with no sink while it is still serving the others —
    /// a sign-in raised afterwards would fall through the `attention` guard
    /// and tell nobody until the next turn's `register_relay` reinstalled one.
    #[tokio::test]
    async fn retiring_the_routed_view_keeps_the_sink_for_the_views_that_remain() {
        let (producer, registry, recorder) = relayed_view_watching_signin().await;
        // A second turn registers, so the route moves to the newest view.
        registry
            .register_relay("conv-2", Arc::clone(&producer))
            .await;

        // That newest view retires while `conv-1` is still served by this
        // same process.
        producer.detach_signin_attention(Some("conv-2")).await;

        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;
        assert_eq!(
            recorder.calls(),
            vec![(
                crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                Some("conv-1".to_string()),
                Some("Sign in".to_string()),
            )],
            "the route moves to a surviving view instead of nulling the sink"
        );
    }

    /// A wait pins `conversation_id` to the turn that raised it while later
    /// turns move `latest_conversation_id` on ahead of it. When that pinned
    /// view is finally retired past the cap, the route must land on the
    /// NEWEST turn — the one the operator is actually looking at — and the
    /// wait must be re-raised there, because the browser is still blocked.
    #[tokio::test]
    async fn retiring_the_routed_view_hands_the_route_to_the_newest_turn() {
        let (producer, registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(
                2,
                Some("Sign in at https://example.com/login"),
            ))
            .await;

        // Eight further turns. `set_signin_attention` keeps the announced wait
        // pinned to conv-1 and moves only `latest_conversation_id`.
        for turn in 2..=9 {
            registry
                .register_relay(format!("conv-{turn}"), Arc::clone(&producer))
                .await;
        }
        // conv-1 is now past MAX_VIEWS_PER_PRODUCER. `retire_views_past_the_cap`
        // is what reaches this in production.
        producer.detach_signin_attention(Some("conv-1")).await;

        assert_eq!(
            recorder.calls(),
            vec![
                (
                    crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                    Some("conv-1".to_string()),
                    Some("Sign in at https://example.com/login".to_string()),
                ),
                (
                    crate::browser_attention::BROWSER_SIGNIN_RESOLVED.to_string(),
                    Some("conv-1".to_string()),
                    None,
                ),
                (
                    crate::browser_attention::BROWSER_SIGNIN_NEEDED.to_string(),
                    Some("conv-9".to_string()),
                    Some("Sign in at https://example.com/login".to_string()),
                ),
            ],
            "the badge moves to the newest turn, not the oldest survivor, and \
             the still-blocked browser is re-raised rather than left cleared"
        );
        assert_eq!(
            producer
                .signin_snapshot()
                .await
                .expect("the browser is still blocked")
                .conversation_id,
            "conv-9"
        );
        assert_eq!(
            registry.pending_signins().await[0].conversation_id,
            "conv-9",
            "and a host reconnecting mid-wait is pointed at the same turn"
        );
    }

    /// The survivor set has to be read UNDER the attention lock. Read before
    /// it, a `register_relay` landing in the window was invisible here, and
    /// the sink was nulled with that new view live — the exact state the
    /// survivor check exists to prevent, reached by a race.
    #[tokio::test]
    async fn a_registration_landing_during_a_detach_still_leaves_a_live_sink() {
        let (producer, _registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;

        // A conv-2 view built on a throwaway producer, so making it live
        // below touches `views` only and never the lock the test is holding.
        let (scratch_channel, _scratch_frames) = agent_channel();
        let scratch_producer =
            RelayProducer::new("other-conn".into(), "car-assistant".into(), scratch_channel);
        let scratch_registry = Arc::new(crate::browser_view::BrowserViewRegistry::default());
        let view_two = scratch_registry
            .register_relay("conv-2", Arc::clone(&scratch_producer))
            .await;

        // Freeze the detach at its lock acquisition — precisely where the old
        // order had already read the survivors and found none.
        let guard = producer.signin_attention.lock().await;
        let detach = tokio::spawn({
            let producer = Arc::clone(&producer);
            async move { producer.detach_signin_attention(Some("conv-1")).await }
        });
        tokio::task::yield_now().await;

        // conv-2 becomes live inside that window.
        producer.attach_view(&view_two).await;
        drop(guard);
        detach.await.unwrap();

        assert_eq!(
            producer
                .signin_snapshot()
                .await
                .expect("a live view remains, so the wait keeps a route")
                .conversation_id,
            "conv-2"
        );
        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED,
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
            ],
            "the badge moves to the surviving view instead of the sink being nulled"
        );
    }

    /// The other half of the rule: when the retiring view IS the last one this
    /// producer serves, the sink goes with it.
    #[tokio::test]
    async fn retiring_the_last_view_detaches_the_sink() {
        let (producer, _registry, recorder) = relayed_view_watching_signin().await;
        producer
            .push_presentation(presentation_at(2, Some("Sign in")))
            .await;

        producer.detach_signin_attention(Some("conv-1")).await;
        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED
            ],
            "the wait it owned is resolved on the way out"
        );

        producer
            .push_presentation(presentation_at(4, Some("Sign in again")))
            .await;
        assert_eq!(
            recorder.kinds(),
            vec![
                crate::browser_attention::BROWSER_SIGNIN_NEEDED,
                crate::browser_attention::BROWSER_SIGNIN_RESOLVED
            ],
            "nothing left to serve, so nothing left to announce"
        );
    }

    /// The broadcast must not happen inside the lock every relayed input and
    /// every presentation push contends on. `HostState::record_event` awaits
    /// each `host.subscribe` socket in turn, bounded at 10s apiece, so holding
    /// `signin_attention` across it made N backpressured hosts an N x 10s
    /// stall on the very next push — for a call with nothing to announce.
    #[tokio::test]
    async fn a_stalled_broadcast_does_not_block_the_next_presentation() {
        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 registry = Arc::new(crate::browser_view::BrowserViewRegistry::default());
        registry.set_signin_attention(Arc::new(BlockingAttention {
            entered: Arc::clone(&entered),
            release: Arc::clone(&release),
        }));
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        let blocked = tokio::spawn({
            let producer = Arc::clone(&producer);
            async move {
                producer
                    .push_presentation(presentation_at(2, Some("Sign in")))
                    .await
            }
        });
        // The announcement is now in flight and wedged on a host socket.
        entered.notified().await;

        // Same pending prompt at a later revision: the republish case, which
        // has nothing to announce and must settle without waiting on it.
        let mut moved = presentation_at(3, Some("Sign in"));
        moved.url = Some("https://example.com/login?step=2".into());
        tokio::time::timeout(Duration::from_secs(5), producer.push_presentation(moved))
            .await
            .expect("a non-transition push must not queue behind a stalled broadcast");

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

    #[tokio::test]
    async fn input_crosses_to_the_agent_process_as_a_reverse_call_and_returns_its_answer() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);

        let relayed = tokio::spawn({
            let producer = Arc::clone(&producer);
            async move { producer.input(ViewInput::TabOpen).await }
        });
        let request =
            answer_next_call(&producer.channel, &frames, json!({ "tab_id": "tab-4" })).await;

        assert_eq!(request["method"], "agent.browser.input");
        assert_eq!(request["params"]["op"], "tab_open");
        assert_eq!(relayed.await.unwrap().unwrap().as_deref(), Some("tab-4"));
    }

    #[tokio::test]
    async fn the_agent_s_error_reaches_the_caller_verbatim() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);

        let relayed = tokio::spawn({
            let producer = Arc::clone(&producer);
            async move { producer.input(ViewInput::Click { x: 1.0, y: 2.0 }).await }
        });
        // The agent answers with an error frame; the daemon's demuxer turns
        // that into `ToolExecuteResponse.error`.
        for _ in 0..200 {
            let id = frames
                .lock()
                .unwrap()
                .iter()
                .filter_map(|t| serde_json::from_str::<Value>(t).ok())
                .find_map(|v| v.get("id").and_then(Value::as_str).map(str::to_string));
            if let Some(id) = id {
                if let Some(waiter) = producer.channel.pending.lock().await.remove(&id) {
                    let _ = waiter.send(car_proto::ToolExecuteResponse {
                        action_id: id,
                        output: None,
                        error: Some("no browser is running for this view".into()),
                        terminal: false,
                    });
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        assert_eq!(
            relayed.await.unwrap().unwrap_err(),
            "no browser is running for this view"
        );
    }

    #[tokio::test]
    async fn control_relays_the_transition_and_brings_its_effects_back() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);

        let relayed = tokio::spawn({
            let producer = Arc::clone(&producer);
            async move { producer.control(ViewControl::HolderDisconnected).await }
        });
        let request = answer_next_call(
            &producer.channel,
            &frames,
            json!({
                "presentation": presentation(WireOwner::User, "https://x.test/"),
                "effects": [{ "effect": "start_grace_period" }],
            }),
        )
        .await;

        assert_eq!(request["method"], "agent.browser.control");
        assert_eq!(request["params"]["action"], "holder_disconnected");
        assert_eq!(
            relayed.await.unwrap(),
            Ok((ControlOwner::User, vec![ControlEffect::StartGracePeriod])),
            "the daemon owns the clock, so the effect has to cross back — and the owner \
             comes from THIS response, not from a re-read of the cache"
        );
        // And the answer updated the cache the input path reads.
        assert_eq!(
            producer.control_status().await.owner,
            crate::assistant::browser_control::ControlOwner::User
        );
    }

    #[tokio::test]
    async fn a_dead_producer_refuses_input_instead_of_hanging_on_a_call() {
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        producer
            .set_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;

        producer.note_disconnected().await;

        assert!(!producer.is_alive());
        let err = producer
            .input(ViewInput::Navigate {
                url: "https://y.test".into(),
            })
            .await
            .unwrap_err();
        assert_eq!(err, PRODUCER_GONE);
        let cleared = producer.presentation().await;
        assert_eq!(cleared.owner, WireOwner::None);
        assert_eq!(cleared.url, None);
        assert!(
            cleared.revision > 3,
            "the revision never moves backwards, even when the browser vanishes"
        );
    }

    /// `take_control` used to decide whether to record the control holder
    /// from a SECOND, independent `control_status()` read — and on the relay
    /// path that projects the producer's CACHED presentation, which the
    /// agent's own presentation pump also writes, unordered against the
    /// transition. A push carrying a pre-take snapshot landing in that window
    /// made the daemon believe nobody held a browser a person had just taken,
    /// which re-opened the hand-back gate from the recording side.
    #[tokio::test]
    async fn a_control_transition_reports_the_owner_from_its_own_answer() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);

        let relayed = {
            let producer = Arc::clone(&producer);
            tokio::spawn(async move { producer.control(ViewControl::TakeControl).await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({
                "presentation": presentation(WireOwner::User, "https://x.test/"),
                "effects": [],
            }),
        )
        .await;
        let (owner, _) = relayed.await.unwrap().unwrap();
        assert_eq!(
            owner,
            ControlOwner::User,
            "the transition landed on User, and that is what the caller must act on"
        );

        // The agent's presentation pump now overwrites the cache with a
        // pre-take snapshot — the race that used to decide the holder.
        producer
            .push_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;
        assert_eq!(
            producer.control_status().await.owner,
            ControlOwner::Agent,
            "the cache really can go backwards, which is why it cannot be the decider"
        );
    }

    /// `note_disconnect` armed the grace timer only from an effect returned
    /// by the very process it has just concluded is not replying — and
    /// `control_best_effort` swallows a relayed transition that never landed,
    /// returning none. The view then sat on `owner: user` with the blackout
    /// up, the connection provably gone, and nothing left to clear it.
    ///
    /// Driven on the relay path because that is the only one where the
    /// transition can genuinely fail: a local reducer never does.
    #[tokio::test(start_paused = true)]
    async fn a_disconnect_arms_the_grace_timer_even_when_the_transition_never_lands() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        // The person takes control while the process is still answering.
        let taking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-1").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({
                "presentation": presentation(WireOwner::User, "https://x.test/"),
                "effects": [],
            }),
        )
        .await;
        taking.await.unwrap().expect("take control");

        // The process then stops answering, and the holder's connection drops.
        producer.note_disconnected().await;
        let before = view.grace_generation_for_test().await;
        view.note_disconnect("host-1", false).await;
        let after = view.grace_generation_for_test().await;

        assert!(
            view.control_holder_for_test().await.is_none(),
            "the holder is cleared at disconnect — the connection is provably gone"
        );
        // Two bumps: one clearing the holder, one ARMING the timer.
        // `control_best_effort` returned no effects here (the process is
        // gone), so a single bump means no timer was spawned and nothing
        // would ever have reverted ownership.
        assert_eq!(
            after - before,
            2,
            "the timer must be armed from what the daemon knows, not from the reply of a \
             process that is not answering"
        );
    }

    /// The grace timer is armed before the bounded relay reconciliation. A
    /// `take_control` landing while that detached call is pending must remain
    /// the newer generation and survive the stale expiry.
    #[tokio::test(start_paused = true)]
    async fn a_take_control_inside_the_disconnect_window_is_not_revoked_by_the_grace_timer() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        // host-1 holds control.
        let taking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-1").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        taking.await.unwrap().expect("take control");

        // Its connection drops. The agent process stops answering; teardown
        // returns after arming recovery while reconciliation stays bounded in
        // a detached task.
        let disconnecting = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.note_disconnect("host-1", false).await })
        };
        // Its `HolderDisconnected` is left PENDING, but disconnect teardown is
        // no longer parked behind it.
        park_next_call(&frames).await;
        disconnecting.await.unwrap();

        // The app reconnects and takes control again INSIDE that window.
        let retaking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-2").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        retaking.await.unwrap().expect("re-take control");
        assert_eq!(
            view.control_holder_for_test().await.as_deref(),
            Some("host-2")
        );

        // The detached relayed call finally times out; then the whole grace
        // window elapses.
        tokio::time::sleep(RELAY_CALL_TIMEOUT + Duration::from_secs(1)).await;
        tokio::time::sleep(crate::browser_view::CONTROL_GRACE + Duration::from_secs(1)).await;

        assert_eq!(
            view.control_holder_for_test().await.as_deref(),
            Some("host-2"),
            "a holder who took control legitimately must not be revoked by a timer armed \
             for the connection they replaced"
        );
    }

    /// How many `agent.browser.control` calls carrying `action` reached the
    /// wire. Unanswered calls stay in `frames`, so this counts attempts, not
    /// completions — which is what a duplicate reconciliation looks like.
    fn count_control_calls(
        frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>,
        action: &str,
    ) -> usize {
        frames
            .lock()
            .unwrap()
            .iter()
            .filter_map(|text| serde_json::from_str::<Value>(text).ok())
            .filter(|value| {
                value["method"] == "agent.browser.control" && value["params"]["action"] == action
            })
            .count()
    }

    /// One disconnect, one grace-timer semantics — and teardown that does not
    /// park on a silent process.
    ///
    /// The ordinary drawer is BOTH the control holder and a subscriber, so
    /// disconnect teardown used to run `note_disconnect` AND
    /// `note_watcher_disconnect` against the same view. Each relays its own
    /// `holder_disconnected` and arms a grace timer — one with
    /// `require_unwatched`, one without — and each bumps the generation that
    /// retires the other's. Which semantics survived was decided by which
    /// relay reply landed last. `note_watcher_disconnect`'s `holder.is_some()`
    /// guard cannot catch this: `note_disconnect` clears the holder before it
    /// runs.
    ///
    /// Driven on the relay path because that is where the duplicate is
    /// observable — a second reconciliation is a second call on the wire — and
    /// where the second one was still INLINE, so teardown inherited its bound.
    /// Nothing answers here: a silent supervised process is the case both
    /// halves are about.
    #[tokio::test]
    async fn a_holder_that_was_also_watching_reconciles_its_disconnect_exactly_once() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        let taking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-1").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        taking.await.unwrap().expect("take control");

        // The same connection is also the drawer watching this view — the
        // shape that makes `was_watching` and "held control" both true.
        let (host_channel, _host_rx) = capture_channel();
        view.subscribe_for_test("host-1", host_channel).await;
        frames.lock().unwrap().clear();

        tokio::time::timeout(
            Duration::from_secs(2),
            registry.drop_subscriptions_for_client("host-1"),
        )
        .await
        .expect("teardown must not park on a relay call nothing is going to answer");

        // Let the detached reconciliation reach the wire, then leave a
        // duplicate every chance to follow it.
        for _ in 0..200 {
            if count_control_calls(&frames, "holder_disconnected") > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(
            count_control_calls(&frames, "holder_disconnected"),
            1,
            "exactly one path owns a disconnect — two arm two grace timers with different \
             semantics and let the relay decide which one survives"
        );
    }

    /// One disconnect, one timer — but it has to be the timer that carries the
    /// WATCHER semantics, or collapsing the two calls quietly costs a person
    /// their sign-in.
    ///
    /// The ordinary drawer is both the control holder and the only subscriber.
    /// `note_watcher_disconnect` used to run second and arm
    /// `require_unwatched = true` last, so a drawer that came back inside the
    /// window suppressed the expiry — which is the whole point: the person
    /// returning is what says their sign-in window is still theirs. Arming only
    /// the holder variant instead never consults the subscriber set, so at
    /// t=`CONTROL_GRACE` it relays `GraceExpired`, the reducer resolves
    /// `pending_signin` as `signed_in: false`, and the page goes back to the
    /// agent while the person is mid-credential-entry.
    ///
    /// Driven on the relay path so the expiry is observable as a call on the
    /// wire rather than as local state.
    #[tokio::test(start_paused = true)]
    async fn a_drawer_that_returns_inside_the_window_cancels_its_own_grace_expiry() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        // host-1 presses Take control at a credential form.
        let taking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-1").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        taking.await.unwrap().expect("take control");

        // The same connection is the drawer watching it — holder AND watcher,
        // the shape the de-duplication is about.
        let (host_channel, _host_rx) = capture_channel();
        view.subscribe_for_test("host-1", host_channel).await;

        // Its daemon connection blips.
        registry.drop_subscriptions_for_client("host-1").await;

        // CarHost reconnects inside the window and re-subscribes. The person is
        // back at the form.
        let (again, _again_rx) = capture_channel();
        view.subscribe_for_test("host-2", again).await;

        frames.lock().unwrap().clear();
        tokio::time::sleep(crate::browser_view::CONTROL_GRACE + Duration::from_secs(1)).await;

        assert_eq!(
            count_control_calls(&frames, "grace_expired"),
            0,
            "a drawer watching inside the window is the person coming back — expiring under \
             them resolves their pending sign-in as failed and hands the page to the agent"
        );
    }

    /// The id of the first reverse call carrying `action`, left PENDING.
    ///
    /// Unlike `park_next_call` this does not forget the request, so the test
    /// can answer it later — after something else has landed in between.
    async fn pending_call_id(
        frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>,
        action: &str,
    ) -> String {
        for _ in 0..200 {
            let found = frames
                .lock()
                .unwrap()
                .iter()
                .filter_map(|text| serde_json::from_str::<Value>(text).ok())
                .find(|value| {
                    value["method"] == "agent.browser.control"
                        && value["params"]["action"] == action
                        && value.get("id").and_then(Value::as_str).is_some()
                })
                .map(|value| value["id"].as_str().unwrap().to_string());
            if let Some(id) = found {
                return id;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        panic!("no '{action}' reverse call arrived on the agent channel");
    }

    /// Resolve one specific parked reverse call by request id.
    async fn answer_call_by_id(channel: &Arc<WsChannel>, id: &str, result: Value) {
        let waiter = channel
            .pending
            .lock()
            .await
            .remove(id)
            .expect("the parked call is still pending");
        let _ = waiter.send(car_proto::ToolExecuteResponse {
            action_id: id.to_string(),
            output: Some(result),
            error: None,
            terminal: false,
        });
    }

    /// The slow-but-alive agent process: its `holder_disconnected` reply lands
    /// AFTER a legitimate `take_control`, and asks for a grace period.
    ///
    /// `note_disconnect` arms the clock synchronously, before relaying, so its
    /// generation precedes any re-take. That only holds if the disconnect arms
    /// EXACTLY ONCE — a second arming from this detached reply would capture
    /// the re-taker's generation and leave nothing but the holder check between
    /// host-2 and having control revoked under them. Nothing covered that
    /// before: every other disconnect test leaves the relayed call parked
    /// forever, so the reducer never returns effects and this path never runs.
    #[tokio::test(start_paused = true)]
    async fn an_answered_holder_disconnect_does_not_re_arm_over_a_landed_take_control() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        let taking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-1").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        taking.await.unwrap().expect("take control");

        // host-1 drops. Teardown arms the clock and detaches reconciliation, so
        // this returns without waiting for the process.
        view.note_disconnect("host-1", false).await;
        let disconnect_id = pending_call_id(&frames, "holder_disconnected").await;
        // Cleared so `answer_next_call` below cannot answer the disconnect by
        // mistake; the request stays pending on the channel.
        frames.lock().unwrap().clear();

        // host-2 legitimately takes control inside the window.
        let retaking = {
            let view = Arc::clone(&view);
            tokio::spawn(async move { view.take_control_for_test("host-2").await })
        };
        answer_next_call(
            &producer.channel,
            &frames,
            json!({ "presentation": presentation(WireOwner::User, "https://x.test/"), "effects": [] }),
        )
        .await;
        retaking.await.unwrap().expect("re-take control");
        assert_eq!(
            view.control_holder_for_test().await.as_deref(),
            Some("host-2")
        );
        let generation_after_retake = view.grace_generation_for_test().await;

        // Only NOW does the disconnect's reply arrive, asking for the clock.
        answer_call_by_id(
            &producer.channel,
            &disconnect_id,
            json!({
                "presentation": presentation(WireOwner::User, "https://x.test/"),
                "effects": [{ "effect": "start_grace_period" }],
            }),
        )
        .await;
        for _ in 0..200 {
            if view.grace_generation_for_test().await != generation_after_retake {
                break;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        assert_eq!(
            view.grace_generation_for_test().await,
            generation_after_retake,
            "the disconnect already armed its clock before relaying; re-arming here would \
             capture host-2's generation and disarm the stale-expiry check"
        );

        tokio::time::sleep(crate::browser_view::CONTROL_GRACE + Duration::from_secs(1)).await;
        assert_eq!(
            view.control_holder_for_test().await.as_deref(),
            Some("host-2"),
            "a holder who took control legitimately must survive a timer armed for the \
             connection they replaced, however late that connection's process answers"
        );
    }

    /// The cache the INPUT GATE reads had no monotonicity guard, unlike both
    /// its siblings. A `browser.producer.register` carrying a snapshot taken
    /// before its round trip could therefore rewind a newer push that landed
    /// during it — and a user who had just taken control would have `owner`
    /// read back as `Agent`, so every click came back "the agent holds
    /// control of this browser".
    #[tokio::test]
    async fn an_older_presentation_never_rewinds_the_cache_the_input_gate_reads() {
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);

        let mut taken = presentation(WireOwner::User, "https://x.test/");
        taken.revision = 9;
        producer.push_presentation(taken).await;
        assert_eq!(producer.control_status().await.owner, ControlOwner::User);

        // The in-flight register's pre-take snapshot lands afterwards.
        let mut stale = presentation(WireOwner::Agent, "https://x.test/");
        stale.revision = 8;
        producer.set_presentation(stale).await;

        assert_eq!(
            producer.control_status().await.owner,
            ControlOwner::User,
            "the person still holds control, so their input must still be admitted"
        );

        // A genuinely newer one still lands.
        let mut newer = presentation(WireOwner::Agent, "https://x.test/");
        newer.revision = 10;
        producer.push_presentation(newer).await;
        assert_eq!(producer.control_status().await.owner, ControlOwner::Agent);
    }

    /// The count and the signal were two unsynchronized steps, so an
    /// unsubscribe/resubscribe pair could settle as
    /// stop(1→0) … start(0→1, send true) … stop's send(false): the agent
    /// process told to stop capturing while a view still had a live
    /// subscriber and believed capture was on, which `ensure_streamer` then
    /// never restarts. Publishing the DERIVED state under the count's own
    /// lock makes the settled signal a function of the settled count,
    /// whatever order the two calls land in.
    #[tokio::test]
    async fn the_capture_signal_always_matches_the_settled_watcher_count() {
        let (channel, _frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let mut capture = producer.capture.subscribe();

        // The interleaving, in the order that used to lose: the LAST call to
        // land is the decrement, but the count it settles on is 1.
        producer.start_capture(); // T2's subscribe wins the race …
        producer.stop_capture(); // … and T1's stale stop lands after it.
        assert!(
            !*capture.borrow_and_update(),
            "count is 0, so capture is off"
        );

        producer.start_capture();
        assert!(*capture.borrow_and_update());
        producer.start_capture();
        producer.stop_capture();
        assert!(
            *capture.borrow_and_update(),
            "one watcher remains, so the process must still be capturing"
        );
        producer.stop_capture();
        assert!(!*capture.borrow_and_update());
    }

    #[tokio::test]
    async fn capture_is_asked_for_only_while_somebody_is_watching() {
        let (channel, frames) = agent_channel();
        let producer = RelayProducer::new("agent-conn".into(), "car-assistant".into(), channel);
        let mut capture = producer.capture.subscribe();

        producer.start_capture();
        capture
            .changed()
            .await
            .expect("the first watcher changes capture");
        capture.borrow_and_update();
        let request = answer_next_call(&producer.channel, &frames, json!({ "ok": true })).await;
        assert_eq!(request["method"], "agent.browser.capture");
        assert_eq!(request["params"]["enabled"], true);

        // A second watcher does not ask again. Two assertions, deliberately:
        // the watch channel says no signal was PUBLISHED, and `frames` says
        // nothing reached the WIRE. The second is the one the agent process
        // actually experiences, and a regression that published nothing while
        // still relaying a call would pass the first alone.
        producer.start_capture();
        assert!(
            !capture.has_changed().expect("capture sender remains live"),
            "capture is a producer-level state, not a per-subscriber one; no signal was published"
        );
        tokio::time::sleep(Duration::from_millis(30)).await;
        assert!(
            frames.lock().unwrap().is_empty(),
            "capture is a producer-level state, not a per-subscriber one"
        );

        producer.stop_capture();
        assert!(
            !capture.has_changed().expect("capture sender remains live"),
            "one watcher left, one remains — no stop signal was published"
        );
        tokio::time::sleep(Duration::from_millis(30)).await;
        assert!(
            frames.lock().unwrap().is_empty(),
            "one watcher left, one remains — the process keeps capturing"
        );

        producer.stop_capture();
        capture
            .changed()
            .await
            .expect("the final watcher changes capture");
        assert!(!*capture.borrow_and_update());
        let request = answer_next_call(&producer.channel, &frames, json!({ "ok": true })).await;
        assert_eq!(request["params"]["enabled"], false);
    }

    /// A channel whose write half is CLOSED — the socket a wedged or dead
    /// process leaves behind. Every send fails immediately, which is the fast
    /// stand-in for the 30s timeout the same code path takes against a process
    /// that is merely silent.
    fn broken_channel() -> Arc<WsChannel> {
        use futures::sink::SinkExt as _;
        let (tx, rx) = futures::channel::mpsc::channel::<Message>(1);
        drop(rx);
        let sink: WsSink =
            Box::pin(tx.sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed));
        Arc::new(WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            active_actions: Mutex::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(0),
        })
    }

    /// FINDING 1. A transition that never reached the process did not happen.
    /// Reporting success would leave the daemon believing the host is driving
    /// a browser whose own reducer still says the agent is — every later input
    /// refused, and a person told the blackout is up when the process never
    /// entered it.
    #[tokio::test]
    async fn a_control_transition_that_never_reached_the_process_fails_and_changes_nothing() {
        let (state, _temp) = test_state().await;
        let (host, _rx) = host_session(&state, "host-1").await;
        let producer = state
            .browser_views
            .producer_for("agent-conn", "car-assistant", &broken_channel())
            .await;
        producer
            .set_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;
        let view = state
            .browser_views
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        let err = crate::browser_view::handle_take_control(
            &request(
                "browser.view.take_control",
                json!({ "conversation_id": "conv-1" }),
            ),
            &host,
            &state,
        )
        .await
        .expect_err("a transition that did not land must not report success");
        assert!(err.contains("unreachable"), "got: {err}");

        // The daemon did NOT record the host as the control holder, and the
        // view still says what the process says: the agent is driving.
        assert_eq!(
            view.snapshot_for_test().await.0.owner,
            WireOwner::Agent,
            "the drawer must not be told the user took control of a browser that never heard"
        );
        let err = crate::browser_view::handle_input(
            crate::browser_view::InputOp::Click,
            &request(
                "browser.view.click",
                json!({ "conversation_id": "conv-1", "x": 1.0, "y": 2.0 }),
            ),
            &host,
            &state,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("take_control"),
            "both sides agree the agent still holds it; got: {err}"
        );
    }

    /// FINDING 4, daemon half. `chat_sessions` is per TURN, but a published
    /// browser outlives its run. A process whose daemon session dropped
    /// BETWEEN turns must be able to republish its own conversation — without
    /// that, the drawer sits on "its browser is gone" pointing at a live
    /// browser until the user happens to send another message.
    #[tokio::test]
    async fn a_reconnecting_process_republishes_between_turns_without_a_new_user_turn() {
        let (state, _temp) = test_state().await;
        let (agent, _frames) = agent_session(&state, "conn-1", "car-assistant", "conv-1").await;
        let (host, mut host_rx) = host_session(&state, "host-1").await;

        handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1",
                        "presentation": presentation(WireOwner::Agent, "https://x.test/") }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();
        let subscribed = crate::browser_view::handle_subscribe(
            &request(
                "browser.view.subscribe",
                json!({ "conversation_id": "conv-1" }),
            ),
            &host,
            &state,
        )
        .await
        .unwrap();
        let before = subscribed["cursor"].as_u64().unwrap();

        // The turn ends — the daemon drops the chat-session routing entry —
        // and only THEN does the process's session drop.
        state.chat_sessions.lock().await.remove("conv-1");
        state.remove_session("conn-1").await;
        assert_eq!(
            state
                .browser_views
                .get(Some("conv-1"))
                .await
                .unwrap()
                .snapshot_for_test()
                .await
                .0
                .owner,
            WireOwner::None,
            "the view reports the browser as gone while the process is away"
        );

        // The process reconnects on a NEW connection and republishes on its
        // own, with no chat session anywhere.
        let (channel, _frames) = agent_channel();
        let reconnected = state.create_session("conn-2", channel).await.unwrap();
        *reconnected.agent_id.lock().await = Some("car-assistant".to_string());
        handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1",
                        "presentation": presentation(WireOwner::Agent, "https://back.test/") }),
            ),
            &reconnected,
            &state,
        )
        .await
        .expect("the agent that established this conversation may republish it");

        let view = state.browser_views.get(Some("conv-1")).await.unwrap();
        assert_eq!(
            view.snapshot_for_test().await.0.url.as_deref(),
            Some("https://back.test/")
        );
        assert_eq!(
            view.subscriber_count_for_test().await,
            1,
            "the drawer came across without re-subscribing"
        );
        let mut seen = next_event(&mut host_rx).await;
        while seen.cursor <= before {
            seen = next_event(&mut host_rx).await;
        }
        assert!(seen.cursor > before, "the cursor never moves backwards");
    }

    /// FINDING 4, the authorization half: the relaxation is about LIVENESS,
    /// not about who is entitled. A conversation nobody served, and one served
    /// by somebody else, are refused exactly as before.
    #[tokio::test]
    async fn republishing_is_still_refused_to_an_agent_that_never_served_the_conversation() {
        let (state, _temp) = test_state().await;
        let (agent, _frames) = agent_session(&state, "conn-1", "car-assistant", "conv-1").await;
        handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();
        state.chat_sessions.lock().await.remove("conv-1");

        // A different agent, with the turn long over, cannot take the drawer.
        let (channel, _frames) = agent_channel();
        let impostor = state.create_session("conn-x", channel).await.unwrap();
        *impostor.agent_id.lock().await = Some("some-other-agent".to_string());
        let err = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &impostor,
            &state,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("is served by agent 'car-assistant'"),
            "got: {err}"
        );

        // And a conversation nobody ever served is still not claimable.
        let err = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "never-seen" }),
            ),
            &impostor,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("not an active chat session"), "got: {err}");
    }

    #[test]
    fn a_conversation_claim_needs_a_live_turn_or_a_binding_this_agent_established() {
        // First registration: entitled by the live turn.
        assert!(
            authorize_conversation_claim("c", "a", Some("a"), None).is_ok(),
            "the agent the daemon dispatched the turn to"
        );
        // Re-registration between turns: entitled by its own binding.
        assert!(authorize_conversation_claim("c", "a", None, Some("a")).is_ok());
        // A live turn for somebody else beats a stale binding.
        assert!(authorize_conversation_claim("c", "a", Some("b"), Some("a")).is_err());
        // Somebody else's binding, no live turn.
        assert!(authorize_conversation_claim("c", "a", None, Some("b")).is_err());
        // Never served by anyone.
        assert!(authorize_conversation_claim("c", "a", None, None).is_err());
    }

    // ---- the codec ------------------------------------------------------

    #[test]
    fn every_input_round_trips_through_the_wire() {
        for input in [
            ViewInput::Navigate {
                url: "https://x.test/".into(),
            },
            ViewInput::Click { x: 1.5, y: 2.5 },
            ViewInput::Type {
                text: "hello".into(),
            },
            ViewInput::Keypress {
                key: "Enter".into(),
                modifiers: vec![Modifier::Meta, Modifier::Shift],
            },
            ViewInput::Scroll { delta_y: -120 },
            ViewInput::Paste {
                text: "pasted".into(),
            },
            ViewInput::Back,
            ViewInput::Forward,
            ViewInput::Reload,
            ViewInput::TabOpen,
            ViewInput::TabClose {
                tab_id: "tab-2".into(),
            },
            ViewInput::TabSwitch {
                tab_id: "tab-3".into(),
            },
        ] {
            let wire = input_to_wire(&input);
            assert_eq!(
                input_from_wire(&wire).expect("decodes"),
                input,
                "round trip failed for {wire}"
            );
        }
    }

    #[test]
    fn every_control_action_and_effect_round_trips() {
        for control in [
            ViewControl::TakeControl,
            ViewControl::HandBack,
            ViewControl::RunEnded,
            ViewControl::HolderDisconnected,
            ViewControl::GraceExpired,
        ] {
            assert_eq!(
                control_from_wire(control_to_wire(control)).unwrap(),
                control
            );
        }
        let effects = vec![
            ControlEffect::StartGracePeriod,
            ControlEffect::SignInResolved { signed_in: true },
        ];
        assert_eq!(effects_from_wire(&effects_to_wire(&effects)), effects);
    }

    #[test]
    fn an_unknown_effect_is_dropped_rather_than_failing_the_transition() {
        let wire = json!([{ "effect": "teleport" }, { "effect": "start_grace_period" }]);
        assert_eq!(
            effects_from_wire(&wire),
            vec![ControlEffect::StartGracePeriod]
        );
    }

    #[test]
    fn an_unknown_input_op_is_a_clean_error() {
        let err = input_from_wire(&json!({ "op": "read_dom" })).unwrap_err();
        assert!(err.contains("unknown agent.browser.input op"), "got: {err}");
    }

    // ---- registry -------------------------------------------------------

    #[tokio::test]
    async fn a_registered_conversation_resolves_to_the_process_s_browser() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        producer
            .set_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;

        let found = registry.get(Some("conv-1")).await.expect("registered");
        assert!(Arc::ptr_eq(&view, &found));
        let (snapshot, _) = found.snapshot_for_test().await;
        assert_eq!(snapshot.url.as_deref(), Some("https://x.test/"));
        assert_eq!(snapshot.owner, WireOwner::Agent);
    }

    #[tokio::test]
    async fn re_registering_the_same_conversation_is_a_no_op_for_the_drawer() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        let first = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        // Every turn registers again; the drawer must not be churned for it.
        let second = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        assert!(
            Arc::ptr_eq(&first, &second),
            "the same process re-claiming its own conversation keeps the view"
        );
    }

    #[tokio::test]
    async fn one_process_backs_every_conversation_it_registers() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        registry
            .register_relay("conv-2", Arc::clone(&producer))
            .await;

        producer
            .push_presentation(presentation(WireOwner::Agent, "https://shared.test/"))
            .await;

        for key in ["conv-1", "conv-2"] {
            let view = registry.get(Some(key)).await.expect("registered");
            assert_eq!(
                view.snapshot_for_test().await.0.url.as_deref(),
                Some("https://shared.test/"),
                "a supervised process has ONE browser; both of its conversations show it"
            );
        }
    }

    /// The isolation the outcomes ask for: two conversations served by two
    /// DIFFERENT agents are two processes with two browsers, and neither
    /// drawer ever sees the other's page or frames.
    #[tokio::test]
    async fn two_processes_two_conversations_never_see_each_other() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel_a, _frames_a) = agent_channel();
        let (channel_b, _frames_b) = agent_channel();
        let alpha = registry
            .producer_for("conn-a", "agent-alpha", &channel_a)
            .await;
        let beta = registry
            .producer_for("conn-b", "agent-beta", &channel_b)
            .await;
        let view_a = registry.register_relay("conv-a", Arc::clone(&alpha)).await;
        let view_b = registry.register_relay("conv-b", Arc::clone(&beta)).await;

        let (host_a, mut rx_a) = capture_channel();
        let (host_b, mut rx_b) = capture_channel();
        view_a.subscribe_for_test("host-1", host_a).await;
        view_b.subscribe_for_test("host-1", host_b).await;

        alpha
            .push_presentation(presentation(WireOwner::Agent, "https://alpha.test/"))
            .await;
        beta.push_presentation(presentation(WireOwner::User, "https://beta.test/"))
            .await;
        alpha
            .push_frame(WireFrame {
                jpeg_base64: "QQ==".into(),
                width: 800,
                height: 600,
                device_pixel_ratio: 1.0,
                captured_at: 0.0,
            })
            .await;

        assert_eq!(
            view_a.snapshot_for_test().await.0.url.as_deref(),
            Some("https://alpha.test/")
        );
        assert_eq!(
            view_b.snapshot_for_test().await.0.url.as_deref(),
            Some("https://beta.test/")
        );

        // Alpha's drawer saw alpha's page and alpha's frame, in that order.
        match next_event(&mut rx_a).await.payload {
            crate::browser_view::BrowserViewPayload::Presentation { presentation } => {
                assert_eq!(presentation.url.as_deref(), Some("https://alpha.test/"));
            }
            crate::browser_view::BrowserViewPayload::Frame { .. } => {
                panic!("expected alpha's presentation first")
            }
        }
        match next_event(&mut rx_a).await.payload {
            crate::browser_view::BrowserViewPayload::Frame { frame } => {
                assert_eq!(frame.jpeg_base64, "QQ==");
            }
            crate::browser_view::BrowserViewPayload::Presentation { .. } => {
                panic!("expected alpha's frame")
            }
        }
        // Beta's drawer saw exactly one event — its own presentation. Alpha's
        // frame never reached it.
        match next_event(&mut rx_b).await.payload {
            crate::browser_view::BrowserViewPayload::Presentation { presentation } => {
                assert_eq!(presentation.url.as_deref(), Some("https://beta.test/"));
                assert_eq!(presentation.owner, WireOwner::User);
            }
            crate::browser_view::BrowserViewPayload::Frame { .. } => {
                panic!("beta's drawer must never receive alpha's frame")
            }
        }
        assert!(
            tokio::time::timeout(Duration::from_millis(100), rx_b.next())
                .await
                .is_err(),
            "nothing else crossed between the two conversations"
        );
    }

    #[tokio::test]
    async fn a_restarted_process_replaces_the_view_and_carries_the_drawer_across() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (first_channel, _first_frames) = agent_channel();
        let first = registry
            .producer_for("agent-conn-1", "car-assistant", &first_channel)
            .await;
        let view = registry.register_relay("conv-1", Arc::clone(&first)).await;
        let (host, mut host_rx) = capture_channel();
        view.subscribe_for_test("host-1", host).await;
        first
            .push_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;
        let before = next_event(&mut host_rx).await.cursor;

        // The process dies and the supervisor restarts it: a NEW connection,
        // so a new producer, claiming the same conversation.
        registry.note_producer_disconnected("agent-conn-1").await;
        let (second_channel, _second_frames) = agent_channel();
        let second = registry
            .producer_for("agent-conn-2", "car-assistant", &second_channel)
            .await;
        second
            .set_presentation(presentation(
                WireOwner::Agent,
                "https://after-restart.test/",
            ))
            .await;
        let replacement = registry.register_relay("conv-1", Arc::clone(&second)).await;

        assert!(
            !Arc::ptr_eq(&view, &replacement),
            "a different process is a different producer, so a different view"
        );
        assert_eq!(
            replacement.subscriber_count_for_test().await,
            1,
            "the drawer came across without re-subscribing"
        );
        // Assert every intermediate cursor instead of fast-forwarding until
        // one happens to exceed `before`.
        let mut previous_cursor = before;
        loop {
            let seen = next_event(&mut host_rx).await;
            assert_eq!(
                seen.cursor,
                previous_cursor + 1,
                "the cursor sequence must be contiguous and monotonic"
            );
            previous_cursor = seen.cursor;
            if matches!(
                seen.payload,
                crate::browser_view::BrowserViewPayload::Presentation { ref presentation }
                    if presentation.url.as_deref() == Some("https://after-restart.test/")
            ) {
                break;
            }
        }
        assert_eq!(
            replacement.snapshot_for_test().await.0.url.as_deref(),
            Some("https://after-restart.test/")
        );
    }

    #[tokio::test]
    async fn a_pushed_frame_reaches_the_drawer_with_the_next_cursor() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        let (host, mut host_rx) = capture_channel();
        let (_, cursor) = view.subscribe_for_test("host-1", host).await;

        producer
            .push_frame(WireFrame {
                jpeg_base64: "AQID".into(),
                width: 1920,
                height: 1080,
                device_pixel_ratio: 2.0,
                captured_at: 1.5,
            })
            .await;

        let event = next_event(&mut host_rx).await;
        assert_eq!(event.cursor, cursor + 1);
        assert_eq!(event.conversation_id.as_deref(), Some("conv-1"));
        match event.payload {
            crate::browser_view::BrowserViewPayload::Frame { frame } => {
                assert_eq!(frame.jpeg_base64, "AQID");
                assert_eq!(frame.width, 1920);
                assert_eq!(frame.device_pixel_ratio, 2.0);
            }
            crate::browser_view::BrowserViewPayload::Presentation { .. } => {
                panic!("expected a frame event")
            }
        }
    }

    /// `agents.chat`'s `session_id` is minted fresh PER TURN, so "one view per
    /// conversation" was in practice one view per turn: a long-lived
    /// `car do --serve` process accumulated a `BrowserView`, a registry entry
    /// and a binding for every turn it had ever served, and every screencast
    /// frame was cloned into all of them. A registration retires the older
    /// ones — replacement, at the same recency the agent side republishes at.
    #[tokio::test]
    async fn a_registration_retires_this_producer_s_oldest_views() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;

        // One turn per key, exactly as `register_conversation` does.
        for turn in 0..MAX_VIEWS_PER_PRODUCER {
            let key = format!("turn-{turn}");
            registry.bind_conversation(&key, "car-assistant").await;
            registry.register_relay(key, Arc::clone(&producer)).await;
        }
        assert!(
            registry.get(Some("turn-0")).await.is_some(),
            "precondition: nothing is retired while the producer is within its cap"
        );

        // One more turn is one too many.
        registry.bind_conversation("turn-8", "car-assistant").await;
        registry
            .register_relay("turn-8", Arc::clone(&producer))
            .await;

        assert!(
            registry.get(Some("turn-0")).await.is_none(),
            "the oldest turn's view must be retired, not accumulated"
        );
        assert!(
            registry.conversation_owner("turn-0").await.is_none(),
            "and its binding with it — nothing can re-register a key past the cap"
        );
        assert!(
            registry.get(Some("turn-1")).await.is_some(),
            "only what is PAST the cap goes"
        );
        assert!(registry.get(Some("turn-8")).await.is_some());
    }

    /// Retiring is not deleting. A drawer still watching an older key keeps it
    /// — `release_if_idle` refuses to drop a view anybody is subscribed to —
    /// which is what preserves the candidate chain's one-turn fallback and the
    /// restart-adoption behaviour.
    #[tokio::test]
    async fn a_retired_view_somebody_is_watching_survives() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;

        let watched = registry
            .register_relay("turn-0", Arc::clone(&producer))
            .await;
        let (host, _rx) = capture_channel();
        watched.subscribe_for_test("host-1", host).await;

        for turn in 1..=MAX_VIEWS_PER_PRODUCER {
            registry
                .register_relay(format!("turn-{turn}"), Arc::clone(&producer))
                .await;
        }

        assert!(
            registry.get(Some("turn-0")).await.is_some(),
            "a view the drawer is subscribed to is never taken out from under it"
        );
    }

    /// A frame is a base64 full-viewport JPEG, so every fan-out clone is a
    /// memcpy of a few hundred KB — and it went to every view the producer had
    /// ever registered, watched or not.
    #[tokio::test]
    async fn a_frame_only_reaches_views_somebody_is_watching() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        let unwatched = registry
            .register_relay("turn-0", Arc::clone(&producer))
            .await;
        let watched = registry
            .register_relay("turn-1", Arc::clone(&producer))
            .await;
        let (host, mut host_rx) = capture_channel();
        let (_snapshot, cursor) = watched.subscribe_for_test("host-1", host).await;
        let (_, unwatched_cursor) = unwatched.snapshot_for_test().await;

        producer
            .push_frame(WireFrame {
                jpeg_base64: "AQID".into(),
                width: 8,
                height: 8,
                device_pixel_ratio: 1.0,
                captured_at: 0.5,
            })
            .await;

        let event = next_event(&mut host_rx).await;
        assert_eq!(event.cursor, cursor + 1, "the watched view is served");
        let (_, after) = unwatched.snapshot_for_test().await;
        assert_eq!(
            after, unwatched_cursor,
            "a view nobody is watching pays nothing — not even a cursor bump"
        );
    }

    /// The restart-adoption guarantee, which is about a view somebody is
    /// WATCHING: it stays registered so a restarted process replaces it in
    /// place and the drawer follows across without re-subscribing.
    #[tokio::test]
    async fn a_disconnected_producer_is_dropped_from_the_registry_and_clears_its_views() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        producer
            .set_presentation(presentation(WireOwner::Agent, "https://x.test/"))
            .await;
        let view = registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        let (host, _rx) = capture_channel();
        view.subscribe_for_test("host-1", host).await;

        registry.note_producer_disconnected("agent-conn").await;

        assert!(registry.producer("agent-conn").await.is_none());
        let view = registry
            .get(Some("conv-1"))
            .await
            .expect("the view stays so a restarted process can replace it");
        let (snapshot, _) = view.snapshot_for_test().await;
        assert_eq!(snapshot.owner, WireOwner::None);
        assert_eq!(snapshot.url, None);
    }

    /// The other half, and the leak. `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. 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.
    ///
    /// Nobody is watching this one, so nothing is owed to a restart.
    #[tokio::test]
    async fn a_disconnected_producer_s_unwatched_views_are_released_with_their_socket() {
        let registry = BrowserViewRegistry::new(std::env::temp_dir());
        let (channel, _frames) = agent_channel();
        let producer = registry
            .producer_for("agent-conn", "car-assistant", &channel)
            .await;
        let weak = Arc::downgrade(&producer);
        registry
            .register_relay("conv-1", Arc::clone(&producer))
            .await;
        drop(producer);

        registry.note_producer_disconnected("agent-conn").await;

        assert!(
            registry.get(Some("conv-1")).await.is_none(),
            "an unwatched view for a process that is gone must not stay registered"
        );
        assert!(
            weak.upgrade().is_none(),
            "and the producer — with the dead connection's WsChannel — must actually be released"
        );
    }

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

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

    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,
        }
    }

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

    /// An attached supervised agent mid-turn on `conversation`: the session
    /// binding `session.auth { agent_id }` makes, plus the `chat_sessions`
    /// routing entry `agents.chat` creates before it dispatches the turn.
    async fn agent_session(
        state: &Arc<ServerState>,
        client_id: &str,
        agent_id: &str,
        conversation: &str,
    ) -> (
        Arc<ClientSession>,
        std::sync::Arc<std::sync::Mutex<Vec<String>>>,
    ) {
        let (channel, frames) = agent_channel();
        let session = state.create_session(client_id, channel).await.unwrap();
        *session.agent_id.lock().await = Some(agent_id.to_string());
        state.chat_sessions.lock().await.insert(
            conversation.to_string(),
            crate::session::ChatSession {
                agent_id: agent_id.to_string(),
                host_client_id: "host-1".to_string(),
                created_at: 0,
                local_cancel: None,
            },
        );
        (session, frames)
    }

    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)
    }

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

        let err = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("not a supervised agent"), "got: {err}");
        assert!(state.browser_views.get(Some("conv-1")).await.is_none());
    }

    #[tokio::test]
    async fn an_agent_cannot_claim_a_conversation_it_is_not_serving() {
        let (state, _temp) = test_state().await;
        let (session, _frames) =
            agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;

        // Someone else's conversation.
        state.chat_sessions.lock().await.insert(
            "conv-other".to_string(),
            crate::session::ChatSession {
                agent_id: "some-other-agent".to_string(),
                host_client_id: "host-1".to_string(),
                created_at: 0,
                local_cancel: None,
            },
        );
        let err = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-other" }),
            ),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("is served by agent 'some-other-agent'"),
            "got: {err}"
        );

        // And a conversation nobody is serving.
        let err = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "ghost" }),
            ),
            &session,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("not an active chat session"), "got: {err}");
        assert!(state.browser_views.get(Some("ghost")).await.is_none());
    }

    /// The whole hop, end to end through the real handlers: a supervised
    /// process registers its browser for the conversation it is serving, a
    /// Command Deck subscribes by that conversation id, the process pushes,
    /// and the drawer receives it.
    #[tokio::test]
    async fn the_drawer_subscribes_by_conversation_and_receives_the_process_s_pushes() {
        let (state, _temp) = test_state().await;
        let (agent, agent_frames) =
            agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;
        let (host, mut host_rx) = host_session(&state, "host-1").await;

        let out = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({
                    "conversation_id": "conv-1",
                    "presentation": presentation(WireOwner::Agent, "https://x.test/"),
                }),
            ),
            &agent,
            &state,
        )
        .await
        .expect("the agent may publish the conversation it serves");
        assert_eq!(out["ok"], true);

        // The drawer's own surface, unchanged, now reaches into the process.
        let snapshot = crate::browser_view::handle_subscribe(
            &request(
                "browser.view.subscribe",
                json!({ "conversation_id": "conv-1" }),
            ),
            &host,
            &state,
        )
        .await
        .expect("a supervised agent's browser is subscribable by conversation");
        assert_eq!(snapshot["standing_session"], false);
        assert_eq!(snapshot["presentation"]["url"], "https://x.test/");
        assert_eq!(snapshot["presentation"]["owner"], "agent");
        let cursor = snapshot["cursor"].as_u64().unwrap();

        // Somebody is watching now — and only now does the process get asked
        // to capture, so a browser nobody has open pays for no screencast and
        // no WS traffic.
        let capture = answer_next_call(&agent.channel, &agent_frames, json!({ "ok": true })).await;
        assert_eq!(capture["method"], "agent.browser.capture");
        assert_eq!(capture["params"]["enabled"], true);

        // The process pushes a presentation delta and a frame.
        assert!(
            try_handle_producer_push(
                &notification(
                    "browser.producer.presentation",
                    json!({ "presentation": presentation(WireOwner::Agent, "https://moved.test/") }),
                ),
                &state,
                &agent,
            )
            .await
        );
        let event = next_event(&mut host_rx).await;
        assert_eq!(event.cursor, cursor + 1);
        match event.payload {
            crate::browser_view::BrowserViewPayload::Presentation { presentation } => {
                assert_eq!(presentation.url.as_deref(), Some("https://moved.test/"));
            }
            crate::browser_view::BrowserViewPayload::Frame { .. } => {
                panic!("expected a presentation event")
            }
        }
    }

    /// Task 7: the supervised process has no direct read of the daemon's
    /// session set, so `browser.producer.register`'s acknowledgment is the
    /// channel it learns "is a host connected" through. This binds that this
    /// existing round trip actually carries the real, live answer both ways
    /// — not a stale or hardcoded one.
    #[tokio::test]
    async fn producer_register_reports_whether_a_host_is_currently_connected() {
        let (state, _temp) = test_state().await;
        let (agent, _frames) = agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;

        // No host connected yet.
        let out = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();
        assert_eq!(out["host_connected"], false);

        // A host connects — the DAEMON's own answer changes immediately
        // (this is `handle_producer_register`, driven directly; the process
        // side's own freshness bound — refreshed only when it actually
        // calls this again — is documented and tested separately in
        // `assistant::browser_producer`).
        let (_host, _host_rx) = host_session(&state, "host-1").await;
        let out = handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();
        assert_eq!(out["host_connected"], true);
    }

    /// How many `agent.browser.host_connected` pushes reached the wire, and
    /// what the last one said.
    fn host_connected_calls(
        frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>,
    ) -> (usize, Option<Value>) {
        let seen: Vec<Value> = frames
            .lock()
            .unwrap()
            .iter()
            .filter_map(|text| serde_json::from_str::<Value>(text).ok())
            .filter(|value| value["method"] == "agent.browser.host_connected")
            .collect();
        let last = seen
            .last()
            .map(|value| value["params"]["connected"].clone());
        (seen.len(), last)
    }

    /// `remove_session` fanned a `host_connected` reverse call to EVERY
    /// producer on every disconnect — agent and CLI connections included,
    /// none of which can change host connectivity. Gating that on the removed
    /// session having actually held the host role is only correct if a real
    /// host removal still broadcasts, so both halves are asserted here.
    #[tokio::test]
    async fn only_a_host_removal_tells_the_producers_that_host_connectivity_changed() {
        let (state, _temp) = test_state().await;
        let (agent, frames) = agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;
        handle_producer_register(
            &request(
                "browser.producer.register",
                json!({ "conversation_id": "conv-1" }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();

        let (_host, _host_rx) = host_session(&state, "host-1").await;
        let (other_channel, _other_rx) = capture_channel();
        state
            .create_session("other-1", other_channel)
            .await
            .unwrap();
        frames.lock().unwrap().clear();

        // A non-host connection going away cannot have changed the answer.
        state
            .remove_session("other-1")
            .await
            .expect("the non-host session was registered");
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            host_connected_calls(&frames).0,
            0,
            "a non-host disconnect must not fan a reverse call to every producer"
        );

        // The last host going away does — and carries the post-removal truth.
        state
            .remove_session("host-1")
            .await
            .expect("the host session was registered");
        for _ in 0..200 {
            if host_connected_calls(&frames).0 > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        let (count, connected) = host_connected_calls(&frames);
        assert_eq!(
            count, 1,
            "removing the last host must still tell the producers"
        );
        assert_eq!(
            connected,
            Some(json!(false)),
            "the session is already out of `sessions`, so the broadcast reads the \
             post-removal truth"
        );
    }

    #[tokio::test]
    async fn a_push_from_a_connection_with_no_producer_is_consumed_and_dropped() {
        let (state, _temp) = test_state().await;
        let (agent, _frames) = agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;

        // Recognized (so the dispatcher does not answer a notification with
        // method-not-found) but nothing happens: no producer, no view.
        assert!(
            try_handle_producer_push(
                &notification(
                    "browser.producer.frame",
                    json!({ "frame": { "jpeg_base64": "AQ==", "width": 1, "height": 1,
                                       "device_pixel_ratio": 1.0, "captured_at": 0.0 } }),
                ),
                &state,
                &agent,
            )
            .await
        );
        assert!(state.browser_views.get(Some("conv-1")).await.is_none());
    }

    #[tokio::test]
    async fn a_producer_request_with_an_id_is_left_to_the_dispatcher() {
        let (state, _temp) = test_state().await;
        let (agent, _frames) = agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;
        assert!(
            !try_handle_producer_push(
                &request("browser.producer.presentation", json!({})),
                &state,
                &agent,
            )
            .await,
            "a frame with an id is a request; it must get a real reply, not be swallowed"
        );
    }

    /// The drawer's input path, on a relayed browser: the control gate is
    /// applied in the daemon against the pushed presentation, and only then
    /// does the call cross to the process.
    #[tokio::test]
    async fn input_reaches_the_process_only_after_the_control_gate_passes() {
        let (state, _temp) = test_state().await;
        let (agent, agent_frames) =
            agent_session(&state, "agent-conn", "car-assistant", "conv-1").await;
        let (host, _host_rx) = host_session(&state, "host-1").await;

        handle_producer_register(
            &request(
                "browser.producer.register",
                json!({
                    "conversation_id": "conv-1",
                    "presentation": presentation(WireOwner::Agent, "https://x.test/"),
                }),
            ),
            &agent,
            &state,
        )
        .await
        .unwrap();

        // The agent is driving: refused in the daemon, nothing crosses.
        let click = request(
            "browser.view.click",
            json!({ "conversation_id": "conv-1", "x": 4.0, "y": 5.0 }),
        );
        let err = crate::browser_view::handle_input(
            crate::browser_view::InputOp::Click,
            &click,
            &host,
            &state,
        )
        .await
        .unwrap_err();
        assert!(err.contains("take_control"), "got: {err}");
        assert!(
            agent_frames.lock().unwrap().is_empty(),
            "a refused input must never reach the agent process"
        );

        // Take control — which itself is a relayed transition — then click.
        let taking = tokio::spawn({
            let state = Arc::clone(&state);
            let host = Arc::clone(&host);
            async move {
                crate::browser_view::handle_take_control(
                    &request(
                        "browser.view.take_control",
                        json!({ "conversation_id": "conv-1" }),
                    ),
                    &host,
                    &state,
                )
                .await
            }
        });
        let request_frame = answer_next_call(
            &agent.channel,
            &agent_frames,
            json!({
                "presentation": presentation(WireOwner::User, "https://x.test/"),
                "effects": [],
            }),
        )
        .await;
        assert_eq!(request_frame["method"], "agent.browser.control");
        assert_eq!(request_frame["params"]["action"], "take_control");
        assert_eq!(
            taking.await.unwrap().unwrap()["presentation"]["owner"],
            "user"
        );

        let clicking = tokio::spawn({
            let state = Arc::clone(&state);
            let host = Arc::clone(&host);
            async move {
                crate::browser_view::handle_input(
                    crate::browser_view::InputOp::Click,
                    &click,
                    &host,
                    &state,
                )
                .await
            }
        });
        let request_frame = answer_next_call(&agent.channel, &agent_frames, json!({})).await;
        assert_eq!(request_frame["method"], "agent.browser.input");
        assert_eq!(request_frame["params"]["op"], "click");
        assert_eq!(request_frame["params"]["x"], 4.0);
        assert_eq!(clicking.await.unwrap().unwrap()["ok"], true);
    }
}